[SPARK-32730][SQL] Improve LeftSemi and Existence SortMergeJoin right side buffering #29572

peter-toth · 2020-08-28T20:15:46Z

What changes were proposed in this pull request?

LeftSemi and Existence SortMergeJoin should not buffer all matching right side rows when bound condition is empty, this is unnecessary and can lead to performance degradation especially when spilling happens.

Why are the changes needed?

Performance improvement.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New UT and TPCDS benchmarks.

SparkQA · 2020-08-29T00:31:08Z

Test build #128000 has finished for PR 29572 at commit 1a49356.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait AppendOnlyUnsafeRowArray

peter-toth · 2020-08-29T08:33:01Z

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala


+      // LEFT SEMI JOIN without bound condition does not use [[ExternalAppendOnlyUnsafeRowArray]]
+      // so should not cause any spill
+      assertNotSpilled(sparkContext, "left semi join") {


Without this fix this UT fails.

SparkQA · 2020-08-29T12:50:46Z

Test build #128014 has finished for PR 29572 at commit acc6646.

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

peter-toth · 2020-09-01T14:37:13Z

cc @cloud-fan, @maropu, @viirya

cloud-fan · 2020-09-02T08:54:39Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

+      }
+
+      override def add(row: UnsafeRow): Unit = {
+        assert(buffer == null)


Are you saying that ExternalAppendOnlyUnsafeRowArray will do spill even if we only add one row?

No, its threshold parameters do work as expected. Just ExternalAppendOnlyUnsafeRowArray looked a bit heavy weight for this case when we want to store only one row. But we can also use new ExternalAppendOnlyUnsafeRowArray(1, 1) for this case.

How did you test it if ExternalAppendOnlyUnsafeRowArray doesn't spill either?

viirya · 2020-09-02T16:21:03Z

sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala

    pageSizeBytes: Long,
    numRowsInMemoryBufferThreshold: Int,
-    numRowsSpillThreshold: Int) extends Logging {
+    numRowsSpillThreshold: Int) extends AppendOnlyUnsafeRowArray with Logging {


For core component like this, I remember we rarely change its inheritance. It is easily to have performance regression.

All right, in that case let's drop that new trait and stick to the important part.

SparkQA · 2020-09-03T01:36:45Z

Test build #128212 has finished for PR 29572 at commit 037b876.

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

cloud-fan · 2020-09-03T06:10:58Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

-  private[this] val bufferedMatches =
-    new ExternalAppendOnlyUnsafeRowArray(inMemoryThreshold, spillThreshold)
+  private[this] val bufferedMatches: ExternalAppendOnlyUnsafeRowArray =
+    new ExternalAppendOnlyUnsafeRowArray(if (bufferFirstOnly) 1 else inMemoryThreshold,


how does this change avoid spilling?

Not this change does it, but the one in bufferMatchingRows(). This change just avoids creating a buffer larger than 1.

maropu · 2020-09-03T06:23:47Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

    spillThreshold: Int,
-    eagerCleanupResources: () => Unit) {
+    eagerCleanupResources: () => Unit,
+    bufferFirstOnly: Boolean) {


nit: bufferFirstOnly -> matchedBufferFirstOnly? And, please add @param, too.

bufferFirstOnly: Boolean = false to avoid the unnecessary changes.

Thanks, fixed.

maropu · 2020-09-03T06:26:10Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

-  private[this] val bufferedMatches =
-    new ExternalAppendOnlyUnsafeRowArray(inMemoryThreshold, spillThreshold)
+  private[this] val bufferedMatches: ExternalAppendOnlyUnsafeRowArray =
+    new ExternalAppendOnlyUnsafeRowArray(if (bufferFirstOnly) 1 else inMemoryThreshold,


We need this change? How about just setting 1 in the caller side? https://github.com/apache/spark/pull/29572/files#diff-2649ee0724bbc5ee6a64d7de308c527eR255

There is another call site in existence join: https://github.com/apache/spark/pull/29572/files#diff-2649ee0724bbc5ee6a64d7de308c527eR336

cloud-fan · 2020-09-03T07:30:03Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

    spillThreshold: Int,
-    eagerCleanupResources: () => Unit) {
+    eagerCleanupResources: () => Unit,
+    matchedBufferFirstOnly: Boolean = false) {


how about onlyBufferFirstMatch?

Ok, renamed.

maropu · 2020-09-03T07:48:03Z

Ur one more; please update the title/description (this PR is not only for leftsemi, right?), too.

maropu · 2020-09-03T07:48:27Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

              inMemoryThreshold,
              spillThreshold,
-              cleanupResources
-            )


nit: please avoid the unnecessary changes.

Sorry, reverted them.

peter-toth · 2020-09-03T07:56:26Z

Ur one more; please update the title/description (this PR is not only for leftsemi, right?), too.

Updated, thanks.

SparkQA · 2020-09-03T12:08:00Z

Test build #128238 has finished for PR 29572 at commit 3937e4c.

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

SparkQA · 2020-09-03T12:55:51Z

Test build #128243 has finished for PR 29572 at commit f699118.

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

SparkQA · 2020-09-03T13:05:37Z

Test build #128242 has finished for PR 29572 at commit 5cf3ab3.

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

cloud-fan · 2020-09-03T14:17:31Z

thanks, merging to master!

peter-toth · 2020-09-03T14:52:22Z

Thanks for the review @cloud-fan, @maropu, @viirya.

gatorsmile · 2020-09-09T02:17:32Z

@peter-toth Nice fix! Could you share the perf difference when you run the TPC-DS ?

juliuszsompolski · 2020-09-10T16:49:55Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

              spillThreshold,
-              cleanupResources
+              cleanupResources,
+              condition.isEmpty


Thanks @peter-toth !
I think this could be also added to LeftAnti join, which is also only interested in the existence of a match and doesn't need to buffer them.

Thanks @juliuszsompolski, I think you are right. Shall I open a follow-up PR or a different ticket?

I think a followup PR is fine.

I've opened #29727

peter-toth · 2020-09-10T19:46:06Z

@peter-toth Nice fix! Could you share the perf difference when you run the TPC-DS ?

Thanks @gatorsmile. Yes, I will try to run some benchmarks for this particular change and share the results.

BTW, I have another PR open that brings ~30% improvement to some of the TPCDS queries: #28885

…de buffering ### What changes were proposed in this pull request? This is a follow-up to #29572. LeftAnti SortMergeJoin should not buffer all matching right side rows when bound condition is empty, this is unnecessary and can lead to performance degradation especially when spilling happens. ### Why are the changes needed? Performance improvement. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New UT. Closes #29727 from peter-toth/SPARK-32730-improve-leftsemi-sortmergejoin-followup. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…de buffering ### What changes were proposed in this pull request? This is a follow-up to apache/spark#29572. LeftAnti SortMergeJoin should not buffer all matching right side rows when bound condition is empty, this is unnecessary and can lead to performance degradation especially when spilling happens. ### Why are the changes needed? Performance improvement. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New UT. Closes #29727 from peter-toth/SPARK-32730-improve-leftsemi-sortmergejoin-followup. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

[SPARK-32730][SQL] Improve LeftSemi SortMergeJoin right side buffering

1a49356

probot-autolabeler bot added the SQL label Aug 28, 2020

add UT

acc6646

peter-toth changed the title ~~[WIP][SPARK-32730][SQL] Improve LeftSemi SortMergeJoin right side buffering~~ [SPARK-32730][SQL] Improve LeftSemi SortMergeJoin right side buffering Aug 29, 2020

peter-toth commented Aug 29, 2020

View reviewed changes

cloud-fan reviewed Sep 2, 2020

View reviewed changes

viirya reviewed Sep 2, 2020

View reviewed changes

peter-toth force-pushed the SPARK-32730-improve-leftsemi-sortmergejoin branch from 689b5b7 to d893580 Compare September 2, 2020 19:15

drop AppendOnlyUnsafeRowArray trait

037b876

peter-toth force-pushed the SPARK-32730-improve-leftsemi-sortmergejoin branch from d893580 to 037b876 Compare September 2, 2020 19:17

cloud-fan reviewed Sep 3, 2020

View reviewed changes

maropu reviewed Sep 3, 2020

View reviewed changes

address review comments

3937e4c

cloud-fan reviewed Sep 3, 2020

View reviewed changes

cloud-fan approved these changes Sep 3, 2020

View reviewed changes

rename param

5cf3ab3

maropu reviewed Sep 3, 2020

View reviewed changes

revert accidental changes

f699118

peter-toth changed the title ~~[SPARK-32730][SQL] Improve LeftSemi SortMergeJoin right side buffering~~ [SPARK-32730][SQL] Improve LeftSemi and Existence SortMergeJoin right side buffering Sep 3, 2020

cloud-fan closed this in ffd5227 Sep 3, 2020

juliuszsompolski reviewed Sep 10, 2020

View reviewed changes

peter-toth mentioned this pull request Sep 11, 2020

[SPARK-32730][SQL][FOLLOW-UP] Improve LeftAnti SortMergeJoin right side buffering #29727

Closed

[SPARK-32730][SQL] Improve LeftSemi and Existence SortMergeJoin right side buffering #29572

[SPARK-32730][SQL] Improve LeftSemi and Existence SortMergeJoin right side buffering #29572

Conversation

peter-toth commented Aug 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

SparkQA commented Aug 29, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 29, 2020

Uh oh!

peter-toth commented Sep 1, 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!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 3, 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!

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!

Choose a reason for hiding this comment

Uh oh!

maropu commented Sep 3, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

peter-toth commented Sep 3, 2020

Uh oh!

SparkQA commented Sep 3, 2020

Uh oh!

SparkQA commented Sep 3, 2020

Uh oh!

SparkQA commented Sep 3, 2020

Uh oh!

cloud-fan commented Sep 3, 2020

Uh oh!

peter-toth commented Sep 3, 2020

Uh oh!

gatorsmile commented Sep 9, 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!

peter-toth commented Sep 10, 2020

Uh oh!

Reviewers

Assignees

peter-toth commented Aug 28, 2020 •

edited

Loading