[SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order #16985

tejasapatil · 2017-02-18T18:35:43Z

What changes were proposed in this pull request?

Jira : https://issues.apache.org/jira/browse/SPARK-19122

leftKeys and rightKeys in SortMergeJoinExec are altered based on the ordering of join keys in the child's outputPartitioning. This is done everytime requiredChildDistribution is invoked during query planning.

How was this patch tested?

Added new test case
Existing tests

tejasapatil · 2017-02-18T18:35:50Z

ok to test

SparkQA · 2017-02-18T20:36:26Z

Test build #73112 has finished for PR 16985 at commit 88d2025.

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

tejasapatil · 2017-02-19T00:26:34Z

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

This seems ugly but I can't think of a better way. The problem is: I want to mutate this ordering at some point in the query planning. I cannot do that when SortMergeJoinExec object is generated because there wont be ample information available at that time.

I tried to add class attributes which would be altered and don't mutate this. Doing that, I saw that that tasks on executor do not see the updated values of the local class attributes.

What information are you missing? The SortMergeExec is replaced after each planning iteration.

I would prefer that we use a lazy val here instead.

@hvanhovell : I had tried that but for some class of queries that didn't work. When I try to get the outputPartitioning for a SMB node, in case of inner-join it is PartioniningCollection. Now one of its children can have a ReusedExchange which is yet to be resolved but if the other child is resolved, then this check fails.

Example query:

SELECT a.i, b.i, c.i FROM mytable a, mytable b, mytable c where a.i = b.i and a.i = c.i

Example query plan:

:- *SortMergeJoin [i#8], [i#9], Inner : :- *Sort [i#8 ASC NULLS FIRST], false, 0 : : +- Exchange hashpartitioning(i#8, 200) : : +- *Project [i#8] : : +- *Filter isnotnull(i#8) : : +- *FileScan orc default.one_column[i#8] Batched: false, Format: ORC, Location: InMemoryFileIndex[file:/Users/tejasp/Desktop/dev/tp-spark/spark-warehouse/one_column], PartitionFilters: [], PushedFilters: [IsNotNull(i)], ReadSchema: struct<i:int> : +- *Sort [i#9 ASC NULLS FIRST], false, 0 : +- ReusedExchange [i#9], Exchange hashpartitioning(i#8, 200)

Just to be on the same page, sharing what I had tried (will update the PR with the change anyways. I know that there would be some unit tests which would fail):

case class SortMergeJoinExec( leftKeys: Seq[Expression], rightKeys: Seq[Expression], ....) { lazy val (reorderedLeftKeys, reorderedRightKeys) = { def reorder( expectedOrderOfKeys: Seq[Expression], currentOrderOfKeys: Seq[Expression]): (Seq[Expression], Seq[Expression]) = { val leftKeysBuffer = ArrayBuffer[Expression]() val rightKeysBuffer = ArrayBuffer[Expression]() expectedOrderOfKeys.foreach(expression => { val index = currentOrderOfKeys.indexWhere(e => e.semanticEquals(expression)) leftKeysBuffer.append(leftKeys(index)) rightKeysBuffer.append(rightKeys(index)) }) (leftKeysBuffer, rightKeysBuffer) } left.outputPartitioning match { case HashPartitioning(leftExpressions, _) if leftExpressions.length == leftKeys.length && leftKeys.forall(x => leftExpressions.exists(_.semanticEquals(x))) => reorder(leftExpressions, leftKeys) case _ => right.outputPartitioning match { case HashPartitioning(rightExpressions, _) if rightExpressions.length == rightKeys.length && rightKeys.forall(x => rightExpressions.exists(_.semanticEquals(x))) => reorder(rightExpressions, rightKeys) case _ => (leftKeys, rightKeys) } } }

@hvanhovell : ping !!! As expected, that doesn't work in all cases and few unit test are failing. I would go back to my original version if you have don't any other idea(s).

@tejasapatil I see your point. Let me think of another way.

tejasapatil · 2017-02-19T00:27:02Z

@hvanhovell : This is as per our discussion in the jira : https://issues.apache.org/jira/browse/SPARK-19122

SparkQA · 2017-02-19T02:00:22Z

Test build #73119 has finished for PR 16985 at commit 2ffaf17.

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

SparkQA · 2017-02-19T02:21:45Z

Test build #73121 has finished for PR 16985 at commit 7f3ea36.

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

SparkQA · 2017-02-21T17:37:49Z

Test build #73221 has finished for PR 16985 at commit 41aa767.

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

tejasapatil · 2017-03-14T21:39:46Z

Jenkins test this please

SparkQA · 2017-03-14T23:41:44Z

Test build #74556 has finished for PR 16985 at commit ba5119b.

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

tejasapatil · 2017-03-14T23:47:14Z

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

@hvanhovell : continuing our discussion at #16985 (comment) :

I found that ReusedExchangeExec does not use child's partitioning which was causing the tests to fail. This version of the PR is with your suggestion + works in all cases (all unit tests are passing).

good catch!

SparkQA · 2017-03-28T17:38:59Z

Test build #75319 has finished for PR 16985 at commit 7cbf0d0.

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

tejasapatil · 2017-03-28T20:46:26Z

@hvanhovell @cloud-fan @gatorsmile can anyone please review this PR ?

cloud-fan · 2017-04-03T14:38:20Z

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

is it fixed by #17339 ?

I looked at #17339 and its doing something orthogonal to whats done here.

#17339 is ensuring that the join outputs' sort ordering has attributes from both relations.
This PR is ensuring that the order of join kets (in both distribution and sort order) is not blindly picked from the order of occurrence in the query string.

cloud-fan · 2017-04-03T14:40:12Z

sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala

Since this is test, I think it's ok to always return the result

tejasapatil · 2017-05-08T21:40:40Z

Jenkins test this please

SparkQA · 2017-05-08T23:46:38Z

Test build #76593 has finished for PR 16985 at commit b2a6f1c.

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

tejasapatil · 2017-05-09T00:19:01Z

Jenkins test this please

SparkQA · 2017-05-09T02:31:12Z

Test build #76601 has finished for PR 16985 at commit b2a6f1c.

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

cloud-fan · 2017-05-09T05:37:31Z

shall we introduce a physical optimizer rule which reorders join predicates based on child.outputOrdering and outputPartitioning?

SparkQA · 2017-05-09T05:52:25Z

Test build #76614 has finished for PR 16985 at commit e202ac1.

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

SparkQA · 2017-05-09T17:15:03Z

Test build #76685 has finished for PR 16985 at commit bd1aa6d.

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

SparkQA · 2017-06-09T05:42:37Z

Test build #77836 has started for PR 16985 at commit 72325b8.

tejasapatil · 2017-06-09T15:34:39Z

Jenkins test this please

SparkQA · 2017-06-09T18:09:06Z

Test build #77845 has finished for PR 16985 at commit 72325b8.

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

tejasapatil · 2017-06-09T20:53:01Z

@cloud-fan : ping

cloud-fan · 2017-06-29T09:05:31Z

Sorry for the delay, pretty busy this month...

So a summary of this patch: order does matter for hash partitioning keys, but doesn't matter for join keys. So we can reorder the join keys to match the child hash partitioning to avoid shuffle.

My last concern is #16985 (comment) . Ideally it is a valid case I think, but we may have some problems in the implementation. Can we investigate it?

…ordering differ from bucketing and sorting order

…s done

… on child.outputOrdering and outputPartitioning

tejasapatil · 2017-08-11T06:02:52Z

@cloud-fan : I was on a long vacation for a quite sometime so couldn't get to this. Wrt to the concern you had, I have replied to that discussion in the PR : https://github.com/apache/spark/pull/16985/files#r132620278 along with a test case which covers the scenario you had mentioned.

tejasapatil · 2017-08-11T06:04:31Z

BTW: The "summary of this patch" in your comment accurately captures what this PR is doing.

cloud-fan · 2017-08-11T06:20:59Z

retest this please

cloud-fan · 2017-08-11T06:47:30Z

sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala

+    }
+
+    // Irrespective of the ordering of keys in the join predicate, the query plan and
+    // query results should always be the same


I think we don't need to test this here, this is an existing property of join implementation in Spark SQL, and should have already been well tested. Then we don't need to change testBucketing

removed validation of query result

cloud-fan · 2017-08-11T06:56:20Z

LGTM except one comment for the test

tejasapatil · 2017-08-11T15:36:57Z

jenkins test this please

SparkQA · 2017-08-11T19:18:47Z

Test build #80541 has finished for PR 16985 at commit 7171b58.

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

gatorsmile · 2017-08-11T22:13:25Z

Thanks! Merging to master.

tejasapatil changed the title ~~[WIP] [SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order~~ [SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order Feb 18, 2017

tejasapatil changed the title ~~[SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order~~ [WIP] [SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order Feb 18, 2017

tejasapatil changed the title ~~[WIP] [SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order~~ [SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order Feb 19, 2017

tejasapatil commented Feb 19, 2017

View reviewed changes

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from 7f3ea36 to 41aa767 Compare February 21, 2017 16:08

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from 41aa767 to ba5119b Compare March 14, 2017 21:38

tejasapatil commented Mar 14, 2017

View reviewed changes

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from ba5119b to 7cbf0d0 Compare March 28, 2017 15:35

cloud-fan reviewed Apr 3, 2017

View reviewed changes

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from 7cbf0d0 to b2a6f1c Compare May 8, 2017 21:40

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from b2a6f1c to e202ac1 Compare May 9, 2017 03:42

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from e202ac1 to bd1aa6d Compare May 9, 2017 15:00

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from 4f76bd0 to 72325b8 Compare June 9, 2017 05:39

tejasapatil added 8 commits August 10, 2017 22:59

[SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates …

217a843

…ordering differ from bucketing and sorting order

resolve ordering of join keys everytime requiredChildDistribution i…

d11fe32

…s done

minor refac

060730c

use lazy val

213c273

always return result in test case

3f27ff2

create a physical optimizer rule which reorders join predicates based…

c36cd1a

… on child.outputOrdering and outputPartitioning

checkstyle

acd4dcb

new test case

b295093

tejasapatil force-pushed the SPARK-19122_join_order_shuffle branch from 72325b8 to b295093 Compare August 11, 2017 05:59

cloud-fan reviewed Aug 11, 2017

View reviewed changes

removed checking for qury result in the test case

7171b58

asfgit closed this in 7f16c69 Aug 11, 2017

tejasapatil deleted the SPARK-19122_join_order_shuffle branch August 11, 2017 23:08

[SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order #16985

[SPARK-19122][SQL] Unnecessary shuffle+sort added if join predicates ordering differ from bucketing and sorting order #16985

Uh oh!

Conversation

tejasapatil commented Feb 18, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

tejasapatil commented Feb 18, 2017

Uh oh!

SparkQA commented Feb 18, 2017

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!

tejasapatil commented Feb 19, 2017

Uh oh!

SparkQA commented Feb 19, 2017

Uh oh!

SparkQA commented Feb 19, 2017

Uh oh!

SparkQA commented Feb 21, 2017

Uh oh!

tejasapatil commented Mar 14, 2017

Uh oh!

SparkQA commented Mar 14, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Mar 28, 2017

Uh oh!

tejasapatil commented Mar 28, 2017

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!

tejasapatil commented May 8, 2017

Uh oh!

SparkQA commented May 8, 2017

Uh oh!

tejasapatil commented May 9, 2017

Uh oh!

SparkQA commented May 9, 2017

Uh oh!

cloud-fan commented May 9, 2017

Uh oh!

SparkQA commented May 9, 2017

Uh oh!

SparkQA commented May 9, 2017

Uh oh!

SparkQA commented Jun 9, 2017

Uh oh!

tejasapatil commented Jun 9, 2017

Uh oh!

SparkQA commented Jun 9, 2017

Uh oh!

tejasapatil commented Jun 9, 2017

Uh oh!

cloud-fan commented Jun 29, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

tejasapatil commented Aug 11, 2017

Uh oh!

tejasapatil commented Aug 11, 2017

Uh oh!

cloud-fan commented Aug 11, 2017

Uh oh!

tejasapatil commented Feb 18, 2017 •

edited

Loading

cloud-fan commented Jun 29, 2017 •

edited

Loading