[zzylol]

Part of #879

Why

--search memo-milp (experimental, #915) searches the sub-expression memo with a MILP lower bound and a lazy loop, but on main it could not plan whole evaluation workloads: every round enumerated every plan of a group choice in Stage 3 (q24h60 checked 1.31M candidates; RangeHeavy grew past 59 GB), coefficient derivation spent ~10 min in the rollup rules on RollupHeavy, a time-limited run with no evaluated round returned no plan (RollupHeavy t11-5m and t10: "time limit ... before finding a valid plan" after 1800 s), and the bound was weak enough that small rows took hundreds of rounds (q24h60 689 s, rh8-q1h60 772 s, i840 not certified after 1800 s). #879 needs each whole workload to return a plan with an honest gap within 300 s, and the components to certify.

What

--search memo-milp (still opt-in; the default search is unchanged and still legacy) now:

• splits the memo into independent clusters (queries sharing no group, coupling site, candidate read or disk-charged raw source) and runs one model and lazy loop per cluster, solves in parallel; under cost model 1 the workload cost is the sum, each cluster reported in search.milp.clusters;
• decodes each MILP solution into one physical plan and evaluates it in Stage 3 once per round (solution decoding + incumbent evaluation), instead of enumerating every plan of the group choice;
• starts each cluster from a warm-start incumbent (every query planned alone by the legacy search), so a time-limited run always returns a valid plan and its gap;
• keeps the solver's proven bound when a solve stops at the time limit, so a stopped run reports its gap without an unlimited relaxation solve;
• encodes which child choices keep each choice's accuracy guarantee (pair and failure-probability rules, read at the largest size Stage 3 can give) as constraints, instead of cutting them one at a time;
• charges more of each plan's real cost in the model so the bound is tighter, and still a lower bound: scans read by built-once nodes or slots at the union rate, per-query latency per node key, held-memory shares over a node's readers, ingestion-time parts at their ingestion price, reach as flows, an incumbent cutoff within the solver's tolerance;
• adds term cuts after each round for the decoded plan's underestimated terms (a source's scans, a query's held memory, a built-once group's ingestion copy), each valid for every plan sharing that term; the no-good cut stays as the fallback;
• computes the memo's rollup and pane-chain rules per coarse build, cached, instead of for every subset (RollupHeavy coefficient derivation 662 s -> 15 s).

The selection report gains search.milp.decode_failures and search.milp.clusters.

How

search::milp::model::search builds the memo, clusters::split partitions its queries by union-find, and search_clusters runs one ClusterLoop per cluster (new -> offer_independent_plan warm start -> next_round / take_solution -> finish). run_rounds solves the clusters' models on scoped threads; incumbent evaluation stays on the main thread. take_solution calls decode::decode, which composes the chosen choice versions into one CandidateLogicalASAPDAG (identical non-scan nodes one node via with_shared_candidates, plus the widened-scans variant), materializes it with optimize_physical_asap_chosen from the solution's built-once and slot choices, and offers each variant to the Stage 3 selector. The loop then adds cuts::term_cuts_of / held_cuts_of / ingestion_copy_cuts_of and the no-good cut of the decoded choice set. clusters::combine puts the clusters' plans back in workload order and checks that the whole-workload cost equals the sum (else the memo is searched as one).

Key code interfaces added / modified

crates/plan-selection/src/lib.rs

/// The MILP search's report (`search.milp`). The type stays.
pub struct MilpReport {
    /// Last MILP lower bound. // unchanged
    pub lower_bound: Option<f64>,
    /// First round's lower bound. // unchanged
    pub first_lower_bound: Option<f64>,
    /// (incumbent − lower bound) / incumbent, at least 0. // unchanged
    pub gap: Option<f64>,
    /// Lazy-loop rounds (summed over clusters). // unchanged
    pub rounds: u32,
    /// Model variables (summed over clusters). // unchanged
    pub variable_count: u64,
    /// Model constraints (summed over clusters). // unchanged
    pub constraint_count: u64,
    /// Wall time of the solves in ms. // unchanged
    pub solver_time_ms: u64,
    /// Whether the search stopped at its time limit. // unchanged
    pub time_limit_reached: bool,
    /// Solutions decoding could not compose into a plan; each keeps its
    /// objective as a lower bound so the gap stays honest. Default 0. // new
    pub decode_failures: u32,
    /// Each independent cluster's search, ordered by first query; empty
    /// (not serialized) when the memo was searched as one. // new
    pub clusters: Vec<MilpClusterReport>,
}

