Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11121#discussion_r52245464
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
---
@@ -130,6 +131,48 @@ object EliminateSerialization extends
Rule[LogicalPlan] {
}
/**
+ * Pushes down [[LocalLimit]] beneath UNION ALL and beneath the streamed
inputs of outer joins.
+ */
+object LimitPushDown extends Rule[LogicalPlan] {
+
+ private def stripGlobalLimitIfPresent(plan: LogicalPlan): LogicalPlan = {
+ plan match {
+ case GlobalLimit(expr, child) => child
+ case _ => plan
+ }
+ }
+
+ private def maybePushLimit(limitExp: Expression, plan: LogicalPlan):
LogicalPlan = {
+ (limitExp, plan.maxRows) match {
+ case (IntegerLiteral(maxRow), Some(IntegerLiteral(childMaxRows))) if
maxRow < childMaxRows =>
+ LocalLimit(limitExp, stripGlobalLimitIfPresent(plan))
+ case (_, None) =>
+ LocalLimit(limitExp, stripGlobalLimitIfPresent(plan))
+ case _ => plan
+ }
+ }
+
+ def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ // Adding extra Limits below UNION ALL for children which are not
Limit or do not have Limit
+ // descendants whose maxRow is larger. This heuristic is valid
assuming there does not exist any
+ // Limit push-down rule that is unable to infer the value of maxRows.
+ // Note: right now Union means UNION ALL, which does not de-duplicate
rows, so it is safe to
+ // pushdown Limit through it. Once we add UNION DISTINCT, however, we
will not be able to
+ // pushdown Limit.
+ case LocalLimit(exp, Union(children)) =>
+ LocalLimit(exp, Union(children.map(maybePushLimit(exp, _))))
+ case LocalLimit(exp, join @ Join(left, right, joinType, condition)) =>
+ joinType match {
+ case RightOuter => join.copy(right = maybePushLimit(exp, right))
+ case LeftOuter => join.copy(left = maybePushLimit(exp, left))
+ case FullOuter =>
+ join.copy(left = maybePushLimit(exp, left), right =
maybePushLimit(exp, right))
--- End diff --
I have one concern about the rule as implemented in your PR:
If we have a full outer join which initially has neither of its children
limited and then we push a limit to the side with larger statistics, then a
second firing of the `LimitPushDown` rule would match on one of the cases where
only a single side is limited and would push a limit to the other side, leading
to the wrong answer because we would have limited both sides.
Therefore, I think we might want to restrict this rule to only fire in the
case where a neither side of the full outer join has a pre-existing limit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
