Tune varbinview compaction for Arrow export (#8585)

## Summary

Compact `VarBinViewArray` before exporting Arrow byte-view arrays so
filtered arrays do not retain unselected backing buffers. This keeps
Arrow `Utf8View` / `BinaryView` exports from holding onto large payload
buffers that no selected row references.

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Author: gatesn

vortex-array/src/arrays/varbinview/compact.rs

@@ -18,6 +18,9 @@ use crate::arrays::varbinview::Ref;
 use crate::builders::ArrayBuilder;
 use crate::builders::VarBinViewBuilder;
 
+const DEFAULT_COMPACTION_THRESHOLD: f64 = 0.5;
+const MIN_RETAINED_BYTES_PER_ROW_TO_CHECK_COMPACTION: u64 = 128;
+
 impl VarBinViewArray {
     /// Returns a compacted copy of the input array, where all wasted space has been cleaned up. This
     /// operation can be very expensive, in the worst case copying all existing string data into
@@ -33,8 +36,7 @@ impl VarBinViewArray {
             return Ok(self.clone());
         }
 
-        // Use selective compaction with threshold of 1.0 (compact any buffer with any waste)
-        self.compact_with_threshold(1.0)
+        self.compact_with_threshold(DEFAULT_COMPACTION_THRESHOLD)
     }
 
     fn should_compact(&self) -> VortexResult<bool> {
@@ -50,12 +52,18 @@ impl VarBinViewArray {
             return Ok(true);
         }
 
-        let bytes_referenced: u64 = self.count_referenced_bytes()?;
         let buffer_total_bytes: u64 = self.buffers.iter().map(|buf| buf.len() as u64).sum();
+        if buffer_total_bytes == 0 {
+            return Ok(true);
+        }
 
-        // If there is any wasted space, we want to repack.
-        // This is very aggressive.
-        Ok(bytes_referenced < buffer_total_bytes || buffer_total_bytes == 0)
+        let len = u64::try_from(self.len()).unwrap_or(u64::MAX);
+        if len > 0 && buffer_total_bytes / len <= MIN_RETAINED_BYTES_PER_ROW_TO_CHECK_COMPACTION {
+            return Ok(false);
+        }
+
+        let bytes_referenced: u64 = self.count_referenced_bytes()?;
+        Ok((bytes_referenced as f64 / buffer_total_bytes as f64) < DEFAULT_COMPACTION_THRESHOLD)
     }
 
     /// Iterates over all valid, non-inlined views, calling the provided
@@ -264,8 +272,7 @@ mod tests {
             .execute::<VarBinViewArray>(&mut array_session().create_execution_ctx())
             .unwrap();
 
-        // Optimize the taken array
-        let optimized_array = taken_array.compact_buffers().unwrap();
+        let optimized_array = taken_array.compact_with_threshold(1.0).unwrap();
 
         // The optimized array should have exactly 1 buffer (consolidated)
         assert_eq!(optimized_array.data_buffers().len(), 1);

vortex-array/src/arrow/executor/byte.rs

@@ -80,8 +80,11 @@ mod tests {
     use arrow_array::cast::AsArray;
     use arrow_schema::DataType;
     use rstest::rstest;
+    use vortex_error::VortexResult;
+    use vortex_mask::Mask;
 
     use crate::IntoArray;
+    use crate::LEGACY_SESSION;
     use crate::VortexSessionExecute;
     use crate::array_session;
     use crate::arrow::ArrowArrayExecutor;
@@ -179,4 +182,27 @@ mod tests {
         assert!(arrow.is_null(1));
         assert!(!arrow.is_null(2));
     }
+
+    #[test]
+    fn filtered_utf8_view_export_does_not_retain_unselected_buffers() -> VortexResult<()> {
+        let unselected = "x".repeat(1 << 20);
+        let array =
+            VarBinViewArray::from_iter_str(["selected", unselected.as_str(), unselected.as_str()]);
+        let filtered = array
+            .into_array()
+            .filter(Mask::from_iter([true, false, false]))?;
+
+        let arrow = filtered.execute_arrow(
+            Some(&DataType::Utf8View),
+            &mut LEGACY_SESSION.create_execution_ctx(),
+        )?;
+
+        assert_eq!(arrow.as_string_view().value(0), "selected");
+        assert!(
+            arrow.get_array_memory_size() < unselected.len(),
+            "filtered export retained unselected payload: {} bytes",
+            arrow.get_array_memory_size()
+        );
+        Ok(())
+    }
 }

vortex-array/src/arrow/executor/byte_view.rs

@@ -12,7 +12,6 @@ use vortex_error::VortexResult;
 use crate::ArrayRef;
 use crate::ExecutionCtx;
 use crate::arrays::VarBinViewArray;
-use crate::arrow::executor::validity::to_arrow_null_buffer;
 use crate::arrow::null_buffer::to_null_buffer;
 use crate::builtins::ArrayBuiltins;
 use crate::dtype::DType;
@@ -48,19 +47,8 @@ pub fn execute_varbinview_to_arrow<T: ByteViewType>(
     array: &VarBinViewArray,
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<ArrowArrayRef> {
-    let views =
-        ScalarBuffer::<u128>::from(array.views_handle().as_host().clone().into_arrow_buffer());
-    let buffers: Vec<_> = array
-        .data_buffers()
-        .iter()
-        .map(|buffer| buffer.as_host().clone().into_arrow_buffer())
-        .collect();
-    let nulls = to_arrow_null_buffer(array.validity()?, array.len(), ctx)?;
-
-    // SAFETY: our own VarBinView array is considered safe.
-    Ok(Arc::new(unsafe {
-        GenericByteViewArray::<T>::new_unchecked(views, buffers, nulls)
-    }))
+    let compacted = array.compact_buffers()?;
+    canonical_varbinview_to_arrow::<T>(&compacted, ctx)
 }
 
 pub(super) fn to_arrow_byte_view<T: ByteViewType>(
@@ -73,6 +61,7 @@ pub(super) fn to_arrow_byte_view<T: ByteViewType>(
     // flexible since there's no prescribed nullability in Arrow types.
     let array = array.cast(DType::from_arrow((&T::DATA_TYPE, Nullability::Nullable)))?;
 
+    let array = array.execute::<ArrayRef>(ctx)?;
     let varbinview = array.execute::<VarBinViewArray>(ctx)?;
-    canonical_varbinview_to_arrow::<T>(&varbinview, ctx)
+    execute_varbinview_to_arrow::<T>(&varbinview, ctx)
 }