feat: route compatible xxhash64 args through SparkXxhash64

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

@@ -70,5 +70,6 @@
 - Spark 3.5.8 (audited 2026-05-27): baseline. `XxHash64(children, seed) extends HashExpression[Long]`; produces an xxHash64 hash with a configurable Long seed and `LongType` result. Comet routes via `CometXxHash64` to the native `xxhash64` UDF.
 - Spark 4.0.1 (audited 2026-05-27): semantics unchanged.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
+- Upstream (2026-09-15): Comet's native `xxhash64` UDF delegates compatible arguments at Spark's default seed (`42`) to `datafusion-spark::SparkXxhash64`. Differential tests in `native/spark-expr/src/hash_funcs/xxhash64_diff.rs` compare Comet's kernel against `SparkXxhash64` for primitives, both Decimal128 widths, dictionaries, lists, maps, and nested combinations. The Comet kernel is retained for a non-default seed (`SparkXxhash64` hardcodes 42), `Struct` (upstream does not push a parent null mask into children; see #5753), a `Dictionary` nested in a list/map (upstream restarts those hashes from 42), and `Time64`. `murmur3` is unchanged; `create_xxhash64_hashes` remains for `approx_count_distinct` and the fallback path.
 
 [Spark Expression Support]: ../../user-guide/latest/expressions.md

native/spark-expr/benches/xxhash64.rs

@@ -15,30 +15,53 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! `xxhash64` is the alternative Spark hash (e.g. `xxhash64()` and bucketing). Same shape as the
-//! murmur3 benchmark: a representative multi-column key across row counts and null ratios.
+//! `xxhash64` is the alternative Spark hash (e.g. `xxhash64()` and bucketing). It covers a
+//! representative multi-column key across row counts and null ratios, plus the compatible type
+//! families and fallback paths used by `spark_xxhash64`.
 
+use arrow::array::{
+    ArrayRef, BinaryArray, Decimal128Array, DictionaryArray, Int32Array, StringArray,
+};
+use arrow::datatypes::Int32Type;
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
 use datafusion::common::ScalarValue;
 use datafusion::physical_plan::ColumnarValue;
 use datafusion_comet_spark_expr::spark_xxhash64;
 use std::hint::black_box;
+use std::sync::Arc;
 
 #[path = "common/mod.rs"]
 mod common;
 use common::{f64_array, i64_array, string_array, NULL_RATIOS, ROW_COUNTS};
 
