[SPARK-7871][SQL]Improve the outputPartitioning for HashOuterJoin #6413

chenghao-intel · 2015-05-26T15:16:54Z

https://issues.apache.org/jira/browse/SPARK-7871
Optimize the Full outer join followed by another join with the same equi-join key.
For example:

explain SELECT l.key, m.key, r.key from src l full outer join src m on l.key=m.key full outer join src r on l.key=r.key;

Before this PR: (4 Exchange, and include one Exchange for intermediate result repartitioning, this probably cause huge performance problem)
== Physical Plan ==
Project [key#1,key#3,key#5]
 HashOuterJoin [key#1], [key#5], FullOuter, None
  Exchange (HashPartitioning [key#1], 200), []
   Project [key#1,key#3]
    HashOuterJoin [key#1], [key#3], FullOuter, None
     Exchange (HashPartitioning [key#1], 200), []
      HiveTableScan [key#1], (MetastoreRelation default, src, Some(l)), None
     Exchange (HashPartitioning [key#3], 200), []
      HiveTableScan [key#3], (MetastoreRelation default, src, Some(m)), None
  Exchange (HashPartitioning [key#5], 200), []
   HiveTableScan [key#5], (MetastoreRelation default, src, Some(r)), None

Applied this PR (3 Exchange only for raw table data repartitioning)
== Physical Plan ==
Project [key#1,key#3,key#5]
 HashOuterJoin [key#1], [key#5], FullOuter, None
  Project [key#1,key#3]
   HashOuterJoin [key#1], [key#3], FullOuter, None
    Exchange (HashPartitioning 200)
     HiveTableScan [key#1], (MetastoreRelation default, src, Some(l)), None
    Exchange (HashPartitioning 200)
     HiveTableScan [key#3], (MetastoreRelation default, src, Some(m)), None
  Exchange (HashPartitioning 200)
   HiveTableScan [key#5], (MetastoreRelation default, src, Some(r)), None

However, we don't want to change the logic for being followed by a GROUP BY, whose keys are exactly the same as the join key.


explain SELECT l.key, count(m.value) from src l full outer join src m on l.key=m.key group by l.key;
== Physical Plan ==
Aggregate false, [key#9], [key#9,Coalesce(SUM(PartialCount#14L),0) AS _c1#7L]
 Exchange (HashPartitioning 200)
  Aggregate true, [key#9], [key#9,COUNT(value#12) AS PartialCount#14L]
   Project [key#9,value#12]
    HashOuterJoin [key#9], [key#11], FullOuter, None
     Exchange (HashPartitioning 200)
      HiveTableScan [key#9], (MetastoreRelation default, src, Some(l)), None
     Exchange (HashPartitioning 200)
      HiveTableScan [key#11,value#12], (MetastoreRelation default, src, Some(m)), None
Even the join key exactly the same with the group by key, however, the full outer join probably produce the null value for (l.key), hence we have to add another Exchange for Aggregate.

As we probably involve more factors to determine if data shuffle needed, I've also refactor the entire code for EnsureRequirements, by introducing the Gap of the requiredDistribution and the outputPartitioning.

// to describe the output data distribution.
abstract class Partitioning {
/** the number of partitions that the data is split across */
  numPartitions: Option[Int] = None,

  /** the expressions that are used to key the partitioning. */
  clusterKeys: Seq[Expression] = Nil,

  /** the expression that are used to sort the data. */
  sortKeys: Seq[SortOrder] = Nil,

  /** work with `sortKeys` if the sorting cross or just within the partition. */
  globalOrdered: Boolean = false,

  /** to indicate if null clustering key will be generated. */
  additionalNullClusterKeyGenerated: Boolean = true
}

// to describe the required data distribution to the child operator.
trait Distribution (UnspecifiedDistribution, ClusteredDistribution, OrderedDistribution) 

// to describe the additional operations needed to the child operator,
// according to the associated `requiredDistribution` and the child's `outputPartitioning`.
trait Gap(NoGap, SortKeyWithinPartition, GlobalOrdering, RepartitionKey, RepartitionKeyAndSort)

This contains the code refactor for exchange in Spark SQL. It's a WIP PR and still need to do:

Enable more unit tests
Add More Scala Doc
Review all of the existed physical Plan for its requiredDistribution and outputPartitioning

A known issue will be solved in another PR once this PR merged. (https://issues.apache.org/jira/browse/SPARK-2205)

SparkQA · 2015-05-26T15:37:47Z

Test build #33519 has finished for PR 6413 at commit 5e6516f.

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

SparkQA · 2015-05-27T15:00:55Z

Test build #33589 has finished for PR 6413 at commit 970a2dc.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(

jeanlyn · 2015-05-27T17:25:22Z

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

+                  .queryExecution.executedPlan
+    val exchanges = planned.collect { case n: Exchange => n }
+
+    assert(exchanges.size === 3)


Is these changs doesn't effect to

testData .join(testData2, testData("key") === testData2("a"), "outer") .join(testData2, testData("a") === testData3("a"), "outer")

?

This requires some further change I think, @yhuai should have some idea on this.

SparkQA · 2015-05-27T17:36:43Z

Test build #33593 has finished for PR 6413 at commit a69f2ae.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(

scwf · 2015-05-28T01:53:33Z

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

+ * as a valid value if `nullKeysSensitive` == true.
+ *
+ * For examples:
+ * JOIN KEYS: values contains null will be considered as invalid values, which means


here values means the original value of the table or the intermediate value of the join?
is the null in original data of table also considered as invalid?

Should be the input value. (Either the original data from table or the intermediate result(e.g. join outputs)).

Validity of the null in the original table, depends on the semantics, in Join, it's should also be invalid, but it's valid for Group BY.

It would an other optimization for repartition, contains null in the join keys.

chenghao-intel · 2015-05-28T14:12:10Z

sorry, it will causes performance regression for case like

left join a.key=b.key group by a.key, will figure out how to fix it soon.

SparkQA · 2015-06-22T16:00:28Z

Test build #35457 has finished for PR 6413 at commit b17a74d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(
- sealed case class Partitioning(

chenghao-intel · 2015-06-23T00:22:58Z

retest this please

SparkQA · 2015-06-23T01:48:54Z

Test build #35498 has finished for PR 6413 at commit b17a74d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(
- sealed case class Partitioning(

SparkQA · 2015-06-29T07:43:43Z

Test build #35961 has finished for PR 6413 at commit 32d8af0.

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

SparkQA · 2015-06-29T07:48:36Z

Test build #35965 has finished for PR 6413 at commit e59b4d4.

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

chenghao-intel · 2015-06-29T08:01:03Z

retest this please

SparkQA · 2015-06-29T12:01:45Z

Test build #35975 has finished for PR 6413 at commit e59b4d4.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(
- sealed case class Partitioning(

chenghao-intel · 2015-06-30T00:13:42Z

@yhuai sorry it's a big change. :) can you review this for me?

chenghao-intel · 2015-06-30T14:14:03Z

Sorry, I found another bug, will solve it soon.

SparkQA · 2015-07-02T08:00:46Z

Test build #36368 has finished for PR 6413 at commit bd37778.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(
- sealed case class Partitioning(

SparkQA · 2015-07-02T10:30:15Z

Test build #36379 has finished for PR 6413 at commit bd4541d.

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

SparkQA · 2015-07-02T14:33:31Z

Test build #36392 has finished for PR 6413 at commit fcb9aed.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(
- sealed case class Partitioning(

SparkQA · 2015-07-03T14:54:39Z

Test build #36499 has finished for PR 6413 at commit ec4e5c2.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ClusteredDistribution(
- sealed case class Partitioning(

liancheng · 2015-07-07T00:48:16Z

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

+      clusterKeys,
+      sortKeys,
+      globalOrdered,
+      additionalNullClusterKeyGenerated)


You may use the copy method comes with all case classes:

this.copy(numPartitions = Some(num))

JoshRosen · 2015-08-01T17:46:39Z

If I'm not mistaken, I think that this patch's changes will be subsumed by the combination of #7773 and the null-unsafe/safe parts of #7685. Are there any changes in this patch that are missed by the combination of those other two patches?

yhuai · 2015-08-01T18:12:14Z

@JoshRosen Right, #7773 and the null-unsafe/safe parts of #7685 will address the issue.

chenghao-intel · 2015-08-02T13:15:50Z

ok, I am closing this pr.

jeanlyn reviewed May 27, 2015
View reviewed changes

chenghao-intel mentioned this pull request May 28, 2015

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

Closed

scwf reviewed May 28, 2015
View reviewed changes

jeanlyn mentioned this pull request Jun 8, 2015

[SPARK-2205][SPARK-7871][SQL]Advoid redundancy exchange #6682

Closed

chenghao-intel force-pushed the exchange branch from a69f2ae to b17a74d Compare June 22, 2015 15:20

chenghao-intel changed the title ~~[SPARK-7871] [SQL] Improve the outputPartitioning for HashOuterJoin~~ [SPARK-7871][SQL][WIP] Improve the outputPartitioning for HashOuterJoin Jun 22, 2015

chenghao-intel force-pushed the exchange branch from b17a74d to 32d8af0 Compare June 29, 2015 05:48

chenghao-intel changed the title ~~[SPARK-7871][SQL][WIP] Improve the outputPartitioning for HashOuterJoin~~ [SPARK-7871][SQL]Improve the outputPartitioning for HashOuterJoin Jun 29, 2015

chenghao-intel added 9 commits July 2, 2015 00:46

Improve the outputPartitioning for HashOuterJoin

85b8f75

optimize for full outer join

491a890

scalastyle

61a8936

refactor the exchange to support null clustering key as the optimization

ee5b991

remove the AllTuples

42928e9

Add more unit test

2f1e9e4

Add comment for unit test

c61b157

fix bug for the empty clustering key

de9cde0

Add more comment

bd37778

chenghao-intel force-pushed the exchange branch from e59b4d4 to bd37778 Compare July 2, 2015 07:54

scalastyle

bd4541d

Fix bug in GeneratedAggregate

fcb9aed

update the unit test as the default value changed

ec4e5c2

liancheng reviewed Jul 7, 2015
View reviewed changes

chenghao-intel closed this Aug 2, 2015

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-7871][SQL]Improve the outputPartitioning for HashOuterJoin #6413

[SPARK-7871][SQL]Improve the outputPartitioning for HashOuterJoin #6413

chenghao-intel commented May 26, 2015

SparkQA commented May 26, 2015

SparkQA commented May 27, 2015

jeanlyn May 27, 2015

chenghao-intel May 28, 2015

SparkQA commented May 27, 2015

scwf May 28, 2015

chenghao-intel May 28, 2015

chenghao-intel commented May 28, 2015

SparkQA commented Jun 22, 2015

chenghao-intel commented Jun 23, 2015

SparkQA commented Jun 23, 2015

SparkQA commented Jun 29, 2015

SparkQA commented Jun 29, 2015

chenghao-intel commented Jun 29, 2015

SparkQA commented Jun 29, 2015

chenghao-intel commented Jun 30, 2015

chenghao-intel commented Jun 30, 2015

SparkQA commented Jul 2, 2015

SparkQA commented Jul 2, 2015

SparkQA commented Jul 2, 2015

SparkQA commented Jul 3, 2015

liancheng Jul 7, 2015

JoshRosen commented Aug 1, 2015

yhuai commented Aug 1, 2015

chenghao-intel commented Aug 2, 2015

[SPARK-7871][SQL]Improve the outputPartitioning for HashOuterJoin #6413

[SPARK-7871][SQL]Improve the outputPartitioning for HashOuterJoin #6413

Conversation

chenghao-intel commented May 26, 2015

SparkQA commented May 26, 2015

SparkQA commented May 27, 2015

jeanlyn May 27, 2015

Choose a reason for hiding this comment

chenghao-intel May 28, 2015

Choose a reason for hiding this comment

SparkQA commented May 27, 2015

scwf May 28, 2015

Choose a reason for hiding this comment

chenghao-intel May 28, 2015

Choose a reason for hiding this comment

chenghao-intel commented May 28, 2015

SparkQA commented Jun 22, 2015

chenghao-intel commented Jun 23, 2015

SparkQA commented Jun 23, 2015

SparkQA commented Jun 29, 2015

SparkQA commented Jun 29, 2015

chenghao-intel commented Jun 29, 2015

SparkQA commented Jun 29, 2015

chenghao-intel commented Jun 30, 2015

chenghao-intel commented Jun 30, 2015

SparkQA commented Jul 2, 2015

SparkQA commented Jul 2, 2015

SparkQA commented Jul 2, 2015

SparkQA commented Jul 3, 2015

liancheng Jul 7, 2015

Choose a reason for hiding this comment

JoshRosen commented Aug 1, 2015

yhuai commented Aug 1, 2015

chenghao-intel commented Aug 2, 2015