perf: avoid cloning ByteView buffer lists in take and filter

arrow-array/src/array/byte_view_array.rs

@@ -306,6 +306,16 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
         &self.buffers
     }
 
+    /// Returns shared ownership of the buffers storing non-inline values
+    ///
+    /// This operation is `O(1)` and does not clone the individual buffers or
+    /// their contents. See [`Self::data_buffers`] to inspect the buffers
+    /// without taking shared ownership.
+    #[inline]
+    pub fn data_buffers_shared(&self) -> Arc<[Buffer]> {
+        Arc::clone(&self.buffers)
+    }
+
     /// Returns the element at index `i`
     ///
     /// Note: This method does not check for nulls and the value is arbitrary

arrow-select/src/filter.rs

@@ -934,7 +934,7 @@ fn filter_byte_view<T: ByteViewType>(
 ) -> GenericByteViewArray<T> {
     let new_view_buffer = filter_native(array.views(), predicate);
     let views = ScalarBuffer::new(new_view_buffer, 0, predicate.count);
-    let buffers = array.data_buffers().to_vec();
+    let buffers = array.data_buffers_shared();
     let nulls = predicate.filter_nulls(array.nulls());
 
     // SAFETY: each view is copied unchanged from `array.views()` and `buffers`
@@ -1297,6 +1297,9 @@ mod tests {
             let actual = filter(&array, &predicate).unwrap();
 
             assert_eq!(actual.len(), 3);
+            let actual_buffers = actual.as_byte_view::<T>().data_buffers_shared();
+            let input_buffers = array.data_buffers_shared();
+            assert!(Arc::ptr_eq(&actual_buffers, &input_buffers));
 
             let expected = {
                 // ["hello", null, "large payload over 12 bytes"]

arrow-select/src/take.rs

@@ -636,10 +636,9 @@ fn take_byte_view<T: ByteViewType, IndexType: ArrowPrimitiveType>(
 ) -> Result<GenericByteViewArray<T>, ArrowError> {
     let new_views = take_native(array.views(), indices);
     let new_nulls = take_nulls(array.nulls(), indices);
+    let buffers = array.data_buffers_shared();
     // Safety:  array.views was valid, and take_native copies only valid values, and verifies bounds
-    Ok(unsafe {
-        GenericByteViewArray::new_unchecked(new_views, array.data_buffers().to_vec(), new_nulls)
-    })
+    Ok(unsafe { GenericByteViewArray::new_unchecked(new_views, buffers, new_nulls) })
 }
 
 /// `take` implementation for list arrays
@@ -1806,6 +1805,9 @@ mod tests {
         let actual = take(&array, &index, None).unwrap();
 
         assert_eq!(actual.len(), index.len());
+        let actual_buffers = actual.as_byte_view::<T>().data_buffers_shared();
+        let input_buffers = array.data_buffers_shared();
+        assert!(Arc::ptr_eq(&actual_buffers, &input_buffers));
 
         let expected = {
             // ["large payload over 12 bytes", null, "world", "large payload over 12 bytes", "lulu", null]

arrow/benches/take_kernels.rs

@@ -17,10 +17,11 @@
 
 #[macro_use]
 extern crate criterion;
-use criterion::Criterion;
+use criterion::{BenchmarkId, Criterion};
 
 use rand::RngExt;
 
+use arrow::buffer::{Buffer, ScalarBuffer};
 use arrow::compute::{TakeOptions, take, take_record_batch};
 use arrow::datatypes::*;
 use arrow::record_batch::RecordBatch;
@@ -71,6 +72,26 @@ fn bench_take_bounds_check(values: &dyn Array, indices: &UInt32Array) {
     hint::black_box(take(values, indices, Some(TakeOptions { check_bounds: true })).unwrap());
 }
 
+/// Creates an array whose buffer entries all share the same payload allocation.
+///
+/// Views reference every entry in turn, isolating the cost of cloning the buffer
+/// collection from the size of the underlying string payload.
+fn create_string_view_array_with_buffers(size: usize, buffer_count: usize) -> StringViewArray {
+    const VALUE: &[u8] = b"a string longer than twelve bytes";
+
+    let buffer = Buffer::from(VALUE);
+    let buffers = vec![buffer; buffer_count];
+    let views = (0..size)
+        .map(|i| {
+            ByteView::new(VALUE.len() as u32, &VALUE[..4])
+                .with_buffer_index((i % buffer_count) as u32)
+                .as_u128()
+        })
+        .collect::<ScalarBuffer<_>>();
+
+    StringViewArray::new(views, buffers, None)
+}
+
 fn add_benchmark(c: &mut Criterion) {
     let values = create_primitive_array::<Int32Type>(512, 0.0);
     let indices = create_random_index(512, 0.0);
@@ -206,6 +227,18 @@ fn add_benchmark(c: &mut Criterion) {
         b.iter(|| bench_take(&values, &indices))
     });
 
+    let indices = create_random_index(8192, 0.0);
+    let mut group = c.benchmark_group("take stringview by buffer count");
+    for buffer_count in [1, 16, 256, 4096] {
+        let values = create_string_view_array_with_buffers(8192, buffer_count);
+        group.bench_with_input(
+            BenchmarkId::from_parameter(buffer_count),
+            &buffer_count,
+            |b, _| b.iter(|| bench_take(&values, &indices)),
+        );
+    }
+    group.finish();
+
     let values = create_primitive_list_array::<i32, Int32Type>(512, 0.0, 0.0, 20);
     let indices = create_random_index(512, 0.0);
     c.bench_function("take list i32 512", |b| {