Skip to content

[SPARK-29552][SQL] Execute the "OptimizeLocalShuffleReader" rule when creating new query stage and then can optimize the shuffle reader to local shuffle reader as much as possible. #26207

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 3 commits into from
Closed

[SPARK-29552][SQL] Execute the "OptimizeLocalShuffleReader" rule when creating new query stage and then can optimize the shuffle reader to local shuffle reader as much as possible. #26207

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
997e994
fix the flaky test in multi-joins with local shuffle reader
JkSelf Oct 22, 2019
1bc418e
update the comments
JkSelf Oct 23, 2019
b372636
polish the comment for why execute the OptimizaLocalShuffleReader rul…
JkSelf Oct 23, 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
polish the comment for why execute the OptimizaLocalShuffleReader rul… 
…e twice and before ReduceNumShufflePartitions
  • Loading branch information
@JkSelf
JkSelf committed Oct 23, 2019
commit b372636f8820ece343e7fd7c1e1a0d1f60fadb8f
15 changes: 8 additions & 7 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,13 +92,14 @@ case class AdaptiveSparkPlanExec(
// optimizations should be stage-independent.
@transient private val queryStageOptimizerRules: Seq[Rule[SparkPlan]] = Seq(
ReuseAdaptiveSubquery(conf, subqueryCache),
// We will revert the all local shuffle reader node in OptimizeLocalShuffleReader rule
// when introduce additional shuffle. It may be too conservative. So we also re-execute this
// rule when creating new query stage.
// Why need put the OptimizeLocalShuffleReader rule before ReduceNumShufflePartitions rule:
// After we optimize the shuffle reader to local shuffle reader(leaf node),
// it does not need to further optimize the reduce number. So we need to put
// the OptimizeLocalShuffleReader rule before ReduceNumShufflePartitions rule.

// When adding local shuffle readers in 'OptimizeLocalShuffleReader`, we revert all the local
// readers if additional shuffles are introduced. This may be too conservative: maybe there is
// only one local reader that introduces shuffle, and we can still keep other local readers.
// Here we re-execute this rule with the sub-plan-tree of a query stage, to make sure necessary
// local readers are added before executing the query stage.
// This rule must be executed before `ReduceNumShufflePartitions`, as local shuffle readers
// can't change number of partitions.
OptimizeLocalShuffleReader(conf),
ReduceNumShufflePartitions(conf),
ApplyColumnarRulesAndInsertTransitions(session.sessionState.conf,
Expand Down