[Task]: Build a robust, resource-aware optimizer beyond TPCH-shaped workloads

## Summary

TPC-H 1TB currently performs well, but the TPC-DS 1TB failures in #26742, #26743,
#26744, and #26745 expose optimizer assumptions that are reliable mainly for relatively
flat TPCH-shaped star joins. The current behavior is better described as workload-shape
overfitting than as four query-specific defects.

The static evidence and incident symptoms support these working failure chains:

```text
invalid/lossy statistics and cardinality estimates
-> fragile greedy join order and build/probe choice
-> local hash join instead of shuffle
-> selected topology has no effective spill path
-> HashBuild exhausts the query budget

ROLLUP/CUBE and CTE expansion
-> duplicate large scan/join subtrees
-> concurrent HashJoin/Aggregate fan-out
-> local per-operator admission misses query-wide peak memory
-> operators collectively exhaust the 40 GiB query budget
```

Q64 has verbose cardinality evidence. For Q4/Q74/Q80, the structural plan problems are
confirmed but the complete runtime causal chains still require controlled reruns and operator
timelines.

This issue tracks the systematic optimizer roadmap. It must not be closed by query IDs,
SQL-text matching, benchmark-specific hints, or isolated threshold tuning.

## Evidence

The investigation used 103 logical `EXPLAIN` plans covering all 99 TPC-DS queries,
a verbose Q64 plan, and optimizer statistics for all 25 TPC-DS tables. The plan corpus
contains:

| Property | Count |
|---|---:|
| Table scans | 1,445 |
| Joins | 1,156 |
| Shuffled joins | 221 |
| Aggregates | 356 |
| `UNION ALL` nodes | 141 |
| Large fact/returns/inventory scans | 482 |

Confirmed findings:

- ROLLUP/CUBE creates one complete Select/From/Where subtree per grouping set and joins
the branches with `UNION ALL`.
- Q14 statement 1 contains 285 scans and 210 joins; its three sales fact tables are each
scanned 35 times (105 fact scans in total).
- Q80 creates three grouping-set copies of three channel subtrees, producing nine large
sales-left-returns joins.
- CTE reuse rejects estimated materializations above 32 MiB and all variable-width output,
so expensive analytical CTEs such as Q64 `cross_sales` remain inline.
- 23 integer-column NDVs exceed the discrete domain allowed by their collected min/max.
- Join predicate NDV is calculated, but the main inner-join cardinality formula divides by
`min(left.Outcnt, right.Outcnt)` instead of using the join-key NDVs.
- Standalone Filter uses a fixed 5% estimate, aggregate HAVING uses 0.01%, and
`UNION DISTINCT` retains a fixed 70%.
- Q64 is estimated at 712,868 rows where the corresponding actual stream is approximately
411,865,042 rows (about 578x underestimation); this changes the shuffle/spill topology.
- Join ordering uses a high-NDV graph plus selectivity/out-count sorting and first-eligible
greedy expansion rather than candidate-plan cost search.
- Of 482 large scans, 406 have no direct filter and only one large fact scan has a runtime
filter probe. Runtime-filter placement requires a direct table-scan child.
- Q40, Q78, and Q80 contain 13 instances of a filtered dimension being joined after a large
sales-left-returns outer join.
- Shuffle admission is row-count/threshold based and does not estimate query-wide concurrent
build bytes.
- Q4, Q11, and Q74 retain constant-true/false branches through physical plan construction.

Only Q64 currently has verbose actual/estimated evidence. Structural findings for other queries
are confirmed, but their exact per-node estimation error still requires analyzed plans or runtime
profiles.

The collected plans/stats are not yet committed with a reproducible collector and checksums.
M0 must preserve the raw artifacts, collection commands, commit/config/data identity, and parser
used to derive the aggregate counts. Until then, those counts are investigation evidence rather
than a self-contained repository regression.

## System invariants

The roadmap should restore these general contracts:

