apache / apache/datafusion-ballista
Reuse repeated exchanges/subplans in the distributed planner (ReuseExchange/ReuseSubquery analog)
- Dominant language
- Rust
- Stars
- 2.1k
- Forks
- 320
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 66
Description
**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.
Contributor guide
Research direction
Start in ballista/scheduler/src/planner.rs, especially plan_query_stages, and inspect how ShuffleWriterExec and ShuffleReaderExec stages are formed. Compare the existing stage DAG with the requested reuse behavior, then benchmark TPC-H Q2, Q11, Q14, Q15, Q17, Q20, and Q21; done means identical exchanges or subplans are materialized once and shared by downstream stages.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100