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