andygrove opened a new issue, #2742: URL: https://github.com/apache/datafusion-ballista/issues/2742
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** 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, and has no way to add rules that would. Ballista's adaptive rules match DataFusion's own nodes, such as `HashJoinExec` and `RepartitionExec`. A custom node is a leaf to them. The rule list is fixed: `AdaptivePlanner::default_optimizers` is `pub(super)`, the planner's constructors take no rules, and `SchedulerConfig` has no field for them. `ExchangeExec`, which a rule would need in order to read a finished stage's measured statistics or to hold a stage back, is private to the crate. The case that prompted this is [Comet Native Executors](https://github.com/apache/datafusion-comet/issues/7288), which run Spark queries on Ballista. Each stage is a `CometFragmentExec` whose joins are inside a serialized Comet plan, so Ballista's join selection can't see them, and a join whose build side turns out small still shuffles its whole probe side ([apache/datafusion-comet#7391](https://github.com/apache/datafusion-comet/issues/7391)). On TPC-H q17 at SF1000, `part` filters to 199,925 rows while `lineitem` shuffles all of its 6 billion rows to join it. A rule of Comet's own, run in Ballista's adaptive planner, could hold the `lineitem` stage back until `part` is measured, and then broadcast `part` into the `lineitem` scan. **Describe the solution you'd like** A way for the embedder to add rules to adaptive planning, without replacing Ballista's: - An `AdaptiveRules` trait with two kinds of `PhysicalOptimizerRule`: preparation rules, which run once while the job's plan is created, and replan rules, which run each time the plan is optimized, including after each stage completes. - Replan rules run last, after Ballista has placed the stage boundaries, so they see every stage before it can start and can hold one back by marking its exchange inactive. - A `SchedulerConfig::adaptive_rules` field (and `with_adaptive_rules`), passed through `TaskManager` to the adaptive planner, like `override_physical_codec` but adding rules rather than replacing them. - `ExchangeExec` exported, with accessors for whether it is a broadcast, its partitioning and plan id, and `with_inactive_stage` to hold or release its stage. **Describe alternatives you've considered** - Lifting the joins out of Comet's fragments as DataFusion join nodes, so Ballista's existing rules apply. That is a much larger change on Comet's side, and joins with non-equi conditions would stay opaque. - Replacing the whole rule list. That would make an embedder copy Ballista's rules and track their changes. **Additional context** A prototype of the hook, on Ballista's `comet-native` branch, together with Comet's rule cut q17 at SF1000 from 383 s to 257 s and 294 s (two runs) on a two-node cluster in staged mode. For plans submitted as physical plans (#2682), the same rules would also be passed to that path, with preparation rules running after the submitted shuffles are pinned. That can follow once both are merged. -- 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]