1. **Statistics validity**: NDV, null counts, min/max, uniqueness, and key domains satisfy
type-aware mathematical and trusted-schema constraints before entering the cost model.
2. **Cardinality relevance**: physical decisions use the statistics of the actual join/filter
keys and preserve known uniqueness/domain constraints.
3. **Shared logical input**: grouping-set count and CTE reference count do not multiply a
common expensive input when sharing/materialization is semantically valid and cheaper.
4. **Resource feasibility**: every memory-sensitive physical candidate has a byte estimate,
concurrent lifetime, and explicit spill capability; selected plans fit the query budget with
reserve or spill before exhausting it.
5. **Rewrite stability**: semantically equivalent SQL forms do not cause order-of-magnitude
changes in scans, build bytes, or peak memory without a data-dependent reason.
6. **Semantic safety**: join reorder, runtime-filter placement, and common-subtree reuse preserve
NULL, outer-join, correlation, volatility, and early-stop semantics.

## Roadmap

### P0 independent correctness work

The optimizer roadmap does not replace the execution-layer fixes in the original incidents:

- preserve the original HashBuild resource error instead of returning only `context canceled`;
- make pipeline terminal delivery and cleanup correct under failure/cancellation;
- preserve spill forward progress and recovery headroom under the shared HashBuild budget;
- verify the local/non-shuffle versus shuffle spill contract explicitly.

These remain independently testable even when a better plan avoids the original failure shape.

### M0: measurement and regression baseline

- Reproduce each original incident under the same commit/config/data snapshot and preserve the
plan, optimizer trace, HashBuild budget-owner timeline, spill counters, and terminal result.
- Commit or publish the raw plan/stats artifacts, checksums, collection commands, and analysis
script needed to reproduce the static findings.
- Add typed optimizer decision traces for cardinality, build/probe, distribution, runtime-filter,
materialization, spillability, candidate rejection, and final selection.
- Collect per-operator estimated/actual rows, row width, build bytes, hash peak, shuffle bytes,
spill bytes, DOP, and wall time.
- Build fixed TPCH 1TB, TPC-DS 1TB, minimized counterexample, and semantic-equivalence corpora.
- Track cardinality q-error, plan size, fact scans/bytes, compile time, peak memory, spill,
completion rate, and execution time.

M0 is the acceptance foundation for all later milestones and must avoid unbounded per-row,
per-batch, or high-cardinality diagnostic cost.

### M1: eliminate deterministic plan multiplication

- Propagate an empty-relation representation through Filter, Join, and Union; eliminate
`Filter(true)` and statically empty branches before physical pipeline construction.
- Reuse the existing spillable materialized source when costing deterministic, explicit,
multi-reference, full-drain CTEs. Remove the fixed-width/32 MiB hard rejection only after
memory/disk/read cost and cancellation/error behavior are covered.
- Design and prototype shared-input grouping sets. Compare extending the existing Group
`GroupingFlag`/rollup machinery with an existing materialized producer plus multiple aggregate
consumers before introducing a new operator family.

### M2: make statistics trustworthy

- Enforce type-aware NDV/domain/null/uniqueness invariants at the stats ownership boundary.
Single-column uniqueness and tuple uniqueness for composite keys must be handled separately;
integer-domain checks must not use lossy `float64` min/max outside its exact range.
- Replace object-count divisors and arbitrary NDV multipliers with mergeable, bounded-error
sketches or a validated extrapolation model.
- Add confidence/freshness metadata and conservative fallback when validation fails.
- Prioritize trusted PK/FK, uniqueness, multi-column NDV/correlation, MCV/range distribution,
null fraction, and join-domain overlap for topology-changing keys. Unverified constraints must
not be treated as data facts.

### M3: repair topology-critical cardinality estimation

- Add an explicit two-side key-stat API and use left-key NDV, right-key NDV, overlap, null
fraction, uniqueness, and trusted PK/FK relationships for equi-join estimates. The existing
predicate NDV is not a substitute for two-side key statistics.
- Replace universal Filter/HAVING factors when column statistics or expression structure are
available.
- Estimate multi-column groups without unconditional independent-NDV multiplication.
- Represent estimate uncertainty or bounds so low-confidence underestimates cannot select a
non-spillable high-risk topology.

