docs(plans): TPC-H SF1000 query plans for community review (not for merge)

[andygrove]

Draft — not intended to merge. Opening this as a review surface so
the plans render inline on GitHub with per-file diffs and comments;
I'll close it once the discussion wraps.

What this is

A snapshot of the logical, pre-execution physical, and executed stage
plans for all 22 TPC-H queries, captured from a single run against
Ballista 54.0.0 with a SF1000 Parquet/Iceberg-partitioned dataset. Plans
were pulled from the scheduler event log added in #2264.

Each qNN.md has:

1. Logical plan (from JobStart)
2. Pre-execution physical plan (from JobStart)
3. One section per executed shuffle stage: stage_plan + input/output
   rows, elapsed compute, and task duration / task input percentiles
   (from JobEnd)

Env, config, and wall-clock summary are in the
README.

Wall-clock (single run, one job at a time, no warm-up)

q1 9.1s · q2 39.5s · q3 25.5s · q4 9.0s · q5 35.5s · q6 5.2s · q7 39.6s
· q8 164.2s · q9 91.7s · q10 39.6s · q11 15.2s · q12 11.3s ·
q13 12.2s · q14 5.1s · q15 10.8s · q16 16.3s · q17 25.8s · q18 50.6s ·
q19 16.0s · q20 26.2s · q21 59.6s · q22 10.0s

Observations from a first pass over the plans

Two themes explain most of the wall-clock outliers. Would love the
community's read on both.

Theme 1 — SortShuffleWriterExec over lineitem-scale stages shows a large time tail on balanced input

Q8 stage 1 is the worst case: 34 tasks, task duration min 14s / median 62s / max 131s, but task-input spread is only 1.36×. A 9.4× time spread
on balanced data. Q9 s1 and Q3 s3 show the same shape (3–3.5× time
spread on ~1× input). Not data skew — looks like sort/spill/memory
pressure on the writer.

• Is SortShuffleWriterExec the intended writer for stages of this size?
• Would raising ballista.shuffle.sort_based.memory_limit_per_task_bytes
  (currently 256 MiB) help, or is this the wrong strategy for
  lineitem-scale scans?

Theme 2 — with coalesce.enabled=false, most tasks for small partitioned dim tables read zero rows

Iceberg-partitioned dim tables (customer by c_mktsegment, supplier
by nation, nation, region) produce a handful of useful tasks and 30+
empty ones. Examples:

Query / Stage	Non-empty tasks / total	Task duration skew (max ÷ median)
q10 s0 (customer)	~2 / 37	85×
q15 s0 (supplier)	~1 / 38	62×
q12 s0 (orders)	~5 / 34	522×
q16 s4	tail of empties	640×

• Is coalesce.enabled=false intentional at SF1000, or should empty /
  tiny partitions be merged automatically?

Metric definitions I couldn't pin down from the event log

Both look like reporting issues rather than execution issues, but they
make the numbers hard to reason about.

• Stage output_rows appears to sum across operators, not report
  the terminal operator. Q17 s2: input_rows=5,999,981,877,
  output_rows=17,999,953,463 (3× lineitem cardinality). Q1 s0:
  output_rows=17,746,748,965 for a stage whose scan is ~6B. The
  pattern matches sum(op.output_rows for op in stage) where the intent
  looks like it was the terminal op.
• Source-stage input_rows is inconsistent. Same plan shape
  (SortShuffleWriter ← PartialAgg ← … ← DataSource(lineitem)) reports
  401 in Q1 s0 but 6B in Q17 s2. Something switches between "batch /
  group count" and "row count" depending on plan shape.

Is either of these an intentional definition, or worth an issue?

Smaller items worth a look

• Q17 does two full lineitem scans (12B rows total) — standard
  correlated-subquery rewrite, no scan-sharing. Any planned work on
  shared scans / CTE materialization in DataFusion / Ballista?
• Q8 stage 5 funnels 80M probe rows through a single task on a
  CollectLeft hash join. Expected for broadcast, but that's a lot for
  one partition — is there a threshold at which the probe should
  re-parallelize?
• Q4 / Q21 report input_rows=3,793,363,900 on lineitem in three
  different stages with three different join keys and no explicit
  lineitem date filter. Same exact count in all three suggests it's
  post-filter cardinality of l_receiptdate > l_commitdate (~63% of
  6B). Just wanted to confirm where in the pipeline that count is
  measured.
• DataSourceExec file-groups sections are huge — every plan
  inlines the full parquet file list per group. Any appetite for a
  compact representation in the event log?

Positive: AdaptiveDatafusionExec appears once at the root of every
plan, no fragmentation.

Not asking for

• Code review — no code changes here, just Markdown under docs/.
• A merge decision — happy to close once folks have had a look, unless
  there's appetite for keeping a versioned plan baseline in-tree.

[milenkovicm]

I'm sorry for being backseat driver, adding spark plans would help to narrow does ballista have planning logic or execution issue

[milenkovicm]

note, maybe we could emit final plan to event log when AQE used

[avantgardnerio]

AI slop good enough it's worth reposting verbatim:

