[SPARK-21998][SQL] SortMergeJoinExec did not calculate its outputOrdering correctly during physical planning #19281

maryannxue · 2017-09-19T18:53:16Z

What changes were proposed in this pull request?

Right now the calculation of SortMergeJoinExec's outputOrdering relies on the fact that its children have already been sorted on the join keys, while this is often not true until EnsureRequirements has been applied. So we ended up not getting the correct outputOrdering during physical planning stage before Sort nodes are added to the children.

For example, J = {A join B on key1 = key2}

if A is NOT ordered on key1 ASC, J's outputOrdering should include "key1 ASC"
if A is ordered on key1 ASC, J's outputOrdering should include "key1 ASC"
if A is ordered on key1 ASC, with sameOrderExp=c1, J's outputOrdering should include "key1 ASC, sameOrderExp=c1"

So to fix this I changed the behavior of getKeyOrdering(keys, childOutputOrdering) to:

If the childOutputOrdering satisfies (is a superset of) the required child ordering => childOutputOrdering
Otherwise => required child ordering

In addition, I organized the logic for deciding the relationship between two orderings into SparkPlan, so that it can be reused by EnsureRequirements and SortMergeJoinExec, and potentially other classes.

How was this patch tested?

Added new test cases.
Passed all integration tests.

…ring correctly during physical planning

tejasapatil · 2017-09-19T21:18:20Z

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