+#[path = "common/hash_shapes.rs"]
+mod hash_shapes;
+
+const TYPE_FAMILY_ROWS: usize = 8_192;
+
+fn seeded_args(arrays: impl IntoIterator<Item = ArrayRef>, seed: i64) -> Vec<ColumnarValue> {
+    arrays
+        .into_iter()
+        .map(ColumnarValue::Array)
+        .chain(std::iter::once(ColumnarValue::Scalar(ScalarValue::Int64(
+            Some(seed),
+        ))))
+        .collect()
+}
+
 fn criterion_benchmark(c: &mut Criterion) {
     let mut group = c.benchmark_group("spark_xxhash64");
     for rows in ROW_COUNTS {
         for (null_ratio, tag) in NULL_RATIOS {
             // Trailing Int64 scalar is the seed; preceding columns are the key being hashed.
-            let args = vec![
-                ColumnarValue::Array(i64_array(rows, null_ratio, |i| i as i64)),
-                ColumnarValue::Array(string_array(rows, null_ratio, |i| format!("k{}", i % 1024))),
-                ColumnarValue::Array(f64_array(rows, null_ratio, |i| i as f64 * 1.5)),
-                ColumnarValue::Scalar(ScalarValue::Int64(Some(42))),
-            ];
+            let args = seeded_args(
+                [
+                    i64_array(rows, null_ratio, |i| i as i64),
+                    string_array(rows, null_ratio, |i| format!("k{}", i % 1024)),
+                    f64_array(rows, null_ratio, |i| i as f64 * 1.5),
+                ],
+                42,
+            );
             group.bench_with_input(
                 BenchmarkId::from_parameter(format!("{rows}/{tag}")),
                 &args,
@@ -47,6 +70,74 @@ fn criterion_benchmark(c: &mut Criterion) {
         }
     }
     group.finish();
+
+    let primitive = i64_array(TYPE_FAMILY_ROWS, 0.0, |i| i as i64);
+    let strings = string_array(TYPE_FAMILY_ROWS, 0.0, |i| format!("value_{}", i % 1024));
+    let binary: ArrayRef = Arc::new(BinaryArray::from_iter_values(
+        (0..TYPE_FAMILY_ROWS).map(|i| format!("bytes_{}", i % 1024).into_bytes()),
+    ));
+    let decimal_narrow: ArrayRef = Arc::new(
+        Decimal128Array::from_iter_values((0..TYPE_FAMILY_ROWS).map(|i| i as i128 * 100))
+            .with_precision_and_scale(10, 2)
+            .unwrap(),
+    );
+    let decimal_wide: ArrayRef = Arc::new(
+        Decimal128Array::from_iter_values(
+            (0..TYPE_FAMILY_ROWS).map(|i| 10_000_000_000_000_000_000i128 + i as i128),
+        )
+        .with_precision_and_scale(38, 10)
+        .unwrap(),
+    );
+    let dictionary_values: ArrayRef = Arc::new(StringArray::from(
+        (0..1024)
+            .map(|i| format!("dictionary_value_{i}"))
+            .collect::<Vec<_>>(),
+    ));
+    let dictionary_keys =
+        Int32Array::from_iter_values((0..TYPE_FAMILY_ROWS).map(|i| (i % 1024) as i32));
+    let dictionary: ArrayRef = Arc::new(
+        DictionaryArray::<Int32Type>::try_new(dictionary_keys, dictionary_values).unwrap(),
+    );
+
+    let type_family_cases = vec![
+        (
+            "compatible/primitive_i64",
+            seeded_args([Arc::clone(&primitive)], 42),
+        ),
+        (
+            "compatible/string_binary",
+            seeded_args([strings, binary], 42),
+        ),
+        (
+            "compatible/decimal128_narrow_wide",
+            seeded_args([decimal_narrow, decimal_wide], 42),
+        ),
+        (
+            "compatible/dictionary_i32_utf8",
+            seeded_args([dictionary], 42),
+        ),
+        (
+            "compatible/list_i32_x10",
+            seeded_args([hash_shapes::list_of_primitive(TYPE_FAMILY_ROWS, 10)], 42),
+        ),
+        (
+            "compatible/map_utf8_i32_x10",
+            seeded_args([hash_shapes::maps(TYPE_FAMILY_ROWS, 10)], 42),
+        ),
+        (
+            "fallback/struct_i32_utf8",
+            seeded_args([hash_shapes::structs(TYPE_FAMILY_ROWS)], 42),
+        ),
+        ("fallback/custom_seed_i64", seeded_args([primitive], 7)),
+    ];
+
+    let mut group = c.benchmark_group("spark_xxhash64_type_families");
+    for (name, args) in type_family_cases {
+        group.bench_with_input(BenchmarkId::from_parameter(name), &args, |b, args| {
+            b.iter(|| black_box(spark_xxhash64(black_box(args)).unwrap()))
+        });
+    }
+    group.finish();
 }
 
 criterion_group!(benches, criterion_benchmark);

native/spark-expr/src/hash_funcs/mod.rs

@@ -18,6 +18,8 @@
 pub mod murmur3;
 pub(super) mod utils;
 mod xxhash64;
+#[cfg(test)]
+mod xxhash64_diff;
 
 pub use murmur3::spark_murmur3_hash;
 pub(crate) use xxhash64::create_xxhash64_hashes;

native/spark-expr/src/hash_funcs/xxhash64.rs

@@ -21,34 +21,53 @@ use twox_hash::XxHash64;
 use datafusion::{
     arrow::{
         array::*,
-        datatypes::{ArrowDictionaryKeyType, ArrowNativeType},
+        datatypes::{ArrowDictionaryKeyType, ArrowNativeType, DataType, Field},
     },
     common::{internal_err, ScalarValue},
+    config::ConfigOptions,
     error::{DataFusionError, Result},
+    logical_expr::{ScalarFunctionArgs, ScalarUDFImpl},
 };
 
 use crate::create_hashes_internal;
 use arrow::array::{Array, ArrayRef, Int64Array};
 use datafusion::physical_plan::ColumnarValue;
-use std::sync::Arc;
+use datafusion_spark::function::hash::xxhash64::SparkXxhash64;
+use std::sync::{Arc, OnceLock};
 
-/// Spark compatible xxhash64 in vectorized execution fashion
+/// Spark's default `XxHash64` seed. `SparkXxhash64` hardcodes this and does not accept a
+/// trailing seed argument, unlike Comet's native UDF (seed is appended by `CometXxHash64`).
+const SPARK_DEFAULT_SEED: i64 = 42;
+
+/// Spark compatible xxhash64 in vectorized execution fashion.
+///
+/// Compatible arguments at the default seed are delegated to `datafusion-spark`'s
+/// `SparkXxhash64`. The Comet kernel is kept for:
+/// - a non-default seed (`SparkXxhash64` always starts from 42)
+/// - `Struct` (and anything containing one): `SparkXxhash64` does not push a parent null
+///   mask into children, so hidden values of a NULL struct would affect the hash
+/// - a `Dictionary` nested in a list/map: `SparkXxhash64` restarts those hashes from 42
+/// - `Time64`, which `SparkXxhash64` does not dispatch
 pub fn spark_xxhash64(args: &[ColumnarValue]) -> Result<ColumnarValue, DataFusionError> {
     let length = args.len();
     let seed = &args[length - 1];
     match seed {
         ColumnarValue::Scalar(ScalarValue::Int64(Some(seed))) => {
             // iterate over the arguments to find out the length of the array
-            let num_rows = args[0..args.len() - 1]
+            let data_args = &args[..length - 1];
+            let num_rows = data_args
                 .iter()
                 .find_map(|arg| match arg {
                     ColumnarValue::Array(array) => Some(array.len()),
                     ColumnarValue::Scalar(_) => None,
                 })
                 .unwrap_or(1);
+            if *seed == SPARK_DEFAULT_SEED && args_compatible_with_spark_xxhash64(data_args) {
+                return invoke_spark_xxhash64(data_args, num_rows);
+            }
             let mut hashes: Vec<u64> = vec![0_u64; num_rows];
             hashes.fill(*seed as u64);
-            let arrays = args[0..args.len() - 1]
+            let arrays = data_args
                 .iter()
                 .map(|arg| match arg {
                     ColumnarValue::Array(array) => Arc::clone(array),
@@ -76,6 +95,64 @@ pub fn spark_xxhash64(args: &[ColumnarValue]) -> Result<ColumnarValue, DataFusio
     }
 }
 
+fn spark_xxhash64_config() -> Arc<ConfigOptions> {
+    static CFG: OnceLock<Arc<ConfigOptions>> = OnceLock::new();
+    Arc::clone(CFG.get_or_init(|| Arc::new(ConfigOptions::default())))
+}
+
+fn invoke_spark_xxhash64(
+    args: &[ColumnarValue],
+    num_rows: usize,
+) -> Result<ColumnarValue, DataFusionError> {
+    let arg_fields = args
+        .iter()
+        .enumerate()
+        .map(|(i, a)| Arc::new(Field::new(format!("arg{i}"), a.data_type(), true)))
+        .collect();
+    SparkXxhash64::new().invoke_with_args(ScalarFunctionArgs {
+        args: args.to_vec(),
+        arg_fields,
+        number_rows: num_rows,
+        return_field: Arc::new(Field::new("xxhash64", DataType::Int64, false)),
+        config_options: spark_xxhash64_config(),
+    })
+}
+
+/// Types whose `SparkXxhash64` hashes are bit-identical to Comet at seed 42.
+///
+/// `in_list_or_map` is set when walking list/map element types. A dictionary hashed as a
+/// list/map element is sliced to one row per recursive call, and `SparkXxhash64` then
+/// treats it as a first column and restarts from seed 42.
+fn type_compatible_with_spark_xxhash64(dt: &DataType, in_list_or_map: bool) -> bool {
+    use DataType::*;
+    match dt {
+        Boolean | Int8 | Int16 | Int32 | Int64 | Float32 | Float64 => true,
+        Utf8 | LargeUtf8 | Binary | LargeBinary | FixedSizeBinary(_) => true,
+        Date32 | Date64 | Timestamp(_, _) => true,
+        Decimal128(_, _) => true,
+        Dictionary(_, value) if !in_list_or_map => {
+            type_compatible_with_spark_xxhash64(value.as_ref(), true)
+        }
+        List(field) | LargeList(field) | FixedSizeList(field, _) => {
+            type_compatible_with_spark_xxhash64(field.data_type(), true)
+        }
+        Map(field, _) => match field.data_type() {
+            Struct(fields) if fields.len() == 2 => fields
+                .iter()
+                .all(|f| type_compatible_with_spark_xxhash64(f.data_type(), true)),
+            _ => false,
+        },
+        // Struct: `SparkXxhash64` hashes child buffers without applying the parent null
+        // mask. Time64 is a Comet-only dispatch arm.
+        _ => false,
+    }
+}
+
+fn args_compatible_with_spark_xxhash64(args: &[ColumnarValue]) -> bool {
+    args.iter()
+        .all(|a| type_compatible_with_spark_xxhash64(&a.data_type(), false))
+}
+
 #[inline]
 fn spark_compatible_xxhash64<T: AsRef<[u8]>>(data: T, seed: u64) -> u64 {
     XxHash64::oneshot(seed, data.as_ref())