andygrove opened a new pull request, #2743:
URL: https://github.com/apache/datafusion-ballista/pull/2743
# Which issue does this PR close?
Closes #2742.
# Rationale for this change
A project that embeds Ballista's scheduler and plans with `ExecutionPlan`
nodes of its own gets none of Ballista's adaptive planning for those nodes.
Ballista's rules see such a node as a leaf, and there is no way to add rules
that understand it: the rule list is fixed, and `ExchangeExec`, which such a
rule needs, is private to the crate. #2742 has the motivating case, Comet
Native Executors, where a join inside a Comet fragment shuffles its whole probe
side even when the build side turns out small.
# What changes are included in this PR?
- `AdaptiveRules` (in `state::aqe::adaptive_rules`): two lists of
`PhysicalOptimizerRule`, both empty by default.
- Preparation rules run once while the job's physical plan is created,
after Ballista's own preparation rules.
- Replan rules run each time the plan is optimized, both when the job is
planned and after each stage completes. They run last, after
`DistributedExchangeRule` has placed the stage boundaries. Placing them earlier
wouldn't work: the exchange of a stage whose inputs just finished is placed in
the same pass, and the stage would start before an earlier rule could see it.
Running last, a rule sees every stage before it can start, and can hold one
back by marking its exchange inactive. In return a rule must leave the stage
boundaries valid, since nothing runs after it.
- `SchedulerConfig::adaptive_rules` and `with_adaptive_rules`. `TaskManager`
passes them through `AdaptiveExecutionGraph::try_new` to
`AdaptivePlanner::try_new`, which now take them as a parameter (`None` in the
existing tests). The static planner ignores them.
- `ExchangeExec` accessors: `is_broadcast`, `is_inactive_stage`,
`partitioning`, `plan_id`, and `with_inactive_stage`, which holds or releases
the exchange's stage while sharing its state. Docs were added to `to_broadcast`
and `input`, which are now reachable from outside the crate.
- A public `state::adaptive` module that exports `AdaptiveRules`,
`AdaptiveRule` and `ExchangeExec`. The `aqe` module itself stays private.
- Tests in `state/aqe/test/adaptive_rules.rs`:
- Preparation rules run once, replan rules run at planning and again after
a stage completes, and they see the finished stage's measured row count.
- A replan rule can hold a stage back, and a held stage doesn't run.
A rule of Comet's own, built on a version of this hook on the `comet-native`
branch, moves a join into its probe side's stage when the build side measures
small. On TPC-H q17 at SF1000 it cut the query from 383 s to 257 s and 294 s
(two runs) on a two-node cluster. Plans submitted as physical plans (#2682) can
pass the same rules to that path once both are merged.
# Are there any user-facing changes?
New public API for embedders: `SchedulerConfig::adaptive_rules`,
`SchedulerConfig::with_adaptive_rules`, and the `state::adaptive` module.
Nothing changes for a scheduler that doesn't set them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
