[SPARK-12505] [SQL] Pushdown a Limit on top of an Outer-Join #10454

gatorsmile · 2015-12-23T19:54:56Z

"Rule that applies to a Limit on top of an OUTER Join. The original Limit won't go away after applying this rule, but additional Limit node(s) will be created on top of the outer-side child (or children if it's a FULL OUTER Join). "
– from https://issues.apache.org/jira/browse/CALCITE-832

Also, the same topic in Hive: https://issues.apache.org/jira/browse/HIVE-11684 This has been merged to HIVE.

This PR is for performance improvement. The idea is like predicate pushdown. It can reduce the number of rows processed by Outer Joins. The performance improvement is significant when the number of rows in Limit is small but the cost of Join is large.

For example,

    val df = Seq((1, 2, "1"), (3, 4, "3")).toDF("int", "int2", "str").as('a)
    val df2 = Seq((1, 3, "1"), (5, 6, "5")).toDF("int", "int2", "str").as('b)
    val join = df.join(df2, $"a.int" === $"a.int", "outer").limit(1)

After the improvement, we can see the changes in the optimized plan:

== Analyzed Logical Plan ==
int: int, int2: int, str: string, int: int, int2: int, str: string
Limit 1
+- Join FullOuter, Some((int#3 = int#3))
   :- Subquery a
   :  +- Project [_1#0 AS int#3,_2#1 AS int2#4,_3#2 AS str#5]
   :     +- LocalRelation [_1#0,_2#1,_3#2], [[1,2,1],[3,4,3]]
   +- Subquery b
      +- Project [_1#6 AS int#9,_2#7 AS int2#10,_3#8 AS str#11]
         +- LocalRelation [_1#6,_2#7,_3#8], [[1,3,1],[5,6,5]]

== Optimized Logical Plan ==
Limit 1
+- Join FullOuter, Some((int#3 = int#3))
   :- LocalRelation [int#3,int2#4,str#5], [[1,2,1],[3,4,3]]
   +- Limit 1  <---- extra limit is added
      +- LocalRelation [int#9,int2#10,str#11], [[1,3,1],[5,6,5]]

yhuai · 2015-12-23T20:25:38Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+        case _ => f // DO Nothing for the other join types
+      }
+  }
+}


Is it possible that we add an extra limit in every iteration ?

You are right, but, after this rule, there exists another rule CombineLimits, which can combine the extra limits. I think we should still fix it anyway. Thanks!

yhuai · 2015-12-23T20:57:18Z

Can you add an example in the description to illustrate this optimization (and the rational of it)?

SparkQA · 2015-12-23T21:33:05Z

Test build #48252 has finished for PR 10454 at commit 3779874.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2015-12-23T23:03:05Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+object PushLimitThroughOuterJoin extends Rule[LogicalPlan] with PredicateHelper {
+
+  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    case f @ Limit(expr, Join(left, right, joinType, joinCondition)) =>


We also can push it down through the Join's Projection.

SparkQA · 2015-12-24T00:39:43Z

Test build #48263 has finished for PR 10454 at commit 72f73fb.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2015-12-24T02:02:30Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+    case f @ Limit(expr, Join(left, right, joinType, joinCondition)) =>
+      joinType match {
+        case RightOuter =>
+          Limit(expr, Join(left, CombineLimits(Limit(expr, right)), joinType, joinCondition))


need a stop condition to stop pushing Limit if it's already pushed.

Thank you for your review! Since we call CombineLimits here to combine two consecutive Limits, we will not add extra Limit in the subsequent iteration. Thus, no change will be made in the plan. Thus, RuleExecutor will stop automatically. Am I right?

ah, it's right, nvm

That is right. However I think checking if it is already pushed would reduce unnecessary multiple applying this rule and CombineLimits. But the result should be the same. So it is minor.

SparkQA · 2015-12-24T09:35:59Z

Test build #48298 has finished for PR 10454 at commit ee91303.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2015-12-24T17:04:34Z

The new fix contains two parts:

Since the full outer join will remove the duplicates, we are unable to add the extra limit to both sides. As long as we can ensure the completeness of one Child, the generated results will be still correct like the left/right outer join.
There are Projection on top of most Join operators. We also can push through this kind of combination.

Thank you for your reviews and comments! @yhuai @cloud-fan @viirya

rxin · 2016-01-12T01:57:35Z

@gatorsmile let's close the limit push down pull requests. We will need to design this more properly because it is expensive to push down large limits.

gatorsmile · 2016-01-12T02:00:14Z

Sure, let me close it.

This patch adds a new optimizer rule for performing limit pushdown. Limits will now be pushed down in two cases: - If a limit is on top of a `UNION ALL` operator, then a partition-local limit operator will be pushed to each of the union operator's children. - If a limit is on top of an `OUTER JOIN` then a partition-local limit will be pushed to one side of the join. For `LEFT OUTER` and `RIGHT OUTER` joins, the limit will be pushed to the left and right side, respectively. For `FULL OUTER` join, we will only push limits when at most one of the inputs is already limited: if one input is limited we will push a smaller limit on top of it and if neither input is limited then we will limit the input which is estimated to be larger. These optimizations were proposed previously by gatorsmile in #10451 and #10454, but those earlier PRs were closed and deferred for later because at that time Spark's physical `Limit` operator would trigger a full shuffle to perform global limits so there was a chance that pushdowns could actually harm performance by causing additional shuffles/stages. In #7334, we split the `Limit` operator into separate `LocalLimit` and `GlobalLimit` operators, so we can now push down only local limits (which don't require extra shuffles). This patch is based on both of gatorsmile's patches, with changes and simplifications due to partition-local-limiting. When we push down the limit, we still keep the original limit in place, so we need a mechanism to ensure that the optimizer rule doesn't keep pattern-matching once the limit has been pushed down. In order to handle this, this patch adds a `maxRows` method to `SparkPlan` which returns the maximum number of rows that the plan can compute, then defines the pushdown rules to only push limits to children if the children's maxRows are greater than the limit's maxRows. This idea is carried over from #10451; see that patch for additional discussion. Author: Josh Rosen <joshrosen@databricks.com> Closes #11121 from JoshRosen/limit-pushdown-2.

gatorsmile added 2 commits December 23, 2015 11:48

outerjoin limit pushdown.

8df3e0d

revert.

3779874

yhuai reviewed Dec 23, 2015
View reviewed changes

combine the limits.

72f73fb

gatorsmile reviewed Dec 23, 2015
View reviewed changes

cloud-fan reviewed Dec 24, 2015
View reviewed changes

fix the bug.

ee91303

gatorsmile closed this Jan 12, 2016

gatorsmile mentioned this pull request Feb 5, 2016

[SPARK-8964] [SQL] Use Exchange to perform shuffle in Limit #7334

Closed

JoshRosen mentioned this pull request Feb 8, 2016

[SPARK-12503][SPARK-12505] Limit pushdown in UNION ALL and OUTER JOIN #11121

Closed

gatorsmile mentioned this pull request Nov 3, 2017

[SPARK-22211][SQL] Remove incorrect FOJ limit pushdown #19647

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-12505] [SQL] Pushdown a Limit on top of an Outer-Join #10454

[SPARK-12505] [SQL] Pushdown a Limit on top of an Outer-Join #10454

gatorsmile commented Dec 23, 2015

yhuai Dec 23, 2015

gatorsmile Dec 23, 2015

yhuai commented Dec 23, 2015

SparkQA commented Dec 23, 2015

gatorsmile Dec 23, 2015

SparkQA commented Dec 24, 2015

cloud-fan Dec 24, 2015

gatorsmile Dec 24, 2015

cloud-fan Dec 24, 2015

gatorsmile Dec 24, 2015

viirya Dec 24, 2015

SparkQA commented Dec 24, 2015

gatorsmile commented Dec 24, 2015

rxin commented Jan 12, 2016

gatorsmile commented Jan 12, 2016

[SPARK-12505] [SQL] Pushdown a Limit on top of an Outer-Join #10454

[SPARK-12505] [SQL] Pushdown a Limit on top of an Outer-Join #10454

Conversation

gatorsmile commented Dec 23, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yhuai commented Dec 23, 2015

SparkQA commented Dec 23, 2015

Choose a reason for hiding this comment

SparkQA commented Dec 24, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 24, 2015

gatorsmile commented Dec 24, 2015

rxin commented Jan 12, 2016

gatorsmile commented Jan 12, 2016