/// One independent cluster of the MILP search. New type.
pub struct MilpClusterReport {
    /// Its queries, by workload index. // new
    pub queries: Vec<usize>,
    /// The memo groups of its own workload. // new
    pub groups: usize,
    /// Its lazy loop's rounds. // new
    pub rounds: u32,
    /// Its gap, at least 0. // new
    pub gap: Option<f64>,
    /// Its MILP lower bound at the end. // new
    pub lower_bound: Option<f64>,
    /// Its selected plan's cost per second. // new
    pub incumbent_cost: Option<f64>,
    /// Wall ms its solves took, together. // new
    pub solver_time_ms: u64,
    /// Whether it stopped at the time limit. // new
    pub time_limit_reached: bool,
    /// Solutions decoding could not compose into a plan. // new
    pub decode_failures: u32,
}

crates/physical-asap-optimization/src/lib.rs

/// The caller's materialization per candidate: (logical node it implements,
/// its reader queries, its options) -> the chosen option. // new
pub type ChooseMaterialization<'a> =
    dyn FnMut(*const OperatorNode, &[usize], &[Materialization]) -> Materialization + 'a;

/// Stage 2 for one logical DAG and one materialization assignment chosen
/// node by node; `None` when a choice is not an option or not down-closed.
/// Used by solution decoding to build one plan, not every assignment. // new
pub fn optimize_physical_asap_chosen(
    from_logical: &str,
    roots: &[QueryRoot],
    entries: &[QueryWorkloadEntry],
    data: &DataWorkload,
    deployment: &DeploymentCapabilities,
    choose: &mut ChooseMaterialization<'_>,
) -> Result<Option<PhysicalCandidate>, PhysicalASAPOptimizationError>;

crates/logical-asap-optimization/src/asap_aware_cse.rs

/// `roots` with the sharing candidates `subset` selects made one node each:
/// one identical-expression Pass 2 candidate, without listing every closed set. // new
pub fn with_shared_candidates(
    roots: &[QueryRoot],
    candidates: &[SharingCandidate],
    subset: &[bool],
) -> Vec<QueryRoot>;

/// The window-composition Pass 2 candidates in which every one of
/// `candidates` is in panes and reads a shared chain, without the other subsets'. // new
pub fn window_composition_shared_chain_candidates(
    roots: &[QueryRoot],
    candidates: &[PaneCandidate],
) -> Vec<WindowCompositionPass2Candidate>;

/// The rollup candidate of one coarse build, computed for it alone. // new
pub fn rollup_candidate_for(
    roots: &[QueryRoot],
    coarse: *const OperatorNode,
) -> Option<RollupCandidate>;

/// Whether adding `root` may change `coarse`'s rollup candidate; false lets
/// the memo reuse a cached one. // new
pub fn may_change_rollup_candidate(coarse: &Rc<OperatorNode>, root: &Rc<OperatorNode>) -> bool;

No CLI flag changes: --search memo-milp and --milp-time-limit-s are as on main.

Usage

# w840: the #840 workload (four queries over one metric `data`, two of them reading
# the same scan), raw data retained, default cost model, MILP search, 280 s limit.
stage_pipeline --workload w840.json \
  --deployment-capabilities '{"raw_data_retained":true,"query_time_results_retained":false}' \
  --search memo-milp --milp-time-limit-s 280 --out w840.out.json
