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