From b40c0cf476acbbc1c21ecf40d89b1e2da303cf8a Mon Sep 17 00:00:00 2001 From: Robert Kruszewski Date: Thu, 6 Aug 2026 14:47:44 +0100 Subject: [PATCH 1/5] VarBinViewBuilder: gather and scatter views for Dict and Sparse appends MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dict was the one string encoding with no direct VarBinViewBuilder path: it took the dictionary to full logical length (allocating an intermediate views buffer proportional to the row count) and then append_varbinview_array walked all those views a second time to rebase their buffer indices. Sparse had no append_to_builder override at all and paid the same double pass over its scattered views. - VarBinViewBuilder grows push_buffers (flush-aware, dedup-aware buffer adoption returning the index each buffer landed at) and two bulk view appends built on it: append_views_gathered (one rebased view write per row through an index lookup, null rows skip the lookup) and append_views_scattered (one bulk fill-view write plus one write per patch). - Dict gathers views through its codes straight into the builder: the dictionary's buffers are adopted once — deduplicated across chunks sharing the dictionary — with no byte copy and no intermediate array. - Sparse overrides append_to_builder for strings: the view builder gets the scatter directly, and VarBinBuilder gets an in-order walk of fill runs (append_n_values) and patches, preserving last-wins semantics for duplicate patch indices. Non-string dtypes keep the canonicalize fallback. - execute_varbin_inner no longer pushes a data buffer for a fill value short enough to inline — the view never referenced it, so it was pure dead weight in every canonical decode of a short-filled sparse array. Checks: cargo nextest -p vortex-array -p vortex-sparse; cargo +nightly fmt --all; cargo clippy --all-targets on the touched crates. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Uet46qdrbQcF61HXRG7EKc Signed-off-by: Robert Kruszewski --- encodings/sparse/src/canonical.rs | 280 ++++++++++++++++++++- encodings/sparse/src/lib.rs | 32 +++ vortex-array/src/arrays/dict/vtable/mod.rs | 99 ++++++++ vortex-array/src/builders/varbinview.rs | 221 +++++++++++++++- 4 files changed, 621 insertions(+), 11 deletions(-) diff --git a/encodings/sparse/src/canonical.rs b/encodings/sparse/src/canonical.rs index 2cfed7fd25b..c0779a2781b 100644 --- a/encodings/sparse/src/canonical.rs +++ b/encodings/sparse/src/canonical.rs @@ -4,8 +4,11 @@ use std::sync::Arc; use itertools::Itertools; +use num_traits::AsPrimitive; use num_traits::NumCast; use vortex_array::ArrayRef; +use vortex_array::ArrayView; +use vortex_array::Canonical; use vortex_array::ExecutionCtx; use vortex_array::IntoArray; use vortex_array::arrays::BoolArray; @@ -31,6 +34,8 @@ use vortex_array::builders::ArrayBuilder; use vortex_array::builders::DecimalBuilder; use vortex_array::builders::FixedSizeListBuilder; use vortex_array::builders::ListViewBuilder; +use vortex_array::builders::VarBinBuilder; +use vortex_array::builders::VarBinViewBuilder; use vortex_array::builders::builder_with_capacity; use vortex_array::dtype::DType; use vortex_array::dtype::DecimalDType; @@ -62,11 +67,198 @@ use vortex_error::VortexError; use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_bail; +use vortex_error::vortex_ensure; +use vortex_mask::Mask; use crate::ConstantArray; use crate::Sparse; +use crate::SparseExt; use crate::SparseParts; +/// The pieces every string decode of a sparse array starts from: the offset-resolved patch +/// indices and canonical patch values, plus the fill value's bytes (`None` for a null fill). +struct SparseStringParts { + len: usize, + patches: Patches, + indices: PrimitiveArray, + values: VarBinViewArray, + fill: Option, +} + +impl SparseStringParts { + fn new(array: ArrayView<'_, Sparse>, ctx: &mut ExecutionCtx) -> VortexResult { + let patches = array.resolved_patches()?; + let indices = patches.indices().clone().execute::(ctx)?; + let values = patches + .values() + .clone() + .execute::(ctx)? + .into_varbinview(); + let fill_scalar = array.data().fill_scalar(); + let fill = match array.dtype() { + DType::Utf8(_) => fill_scalar + .as_utf8() + .value() + .cloned() + .map(BufferString::into_inner), + DType::Binary(_) => fill_scalar.as_binary().value().cloned(), + dtype => vortex_bail!("Sparse string decode of non-string dtype {dtype}"), + }; + Ok(Self { + len: array.len(), + patches, + indices, + values, + fill, + }) + } +} + +/// Scatters the patch views over the fill value straight into `builder`. +/// +/// The canonical route materializes the same scattered views buffer, wraps it in an array, and +/// then `append_varbinview_array` walks all `len` views a second time to rebase their buffer +/// indices onto the builder's. Scattering into the builder instead adopts the patch buffers once +/// and costs one bulk view fill plus one view write per patch, with no byte copy. +pub(super) fn append_sparse_to_varbinview( + array: ArrayView<'_, Sparse>, + builder: &mut VarBinViewBuilder, + ctx: &mut ExecutionCtx, +) -> VortexResult<()> { + let parts = SparseStringParts::new(array, ctx)?; + let validity = sparse_validity( + &parts.patches, + array.data().fill_scalar(), + array.dtype().nullability(), + parts.len, + ctx, + )? + .execute_mask(parts.len, ctx)?; + + let mut buffers = parts + .values + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + // A fill value short enough to inline lives in its view; only a longer one needs its bytes + // adopted as a (tiny) data buffer. + let fill_view = match &parts.fill { + Some(bytes) if bytes.len() > BinaryView::MAX_INLINED_SIZE => { + buffers.push(bytes.clone()); + BinaryView::make_view( + bytes.as_slice(), + u32::try_from(buffers.len() - 1).vortex_expect("too many buffers"), + 0, + ) + } + Some(bytes) => BinaryView::make_view(bytes.as_slice(), 0, 0), + None => BinaryView::make_view(&[], 0, 0), + }; + + let views = parts.values.views(); + match_each_integer_ptype!(parts.indices.ptype(), |I| { + let indices = parts.indices.as_slice::(); + builder.append_views_scattered( + buffers, + parts.len, + fill_view, + indices + .iter() + .zip(views.iter()) + .map(|(row, view)| (AsPrimitive::::as_(*row), *view)), + &validity, + ); + }); + Ok(()) +} + +/// Walks the rows in order, appending fill runs between the patches, straight into `builder`. +/// +/// The canonical route materializes the scattered views array first; the fill-value copies per +/// row are inherent to the offsets layout either way, so appending directly just skips that +/// intermediate. +pub(super) fn append_sparse_to_varbin( + array: ArrayView<'_, Sparse>, + builder: &mut VarBinBuilder, + ctx: &mut ExecutionCtx, +) -> VortexResult<()> +where + usize: AsPrimitive, +{ + let parts = SparseStringParts::new(array, ctx)?; + let patch_validity = parts + .values + .validity()? + .execute_mask(parts.values.len(), ctx)?; + + let views = parts.values.views(); + let buffers = parts + .values + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().as_slice()) + .collect::>(); + + match_each_integer_ptype!(parts.indices.ptype(), |I| { + append_patch_walk_to_varbin( + builder, + parts.indices.as_slice::(), + parts.len, + views, + &buffers, + &patch_validity, + parts.fill.as_ref(), + ) + }) +} + +/// The typed patch walk behind [`append_sparse_to_varbin`]. +fn append_patch_walk_to_varbin>( + builder: &mut VarBinBuilder, + indices: &[I], + len: usize, + views: &[BinaryView], + buffers: &[&[u8]], + patch_validity: &Mask, + fill: Option<&ByteBuffer>, +) -> VortexResult<()> +where + usize: AsPrimitive, +{ + let fill_run = |builder: &mut VarBinBuilder, n: usize| -> VortexResult<()> { + match fill { + Some(bytes) => builder.append_n_values(bytes, n), + None => { + builder.push_nulls(n); + Ok(()) + } + } + }; + + let mut previous = 0usize; + for (patch, row) in indices.iter().enumerate() { + let row = AsPrimitive::::as_(*row); + // The canonical decode scatters the patches in order, so of duplicate indices the + // last one wins; skip a patch that the next one lands on top of. + if patch + 1 < indices.len() && AsPrimitive::::as_(indices[patch + 1]) == row { + continue; + } + vortex_ensure!( + previous <= row && row < len, + "Sparse patch indices must be ascending within the array length {len}" + ); + fill_run(builder, row - previous)?; + if patch_validity.value(patch) { + builder.append_value(views[patch].bytes(buffers)); + } else { + builder.push_null(); + } + previous = row + 1; + } + fill_run(builder, len - previous) +} + fn sparse_validity( patches: &Patches, fill_value: &Scalar, @@ -571,16 +763,20 @@ fn execute_varbin_inner( let n_patch_buffers = values.data_buffers().len(); let mut buffers = values.data_buffers().to_vec(); - let fill = if let Some(buffer) = &fill_value { - buffers.push(BufferHandle::new_host(buffer.clone())); - BinaryView::make_view( - buffer.as_ref(), - u32::try_from(n_patch_buffers).vortex_expect("too many buffers"), - 0, - ) - } else { + let fill = match &fill_value { + // A fill value short enough to inline lives in its view; pushing its buffer would leave + // that buffer entirely unreferenced. + Some(buffer) if buffer.len() > BinaryView::MAX_INLINED_SIZE => { + buffers.push(BufferHandle::new_host(buffer.clone())); + BinaryView::make_view( + buffer.as_ref(), + u32::try_from(n_patch_buffers).vortex_expect("too many buffers"), + 0, + ) + } + Some(buffer) => BinaryView::make_view(buffer.as_ref(), 0, 0), // any <=12 character value will do - BinaryView::make_view(&[], 0, 0) + None => BinaryView::make_view(&[], 0, 0), }; let mut views = buffer_mut![fill; len]; @@ -617,6 +813,8 @@ mod test { use vortex_array::arrays::listview::ListViewArrayExt; use vortex_array::arrays::listview::ListViewArraySlotsExt; use vortex_array::assert_arrays_eq; + use vortex_array::builders::VarBinBuilder; + use vortex_array::builders::VarBinViewBuilder; use vortex_array::dtype::DType; use vortex_array::dtype::DecimalDType; use vortex_array::dtype::FieldNames; @@ -1761,4 +1959,68 @@ mod test { // when dealing with sliced arrays that have non-zero starting offsets. Ok(()) } + + /// Appending a sparse string array to either variable-binary builder must match its + /// canonical decode exactly — across long, short (inlinable), and null fill values, null + /// patch values, and a sliced array whose patch indices carry an offset. + #[rstest] + #[case::long_fill(Some("a fill value that is too long to inline"))] + #[case::short_fill(Some("123"))] + #[case::null_fill(None)] + fn test_sparse_append_to_string_builders(#[case] fill: Option<&str>) -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + let long = "a patch value that is far too long to inline"; + let strings = + >::from_iter([Some(long), None, Some("tiny")]) + .into_array(); + let fill_scalar = match fill { + Some(value) => Scalar::from(value.to_owned()).into_nullable(), + None => Scalar::null(DType::Utf8(Nullable)), + }; + let array = Sparse::try_new(buffer![1u16, 4, 8].into_array(), strings, 10, fill_scalar)? + .into_array(); + + for candidate in [array.clone(), array.slice(1..9)?] { + let expected = candidate.clone().execute::(&mut ctx)?; + + let mut view_builder = VarBinViewBuilder::with_capacity(candidate.dtype().clone(), 4); + candidate.append_to_builder(&mut view_builder, &mut ctx)?; + assert_arrays_eq!(view_builder.finish_into_varbinview(), expected, &mut ctx); + + let mut varbin_builder = VarBinBuilder::::new(candidate.dtype().clone()); + candidate.append_to_builder(&mut varbin_builder, &mut ctx)?; + assert_arrays_eq!(varbin_builder.finish_into_varbin(), expected, &mut ctx); + } + Ok(()) + } + + /// A fill value short enough to inline lives in its views; the canonical decode must not + /// retain a data buffer nothing references. + #[test] + fn test_sparse_short_fill_pushes_no_buffer() -> VortexResult<()> { + let mut ctx = SESSION.create_execution_ctx(); + let strings = VarBinViewArray::from_iter_str(["hello", "goodbye"]); + assert!(strings.data_buffers().is_empty()); + + let array = Sparse::try_new( + buffer![1u16, 3].into_array(), + strings.into_array(), + 6, + Scalar::from("123".to_owned()), + )?; + + let actual = array + .as_array() + .clone() + .execute::(&mut ctx)?; + assert!( + actual.data_buffers().is_empty(), + "an inlinable fill value must not push an unreferenced buffer" + ); + + let expected = + VarBinViewArray::from_iter_str(["123", "hello", "123", "goodbye", "123", "123"]); + assert_arrays_eq!(actual, expected, &mut ctx); + Ok(()) + } } diff --git a/encodings/sparse/src/lib.rs b/encodings/sparse/src/lib.rs index 733eb18da0d..f3a488ea9ba 100644 --- a/encodings/sparse/src/lib.rs +++ b/encodings/sparse/src/lib.rs @@ -28,9 +28,12 @@ use vortex_array::arrays::Primitive; use vortex_array::arrays::PrimitiveArray; use vortex_array::arrays::bool::BoolArrayExt; use vortex_array::buffer::BufferHandle; +use vortex_array::builders::ArrayBuilder; +use vortex_array::builders::VarBinViewBuilder; use vortex_array::builtins::ArrayBuiltins; use vortex_array::dtype::DType; use vortex_array::dtype::Nullability; +use vortex_array::match_each_varbin_builder; use vortex_array::patches::PatchSlotIndices; use vortex_array::patches::Patches; use vortex_array::patches::PatchesData; @@ -57,6 +60,8 @@ use vortex_mask::Mask; use vortex_session::VortexSession; use vortex_session::registry::CachedId; +use crate::canonical::append_sparse_to_varbin; +use crate::canonical::append_sparse_to_varbinview; use crate::canonical::execute_sparse; use crate::rules::RULES; @@ -353,6 +358,33 @@ impl VTable for Sparse { // TODO(joe): remove ctx from execute_sparse since all slots should be canonical. execute_sparse(parts, ctx).map(ExecutionResult::done) } + + fn append_to_builder( + array: ArrayView<'_, Self>, + builder: &mut dyn ArrayBuilder, + ctx: &mut ExecutionCtx, + ) -> VortexResult<()> { + // Strings decode as a scatter over the fill value; appending that scatter straight into + // a variable-binary builder skips materializing the canonical intermediate. + if matches!(array.dtype(), DType::Utf8(_) | DType::Binary(_)) { + if let Some(builder) = builder.as_any_mut().downcast_mut::() { + return append_sparse_to_varbinview(array, builder, ctx); + } + if let Some(result) = match_each_varbin_builder!(builder, |builder| { + append_sparse_to_varbin(array, builder, ctx) + }) { + return result; + } + } + + // Everything else decodes through the canonical array, like the default implementation. + array + .array() + .clone() + .execute::(ctx)? + .into_array() + .append_to_builder(builder, ctx) + } } const PATCH_SLOTS: PatchSlotIndices = PatchSlotIndices { diff --git a/vortex-array/src/arrays/dict/vtable/mod.rs b/vortex-array/src/arrays/dict/vtable/mod.rs index 62e324f4ba5..92f80f45908 100644 --- a/vortex-array/src/arrays/dict/vtable/mod.rs +++ b/vortex-array/src/arrays/dict/vtable/mod.rs @@ -46,6 +46,7 @@ use crate::arrays::dict::execute::take_canonical; use crate::buffer::BufferHandle; use crate::builders::ArrayBuilder; use crate::builders::VarBinBuilder; +use crate::builders::VarBinViewBuilder; use crate::dtype::DType; use crate::dtype::Nullability; use crate::dtype::OffsetBuilderPType; @@ -239,6 +240,12 @@ impl VTable for Dict { { return result; } + if let CanonicalView::VarBinView(values) = values + && let Some(builder) = builder.as_any_mut().downcast_mut::() + { + let validity = array.validity()?.execute_mask(array.len(), ctx)?; + return append_dict_to_varbinview(codes, values, validity, builder); + } let canonical = take_canonical(values, codes, ctx)?.into_array(); canonical.append_to_builder(builder, ctx)?; return Ok(()); @@ -262,6 +269,36 @@ impl VTable for Dict { } } +/// Gathers the dictionary values straight into `builder` as views. +/// +/// The canonical route first takes the values to full logical length — allocating an intermediate +/// views buffer proportional to the row count — and then `append_varbinview_array` walks all +/// those views a second time to rebase their buffer indices onto the builder's. Gathering through +/// the codes directly instead adopts the dictionary's buffers once (deduplicated across chunks +/// sharing the dictionary, when the builder deduplicates) and writes each row's rebased view in a +/// single pass, with no byte copy at all. +fn append_dict_to_varbinview( + codes: ArrayView<'_, Primitive>, + values: ArrayView<'_, VarBinView>, + validity: Mask, + builder: &mut VarBinViewBuilder, +) -> VortexResult<()> { + let views = values.views(); + let buffers = values + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + + match_each_integer_ptype!(codes.ptype(), |C| { + let codes = codes.as_slice::(); + builder.append_views_gathered(buffers, views, &validity, |row| { + AsPrimitive::::as_(codes[row]) + }); + }); + Ok(()) +} + /// Gathers the dictionary values straight into `builder`. /// /// The canonical route first takes the values to full logical length, which allocates and then @@ -343,4 +380,66 @@ mod tests { assert_arrays_eq!(builder.finish_into_varbin(), expected, &mut ctx); Ok(()) } + + /// The view-builder path gathers views through the codes without materializing the taken + /// array, so the result must still match the canonical take — including rows whose code is + /// null and rows whose dictionary value is null. + #[test] + fn append_to_view_builder_gathers_through_the_dictionary() -> VortexResult<()> { + let mut ctx = array_session().create_execution_ctx(); + let dict = DictArray::try_new( + PrimitiveArray::from_option_iter([Some(0u32), Some(2), None, Some(1), Some(0)]) + .into_array(), + VarBinViewArray::from_iter([Some(LONG), None, Some("short")], DType::Utf8(Nullable)) + .into_array(), + )?; + + let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullable), 8); + builder.append_value(LONG); + dict.append_to_builder(&mut builder, &mut ctx)?; + + let expected = VarBinViewArray::from_iter( + [ + Some(LONG), + Some(LONG), + Some("short"), + None, + None, + Some(LONG), + ], + DType::Utf8(Nullable), + ); + assert_arrays_eq!(builder.finish_into_varbinview(), expected, &mut ctx); + Ok(()) + } + + /// Two chunks gathered through the same dictionary into a deduplicating builder must adopt + /// the dictionary's data buffers once. + #[test] + fn append_to_dedup_view_builder_adopts_the_dictionary_once() -> VortexResult<()> { + let mut ctx = array_session().create_execution_ctx(); + let values = VarBinViewArray::from_iter([Some(LONG), Some("short")], DType::Utf8(Nullable)); + assert_eq!(values.data_buffers().len(), 1); + + let first = DictArray::try_new( + PrimitiveArray::from_option_iter([Some(0u32), Some(1)]).into_array(), + values.clone().into_array(), + )?; + let second = DictArray::try_new( + PrimitiveArray::from_option_iter([Some(1u32), Some(0)]).into_array(), + values.into_array(), + )?; + + let mut builder = VarBinViewBuilder::with_buffer_deduplication(DType::Utf8(Nullable), 8); + first.append_to_builder(&mut builder, &mut ctx)?; + second.append_to_builder(&mut builder, &mut ctx)?; + assert_eq!(builder.completed_block_count(), 1); + + let expected = VarBinViewArray::from_iter( + [Some(LONG), Some("short"), Some("short"), Some(LONG)], + DType::Utf8(Nullable), + ); + assert_arrays_eq!(builder.finish_into_varbinview(), expected, &mut ctx); + Ok(()) + } } diff --git a/vortex-array/src/builders/varbinview.rs b/vortex-array/src/builders/varbinview.rs index 57d4ff64637..f043860d594 100644 --- a/vortex-array/src/builders/varbinview.rs +++ b/vortex-array/src/builders/varbinview.rs @@ -194,8 +194,9 @@ impl VarBinViewBuilder { } /// Whether this builder compacts the data buffers it is handed. The lengths-driven appends - /// use this to gate their utilization measurement; the buffer-adopting escape hatch - /// ([`append_views_built_at`](Self::append_views_built_at)) always bypasses it. + /// use this to gate their utilization measurement; the buffer-adopting escape hatches + /// ([`append_views_built_at`](Self::append_views_built_at) and + /// [`push_buffers`](Self::push_buffers)) always bypass it. fn compacts_buffers(&self) -> bool { self.compaction_threshold > 0.0 } @@ -248,6 +249,106 @@ impl VarBinViewBuilder { Ok(()) } + /// Adopts `buffers` as completed data buffers, returning the index each landed at. + /// + /// Views appended afterwards (e.g. via [`append_views_gathered`](Self::append_views_gathered) + /// or [`append_views_scattered`](Self::append_views_scattered)) reference values through the + /// returned indices. A deduplicating builder returns the existing index for a buffer it + /// already holds, so repeated appends over shared storage — chunks gathered through one + /// dictionary, slices of one array — adopt it once. + /// + /// # Warning + /// + /// Buffers are taken as they are, without utilization measurement; like + /// [`append_views_built_at`](Self::append_views_built_at), a caller whose views may leave a + /// buffer mostly unreferenced silently opts the builder out of any compaction it was + /// configured for. + pub fn push_buffers(&mut self, buffers: impl IntoIterator) -> Vec { + self.flush_in_progress(); + buffers + .into_iter() + .map(|buffer| self.completed.push(buffer)) + .collect() + } + + /// Appends `validity.len()` values, gathering each valid row's view from `views` through the + /// index `view_at` returns for it. + /// + /// `buffers` — the data buffers the views reference, in the numbering the views use — are + /// adopted via [`push_buffers`](Self::push_buffers), and every gathered view is rebased onto + /// the indices they land at as it is written, so the whole append is one view per row with no + /// byte copy and no intermediate array. `view_at` is only called for valid rows; null rows get + /// an empty view. + /// + /// # Panics + /// + /// Panics if `view_at` returns an index out of bounds of `views`, or if a gathered view + /// references a buffer index outside `buffers`. + pub fn append_views_gathered( + &mut self, + buffers: impl IntoIterator, + views: &[BinaryView], + validity: &Mask, + view_at: impl Fn(usize) -> usize, + ) { + let mapping = self.push_buffers(buffers); + + self.views_builder.reserve(validity.len()); + match validity { + Mask::AllTrue(len) => self + .views_builder + .extend_trusted((0..*len).map(|row| remap_view(views[view_at(row)], &mapping))), + Mask::AllFalse(len) => self.views_builder.push_n(BinaryView::empty_view(), *len), + Mask::Values(values) => { + for (row, is_valid) in values.bit_buffer().iter().enumerate() { + let view = if is_valid { + remap_view(views[view_at(row)], &mapping) + } else { + BinaryView::empty_view() + }; + self.views_builder.push(view); + } + } + } + + self.push_only_validity_mask(validity); + debug_assert_eq!(self.nulls.len(), self.views_builder.len()); + } + + /// Appends `len` rows that are `fill` everywhere except at the given patch rows. + /// + /// `buffers` — the data buffers `fill` and the patch views reference, in the numbering they + /// use — are adopted via [`push_buffers`](Self::push_buffers) and the views are rebased onto + /// the indices they land at. `patches` yields `(row, view)` pairs with rows below `len`; the + /// fill rows cost one bulk view fill and each patch one view write, with no byte copy. + /// Validity is appended exactly as given — invalid rows keep whichever view they got. + /// + /// # Panics + /// + /// Panics if `validity` is not `len` long, a patch row is out of bounds, or a view references + /// a buffer index outside `buffers`. + pub fn append_views_scattered( + &mut self, + buffers: impl IntoIterator, + len: usize, + fill: BinaryView, + patches: impl Iterator, + validity: &Mask, + ) { + assert_eq!(validity.len(), len, "Must have one validity entry per row"); + let mapping = self.push_buffers(buffers); + + let start = self.views_builder.len(); + self.views_builder.push_n(remap_view(fill, &mapping), len); + let scattered = &mut self.views_builder[start..]; + for (row, view) in patches { + scattered[row] = remap_view(view, &mapping); + } + + self.push_only_validity_mask(validity); + debug_assert_eq!(self.nulls.len(), self.views_builder.len()); + } + /// Appends values laid end-to-end in `bytes`, one per entry of `lengths`. /// /// The builder adopts `bytes` as a data buffer without copying it (splitting it only past the @@ -598,6 +699,21 @@ impl VarBinViewBuilder { } } +/// Rebases a view built against a local buffer numbering onto the builder indices those buffers +/// landed at, i.e. `mapping[i]` is where the caller's buffer `i` went. Inlined views carry no +/// buffer reference and pass through unchanged. +#[inline] +fn remap_view(view: BinaryView, mapping: &[u32]) -> BinaryView { + if view.is_inlined() { + view + } else { + let view_ref = view.as_view(); + view_ref + .with_buffer_and_offset(mapping[view_ref.buffer_index as usize], view_ref.offset) + .into() + } +} + pub enum CompletedBuffers { Default(Vec), Deduplicated(DeduplicatedBuffers), @@ -1188,6 +1304,107 @@ mod tests { assert_arrays_eq!(actual, expected, &mut ctx); } + /// `push_buffers` returns where each buffer landed, and a deduplicating builder maps a + /// re-pushed buffer back to its existing index instead of holding it twice. + #[test] + fn test_push_buffers_deduplicates() { + let mut builder = + VarBinViewBuilder::with_buffer_deduplication(DType::Utf8(Nullability::Nullable), 8); + + let first = ByteBuffer::copy_from(LONG); + let second = ByteBuffer::copy_from("another value far too long to inline"); + + assert_eq!( + builder.push_buffers([first.clone(), second.clone()]), + [0, 1] + ); + assert_eq!(builder.push_buffers([second, first]), [1, 0]); + assert_eq!(builder.completed_block_count(), 2); + } + + /// Gathered views are rebased onto wherever the pushed buffers landed, and null rows never + /// resolve their index. + #[test] + fn test_append_views_gathered() { + let mut ctx = array_session().create_execution_ctx(); + let dictionary = >::from_iter([ + Some("tiny"), + Some(LONG), + Some("small"), + ]); + let buffers = dictionary + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + + let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8); + // Stages an in-progress buffer that the gather has to flush ahead of its own buffers. + builder.append_value(LONG); + + let codes: [usize; 4] = [1, 0, usize::MAX, 2]; + let views = dictionary.views(); + builder.append_views_gathered( + buffers, + views, + &Mask::from_iter([true, true, false, true]), + // The null row's code is garbage; the builder must not look it up. + |row| codes[row], + ); + + let expected = >::from_iter([ + Some(LONG), + Some(LONG), + Some("tiny"), + None, + Some("small"), + ]); + assert_arrays_eq!(builder.finish_into_varbinview(), expected, &mut ctx); + } + + /// Scattered patches overwrite the fill view at their rows, and both are rebased onto the + /// adopted buffers. + #[test] + fn test_append_views_scattered() { + use crate::arrays::varbinview::build_views::BinaryView; + + let mut ctx = array_session().create_execution_ctx(); + let patch_values = + >::from_iter([Some(LONG), Some("tiny")]); + let mut buffers = patch_values + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + + let fill_bytes = ByteBuffer::copy_from("a fill value too long to inline"); + buffers.push(fill_bytes.clone()); + let fill = BinaryView::make_view( + fill_bytes.as_slice(), + u32::try_from(buffers.len() - 1).unwrap(), + 0, + ); + + let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8); + let views = patch_values.views(); + builder.append_views_scattered( + buffers, + 5, + fill, + [(1usize, views[0]), (3usize, views[1])].into_iter(), + &Mask::from_iter([true, true, false, true, true]), + ); + + let expected = >::from_iter([ + Some("a fill value too long to inline"), + Some(LONG), + None, + Some("tiny"), + Some("a fill value too long to inline"), + ]); + assert_arrays_eq!(builder.finish_into_varbinview(), expected, &mut ctx); + } + /// A fully-inlined heap has zero utilization; a compacting builder must not retain it at all. #[test] fn test_append_buffer_with_lengths_drops_fully_inlined_heap() { From 229b0c1750cfed21a86e76f0297152fbf50b5573 Mon Sep 17 00:00:00 2001 From: Robert Kruszewski Date: Fri, 7 Aug 2026 00:21:57 +0100 Subject: [PATCH 2/5] fixes Signed-off-by: Robert Kruszewski --- vortex-array/src/arrays/varbinview/compact.rs | 4 +- vortex-array/src/builders/varbinview.rs | 433 ++++++++++++++++-- 2 files changed, 409 insertions(+), 28 deletions(-) diff --git a/vortex-array/src/arrays/varbinview/compact.rs b/vortex-array/src/arrays/varbinview/compact.rs index 2aba21ae448..9f1fa8c350f 100644 --- a/vortex-array/src/arrays/varbinview/compact.rs +++ b/vortex-array/src/arrays/varbinview/compact.rs @@ -159,7 +159,7 @@ pub(crate) struct BufferUtilization { } impl BufferUtilization { - fn zero(len: u32) -> Self { + pub(crate) fn zero(len: u32) -> Self { BufferUtilization { len, used: 0u32, @@ -168,7 +168,7 @@ impl BufferUtilization { } } - fn add(&mut self, offset: u32, size: u32) { + pub(crate) fn add(&mut self, offset: u32, size: u32) { self.used += size; self.min_offset = self.min_offset.min(offset); self.max_offset_end = self.max_offset_end.max(offset + size); diff --git a/vortex-array/src/builders/varbinview.rs b/vortex-array/src/builders/varbinview.rs index f043860d594..076ec76efe0 100644 --- a/vortex-array/src/builders/varbinview.rs +++ b/vortex-array/src/builders/varbinview.rs @@ -194,8 +194,8 @@ impl VarBinViewBuilder { } /// Whether this builder compacts the data buffers it is handed. The lengths-driven appends - /// use this to gate their utilization measurement; the buffer-adopting escape hatches - /// ([`append_views_built_at`](Self::append_views_built_at) and + /// and the gather/scatter appends use this to gate their utilization measurement; the + /// buffer-adopting escape hatches ([`append_views_built_at`](Self::append_views_built_at) and /// [`push_buffers`](Self::push_buffers)) always bypass it. fn compacts_buffers(&self) -> bool { self.compaction_threshold > 0.0 @@ -260,10 +260,11 @@ impl VarBinViewBuilder { /// # Warning /// /// Buffers are taken as they are, without utilization measurement; like - /// [`append_views_built_at`](Self::append_views_built_at), a caller whose views may leave a - /// buffer mostly unreferenced silently opts the builder out of any compaction it was - /// configured for. - pub fn push_buffers(&mut self, buffers: impl IntoIterator) -> Vec { + /// [`append_views_built_at`](Self::append_views_built_at), this bypasses any compaction the + /// builder was configured for. A caller whose views may leave a buffer mostly unreferenced + /// must go through [`push_buffers_compacted`](Self::push_buffers_compacted) on a compacting + /// builder instead. + fn push_buffers(&mut self, buffers: impl IntoIterator) -> Vec { self.flush_in_progress(); buffers .into_iter() @@ -271,6 +272,60 @@ impl VarBinViewBuilder { .collect() } + /// The compacting counterpart of [`push_buffers`](Self::push_buffers): adopts each buffer + /// according to the [`CompactionStrategy`] its measured utilization earns — whole, sliced to + /// its referenced range, or not at all. Views into the buffers must then be rebased through + /// [`remap_view_compacted`](Self::remap_view_compacted), which routes views into unadopted + /// buffers through the builder's own storage. + fn push_buffers_compacted( + &mut self, + buffers: Vec, + utilizations: &[BufferUtilization], + ) -> Vec { + self.flush_in_progress(); + buffers + .into_iter() + .zip(utilizations) + .map(|(buffer, utilization)| { + match compaction_strategy(utilization, self.compaction_threshold) { + CompactionStrategy::KeepFull => CompactedSlot::Kept { + index: self.completed.push(buffer), + }, + CompactionStrategy::Slice { start, end } => CompactedSlot::Sliced { + index: self + .completed + .push(buffer.slice(start as usize..end as usize)), + shift: start, + }, + CompactionStrategy::Rewrite => CompactedSlot::Rewrite { source: buffer }, + } + }) + .collect() + } + + /// [`remap_view`] against [`CompactedSlot`]s: views into kept or sliced buffers are rebased + /// onto the index those landed at, and a view into a rewritten buffer has its bytes copied + /// into the builder's own storage. Inlined views pass through unchanged. + fn remap_view_compacted(&mut self, view: BinaryView, slots: &[CompactedSlot]) -> BinaryView { + if view.is_inlined() { + return view; + } + let view_ref = view.as_view(); + match &slots[view_ref.buffer_index as usize] { + CompactedSlot::Kept { index } => view_ref + .with_buffer_and_offset(*index, view_ref.offset) + .into(), + CompactedSlot::Sliced { index, shift } => view_ref + .with_buffer_and_offset(*index, view_ref.offset - shift) + .into(), + CompactedSlot::Rewrite { source } => { + let bytes = &source.as_slice()[view_ref.as_range()]; + let (buffer_idx, offset) = self.append_value_to_buffer(bytes); + BinaryView::make_view(bytes, buffer_idx, offset) + } + } + } + /// Appends `validity.len()` values, gathering each valid row's view from `views` through the /// index `view_at` returns for it. /// @@ -280,6 +335,11 @@ impl VarBinViewBuilder { /// byte copy and no intermediate array. `view_at` is only called for valid rows; null rows get /// an empty view. /// + /// When the builder is configured to compact buffers, utilization is measured from the views + /// the gather actually references — duplicate indices count once — and each buffer is + /// adopted, sliced, or rewritten accordingly, so callers never need a + /// canonicalize-and-compact fallback. + /// /// # Panics /// /// Panics if `view_at` returns an index out of bounds of `views`, or if a gathered view @@ -291,6 +351,15 @@ impl VarBinViewBuilder { validity: &Mask, view_at: impl Fn(usize) -> usize, ) { + if self.compacts_buffers() { + return self.append_views_gathered_compacted( + buffers.into_iter().collect(), + views, + validity, + view_at, + ); + } + let mapping = self.push_buffers(buffers); self.views_builder.reserve(validity.len()); @@ -300,18 +369,72 @@ impl VarBinViewBuilder { .extend_trusted((0..*len).map(|row| remap_view(views[view_at(row)], &mapping))), Mask::AllFalse(len) => self.views_builder.push_n(BinaryView::empty_view(), *len), Mask::Values(values) => { - for (row, is_valid) in values.bit_buffer().iter().enumerate() { - let view = if is_valid { - remap_view(views[view_at(row)], &mapping) - } else { - BinaryView::empty_view() - }; - self.views_builder.push(view); - } + let bits = values.bit_buffer(); + let start = self.views_builder.len(); + self.views_builder + .push_n(BinaryView::empty_view(), bits.len()); + let gathered = &mut self.views_builder[start..]; + bits.for_each_set_index(|row| { + gathered[row] = remap_view(views[view_at(row)], &mapping); + }); } } - self.push_only_validity_mask(validity); + self.nulls.append_validity_mask(validity); + debug_assert_eq!(self.nulls.len(), self.views_builder.len()); + } + + /// The compacting arm of [`append_views_gathered`](Self::append_views_gathered): marks the + /// source views the gather references, measures each buffer's utilization from just those — + /// so entries gathered by many rows count once — and adopts the buffers through + /// [`push_buffers_compacted`](Self::push_buffers_compacted). Each referenced source view is + /// remapped once, so rows sharing an entry of a rewritten buffer share one copy of its bytes. + fn append_views_gathered_compacted( + &mut self, + buffers: Vec, + views: &[BinaryView], + validity: &Mask, + view_at: impl Fn(usize) -> usize, + ) { + let mut referenced = vec![false; views.len()]; + match validity { + Mask::AllTrue(len) => (0..*len).for_each(|row| referenced[view_at(row)] = true), + Mask::AllFalse(_) => {} + Mask::Values(values) => values + .bit_buffer() + .for_each_set_index(|row| referenced[view_at(row)] = true), + } + + let mut utilizations = unmeasured_utilizations(&buffers); + for (view, _) in views.iter().zip(&referenced).filter(|(_, used)| **used) { + measure_view(&mut utilizations, view); + } + let slots = self.push_buffers_compacted(buffers, &utilizations); + + let mut remapped = vec![BinaryView::empty_view(); views.len()]; + for (idx, view) in views.iter().enumerate() { + if referenced[idx] { + remapped[idx] = self.remap_view_compacted(*view, &slots); + } + } + + self.views_builder.reserve(validity.len()); + match validity { + Mask::AllTrue(len) => self + .views_builder + .extend_trusted((0..*len).map(|row| remapped[view_at(row)])), + Mask::AllFalse(len) => self.views_builder.push_n(BinaryView::empty_view(), *len), + Mask::Values(values) => { + let bits = values.bit_buffer(); + let start = self.views_builder.len(); + self.views_builder + .push_n(BinaryView::empty_view(), bits.len()); + let gathered = &mut self.views_builder[start..]; + bits.for_each_set_index(|row| gathered[row] = remapped[view_at(row)]); + } + } + + self.nulls.append_validity_mask(validity); debug_assert_eq!(self.nulls.len(), self.views_builder.len()); } @@ -323,6 +446,10 @@ impl VarBinViewBuilder { /// fill rows cost one bulk view fill and each patch one view write, with no byte copy. /// Validity is appended exactly as given — invalid rows keep whichever view they got. /// + /// When the builder is configured to compact buffers, utilization is measured from the fill + /// view and the patch views and each buffer is adopted, sliced, or rewritten accordingly, so + /// callers never need a canonicalize-and-compact fallback. + /// /// # Panics /// /// Panics if `validity` is not `len` long, a patch row is out of bounds, or a view references @@ -336,6 +463,16 @@ impl VarBinViewBuilder { validity: &Mask, ) { assert_eq!(validity.len(), len, "Must have one validity entry per row"); + if self.compacts_buffers() { + return self.append_views_scattered_compacted( + buffers.into_iter().collect(), + len, + fill, + patches, + validity, + ); + } + let mapping = self.push_buffers(buffers); let start = self.views_builder.len(); @@ -345,7 +482,41 @@ impl VarBinViewBuilder { scattered[row] = remap_view(view, &mapping); } - self.push_only_validity_mask(validity); + self.nulls.append_validity_mask(validity); + debug_assert_eq!(self.nulls.len(), self.views_builder.len()); + } + + /// The compacting arm of [`append_views_scattered`](Self::append_views_scattered): measures + /// each buffer's utilization from the fill view and the patch views, then adopts the buffers + /// through [`push_buffers_compacted`](Self::push_buffers_compacted). Validity is appended as + /// given — an invalid row keeps its view — so every scattered view counts as referenced, + /// which keeps a rewrite from stranding an invalid row's view. + fn append_views_scattered_compacted( + &mut self, + buffers: Vec, + len: usize, + fill: BinaryView, + patches: impl Iterator, + validity: &Mask, + ) { + let patches = patches.collect::>(); + + let mut utilizations = unmeasured_utilizations(&buffers); + measure_view(&mut utilizations, &fill); + for (_, view) in &patches { + measure_view(&mut utilizations, view); + } + let slots = self.push_buffers_compacted(buffers, &utilizations); + + let fill = self.remap_view_compacted(fill, &slots); + let start = self.views_builder.len(); + self.views_builder.push_n(fill, len); + for (row, view) in patches { + let view = self.remap_view_compacted(view, &slots); + self.views_builder[start + row] = view; + } + + self.nulls.append_validity_mask(validity); debug_assert_eq!(self.nulls.len(), self.views_builder.len()); } @@ -589,16 +760,14 @@ impl VarBinViewBuilder { .push_n(BinaryView::empty_view(), array.len()); } Mask::Values(v) => { - for (idx, (&view, is_valid)) in - array.views().iter().zip(v.bit_buffer().iter()).enumerate() - { - let new_view = if !is_valid { - BinaryView::empty_view() - } else { - self.push_view(view, &adjustment, array, idx) - }; - self.views_builder.push(new_view); - } + let views = array.views(); + let start = self.views_builder.len(); + self.views_builder + .push_n(BinaryView::empty_view(), array.len()); + v.bit_buffer().for_each_set_index(|idx| { + let new_view = self.push_view(views[idx], &adjustment, array, idx); + self.views_builder[start + idx] = new_view; + }); } }, } @@ -714,6 +883,38 @@ fn remap_view(view: BinaryView, mapping: &[u32]) -> BinaryView { } } +/// Where a caller's buffer went under `VarBinViewBuilder::push_buffers_compacted`. +enum CompactedSlot { + /// Adopted whole at this index. + Kept { index: u32 }, + /// Adopted at this index as the slice starting `shift` bytes in, so view offsets shift down. + Sliced { index: u32, shift: u32 }, + /// Not adopted: each referenced view's bytes are copied out of the source buffer into the + /// builder's own storage. + Rewrite { source: ByteBuffer }, +} + +/// One unmeasured [`BufferUtilization`] per buffer, ready for [`measure_view`] passes. +fn unmeasured_utilizations(buffers: &[ByteBuffer]) -> Vec { + buffers + .iter() + .map(|buffer| { + // Views address at most `u32` offsets, so measuring an oversized buffer against the + // saturated length under-reports utilization, which can only compact harder — and its + // unaddressable tail is dead weight worth compacting anyway. + BufferUtilization::zero(u32::try_from(buffer.len()).unwrap_or(u32::MAX)) + }) + .collect() +} + +/// Counts `view`'s bytes against the buffer it references; inlined views reference none. +fn measure_view(utilizations: &mut [BufferUtilization], view: &BinaryView) { + if !view.is_inlined() { + let view_ref = view.as_view(); + utilizations[view_ref.buffer_index as usize].add(view_ref.offset, view_ref.size); + } +} + pub enum CompletedBuffers { Default(Vec), Deduplicated(DeduplicatedBuffers), @@ -1405,6 +1606,186 @@ mod tests { assert_arrays_eq!(builder.finish_into_varbinview(), expected, &mut ctx); } + /// A compacting builder must measure the buffers a gather hands it: with the middle + /// dictionary value never referenced, adopting the heap whole would keep its bytes alive. + /// Duplicate codes must share one rewritten copy, and null rows must not resolve their code. + #[test] + fn test_append_views_gathered_compacts_unreferenced_values() { + const DEAD: &str = "a dead value nobody gathers, far too long to inline"; + const OTHER: &str = "another value far too long to inline"; + + let mut ctx = array_session().create_execution_ctx(); + let dictionary = + >::from_iter([Some(LONG), Some(DEAD), Some(OTHER)]); + assert_eq!(dictionary.data_buffers().len(), 1); + let buffers = dictionary + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0); + let codes: [usize; 5] = [2, 0, usize::MAX, 2, 0]; + builder.append_views_gathered( + buffers, + dictionary.views(), + &Mask::from_iter([true, true, false, true, true]), + |row| codes[row], + ); + + let actual = builder.finish_into_varbinview(); + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!( + actual.data_buffers()[0].len(), + LONG.len() + OTHER.len(), + "the rewritten buffer must hold each referenced value exactly once" + ); + + let expected = >::from_iter([ + Some(OTHER), + Some(LONG), + None, + Some(OTHER), + Some(LONG), + ]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + + /// A gather that references only a contiguous tail of the heap must adopt just that slice, + /// zero-copy. + #[test] + fn test_append_views_gathered_slices_contiguous_range() { + const TAIL: &str = "the referenced tail value, too long to inline"; + + let mut ctx = array_session().create_execution_ctx(); + let dictionary = >::from_iter([Some(LONG), Some(TAIL)]); + let heap = dictionary.data_buffers()[0].as_host().clone(); + + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0); + builder.append_views_gathered( + [heap.clone()], + dictionary.views(), + &Mask::new_true(2), + |_| 1, + ); + + let actual = builder.finish_into_varbinview(); + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!(actual.data_buffers()[0].len(), TAIL.len()); + // SAFETY: LONG.len() is in bounds of the heap. + assert_eq!(actual.data_buffers()[0].as_host().as_ptr(), unsafe { + heap.as_ptr().add(LONG.len()) + }); + + let expected = >::from_iter([Some(TAIL), Some(TAIL)]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + + /// A buffer whose referenced share clears the threshold is adopted whole, zero-copy. + #[test] + fn test_append_views_gathered_keeps_utilized_buffer() { + const SHORTER: &str = "a shorter long value"; + + let mut ctx = array_session().create_execution_ctx(); + let dictionary = + >::from_iter([Some(LONG), Some(SHORTER)]); + let heap = dictionary.data_buffers()[0].as_host().clone(); + + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 0.5); + builder.append_views_gathered( + [heap.clone()], + dictionary.views(), + &Mask::new_true(1), + |_| 0, + ); + + let actual = builder.finish_into_varbinview(); + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!(actual.data_buffers()[0].as_host().as_ptr(), heap.as_ptr()); + assert_eq!(actual.data_buffers()[0].len(), heap.len()); + + let expected = >::from_iter([Some(LONG)]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + + /// An all-null gather references nothing, so a compacting builder must not retain any of the + /// buffers — and must never resolve a code. + #[test] + fn test_append_views_gathered_all_null_drops_buffers() { + let mut ctx = array_session().create_execution_ctx(); + let dictionary = >::from_iter([Some(LONG)]); + let buffers = dictionary + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0); + builder.append_views_gathered(buffers, dictionary.views(), &Mask::new_false(3), |_| { + usize::MAX + }); + + let actual = builder.finish_into_varbinview(); + assert!( + actual.data_buffers().is_empty(), + "an all-null gather must not retain any buffers" + ); + + let expected = >::from_iter([None::<&str>, None, None]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + + /// A compacting builder must measure the buffers a scatter hands it: a value no patch + /// references must not survive, while the fill and patch views are rebased onto the + /// rewritten copy. + #[test] + fn test_append_views_scattered_compacts_unreferenced_values() { + use crate::arrays::varbinview::build_views::BinaryView; + + const DEAD: &str = "a dead value nobody patches in, far too long to inline"; + const OTHER: &str = "another value far too long to inline"; + + let mut ctx = array_session().create_execution_ctx(); + let patch_values = + >::from_iter([Some(LONG), Some(DEAD), Some(OTHER)]); + let buffers = patch_values + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0); + let views = patch_values.views(); + builder.append_views_scattered( + buffers, + 4, + BinaryView::make_view(b"fill", 0, 0), + [(1usize, views[0]), (3usize, views[2])].into_iter(), + &Mask::from_iter([true, true, false, true]), + ); + + let actual = builder.finish_into_varbinview(); + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!( + actual.data_buffers()[0].len(), + LONG.len() + OTHER.len(), + "the rewritten buffer must hold only the referenced values" + ); + + let expected = >::from_iter([ + Some("fill"), + Some(LONG), + None, + Some(OTHER), + ]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + /// A fully-inlined heap has zero utilization; a compacting builder must not retain it at all. #[test] fn test_append_buffer_with_lengths_drops_fully_inlined_heap() { From 4e814f484d4c4b30f708b490a0da7aed7a7a8388 Mon Sep 17 00:00:00 2001 From: Robert Kruszewski Date: Fri, 7 Aug 2026 15:51:21 +0100 Subject: [PATCH 3/5] Measure effective VarBinView scatter utilization Signed-off-by: Robert Kruszewski --- encodings/sparse/src/canonical.rs | 60 +++++++++++++- vortex-array/src/builders/varbinview.rs | 101 +++++++++++++++++++++--- 2 files changed, 148 insertions(+), 13 deletions(-) diff --git a/encodings/sparse/src/canonical.rs b/encodings/sparse/src/canonical.rs index c0779a2781b..a63f2179e97 100644 --- a/encodings/sparse/src/canonical.rs +++ b/encodings/sparse/src/canonical.rs @@ -157,6 +157,12 @@ pub(super) fn append_sparse_to_varbinview( }; let views = parts.values.views(); + // Null patch views may still reference their source buffers. Empty them while walking the + // patches so scatter compaction does not retain bytes that no output value can observe. + let patch_validity = parts + .values + .validity()? + .execute_mask(parts.values.len(), ctx)?; match_each_integer_ptype!(parts.indices.ptype(), |I| { let indices = parts.indices.as_slice::(); builder.append_views_scattered( @@ -166,7 +172,17 @@ pub(super) fn append_sparse_to_varbinview( indices .iter() .zip(views.iter()) - .map(|(row, view)| (AsPrimitive::::as_(*row), *view)), + .zip(patch_validity.iter()) + .map(|((row, view), is_valid)| { + ( + AsPrimitive::::as_(*row), + if is_valid { + *view + } else { + BinaryView::empty_view() + }, + ) + }), &validity, ); }); @@ -1994,6 +2010,48 @@ mod test { Ok(()) } + /// Null patch views can reference real source bytes; the direct scatter empties those views + /// so compaction retains only bytes belonging to observable output values. + #[test] + fn test_sparse_append_to_compacting_view_builder_skips_null_patch_bytes() -> VortexResult<()> { + const NULL_BYTES: &str = "a null patch value that is far too long to inline"; + const VALID_BYTES: &str = "a valid patch value that is far too long to inline"; + + let mut ctx = SESSION.create_execution_ctx(); + let source = VarBinViewArray::from_iter_str([NULL_BYTES, VALID_BYTES]); + let patch_values = VarBinViewArray::try_new( + source.views().iter().copied().collect(), + Arc::from( + source + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(), + ), + DType::Utf8(Nullable), + Validity::from_mask(Mask::from_iter([false, true]), Nullable), + )?; + let array = Sparse::try_new( + buffer![0u16, 1].into_array(), + patch_values.into_array(), + 2, + Scalar::from("an unused fill value far too long to inline".to_owned()).into_nullable(), + )? + .into_array(); + + let mut builder = + VarBinViewBuilder::with_compaction(array.dtype().clone(), array.len(), 1.0); + array.append_to_builder(&mut builder, &mut ctx)?; + let actual = builder.finish_into_varbinview(); + + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!(actual.data_buffers()[0].len(), VALID_BYTES.len()); + let expected = + >::from_iter([None::<&str>, Some(VALID_BYTES)]); + assert_arrays_eq!(actual, expected, &mut ctx); + Ok(()) + } + /// A fill value short enough to inline lives in its views; the canonical decode must not /// retain a data buffer nothing references. #[test] diff --git a/vortex-array/src/builders/varbinview.rs b/vortex-array/src/builders/varbinview.rs index 076ec76efe0..d591b5a8406 100644 --- a/vortex-array/src/builders/varbinview.rs +++ b/vortex-array/src/builders/varbinview.rs @@ -20,6 +20,7 @@ use vortex_mask::AllOr; use vortex_mask::Mask; use vortex_utils::aliases::hash_map::Entry; use vortex_utils::aliases::hash_map::HashMap; +use vortex_utils::aliases::hash_set::HashSet; use crate::ArrayRef; use crate::ExecutionCtx; @@ -486,11 +487,9 @@ impl VarBinViewBuilder { debug_assert_eq!(self.nulls.len(), self.views_builder.len()); } - /// The compacting arm of [`append_views_scattered`](Self::append_views_scattered): measures - /// each buffer's utilization from the fill view and the patch views, then adopts the buffers - /// through [`push_buffers_compacted`](Self::push_buffers_compacted). Validity is appended as - /// given — an invalid row keeps its view — so every scattered view counts as referenced, - /// which keeps a rewrite from stranding an invalid row's view. + /// The compacting arm of [`append_views_scattered`](Self::append_views_scattered): resolves + /// repeated patches with last-write-wins semantics, then measures each buffer from only the + /// distinct views in the final scatter. The fill is measured only if an unpatched row remains. fn append_views_scattered_compacted( &mut self, buffers: Vec, @@ -500,20 +499,54 @@ impl VarBinViewBuilder { validity: &Mask, ) { let patches = patches.collect::>(); + let mut patched_rows = HashSet::new(); + let mut effective_patches = Vec::with_capacity(patches.len()); + for (row, view) in patches.into_iter().rev() { + assert!( + row < len, + "Patch row {row} is out of bounds for length {len}" + ); + if patched_rows.insert(row) { + effective_patches.push((row, view)); + } + } + effective_patches.reverse(); + + let fill_used = patched_rows.len() < len; + let mut referenced = HashSet::new(); + if fill_used { + referenced.insert(fill); + } let mut utilizations = unmeasured_utilizations(&buffers); - measure_view(&mut utilizations, &fill); - for (_, view) in &patches { - measure_view(&mut utilizations, view); + if fill_used { + measure_view(&mut utilizations, &fill); + } + for (_, view) in &effective_patches { + if referenced.insert(*view) { + measure_view(&mut utilizations, view); + } } let slots = self.push_buffers_compacted(buffers, &utilizations); - let fill = self.remap_view_compacted(fill, &slots); + let mut remapped = HashMap::new(); + if fill_used { + remapped.insert(fill, self.remap_view_compacted(fill, &slots)); + } + for (_, view) in &effective_patches { + if let Entry::Vacant(entry) = remapped.entry(*view) { + entry.insert(self.remap_view_compacted(*view, &slots)); + } + } + let fill = if fill_used { + remapped[&fill] + } else { + BinaryView::empty_view() + }; let start = self.views_builder.len(); self.views_builder.push_n(fill, len); - for (row, view) in patches { - let view = self.remap_view_compacted(view, &slots); - self.views_builder[start + row] = view; + for (row, view) in effective_patches { + self.views_builder[start + row] = remapped[&view]; } self.nulls.append_validity_mask(validity); @@ -1786,6 +1819,50 @@ mod tests { assert_arrays_eq!(actual, expected, &mut ctx); } + /// Compaction measures the last patch at each row and omits the fill when patches cover the + /// whole output, so overwritten patch bytes and unused fill bytes are both dropped. + #[test] + fn test_append_views_scattered_compaction_measures_final_scatter() { + use crate::arrays::varbinview::build_views::BinaryView; + + const DEAD: &str = "an overwritten patch value far too long to inline"; + const OTHER: &str = "another value far too long to inline"; + + let mut ctx = array_session().create_execution_ctx(); + let patch_values = + >::from_iter([Some(DEAD), Some(LONG), Some(OTHER)]); + let mut buffers = patch_values + .data_buffers() + .iter() + .map(|buffer| buffer.as_host().clone()) + .collect::>(); + let fill_bytes = ByteBuffer::copy_from("an unused fill value far too long to inline"); + let fill = BinaryView::make_view( + fill_bytes.as_slice(), + u32::try_from(buffers.len()).unwrap(), + 0, + ); + buffers.push(fill_bytes); + + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 2, 1.0); + let views = patch_values.views(); + builder.append_views_scattered( + buffers, + 2, + fill, + [(0usize, views[0]), (0, views[1]), (1, views[2])].into_iter(), + &Mask::new_true(2), + ); + + let actual = builder.finish_into_varbinview(); + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!(actual.data_buffers()[0].len(), LONG.len() + OTHER.len()); + + let expected = >::from_iter([Some(LONG), Some(OTHER)]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + /// A fully-inlined heap has zero utilization; a compacting builder must not retain it at all. #[test] fn test_append_buffer_with_lengths_drops_fully_inlined_heap() { From 378b7a038ec61b9328a6339ea5f12bbb2293cb78 Mon Sep 17 00:00:00 2001 From: Robert Kruszewski Date: Fri, 7 Aug 2026 16:12:56 +0100 Subject: [PATCH 4/5] Fix VarBinView rustdoc links Signed-off-by: Robert Kruszewski --- vortex-array/src/builders/varbinview.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/vortex-array/src/builders/varbinview.rs b/vortex-array/src/builders/varbinview.rs index d591b5a8406..8e6df51ddbe 100644 --- a/vortex-array/src/builders/varbinview.rs +++ b/vortex-array/src/builders/varbinview.rs @@ -331,7 +331,7 @@ impl VarBinViewBuilder { /// index `view_at` returns for it. /// /// `buffers` — the data buffers the views reference, in the numbering the views use — are - /// adopted via [`push_buffers`](Self::push_buffers), and every gathered view is rebased onto + /// adopted through the builder's buffer storage, and every gathered view is rebased onto /// the indices they land at as it is written, so the whole append is one view per row with no /// byte copy and no intermediate array. `view_at` is only called for valid rows; null rows get /// an empty view. @@ -442,7 +442,7 @@ impl VarBinViewBuilder { /// Appends `len` rows that are `fill` everywhere except at the given patch rows. /// /// `buffers` — the data buffers `fill` and the patch views reference, in the numbering they - /// use — are adopted via [`push_buffers`](Self::push_buffers) and the views are rebased onto + /// use — are adopted through the builder's buffer storage and the views are rebased onto /// the indices they land at. `patches` yields `(row, view)` pairs with rows below `len`; the /// fill rows cost one bulk view fill and each patch one view write, with no byte copy. /// Validity is appended exactly as given — invalid rows keep whichever view they got. From f442d48880a22f3dec2fe11a7b3ab1681d309fa1 Mon Sep 17 00:00:00 2001 From: Robert Kruszewski Date: Fri, 7 Aug 2026 16:19:08 +0100 Subject: [PATCH 5/5] no Signed-off-by: Robert Kruszewski --- encodings/sparse/src/canonical.rs | 60 +------- vortex-array/src/builders/varbinview.rs | 186 ++---------------------- 2 files changed, 10 insertions(+), 236 deletions(-) diff --git a/encodings/sparse/src/canonical.rs b/encodings/sparse/src/canonical.rs index a63f2179e97..c0779a2781b 100644 --- a/encodings/sparse/src/canonical.rs +++ b/encodings/sparse/src/canonical.rs @@ -157,12 +157,6 @@ pub(super) fn append_sparse_to_varbinview( }; let views = parts.values.views(); - // Null patch views may still reference their source buffers. Empty them while walking the - // patches so scatter compaction does not retain bytes that no output value can observe. - let patch_validity = parts - .values - .validity()? - .execute_mask(parts.values.len(), ctx)?; match_each_integer_ptype!(parts.indices.ptype(), |I| { let indices = parts.indices.as_slice::(); builder.append_views_scattered( @@ -172,17 +166,7 @@ pub(super) fn append_sparse_to_varbinview( indices .iter() .zip(views.iter()) - .zip(patch_validity.iter()) - .map(|((row, view), is_valid)| { - ( - AsPrimitive::::as_(*row), - if is_valid { - *view - } else { - BinaryView::empty_view() - }, - ) - }), + .map(|(row, view)| (AsPrimitive::::as_(*row), *view)), &validity, ); }); @@ -2010,48 +1994,6 @@ mod test { Ok(()) } - /// Null patch views can reference real source bytes; the direct scatter empties those views - /// so compaction retains only bytes belonging to observable output values. - #[test] - fn test_sparse_append_to_compacting_view_builder_skips_null_patch_bytes() -> VortexResult<()> { - const NULL_BYTES: &str = "a null patch value that is far too long to inline"; - const VALID_BYTES: &str = "a valid patch value that is far too long to inline"; - - let mut ctx = SESSION.create_execution_ctx(); - let source = VarBinViewArray::from_iter_str([NULL_BYTES, VALID_BYTES]); - let patch_values = VarBinViewArray::try_new( - source.views().iter().copied().collect(), - Arc::from( - source - .data_buffers() - .iter() - .map(|buffer| buffer.as_host().clone()) - .collect::>(), - ), - DType::Utf8(Nullable), - Validity::from_mask(Mask::from_iter([false, true]), Nullable), - )?; - let array = Sparse::try_new( - buffer![0u16, 1].into_array(), - patch_values.into_array(), - 2, - Scalar::from("an unused fill value far too long to inline".to_owned()).into_nullable(), - )? - .into_array(); - - let mut builder = - VarBinViewBuilder::with_compaction(array.dtype().clone(), array.len(), 1.0); - array.append_to_builder(&mut builder, &mut ctx)?; - let actual = builder.finish_into_varbinview(); - - assert_eq!(actual.data_buffers().len(), 1); - assert_eq!(actual.data_buffers()[0].len(), VALID_BYTES.len()); - let expected = - >::from_iter([None::<&str>, Some(VALID_BYTES)]); - assert_arrays_eq!(actual, expected, &mut ctx); - Ok(()) - } - /// A fill value short enough to inline lives in its views; the canonical decode must not /// retain a data buffer nothing references. #[test] diff --git a/vortex-array/src/builders/varbinview.rs b/vortex-array/src/builders/varbinview.rs index 8e6df51ddbe..aafb2f7a6f3 100644 --- a/vortex-array/src/builders/varbinview.rs +++ b/vortex-array/src/builders/varbinview.rs @@ -20,7 +20,6 @@ use vortex_mask::AllOr; use vortex_mask::Mask; use vortex_utils::aliases::hash_map::Entry; use vortex_utils::aliases::hash_map::HashMap; -use vortex_utils::aliases::hash_set::HashSet; use crate::ArrayRef; use crate::ExecutionCtx; @@ -194,9 +193,9 @@ impl VarBinViewBuilder { self.completed.len() } - /// Whether this builder compacts the data buffers it is handed. The lengths-driven appends - /// and the gather/scatter appends use this to gate their utilization measurement; the - /// buffer-adopting escape hatches ([`append_views_built_at`](Self::append_views_built_at) and + /// Whether this builder compacts the data buffers it is handed. The lengths-driven and + /// gather appends use this to gate their utilization measurement; the buffer-adopting escape + /// hatches ([`append_views_built_at`](Self::append_views_built_at) and /// [`push_buffers`](Self::push_buffers)) always bypass it. fn compacts_buffers(&self) -> bool { self.compaction_threshold > 0.0 @@ -262,9 +261,9 @@ impl VarBinViewBuilder { /// /// Buffers are taken as they are, without utilization measurement; like /// [`append_views_built_at`](Self::append_views_built_at), this bypasses any compaction the - /// builder was configured for. A caller whose views may leave a buffer mostly unreferenced - /// must go through [`push_buffers_compacted`](Self::push_buffers_compacted) on a compacting - /// builder instead. + /// builder was configured for. Scattered appends deliberately have these semantics; + /// compacting paths that measure utilization use + /// [`push_buffers_compacted`](Self::push_buffers_compacted) instead. fn push_buffers(&mut self, buffers: impl IntoIterator) -> Vec { self.flush_in_progress(); buffers @@ -447,9 +446,9 @@ impl VarBinViewBuilder { /// fill rows cost one bulk view fill and each patch one view write, with no byte copy. /// Validity is appended exactly as given — invalid rows keep whichever view they got. /// - /// When the builder is configured to compact buffers, utilization is measured from the fill - /// view and the patch views and each buffer is adopted, sliced, or rewritten accordingly, so - /// callers never need a canonicalize-and-compact fallback. + /// This append deliberately adopts the supplied buffers as-is, even when the builder is + /// configured for compaction. Callers should use it only when retaining those buffers is + /// acceptable. /// /// # Panics /// @@ -464,16 +463,6 @@ impl VarBinViewBuilder { validity: &Mask, ) { assert_eq!(validity.len(), len, "Must have one validity entry per row"); - if self.compacts_buffers() { - return self.append_views_scattered_compacted( - buffers.into_iter().collect(), - len, - fill, - patches, - validity, - ); - } - let mapping = self.push_buffers(buffers); let start = self.views_builder.len(); @@ -487,72 +476,6 @@ impl VarBinViewBuilder { debug_assert_eq!(self.nulls.len(), self.views_builder.len()); } - /// The compacting arm of [`append_views_scattered`](Self::append_views_scattered): resolves - /// repeated patches with last-write-wins semantics, then measures each buffer from only the - /// distinct views in the final scatter. The fill is measured only if an unpatched row remains. - fn append_views_scattered_compacted( - &mut self, - buffers: Vec, - len: usize, - fill: BinaryView, - patches: impl Iterator, - validity: &Mask, - ) { - let patches = patches.collect::>(); - let mut patched_rows = HashSet::new(); - let mut effective_patches = Vec::with_capacity(patches.len()); - for (row, view) in patches.into_iter().rev() { - assert!( - row < len, - "Patch row {row} is out of bounds for length {len}" - ); - if patched_rows.insert(row) { - effective_patches.push((row, view)); - } - } - effective_patches.reverse(); - - let fill_used = patched_rows.len() < len; - let mut referenced = HashSet::new(); - if fill_used { - referenced.insert(fill); - } - - let mut utilizations = unmeasured_utilizations(&buffers); - if fill_used { - measure_view(&mut utilizations, &fill); - } - for (_, view) in &effective_patches { - if referenced.insert(*view) { - measure_view(&mut utilizations, view); - } - } - let slots = self.push_buffers_compacted(buffers, &utilizations); - - let mut remapped = HashMap::new(); - if fill_used { - remapped.insert(fill, self.remap_view_compacted(fill, &slots)); - } - for (_, view) in &effective_patches { - if let Entry::Vacant(entry) = remapped.entry(*view) { - entry.insert(self.remap_view_compacted(*view, &slots)); - } - } - let fill = if fill_used { - remapped[&fill] - } else { - BinaryView::empty_view() - }; - let start = self.views_builder.len(); - self.views_builder.push_n(fill, len); - for (row, view) in effective_patches { - self.views_builder[start + row] = remapped[&view]; - } - - self.nulls.append_validity_mask(validity); - debug_assert_eq!(self.nulls.len(), self.views_builder.len()); - } - /// Appends values laid end-to-end in `bytes`, one per entry of `lengths`. /// /// The builder adopts `bytes` as a data buffer without copying it (splitting it only past the @@ -1772,97 +1695,6 @@ mod tests { assert_arrays_eq!(actual, expected, &mut ctx); } - /// A compacting builder must measure the buffers a scatter hands it: a value no patch - /// references must not survive, while the fill and patch views are rebased onto the - /// rewritten copy. - #[test] - fn test_append_views_scattered_compacts_unreferenced_values() { - use crate::arrays::varbinview::build_views::BinaryView; - - const DEAD: &str = "a dead value nobody patches in, far too long to inline"; - const OTHER: &str = "another value far too long to inline"; - - let mut ctx = array_session().create_execution_ctx(); - let patch_values = - >::from_iter([Some(LONG), Some(DEAD), Some(OTHER)]); - let buffers = patch_values - .data_buffers() - .iter() - .map(|buffer| buffer.as_host().clone()) - .collect::>(); - - let mut builder = - VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 8, 1.0); - let views = patch_values.views(); - builder.append_views_scattered( - buffers, - 4, - BinaryView::make_view(b"fill", 0, 0), - [(1usize, views[0]), (3usize, views[2])].into_iter(), - &Mask::from_iter([true, true, false, true]), - ); - - let actual = builder.finish_into_varbinview(); - assert_eq!(actual.data_buffers().len(), 1); - assert_eq!( - actual.data_buffers()[0].len(), - LONG.len() + OTHER.len(), - "the rewritten buffer must hold only the referenced values" - ); - - let expected = >::from_iter([ - Some("fill"), - Some(LONG), - None, - Some(OTHER), - ]); - assert_arrays_eq!(actual, expected, &mut ctx); - } - - /// Compaction measures the last patch at each row and omits the fill when patches cover the - /// whole output, so overwritten patch bytes and unused fill bytes are both dropped. - #[test] - fn test_append_views_scattered_compaction_measures_final_scatter() { - use crate::arrays::varbinview::build_views::BinaryView; - - const DEAD: &str = "an overwritten patch value far too long to inline"; - const OTHER: &str = "another value far too long to inline"; - - let mut ctx = array_session().create_execution_ctx(); - let patch_values = - >::from_iter([Some(DEAD), Some(LONG), Some(OTHER)]); - let mut buffers = patch_values - .data_buffers() - .iter() - .map(|buffer| buffer.as_host().clone()) - .collect::>(); - let fill_bytes = ByteBuffer::copy_from("an unused fill value far too long to inline"); - let fill = BinaryView::make_view( - fill_bytes.as_slice(), - u32::try_from(buffers.len()).unwrap(), - 0, - ); - buffers.push(fill_bytes); - - let mut builder = - VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 2, 1.0); - let views = patch_values.views(); - builder.append_views_scattered( - buffers, - 2, - fill, - [(0usize, views[0]), (0, views[1]), (1, views[2])].into_iter(), - &Mask::new_true(2), - ); - - let actual = builder.finish_into_varbinview(); - assert_eq!(actual.data_buffers().len(), 1); - assert_eq!(actual.data_buffers()[0].len(), LONG.len() + OTHER.len()); - - let expected = >::from_iter([Some(LONG), Some(OTHER)]); - assert_arrays_eq!(actual, expected, &mut ctx); - } - /// A fully-inlined heap has zero utilization; a compacting builder must not retain it at all. #[test] fn test_append_buffer_with_lengths_drops_fully_inlined_heap() {