### M4: resource-aware physical candidates

- Propagate output row width through Project, Filter, Join, Aggregate, and Union.
- Cost HashBuild, exchange, materialization, and aggregation in bytes, including metadata,
DOP, and concurrency.
- Generate local-hash, broadcast, shuffle-hash, and spillable alternatives where legal.
- First use the existing statement/CN-scoped `HashBuildBudget` owner data to measure overlap,
admission, and spill-recovery headroom. Add conservative query-level peak estimation only where
those measurements demonstrate that local candidate cost is insufficient.
- Treat spillability as a property of a physical candidate, not an after-the-fact flag.

### M5: reduce fact data early

- Place runtime filters through supported Project/Filter/join subtrees to their owning scans.
- Use expected saved rows/bytes/blocks minus build/delivery cost instead of only a 10% NDV ratio.
- Add legality-checked preserved-side outer-join reorder for shapes such as
`(sales LEFT JOIN returns) JOIN filtered_dimension`.

### M6: bounded join-plan search

- Use bounded DP for small connected inner equi-join components (initially a measured threshold
such as `N <= 8`) and retain a hard candidate/time budget.
- Prune dominated candidates using intermediate bytes, network, peak memory, and spill cost.
- For large graphs, use an explicit uncertainty-aware heuristic fallback that considers multiple
starts and retains a robust spillable candidate.
- Keep outer, semi, anti, mark, single, and correlated joins outside the initial enumeration;
handle their legality through separately tested rewrite boundaries.

This milestone comes after trustworthy statistics/cardinality and byte-based candidate cost. A
general memo/property framework should not be introduced merely to search the current invalid estimates and
duplicated logical trees more expensively.

### M7: runtime feedback and adaptive behavior

- Feed actual cardinalities into later statements and stats refresh decisions.
- Consider parameter-sensitive plans and bounded stage-boundary re-optimization.
- Add dynamic repartition/spill fallback when build size exceeds the selected estimate.

M7 must not be used to hide deterministic logical expansion or invalid base statistics.

## Proposed implementation subissues

The umbrella should be implemented as independently reviewable issues, not one monolithic PR:

- [ ] Stats: enforce type-aware NDV invariants and publish invalid-stat telemetry.
- [ ] Cardinality: estimate equi-joins from both-side NDV, overlap, NULLs, and trusted keys.
- [ ] CTE: cost spillable multi-reference CTE reuse with the existing materialized source.
- [ ] Grouping sets: execute a shared input once without cloning its relational subtree.
- [ ] Relational rules: prune compile-time empty relations and identity filters before pipeline
construction.
- [ ] Runtime filter: place profitable probes through transparent nodes and explicitly safe join
paths.
- [ ] Join rewrite: reassociate INNER over LEFT JOIN when predicates use only the preserved side.
- [ ] Physical cost: propagate output widths and cost hash/shuffle candidates in bytes.
- [ ] Join order: bounded DP for small connected inner equi-join components, dependent on Stats,
Cardinality, and byte cost.
- [ ] Execution/HashBuild: preserve spill recovery headroom under the existing shared query budget.

The general memo optimizer, generic common-subexpression elimination, and a complete pipeline
lifetime framework are intentionally deferred until measured recurring needs justify them.

## Validation matrix

Every implementation subtask needs an invariant, its negation, a public-path witness where
externally visible, a typed white-box oracle, and an independent SQL/result or resource oracle.
Whole `EXPLAIN` text snapshots are not sufficient.

