Skip to main content

vortex_array/builders/
varbinview.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::any::Any;
5use std::ops::Range;
6use std::sync::Arc;
7
8use itertools::Itertools;
9use num_traits::AsPrimitive;
10use vortex_buffer::Alignment;
11use vortex_buffer::Buffer;
12use vortex_buffer::BufferMut;
13use vortex_buffer::ByteBuffer;
14use vortex_buffer::ByteBufferMut;
15use vortex_error::VortexExpect;
16use vortex_error::VortexResult;
17use vortex_error::vortex_bail;
18use vortex_error::vortex_ensure;
19use vortex_mask::AllOr;
20use vortex_mask::Mask;
21use vortex_utils::aliases::hash_map::Entry;
22use vortex_utils::aliases::hash_map::HashMap;
23
24use crate::ArrayRef;
25use crate::ExecutionCtx;
26use crate::IntoArray;
27use crate::arrays::VarBinViewArray;
28use crate::arrays::varbinview::VarBinViewArrayExt;
29use crate::arrays::varbinview::build_views::BinaryView;
30use crate::arrays::varbinview::build_views::MAX_BUFFER_LEN;
31use crate::arrays::varbinview::build_views::extend_views;
32use crate::arrays::varbinview::compact::BufferUtilization;
33use crate::builders::ArrayBuilder;
34use crate::builders::LazyBitBufferBuilder;
35use crate::canonical::Canonical;
36use crate::dtype::DType;
37use crate::dtype::NativePType;
38use crate::scalar::Scalar;
39
40/// The builder for building a [`VarBinViewArray`].
41pub struct VarBinViewBuilder {
42    dtype: DType,
43    views_builder: BufferMut<BinaryView>,
44    nulls: LazyBitBufferBuilder,
45    completed: CompletedBuffers,
46    in_progress: Option<ByteBufferMut>,
47    growth_strategy: BufferGrowthStrategy,
48    compaction_threshold: f64,
49}
50
51impl VarBinViewBuilder {
52    pub fn with_capacity(dtype: DType, capacity: usize) -> Self {
53        Self::new(dtype, capacity, Default::default(), Default::default(), 0.0)
54    }
55
56    pub fn with_buffer_deduplication(dtype: DType, capacity: usize) -> Self {
57        Self::new(
58            dtype,
59            capacity,
60            CompletedBuffers::Deduplicated(Default::default()),
61            Default::default(),
62            0.0,
63        )
64    }
65
66    pub fn with_compaction(dtype: DType, capacity: usize, compaction_threshold: f64) -> Self {
67        Self::new(
68            dtype,
69            capacity,
70            Default::default(),
71            Default::default(),
72            compaction_threshold,
73        )
74    }
75
76    pub fn new(
77        dtype: DType,
78        capacity: usize,
79        completed: CompletedBuffers,
80        growth_strategy: BufferGrowthStrategy,
81        compaction_threshold: f64,
82    ) -> Self {
83        assert!(
84            matches!(dtype, DType::Utf8(_) | DType::Binary(_)),
85            "VarBinViewBuilder DType must be Utf8 or Binary."
86        );
87        Self {
88            views_builder: BufferMut::with_capacity_preferred_aligned(
89                capacity,
90                Alignment::of::<BinaryView>(),
91                None,
92            ),
93            nulls: LazyBitBufferBuilder::new(capacity),
94            completed,
95            in_progress: None,
96            dtype,
97            growth_strategy,
98            compaction_threshold,
99        }
100    }
101
102    fn append_value_view(&mut self, value: &[u8]) {
103        let length =
104            u32::try_from(value.len()).vortex_expect("cannot have a single string >2^32 in length");
105        if length <= 12 {
106            self.views_builder.push(BinaryView::make_view(value, 0, 0));
107            return;
108        }
109
110        let (buffer_idx, offset) = self.append_value_to_buffer(value);
111        let view = BinaryView::make_view(value, buffer_idx, offset);
112        self.views_builder.push(view);
113    }
114
115    /// Appends a value to the builder.
116    pub fn append_value<S: AsRef<[u8]>>(&mut self, value: S) {
117        self.append_value_view(value.as_ref());
118        self.nulls.append_non_null();
119    }
120
121    /// Appends `n` copies of `value` as non-null entries.
122    pub fn append_n_values<S: AsRef<[u8]>>(&mut self, value: S, n: usize) {
123        if n == 0 {
124            return;
125        }
126        let bytes = value.as_ref();
127        let view = if bytes.len() <= BinaryView::MAX_INLINED_SIZE {
128            BinaryView::make_view(bytes, 0, 0)
129        } else {
130            let (buffer_idx, offset) = self.append_value_to_buffer(bytes);
131            BinaryView::make_view(bytes, buffer_idx, offset)
132        };
133        self.views_builder.push_n(view, n);
134        self.nulls.append_n_non_nulls(n);
135    }
136
137    fn flush_in_progress(&mut self) {
138        let Some(block) = self.in_progress.take() else {
139            return;
140        };
141
142        assert!(block.len() < u32::MAX as usize, "Block too large");
143
144        let initial_len = self.completed.len();
145        self.completed.push(block.freeze());
146        assert_eq!(
147            self.completed.len(),
148            initial_len + 1,
149            "Invalid state, just completed block already exists"
150        );
151    }
152
153    fn init_in_progress(&mut self, min_len: usize) {
154        let next_buffer_size = self.growth_strategy.next_size() as usize;
155        let to_reserve = next_buffer_size.max(min_len);
156        self.in_progress = Some(ByteBufferMut::with_capacity_preferred_aligned(
157            to_reserve,
158            Alignment::of::<u8>(),
159            None,
160        ));
161    }
162
163    /// append a non inlined value to self.in_progress.
164    fn append_value_to_buffer(&mut self, value: &[u8]) -> (u32, u32) {
165        assert!(
166            value.len() > BinaryView::MAX_INLINED_SIZE,
167            "must inline small strings"
168        );
169
170        if let Some(in_progress) = &mut self.in_progress {
171            let required_cap = in_progress.len() + value.len();
172            if in_progress.capacity() < required_cap {
173                self.flush_in_progress();
174                self.init_in_progress(value.len());
175            }
176        } else {
177            self.init_in_progress(value.len())
178        };
179
180        let in_progress = self
181            .in_progress
182            .as_mut()
183            .vortex_expect("in_progress just set");
184
185        let buffer_idx = self.completed.len();
186        let offset = u32::try_from(in_progress.len()).vortex_expect("too many buffers");
187        in_progress.extend_from_slice(value);
188
189        (buffer_idx, offset)
190    }
191
192    pub fn completed_block_count(&self) -> u32 {
193        self.completed.len()
194    }
195
196    /// Whether this builder compacts the data buffers it is handed. The lengths-driven and
197    /// gather appends use this to gate their utilization measurement; the buffer-adopting escape
198    /// hatches ([`append_views_built_at`](Self::append_views_built_at) and
199    /// [`push_buffers`](Self::push_buffers)) always bypass it.
200    fn compacts_buffers(&self) -> bool {
201        self.compaction_threshold > 0.0
202    }
203
204    /// Adopts the buffers and views that `build` produces against the index its first buffer
205    /// will land at.
206    ///
207    /// The builder flushes its staged bytes, then hands `build` the index the next data buffer
208    /// will occupy; `build` returns data buffers — which land contiguously from that index — and
209    /// one view per entry of `validity`, already referencing them. This is the escape hatch for
210    /// an encoding that only discovers its views while walking its own byte format (e.g.
211    /// length-prefixed frames), where the lengths-driven appends cannot apply; keeping the
212    /// numbering inside this call is what makes the views come out right without a rebase pass.
213    ///
214    /// # Warning
215    ///
216    /// This method does not check utilization of the returned buffers. `build` must return
217    /// buffers that are fully utilized by its views.
218    ///
219    /// # Panics
220    ///
221    /// Panics if `build` returns a different view count than `validity.len()`, or if this
222    /// builder deduplicates buffers and already holds one of the returned buffers.
223    pub fn append_views_built_at(
224        &mut self,
225        validity: &Mask,
226        build: impl FnOnce(u32) -> VortexResult<(Vec<ByteBuffer>, Buffer<BinaryView>)>,
227    ) -> VortexResult<()> {
228        self.flush_in_progress();
229
230        let start_index = self.completed.len();
231        let (buffers, views) = build(start_index)?;
232        assert_eq!(
233            views.len(),
234            validity.len(),
235            "Must build one view per validity entry"
236        );
237
238        let expected_completed_len = start_index as usize + buffers.len();
239        self.completed.extend_from_slice_unchecked(&buffers);
240        assert_eq!(
241            self.completed.len() as usize,
242            expected_completed_len,
243            "Some buffers already exist",
244        );
245        self.views_builder.extend_trusted(views.iter().copied());
246        self.nulls.append_validity_mask(validity);
247
248        debug_assert_eq!(self.nulls.len(), self.views_builder.len());
249        Ok(())
250    }
251
252    /// Adopts `buffers` as completed data buffers, returning the index each landed at.
253    ///
254    /// Views appended afterwards (e.g. via [`append_views_gathered`](Self::append_views_gathered)
255    /// or [`append_views_scattered`](Self::append_views_scattered)) reference values through the
256    /// returned indices. A deduplicating builder returns the existing index for a buffer it
257    /// already holds, so repeated appends over shared storage — chunks gathered through one
258    /// dictionary, slices of one array — adopt it once.
259    ///
260    /// # Warning
261    ///
262    /// Buffers are taken as they are, without utilization measurement; like
263    /// [`append_views_built_at`](Self::append_views_built_at), this bypasses any compaction the
264    /// builder was configured for. Scattered appends deliberately have these semantics;
265    /// compacting paths that measure utilization use
266    /// [`push_buffers_compacted`](Self::push_buffers_compacted) instead.
267    fn push_buffers(&mut self, buffers: impl IntoIterator<Item = ByteBuffer>) -> Vec<u32> {
268        self.flush_in_progress();
269        buffers
270            .into_iter()
271            .map(|buffer| self.completed.push(buffer))
272            .collect()
273    }
274
275    /// The compacting counterpart of [`push_buffers`](Self::push_buffers): adopts each buffer
276    /// according to the [`CompactionStrategy`] its measured utilization earns — whole, sliced to
277    /// its referenced range, or not at all. Views into the buffers must then be rebased through
278    /// [`remap_view_compacted`](Self::remap_view_compacted), which routes views into unadopted
279    /// buffers through the builder's own storage.
280    fn push_buffers_compacted(
281        &mut self,
282        buffers: Vec<ByteBuffer>,
283        utilizations: &[BufferUtilization],
284    ) -> Vec<CompactedSlot> {
285        self.flush_in_progress();
286        buffers
287            .into_iter()
288            .zip(utilizations)
289            .map(|(buffer, utilization)| {
290                match compaction_strategy(utilization, self.compaction_threshold) {
291                    CompactionStrategy::KeepFull => CompactedSlot::Kept {
292                        index: self.completed.push(buffer),
293                    },
294                    CompactionStrategy::Slice { start, end } => CompactedSlot::Sliced {
295                        index: self
296                            .completed
297                            .push(buffer.slice(start as usize..end as usize)),
298                        shift: start,
299                    },
300                    CompactionStrategy::Rewrite => CompactedSlot::Rewrite { source: buffer },
301                }
302            })
303            .collect()
304    }
305
306    /// [`remap_view`] against [`CompactedSlot`]s: views into kept or sliced buffers are rebased
307    /// onto the index those landed at, and a view into a rewritten buffer has its bytes copied
308    /// into the builder's own storage. Inlined views pass through unchanged.
309    fn remap_view_compacted(&mut self, view: BinaryView, slots: &[CompactedSlot]) -> BinaryView {
310        if view.is_inlined() {
311            return view;
312        }
313        let view_ref = view.as_view();
314        match &slots[view_ref.buffer_index as usize] {
315            CompactedSlot::Kept { index } => view_ref
316                .with_buffer_and_offset(*index, view_ref.offset)
317                .into(),
318            CompactedSlot::Sliced { index, shift } => view_ref
319                .with_buffer_and_offset(*index, view_ref.offset - shift)
320                .into(),
321            CompactedSlot::Rewrite { source } => {
322                let bytes = &source.as_slice()[view_ref.as_range()];
323                let (buffer_idx, offset) = self.append_value_to_buffer(bytes);
324                BinaryView::make_view(bytes, buffer_idx, offset)
325            }
326        }
327    }
328
329    /// Appends `validity.len()` values, gathering each valid row's view from `views` through the
330    /// index `view_at` returns for it.
331    ///
332    /// `buffers` — the data buffers the views reference, in the numbering the views use — are
333    /// adopted through the builder's buffer storage, and every gathered view is rebased onto
334    /// the indices they land at as it is written, so the whole append is one view per row with no
335    /// byte copy and no intermediate array. `view_at` is only called for valid rows; null rows get
336    /// an empty view.
337    ///
338    /// When the builder is configured to compact buffers, utilization is measured from the views
339    /// the gather actually references — duplicate indices count once — and each buffer is
340    /// adopted, sliced, or rewritten accordingly, so callers never need a
341    /// canonicalize-and-compact fallback.
342    ///
343    /// # Panics
344    ///
345    /// Panics if `view_at` returns an index out of bounds of `views`, or if a gathered view
346    /// references a buffer index outside `buffers`.
347    pub fn append_views_gathered(
348        &mut self,
349        buffers: impl IntoIterator<Item = ByteBuffer>,
350        views: &[BinaryView],
351        validity: &Mask,
352        view_at: impl Fn(usize) -> usize,
353    ) {
354        if self.compacts_buffers() {
355            return self.append_views_gathered_compacted(
356                buffers.into_iter().collect(),
357                views,
358                validity,
359                view_at,
360            );
361        }
362
363        let mapping = self.push_buffers(buffers);
364
365        self.views_builder.reserve(validity.len());
366        match validity {
367            Mask::AllTrue(len) => self
368                .views_builder
369                .extend_trusted((0..*len).map(|row| remap_view(views[view_at(row)], &mapping))),
370            Mask::AllFalse(len) => self.views_builder.push_n(BinaryView::empty_view(), *len),
371            Mask::Values(values) => {
372                let bits = values.bit_buffer();
373                let start = self.views_builder.len();
374                self.views_builder
375                    .push_n(BinaryView::empty_view(), bits.len());
376                let gathered = &mut self.views_builder[start..];
377                bits.for_each_set_index(|row| {
378                    gathered[row] = remap_view(views[view_at(row)], &mapping);
379                });
380            }
381        }
382
383        self.nulls.append_validity_mask(validity);
384        debug_assert_eq!(self.nulls.len(), self.views_builder.len());
385    }
386
387    /// The compacting arm of [`append_views_gathered`](Self::append_views_gathered): marks the
388    /// source views the gather references, measures each buffer's utilization from just those —
389    /// so entries gathered by many rows count once — and adopts the buffers through
390    /// [`push_buffers_compacted`](Self::push_buffers_compacted). Each referenced source view is
391    /// remapped once, so rows sharing an entry of a rewritten buffer share one copy of its bytes.
392    fn append_views_gathered_compacted(
393        &mut self,
394        buffers: Vec<ByteBuffer>,
395        views: &[BinaryView],
396        validity: &Mask,
397        view_at: impl Fn(usize) -> usize,
398    ) {
399        let mut referenced = vec![false; views.len()];
400        match validity {
401            Mask::AllTrue(len) => (0..*len).for_each(|row| referenced[view_at(row)] = true),
402            Mask::AllFalse(_) => {}
403            Mask::Values(values) => values
404                .bit_buffer()
405                .for_each_set_index(|row| referenced[view_at(row)] = true),
406        }
407
408        let mut utilizations = unmeasured_utilizations(&buffers);
409        for (view, _) in views.iter().zip(&referenced).filter(|(_, used)| **used) {
410            measure_view(&mut utilizations, view);
411        }
412        let slots = self.push_buffers_compacted(buffers, &utilizations);
413
414        let mut remapped = vec![BinaryView::empty_view(); views.len()];
415        for (idx, view) in views.iter().enumerate() {
416            if referenced[idx] {
417                remapped[idx] = self.remap_view_compacted(*view, &slots);
418            }
419        }
420
421        self.views_builder.reserve(validity.len());
422        match validity {
423            Mask::AllTrue(len) => self
424                .views_builder
425                .extend_trusted((0..*len).map(|row| remapped[view_at(row)])),
426            Mask::AllFalse(len) => self.views_builder.push_n(BinaryView::empty_view(), *len),
427            Mask::Values(values) => {
428                let bits = values.bit_buffer();
429                let start = self.views_builder.len();
430                self.views_builder
431                    .push_n(BinaryView::empty_view(), bits.len());
432                let gathered = &mut self.views_builder[start..];
433                bits.for_each_set_index(|row| gathered[row] = remapped[view_at(row)]);
434            }
435        }
436
437        self.nulls.append_validity_mask(validity);
438        debug_assert_eq!(self.nulls.len(), self.views_builder.len());
439    }
440
441    /// Appends `len` rows that are `fill` everywhere except at the given patch rows.
442    ///
443    /// `buffers` — the data buffers `fill` and the patch views reference, in the numbering they
444    /// use — are adopted through the builder's buffer storage and the views are rebased onto
445    /// the indices they land at. `patches` yields `(row, view)` pairs with rows below `len`; the
446    /// fill rows cost one bulk view fill and each patch one view write, with no byte copy.
447    /// Validity is appended exactly as given — invalid rows keep whichever view they got.
448    ///
449    /// This append deliberately adopts the supplied buffers as-is, even when the builder is
450    /// configured for compaction. Callers should use it only when retaining those buffers is
451    /// acceptable.
452    ///
453    /// # Panics
454    ///
455    /// Panics if `validity` is not `len` long, a patch row is out of bounds, or a view references
456    /// a buffer index outside `buffers`.
457    pub fn append_views_scattered(
458        &mut self,
459        buffers: impl IntoIterator<Item = ByteBuffer>,
460        len: usize,
461        fill: BinaryView,
462        patches: impl Iterator<Item = (usize, BinaryView)>,
463        validity: &Mask,
464    ) {
465        assert_eq!(validity.len(), len, "Must have one validity entry per row");
466        let mapping = self.push_buffers(buffers);
467
468        let start = self.views_builder.len();
469        self.views_builder.push_n(remap_view(fill, &mapping), len);
470        let scattered = &mut self.views_builder[start..];
471        for (row, view) in patches {
472            scattered[row] = remap_view(view, &mapping);
473        }
474
475        self.nulls.append_validity_mask(validity);
476        debug_assert_eq!(self.nulls.len(), self.views_builder.len());
477    }
478
479    /// Appends values laid end-to-end in `bytes`, one per entry of `lengths`.
480    ///
481    /// The builder adopts `bytes` as a data buffer without copying it (splitting it only past the
482    /// `u32` view-offset limit) and builds the views directly into its own storage, so the whole
483    /// append costs one view per row. Every length is consumed, including those for null rows, so
484    /// the lengths must describe `bytes` exactly; bytes belonging to null rows are not retained
485    /// when compaction rewrites the buffer.
486    ///
487    /// When the builder is configured to compact buffers, the utilization is measured from the
488    /// lengths alone — only values too long to inline reference the buffer — and a heap below
489    /// the threshold is rewritten to just those values instead of adopted, so callers never need
490    /// a canonicalize-and-compact fallback.
491    ///
492    /// # Panics
493    ///
494    /// Panics if `lengths` and `validity` disagree in length, if the lengths do not describe
495    /// `bytes` exactly, or if this builder deduplicates buffers and already holds `bytes`.
496    pub fn append_buffer_with_lengths<P: NativePType + AsPrimitive<usize>>(
497        &mut self,
498        bytes: ByteBuffer,
499        lengths: &[P],
500        validity: &Mask,
501    ) {
502        assert_eq!(
503            lengths.len(),
504            validity.len(),
505            "Must have one length per validity entry"
506        );
507        self.append_buffer_views(&bytes, lengths.len(), validity, |i| lengths[i].as_());
508    }
509
510    /// [`append_buffer_with_lengths`](Self::append_buffer_with_lengths) for values described by
511    /// an offsets buffer instead of lengths.
512    ///
513    /// `offsets` are absolute positions into `bytes` — the layout a
514    /// [`VarBinArray`](crate::arrays::VarBinArray) stores — so there is one more offset than there
515    /// are values, and only the `offsets[0]..offsets[last]` range of `bytes` is adopted, again
516    /// without copying.
517    ///
518    /// # Panics
519    ///
520    /// Panics if `offsets` does not hold exactly one more entry than `validity`, or if the offsets
521    /// are not monotonically non-decreasing positions within `bytes`.
522    pub fn append_buffer_with_offsets<P: NativePType + AsPrimitive<usize>>(
523        &mut self,
524        bytes: ByteBuffer,
525        offsets: &[P],
526        validity: &Mask,
527    ) {
528        assert_eq!(
529            offsets.len(),
530            validity.len() + 1,
531            "Must have one more offset than validity entries"
532        );
533        let first: usize = offsets[0].as_();
534        let last: usize = offsets[offsets.len() - 1].as_();
535        let bytes = bytes.slice(first..last);
536        // Wrapping keeps corrupt non-monotonic offsets from panicking on the subtraction itself;
537        // the wrapped length then fails the in-bounds checks of the view-building loop.
538        self.append_buffer_views(&bytes, validity.len(), validity, |i| {
539            AsPrimitive::<usize>::as_(offsets[i + 1])
540                .wrapping_sub(AsPrimitive::<usize>::as_(offsets[i]))
541        });
542    }
543
544    /// Shared tail of the bulk buffer appends: builds the views straight into the builder's views
545    /// storage, then adopts the buffer segments and the validity.
546    fn append_buffer_views(
547        &mut self,
548        bytes: &ByteBuffer,
549        count: usize,
550        validity: &Mask,
551        len_at: impl Fn(usize) -> usize,
552    ) {
553        self.flush_in_progress();
554
555        // A compacting builder measures utilization before adopting the buffer. Only values too
556        // long to inline reference the heap, so the measurement is one pass over the lengths and
557        // never touches the bytes. Heaps past the single-buffer limit are adopted as they are:
558        // they roll over into multiple buffers, and per-segment accounting is not worth the rare
559        // >2GiB case.
560        if self.compacts_buffers() && bytes.len() <= MAX_BUFFER_LEN {
561            let referenced: usize = match validity.bit_buffer() {
562                AllOr::All => (0..count)
563                    .map(&len_at)
564                    .filter(|len| *len > BinaryView::MAX_INLINED_SIZE)
565                    .sum(),
566                AllOr::None => 0,
567                AllOr::Some(b) => {
568                    let mut sum = 0;
569                    b.for_each_set_index(|idx| {
570                        let len = len_at(idx);
571                        if len > BinaryView::MAX_INLINED_SIZE {
572                            sum += len;
573                        }
574                    });
575                    sum
576                }
577            };
578            #[expect(clippy::cast_precision_loss)]
579            if (referenced as f64) < self.compaction_threshold * (bytes.len() as f64) {
580                return self
581                    .append_buffer_views_rewritten(bytes, count, validity, len_at, referenced);
582            }
583        }
584
585        let start_index = self.completed.len();
586        let segments = extend_views(
587            &mut self.views_builder,
588            start_index,
589            MAX_BUFFER_LEN,
590            bytes,
591            count,
592            len_at,
593        );
594
595        let expected_completed_len = start_index as usize + segments.len();
596        self.completed.extend_from_slice_unchecked(&segments);
597        assert_eq!(
598            self.completed.len() as usize,
599            expected_completed_len,
600            "Some buffers already exist",
601        );
602
603        self.nulls.append_validity_mask(validity);
604        debug_assert_eq!(self.nulls.len(), self.views_builder.len());
605    }
606
607    /// The under-utilized arm of [`append_buffer_views`](Self::append_buffer_views): copies only
608    /// the values that actually reference the heap into a compact buffer, sized `referenced`,
609    /// instead of adopting the whole heap. A fully-inlined append pushes no buffer at all.
610    fn append_buffer_views_rewritten(
611        &mut self,
612        bytes: &ByteBuffer,
613        count: usize,
614        validity: &Mask,
615        len_at: impl Fn(usize) -> usize,
616        referenced: usize,
617    ) {
618        let buf_index = self.completed.len();
619        let mut compact = ByteBufferMut::with_capacity(referenced);
620        self.views_builder.reserve(count);
621
622        let data = bytes.as_slice();
623        let mut offset = 0usize;
624        for (i, is_valid) in validity.iter().enumerate() {
625            let len = len_at(i);
626            let value = &data[offset..offset + len];
627            let view = if !is_valid {
628                BinaryView::empty_view()
629            } else if len > BinaryView::MAX_INLINED_SIZE {
630                // In `u32` range: `referenced <= bytes.len() <= MAX_BUFFER_LEN` (checked by the
631                // caller), and `compact` never grows past `referenced`.
632                #[expect(clippy::cast_possible_truncation)]
633                let view = BinaryView::make_view(value, buf_index, compact.len() as u32);
634                compact.extend_from_slice(value);
635                view
636            } else {
637                BinaryView::make_view(value, buf_index, 0)
638            };
639            self.views_builder.push(view);
640            offset += len;
641        }
642        assert_eq!(
643            offset,
644            data.len(),
645            "value lengths must describe the byte heap exactly"
646        );
647
648        if !compact.is_empty() {
649            let pushed_index = self.completed.push(compact.freeze());
650            assert_eq!(pushed_index, buf_index, "Buffer already exists");
651        }
652
653        self.nulls.append_validity_mask(validity);
654        debug_assert_eq!(self.nulls.len(), self.views_builder.len());
655    }
656
657    /// Finishes the builder directly into a [`VarBinViewArray`].
658    pub fn finish_into_varbinview(&mut self) -> VarBinViewArray {
659        self.flush_in_progress();
660        let buffers = std::mem::take(&mut self.completed);
661
662        assert_eq!(
663            self.views_builder.len(),
664            self.nulls.len(),
665            "View and validity length must match"
666        );
667
668        let validity = self.nulls.finish_with_nullability(self.dtype.nullability());
669
670        // SAFETY: the builder methods check safety at each step.
671        unsafe {
672            VarBinViewArray::new_unchecked(
673                std::mem::take(&mut self.views_builder).freeze(),
674                buffers.finish(),
675                self.dtype.clone(),
676                validity,
677            )
678        }
679    }
680
681    pub(crate) fn append_varbinview_array(
682        &mut self,
683        array: &VarBinViewArray,
684        ctx: &mut ExecutionCtx,
685    ) -> VortexResult<()> {
686        self.flush_in_progress();
687
688        let mask = array.varbinview_validity().execute_mask(array.len(), ctx)?;
689
690        self.nulls.append_validity_mask(&mask);
691
692        let view_adjustment =
693            self.completed
694                .extend_from_compaction(BuffersWithOffsets::from_array(
695                    array,
696                    self.compaction_threshold,
697                    ctx,
698                ));
699
700        match view_adjustment {
701            ViewAdjustment::Precomputed(adjustment) => self.views_builder.extend_trusted(
702                array
703                    .views()
704                    .iter()
705                    .map(|view| adjustment.adjust_view(view)),
706            ),
707            ViewAdjustment::Rewriting(adjustment) => match mask {
708                Mask::AllTrue(_) => {
709                    for (idx, &view) in array.views().iter().enumerate() {
710                        let new_view = self.push_view(view, &adjustment, array, idx);
711                        self.views_builder.push(new_view);
712                    }
713                }
714                Mask::AllFalse(_) => {
715                    self.views_builder
716                        .push_n(BinaryView::empty_view(), array.len());
717                }
718                Mask::Values(v) => {
719                    let views = array.views();
720                    let start = self.views_builder.len();
721                    self.views_builder
722                        .push_n(BinaryView::empty_view(), array.len());
723                    v.bit_buffer().for_each_set_index(|idx| {
724                        let new_view = self.push_view(views[idx], &adjustment, array, idx);
725                        self.views_builder[start + idx] = new_view;
726                    });
727                }
728            },
729        }
730
731        Ok(())
732    }
733}
734
735impl ArrayBuilder for VarBinViewBuilder {
736    fn as_any(&self) -> &dyn Any {
737        self
738    }
739
740    fn as_any_mut(&mut self) -> &mut dyn Any {
741        self
742    }
743
744    fn dtype(&self) -> &DType {
745        &self.dtype
746    }
747
748    fn len(&self) -> usize {
749        self.nulls.len()
750    }
751
752    fn append_zeros(&mut self, n: usize) {
753        self.views_builder.push_n(BinaryView::empty_view(), n);
754        self.nulls.append_n_non_nulls(n);
755    }
756
757    unsafe fn append_nulls_unchecked(&mut self, n: usize) {
758        self.views_builder.push_n(BinaryView::empty_view(), n);
759        self.nulls.append_n_nulls(n);
760    }
761
762    fn append_scalar(&mut self, scalar: &Scalar) -> VortexResult<()> {
763        vortex_ensure!(
764            scalar.dtype() == self.dtype(),
765            "VarBinViewBuilder expected scalar with dtype {}, got {}",
766            self.dtype(),
767            scalar.dtype()
768        );
769
770        match self.dtype() {
771            DType::Utf8(_) => match scalar.as_utf8().value() {
772                Some(value) => self.append_value(value),
773                None => self.append_null(),
774            },
775            DType::Binary(_) => match scalar.as_binary().value() {
776                Some(value) => self.append_value(value),
777                None => self.append_null(),
778            },
779            _ => vortex_bail!(
780                "VarBinViewBuilder can only handle Utf8 or Binary scalars, got {:?}",
781                scalar.dtype()
782            ),
783        }
784
785        Ok(())
786    }
787
788    fn reserve_exact(&mut self, additional: usize) {
789        self.views_builder.reserve(additional);
790        self.nulls.reserve_exact(additional);
791    }
792
793    fn finish(&mut self) -> ArrayRef {
794        self.finish_into_varbinview().into_array()
795    }
796
797    fn finish_into_canonical(&mut self, _ctx: &mut ExecutionCtx) -> Canonical {
798        Canonical::VarBinView(self.finish_into_varbinview())
799    }
800}
801
802impl VarBinViewBuilder {
803    #[inline]
804    fn push_view(
805        &mut self,
806        view: BinaryView,
807        adjustment: &RewritingViewAdjustment,
808        array: &VarBinViewArray,
809        idx: usize,
810    ) -> BinaryView {
811        if view.is_inlined() {
812            view
813        } else if let Some(adjusted) = adjustment.adjust_view(&view) {
814            adjusted
815        } else {
816            let bytes = array.bytes_at(idx);
817            let (new_buf_idx, new_offset) = self.append_value_to_buffer(&bytes);
818            BinaryView::make_view(bytes.as_slice(), new_buf_idx, new_offset)
819        }
820    }
821}
822
823/// Rebases a view built against a local buffer numbering onto the builder indices those buffers
824/// landed at, i.e. `mapping[i]` is where the caller's buffer `i` went. Inlined views carry no
825/// buffer reference and pass through unchanged.
826#[inline]
827fn remap_view(view: BinaryView, mapping: &[u32]) -> BinaryView {
828    if view.is_inlined() {
829        view
830    } else {
831        let view_ref = view.as_view();
832        view_ref
833            .with_buffer_and_offset(mapping[view_ref.buffer_index as usize], view_ref.offset)
834            .into()
835    }
836}
837
838/// Where a caller's buffer went under `VarBinViewBuilder::push_buffers_compacted`.
839enum CompactedSlot {
840    /// Adopted whole at this index.
841    Kept { index: u32 },
842    /// Adopted at this index as the slice starting `shift` bytes in, so view offsets shift down.
843    Sliced { index: u32, shift: u32 },
844    /// Not adopted: each referenced view's bytes are copied out of the source buffer into the
845    /// builder's own storage.
846    Rewrite { source: ByteBuffer },
847}
848
849/// One unmeasured [`BufferUtilization`] per buffer, ready for [`measure_view`] passes.
850fn unmeasured_utilizations(buffers: &[ByteBuffer]) -> Vec<BufferUtilization> {
851    buffers
852        .iter()
853        .map(|buffer| {
854            // Views address at most `u32` offsets, so measuring an oversized buffer against the
855            // saturated length under-reports utilization, which can only compact harder — and its
856            // unaddressable tail is dead weight worth compacting anyway.
857            BufferUtilization::zero(u32::try_from(buffer.len()).unwrap_or(u32::MAX))
858        })
859        .collect()
860}
861
862/// Counts `view`'s bytes against the buffer it references; inlined views reference none.
863fn measure_view(utilizations: &mut [BufferUtilization], view: &BinaryView) {
864    if !view.is_inlined() {
865        let view_ref = view.as_view();
866        utilizations[view_ref.buffer_index as usize].add(view_ref.offset, view_ref.size);
867    }
868}
869
870pub enum CompletedBuffers {
871    Default(Vec<ByteBuffer>),
872    Deduplicated(DeduplicatedBuffers),
873}
874
875impl Default for CompletedBuffers {
876    fn default() -> Self {
877        Self::Default(Vec::new())
878    }
879}
880
881// Self::push enforces len < u32::max
882#[expect(clippy::cast_possible_truncation)]
883impl CompletedBuffers {
884    fn len(&self) -> u32 {
885        match self {
886            Self::Default(buffers) => buffers.len() as u32,
887            Self::Deduplicated(buffers) => buffers.len(),
888        }
889    }
890
891    /// Push a new block, returning the index it landed at (or, when deduplicating, the index of
892    /// the identical block already held).
893    fn push(&mut self, block: ByteBuffer) -> u32 {
894        match self {
895            Self::Default(buffers) => {
896                assert!(buffers.len() < u32::MAX as usize, "Too many blocks");
897                buffers.push(block);
898                self.len() - 1
899            }
900            Self::Deduplicated(buffers) => buffers.push(block),
901        }
902    }
903
904    /// Does not compact buffers, bypasses utilization checks.
905    fn extend_from_slice_unchecked(&mut self, buffers: &[ByteBuffer]) {
906        for buffer in buffers {
907            self.push(buffer.clone());
908        }
909    }
910
911    fn extend_from_compaction(&mut self, buffers: BuffersWithOffsets) -> ViewAdjustment {
912        match (self, buffers) {
913            (
914                Self::Default(completed_buffers),
915                BuffersWithOffsets::AllKept { buffers, offsets },
916            ) => {
917                let buffer_offset = completed_buffers.len() as u32;
918                completed_buffers.extend_from_slice(&buffers);
919                ViewAdjustment::shift(buffer_offset, offsets)
920            }
921            (
922                Self::Default(completed_buffers),
923                BuffersWithOffsets::SomeCompacted { buffers, offsets },
924            ) => {
925                let lookup = buffers
926                    .iter()
927                    .map(|maybe_buffer| {
928                        maybe_buffer.as_ref().map(|buffer| {
929                            completed_buffers.push(buffer.clone());
930                            completed_buffers.len() as u32 - 1
931                        })
932                    })
933                    .collect();
934                ViewAdjustment::rewriting(lookup, offsets)
935            }
936
937            (
938                Self::Deduplicated(completed_buffers),
939                BuffersWithOffsets::AllKept { buffers, offsets },
940            ) => {
941                let buffer_lookup = completed_buffers.extend_from_iter(buffers.iter().cloned());
942                ViewAdjustment::lookup(buffer_lookup, offsets)
943            }
944            (
945                Self::Deduplicated(completed_buffers),
946                BuffersWithOffsets::SomeCompacted { buffers, offsets },
947            ) => {
948                let buffer_lookup = completed_buffers.extend_from_option_slice(&buffers);
949                ViewAdjustment::rewriting(buffer_lookup, offsets)
950            }
951        }
952    }
953
954    fn finish(self) -> Arc<[ByteBuffer]> {
955        match self {
956            Self::Default(buffers) => Arc::from(buffers),
957            Self::Deduplicated(buffers) => buffers.finish(),
958        }
959    }
960}
961
962#[derive(Default)]
963pub struct DeduplicatedBuffers {
964    buffers: Vec<ByteBuffer>,
965    buffer_to_idx: HashMap<BufferId, u32>,
966}
967
968impl DeduplicatedBuffers {
969    // Self::push enforces len < u32::max
970    #[expect(clippy::cast_possible_truncation)]
971    fn len(&self) -> u32 {
972        self.buffers.len() as u32
973    }
974
975    /// Push a new block if not seen before. Returns the idx of the block.
976    pub(crate) fn push(&mut self, block: ByteBuffer) -> u32 {
977        assert!(self.buffers.len() < u32::MAX as usize, "Too many blocks");
978
979        let initial_len = self.len();
980        let id = BufferId::from(&block);
981        match self.buffer_to_idx.entry(id) {
982            Entry::Occupied(idx) => *idx.get(),
983            Entry::Vacant(entry) => {
984                let idx = initial_len;
985                entry.insert(idx);
986                self.buffers.push(block);
987                idx
988            }
989        }
990    }
991
992    pub(crate) fn extend_from_option_slice(
993        &mut self,
994        buffers: &[Option<ByteBuffer>],
995    ) -> Vec<Option<u32>> {
996        buffers
997            .iter()
998            .map(|buffer| buffer.as_ref().map(|buf| self.push(buf.clone())))
999            .collect()
1000    }
1001
1002    pub(crate) fn extend_from_iter(
1003        &mut self,
1004        buffers: impl Iterator<Item = ByteBuffer>,
1005    ) -> Vec<u32> {
1006        buffers.map(|buffer| self.push(buffer)).collect()
1007    }
1008
1009    pub(crate) fn finish(self) -> Arc<[ByteBuffer]> {
1010        Arc::from(self.buffers)
1011    }
1012}
1013
1014#[derive(PartialEq, Eq, Hash)]
1015struct BufferId {
1016    // *const u8 stored as usize for `Send`
1017    ptr: usize,
1018    len: usize,
1019}
1020
1021impl BufferId {
1022    fn from(buffer: &ByteBuffer) -> Self {
1023        let slice = buffer.as_slice();
1024        Self {
1025            ptr: slice.as_ptr() as usize,
1026            len: slice.len(),
1027        }
1028    }
1029}
1030
1031#[derive(Debug, Clone)]
1032pub enum BufferGrowthStrategy {
1033    /// Use a fixed buffer size for all allocations.
1034    Fixed { size: u32 },
1035    /// Use exponential growth starting from initial_size, doubling until max_size.
1036    Exponential { current_size: u32, max_size: u32 },
1037}
1038
1039impl Default for BufferGrowthStrategy {
1040    fn default() -> Self {
1041        Self::Exponential {
1042            current_size: 4 * 1024,    // 4KB starting size
1043            max_size: 2 * 1024 * 1024, // 2MB max size
1044        }
1045    }
1046}
1047
1048impl BufferGrowthStrategy {
1049    pub fn fixed(size: u32) -> Self {
1050        Self::Fixed { size }
1051    }
1052
1053    pub fn exponential(initial_size: u32, max_size: u32) -> Self {
1054        Self::Exponential {
1055            current_size: initial_size,
1056            max_size,
1057        }
1058    }
1059
1060    /// Returns the next buffer size to allocate and updates internal state.
1061    pub fn next_size(&mut self) -> u32 {
1062        match self {
1063            Self::Fixed { size } => *size,
1064            Self::Exponential {
1065                current_size,
1066                max_size,
1067            } => {
1068                let result = *current_size;
1069                if *current_size < *max_size {
1070                    *current_size = current_size.saturating_mul(2).min(*max_size);
1071                }
1072                result
1073            }
1074        }
1075    }
1076}
1077
1078enum BuffersWithOffsets {
1079    AllKept {
1080        buffers: Arc<[ByteBuffer]>,
1081        offsets: Option<Vec<u32>>,
1082    },
1083    SomeCompacted {
1084        buffers: Vec<Option<ByteBuffer>>,
1085        offsets: Option<Vec<u32>>,
1086    },
1087}
1088
1089impl BuffersWithOffsets {
1090    pub fn from_array(
1091        array: &VarBinViewArray,
1092        compaction_threshold: f64,
1093        ctx: &mut ExecutionCtx,
1094    ) -> Self {
1095        if compaction_threshold == 0.0 {
1096            return Self::AllKept {
1097                buffers: Arc::from(
1098                    array
1099                        .data_buffers()
1100                        .iter()
1101                        .cloned()
1102                        .map(|b| b.unwrap_host())
1103                        .collect_vec(),
1104                ),
1105                offsets: None,
1106            };
1107        }
1108
1109        let buffer_utilizations = array
1110            .buffer_utilizations(ctx)
1111            .vortex_expect("buffer_utilizations in BuffersWithOffsets::from_array");
1112        let mut has_rewrite = false;
1113        let mut has_nonzero_offset = false;
1114        for utilization in buffer_utilizations.iter() {
1115            match compaction_strategy(utilization, compaction_threshold) {
1116                CompactionStrategy::KeepFull => continue,
1117                CompactionStrategy::Slice { .. } => has_nonzero_offset = true,
1118                CompactionStrategy::Rewrite => has_rewrite = true,
1119            }
1120        }
1121
1122        let buffers_with_offsets_iter = buffer_utilizations
1123            .iter()
1124            .zip(array.data_buffers().iter())
1125            .map(|(utilization, buffer)| {
1126                match compaction_strategy(utilization, compaction_threshold) {
1127                    CompactionStrategy::KeepFull => (Some(buffer.as_host().clone()), 0),
1128                    CompactionStrategy::Slice { start, end } => (
1129                        Some(buffer.as_host().slice(start as usize..end as usize)),
1130                        start,
1131                    ),
1132                    CompactionStrategy::Rewrite => (None, 0),
1133                }
1134            });
1135
1136        match (has_rewrite, has_nonzero_offset) {
1137            // keep all buffers
1138            (false, false) => {
1139                let buffers: Vec<_> = buffers_with_offsets_iter
1140                    .map(|(b, _)| b.vortex_expect("already checked for rewrite"))
1141                    .collect();
1142                Self::AllKept {
1143                    buffers: Arc::from(buffers),
1144                    offsets: None,
1145                }
1146            }
1147            // rewrite, all zero offsets
1148            (true, false) => {
1149                let buffers: Vec<_> = buffers_with_offsets_iter.map(|(b, _)| b).collect();
1150                Self::SomeCompacted {
1151                    buffers,
1152                    offsets: None,
1153                }
1154            }
1155            // keep all buffers, but some have offsets
1156            (false, true) => {
1157                let (buffers, offsets): (Vec<_>, _) = buffers_with_offsets_iter
1158                    .map(|(buffer, offset)| {
1159                        (buffer.vortex_expect("already checked for rewrite"), offset)
1160                    })
1161                    .collect();
1162                Self::AllKept {
1163                    buffers: Arc::from(buffers),
1164                    offsets: Some(offsets),
1165                }
1166            }
1167            // rewrite and some have offsets
1168            (true, true) => {
1169                let (buffers, offsets) = buffers_with_offsets_iter.collect();
1170                Self::SomeCompacted {
1171                    buffers,
1172                    offsets: Some(offsets),
1173                }
1174            }
1175        }
1176    }
1177}
1178
1179#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1180enum CompactionStrategy {
1181    KeepFull,
1182    /// Slice the buffer to [start, end) range
1183    Slice {
1184        start: u32,
1185        end: u32,
1186    },
1187    /// Rewrite data into new compacted buffer
1188    Rewrite,
1189}
1190
1191fn compaction_strategy(
1192    buffer_utilization: &BufferUtilization,
1193    threshold: f64,
1194) -> CompactionStrategy {
1195    match buffer_utilization.overall_utilization() {
1196        // rewrite empty or not used buffers TODO(os): maybe keep them
1197        0.0 => CompactionStrategy::Rewrite,
1198        utilised if utilised >= threshold => CompactionStrategy::KeepFull,
1199        _ if buffer_utilization.range_utilization() >= threshold => {
1200            let Range { start, end } = buffer_utilization.range();
1201            CompactionStrategy::Slice { start, end }
1202        }
1203        _ => CompactionStrategy::Rewrite,
1204    }
1205}
1206
1207enum ViewAdjustment {
1208    Precomputed(PrecomputedViewAdjustment),
1209    Rewriting(RewritingViewAdjustment),
1210}
1211
1212impl ViewAdjustment {
1213    fn shift(buffer_offset: u32, offsets: Option<Vec<u32>>) -> Self {
1214        Self::Precomputed(PrecomputedViewAdjustment::Shift {
1215            buffer_offset,
1216            offsets,
1217        })
1218    }
1219
1220    fn lookup(buffer_lookup: Vec<u32>, offsets: Option<Vec<u32>>) -> Self {
1221        Self::Precomputed(PrecomputedViewAdjustment::Lookup {
1222            buffer_lookup,
1223            offsets,
1224        })
1225    }
1226
1227    fn rewriting(buffer_lookup: Vec<Option<u32>>, offsets: Option<Vec<u32>>) -> Self {
1228        Self::Rewriting(RewritingViewAdjustment {
1229            buffer_lookup,
1230            offsets,
1231        })
1232    }
1233}
1234
1235// Care when adding new variants or fields in this enum, it will mess with inlining if it gets too big
1236enum PrecomputedViewAdjustment {
1237    Shift {
1238        buffer_offset: u32,
1239        offsets: Option<Vec<u32>>,
1240    },
1241    Lookup {
1242        buffer_lookup: Vec<u32>,
1243        offsets: Option<Vec<u32>>,
1244    },
1245}
1246
1247impl PrecomputedViewAdjustment {
1248    #[inline]
1249    fn adjust_view(&self, view: &BinaryView) -> BinaryView {
1250        if view.is_inlined() {
1251            return *view;
1252        }
1253        let view_ref = view.as_view();
1254        match self {
1255            Self::Shift {
1256                buffer_offset,
1257                offsets,
1258            } => {
1259                let b_idx = view_ref.buffer_index;
1260                let offset_shift = offsets
1261                    .as_ref()
1262                    .map(|o| o[b_idx as usize])
1263                    .unwrap_or_default();
1264
1265                // If offset < offset_shift, this view was invalid and wasn't counted in buffer_utilizations.
1266                // Return an empty view to match how invalid views are handled in the Rewriting path.
1267                if view_ref.offset < offset_shift {
1268                    return BinaryView::empty_view();
1269                }
1270
1271                view_ref
1272                    .with_buffer_and_offset(b_idx + buffer_offset, view_ref.offset - offset_shift)
1273            }
1274            Self::Lookup {
1275                buffer_lookup,
1276                offsets,
1277            } => {
1278                let b_idx = view_ref.buffer_index;
1279                let buffer = buffer_lookup[b_idx as usize];
1280                let offset_shift = offsets
1281                    .as_ref()
1282                    .map(|o| o[b_idx as usize])
1283                    .unwrap_or_default();
1284
1285                // If offset < offset_shift, this view was invalid and wasn't counted in buffer_utilizations.
1286                // Return an empty view to match how invalid views are handled in the Rewriting path.
1287                if view_ref.offset < offset_shift {
1288                    return BinaryView::empty_view();
1289                }
1290
1291                view_ref.with_buffer_and_offset(buffer, view_ref.offset - offset_shift)
1292            }
1293        }
1294        .into()
1295    }
1296}
1297
1298struct RewritingViewAdjustment {
1299    buffer_lookup: Vec<Option<u32>>,
1300    offsets: Option<Vec<u32>>,
1301}
1302
1303impl RewritingViewAdjustment {
1304    /// Can return None if this view can't be adjusted, because there is no precomputed lookup
1305    /// for the current buffer.
1306    #[inline]
1307    fn adjust_view(&self, view: &BinaryView) -> Option<BinaryView> {
1308        if view.is_inlined() {
1309            return Some(*view);
1310        }
1311
1312        let view_ref = view.as_view();
1313        self.buffer_lookup[view_ref.buffer_index as usize].map(|buffer| {
1314            let offset_shift = self
1315                .offsets
1316                .as_ref()
1317                .map(|o| o[view_ref.buffer_index as usize])
1318                .unwrap_or_default();
1319            view_ref
1320                .with_buffer_and_offset(buffer, view_ref.offset - offset_shift)
1321                .into()
1322        })
1323    }
1324}
1325
1326#[cfg(test)]
1327mod tests {
1328    use vortex_buffer::ByteBuffer;
1329    use vortex_error::VortexResult;
1330    use vortex_mask::Mask;
1331
1332    use crate::IntoArray;
1333    use crate::VortexSessionExecute;
1334    use crate::array_session;
1335    use crate::assert_arrays_eq;
1336    use crate::builders::ArrayBuilder;
1337    use crate::builders::VarBinViewBuilder;
1338    use crate::builders::varbinview::VarBinViewArray;
1339    use crate::dtype::DType;
1340    use crate::dtype::Nullability;
1341
1342    /// A long-enough value that a view over it must reference a data buffer.
1343    const LONG: &str = "a value that is far too long to inline";
1344
1345    /// The heap is adopted zero-copy as a data buffer, the views are built against it in place,
1346    /// and the append composes with staged in-progress bytes on either side.
1347    #[test]
1348    fn test_append_buffer_with_lengths() {
1349        let mut ctx = array_session().create_execution_ctx();
1350        let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8);
1351
1352        // Stages an in-progress buffer the bulk append has to flush first.
1353        builder.append_value(LONG);
1354
1355        let heap = ByteBuffer::copy_from([LONG.as_bytes(), b"", b"tiny"].concat());
1356        let heap_ptr = heap.as_ptr();
1357        let lengths = [u32::try_from(LONG.len()).unwrap(), 0, 4];
1358        builder.append_buffer_with_lengths(heap, &lengths, &Mask::from_iter([true, false, true]));
1359
1360        builder.append_value("tail");
1361
1362        let actual = builder.finish_into_varbinview();
1363        // The adopted heap sits after the flushed in-progress buffer, untouched.
1364        assert_eq!(actual.data_buffers()[1].as_host().as_ptr(), heap_ptr);
1365
1366        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1367            Some(LONG),
1368            Some(LONG),
1369            None,
1370            Some("tiny"),
1371            Some("tail"),
1372        ]);
1373        assert_arrays_eq!(actual, expected, &mut ctx);
1374    }
1375
1376    /// Offsets need not start at zero: only the referenced range of the heap is adopted.
1377    #[test]
1378    fn test_append_buffer_with_offsets() {
1379        let mut ctx = array_session().create_execution_ctx();
1380        let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8);
1381
1382        let heap = ByteBuffer::copy_from(format!("..{LONG}tiny!!"));
1383        let long_len = u32::try_from(LONG.len()).unwrap();
1384        let offsets = [2u32, 2 + long_len, 2 + long_len, 2 + long_len + 4];
1385        builder.append_buffer_with_offsets(
1386            heap.clone(),
1387            &offsets,
1388            &Mask::from_iter([true, false, true]),
1389        );
1390
1391        let actual = builder.finish_into_varbinview();
1392        // Zero-copy adoption of just the `offsets[0]..offsets[last]` range.
1393        // SAFETY: offset 2 is in bounds of the heap.
1394        assert_eq!(actual.data_buffers()[0].as_host().as_ptr(), unsafe {
1395            heap.as_ptr().add(2)
1396        });
1397        assert_eq!(
1398            actual.data_buffers()[0].len(),
1399            LONG.len() + 4,
1400            "only the referenced range must be adopted"
1401        );
1402
1403        let expected =
1404            <VarBinViewArray as FromIterator<_>>::from_iter([Some(LONG), None, Some("tiny")]);
1405        assert_arrays_eq!(actual, expected, &mut ctx);
1406    }
1407
1408    /// A compacting builder must measure the heap it is handed: values short enough to inline
1409    /// never reference it, so an under-utilized heap is rewritten to just the referencing values.
1410    #[test]
1411    fn test_append_buffer_with_lengths_compacts_underutilized_heap() {
1412        let mut ctx = array_session().create_execution_ctx();
1413        let mut builder =
1414            VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 4, 1.0);
1415
1416        let heap = ByteBuffer::copy_from([b"short".as_slice(), LONG.as_bytes(), b"tiny"].concat());
1417        let lengths = [5u32, u32::try_from(LONG.len()).unwrap(), 4];
1418        builder.append_buffer_with_lengths(heap, &lengths, &Mask::new_true(3));
1419
1420        let actual = builder.finish_into_varbinview();
1421        assert_eq!(actual.data_buffers().len(), 1);
1422        assert_eq!(
1423            actual.data_buffers()[0].len(),
1424            LONG.len(),
1425            "the compact buffer must hold only the non-inlined values"
1426        );
1427
1428        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1429            Some("short"),
1430            Some(LONG),
1431            Some("tiny"),
1432        ]);
1433        assert_arrays_eq!(actual, expected, &mut ctx);
1434    }
1435
1436    /// Rewriting an under-utilized heap must consume null values' spans without retaining their
1437    /// bytes or producing views that reference them.
1438    #[test]
1439    fn test_append_buffer_with_lengths_compaction_skips_null_bytes() {
1440        let mut ctx = array_session().create_execution_ctx();
1441        let mut builder =
1442            VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 2, 1.0);
1443
1444        let heap = ByteBuffer::copy_from([LONG.as_bytes(), LONG.as_bytes()].concat());
1445        let lengths = [u32::try_from(LONG.len()).unwrap(); 2];
1446        builder.append_buffer_with_lengths(heap, &lengths, &Mask::from_iter([false, true]));
1447
1448        let actual = builder.finish_into_varbinview();
1449        assert_eq!(actual.data_buffers().len(), 1);
1450        assert_eq!(
1451            actual.data_buffers()[0].len(),
1452            LONG.len(),
1453            "the compact buffer must omit bytes belonging to null rows"
1454        );
1455
1456        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([None::<&str>, Some(LONG)]);
1457        assert_arrays_eq!(actual, expected, &mut ctx);
1458    }
1459
1460    /// `push_buffers` returns where each buffer landed, and a deduplicating builder maps a
1461    /// re-pushed buffer back to its existing index instead of holding it twice.
1462    #[test]
1463    fn test_push_buffers_deduplicates() {
1464        let mut builder =
1465            VarBinViewBuilder::with_buffer_deduplication(DType::Utf8(Nullability::Nullable), 8);
1466
1467        let first = ByteBuffer::copy_from(LONG);
1468        let second = ByteBuffer::copy_from("another value far too long to inline");
1469
1470        assert_eq!(
1471            builder.push_buffers([first.clone(), second.clone()]),
1472            [0, 1]
1473        );
1474        assert_eq!(builder.push_buffers([second, first]), [1, 0]);
1475        assert_eq!(builder.completed_block_count(), 2);
1476    }
1477
1478    /// Gathered views are rebased onto wherever the pushed buffers landed, and null rows never
1479    /// resolve their index.
1480    #[test]
1481    fn test_append_views_gathered() {
1482        let mut ctx = array_session().create_execution_ctx();
1483        let dictionary = <VarBinViewArray as FromIterator<_>>::from_iter([
1484            Some("tiny"),
1485            Some(LONG),
1486            Some("small"),
1487        ]);
1488        let buffers = dictionary
1489            .data_buffers()
1490            .iter()
1491            .map(|buffer| buffer.as_host().clone())
1492            .collect::<Vec<_>>();
1493
1494        let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8);
1495        // Stages an in-progress buffer that the gather has to flush ahead of its own buffers.
1496        builder.append_value(LONG);
1497
1498        let codes: [usize; 4] = [1, 0, usize::MAX, 2];
1499        let views = dictionary.views();
1500        builder.append_views_gathered(
1501            buffers,
1502            views,
1503            &Mask::from_iter([true, true, false, true]),
1504            // The null row's code is garbage; the builder must not look it up.
1505            |row| codes[row],
1506        );
1507
1508        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1509            Some(LONG),
1510            Some(LONG),
1511            Some("tiny"),
1512            None,
1513            Some("small"),
1514        ]);
1515        assert_arrays_eq!(builder.finish_into_varbinview(), expected, &mut ctx);
1516    }
1517
1518    /// Scattered patches overwrite the fill view at their rows, and both are rebased onto the
1519    /// adopted buffers.
1520    #[test]
1521    fn test_append_views_scattered() {
1522        use crate::arrays::varbinview::build_views::BinaryView;
1523
1524        let mut ctx = array_session().create_execution_ctx();
1525        let patch_values =
1526            <VarBinViewArray as FromIterator<_>>::from_iter([Some(LONG), Some("tiny")]);
1527        let mut buffers = patch_values
1528            .data_buffers()
1529            .iter()
1530            .map(|buffer| buffer.as_host().clone())
1531            .collect::<Vec<_>>();
1532
1533        let fill_bytes = ByteBuffer::copy_from("a fill value too long to inline");
1534        buffers.push(fill_bytes.clone());
1535        let fill = BinaryView::make_view(
1536            fill_bytes.as_slice(),
1537            u32::try_from(buffers.len() - 1).unwrap(),
1538            0,
1539        );
1540
1541        let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8);
1542        let views = patch_values.views();
1543        builder.append_views_scattered(
1544            buffers,
1545            5,
1546            fill,
1547            [(1usize, views[0]), (3usize, views[1])].into_iter(),
1548            &Mask::from_iter([true, true, false, true, true]),
1549        );
1550
1551        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1552            Some("a fill value too long to inline"),
1553            Some(LONG),
1554            None,
1555            Some("tiny"),
1556            Some("a fill value too long to inline"),
1557        ]);
1558        assert_arrays_eq!(builder.finish_into_varbinview(), expected, &mut ctx);
1559    }
1560
1561    /// A compacting builder must measure the buffers a gather hands it: with the middle
1562    /// dictionary value never referenced, adopting the heap whole would keep its bytes alive.
1563    /// Duplicate codes must share one rewritten copy, and null rows must not resolve their code.
1564    #[test]
1565    fn test_append_views_gathered_compacts_unreferenced_values() {
1566        const DEAD: &str = "a dead value nobody gathers, far too long to inline";
1567        const OTHER: &str = "another value far too long to inline";
1568
1569        let mut ctx = array_session().create_execution_ctx();
1570        let dictionary =
1571            <VarBinViewArray as FromIterator<_>>::from_iter([Some(LONG), Some(DEAD), Some(OTHER)]);
1572        assert_eq!(dictionary.data_buffers().len(), 1);
1573        let buffers = dictionary
1574            .data_buffers()
1575            .iter()
1576            .map(|buffer| buffer.as_host().clone())
1577            .collect::<Vec<_>>();
1578
1579        let mut builder =
1580            VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0);
1581        let codes: [usize; 5] = [2, 0, usize::MAX, 2, 0];
1582        builder.append_views_gathered(
1583            buffers,
1584            dictionary.views(),
1585            &Mask::from_iter([true, true, false, true, true]),
1586            |row| codes[row],
1587        );
1588
1589        let actual = builder.finish_into_varbinview();
1590        assert_eq!(actual.data_buffers().len(), 1);
1591        assert_eq!(
1592            actual.data_buffers()[0].len(),
1593            LONG.len() + OTHER.len(),
1594            "the rewritten buffer must hold each referenced value exactly once"
1595        );
1596
1597        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1598            Some(OTHER),
1599            Some(LONG),
1600            None,
1601            Some(OTHER),
1602            Some(LONG),
1603        ]);
1604        assert_arrays_eq!(actual, expected, &mut ctx);
1605    }
1606
1607    /// A gather that references only a contiguous tail of the heap must adopt just that slice,
1608    /// zero-copy.
1609    #[test]
1610    fn test_append_views_gathered_slices_contiguous_range() {
1611        const TAIL: &str = "the referenced tail value, too long to inline";
1612
1613        let mut ctx = array_session().create_execution_ctx();
1614        let dictionary = <VarBinViewArray as FromIterator<_>>::from_iter([Some(LONG), Some(TAIL)]);
1615        let heap = dictionary.data_buffers()[0].as_host().clone();
1616
1617        let mut builder =
1618            VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0);
1619        builder.append_views_gathered(
1620            [heap.clone()],
1621            dictionary.views(),
1622            &Mask::new_true(2),
1623            |_| 1,
1624        );
1625
1626        let actual = builder.finish_into_varbinview();
1627        assert_eq!(actual.data_buffers().len(), 1);
1628        assert_eq!(actual.data_buffers()[0].len(), TAIL.len());
1629        // SAFETY: LONG.len() is in bounds of the heap.
1630        assert_eq!(actual.data_buffers()[0].as_host().as_ptr(), unsafe {
1631            heap.as_ptr().add(LONG.len())
1632        });
1633
1634        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([Some(TAIL), Some(TAIL)]);
1635        assert_arrays_eq!(actual, expected, &mut ctx);
1636    }
1637
1638    /// A buffer whose referenced share clears the threshold is adopted whole, zero-copy.
1639    #[test]
1640    fn test_append_views_gathered_keeps_utilized_buffer() {
1641        const SHORTER: &str = "a shorter long value";
1642
1643        let mut ctx = array_session().create_execution_ctx();
1644        let dictionary =
1645            <VarBinViewArray as FromIterator<_>>::from_iter([Some(LONG), Some(SHORTER)]);
1646        let heap = dictionary.data_buffers()[0].as_host().clone();
1647
1648        let mut builder =
1649            VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 0.5);
1650        builder.append_views_gathered(
1651            [heap.clone()],
1652            dictionary.views(),
1653            &Mask::new_true(1),
1654            |_| 0,
1655        );
1656
1657        let actual = builder.finish_into_varbinview();
1658        assert_eq!(actual.data_buffers().len(), 1);
1659        assert_eq!(actual.data_buffers()[0].as_host().as_ptr(), heap.as_ptr());
1660        assert_eq!(actual.data_buffers()[0].len(), heap.len());
1661
1662        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([Some(LONG)]);
1663        assert_arrays_eq!(actual, expected, &mut ctx);
1664    }
1665
1666    /// An all-null gather references nothing, so a compacting builder must not retain any of the
1667    /// buffers — and must never resolve a code.
1668    #[test]
1669    fn test_append_views_gathered_all_null_drops_buffers() {
1670        let mut ctx = array_session().create_execution_ctx();
1671        let dictionary = <VarBinViewArray as FromIterator<_>>::from_iter([Some(LONG)]);
1672        let buffers = dictionary
1673            .data_buffers()
1674            .iter()
1675            .map(|buffer| buffer.as_host().clone())
1676            .collect::<Vec<_>>();
1677
1678        let mut builder =
1679            VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0);
1680        builder.append_views_gathered(buffers, dictionary.views(), &Mask::new_false(3), |_| {
1681            usize::MAX
1682        });
1683
1684        let actual = builder.finish_into_varbinview();
1685        assert!(
1686            actual.data_buffers().is_empty(),
1687            "an all-null gather must not retain any buffers"
1688        );
1689
1690        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([None::<&str>, None, None]);
1691        assert_arrays_eq!(actual, expected, &mut ctx);
1692    }
1693
1694    /// A fully-inlined heap has zero utilization; a compacting builder must not retain it at all.
1695    #[test]
1696    fn test_append_buffer_with_lengths_drops_fully_inlined_heap() {
1697        let mut ctx = array_session().create_execution_ctx();
1698        let mut builder =
1699            VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 4, 1.0);
1700
1701        let heap = ByteBuffer::copy_from(b"shorttinysmall".as_slice());
1702        builder.append_buffer_with_lengths(heap, &[5u32, 4, 5], &Mask::new_true(3));
1703
1704        let actual = builder.finish_into_varbinview();
1705        assert!(
1706            actual.data_buffers().is_empty(),
1707            "a fully-inlined append must not retain any value bytes"
1708        );
1709
1710        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1711            Some("short"),
1712            Some("tiny"),
1713            Some("small"),
1714        ]);
1715        assert_arrays_eq!(actual, expected, &mut ctx);
1716    }
1717
1718    #[test]
1719    fn test_utf8_builder() {
1720        let mut ctx = array_session().create_execution_ctx();
1721        let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 10);
1722
1723        builder.append_value("Hello");
1724        builder.append_null();
1725        builder.append_value("World");
1726
1727        builder.append_nulls(2);
1728
1729        builder.append_zeros(2);
1730        builder.append_value("test");
1731
1732        let actual = builder.finish();
1733        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1734            Some("Hello"),
1735            None,
1736            Some("World"),
1737            None,
1738            None,
1739            Some(""),
1740            Some(""),
1741            Some("test"),
1742        ]);
1743        assert_arrays_eq!(actual, expected, &mut ctx);
1744    }
1745
1746    #[test]
1747    fn test_utf8_builder_with_extend() {
1748        let mut ctx = array_session().create_execution_ctx();
1749        let array = {
1750            let mut builder =
1751                VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 10);
1752            builder.append_null();
1753            builder.append_value("Hello2");
1754            builder.finish()
1755        };
1756        let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 10);
1757
1758        builder.append_value("Hello1");
1759        array.append_to_builder(&mut builder, &mut ctx).unwrap();
1760        builder.append_nulls(2);
1761        builder.append_value("Hello3");
1762
1763        let actual = builder.finish_into_canonical(&mut ctx);
1764        let expected = <VarBinViewArray as FromIterator<_>>::from_iter([
1765            Some("Hello1"),
1766            None,
1767            Some("Hello2"),
1768            None,
1769            None,
1770            Some("Hello3"),
1771        ]);
1772        assert_arrays_eq!(actual.into_array(), expected.into_array(), &mut ctx);
1773    }
1774
1775    #[test]
1776    fn test_buffer_deduplication() -> VortexResult<()> {
1777        let array = {
1778            let mut builder =
1779                VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 10);
1780            builder.append_value("This is a long string that should not be inlined");
1781            builder.append_value("short string");
1782            builder.finish_into_varbinview()
1783        };
1784
1785        assert_eq!(array.data_buffers().len(), 1);
1786        let mut builder =
1787            VarBinViewBuilder::with_buffer_deduplication(DType::Utf8(Nullability::Nullable), 10);
1788
1789        let mut ctx = array_session().create_execution_ctx();
1790
1791        array.append_to_builder(&mut builder, &mut ctx)?;
1792        assert_eq!(builder.completed_block_count(), 1);
1793
1794        array
1795            .slice(1..2)?
1796            .append_to_builder(&mut builder, &mut ctx)?;
1797        array
1798            .slice(0..1)?
1799            .append_to_builder(&mut builder, &mut ctx)?;
1800        assert_eq!(builder.completed_block_count(), 1);
1801
1802        let array2 = {
1803            let mut builder =
1804                VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 10);
1805            builder.append_value("This is a long string that should not be inlined");
1806            builder.finish_into_varbinview()
1807        };
1808
1809        array2.append_to_builder(&mut builder, &mut ctx)?;
1810        assert_eq!(builder.completed_block_count(), 2);
1811
1812        array
1813            .slice(0..1)?
1814            .append_to_builder(&mut builder, &mut ctx)?;
1815        array2
1816            .slice(0..1)?
1817            .append_to_builder(&mut builder, &mut ctx)?;
1818        assert_eq!(builder.completed_block_count(), 2);
1819        Ok(())
1820    }
1821
1822    #[test]
1823    fn test_append_scalar() {
1824        let mut ctx = array_session().create_execution_ctx();
1825        use crate::scalar::Scalar;
1826
1827        // Test with Utf8 builder.
1828        let mut utf8_builder =
1829            VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 10);
1830
1831        // Test appending a valid utf8 value.
1832        let utf8_scalar1 = Scalar::utf8("hello", Nullability::Nullable);
1833        utf8_builder.append_scalar(&utf8_scalar1).unwrap();
1834
1835        // Test appending another value.
1836        let utf8_scalar2 = Scalar::utf8("world", Nullability::Nullable);
1837        utf8_builder.append_scalar(&utf8_scalar2).unwrap();
1838
1839        // Test appending null value.
1840        let null_scalar = Scalar::null(DType::Utf8(Nullability::Nullable));
1841        utf8_builder.append_scalar(&null_scalar).unwrap();
1842
1843        let array = utf8_builder.finish();
1844        let expected =
1845            <VarBinViewArray as FromIterator<_>>::from_iter([Some("hello"), Some("world"), None]);
1846        assert_arrays_eq!(&array, &expected, &mut ctx);
1847
1848        // Test with Binary builder.
1849        let mut binary_builder =
1850            VarBinViewBuilder::with_capacity(DType::Binary(Nullability::Nullable), 10);
1851
1852        let binary_scalar = Scalar::binary(vec![1u8, 2, 3], Nullability::Nullable);
1853        binary_builder.append_scalar(&binary_scalar).unwrap();
1854
1855        let binary_null = Scalar::null(DType::Binary(Nullability::Nullable));
1856        binary_builder.append_scalar(&binary_null).unwrap();
1857
1858        let binary_array = binary_builder.finish();
1859        let expected =
1860            <VarBinViewArray as FromIterator<_>>::from_iter([Some(vec![1u8, 2, 3]), None]);
1861        assert_arrays_eq!(&binary_array, &expected, &mut ctx);
1862
1863        // Test wrong dtype error.
1864        let mut builder =
1865            VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::NonNullable), 10);
1866        let wrong_scalar = Scalar::from(42i32);
1867        assert!(builder.append_scalar(&wrong_scalar).is_err());
1868    }
1869
1870    #[test]
1871    fn test_buffer_growth_strategies() {
1872        use super::BufferGrowthStrategy;
1873
1874        // Test Fixed strategy
1875        let mut strategy = BufferGrowthStrategy::fixed(1024);
1876
1877        // Should always return the fixed size
1878        assert_eq!(strategy.next_size(), 1024);
1879        assert_eq!(strategy.next_size(), 1024);
1880        assert_eq!(strategy.next_size(), 1024);
1881
1882        // Test Exponential strategy
1883        let mut strategy = BufferGrowthStrategy::exponential(1024, 8192);
1884
1885        // Should double each time until hitting max_size
1886        assert_eq!(strategy.next_size(), 1024); // First: 1024
1887        assert_eq!(strategy.next_size(), 2048); // Second: 2048
1888        assert_eq!(strategy.next_size(), 4096); // Third: 4096
1889        assert_eq!(strategy.next_size(), 8192); // Fourth: 8192 (max)
1890        assert_eq!(strategy.next_size(), 8192); // Fifth: 8192 (capped)
1891    }
1892
1893    #[test]
1894    fn test_large_value_allocation() {
1895        use super::BufferGrowthStrategy;
1896        use super::VarBinViewBuilder;
1897
1898        let mut builder = VarBinViewBuilder::new(
1899            DType::Binary(Nullability::Nullable),
1900            10,
1901            Default::default(),
1902            BufferGrowthStrategy::exponential(1024, 4096),
1903            0.0,
1904        );
1905
1906        // Create a value larger than max_size
1907        let large_value = vec![0u8; 8192];
1908
1909        // Should successfully append the large value
1910        builder.append_value(&large_value);
1911
1912        let array = builder.finish_into_varbinview();
1913        assert_eq!(array.len(), 1);
1914
1915        // Verify the value was stored correctly
1916        let retrieved = array
1917            .execute_scalar(0, &mut array_session().create_execution_ctx())
1918            .unwrap()
1919            .as_binary()
1920            .value()
1921            .cloned()
1922            .unwrap();
1923        assert_eq!(retrieved.len(), 8192);
1924        assert_eq!(retrieved.as_slice(), &large_value);
1925    }
1926}