feat(arrow): export strings and binary arrays with offsets

Problem

Vortex exports UTF-8 and binary data as Arrow view arrays by default.
Some consumers require Arrow arrays that use 32-bit or 64-bit offsets.
These consumers must cast each view array and rebuild its buffers.

Implementation

Add VarBinBufferBuilder for 32-bit and 64-bit offsets.
Write Arrow value, offset, and validity buffers in this builder.
Decode FSST, OnPair, Zstd, Dict, RunEnd, and Sparse arrays directly into this builder.
Do not create an intermediate VarBinViewArray for these paths.

Interface

Add an optional Arrow type to the Rust and Python array APIs.
Add an optional Arrow schema to the Python file API.
Keep Arrow view arrays as the default output.

Result

Arrow uses the builder buffers without another copy.
Callers can select StringArray, BinaryArray, LargeStringArray, or LargeBinaryArray.

Signed-off-by: cl <cailue@apache.org>

Author: ClSlaid

encodings/fsst/src/array.rs

@@ -11,6 +11,7 @@ use std::sync::OnceLock;
 use fsst::Compressor;
 use fsst::Decompressor;
 use fsst::Symbol;
+use num_traits::AsPrimitive;
 use prost::Message as _;
 use vortex_array::Array;
 use vortex_array::ArrayEq;
@@ -32,11 +33,13 @@ use vortex_array::arrays::VarBinArray;
 use vortex_array::arrays::varbin::VarBinArrayExt;
 use vortex_array::buffer::BufferHandle;
 use vortex_array::builders::ArrayBuilder;
+use vortex_array::builders::VarBinBufferBuilder;
 use vortex_array::builders::VarBinViewBuilder;
 use vortex_array::dtype::DType;
 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::serde::ArrayChildren;
 use vortex_array::smallvec::smallvec;
 use vortex_array::validity::Validity;
@@ -56,6 +59,7 @@ use vortex_session::VortexSession;
 use vortex_session::registry::CachedId;
 
 use crate::canonical::canonicalize_fsst;
+use crate::canonical::fsst_decode_bytes;
 use crate::canonical::fsst_decode_views;
 use crate::rules::RULES;
 
@@ -304,6 +308,25 @@ impl VTable for FSST {
         builder: &mut dyn ArrayBuilder,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<()> {
+        if let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinBufferBuilder>() {
+            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()
+                        .map(|length| AsPrimitive::<usize>::as_(*length)),
+                    &validity,
+                );
+            });
+            return Ok(());
+        }
+
         let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
             return array
                 .array()

encodings/fsst/src/canonical.rs

@@ -18,6 +18,7 @@ use vortex_buffer::Buffer;
 use vortex_buffer::ByteBuffer;
 use vortex_buffer::ByteBufferMut;
 use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
 
 use crate::FSST;
 use crate::FSSTArrayExt;
@@ -40,20 +41,11 @@ pub(super) fn canonicalize_fsst(
     })
 }
 
-pub(crate) fn fsst_decode_views(
+pub(crate) fn fsst_decode_bytes(
     fsst_array: ArrayView<'_, FSST>,
-    start_buf_index: u32,
     ctx: &mut ExecutionCtx,
-) -> VortexResult<(Vec<ByteBuffer>, Buffer<BinaryView>)> {
-    // FSSTArray has two child arrays:
-    //  1. A VarBinArray, which holds the string heap of the compressed codes.
-    //  2. An uncompressed_lengths primitive array, storing the length of each original
-    //     string element.
-    // To speed up canonicalization, we can decompress the entire string-heap in a single
-    // call. We then turn our uncompressed_lengths into an offsets buffer
-    // necessary for a VarBinViewArray and construct the canonical array.
+) -> VortexResult<(ByteBufferMut, PrimitiveArray)> {
     let bytes = fsst_array.codes().sliced_bytes();
-
     let uncompressed_lens_array = fsst_array
         .uncompressed_lengths()
         .clone()
@@ -68,14 +60,24 @@ pub(crate) fn fsst_decode_views(
             .sum()
     });
 
-    // Bulk-decompress the entire array.
     let decompressor = fsst_array.decompressor();
     let mut uncompressed_bytes = ByteBufferMut::with_capacity(total_size + 7);
     let len =
         decompressor.decompress_into(bytes.as_slice(), uncompressed_bytes.spare_capacity_mut());
+    vortex_ensure!(
+        len == total_size,
+        "FSST decoded {len} bytes, expected {total_size}"
+    );
     unsafe { uncompressed_bytes.set_len(len) };
+    Ok((uncompressed_bytes, uncompressed_lens_array))
+}
 
