perf: optimize map_sort for multi-entry string maps (up to 3x faster)

docs/source/contributor-guide/expression-audits/map_funcs.md

@@ -66,6 +66,7 @@
 
 ## map_sort
 
+- Performance (tuned locally 2026-09-13; [PR #5901](https://github.com/apache/datafusion-comet/pull/5901), related to [#5900](https://github.com/apache/datafusion-comet/issues/5900)): reuse per-batch prefix-tuple sorting scratch for multi-entry `Utf8`/`Int32` maps and bulk-fill all-empty offsets, preserving the singleton path from [#5887](https://github.com/apache/datafusion-comet/pull/5887). Against upstream including #5887, matched 2–10-entry forward normalization was about 3x faster; 2–50-entry maps improved 28–39% in the full run and 33–39% in independent paired confirmation. Benchmarks: `native/spark-expr/benches/map_sort.rs`, `hash.rs`, and `common/matched_maps.rs`.
 - Performance (tuned locally 2026-09-12; [PR #5887](https://github.com/apache/datafusion-comet/pull/5887)): skip Arrow sort dispatch for eligible flat singleton keys, with a batch check and specialized fallback loop for batches without singletons. In the local DataFusion 55.0.0 development cohort, matched singleton normalization measured 19–22x faster in the full run and 18.4x in an independent forward-order confirmation. Benchmarks: `native/spark-expr/benches/map_sort.rs`, `hash.rs`, and `common/matched_maps.rs`; 92 cases cover normalization, hashing, combined execution, nulls, slices, mixed cardinalities, and long Unicode values. Flagged regressions did not remain stable through independent and reversed-order confirmation.
 
 ## map_values

native/spark-expr/benches/common/matched_maps.rs

@@ -379,3 +379,120 @@ pub fn bench_regression_maps(c: &mut Criterion, stage: Stage) {
     }
     group.finish();
 }
+
+// Int-value variants retain the shared regression fixture conventions.
+pub fn bench_multi_entry_int_maps(c: &mut Criterion, stage: Stage) {
+    let ints: ArrayRef = Arc::new(Int32Array::from_iter_values((0..ROWS).map(c1)));
+    let mut group = c.benchmark_group(format!("multi_entry_int_maps/{}", stage.name()));
+    group.throughput(Throughput::Elements(ROWS as u64));
+    for (name, null_every, mixed, long) in [
+        ("wide_no_null", 0, false, false),
+        ("wide_sparse_null", 100, false, false),
+        ("wide_dense_null", 2, false, false),
+        ("wide_long_unicode", 0, false, true),
+        ("wide_long_unicode_dense_null", 2, false, true),
+        ("mixed_dense_null", 2, true, false),
+    ] {
+        let mut builder = MapBuilder::new(
+            Some(crate::common::map_field_names()),
+            StringBuilder::new(),
+            Int32Builder::new(),
+        );
+        let prefix = if long {
+            "資料é".repeat(128)
+        } else {
+            String::new()
+        };
+        for row in 0..=ROWS {
+            let count = if row == 0 {
+                3 // Sliced-away prefix must have entries, even for mixed cardinalities.
+            } else if mixed {
+                [0, 1, 1, 2, 10, 50][row % 6]
+            } else {
+                2 + c1(row) as usize % 49
+            };
+            for entry in (0..count).rev() {
+                builder.keys().append_value(format!("{prefix}{entry:04}"));
+                if null_every != 0 && (row + entry + 1) % null_every == 0 {
+                    builder.values().append_null();
+                } else {
+                    builder.values().append_value((row * 100 + entry) as i32);
+                }
+            }
+            builder
+                .append(null_every == 0 || row % null_every != 0)
+                .unwrap();
+        }
+        let raw: ArrayRef = Arc::new(builder.finish().slice(1, ROWS));
+        let args = [ColumnarValue::Array(Arc::clone(&raw))];
+        let normalized = normalize(&args);
+        // Independent expected permutation checks values, child nulls, and schema as well
+        // as ordering; explicit checks cover map validity and rebased sliced offsets.
+        let map = raw.as_any().downcast_ref::<MapArray>().unwrap();
+        let keys = map
+            .keys()
+            .as_any()
+            .downcast_ref::<arrow::array::StringArray>()
+            .unwrap();
+        let mut permutation = Vec::new();
+        let mut offsets = vec![0i32];
+        for pair in map.value_offsets().windows(2) {
+            let mut indices: Vec<u32> = (pair[0] as u32..pair[1] as u32).collect();
+            indices.sort_by(|a, b| keys.value(*a as usize).cmp(keys.value(*b as usize)));
+            permutation.extend(indices);
+            offsets.push(permutation.len() as i32);
+        }
+        let expected_entries = arrow::compute::take(
+            map.entries(),
+            &arrow::array::UInt32Array::from(permutation),
+            None,
+        )
+        .unwrap();
+        let actual = normalized.as_any().downcast_ref::<MapArray>().unwrap();
+        assert_eq!(actual.entries().to_data(), expected_entries.to_data());
+        assert_eq!(actual.value_offsets(), offsets);
+        assert_eq!(actual.nulls(), map.nulls());
+        assert_eq!(actual.data_type(), map.data_type());
+        for shape in [Shape::Map, Shape::StructMapInt] {
+            if matches!((stage, shape), (Stage::NormalizeOnly, Shape::StructMapInt)) {
+                continue;
+            }
+            group.bench_function(BenchmarkId::new(shape.name(), name), |b| match stage {
+                Stage::NormalizeOnly => b.iter(|| black_box(normalize(black_box(&args)))),
+                Stage::HashOnly => {
+                    let input = match shape {
+                        Shape::Map => Arc::clone(&normalized),
+                        Shape::StructMapInt => wrap(Arc::clone(&normalized), &ints),
+                    };
+                    let mut buffer = vec![42; ROWS];
+                    b.iter(|| {
+                        buffer.fill(42);
+                        create_murmur3_hashes(std::slice::from_ref(black_box(&input)), &mut buffer)
+                            .unwrap();
+                        black_box(&buffer);
+                    });
+                }
+                Stage::NormalizeHash => {
+                    let mut buffer = vec![42; ROWS];
+                    match shape {
+                        Shape::Map => b.iter(|| {
+                            buffer.fill(42);
+                            let input = normalize(black_box(&args));
+                            create_murmur3_hashes(std::slice::from_ref(&input), &mut buffer)
+                                .unwrap();
+                            black_box(&buffer);
+                        }),
+                        Shape::StructMapInt => b.iter(|| {
+                            buffer.fill(42);
+                            let input = wrap(normalize(black_box(&args)), &ints);
+                            create_murmur3_hashes(std::slice::from_ref(&input), &mut buffer)
+                                .unwrap();
+                            black_box(&buffer);
+                        }),
+                    }
+                }
+            });
+        }
+    }
+    group.finish();
+}

native/spark-expr/benches/hash.rs

@@ -116,7 +116,9 @@ fn bench_matched_maps(c: &mut Criterion) {
     matched_maps::bench_maps(c, matched_maps::Stage::HashOnly);
     matched_maps::bench_maps(c, matched_maps::Stage::NormalizeHash);
     matched_maps::bench_regression_maps(c, matched_maps::Stage::HashOnly);
+    matched_maps::bench_multi_entry_int_maps(c, matched_maps::Stage::HashOnly);
     matched_maps::bench_regression_maps(c, matched_maps::Stage::NormalizeHash);
+    matched_maps::bench_multi_entry_int_maps(c, matched_maps::Stage::NormalizeHash);
     c.bench_function("matched_maps/hash_buffer_seed_reset", |b| {
         let mut hashes = vec![42u32; matched_maps::ROWS];
         b.iter(|| {

native/spark-expr/benches/map_sort.rs

@@ -145,6 +145,7 @@ fn bench_unsupported_singleton(c: &mut Criterion) {
 fn bench_matched_maps(c: &mut Criterion) {
     matched_maps::bench_maps(c, matched_maps::Stage::NormalizeOnly);
     matched_maps::bench_regression_maps(c, matched_maps::Stage::NormalizeOnly);
+    matched_maps::bench_multi_entry_int_maps(c, matched_maps::Stage::NormalizeOnly);
 }
 
 criterion_group!(