holdenk commented on code in PR #59403:
URL: https://github.com/apache/spark/pull/59403#discussion_r4342218632
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala:
##########
@@ -78,9 +79,11 @@ object NormalizeFloatingNumbers extends Rule[LogicalPlan] {
// join keys, assuming Joins with join keys are always planned as hash
join or sort merge
// join. It's very unlikely that we will break this assumption in the
near future.
case j @ ExtractEquiJoinKeys(_, leftKeys, rightKeys, condition, _, _, _,
_)
- // The analyzer guarantees left and right joins keys are of the same
data type. Here we
- // only need to check join keys of one side.
- if leftKeys.exists(k => needNormalize(k)) =>
+ // The analyzer guarantees left and right join keys are of the same
data type, but a key
+ // can already be normalized on one side only (e.g. a filter pushed
below an Aggregate
+ // reads the normalized grouping key, and a subquery rewrite turns
it into a join key),
+ // so check both sides.
+ if leftKeys.exists(needNormalize) || rightKeys.exists(needNormalize)
=>
val newLeftJoinKeys = leftKeys.map(normalize)
val newRightJoinKeys = rightKeys.map(normalize)
Review Comment:
Note for self: normalize will skip re-normalizing something that's already
been normalized so we don't need to worry about double applications being
inefficent.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/NormalizeFloatingNumbers.scala:
##########
@@ -100,6 +103,68 @@ object NormalizeFloatingNumbers extends Rule[LogicalPlan] {
// Scaladoc just above.
}
+ /**
+ * Whether `aggregate` outputs its float/double grouping keys normalized.
The planner normalizes
+ * the grouping keys of every Aggregate (see `SparkStrategies.Aggregation`),
except one whose
+ * aggregate functions are all grouped-agg pandas UDFs, planned as
`ArrowAggregatePythonExec` on
+ * the raw keys. Rules that replace an Aggregate with a Project check
[[readsNormalizedKeys]].
Review Comment:
non-blocking: This seems like a good thing to file a follow up JIRA for
consistency; I think we should probably do the normalization on the
ArrowAggregatePythonExec path too and then we can delete some of this code (but
follow up item, non-blocking). I filed
https://issues.apache.org/jira/browse/SPARK-60444 so we don't loose track of
this (if you want to take this on after heck yeah, otherwise I'll add it to our
pile :D )
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -954,6 +954,23 @@ object LimitPushDown extends Rule[LogicalPlan] {
}
}
+ // Pushes `LocalLimit(1)` below a group-only Aggregate under `LIMIT 1` and
replaces the Aggregate
+ // with a Project of its child. If that Project would read float/double keys
that the Aggregate
+ // normalizes, the Aggregate is kept, with the limit still pushed below it.
Review Comment:
This seems fine to me, the other option here would have been to inject
normalization into the projection but this works too.
--
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]
