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::ConstantArray;
738    use crate::arrays::VarBinArray;
739    use crate::arrays::VarBinViewArray;
740    use crate::arrays::varbin::VarBinArraySlotsExt;
741    use crate::arrays::varbin::builder::VarBinBuilder;
742    use crate::assert_arrays_eq;
743    use crate::builders::ArrayBuilder;
744    use crate::dtype::DType;
745    use crate::dtype::Nullability::Nullable;
746    use crate::expr::stats::Precision;
747    use crate::expr::stats::Stat;
748    use crate::expr::stats::StatsProviderExt;
749    use crate::scalar::Scalar;
750
751    #[test]
752    fn test_builder() {
753        let mut builder = VarBinBuilder::<i32>::with_capacity_in(
754            DType::Utf8(Nullable),
755            0,
756            vortex_buffer::BufferAllocatorRef::static_ref(),
757        );
758        builder.append(Some(b"hello"));
759        builder.append(None);
760        builder.append(Some(b"world"));
761        let array = builder.finish_into_varbin();
762
763        assert_eq!(array.len(), 3);
764        assert_eq!(array.dtype().nullability(), Nullable);
765        assert_eq!(
766            array
767                .execute_scalar(0, &mut array_session().create_execution_ctx())
768                .unwrap(),
769            Scalar::utf8("hello".to_string(), Nullable)
770        );
771        assert!(
772            array
773                .execute_scalar(1, &mut array_session().create_execution_ctx())
774                .unwrap()
775                .is_null()
776        );
777        assert_arrays_eq!(
778            array,
779            VarBinViewArray::from_iter_nullable_str([Some("hello"), None, Some("world")]),
780            &mut array_session().create_execution_ctx()
781        );
782    }
783
784    #[rstest]
785    #[case(false)]
786    #[case(true)]
787    fn test_append_varbin_to_builder(#[case] large_offsets: bool) -> VortexResult<()> {
788        let source = VarBinArray::from_iter(
789            [
790                Some("prefix"),
791                Some("hello"),
792                None,
793                Some("world"),
794                Some("suffix"),
795            ],
796            DType::Utf8(Nullable),
797        )
798        .into_array()
799        .slice(1..4)?;
800        let mut ctx = array_session().create_execution_ctx();
801
802        let actual = with_offsets(large_offsets, source.dtype().clone(), |builder| {
803            source.append_to_builder(builder, &mut ctx)
804        })?;
805
806        assert_arrays_eq!(
807            actual,
808            VarBinViewArray::from_iter_nullable_str([Some("hello"), None, Some("world")]),
809            &mut ctx
810        );
811        Ok(())
812    }
813
814    #[test]
815    fn append_n_values_offset_overflow_returns_error() {
816        let mut builder = VarBinBuilder::<i32>::new_in(
817            DType::Utf8(Nullable),
818            vortex_buffer::BufferAllocatorRef::static_ref(),
819        );
820
821        // The limit is checked before anything is reserved, so the huge byte total is free.
822        let result = builder.append_n_values(b"hello", i32::MAX as usize / 5 + 1);
823
824        assert!(result.is_err());
825        assert_eq!(builder.offsets.len(), 1);
826        assert!(builder.data.is_empty());
827        assert_eq!(builder.validity.len(), 0);
828    }
829
830    #[test]
831    fn append_values_rejects_a_short_offset_count() {
832        let mut builder = VarBinBuilder::<i32>::new_in(
833            DType::Utf8(Nullable),
834            vortex_buffer::BufferAllocatorRef::static_ref(),
835        );
836
837        let result = builder.append_values(b"ab", [1usize, 2].into_iter(), &Mask::new_true(3));
838
839        assert!(result.is_err());
840        assert_eq!(builder.offsets.len(), 1);
841        assert!(builder.data.is_empty());
842    }
843
844    #[test]
845    fn append_values_rejects_non_monotonic_offsets() {
846        let mut builder = VarBinBuilder::<i32>::new_in(
847            DType::Utf8(Nullable),
848            vortex_buffer::BufferAllocatorRef::static_ref(),
849        );
850
851        let result = builder.append_values(b"ab", [2usize, 1].into_iter(), &Mask::new_true(2));
852
853        assert!(result.is_err());
854        assert_eq!(builder.offsets.len(), 1);
855    }
856
857    #[test]
858    fn append_decoded_writes_into_the_builder_storage() -> VortexResult<()> {
859        let mut ctx = array_session().create_execution_ctx();
860        let mut builder = VarBinBuilder::<i32>::new_in(
861            DType::Utf8(Nullable),
862            vortex_buffer::BufferAllocatorRef::static_ref(),
863        );
864
865        // SAFETY: the closure initializes exactly the 6 bytes it reports.
866        unsafe {
867            builder.append_decoded(
868                6,
869                4,
870                &[3usize, 0, 3],
871                &Mask::from_iter([true, false, true]),
872                &mut |spare: &mut [MaybeUninit<u8>]| {
873                    for (slot, byte) in spare.iter_mut().zip(b"foobar") {
874                        slot.write(*byte);
875                    }
876                    Ok(6)
877                },
878            )?;
879        }
880
881        let expected =
882            VarBinViewArray::from_iter([Some("foo"), None, Some("bar")], DType::Utf8(Nullable));
883        assert_arrays_eq!(builder.finish_into_varbin(), expected, &mut ctx);
884        Ok(())
885    }
886
887    /// The offset width has to be checked before `decode` runs: an overflowing `i32` builder
888    /// should not pay for a full decompression first.
889    #[test]
890    fn append_decoded_rejects_an_offset_overflow_without_decoding() {
891        let mut builder = VarBinBuilder::<i32>::new_in(
892            DType::Utf8(Nullable),
893            vortex_buffer::BufferAllocatorRef::static_ref(),
894        );
895        let num_bytes = i32::MAX as usize + 1;
896        let mut decoded = false;
897
898        // SAFETY: the closure is never called, and reports only what it initializes if it were.
899        let result = unsafe {
900            builder.append_decoded(num_bytes, 0, &[num_bytes], &Mask::new_true(1), &mut |_| {
901                decoded = true;
902                Ok(0)
903            })
904        };
905
906        assert!(result.is_err());
907        assert!(!decoded, "decode ran before the offset width was checked");
908        assert_eq!(builder.offsets.len(), 1);
909        assert_eq!(builder.validity.len(), 0);
910    }
911
912    /// `append_scalar_repeated` reports an offset overflow rather than panicking, so a constant
913    /// column wider than the offset type errors like the bulk appends do.
914    #[test]
915    fn append_scalar_repeated_rejects_an_offset_overflow() {
916        let mut builder = VarBinBuilder::<i32>::new_in(
917            DType::Utf8(Nullable),
918            vortex_buffer::BufferAllocatorRef::static_ref(),
919        );
920
921        let result = builder
922            .append_scalar_repeated(&Scalar::utf8("hello", Nullable), i32::MAX as usize / 5 + 1);
923
924        assert!(result.is_err());
925        assert_eq!(builder.offsets.len(), 1);
926        assert!(builder.data.is_empty());
927        assert_eq!(builder.validity.len(), 0);
928    }
929
930    #[test]
931    fn append_decoded_rejects_a_short_decode() {
932        let mut builder = VarBinBuilder::<i32>::new_in(
933            DType::Utf8(Nullable),
934            vortex_buffer::BufferAllocatorRef::static_ref(),
935        );
936
937        // SAFETY: the closure initializes the 3 bytes it reports (none, and it reports 3 — but the
938        // length mismatch is rejected before anything is published).
939        let result = unsafe {
940            builder.append_decoded(6, 0, &[3usize, 3], &Mask::new_true(2), &mut |spare| {
941                spare[..3].fill(MaybeUninit::new(b'x'));
942                Ok(3)
943            })
944        };
945
946        assert!(result.is_err());
947        assert_eq!(builder.offsets.len(), 1);
948        assert_eq!(builder.validity.len(), 0);
949    }
950
951    /// Slices that overrun the declared byte count are rejected, and the builder is left exactly
952    /// as it was — the overrun bytes are copied before the total can be checked, so the unwind has
953    /// to put them back.
954    #[rstest]
955    #[case::all_valid(Mask::new_true(2))]
956    #[case::some_valid(Mask::from_iter([true, false, true]))]
957    fn append_valid_slices_rejects_a_byte_count_mismatch(#[case] validity: Mask) {
958        let mut builder = VarBinBuilder::<i32>::new_in(
959            DType::Utf8(Nullable),
960            vortex_buffer::BufferAllocatorRef::static_ref(),
961        );
962        let values = [b"foo".as_slice(), b"quux".as_slice()];
963
964        let mut next = 0;
965        let result = builder.append_valid_slices(6, &validity, |_| {
966            next += 1;
967            values[next - 1]
968        });
969
970        assert!(result.is_err());
971        assert_eq!(builder.offsets.len(), 1);
972        assert!(builder.data.is_empty());
973        assert_eq!(builder.validity.len(), 0);
974    }
975
976    #[test]
977    #[should_panic(expected = "The offset count must be one more than the validity length")]
978    fn finish_rejects_mismatched_validity() {
979        let mut builder = VarBinBuilder::<i32>::new_in(
980            DType::Utf8(Nullable),
981            vortex_buffer::BufferAllocatorRef::static_ref(),
982        );
983        builder.validity.append_true();
984        drop(builder.finish_into_varbin());
985    }
986
987    #[rstest]
988    #[case(false)]
989    #[case(true)]
990    fn test_array_builder_methods(#[case] large_offsets: bool) -> VortexResult<()> {
991        let mut ctx = array_session().create_execution_ctx();
992        let result = with_offsets(large_offsets, DType::Utf8(Nullable), |builder| {
993            builder.reserve_exact(3);
994            builder.append_zero();
995            builder.append_scalar(&Scalar::utf8("hello", Nullable))?;
996            builder.append_null();
997            assert_eq!(builder.len(), 3);
998            Ok(())
999        })?;
1000
1001        assert_eq!(
1002            result.validity()?.execute_mask(3, &mut ctx)?,
1003            Mask::from_iter([true, true, false])
1004        );
1005        assert_arrays_eq!(
1006            result,
1007            VarBinViewArray::from_iter_nullable_str([Some(""), Some("hello"), None]),
1008            &mut ctx
1009        );
1010        Ok(())
1011    }
1012
1013    #[rstest]
1014    #[case(false)]
1015    #[case(true)]
1016    fn test_append_varbinview_validity_to_builder(#[case] large_offsets: bool) -> VortexResult<()> {
1017        let long = "a value that does not fit inline";
1018        let all_null = VarBinViewArray::from_iter([None::<&str>, None], DType::Utf8(Nullable));
1019        let mixed =
1020            VarBinViewArray::from_iter([Some("hello"), None, Some(long)], DType::Utf8(Nullable));
1021        let expected = VarBinViewArray::from_iter(
1022            [None, None, Some("hello"), None, Some(long)],
1023            DType::Utf8(Nullable),
1024        );
1025        let mut ctx = array_session().create_execution_ctx();
1026
1027        let actual = with_offsets(large_offsets, expected.dtype().clone(), |builder| {
1028            all_null
1029                .clone()
1030                .into_array()
1031                .append_to_builder(builder, &mut ctx)?;
1032            mixed
1033                .clone()
1034                .into_array()
1035                .append_to_builder(builder, &mut ctx)
1036        })?;
1037
1038        assert_arrays_eq!(actual, expected, &mut ctx);
1039        Ok(())
1040    }
1041
1042    /// `match_each_varbin_builder!` covers every offset width a `VarBinBuilder` can be built with,
1043    /// so both canonical appends must reach their specialization for all of them. A macro that
1044    /// only matched the signed pair would send an unsigned builder down a downcast that assumes
1045    /// `VarBinViewBuilder` and panic.
1046    #[rstest]
1047    #[case::u32(VarBinBuilder::<u32>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1048    #[case::u64(VarBinBuilder::<u64>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1049    #[case::i32(VarBinBuilder::<i32>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1050    #[case::i64(VarBinBuilder::<i64>::new_in(DType::Utf8(Nullable), vortex_buffer::BufferAllocatorRef::static_ref()))]
1051    fn append_to_every_offset_width(#[case] mut builder: impl ArrayBuilder) -> VortexResult<()> {
1052        let mut ctx = array_session().create_execution_ctx();
1053        let long = "a string that is far too long to be inlined in a view";
1054        let values = [Some("hello"), None, Some(long), Some("")];
1055
1056        let view = VarBinViewArray::from_iter(values, DType::Utf8(Nullable)).into_array();
1057        let varbin = VarBinArray::from_iter(values, DType::Utf8(Nullable)).into_array();
1058        let constant = ConstantArray::new(Scalar::from("hello").into_nullable(), 2).into_array();
1059
1060        for chunk in [&view, &varbin, &constant] {
1061            chunk.append_to_builder(&mut builder, &mut ctx)?;
1062        }
1063
1064        let expected = VarBinViewArray::from_iter_nullable_str([
1065            Some("hello"),
1066            None,
1067            Some(long),
1068            Some(""),
1069            Some("hello"),
1070            None,
1071            Some(long),
1072            Some(""),
1073            Some("hello"),
1074            Some("hello"),
1075        ]);
1076        assert_arrays_eq!(builder.finish(), expected, &mut ctx);
1077        Ok(())
1078    }
1079
1080    #[test]
1081    fn offsets_have_is_sorted_stat() -> VortexResult<()> {
1082        let mut builder = VarBinBuilder::<i32>::with_capacity_in(
1083            DType::Utf8(Nullable),
1084            0,
1085            vortex_buffer::BufferAllocatorRef::static_ref(),
1086        );
1087        builder.append_value(b"aaa");
1088        builder.push_null();
1089        builder.append_value(b"bbb");
1090        let array = builder.finish_into_varbin();
1091
1092        let is_sorted = array
1093            .offsets()
1094            .statistics()
1095            .with_typed_stats_set(|s| s.get_as::<bool>(Stat::IsSorted));
1096        assert_eq!(is_sorted, Precision::Exact(true));
1097        Ok(())
1098    }
1099
1100    #[test]
1101    fn empty_builder_offsets_have_is_sorted_stat() -> VortexResult<()> {
1102        let mut builder = VarBinBuilder::<i32>::new_in(
1103            DType::Utf8(Nullable),
1104            vortex_buffer::BufferAllocatorRef::static_ref(),
1105        );
1106        let array = builder.finish_into_varbin();
1107
1108        let is_sorted = array
1109            .offsets()
1110            .statistics()
1111            .with_typed_stats_set(|s| s.get_as::<bool>(Stat::IsSorted));
1112        assert_eq!(is_sorted, Precision::Exact(true));
1113        Ok(())
1114    }
1115
1116    /// Runs `f` against an `i32` or `i64` builder and returns the finished array.
1117    fn with_offsets(
1118        large_offsets: bool,
1119        dtype: DType,
1120        f: impl FnOnce(&mut dyn ArrayBuilder) -> VortexResult<()>,
1121    ) -> VortexResult<VarBinArray> {
1122        if large_offsets {
1123            let mut builder = VarBinBuilder::<i64>::with_capacity_in(
1124                dtype,
1125                8,
1126                vortex_buffer::BufferAllocatorRef::static_ref(),
1127            );
1128            f(&mut builder)?;
1129            Ok(builder.finish_into_varbin())
1130        } else {
1131            let mut builder = VarBinBuilder::<i32>::with_capacity_in(
1132                dtype,
1133                8,
1134                vortex_buffer::BufferAllocatorRef::static_ref(),
1135            );
1136            f(&mut builder)?;
1137            Ok(builder.finish_into_varbin())
1138        }
1139    }
1140}