[SPARK-9703] [SQL] Refactor EnsureRequirements to avoid certain unnecessary shuffles #7988

JoshRosen · 2015-08-06T07:20:52Z

This pull request refactors the EnsureRequirements planning rule in order to avoid the addition of certain unnecessary shuffles.

As an example of how unnecessary shuffles can occur, consider SortMergeJoin, which requires clustered distribution and sorted ordering of its children's input rows. Say that both of SMJ's children produce unsorted output but are both SinglePartition. In this case, we will need to inject sort operators but should not need to inject Exchanges. Unfortunately, it looks like the EnsureRequirements unnecessarily repartitions using a hash partitioning.

This patch solves this problem by refactoring EnsureRequirements to properly implement the compatibleWith checks that were broken in earlier implementations. See the significant inline comments for a better description of how this works. The majority of this PR is new comments and test cases, with few actual changes to the code.

…es not help.

SparkQA · 2015-08-06T07:42:02Z

Test build #40007 has finished for PR 7988 at commit 0675956.

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

SparkQA · 2015-08-06T09:12:23Z

Test build #40024 has finished for PR 7988 at commit c628daf.

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

SparkQA · 2015-08-06T11:37:56Z

Test build #40026 has finished for PR 7988 at commit 752b8de.

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

SparkQA · 2015-08-06T18:53:42Z

Test build #40045 has finished for PR 7988 at commit 0725a34.

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

SparkQA · 2015-08-06T18:57:35Z

Test build #40044 has finished for PR 7988 at commit 2e0f33a.

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

JoshRosen · 2015-08-06T21:12:02Z

There's one more test that I need to add.

SparkQA · 2015-08-06T22:00:58Z

Test build #40071 has finished for PR 7988 at commit 4f08278.

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

JoshRosen · 2015-08-06T22:57:23Z

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

