From a1152566ecd152420cdbe6dc8f8627037ac95e07 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Tue, 15 Sep 2026 22:22:20 +0800 Subject: [PATCH 1/2] feat: route compatible xxhash64 args through SparkXxhash64 Keep Comet's kernel for custom seeds, structs, nested dictionaries, and Time64. Partial-closes #5103. --- .../expression-audits/hash_funcs.md | 1 + native/spark-expr/src/hash_funcs/mod.rs | 2 + native/spark-expr/src/hash_funcs/xxhash64.rs | 87 +- .../src/hash_funcs/xxhash64_diff.rs | 1143 +++++++++++++++++ .../comet/CometHashExpressionSuite.scala | 90 +- 5 files changed, 1276 insertions(+), 47 deletions(-) create mode 100644 native/spark-expr/src/hash_funcs/xxhash64_diff.rs diff --git a/docs/source/contributor-guide/expression-audits/hash_funcs.md b/docs/source/contributor-guide/expression-audits/hash_funcs.md index 35337b8fe86..989d1d20668 100644 --- a/docs/source/contributor-guide/expression-audits/hash_funcs.md +++ b/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 diff --git a/native/spark-expr/src/hash_funcs/mod.rs b/native/spark-expr/src/hash_funcs/mod.rs index c6eba4a46bf..cca5cd2b6aa 100644 --- a/native/spark-expr/src/hash_funcs/mod.rs +++ b/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; diff --git a/native/spark-expr/src/hash_funcs/xxhash64.rs b/native/spark-expr/src/hash_funcs/xxhash64.rs index 45c273bb9f5..bdda7c85f1d 100644 --- a/native/spark-expr/src/hash_funcs/xxhash64.rs +++ b/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 { 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 = 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 Arc { + static CFG: OnceLock> = OnceLock::new(); + Arc::clone(CFG.get_or_init(|| Arc::new(ConfigOptions::default()))) +} + +fn invoke_spark_xxhash64( + args: &[ColumnarValue], + num_rows: usize, +) -> Result { + 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>(data: T, seed: u64) -> u64 { XxHash64::oneshot(seed, data.as_ref()) diff --git a/native/spark-expr/src/hash_funcs/xxhash64_diff.rs b/native/spark-expr/src/hash_funcs/xxhash64_diff.rs new file mode 100644 index 00000000000..3c51a0e4862 --- /dev/null +++ b/native/spark-expr/src/hash_funcs/xxhash64_diff.rs @@ -0,0 +1,1143 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Differential coverage of Comet `xxhash64` against `datafusion-spark::SparkXxhash64`. +//! +//! `SparkXxhash64` always starts from Spark's default seed (`42`) and hashes every argument; +//! Comet's native UDF takes the seed as a trailing Int64 scalar. Kernel comparisons therefore +//! go through [`create_xxhash64_hashes`] (seed 42) vs `SparkXxhash64::invoke_with_args`. +//! +//! Compatibility at seed 42 (bit-identical): +//! +//! | Type | Compatible | +//! | --- | --- | +//! | Boolean, Int8/16/32/64, Float32/64 | yes | +//! | Utf8, LargeUtf8, Binary, LargeBinary, FixedSizeBinary | yes | +//! | Date32, Date64, Timestamp | yes | +//! | Decimal128 precision ≤ 18 | yes | +//! | Decimal128 precision > 18 | yes | +//! | Dictionary (top-level) | yes | +//! | List / LargeList / FixedSizeList of primitives | yes | +//! | Map<Utf8, Int32> / Map<Int32, Utf8> / Map<Utf8, Utf8> / Map<Int32, Int32> | yes | +//! | Map<Utf8, Decimal128> | yes | +//! | Struct (non-null parent) | yes, but not routed (see below) | +//! | Struct NULL with hidden children | **no** — `SparkXxhash64` hashes hidden children | +//! | List<Dictionary> | **no** — upstream restarts from seed 42 | +//! | Time64(ns) | **no** — upstream does not dispatch | +//! | custom seed | **no** — `SparkXxhash64` hardcodes 42 | + +use super::{create_xxhash64_hashes, spark_xxhash64}; +use arrow::array::builder::{ + Decimal128Builder, Int32Builder, ListBuilder, MapBuilder, StringBuilder, StructBuilder, +}; +use arrow::array::{ + Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, DictionaryArray, FixedSizeBinaryArray, + FixedSizeListArray, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, + LargeBinaryArray, LargeListArray, LargeStringArray, ListArray, StringArray, StructArray, + Time64NanosecondArray, TimestampMicrosecondArray, +}; +use arrow::buffer::{NullBuffer, OffsetBuffer}; +use arrow::datatypes::{DataType, Field, Fields, Int32Type, Int8Type}; +use datafusion::common::{Result, ScalarValue}; +use datafusion::config::ConfigOptions; +use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl}; +use datafusion_spark::function::hash::xxhash64::SparkXxhash64; +use std::sync::Arc; + +const SPARK_DEFAULT_SEED: u64 = 42; + +fn comet_kernel(arrays: &[ArrayRef], seed: u64) -> Result> { + let n = arrays.first().map(|a| a.len()).unwrap_or(0); + let mut hashes = vec![seed; n]; + create_xxhash64_hashes(arrays, &mut hashes)?; + Ok(hashes) +} + +fn columnar_u64s(value: ColumnarValue, n: usize) -> Vec { + match value { + ColumnarValue::Scalar(ScalarValue::Int64(Some(v))) => vec![v as u64; n.max(1)], + ColumnarValue::Array(array) => { + let typed = array + .as_any() + .downcast_ref::() + .expect("xxhash64 result is Int64"); + typed.values().iter().map(|v| *v as u64).collect() + } + other => panic!("unexpected xxhash64 result: {other:?}"), + } +} + +fn spark_xxhash64_upstream(arrays: &[ArrayRef]) -> Result> { + let n = arrays.first().map(|a| a.len()).unwrap_or(1); + let args: Vec = arrays + .iter() + .map(|a| ColumnarValue::Array(Arc::clone(a))) + .collect(); + let arg_fields = arrays + .iter() + .enumerate() + .map(|(i, a)| Arc::new(Field::new(format!("c{i}"), a.data_type().clone(), true))) + .collect(); + let result = SparkXxhash64::new().invoke_with_args(ScalarFunctionArgs { + args, + arg_fields, + number_rows: n, + return_field: Arc::new(Field::new("xxhash64", DataType::Int64, false)), + config_options: Arc::new(ConfigOptions::default()), + })?; + Ok(columnar_u64s(result, n)) +} + +fn comet_expr(arrays: &[ArrayRef], seed: i64) -> Result> { + let n = arrays.first().map(|a| a.len()).unwrap_or(1); + let mut args: Vec = arrays + .iter() + .map(|a| ColumnarValue::Array(Arc::clone(a))) + .collect(); + args.push(ColumnarValue::Scalar(ScalarValue::Int64(Some(seed)))); + Ok(columnar_u64s(spark_xxhash64(&args)?, n)) +} + +/// Compare Comet's kernel (seed 42) with `SparkXxhash64` on the same columns. +fn assert_compatible(label: &str, arrays: &[ArrayRef]) { + let comet = comet_kernel(arrays, SPARK_DEFAULT_SEED) + .unwrap_or_else(|e| panic!("{label}: Comet kernel failed: {e}")); + let upstream = spark_xxhash64_upstream(arrays) + .unwrap_or_else(|e| panic!("{label}: SparkXxhash64 failed: {e}")); + assert_eq!(comet, upstream, "{label}: kernel mismatch"); + let expr = comet_expr(arrays, SPARK_DEFAULT_SEED as i64) + .unwrap_or_else(|e| panic!("{label}: Comet expression failed: {e}")); + assert_eq!(expr, upstream, "{label}: expression mismatch"); +} + +fn col(array: impl Array + 'static) -> Vec { + vec![Arc::new(array) as ArrayRef] +} + +fn list_i32(rows: Vec>>>) -> ArrayRef { + let mut b = ListBuilder::new(Int32Builder::new()); + for row in rows { + match row { + None => b.append(false), + Some(values) => { + for v in values { + match v { + Some(x) => b.values().append_value(x), + None => b.values().append_null(), + } + } + b.append(true); + } + } + } + Arc::new(b.finish()) +} + +fn large_list_i32(rows: Vec>>>) -> ArrayRef { + let mut offsets = vec![0i64]; + let mut values: Vec> = Vec::new(); + let mut validity = Vec::new(); + for row in rows { + match row { + None => { + validity.push(false); + offsets.push(values.len() as i64); + } + Some(elems) => { + validity.push(true); + values.extend(elems); + offsets.push(values.len() as i64); + } + } + } + let values = Int32Array::from(values); + Arc::new(LargeListArray::new( + Arc::new(Field::new("item", DataType::Int32, true)), + OffsetBuffer::new(offsets.into()), + Arc::new(values), + Some(NullBuffer::from(validity)), + )) +} + +fn decimal128(precision: u8, scale: i8, values: Vec>) -> ArrayRef { + let mut b = Decimal128Builder::with_capacity(values.len()) + .with_data_type(DataType::Decimal128(precision, scale)); + for v in values { + match v { + Some(x) => b.append_value(x), + None => b.append_null(), + } + } + Arc::new(b.finish()) +} + +fn struct_ab(a: Vec>, b: Vec>, nulls: Option>) -> ArrayRef { + let fields: Fields = vec![ + Arc::new(Field::new("a", DataType::Int32, true)), + Arc::new(Field::new("b", DataType::Utf8, true)), + ] + .into(); + let children: Vec = vec![ + Arc::new(Int32Array::from(a)), + Arc::new(StringArray::from(b)), + ]; + let nulls = nulls.map(NullBuffer::from); + Arc::new(StructArray::new(fields, children, nulls)) +} + +type Utf8I32Entries = Vec<(&'static str, Option)>; +type I32Utf8Entries = Vec<(i32, Option<&'static str>)>; +type Utf8Utf8Entries = Vec<(&'static str, Option<&'static str>)>; +type I32I32Entries = Vec<(i32, Option)>; +type Utf8DecimalEntries = Vec<(&'static str, Option)>; + +fn map_utf8_i32(rows: Vec>) -> ArrayRef { + let mut mb = MapBuilder::new(None, StringBuilder::new(), Int32Builder::new()); + for row in rows { + match row { + None => { + mb.append(false).unwrap(); + } + Some(entries) => { + for (k, v) in entries { + mb.keys().append_value(k); + match v { + Some(x) => mb.values().append_value(x), + None => mb.values().append_null(), + } + } + mb.append(true).unwrap(); + } + } + } + Arc::new(mb.finish()) +} + +fn map_i32_utf8(rows: Vec>) -> ArrayRef { + let mut mb = MapBuilder::new(None, Int32Builder::new(), StringBuilder::new()); + for row in rows { + match row { + None => { + mb.append(false).unwrap(); + } + Some(entries) => { + for (k, v) in entries { + mb.keys().append_value(k); + match v { + Some(x) => mb.values().append_value(x), + None => mb.values().append_null(), + } + } + mb.append(true).unwrap(); + } + } + } + Arc::new(mb.finish()) +} + +fn map_utf8_utf8(rows: Vec>) -> ArrayRef { + let mut mb = MapBuilder::new(None, StringBuilder::new(), StringBuilder::new()); + for row in rows { + match row { + None => { + mb.append(false).unwrap(); + } + Some(entries) => { + for (k, v) in entries { + mb.keys().append_value(k); + match v { + Some(x) => mb.values().append_value(x), + None => mb.values().append_null(), + } + } + mb.append(true).unwrap(); + } + } + } + Arc::new(mb.finish()) +} + +fn map_i32_i32(rows: Vec>) -> ArrayRef { + let mut mb = MapBuilder::new(None, Int32Builder::new(), Int32Builder::new()); + for row in rows { + match row { + None => { + mb.append(false).unwrap(); + } + Some(entries) => { + for (k, v) in entries { + mb.keys().append_value(k); + match v { + Some(x) => mb.values().append_value(x), + None => mb.values().append_null(), + } + } + mb.append(true).unwrap(); + } + } + } + Arc::new(mb.finish()) +} + +fn map_utf8_decimal(precision: u8, scale: i8, rows: Vec>) -> ArrayRef { + let mut mb = MapBuilder::new( + None, + StringBuilder::new(), + Decimal128Builder::new().with_data_type(DataType::Decimal128(precision, scale)), + ); + for row in rows { + match row { + None => { + mb.append(false).unwrap(); + } + Some(entries) => { + for (k, v) in entries { + mb.keys().append_value(k); + match v { + Some(x) => mb.values().append_value(x), + None => mb.values().append_null(), + } + } + mb.append(true).unwrap(); + } + } + } + Arc::new(mb.finish()) +} + +// --------------------------------------------------------------------------- +// Primitive types +// --------------------------------------------------------------------------- + +#[test] +fn boolean() { + assert_compatible( + "Boolean", + &col(BooleanArray::from(vec![ + Some(true), + Some(false), + None, + Some(true), + ])), + ); +} + +#[test] +fn int8() { + assert_compatible( + "Int8", + &col(Int8Array::from(vec![ + Some(1), + Some(0), + Some(-1), + Some(i8::MAX), + Some(i8::MIN), + None, + ])), + ); +} + +#[test] +fn int16() { + assert_compatible( + "Int16", + &col(Int16Array::from(vec![ + Some(1), + Some(0), + Some(-1), + Some(i16::MAX), + Some(i16::MIN), + None, + ])), + ); +} + +#[test] +fn int32() { + assert_compatible( + "Int32", + &col(Int32Array::from(vec![ + Some(1), + Some(0), + Some(-1), + Some(i32::MAX), + Some(i32::MIN), + None, + ])), + ); +} + +#[test] +fn int64() { + assert_compatible( + "Int64", + &col(Int64Array::from(vec![ + Some(1), + Some(0), + Some(-1), + Some(i64::MAX), + Some(i64::MIN), + None, + ])), + ); +} + +#[test] +fn float32() { + assert_compatible( + "Float32", + &col(Float32Array::from(vec![ + Some(1.0), + Some(0.0), + Some(-0.0), + Some(-1.0), + Some(f32::NAN), + Some(f32::INFINITY), + Some(f32::NEG_INFINITY), + None, + ])), + ); +} + +#[test] +fn float64() { + assert_compatible( + "Float64", + &col(Float64Array::from(vec![ + Some(1.0), + Some(0.0), + Some(-0.0), + Some(-1.0), + Some(f64::NAN), + Some(f64::INFINITY), + Some(f64::NEG_INFINITY), + None, + ])), + ); +} + +#[test] +fn utf8() { + assert_compatible( + "Utf8", + &col(StringArray::from(vec![ + Some("hello"), + Some(""), + Some("😁"), + Some("天地"), + Some("abc"), + None, + ])), + ); +} + +#[test] +fn large_utf8() { + assert_compatible( + "LargeUtf8", + &col(LargeStringArray::from(vec![ + Some("hello"), + Some(""), + Some("😁"), + Some("天地"), + None, + ])), + ); +} + +#[test] +fn binary() { + assert_compatible( + "Binary", + &col(BinaryArray::from_opt_vec(vec![ + Some(b"hello".as_slice()), + Some(b"".as_slice()), + Some(&[0u8, 1, 2][..]), + None, + ])), + ); +} + +#[test] +fn large_binary() { + assert_compatible( + "LargeBinary", + &col(LargeBinaryArray::from_opt_vec(vec![ + Some(b"hello".as_slice()), + Some(b"".as_slice()), + Some(&[0u8, 1, 2][..]), + None, + ])), + ); +} + +#[test] +fn fixed_size_binary() { + let array = FixedSizeBinaryArray::try_from_sparse_iter_with_size( + vec![ + Some(&[0x01, 0x02, 0x03, 0x04][..]), + Some(&[0x00, 0x00, 0x00, 0x00][..]), + None, + ] + .into_iter(), + 4, + ) + .unwrap(); + assert_compatible("FixedSizeBinary", &col(array)); +} + +#[test] +fn date32() { + assert_compatible( + "Date32", + &col(Date32Array::from(vec![ + Some(0), + Some(1), + Some(-1), + Some(i32::MAX), + Some(i32::MIN), + None, + ])), + ); +} + +#[test] +fn date64() { + assert_compatible( + "Date64", + &col(arrow::array::Date64Array::from(vec![ + Some(0), + Some(86_400_000), + Some(-86_400_000), + None, + ])), + ); +} + +#[test] +fn timestamp_microsecond() { + let values = vec![ + Some(0i64), + Some(1), + Some(-1), + Some(i64::MAX), + Some(i64::MIN), + None, + ]; + assert_compatible( + "Timestamp(us, None)", + &col(TimestampMicrosecondArray::from(values.clone())), + ); + let tz = TimestampMicrosecondArray::from(values).with_timezone("UTC"); + assert_compatible("Timestamp(us, UTC)", &col(tz)); +} + +// --------------------------------------------------------------------------- +// Decimal128 (small / large split) +// --------------------------------------------------------------------------- + +#[test] +fn decimal128_precision_10_fits_i64() { + assert_compatible( + "Decimal128(10,2)", + &col(decimal128( + 10, + 2, + vec![Some(0), Some(123), Some(-123), Some(9_999_999_999), None], + )), + ); +} + +#[test] +fn decimal128_precision_18_fits_i64() { + assert_compatible( + "Decimal128(18,2)", + &col(decimal128( + 18, + 2, + vec![ + Some(0), + Some(123), + Some(-123), + Some(i64::MAX as i128), + Some(i64::MIN as i128), + None, + ], + )), + ); +} + +#[test] +fn decimal128_precision_20_does_not_fit_i64() { + let too_big = 10_000_000_000_000_000_000i128; // 1e19, beyond i64::MAX + assert_compatible( + "Decimal128(20,2)", + &col(decimal128( + 20, + 2, + vec![ + Some(0), + Some(123), + Some(-123), + Some(too_big), + Some(-too_big), + None, + ], + )), + ); +} + +#[test] +fn decimal128_precision_38() { + let wide = 10i128.pow(28); + assert_compatible( + "Decimal128(38,10)", + &col(decimal128( + 38, + 10, + vec![ + Some(0), + Some(123), + Some(-123), + Some(wide), + Some(-wide), + None, + ], + )), + ); +} + +// --------------------------------------------------------------------------- +// Dictionary +// --------------------------------------------------------------------------- + +#[test] +fn dictionary_int8_utf8() { + let values: ArrayRef = Arc::new(StringArray::from(vec!["hello", "world", "abc"])); + let keys = Int8Array::from(vec![Some(0), Some(1), Some(2), Some(0), None, Some(1)]); + let dict = DictionaryArray::::try_new(keys, values).unwrap(); + assert_compatible("Dictionary", &col(dict)); +} + +#[test] +fn dictionary_int32_utf8() { + let values: ArrayRef = Arc::new(StringArray::from(vec!["hello", "world"])); + let keys = Int32Array::from(vec![Some(0), Some(1), Some(0), None, Some(1)]); + let dict = DictionaryArray::::try_new(keys, values).unwrap(); + assert_compatible("Dictionary", &col(dict)); +} + +#[test] +fn dictionary_int32_int64() { + let values: ArrayRef = Arc::new(Int64Array::from(vec![Some(10), Some(20), None])); + let keys = Int32Array::from(vec![Some(0), Some(1), Some(2), None, Some(0)]); + let dict = DictionaryArray::::try_new(keys, values).unwrap(); + assert_compatible("Dictionary", &col(dict)); +} + +#[test] +fn dictionary_matches_decoded() { + let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20, 30])); + let keys = Int8Array::from(vec![Some(0), Some(1), Some(2), Some(0), None]); + let dict: ArrayRef = + Arc::new(DictionaryArray::::try_new(keys, Arc::clone(&values)).unwrap()); + let decoded: ArrayRef = Arc::new(Int32Array::from(vec![ + Some(10), + Some(20), + Some(30), + Some(10), + None, + ])); + let from_dict = comet_kernel(&[Arc::clone(&dict)], SPARK_DEFAULT_SEED).unwrap(); + let from_decoded = comet_kernel(&[Arc::clone(&decoded)], SPARK_DEFAULT_SEED).unwrap(); + assert_eq!(from_dict, from_decoded); + assert_compatible("Dictionary decoded equivalent", &[dict]); + assert_compatible("decoded Int32", &[decoded]); +} + +#[test] +fn dictionary_nonuniform_seeds_match_decoded() { + let values: ArrayRef = Arc::new(Int32Array::from(vec![Some(10), Some(20), None])); + let keys = Int8Array::from(vec![Some(0), Some(1), Some(2), None, Some(0)]); + let dict: ArrayRef = Arc::new(DictionaryArray::::try_new(keys, values).unwrap()); + let decoded: ArrayRef = Arc::new(Int32Array::from(vec![ + Some(10), + Some(20), + None, + None, + Some(10), + ])); + let seeds = vec![7u64, 38, 69, 100, 131]; + let mut from_dict = seeds.clone(); + create_xxhash64_hashes(&[dict], &mut from_dict).unwrap(); + let mut from_decoded = seeds; + create_xxhash64_hashes(&[decoded], &mut from_decoded).unwrap(); + assert_eq!(from_dict, from_decoded); +} + +// --------------------------------------------------------------------------- +// Struct +// --------------------------------------------------------------------------- + +#[test] +fn struct_non_null() { + assert_compatible( + "Struct non-null", + &[struct_ab( + vec![Some(1), Some(2), Some(3)], + vec![Some("a"), Some("b"), Some("c")], + None, + )], + ); +} + +#[test] +fn struct_null_fields() { + assert_compatible( + "Struct with null fields", + &[struct_ab( + vec![Some(1), None, Some(3)], + vec![None, Some("b"), Some("c")], + None, + )], + ); +} + +/// Spark-compatible: a null struct must ignore hidden child values. +/// `SparkXxhash64` hashes the child buffers without pushing the parent null mask, so this +/// case is *not* routed upstream. +#[test] +fn null_struct_ignores_hidden_child_values_comet() { + let fields: Fields = vec![Arc::new(Field::new("a", DataType::Int32, true))].into(); + let nulls = NullBuffer::from(vec![true, false]); + let hidden: ArrayRef = Arc::new(StructArray::new( + fields.clone(), + vec![Arc::new(Int32Array::from(vec![Some(1), Some(999)])) as ArrayRef], + Some(nulls.clone()), + )); + let plain: ArrayRef = Arc::new(StructArray::new( + fields, + vec![Arc::new(Int32Array::from(vec![Some(1), None])) as ArrayRef], + Some(nulls), + )); + + let mut a = vec![SPARK_DEFAULT_SEED; 2]; + create_xxhash64_hashes(&[hidden], &mut a).unwrap(); + let mut b = vec![SPARK_DEFAULT_SEED; 2]; + create_xxhash64_hashes(&[plain], &mut b).unwrap(); + assert_eq!(a, b, "a null struct must hash the same either way"); + assert_eq!( + a[1], SPARK_DEFAULT_SEED, + "a null struct must leave the seed untouched" + ); +} + +#[test] +fn null_struct_hidden_children_diverge_from_spark_xxhash64() { + let fields: Fields = vec![Arc::new(Field::new("a", DataType::Int32, true))].into(); + let nulls = NullBuffer::from(vec![true, false]); + let hidden: ArrayRef = Arc::new(StructArray::new( + fields, + vec![Arc::new(Int32Array::from(vec![Some(1), Some(999)])) as ArrayRef], + Some(nulls), + )); + let comet = comet_kernel(&[Arc::clone(&hidden)], SPARK_DEFAULT_SEED).unwrap(); + let upstream = spark_xxhash64_upstream(&[hidden]).unwrap(); + assert_eq!(comet[0], upstream[0], "non-null struct row still matches"); + assert_eq!(comet[1], SPARK_DEFAULT_SEED); + assert_ne!( + comet[1], upstream[1], + "SparkXxhash64 hashes hidden child values of a NULL struct" + ); +} + +// --------------------------------------------------------------------------- +// List / LargeList / FixedSizeList +// --------------------------------------------------------------------------- + +#[test] +fn list_int32() { + assert_compatible( + "List", + &[list_i32(vec![ + Some(vec![Some(1), Some(2)]), + Some(vec![]), + None, + Some(vec![Some(1), None, Some(3)]), + Some(vec![Some(-1)]), + ])], + ); +} + +#[test] +fn large_list_int32() { + assert_compatible( + "LargeList", + &[large_list_i32(vec![ + Some(vec![Some(1), Some(2)]), + Some(vec![]), + None, + Some(vec![Some(1), None, Some(3)]), + ])], + ); +} + +#[test] +fn fixed_size_list_int32() { + let values = Int32Array::from(vec![ + Some(1), + Some(2), + Some(3), + Some(4), + None, + Some(6), + Some(0), + Some(0), + Some(0), + ]); + let mut validity = arrow::array::BooleanBufferBuilder::new(3); + validity.append(true); + validity.append(true); + validity.append(false); + let array = FixedSizeListArray::new( + Arc::new(Field::new("item", DataType::Int32, true)), + 3, + Arc::new(values), + Some(NullBuffer::new(validity.finish())), + ); + assert_compatible("FixedSizeList", &col(array)); +} + +#[test] +fn nested_list() { + let inner = ListBuilder::new(Int32Builder::new()); + let mut outer = ListBuilder::new(inner); + // [[1,2],[3]] + { + let inner = outer.values(); + inner.values().append_value(1); + inner.values().append_value(2); + inner.append(true); + inner.values().append_value(3); + inner.append(true); + outer.append(true); + } + // [] + outer.append(true); + // NULL + outer.append(false); + assert_compatible("List>", &[Arc::new(outer.finish())]); +} + +// --------------------------------------------------------------------------- +// Map +// --------------------------------------------------------------------------- + +#[test] +fn map_utf8_int32() { + assert_compatible( + "Map", + &[map_utf8_i32(vec![ + Some(vec![("a", Some(1)), ("b", Some(2))]), + Some(vec![]), + None, + Some(vec![("k", None)]), + Some(vec![("x", Some(0)), ("y", Some(-1))]), + ])], + ); +} + +#[test] +fn map_int32_utf8() { + assert_compatible( + "Map", + &[map_i32_utf8(vec![ + Some(vec![(1, Some("a")), (2, Some("b"))]), + Some(vec![]), + None, + Some(vec![(0, None)]), + Some(vec![(-1, Some("")), (3, Some("x"))]), + ])], + ); +} + +#[test] +fn map_utf8_to_utf8() { + assert_compatible( + "Map", + &[map_utf8_utf8(vec![ + Some(vec![("a", Some("x")), ("b", Some("y"))]), + Some(vec![]), + None, + Some(vec![("k", None)]), + Some(vec![("empty", Some("")), ("z", Some("zz"))]), + ])], + ); +} + +#[test] +fn map_int32_int32() { + assert_compatible( + "Map", + &[map_i32_i32(vec![ + Some(vec![(1, Some(10)), (2, Some(20))]), + Some(vec![]), + None, + Some(vec![(0, None)]), + Some(vec![(-1, Some(0)), (3, Some(-3))]), + ])], + ); +} + +#[test] +fn map_utf8_decimal128_small() { + assert_compatible( + "Map", + &[map_utf8_decimal( + 10, + 2, + vec![ + Some(vec![("a", Some(123)), ("b", Some(-4))]), + Some(vec![]), + None, + Some(vec![("k", None)]), + ], + )], + ); +} + +#[test] +fn map_utf8_decimal128_large() { + let wide = 10_000_000_000_000_000_000i128; + assert_compatible( + "Map", + &[map_utf8_decimal( + 20, + 2, + vec![ + Some(vec![("a", Some(wide)), ("b", Some(-wide))]), + Some(vec![]), + None, + ], + )], + ); +} + +// --------------------------------------------------------------------------- +// Nested combinations +// --------------------------------------------------------------------------- + +#[test] +fn list_of_struct_non_null() { + let mut lb = ListBuilder::new(StructBuilder::new( + vec![ + Arc::new(Field::new("a", DataType::Int32, true)), + Arc::new(Field::new("b", DataType::Utf8, true)), + ], + vec![ + Box::new(Int32Builder::new()), + Box::new(StringBuilder::new()), + ], + )); + for (a, b) in [(1, "x"), (2, "y")] { + let sb = lb.values(); + sb.field_builder::(0).unwrap().append_value(a); + sb.field_builder::(1) + .unwrap() + .append_value(b); + sb.append(true); + } + lb.append(true); + lb.append(true); // empty list + lb.append(false); // null list + assert_compatible("List non-null elements", &[Arc::new(lb.finish())]); +} + +#[test] +fn struct_of_list() { + let list = list_i32(vec![ + Some(vec![Some(1), Some(2)]), + Some(vec![]), + None, + Some(vec![Some(3)]), + ]); + let fields: Fields = vec![Arc::new(Field::new("xs", list.data_type().clone(), true))].into(); + let array: ArrayRef = Arc::new(StructArray::new(fields, vec![list], None)); + assert_compatible("Struct>", &[array]); +} + +#[test] +fn list_of_dictionary_is_incompatible() { + let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20])); + let keys = Int8Array::from(vec![0i8, 1]); + let dict: ArrayRef = Arc::new(DictionaryArray::::try_new(keys, values).unwrap()); + let as_list = |elems: ArrayRef| -> ArrayRef { + Arc::new(ListArray::new( + Arc::new(Field::new("item", elems.data_type().clone(), true)), + OffsetBuffer::new(vec![0i32, 2].into()), + elems, + None, + )) + }; + let decoded: ArrayRef = Arc::new(Int32Array::from(vec![10, 20])); + let comet_dict = comet_kernel(&[as_list(Arc::clone(&dict))], SPARK_DEFAULT_SEED).unwrap(); + let comet_decoded = comet_kernel(&[as_list(decoded)], SPARK_DEFAULT_SEED).unwrap(); + assert_eq!( + comet_dict, comet_decoded, + "Comet hashes a dictionary list element as its decoded values" + ); + + let upstream = spark_xxhash64_upstream(&[as_list(dict)]).unwrap(); + assert_ne!( + comet_dict, upstream, + "SparkXxhash64 restarts nested dictionary hashes from seed 42" + ); +} + +#[test] +fn struct_of_dictionary() { + let values: ArrayRef = Arc::new(StringArray::from(vec!["hello", "world"])); + let keys = Int32Array::from(vec![Some(0), Some(1), Some(0), None]); + let dict: ArrayRef = Arc::new(DictionaryArray::::try_new(keys, values).unwrap()); + let fields: Fields = vec![Arc::new(Field::new("d", dict.data_type().clone(), true))].into(); + let array: ArrayRef = Arc::new(StructArray::new(fields, vec![dict], None)); + assert_compatible("Struct> non-null", &[array]); +} + +// --------------------------------------------------------------------------- +// Seed / multi-column chaining +// --------------------------------------------------------------------------- + +#[test] +fn multi_column_chain_matches_sequential() { + let a: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(2), None, Some(-1)])); + let b: ArrayRef = Arc::new(StringArray::from(vec![ + Some("x"), + None, + Some("y"), + Some("z"), + ])); + let c: ArrayRef = Arc::new(Float64Array::from(vec![ + Some(1.5), + Some(-0.0), + Some(0.0), + None, + ])); + let cols = [Arc::clone(&a), Arc::clone(&b), Arc::clone(&c)]; + assert_compatible("multi-column Int32,Utf8,Float64", &cols); + + let chained = comet_kernel(&cols, SPARK_DEFAULT_SEED).unwrap(); + let mut sequential = vec![SPARK_DEFAULT_SEED; a.len()]; + create_xxhash64_hashes(&[a], &mut sequential).unwrap(); + create_xxhash64_hashes(&[b], &mut sequential).unwrap(); + create_xxhash64_hashes(&[c], &mut sequential).unwrap(); + assert_eq!( + chained, sequential, + "hash(a,b,c) must fold each argument into the running seed" + ); +} + +#[test] +fn custom_seed_is_honored_by_comet() { + let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(0), None, Some(-1)])); + for seed in [0i64, 1, 7, 42, -1, i64::MIN] { + let expr = comet_expr(&[Arc::clone(&array)], seed).unwrap(); + let kernel = comet_kernel(&[Arc::clone(&array)], seed as u64).unwrap(); + assert_eq!(expr, kernel, "seed={seed}"); + if seed as u64 != SPARK_DEFAULT_SEED { + let default = comet_kernel(&[Arc::clone(&array)], SPARK_DEFAULT_SEED).unwrap(); + assert_ne!(expr, default, "custom seed {seed} must change the hash"); + } + } +} + +#[test] +fn row_dependent_starting_seeds() { + let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(1), Some(1)])); + let mut hashes = vec![1u64, 2, 3]; + create_xxhash64_hashes(&[array], &mut hashes).unwrap(); + assert_ne!(hashes[0], hashes[1]); + assert_ne!(hashes[1], hashes[2]); +} + +#[test] +fn spark_xxhash64_does_not_take_a_trailing_seed_argument() { + // A trailing seed scalar is a Comet UDF convention. Passing it to SparkXxhash64 would + // hash the seed as another column, starting from 42. + let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(2)])); + let comet = comet_expr(&[Arc::clone(&array)], 7).unwrap(); + let upstream_default = spark_xxhash64_upstream(&[array]).unwrap(); + assert_ne!(comet, upstream_default); +} + +// --------------------------------------------------------------------------- +// Types Comet supports that SparkXxhash64 does not +// --------------------------------------------------------------------------- + +#[test] +fn time64_nanosecond_is_comet_only() { + let array: ArrayRef = Arc::new(Time64NanosecondArray::from(vec![ + Some(0), + Some(1_000), + None, + Some(-1), + ])); + let comet = + comet_kernel(&[Arc::clone(&array)], SPARK_DEFAULT_SEED).expect("Comet hashes Time64(ns)"); + let expr = comet_expr(&[Arc::clone(&array)], SPARK_DEFAULT_SEED as i64).unwrap(); + assert_eq!(expr, comet); + let upstream = spark_xxhash64_upstream(&[array]); + assert!( + upstream.is_err(), + "SparkXxhash64 is not expected to hash Time64: {upstream:?}" + ); +} + +/// After routing compatible types to `SparkXxhash64`, null structs must still use the Comet +/// kernel so hidden child values do not affect the hash. +#[test] +fn null_struct_expression_matches_comet_kernel() { + let fields: Fields = vec![Arc::new(Field::new("a", DataType::Int32, true))].into(); + let nulls = NullBuffer::from(vec![true, false]); + let hidden: ArrayRef = Arc::new(StructArray::new( + fields, + vec![Arc::new(Int32Array::from(vec![Some(1), Some(999)])) as ArrayRef], + Some(nulls), + )); + let kernel = comet_kernel(&[Arc::clone(&hidden)], SPARK_DEFAULT_SEED).unwrap(); + let expr = comet_expr(&[Arc::clone(&hidden)], SPARK_DEFAULT_SEED as i64).unwrap(); + assert_eq!(expr, kernel); + assert_eq!(expr[1], SPARK_DEFAULT_SEED); + let upstream = spark_xxhash64_upstream(&[hidden]).unwrap(); + assert_ne!(expr[1], upstream[1]); +} + +#[test] +fn list_of_dictionary_expression_matches_comet_kernel() { + let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20])); + let keys = Int8Array::from(vec![0i8, 1]); + let dict: ArrayRef = Arc::new(DictionaryArray::::try_new(keys, values).unwrap()); + let list: ArrayRef = Arc::new(ListArray::new( + Arc::new(Field::new("item", dict.data_type().clone(), true)), + OffsetBuffer::new(vec![0i32, 2].into()), + dict, + None, + )); + let kernel = comet_kernel(&[Arc::clone(&list)], SPARK_DEFAULT_SEED).unwrap(); + let expr = comet_expr(&[list], SPARK_DEFAULT_SEED as i64).unwrap(); + assert_eq!(expr, kernel); +} diff --git a/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala index 68c4471e05d..bbf6c4289ad 100644 --- a/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometHashExpressionSuite.scala @@ -28,10 +28,10 @@ import org.apache.spark.sql.types.{IntegerType, StructField, StructType} import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions} /** - * Test suite for Spark murmur3 hash function compatibility between Spark and Comet. + * Test suite for Spark `hash` (murmur3) and `xxhash64` compatibility between Spark and Comet. * - * These tests verify that Comet's native implementation of murmur3 hash produces identical - * results to Spark's implementation for all supported data types. + * These tests verify that Comet's native implementations produce identical results to Spark for + * all supported data types. */ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper { @@ -39,7 +39,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c BOOLEAN) USING parquet") sql("INSERT INTO t VALUES (true), (false), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -47,7 +47,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c TINYINT) USING parquet") sql("INSERT INTO t VALUES (1), (0), (-1), (127), (-128), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -55,7 +55,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c SMALLINT) USING parquet") sql("INSERT INTO t VALUES (1), (0), (-1), (32767), (-32768), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -63,7 +63,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c INT) USING parquet") sql("INSERT INTO t VALUES (1), (0), (-1), (2147483647), (-2147483648), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -72,7 +72,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe sql("CREATE TABLE t(c BIGINT) USING parquet") sql( "INSERT INTO t VALUES (1), (0), (-1), (9223372036854775807), (-9223372036854775808), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -80,7 +80,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c FLOAT) USING parquet") sql("INSERT INTO t VALUES (1.0), (0.0), (-0.0), (-1.0), (3.14159), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -88,7 +88,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c DOUBLE) USING parquet") sql("INSERT INTO t VALUES (1.0), (0.0), (-0.0), (-1.0), (3.14159265358979), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -96,7 +96,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c STRING) USING parquet") sql("INSERT INTO t VALUES ('hello'), (''), ('Spark SQL'), ('苹果手机'), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -104,7 +104,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c BINARY) USING parquet") sql("INSERT INTO t VALUES (X''), (X'00'), (X'0102030405'), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -113,7 +113,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe sql("CREATE TABLE t(c DATE) USING parquet") sql( "INSERT INTO t VALUES (DATE '2023-01-01'), (DATE '1970-01-01'), (DATE '2000-12-31'), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -125,7 +125,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (TIMESTAMP '1970-01-01 00:00:00'), (TIMESTAMP '2000-12-31 23:59:59'), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } @@ -134,7 +134,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql(s"CREATE TABLE t(c DECIMAL($precision, $scale)) USING parquet") sql("INSERT INTO t VALUES (1.23), (-1.23), (0.0), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } } @@ -145,7 +145,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe sql(s"CREATE TABLE t(c DECIMAL($precision, $scale)) USING parquet") sql("INSERT INTO t VALUES (1.23), (-1.23), (0.0), (null)") // Large decimals may fall back to Spark, so just check the answer - checkSparkAnswer("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswer("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } } @@ -155,7 +155,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe sql("CREATE TABLE t(c ARRAY) USING parquet") sql("INSERT INTO t VALUES (array(1.23, 2.34)), (null)") // Should fall back to Spark due to nested high-precision decimal - checkSparkAnswerAndFallbackReason("SELECT c, hash(c) FROM t", "precision > 18") + checkSparkAnswerAndFallbackReason("SELECT c, hash(c), xxhash64(c) FROM t", "precision > 18") } } @@ -164,7 +164,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe sql("CREATE TABLE t(c STRUCT) USING parquet") sql("INSERT INTO t VALUES (named_struct('a', 1, 'b', 1.23)), (null)") // Should fall back to Spark due to nested high-precision decimal - checkSparkAnswerAndFallbackReason("SELECT c, hash(c) FROM t", "precision > 18") + checkSparkAnswerAndFallbackReason("SELECT c, hash(c), xxhash64(c) FROM t", "precision > 18") } } @@ -174,7 +174,9 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe sql("CREATE TABLE t(c MAP) USING parquet") sql("INSERT INTO t VALUES (map('a', 1.23)), (null)") // Should fall back to Spark due to nested high-precision decimal - checkSparkAnswerAndFallbackReason("SELECT c, hash(c) FROM t", "precision > 18") + checkSparkAnswerAndFallbackReason( + "SELECT c, hash(c), xxhash64(c) FROM t", + "precision > 18") } } } @@ -189,7 +191,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (null), (array(null)), (array(1, null, 3))""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -204,7 +206,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (null), (array(null)), (array('a', null, 'b'))""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -216,7 +218,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (array(-1.0, 0.0, 1.0)), (array()), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -228,7 +230,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (array(array(), array(1))), (array()), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -241,7 +243,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (named_struct('a', null, 'b', 'test')), (named_struct('a', 42, 'b', null)), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -253,7 +255,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (named_struct('a', 2, 'b', named_struct('x', '', 'y', 0.0))), (named_struct('a', 3, 'b', null)), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -359,7 +361,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (named_struct('a', 2, 'b', array())), (named_struct('a', 3, 'b', null)), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -371,7 +373,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (array(named_struct('a', 3, 'b', ''))), (array()), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -385,7 +387,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (map('x', -1)), (map()), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } } @@ -400,7 +402,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (map('x', array())), (map()), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } } @@ -414,7 +416,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (null, null, null), (-1, 'test', -1.5)""") checkSparkAnswerAndOperator( - "SELECT hash(a, b, c), hash(c, b, a), hash(a), hash(b), hash(c) FROM t") + "SELECT hash(a, b, c), xxhash64(a, b, c), hash(c, b, a), xxhash64(c, b, a), hash(a), xxhash64(a), hash(b), xxhash64(b), hash(c), xxhash64(c) FROM t") } } @@ -426,7 +428,8 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (2, array(), ''), (null, null, null), (3, array(-1, 0, 1), 'test')""") - checkSparkAnswerAndOperator("SELECT hash(a, b, c), hash(b), hash(a, c) FROM t") + checkSparkAnswerAndOperator( + "SELECT hash(a, b, c), xxhash64(a, b, c), hash(b), xxhash64(b), hash(a, c), xxhash64(a, c) FROM t") } } @@ -438,7 +441,8 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (2, named_struct('x', 20, 'y', '')), (null, null), (3, named_struct('x', null, 'y', 'test'))""") - checkSparkAnswerAndOperator("SELECT hash(a, b), hash(b, a), hash(b) FROM t") + checkSparkAnswerAndOperator( + "SELECT hash(a, b), xxhash64(a, b), hash(b, a), xxhash64(b, a), hash(b), xxhash64(b) FROM t") } } @@ -446,7 +450,8 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(s STRING, a ARRAY) USING parquet") sql("INSERT INTO t VALUES ('', array()), ('a', array(1))") - checkSparkAnswerAndOperator("SELECT hash(s), hash(a), hash(s, a) FROM t") + checkSparkAnswerAndOperator( + "SELECT hash(s), xxhash64(s), hash(a), xxhash64(a), hash(s, a), xxhash64(s, a) FROM t") } } @@ -454,7 +459,8 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(a INT, b STRING, c ARRAY) USING parquet") sql("INSERT INTO t VALUES (null, null, null)") - checkSparkAnswerAndOperator("SELECT hash(a), hash(b), hash(c), hash(a, b, c) FROM t") + checkSparkAnswerAndOperator( + "SELECT hash(a), xxhash64(a), hash(b), xxhash64(b), hash(c), xxhash64(c), hash(a, b, c), xxhash64(a, b, c) FROM t") } } @@ -462,9 +468,9 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe withTable("t") { sql("CREATE TABLE t(c INT) USING parquet") sql("INSERT INTO t VALUES (1), (2), (3), (null)") - // hash() with seed 42 (default) and seed 0 + // Extra integer arguments are hashed as additional children, not as the Spark seed. checkSparkAnswerAndOperator( - "SELECT hash(c), hash(c, 0), hash(c, 42), hash(c, -1) FROM t ORDER BY c") + "SELECT hash(c), xxhash64(c), hash(c, 0), xxhash64(c, 0), hash(c, 42), xxhash64(c, 42), hash(c, -1), xxhash64(c, -1) FROM t ORDER BY c") } } @@ -474,7 +480,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe // Create an array with 1000 elements val largeArray = (1 to 1000).mkString("array(", ", ", ")") sql(s"INSERT INTO t VALUES ($largeArray)") - checkSparkAnswerAndOperator("SELECT hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT hash(c), xxhash64(c) FROM t") } } @@ -491,7 +497,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (named_struct('a', 1, 'b', named_struct('x', 'hello', 'y', array(named_struct('p', 10, 'q', 'foo'), named_struct('p', 20, 'q', 'bar'))))), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } @@ -502,7 +508,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe sql("CREATE TABLE t(c STRING) USING parquet") // Repeated values to trigger dictionary encoding sql("INSERT INTO t VALUES ('a'), ('b'), ('a'), ('b'), ('a'), ('c'), (null)") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t ORDER BY c") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t ORDER BY c") } } } @@ -518,7 +524,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (array('a', 'b')), (array('c')), (null)""") - checkSparkAnswerAndOperator("SELECT c, hash(c) FROM t") + checkSparkAnswerAndOperator("SELECT c, hash(c), xxhash64(c) FROM t") } } } @@ -564,7 +570,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe (3, array(named_struct('l', cast(null as array))), null), (4, null, array(named_struct('a', 3, 'b', 'z')))""") checkSparkAnswerAndOperator( - "SELECT id, hash(nested), xxhash64(nested), hash(plain, nested) FROM t ORDER BY id") + "SELECT id, hash(nested), xxhash64(nested), hash(plain, nested), xxhash64(plain, nested) FROM t ORDER BY id") } } @@ -583,7 +589,7 @@ class CometHashExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelpe spark.read.parquet(filename.toString).createOrReplaceTempView("t1") for (col <- schema.fields) { val name = col.name - checkSparkAnswer(s"select $name, hash($name) from t1 order by $name") + checkSparkAnswer(s"select $name, hash($name), xxhash64($name) from t1 order by $name") } } } From d2ebecd73ab3016c10c688d942ac84066befd7b8 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Wed, 16 Sep 2026 02:29:07 +0800 Subject: [PATCH 2/2] test: expand xxhash64 benchmark coverage --- native/spark-expr/benches/xxhash64.rs | 107 ++++++++++++++++++++++++-- 1 file changed, 99 insertions(+), 8 deletions(-) diff --git a/native/spark-expr/benches/xxhash64.rs b/native/spark-expr/benches/xxhash64.rs index 308eb05807e..94f65516142 100644 --- a/native/spark-expr/benches/xxhash64.rs +++ b/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, seed: i64) -> Vec { + 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::>(), + )); + let dictionary_keys = + Int32Array::from_iter_values((0..TYPE_FAMILY_ROWS).map(|i| (i % 1024) as i32)); + let dictionary: ArrayRef = Arc::new( + DictionaryArray::::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);