Replace DynVarBinBuilder with a generic VarBinBuilder

`VarBinBuilder<O>` becomes an actual `ArrayBuilder`, carrying its own `DType`
and offset width, so `DynVarBinBuilder` is no longer needed as a separate
type-erasing wrapper.

Encodings recover the concrete offset width from a `&mut dyn ArrayBuilder`
with the new `match_each_varbin_builder!` (i32/i64) and
`match_each_any_varbin_builder!` (every width) macros, which monomorphize the
append body over the width instead of dispatching through `dyn`. The existing
`append_to_builder` specializations for FSST, OnPair and Zstd are ported over
mechanically; their optimizations follow in later commits.

Because the builder's offset type now matches the Arrow target, the Arrow byte
export hands the offsets buffer straight to Arrow with no cast, which also
removes the `VarBinView` + `arrow_cast` round trip a `Utf8`/`Binary` dtype
mismatch used to take. A `Binary` source exported to `Utf8` instead validates
the bytes of its live values, skipping the extents of null slots that
`GenericByteArray::try_new` would have rejected.

Also adds `BinaryView::bytes`, which borrows from already-resolved buffer
slices instead of cloning a buffer handle per row.

Signed-off-by: Robert Kruszewski <robert@spiraldb.com>

Author: robert3005

Cargo.lock

@@ -34,13 +34,15 @@ use vortex_array::arrays::VarBinArray;
 use vortex_array::arrays::varbin::VarBinArraySlotsExt;
 use vortex_array::buffer::BufferHandle;
 use vortex_array::builders::ArrayBuilder;
-use vortex_array::builders::DynVarBinBuilder;
+use vortex_array::builders::VarBinBuilder;
 use vortex_array::builders::VarBinViewBuilder;
 use vortex_array::dtype::DType;
+use vortex_array::dtype::IntegerPType;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::PType;
 use vortex_array::legacy_session;
 use vortex_array::match_each_integer_ptype;
+use vortex_array::match_each_varbin_builder;
 use vortex_array::serde::ArrayChildren;
 use vortex_array::validity::Validity;
 use vortex_array::vtable::VTable;