-        case (OrderedDistribution(ordering), rowOrdering, child) =>
-          addOperatorsIfNecessary(RangePartitioning(ordering, numPartitions), rowOrdering, child)
+    def addShuffleIfNecessary(child: SparkPlan, requiredDistribution: Distribution): SparkPlan = {
+      // A pre-condition of ensureDistributionAndOrdering is that joins' children have compatible


This piece of reasoning is the trickiest part of this entire patch. Is this a valid argument given the current semantics of guarantees() and satisfies()?

This will be cleaned up tomorrow when I consolidate the shuffle steps and make the assertions / invariants clearer.

SparkQA · 2015-08-06T23:33:52Z

Test build #40077 has finished for PR 7988 at commit 8dbc845.

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

yhuai · 2015-08-06T23:46:45Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

+      case Seq(a) => true
+      case Seq(a, b) =>
+        if (a.numPartitions != b.numPartitions) {
+          assert(!a.guarantees(b) && !b.guarantees(a))


Since we are using guarantees to check if sibling operators are compatible or not. How about we just get back compatible with?

This sounds good to me. Let's revisit this naming later once we actually break this symmetry.

SparkQA · 2015-08-07T00:03:51Z

Test build #40087 has finished for PR 7988 at commit 642b0bb.

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

SparkQA · 2015-08-07T00:24:24Z

Test build #40080 has finished for PR 7988 at commit 06aba0c.

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

SparkQA · 2015-08-07T00:56:00Z

Test build #40098 has finished for PR 7988 at commit fee65c4.

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

JoshRosen · 2015-08-07T02:12:54Z

Yin and I discussed offline and I think that I see a way to greatly simplify this patch while making it easier to understand. Will update after I unblock my SMJ patch.

SparkQA · 2015-08-08T05:32:04Z

Test build #40224 has finished for PR 7988 at commit 8784bd9.

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

JoshRosen · 2015-08-08T06:40:22Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

@@ -75,6 +75,37 @@ case class OrderedDistribution(ordering: Seq[SortOrder]) extends Distribution {
  def clustering: Set[Expression] = ordering.map(_.child).toSet
 }

+/**
+ * Describes how an operator's output is split across partitions. The `satisfies`,


Editing note: should exploit parallelism between these two lists, since that's better writing. Will copyedit tomorrow.

…ing.

JoshRosen · 2015-08-09T07:11:41Z

Alright, I've simplified these EnsureRequirements changes and I think this patch is now ready for review. The final change was quite small, essentially adding back the compatibleWith checks and fixing them. The rest of the changes are tests, documentation, and some small refactoring to improve clarity.

SparkQA · 2015-08-09T09:06:27Z

Test build #40262 has finished for PR 7988 at commit 38006e7.

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

JoshRosen · 2015-08-09T20:46:41Z

/cc @yhuai for final proofreading and sign-off.

yhuai · 2015-08-09T21:25:41Z

Looks very good! I am merging it to master and branch 1.5.

…essary shuffles This pull request refactors the `EnsureRequirements` planning rule in order to avoid the addition of certain unnecessary shuffles. As an example of how unnecessary shuffles can occur, consider SortMergeJoin, which requires clustered distribution and sorted ordering of its children's input rows. Say that both of SMJ's children produce unsorted output but are both SinglePartition. In this case, we will need to inject sort operators but should not need to inject Exchanges. Unfortunately, it looks like the EnsureRequirements unnecessarily repartitions using a hash partitioning. This patch solves this problem by refactoring `EnsureRequirements` to properly implement the `compatibleWith` checks that were broken in earlier implementations. See the significant inline comments for a better description of how this works. The majority of this PR is new comments and test cases, with few actual changes to the code. Author: Josh Rosen <joshrosen@databricks.com> Closes #7988 from JoshRosen/exchange-fixes and squashes the following commits: 38006e7 [Josh Rosen] Rewrite EnsureRequirements _yet again_ to make things even simpler 0983f75 [Josh Rosen] More guarantees vs. compatibleWith cleanup; delete BroadcastPartitioning. 8784bd9 [Josh Rosen] Giant comment explaining compatibleWith vs. guarantees 1307c50 [Josh Rosen] Update conditions for requiring child compatibility. 18cddeb [Josh Rosen] Rename DummyPlan to DummySparkPlan. 2c7e126 [Josh Rosen] Merge remote-tracking branch 'origin/master' into exchange-fixes fee65c4 [Josh Rosen] Further refinement to comments / reasoning 642b0bb [Josh Rosen] Further expand comment / reasoning 06aba0c [Josh Rosen] Add more comments 8dbc845 [Josh Rosen] Add even more tests. 4f08278 [Josh Rosen] Fix the test by adding the compatibility check to EnsureRequirements a1c12b9 [Josh Rosen] Add failing test to demonstrate allCompatible bug 0725a34 [Josh Rosen] Small assertion cleanup. 5172ac5 [Josh Rosen] Add test for requiresChildrenToProduceSameNumberOfPartitions. 2e0f33a [Josh Rosen] Write a more generic test for EnsureRequirements. 752b8de [Josh Rosen] style fix c628daf [Josh Rosen] Revert accidental ExchangeSuite change. c9fb231 [Josh Rosen] Rewrite exchange to fix better handle this case. adcc742 [Josh Rosen] Move test to PlannerSuite. 0675956 [Josh Rosen] Preserving ordering and partitioning in row format converters also does not help. cc5669c [Josh Rosen] Adding outputPartitioning to Repartition does not fix the test. 2dfc648 [Josh Rosen] Add failing test illustrating bad exchange planning. (cherry picked from commit 23cf5af) Signed-off-by: Yin Huai <yhuai@databricks.com>

cloud-fan · 2015-08-10T06:36:58Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

+object Partitioning {
+  def allCompatible(partitionings: Seq[Partitioning]): Boolean = {
+    // Note: this assumes transitivity
+    partitionings.sliding(2).map {


Nit: we can use forall instead of map here and remove the forall at the end.

…essary shuffles This pull request refactors the `EnsureRequirements` planning rule in order to avoid the addition of certain unnecessary shuffles. As an example of how unnecessary shuffles can occur, consider SortMergeJoin, which requires clustered distribution and sorted ordering of its children's input rows. Say that both of SMJ's children produce unsorted output but are both SinglePartition. In this case, we will need to inject sort operators but should not need to inject Exchanges. Unfortunately, it looks like the EnsureRequirements unnecessarily repartitions using a hash partitioning. This patch solves this problem by refactoring `EnsureRequirements` to properly implement the `compatibleWith` checks that were broken in earlier implementations. See the significant inline comments for a better description of how this works. The majority of this PR is new comments and test cases, with few actual changes to the code. Author: Josh Rosen <joshrosen@databricks.com> Closes apache#7988 from JoshRosen/exchange-fixes and squashes the following commits: 38006e7 [Josh Rosen] Rewrite EnsureRequirements _yet again_ to make things even simpler 0983f75 [Josh Rosen] More guarantees vs. compatibleWith cleanup; delete BroadcastPartitioning. 8784bd9 [Josh Rosen] Giant comment explaining compatibleWith vs. guarantees 1307c50 [Josh Rosen] Update conditions for requiring child compatibility. 18cddeb [Josh Rosen] Rename DummyPlan to DummySparkPlan. 2c7e126 [Josh Rosen] Merge remote-tracking branch 'origin/master' into exchange-fixes fee65c4 [Josh Rosen] Further refinement to comments / reasoning 642b0bb [Josh Rosen] Further expand comment / reasoning 06aba0c [Josh Rosen] Add more comments 8dbc845 [Josh Rosen] Add even more tests. 4f08278 [Josh Rosen] Fix the test by adding the compatibility check to EnsureRequirements a1c12b9 [Josh Rosen] Add failing test to demonstrate allCompatible bug 0725a34 [Josh Rosen] Small assertion cleanup. 5172ac5 [Josh Rosen] Add test for requiresChildrenToProduceSameNumberOfPartitions. 2e0f33a [Josh Rosen] Write a more generic test for EnsureRequirements. 752b8de [Josh Rosen] style fix c628daf [Josh Rosen] Revert accidental ExchangeSuite change. c9fb231 [Josh Rosen] Rewrite exchange to fix better handle this case. adcc742 [Josh Rosen] Move test to PlannerSuite. 0675956 [Josh Rosen] Preserving ordering and partitioning in row format converters also does not help. cc5669c [Josh Rosen] Adding outputPartitioning to Repartition does not fix the test. 2dfc648 [Josh Rosen] Add failing test illustrating bad exchange planning.

JoshRosen added 4 commits August 6, 2015 00:05

Add failing test illustrating bad exchange planning.

2dfc648

Adding outputPartitioning to Repartition does not fix the test.

cc5669c

Preserving ordering and partitioning in row format converters also do…

0675956

…es not help.

Move test to PlannerSuite.

adcc742

JoshRosen added 2 commits August 6, 2015 02:04

Rewrite exchange to fix better handle this case.

c9fb231

Revert accidental ExchangeSuite change.

c628daf

JoshRosen mentioned this pull request Aug 6, 2015

[SQL] [WIP] Join test coverage improvements #7985

Closed

style fix

752b8de

JoshRosen added 3 commits August 6, 2015 09:46

Write a more generic test for EnsureRequirements.

2e0f33a

Add test for requiresChildrenToProduceSameNumberOfPartitions.

5172ac5

Small assertion cleanup.

0725a34

JoshRosen added 2 commits August 6, 2015 12:55

Add failing test to demonstrate allCompatible bug

a1c12b9

Fix the test by adding the compatibility check to EnsureRequirements

4f08278

JoshRosen changed the title ~~[SQL] [WIP] Demonstration of issue in Exchange planning~~ [SPARK-9703] [SQL] [WIP] EnsureRequirements should not add unnecessary shuffles when only ordering requirements are unsatisfied Aug 6, 2015

JoshRosen added 2 commits August 6, 2015 14:40

Add even more tests.

8dbc845

Add more comments

06aba0c

JoshRosen added 2 commits August 6, 2015 15:22

Further expand comment / reasoning

642b0bb

Further refinement to comments / reasoning

fee65c4

JoshRosen reviewed Aug 6, 2015
View reviewed changes

JoshRosen changed the title ~~[SPARK-9703] [SQL] [WIP] EnsureRequirements should not add unnecessary shuffles when only ordering requirements are unsatisfied~~ [SPARK-9703] [SQL] EnsureRequirements should not add unnecessary shuffles when only ordering requirements are unsatisfied Aug 6, 2015

yhuai reviewed Aug 6, 2015
View reviewed changes

JoshRosen added 4 commits August 7, 2015 19:35

Merge remote-tracking branch 'origin/master' into exchange-fixes

2c7e126

Rename DummyPlan to DummySparkPlan.

18cddeb

Update conditions for requiring child compatibility.

1307c50

Giant comment explaining compatibleWith vs. guarantees

8784bd9

JoshRosen reviewed Aug 8, 2015
View reviewed changes

JoshRosen added 2 commits August 8, 2015 23:29

More guarantees vs. compatibleWith cleanup; delete BroadcastPartition…

0983f75

…ing.

Rewrite EnsureRequirements _yet again_ to make things even simpler

38006e7

JoshRosen changed the title ~~[SPARK-9703] [SQL] EnsureRequirements should not add unnecessary shuffles when only ordering requirements are unsatisfied~~ [SPARK-9703] [SQL] Refactor EnsureRequirements to avoid certain unnecessary shuffles Aug 9, 2015

asfgit closed this in 23cf5af Aug 9, 2015

cloud-fan reviewed Aug 10, 2015
View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-9703] [SQL] Refactor EnsureRequirements to avoid certain unnecessary shuffles #7988

[SPARK-9703] [SQL] Refactor EnsureRequirements to avoid certain unnecessary shuffles #7988

JoshRosen commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

JoshRosen commented Aug 6, 2015

SparkQA commented Aug 6, 2015

JoshRosen Aug 6, 2015

JoshRosen Aug 8, 2015

SparkQA commented Aug 6, 2015

yhuai Aug 6, 2015

JoshRosen Aug 6, 2015

SparkQA commented Aug 7, 2015

SparkQA commented Aug 7, 2015

SparkQA commented Aug 7, 2015

JoshRosen commented Aug 7, 2015

SparkQA commented Aug 8, 2015

JoshRosen Aug 8, 2015

JoshRosen commented Aug 9, 2015

SparkQA commented Aug 9, 2015

JoshRosen commented Aug 9, 2015

yhuai commented Aug 9, 2015

cloud-fan Aug 10, 2015

[SPARK-9703] [SQL] Refactor EnsureRequirements to avoid certain unnecessary shuffles #7988

[SPARK-9703] [SQL] Refactor EnsureRequirements to avoid certain unnecessary shuffles #7988

Conversation

JoshRosen commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

SparkQA commented Aug 6, 2015

JoshRosen commented Aug 6, 2015

SparkQA commented Aug 6, 2015

JoshRosen Aug 6, 2015

Choose a reason for hiding this comment

JoshRosen Aug 8, 2015

Choose a reason for hiding this comment

SparkQA commented Aug 6, 2015

yhuai Aug 6, 2015

Choose a reason for hiding this comment

JoshRosen Aug 6, 2015

Choose a reason for hiding this comment

SparkQA commented Aug 7, 2015

SparkQA commented Aug 7, 2015

SparkQA commented Aug 7, 2015

JoshRosen commented Aug 7, 2015

SparkQA commented Aug 8, 2015

JoshRosen Aug 8, 2015

Choose a reason for hiding this comment

JoshRosen commented Aug 9, 2015

SparkQA commented Aug 9, 2015

JoshRosen commented Aug 9, 2015

yhuai commented Aug 9, 2015

cloud-fan Aug 10, 2015

Choose a reason for hiding this comment