Add StructArray and RunArray benchmark tests to with_hashes (#20182)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #20181 

## Rationale for this change

<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
Issue #20152 shows some areas of optimization for `RunArray` and
`StructArray` hashing. But the existing `with_hashes` benchmark tests
don't include coverage for these!


## What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
Added benchmarks to `with_hashes.rs`:
  - **StructArray**: 4-column struct (bool, int32, int64, string)
  - **RunArray**: Int32 run-encoded array
  - Both include single/multiple columns and with/without nulls

## Are these changes tested?

<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

No additional tests added, but the benchmarks both compile and run. 

<details>
<summary>a sample run:</summary>

```
❯ cargo bench --features=parquet --bench with_hashes -- array
   Compiling datafusion-common v52.1.0 (/Users/notashes/dev/datafusion/datafusion/common)
    Finished `bench` profile [optimized] target(s) in 34.49s
     Running benches/with_hashes.rs (target/release/deps/with_hashes-2f180744d22084f3)
Gnuplot not found, using plotters backend
struct_array: single, no nulls
                        time:   [38.389 µs 38.437 µs 38.485 µs]
Found 5 outliers among 100 measurements (5.00%)
  1 (1.00%) low severe
  2 (2.00%) low mild
  2 (2.00%) high mild

struct_array: single, nulls
                        time:   [46.108 µs 46.197 µs 46.291 µs]
Found 4 outliers among 100 measurements (4.00%)
  3 (3.00%) high mild
  1 (1.00%) high severe

struct_array: multiple, no nulls
                        time:   [114.64 µs 114.79 µs 114.93 µs]
Found 4 outliers among 100 measurements (4.00%)
  1 (1.00%) low severe
  2 (2.00%) low mild
  1 (1.00%) high mild

struct_array: multiple, nulls
                        time:   [138.29 µs 138.62 µs 139.07 µs]
Found 8 outliers among 100 measurements (8.00%)
  1 (1.00%) low severe
  4 (4.00%) low mild
  1 (1.00%) high mild
  2 (2.00%) high severe

run_array_int32: single, no nulls
                        time:   [1.8777 µs 1.9098 µs 1.9457 µs]
Found 3 outliers among 100 measurements (3.00%)
  3 (3.00%) high mild

run_array_int32: single, nulls
                        time:   [2.0110 µs 2.0417 µs 2.0751 µs]
Found 7 outliers among 100 measurements (7.00%)
  6 (6.00%) high mild
  1 (1.00%) high severe

run_array_int32: multiple, no nulls
                        time:   [5.0511 µs 5.0603 µs 5.0693 µs]
Found 6 outliers among 100 measurements (6.00%)
  1 (1.00%) low mild
  5 (5.00%) high mild

run_array_int32: multiple, nulls
                        time:   [5.6052 µs 5.6201 µs 5.6353 µs]
Found 4 outliers among 100 measurements (4.00%)
  3 (3.00%) high mild
  1 (1.00%) high severe
  ```


</details>

## Are there any user-facing changes?

<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->

Author: notashes

datafusion/common/benches/with_hashes.rs

@@ -20,10 +20,13 @@
 use ahash::RandomState;
 use arrow::array::{
     Array, ArrayRef, ArrowPrimitiveType, DictionaryArray, GenericStringArray,
-    NullBufferBuilder, OffsetSizeTrait, PrimitiveArray, StringViewArray, make_array,
+    NullBufferBuilder, OffsetSizeTrait, PrimitiveArray, RunArray, StringViewArray,
+    StructArray, make_array,
 };
 use arrow::buffer::NullBuffer;
-use arrow::datatypes::{ArrowDictionaryKeyType, Int32Type, Int64Type};
+use arrow::datatypes::{
+    ArrowDictionaryKeyType, DataType, Field, Fields, Int32Type, Int64Type,
+};
 use criterion::{Bencher, Criterion, criterion_group, criterion_main};
 use datafusion_common::hash_utils::with_hashes;
 use rand::Rng;
@@ -37,6 +40,7 @@ const BATCH_SIZE: usize = 8192;
 struct BenchData {
     name: &'static str,
     array: ArrayRef,
+    supports_nulls: bool,
 }
 
 fn criterion_benchmark(c: &mut Criterion) {
@@ -47,50 +51,74 @@ fn criterion_benchmark(c: &mut Criterion) {
         BenchData {
             name: "int64",
             array: primitive_array::<Int64Type>(BATCH_SIZE),
+            supports_nulls: true,
         },
         BenchData {
             name: "utf8",
             array: pool.string_array::<i32>(BATCH_SIZE),
+            supports_nulls: true,
         },
         BenchData {
             name: "large_utf8",
             array: pool.string_array::<i64>(BATCH_SIZE),
+            supports_nulls: true,
         },
         BenchData {
             name: "utf8_view",
             array: pool.string_view_array(BATCH_SIZE),
+            supports_nulls: true,
         },
         BenchData {
             name: "utf8_view (small)",
             array: small_pool.string_view_array(BATCH_SIZE),
+            supports_nulls: true,
         },
         BenchData {
             name: "dictionary_utf8_int32",
             array: pool.dictionary_array::<Int32Type>(BATCH_SIZE),
+            supports_nulls: true,
+        },
+        BenchData {
+            name: "struct_array",
+            array: create_struct_array(&pool, BATCH_SIZE),
+            supports_nulls: true,
+        },
+        BenchData {
+            name: "run_array_int32",
+            array: create_run_array::<Int32Type>(BATCH_SIZE),
+            supports_nulls: true,
         },
     ];
 
-    for BenchData { name, array } in cases {
-        // with_hash has different code paths for single vs multiple arrays and nulls vs no nulls
-        let nullable_array = add_nulls(&array);
+    for BenchData {
+        name,
+        array,
+        supports_nulls,
+    } in cases
+    {
         c.bench_function(&format!("{name}: single, no nulls"), |b| {
             do_hash_test(b, std::slice::from_ref(&array));
         });
-        c.bench_function(&format!("{name}: single, nulls"), |b| {
-            do_hash_test(b, std::slice::from_ref(&nullable_array));
-        });
         c.bench_function(&format!("{name}: multiple, no nulls"), |b| {
             let arrays = vec![array.clone(), array.clone(), array.clone()];
             do_hash_test(b, &arrays);
         });
-        c.bench_function(&format!("{name}: multiple, nulls"), |b| {
-            let arrays = vec![
-                nullable_array.clone(),
-                nullable_array.clone(),
-                nullable_array.clone(),
-            ];
-            do_hash_test(b, &arrays);
-        });
+
+        if supports_nulls {
+            let nullable_array = add_nulls(&array);
+
+            c.bench_function(&format!("{name}: single, nulls"), |b| {
+                do_hash_test(b, std::slice::from_ref(&nullable_array));
+            });
+            c.bench_function(&format!("{name}: multiple, nulls"), |b| {
+                let arrays = vec![
+                    nullable_array.clone(),
+                    nullable_array.clone(),
+                    nullable_array.clone(),
+                ];
+                do_hash_test(b, &arrays);
+            });
+        }
     }
 }
 
@@ -122,16 +150,51 @@ where
     builder.finish().expect("should be nulls in buffer")
 }
 
-// Returns an new array that is the same as array, but with nulls
+// Returns a new array that is the same as array, but with nulls
+// Handles the special case of RunArray where nulls must be in the values array
 fn add_nulls(array: &ArrayRef) -> ArrayRef {
-    let array_data = array
-        .clone()
-        .into_data()
-        .into_builder()
-        .nulls(Some(create_null_mask(array.len())))
-        .build()
-        .unwrap();
-    make_array(array_data)
+    use arrow::datatypes::DataType;
+
+    match array.data_type() {
+        DataType::RunEndEncoded(_, _) => {
+            // RunArray can't have top-level nulls, so apply nulls to the values array
+            let run_array = array
+                .as_any()
+                .downcast_ref::<RunArray<Int32Type>>()
+                .expect("Expected RunArray");
+
+            let run_ends_buffer = run_array.run_ends().inner().clone();
+            let run_ends_array = PrimitiveArray::<Int32Type>::new(run_ends_buffer, None);
+            let values = run_array.values().clone();
+
+            // Add nulls to the values array
+            let values_with_nulls = {
+                let array_data = values
+                    .clone()
+                    .into_data()
+                    .into_builder()
+                    .nulls(Some(create_null_mask(values.len())))
+                    .build()
+                    .unwrap();
+                make_array(array_data)
+            };
+
+            Arc::new(
+                RunArray::try_new(&run_ends_array, values_with_nulls.as_ref())
+                    .expect("Failed to create RunArray with null values"),
+            )
+        }
+        _ => {
+            let array_data = array
+                .clone()
+                .into_data()
+                .into_builder()
+                .nulls(Some(create_null_mask(array.len())))
+                .build()
+                .unwrap();
+            make_array(array_data)
+        }
+    }
 }
 
 pub fn make_rng() -> StdRng {
@@ -205,5 +268,66 @@ where
     Arc::new(array)
 }
 
+fn boolean_array(array_len: usize) -> ArrayRef {
+    let mut rng = make_rng();
+    Arc::new(
+        (0..array_len)
+            .map(|_| Some(rng.random::<bool>()))
+            .collect::<arrow::array::BooleanArray>(),
+    )
+}
+
+/// Create a StructArray with multiple columns
+fn create_struct_array(pool: &StringPool, array_len: usize) -> ArrayRef {
+    let bool_array = boolean_array(array_len);
+    let int32_array = primitive_array::<Int32Type>(array_len);
+    let int64_array = primitive_array::<Int64Type>(array_len);
+    let str_array = pool.string_array::<i32>(array_len);
+
+    let fields = Fields::from(vec![
+        Field::new("bool_col", DataType::Boolean, false),
+        Field::new("int32_col", DataType::Int32, false),
+        Field::new("int64_col", DataType::Int64, false),
+        Field::new("string_col", DataType::Utf8, false),
+    ]);
+
+    Arc::new(StructArray::new(
+        fields,
+        vec![bool_array, int32_array, int64_array, str_array],
+        None,
+    ))
+}
+
+/// Create a RunArray to test run array hashing.
+fn create_run_array<T>(array_len: usize) -> ArrayRef
+where
+    T: ArrowPrimitiveType,
+    StandardUniform: Distribution<T::Native>,
+{
+    let mut rng = make_rng();
+
+    // Create runs of varying lengths
+    let mut run_ends = Vec::new();
+    let mut values = Vec::new();
+    let mut current_end = 0;
+
+    while current_end < array_len {
+        // Random run length between 1 and 50
+        let run_length = rng.random_range(1..=50).min(array_len - current_end);
+        current_end += run_length;
+        run_ends.push(current_end as i32);
+        values.push(Some(rng.random::<T::Native>()));
+    }
+
+    let run_ends_array = Arc::new(PrimitiveArray::<Int32Type>::from(run_ends));
+    let values_array: Arc<dyn Array> =
+        Arc::new(values.into_iter().collect::<PrimitiveArray<T>>());
+
+    Arc::new(
+        RunArray::try_new(&run_ends_array, values_array.as_ref())
+            .expect("Failed to create RunArray"),
+    )
+}
+
 criterion_group!(benches, criterion_benchmark);
 criterion_main!(benches);