q08 s1 and q09 s2, same run, same cluster, same SortShuffleWriterExec, both into 256 partitions:

  ┌────────┬─────────────────────────────────────────────────┬────────────────────┬────────┬────────┐
  │        │                      plan                       │ rows/task (median) │ median │ µs/row │
  ├────────┼─────────────────────────────────────────────────┼────────────────────┼────────┼────────┤
  │ q08 s1 │ SortShuffleWriter ← DataSourceExec              │             187.0M │ 62.45s │  0.334 │
  ├────────┼─────────────────────────────────────────────────┼────────────────────┼────────┼────────┤
  │ q09 s1 │ SortShuffleWriter ← DataSourceExec              │             142.1M │ 19.18s │  0.135 │
  ├────────┼─────────────────────────────────────────────────┼────────────────────┼────────┼────────┤
  │ q09 s2 │ SortShuffleWriter ← HashJoin ← 2× ShuffleReader │             197.7M │  6.88s │  0.035 │
  └────────┴─────────────────────────────────────────────────┴────────────────────┴────────┴────────┘

  q09 s2 pushes more rows through the same writer and runs a partitioned hash join on top — and it's 9.6× faster per row than q08 s1. If buffering/spilling/consolidating ~190M rows
  into 256 partitions were inherently a 60s operation, q09 s2 could not be 6.9s.

  And q08 s1 vs q09 s1 are the identical operator pair — SortShuffleWriterExec: partitioning=Hash([l_partkey@1], 256) ← DataSourceExec, same 5,999,989,709 input rows, same
  11,999,979,418 output — differing 3.3× on the median.

  The variable that tracks it is the source: the two slow stages read Parquet, the fast one reads ShuffleReaderExec. Task wall includes waiting on S3, and elapsed_compute is task
  wall.

[andygrove]

I'm sorry for being backseat driver, adding spark plans would help to narrow does ballista have planning logic or execution issue

I'll try and get those tomorrow

[avantgardnerio]

I think bumping broadcast_join_threshold_rows to ~1.5M might fix q8. I'll try as soon as I get a chance.

[andygrove]

LLM-assisted reply (Claude Code), reviewed by me before posting.

Good call — filed as #2422 with a design sketch.

Short version of what I found while writing it up: the post-AQE plan is already in the log, just unlabelled. graph_to_job_response renders graph.physical_plan(), and for AdaptiveExecutionGraph that getter returns the live planner.plan, so JobEnd.job.physical_plan is the plan after the last replan — whereas JobStart.physical_plan is the pre-execution one. Same field name, two meanings, and nothing marking the job as adaptive, which is why the plan dumps in this PR only ever showed the pre-execution plan.

So the proposal is mostly about making that explicit (final_physical_plan + an adaptive flag on JobResponse, which rides inside the existing JobEnd.job blob and so needs no event-log schema bump), plus an opt-in per-replan JobPlanRevision event for the actual AQE decision history — off by default, since a full plan render per replan is expensive with the file-group listings.

Open questions are in the issue; question 2 in particular is one for you if you have a view: replan_stages re-optimizes the whole tree after every stage completion and ExchangeExec::with_new_children carries the resolved shuffle state across a rewrite, so in principle a subtree under an already-resolved exchange could be rewritten after that stage ran — which would make the "final plan" disagree with the frozen stage_plan. Is that reachable, or prevented somewhere I missed?

[andygrove]

q8/q9 time per query is highly variable

        {
          "query_id": "TPCH-8",
          "has_errors": false,
          "has_warnings": false,
          "mean_s": 99.662688241,
          "min_s": 57.147825423,
          "max_s": 142.177551059,
          "median_s": 99.662688241,
          "stdev_s": 60.12509559964722,
          "iterations": 2
        },
        {
          "query_id": "TPCH-9",
          "has_errors": false,
          "has_warnings": false,
          "mean_s": 102.72393888650001,
          "min_s": 85.839148227,
          "max_s": 119.608729546,
          "median_s": 102.72393888650001,
          "stdev_s": 23.878699948495463,
          "iterations": 2
        },

[andygrove]

Digging into q8: the 6B-row lineitem shuffle should not exist

Following up on Theme 1 with a closer look at where q8's 164s goes and why.

Summing max task duration per stage gets to ~170s, so the DAG is effectively serialized stage by stage. One stage is 80% of the query:

Stage	What	Max task
1	SortShuffleWriter Hash([l_partkey], 256) over the raw lineitem scan	131.1s
2	part ⋈ lineitem, Partitioned	11.4s
5	⋈ supplier, CollectLeft	7.4s
7	customer scan	7.0s
8	⋈ customer / nation	5.4s
rest		< 4s each

So q8 is really a single-stage problem.

Why we shuffle 6B rows to join against 1.3M

The build side of join plan_id=0 is FilterExec(p_type = 'ECONOMY ANODIZED STEEL') over part. Stage 0's runtime input_rows is 1,332,379, which is exactly 200M / 150 distinct p_type values. That is one p_partkey i64 column, roughly 10.7 MB, about 12x under the configured 128 MiB broadcast threshold. This should have been a broadcast join with no lineitem shuffle at all.

