[zzylol]

Stacked on #1234 (refactor/1207-pass2, PR 4). Part of the #1207 merge unit: do not merge on its own.

Why

#1207's Order, row "5. Stage 2": expand each logical candidate Pass 2 kept into its physical alternatives $p^{\mathrm{phys}} = (\tau, \ell)$, each logical edge into physical edges with $\alpha^{\log}$ and $\alpha^{\mathrm{phys}}$, and record every rejected alternative with its PruneReason. Concepts doc §2 "Decision tree" (levels 3–4), "Physical requirements", and §3 "Stage 2: physical alternatives" (data arrival, prune reasons). Decisions applied: F1–F3 and D1–D13 of the PR 5 review (2026-11-07), G2-counter and G4/H3 (no executor-support check anywhere), L8a (Count-Min prune via column_sign), time indexing (any grid-evaluated group may be AcrossEvaluations), and "WeightDomain deleted in PR 5" (2026-11-09).

What

generate_physical_asap_memo_candidates(logical, data, capabilities) (was todo!()):

Step	Code	What it decides
1. Offered values	catalog/configurations.rs enumerate_candidate_configurations	τ = QueryTime always; IngestionTime only for a candidate whose group is time-independent (GroupProperties::depends_on_evaluation_time false) or an input of one (ingestion_time_offered_per_group), and for a counter state (rate/increase/irate) only when its rows are ordered (max_out_of_order_ms declared, or PromQL rows); ℓ = None, OneEvaluation(Memory), and AcrossEvaluations(Memory) only for a chain or a group with a known evaluation_grid_of_group. Values not offered are not recorded
2. Prune rules (levels 3–4)	same	CountMinWeightsNotProvenNonNegative (a Cms/CmsWithHeap/HydraCms build whose state column's column_sign is not NonNegative), NoIngestionPipeline (IT, arrival AtRest/Unknown), RawDataNotRetained (QT Scan), QueryTimeResultsNotRetained (QT AcrossEvaluations)
3. Physical edges	catalog/mod.rs	Pass 2's edges in (consumer, input) chunks; every pair of the two candidates' alternatives, $\alpha^{\log}$ copied, $\alpha^{\mathrm{phys}}$ = producer_satisfies_physical_requirements(producer, consumer_timing)
4. Unusable alternatives	catalog/pruning.rs	InputHasNoBuildableProducer (least fixed point of "every input has a buildable admissible producer"), ReachesNoQueryOutput (not reached from a query output, an alternative of a root group a query-time consumer may read)
5. Output	PhysicalASAPMemoCandidates::new	emitted candidates renumbered; edges between them grouped by (consumer, input); admissible-edge indexes built once (admissible_edges_into, admissible_edges_out_of)

Candidates Pass 2 eliminated (LogicalASAPMemoCandidates::pruned) are not expanded. Stage 2 checks no executor support and costs nothing; alignment with each execution context stays Stage 3's (PhysicalCandidate::delivers_on, implemented in PR 1).

Deleted (replaced, no shims)

Deleted	Replaced by
SummaryUpdate::weight_domain, WeightDomain (UnknownOrSigned, NonNegative, RawDataSamplesOfMetric, CounterDerivativeOfMetric), NonNegativeWeightProof (5 variants)	the state group's column_sign (derived from the weights' sign and the declared source signs) read by Stage 2's CountMinWeightsNotProvenNonNegative (decision 2026-11-09)
Pass 1: the CmsWithHeap gate on unproven weights in realize_keyed_additive_summary_input, raw_data_samples_of_metric, with_raw_data_samples_requirement, SingleMeasure::raw_data_samples_of	Stage 2's Count-Min prune (Pass 1 offers the build; Stage 2 records why it is pruned)
Pass 1's private reads_only_time_series(memo, input)	asap_types::stages::reads_only_time_series(memo, group), which calls the one generic copy in ir::schema::topk (D12)
Executor: the CmsWithHeap "nonnegative weight contract" check (physical_planner), the rate-weight refusal in create_planner_accumulator (keyed off ResetAwareCounterDerivative), the Hydra "weight's sign is unknown" binding check; tests weighted_topk_binding::rate_updates_cannot_enter_integer_heap_factory, hydra_cms_execution::unknown_sign_weight_is_rejected	the planner proves the sign before offering a Count-Min build; the Hydra build still rejects a negative weight value at run time (invalid_weight_values_fail_the_build)
Test fixtures copied per crate: Pass 2's try_candidate, candidate, group, series_schema, scan_operator; Pass 1's scan_operator	asap_types::test_support::{add_candidate, group_with_own_candidate, series_schema, time_series_scan} (D11)
thiserror dependency of asap-stage2-physical-optimization	unused (Stage 2 cannot fail)