-    // Directly create the binary views.
+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,
@@ -97,15 +99,21 @@ mod tests {
     use vortex_array::IntoArray;
     use vortex_array::VortexSessionExecute;
     use vortex_array::arrays::ChunkedArray;
+    use vortex_array::arrays::PrimitiveArray;
     use vortex_array::arrays::VarBinArray;
     use vortex_array::arrays::VarBinViewArray;
+    use vortex_array::arrays::varbin::VarBinArrayExt;
     use vortex_array::builders::ArrayBuilder;
+    use vortex_array::builders::VarBinBufferBuilder;
     use vortex_array::builders::VarBinViewBuilder;
     use vortex_array::dtype::DType;
     use vortex_array::dtype::Nullability;
     use vortex_error::VortexResult;
     use vortex_session::VortexSession;
 
+    use super::fsst_decode_bytes;
+    use crate::FSST;
+    use crate::FSSTArrayExt;
     use crate::fsst_compress;
     use crate::fsst_train_compressor;
 
@@ -199,6 +207,20 @@ mod tests {
                 .collect::<Vec<_>>();
             assert_eq!(data, res2)
         };
+
+        {
+            let mut builder =
+                VarBinBufferBuilder::with_capacity(chunked_arr.dtype().clone(), false, data.len());
+            chunked_arr
+                .into_array()
+                .append_to_builder(&mut builder, &mut ctx)?;
+            let arr = builder.finish_into_varbin();
+            let mask = arr.validity()?.execute_mask(arr.len(), &mut ctx)?;
+            let actual = (0..arr.len())
+                .map(|i| mask.value(i).then(|| arr.bytes_at(i).to_vec()))
+                .collect::<Vec<_>>();
+            assert_eq!(data, actual);
+        }
         Ok(())
     }
 
@@ -225,4 +247,22 @@ mod tests {
         let _result = builder.finish_into_varbinview();
         Ok(())
     }
+
+    #[test]
+    fn rejects_incorrect_uncompressed_lengths() -> VortexResult<()> {
+        let input = VarBinViewArray::from_iter_str(["hello"]).into_array();
+        let mut ctx = SESSION.create_execution_ctx();
+        let encoded = fsst_compress(&input, &fsst_train_compressor(&input, &mut ctx)?, &mut ctx)?;
+        let invalid = FSST::try_new(
+            encoded.dtype().clone(),
+            encoded.symbols().clone(),
+            encoded.symbol_lengths().clone(),
+            encoded.codes(),
+            PrimitiveArray::from_iter([4i32]).into_array(),
+            &mut ctx,
+        )?;
+
+        assert!(fsst_decode_bytes(invalid.as_view(), &mut ctx).is_err());
+        Ok(())
+    }
 }

encodings/onpair/src/array.rs

@@ -8,6 +8,7 @@ use std::hash::Hasher;
 use std::sync::Arc;
 use std::sync::OnceLock;
 
+use num_traits::AsPrimitive;
 use onpair::CompactDictionaryView;
 use prost::Message as _;
 use vortex_array::Array;
@@ -25,10 +26,12 @@ use vortex_array::IntoArray;
 use vortex_array::array_slots;
 use vortex_array::buffer::BufferHandle;
 use vortex_array::builders::ArrayBuilder;
+use vortex_array::builders::VarBinBufferBuilder;
 use vortex_array::builders::VarBinViewBuilder;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::PType;
+use vortex_array::match_each_integer_ptype;
 use vortex_array::serde::ArrayChildren;
 use vortex_array::validity::Validity;
 use vortex_array::vtable::VTable;