Two independent gates block it, both in to_actual_join (dynamic_join.rs:296):

Gate 1: the pre-execution estimate is wrong by 30x. There are no runtime stats when the initial plan is built, so FilterExec falls back to datafusion.execution.default_filter_selectivity (20%). Estimate: 40M rows, ~320 MB. Over threshold, so under_threshold = false, which gives PartitionMode::Partitioned and therefore JoinSelectionAction::Repartition. That arm at join_selection.rs:297 unconditionally wraps both children in an ExchangeExec. The 131s is committed right there, before a single byte of measured statistics exists.

Gate 2: the 1M row ceiling is a hard AND, evaluated before the byte estimate. In supports_collect_by_thresholds, dynamic_join.rs:493:

if num_rows == 0 || num_rows >= threshold_num_rows {
    return false;
}
estimate_output_byte_size(...).is_some_and(|est| est < threshold_byte_size)

Even with a perfect 1.33M row estimate, the row check rejects the broadcast before the byte estimate runs. broadcast_join_threshold_rows defaults to 1,000,000.

@avantgardnerio your instinct about bumping broadcast_join_threshold_rows to ~1.5M is right about gate 2, but I do not think it fixes q8 on its own, because gate 1 hands the check 40M rows rather than 1.33M.

One more thing worth knowing: once Repartition fires there is no way back. On re-resolution selection_state == Repartitioned, so to_actual_join only chooses between Hash and SortMerge. AQE cannot recover the broadcast even after stage 0 finishes in 1.4s and reveals the true size.

Possible fixes, cheapest first

1. Config-only experiment to confirm the diagnosis. Set datafusion.execution.default_filter_selectivity=1 and ballista.optimizer.broadcast_join_threshold_rows=5000000. That drops the part estimate to 2M rows / 16 MB and clears both gates. If q8 falls from 164s to roughly 35s the diagnosis holds. Neither is a good permanent default, but it isolates the cause in one run. I can do this next.

2. Make the byte estimate authoritative when it is available. Reorder supports_collect_by_thresholds so the row ceiling only applies when estimate_output_byte_size returns None. A 1M-row ceiling on a single i64 key column is 8 MB, which is 16x under the byte threshold. The row rule reads like a fallback for unknown widths, but today it vetoes narrow build sides the byte rule would accept.

3. Asymmetric exchange: resolve the cheap side first. This is the structural fix. Today Repartition shuffles both sides at once. When one side's estimate is Inexact and the other is orders of magnitude larger, we could insert the exchange on the cheap side only and leave the big side unresolved. For q8 that means stage 0 runs for 1.4s, returns an exact 1.33M rows / 10.7 MB, and the join flips to CollectLeft. Lineitem is then scanned once inside the join stage against a broadcast build side, and stages 1 and 2 (142s combined) collapse into roughly 12s. Same shape should help q9 s1, q3 s3, and q17.

The distinction worth encoding is that "estimated large but inexact" is not the same as "known large". Right now both take the same branch.

4. Cross-stage dynamic filters (#1375). Biggest lever for the whole suite rather than just q8. enable_dynamic_filter_pushdown is hard-disabled at extension.rs:775 because a filter produced in one stage never reaches a scan in another. The scheduler is arguably the right place to bridge that, though: stage 0 finishes holding 1.33M p_partkey values, and we could ship a bloom filter of maybe 2 MB into stage 1's DataSourceExec before launching it. Stage 1 would then read ~40M rows instead of 6B. Even keeping the shuffle, that is a ~100x reduction. This is essentially Spark's runtime filter / DPP.

Two secondary findings

Stage 5 runs the whole join in a single task. The metrics are unambiguous: min=median=max=7429ms, one task, 79.9M rows. ShuffleReaderExec::try_new_broadcast gives the reader UnknownPartitioning(1) (shuffle_reader.rs:169-178), and in stage 5 that broadcast reader sits on the probe side (upstream_stage: 2), not the build. The build (supplier, stage 3) is read through a plain 256-partition reader under CoalescePartitionsExec. So both inputs gather everything, the stage collapses to one core, and it also serializes a 256-way sort shuffle write. Every path I traced through SelectJoinRule looks like it should put the broadcast on the build side, so the executed shape does not match what the code appears to produce. Worth tracing which branch actually built this.

Stage 1's time tail looks like contention rather than shuffle cost. Compare against q9 s1, which is the identical operator over the identical 6B rows with six projected columns instead of five:

	cols	median	max	task input min/max
q8 s1	5	62.5s	131.1s	1.36x
q9 s1	6	19.2s	58.1s	8.4x

q8 moves less data and takes 3.3x longer per task, on more balanced input. Together with the repeat run above showing q8 at 57s and 142s across two identical iterations, that reads as variance rather than a deterministic cost. Candidates: 34 concurrent tasks each buffering up to 256 MiB in the sort shuffle writer (up to ~8.7 GiB per executor before spill), or S3 read bandwidth. Separately, 34 tasks for the most expensive stage in the query seems low for 32-core executors, so the scan's file-group count may be worth a look on its own.