[SPARK-7165] [SQL] use sort merge join for outer join #5717

adrian-wang · 2015-04-27T09:52:39Z

This is an extended version of #5208
In this patch, we are introducing sort merge join for not only inner joins, but left outer/ right outer/ full outer joins.
Using sort merge join could resolve the OOM which is quite common as the memory easily becomes too small for joins of large tables.

Test cases are always available in SortMergeCompatibilitySuite.
Also , This patch would benefit from #3438 quite a lot.

/cc @chenghao-intel

SparkQA · 2015-04-27T11:59:24Z

Test build #30964 has finished for PR 5717 at commit fc862f4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

SparkQA · 2015-04-28T04:33:55Z

Test build #31103 has finished for PR 5717 at commit ae68ee7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

SparkQA · 2015-05-20T04:30:20Z

Test build #33121 has finished for PR 5717 at commit 44fd7cf.

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

rxin · 2015-05-27T07:58:30Z

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

+    case FullOuter =>
+      left.output.map(_.withNullability(true)) ++ right.output.map(_.withNullability(true))
+    case x =>
+      throw new Exception(s"SortMergeJoin should not take $x as the JoinType")


SparkQA · 2015-05-27T10:49:23Z

Test build #33580 has finished for PR 5717 at commit 6aaa593.

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

adrian-wang · 2015-05-27T10:56:19Z

retest this please.

SparkQA · 2015-05-27T12:51:21Z

Test build #33582 has finished for PR 5717 at commit 6aaa593.

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

chenghao-intel · 2015-05-28T00:50:25Z

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


