perf: optimize slow path and fix benchmark in ByteViewGroupValueBuilder

Three follow-up changes addressing PR review and CI feedback on #21794:

1. Remove `required-features = ["test_utils"]` from the `aggregate_vectorized`
   bench target. The gate caused the bench to be silently skipped when CI ran
   `cargo bench --features=parquet --bench aggregate_vectorized`. The bench
   compiles fine in a workspace build where `arrow/test_utils` is already
   enabled via feature unification through `datafusion-physical-expr`.

2. Replace `arrow::util::test_util::seedable_rng()` with
   `StdRng::seed_from_u64(42)` from the already-present `rand` crate.
   This was the only import from `test_util` (as opposed to `bench_util`)
   and removes the last explicit dependency on the `test_utils` feature
   in the benchmark.

3. Optimize the slow path in `vectorized_append_inner` (non-inline strings,
   `!data_buffers().is_empty()`), addressing Dandandan's review comment.
   Instead of calling `do_append_val_inner` which goes through
   `array.value(row)` (buffer lookup + slice construction) and then
   `make_view` (re-reads the first 4 bytes to build the prefix), the new
   path reads `arr.views()[row]` directly and copies `src.prefix` from the
   source `ByteView`. Benchmarks show 41–53% improvement for 64-byte strings
   and 6–16% improvement for random-length strings (null_density=0.0).

Author: EeshanBembi

datafusion/physical-plan/Cargo.toml

@@ -107,4 +107,3 @@ required-features = ["test_utils"]
 [[bench]]
 harness = false
 name = "aggregate_vectorized"
-required-features = ["test_utils"]

datafusion/physical-plan/benches/aggregate_vectorized.rs

@@ -21,7 +21,6 @@ use arrow::util::bench_util::{
     create_primitive_array, create_string_view_array_with_len,
     create_string_view_array_with_max_len,
 };
-use arrow::util::test_util::seedable_rng;
 use arrow_schema::DataType;
 use criterion::measurement::WallTime;
 use criterion::{
@@ -30,7 +29,9 @@ use criterion::{
 use datafusion_physical_plan::aggregates::group_values::multi_group_by::GroupColumn;
 use datafusion_physical_plan::aggregates::group_values::multi_group_by::bytes_view::ByteViewGroupValueBuilder;
 use datafusion_physical_plan::aggregates::group_values::multi_group_by::primitive::PrimitiveGroupValueBuilder;
+use rand::SeedableRng;
 use rand::distr::{Bernoulli, Distribution};
+use rand::rngs::StdRng;
 use std::hint::black_box;
 use std::sync::Arc;
 
@@ -128,7 +129,7 @@ fn bytes_bench(
         input,
         "0.75 true",
         {
-            let mut rng = seedable_rng();
+            let mut rng = StdRng::seed_from_u64(42);
             let d = Bernoulli::new(0.75).unwrap();
             (0..size).map(|_| d.sample(&mut rng)).collect::<Vec<_>>()
         },
@@ -141,7 +142,7 @@ fn bytes_bench(
         input,
         "0.5 true",
         {
-            let mut rng = seedable_rng();
+            let mut rng = StdRng::seed_from_u64(42);
             let d = Bernoulli::new(0.5).unwrap();
             (0..size).map(|_| d.sample(&mut rng)).collect::<Vec<_>>()
         },
@@ -154,7 +155,7 @@ fn bytes_bench(
         input,
         "0.25 true",
         {
-            let mut rng = seedable_rng();
+            let mut rng = StdRng::seed_from_u64(42);
             let d = Bernoulli::new(0.25).unwrap();
             (0..size).map(|_| d.sample(&mut rng)).collect::<Vec<_>>()
         },
@@ -236,7 +237,7 @@ fn bench_single_primitive<const NULLABLE: bool>(
         &input,
         "0.75 true",
         {
-            let mut rng = seedable_rng();
+            let mut rng = StdRng::seed_from_u64(42);
             let d = Bernoulli::new(0.75).unwrap();
             (0..size).map(|_| d.sample(&mut rng)).collect::<Vec<_>>()
         },
@@ -249,7 +250,7 @@ fn bench_single_primitive<const NULLABLE: bool>(
         &input,
         "0.5 true",
         {
-            let mut rng = seedable_rng();
+            let mut rng = StdRng::seed_from_u64(42);
             let d = Bernoulli::new(0.5).unwrap();
             (0..size).map(|_| d.sample(&mut rng)).collect::<Vec<_>>()
         },
@@ -262,7 +263,7 @@ fn bench_single_primitive<const NULLABLE: bool>(
         &input,
         "0.25 true",
         {
-            let mut rng = seedable_rng();
+            let mut rng = StdRng::seed_from_u64(42);
             let d = Bernoulli::new(0.25).unwrap();
             (0..size).map(|_| d.sample(&mut rng)).collect::<Vec<_>>()
         },

datafusion/physical-plan/src/aggregates/group_values/multi_group_by/bytes_view.rs

@@ -173,10 +173,37 @@ impl<B: ByteViewType> ByteViewGroupValueBuilder<B> {
                     self.views.extend(rows.iter().map(|&row| arr.views()[row]));
                 } else {
                     // Slow path: some strings are non-inline (>12 bytes).
-                    // Pre-reserve views capacity to avoid repeated reallocation.
+                    // Read views directly to avoid array.value(row) overhead and
+                    // reuse the source view's prefix instead of recomputing it via make_view.
                     self.views.reserve(rows.len());
                     for &row in rows {
-                        self.do_append_val_inner(arr, row);
+                        let view = arr.views()[row];
+                        let len = view as u32;
+                        if len <= 12 {
+                            // This row happens to be inline; copy view directly.
+                            self.views.push(view);
+                        } else {
+                            let src = ByteView::from(view);
+                            // ensure_in_progress_big_enough must be called before computing
+                            // new_buffer_index / new_offset — it may flush in_progress to completed.
+                            self.ensure_in_progress_big_enough(len as usize);
+                            let new_buffer_index = self.completed.len() as u32;
+                            let new_offset = self.in_progress.len() as u32;
+                            let src_buf = &arr.data_buffers()[src.buffer_index as usize];
+                            self.in_progress.extend_from_slice(
+                                &src_buf[src.offset as usize
+                                    ..(src.offset + src.length) as usize],
+                            );
+                            // Reuse prefix from the source view — avoids re-reading first 4 bytes.
+                            let new_view = ByteView {
+                                length: src.length,
+                                prefix: src.prefix,
+                                buffer_index: new_buffer_index,
+                                offset: new_offset,
+                            }
+                            .as_u128();
+                            self.views.push(new_view);
+                        }
                     }
                 }
             }