# Stage 1 builds the memo; search::milp::model::search splits it into clusters,
# each ClusterLoop offers the warm start, then each round solves the model,
# decodes the solution into one plan (decode::decode -> optimize_physical_asap_chosen),
# evaluates it in Stage 3 and adds term cuts + the no-good cut.
# w840.out.json -> stage3_plan_selection.search (measured on 0f8bf09f):
#   incumbent_cost 0.1477616 (= --exhaustive), milp { rounds 15, gap 1.3e-06 (certified,
#   <= 1e-4), variable_count 1135, constraint_count 2008, decode_failures 0, clusters [...] },
#   plus the selected plan, its sizes and cost, and each candidate's accepted or
#   rejected line, as before.

Major algorithm changes

Per-cluster MILP

• Problem: one model over the whole memo, even when queries share nothing; W-Mixed's two halves (disjoint sources) were solved as one.
• Before / after: before: one model_input_of + one lazy loop. After: (1) clusters::split union-finds queries over shared groups, candidate reads, coupling sites and disk-charged raw sources; (2) one ClusterLoop per cluster, solves on threads; (3) clusters::combine merges plans and checks cost = sum.
• Correctness: under cost model 1 no cost term spans clusters, so the sum of per-cluster optima is the optimum; combine returns None (search as one) unless the combined cost equals the sum within 1e-9. Test: three_way_on_two_independent_clusters, clusters unit tests.
• Complexity: model size per cluster instead of whole; RangeHeavy and RollupHeavy remain one cluster each (one metric / one table).
• Assumptions and limits: cost models 2/3 do not run the MILP (Memo search solver: MILP over the memo for all cost models #888); calibrated measured_operation_costs put everything in one cluster.

Solution decoding and incumbent evaluation

• Problem: each round's exact evaluation enumerated every plan of the chosen group choice (offer_group_choice): q24h60 1,310,528 candidates; RangeHeavy 59 GB in identical-expression variants.
• Before / after: before: enumerate all compositions + Stage 2 assignments of a group choice. After: decode::decode builds one logical DAG from the chosen versions, shares identical nodes the solution shares (with_shared_candidates), materializes per the solution (optimize_physical_asap_chosen), plus the widened-scans variant; nodes no variable fixes get every Stage 2 option combination up to 16.
• Correctness: the cut excludes exactly the decoded choice set; when a set has more than 16 free combinations or fails to decode, its objective is kept as unevaluated_bound and the reported bound is min(bound, that), so no false certificate. Tests: decode and model unit tests (site threshold, shared-scan readers, built-once fidelity, unevaluated bound), 20-seed three-way.
• Complexity: one Stage 3 evaluation (a few variants) per round instead of up to millions; RangeHeavy peak ~1.1 GB instead of 59 GB.
• Assumptions and limits: widened scans and free materializations are a small local set, not model variables.

Warm start and time-limit bound

• Problem: a time-limited run with no evaluated round returned no plan (t11-5m, t10 on main).
• Before / after: after: ClusterLoop::offer_independent_plan offers every query's own legacy plan before round 1; a solve stopped at the time limit keeps HiGHS's proven bound lowered by its tolerance.
• Correctness: the warm start is only an incumbent; certification still compares bound with incumbent; the gap is None when no valid bound exists. Test: a_time_limited_memo_milp_run_returns_the_warm_start_plan.
• Complexity: RollupHeavy 1.3 s, RangeHeavy ~16 s extra before round 1.
• Assumptions and limits: a query the legacy search cannot plan leaves no warm start.

Accuracy rules as constraints

• Problem: the loop cut one whole group choice per round on AccuracyTargetMissed (RollupHeavy t10: every round failed on main).
• Before / after: after: guarantee_requirements and the rollup/pair rules require child choices that keep each choice's guarantee, read at the largest summary size Stage 3 can give (resized_to_the_largest); exact compositions and custom accuracy models are skipped; a guarantee lost only at the largest size rules nothing out.
• Correctness: a combination is ruled out only if it fails at the largest size, so no valid plan is lost. Tests: lower-bound tests, d990e77a's largest-size test.
• Complexity: t10 constraints 4966 -> 5115; ~4 s per solve instead of ~1.1 s.
• Assumptions and limits: pairwise only; three-level failures are still cut lazily. Per-size pair checks are in "Known open problems".

Tighter lower bound and term cuts

• Problem: weak bounds: w840_4 / i840 took ~180 rounds and did not certify at 280 s (gap 0.0175); j839 raw first bound 72 vs 113.95.
• Before / after: after: union-rate charging for built-once readers, latency per node key (ChoiceCoefficients::query_work), held-memory shares, ingestion prices, reach flows, strict cutoff; then per round term_cuts_of (SourceScans), held_cuts_of (HeldPerReader, HeldLongest, SharedBuildHeld), ingestion_copy_cuts_of: charge(term) >= bound * (sum of conditions - (n-1)).
• Correctness: each cut's bound is a lower bound of the term in every plan with those choices; every lower-bound fix comes with a test that the model cost is <= Stage 3 cost of a known optimal plan (lower_bound tests), plus term_cuts_certify_the_brute_force_price_on_839.
• Complexity: w840_4 / i840 rounds 176 -> 28, 280 s -> 44 s.
• Assumptions and limits: per-consumer multiplicity of scans/builds is not cut yet (t11-5m gap ~0.15).

Rollup and pane-chain rules per coarse build

• Problem: rollup_rules rebuilt every subset's rollup candidates (RollupHeavy coefficient derivation 662 s).
• Before / after: after: rollup_candidate_for per coarse build, cached per plan, skipping candidates may_change_rollup_candidate says cannot change it; the pane rule builds only the pair's shared-chain candidates (window_composition_shared_chain_candidates).
• Correctness: same rules (checked against the old function during development; test 72a873ae compares shared-chain roots).
• Complexity: RollupHeavy coefficient derivation 662 s -> 15 s.
• Assumptions and limits: none beyond the old rules'.

Before this PR

main (--search memo-milp --milp-time-limit-s 1800, shared host):

row	wall s	rounds	gap / certified	cost	reference
q24h60 (fargate, raw, enforce)	689	129	0 / yes	8.73884e-05	8.739e-05
w840	276	73	0 / yes	0.14776156	0.1477616
i840	1800	172	0.0425 / no	0.15078779	0.150788
rh8-q1h60 (fargate, raw, enforce)	772	97	0 / yes	3.832840e-06	3.832852e-06
t11-5m (RollupHeavy 4 SQL)	1800	142	no plan	—	1.703597e-06
t10 (RollupHeavy 6 SQL)	1801	503	no plan	—	none

Whole workloads (DRYRUN, 180 s / 30 min): 12 of 12 time out with no plan.

After this PR

Merged head 0f8bf09, --milp-time-limit-s 280, same host (shared, load ~15):

Fixed regression set (component rows):

row	reference	cost	rounds	gap	wall s	main (1800 s limit)
w840	0.1477616	0.1477616	15	1.3e-6, certified	14	276 s, certified
w840_3	0.1477616	0.1477616	15	1.3e-6, certified	14	–
w840_4	0.1507878	0.1507878	28	1.3e-6, certified	40	–
i840	0.150788	0.1507878	28	1.3e-6, certified	42	1800 s, gap 0.0425
j839 raw	113.9514	113.9514	5	2.1e-6, certified	2.6	–
j839 no raw	115.6695	115.6695	2	2.0e-6, certified	0.5	–
t11-5m (RollupHeavy 4 SQL)	1.703597e-06 (legacy)	1.703783e-06	54	0.145, time limit	280	no plan
q24h60 (fargate, raw, enforce)	8.739e-05	3.546e-04	452	0.849, time limit	281	689 s, certified
rh8-q1h60 (fargate, raw, enforce)	3.832852e-06	1.342e-05	162	0.814, time limit	280	772 s, certified

Whole workloads (raw retained, fargate, enforced; each returns a plan and its gap within 300 s, none certifies):

workload	clusters	cost	rounds	gap	wall s	RSS GB	main
RollupHeavy	2	4.628e-05	10	0.918	283	1.5	no plan
RollupHeavy 8 users	2	7.112e-05	2	0.994	285	2.2	no plan
RangeHeavy	1	4.708e-04 (warm start)	1	0.968	285	0.8	no plan (OOM at 59-94 GB)
RangeHeavy 8 users	1	5.674e-04 (warm start)	1	0.99997	287	1.5	no plan

decode_failures is 0 in every row. q24h60 and rh8-q1h60 are worse than main at a 280 s limit (main needed 689 s / 772 s to certify them); see Known open problems.

Known open problems

1. q24h60 ratio-validity gap (regression vs main at a 280 s limit). The model keeps offering choices where a ratio of two approximate children cannot meet its target, so Stage 3 rejects them round after round. Result: 3.546e-04, gap 0.849 at 280 s. The reference is 8.739e-05, which main certified in 689 s. This is a validity gap, not a lower-bound bug: the reported gap is honest. The fix is on the unmerged branch fix/memo-q24h-ingestion, an exact root pair rule that restores 8.738845e-05 certified in 113 s. A review found that it can judge scalar-root queries at the operator group; a WIP fix for that is on the same branch, not gated yet. The per-size form of the same rule is tracked in memo-milp misses rh8-q1h60 at 280 s (1.34e-05 vs 3.83e-06): suspected decode cut of a valid stricter site threshold #928.
2. rh8-q1h60 misses its reference: memo-milp misses rh8-q1h60 at 280 s (1.34e-05 vs 3.83e-06): suspected decode cut of a valid stricter site threshold #928. Result: 1.342e-05, gap 0.814 at 280 s; the reference is 3.832852e-06. The suspected cause is a decode exactness bug: decoding evaluates one site threshold per choice set, and its cut may remove the same choices at a stricter, valid threshold. If so, a run could certify wrongly. The diagnosis so far says the cut is per threshold version, and the time goes to the pair accuracy rule firing only when every size fails. A WIP fix (fix/memo-rh8) certifies 3.832840e-06 in 17.5 s. memo-milp misses rh8-q1h60 at 280 s (1.34e-05 vs 3.83e-06): suspected decode cut of a valid stricter site threshold #928 stays open until a test settles the decode question.
3. Whole workloads are far from certified: memo-milp: certify whole evaluation workloads within 300 s #927. Each returns a plan and its gap within 300 s, but the gaps are 0.918–0.99997, and RangeHeavy returns its warm start. Certifying them within 300 s is the follow-up the user decided on.
4. Sizing once (Option A): memo-milp: size summaries once (final config per reader set and summary count; Stage 3 never resizes) #929. The model prices a choice at a size that Stage 3 may change after selection. The model stays a lower bound, but it is loose. In progress on feat/memo-sizing-once.

The following are not in this PR: fix/memo-q24h-ingestion, fix/memo-rh8, perf/memo-warm-start-cheap, w2/held-once, feat/memo-sizing-once and feat/memo-validity.

Verification

On the merged head (main merged in, 0f8bf09), on the remote build host:

• cargo fmt --all -- --check: clean.
• cargo clippy --release -p asap-plan-selection -p asap-integration-tests -p asap-logical-asap-optimization -p asap-physical-asap-optimization --all-targets -- -D warnings: clean.
• cargo test --release -p asap-plan-selection --lib: 223 passed.
• cargo test --release -p asap-integration-tests --test joint_search_equivalence three_way_twenty_smallest -- --ignored: passed.
• cargo test --release -p asap-devtools: all pass, after updating two stale stage_pipeline expectations to PR-B's intended behavior (search_flag_picks_the_search_algorithm: a certified gap is within 1e-4, not 1e-9; milp_time_limit_is_reported_and_needs_the_milp_search: a run stopped before round 1 now returns the warm-start plan with its gap instead of an error).
• cargo test --release -p asap-integration-tests -p asap-logical-asap-optimization -p asap-physical-asap-optimization (non-ignored): 556 passed.
• Rows above (fixed regression set + whole workloads), outputs kept on the build host.
• Each item was reviewed (correctness/exactness) and gated (lower-bound tests, 20-seed three-way, regression rows) as it merged into the integration branch. The 60-seed three-way and the slow ignored equivalence tests were not rerun for this PR.