[SPARK-32282][SQL] Improve EnsureRquirement.reorderJoinKeys to handle more scenarios such as PartitioningCollection #29074

imback82 · 2020-07-12T02:59:14Z

What changes were proposed in this pull request?

This PR proposes to improve EnsureRquirement.reorderJoinKeys to handle the following scenarios:

If the keys cannot be reordered to match the left-side HashPartitioning, consider the right-side HashPartitioning.
Handle PartitioningCollection, which may contain HashPartitioning

Why are the changes needed?

For the scenario 1), the current behavior matches either the left-side HashPartitioning or the right-side HashPartitioning. This means that if both sides are HashPartitioning, it will try to match only the left side.
The following will not consider the right-side HashPartitioning:

val df1 = (0 until 10).map(i => (i % 5, i % 13)).toDF("i1", "j1")
val df2 = (0 until 10).map(i => (i % 7, i % 11)).toDF("i2", "j2")
df1.write.format("parquet").bucketBy(4, "i1", "j1").saveAsTable("t1")df2.write.format("parquet").bucketBy(4, "i2", "j2").saveAsTable("t2")
val t1 = spark.table("t1")
val t2 = spark.table("t2")
val join = t1.join(t2, t1("i1") === t2("j2") && t1("i1") === t2("i2"))
 join.explain

