diff --git a/encodings/fsst/src/array.rs b/encodings/fsst/src/array.rs index 322030789b9..ef25c559b99 100644 --- a/encodings/fsst/src/array.rs +++ b/encodings/fsst/src/array.rs @@ -63,7 +63,7 @@ use vortex_session::registry::CachedId; use crate::canonical::FSST_DECODE_SLACK; use crate::canonical::FsstDecodePlan; use crate::canonical::canonicalize_fsst; -use crate::canonical::fsst_decode_views; +use crate::canonical::fsst_decode_bytes; use crate::rules::RULES; /// A [`FSST`]-encoded Vortex array. @@ -337,20 +337,20 @@ impl VTable for FSST { vortex_bail!("append_to_builder for FSST requires a variable-binary builder") }; - // Decompress the whole block of data into a new buffer, and create some views - // from it instead. The new buffer lands after any pending in-progress - // buffer that push_buffer_and_adjusted_views will flush first. - let next_buffer_index = builder.completed_block_count() + u32::from(builder.in_progress()); - let (buffers, views) = fsst_decode_views(array, next_buffer_index, ctx)?; - - builder.push_buffer_and_adjusted_views( - &buffers, - &views, - array - .array() - .validity()? - .execute_mask(array.array().len(), ctx)?, - ); + // Decompress the whole block of data into a new buffer, which the builder adopts as a + // data buffer with views built over it in place. + let validity = array + .array() + .validity()? + .execute_mask(array.array().len(), ctx)?; + let (uncompressed_bytes, uncompressed_lens) = fsst_decode_bytes(array, ctx)?; + match_each_integer_ptype!(uncompressed_lens.ptype(), |P| { + builder.append_buffer_with_lengths( + uncompressed_bytes.freeze(), + uncompressed_lens.as_slice::

(), + &validity, + ) + }); Ok(()) } diff --git a/encodings/fsst/src/canonical.rs b/encodings/fsst/src/canonical.rs index a0f97b37d99..28942648aa4 100644 --- a/encodings/fsst/src/canonical.rs +++ b/encodings/fsst/src/canonical.rs @@ -12,11 +12,9 @@ use vortex_array::IntoArray; use vortex_array::arrays::PrimitiveArray; use vortex_array::arrays::VarBinViewArray; use vortex_array::arrays::varbin::VarBinArrayExt; -use vortex_array::arrays::varbinview::build_views::BinaryView; use vortex_array::arrays::varbinview::build_views::MAX_BUFFER_LEN; use vortex_array::arrays::varbinview::build_views::build_views; use vortex_array::match_each_integer_ptype; -use vortex_buffer::Buffer; use vortex_buffer::ByteBuffer; use vortex_buffer::ByteBufferMut; use vortex_error::VortexResult; @@ -30,7 +28,15 @@ pub(super) fn canonicalize_fsst( array: ArrayView<'_, FSST>, ctx: &mut ExecutionCtx, ) -> VortexResult { - let (buffers, views) = fsst_decode_views(array, 0, ctx)?; + let (uncompressed_bytes, uncompressed_lens) = fsst_decode_bytes(array, ctx)?; + let (buffers, views) = match_each_integer_ptype!(uncompressed_lens.ptype(), |P| { + build_views( + 0, + MAX_BUFFER_LEN, + uncompressed_bytes.freeze(), + uncompressed_lens.as_slice::

(), + ) + }); // SAFETY: FSST already validates the bytes for binary/UTF-8. We build views directly on // top of them, so the view pointers will all be valid. Ok(unsafe { @@ -122,22 +128,6 @@ pub(crate) fn fsst_decode_bytes( Ok((uncompressed_bytes, plan.lengths)) } -pub(crate) fn fsst_decode_views( - fsst_array: ArrayView<'_, FSST>, - start_buf_index: u32, - ctx: &mut ExecutionCtx, -) -> VortexResult<(Vec, Buffer)> { - let (uncompressed_bytes, uncompressed_lens_array) = fsst_decode_bytes(fsst_array, ctx)?; - match_each_integer_ptype!(uncompressed_lens_array.ptype(), |P| { - Ok(build_views( - start_buf_index, - MAX_BUFFER_LEN, - uncompressed_bytes, - uncompressed_lens_array.as_slice::

(), - )) - }) -} - #[cfg(test)] mod tests { use std::sync::LazyLock; diff --git a/encodings/onpair/src/array.rs b/encodings/onpair/src/array.rs index 58c38344532..e57bed91827 100644 --- a/encodings/onpair/src/array.rs +++ b/encodings/onpair/src/array.rs @@ -54,7 +54,7 @@ use vortex_session::registry::CachedId; use crate::canonical::OnPairDecodePlan; use crate::canonical::canonicalize_onpair; -use crate::canonical::onpair_decode_views; +use crate::canonical::onpair_decode_bytes; use crate::decode::collect_widened; use crate::rules::RULES; @@ -605,16 +605,20 @@ impl VTable for OnPair { vortex_bail!("append_to_builder for OnPair requires a variable-binary builder") }; - let next_buffer_index = builder.completed_block_count() + u32::from(builder.in_progress()); - let (buffers, views) = onpair_decode_views(array, next_buffer_index, ctx)?; - builder.push_buffer_and_adjusted_views( - &buffers, - &views, - array - .array() - .validity()? - .execute_mask(array.array().len(), ctx)?, - ); + // Decode the whole code stream into a new buffer, which the builder adopts as a data + // buffer with views built over it in place. + let validity = array + .array() + .validity()? + .execute_mask(array.array().len(), ctx)?; + let (out_bytes, lengths) = onpair_decode_bytes(array, ctx)?; + match_each_integer_ptype!(lengths.ptype(), |P| { + builder.append_buffer_with_lengths( + out_bytes.freeze(), + lengths.as_slice::

