andygrove opened a new issue, #1953: URL: https://github.com/apache/datafusion-ballista/issues/1953
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** The distributed planner (`ballista/scheduler/src/planner.rs`, `plan_query_stages`) inserts shuffle boundaries by walking the physical plan, but it has **no mechanism to detect and reuse identical repeated subplans** — there is no `ReuseExchange` / `ReuseSubquery` analog anywhere in the codebase (`grep -r "Reuse" ballista/` finds nothing). When a query contains the same subtree more than once (common in TPC-H: Q2, Q11, Q14, Q15, Q17, Q20, Q21 all reference a shared aggregate/scan subquery), Ballista plans and **executes each occurrence independently** — redundant scans, redundant shuffles, and redundant compute for work that produces identical results. Spark (and Comet) avoid this with `ReuseExchange` and `ReuseSubquery` rules that materialize a shared exchange once and fan its output out to every consumer. In a SF100 TPC-H comparison (2 executors × 8 slots), a properly-parallelized Ballista (`target_partitions=32`, 552 s) roughly matches vanilla Spark (593 s) but trails Comet (340 s). Part of that residual gap is redundant execution of shared subplans that Spark/Comet reuse — it shows up on exactly the multi-subquery queries listed above. **Describe the solution you'd like** Add exchange/subplan reuse to the distributed planning stage: - During `plan_query_stages`, detect structurally-identical `ShuffleWriterExec` subtrees (same input plan, same partitioning) and materialize the stage **once**, wiring every consumer stage's `ShuffleReaderExec` to the single shared output rather than producing a duplicate writer stage. - Equivalently, port Spark's `ReuseExchange`: canonicalize subplans, deduplicate identical exchanges in the stage DAG, and let multiple downstream stages read the same shuffle output. - Benchmark the multi-subquery TPC-H queries (Q2/Q11/Q14/Q15/Q17/Q20/Q21) before/after to quantify the win. **Describe alternatives you've considered** - **Logical-plan-level CTE/common-subexpression elimination** before physical planning. DataFusion has some common-subexpression elimination for expressions, but not distributed-exchange dedup across stages; the reuse has to happen where stages are formed (the planner), because that is where the shuffle boundary — the reusable unit — is created. - **Rely on the query author to rewrite with CTEs.** Doesn't help the general case or the standard TPC-H query text, and Spark/Comet don't require it. **Additional context** Discovered while analyzing why Ballista trails Comet on SF100 TPC-H (the cluster benchmark tooling in the DataFusion benchmark automation). Related: #1375 (dynamic/runtime filters — the other major planner-level gap on join/scan-heavy queries). This issue is specifically about deduplicating repeated exchanges/subplans, which is orthogonal to dynamic filtering. -- 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]
