HyukjinKwon commented on code in PR #59403:
URL: https://github.com/apache/spark/pull/59403#discussion_r4272967058
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala:
##########
@@ -1271,6 +1271,64 @@ class DataFrameAggregateSuite extends SharedSparkSession
)
}
+ test("SPARK-60150: filter above an aggregate sees normalized float/double
grouping keys") {
+ // Two rows, so that OptimizeOneRowPlan does not remove the aggregate.
Review Comment:
**Nit (P3):** The two-row input here, and in the subquery test, is chosen to
avoid `OptimizeOneRowPlan`, and every new test uses built-in aggregates. So
none of them runs a filter pushed below an Aggregate that is later replaced by
a Project (`OptimizeOneRowPlan`, `RemoveRedundantAggregates`, `LIMIT 1`
push-down) or planned as `ArrowAggregatePythonExec` (grouped-agg pandas UDF).
Those are the paths where the normalized pushed filter disagrees with the
output, so their wrong results pass the suite. Tests for one-row, `LIMIT 1` and
`limit(1)` queries, plus a grouped-agg pandas UDF case (for example
`IntegratedUDFTestUtils.TestGroupedAggPandasUDF` in `PythonUDFSuite`), each
with a sign-sensitive predicate, would cover them.
**Verification:**
- **Regression:** One-row, LIMIT 1-child and limit(1)/first() DISTINCT or
GROUP BY queries over -0.0 filtered with a sign-sensitive predicate return only
rows that satisfy it.
- **Regression:** A grouped-agg pandas UDF aggregation over a -0.0 key
filtered with a sign-sensitive predicate returns the same rows as the unpushed
filter.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala:
##########
@@ -239,11 +239,11 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends
Rule[LogicalPlan] with
}
if (removedAny) (u.withNewChildren(newChildren), true) else (u, false)
case agg: Aggregate =>
- // Mirror PushPredicateThroughNonJoin: translate the target through
the grouping
- // aliases with the same helper it uses to move grouping-key filters
below the
- // aggregate (filters referencing aggregate expressions stay up, so
they are always
- // found above this node).
- val translated = replaceAlias(target, getAliasMap(agg))
+ // Mirror PushPredicateThroughNonJoin: translate the target with the
same helper it uses
+ // to move grouping-key filters below the aggregate (filters
referencing aggregate
+ // expressions stay up, so they are always found above this node).
+ val translated =
+
PushPredicateThroughNonJoin.pushBelowAggregate(splitConjunctivePredicates(target),
agg)
Review Comment:
**Nit (P3):** The Aggregate case now goes through `pushBelowAggregate`,
which also normalizes float/double grouping keys. The `removePushedDownFilter`
Scaladoc above (lines 186-191) still says the `Project` and `Aggregate` cases
use "the very same `AliasHelper` utilities" and translate only "through
projection and grouping-key aliases". The class doc of
`PushPredicateThroughNonJoin` asks that this mirror be kept in sync, so the
Scaladoc should say that the Aggregate case reuses `pushBelowAggregate`,
including its grouping-key normalization.
**Verification:**
- **Inspection:** The removePushedDownFilter Scaladoc matches the Aggregate
case's use of pushBelowAggregate, including grouping-key normalization.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2796,6 +2797,49 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
case _ => false
}
+ // Rewrites conjuncts pushed below `aggregate` in terms of the aggregate's
child and ANDs them.
+ // Expressions above the aggregate see its float/double grouping keys
normalized (see
+ // NormalizeFloatingNumbers: -0.0 becomes 0.0 and every NaN becomes one
canonical NaN), so the
+ // pushed conjuncts read them normalized too, wherever the raw value could
change the result.
+ private[optimizer] def pushBelowAggregate(
+ conds: Seq[Expression],
+ aggregate: Aggregate): Expression = {
+ val aliasMap = getAliasMap(aggregate)
+ val normalizedGrouping = aggregate.groupingExpressions
Review Comment:
**Non-blocking (P2):** `normalizedGrouping` is applied to every Aggregate
with float/double keys. In `SparkStrategies.Aggregation`, though, the
all-`PythonUDAF` branch passes the raw grouping expressions to
`ArrowAggregatePythonExec`. Unlike the regular and `PythonAggregate` branches,
it never calls `NormalizeFloatingNumbers.normalize`, so a grouped-agg pandas
UDF aggregate outputs raw keys. `PythonUDAF` is deterministic by default, so
filters are still pushed below it, now in normalized form.
Example with rows `(x=-0.0, v=1.0), (x=-0.0, v=2.0)`:
`df.groupBy("x").agg(mean_udf("v")).where(col("x").cast("string") == "0.0")`
returns the `-0.0` group with this PR, and that key casts to `'-0.0'`. Master
returns nothing. Built-in and `PythonAggregate` aggregates are unaffected.
**Recommended change:** In pushBelowAggregate, skip
normalizeGroupingExpressions when every aggregate function of the Aggregate is
a PythonUDAF, mirroring the SparkStrategies.Aggregation branch that plans it
without key normalization, and document that condition next to the helper. Add
an optimizer plan test for a PythonUDAF aggregate.
**Why this works:** The pushed filter then reads the same raw keys
ArrowAggregatePythonExec outputs, as before the PR, while every normalizing
branch keeps the fix.
**Scope:** Keep pushed filters raw for Aggregates the planner executes
without key normalization.
**Compatibility:** Preserve behavior outside the stated repair boundary.
**Risks:** The skip condition must stay in sync with
SparkStrategies.Aggregation's branch selection if that branch later normalizes
keys.
**Constraints:** Do not change the plans for Aggregates planned by the
normalizing branches.
**Success:** The pushed filter below an all-PythonUDAF Aggregate is the raw
alias-expanded predicate. A grouped-agg pandas UDF aggregation over a -0.0 key
filtered with a sign-sensitive predicate returns the same rows as the unpushed
filter.
**Verification:**
- **Regression:** A grouped-agg pandas UDF aggregation over a -0.0 key
filtered with a sign-sensitive predicate returns the same rows as the unpushed
filter.
- **Behavior:** The pushed filter below a PythonUDAF aggregate is the raw
alias-expanded predicate, while built-in aggregates still get the normalized
form.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2796,6 +2797,49 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
case _ => false
}
+ // Rewrites conjuncts pushed below `aggregate` in terms of the aggregate's
child and ANDs them.
+ // Expressions above the aggregate see its float/double grouping keys
normalized (see
+ // NormalizeFloatingNumbers: -0.0 becomes 0.0 and every NaN becomes one
canonical NaN), so the
+ // pushed conjuncts read them normalized too, wherever the raw value could
change the result.
+ private[optimizer] def pushBelowAggregate(
+ conds: Seq[Expression],
+ aggregate: Aggregate): Expression = {
+ val aliasMap = getAliasMap(aggregate)
+ val normalizedGrouping = aggregate.groupingExpressions
+ .filter(e => NormalizeFloatingNumbers.needNormalize(e.dataType))
+ .map(trimAliases)
+ conds.map { cond =>
+ val replaced = replaceAlias(cond, aliasMap)
+ if (normalizedGrouping.isEmpty) {
+ replaced
+ } else {
+ normalizeGroupingExpressions(replaced, normalizedGrouping)
+ }
+ }.reduce(And)
+ }
+
+ // Rewrites `e` so that it gives the same result on the raw values of
`grouping` as `e` gives on
+ // their normalized values (see NormalizeFloatingNumbers). Each occurrence
of a grouping
+ // expression is wrapped in its normalized form, with one exception: a
float/double attribute
+ // that is a direct operand of one of the comparisons matched below is left
as is. Those
+ // comparisons already treat -0.0 and 0.0 as equal and all NaNs as equal (see
+ // SQLOrderingUtil.compareDoubles and CodegenContext.genEqual), and IsNull,
IsNotNull and IsNaN
+ // cannot tell normalized values apart either, so wrapping would not change
the result. Leaving
+ // them as is keeps plain comparisons usable by data sources and partition
pruning.
+ private[optimizer] def normalizeGroupingExpressions(
+ e: Expression,
+ grouping: Seq[Expression]): Expression = e match {
+ case _: KnownFloatingPointNormalized => e
+ case _: EqualTo | _: EqualNullSafe | _: LessThan | _: LessThanOrEqual | _:
GreaterThan |
+ _: GreaterThanOrEqual | _: In | _: IsNull | _: IsNotNull | _: IsNaN =>
+ e.mapChildren {
+ case a: Attribute if a.dataType == FloatType || a.dataType ==
DoubleType => a
Review Comment:
**Non-blocking (P2):** Only a bare float/double attribute is exempt here. A
FLOAT key compared with a double or decimal literal (`f > 1.5` in SQL, `$"f" >
1.5`) is analyzed as `cast(f AS DOUBLE) > 1.5`, so the key under the `Cast` is
wrapped. `UnwrapCastInBinaryComparison` runs after `PushDownPredicates` and
then produces `knownfloatingpointnormalized(normalizenanandzero(f)) > 1.5f`
instead of `f > 1.5f`. Neither `DataSourceStrategy` nor V2 filter translation
can push that.
As a result, `SELECT * FROM (SELECT f, count(*) FROM parquet_t GROUP BY f)
WHERE f > 1.5` loses Parquet/ORC filter pushdown and partition pruning on a
FLOAT partition column. For DSv2 sources, the leftover post-scan filter also
blocks aggregate pushdown. Results are unchanged. A widening float-to-double
`Cast` of a direct attribute keeps the -0.0/0.0 and NaN equivalence these
comparisons rely on, so it can also be left as is. A non-widening cast, such as
to STRING, must still be normalized.
**Verification:**
- **Behavior:** Comparisons and IN lists between a FLOAT grouping key and
double or decimal literals are pushed below the Aggregate in a form data
sources can translate.
- **Regression:** A sign-sensitive predicate through a non-widening cast
still reads the normalized key.
- **Behavior:** The widened comparison gives the same result for raw and
normalized FLOAT values, including -0.0 and NaN bit patterns.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/PredicateSuite.scala:
##########
@@ -571,6 +572,39 @@ class PredicateSuite extends SparkFunSuite with
ExpressionEvalHelper {
checkEvaluation(GreaterThan(Literal(0.0F), Literal(-0.0F)), false)
}
+ test("SPARK-60150: comparisons exempt from normalization agree with
normalized values") {
+ // PushPredicateThroughNonJoin pushes these comparisons below an Aggregate
without
+ // normalizing a float/double grouping key that is a direct operand, so
each must give the
+ // same result for a value and for its normalized form: -0.0 and 0.0, and
NaNs with different
+ // bit patterns.
+ val binary: Seq[(Expression, Expression) => Expression] = Seq(
+ EqualTo, EqualNullSafe, LessThan, LessThanOrEqual, GreaterThan,
GreaterThanOrEqual,
+ (l, r) => In(l, Seq(r)))
+ val unary: Seq[Expression => Expression] = Seq(IsNull, IsNotNull, IsNaN)
+
+ def check(dataType: DataType, values: Seq[Any]): Unit = {
+ def lit(v: Any): Expression = Literal.create(v, dataType)
Review Comment:
**Nit (P3):** `lit` builds every operand as a `Literal`, and `Literal`
codegen emits any NaN as `Double.NaN` / `Float.NaN`. The generated-code checks
inside `checkEvaluation` therefore never see the bits of `otherDoubleNaN` /
`otherFloatNaN`; only interpreted evaluation does. Pushed filters run as
generated code on input-row values that keep their raw bits. If generated code
for one of these comparisons started to distinguish NaN bit patterns, this test
would still pass. Passing the values through an input row (bound references)
instead of literals would cover both paths.
**Verification:**
- **Behavior:** Each exempt comparison gives the same result for a raw value
and its normalized form under interpreted and generated evaluation, with raw
NaN bits supplied through the input row.
--
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]