| Contract | Minimal witness | Nearby control | White-box oracle | Black-box/metamorphic oracle |
|---|---|---|---|---|
| Grouping sets share input | one table with two-key ROLLUP | plain GROUP BY; DISTINCT aggregate | one input scan and native grouping metadata | result equals explicit `UNION ALL` reference |
| CTE reuse is cost based | deterministic wide CTE referenced twice | volatile/correlated/early-stop consumer | shared producer only when legal | result equals inline form; bounded materialization memory |
| Stats respect domain | sampled bounded integer column | sparse range, NULL, unique key | validated stats object | refreshed `table_stats` satisfies invariants |
| Join estimate uses keys | deterministic PK/FK join | many-to-many, disjoint ranges, NULL | estimate follows key metadata | actual/estimate remains within agreed bounds |
| Runtime filter reaches scan | filtered dimension behind Project/Filter | unsupported expression/unsafe join | probe reaches owning fact scan | same rows with fewer scanned rows/blocks |
| Outer reorder is legal | filter on LEFT JOIN preserved side | predicate references nullable side | typed tree changes only when legal | results equal rewrite-disabled control including unmatched rows |
| Plan fits memory | two concurrent large hash builds | one small or skewed build | byte/lifetime estimates exist | bounded-memory execution completes or spills without OOM |
| Empty branches disappear | `UNION ALL` branch with `WHERE false` | plan-time unknown parameter | empty removed, parameter branch retained | same rows and result types |

## Acceptance gates

This tracking issue is complete only when the following are demonstrated without query-specific
rules:

- each original incident has a controlled baseline with preserved raw artifacts and final result;
- zero invalid base-statistics invariants in the fixed TPC-H/TPC-DS datasets;
- Q14/Q67/Q77/Q80 fact scans no longer multiply with grouping-set count;
- Q64 no longer selects a non-spillable topology from the observed low-confidence underestimate;
- Q80 completes or spills under the same 40 GiB query limit without exceeding its memory budget;
- semantic-equivalent query rewrites do not cause order-of-magnitude scan/build/peak-memory changes;
- TPCH 1TB completion and agreed core-query performance do not regress;
- TPC-DS 1TB completion, peak memory, compile time, and runtime changes are reported per milestone;
- #26742–#26745 retain independent execution/error/cleanup regressions where applicable.

Milestone-specific gates should include:

- Stats deterministic fixtures: join-cardinality median q-error at most 2 and p95 at most 10,
reported separately for PK/FK, many-to-many, disjoint-range, NULL, stale, and missing Stats;
- grouping sets: the common input executes once, input scan bytes do not grow linearly with the
grouping-set count, and plan size is `O(base subtree + grouping metadata)`;
- CTE reuse above the existing in-memory retention bound: spill is observed, retained memory
remains bounded, and variable-width/cancel/error/prepared-reuse paths pass;
- bounded DP: the chosen plan is not dominated by another enumerated legal candidate in build
bytes, network bytes, and peak hash bytes, while planner p95 latency stays within a measured
hard budget;
- Q64/Q80 under 40 GiB: execution completes or spills without memory-admission failure, peak
query memory stays within the configured cap, and spill disk/FD use stays within budget;
- the full suite reports all 99 TPC-DS queries, geometric-mean runtime/scan bytes, maximum
single-query regression, planning time, plan nodes, peak memory, and spill bytes.

Exact q-error and performance thresholds should be set from the M0 baseline rather than chosen
without measurement. Each milestone must report before/after distributions, not only one query.

## Scope boundaries

In scope:

- optimizer observability needed to evaluate the roadmap;
- statistics validity and topology-critical cardinality estimation;
- logical plan sharing and dead-branch elimination;
- resource-aware physical candidates, runtime filters, join reorder, and bounded join search;
- regression and benchmark gates preventing TPCH-shaped overfitting.

Out of scope for this umbrella issue:

- query-ID or SQL-text special cases;
- closing execution lifecycle/error-propagation defects solely because a faster plan avoids them;
- a full optimizer framework rewrite before the measured invariants require it;
- runtime adaptivity used as a substitute for fixing deterministic planner defects;
- assigning implementation to one monolithic PR. Each milestone should be split into independently
reviewable subissues with its own owner and acceptance evidence.

## Related issues

- #26742
- #26743
- #26744
- #26745
- #25782