+   *
+   * Ordering A satisfies ordering B if and only if B is an equivalent of A or of A's prefix.
+   */
+  def orderingSatisfies(actual: Seq[SortOrder], required: Seq[SortOrder]): Boolean = {


how about moving this to SortOrder object :

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SortOrder.scala

Line 92 in e9c91ba

object SortOrder {

Actually I had struggled where to put this, in SortOrder or SparkPlan. It doesn't look like anywhere else is using Seq[SortOrder] so far, so I chose to leave this out of SortOrder. I think, though, if we see potential usage of Seq[SortOrder] elsewhere, it might be worth to wrap it as a class. Agree? Either way, I could put it into SortOrder for now.

SparkPlan is the node for physical operator in SQL so doesn't feel like a good place to have this. Since one would want to have all methods related to SortOrder in a single place, the object class feels better option. We can revisit if there are more such methods being added to that object and refac to a class.

Let us first move it to SortOrder.

tejasapatil · 2017-09-20T00:23:30Z

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

@@ -64,6 +67,42 @@ class JoinSuite extends QueryTest with SharedSQLContext {
    }
  }

+  def assertJoinOrdering(sqlString: String, numOfJoin: Int, numOfSort: Int): Any = {


nit: comparing the counts does not ensure that the sorts are in right place. I wish there was an easier way to pass that here but I can't think of any

BTW: since this is only used by one test case, we could put it inside the test case method and not make it class level

tejasapatil · 2017-09-20T00:28:54Z

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

-    keys.zip(childOutputOrdering).map { case (key, childOrder) =>
-      SortOrder(key, Ascending, childOrder.sameOrderExpressions + childOrder.child - key)
+    val requiredOrdering = requiredOrders(keys)
+    if (SparkPlan.orderingSatisfies(childOutputOrdering, requiredOrdering)) {


This looks good to me. cc @wzhfy who last touched this code

maropu · 2017-09-20T00:47:08Z

@gatorsmile @cloud-fan could you trigger tests if it is worth fixing? Thanks.

gatorsmile · 2017-09-20T04:31:57Z

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

+      }
+    } else {
+      true
+    }


Please simplify it to

if (required.isEmpty) { true } else if (required.length > actual.length) { false } else { required.zip(actual).forall { case (requiredOrder, actualOrder) => actualOrder.satisfies(requiredOrder) } }

gatorsmile · 2017-09-20T04:37:34Z

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

+   *
+   * Ordering A satisfies ordering B if and only if B is an equivalent of A or of A's prefix.
+   */
+  def orderingSatisfies(actual: Seq[SortOrder], required: Seq[SortOrder]): Boolean = {


Let us first move it to SortOrder.

gatorsmile · 2017-09-20T04:53:13Z

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

-    keys.zip(childOutputOrdering).map { case (key, childOrder) =>
-      SortOrder(key, Ascending, childOrder.sameOrderExpressions + childOrder.child - key)
+    val requiredOrdering = requiredOrders(keys)
+    if (SparkPlan.orderingSatisfies(childOutputOrdering, requiredOrdering)) {


Please add a comment here to explain the reason.

gatorsmile · 2017-09-20T04:54:49Z

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

@@ -64,6 +67,42 @@ class JoinSuite extends QueryTest with SharedSQLContext {
    }
  }

+  def assertJoinOrdering(sqlString: String, numOfJoin: Int, numOfSort: Int): Any = {


Please add function comments to explain what it does

gatorsmile · 2017-09-20T05:11:01Z

If the childOutputOrdering satisfies (is a superset of) the required child ordering => childOutputOrdering

This is not accurate. It depends on the length of required ordering and the length of child ordering.

maryannxue · 2017-09-20T05:18:38Z

bq. This is not accurate. It depends on the length of required ordering and the length of child ordering.

You are right. I did it right in the code but made a mistake in the description here. Thanks for pointing out, @gatorsmile!

gatorsmile · 2017-09-20T05:20:42Z

ok to test

SparkQA · 2017-09-20T07:04:44Z

Test build #81966 has finished for PR 19281 at commit 7893935.

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

cloud-fan · 2017-09-20T07:20:39Z

So we ended up not getting the correct outputOrdering during physical planning stage before Sort nodes are added to the children.

What's the harm of this? I think only EnsureRequirements cares about outputOrdering.

maryannxue · 2017-09-20T07:30:43Z

@cloud-fan Please refer to https://issues.apache.org/jira/browse/SPARK-18591. The plan is to apply the optimization during physical planning stage, and specifically, when creating Aggregate physical operator. So without the outputOrdering being accurate, such optimizations cannot be done at physical planning stage. And even in post-physical-planning stage, it has to be pushed after EnsureRequirements.

maryannxue · 2017-09-20T07:32:26Z

Thank you very much for the feedback, @tejasapatil, @gatorsmile! All the suggestions/comments have been addressed by my latest check-in.

SparkQA · 2017-09-20T10:01:25Z

Test build #81980 has finished for PR 19281 at commit a14c9be.

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

wzhfy · 2017-09-21T09:15:12Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SortOrder.scala

+   * Returns if a sequence of SortOrder satisfies another sequence of SortOrder.
+   *
+   * SortOrder sequence A satisfies SortOrder sequence B if and only if B is an equivalent of A
+   * or of A's prefix.


It would be better to add an example in comment: E.g. If ordering A is [x, y] and required ordering B is [x], then A satisfies B.

wzhfy · 2017-09-21T09:16:56Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SortOrder.scala

+   * SortOrder sequence A satisfies SortOrder sequence B if and only if B is an equivalent of A
+   * or of A's prefix.
+   */
+  def orderingSatisfies(seq1: Seq[SortOrder], seq2: Seq[SortOrder]): Boolean = {


nit: seq1 => ordering, seq2 => requiredOrdering

I had originally named them "actualOrdering" and "requiredOrdering" in SparkPlan, but since I've moved the function here to SortOrder, thought the names should be more general.

wzhfy · 2017-09-21T09:25:39Z

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

+   * The utility method to get output ordering for left or right side of the join.
+   *
+   * Returns the required ordering for left or right child if childOutputOrdering does not
+   * satisfy the required ordering; otherwise, in which case the child will not be re-sorted,


nit: in case the child does not need to be sorted

wzhfy · 2017-09-21T11:27:12Z

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

+          // outputOrdering should be consistent between physical plan and executed plan
+          assert(outputOrderingPhysical == outputOrderingExecuted,
+            s"Physical operator $join1 did not have the same output ordering as " +
+            s"corresponding executed operator $join2")


nit: physical/executed operator => physical/executed plan?

maryannxue · 2017-09-21T23:13:13Z

Thank you, @wzhfy, very much for the comments! I have made corresponding changes in my latest commit.

SparkQA · 2017-09-22T01:47:13Z

Test build #82054 has finished for PR 19281 at commit 78f41e0.

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

gatorsmile · 2017-09-22T06:53:45Z

LGTM

gatorsmile · 2017-09-22T06:54:14Z

Thanks! Merged to master.

[SPARK-21998][SQL] SortMergeJoinExec did not calculate its outputOrde…

7893935

…ring correctly during physical planning

tejasapatil reviewed Sep 19, 2017

View reviewed changes

tejasapatil reviewed Sep 20, 2017

View reviewed changes

gatorsmile reviewed Sep 20, 2017

View reviewed changes

Make changes according to review comments

a14c9be

wzhfy reviewed Sep 21, 2017

View reviewed changes

Make changes according to review comments

78f41e0

asfgit closed this in 5960686 Sep 22, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-21998][SQL] SortMergeJoinExec did not calculate its outputOrdering correctly during physical planning #19281

[SPARK-21998][SQL] SortMergeJoinExec did not calculate its outputOrdering correctly during physical planning #19281

maryannxue commented Sep 19, 2017

tejasapatil Sep 19, 2017

maryannxue Sep 19, 2017

tejasapatil Sep 20, 2017

gatorsmile Sep 20, 2017

tejasapatil Sep 20, 2017

tejasapatil Sep 20, 2017

tejasapatil Sep 20, 2017

maropu commented Sep 20, 2017

gatorsmile Sep 20, 2017

gatorsmile Sep 20, 2017

gatorsmile Sep 20, 2017

gatorsmile Sep 20, 2017

gatorsmile commented Sep 20, 2017

maryannxue commented Sep 20, 2017

gatorsmile commented Sep 20, 2017

SparkQA commented Sep 20, 2017

cloud-fan commented Sep 20, 2017

maryannxue commented Sep 20, 2017

maryannxue commented Sep 20, 2017

SparkQA commented Sep 20, 2017

wzhfy Sep 21, 2017 •

edited

wzhfy Sep 21, 2017

maryannxue Sep 21, 2017

wzhfy Sep 21, 2017

wzhfy Sep 21, 2017

maryannxue commented Sep 21, 2017

SparkQA commented Sep 22, 2017

gatorsmile commented Sep 22, 2017

gatorsmile commented Sep 22, 2017

[SPARK-21998][SQL] SortMergeJoinExec did not calculate its outputOrdering correctly during physical planning #19281

[SPARK-21998][SQL] SortMergeJoinExec did not calculate its outputOrdering correctly during physical planning #19281

Conversation

maryannxue commented Sep 19, 2017

What changes were proposed in this pull request?

How was this patch tested?

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

maropu commented Sep 20, 2017

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

gatorsmile commented Sep 20, 2017

maryannxue commented Sep 20, 2017

gatorsmile commented Sep 20, 2017

SparkQA commented Sep 20, 2017

cloud-fan commented Sep 20, 2017

maryannxue commented Sep 20, 2017

maryannxue commented Sep 20, 2017

SparkQA commented Sep 20, 2017

wzhfy Sep 21, 2017 • edited

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

maryannxue commented Sep 21, 2017

SparkQA commented Sep 22, 2017

gatorsmile commented Sep 22, 2017

gatorsmile commented Sep 22, 2017

wzhfy Sep 21, 2017 •

edited