VarBinViewBuilder: support appending buffers directly into the builder storage

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::<P>(),
+                &validity,
+            )
+        });
         Ok(())
     }
 

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<ArrayRef> {
-    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::<P>(),
+        )
+    });
     // 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<ByteBuffer>, Buffer<BinaryView>)> {
-    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::<P>(),
-        ))
-    })
-}
-
 #[cfg(test)]
 mod tests {
     use std::sync::LazyLock;

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::<P>(),
+                &validity,
+            )
+        });
         Ok(())
     }
 

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<ArrayRef> {
-    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::<P>(),
+        )
+    });
     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<ByteBuffer>, Buffer<BinaryView>)> {
-    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::<P>(),
-        ))
-    })
-}

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::<BinaryView>::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::<BinaryView>::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)]

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();

vortex-array/src/arrays/varbin/vtable/canonical.rs

@@ -3,65 +3,44 @@
 
 use std::sync::Arc;
 
-use num_traits::AsPrimitive;
-use vortex_buffer::Buffer;
-use vortex_buffer::ByteBuffer;
 use vortex_error::VortexResult;
 
 use crate::ExecutionCtx;
 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<VarBinViewArray> {
     let parts = array.into_owned().into_data_parts();
     let offsets = parts.offsets.execute::<PrimitiveArray>(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::<P>(),
+        )
+    });
 
     // SAFETY: views are correctly computed from valid offsets
     Ok(unsafe {
         VarBinViewArray::new_unchecked(views, Arc::from(buffers), parts.dtype, parts.validity)
     })
 }
 
-/// 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<ByteBuffer>, Buffer<BinaryView>) {
-    match_each_integer_ptype!(offsets.ptype(), |P| {
-        let offsets_slice = offsets.as_slice::<P>();
-        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();