1use 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
40pub 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 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 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 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 fn compacts_buffers(&self) -> bool {
201 self.compaction_threshold > 0.0
202 }
203
204 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 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 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 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 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 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 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 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 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 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 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 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 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 #[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 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 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#[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
838enum CompactedSlot {
840 Kept { index: u32 },
842 Sliced { index: u32, shift: u32 },
844 Rewrite { source: ByteBuffer },
847}
848
849fn unmeasured_utilizations(buffers: &[ByteBuffer]) -> Vec<BufferUtilization> {
851 buffers
852 .iter()
853 .map(|buffer| {
854 BufferUtilization::zero(u32::try_from(buffer.len()).unwrap_or(u32::MAX))
858 })
859 .collect()
860}
861
862fn 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#[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 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 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 #[expect(clippy::cast_possible_truncation)]
971 fn len(&self) -> u32 {
972 self.buffers.len() as u32
973 }
974
975 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 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 Fixed { size: u32 },
1035 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, max_size: 2 * 1024 * 1024, }
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 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 (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 (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 (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 (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 {
1184 start: u32,
1185 end: u32,
1186 },
1187 Rewrite,
1189}
1190
1191fn compaction_strategy(
1192 buffer_utilization: &BufferUtilization,
1193 threshold: f64,
1194) -> CompactionStrategy {
1195 match buffer_utilization.overall_utilization() {
1196 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
1235enum 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 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 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 #[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 const LONG: &str = "a value that is far too long to inline";
1344
1345 #[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 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 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 #[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 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 #[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 #[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 #[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 #[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 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 |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 #[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 #[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 #[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 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 #[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 #[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 #[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 let mut utf8_builder =
1829 VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 10);
1830
1831 let utf8_scalar1 = Scalar::utf8("hello", Nullability::Nullable);
1833 utf8_builder.append_scalar(&utf8_scalar1).unwrap();
1834
1835 let utf8_scalar2 = Scalar::utf8("world", Nullability::Nullable);
1837 utf8_builder.append_scalar(&utf8_scalar2).unwrap();
1838
1839 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 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 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 let mut strategy = BufferGrowthStrategy::fixed(1024);
1876
1877 assert_eq!(strategy.next_size(), 1024);
1879 assert_eq!(strategy.next_size(), 1024);
1880 assert_eq!(strategy.next_size(), 1024);
1881
1882 let mut strategy = BufferGrowthStrategy::exponential(1024, 8192);
1884
1885 assert_eq!(strategy.next_size(), 1024); assert_eq!(strategy.next_size(), 2048); assert_eq!(strategy.next_size(), 4096); assert_eq!(strategy.next_size(), 8192); assert_eq!(strategy.next_size(), 8192); }
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 let large_value = vec![0u8; 8192];
1908
1909 builder.append_value(&large_value);
1911
1912 let array = builder.finish_into_varbinview();
1913 assert_eq!(array.len(), 1);
1914
1915 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}