Dandandan opened a new pull request, #2345:
URL: https://github.com/apache/datafusion-ballista/pull/2345
## Which issue does this PR close?
None filed. Found by profiling the executed TPC-H plans: q15 computes its
`revenue0` view twice.
**Stacked on #2344** — the first commit here is that PR's, so review the
second (`perf(scheduler): run one stage when the plan asks for the same stage
twice`). The dependency is real, not just convenience: without #2344 a scan's
rendered predicate carries one copy per replan, so two structurally identical
subplans can render *differently* and the match below misses them.
## Rationale for this change
A query that reads the same data the same way twice plans two exchanges over
identical inputs, and each becomes its own stage. TPC-H q15 references the
`revenue0` view twice — once joined to `supplier`, once for
`max(total_revenue)` — so the whole view is computed twice, a filtered lineitem
scan and a group-by per copy:
```
SortShuffleWriterExec: partitioning=Hash([l_suppkey@0], 4)
AggregateExec: mode=Partial, gby=[l_suppkey], aggr=[sum(l_extendedprice *
(1 - l_discount))]
FilterExec: l_shipdate >= 1996-01-01 AND l_shipdate < 1996-04-01
DataSourceExec: lineitem
```
Spark plans the same query with a `ReusedExchange`. Ballista had no
equivalent: across the 22 executed plans, no stage was ever consumed by more
than one stage.
## What changes are included in this PR?
`ReuseIdenticalStagesRule`, after the rule that creates the exchanges: where
two exchanges cover structurally identical inputs and want the same
partitioning, the duplicates are pointed at the first one's stage by sharing
its stage id and its resolved-partition slot. The stage runs once and both
consumers read its output. `output_links` was already a list, so a stage
feeding several consumers needed no new plumbing; `identify_runnable_stages`
now returns a shared stage once so it is not launched twice.
**Matching is gated on file-backed scans, and that gate is the part to
review.** Two *different* in-memory tables both render as `DataSourceExec:
partitions=4, partition_sizes=[1, 1, 1, 1]`, so matching on rendered text alone
merges unrelated scans and turns a join into a self-join — eight AQE tests
failed exactly that way before the gate. A file scan names its files, which
makes the text identifying. A stronger fingerprint (comparing the serialized
physical plan, which is what the executor receives) would widen this safely and
is the obvious follow-up.
## Are these changes tested?
`cargo test -p ballista-scheduler` passes (362 + 25); clippy clean.
Plan effect: q15 goes from 7 stages to 6 and from two lineitem scans to one;
q11 from 6 stages to 5. All 22 TPC-H queries return unchanged row counts.
Paired A/B at SF10 on two executors x 4 vcores, baseline = the parent commit
so this measures the rule alone. Four repetitions per query, order flipped each
rep, best of 3 iterations per run:
| query | base | this PR | median paired ratio | per-pair ratios |
| ----- | ---- | ------- | ------------------- | --------------- |
| **q15** | 0.442s | 0.263s | **0.588** | 0.56, 0.61, 0.59, 0.59 |
| q11 | 0.133s | 0.127s | 0.966 | 0.97, 0.96, 0.98, 0.91 |
| q1 (control) | 0.631s | 0.630s | 0.998 | 1.00, 1.00, 1.00, 0.99 |
| q3 (control) | 0.790s | 0.825s | 0.993 | 0.97, 1.01, 1.21, 0.98 |
q15 is **1.7x faster**, all four pairs within 0.56–0.61, with the controls
pinned at 0.998 and 0.993. The win is structural — an entire filtered scan and
group-by stop being computed — so it should hold at scale, where q15 is 23.7s
in the SF1000 benchmark.
## Are there any user-facing changes?
No API or configuration change. A query that plans the same stage twice now
runs it once, so plans have fewer stages and stage ids shift accordingly.
--
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]