-  override def outputPartitioning: Partitioning = left.outputPartitioning
+  override def outputPartitioning: Partitioning = joinType match {


NOTICE: Should always be streamed.outputPartitioning once #6413 merged, see https://github.com/apache/spark/pull/6413/files#diff-48230fdc68c8c172d22709ed90f8817dR50

SparkQA · 2015-05-29T04:24:23Z

Test build #33707 has finished for PR 5717 at commit add49a2.

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

JoshRosen · 2015-06-12T15:40:53Z

Jenkins, retest this please.

JoshRosen · 2015-06-12T15:42:11Z

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


  private def requiredOrders(keys: Seq[Expression]): Seq[SortOrder] =
    keys.map(SortOrder(_, Ascending))

  protected override def doExecute(): RDD[Row] = {
-    val leftResults = left.execute().map(_.copy())
-    val rightResults = right.execute().map(_.copy())
+    val streamResults = streamed.execute().map(_.copy())


Why do we need to copy the streamed rows? I understand why we need to do the copy for the buffered results, since we might be dealing with mutable input rows, but that shouldn't be a problem for the stream side, right?

I think we need to copy this, it has something to do with the external sort.

We certainly need to copy the inputs that are passed to external sort, but the ExternalSort operator itself should take care of that. Here, I think we're consuming the result of a sort operator and are not buffering rows from streamResults (unless I've overlooked other buffering inside of zipPartitions somehow).

That's true for Left/Right Outer and even inner join, however, in full outer join, we probably need to cache the streamed row once, but you're right, we can do the copy whenever necessary during the iterating, not here.

SparkQA · 2015-06-12T17:36:46Z

Test build #34777 has finished for PR 5717 at commit add49a2.

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

adrian-wang · 2015-06-12T17:41:09Z

@JoshRosen Thanks for your comments, I will refine the code accordingly.

JoshRosen · 2015-06-12T18:56:53Z

By the way, to provide a bit of context for why I'm reviewing this PR: I'm working on some optimizations to sorting in Spark SQL which should benefit sort-merge-join, so I've looked over all of this code pretty recently.

JoshRosen · 2015-06-19T02:44:42Z

@adrian-wang, I'm planning to take another pass on this pretty soon. At a high level, this patch is in very good shape since most of its code is modeled after other existing join implementations in Spark SQL. If you update this in the next couple of days, I'll try my best to be responsive with my reviews so we can get this in soon and not have too many merge conflicts.

SparkQA · 2015-06-19T07:28:53Z

Test build #35233 has finished for PR 5717 at commit 211e101.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class SerializableConfiguration(@transient var value: Configuration) extends Serializable
- class SerializableJobConf(@transient var value: JobConf) extends Serializable
- class ElementwiseProduct(VectorTransformer):
- case class CreateStruct(children: Seq[Expression]) extends Expression
- case class Sqrt(child: Expression) extends UnaryMathExpression(math.sqrt, "SQRT")
- case class Logarithm(left: Expression, right: Expression)
- case class SetCommand(kv: Option[(String, Option[String])]) extends RunnableCommand with Logging

adrian-wang · 2015-06-19T23:17:29Z

retest this please.

SparkQA · 2015-06-20T01:12:30Z

Test build #35337 has finished for PR 5717 at commit 211e101.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class SerializableConfiguration(@transient var value: Configuration) extends Serializable
- class SerializableJobConf(@transient var value: JobConf) extends Serializable

JoshRosen · 2015-06-20T06:54:19Z

Thanks for updating this; I'll try to take another review pass tomorrow.

jeanlyn · 2015-06-21T07:33:11Z

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

+              if (bufferedPosition >= bufferedMatches.size) {
+                bufferedPosition = 0
+                if (joinType != FullOuter || secondStreamedElement == null) {
+                  fetchStreamed()


I think we should use boundCondition to update bufferedMatches after we fetchStreamed () .Otherwise we may get wrong answer.For example

table a(key int,value int);table b(key int,value int) data of a 1 3 1 1 2 1 2 3 data of b 1 1 2 1 select a.key,b.key,a.value-b.value from a left outer join b on a.key=b.key and a.value - b.value > 1

Good catch, I'll rewrite this part.

SparkQA · 2015-07-30T08:11:55Z

Test build #39030 has finished for PR 5717 at commit fd73084.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-07-31T06:34:13Z

Test build #39159 has finished for PR 5717 at commit f520079.

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

SparkQA · 2015-08-04T07:40:10Z

Test build #39678 has finished for PR 5717 at commit d0e65c5.

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

SparkQA · 2015-08-04T10:28:48Z

Test build #39689 has finished for PR 5717 at commit bff834a.

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

adrian-wang · 2015-08-04T10:37:35Z

@JoshRosen I've fixed the bug that @jeanlyn mentioned, can you merge this first and then do the following steps in #7904 ?

JoshRosen · 2015-08-04T18:51:52Z

We should have a test to guard against reintroduction of the the bug that @jeanlyn mentioned.

I find the code here to be really dense and hard to understand, so I'd like to try to pursue my design first. There's another 1.5 blocker / critical related to eliminating JoinedRow in favor of Tungsten's RowJoiner when UnsafeRows are used, and I think that the code re-use enabled by my design will make this significantly easier to accomplish.

JoshRosen · 2015-08-04T19:08:51Z

Also, I think that it might be a little clearer to introduce a separate SortMergeOuterJoin operator rather than trying to combine the inner and outer joins into the same operator. This would be consistent with what we've done for other joins.

JoshRosen · 2015-08-04T20:42:55Z

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

      ("SELECT * FROM testData RIGHT JOIN testData2 ON key = a where key = 2",
-        classOf[BroadcastHashOuterJoin]),


It looks like this patch causes us to plan SortMergeJoin for outer joins that are capable of using BroadcastHashOuterJoin, which seems like it could lead to performance issues by triggering unnecessary shuffling of the large table.

As a result, I think that we should not change the broadcast-enabled half of the test, but, rather, should update the broadcast-disabled half to test both the sort-merge-join enabled and sort-merge-join-disabled configurations.

adrian-wang · 2015-08-05T07:57:19Z

retest this please.

SparkQA · 2015-08-05T08:07:30Z

Test build #227 has finished for PR 5717 at commit 549796e.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-08-05T08:10:41Z

Test build #39844 has finished for PR 5717 at commit 549796e.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-08-06T04:54:44Z

Test build #39983 has finished for PR 5717 at commit d02f6bb.

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

viirya · 2015-08-07T06:07:57Z

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

        }

        /**
-         * Searches the right iterator for the next rows that have matches in left side, and store
-         * them in a buffer.
+         * Searches the right iterator for the next rows that have matches in left side (only check


Is it confusing because right and left can be both streamed or buffered here? Do we need to use streamed and buffered in the comments as well?

This may be clarified slightly in my own SMJ patch, #7904.

…t outer join This patch adds a new `SortMergeOuterJoin` operator that performs left and right outer joins using sort merge join. It also refactors `SortMergeJoin` in order to improve performance and code clarity. Along the way, I also performed a couple pieces of minor cleanup and optimization: - Rename the `HashJoin` physical planner rule to `EquiJoinSelection`, since it's also used for non-hash joins. - Rewrite the comment at the top of `HashJoin` to better explain the precedence for choosing join operators. - Update `JoinSuite` to use `SqlTestUtils.withConf` for changing SQLConf settings. This patch incorporates several ideas from adrian-wang's patch, #5717. Closes #5717.  [<img src="https://reviewable.io/review_button.png" height=40 alt="Review on Reviewable"/>](https://reviewable.io/reviews/apache/spark/7904)  Author: Josh Rosen <[email protected]> Author: Daoyuan Wang <[email protected]> Closes #7904 from JoshRosen/outer-join-smj and squashes 1 commits. (cherry picked from commit 91e9389) Signed-off-by: Reynold Xin <[email protected]>

…t outer join This patch adds a new `SortMergeOuterJoin` operator that performs left and right outer joins using sort merge join. It also refactors `SortMergeJoin` in order to improve performance and code clarity. Along the way, I also performed a couple pieces of minor cleanup and optimization: - Rename the `HashJoin` physical planner rule to `EquiJoinSelection`, since it's also used for non-hash joins. - Rewrite the comment at the top of `HashJoin` to better explain the precedence for choosing join operators. - Update `JoinSuite` to use `SqlTestUtils.withConf` for changing SQLConf settings. This patch incorporates several ideas from adrian-wang's patch, apache#5717. Closes apache#5717.  [<img src="https://reviewable.io/review_button.png" height=40 alt="Review on Reviewable"/>](https://reviewable.io/reviews/apache/spark/7904)  Author: Josh Rosen <[email protected]> Author: Daoyuan Wang <[email protected]> Closes apache#7904 from JoshRosen/outer-join-smj and squashes 1 commits.

adrian-wang force-pushed the outersmj branch from ae68ee7 to 44fd7cf Compare May 20, 2015 02:32

rxin reviewed May 27, 2015
View reviewed changes

chenghao-intel reviewed May 28, 2015
View reviewed changes

JoshRosen reviewed Jun 12, 2015
View reviewed changes

adrian-wang force-pushed the outersmj branch from add49a2 to 211e101 Compare June 19, 2015 06:22

jeanlyn reviewed Jun 21, 2015
View reviewed changes

JoshRosen mentioned this pull request Aug 3, 2015

[SPARK-9729] [SPARK-9363] [SQL] Use sort merge join for left and right outer join #7904

Closed

adrian-wang force-pushed the outersmj branch from f520079 to d0e65c5 Compare August 4, 2015 07:07

JoshRosen reviewed Aug 4, 2015
View reviewed changes

use sort merge join for outer join

d95417e

adrian-wang force-pushed the outersmj branch from 549796e to d02f6bb Compare August 6, 2015 03:13

adrian-wang added 9 commits August 5, 2015 20:14

rebase

71ff4e9

bring it up to date

a8d1ff7

fix default setting change

fdea91d

fix style

a4cf5cd

fix comments from @jeanlyn

53c2bdb

Use withSQLConf in JoinSuite

6a771e9

minor fixes

13f86bd

bug fix

d2a1d12

fix broadcast selection

d02f6bb

viirya reviewed Aug 7, 2015
View reviewed changes

asfgit closed this in 91e9389 Aug 11, 2015


		override def outputPartitioning: Partitioning = left.outputPartitioning
		override def outputPartitioning: Partitioning = joinType match {

		("SELECT * FROM testData RIGHT JOIN testData2 ON key = a where key = 2",
		classOf[BroadcastHashOuterJoin]),

[SPARK-7165] [SQL] use sort merge join for outer join #5717

[SPARK-7165] [SQL] use sort merge join for outer join #5717

Conversation

adrian-wang commented Apr 27, 2015

SparkQA commented Apr 27, 2015

SparkQA commented Apr 28, 2015

SparkQA commented May 20, 2015

Choose a reason for hiding this comment

SparkQA commented May 27, 2015

adrian-wang commented May 27, 2015

SparkQA commented May 27, 2015

Choose a reason for hiding this comment

SparkQA commented May 29, 2015

JoshRosen commented Jun 12, 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

SparkQA commented Jun 12, 2015

adrian-wang commented Jun 12, 2015

JoshRosen commented Jun 12, 2015

JoshRosen commented Jun 19, 2015

SparkQA commented Jun 19, 2015

adrian-wang commented Jun 19, 2015

SparkQA commented Jun 20, 2015

JoshRosen commented Jun 20, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jul 30, 2015

SparkQA commented Jul 31, 2015

SparkQA commented Aug 4, 2015

SparkQA commented Aug 4, 2015

adrian-wang commented Aug 4, 2015

JoshRosen commented Aug 4, 2015

JoshRosen commented Aug 4, 2015

Choose a reason for hiding this comment

adrian-wang commented Aug 5, 2015

SparkQA commented Aug 5, 2015

SparkQA commented Aug 5, 2015

SparkQA commented Aug 6, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment