andygrove opened a new pull request, #6447:
URL: https://github.com/apache/datafusion-comet/pull/6447
## Which issue does this PR close?
Closes #6157.
Part of #6385: the "normalize comparisons in the native comparison builder
and remove `CometExecRule.normalize`" step.
## Rationale for this change
Spark compares floats with `SQLOrderingUtil.compareDoubles`: `-0.0` equals
`0.0`, all NaNs are equal, and NaN sorts above every other value. Arrow's
comparison kernels use IEEE 754 total order instead. `CometExecRule.normalize`
bridged the two by wrapping comparison operands in `NormalizeNaNAndZero`, but
only inside `ProjectExec` and `FilterExec`, so a comparison anywhere else
compared raw Arrow values. DataFusion 55 folds `-0.0` in scalar comparisons,
which fixed the zero cases, but a NaN with the sign bit set still sorts below
every other value, and on x86-64 every NaN that arithmetic produces has that
bit set. From #6385, where `-d` turns the NaN in row 3 into a sign-bit NaN on
any platform:
| Query | Spark | Comet before |
| --- | --- | --- |
| `SELECT sum(if(-d > 0.0, 1, 0)) FROM t` | `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` | `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` | `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` |
Arrays and structs had the same gap for `<=>`, `<`, `<=`, `>` and `>=` in
every operator, Project and Filter included, because the rewrite only wrapped
scalar operands (#6157).
## What changes are included in this PR?
- `spark_comparison`, which builds every native comparison, now normalizes
float operands for all eight comparison operators: a `FLOAT` or `DOUBLE`
operand through `NormalizeNaNAndZero`, and an array or struct with a float leaf
through `NormalizeNestedFloats`. Once `-0.0` is folded and NaN canonicalized,
Arrow's total order agrees with Spark's. A literal operand is normalized while
the plan is built, so it stays a literal. Nested `=` and `<>` keep their
`spark_equality` path, which compares without copying.
- The data filters that a native scan pushes into the Parquet reader are
built without that normalization. DataFusion's pruning only recognizes a column
compared with a literal, so a wrapped column would stop row-group and page
pruning for every float predicate. The Filter above the scan still evaluates
each of these filters with Spark's semantics. `PhysicalPlanner` becomes `Clone`
so that `create_data_filter` can plan them with a copy set to
`FloatOperands::Raw`. With row-level pushdown, which is off by default, the
reader filters rows by the raw comparison, as it did before this change.
- `CometExecRule.normalize` is removed. It also normalized the divisor of
`Divide` and `Remainder`, which is no longer needed: native remainder and ANSI
division already treat `-0.0` as a zero divisor, and non-ANSI division checks
for a zero divisor with an `=` comparison, which is now normalized natively.
- `NormalizeNaNAndZero` keeps a scalar child a scalar instead of expanding
it into a column.
- The floating-point compatibility guide describes the new behavior.
- TPC-DS golden files, regenerated with `dev/regenerate-golden-files.sh` for
every Spark version. The plans are unchanged. Queries whose Project or Filter
compared or divided doubles (q21, q34, q39a, q39b, q73, q83) count one fewer
native expression, the `normalizenanandzero` the rewrite added, and q78 drops
`knownfloatingpointnormalized` and `normalizenanandzero` from its dispatched
functions.
## How are these changes tested?
- A new `float_comparisons.sql` fixture compares `DOUBLE` and `FLOAT` edge
values, including negated NaNs, with every operator, against columns and
against literals on either side. It covers Project, Filter, an aggregate
argument, a `FILTER` clause, a grouping key, broadcast hash, shuffled hash,
sort-merge and nested loop join conditions, a sort key and `explode`. It also
compares arrays and structs of floats under all six operators (#6157). On
`main` it fails at the first query outside Project and Filter.
- `arithmetic.sql` divides and takes the remainder by `-0.0D` and
`double('-0.0')`, the divisors the removed rewrite used to normalize.
- A `CometNativeReaderSuite` test checks that row-group statistics pruning
still fires for `d > 500.0D` on a `DOUBLE` column, and a planner unit test
checks that `create_data_filter` leaves the column unwrapped where
`create_expr` wraps it. The unit test fails if the data filters are built with
normalization.
- Unit tests:
- `spark_comparison` on every pair of edge values (both zeros, canonical,
sign-bit and payload NaNs, infinities and null) under all eight operators, as
two columns and as a column and a literal on either side, for `Float64` and
`Float32`, checked against `compare_floats`.
- `<=>`, `<`, `<=`, `>` and `>=` on lists, including lists of different
lengths, and on structs, checked against `spark_comparator`.
- Literal folding, wrapping each operand once, `FloatOperands::Raw`
leaving operands alone, and non-comparison operators left alone.
- `NormalizeNaNAndZero` keeping scalars.
- The three new `spark_comparison` tests each fail with the normalization
turned off.
- Results on macOS aarch64 with the default Spark 4.1 profile:
- Unit tests: 1027 in spark-expr, 566 in core.
- `CometSqlFileTestSuite`: 588. `CometNativeReaderSuite`: 86, plus one
test that cancels itself on this Spark version. `CometJoinSuite`: 55.
- `CometExpressionSuite` divide-by-zero tests: 3.
- TPC-DS plan stability v1.4 and v2.7: 129, after regenerating for every
Spark version.
Performance: comparisons in Project and Filter cost what they did before,
since the rewrite already wrapped their operands in the same
`NormalizeNaNAndZero`. Elsewhere a comparison now pays for normalizing its
operands. On an M3 Max, a batch of 8192 doubles takes 11.0 µs for a column
compared with a column, up from 7.2 µs, and 5.9 µs for a column compared with a
literal, up from 4.0 µs. A kernel that compares in Spark's order directly,
without copying the operands, could win that back later.
--
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]
