perf(scheduler): run one stage when the plan asks for the same stage twice

[Dandandan]

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.

How two exchanges are judged identical is the part to review. DataFusion gives physical plans no structural equality — ExecutionPlan: Any + Debug + DisplayAs + Send + Sync, no DynEq/DynHash — so something has to stand in for it, and rendered text is not safe: two different in-memory tables both display as DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1], so matching on text merges unrelated scans and turns a join into a self-join. Eight AQE tests failed exactly that way on a first attempt.

This matches on the serialized physical plan instead, encoded with the same codec that ships plans to executors. Those bytes are what a task actually runs, so equal bytes mean either copy computes the same rows, and a plan the codec cannot encode is one the rule leaves alone. Scope is additionally kept to pipelines whose scans read files — not needed for correctness once identity is the encoded plan, but it keeps the rule to the case it was written for. Lifting it would also merge in-memory scans of identical content, which is sound and needs the stage counts in two AQE tests updated.

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.493s	0.327s	0.675	0.65, 0.62, 0.71, 0.70
q1 (control)	0.754s	0.748s	0.974	1.00, 0.95, 1.11, 0.95

q15 is 1.5x faster with the control clean. An earlier revision that matched on plan text measured 0.588 on q15 (0.56–0.61 across four pairs) with controls at 0.998 and 0.993, and merged exactly the same stages; the difference between the two figures is run-to-run variation plus the planning cost of encoding each exchange input. q11 measured 0.966 (0.91–0.98) — its duplicate is a 25-row nation scan, so there is little there to win.

The win is structural: an entire filtered lineitem scan and group-by stop being computed. That should hold at scale, where q15 takes 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.

[milenkovicm]

is this similar to #1954 ?

[Dandandan]

is this similar to #1954 ?

Yes it is!