(), + &validity, + ) + }); Ok(()) } diff --git a/encodings/onpair/src/canonical.rs b/encodings/onpair/src/canonical.rs index 50bdd70eb6c..7ed5bfe6aa6 100644 --- a/encodings/onpair/src/canonical.rs +++ b/encodings/onpair/src/canonical.rs @@ -17,12 +17,10 @@ use vortex_array::ExecutionCtx; use vortex_array::IntoArray; use vortex_array::arrays::PrimitiveArray; use vortex_array::arrays::VarBinViewArray; -use vortex_array::arrays::varbinview::build_views::BinaryView; use vortex_array::arrays::varbinview::build_views::MAX_BUFFER_LEN; use vortex_array::arrays::varbinview::build_views::build_views; use vortex_array::match_each_integer_ptype; use vortex_buffer::Buffer; -use vortex_buffer::ByteBuffer; use vortex_buffer::ByteBufferMut; use vortex_error::VortexResult; use vortex_error::vortex_bail; @@ -38,7 +36,15 @@ pub(super) fn canonicalize_onpair( array: ArrayView<'_, OnPair>, ctx: &mut ExecutionCtx, ) -> VortexResult { - let (buffers, views) = onpair_decode_views(array, 0, ctx)?; + let (out_bytes, lengths) = onpair_decode_bytes(array, ctx)?; + let (buffers, views) = match_each_integer_ptype!(lengths.ptype(), |P| { + build_views( + 0, + MAX_BUFFER_LEN, + out_bytes.freeze(), + lengths.as_slice::

(), + ) + }); let validity = array.array().validity()?; Ok(unsafe { VarBinViewArray::new_unchecked(views, Arc::from(buffers), array.dtype().clone(), validity) @@ -144,19 +150,3 @@ pub(crate) fn onpair_decode_bytes( unsafe { out_bytes.set_len(written) }; Ok((out_bytes, plan.lengths)) } - -pub(crate) fn onpair_decode_views( - array: ArrayView<'_, OnPair>, - start_buf_index: u32, - ctx: &mut ExecutionCtx, -) -> VortexResult<(Vec, Buffer)> { - let (out_bytes, lengths) = onpair_decode_bytes(array, ctx)?; - match_each_integer_ptype!(lengths.ptype(), |P| { - Ok(build_views( - start_buf_index, - MAX_BUFFER_LEN, - out_bytes, - lengths.as_slice::

(), - )) - }) -} diff --git a/encodings/zstd/src/array.rs b/encodings/zstd/src/array.rs index 0aad8c8c4e0..0d831e58775 100644 --- a/encodings/zstd/src/array.rs +++ b/encodings/zstd/src/array.rs @@ -373,45 +373,44 @@ fn append_to_varbinview( return Ok(()); } - // Build the views against the index the pushed buffers will land at, so the builder does not - // have to rebase them afterwards. - let next_buffer_index = builder.completed_block_count() + u32::from(builder.in_progress()); // The decompressed frames cover whole frames and so can extend past the requested values on // either side. Reconstructing over just the requested region keeps the pushed buffers fully - // utilized, which is what `push_buffer_and_adjusted_views` requires: it hands them to the - // finished array as they are, without compacting them. + // utilized, which is what `append_views_built_at` requires: it hands them to the finished + // array as they are, without compacting them. let value_bytes = slice.bytes.slice(slice.value_byte_range()?); - let (buffers, valid_views) = - try_reconstruct_views(&value_bytes, next_buffer_index, MAX_BUFFER_LEN)?; - vortex_ensure!( - valid_views.len() == mask.true_count(), - "Corrupt zstd metadata: the decompressed frames hold {} values for the {} valid rows of \ - the slice", - valid_views.len(), - mask.true_count() - ); - - let views = match mask.bit_buffer() { - AllOr::All => valid_views, - AllOr::None => unreachable!("handled above"), - AllOr::Some(bits) => { - // Null rows carry an empty view, so scatter the stored values into their rows. Walking - // the set bits a word at a time avoids materializing the mask's indices, which the - // views are the only consumer of. - let mut views = BufferMut::::zeroed(slice.n_rows); - let mut valid_row = 0; - bits.for_each_set_index(|index| { - // In bounds: `valid_views.len() == mask.true_count()` was checked above, and - // `index < slice.n_rows` because `bits` is the mask over those rows. - views[index] = valid_views[valid_row]; - valid_row += 1; - }); - views.freeze() - } - }; + // The values only reveal themselves while walking the length-prefixed frames, so the views + // are built inside the builder's numbering callback rather than from a lengths slice. + builder.append_views_built_at(&mask, |next_buffer_index| { + let (buffers, valid_views) = + try_reconstruct_views(&value_bytes, next_buffer_index, MAX_BUFFER_LEN)?; + vortex_ensure!( + valid_views.len() == mask.true_count(), + "Corrupt zstd metadata: the decompressed frames hold {} values for the {} valid rows \ + of the slice", + valid_views.len(), + mask.true_count() + ); - builder.push_buffer_and_adjusted_views(&buffers, &views, mask); - Ok(()) + let views = match mask.bit_buffer() { + AllOr::All => valid_views, + AllOr::None => unreachable!("handled above"), + AllOr::Some(bits) => { + // Null rows carry an empty view, so scatter the stored values into their rows. + // Walking the set bits a word at a time avoids materializing the mask's indices, + // which the views are the only consumer of. + let mut views = BufferMut::::zeroed(slice.n_rows); + let mut valid_row = 0; + bits.for_each_set_index(|index| { + // In bounds: `valid_views.len() == mask.true_count()` was checked above, and + // `index < slice.n_rows` because `bits` is the mask over those rows. + views[index] = valid_views[valid_row]; + valid_row += 1; + }); + views.freeze() + } + }; + Ok((buffers, views)) + }) } #[derive(Clone, Debug)] diff --git a/encodings/zstd/src/test.rs b/encodings/zstd/src/test.rs index e600cff020f..f65e5f6fe98 100644 --- a/encodings/zstd/src/test.rs +++ b/encodings/zstd/src/test.rs @@ -246,8 +246,8 @@ fn test_zstd_append_to_offset_builder() { } /// A slice decompresses whole frames, so the frames hold values on either side of the ones it -/// requests. `push_buffer_and_adjusted_views` publishes the buffers it is handed as they are, so -/// only the requested region may reach it — otherwise the finished array retains the whole frames. +/// requests. `append_views_built_at` publishes the buffers it is handed as they are, so only the +/// requested region may reach it — otherwise the finished array retains the whole frames. #[test] fn test_zstd_append_to_view_builder_keeps_only_the_sliced_bytes() -> VortexResult<()> { let mut ctx = array_session().create_execution_ctx(); diff --git a/vortex-array/src/arrays/varbin/vtable/canonical.rs b/vortex-array/src/arrays/varbin/vtable/canonical.rs index 6c195fe16ce..971d4397381 100644 --- a/vortex-array/src/arrays/varbin/vtable/canonical.rs +++ b/vortex-array/src/arrays/varbin/vtable/canonical.rs @@ -3,9 +3,6 @@ use std::sync::Arc; -use num_traits::AsPrimitive; -use vortex_buffer::Buffer; -use vortex_buffer::ByteBuffer; use vortex_error::VortexResult; use crate::ExecutionCtx; @@ -13,23 +10,30 @@ use crate::array::ArrayView; use crate::arrays::PrimitiveArray; use crate::arrays::VarBin; use crate::arrays::VarBinViewArray; -use crate::arrays::varbinview::BinaryView; use crate::arrays::varbinview::build_views::MAX_BUFFER_LEN; -use crate::arrays::varbinview::build_views::build_views; -use crate::arrays::varbinview::build_views::offsets_to_lengths; -use crate::buffer::BufferHandle; +use crate::arrays::varbinview::build_views::build_views_from_offsets; use crate::match_each_integer_ptype; /// Converts a VarBinArray to its canonical form (VarBinViewArray). /// /// This is a shared helper used by both `canonicalize` and `execute`. +/// +/// The value bytes are handed over as they are — only the offsets are consumed, to derive the view +/// lengths — so this costs one view per row and no byte copy. pub(crate) fn varbin_to_canonical( array: ArrayView<'_, VarBin>, ctx: &mut ExecutionCtx, ) -> VortexResult { let parts = array.into_owned().into_data_parts(); let offsets = parts.offsets.execute::(ctx)?; - let (buffers, views) = varbin_decode_views(&offsets, parts.bytes, 0); + let (buffers, views) = match_each_integer_ptype!(offsets.ptype(), |P| { + build_views_from_offsets( + 0, + MAX_BUFFER_LEN, + parts.bytes.unwrap_host(), + offsets.as_slice::