Not reintroduced from the old PR 5 head (decisions after it): executor_can_run / PerEntityAggregateNotExecutable (G4/H3: no executor-support check), AcrossEvaluationsWithoutWindowLayout (F1/C7: a value-set rule, not a prune reason), PhysicalCandidateKey, PrunedCandidate.group, PhysicalASAPMemoCandidates::query_roots, the private Deployment struct (D1–D4, F2, F3; the PR 0 types already have these shapes).

Pending approval (new pub items not in the concepts doc)

N1. PhysicalEdge::is_admissible (review D6, approved; not in the doc's interface listing):

impl PhysicalEdge {
    /// Whether a plan may use the edge: α^log = α^phys = 1.
    pub fn is_admissible(&self) -> bool;
}

Example: in query A below, the edge from the scan's (QT, None) alternative into the chain's (IT, AcrossEvaluations) has satisfies_logical_requirement: true, satisfies_physical_requirement: false (an ingestion-time build cannot read a query-time scan), so is_admissible() is false and neither Stage 2's pruning nor admissible_edges_into lists it.

N2. asap_types::stages::reads_only_time_series (review D12, approved; not in the doc):

/// Whether every scan below `group` reads a time series (PromQL rows).
pub fn reads_only_time_series(memo: &Memo, group: MemoGroupId) -> bool;

Example: for rate(c_total[5m])'s chain over Scan(TimeSeries c_total) it is true, so the counter chain is offered ingestion time with max_out_of_order_ms: None; over Scan(Table events) it is false, and the counter chain is offered ingestion time only with max_out_of_order_ms: Some(_). Pass 1 uses it for the PromQL top-k rule.

N3. Shared memo fixtures under the test-support feature (review D11, approved; test-only):

// crates/types/src/test_support.rs
/// [ts, value, labels…], time index 0, series identity realized (closed, `$promql_series_identity` last).
pub fn series_schema(labels: &[&str]) -> Schema;
/// A `TimeSeries(metric)` scan with `schema`, filtered by `predicates`.
pub fn time_series_scan<C>(metric: &str, schema: Schema, predicates: Vec<Predicate<C>>) -> Operator<C>;
/// Adds `operator` with `parameters` to `group` as Pass 1 does (no requirement, no fused group).
pub fn add_candidate(memo: &mut Memo, group: MemoGroupId, operator: Operator<MemoGroupId>,
    parameters: Option<SketchParams>) -> Result<CandidateForMemoGroupId, CandidateError>;
/// The group of `operator`, with `operator` as its own candidate.
pub fn group_with_own_candidate(memo: &mut Memo, operator: Operator<MemoGroupId>) -> MemoGroupId;

Example: group_with_own_candidate(&mut memo, time_series_scan("latency", series_schema(&["job"]), vec![])) is query A's scan group; Pass 2's and Stage 2's tests build their memos with these.

Design choices taken (conservative; please confirm)

C1. QueryTimeResultsNotRetained prunes only (QueryTime, AcrossEvaluations). The concepts doc's prune-reason table (and PR 0's comment on the variant) says OneEvaluation too; the doc's decision-tree diagram, DeploymentCapabilities::query_time_results_retained ("materialized across evaluations"; default false) and the old PR 5 say AcrossEvaluations only. Taken: AcrossEvaluations only, because with the default capabilities pruning OneEvaluation would remove every within-evaluation sharing (decision C3: a None producer has at most one consumer per context). Example: query A with default capabilities keeps the readout's (QT, OneEvaluation) and prunes its (QT, AcrossEvaluations). If approved, the doc's table row reads "QueryTime with ℓ = AcrossEvaluations".

C2. Ingestion time is offered through inputs transitively. The doc says "a candidate that is time-independent or an input of one". Taken: the closure: an input of a candidate of an offered group is offered too (an ingestion-time candidate reads only ingestion-time producers, so a one-level rule could not build anything below an @ T). Example: sum_over_time(m[1m] @ 1000): TimeShift @1000 is time-independent; its input TimeRange 1m and that one's Scan are offered (IT, None) (test an_at_t_subtree_is_offered_ingestion_time). Without @ T the closure adds nothing: inputs of time-independent groups are time-independent.

C3. Count-Min weight sign read from the state group. The doc says "whose weight column's column_sign". Taken: the sign the build's own group records for its state column (Stage 0 derives it from the weight: a constant's sign, or the weight column's column_sign in the input group), so no second derivation of the weight's sign. Equal for every candidate of the group, since they produce the same state. Example: CmsWithHeap keyed by series, weighted by SampleValue of m: pruned in all three alternatives (QT None, QT One, IT None) without a declaration, offered with column_signs: {"value": "non_negative"} for m (test a_count_min_build_needs_weights_proven_non_negative). It is checked first: it holds for every (τ, ℓ).

