Skip to main content

vortex_array/arrays/varbin/
builder.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::any::Any;
5use std::mem::MaybeUninit;
6
7use num_traits::AsPrimitive;
8use vortex_buffer::Alignment;
9use vortex_buffer::BitBufferMut;
10use vortex_buffer::BufferAllocatorRef;
11use vortex_buffer::BufferMut;
12use vortex_buffer::ByteBuffer;
13use vortex_buffer::ByteBufferMut;
14use vortex_error::VortexExpect;
15use vortex_error::VortexResult;
16use vortex_error::vortex_bail;
17use vortex_error::vortex_ensure;
18use vortex_error::vortex_ensure_eq;
19use vortex_error::vortex_panic;
20use vortex_mask::AllOr;
21use vortex_mask::Mask;
22
23use crate::ArrayRef;
24use crate::ArrayView;
25use crate::Canonical;
26use crate::ExecutionCtx;
27use crate::IntoArray;
28#[cfg(debug_assertions)]
29use crate::VortexSessionExecute;
30use crate::arrays::PrimitiveArray;
31use crate::arrays::VarBin;
32use crate::arrays::VarBinArray;
33use crate::arrays::VarBinView;
34use crate::arrays::varbin::VarBinArrayExt;
35use crate::arrays::varbin::VarBinArraySlotsExt;
36use crate::arrays::varbinview::VarBinViewArrayExt;
37use crate::builders::ArrayBuilder;
38use crate::dtype::DType;
39use crate::dtype::OffsetBuilderPType;
40use crate::expr::stats::Precision;
41use crate::expr::stats::Stat;
42#[cfg(debug_assertions)]
43use crate::legacy_session;
44use crate::match_each_integer_ptype;
45use crate::scalar::Scalar;
46use crate::validity::Validity;
47
48/// Builder for [`VarBinArray`] values with `O`-typed offsets.
49///
50/// This is the offset-based counterpart to
51/// [`VarBinViewBuilder`](crate::builders::VarBinViewBuilder): values are laid out contiguously in
52/// a single byte buffer described by a monotonically increasing offsets buffer. An `i32` or `i64`
53/// builder can therefore be handed straight to Arrow as a `Utf8`/`LargeUtf8`/`Binary`/
54/// `LargeBinary` array without re-laying out the bytes.
55///
56/// Encodings that can decode into this layout should specialize
57/// [`append_to_builder`](crate::vtable::VTable::append_to_builder) with
58/// [`match_each_varbin_builder!`](crate::match_each_varbin_builder), which recovers the concrete
59/// offset type from a `&mut dyn ArrayBuilder` so the decode loop is monomorphized over it.
60pub struct VarBinBuilder<O: OffsetBuilderPType> {
61    dtype: DType,
62    offsets: BufferMut<O>,
63    data: ByteBufferMut,
64    validity: BitBufferMut,
65}
66
67impl<O: OffsetBuilderPType> VarBinBuilder<O> {
68    /// Creates an empty builder for `dtype`.
69    #[deprecated(note = "use `new_in` with an explicit allocator")]
70    pub fn new(dtype: DType) -> Self {
71        Self::new_in(dtype, BufferAllocatorRef::static_ref())
72    }
73
74    /// Creates an empty builder for `dtype` using `allocator`.
75    pub fn new_in(dtype: DType, allocator: &BufferAllocatorRef) -> Self {
76        Self::with_capacity_in(dtype, 0, allocator)
77    }
78
79    /// Creates a builder for `dtype` with room for `capacity` values.
80    #[deprecated(note = "use `with_capacity_in` with an explicit allocator")]
81    pub fn with_capacity(dtype: DType, capacity: usize) -> Self {
82        Self::with_capacity_in(dtype, capacity, BufferAllocatorRef::static_ref())
83    }
84
85    /// Creates a builder for `dtype` with room for `capacity` values using `allocator`.
86    pub fn with_capacity_in(dtype: DType, capacity: usize, allocator: &BufferAllocatorRef) -> Self {
87        assert!(
88            matches!(dtype, DType::Utf8(_) | DType::Binary(_)),
89            "VarBinBuilder dtype must be Utf8 or Binary, got {dtype}"
90        );
91        let mut offsets = BufferMut::with_capacity_in(capacity + 1, allocator.clone());
92        offsets.push(O::zero());
93        Self {
94            dtype,
95            offsets,
96            data: BufferMut::empty_aligned_in(Alignment::of::<u8>(), allocator.clone()),
97            validity: BitBufferMut::with_capacity_in(capacity, allocator.clone()),
98        }
99    }
100
101    /// Creates a builder for `dtype` with room for `capacity` values totalling `bytes` bytes.
102    #[deprecated(note = "use `with_capacity_bytes_in` with an explicit allocator")]
103    pub fn with_capacity_bytes(dtype: DType, capacity: usize, bytes: usize) -> Self {
104        Self::with_capacity_bytes_in(dtype, capacity, bytes, BufferAllocatorRef::static_ref())
105    }
106
107    /// Creates a builder using `allocator` with room for `capacity` values and `bytes` bytes.
108    pub fn with_capacity_bytes_in(
109        dtype: DType,
110        capacity: usize,
111        bytes: usize,
112        allocator: &BufferAllocatorRef,
113    ) -> Self {
114        let mut builder = Self::with_capacity_in(dtype, capacity, allocator);
115        builder.reserve_data(bytes);
116        builder
117    }
118
119    /// Reserves room for `additional` value bytes.
120    ///
121    /// [`reserve_exact`](ArrayBuilder::reserve_exact) takes a row count and so cannot size the
122    /// value bytes; callers that know the byte total should use this to size the buffer once
123    /// instead of letting it grow.
124    pub fn reserve_data(&mut self, additional: usize) {
125        self.data.reserve(additional);
126    }
127
128    /// Appends one value, or a null when it is `None`.
129    ///
130    /// # Panics
131    ///
132    /// Panics if the resulting end offset does not fit in `O`. See
133    /// [`append_value`](Self::append_value).
134    #[inline]
135    pub fn append(&mut self, value: Option<&[u8]>) {
136        match value {
137            Some(v) => self.append_value(v),
138            None => self.push_null(),
139        }
140    }
141
142    /// Appends one non-null value.
143    ///
144    /// # Panics
145    ///
146    /// Panics if the resulting end offset does not fit in `O`. The bulk appends report that as an
147    /// error instead; use [`append_n_values`](Self::append_n_values) for a fallible single append,
148    /// or an `i64` builder for byte totals past `i32::MAX`.
149    #[inline]
150    pub fn append_value(&mut self, value: impl AsRef<[u8]>) {
151        self.push_value(value.as_ref());
152        self.validity.append_true();
153    }
154
155    /// Appends the same non-null value `n` times.
156    ///
157    /// # Errors
158    ///
159    /// Returns an error, leaving the builder unchanged, if the resulting end offsets do not fit
160    /// in `O`.
161    pub fn append_n_values(&mut self, value: impl AsRef<[u8]>, n: usize) -> VortexResult<()> {
162        let value = value.as_ref();
163        let Some(num_bytes) = value.len().checked_mul(n) else {
164            vortex_bail!("Byte count overflow: {} values of {} bytes", n, value.len());
165        };
166        // Checking the total up front is what keeps `push_value` below from panicking.
167        self.check_offset_limit(self.data.len(), num_bytes)?;
168        self.offsets.reserve(n);
169        self.data.reserve(num_bytes);
170        for _ in 0..n {
171            self.push_value(value);
172        }
173        self.validity.append_n(true, n);
174        Ok(())
175    }
176
177    /// Appends a null value.
178    ///
179    /// Unlike [`append_null`](ArrayBuilder::append_null) this does not check that the builder is
180    /// nullable; the offsets and validity stay consistent either way, and a non-nullable `dtype`
181    /// discards the validity bits on [`finish_into_varbin`](Self::finish_into_varbin).
182    #[inline]
183    pub fn push_null(&mut self) {
184        self.push_nulls(1)
185    }
186
187    /// Appends `n` null values. See [`push_null`](Self::push_null).
188    #[inline]
189    pub fn push_nulls(&mut self, n: usize) {
190        self.offsets.push_n(self.last_offset(), n);
191        self.validity.append_n(false, n);
192    }
193
194    /// Appends the same UTF-8 or binary scalar `n` times.
195    ///
196    /// # Errors
197    ///
198    /// Returns an error if `scalar` has a different dtype than the builder, or if the resulting
199    /// end offsets do not fit in `O`.
200    pub fn append_scalar_repeated(&mut self, scalar: &Scalar, n: usize) -> VortexResult<()> {
201        vortex_ensure_eq!(
202            scalar.dtype(),
203            &self.dtype,
204            "VarBinBuilder scalar dtype does not match builder dtype",
205        );
206        match &self.dtype {
207            DType::Utf8(_) => match scalar.as_utf8().value() {
208                Some(value) => self.append_n_values(value, n)?,
209                None => self.push_nulls(n),
210            },
211            DType::Binary(_) => match scalar.as_binary().value() {
212                Some(value) => self.append_n_values(value, n)?,
213                None => self.push_nulls(n),
214            },
215            dtype => vortex_bail!("VarBinBuilder cannot append scalar of dtype {dtype}"),
216        }
217        Ok(())
218    }
219
220    /// Appends values from one contiguous byte buffer described by relative end offsets.
221    ///
222    /// Each entry of `end_offsets` marks the end of one value relative to the start of `values`,
223    /// and there must be exactly one per entry in `validity`.
224    #[inline]
225    pub fn append_values<P>(
226        &mut self,
227        values: &[u8],
228        end_offsets: impl Iterator<Item = P>,
229        validity: &Mask,
230    ) -> VortexResult<()>
231    where
232        P: AsPrimitive<usize>,
233        usize: AsPrimitive<O>,
234    {
235        // Offsets are committed first: they are the only fallible part, and failing before the
236        // bytes are appended leaves the builder untouched.
237        self.extend_offsets(values.len(), validity.len(), end_offsets)?;
238        self.data.extend_from_slice(values);
239        self.append_validity(validity);
240        Ok(())
241    }
242
243    /// Appends values whose bytes `decode` writes straight into the builder's byte storage.
244    ///
245    /// `decode` is handed at least `num_bytes + slack` bytes of uninitialized storage and returns
246    /// the number of bytes it initialized, which must be exactly `num_bytes`. Each entry of
247    /// `lengths` is the byte length of one value — including nulls, which are zero-length — so
248    /// there must be one per entry in `validity`.
249    ///
250    /// `slack` is spare headroom past the values themselves. It exists so that a decoder which
251    /// stores in wide fixed-size chunks can keep using them through the final value instead of
252    /// dropping into a byte-at-a-time tail; a decoder must still never write past the slice it is
253    /// handed, so `0` is correct for one that is already exact.
254    ///
255    /// Decoding in place saves staging the decoded heap in a temporary buffer and copying it in.
256    ///
257    /// `decode` is taken as a `dyn` reference so a caller that resolves the `lengths` type with
258    /// [`match_each_integer_ptype!`](crate::match_each_integer_ptype) can build the closure once
259    /// outside that match rather than inlining a whole decoder into each of its arms.
260    ///
261    /// # Errors
262    ///
263    /// Returns an error, leaving the builder unchanged, if `num_bytes` would push an offset past
264    /// what `O` can hold — checked before `decode` runs — or if `decode` reports a different byte
265    /// count than `num_bytes`, or if `lengths` does not describe those bytes.
266    ///
267    /// # Safety
268    ///
269    /// `decode` must initialize the first `n` bytes of the slice it is passed, where `n` is the
270    /// value it returns. Those bytes are published as initialized without ever being read first,
271    /// so a `decode` that over-reports leaves the builder holding uninitialized memory.
272    pub unsafe fn append_decoded<P>(
273        &mut self,
274        num_bytes: usize,
275        slack: usize,
276        lengths: &[P],
277        validity: &Mask,
278        decode: &mut dyn FnMut(&mut [MaybeUninit<u8>]) -> VortexResult<usize>,
279    ) -> VortexResult<()>
280    where
281        P: AsPrimitive<usize>,
282        usize: AsPrimitive<O>,
283    {
284        let Some(capacity) = num_bytes.checked_add(slack) else {
285            vortex_bail!("Decoded size overflow: {num_bytes} + {slack}");
286        };
287        // Checked before decoding: an `i32` builder that the decoded bytes would overflow should
288        // not pay for the decompression first.
289        self.check_offset_limit(self.data.len(), num_bytes)?;
290        self.data.reserve(capacity);
291
292        let data_len = self.data.len();
293        let written = decode(self.data.spare_capacity_mut())?;
294        vortex_ensure_eq!(
295            written,
296            num_bytes,
297            "Decoded byte count does not match expected byte count",
298        );
299
300        // The decoded bytes live in spare capacity until `set_len` below, so an invalid `lengths`
301        // still leaves the builder unchanged.
302        self.extend_offsets(num_bytes, validity.len(), prefix_sums(lengths))?;
303
304        // SAFETY: `decode` reported initializing `written` spare bytes, and the caller guarantees
305        // that report is accurate; `written == num_bytes` was checked above.
306        unsafe { self.data.set_len(data_len + num_bytes) };
307        self.append_validity(validity);
308        Ok(())
309    }
310
311    /// Appends `validity.len()` values, taking each valid row's bytes from `value`.
312    ///
313    /// `value` is called once per valid row in ascending row order and the slices it returns must
314    /// total `num_bytes` bytes. Null rows consume no bytes and repeat the preceding offset.
315    ///
316    /// The valid rows are walked a `u64` word at a time rather than pulled one at a time through
317    /// an iterator, and both buffers are sized once up front, so the copy loop costs one offset
318    /// store plus one `memcpy` per value. A caller whose values only exist as a stream can still
319    /// use this by pulling from the stream inside `value`: the rows are visited in ascending
320    /// order.
321    pub fn append_valid_slices<'a, F>(
322        &mut self,
323        num_bytes: usize,
324        validity: &Mask,
325        value: F,
326    ) -> VortexResult<()>
327    where
328        F: FnMut(usize) -> &'a [u8],
329        usize: AsPrimitive<O>,
330    {
331        let data_start = self.data.len();
332        match self.gather_valid_slices(num_bytes, validity, value) {
333            Ok(()) => {
334                self.append_validity(validity);
335                Ok(())
336            }
337            Err(error) => {
338                // The offsets are never committed on failure, so only the copied bytes need to go.
339                self.data.truncate(data_start);
340                Err(error)
341            }
342        }
343    }
344
345    /// Appends a [`VarBinArray`], reusing its offsets instead of walking its values.
346    pub fn append_varbin(
347        &mut self,
348        array: ArrayView<'_, VarBin>,
349        ctx: &mut ExecutionCtx,
350    ) -> VortexResult<()>
351    where
352        usize: AsPrimitive<O>,
353    {
354        let offsets = array.offsets().clone().execute::<PrimitiveArray>(ctx)?;
355        let bytes: ByteBuffer = array.sliced_bytes();
356        let validity = array
357            .varbin_validity()
358            .execute_mask(array.as_ref().len(), ctx)?;
359        match_each_integer_ptype!(offsets.ptype(), |P| {
360            let offsets = offsets.as_slice::<P>();
361            let first: usize = offsets[0].as_();
362            self.append_values(
363                bytes.as_slice(),
364                // Wrapping keeps a corrupt offsets child from panicking here; `append_values`
365                // rejects the resulting non-monotonic offsets.
366                offsets[1..]
367                    .iter()
368                    .map(|offset| AsPrimitive::<usize>::as_(*offset).wrapping_sub(first)),
369                &validity,
370            )
371        })
372    }
373
374    /// Appends a [`VarBinViewArray`](crate::arrays::VarBinViewArray), compacting its values.
375    pub fn append_varbinview(
376        &mut self,
377        array: ArrayView<'_, VarBinView>,
378        ctx: &mut ExecutionCtx,
379    ) -> VortexResult<()>
380    where
381        usize: AsPrimitive<O>,
382    {
383        let len = array.as_ref().len();
384        let validity = array.varbinview_validity().execute_mask(len, ctx)?;
385
386        // Resolve the views slice and the data buffers once. Reading them per row costs a buffer
387        // handle clone, and the byte total needs only the fixed-width view headers.
388        let views = array.views();
389        let buffers = array
390            .data_buffers()
391            .iter()
392            .map(|buffer| buffer.as_host().as_slice())
393            .collect::<Vec<_>>();
394
395        let num_bytes = match validity.bit_buffer() {
396            AllOr::All => views.iter().map(|view| view.len() as usize).sum(),
397            AllOr::None => 0,
398            AllOr::Some(bits) => {
399                let mut total = 0;
400                bits.for_each_set_index(|index| total += views[index].len() as usize);
401                total
402            }
403        };
404
405        self.append_valid_slices(num_bytes, &validity, |index| views[index].bytes(&buffers))
406    }
407
408    /// Finishes the appended values into a [`VarBinArray`] and resets the builder.
409    #[allow(clippy::disallowed_methods)]
410    pub fn finish_into_varbin(&mut self) -> VarBinArray {
411        assert_eq!(
412            self.offsets.len() - 1,
413            self.validity.len(),
414            "The offset count must be one more than the validity length"
415        );
416
417        let offsets = self.offsets.take();
418        self.offsets.push(O::zero());
419        let offsets = PrimitiveArray::new(offsets.freeze(), Validity::NonNullable);
420        let data = self.data.take();
421        let nulls = self.validity.take().freeze();
422
423        let validity = Validity::from_bit_buffer(nulls, self.dtype.nullability());
424
425        // The builder adds offsets in monotonically increasing order. Store this statistic to
426        // prevent VarBinArray::validate from recomputing it after deserialization.
427        #[cfg(debug_assertions)]
428        {
429            let offsets_are_sorted = offsets
430                .statistics()
431                .compute_is_sorted(&mut legacy_session().create_execution_ctx())
432                .unwrap_or(false);
433            debug_assert!(offsets_are_sorted, "VarBinBuilder offsets must be sorted");
434        }
435        offsets
436            .statistics()
437            .set(Stat::IsSorted, Precision::Exact(true.into()));
438
439        // SAFETY: The builder maintains all invariants:
440        // - Offsets are monotonically increasing starting from 0 (guaranteed by builder logic).
441        // - Bytes buffer contains exactly the data referenced by offsets.
442        // - Validity matches the dtype nullability.
443        // - UTF-8 validity is ensured by the caller when using DType::Utf8.
444        unsafe {
445            VarBinArray::new_unchecked(
446                offsets.into_array(),
447                data.freeze(),
448                self.dtype.clone(),
449                validity,
450            )
451        }
452    }
453
454    #[inline]
455    fn last_offset(&self) -> O {
456        self.offsets[self.offsets.len() - 1]
457    }
458
459    /// Appends `value`'s bytes and its end offset, leaving the validity bits to the caller.
460    #[inline]
461    fn push_value(&mut self, value: &[u8]) {
462        self.offsets
463            .push(O::from(self.data.len() + value.len()).unwrap_or_else(|| {
464                offset_overflow(self.data.len(), value.len(), std::any::type_name::<O>())
465            }));
466        self.data.extend_from_slice(value);
467    }
468
469    fn append_validity(&mut self, validity: &Mask) {
470        match validity {
471            Mask::AllTrue(len) => self.validity.append_n(true, *len),
472            Mask::AllFalse(len) => self.validity.append_n(false, *len),
473            Mask::Values(values) => self.validity.append_buffer(values.bit_buffer()),
474        }
475    }
476
477    /// Appends `count` end offsets derived from `end_offsets`, shifted past the current data end.
478    ///
479    /// `end_offsets` must be monotonically non-decreasing and end at exactly `num_bytes`. Offsets
480    /// are written through the reserved spare capacity and committed with a single `set_len`, so a
481    /// rejected input leaves the buffer untouched and the caller can propagate the error.
482    fn extend_offsets<P>(
483        &mut self,
484        num_bytes: usize,
485        count: usize,
486        end_offsets: impl Iterator<Item = P>,
487    ) -> VortexResult<()>
488    where
489        P: AsPrimitive<usize>,
490        usize: AsPrimitive<O>,
491    {
492        let data_start = self.data.len();
493        let offsets_len = self.offsets.len();
494        self.check_offset_limit(data_start, num_bytes)?;
495        self.offsets.reserve(count);
496
497        // Writing into the spare capacity keeps the output cursor in a register: `push` rewrites
498        // the buffer length on every value, which the optimizer cannot hoist out of the loop.
499        let spare = &mut self.offsets.spare_capacity_mut()[..count];
500        let mut end_offsets = end_offsets;
501        let mut previous = 0usize;
502        for slot in spare.iter_mut() {
503            let Some(end) = end_offsets.next() else {
504                vortex_bail!("End offset count is less than the validity length {count}");
505            };
506            let end = end.as_();
507            vortex_ensure!(
508                end >= previous && end <= num_bytes,
509                "End offsets must be monotonically increasing within {num_bytes} bytes, \
510                 got {end} after {previous}"
511            );
512            slot.write((data_start + end).as_());
513            previous = end;
514        }
515        vortex_ensure!(
516            end_offsets.next().is_none(),
517            "End offset count exceeds the validity length {count}"
518        );
519        vortex_ensure_eq!(
520            previous,
521            num_bytes,
522            "Final end offset does not match the value byte count",
523        );
524
525        // SAFETY: the loop initialized the first `count` spare slots.
526        unsafe { self.offsets.set_len(offsets_len + count) };
527        Ok(())
528    }
529
530    /// Copies each valid row's bytes from `value` into the byte storage and records the offsets.
531    /// See [`append_valid_slices`](Self::append_valid_slices); the validity bits are the caller's
532    /// job so that a failure here can be unwound.
533    fn gather_valid_slices<'a, F>(
534        &mut self,
535        num_bytes: usize,
536        validity: &Mask,
537        mut value: F,
538    ) -> VortexResult<()>
539    where
540        F: FnMut(usize) -> &'a [u8],
541        usize: AsPrimitive<O>,
542    {
543        let count = validity.len();
544        let data_start = self.data.len();
545        let offsets_len = self.offsets.len();
546        self.check_offset_limit(data_start, num_bytes)?;
547        self.offsets.reserve(count);
548        self.data.reserve(num_bytes);
549
550        // Disjoint field borrows: the offsets spare capacity stays valid while the byte buffer
551        // grows, since the two are separate allocations.
552        let Self { offsets, data, .. } = self;
553        let spare = &mut offsets.spare_capacity_mut()[..count];
554
555        match validity.bit_buffer() {
556            AllOr::All => {
557                for (row, slot) in spare.iter_mut().enumerate() {
558                    data.extend_from_slice(value(row));
559                    slot.write(data.len().as_());
560                }
561            }
562            AllOr::None => {
563                spare.fill(MaybeUninit::new(data_start.as_()));
564            }
565            AllOr::Some(bits) => {
566                let mut row = 0;
567                bits.for_each_set_index(|index| {
568                    // Null rows between the previous valid row and this one repeat its end offset.
569                    spare[row..index].fill(MaybeUninit::new(data.len().as_()));
570                    data.extend_from_slice(value(index));
571                    spare[index].write(data.len().as_());
572                    row = index + 1;
573                });
574                spare[row..].fill(MaybeUninit::new(data.len().as_()));
575            }
576        }
577
578        // A caller whose slices overrun `num_bytes` only grows the byte buffer past the reservation,
579        // so the overrun is caught here rather than per value. The offsets are still uncommitted, so
580        // rejecting it now leaves them untouched.
581        vortex_ensure_eq!(
582            data.len(),
583            data_start + num_bytes,
584            "Value slices byte count does not match expected byte count",
585        );
586
587        // SAFETY: every branch above initialized all `count` spare slots.
588        unsafe { self.offsets.set_len(offsets_len + count) };
589        Ok(())
590    }
591
592    /// Checks that an offset past `num_bytes` more bytes is representable as an `O`.
593    fn check_offset_limit(&self, data_start: usize, num_bytes: usize) -> VortexResult<()> {
594        let Some(limit) = data_start.checked_add(num_bytes) else {
595            vortex_bail!("Byte offset overflow: {data_start} + {num_bytes}");
596        };
597        vortex_ensure!(
598            u64::try_from(limit).is_ok_and(|limit| limit <= O::max_value_as_u64()),
599            "Byte offset {limit} does not fit in {}",
600            std::any::type_name::<O>()
601        );
602        Ok(())
603    }
604}
605
606impl<O: OffsetBuilderPType> ArrayBuilder for VarBinBuilder<O> {
607    fn as_any(&self) -> &dyn Any {
608        self
609    }
610
611    fn as_any_mut(&mut self) -> &mut dyn Any {
612        self
613    }
614
615    fn dtype(&self) -> &DType {
616        &self.dtype
617    }
618
619    fn len(&self) -> usize {
620        self.validity.len()
621    }
622
623    fn append_zeros(&mut self, n: usize) {
624        self.offsets.push_n(self.last_offset(), n);
625        self.validity.append_n(true, n);
626    }
627
628    unsafe fn append_nulls_unchecked(&mut self, n: usize) {
629        self.push_nulls(n);
630    }
631
632    fn append_scalar(&mut self, scalar: &Scalar) -> VortexResult<()> {
633        self.append_scalar_repeated(scalar, 1)
634    }
635
636    fn reserve_exact(&mut self, additional: usize) {
637        self.offsets.reserve(additional);
638        self.validity.reserve(additional);
639    }
640
641    fn finish(&mut self) -> ArrayRef {
642        self.finish_into_varbin().into_array()
643    }
644
645    fn finish_into_canonical(&mut self, ctx: &mut ExecutionCtx) -> Canonical {
646        self.finish()
647            .execute::<Canonical>(ctx)
648            .vortex_expect("varbin builder should canonicalize")
649    }
650}
651
652/// Recovers a concrete [`VarBinBuilder`] from a `&mut dyn ArrayBuilder`.
653///
654/// Evaluates to `Some($body)` with `$typed` bound to the `&mut VarBinBuilder<O>` when the builder
655/// is one, and `None` otherwise, so an encoding can fall through to its other output paths:
656///
657/// ```ignore
658/// if let Some(result) = match_each_varbin_builder!(builder, |builder| {
659///     append_my_encoding(array, builder, ctx)
660/// }) {
661///     return result;
662/// }
663/// ```
664#[macro_export]
665macro_rules! match_each_varbin_builder {
666    ($builder:expr, | $typed:ident | $body:expr) => {
667        $crate::__match_varbin_builder_widths!($builder, |$typed| $body, [u32, u64, i32, i64])
668    };
669}
670
671/// Expands `$body` once per listed offset width, guarded by a downcast. See
672/// [`match_each_varbin_builder!`].
673#[doc(hidden)]
674#[macro_export]
675macro_rules! __match_varbin_builder_widths {
676    ($builder:expr, | $typed:ident | $body:expr, [$($width:ty),+ $(,)?]) => {{
677        let __varbin_builder: &mut dyn $crate::builders::ArrayBuilder = $builder;
678        $crate::__match_varbin_builder_arms!(__varbin_builder, |$typed| $body, [$($width),+])
679    }};
680}
681
682/// The `if`/`else if` chain behind [`__match_varbin_builder_widths!`], one arm per width.
683#[doc(hidden)]
684#[macro_export]
685macro_rules! __match_varbin_builder_arms {
686    ($builder:expr, | $typed:ident | $body:expr, []) => {
687        None
688    };
689    ($builder:expr, | $typed:ident | $body:expr, [$head:ty $(, $tail:ty)*]) => {
690        if $builder
691            .as_any()
692            .is::<$crate::builders::VarBinBuilder<$head>>()
693        {
694            let $typed = match $builder
695                .as_any_mut()
696                .downcast_mut::<$crate::builders::VarBinBuilder<$head>>()
697            {
698                Some(typed) => typed,
699                None => unreachable!("builder type checked above"),
700            };
701            Some($body)
702        } else {
703            $crate::__match_varbin_builder_arms!($builder, |$typed| $body, [$($tail),*])
704        }
705    };
706}
707
708#[cold]
709#[inline(never)]
710fn offset_overflow(data_len: usize, value_len: usize, offset_type: &'static str) -> ! {
711    vortex_panic!(
712        "Failed to convert sum of {data_len} and {value_len} to offset of type {offset_type}"
713    )
714}
715
716/// Running totals of `lengths`, wrapping so a corrupt lengths child is rejected rather than
717/// panicking; [`VarBinBuilder::extend_offsets`] catches the resulting non-monotonic offsets.
718#[inline]
719fn prefix_sums<P: AsPrimitive<usize>>(lengths: &[P]) -> impl Iterator<Item = usize> {
720    lengths.iter().scan(0usize, |end, length| {
721        *end = end.wrapping_add(length.as_());
722        Some(*end)
723    })
724}
725
726#[cfg(test)]
727mod tests {
728    use std::mem::MaybeUninit;
729
730    use rstest::rstest;
731    use vortex_error::VortexResult;
732    use vortex_mask::Mask;
733
734    use crate::IntoArray;
735    use crate::VortexSessionExecute;
736    use crate::array_session;
737    use crate::arrays::ChunkedArray;
738    use crate::arrays::ConstantArray;
739    use crate::arrays::VarBinArray;
740    use crate::arrays::VarBinViewArray;
741    use crate::arrays::varbin::VarBinArraySlotsExt;
742    use crate::arrays::varbin::builder::VarBinBuilder;
743    use crate::assert_arrays_eq;
744    use crate::builders::ArrayBuilder;
745    use crate::dtype::DType;
746    use crate::dtype::Nullability::Nullable;
747    use crate::expr::stats::Precision;
748    use crate::expr::stats::Stat;
749    use crate::expr::stats::StatsProviderExt;
750    use crate::scalar::Scalar;
751
752    #[test]
753    fn test_builder() {
754        let mut builder = VarBinBuilder::<i32>::with_capacity_in(
755            DType::Utf8(Nullable),
756            0,
757            vortex_buffer::BufferAllocatorRef::static_ref(),
758        );
759        builder.append(Some(b"hello"));
760        builder.append(None);
761        builder.append(Some(b"world"));
762        let array = builder.finish_into_varbin();
763
764        assert_eq!(array.len(), 3);
765        assert_eq!(array.dtype().nullability(), Nullable);
766        assert_eq!(
767            array
768                .execute_scalar(0, &mut array_session().create_execution_ctx())
769                .unwrap(),
770            Scalar::utf8("hello".to_string(), Nullable)
771        );
772        assert!(
773            array
774                .execute_scalar(1, &mut array_session().create_execution_ctx())
775                .unwrap()
776                .is_null()
777        );
778    }
779
780    #[rstest]
781    #[case(false)]
782    #[case(true)]
783    fn test_append_varbin_to_builder(#[case] large_offsets: bool) -> VortexResult<()> {
784        let source = VarBinArray::from_iter(
785            [
786                Some("prefix"),
787                Some("hello"),
788                None,
789                Some("world"),
790                Some("suffix"),
791            ],
792            DType::Utf8(Nullable),
793        )
794        .into_array()
795        .slice(1..4)?;
796        let mut ctx = array_session().create_execution_ctx();
797
798        let actual = with_offsets(large_offsets, source.dtype().clone(), |builder| {
799            source.append_to_builder(builder, &mut ctx)
800        })?;
801
802        assert_arrays_eq!(actual, source, &mut ctx);
803        Ok(())
804    }
805
806    #[test]
807    fn append_n_values_offset_overflow_returns_error() {
808        let mut builder = VarBinBuilder::<i32>::new_in(
809            DType::Utf8(Nullable),
810            vortex_buffer::BufferAllocatorRef::static_ref(),
811        );
812
813        // The limit is checked before anything is reserved, so the huge byte total is free.
814        let result = builder.append_n_values(b"hello", i32::MAX as usize / 5 + 1);
815
816        assert!(result.is_err());
817        assert_eq!(builder.offsets.len(), 1);
818        assert!(builder.data.is_empty());
819        assert_eq!(builder.validity.len(), 0);
820    }
821
822    #[test]
823    fn append_values_rejects_a_short_offset_count() {
824        let mut builder = VarBinBuilder::<i32>::new_in(
825            DType::Utf8(Nullable),
826            vortex_buffer::BufferAllocatorRef::static_ref(),
827        );
828
829        let result = builder.append_values(b"ab", [1usize, 2].into_iter(), &Mask::new_true(3));
830
831        assert!(result.is_err());
832        assert_eq!(builder.offsets.len(), 1);
833        assert!(builder.data.is_empty());
834    }
835
836    #[test]
837    fn append_values_rejects_non_monotonic_offsets() {
838        let mut builder = VarBinBuilder::<i32>::new_in(
839            DType::Utf8(Nullable),
840            vortex_buffer::BufferAllocatorRef::static_ref(),
841        );
842
843        let result = builder.append_values(b"ab", [2usize, 1].into_iter(), &Mask::new_true(2));
844
845        assert!(result.is_err());
846        assert_eq!(builder.offsets.len(), 1);
847    }
848
849    #[test]
850    fn append_decoded_writes_into_the_builder_storage() -> VortexResult<()> {
851        let mut ctx = array_session().create_execution_ctx();
852        let mut builder = VarBinBuilder::<i32>::new_in(
853            DType::Utf8(Nullable),
854            vortex_buffer::BufferAllocatorRef::static_ref(),
855        );
856
857        // SAFETY: the closure initializes exactly the 6 bytes it reports.
858        unsafe {
859            builder.append_decoded(
860                6,
861                4,
862                &[3usize, 0, 3],
863                &Mask::from_iter([true, false, true]),
864                &mut |spare: &mut [MaybeUninit<u8>]| {
865                    for (slot, byte) in spare.iter_mut().zip(b"foobar") {
866                        slot.write(*byte);
867                    }
868                    Ok(6)
869                },
870            )?;
871        }
872
873        let expected =
874            VarBinViewArray::from_iter([Some("foo"), None, Some("bar")], DType::Utf8(Nullable));
875        assert_arrays_eq!(builder.finish_into_varbin(), expected, &mut ctx);
876        Ok(())
877    }
878
879    /// The offset width has to be checked before `decode` runs: an overflowing `i32` builder
880    /// should not pay for a full decompression first.
881    #[test]
882    fn append_decoded_rejects_an_offset_overflow_without_decoding() {
883        let mut builder = VarBinBuilder::<i32>::new_in(
884            DType::Utf8(Nullable),
885            vortex_buffer::BufferAllocatorRef::static_ref(),
886        );
887        let num_bytes = i32::MAX as usize + 1;
888        let mut decoded = false;
889
890        // SAFETY: the closure is never called, and reports only what it initializes if it were.
891        let result = unsafe {
892            builder.append_decoded(num_bytes, 0, &[num_bytes], &Mask::new_true(1), &mut |_| {
893                decoded = true;
894                Ok(0)
895            })
896        };
897
898        assert!(result.is_err());
899        assert!(!decoded, "decode ran before the offset width was checked");
900        assert_eq!(builder.offsets.len(), 1);
901        assert_eq!(builder.validity.len(), 0);
902    }
903
904    /// `append_scalar_repeated` reports an offset overflow rather than panicking, so a constant
905    /// column wider than the offset type errors like the bulk appends do.
906    #[test]
907    fn append_scalar_repeated_rejects_an_offset_overflow() {
908        let mut builder = VarBinBuilder::<i32>::new_in(
909            DType::Utf8(Nullable),
910            vortex_buffer::BufferAllocatorRef::static_ref(),
911        );
912
913        let result = builder
914            .append_scalar_repeated(&Scalar::utf8("hello", Nullable), i32::MAX as usize / 5 + 1);
915
916        assert!(result.is_err());
917        assert_eq!(builder.offsets.len(), 1);
918        assert!(builder.data.is_empty());
919        assert_eq!(builder.validity.len(), 0);
920    }
921
922    #[test]
923    fn append_decoded_rejects_a_short_decode() {
924        let mut builder = VarBinBuilder::<i32>::new_in(
925            DType::Utf8(Nullable),
926            vortex_buffer::BufferAllocatorRef::static_ref(),
927        );
928
929        // SAFETY: the closure initializes the 3 bytes it reports (none, and it reports 3 — but the
930        // length mismatch is rejected before anything is published).
931        let result = unsafe {
932            builder.append_decoded(6, 0, &[3usize, 3], &Mask::new_true(2), &mut |spare| {
933                spare[..3].fill(MaybeUninit::new(b'x'));
934                Ok(3)
935            })
936        };
937
938        assert!(result.is_err());
939        assert_eq!(builder.offsets.len(), 1);
940        assert_eq!(builder.validity.len(), 0);
941    }
942
943    /// Slices that overrun the declared byte count are rejected, and the builder is left exactly
944    /// as it was — the overrun bytes are copied before the total can be checked, so the unwind has
945    /// to put them back.
946    #[rstest]
947    #[case::all_valid(Mask::new_true(2))]
948    #[case::some_valid(Mask::from_iter([true, false, true]))]
949    fn append_valid_slices_rejects_a_byte_count_mismatch(#[case] validity: Mask) {
950        let mut builder = VarBinBuilder::<i32>::new_in(
951            DType::Utf8(Nullable),
952            vortex_buffer::BufferAllocatorRef::static_ref(),
953        );
954        let values = [b"foo".as_slice(), b"quux".as_slice()];
955
956        let mut next = 0;
957        let result = builder.append_valid_slices(6, &validity, |_| {
958            next += 1;
959            values[next - 1]
960        });
961
962        assert!(result.is_err());
963        assert_eq!(builder.offsets.len(), 1);
964        assert!(builder.data.is_empty());
965        assert_eq!(builder.validity.len(), 0);
966    }
967
968    #[test]
969    #[should_panic(expected = "The offset count must be one more than the validity length")]
970    fn finish_rejects_mismatched_validity() {
971        let mut builder = VarBinBuilder::<i32>::new_in(
972            DType::Utf8(Nullable),
973            vortex_buffer::BufferAllocatorRef::static_ref(),
974        );
975        builder.validity.append_true();
976        drop(builder.finish_into_varbin());
977    }
978
979    #[rstest]
980    #[case(false)]
981    #[case(true)]
982    fn test_array_builder_methods(#[case] large_offsets: bool) -> VortexResult<()> {
983        let mut ctx = array_session().create_execution_ctx();
984        let result = with_offsets(large_offsets, DType::Utf8(Nullable), |builder| {
985            builder.reserve_exact(3);
986            builder.append_zero();
987            builder.append_scalar(&Scalar::utf8("hello", Nullable))?;
988            builder.append_null();
989            assert_eq!(builder.len(), 3);
990            Ok(())
991        })?;
992
993        assert_eq!(
994            result.validity()?.execute_mask(3, &mut ctx)?,
995            Mask::from_iter([true, true, false])
996        );
997        Ok(())
998    }
999
1000    #[rstest]
1001    #[case(false)]
1002    #[case(true)]
1003    fn test_append_varbinview_validity_to_builder(#[case] large_offsets: bool) -> VortexResult<()> {
1004        let long = "a value that does not fit inline";
1005        let all_null = VarBinViewArray::from_iter([None::<&str>, None], DType::Utf8(Nullable));
1006        let mixed =
1007            VarBinViewArray::from_iter([Some("hello"), None, Some(long)], DType::Utf8(Nullable));
1008        let expected = VarBinViewArray::from_iter(
1009            [None, None, Some("hello"), None, Some(long)],
1010            DType::Utf8(Nullable),
1011        );
1012        let mut ctx = array_session().create_execution_ctx();
1013
1014        let actual = with_offsets(large_offsets, expected.dtype().clone(), |builder| {
1015            all_null
1016                .clone()
1017                .into_array()
1018                .append_to_builder(builder, &mut ctx)?;
1019            mixed
1020                .clone()
1021                .into_array()
1022                .append_to_builder(builder, &mut ctx)
1023        })?;
1024
1025        assert_arrays_eq!(actual, expected, &mut ctx);
1026        Ok(())
1027    }
1028
1029    /// `match_each_varbin_builder!` covers every offset width a `VarBinBuilder` can be built with,
1030    /// so both canonical appends must reach their specialization for all of them. A macro that
1031    /// only matched the signed pair would send an unsigned builder down a downcast that assumes
1032    /// `VarBinViewBuilder` and panic.
1033    #[rstest]
1034    #[case::u32(VarBinBuilder::<u32>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1035    #[case::u64(VarBinBuilder::<u64>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1036    #[case::i32(VarBinBuilder::<i32>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1037    #[case::i64(VarBinBuilder::<i64>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1038    fn append_to_every_offset_width(#[case] mut builder: impl ArrayBuilder) -> VortexResult<()> {
1039        let mut ctx = array_session().create_execution_ctx();
1040        let long = "a string that is far too long to be inlined in a view";
1041        let values = [Some("hello"), None, Some(long), Some("")];
1042
1043        let view = VarBinViewArray::from_iter(values, DType::Utf8(Nullable)).into_array();
1044        let varbin = VarBinArray::from_iter(values, DType::Utf8(Nullable)).into_array();
1045        let constant = ConstantArray::new(Scalar::from("hello").into_nullable(), 2).into_array();
1046
1047        for chunk in [&view, &varbin, &constant] {
1048            chunk.append_to_builder(&mut builder, &mut ctx)?;
1049        }
1050
1051        let expected = ChunkedArray::try_new(vec![view, varbin, constant], DType::Utf8(Nullable))?;
1052        assert_arrays_eq!(builder.finish(), expected, &mut ctx);
1053        Ok(())
1054    }
1055
1056    #[test]
1057    fn offsets_have_is_sorted_stat() -> VortexResult<()> {
1058        let mut builder = VarBinBuilder::<i32>::with_capacity_in(
1059            DType::Utf8(Nullable),
1060            0,
1061            vortex_buffer::BufferAllocatorRef::static_ref(),
1062        );
1063        builder.append_value(b"aaa");
1064        builder.push_null();
1065        builder.append_value(b"bbb");
1066        let array = builder.finish_into_varbin();
1067
1068        let is_sorted = array
1069            .offsets()
1070            .statistics()
1071            .with_typed_stats_set(|s| s.get_as::<bool>(Stat::IsSorted));
1072        assert_eq!(is_sorted, Precision::Exact(true));
1073        Ok(())
1074    }
1075
1076    #[test]
1077    fn empty_builder_offsets_have_is_sorted_stat() -> VortexResult<()> {
1078        let mut builder = VarBinBuilder::<i32>::new_in(
1079            DType::Utf8(Nullable),
1080            vortex_buffer::BufferAllocatorRef::static_ref(),
1081        );
1082        let array = builder.finish_into_varbin();
1083
1084        let is_sorted = array
1085            .offsets()
1086            .statistics()
1087            .with_typed_stats_set(|s| s.get_as::<bool>(Stat::IsSorted));
1088        assert_eq!(is_sorted, Precision::Exact(true));
1089        Ok(())
1090    }
1091
1092    /// Runs `f` against an `i32` or `i64` builder and returns the finished array.
1093    fn with_offsets(
1094        large_offsets: bool,
1095        dtype: DType,
1096        f: impl FnOnce(&mut dyn ArrayBuilder) -> VortexResult<()>,
1097    ) -> VortexResult<VarBinArray> {
1098        if large_offsets {
1099            let mut builder = VarBinBuilder::<i64>::with_capacity_in(
1100                dtype,
1101                8,
1102                vortex_buffer::BufferAllocatorRef::static_ref(),
1103            );
1104            f(&mut builder)?;
1105            Ok(builder.finish_into_varbin())
1106        } else {
1107            let mut builder = VarBinBuilder::<i32>::with_capacity_in(
1108                dtype,
1109                8,
1110                vortex_buffer::BufferAllocatorRef::static_ref(),
1111            );
1112            f(&mut builder)?;
1113            Ok(builder.finish_into_varbin())
1114        }
1115    }
1116}