Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
JkSelf wants to merge 8 commits into from
Closed

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5b7ff2d
support the build side and probe side to local shuffle reader as far …
JkSelf Oct 29, 2019
e13f637
update the implementation of OptimizeLocalShuffleReader rule
JkSelf Oct 29, 2019
782827a
add the comments
JkSelf Oct 29, 2019
f3bb9ce
re-optimize the rule of OptimizeLocalShuffleReader
JkSelf Oct 30, 2019
152aaa6
resolve the comments
JkSelf Oct 30, 2019
e510e96
resolve the comment and update the package of ReduceNumShufflePartiti…
JkSelf Oct 31, 2019
1e947db
update the package dependency
JkSelf Oct 31, 2019
573ffcd
resolve the comments and build failed fix
JkSelf Oct 31, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
resolve the comment and update the package of ReduceNumShufflePartiti… 
…ons rule
  • Loading branch information
@JkSelf
JkSelf committed Oct 31, 2019
commit e510e96d6ea489092385e4f8e50da05df8e0b6fb
44 changes: 29 additions & 15 deletions ...e/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeLocalShuffleReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,25 @@ import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, UnknownPartit
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
import org.apache.spark.sql.execution.exchange.{EnsureRequirements, ShuffleExchangeExec}
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BuildLeft, BuildRight}
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BuildLeft, BuildRight, BuildSide}
import org.apache.spark.sql.internal.SQLConf

object BroadcastJoinWithShuffleLeft {
def unapply(plan: SparkPlan): Option[(QueryStageExec, BuildSide)] = plan match {
case join: BroadcastHashJoinExec if ShuffleQueryStageExec.isShuffleQueryStageExec(join.left) =>
Some((join.left.asInstanceOf[QueryStageExec], join.buildSide))
case _ => None
}
}

object BroadcastJoinWithShuffleRight {
def unapply(plan: SparkPlan): Option[(QueryStageExec, BuildSide)] = plan match {
case join: BroadcastHashJoinExec if ShuffleQueryStageExec.isShuffleQueryStageExec(join.right) =>
Some((join.right.asInstanceOf[QueryStageExec], join.buildSide))
case _ => None
}
}

/**
* A rule to optimize the shuffle reader to local reader as far as possible
* when converting the 'SortMergeJoinExec' to 'BroadcastHashJoinExec' in runtime.
Expand Down Expand Up @@ -55,12 +71,12 @@ case class OptimizeLocalShuffleReader(conf: SQLConf) extends Rule[SparkPlan] {
}
// Add local reader in probe side.
val withProbeSideLocalReader = plan.transformDown {
case join: BroadcastHashJoinExec if canUseLocalShuffleReaderProbeRight(join) =>
val localReader = LocalShuffleReaderExec(join.right.asInstanceOf[QueryStageExec])
join.copy(right = localReader)
case join: BroadcastHashJoinExec if canUseLocalShuffleReaderProbeLeft(join) =>
val localReader = LocalShuffleReaderExec(join.left.asInstanceOf[QueryStageExec])
join.copy(left = localReader)
case join @ BroadcastJoinWithShuffleLeft(shuffleStage, BuildRight) =>
val localReader = LocalShuffleReaderExec(shuffleStage)
join.asInstanceOf[BroadcastHashJoinExec].copy(left = localReader)
case join @ BroadcastJoinWithShuffleRight(shuffleStage, BuildLeft) =>
val localReader = LocalShuffleReaderExec(shuffleStage)
join.asInstanceOf[BroadcastHashJoinExec].copy(right = localReader)
}

def numExchanges(plan: SparkPlan): Int = {
Expand All @@ -81,14 +97,12 @@ case class OptimizeLocalShuffleReader(conf: SQLConf) extends Rule[SparkPlan] {
// 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 &&
ShuffleQueryStageExec.isShuffleQueryStageExec(join.left)) =>
val localReader = LocalShuffleReaderExec(join.left.asInstanceOf[QueryStageExec])
join.copy(left = localReader)
case join: BroadcastHashJoinExec if (join.buildSide == BuildRight &&
ShuffleQueryStageExec.isShuffleQueryStageExec(join.right)) =>
val localReader = LocalShuffleReaderExec(join.right.asInstanceOf[QueryStageExec])
join.copy(right = localReader)
case join @ BroadcastJoinWithShuffleLeft(shuffleStage, BuildLeft) =>
val localReader = LocalShuffleReaderExec(shuffleStage)
join.asInstanceOf[BroadcastHashJoinExec].copy(left = localReader)
case join @ BroadcastJoinWithShuffleRight(shuffleStage, BuildRight) =>
val localReader = LocalShuffleReaderExec(shuffleStage)
join.asInstanceOf[BroadcastHashJoinExec].copy(right = localReader)
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion ...e/src/main/scala/org/apache/spark/sql/execution/adaptive/ReduceNumShufflePartitions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.spark.sql.execution.adaptive.rule
package org.apache.spark.sql.execution.adaptive

import scala.collection.mutable.ArrayBuffer
import scala.concurrent.duration.Duration
Expand Down