C4. A query reads its output as a query-time consumer (review G3, doc line pending). A query output is an alternative of a root group that producer_satisfies_physical_requirements(_, QueryTime) admits: any query-time alternative and (IngestionTime, AcrossEvaluations); (IngestionTime, None) is not, since its result is gone when the query runs. Example: sum_over_time(m[1m] @ 1000)'s root TimeShift keeps (QT, None) and (QT, OneEvaluation); its (IT, None) is ReachesNoQueryOutput.

C5. The executor no longer refuses rate-weighted updates in create_planner_accumulator. That refusal read NonNegativeWeightProof::ResetAwareCounterDerivative; with the proof gone, the factory cannot tell a rate weight from a sum weight. Listed for PR 6 (typed binding of window-weighted heaps).

How

• types: PhysicalASAPMemoCandidates::new builds the admissible-edge indexes (HashMap<(consumer, input), Vec<PhysicalEdgeId>>, Vec<Vec<PhysicalEdgeId>> by producer) through PhysicalEdge::is_admissible; the two accessors read them (were todo!()).
• Stage 2: one pass over the memo's groups and candidates (skipping Pass 2's eliminated ones) emits each candidate's configurations; one pass over Pass 2's edge chunks emits physical edges and the admissible producers per input; pruning; renumbering; a stable sort of the kept edges by (consumer, input).

Key code interfaces added / modified

crates/stage2-physical-optimization/src/catalog/mod.rs

/// Stage 2: expand every logical candidate Pass 2 kept into its offered (τ, ℓ), every logical
/// edge into physical edges; record each rejected alternative with its reason. Cannot fail.
pub fn generate_physical_asap_memo_candidates(
    logical: &LogicalASAPMemoCandidates,
    data: &DataWorkload,
    capabilities: &DeploymentCapabilities,
) -> PhysicalASAPMemoCandidates;                                   // unchanged signature, was todo!()

crates/types/src/stages/physical/memo_candidates.rs, stages/memo.rs, stages/pruning.rs

impl PhysicalASAPMemoCandidates {
    /// Stage 2's output with the admissible-edge indexes built once.
    pub fn new(logical_candidates: LogicalASAPMemoCandidates, physical_candidates: Vec<PhysicalCandidate>,
        edges: Vec<PhysicalEdge>, pruned_candidates: Vec<PrunedCandidate<PhysicalCandidate>>) -> Self; // was todo!()
    /// P^adm_{c,i}: the admissible edges into input i of `consumer`; empty when none.
    pub fn admissible_edges_into(&self, consumer: PhysicalCandidateId, input_index_of_consumer: usize)
        -> &[PhysicalEdgeId];                                      // was todo!()
    /// The admissible edges out of `producer`.
    pub fn admissible_edges_out_of(&self, producer: PhysicalCandidateId) -> &[PhysicalEdgeId]; // was todo!()
}
impl PhysicalEdge { pub fn is_admissible(&self) -> bool; }        // new (N1)
pub fn reads_only_time_series(memo: &Memo, group: MemoGroupId) -> bool; // new (N2)
/// Why a stage removed a candidate. Stays; variants unchanged (doc of one variant narrowed, C1).
pub enum PruneReason {
    /// Stage 2: IT, and the data has no ingestion pipeline (at rest or unknown arrival). Input fact.
    NoIngestionPipeline,                     // unchanged, now produced
    /// Stage 2: QT `Scan`, and the deployment does not retain raw data. Input fact.
    RawDataNotRetained,                      // unchanged, now produced
    /// Stage 2: QT AcrossEvaluations, and the deployment retains no query-time result across evaluations.
    QueryTimeResultsNotRetained,             // unchanged, now produced (doc: AcrossEvaluations only, C1)
    /// Stage 2: an input has no buildable admissible producer.
    InputHasNoBuildableProducer,             // unchanged, now produced
    /// Stage 2: no query output reads it over admissible edges.
    ReachesNoQueryOutput,                    // unchanged, now produced
    /// Stage 2: a Count-Min build whose weights are not proven non-negative.
    CountMinWeightsNotProvenNonNegative,     // unchanged, now produced
    /// Stage 3: dominated by the named candidate.
    DominatedBy(PhysicalCandidateId),        // unchanged
    /// Stage 3: `asap-cost` could not cost it.
    UsageNotCostable,                        // unchanged
}