== Physical Plan ==
*(5) SortMergeJoin [i1#26, i1#26], [j2#31, i2#30], Inner
:- *(2) Sort [i1#26 ASC NULLS FIRST, i1#26 ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(i1#26, i1#26, 4), true, [id=#69]
:     +- *(1) Project [i1#26, j1#27]
:        +- *(1) Filter isnotnull(i1#26)
:           +- *(1) ColumnarToRow
:              +- FileScan parquet default.t1[i1#26,j1#27] Batched: true, DataFilters: [isnotnull(i1#26)], Format: Parquet, Location: InMemoryFileIndex[..., PartitionFilters: [], PushedFilters: [IsNotNull(i1)], ReadSchema: struct<i1:int,j1:int>, SelectedBucketsCount: 4 out of 4
+- *(4) Sort [j2#31 ASC NULLS FIRST, i2#30 ASC NULLS FIRST], false, 0.
   +- Exchange hashpartitioning(j2#31, i2#30, 4), true, [id=#79].       <===== This can be removed
      +- *(3) Project [i2#30, j2#31]
         +- *(3) Filter (((j2#31 = i2#30) AND isnotnull(j2#31)) AND isnotnull(i2#30))
            +- *(3) ColumnarToRow
               +- FileScan parquet default.t2[i2#30,j2#31] Batched: true, DataFilters: [(j2#31 = i2#30), isnotnull(j2#31), isnotnull(i2#30)], Format: Parquet, Location: InMemoryFileIndex[..., PartitionFilters: [], PushedFilters: [IsNotNull(j2), IsNotNull(i2)], ReadSchema: struct<i2:int,j2:int>, SelectedBucketsCount: 4 out of 4

For the scenario 2), the current behavior does not handle PartitioningCollection:

val df1 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i1", "j1")
val df2 = (0 until 100).map(i => (i % 7, i % 11)).toDF("i2", "j2")
val df3 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i3", "j3")
val join = df1.join(df2, df1("i1") === df2("i2") && df1("j1") === df2("j2")) // PartitioningCollection
val join2 = join.join(df3, join("j1") === df3("j3") && join("i1") === df3("i3"))
join2.explain

== Physical Plan ==
*(9) SortMergeJoin [j1#8, i1#7], [j3#30, i3#29], Inner
:- *(6) Sort [j1#8 ASC NULLS FIRST, i1#7 ASC NULLS FIRST], false, 0.       <===== This can be removed
:  +- Exchange hashpartitioning(j1#8, i1#7, 5), true, [id=#58]             <===== This can be removed
:     +- *(5) SortMergeJoin [i1#7, j1#8], [i2#18, j2#19], Inner
:        :- *(2) Sort [i1#7 ASC NULLS FIRST, j1#8 ASC NULLS FIRST], false, 0
:        :  +- Exchange hashpartitioning(i1#7, j1#8, 5), true, [id=#45]
:        :     +- *(1) Project [_1#2 AS i1#7, _2#3 AS j1#8]
:        :        +- *(1) LocalTableScan [_1#2, _2#3]
:        +- *(4) Sort [i2#18 ASC NULLS FIRST, j2#19 ASC NULLS FIRST], false, 0
:           +- Exchange hashpartitioning(i2#18, j2#19, 5), true, [id=#51]
:              +- *(3) Project [_1#13 AS i2#18, _2#14 AS j2#19]
:                 +- *(3) LocalTableScan [_1#13, _2#14]
+- *(8) Sort [j3#30 ASC NULLS FIRST, i3#29 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(j3#30, i3#29, 5), true, [id=#64]
      +- *(7) Project [_1#24 AS i3#29, _2#25 AS j3#30]
         +- *(7) LocalTableScan [_1#24, _2#25]

Does this PR introduce any user-facing change?

Yes, now from the above examples, the shuffle/sort nodes pointed by This can be removed are now removed:

Senario 1):

== Physical Plan ==
*(4) SortMergeJoin [i1#26, i1#26], [i2#30, j2#31], Inner
:- *(2) Sort [i1#26 ASC NULLS FIRST, i1#26 ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(i1#26, i1#26, 4), true, [id=#67]
:     +- *(1) Project [i1#26, j1#27]
:        +- *(1) Filter isnotnull(i1#26)
:           +- *(1) ColumnarToRow
:              +- FileScan parquet default.t1[i1#26,j1#27] Batched: true, DataFilters: [isnotnull(i1#26)], Format: Parquet, Location: InMemoryFileIndex[..., PartitionFilters: [], PushedFilters: [IsNotNull(i1)], ReadSchema: struct<i1:int,j1:int>, SelectedBucketsCount: 4 out of 4
+- *(3) Sort [i2#30 ASC NULLS FIRST, j2#31 ASC NULLS FIRST], false, 0
   +- *(3) Project [i2#30, j2#31]
      +- *(3) Filter (((j2#31 = i2#30) AND isnotnull(j2#31)) AND isnotnull(i2#30))
         +- *(3) ColumnarToRow
            +- FileScan parquet default.t2[i2#30,j2#31] Batched: true, DataFilters: [(j2#31 = i2#30), isnotnull(j2#31), isnotnull(i2#30)], Format: Parquet, Location: InMemoryFileIndex[..., PartitionFilters: [], PushedFilters: [IsNotNull(j2), IsNotNull(i2)], ReadSchema: struct<i2:int,j2:int>, SelectedBucketsCount: 4 out of 4

Scenario 2):

== Physical Plan ==
*(8) SortMergeJoin [i1#7, j1#8], [i3#29, j3#30], Inner
:- *(5) SortMergeJoin [i1#7, j1#8], [i2#18, j2#19], Inner
:  :- *(2) Sort [i1#7 ASC NULLS FIRST, j1#8 ASC NULLS FIRST], false, 0
:  :  +- Exchange hashpartitioning(i1#7, j1#8, 5), true, [id=#43]
:  :     +- *(1) Project [_1#2 AS i1#7, _2#3 AS j1#8]
:  :        +- *(1) LocalTableScan [_1#2, _2#3]
:  +- *(4) Sort [i2#18 ASC NULLS FIRST, j2#19 ASC NULLS FIRST], false, 0
:     +- Exchange hashpartitioning(i2#18, j2#19, 5), true, [id=#49]
:        +- *(3) Project [_1#13 AS i2#18, _2#14 AS j2#19]
:           +- *(3) LocalTableScan [_1#13, _2#14]
+- *(7) Sort [i3#29 ASC NULLS FIRST, j3#30 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(i3#29, j3#30, 5), true, [id=#58]
      +- *(6) Project [_1#24 AS i3#29, _2#25 AS j3#30]
         +- *(6) LocalTableScan [_1#24, _2#25]

How was this patch tested?

Added tests.

imback82 · 2020-07-12T03:22:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

+            leftKeys, rightKeys, UnknownPartitioning(0), rightPartitioning))
+      case (_, HashPartitioning(rightExpressions, _)) =>
+        reorder(leftKeys.toIndexedSeq, rightKeys.toIndexedSeq, rightExpressions, rightKeys)
+          .orElse(reorderJoinKeysRecursively(


This can be also implemented by looking at left partitioning first then move to the right partitionoing:

(leftPartitioning, rightPartitioning) match { case (HashPartitioning(leftExpressions, _), _) => reorder(leftKeys.toIndexedSeq, rightKeys.toIndexedSeq, leftExpressions, leftKeys) .orElse(reorderJoinKeysRecursively( leftKeys, rightKeys, UnknownPartitioning(0), rightPartitioning)) case (PartitioningCollection(partitionings), _) => partitionings.foreach { p => reorderJoinKeysRecursively(leftKeys, rightKeys, p, rightPartitioning).map { k => return Some(k) } } reorderJoinKeysRecursively(leftKeys, rightKeys, UnknownPartitioning(0), rightPartitioning) case (_, HashPartitioning(rightExpressions, _)) => reorder(leftKeys.toIndexedSeq, rightKeys.toIndexedSeq, rightExpressions, rightKeys) case (_, PartitioningCollection(partitionings)) => partitionings.foreach { p => reorderJoinKeysRecursively(leftKeys, rightKeys, leftPartitioning, p).map { k => return Some(k) } } None case _ => None }

However, I chose this way so that the behavior remains the same. If you have leftPartitioning = PartitioningCollection and rightPartitioning = HashPartitioning, it will match the rightPartitioning first, which is the existing behavior.

SparkQA · 2020-07-12T07:05:02Z

Test build #125704 has finished for PR 29074 at commit 99493e4.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

imback82 · 2020-07-12T18:09:24Z

retest this please

SparkQA · 2020-07-12T23:49:43Z

Test build #125722 has finished for PR 29074 at commit 99493e4.

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

imback82 · 2020-07-13T00:09:40Z

cc: @maropu @cloud-fan

maropu · 2020-07-13T00:44:04Z

also cc: @viirya

SparkQA · 2020-07-15T16:22:03Z

Test build #125890 has finished for PR 29074 at commit ab237bc.

This patch fails to build.
This patch does not merge cleanly.
This patch adds no public classes.

SparkQA · 2020-07-16T06:59:36Z

Test build #125917 has finished for PR 29074 at commit 8308649.

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

imback82 · 2020-07-27T20:32:24Z

Gentle ping @cloud-fan / @maropu / @viirya

SparkQA · 2020-08-02T04:35:17Z

Test build #126928 has finished for PR 29074 at commit 268326b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

SparkQA · 2020-08-07T06:56:57Z

Test build #127164 has finished for PR 29074 at commit e5b078f.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class EnsureRequirementsSuite extends SharedSparkSession

maropu

Looks okay except for the existing minor comments. @cloud-fan @viirya

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala

SparkQA · 2020-08-07T22:27:46Z

Test build #127213 has finished for PR 29074 at commit 89ad6ef.

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

maropu · 2020-08-22T11:25:54Z

retest this please

SparkQA · 2020-08-22T16:16:25Z

Test build #127775 has finished for PR 29074 at commit 89ad6ef.

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

c21

Thanks @imback82 for adding this. This is important for complex bucketed-related queries to avoid shuffle. LGTM, just a nit.

sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala

SparkQA · 2020-08-24T22:04:02Z

Test build #127850 has finished for PR 29074 at commit 1729c8b.

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

imback82 · 2020-08-31T05:33:52Z

Gentle ping. @cloud-fan, do you think this PR can move forward? Thanks in advance!

maropu · 2020-09-09T13:50:59Z

retest this please

SparkQA · 2020-09-09T19:34:17Z

Test build #128455 has finished for PR 29074 at commit 1729c8b.

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

maropu · 2020-09-28T12:14:10Z

retest this please

SparkQA · 2020-09-28T13:05:46Z

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

SparkQA · 2020-09-28T13:30:48Z

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

SparkQA · 2020-09-28T16:50:13Z

Test build #129183 has finished for PR 29074 at commit 1729c8b.

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

cloud-fan · 2020-10-06T15:34:07Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

+      case (Some(PartitioningCollection(partitionings)), _) =>
+        partitionings.foreach { p =>
+          reorderJoinKeysRecursively(leftKeys, rightKeys, Some(p), rightPartitioning).map { k =>
+            return Some(k)


nit:

partitionings.foldLeft(None) { (res, p) => res.orElse(reorderJoinKeysRecursively...) }.getOrElse(reorderJoinKeysRecursively(leftKeys, rightKeys, None, rightPartitioning))

Thanks, updated.

cloud-fan · 2020-10-06T15:36:03Z

sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala

+          ShuffleExchangeExec(HashPartitioning(rightPartitioningExpressions, _), _, _), _), _) =>
+        assert(leftKeys !== smjExec1.leftKeys)
+        assert(rightKeys !== smjExec1.rightKeys)
+        assert(leftKeys === leftPartitionings.head.asInstanceOf[HashPartitioning].expressions)


can we simply check leftKeys === Seq(exprA, exprB)?

OK. I simplified the checks in this test.

cloud-fan

LGTM except some minor comments

SparkQA · 2020-10-07T02:28:02Z

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

SparkQA · 2020-10-07T02:52:08Z

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

SparkQA · 2020-10-07T06:19:19Z

Test build #129483 has finished for PR 29074 at commit 10b4d5a.

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

cloud-fan · 2020-10-07T06:45:33Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

+          res.orElse(reorderJoinKeysRecursively(leftKeys, rightKeys, Some(p), rightPartitioning))
+        }.orElse(reorderJoinKeysRecursively(leftKeys, rightKeys, None, rightPartitioning))
+      case (_, Some(PartitioningCollection(partitionings))) =>
+        partitionings.foreach { p =>


can you do the same refactor here?

cloud-fan · 2020-10-07T06:46:29Z

sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala

+      case SortMergeJoinExec(leftKeys, rightKeys, _, _,
+        SortExec(_, _, DummySparkPlan(_, _, _: PartitioningCollection, _, _), _),
+        SortExec(_, _, ShuffleExchangeExec(_: HashPartitioning, _, _), _), _) =>
+        assert(leftKeys !== smjExec1.leftKeys)


is this check needed? We already check leftKeys === Seq(exprA, exprB) and it's obvious that leftKeys !== smjExec1.leftKeys

Removed, thanks!

SparkQA · 2020-10-07T17:40:51Z

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

SparkQA · 2020-10-07T17:59:21Z

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

SparkQA · 2020-10-07T21:11:31Z

Test build #129523 has finished for PR 29074 at commit 3cd6df9.

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

cloud-fan · 2020-10-08T04:58:32Z

thanks, merging to master!

imback82 added 2 commits July 11, 2020 19:04

initial commit

a0366f2

update comments

99493e4

probot-autolabeler bot added the SQL label Jul 12, 2020

imback82 changed the title ~~[SPARK-xxx][SQL] Improve EnsureRquirement.reorderJoinKeys to handle more scenarios such as PartitioningCollection~~ [SPARK-32282][SQL] Improve EnsureRquirement.reorderJoinKeys to handle more scenarios such as PartitioningCollection Jul 12, 2020

imback82 commented Jul 12, 2020

View reviewed changes

Merge branch 'master' into reorder_keys

ab237bc

Fix merge error

8308649

Merge branch 'master' into reorder_keys

268326b

maropu reviewed Aug 3, 2020

View reviewed changes

imback82 added 2 commits August 6, 2020 13:50

Merge branch 'master' into reorder_keys

d91bcdd

address PR comments

e5b078f

maropu reviewed Aug 7, 2020

View reviewed changes

address comments

89ad6ef

c21 reviewed Aug 23, 2020

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala Outdated Show resolved Hide resolved

imback82 added 2 commits August 24, 2020 10:04

Merge branch 'master' into reorder_keys

fa3aafa

Address PR comment

1729c8b

cloud-fan reviewed Oct 6, 2020

View reviewed changes

cloud-fan approved these changes Oct 6, 2020

View reviewed changes

imback82 added 2 commits October 6, 2020 17:21

Merge branch 'master' into reorder_keys

e2f7e44

Address PR comments

10b4d5a

cloud-fan reviewed Oct 7, 2020

View reviewed changes

Address PR comments

3cd6df9

cloud-fan closed this in 1c781a4 Oct 8, 2020

[SPARK-32282][SQL] Improve EnsureRquirement.reorderJoinKeys to handle more scenarios such as PartitioningCollection #29074

[SPARK-32282][SQL] Improve EnsureRquirement.reorderJoinKeys to handle more scenarios such as PartitioningCollection #29074

Uh oh!

Conversation

imback82 commented Jul 12, 2020

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?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 12, 2020

Uh oh!

imback82 commented Jul 12, 2020

Uh oh!

SparkQA commented Jul 12, 2020

Uh oh!

imback82 commented Jul 13, 2020

Uh oh!

maropu commented Jul 13, 2020

Uh oh!

SparkQA commented Jul 15, 2020

Uh oh!

SparkQA commented Jul 16, 2020

Uh oh!

imback82 commented Jul 27, 2020

Uh oh!

SparkQA commented Aug 2, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Aug 7, 2020

Uh oh!

maropu left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Aug 7, 2020

Uh oh!

maropu commented Aug 22, 2020

Uh oh!

SparkQA commented Aug 22, 2020

Uh oh!

c21 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

SparkQA commented Aug 24, 2020

Uh oh!

imback82 commented Aug 31, 2020

Uh oh!

maropu commented Sep 9, 2020

Uh oh!

SparkQA commented Sep 9, 2020

Uh oh!

maropu commented Sep 28, 2020

Uh oh!

SparkQA commented Sep 28, 2020

Uh oh!

SparkQA commented Sep 28, 2020

Uh oh!

SparkQA commented Sep 28, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Uh oh!

maropu left a comment •

edited

Loading