(), + ) + }); // SAFETY: views are correctly computed from valid offsets Ok(unsafe { @@ -37,31 +41,6 @@ pub(crate) fn varbin_to_canonical( }) } -/// Lays a `VarBin` array's value bytes out as `VarBinView` buffers plus the views over them. -/// -/// `start_buf_index` is the index the first returned buffer will occupy in its destination, so the -/// views come out already referencing the right buffer and never need rebasing. Canonicalization -/// passes `0`; appending into a [`VarBinViewBuilder`](crate::builders::VarBinViewBuilder) passes the -/// index its next buffer will land at. -/// -/// The value bytes are handed over as they are — only the offsets are consumed, to derive the view -/// lengths — so this costs one view per row and no byte copy when the buffer is uniquely held. -pub(crate) fn varbin_decode_views( - offsets: &PrimitiveArray, - bytes: BufferHandle, - start_buf_index: u32, -) -> (Vec, Buffer) { - match_each_integer_ptype!(offsets.ptype(), |P| { - let offsets_slice = offsets.as_slice::

(); - let first: usize = offsets_slice[0].as_(); - let last: usize = offsets_slice[offsets_slice.len() - 1].as_(); - let bytes = bytes.unwrap_host().slice(first..last).into_mut(); - - let lens = offsets_to_lengths(offsets_slice); - build_views(start_buf_index, MAX_BUFFER_LEN, bytes, lens.as_slice()) - }) -} - #[cfg(test)] mod tests { use rstest::rstest; @@ -179,7 +158,8 @@ mod tests { /// A builder configured to compact must not be handed a raw buffer behind its back: the /// inlined values leave the pushed buffer only partly referenced, and skipping compaction - /// would keep those bytes alive. Appending through the canonical array instead drops them. + /// would keep those bytes alive. The builder measures utilization from the value lengths and + /// drops the fully-inlined heap. #[test] fn append_varbin_to_a_compacting_builder_still_compacts() -> VortexResult<()> { let mut ctx = array_session().create_execution_ctx(); diff --git a/vortex-array/src/arrays/varbin/vtable/mod.rs b/vortex-array/src/arrays/varbin/vtable/mod.rs index 1e2065e29c2..397b0f8084b 100644 --- a/vortex-array/src/arrays/varbin/vtable/mod.rs +++ b/vortex-array/src/arrays/varbin/vtable/mod.rs @@ -31,6 +31,7 @@ use crate::builders::VarBinViewBuilder; use crate::dtype::DType; use crate::dtype::Nullability; use crate::dtype::PType; +use crate::match_each_integer_ptype; use crate::match_each_varbin_builder; use crate::serde::ArrayChildren; use crate::validity::Validity; @@ -39,7 +40,6 @@ mod kernel; mod operations; mod validity; -use canonical::varbin_decode_views; use canonical::varbin_to_canonical; use vortex_session::VortexSession; @@ -221,23 +221,9 @@ impl VTable for VarBin { // The two arms here are every builder a `Utf8`/`Binary` dtype has: all four // `VarBinBuilder` widths above, and `VarBinViewBuilder` below. - let Some(view_builder) = builder.as_any().downcast_ref::() else { + let Some(builder) = builder.as_any_mut().downcast_mut::() else { vortex_bail!("append_to_builder for VarBin requires a variable-binary builder") }; - - if view_builder.compacts_buffers() { - // A compacting builder decides per buffer whether to keep, slice or rewrite it, which - // it can only do by measuring the finished views against the buffer. Go through the - // canonical array so that policy still applies. - return varbin_to_canonical(array, ctx)? - .into_array() - .append_to_builder(builder, ctx); - } - - let builder = builder - .as_any_mut() - .downcast_mut::() - .vortex_expect("builder type checked above"); append_to_varbinview(array, builder, ctx) } @@ -253,8 +239,9 @@ impl VTable for VarBin { /// Canonicalizing first would build the same views, then pay for them twice more: once to wrap /// them in a `VarBinViewArray` the builder immediately unwraps, and once for /// `append_varbinview_array` to rewrite every view so its buffer index is rebased onto the -/// builder's. Numbering the buffer up front instead makes the whole append one view per row plus -/// pushing the byte buffer. +/// builder's. Handing the heap and offsets to the builder instead makes the whole append one view +/// per row with no byte copy — the builder adopts the referenced range of the heap as it is. That +/// range is fully covered by the new views, so this stays valid for a compacting builder too. fn append_to_varbinview( array: ArrayView<'_, VarBin>, builder: &mut VarBinViewBuilder, @@ -263,15 +250,15 @@ fn append_to_varbinview( let len = array.as_ref().len(); let validity = array.varbin_validity().execute_mask(len, ctx)?; - // Build the views against the index the pushed buffer will land at, so the builder does not - // have to rebase them afterwards. - let next_buffer_index = builder.completed_block_count() + u32::from(builder.in_progress()); - let parts = array.into_owned().into_data_parts(); let offsets = parts.offsets.execute::(ctx)?; - let (buffers, views) = varbin_decode_views(&offsets, parts.bytes, next_buffer_index); - - builder.push_buffer_and_adjusted_views(&buffers, &views, validity); + match_each_integer_ptype!(offsets.ptype(), |P| { + builder.append_buffer_with_offsets( + parts.bytes.unwrap_host(), + offsets.as_slice::

