andygrove opened a new issue, #6385:
URL: https://github.com/apache/datafusion-comet/issues/6385
### Describe the bug
Signed zero and NaN differences between Spark and Arrow/DataFusion keep
producing silent wrong results, and we fix them one expression at a time.
Eleven PRs since August 21 went into this (#5393, #5404, #5469, #5166, #5235,
#5981, #6049, #6055, #6073, #5472, #5403), and the queries below still return
different answers from Spark on `main` (222609d8f, default Spark 4.1 profile,
default configuration).
Two things keep this from converging.
**1. Spark has several float equality rules.** Each function inherits one
from the Java API its implementation calls, and some changed in patch releases:
| Spark rule | `-0.0` vs `0.0` | NaN | Used by |
|---|---|---|---|
| SQL ordering (`SQLOrderingUtil.compareDoubles`, `genEqual`) | equal | all
NaNs equal, NaN sorts highest | comparisons, `IN`, `ORDER BY`, rank,
`min`/`max`, `greatest`/`least`, `sort_array`,
`array_contains`/`array_position`/`array_remove`, nested comparisons |
| `NormalizeNaNAndZero` (inserted by Spark's optimizer) | folded to `0.0` |
canonical NaN | grouping keys, join keys, window partition keys |
| `Murmur3Hash` / `XxHash64` | same hash | hashed via `doubleToLongBits`
(canonical) | `hash`, `xxhash64` |
| `java.lang.Double.equals` (`OpenHashSet` after SPARK-45599) | distinct |
all NaNs equal | `collect_set` and `mode` before 4.2 (SPARK-57298,
SPARK-57329); array set functions before SPARK-54918 (4.0.5, 4.1.4, 4.2.0) |
Arrow compares floats with IEEE 754 total order: the two zeros are distinct,
NaNs compare by bit pattern, and a NaN with the sign bit set sorts below
`-Infinity`. DataFusion 55 added `-0.0` folding, but not NaN canonicalization,
in `apply_cmp`, the array set functions, `GroupValuesRows` and join key
comparison. That fixed some Comet cases without any Comet change and changed
others (#5701). iceberg-rust's `OrderedFloat` follows yet another rule (#6138).
**2. Sign-bit NaNs are the normal case on x86-64.** Every NaN produced by
arithmetic (`sqrt(-1)`, `ln(-1)`, `0/0`, `inf - inf`) is `0xfff8000000000000`
in both Rust and the JVM on x86-64, while aarch64 produces
`0x7ff8000000000000`. Spark hides the sign through `doubleToLongBits`. Comet
paths that compare raw Arrow values put these NaNs below every other value, so
a query that passes on an Apple Silicon laptop can fail on a Linux x86 cluster.
The current handling is spread across:
- Scala
- `CometExecRule.normalize` wraps comparison operands in
`NormalizeNaNAndZero`, but only inside `ProjectExec` and `FilterExec`.
- `normalizeInOperand` in `predicates.scala` repeats that logic for `IN`.
- `collect_set` calls Spark's `NormalizeFloatingNumbers.normalize`, and
`mode` gets a flag.
- The opt-in `strictFloatingPoint` fallback is checked in six places.
- Native
- `normalize_float` / `NormalizeNaNAndZero::normalize_array`
(`math_funcs/internal/normalize_nan.rs`), with copies in `hll_plus_plus.rs`
(`normalize_floats`) and `max_min_by.rs` (`canonicalize_float_ordering`), and a
macro variant in `mode.rs` (`normalize_key`).
- `normalize_nested_floats` (`nested_float_normalize.rs`), which does not
handle `Map`.
- Two separate Spark comparators: `nested_equality`
(`nested_comparison.rs`) and `spark_comparator` (`array_extrema.rs`).
- `OverlapKey` (`arrays_overlap.rs`).
- Six inline `-0.0` checks in `hash_funcs/utils.rs`, with no NaN
canonicalization.
- `create_normalized_key_expr` (`planner.rs`).
### Steps to reproduce
```sql
CREATE TABLE t (id INT, d DOUBLE) USING parquet;
INSERT INTO t VALUES (1, 0.0D), (2, double('-0.0')), (3, double('NaN')), (4,
1.0D), (5, -1.0D);
```
`-d` flips the sign bit, so row 3 yields a sign-bit NaN on any platform.
Negation is the portable way to get one in a test, because a Parquet round trip
through Spark's writer canonicalizes NaN.
Same results on macOS aarch64 and Linux x86-64:
| Query | Spark | Comet |
|---|---|---|
| `SELECT max(-d), min(-d) FROM t WHERE id IN (3, 4, 5)` | `NaN, -1.0` |
`1.0, NaN` |
| `SELECT hash(-d), xxhash64(-d) FROM t WHERE id = 3` | `-1281358385,
-3127944061524951246` | `-1489914710, 9200374361256412029` |
| `SELECT sum(if(-d > 0.0, 1, 0)) FROM t` (comparison in an aggregate
argument) | `2` | `1` |
| `SELECT count(*) FILTER (WHERE -d >= d) FROM t` | `4` | `3` |
| `SELECT a.id FROM t a JOIN t b ON a.id = b.id AND -a.d >= b.d` (hash join
condition) | `1, 2, 3, 5` | `1, 2, 5` |
| `SELECT b.id FROM t a JOIN t b ON -a.d > b.d WHERE a.id = 3` (nested loop
join) | `1, 2, 4, 5` | (no rows) |
| `SELECT id FROM t ORDER BY -d > 0.0, id` | `1, 2, 4, 3, 5` | `1, 2, 3, 4,
5` |
| `SELECT id, x FROM t LATERAL VIEW explode(array(-d > 0.0)) v AS x WHERE id
= 3` | `3, true` | `3, false` |
| `SELECT id, greatest(-d, 0.0D), least(-d, 0.0D) FROM t WHERE id IN (1, 3)`
| `1, -0.0, -0.0` and `3, NaN, 0.0` | `1, 0.0, -0.0` and `3, 0.0, NaN` |
| `SELECT array_remove(array(d, 1.0D), 0.0D) FROM t WHERE id = 2` | `[1.0]`
| `[-0.0, 1.0]` |
| `SELECT array_remove(array(-d, 1.0D), d) FROM t WHERE id = 3` | `[1.0]` |
`[NaN, 1.0]` |
| `SELECT array_distinct(array(d, -d)), array_union(array(d), array(-d))
FROM t WHERE id = 3` | `[NaN], [NaN]` | `[NaN, NaN], [NaN, NaN]` |
| `SELECT sort_array(array(-d, d, 1.0D)) FROM t WHERE id = 3` | `[1.0, NaN,
NaN]` | `[NaN, 1.0, NaN]` |
On Linux x86-64 only, ordinary arithmetic triggers the same failures (all of
these match on aarch64):
| Query | Spark | Comet on x86-64 |
|---|---|---|
| `SELECT max(sqrt(d)), min(sqrt(d)) FROM t WHERE id IN (4, 5)` | `NaN, 1.0`
| `1.0, NaN` |
| `SELECT hash(sqrt(d)), xxhash64(sqrt(d)) FROM t WHERE id = 5` |
`-1281358385, -3127944061524951246` | `-1489914710, 9200374361256412029` |
| `SELECT sum(if(sqrt(d) > 100.0D, 1, 0)) FROM t` | `2` | `1` |
| `SELECT id, greatest(sqrt(d), 0.0D) FROM t WHERE id = 5` | `5, NaN` | `5,
0.0` |
Already tracked separately: nested ordering comparisons (#6157), nested sort
keys and rank (#5507), `-0.0` in `array_distinct` / `array_union` (#5701),
`collect_set` before 4.2 (#5312), map lookups with float keys (#5580).
These matched Spark in the same runs: comparisons in `Project` and `Filter`,
`GROUP BY`, scalar `ORDER BY` and rank, equi-joins, `IN`, nested `=`,
`array_contains`, `array_position`, `array_max`/`array_min`, `arrays_overlap`,
`max_by`/`min_by`, `approx_count_distinct`, `count(DISTINCT ...)`.
### Expected behavior
Every native path follows the same Spark rule as the Spark function it
replaces, on every platform, and that rule is implemented once instead of being
re-derived per expression.
Proposed approach:
1. **One native module for Spark float semantics** in `spark-expr`. It
provides the two key normalizations (SQL: fold `-0.0` and canonicalize NaN;
boxed: canonicalize NaN only), flat and nested including `Map`. It also
provides one nested-aware Spark comparator (`compareDoubles` order) and the
hash input for doubles and floats. The helpers listed above move onto it.
2. **Normalize float operands in the native comparison builder**
(`binary_expr_builder` / `spark_comparison`), scalar and nested, for all six
comparison operators. That covers every operator that hosts a comparison
(aggregates, `FILTER`, join conditions, sort keys, `Generate`), so
`CometExecRule.normalize` can be removed. Scan predicates need care so
row-group pruning is not lost.
3. **Keep the Spark version dependence in Scala** as one policy object that
passes a mode to native code. It covers SPARK-45599, SPARK-54918, SPARK-57298
and SPARK-57329, checked at runtime down to the patch version, and replaces
version checks scattered across serdes.
4. **Until a function has a native fix**, report it `Incompatible` for float
inputs so it runs through the codegen dispatcher.
Suggested order:
- [ ] Add a sweep suite: edge values (`0.0`, `-0.0`, canonical NaN, negated
NaN) crossed with operator contexts (`Project`, `Filter`, aggregate argument,
`FILTER`, hash/sort-merge/nested loop join condition, sort key, `Generate`,
window) and every serde that accepts float input, with known gaps linked to
issues. This also catches the next DataFusion upgrade that shifts float
behavior.
- [ ] Consolidate the native helpers into the shared module (no behavior
change)
- [ ] Canonicalize NaN in `hash` / `xxhash64`
- [ ] Normalize comparisons in the native comparison builder and remove
`CometExecRule.normalize`
- [ ] Fix `min`/`max`, `greatest`/`least`, `array_remove`, NaN handling in
`array_distinct`/`array_union`, and `sort_array`
- [ ] Nested ordering (#6157, #5507)
- [ ] Contributor guide section on which Spark rule an expression follows
and how to test for it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
