andygrove opened a new issue, #2678: URL: https://github.com/apache/datafusion-ballista/issues/2678
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** A client that builds its own physical plan can submit it with `submit_physical_plan` or a `PhysicalPlan` query (#1924), but that path always uses the static planner, even though adaptive planning is on by default. `TaskManager::submit_plan` sends `SubmitPlan::Physical` straight to `StaticExecutionGraph` ([task_manager.rs#L315-L333](https://github.com/apache/datafusion-ballista/blob/98d724b3bd9b603211f444d992b0bcca13c75bca/ballista/scheduler/src/state/task_manager.rs#L315-L333)), with a comment noting that AQE plans from the logical plan. These jobs therefore get none of AQE's runtime decisions: join selection from measured sizes, empty-stage propagation and, when enabled, partition coalescing. They also skip `BalanceAndPruneFileGroups`, which the static path applies to logical submissions. Comet Native Executors (apache/datafusion-comet#7288) run whole Spark queries on Ballista by submitting physical plans that Comet builds: DataFusion repartitions and coalesces at the stage boundaries, with Comet's own extension nodes in between. The prototype is apache/datafusion-comet#7289. **Describe the solution you'd like** AQE already works on physical plans after its first step. `AdaptivePlanner::try_new` uses the logical plan only to build the initial physical plan, with the plan-preparation optimizers as the session's physical optimizer rules ([aqe/planner.rs#L130-L161](https://github.com/apache/datafusion-ballista/blob/98d724b3bd9b603211f444d992b0bcca13c75bca/ballista/scheduler/src/state/aqe/planner.rs#L130-L161)). It then hands that plan to `AdaptivePlanner::try_new_with_optimizers`, which takes an `Arc<dyn ExecutionPlan>` ([aqe/planner.rs#L98-L118](https://github.com/apache/datafusion-ballista/blob/98d724b3bd9b603211f444d992b0bcca13c75bca/ballista/scheduler/src/state/aqe/planner.rs#L98-L118)). A physical-plan entry point could: 1. Run the plan-preparation rules over the submitted plan, as `create_physical_plan` does for a logical one. 2. Build the `AdaptivePlanner` with `try_new_with_optimizers` and the default optimizers. 3. Add a matching `AdaptiveExecutionGraph` constructor, and route `SubmitPlan::Physical` through it when `ballista.planner.adaptive.enabled` is set, keeping the static planner otherwise. Open questions: - **Extension nodes.** The AQE rules match DataFusion operators by type. A node they don't know should be left alone, and its `input_distribution_requirements` respected. For example, coalescing the partitions of a shuffle that feeds an extension node changes how many partitions the node sees, which is only safe if the node doesn't depend on that count. Comet's fragments do depend on it. - **Explain.** `handle_explain_plan` takes a logical plan, which a physical submission doesn't have. **Describe alternatives you've considered** - Submitting a logical plan, with a `UserDefinedLogicalNode` for each extension node and a custom `QueryPlanner` on the scheduler. The client already has the physical plan, so this means writing logical wrappers and a planner only to get back to it. - Keeping physical submissions on the static planner, as today, and documenting it. **Additional context** Part of apache/datafusion-comet#7288. Physical-plan submission came from #1924. -- 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]
