[SPARK-28560][SQL][followup] support the build side to local shuffle reader as far as possible in BroadcastHashJoin #26289

JkSelf · 2019-10-29T03:52:57Z

What changes were proposed in this pull request?

PR#25295 already implement the rule of converting the shuffle reader to local reader for the BroadcastHashJoin in probe side. This PR support converting the shuffle reader to local reader in build side.

Why are the changes needed?

Improve performance

Does this PR introduce any user-facing change?

No

How was this patch tested?

existing unit tests

…as possible in BroadcastHashJoin

JkSelf · 2019-10-29T03:53:26Z

@cloud-fan Please help me review. Thanks.

SparkQA · 2019-10-29T05:34:27Z

Test build #112816 has finished for PR 26289 at commit 5b7ff2d.

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

JkSelf · 2019-10-29T05:38:11Z

The failed test may be not related.

cloud-fan · 2019-10-29T12:44:03Z

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala

@@ -525,7 +525,7 @@ object AdaptiveSparkPlanExec {
   * Apply a list of physical operator rules on a [[SparkPlan]].
   */
  def applyPhysicalRules(plan: SparkPlan, rules: Seq[Rule[SparkPlan]]): SparkPlan = {
-    rules.foldLeft(plan) { case (sp, rule) => rule.apply(sp) }


nit: the space should be there.

cloud-fan · 2019-10-29T12:48:15Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

+    val shuffleStages = collectShuffleStages(plan)
+
+    val optimizedPlan = if (shuffleStages.isEmpty ||
+      !shuffleStages.forall(_.plan.canChangeNumPartitions)) {


This is different from ReduceNumShufflePartitions. ReduceNumShufflePartitions needs to change all the shuffles together, so as long as there is a user-added shuffle, we need to skip it.

OptimizeLocalShuffleReader can add local reader to any shuffle, so it's simple

private def canAddLocalReader(stage: QueryStage): Boolean = stage match { case s: ShuffleQueryStage => s.plan.canChangeNumPartitions case ReusedQueryStage(s: ShuffleQueryStage) => s.plan.canChangeNumPartitions } plan.transformUp { case stage: QueryStageExec if canAddLocalReader(stage) => LocalShuffleReaderExec(stage) }

SparkQA · 2019-10-29T13:02:47Z

Test build #112840 has finished for PR 26289 at commit e13f637.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class OptimizeLocalShuffleReader(conf: SQLConf) extends Rule[SparkPlan]

SparkQA · 2019-10-29T13:08:34Z

Test build #112841 has finished for PR 26289 at commit 782827a.

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

maryannxue · 2019-10-29T16:59:15Z

How about we parameterize this rule into two versions, one for adding local shuffle reader on probe side which checks for extra exchanges and the other with no such checks at all?

cloud-fan · 2019-10-29T17:51:20Z

and the other with no such checks at all

We always need the check. A query stage can be very complex, though it doesn't have exchanges in the middle. For example, we may have multiple BHJ, aggregate in one query stage, without shuffle between them.

maryannxue · 2019-10-29T17:57:02Z

Why do we need the check on the build side?

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

JkSelf · 2019-10-30T06:50:06Z

@maryannxue after offline discussion with @cloud-fan , agree with the new optimization method in local reader rule: separate adding local reader in probe and build side and then we can ensure the local reader in build side will not be reverted. Already update the PR, please help review. Thanks.

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

SparkQA · 2019-10-30T07:05:02Z

Test build #112891 has finished for PR 26289 at commit f3bb9ce.

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

cloud-fan · 2019-10-30T07:06:33Z

retest this please

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

cloud-fan · 2019-10-30T07:08:27Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

-    val optimizedPlan = plan.transformDown {
-      case join: BroadcastHashJoinExec if canUseLocalShuffleReaderRight(join) =>
+    // Add local reader in probe side.
+    val tmpOptimizedProbeSidePlan = plan.transformDown {


nit: withProbeSideLocalReader

cloud-fan · 2019-10-30T07:09:46Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

+ *       shuffle introduced. If introduced, we will revert all the local
+ *       reader in probe side.
+ * Step2: Add the local reader in build side and will not check whether
+ *        additional shuffle introduced.Because the build side will not introduce


nit: space after ... introduce.

cloud-fan · 2019-10-30T07:11:16Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

-    if (numExchangeAfter > numExchangeBefore) {
-      logDebug("OptimizeLocalShuffleReader rule is not applied due" +
+    val numExchangeAfter = numExchanges(EnsureRequirements(conf).apply(tmpOptimizedProbeSidePlan))
+    val optimizedProbeSidePlan = if (numExchangeAfter > numExchangeBefore) {


maybe simply "optimizedPlan"

cloud-fan · 2019-10-30T07:12:09Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

+    // Add the local reader in  build side and will not check whether
+    // additional shuffle introduced.
+    optimizedProbeSidePlan.transformDown {
+      case join: BroadcastHashJoinExec if canUseLocalShuffleReaderBuildLeft(join) =>


nit: case join: BroadcastHashJoinExec if join.buildSide == BuildLeft && isShuffle(join.left) =>

viirya · 2019-10-30T07:20:58Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

@@ -56,16 +76,25 @@ case class OptimizeLocalShuffleReader(conf: SQLConf) extends Rule[SparkPlan] {
        case e: ShuffleExchangeExec => e
      }.length
    }
-
+    // Check whether additional shuffle introduced. If introduced, revert the local reader.


Now this rule converts local shuffle reader for all BroadcastHashJoinExec and then reverts all local shuffle readers if any of local shuffle reader causes additional shuffle.

Can we just revert the local shuffle readers that cause additional shuffle and keep these not?

This is the best, but I don't know if there is an easy way to do it.

We can implement using revert all the local reader currently and re-optimize later when we find a better way.

cloud-fan · 2019-10-30T07:27:57Z

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

-      // The child of remaining two BroadcastHashJoin is not ShuffleQueryStage.
-      // So only two LocalShuffleReader.
-      checkNumLocalShuffleReaders(adaptivePlan, 1)
+      // *(7) BroadcastHashJoin [b#24], [a#33], Inner, BuildLeft


It's a bit verbose to put the entire query plan here. How about we only put the sketch?

BroadcastHashJoin +- BroadcastExchange +- LocalShuffleReader* +- ShuffleExchange +- BroadcastHashJoin +- BroadcastExchange +- LocalShuffleReader* +- ShuffleExchange +- LocalShuffleReader* +- ShuffleExchange +- BroadcastHashJoin +- LocalShuffleReader* +- ShuffleExchange +- BroadcastExchange +-HashAggregate +- CoalescedShuffleReader +- ShuffleExchange

cloud-fan · 2019-10-30T07:28:06Z

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

-      // The child of remaining two BroadcastHashJoin is not ShuffleQueryStage.
-      // So only two LocalShuffleReader.
-      checkNumLocalShuffleReaders(adaptivePlan, 1)
+      // *(6) BroadcastHashJoin [cast(value#14 as int)], [a#220], Inner, BuildLeft


JkSelf · 2019-10-30T08:56:45Z

@cloud-fan @viirya update the comments. Please help review again. Thanks.

cloud-fan · 2019-10-30T10:06:35Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

+    // Add the local reader in build side and and do not need to check whether
+    // additional shuffle introduced.
+    optimizedPlan.transformDown {
+      case join: BroadcastHashJoinExec if (join.buildSide == BuildLeft &&


nit: we can unify the condition check some how

object BroadcastJoinWithShuffleLeft { def unapply(plan: LogicalPlan): Option[(QueryStageExec, BuildSide)] = plan match { case join: BroadcastHashJoinExec if ShuffleQueryStageExec.isShuffleQueryStageExec(join.left) => Some((join.left.asInstanceOf[QueryStageExec], join.buildSide)) } } object BroadcastJoinWithShuffleRight ... // add probe side local reader case join @ BroadcastJoinWithShuffleLeft(shuffleStage, BuildRight) => val localReader = ... join.copy(left = localReader) ... // add build side local reader case join @ BroadcastJoinWithShuffleLeft(shuffleStage, BuildLeft) => ...

cloud-fan · 2019-10-30T10:07:08Z

LGTM except one minor comment

SparkQA · 2019-10-30T11:00:47Z

Test build #112895 has finished for PR 26289 at commit f3bb9ce.

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

SparkQA · 2019-10-30T12:50:47Z

Test build #112908 has finished for PR 26289 at commit 152aaa6.

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

…ons rule

SparkQA · 2019-10-31T05:09:34Z

Test build #112993 has finished for PR 26289 at commit e510e96.

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

SparkQA · 2019-10-31T06:39:22Z

Test build #112998 has finished for PR 26289 at commit 1e947db.

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

cloud-fan · 2019-10-31T06:55:13Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

 case class OptimizeLocalShuffleReader(conf: SQLConf) extends Rule[SparkPlan] {

-  def canUseLocalShuffleReaderLeft(join: BroadcastHashJoinExec): Boolean = {
+  def canUseLocalShuffleReaderProbeLeft(join: BroadcastHashJoinExec): Boolean = {


We can remove it now

cloud-fan · 2019-10-31T06:55:18Z

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala

    join.buildSide == BuildRight &&  ShuffleQueryStageExec.isShuffleQueryStageExec(join.left)
  }

-  def canUseLocalShuffleReaderRight(join: BroadcastHashJoinExec): Boolean = {
+  def canUseLocalShuffleReaderProbeRight(join: BroadcastHashJoinExec): Boolean = {


SparkQA · 2019-10-31T10:46:28Z

Test build #113002 has finished for PR 26289 at commit 573ffcd.

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

cloud-fan · 2019-10-31T12:48:59Z

I see Spark R failures in other PRs as well, and it's totally unrelated to this PR: checking CRAN incoming feasibility ...Error in .check_package_CRAN_incoming(pkgdir)

thanks, merging to master!

support the build side and probe side to local shuffle reader as far …

5b7ff2d

…as possible in BroadcastHashJoin

JkSelf added 2 commits October 29, 2019 17:23

update the implementation of OptimizeLocalShuffleReader rule

e13f637

add the comments

782827a

cloud-fan reviewed Oct 29, 2019

View reviewed changes

viirya reviewed Oct 30, 2019

View reviewed changes

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala Show resolved Hide resolved

re-optimize the rule of OptimizeLocalShuffleReader

f3bb9ce

viirya reviewed Oct 30, 2019

View reviewed changes

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala Outdated Show resolved Hide resolved

viirya reviewed Oct 30, 2019

View reviewed changes

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala Outdated Show resolved Hide resolved

...core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Oct 30, 2019

View reviewed changes

viirya reviewed Oct 30, 2019

View reviewed changes

cloud-fan reviewed Oct 30, 2019

View reviewed changes

dongjoon-hyun added the SQL label Oct 30, 2019

resolve the comments

152aaa6

cloud-fan reviewed Oct 30, 2019

View reviewed changes

viirya approved these changes Oct 30, 2019

View reviewed changes

resolve the comment and update the package of ReduceNumShufflePartiti…

e510e96

…ons rule

update the package dependency

1e947db

cloud-fan reviewed Oct 31, 2019

View reviewed changes

resolve the comments and build failed fix

573ffcd

cloud-fan closed this in cd39cd4 Oct 31, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-28560][SQL][followup] support the build side to local shuffle reader as far as possible in BroadcastHashJoin #26289

[SPARK-28560][SQL][followup] support the build side to local shuffle reader as far as possible in BroadcastHashJoin #26289

JkSelf commented Oct 29, 2019

JkSelf commented Oct 29, 2019

SparkQA commented Oct 29, 2019

JkSelf commented Oct 29, 2019

cloud-fan Oct 29, 2019

cloud-fan Oct 29, 2019

SparkQA commented Oct 29, 2019

SparkQA commented Oct 29, 2019

maryannxue commented Oct 29, 2019 •

edited

Loading

cloud-fan commented Oct 29, 2019

maryannxue commented Oct 29, 2019

JkSelf commented Oct 30, 2019

SparkQA commented Oct 30, 2019

cloud-fan commented Oct 30, 2019

cloud-fan Oct 30, 2019

cloud-fan Oct 30, 2019

cloud-fan Oct 30, 2019

cloud-fan Oct 30, 2019

viirya Oct 30, 2019

cloud-fan Oct 30, 2019

JkSelf Oct 30, 2019

cloud-fan Oct 30, 2019

cloud-fan Oct 30, 2019

JkSelf commented Oct 30, 2019

cloud-fan Oct 30, 2019

cloud-fan commented Oct 30, 2019

SparkQA commented Oct 30, 2019

SparkQA commented Oct 30, 2019

SparkQA commented Oct 31, 2019

SparkQA commented Oct 31, 2019

cloud-fan Oct 31, 2019

cloud-fan Oct 31, 2019

SparkQA commented Oct 31, 2019

cloud-fan commented Oct 31, 2019

[SPARK-28560][SQL][followup] support the build side to local shuffle reader as far as possible in BroadcastHashJoin #26289

[SPARK-28560][SQL][followup] support the build side to local shuffle reader as far as possible in BroadcastHashJoin #26289

Conversation

JkSelf commented Oct 29, 2019

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

JkSelf commented Oct 29, 2019

SparkQA commented Oct 29, 2019

JkSelf commented Oct 29, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Oct 29, 2019

SparkQA commented Oct 29, 2019

maryannxue commented Oct 29, 2019 • edited Loading

cloud-fan commented Oct 29, 2019

maryannxue commented Oct 29, 2019

JkSelf commented Oct 30, 2019

SparkQA commented Oct 30, 2019

cloud-fan commented Oct 30, 2019

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JkSelf commented Oct 30, 2019

Choose a reason for hiding this comment

cloud-fan commented Oct 30, 2019

SparkQA commented Oct 30, 2019

SparkQA commented Oct 30, 2019

SparkQA commented Oct 31, 2019

SparkQA commented Oct 31, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Oct 31, 2019

cloud-fan commented Oct 31, 2019

maryannxue commented Oct 29, 2019 •

edited

Loading