You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/03 13:17:04 UTC
(datafusion-comet) 01/01: fix: hash NaN the way Spark does in hash, xxhash64 and native shuffle (#6413)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6413-71250f2bc725e8b101650195a0a68acfbc559ddf
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit e32df534a71bd86d1b29e14bf74149c8207a1439
Author: Andy Grove <ag...@apache.org>
AuthorDate: Sat Oct 3 13:03:04 2026 +0000
fix: hash NaN the way Spark does in hash, xxhash64 and native shuffle (#6413)
* refactor: move native -0.0 and NaN handling into one float_semantics module
Part of #6385. Each native expression that follows one of Spark's float
rules carried its own copy of that rule. This moves them onto a new
`float_semantics` module in `spark-expr`, with no change in behavior.
- `normalize_float`, `canonicalize_nan`, `compare_floats` and `hash_input`
are the per-value rules: `NormalizeNaNAndZero`, `Double.equals`,
`compareDoubles`, and the `Murmur3Hash`/`XxHash64` input.
- `normalize_floats`, `normalize_nested_floats` and `has_float_leaf`
normalize arrays. `NormalizeNaNAndZero` and `NormalizeNestedFloats`
move into the module, and `NormalizeNaNAndZero::wrap_if_needed` takes
over the planner's check in `create_normalized_key_expr`.
- `spark_comparator` replaces both `nested_equality` (nested `=` and
`IN`) and the comparator local to `array_min`/`array_max`.
hll_plus_plus, max_min_by, mode, percentile, array_extrema,
array_position, arrays_overlap, nested_comparison, the float hash macros,
and the planner's sort keys and range partition bounds now call the
module.
* Restore hot-loop speed and narrow the float_semantics API
- compare_floats uses the if-chain form again, so `.is_eq()` in
array_position and percentile compiles to a single comparison.
- float_extrema calls new float_lt/float_gt predicates, which compile to
the same instructions as the inline tests they replace. The three-way
comparison was up to 9x slower in that loop.
- spark_equality gives nested `=` and `IN` back the length check that
skips comparing the elements of lists of different lengths.
- float_semantics is a private module again, re-exporting
normalize_floats, canonicalize_nan and the two expressions.
- The Parquet variant cast canonicalizes NaN with canonicalize_nan, and
gains a test.
- Tests share the NaN constants and check the per-value rules in one
table, and the max_by doc defers to the module docs.
* Return an error, not a panic, for mismatched physical types in the comparator
`check_types` compares logical types, so a `Dictionary(_, Float64)` passes
as `Float64` and then panicked when `float_comparator` downcast it. The
comparator now matches on both sides' types and returns the internal error.
`mismatched_types_are_rejected` covers a dictionary on either side, at the
top level and inside a list.
* fix: hash NaN the way Spark does in hash, xxhash64 and native shuffle
Part of #6385. Spark hashes a float through doubleToLongBits or
floatToIntBits, which canonicalize NaN, so every NaN hashes alike. Comet
hashed the raw bits, so `hash(-d)` and `xxhash64(-d)` differed from Spark
for a NaN, and on x86-64 so did the hash of every NaN that arithmetic
produces. `hash_input` now returns `normalize_float`, which covers `hash`,
`xxhash64`, list, struct and dictionary elements, the native shuffle's hash
partitioner and `approx_count_distinct`.
At the default seed, `xxhash64` handed float arguments to datafusion-spark's
`SparkXxhash64`, which also hashes the raw bits, so floats now stay on
Comet's kernel. `approx_count_distinct` no longer normalizes floats before
hashing, since `xxhash64` now does.
* test: cover infinities in hash NaN tests and link the upstream xxhash64 issue
Add Infinity and -Infinity rows to the hash fixture, so hash(-d) and
hash(-f) show that both infinities keep their own hashes through NaN
canonicalization. Link apache/datafusion#25913, the upstream report of
SparkXxhash64 hashing a NaN's raw bits, at the gate that keeps floats on
Comet's kernel, and list floats among the retained cases in the xxhash64
audit entry.
* test: cover maps and nested floats in the hash NaN tests
* test: drop the sweep suite's known gap for hash and xxhash64
hash, xxhash64 and the native shuffle's hash partitioner now hash NaN
the way Spark does, so the suite requires those cases to match.
---
.../expression-audits/hash_funcs.md | 2 +-
native/spark-expr/src/agg_funcs/hll_plus_plus.rs | 87 ++++++++++++++---
native/spark-expr/src/float_semantics/mod.rs | 50 ++++------
native/spark-expr/src/hash_funcs/murmur3.rs | 103 ++++++++++++++++++++-
native/spark-expr/src/hash_funcs/xxhash64.rs | 22 ++++-
native/spark-expr/src/hash_funcs/xxhash64_diff.rs | 21 +++++
.../resources/sql-tests/expressions/hash/hash.sql | 23 +++++
.../apache/comet/CometFloatSemanticsSuite.scala | 6 --
.../comet/exec/CometNativeShuffleSuite.scala | 24 +++++
9 files changed, 281 insertions(+), 57 deletions(-)
diff --git a/docs/source/contributor-guide/expression-audits/hash_funcs.md b/docs/source/contributor-guide/expression-audits/hash_funcs.md
index 989d1d2066..c3ad4efe7b 100644
--- a/docs/source/contributor-guide/expression-audits/hash_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/hash_funcs.md
@@ -70,6 +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 paren [...]
+- 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, and record where the two differ on a NaN whose bits are not canonical. The Comet kernel is retained for a non-default seed ( [...]
[Spark Expression Support]: ../../user-guide/latest/expressions.md
diff --git a/native/spark-expr/src/agg_funcs/hll_plus_plus.rs b/native/spark-expr/src/agg_funcs/hll_plus_plus.rs
index 080127225b..5743148e79 100644
--- a/native/spark-expr/src/agg_funcs/hll_plus_plus.rs
+++ b/native/spark-expr/src/agg_funcs/hll_plus_plus.rs
@@ -17,14 +17,12 @@
//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed with Spark's
-//! `XxHash64` (seed 42, floats normalized first) and the registers are stored using the exact
-//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per 64-bit word). Keeping
-//! the wire format identical means the partial-aggregation state matches Spark's
-//! `aggBufferSchema`, and the cardinality estimate uses the same bias-correction tables, so
-//! results are bit-identical to Spark.
+//! `XxHash64` (seed 42) and the registers are stored using the exact same packed-`Long` buffer
+//! layout Spark uses (10 six-bit registers per 64-bit word). Keeping the wire format identical
+//! means the partial-aggregation state matches Spark's `aggBufferSchema`, and the cardinality
+//! estimate uses the same bias-correction tables, so results are bit-identical to Spark.
use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA, THRESHOLDS};
-use crate::float_semantics::normalize_floats;
use crate::hash_funcs::create_xxhash64_hashes;
use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Int64Array};
use arrow::datatypes::{DataType, Field, FieldRef};
@@ -118,14 +116,17 @@ impl AggregateUDFImpl for HllPlusPlus {
}
}
-/// Hash a value column with Spark's `XxHash64` (seed 42), normalizing floats first, reusing
-/// `buf` as scratch to avoid a per-batch allocation. The buffer is re-seeded (not just cleared)
-/// because `create_xxhash64_hashes` folds each value into the existing seed.
+/// Hash a value column with Spark's `XxHash64` (seed 42), reusing `buf` as scratch to avoid a
+/// per-batch allocation. The buffer is re-seeded (not just cleared) because
+/// `create_xxhash64_hashes` folds each value into the existing seed.
+///
+/// Spark runs float inputs through `NormalizeFloatingNumbers` before hashing, but
+/// `create_xxhash64_hashes` already hashes `-0.0` as `0.0` and every NaN as the canonical NaN, so
+/// the result is the same without a separate pass.
fn hash_values_into(array: &ArrayRef, buf: &mut Vec<u64>) -> Result<()> {
- let normalized = normalize_floats(array);
buf.clear();
- buf.resize(normalized.len(), HASH_SEED);
- create_xxhash64_hashes(&[normalized], buf)?;
+ buf.resize(array.len(), HASH_SEED);
+ create_xxhash64_hashes(&[Arc::clone(array)], buf)?;
Ok(())
}
@@ -463,7 +464,7 @@ impl GroupsAccumulator for HllPlusPlusGroupsAccumulator {
#[cfg(test)]
mod tests {
use super::*;
- use arrow::array::{Int32Array, StringArray};
+ use arrow::array::{Float32Array, Float64Array, Int32Array, StringArray};
fn acc(p: usize) -> HllPlusPlusAccumulator {
HllPlusPlusAccumulator::new(p)
@@ -503,6 +504,66 @@ mod tests {
assert_eq!(eval(&mut a), 2);
}
+ /// Spark counts the two zeros as one value and every NaN as one value.
+ #[test]
+ fn floats_fold_negative_zero_and_nan() {
+ let values: ArrayRef = Arc::new(Float64Array::from(vec![
+ 0.0,
+ -0.0,
+ f64::NAN,
+ f64::from_bits(0xfff8_0000_0000_0000),
+ f64::from_bits(0x7ff0_0000_0000_0001),
+ ]));
+ let mut a = acc(9);
+ a.update_batch(&[values]).unwrap();
+ assert_eq!(eval(&mut a), 2);
+ let values: ArrayRef = Arc::new(Float32Array::from(vec![
+ 0.0,
+ -0.0,
+ f32::NAN,
+ f32::from_bits(0xffc0_0000),
+ ]));
+ let mut a = acc(9);
+ a.update_batch(&[values]).unwrap();
+ assert_eq!(eval(&mut a), 2);
+ }
+
+ /// The hash kernel canonicalizes floats inside nested types, so a struct or list holding a
+ /// negative zero or a non-canonical NaN counts the same as one holding the plain value.
+ #[test]
+ fn nested_floats_fold_negative_zero_and_nan() {
+ use arrow::array::{ListArray, StructArray};
+ use arrow::buffer::OffsetBuffer;
+ use arrow::datatypes::Field;
+
+ let floats: ArrayRef = Arc::new(Float64Array::from(vec![
+ 0.0,
+ -0.0,
+ f64::NAN,
+ f64::from_bits(0xfff8_0000_0000_0000),
+ f64::from_bits(0x7ff0_0000_0000_0001),
+ ]));
+ let field = Arc::new(Field::new("item", DataType::Float64, true));
+ let nested: Vec<ArrayRef> = vec![
+ Arc::new(StructArray::new(
+ vec![Arc::clone(&field)].into(),
+ vec![Arc::clone(&floats)],
+ None,
+ )),
+ Arc::new(ListArray::new(
+ field,
+ OffsetBuffer::from_lengths([1; 5]),
+ floats,
+ None,
+ )),
+ ];
+ for values in nested {
+ let mut a = acc(9);
+ a.update_batch(&[Arc::clone(&values)]).unwrap();
+ assert_eq!(eval(&mut a), 2, "{}", values.data_type());
+ }
+ }
+
#[test]
fn strings_small_cardinality() {
let mut a = acc(9);
diff --git a/native/spark-expr/src/float_semantics/mod.rs b/native/spark-expr/src/float_semantics/mod.rs
index b4ab1ae130..3d0f0e2ec9 100644
--- a/native/spark-expr/src/float_semantics/mod.rs
+++ b/native/spark-expr/src/float_semantics/mod.rs
@@ -120,19 +120,12 @@ pub fn float_gt<T: Float>(left: T, right: T) -> bool {
left > right || (left.is_nan() && !right.is_nan())
}
-/// The value that Spark's `Murmur3Hash` and `XxHash64` hash in place of a float. `-0.0` hashes as
-/// `0.0`, so the two zeros hash alike.
-///
-/// Spark reads the other bits through `doubleToLongBits`, which also canonicalizes NaN. This does
-/// not yet, so a NaN whose bits are not canonical hashes differently from Spark (#6385).
+/// The value that Spark's `Murmur3Hash` and `XxHash64` hash in place of a float. Spark hashes
+/// `-0.0` as `0.0` and reads the other bits through `doubleToLongBits` or `floatToIntBits`, which
+/// canonicalize NaN, so this is the same value [`normalize_float`] returns.
#[inline]
pub fn hash_input<T: Float>(v: T) -> T {
- if v == T::zero() {
- // `-0.0 == 0.0` in IEEE 754, so this catches negative zero as well.
- T::zero()
- } else {
- v
- }
+ normalize_float(v)
}
/// A NaN with the sign bit set, which arithmetic produces on x86-64.
@@ -150,35 +143,26 @@ mod tests {
#[test]
fn per_value_rules() {
- // Each row: the input, then what `normalize_float`, `canonicalize_nan` and `hash_input`
- // return for it.
+ // Each row: the input, then what `normalize_float` and `canonicalize_nan` return for it.
+ // Spark hashes the value `normalize_float` returns.
let rows = [
- (-0.0, 0.0, -0.0, 0.0),
- (0.0, 0.0, 0.0, 0.0),
- (-1.5, -1.5, -1.5, -1.5),
- (
- f64::NEG_INFINITY,
- f64::NEG_INFINITY,
- f64::NEG_INFINITY,
- f64::NEG_INFINITY,
- ),
- // `hash_input` does not canonicalize NaN yet (#6385).
- (NEGATIVE_NAN, f64::NAN, f64::NAN, NEGATIVE_NAN),
- (PAYLOAD_NAN, f64::NAN, f64::NAN, PAYLOAD_NAN),
+ (-0.0, 0.0, -0.0),
+ (0.0, 0.0, 0.0),
+ (-1.5, -1.5, -1.5),
+ (f64::NEG_INFINITY, f64::NEG_INFINITY, f64::NEG_INFINITY),
+ (NEGATIVE_NAN, f64::NAN, f64::NAN),
+ (PAYLOAD_NAN, f64::NAN, f64::NAN),
];
- for (value, normalized, canonical, hashed) in rows {
+ for (value, normalized, canonical) in rows {
assert_eq!(normalize_float(value).to_bits(), normalized.to_bits());
assert_eq!(canonicalize_nan(value).to_bits(), canonical.to_bits());
- assert_eq!(hash_input(value).to_bits(), hashed.to_bits());
+ assert_eq!(hash_input(value).to_bits(), normalized.to_bits());
}
- let rows = [
- (-0.0, 0.0, -0.0, 0.0),
- (NEGATIVE_NAN_F32, f32::NAN, f32::NAN, NEGATIVE_NAN_F32),
- ];
- for (value, normalized, canonical, hashed) in rows {
+ let rows = [(-0.0, 0.0, -0.0), (NEGATIVE_NAN_F32, f32::NAN, f32::NAN)];
+ for (value, normalized, canonical) in rows {
assert_eq!(normalize_float(value).to_bits(), normalized.to_bits());
assert_eq!(canonicalize_nan(value).to_bits(), canonical.to_bits());
- assert_eq!(hash_input(value).to_bits(), hashed.to_bits());
+ assert_eq!(hash_input(value).to_bits(), normalized.to_bits());
}
}
diff --git a/native/spark-expr/src/hash_funcs/murmur3.rs b/native/spark-expr/src/hash_funcs/murmur3.rs
index 29a4820ca5..f891087b24 100644
--- a/native/spark-expr/src/hash_funcs/murmur3.rs
+++ b/native/spark-expr/src/hash_funcs/murmur3.rs
@@ -784,9 +784,14 @@ mod tests {
Some(-1.0),
Some(99999999999.99999999999),
Some(-99999999999.99999999999),
+ // Every NaN hashes as the canonical NaN, whatever its sign or payload.
+ Some(f32::NAN),
+ Some(f32::from_bits(0xffc0_0000)),
+ Some(f32::from_bits(0x7f80_0001)),
],
vec![
- 0xe434cc39, 0x379fae8f, 0x379fae8f, 0xdc0da8eb, 0xcbdc340f, 0xc0361c86,
+ 0xe434cc39, 0x379fae8f, 0x379fae8f, 0xdc0da8eb, 0xcbdc340f, 0xc0361c86, 0xeb2eb18a,
+ 0xeb2eb18a, 0xeb2eb18a,
],
);
}
@@ -801,13 +806,107 @@ mod tests {
Some(-1.0),
Some(99999999999.99999999999),
Some(-99999999999.99999999999),
+ // Every NaN hashes as the canonical NaN, whatever its sign or payload.
+ Some(f64::NAN),
+ Some(f64::from_bits(0xfff8_0000_0000_0000)),
+ Some(f64::from_bits(0x7ff0_0000_0000_0001)),
],
vec![
- 0xe4876492, 0x9c67b85d, 0x9c67b85d, 0x13d81357, 0xb87e1595, 0xa0eef9f9,
+ 0xe4876492, 0x9c67b85d, 0x9c67b85d, 0x13d81357, 0xb87e1595, 0xa0eef9f9, 0xb3a005cf,
+ 0xb3a005cf, 0xb3a005cf,
],
);
}
+ /// Spark hashes a float through `doubleToLongBits` or `floatToIntBits`, which canonicalize
+ /// NaN, so a NaN with the sign bit set or with a payload hashes like the canonical NaN inside
+ /// lists, structs, maps and dictionaries too. Checks both hashes, which expand the same macros.
+ #[test]
+ fn test_non_canonical_nan_hashes_like_canonical_nan() {
+ use crate::hash_funcs::create_xxhash64_hashes;
+ use arrow::array::{
+ DictionaryArray, FixedSizeListArray, Int32Array, LargeListArray, ListArray, MapArray,
+ StructArray,
+ };
+ use arrow::buffer::OffsetBuffer;
+ use arrow::datatypes::{DataType, Field, Fields};
+
+ let shapes = |values: ArrayRef| -> Vec<ArrayRef> {
+ let field = Arc::new(Field::new("item", values.data_type().clone(), true));
+ // One entry per row, with the float as both the key and the value.
+ let entry_fields = Fields::from(vec![
+ Field::new("key", values.data_type().clone(), false),
+ Field::new("value", values.data_type().clone(), true),
+ ]);
+ let entries = StructArray::new(
+ entry_fields.clone(),
+ vec![Arc::clone(&values), Arc::clone(&values)],
+ None,
+ );
+ vec![
+ Arc::clone(&values),
+ Arc::new(ListArray::new(
+ Arc::clone(&field),
+ OffsetBuffer::from_lengths([1; 3]),
+ Arc::clone(&values),
+ None,
+ )),
+ Arc::new(LargeListArray::new(
+ Arc::clone(&field),
+ OffsetBuffer::from_lengths([1; 3]),
+ Arc::clone(&values),
+ None,
+ )),
+ Arc::new(FixedSizeListArray::new(
+ Arc::clone(&field),
+ 1,
+ Arc::clone(&values),
+ None,
+ )),
+ Arc::new(StructArray::new(
+ vec![field].into(),
+ vec![Arc::clone(&values)],
+ None,
+ )),
+ Arc::new(DictionaryArray::new(
+ Int32Array::from(vec![0, 1, 2]),
+ values,
+ )),
+ Arc::new(MapArray::new(
+ Arc::new(Field::new("entries", DataType::Struct(entry_fields), false)),
+ OffsetBuffer::from_lengths([1; 3]),
+ entries,
+ None,
+ false,
+ )),
+ ]
+ };
+ let canonical64: ArrayRef = Arc::new(Float64Array::from(vec![f64::NAN; 3]));
+ let other64: ArrayRef = Arc::new(Float64Array::from(vec![
+ f64::from_bits(0xfff8_0000_0000_0000),
+ f64::from_bits(0x7ff0_0000_0000_0001),
+ f64::from_bits(0xfff0_0000_0000_0001),
+ ]));
+ let canonical32: ArrayRef = Arc::new(Float32Array::from(vec![f32::NAN; 3]));
+ let other32: ArrayRef = Arc::new(Float32Array::from(vec![
+ f32::from_bits(0xffc0_0000),
+ f32::from_bits(0x7f80_0001),
+ f32::from_bits(0xff80_0001),
+ ]));
+ for (canonical, other) in [(canonical64, other64), (canonical32, other32)] {
+ for (canonical, other) in shapes(canonical).into_iter().zip(shapes(other)) {
+ let (mut expected, mut actual) = (vec![42u32; 3], vec![42u32; 3]);
+ create_murmur3_hashes(&[Arc::clone(&canonical)], &mut expected).unwrap();
+ create_murmur3_hashes(&[Arc::clone(&other)], &mut actual).unwrap();
+ assert_eq!(actual, expected, "murmur3 of {}", other.data_type());
+ let (mut expected, mut actual) = (vec![42u64; 3], vec![42u64; 3]);
+ create_xxhash64_hashes(&[canonical], &mut expected).unwrap();
+ create_xxhash64_hashes(&[Arc::clone(&other)], &mut actual).unwrap();
+ assert_eq!(actual, expected, "xxhash64 of {}", other.data_type());
+ }
+ }
+ }
+
#[test]
fn test_str() {
let input = [
diff --git a/native/spark-expr/src/hash_funcs/xxhash64.rs b/native/spark-expr/src/hash_funcs/xxhash64.rs
index bdda7c85f1..2469711462 100644
--- a/native/spark-expr/src/hash_funcs/xxhash64.rs
+++ b/native/spark-expr/src/hash_funcs/xxhash64.rs
@@ -48,6 +48,9 @@ const SPARK_DEFAULT_SEED: i64 = 42;
/// 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
+/// - `Float32`/`Float64` (and anything containing one): `SparkXxhash64` hashes the raw bits of a
+/// NaN, where Spark hashes every NaN as the canonical NaN
+/// (<https://github.com/apache/datafusion/issues/25913>)
pub fn spark_xxhash64(args: &[ColumnarValue]) -> Result<ColumnarValue, DataFusionError> {
let length = args.len();
let seed = &args[length - 1];
@@ -126,7 +129,7 @@ fn invoke_spark_xxhash64(
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,
+ Boolean | Int8 | Int16 | Int32 | Int64 => true,
Utf8 | LargeUtf8 | Binary | LargeBinary | FixedSizeBinary(_) => true,
Date32 | Date64 | Timestamp(_, _) => true,
Decimal128(_, _) => true,
@@ -143,7 +146,8 @@ fn type_compatible_with_spark_xxhash64(dt: &DataType, in_list_or_map: bool) -> b
_ => false,
},
// Struct: `SparkXxhash64` hashes child buffers without applying the parent null
- // mask. Time64 is a Comet-only dispatch arm.
+ // mask. Time64 is a Comet-only dispatch arm. Float32/Float64: `SparkXxhash64` does not
+ // canonicalize NaN.
_ => false,
}
}
@@ -403,6 +407,10 @@ mod tests {
Some(-1.0),
Some(99999999999.99999999999),
Some(-99999999999.99999999999),
+ // Every NaN hashes as the canonical NaN, whatever its sign or payload.
+ Some(f32::NAN),
+ Some(f32::from_bits(0xffc0_0000)),
+ Some(f32::from_bits(0x7f80_0001)),
],
vec![
0x9b92689757fcdbd,
@@ -411,6 +419,9 @@ mod tests {
0xa2becc0e61bb3823,
0x8f20ab82d4f3687f,
0xdce4982d97f7ac4,
+ 0x255d1be3831088f8,
+ 0x255d1be3831088f8,
+ 0x255d1be3831088f8,
],
)
}
@@ -425,6 +436,10 @@ mod tests {
Some(-1.0),
Some(99999999999.99999999999),
Some(-99999999999.99999999999),
+ // Every NaN hashes as the canonical NaN, whatever its sign or payload.
+ Some(f64::NAN),
+ Some(f64::from_bits(0xfff8_0000_0000_0000)),
+ Some(f64::from_bits(0x7ff0_0000_0000_0001)),
],
vec![
0xe1fd6e07fee8ad53,
@@ -433,6 +448,9 @@ mod tests {
0x8cdde022746f8f1f,
0x793c5c88d313eac7,
0xc5e60e7b75d9b232,
+ 0xd4974f72d7833f32,
+ 0xd4974f72d7833f32,
+ 0xd4974f72d7833f32,
],
)
}
diff --git a/native/spark-expr/src/hash_funcs/xxhash64_diff.rs b/native/spark-expr/src/hash_funcs/xxhash64_diff.rs
index 3c51a0e486..bf16263358 100644
--- a/native/spark-expr/src/hash_funcs/xxhash64_diff.rs
+++ b/native/spark-expr/src/hash_funcs/xxhash64_diff.rs
@@ -38,6 +38,7 @@
//! | 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 |
+//! | Float32/64 NaN with non-canonical bits | **no** — upstream hashes the raw bits |
//! | custom seed | **no** — `SparkXxhash64` hardcodes 42 |
use super::{create_xxhash64_hashes, spark_xxhash64};
@@ -430,6 +431,26 @@ fn float64() {
);
}
+/// Spark hashes NaN through `doubleToLongBits`, which canonicalizes it. `SparkXxhash64` hashes the
+/// raw bits, so it differs from Spark for a NaN with the sign bit set.
+#[test]
+fn non_canonical_nan_diverges_from_spark_xxhash64() {
+ let values: ArrayRef = Arc::new(Float64Array::from(vec![
+ f64::NAN,
+ f64::from_bits(0xfff8_0000_0000_0000),
+ ]));
+ let comet = comet_kernel(&[Arc::clone(&values)], SPARK_DEFAULT_SEED).unwrap();
+ let expr = comet_expr(&[Arc::clone(&values)], SPARK_DEFAULT_SEED as i64).unwrap();
+ let upstream = spark_xxhash64_upstream(&[values]).unwrap();
+ assert_eq!(expr, comet, "floats must not be delegated to SparkXxhash64");
+ assert_eq!(comet[0], upstream[0], "the canonical NaN still matches");
+ assert_eq!(comet[1], comet[0], "every NaN hashes as the canonical NaN");
+ assert_ne!(
+ comet[1], upstream[1],
+ "SparkXxhash64 hashes the raw bits of a NaN"
+ );
+}
+
#[test]
fn utf8() {
assert_compatible(
diff --git a/spark/src/test/resources/sql-tests/expressions/hash/hash.sql b/spark/src/test/resources/sql-tests/expressions/hash/hash.sql
index fc93e9d4bf..cbf90b1845 100644
--- a/spark/src/test/resources/sql-tests/expressions/hash/hash.sql
+++ b/spark/src/test/resources/sql-tests/expressions/hash/hash.sql
@@ -15,6 +15,8 @@
-- specific language governing permissions and limitations
-- under the License.
+-- Config: spark.sql.legacy.allowHashOnMapType=true
+
-- hash functions
statement
CREATE TABLE test(col string, a int, b float) USING parquet
@@ -30,3 +32,24 @@ SELECT md5(col), md5(cast(a as string)), md5(cast(b as string)), hash(col), hash
-- native engine as scalar values rather than being folded away by Spark's optimizer.
query
SELECT md5('Spark SQL'), sha1('test'), sha2('test', 0), sha2('test', 256), sha2('test', 224), sha2('test', 384), sha2('test', 512), sha2('test', 128), sha2('test', -1), sha2(cast(null as string), 256), hash('test'), xxhash64('test')
+
+-- Spark hashes a float through doubleToLongBits or floatToIntBits, which canonicalize NaN, so
+-- every NaN hashes alike. Negating a column flips the sign bit of a NaN, giving the bits that
+-- arithmetic produces on x86-64. The infinities share a NaN's exponent bits but are not NaN, so
+-- they keep their own hashes, which negation swaps.
+statement
+CREATE TABLE test_nan(d double, f float) USING parquet
+
+statement
+INSERT INTO test_nan VALUES (double('NaN'), float('NaN')), (0.0, 0.0), (double('-0.0'), float('-0.0')), (1.5, 1.5), (NULL, NULL), (double('Infinity'), float('Infinity')), (double('-Infinity'), float('-Infinity'))
+
+query
+SELECT hash(d), hash(-d), xxhash64(d), xxhash64(-d), hash(f), hash(-f), xxhash64(f), xxhash64(-f), hash(-d, -f), xxhash64(-d, -f) FROM test_nan
+
+query
+SELECT hash(array(-d)), xxhash64(array(-d)), hash(named_struct('a', -d, 'b', -f)), xxhash64(named_struct('a', -d, 'b', -f)) FROM test_nan
+
+-- Spark rejects hashing a map unless the legacy config above is set. Map keys cannot be null,
+-- so the null row is filtered out.
+query
+SELECT hash(map(-d, -f)), xxhash64(map(-d, -f)) FROM test_nan WHERE d IS NOT NULL
diff --git a/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala b/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala
index a5c679b609..8715e06cc8 100644
--- a/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala
@@ -349,12 +349,6 @@ object CometFloatSemanticsSuite {
c.group == "comparison" &&
(c.context.startsWith("array operands") || c.context.startsWith("struct operands")) &&
!Set("=", "!=").contains(c.variant)),
- KnownGap(
- issue(6385),
- "hash, xxhash64 and the native shuffle's hash partitioner hash a NaN's raw bits.",
- c =>
- in("expression", "hash", "xxhash64", "hash of array", "hash of struct")(c) ||
- in("key", "hash repartition")(c)),
KnownGap(
issue(6385),
"min, max, greatest and least order floats by IEEE 754 total order.",
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
index cbacf4368a..03a57c18b4 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeShuffleSuite.scala
@@ -986,6 +986,30 @@ class CometNativeShuffleSuite extends CometTestBase with AdaptiveSparkPlanHelper
}
}
}
+ test("native shuffle on a float hash partitioning key matches Spark's partition assignment") {
+ // Spark hashes a float through doubleToLongBits or floatToIntBits, which canonicalize NaN.
+ // Negating a NaN in a native projection flips its sign bit, which gives the bits arithmetic
+ // produces on x86-64, and the native hash must still send the row where Spark sends it.
+ withParquetTable(
+ Seq(0.0, -0.0, Double.NaN, 1.5, -1.5).zipWithIndex.map { case (d, i) => (i, d, d.toFloat) },
+ "tbl") {
+ Seq("d", "nd", "nf", "nd, nf").foreach { keys =>
+ val repartitioned =
+ s"SELECT /*+ REPARTITION(10, $keys) */ _1, _2 AS d, -_2 AS nd, -_3 AS nf FROM tbl"
+ val query = s"SELECT _1, spark_partition_id() AS pid FROM ($repartitioned)"
+ val cometRows = sql(query).collect().map(r => (r.getInt(0), r.getInt(1))).sorted
+ // `SQLHelper.withSQLConf` returns Unit on Spark 3.x, so capture the rows via a var.
+ var sparkRows: Array[(Int, Int)] = Array.empty
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ sparkRows = sql(query).collect().map(r => (r.getInt(0), r.getInt(1))).sorted
+ }
+ assert(sparkRows.nonEmpty, "Spark produced no rows; the comparison would be vacuous")
+ checkCometExchange(sql(repartitioned), 1, true)
+ assert(cometRows === sparkRows, s"partition assignment differs from Spark for ($keys)")
+ }
+ }
+ }
+
test("native shuffle on nested hash partitioning key with interval leaf falls back") {
// CalendarIntervalType is allowed as a shuffle DATA column but the native hasher has no
// branch for it (https://github.com/apache/datafusion-comet/issues/5059). Because the nested
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org