ctagard commented on code in PR #59403:
URL: https://github.com/apache/spark/pull/59403#discussion_r4273830172
##########
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:
Updated in 734f939: the Scaladoc now says the Aggregate case calls
`pushBelowAggregate`, including its 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
+ .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:
Done in 3ea6c23: a float attribute widened to double is exempt too;
non-widening casts are still normalized. A new test checks that
`GreaterThan(f,1.5)` reaches the Parquet scan.
##########
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:
Done in b577c23: the values now come from the input row, so generated code
sees the raw NaN bits.
##########
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:
Added in 3c36286, 3f543f3 and 3d03a16: one-row, `LIMIT 1`,
`limit(1)`/`first()`, DISTINCT under an aggregate, and the pandas UDF case,
each with a sign-sensitive predicate. Each fails without its fix.
--
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]