@@ -307,23 +309,10 @@ impl VTable for FSST {
         builder: &mut dyn ArrayBuilder,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<()> {
-        if let Some(builder) = builder.as_any_mut().downcast_mut::<DynVarBinBuilder>() {
-            let (bytes, lengths) = fsst_decode_bytes(array, ctx)?;
-            let validity = array
-                .array()
-                .validity()?
-                .execute_mask(array.array().len(), ctx)?;
-            match_each_integer_ptype!(lengths.ptype(), |P| {
-                builder.append_values(
-                    bytes.as_slice(),
-                    lengths.as_slice::<P>().iter().scan(0usize, |end, length| {
-                        *end += AsPrimitive::<usize>::as_(*length);
-                        Some(*end)
-                    }),
-                    &validity,
-                )
-            })?;
-            return Ok(());
+        if let Some(result) =
+            match_each_varbin_builder!(builder, |builder| append_to_varbin(array, builder, ctx))
+        {
+            return result;
         }
 
         let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
@@ -361,6 +350,32 @@ impl VTable for FSST {
     }
 }
 
+/// Decompresses the values and appends them to `builder`.
+fn append_to_varbin<O: IntegerPType>(
+    array: ArrayView<'_, FSST>,
+    builder: &mut VarBinBuilder<O>,
+    ctx: &mut ExecutionCtx,
+) -> VortexResult<()>
+where
+    usize: AsPrimitive<O>,
+{
+    let (bytes, lengths) = fsst_decode_bytes(array, ctx)?;
+    let validity = array
+        .array()
+        .validity()?
+        .execute_mask(array.array().len(), ctx)?;
+    match_each_integer_ptype!(lengths.ptype(), |P| {
+        builder.append_values(
+            bytes.as_slice(),
+            lengths.as_slice::<P>().iter().scan(0usize, |end, length| {
+                *end += AsPrimitive::<usize>::as_(*length);
+                Some(*end)
+            }),
+            &validity,
+        )
+    })
+}
+
 #[array_slots(FSST)]
 pub struct FSSTSlots {
     /// Lengths of the original values before compression, can be compressed.

encodings/fsst/src/array.rs

@@ -106,7 +106,7 @@ mod tests {
     use vortex_array::arrays::VarBinViewArray;
     use vortex_array::arrays::varbin::VarBinArrayExt;
     use vortex_array::builders::ArrayBuilder;
-    use vortex_array::builders::DynVarBinBuilder;
+    use vortex_array::builders::VarBinBuilder;
     use vortex_array::builders::VarBinViewBuilder;
     use vortex_array::dtype::DType;
     use vortex_array::dtype::Nullability;
@@ -212,7 +212,7 @@ mod tests {
 
         {
             let mut builder =
-                DynVarBinBuilder::with_capacity(chunked_arr.dtype().clone(), false, data.len());
+                VarBinBuilder::<i32>::with_capacity(chunked_arr.dtype().clone(), data.len());
             chunked_arr
                 .into_array()
                 .append_to_builder(&mut builder, &mut ctx)?;

encodings/fsst/src/canonical.rs

@@ -199,7 +199,11 @@ fn compress_views<O>(
 where
     O: IntegerPType + 'static,
 {
-    let mut sink = FsstSink::<O>::with_capacity(strings.len(), compressor);
+    let mut sink = FsstSink::<O>::with_capacity(
+        DType::Binary(strings.dtype().nullability()),
+        strings.len(),
+        compressor,
+    );
     let views = strings.views();
     let buffers = strings.data_buffers();
     match mask.bit_buffer() {
@@ -232,7 +236,11 @@ fn compress_varbin<O>(
 where
     O: IntegerPType + 'static,
 {
-    let mut sink = FsstSink::<O>::with_capacity(strings.len(), compressor);
+    let mut sink = FsstSink::<O>::with_capacity(
+        DType::Binary(strings.dtype().nullability()),
+        strings.len(),
+        compressor,
+    );
     let bytes = strings.bytes().as_slice();
     match_each_integer_ptype!(offsets.ptype(), |I| {
         let off = offsets.as_slice::<I>();
@@ -277,10 +285,10 @@ struct FsstSink<'c, O: IntegerPType + 'static> {
 }
 
 impl<'c, O: IntegerPType + 'static> FsstSink<'c, O> {
-    fn with_capacity(len: usize, compressor: &'c Compressor) -> Self {
+    fn with_capacity(dtype: DType, len: usize, compressor: &'c Compressor) -> Self {
         Self {
             buffer: Vec::with_capacity(DEFAULT_BUFFER_LEN),
-            builder: VarBinBuilder::<O>::with_capacity(len),
+            builder: VarBinBuilder::<O>::with_capacity(dtype, len),
             uncompressed_lengths: BufferMut::with_capacity(len),
             compressor,
         }
@@ -289,7 +297,7 @@ impl<'c, O: IntegerPType + 'static> FsstSink<'c, O> {
     #[inline]
     fn emit(&mut self, row: Option<&[u8]>) {
         let Some(s) = row else {
-            self.builder.append_null();
+            self.builder.push_null();
             self.uncompressed_lengths.push(0);
             return;
         };
@@ -310,8 +318,8 @@ impl<'c, O: IntegerPType + 'static> FsstSink<'c, O> {
         self.builder.append_value(&self.buffer);
     }
 
-    fn finish(self, dtype: DType, ctx: &mut ExecutionCtx) -> VortexResult<FSSTArray> {
-        let codes = self.builder.finish(DType::Binary(dtype.nullability()));
+    fn finish(mut self, dtype: DType, ctx: &mut ExecutionCtx) -> VortexResult<FSSTArray> {
+        let codes = self.builder.finish_into_varbin();
         FSST::try_new(
             dtype,
             Buffer::copy_from(self.compressor.symbol_table()),

encodings/fsst/src/compress.rs

@@ -61,7 +61,8 @@ mod tests {
     });
 
     fn build_test_fsst_array() -> ArrayRef {
-        let mut builder = VarBinBuilder::<i32>::with_capacity(10);
+        let mut builder =
+            VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::NonNullable), 10);
         builder.append_value(b"hello world");
         builder.append_value(b"foo bar baz");
         builder.append_value(b"testing fsst compression");
@@ -72,7 +73,7 @@ mod tests {
         builder.append_value(b"qrstuvwxyz");
         builder.append_value(b"0123456789");
         builder.append_value(b"final string");
-        let input = builder.finish(DType::Utf8(Nullability::NonNullable));
+        let input = builder.finish_into_varbin();
 
         let mut ctx = SESSION.create_execution_ctx();
         let arr = input.into_array();
@@ -131,7 +132,8 @@ mod tests {
         // Test case with special characters and nulls
         // Values: ["", "", "", "", "", "", "", "", "", "", "", ",", "A<<<<<<<", "", "", "", "", null, null, null, null, null, null]
         // Mask: only the last element is selected (true at index 22)
-        let mut builder = VarBinBuilder::<i32>::with_capacity(23);
+        let mut builder =
+            VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::Nullable), 23);
         // 11 empty strings
         for _ in 0..11 {
             builder.append_value(b"");
@@ -146,9 +148,9 @@ mod tests {
         }
         // 6 nulls
         for _ in 0..6 {
-            builder.append_null();
+            builder.push_null();
         }
-        let input = builder.finish(DType::Utf8(Nullability::Nullable));
+        let input = builder.finish_into_varbin();
         let array = input.clone().into_array();
 
         let mut ctx = SESSION.create_execution_ctx();
@@ -172,12 +174,13 @@ mod tests {
 
     #[test]
     fn filter_only_null() -> VortexResult<()> {
-        let mut builder = VarBinBuilder::<i32>::with_capacity(3);
-        builder.append_null();
+        let mut builder =
+            VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::Nullable), 3);
+        builder.push_null();
         builder.append_value(b"A");
-        builder.append_null();
+        builder.push_null();
 
-        let input = builder.finish(DType::Utf8(Nullability::Nullable));
+        let input = builder.finish_into_varbin();
         let array = input.clone().into_array();
 
         let mut ctx = SESSION.create_execution_ctx();
@@ -213,15 +216,14 @@ mod tests {
 
     #[test]
     fn test_fsst_byte_length() -> VortexResult<()> {
-        let mut builder = VarBinBuilder::<i32>::with_capacity(3);
+        let mut builder =
+            VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::NonNullable), 3);
         builder.append_value(b"hello");
         builder.append_value(b"world!!");
         builder.append_value("Пуховички"); // 9 characters, 18 bytes
         builder.append_value(b"");
 
-        let varbin = builder
-            .finish(DType::Utf8(Nullability::NonNullable))
-            .into_array();
+        let varbin = builder.finish_into_varbin().into_array();
         let mut ctx = SESSION.create_execution_ctx();
         let compressor = fsst_train_compressor(&varbin, &mut ctx)?;
         let fsst = fsst_compress(&varbin, &compressor, &mut ctx)?.into_array();

encodings/fsst/src/kernel.rs

@@ -30,15 +30,14 @@ static SESSION: LazyLock<VortexSession> = LazyLock::new(|| {
 
 /// this function is VERY slow on miri, so we only want to run it once
 pub(crate) fn build_fsst_array(ctx: &mut ExecutionCtx) -> ArrayRef {
-    let mut input_array = VarBinBuilder::<i32>::with_capacity(3);
+    let mut input_array =
+        VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::NonNullable), 3);
     input_array.append_value(b"The Greeks never said that the limit could not be overstepped");
     input_array.append_value(
         b"They said it existed and that whoever dared to exceed it was mercilessly struck down",
     );
     input_array.append_value(b"Nothing in present history can contradict them");
-    let input_array = input_array
-        .finish(DType::Utf8(Nullability::NonNullable))
-        .into_array();
+    let input_array = input_array.finish_into_varbin().into_array();
 
     let compressor = fsst_train_compressor(&input_array, ctx).unwrap();
     fsst_compress(&input_array, &compressor, ctx)
@@ -162,13 +161,11 @@ fn fsst_compress_offsets_overflow_i32() {
 
     println!("building large VarBinArray");
     let string = vec![b'a'; STRING_LEN];
-    let mut builder = VarBinBuilder::<i64>::with_capacity(N);
+    let mut builder = VarBinBuilder::<i64>::with_capacity(DType::Utf8(Nullability::NonNullable), N);
     for _ in 0..N {
         builder.append_value(&string);
     }
-    let array = builder
-        .finish(DType::Utf8(Nullability::NonNullable))
-        .into_array();
+    let array = builder.finish_into_varbin().into_array();
 
     let compressor = CompressorBuilder::default().build();
     let len = array.len();

encodings/fsst/src/tests.rs

@@ -26,12 +26,14 @@ use vortex_array::IntoArray;
 use vortex_array::array_slots;
 use vortex_array::buffer::BufferHandle;
 use vortex_array::builders::ArrayBuilder;
-use vortex_array::builders::DynVarBinBuilder;
+use vortex_array::builders::VarBinBuilder;
 use vortex_array::builders::VarBinViewBuilder;
 use vortex_array::dtype::DType;
+use vortex_array::dtype::IntegerPType;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::PType;
 use vortex_array::match_each_integer_ptype;
+use vortex_array::match_each_varbin_builder;
 use vortex_array::serde::ArrayChildren;
 use vortex_array::validity::Validity;
 use vortex_array::vtable::VTable;
@@ -553,23 +555,10 @@ impl VTable for OnPair {
         builder: &mut dyn ArrayBuilder,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<()> {
-        if let Some(builder) = builder.as_any_mut().downcast_mut::<DynVarBinBuilder>() {
-            let (bytes, lengths) = onpair_decode_bytes(array, ctx)?;
-            let validity = array
-                .array()
-                .validity()?
-                .execute_mask(array.array().len(), ctx)?;
-            match_each_integer_ptype!(lengths.ptype(), |P| {
-                builder.append_values(
-                    bytes.as_slice(),
-                    lengths.as_slice::<P>().iter().scan(0usize, |end, length| {
-                        *end += AsPrimitive::<usize>::as_(*length);
-                        Some(*end)
-                    }),
-                    &validity,
-                )
-            })?;
-            return Ok(());
+        if let Some(result) =
+            match_each_varbin_builder!(builder, |builder| append_to_varbin(array, builder, ctx))
+        {
+            return result;
         }
 
         let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
@@ -603,6 +592,32 @@ impl VTable for OnPair {
     }
 }
 
+/// Decodes the values and appends them to `builder`.
+fn append_to_varbin<O: IntegerPType>(
+    array: ArrayView<'_, OnPair>,
+    builder: &mut VarBinBuilder<O>,
+    ctx: &mut ExecutionCtx,
+) -> VortexResult<()>
+where
+    usize: AsPrimitive<O>,
+{
+    let (bytes, lengths) = onpair_decode_bytes(array, ctx)?;
+    let validity = array
+        .array()
+        .validity()?
+        .execute_mask(array.array().len(), ctx)?;
+    match_each_integer_ptype!(lengths.ptype(), |P| {
+        builder.append_values(
+            bytes.as_slice(),
+            lengths.as_slice::<P>().iter().scan(0usize, |end, length| {
+                *end += AsPrimitive::<usize>::as_(*length);
+                Some(*end)
+            }),
+            &validity,
+        )
+    })
+}
+
 impl ValidityVTable<OnPair> for OnPair {
     fn validity(array: ArrayView<'_, OnPair>) -> VortexResult<Validity> {
         Ok(child_to_validity(

encodings/onpair/src/array.rs

@@ -16,7 +16,7 @@ use vortex_array::arrays::VarBinViewArray;
 use vortex_array::arrays::filter::FilterKernel;
 use vortex_array::assert_arrays_eq;
 use vortex_array::buffer::BufferHandle;
-use vortex_array::builders::DynVarBinBuilder;
+use vortex_array::builders::VarBinBuilder;
 use vortex_array::builtins::ArrayBuiltins;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
@@ -66,7 +66,7 @@ fn test_direct_offset_builder() -> vortex_error::VortexResult<()> {
     let mut ctx = SESSION.create_execution_ctx();
     let input = sample_input();
     let encoded = compress_onpair(input.as_ref(), &mut ctx)?;
-    let mut builder = DynVarBinBuilder::with_capacity(input.dtype().clone(), false, input.len());
+    let mut builder = VarBinBuilder::<i32>::with_capacity(input.dtype().clone(), input.len());
     encoded
         .into_array()
         .append_to_builder(&mut builder, &mut ctx)?;

encodings/onpair/src/tests.rs

@@ -506,7 +506,7 @@ mod tests {
     use vortex_array::arrays::DictArray;
     use vortex_array::arrays::VarBinViewArray;
     use vortex_array::assert_arrays_eq;
-    use vortex_array::builders::DynVarBinBuilder;
+    use vortex_array::builders::VarBinBuilder;
     use vortex_array::dtype::DType;
     use vortex_array::dtype::Nullability;
     use vortex_array::dtype::PType;
@@ -564,7 +564,7 @@ mod tests {
             Some("c"),
         ])
         .into_array();
-        let mut builder = DynVarBinBuilder::with_capacity(arr.dtype().clone(), false, arr.len());
+        let mut builder = VarBinBuilder::<i32>::with_capacity(arr.dtype().clone(), arr.len());
         arr.append_to_builder(&mut builder, &mut ctx).unwrap();
         assert_arrays_eq!(builder.finish_into_varbin(), expected, &mut ctx);
         assert_arrays_eq!(arr.into_array(), expected, &mut ctx);