crates/types/src/ir/schema/sketch.rs

/// What one input row adds to a summary state. Stays.
pub struct SummaryUpdate {
    /// The key the row is counted under, for keyed summary types.
    pub item: Option<SummaryInputExpr>,      // unchanged
    /// The value added to the state.
    pub weight: SummaryInputExpr,            // unchanged
    pub weight_domain: WeightDomain,         // removed (the state group's column_sign)
}
pub enum WeightDomain { /* … */ }            // removed, with all 4 variants
pub enum NonNegativeWeightProof { /* … */ }  // removed, with all 5 variants

Usage (end to end; test offers_each_candidate_its_timings_and_lifetimes at 616f765, query A of the concepts doc):

// The user's query: quantile_over_time(0.95, latency[1m]), every 60 s at phase 0, over
// ingesting data (rows_per_second 1000, max_out_of_order_ms None); the deployment retains raw
// data and query-time results.
// Stage 0 + Pass 1 + Pass 2 give (built by the test as they build it): G0 Scan latency;
// G1 TimeRange 1m (G0); G2 KLL state with [0] SummaryAgg(G1) and [1] SummaryMerge(G4);
// G3 SummaryTumblingWindowChain[KLL, w = 60 s](G0); G4 TimeRange 1m (G3); G5 p95 readout (root).
// evaluation_grid_of_group: 60 s @ 0 for G1, G2, G4, G5; None for G0 and G3 (time-independent).
let physical = generate_physical_asap_memo_candidates(&logical, &data, &capabilities);
// The planner facade then hands `physical` to Stage 3 (PR 6). What it holds:
// G1, G2[0], G2[1], G4, G5: (QT, None), (QT, OneEvaluation), (QT, AcrossEvaluations) — time-dependent,
//   so no ingestion time; on the 60 s grid, so AcrossEvaluations is offered.
// G0 Scan: (QT, None), (QT, OneEvaluation), (IT, None) — no grid, so no AcrossEvaluations.
// G3 chain: (QT, None), (QT, OneEvaluation), (QT, AcrossEvaluations), (IT, AcrossEvaluations);
//   pruned_candidates: G3 (IT, None) ReachesNoQueryOutput (only an ingestion-time consumer could read it).
// edges: e.g. into G3's four alternatives from G0's three: 12 edges (test `edges_carry_both_requirements`),
//   admissible exactly when both run at the same timing (a QT chain reads the QT scan, the IT chain the IT scan).

Major algorithm changes