@@ -46,6 +49,7 @@ use vortex_session::VortexSession;
 use vortex_session::registry::CachedId;
 
 use crate::canonical::canonicalize_onpair;
+use crate::canonical::onpair_decode_bytes;
 use crate::canonical::onpair_decode_views;
 use crate::decode::collect_widened;
 use crate::rules::RULES;
@@ -544,6 +548,25 @@ impl VTable for OnPair {
         builder: &mut dyn ArrayBuilder,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<()> {
+        if let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinBufferBuilder>() {
+            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()
+                        .map(|length| AsPrimitive::<usize>::as_(*length)),
+                    &validity,
+                );
+            });
+            return Ok(());
+        }
+
         let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
             return array
                 .array()

encodings/onpair/src/canonical.rs

@@ -44,11 +44,10 @@ pub(super) fn canonicalize_onpair(
     })
 }
 
-pub(crate) fn onpair_decode_views(
+pub(crate) fn onpair_decode_bytes(
     array: ArrayView<'_, OnPair>,
-    start_buf_index: u32,
     ctx: &mut ExecutionCtx,
-) -> VortexResult<(Vec<ByteBuffer>, Buffer<BinaryView>)> {
+) -> VortexResult<(ByteBufferMut, PrimitiveArray)> {
     let lengths = array
         .uncompressed_lengths()
         .clone()
@@ -62,15 +61,6 @@ pub(crate) fn onpair_decode_views(
             .sum()
     });
 
-    // `codes_offsets` holds the per-row code boundaries and may itself be a
-    // sliced or filtered view of the original. Its first and last entries
-    // bound the contiguous run of `codes` belonging to the rows present in
-    // this array: `slice` keeps the full `codes` child and only narrows
-    // `codes_offsets` (so `code_start > 0` and/or `code_end < codes.len()`),
-    // while `filter` rebuilds both children so the window is the whole stream.
-    // OnPair has no `TakeExecute`, so a reordering take is served from the
-    // canonical `VarBinView` and never reaches this path. We only need those
-    // two boundaries, so point-look them up rather than decoding every offset.
     let codes_offsets = array.codes_offsets();
     let code_start = code_boundary_at(codes_offsets, 0, ctx)?;
     let code_end = code_boundary_at(codes_offsets, array.len(), ctx)?;
@@ -85,13 +75,8 @@ pub(crate) fn onpair_decode_views(
         array.codes().len()
     );
 
-    // Slice the `codes` child to that window *before* unpacking it, so a sliced
-    // array materialises only its own codes rather than the whole column's. The
-    // contiguous decoder walks `codes` in order and never reads the per-row
-    // boundaries, so an empty boundary slice is sound.
     let codes = collect_widened::<u16>(&array.codes().slice(code_start..code_end)?, ctx)?;
     let dict = dict_view(array, ctx)?;
-
     let mut out_bytes = ByteBufferMut::with_capacity(total_size);
     let written =
         match onpair::try_decode_into(codes.as_slice(), dict, out_bytes.spare_capacity_mut()) {
@@ -105,9 +90,16 @@ pub(crate) fn onpair_decode_views(
             "OnPair codes decoded to {written} bytes but uncompressed_lengths records {total_size}"
         );
     }
-    // SAFETY: `try_decode_into` initialised exactly `written` bytes.
     unsafe { out_bytes.set_len(written) };
+    Ok((out_bytes, 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,

encodings/onpair/src/tests.rs

@@ -16,6 +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::VarBinBufferBuilder;
 use vortex_array::builtins::ArrayBuiltins;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
@@ -60,6 +61,19 @@ fn sample_input() -> VarBinArray {
     )
 }
 
+#[test]
+fn 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 = VarBinBufferBuilder::with_capacity(input.dtype().clone(), false, input.len());
+    encoded
+        .into_array()
+        .append_to_builder(&mut builder, &mut ctx)?;
+    assert_arrays_eq!(builder.finish_into_varbin(), input, &mut ctx);
+    Ok(())
+}
+
 #[test]
 fn test_onpair_rejects_100k_token_dictionary() -> vortex_error::VortexResult<()> {
     let num_tokens = 100_000usize;