(), + &validity, + ) + }); Ok(()) } diff --git a/vortex-array/src/arrays/varbinview/build_views.rs b/vortex-array/src/arrays/varbinview/build_views.rs index 69f1351b781..8a6a5a0530d 100644 --- a/vortex-array/src/arrays/varbinview/build_views.rs +++ b/vortex-array/src/arrays/varbinview/build_views.rs @@ -6,7 +6,6 @@ use num_traits::AsPrimitive; use vortex_buffer::Buffer; use vortex_buffer::BufferMut; use vortex_buffer::ByteBuffer; -use vortex_buffer::ByteBufferMut; pub use crate::arrays::varbinview::BinaryView; use crate::dtype::NativePType; @@ -26,25 +25,107 @@ pub const MAX_BUFFER_LEN: usize = i32::MAX as usize; /// Split a large buffer of input `bytes` holding string data into `VarBinView` buffers and views. /// +/// The values must be laid end-to-end in `bytes`, one per entry of `lens`, describing the whole +/// buffer exactly. The returned buffers are zero-copy slices of `bytes`, numbered sequentially +/// from `start_buf_index`. +/// /// `max_buffer_len` must not exceed [`MAX_BUFFER_LEN`], since every view offset is stored in a /// `u32` and offsets are bounded by `max_buffer_len`. +/// +/// # Panics +/// +/// Panics if the lengths do not describe `bytes` exactly, or if a single value exceeds +/// `max_buffer_len`. pub fn build_views>( start_buf_index: u32, max_buffer_len: usize, - bytes: ByteBufferMut, + bytes: ByteBuffer, lens: &[P], ) -> (Vec, Buffer) { + let mut views = BufferMut::with_capacity(lens.len()); + let buffers = extend_views( + &mut views, + start_buf_index, + max_buffer_len, + &bytes, + lens.len(), + |i| lens[i].as_(), + ); + (buffers, views.freeze()) +} + +/// [`build_views`] for values described by an offsets buffer instead of lengths. +/// +/// `offsets` are absolute positions into `bytes` — the layout a `VarBinArray` stores — so there +/// is one more offset than there are values, and the values need not start at the beginning of +/// `bytes`: only `offsets[0]..offsets[last]` is referenced, and the returned buffers are zero-copy +/// slices of that range. +/// +/// # Panics +/// +/// Panics if `offsets` is empty, not monotonically non-decreasing within `bytes`, or if a single +/// value exceeds `max_buffer_len`. +pub fn build_views_from_offsets>( + start_buf_index: u32, + max_buffer_len: usize, + bytes: ByteBuffer, + offsets: &[P], +) -> (Vec, Buffer) { + assert!(!offsets.is_empty(), "offsets must hold at least one entry"); + let first: usize = offsets[0].as_(); + let last: usize = offsets[offsets.len() - 1].as_(); + let bytes = bytes.slice(first..last); + + let count = offsets.len() - 1; + let mut views = BufferMut::with_capacity(count); + // Wrapping keeps corrupt non-monotonic offsets from panicking on the subtraction itself; the + // wrapped length then fails the in-bounds slicing (or `max_buffer_len`) checks in the loop. + let buffers = extend_views( + &mut views, + start_buf_index, + max_buffer_len, + &bytes, + count, + |i| { + AsPrimitive::::as_(offsets[i + 1]) + .wrapping_sub(AsPrimitive::::as_(offsets[i])) + }, + ); + (buffers, views.freeze()) +} + +/// Appends one view per value straight into `views`, splitting `bytes` into buffers. +/// +/// This is the core behind [`build_views`]: it writes into an existing views buffer so that a +/// [`VarBinViewBuilder`](crate::builders::VarBinViewBuilder) can build views directly into its +/// storage without an intermediate allocation. `len_at(i)` is the byte length of value `i`, and +/// the `count` lengths must describe `bytes` exactly. The returned buffers are zero-copy slices +/// of `bytes`, numbered sequentially from `start_buf_index`. +pub(crate) fn extend_views( + views: &mut BufferMut, + start_buf_index: u32, + max_buffer_len: usize, + bytes: &ByteBuffer, + count: usize, + len_at: impl Fn(usize) -> usize, +) -> Vec { assert!( max_buffer_len <= MAX_BUFFER_LEN, "max_buffer_len cannot exceed MAX_BUFFER_LEN, offsets must fit in u32" ); if bytes.len() <= max_buffer_len { - // Common case: the whole decoded heap fits within a single buffer, so no rollover can occur - // (`bytes.len()` is the total decoded size and therefore an upper bound on every offset). - build_views_single_buffer(start_buf_index, bytes, lens) + // Common case: the whole decoded heap fits within a single buffer, so no rollover can + // occur (`bytes.len()` is the total decoded size and therefore an upper bound on every + // offset). + extend_views_single_buffer(views, start_buf_index, bytes, count, len_at); + if bytes.is_empty() { + Vec::new() + } else { + vec![bytes.clone()] + } } else { - build_views_rolling(start_buf_index, max_buffer_len, bytes, lens) + extend_views_rolling(views, start_buf_index, max_buffer_len, bytes, count, len_at) } } @@ -54,12 +135,15 @@ pub fn build_views>( /// reference views inline, avoiding the out-of-line `BinaryView::make_view` call for the common /// long-string case. Every offset is bounded by `bytes.len()`, which the caller has guaranteed is /// at most [`MAX_BUFFER_LEN`], so the `usize -> u32` conversions cannot truncate. -fn build_views_single_buffer>( - start_buf_index: u32, - bytes: ByteBufferMut, - lens: &[P], -) -> (Vec, Buffer) { - let mut views = BufferMut::::with_capacity(lens.len()); +fn extend_views_single_buffer( + views: &mut BufferMut, + buf_index: u32, + bytes: &ByteBuffer, + count: usize, + len_at: impl Fn(usize) -> usize, +) { + views.reserve(count); + let base = views.len(); let data = bytes.as_slice(); let mut offset = 0usize; @@ -68,72 +152,77 @@ fn build_views_single_buffer>( // loop-invariant, so it reloads and rewrites the output cursor through the stack each // iteration. Writing into the spare slice keeps the cursor in a register and the length is // set once after the loop. - let spare = views.spare_capacity_mut(); - for (slot, &len) in spare.iter_mut().zip(lens) { - let len = len.as_(); + let spare = &mut views.spare_capacity_mut()[..count]; + for (i, slot) in spare.iter_mut().enumerate() { + let len = len_at(i); let value = &data[offset..offset + len]; let view = if len > BinaryView::MAX_INLINED_SIZE { let mut prefix = [0u8; 4]; prefix.copy_from_slice(&value[..4]); - BinaryView::new_ref(len.as_(), prefix, start_buf_index, offset.as_()) + BinaryView::new_ref(len.as_(), prefix, buf_index, offset.as_()) } else { - BinaryView::make_view(value, start_buf_index, offset.as_()) + BinaryView::make_view(value, buf_index, offset.as_()) }; slot.write(view); offset += len; } - // SAFETY: the loop initialized exactly `lens.len()` contiguous views (`spare` has at least - // `lens.len()` slots, and `zip` stops at the shorter operand). - unsafe { views.set_len(lens.len()) }; - - let buffers = if bytes.is_empty() { - Vec::new() - } else { - vec![bytes.freeze()] - }; - (buffers, views.freeze()) + assert_eq!( + offset, + data.len(), + "value lengths must describe the byte heap exactly" + ); + // SAFETY: the loop initialized exactly `count` contiguous views (`spare` has at least + // `count` slots). + unsafe { views.set_len(base + count) }; } /// Build views when the heap exceeds `max_buffer_len` and must be split across multiple buffers. /// /// The buffer is rolled over every `max_buffer_len` bytes so that no view offset overflows the -/// `u32` offset field. -fn build_views_rolling>( +/// `u32` offset field. Each output buffer is a zero-copy slice of `bytes`. +fn extend_views_rolling( + views: &mut BufferMut, start_buf_index: u32, max_buffer_len: usize, - mut bytes: ByteBufferMut, - lens: &[P], -) -> (Vec, Buffer) { - let mut views = BufferMut::::with_capacity(lens.len()); + bytes: &ByteBuffer, + count: usize, + len_at: impl Fn(usize) -> usize, +) -> Vec { + views.reserve(count); let mut buffers = Vec::new(); let mut buf_index = start_buf_index; - let mut offset = 0; - for &len in lens { - let len = len.as_(); + let data = bytes.as_slice(); + // The absolute start of the current segment, and the offset of the next value within it. + let mut segment_start = 0usize; + let mut offset = 0usize; + for i in 0..count { + let len = len_at(i); assert!(len <= max_buffer_len, "values cannot exceed max_buffer_len"); - if (offset + len) > max_buffer_len { + if offset + len > max_buffer_len { // Roll the buffer every 2GiB, to avoid overflowing VarBinView offset field - let rest = bytes.split_off(offset); - - buffers.push(bytes.freeze()); + buffers.push(bytes.slice(segment_start..segment_start + offset)); buf_index += 1; + segment_start += offset; offset = 0; - - bytes = rest; } - let view = BinaryView::make_view(&bytes[offset..][..len], buf_index, offset.as_()); - // SAFETY: we reserved the right capacity beforehand - unsafe { views.push_unchecked(view) }; + let start = segment_start + offset; + let view = BinaryView::make_view(&data[start..start + len], buf_index, offset.as_()); + views.push(view); offset += len; } + assert_eq!( + segment_start + offset, + data.len(), + "value lengths must describe the byte heap exactly" + ); - if !bytes.is_empty() { - buffers.push(bytes.freeze()); + if segment_start < data.len() { + buffers.push(bytes.slice(segment_start..data.len())); } - (buffers, views.freeze()) + buffers } #[cfg(test)] @@ -145,17 +234,18 @@ mod tests { use crate::arrays::varbinview::BinaryView; use crate::arrays::varbinview::build_views::MAX_BUFFER_LEN; use crate::arrays::varbinview::build_views::build_views; + use crate::arrays::varbinview::build_views::build_views_from_offsets; /// Concatenate `values` into a single byte heap and return it alongside the per-element lengths, /// matching the `(bytes, lens)` inputs that `build_views` consumes. - fn flatten(values: &[&[u8]]) -> (ByteBufferMut, Vec) { + fn flatten(values: &[&[u8]]) -> (ByteBuffer, Vec) { let mut bytes = ByteBufferMut::empty(); let mut lens = Vec::with_capacity(values.len()); for v in values { bytes.extend_from_slice(v); lens.push(u32::try_from(v.len()).unwrap()); } - (bytes, lens) + (bytes.freeze(), lens) } /// Reconstruct the logical value behind each view by dereferencing it through the output @@ -206,7 +296,7 @@ mod tests { assert!(buffers.is_empty(), "empty heap must not allocate a buffer"); } else { assert_eq!(buffers.len(), 1, "whole heap must stay in one buffer"); - // The fast path freezes the input heap unchanged. + // The fast path adopts the input heap unchanged. let concatenated: Vec = values.concat(); assert_eq!(buffers[0].as_slice(), concatenated.as_slice()); } @@ -220,6 +310,37 @@ mod tests { assert_eq!(reconstruct(&buffers, &views, start_buf_index), expected); } + /// The output buffers must be zero-copy slices of the input heap, on both paths — a copy here + /// silently doubles the memory cost of every decode that feeds views. + #[test] + fn output_buffers_are_zero_copy() { + let values: &[&[u8]] = &[ + b"first long reference value", + b"tiny", + b"second long reference value!!", + b"third looooong reference value", + ]; + let (bytes, lens) = flatten(values); + let base = bytes.as_ptr(); + + // Fast path: the single output buffer is the input buffer. + let (buffers, _views) = build_views(0, bytes.len() + 1, bytes.clone(), &lens); + assert_eq!(buffers.len(), 1); + assert_eq!(buffers[0].as_ptr(), base, "fast path must not copy"); + + // Rolling path: every output buffer points into the input allocation. + let longest = values.iter().map(|v| v.len()).max().unwrap(); + let (buffers, _views) = build_views(0, longest, bytes, &lens); + assert!(buffers.len() > 1); + let mut expected_ptr = base; + for buffer in &buffers { + assert_eq!(buffer.as_ptr(), expected_ptr, "rolling path must not copy"); + // SAFETY: the buffers partition the input heap, so the next one starts where + // this one ends, still within (or one past) the original allocation. + expected_ptr = unsafe { expected_ptr.add(buffer.len()) }; + } + } + /// Offsets and sizes are written into the `u32` `Ref` fields via `as_` truncation, so we must /// confirm they stay correct once the running offset grows well past the 16-bit range (i.e. is /// not narrowed to a smaller width). A ~9 MiB heap pushes offsets above 2^23 while remaining far @@ -314,7 +435,7 @@ mod tests { #[test] fn fast_path_empty_input() { let lens: Vec = Vec::new(); - let (buffers, views) = build_views(0, 1024, ByteBufferMut::empty(), &lens); + let (buffers, views) = build_views(0, 1024, ByteBuffer::empty(), &lens); assert!(buffers.is_empty()); assert!(views.is_empty()); } @@ -336,6 +457,41 @@ mod tests { assert_eq!(views.as_slice(), &expected); } + /// The offsets-driven variant must agree with the lengths-driven one, reference only the + /// `offsets[0]..offsets[last]` range, and stay zero-copy — it exists so a `VarBinArray` heap + /// can feed views without materializing a lengths buffer or copying its bytes. + #[test] + fn from_offsets_matches_lengths_and_is_zero_copy() { + // A heap with a prefix and suffix outside the offsets range, as a sliced VarBin has. + let heap = ByteBuffer::copy_from(b"..a long value that is referenced!tiny..".as_slice()); + let offsets: Vec = vec![2, 34, 38]; + + let (buffers, views) = build_views_from_offsets(5, MAX_BUFFER_LEN, heap.clone(), &offsets); + + assert_eq!(buffers.len(), 1); + // Zero-copy: the buffer points at offset 2 of the original allocation. + // SAFETY: offset 2 is in bounds of the 40-byte heap. + assert_eq!(buffers[0].as_ptr(), unsafe { heap.as_ptr().add(2) }); + assert_eq!(buffers[0].len(), 36); + + assert_eq!( + reconstruct(&buffers, &views, 5), + vec![ + b"a long value that is referenced!".to_vec(), + b"tiny".to_vec() + ] + ); + } + + /// Lengths that do not cover the heap exactly are a caller bug and must be rejected rather + /// than silently emitting views over a partially-covered buffer. + #[test] + #[should_panic(expected = "value lengths must describe the byte heap exactly")] + fn short_lengths_panic() { + let (bytes, _) = flatten(&[b"a long value that is referenced", b"tiny"]); + build_views(0, MAX_BUFFER_LEN, bytes, &[31u32]); + } + // TODO(someone): ideally CI would run this in release mode as well, since debug builds make the // ~2.25 GiB allocation and fill loop substantially slower. /// Slow regression for the single-buffer fast-path guard. The fast path is only valid when the @@ -380,7 +536,7 @@ mod tests { } let lens = vec![u32::try_from(STRING_LEN).unwrap(); N]; - let (buffers, views) = build_views(0, MAX_BUFFER_LEN, bytes, &lens); + let (buffers, views) = build_views(0, MAX_BUFFER_LEN, bytes.freeze(), &lens); assert_eq!(views.len(), N); assert!( @@ -419,7 +575,7 @@ mod tests { // In real code, this would all fit in one buffer, but to unit test the splitting logic // we split buffers at length 26, which should result in two buffers for the output array. let raw_data = - ByteBufferMut::copy_from("aaaaaaaaaaaaabbbbbbbbbbbbbcccccccccccccddddddddddddd"); + ByteBuffer::copy_from("aaaaaaaaaaaaabbbbbbbbbbbbbcccccccccccccddddddddddddd"); let lens = vec![13u8; 4]; let (buffers, views) = build_views(0, 26, raw_data, &lens); @@ -446,11 +602,6 @@ mod tests { #[test] #[should_panic(expected = "max_buffer_len cannot exceed MAX_BUFFER_LEN")] fn test_max_buffer_len_too_large_panics() { - build_views( - 0, - MAX_BUFFER_LEN + 1, - ByteBufferMut::copy_from("abc"), - &[3u32], - ); + build_views(0, MAX_BUFFER_LEN + 1, ByteBuffer::copy_from("abc"), &[3u32]); } } diff --git a/vortex-array/src/builders/varbinview.rs b/vortex-array/src/builders/varbinview.rs index 1ec39138ebb..57d4ff64637 100644 --- a/vortex-array/src/builders/varbinview.rs +++ b/vortex-array/src/builders/varbinview.rs @@ -6,6 +6,7 @@ use std::ops::Range; use std::sync::Arc; use itertools::Itertools; +use num_traits::AsPrimitive; use vortex_buffer::Alignment; use vortex_buffer::Buffer; use vortex_buffer::BufferMut; @@ -15,6 +16,7 @@ use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_bail; use vortex_error::vortex_ensure; +use vortex_mask::AllOr; use vortex_mask::Mask; use vortex_utils::aliases::hash_map::Entry; use vortex_utils::aliases::hash_map::HashMap; @@ -25,11 +27,14 @@ use crate::IntoArray; use crate::arrays::VarBinViewArray; use crate::arrays::varbinview::VarBinViewArrayExt; use crate::arrays::varbinview::build_views::BinaryView; +use crate::arrays::varbinview::build_views::MAX_BUFFER_LEN; +use crate::arrays::varbinview::build_views::extend_views; use crate::arrays::varbinview::compact::BufferUtilization; use crate::builders::ArrayBuilder; use crate::builders::LazyBitBufferBuilder; use crate::canonical::Canonical; use crate::dtype::DType; +use crate::dtype::NativePType; use crate::scalar::Scalar; /// The builder for building a [`VarBinViewArray`]. @@ -188,57 +193,237 @@ impl VarBinViewBuilder { self.completed.len() } - /// Returns true if a non-empty in-progress buffer is staged (and would - /// become a completed buffer on the next flush), false otherwise. - pub fn in_progress(&self) -> bool { - self.in_progress.is_some() + /// 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. + fn compacts_buffers(&self) -> bool { + self.compaction_threshold > 0.0 } - /// Whether this builder compacts the data buffers it is handed. + /// Adopts the buffers and views that `build` produces against the index its first buffer + /// will land at. /// - /// [`push_buffer_and_adjusted_views`](Self::push_buffer_and_adjusted_views) takes buffers - /// exactly as they are, so an encoding that would push a buffer only partly covered by its - /// views should check this first and fall back to a route that measures utilization — - /// otherwise it silently opts the builder out of the compaction it was configured for. - pub fn compacts_buffers(&self) -> bool { - self.compaction_threshold > 0.0 + /// The builder flushes its staged bytes, then hands `build` the index the next data buffer + /// will occupy; `build` returns data buffers — which land contiguously from that index — and + /// one view per entry of `validity`, already referencing them. This is the escape hatch for + /// an encoding that only discovers its views while walking its own byte format (e.g. + /// length-prefixed frames), where the lengths-driven appends cannot apply; keeping the + /// numbering inside this call is what makes the views come out right without a rebase pass. + /// + /// # Warning + /// + /// This method does not check utilization of the returned buffers. `build` must return + /// buffers that are fully utilized by its views. + /// + /// # Panics + /// + /// Panics if `build` returns a different view count than `validity.len()`, or if this + /// builder deduplicates buffers and already holds one of the returned buffers. + pub fn append_views_built_at( + &mut self, + validity: &Mask, + build: impl FnOnce(u32) -> VortexResult<(Vec, Buffer)>, + ) -> VortexResult<()> { + self.flush_in_progress(); + + let start_index = self.completed.len(); + let (buffers, views) = build(start_index)?; + assert_eq!( + views.len(), + validity.len(), + "Must build one view per validity entry" + ); + + let expected_completed_len = start_index as usize + buffers.len(); + self.completed.extend_from_slice_unchecked(&buffers); + assert_eq!( + self.completed.len() as usize, + expected_completed_len, + "Some buffers already exist", + ); + self.views_builder.extend_trusted(views.iter().copied()); + self.nulls.append_validity_mask(validity); + + debug_assert_eq!(self.nulls.len(), self.views_builder.len()); + Ok(()) } - /// Pushes buffers and pre-adjusted views into the builder. + /// Appends values laid end-to-end in `bytes`, one per entry of `lengths`. /// - /// The provided `buffers` contain sections of data from a `VarBinViewArray`, and the - /// `views` are `BinaryView`s that have already been adjusted to reference the correct buffer - /// indices and offsets for this builder. All views must point to valid sections within the - /// provided buffers, and the validity length must match the view length. + /// The builder adopts `bytes` as a data buffer without copying it (splitting it only past the + /// `u32` view-offset limit) and builds the views directly into its own storage, so the whole + /// append costs one view per row. Every length is consumed, including those for null rows, so + /// the lengths must describe `bytes` exactly; bytes belonging to null rows are not retained + /// when compaction rewrites the buffer. /// - /// # Warning + /// When the builder is configured to compact buffers, the utilization is measured from the + /// lengths alone — only values too long to inline reference the buffer — and a heap below + /// the threshold is rewritten to just those values instead of adopted, so callers never need + /// a canonicalize-and-compact fallback. /// - /// This method does not check utilization of the given buffers. Callers must provide - /// buffers that are fully utilized by the given adjusted views. + /// # Panics + /// + /// Panics if `lengths` and `validity` disagree in length, if the lengths do not describe + /// `bytes` exactly, or if this builder deduplicates buffers and already holds `bytes`. + pub fn append_buffer_with_lengths>( + &mut self, + bytes: ByteBuffer, + lengths: &[P], + validity: &Mask, + ) { + assert_eq!( + lengths.len(), + validity.len(), + "Must have one length per validity entry" + ); + self.append_buffer_views(&bytes, lengths.len(), validity, |i| lengths[i].as_()); + } + + /// [`append_buffer_with_lengths`](Self::append_buffer_with_lengths) for values described by + /// an offsets buffer instead of lengths. + /// + /// `offsets` are absolute positions into `bytes` — the layout a + /// [`VarBinArray`](crate::arrays::VarBinArray) stores — so there is one more offset than there + /// are values, and only the `offsets[0]..offsets[last]` range of `bytes` is adopted, again + /// without copying. /// /// # Panics /// - /// Panics if this builder deduplicates buffers and any of the given buffers already - /// exist in this builder. - pub fn push_buffer_and_adjusted_views( + /// Panics if `offsets` does not hold exactly one more entry than `validity`, or if the offsets + /// are not monotonically non-decreasing positions within `bytes`. + pub fn append_buffer_with_offsets>( + &mut self, + bytes: ByteBuffer, + offsets: &[P], + validity: &Mask, + ) { + assert_eq!( + offsets.len(), + validity.len() + 1, + "Must have one more offset than validity entries" + ); + let first: usize = offsets[0].as_(); + let last: usize = offsets[offsets.len() - 1].as_(); + let bytes = bytes.slice(first..last); + // Wrapping keeps corrupt non-monotonic offsets from panicking on the subtraction itself; + // the wrapped length then fails the in-bounds checks of the view-building loop. + self.append_buffer_views(&bytes, validity.len(), validity, |i| { + AsPrimitive::::as_(offsets[i + 1]) + .wrapping_sub(AsPrimitive::::as_(offsets[i])) + }); + } + + /// Shared tail of the bulk buffer appends: builds the views straight into the builder's views + /// storage, then adopts the buffer segments and the validity. + fn append_buffer_views( &mut self, - buffers: &[ByteBuffer], - views: &Buffer, - validity_mask: Mask, + bytes: &ByteBuffer, + count: usize, + validity: &Mask, + len_at: impl Fn(usize) -> usize, ) { self.flush_in_progress(); - let expected_completed_len = self.completed.len() as usize + buffers.len(); - self.completed.extend_from_slice_unchecked(buffers); + // A compacting builder measures utilization before adopting the buffer. Only values too + // long to inline reference the heap, so the measurement is one pass over the lengths and + // never touches the bytes. Heaps past the single-buffer limit are adopted as they are: + // they roll over into multiple buffers, and per-segment accounting is not worth the rare + // >2GiB case. + if self.compacts_buffers() && bytes.len() <= MAX_BUFFER_LEN { + let referenced: usize = match validity.bit_buffer() { + AllOr::All => (0..count) + .map(&len_at) + .filter(|len| *len > BinaryView::MAX_INLINED_SIZE) + .sum(), + AllOr::None => 0, + AllOr::Some(b) => { + let mut sum = 0; + b.for_each_set_index(|idx| { + let len = len_at(idx); + if len > BinaryView::MAX_INLINED_SIZE { + sum += len; + } + }); + sum + } + }; + #[expect(clippy::cast_precision_loss)] + if (referenced as f64) < self.compaction_threshold * (bytes.len() as f64) { + return self + .append_buffer_views_rewritten(bytes, count, validity, len_at, referenced); + } + } + + let start_index = self.completed.len(); + let segments = extend_views( + &mut self.views_builder, + start_index, + MAX_BUFFER_LEN, + bytes, + count, + len_at, + ); + + let expected_completed_len = start_index as usize + segments.len(); + self.completed.extend_from_slice_unchecked(&segments); assert_eq!( self.completed.len() as usize, expected_completed_len, "Some buffers already exist", ); - self.views_builder.extend_trusted(views.iter().copied()); - self.push_only_validity_mask(&validity_mask); - debug_assert_eq!(self.nulls.len(), self.views_builder.len()) + self.nulls.append_validity_mask(validity); + debug_assert_eq!(self.nulls.len(), self.views_builder.len()); + } + + /// The under-utilized arm of [`append_buffer_views`](Self::append_buffer_views): copies only + /// the values that actually reference the heap into a compact buffer, sized `referenced`, + /// instead of adopting the whole heap. A fully-inlined append pushes no buffer at all. + fn append_buffer_views_rewritten( + &mut self, + bytes: &ByteBuffer, + count: usize, + validity: &Mask, + len_at: impl Fn(usize) -> usize, + referenced: usize, + ) { + let buf_index = self.completed.len(); + let mut compact = ByteBufferMut::with_capacity(referenced); + self.views_builder.reserve(count); + + let data = bytes.as_slice(); + let mut offset = 0usize; + for (i, is_valid) in validity.iter().enumerate() { + let len = len_at(i); + let value = &data[offset..offset + len]; + let view = if !is_valid { + BinaryView::empty_view() + } else if len > BinaryView::MAX_INLINED_SIZE { + // In `u32` range: `referenced <= bytes.len() <= MAX_BUFFER_LEN` (checked by the + // caller), and `compact` never grows past `referenced`. + #[expect(clippy::cast_possible_truncation)] + let view = BinaryView::make_view(value, buf_index, compact.len() as u32); + compact.extend_from_slice(value); + view + } else { + BinaryView::make_view(value, buf_index, 0) + }; + self.views_builder.push(view); + offset += len; + } + assert_eq!( + offset, + data.len(), + "value lengths must describe the byte heap exactly" + ); + + if !compact.is_empty() { + let pushed_index = self.completed.push(compact.freeze()); + assert_eq!(pushed_index, buf_index, "Buffer already exists"); + } + + self.nulls.append_validity_mask(validity); + debug_assert_eq!(self.nulls.len(), self.views_builder.len()); } /// Finishes the builder directly into a [`VarBinViewArray`]. @@ -265,11 +450,6 @@ impl VarBinViewBuilder { } } - // Pushes a validity mask into the builder not affecting the views or buffers - fn push_only_validity_mask(&mut self, validity_mask: &Mask) { - self.nulls.append_validity_mask(validity_mask); - } - pub(crate) fn append_varbinview_array( &mut self, array: &VarBinViewArray, @@ -279,7 +459,7 @@ impl VarBinViewBuilder { let mask = array.varbinview_validity().execute_mask(array.len(), ctx)?; - self.push_only_validity_mask(&mask); + self.nulls.append_validity_mask(&mask); let view_adjustment = self.completed @@ -439,12 +619,14 @@ impl CompletedBuffers { } } + /// Push a new block, returning the index it landed at (or, when deduplicating, the index of + /// the identical block already held). fn push(&mut self, block: ByteBuffer) -> u32 { match self { Self::Default(buffers) => { assert!(buffers.len() < u32::MAX as usize, "Too many blocks"); buffers.push(block); - self.len() + self.len() - 1 } Self::Deduplicated(buffers) => buffers.push(block), } @@ -874,7 +1056,9 @@ impl RewritingViewAdjustment { #[cfg(test)] mod tests { + use vortex_buffer::ByteBuffer; use vortex_error::VortexResult; + use vortex_mask::Mask; use crate::IntoArray; use crate::VortexSessionExecute; @@ -886,6 +1070,148 @@ mod tests { use crate::dtype::DType; use crate::dtype::Nullability; + /// A long-enough value that a view over it must reference a data buffer. + const LONG: &str = "a value that is far too long to inline"; + + /// The heap is adopted zero-copy as a data buffer, the views are built against it in place, + /// and the append composes with staged in-progress bytes on either side. + #[test] + fn test_append_buffer_with_lengths() { + let mut ctx = array_session().create_execution_ctx(); + let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8); + + // Stages an in-progress buffer the bulk append has to flush first. + builder.append_value(LONG); + + let heap = ByteBuffer::copy_from([LONG.as_bytes(), b"", b"tiny"].concat()); + let heap_ptr = heap.as_ptr(); + let lengths = [u32::try_from(LONG.len()).unwrap(), 0, 4]; + builder.append_buffer_with_lengths(heap, &lengths, &Mask::from_iter([true, false, true])); + + builder.append_value("tail"); + + let actual = builder.finish_into_varbinview(); + // The adopted heap sits after the flushed in-progress buffer, untouched. + assert_eq!(actual.data_buffers()[1].as_host().as_ptr(), heap_ptr); + + let expected = >::from_iter([ + Some(LONG), + Some(LONG), + None, + Some("tiny"), + Some("tail"), + ]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + + /// Offsets need not start at zero: only the referenced range of the heap is adopted. + #[test] + fn test_append_buffer_with_offsets() { + let mut ctx = array_session().create_execution_ctx(); + let mut builder = VarBinViewBuilder::with_capacity(DType::Utf8(Nullability::Nullable), 8); + + let heap = ByteBuffer::copy_from(format!("..{LONG}tiny!!")); + let long_len = u32::try_from(LONG.len()).unwrap(); + let offsets = [2u32, 2 + long_len, 2 + long_len, 2 + long_len + 4]; + builder.append_buffer_with_offsets( + heap.clone(), + &offsets, + &Mask::from_iter([true, false, true]), + ); + + let actual = builder.finish_into_varbinview(); + // Zero-copy adoption of just the `offsets[0]..offsets[last]` range. + // SAFETY: offset 2 is in bounds of the heap. + assert_eq!(actual.data_buffers()[0].as_host().as_ptr(), unsafe { + heap.as_ptr().add(2) + }); + assert_eq!( + actual.data_buffers()[0].len(), + LONG.len() + 4, + "only the referenced range must be adopted" + ); + + let expected = + >::from_iter([Some(LONG), None, Some("tiny")]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + + /// A compacting builder must measure the heap it is handed: values short enough to inline + /// never reference it, so an under-utilized heap is rewritten to just the referencing values. + #[test] + fn test_append_buffer_with_lengths_compacts_underutilized_heap() { + let mut ctx = array_session().create_execution_ctx(); + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 4, 1.0); + + let heap = ByteBuffer::copy_from([b"short".as_slice(), LONG.as_bytes(), b"tiny"].concat()); + let lengths = [5u32, u32::try_from(LONG.len()).unwrap(), 4]; + builder.append_buffer_with_lengths(heap, &lengths, &Mask::new_true(3)); + + let actual = builder.finish_into_varbinview(); + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!( + actual.data_buffers()[0].len(), + LONG.len(), + "the compact buffer must hold only the non-inlined values" + ); + + let expected = >::from_iter([ + Some("short"), + Some(LONG), + Some("tiny"), + ]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + + /// Rewriting an under-utilized heap must consume null values' spans without retaining their + /// bytes or producing views that reference them. + #[test] + fn test_append_buffer_with_lengths_compaction_skips_null_bytes() { + let mut ctx = array_session().create_execution_ctx(); + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 2, 1.0); + + let heap = ByteBuffer::copy_from([LONG.as_bytes(), LONG.as_bytes()].concat()); + let lengths = [u32::try_from(LONG.len()).unwrap(); 2]; + builder.append_buffer_with_lengths(heap, &lengths, &Mask::from_iter([false, true])); + + let actual = builder.finish_into_varbinview(); + assert_eq!(actual.data_buffers().len(), 1); + assert_eq!( + actual.data_buffers()[0].len(), + LONG.len(), + "the compact buffer must omit bytes belonging to null rows" + ); + + let expected = >::from_iter([None::<&str>, Some(LONG)]); + 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() { + let mut ctx = array_session().create_execution_ctx(); + let mut builder = + VarBinViewBuilder::with_compaction(DType::Utf8(Nullability::Nullable), 4, 1.0); + + let heap = ByteBuffer::copy_from(b"shorttinysmall".as_slice()); + builder.append_buffer_with_lengths(heap, &[5u32, 4, 5], &Mask::new_true(3)); + + let actual = builder.finish_into_varbinview(); + assert!( + actual.data_buffers().is_empty(), + "a fully-inlined append must not retain any value bytes" + ); + + let expected = >::from_iter([ + Some("short"), + Some("tiny"), + Some("small"), + ]); + assert_arrays_eq!(actual, expected, &mut ctx); + } + #[test] fn test_utf8_builder() { let mut ctx = array_session().create_execution_ctx(); diff --git a/vortex-cuda/src/kernel/encodings/fsst.rs b/vortex-cuda/src/kernel/encodings/fsst.rs index 4c5e2219ccc..f9b90c8ac35 100644 --- a/vortex-cuda/src/kernel/encodings/fsst.rs +++ b/vortex-cuda/src/kernel/encodings/fsst.rs @@ -386,12 +386,7 @@ where let host_bytes = host_bytes.slice(0..total_size); let (buffers, views) = match_each_integer_ptype!(lens.ptype(), |P| { - build_views( - 0, - MAX_BUFFER_LEN, - host_bytes.into_mut(), - lens.as_slice::

(), - ) + build_views(0, MAX_BUFFER_LEN, host_bytes, lens.as_slice::

()) }); Ok(Canonical::VarBinView(unsafe {