Stage 2 expansion and pruning (new; the previous head's version was written against the old memo and is replaced).

• Problem: at PR 4's head every planning run stopped at Stage 2's todo!().
• After: (1) per group, IT offered by the time-independence closure; (2) per kept logical candidate, TIMINGS × LIFETIMES filtered by the offered values, each with the first matching prune rule; (3) per Pass 2 edge chunk (c, i), the product of c's and the producers' alternatives; (4) buildable = least fixed point, reached = DFS from query outputs over admissible edges into buildable producers; (5) renumber, sort edges, index.
• Correctness: an emitted candidate has, for each input, an admissible edge from an emitted producer, and is reachable from a query output (checked by every test's generate helper, which also checks the edge order and the indexes); a pruned alternative is recorded exactly once, with the first reason that applies.
• Complexity: O(|C|·5) alternatives, O(Σ_{(c,i)} |alt(c)|·Σ_u |alt(u)|) ≤ 25·|logical edges| physical edges; the buildable fixed point is O(depth·|E|), reach O(|E|).
• Assumptions and limits: Memory storage only (the capabilities offer no disk, Executor: store Disk-medium materialized views and one-evaluation results on disk #1217); Stage 2 checks no executor support (G4/H3: executor gaps are PR 6's).

Before / After

Before (PR 4 head): Pass 2 produced LogicalASAPMemoCandidates; Stage 2 was todo!(), so every planning run stopped there. Count-Min weight signs were a WeightDomain proof on each update, set by Pass 1 and checked by the executor.

After: Stage 2 produces PhysicalASAPMemoCandidates with physical edges, indexes and the pruned list; planning stops at Stage 3 (select_plan's todo!() of PR 6). A Count-Min build's weight sign is the state group's column_sign, and Stage 2 records an unproven one as CountMinWeightsNotProvenNonNegative.

Representation, rules, enumeration

• Representation: PhysicalCandidate { logical_candidate_id, physical_properties }; PhysicalEdge { …, satisfies_logical_requirement, satisfies_physical_requirement }; PrunedCandidate<PhysicalCandidate> { candidate, reason }.
• Rules: offered values (time-independence closure, counter order, chain or known grid); prune rules (Count-Min sign, ingestion pipeline, raw data and query-time result retention); buildability and reach over admissible edges.
• Enumeration: every offered (τ, ℓ) per kept logical candidate; every producer/consumer alternative pair per logical edge. Stage 3 chooses.

Self-review

• No new abstraction: no Stage 2 input struct (F2), no new prune reason, no executor-support table. Two admissible-producer indexes stay (before pruning by provisional index, after pruning by ID; D6 allows both, both call is_admissible).
• Duplication removed: one reads_only_time_series, one set of memo test fixtures, one weight-sign derivation (the state column's column_sign; WeightDomain gone).
• Dead code: none added; thiserror dropped from Stage 2.
• Docs: planner-layering.md Stage 2 rows and pruning paragraphs (D13.1), asap-primitive-schema.md and glossary.md without WeightDomain; the executor's filtered-ingestion comment no longer claims the planner never offers one. The concepts doc is not changed (C1 needs a decision first; D13.2's crate path is already stage2-physical-optimization there).
• Terms: consumer/query, materialized, cost; no executor-support wording in planner code.

Verification

• cargo fmt --all, cargo clippy --workspace --all-features --all-targets -- -D warnings: clean; crate_graph passes.
• Stage 2: 11 tests (offered values per candidate, no ingestion pipeline, raw data not retained, query-time results not retained, @ T closure, counter order (SQL vs PromQL, bound declared or not), Count-Min sign, edge requirements, an input without an admissible producer, Pass 2's eliminated candidates, a scalar root's operand).

Test status at 616f765 (cargo test -p <crate> --all-features, per crate)

Crate	Passed	Failed	Ignored	Failures stop at
asap-types	204	0	0	—
asap-accuracy	24	0	0	—
asap-cost	31	1	0	#1207 PR 6: check θ, which the exported node carries: 1
asap-frontend-common	49	0	0	—
asap-sql-function-catalog	10	0	0	—
metricsql_common	3	0	0	—
asap-frontend-promql	218	0	0	—
asap-frontend-metricsql	15	0	0	—
asap-frontend-sql	175	0	0	—
asap-executor	209	62	1	#1207 PR 6: check θ, which the exported node carries: 32; #1207 PR 6: the Hydra grid's sizes are θ, which the exported node carries: 12; #1207 PR 6: each materialized node's retention_ms is its h_v, which Stage 3 sets: 8; #1207 PR 6: the weighted-frequency θ, which the exported node carries: 7; #1207 PR 6: check the weighted-frequency θ, which the exported node carries: 2; #1207 PR 6: the baseline MILP over x, y, z, t, and the PhysicalASAPDAG of its solution: 1
asap-planner	8	18	1	#1207 PR 6: the baseline MILP over x, y, z, t, and the PhysicalASAPDAG of its solution: 18
asap-devtools	37	56	3	#1207 PR 6: the baseline MILP over x, y, z, t, and the PhysicalASAPDAG of its solution: 41; #1207 PR 9: a group's node from its expression: 9; #1207 PR 9 (#1205): each pruned PhysicalCandidate with its reason: 2; OTHER: assertion failed: output.status.success(): 2; #1207 PR 9: label a group by its expression: 2
asap-integration-tests	100	50	4	#1207 PR 6: the baseline MILP over x, y, z, t, and the PhysicalASAPDAG of its solution: 43; #1207 PR 6: each materialized node's retention_ms is its h_v, which Stage 3 sets: 3; #1207 PR 6: realize the sum-and-count plan of avg from Pass 1's memo: 3; OTHER: called Result::unwrap() on an Err value: Error("invalid type: sequence, expected string or map", line: 0, column: 0): 1
asap-stage3-plan-selection	3	0	0	—
asap-stage0-memo	15	0	0	—
asap-stage2-physical-optimization	11	0	0	—
asap-stage1-pass2-reuse-candidates	21	0	0	—
asap-stage1-pass1-per-computation-candidates	32	0	0	—
Total	1165	187	9

PR 4 (bcae9e5): 1155 / 188 / 9. By test name: no test that passes on PR 4 fails here. 11 new Stage 2 tests pass. Two executor tests are deleted with the checks they tested (weighted_topk_binding::rate_updates_cannot_enter_integer_heap_factory, passing; hydra_cms_execution::unknown_sign_weight_is_rejected, already failing at PR 6's θ todo!()). Every planning failure that stopped at PR 5's "physical expansion and physical edges" (planner 18, devtools 41, integration 43, executor 1) now stops at PR 6's select_plan ("the baseline MILP over x, y, z, t, …"); 2 devtools tests now stop at PR 9's pruned-candidate printout.