ctagard commented on PR #59403:
URL: https://github.com/apache/spark/pull/59403#issuecomment-6173745357
Thanks @HyukjinKwon, I addressed all six findings and the decision
challenge, one commit each:
**Aggregate replaced by a Project** (3c36286): I tried normalizing the keys
in the replacement `Project` as suggested, but that can't keep the outputs'
`ExprId`s: `normalize(d) AS d#0` over a child that outputs `d#0` fails plan
validation ("An alias reuses the same expression ID as previously present in an
attribute"), and new `ExprId`s would mean rewriting every parent. So instead
`NormalizeFloatingNumbers` now states the contract once:
- `outputsNormalizedKeys(aggregate)`: when an `Aggregate`'s float/double
keys come out normalized (all aggregates except grouped-agg pandas UDF ones).
`pushBelowAggregate` uses it.
- `readsNormalizedKeys(aggregate, outputs)`: whether a `Project` of those
outputs would read such a key raw. `OptimizeOneRowPlan`, `LimitPushDown`'s two
`LIMIT 1` cases and `RemoveRedundantAggregates`' group-only case don't fire in
that case, so the `Aggregate` stays and the planner normalizes as usual. They
still fire when the float key isn't output. The cost is keeping a one-row /
`LIMIT 1` / already-distinct aggregate on a float key.
The same problem exists in `RemoveRedundantAggregates`' other case (removing
a lower `DISTINCT` under a duplicate-insensitive upper aggregate): with a
filter pushed below the lower aggregate, `SELECT cast(x AS STRING) AS k FROM
(SELECT DISTINCT x FROM t) WHERE cast(x AS STRING) = '0.0' GROUP BY 1` over
`t.x = {-0.0, -0.0}` returned `'-0.0'`. I applied the same check there in a
separate commit (3f543f3). It also fixes the SPARK-60093 repro, which was filed
by someone else; I'm happy to drop that commit if you'd rather keep SPARK-60093
separate. Two existing `RemoveRedundantAggregatesSuite` tests used a `rand(0)`
(double) key: one now uses an integer key so it keeps testing the removal, and
the other expects the upper aggregate to be kept.
**FLOAT key vs DOUBLE/DECIMAL literal** (3ea6c23): a float attribute widened
to double is now exempt as well; other casts are still normalized. A new
end-to-end test checks that `GreaterThan(f,1.5)` reaches the Parquet scan (it
got no pushed filters before), and `PredicateSuite` checks the widened
comparison for raw and normalized FLOAT values.
**Grouped-agg pandas UDF aggregates** (3d03a16): `pushBelowAggregate`
doesn't normalize below an aggregate whose functions are all `PythonUDAF`,
through `outputsNormalizedKeys`. Plan test in `FilterPushdownSuite`, end-to-end
test in `PythonUDFSuite` (run locally with pandas/pyarrow; it fails without the
fix).
**Decision challenge, the `KnownFloatingPointNormalized` short-circuit**
(a7cddb9): agreed. It now only stops at the normalized form of a current
grouping key and recurses otherwise, so your `pow(x, -1)` example returns
`(Infinity, 1)`. Plan, unit and end-to-end tests added. For array and map keys
`normalize` creates new lambda variables, so those are normalized again on a
stacked push, which gives the same values.
**Nits:** the `removePushedDownFilter` Scaladoc is updated (734f939); the
`PredicateSuite` test reads its values from the input row, so generated code
sees raw NaN bits (b577c23); and the new end-to-end tests cover one-row, `LIMIT
1`, `limit(1)`/`first()` and the pandas UDF path with sign-sensitive
predicates. Each fails without its fix.
I also updated the PR description. `catalyst/test`, the TPC-DS plan
stability suites (no golden changes), `SQLQueryTestSuite` and the related
`sql/core` suites pass locally.
--
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]
