apache / apache/datafusion-ballista

Reuse repeated exchanges/subplans in the distributed planner (ReuseExchange/ReuseSubquery analog)

Open
#1,953 0 comments 0 reactions 0 assignees View on GitHub
enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.