[SPARK-26138][SQL] Pushdown limit through InnerLike when condition is empty #31567

wangyum · 2021-02-15T12:56:56Z

What changes were proposed in this pull request?

This pr pushdown limit through InnerLike when condition is empty(Origin pr: #23104). For example:

CREATE TABLE t1 using parquet AS SELECT id AS a, id AS b FROM range(2);
CREATE TABLE t2 using parquet AS SELECT id AS d FROM range(2);
SELECT * FROM t1 CROSS JOIN t2 LIMIT 10;

Before this pr:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- CollectLimit 10
   +- BroadcastNestedLoopJoin BuildRight, Cross
      :- FileScan parquet default.t1[a#5L,b#6L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/tg/f5mz46090wg7swzgdc69f8q03965_0/T/warehous..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:bigint,b:bigint>
      +- BroadcastExchange IdentityBroadcastMode, [id=#43]
         +- FileScan parquet default.t2[d#7L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/tg/f5mz46090wg7swzgdc69f8q03965_0/T/warehous..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<d:bigint>

After this pr:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- CollectLimit 10
   +- BroadcastNestedLoopJoin BuildRight, Cross
      :- LocalLimit 10
      :  +- FileScan parquet default.t1[a#5L,b#6L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/tg/f5mz46090wg7swzgdc69f8q03965_0/T/warehous..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:bigint,b:bigint>
      +- BroadcastExchange IdentityBroadcastMode, [id=#51]
         +- LocalLimit 10
            +- FileScan parquet default.t2[d#7L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/private/var/folders/tg/f5mz46090wg7swzgdc69f8q03965_0/T/warehous..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<d:bigint>

Why are the changes needed?

Improve query performance.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Unit test.

SparkQA · 2021-02-15T14:43:43Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39735/

SparkQA · 2021-02-15T15:18:28Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39735/

SparkQA · 2021-02-15T18:09:00Z

Test build #135154 has finished for PR 31567 at commit a0a0685.

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

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

SparkQA · 2021-02-16T12:45:20Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39759/

SparkQA · 2021-02-16T13:16:13Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39759/

SparkQA · 2021-02-16T17:24:13Z

Test build #135178 has finished for PR 31567 at commit 57171fe.

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

HyukjinKwon · 2021-02-17T01:54:56Z

Looks making sense. cc @cloud-fan, @maropu, @viirya FYI

viirya · 2021-02-17T03:52:02Z

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushdownSuite.scala

@@ -194,4 +194,22 @@ class LimitPushdownSuite extends PlanTest {
      LocalLimit(1, y.groupBy(Symbol("b"))(count(1))))).analyze
    comparePlans(expected2, optimized2)
  }
+
+  test("SPARK-26138: pushdown limit through InnerLike when condition is empty") {


Shall we have a e2e test to check answer too?

viirya

Looks okay. Better to have e2e test for this.

c21 · 2021-02-17T05:42:03Z

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

    // We also need to ensure that this limit pushdown rule will not eventually introduce limits
    // on both sides if it is applied multiple times. Therefore:
    //   - If one side is already limited, stack another limit on top if the new limit is smaller.
    //     The redundant limit will be collapsed by the CombineLimits rule.
-    case LocalLimit(exp, join @ Join(left, right, joinType, _, _)) =>
+    case LocalLimit(exp, join @ Join(left, right, joinType, conditionOpt, _)) =>
      val newJoin = joinType match {
        case RightOuter => join.copy(right = maybePushLocalLimit(exp, right))
        case LeftOuter => join.copy(left = maybePushLocalLimit(exp, left))


It seems that we can also push down limit into left side, for LEFT SEMI and LEFT ANTI join, right?
I can create a minor PR if it's not on your plan @wangyum , thanks.

It seems we can not pushdown LEFT SEMI JOIN, for example:

spark.range(20).selectExpr("id % 10 as id").repartition(1).write.saveAsTable("t1") spark.range(5, 9, 1).repartition(1).write.saveAsTable("t2") val df = spark.sql("select * from t1 LEFT SEMI JOIN t2 on t1.id = t2.id limit 3") df.explain() df.show

Current:

== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- CollectLimit 3 +- BroadcastHashJoin [id#10L], [id#11L], LeftSemi, BuildRight, false :- Filter isnotnull(id#10L) : +- FileScan parquet default.t1[id#10L] Batched: true, DataFilters: [isnotnull(id#10L)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/yumwang/spark/SPARK-28169/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(id)], ReadSchema: struct<id:bigint> +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [id=#69] +- Filter isnotnull(id#11L) +- FileScan parquet default.t2[id#11L] Batched: true, DataFilters: [isnotnull(id#11L)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/yumwang/spark/SPARK-28169/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(id)], ReadSchema: struct<id:bigint> +---+ | id| +---+ | 5| | 6| | 7| +---+

Pushdown:

== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- CollectLimit 3 +- BroadcastHashJoin [id#10L], [id#11L], LeftSemi, BuildRight, false :- LocalLimit 3 : +- Filter isnotnull(id#10L) : +- LocalLimit 3 : +- FileScan parquet default.t1[id#10L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/yumwang/spark/SPARK-28169/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:bigint> +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [id=#77] +- Filter isnotnull(id#11L) +- FileScan parquet default.t2[id#11L] Batched: true, DataFilters: [isnotnull(id#11L)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/yumwang/spark/SPARK-28169/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(id)], ReadSchema: struct<id:bigint> +---+ | id| +---+ +---+

@wangyum - yes, you are right. My bad. LEFT OUTER/RIGHT OUTER join will output the left/right side rows anyway no matter of being matched or not. So they are safe for limit push down, but not LEFT SEMI/LEFT ANTI.

Actually, what about LEFT SEMI / LEFT ANTI join without condition? They should be planned into physical operator BroadcastNestedLoopJoin instead.

maropu · 2021-02-18T01:29:44Z

okay, it looks fine except for the last @viirya comment. I think it's better to add an end-2-end test just in case, too.

SparkQA · 2021-02-18T05:02:36Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39797/

SparkQA · 2021-02-18T05:20:58Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39797/

SparkQA · 2021-02-18T08:14:32Z

Test build #135215 has finished for PR 31567 at commit a9a1955.

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

# Conflicts: # sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

SparkQA · 2021-02-23T04:04:07Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39942/

SparkQA · 2021-02-23T04:34:15Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39942/

SparkQA · 2021-02-23T07:45:17Z

Test build #135362 has finished for PR 31567 at commit e0915b9.

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

wangyum · 2021-02-24T01:55:22Z

Thank you all. Merged to master.

cloud-fan · 2021-02-24T04:56:51Z

late LGTM

### What changes were proposed in this pull request? I found out during code review of #31567 (comment), where we can push down limit to the left side of LEFT SEMI and LEFT ANTI join, if the join condition is empty. Why it's safe to push down limit: The semantics of LEFT SEMI join without condition: (1). if right side is non-empty, output all rows from left side. (2). if right side is empty, output nothing. The semantics of LEFT ANTI join without condition: (1). if right side is non-empty, output nothing. (2). if right side is empty, output all rows from left side. With the semantics of output all rows from left side or nothing (all or nothing), it's safe to push down limit to left side. NOTE: LEFT SEMI / LEFT ANTI join with non-empty condition is not safe for limit push down, because output can be a portion of left side rows. Reference: physical operator implementation for LEFT SEMI / LEFT ANTI join without condition - https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNestedLoopJoinExec.scala#L200-L204 . ### Why are the changes needed? Better performance. Save CPU and IO for these joins, as limit being pushed down before join. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added unit test in `LimitPushdownSuite.scala` and `SQLQuerySuite.scala`. Closes #31630 from c21/limit-pushdown. Authored-by: Cheng Su <chengsu@fb.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…s empty ### What changes were proposed in this pull request? Similar to #31567. This PR enhances `LimitPushDown` to support push local limit to both sides if it is outer join and join condition is empty. It is safe to push down because without join condition is actually a cross join. ### Why are the changes needed? Improve query performance. For example: <img width="400" alt="image" src="https://user-images.githubusercontent.com/5399861/184052707-ebf50748-6870-4650-84c3-65d79b18ba9d.png"> ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Unit test. Closes #37475 from wangyum/SPARK-40040. Lead-authored-by: Yuming Wang <yumwang@ebay.com> Co-authored-by: Yuming Wang <wgyumg@gmail.com> Signed-off-by: Yuming Wang <yumwang@ebay.com>

SPARK-26138: pushdown limit through InnerLike when condition is empty

a0a0685

github-actions bot added the SQL label Feb 15, 2021

HyukjinKwon reviewed Feb 16, 2021

View reviewed changes

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

fix comments

57171fe

viirya reviewed Feb 17, 2021

View reviewed changes

c21 reviewed Feb 17, 2021

View reviewed changes

fix

a9a1955

Merge remote-tracking branch 'upstream/master' into SPARK-26138

e0915b9

# Conflicts: # sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

maropu approved these changes Feb 24, 2021

View reviewed changes

viirya approved these changes Feb 24, 2021

View reviewed changes

wangyum closed this in b5afff5 Feb 24, 2021

wangyum deleted the SPARK-26138 branch February 24, 2021 01:55

c21 mentioned this pull request Feb 24, 2021

[SPARK-34514][SQL] Push down limit for LEFT SEMI and LEFT ANTI join #31630

Closed

wangyum mentioned this pull request Aug 23, 2022

[SPARK-40040][SQL] Push local limit to both sides if join condition is empty #37475

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-26138][SQL] Pushdown limit through InnerLike when condition is empty #31567

[SPARK-26138][SQL] Pushdown limit through InnerLike when condition is empty #31567

wangyum commented Feb 15, 2021 •

edited

Loading

SparkQA commented Feb 15, 2021

SparkQA commented Feb 15, 2021

SparkQA commented Feb 15, 2021

SparkQA commented Feb 16, 2021

SparkQA commented Feb 16, 2021

SparkQA commented Feb 16, 2021

HyukjinKwon commented Feb 17, 2021

viirya Feb 17, 2021

viirya left a comment

c21 Feb 17, 2021

wangyum Feb 17, 2021

c21 Feb 17, 2021

c21 Feb 17, 2021

maropu commented Feb 18, 2021

SparkQA commented Feb 18, 2021

SparkQA commented Feb 18, 2021

SparkQA commented Feb 18, 2021

SparkQA commented Feb 23, 2021

SparkQA commented Feb 23, 2021

SparkQA commented Feb 23, 2021

wangyum commented Feb 24, 2021

cloud-fan commented Feb 24, 2021

[SPARK-26138][SQL] Pushdown limit through InnerLike when condition is empty #31567

[SPARK-26138][SQL] Pushdown limit through InnerLike when condition is empty #31567

Conversation

wangyum commented Feb 15, 2021 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

SparkQA commented Feb 15, 2021

SparkQA commented Feb 15, 2021

SparkQA commented Feb 15, 2021

SparkQA commented Feb 16, 2021

SparkQA commented Feb 16, 2021

SparkQA commented Feb 16, 2021

HyukjinKwon commented Feb 17, 2021

viirya Feb 17, 2021

Choose a reason for hiding this comment

viirya left a comment

Choose a reason for hiding this comment

c21 Feb 17, 2021

Choose a reason for hiding this comment

wangyum Feb 17, 2021

Choose a reason for hiding this comment

c21 Feb 17, 2021

Choose a reason for hiding this comment

c21 Feb 17, 2021

Choose a reason for hiding this comment

maropu commented Feb 18, 2021

SparkQA commented Feb 18, 2021

SparkQA commented Feb 18, 2021

SparkQA commented Feb 18, 2021

SparkQA commented Feb 23, 2021

SparkQA commented Feb 23, 2021

SparkQA commented Feb 23, 2021

wangyum commented Feb 24, 2021

cloud-fan commented Feb 24, 2021

wangyum commented Feb 15, 2021 •

edited

Loading