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    fn replace_validity(&mut self, validity: Mask) {
460        self.validity = match validity {
461            Mask::AllTrue(len) => BitBufferMut::new_set(len),
462            Mask::AllFalse(len) => BitBufferMut::new_unset(len),
463            values @ Mask::Values(_) => values
464                .into_bit_buffer()
465                .try_into_mut()
466                .unwrap_or_else(|buffer| BitBufferMut::copy_from(&buffer)),
467        };
468    }
469
470    /// Appends `count` end offsets derived from `end_offsets`, shifted past the current data end.
471    ///
472    /// `end_offsets` must be monotonically non-decreasing and end at exactly `num_bytes`. Offsets
473    /// are written through the reserved spare capacity and committed with a single `set_len`, so a
474    /// rejected input leaves the buffer untouched and the caller can propagate the error.
475    fn extend_offsets<P>(
476        &mut self,
477        num_bytes: usize,
478        count: usize,
479        end_offsets: impl Iterator<Item = P>,
480    ) -> VortexResult<()>
481    where
482        P: AsPrimitive<usize>,
483        usize: AsPrimitive<O>,
484    {
485        let data_start = self.data.len();
486        let offsets_len = self.offsets.len();
487        self.check_offset_limit(data_start, num_bytes)?;
488        self.offsets.reserve(count);
489
490        // Writing into the spare capacity keeps the output cursor in a register: `push` rewrites
491        // the buffer length on every value, which the optimizer cannot hoist out of the loop.
492        let spare = &mut self.offsets.spare_capacity_mut()[..count];
493        let mut end_offsets = end_offsets;
494        let mut previous = 0usize;
495        for slot in spare.iter_mut() {
496            let Some(end) = end_offsets.next() else {
497                vortex_bail!("End offset count is less than the validity length {count}");
498            };
499            let end = end.as_();
500            vortex_ensure!(
501                end >= previous && end <= num_bytes,
502                "End offsets must be monotonically increasing within {num_bytes} bytes, \
503                 got {end} after {previous}"
504            );
505            slot.write((data_start + end).as_());
506            previous = end;
507        }
508        vortex_ensure!(
509            end_offsets.next().is_none(),
510            "End offset count exceeds the validity length {count}"
511        );
512        vortex_ensure!(
513            previous == num_bytes,
514            "Final end offset {previous} does not match the value byte count {num_bytes}"
515        );
516
517        // SAFETY: the loop initialized the first `count` spare slots.
518        unsafe { self.offsets.set_len(offsets_len + count) };
519        Ok(())
520    }
521
522    /// Copies each valid row's bytes from `value` into the byte storage and records the offsets.
523    /// See [`append_valid_slices`](Self::append_valid_slices); the validity bits are the caller's
524    /// job so that a failure here can be unwound.
525    fn gather_valid_slices<'a, F>(
526        &mut self,
527        num_bytes: usize,
528        validity: &Mask,
529        mut value: F,
530    ) -> VortexResult<()>
531    where
532        F: FnMut(usize) -> &'a [u8],
533        usize: AsPrimitive<O>,
534    {
535        let count = validity.len();
536        let data_start = self.data.len();
537        let offsets_len = self.offsets.len();
538        self.check_offset_limit(data_start, num_bytes)?;
539        self.offsets.reserve(count);
540        self.data.reserve(num_bytes);
541
542        // Disjoint field borrows: the offsets spare capacity stays valid while the byte buffer
543        // grows, since the two are separate allocations.
544        let Self { offsets, data, .. } = self;
545        let spare = &mut offsets.spare_capacity_mut()[..count];
546
547        match validity.bit_buffer() {
548            AllOr::All => {
549                for (row, slot) in spare.iter_mut().enumerate() {
550                    data.extend_from_slice(value(row));
551                    slot.write(data.len().as_());
552                }
553            }
554            AllOr::None => {
555                spare.fill(MaybeUninit::new(data_start.as_()));
556            }
557            AllOr::Some(bits) => {
558                let mut row = 0;
559                bits.for_each_set_index(|index| {
560                    // Null rows between the previous valid row and this one repeat its end offset.
561                    spare[row..index].fill(MaybeUninit::new(data.len().as_()));
562                    data.extend_from_slice(value(index));
563                    spare[index].write(data.len().as_());
564                    row = index + 1;
565                });
566                spare[row..].fill(MaybeUninit::new(data.len().as_()));
567            }
568        }
569
570        // A caller whose slices overrun `num_bytes` only grows the byte buffer past the reservation,
571        // so the overrun is caught here rather than per value. The offsets are still uncommitted, so
572        // rejecting it now leaves them untouched.
573        vortex_ensure!(
574            data.len() == data_start + num_bytes,
575            "Value slices total {} bytes, expected {num_bytes}",
576            data.len() - data_start
577        );
578
579        // SAFETY: every branch above initialized all `count` spare slots.
580        unsafe { self.offsets.set_len(offsets_len + count) };
581        Ok(())
582    }
583
584    /// Checks that an offset past `num_bytes` more bytes is representable as an `O`.
585    fn check_offset_limit(&self, data_start: usize, num_bytes: usize) -> VortexResult<()> {
586        let Some(limit) = data_start.checked_add(num_bytes) else {
587            vortex_bail!("Byte offset overflow: {data_start} + {num_bytes}");
588        };
589        vortex_ensure!(
590            u64::try_from(limit).is_ok_and(|limit| limit <= O::max_value_as_u64()),
591            "Byte offset {limit} does not fit in {}",
592            std::any::type_name::<O>()
593        );
594        Ok(())
595    }
596}
597
598impl<O: OffsetBuilderPType> ArrayBuilder for VarBinBuilder<O> {
599    fn as_any(&self) -> &dyn Any {
600        self
601    }
602
603    fn as_any_mut(&mut self) -> &mut dyn Any {
604        self
605    }
606
607    fn dtype(&self) -> &DType {
608        &self.dtype
609    }
610
611    fn len(&self) -> usize {
612        self.validity.len()
613    }
614
615    fn append_zeros(&mut self, n: usize) {
616        self.offsets.push_n(self.last_offset(), n);
617        self.validity.append_n(true, n);
618    }
619
620    unsafe fn append_nulls_unchecked(&mut self, n: usize) {
621        self.push_nulls(n);
622    }
623
624    fn append_scalar(&mut self, scalar: &Scalar) -> VortexResult<()> {
625        self.append_scalar_repeated(scalar, 1)
626    }
627
628    fn reserve_exact(&mut self, additional: usize) {
629        self.offsets.reserve(additional);
630        self.validity.reserve(additional);
631    }
632
633    unsafe fn set_validity_unchecked(&mut self, validity: Mask) {
634        self.replace_validity(validity)
635    }
636
637    fn finish(&mut self) -> ArrayRef {
638        self.finish_into_varbin().into_array()
639    }
640
641    fn finish_into_canonical(&mut self, ctx: &mut ExecutionCtx) -> Canonical {
642        self.finish()
643            .execute::<Canonical>(ctx)
644            .vortex_expect("varbin builder should canonicalize")
645    }
646}
647
648/// Recovers a concrete [`VarBinBuilder`] from a `&mut dyn ArrayBuilder`.
649///
650/// Evaluates to `Some($body)` with `$typed` bound to the `&mut VarBinBuilder<O>` when the builder
651/// is one, and `None` otherwise, so an encoding can fall through to its other output paths:
652///
653/// ```ignore
654/// if let Some(result) = match_each_varbin_builder!(builder, |builder| {
655///     append_my_encoding(array, builder, ctx)
656/// }) {
657///     return result;
658/// }
659/// ```
660#[macro_export]
661macro_rules! match_each_varbin_builder {
662    ($builder:expr, | $typed:ident | $body:expr) => {
663        $crate::__match_varbin_builder_widths!($builder, |$typed| $body, [u32, u64, i32, i64])
664    };
665}
666
667/// Expands `$body` once per listed offset width, guarded by a downcast. See
668/// [`match_each_varbin_builder!`].
669#[doc(hidden)]
670#[macro_export]
671macro_rules! __match_varbin_builder_widths {
672    ($builder:expr, | $typed:ident | $body:expr, [$($width:ty),+ $(,)?]) => {{
673        let __varbin_builder: &mut dyn $crate::builders::ArrayBuilder = $builder;
674        $crate::__match_varbin_builder_arms!(__varbin_builder, |$typed| $body, [$($width),+])
675    }};
676}
677
678/// The `if`/`else if` chain behind [`__match_varbin_builder_widths!`], one arm per width.
679#[doc(hidden)]
680#[macro_export]
681macro_rules! __match_varbin_builder_arms {
682    ($builder:expr, | $typed:ident | $body:expr, []) => {
683        None
684    };
685    ($builder:expr, | $typed:ident | $body:expr, [$head:ty $(, $tail:ty)*]) => {
686        if $builder
687            .as_any()
688            .is::<$crate::builders::VarBinBuilder<$head>>()
689        {
690            let $typed = match $builder
691                .as_any_mut()
692                .downcast_mut::<$crate::builders::VarBinBuilder<$head>>()
693            {
694                Some(typed) => typed,
695                None => unreachable!("builder type checked above"),
696            };
697            Some($body)
698        } else {
699            $crate::__match_varbin_builder_arms!($builder, |$typed| $body, [$($tail),*])
700        }
701    };
702}
703
704/// Running totals of `lengths`, wrapping so a corrupt lengths child is rejected rather than
705/// panicking; [`VarBinBuilder::extend_offsets`] catches the resulting non-monotonic offsets.
706#[inline]
707fn prefix_sums<P: AsPrimitive<usize>>(lengths: &[P]) -> impl Iterator<Item = usize> {
708    lengths.iter().scan(0usize, |end, length| {
709        *end = end.wrapping_add(length.as_());
710        Some(*end)
711    })
712}
713
714#[cfg(test)]
715mod tests {
716    use std::mem::MaybeUninit;
717
718    use rstest::rstest;
719    use vortex_error::VortexResult;
720    use vortex_mask::Mask;
721
722    use crate::IntoArray;
723    use crate::VortexSessionExecute;
724    use crate::array_session;
725    use crate::arrays::ChunkedArray;
726    use crate::arrays::ConstantArray;
727    use crate::arrays::VarBinArray;
728    use crate::arrays::VarBinViewArray;
729    use crate::arrays::varbin::VarBinArraySlotsExt;
730    use crate::arrays::varbin::builder::VarBinBuilder;
731    use crate::assert_arrays_eq;
732    use crate::builders::ArrayBuilder;
733    use crate::dtype::DType;
734    use crate::dtype::Nullability::Nullable;
735    use crate::expr::stats::Precision;
736    use crate::expr::stats::Stat;
737    use crate::expr::stats::StatsProviderExt;
738    use crate::scalar::Scalar;
739
740    #[test]
741    fn test_builder() {
742        let mut builder = VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullable), 0);
743        builder.append(Some(b"hello"));
744        builder.append(None);
745        builder.append(Some(b"world"));
746        let array = builder.finish_into_varbin();
747
748        assert_eq!(array.len(), 3);
749        assert_eq!(array.dtype().nullability(), Nullable);
750        assert_eq!(
751            array
752                .execute_scalar(0, &mut array_session().create_execution_ctx())
753                .unwrap(),
754            Scalar::utf8("hello".to_string(), Nullable)
755        );
756        assert!(
757            array
758                .execute_scalar(1, &mut array_session().create_execution_ctx())
759                .unwrap()
760                .is_null()
761        );
762    }
763
764    #[rstest]
765    #[case(false)]
766    #[case(true)]
767    fn test_append_varbin_to_builder(#[case] large_offsets: bool) -> VortexResult<()> {
768        let source = VarBinArray::from_iter(
769            [
770                Some("prefix"),
771                Some("hello"),
772                None,
773                Some("world"),
774                Some("suffix"),
775            ],
776            DType::Utf8(Nullable),
777        )
778        .into_array()
779        .slice(1..4)?;
780        let mut ctx = array_session().create_execution_ctx();
781
782        let actual = with_offsets(large_offsets, source.dtype().clone(), |builder| {
783            source.append_to_builder(builder, &mut ctx)
784        })?;
785
786        assert_arrays_eq!(actual, source, &mut ctx);
787        Ok(())
788    }
789
790    #[test]
791    fn append_n_values_offset_overflow_returns_error() {
792        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
793
794        // The limit is checked before anything is reserved, so the huge byte total is free.
795        let result = builder.append_n_values(b"hello", i32::MAX as usize / 5 + 1);
796
797        assert!(result.is_err());
798        assert_eq!(builder.offsets.len(), 1);
799        assert!(builder.data.is_empty());
800        assert_eq!(builder.validity.len(), 0);
801    }
802
803    #[test]
804    fn append_values_rejects_a_short_offset_count() {
805        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
806
807        let result = builder.append_values(b"ab", [1usize, 2].into_iter(), &Mask::new_true(3));
808
809        assert!(result.is_err());
810        assert_eq!(builder.offsets.len(), 1);
811        assert!(builder.data.is_empty());
812    }
813
814    #[test]
815    fn append_values_rejects_non_monotonic_offsets() {
816        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
817
818        let result = builder.append_values(b"ab", [2usize, 1].into_iter(), &Mask::new_true(2));
819
820        assert!(result.is_err());
821        assert_eq!(builder.offsets.len(), 1);
822    }
823
824    #[test]
825    fn append_decoded_writes_into_the_builder_storage() -> VortexResult<()> {
826        let mut ctx = array_session().create_execution_ctx();
827        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
828
829        // SAFETY: the closure initializes exactly the 6 bytes it reports.
830        unsafe {
831            builder.append_decoded(
832                6,
833                4,
834                &[3usize, 0, 3],
835                &Mask::from_iter([true, false, true]),
836                &mut |spare: &mut [MaybeUninit<u8>]| {
837                    for (slot, byte) in spare.iter_mut().zip(b"foobar") {
838                        slot.write(*byte);
839                    }
840                    Ok(6)
841                },
842            )?;
843        }
844
845        let expected =
846            VarBinViewArray::from_iter([Some("foo"), None, Some("bar")], DType::Utf8(Nullable));
847        assert_arrays_eq!(builder.finish_into_varbin(), expected, &mut ctx);
848        Ok(())
849    }
850
851    /// The offset width has to be checked before `decode` runs: an overflowing `i32` builder
852    /// should not pay for a full decompression first.
853    #[test]
854    fn append_decoded_rejects_an_offset_overflow_without_decoding() {
855        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
856        let num_bytes = i32::MAX as usize + 1;
857        let mut decoded = false;
858
859        // SAFETY: the closure is never called, and reports only what it initializes if it were.
860        let result = unsafe {
861            builder.append_decoded(num_bytes, 0, &[num_bytes], &Mask::new_true(1), &mut |_| {
862                decoded = true;
863                Ok(0)
864            })
865        };
866
867        assert!(result.is_err());
868        assert!(!decoded, "decode ran before the offset width was checked");
869        assert_eq!(builder.offsets.len(), 1);
870        assert_eq!(builder.validity.len(), 0);
871    }
872
873    /// `append_scalar_repeated` reports an offset overflow rather than panicking, so a constant
874    /// column wider than the offset type errors like the bulk appends do.
875    #[test]
876    fn append_scalar_repeated_rejects_an_offset_overflow() {
877        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
878
879        let result = builder
880            .append_scalar_repeated(&Scalar::utf8("hello", Nullable), i32::MAX as usize / 5 + 1);
881
882        assert!(result.is_err());
883        assert_eq!(builder.offsets.len(), 1);
884        assert!(builder.data.is_empty());
885        assert_eq!(builder.validity.len(), 0);
886    }
887
888    #[test]
889    fn append_decoded_rejects_a_short_decode() {
890        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
891
892        // SAFETY: the closure initializes the 3 bytes it reports (none, and it reports 3 — but the
893        // length mismatch is rejected before anything is published).
894        let result = unsafe {
895            builder.append_decoded(6, 0, &[3usize, 3], &Mask::new_true(2), &mut |spare| {
896                spare[..3].fill(MaybeUninit::new(b'x'));
897                Ok(3)
898            })
899        };
900
901        assert!(result.is_err());
902        assert_eq!(builder.offsets.len(), 1);
903        assert_eq!(builder.validity.len(), 0);
904    }
905
906    /// Slices that overrun the declared byte count are rejected, and the builder is left exactly
907    /// as it was — the overrun bytes are copied before the total can be checked, so the unwind has
908    /// to put them back.
909    #[rstest]
910    #[case::all_valid(Mask::new_true(2))]
911    #[case::some_valid(Mask::from_iter([true, false, true]))]
912    fn append_valid_slices_rejects_a_byte_count_mismatch(#[case] validity: Mask) {
913        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
914        let values = [b"foo".as_slice(), b"quux".as_slice()];
915
916        let mut next = 0;
917        let result = builder.append_valid_slices(6, &validity, |_| {
918            next += 1;
919            values[next - 1]
920        });
921
922        assert!(result.is_err());
923        assert_eq!(builder.offsets.len(), 1);
924        assert!(builder.data.is_empty());
925        assert_eq!(builder.validity.len(), 0);
926    }
927
928    #[test]
929    #[should_panic(expected = "The offset count must be one more than the validity length")]
930    fn finish_rejects_mismatched_validity() {
931        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
932        builder.validity.append_true();
933        drop(builder.finish_into_varbin());
934    }
935
936    #[rstest]
937    #[case(false)]
938    #[case(true)]
939    fn test_array_builder_methods(#[case] large_offsets: bool) -> VortexResult<()> {
940        let mut ctx = array_session().create_execution_ctx();
941        for validity in [
942            Mask::new_true(3),
943            Mask::new_false(3),
944            Mask::from_iter([true, false, true]),
945        ] {
946            let result = with_offsets(large_offsets, DType::Utf8(Nullable), |builder| {
947                builder.reserve_exact(3);
948                builder.append_zero();
949                builder.append_scalar(&Scalar::utf8("hello", Nullable))?;
950                builder.append_null();
951                assert_eq!(builder.len(), 3);
952                builder.set_validity(validity.clone());
953                Ok(())
954            })?;
955            assert_eq!(result.validity()?.execute_mask(3, &mut ctx)?, validity);
956        }
957        Ok(())
958    }
959
960    #[rstest]
961    #[case(false)]
962    #[case(true)]
963    fn test_append_varbinview_validity_to_builder(#[case] large_offsets: bool) -> VortexResult<()> {
964        let long = "a value that does not fit inline";
965        let all_null = VarBinViewArray::from_iter([None::<&str>, None], DType::Utf8(Nullable));
966        let mixed =
967            VarBinViewArray::from_iter([Some("hello"), None, Some(long)], DType::Utf8(Nullable));
968        let expected = VarBinViewArray::from_iter(
969            [None, None, Some("hello"), None, Some(long)],
970            DType::Utf8(Nullable),
971        );
972        let mut ctx = array_session().create_execution_ctx();
973
974        let actual = with_offsets(large_offsets, expected.dtype().clone(), |builder| {
975            all_null
976                .clone()
977                .into_array()
978                .append_to_builder(builder, &mut ctx)?;
979            mixed
980                .clone()
981                .into_array()
982                .append_to_builder(builder, &mut ctx)
983        })?;
984
985        assert_arrays_eq!(actual, expected, &mut ctx);
986        Ok(())
987    }
988
989    /// `match_each_varbin_builder!` covers every offset width a `VarBinBuilder` can be built with,
990    /// so both canonical appends must reach their specialization for all of them. A macro that
991    /// only matched the signed pair would send an unsigned builder down a downcast that assumes
992    /// `VarBinViewBuilder` and panic.
993    #[rstest]
994    #[case::u32(VarBinBuilder::<u32>::new(DType::Utf8(Nullable)))]
995    #[case::u64(VarBinBuilder::<u64>::new(DType::Utf8(Nullable)))]
996    #[case::i32(VarBinBuilder::<i32>::new(DType::Utf8(Nullable)))]
997    #[case::i64(VarBinBuilder::<i64>::new(DType::Utf8(Nullable)))]
998    fn append_to_every_offset_width(#[case] mut builder: impl ArrayBuilder) -> VortexResult<()> {
999        let mut ctx = array_session().create_execution_ctx();
1000        let long = "a string that is far too long to be inlined in a view";
1001        let values = [Some("hello"), None, Some(long), Some("")];
1002
1003        let view = VarBinViewArray::from_iter(values, DType::Utf8(Nullable)).into_array();
1004        let varbin = VarBinArray::from_iter(values, DType::Utf8(Nullable)).into_array();
1005        let constant = ConstantArray::new(Scalar::from("hello").into_nullable(), 2).into_array();
1006
1007        for chunk in [&view, &varbin, &constant] {
1008            chunk.append_to_builder(&mut builder, &mut ctx)?;
1009        }
1010
1011        let expected = ChunkedArray::try_new(vec![view, varbin, constant], DType::Utf8(Nullable))?;
1012        assert_arrays_eq!(builder.finish(), expected, &mut ctx);
1013        Ok(())
1014    }
1015
1016    #[test]
1017    fn offsets_have_is_sorted_stat() -> VortexResult<()> {
1018        let mut builder = VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullable), 0);
1019        builder.append_value(b"aaa");
1020        builder.push_null();
1021        builder.append_value(b"bbb");
1022        let array = builder.finish_into_varbin();
1023
1024        let is_sorted = array
1025            .offsets()
1026            .statistics()
1027            .with_typed_stats_set(|s| s.get_as::<bool>(Stat::IsSorted));
1028        assert_eq!(is_sorted, Precision::Exact(true));
1029        Ok(())
1030    }
1031
1032    #[test]
1033    fn empty_builder_offsets_have_is_sorted_stat() -> VortexResult<()> {
1034        let mut builder = VarBinBuilder::<i32>::new(DType::Utf8(Nullable));
1035        let array = builder.finish_into_varbin();
1036
1037        let is_sorted = array
1038            .offsets()
1039            .statistics()
1040            .with_typed_stats_set(|s| s.get_as::<bool>(Stat::IsSorted));
1041        assert_eq!(is_sorted, Precision::Exact(true));
1042        Ok(())
1043    }
1044
1045    /// Runs `f` against an `i32` or `i64` builder and returns the finished array.
1046    fn with_offsets(
1047        large_offsets: bool,
1048        dtype: DType,
1049        f: impl FnOnce(&mut dyn ArrayBuilder) -> VortexResult<()>,
1050    ) -> VortexResult<VarBinArray> {
1051        if large_offsets {
1052            let mut builder = VarBinBuilder::<i64>::with_capacity(dtype, 8);
1053            f(&mut builder)?;
1054            Ok(builder.finish_into_varbin())
1055        } else {
1056            let mut builder = VarBinBuilder::<i32>::with_capacity(dtype, 8);
1057            f(&mut builder)?;
1058            Ok(builder.finish_into_varbin())
1059        }
1060    }
1061}