feat: Use buffers for opaque data in VarBin and VarBinView (#1935)

Author: AdamGS

bench-vortex/benches/bytes_at.rs

@@ -16,7 +16,7 @@ use vortex::{Context, IntoArrayData, IntoArrayVariant};
 fn array_data_fixture() -> VarBinArray {
     VarBinArray::try_new(
         buffer![0i32, 5i32, 10i32, 15i32, 20i32].into_array(),
-        ByteBuffer::copy_from(b"helloworldhelloworld".as_bytes()).into_array(),
+        ByteBuffer::copy_from(b"helloworldhelloworld".as_bytes()),
         DType::Utf8(Nullability::NonNullable),
         Validity::NonNullable,
     )

docs/quickstart.rst

@@ -35,7 +35,7 @@ Vortex array:
    >>> parquet = pq.read_table("_static/example.parquet")
    >>> vtx = vortex.array(parquet)
    >>> vtx.nbytes
-   141070
+   141069
 
 Compress
 ^^^^^^^^
@@ -46,7 +46,7 @@ Use :func:`~vortex.encoding.compress` to compress the Vortex array and check the
 
    >>> cvtx = vortex.compress(vtx)
    >>> cvtx.nbytes
-   16605
+   16604
    >>> cvtx.nbytes / vtx.nbytes
    0.11...
 

encodings/dict/src/compress.rs

@@ -153,13 +153,8 @@ fn dict_encode_varbin_bytes<'a, I: Iterator<Item = Option<&'a [u8]>>>(
     let values_validity = dict_values_validity(dtype.is_nullable(), offsets.len() - 1);
     (
         PrimitiveArray::new(codes, Validity::NonNullable),
-        VarBinArray::try_new(
-            offsets.into_array(),
-            bytes.into_array(),
-            dtype,
-            values_validity,
-        )
-        .vortex_expect("Failed to create VarBinArray dictionary during encoding"),
+        VarBinArray::try_new(offsets.into_array(), bytes.freeze(), dtype, values_validity)
+            .vortex_expect("Failed to create VarBinArray dictionary during encoding"),
     )
 }
 

encodings/fsst/src/canonical.rs

@@ -1,9 +1,7 @@
 use arrow_array::builder::make_view;
 use vortex_array::array::{BinaryView, VarBinArray, VarBinViewArray};
 use vortex_array::variants::PrimitiveArrayTrait;
-use vortex_array::{
-    ArrayDType, ArrayLen, Canonical, IntoArrayData, IntoArrayVariant, IntoCanonical,
-};
+use vortex_array::{ArrayDType, ArrayLen, Canonical, IntoCanonical};
 use vortex_buffer::{BufferMut, ByteBuffer};
 use vortex_dtype::match_each_integer_ptype;
 use vortex_error::VortexResult;
@@ -23,12 +21,10 @@ impl IntoCanonical for FSSTArray {
             // call. We then turn our uncompressed_lengths into an offsets buffer
             // necessary for a VarBinViewArray and construct the canonical array.
 
-            let compressed_bytes = VarBinArray::try_from(self.codes())?
-                .sliced_bytes()?
-                .into_primitive()?;
+            let bytes = VarBinArray::try_from(self.codes())?.sliced_bytes();
 
             // Bulk-decompress the entire array.
-            let uncompressed_bytes = decompressor.decompress(compressed_bytes.as_slice::<u8>());
+            let uncompressed_bytes = decompressor.decompress(bytes.as_slice());
 
             let uncompressed_lens_array = self
                 .uncompressed_lengths()
@@ -54,7 +50,7 @@ impl IntoCanonical for FSSTArray {
             });
 
             let views = views.freeze();
-            let uncompressed_bytes_array = ByteBuffer::from(uncompressed_bytes).into_array();
+            let uncompressed_bytes_array = ByteBuffer::from(uncompressed_bytes);
 
             VarBinViewArray::try_new(
                 views,

vortex-array/src/array/chunked/canonical.rs

@@ -244,16 +244,12 @@ fn pack_views(
         // merged buffers list.
         let buffers_offset = u32::try_from(buffers.len())?;
         let canonical_chunk = chunk.clone().into_varbinview()?;
+        buffers.extend(canonical_chunk.buffers());
 
-        for buffer in canonical_chunk.buffers() {
-            let canonical_buffer = buffer.into_canonical()?.into_primitive()?.into_array();
-            buffers.push(canonical_buffer);
-        }
-
-        for view in canonical_chunk.binary_views()? {
+        for view in canonical_chunk.views().iter() {
             if view.is_inlined() {
                 // Inlined views can be copied directly into the output
-                views.push(view);
+                views.push(*view);
             } else {
                 // Referencing views must have their buffer_index adjusted with new offsets
                 let view_ref = view.as_view();

vortex-array/src/array/constant/canonical.rs

@@ -81,10 +81,7 @@ fn canonical_byte_view(
             let view = BinaryView::from(make_view(scalar_bytes, 0, 0));
             let mut buffers = Vec::new();
             if scalar_bytes.len() >= BinaryView::MAX_INLINED_SIZE {
-                buffers.push(
-                    PrimitiveArray::new(Buffer::copy_from(scalar_bytes), Validity::NonNullable)
-                        .into_array(),
-                );
+                buffers.push(Buffer::copy_from(scalar_bytes));
             }
 
             // Clone our constant view `len` times.

vortex-array/src/array/varbin/accessor.rs

@@ -12,14 +12,15 @@ impl ArrayAccessor<[u8]> for VarBinArray {
     where
         F: for<'a> FnOnce(&mut (dyn Iterator<Item = Option<&'a [u8]>>)) -> R,
     {
-        // TODO(ngates): what happens if bytes is much larger than sliced_bytes?
-        let primitive = self.bytes().into_primitive()?;
         let offsets = self.offsets().into_primitive()?;
         let validity = self.logical_validity().to_null_buffer()?;
 
+        // TODO(ngates): what happens if bytes is much larger than sliced_bytes?
+        let bytes = self.bytes();
+        let bytes = bytes.as_slice();
+
         match_each_integer_ptype!(offsets.ptype(), |$T| {
             let offsets = offsets.as_slice::<$T>();
-            let bytes = primitive.as_slice::<u8>();
 
             match validity {
                 None => {

vortex-array/src/array/varbin/array.rs

@@ -19,7 +19,7 @@ impl ValidityVTable<VarBinArray> for VarBinEncoding {
 impl VisitorVTable<VarBinArray> for VarBinEncoding {
     fn accept(&self, array: &VarBinArray, visitor: &mut dyn ArrayVisitor) -> VortexResult<()> {
         visitor.visit_child("offsets", &array.offsets())?;
-        visitor.visit_child("bytes", &array.bytes())?;
+        visitor.visit_buffer(&array.bytes())?;
         visitor.visit_validity(&array.validity())
     }
 }

vortex-array/src/array/varbin/arrow.rs

@@ -31,29 +31,22 @@ pub(crate) fn varbin_to_arrow(varbin_array: &VarBinArray) -> VortexResult<ArrayR
         .to_null_buffer()
         .map_err(|err| err.with_context("Failed to get null buffer from logical validity"))?;
 
-    let data = varbin_array
-        .bytes()
-        .into_primitive()
-        .map_err(|err| err.with_context("Failed to canonicalize bytes"))?;
-    if data.dtype() != &DType::BYTES {
-        vortex_bail!("Expected bytes to be of type U8, got {}", data.ptype());
-    }
-    let data = data.byte_buffer();
+    let data = varbin_array.bytes();
 
     // Switch on Arrow DType.
     Ok(match varbin_array.dtype() {
         DType::Binary(_) => match offsets.ptype() {
             PType::I32 => Arc::new(unsafe {
                 BinaryArray::new_unchecked(
                     offsets.buffer::<i32>().into_arrow_offset_buffer(),
-                    data.clone().into_arrow_buffer(),
+                    data.into_arrow_buffer(),
                     nulls,
                 )
             }),
             PType::I64 => Arc::new(unsafe {
                 LargeBinaryArray::new_unchecked(
                     offsets.buffer::<i64>().into_arrow_offset_buffer(),
-                    data.clone().into_arrow_buffer(),
+                    data.into_arrow_buffer(),
                     nulls,
                 )
             }),
@@ -63,14 +56,14 @@ pub(crate) fn varbin_to_arrow(varbin_array: &VarBinArray) -> VortexResult<ArrayR
             PType::I32 => Arc::new(unsafe {
                 StringArray::new_unchecked(
                     offsets.buffer::<i32>().into_arrow_offset_buffer(),
-                    data.clone().into_arrow_buffer(),
+                    data.into_arrow_buffer(),
                     nulls,
                 )
             }),
             PType::I64 => Arc::new(unsafe {
                 LargeStringArray::new_unchecked(
                     offsets.buffer::<i64>().into_arrow_offset_buffer(),
-                    data.clone().into_arrow_buffer(),
+                    data.into_arrow_buffer(),
                     nulls,
                 )
             }),

vortex-array/src/array/varbin/builder.rs

@@ -80,7 +80,6 @@ impl<O: NativePType + PrimInt> VarBinBuilder<O> {
 
     pub fn finish(mut self, dtype: DType) -> VarBinArray {
         let offsets = PrimitiveArray::new(self.offsets.freeze(), Validity::NonNullable);
-        let data = PrimitiveArray::new(self.data.freeze(), Validity::NonNullable);
         let nulls = self.validity.finish();
 
         let validity = if dtype.is_nullable() {
@@ -90,7 +89,7 @@ impl<O: NativePType + PrimInt> VarBinBuilder<O> {
             Validity::NonNullable
         };
 
-        VarBinArray::try_new(offsets.into_array(), data.into_array(), dtype, validity)
+        VarBinArray::try_new(offsets.into_array(), self.data.freeze(), dtype, validity)
             .vortex_expect("Unexpected error while building VarBinArray")
     }
 }