insert query stages dynamically #6
Conversation
cloud-fan
force-pushed
the
help
branch
2 times, most recently
from
d1a3c60 to
5e233e4
Compare
cloud-fan
force-pushed
the
help
branch
2 times, most recently
from
65fd25a to
8f20f69
Compare
| private val postStageCreationOptimizerRules: Seq[Rule[SparkPlan]] = Seq( | ||
| ReduceNumShufflePartitions(conf), | ||
| CollapseCodegenStages(conf), | ||
| ReuseSubquery(conf)) |
There was a problem hiding this comment.
ReuseSubquery will only be applied to a single stage, it seems it can't be reused by another stage now.
| case class InsertAdaptiveSparkPlan(session: SparkSession) extends Rule[SparkPlan] { | ||
|
|
||
| override def apply(plan: SparkPlan): SparkPlan = plan match { | ||
| case _: ExecutedCommandExec => plan |
There was a problem hiding this comment.
We also need support AE for ExecutedCommandExec. For example the last node may be a CreateHiveTableAsSelectCommand, we also want to run the plan in AE mode.
There was a problem hiding this comment.
CreateHiveTableAsSelectCommand is not a ExecutedCommandExec any more.
There was a problem hiding this comment.
CreateHiveTableAsSelectCommand is not a ExecutedCommandExec any more.
| stop() | ||
| } else { | ||
| readyStages += stage.id | ||
| currentPlan = createQueryStages(currentPlan) |
There was a problem hiding this comment.
The initial call and this call to createQueryStages are invoked from two different threads? Only one thread should call it, right?
| val results = plan.children.map(createQueryStages0) | ||
| CreateStageResult( | ||
| newPlan = plan.withNewChildren(results.map(_.newPlan)), | ||
| allChildStagesReady = results.forall(_.allChildStagesReady)) |
There was a problem hiding this comment.
can we add a case for leaf nodes? This also works but is not obvious to me at the first look..
|
Great work, @cloud-fan ! I just left a few comments. |
No description provided.