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