[zzylol]

Merge unit. PR 5 of #1207's merge unit (PRs 0–9, with 0b), stacked on PR 4 (#1234, refactor/1207-pass2). The PRs merge into main together, after PR 9. This PR fills in Stage 2, which PR 0 left as todo!(). Planning now gets through Stage 2 and stops at Stage 3's first todo!() (PR 6); see "Test status".

Why

#1207's Order, row "5. Stage 2": physical expansion (decision-tree levels 4–5), physical edges, pruned_candidates (concepts doc §2 "Physical requirements", "Invalid combinations", "Decision tree"; §3 "Stage 2: physical alternatives"; #1200, #1201). Stage 3 (PR 6) allocates $y_{u,c,i}$ on the admissible physical edges this PR builds.

What

• Expansion, levels 4 and 5 (stage2-physical-optimization/src/catalog/configurations.rs). Each logical candidate gets one PhysicalCandidate per valid $p^{\mathrm{phys}} = (\tau, \ell)$: the product of ExecutionTiming {QueryTime, IngestionTime} and Lifetime {None, OneEvaluation(Memory), AcrossEvaluations(Memory)}. PhysicalProperties::new rejects (IngestionTime, OneEvaluation), so it is neither offered nor pruned. Stage 2 creates no logical candidates.

• Prune rules (each reads only the value it removes and the values above it), each alternative recorded in pruned_candidates with its PruneReason:

  Rule	Level	PruneReason
  the executor has no kernel for an exact per-entity Aggregate (every (τ, ℓ))	1 (checked in Stage 2: Pass 1/2 record no pruned list)	PerEntityAggregateNotExecutable
  τ = IngestionTime and the data has no ingestion pipeline (DataArrival::AtRest or Unknown)	4	NoIngestionPipeline (new)
  τ = QueryTime on a Scan and raw data is not retained	4	RawDataNotRetained
  ℓ = AcrossEvaluations without a window layout ω (doc: "not offered yet: it needs its own delivery rule")	5	AcrossEvaluationsWithoutWindowLayout (new)
  (QueryTime, AcrossEvaluations) and the deployment keeps no query-time results	5	QueryTimeResultsNotRetained (new)
  an input with no buildable producer over admissible edges	after expansion	InputHasNoBuildableProducer
  no query output reads it over admissible edges	after expansion	ReachesNoQueryOutput

  A materialized result is in memory only: the executor does not store Disk results yet (Executor: store Disk-medium materialized views and one-evaluation results on disk #1217). This is a stated rule, not a pruned alternative.

• Physical edges. Each logical edge $(u, c, i)$ becomes one PhysicalEdge per pair of emitted alternatives of $u$ and $c$, grouped by (consumer, input). $\alpha^{\log}$ is copied from the logical edge. $\alpha^{\mathrm{phys}}$ is producer_satisfies_physical_requirements (timing and lifetime only). Every edge is listed, admissible or not ($\mathcal E$ of §1). PhysicalASAPMemoCandidates::new indexes the admissible ones ($P^{\mathrm{adm}}_{c,i}$): admissible_edges_into(c, i), admissible_edges_out_of(u). Alignment with an execution context is not part of $\alpha^{\mathrm{phys}}$: Stage 3 checks it with PhysicalCandidate::delivers_on (PR 1).

• Stage 2 cannot fail. It returns PhysicalASAPMemoCandidates, not a Result, and no longer takes the workload entries (alignment is Stage 3's).

Deleted

Item	Why / replacement
ErrorGeneratingPhysicalASAPMemoCandidates (Pruning(String))	Stage 2 has no failure: every alternative it cannot use is a pruned candidate.
asap_planner::SelectionError::PhysicalASAPCandidates(String)	Only produced from the error above.
entries: &[QueryWorkloadEntry] argument of generate_physical_asap_memo_candidates	Unused: per-query alignment is Stage 3's delivers_on.
PruneReason::TooFewQueryReaders	A consumer-count policy (MinimumQueryReaders), deleted with constraint (J) on 2026-11-06; the cost model decides.
thiserror dependency of asap-stage2-physical-optimization	Unused.
todo!("#1207 PR 5 …") in PhysicalASAPMemoCandidates::{new, admissible_edges_into, admissible_edges_out_of} and generate_physical_asap_memo_candidates	Implemented.

Old Stage 2 rules not ported (deleted in PR 0; see spec questions): per-query ingestion eligibility (NotRepeated, NotPredictable, WindowNotFixed, WindowOffEpochGrid, RangeEvaluatedWholeWindow, TumblingWindowsOff{Phase,Interval} are now delivers_on per context), Filtered, CounterNeedsTimeOrder, readouts forced to query time, and OneEvaluation only for summaries or nodes reached by two or more queries.

New abstractions not in #1207 or the doc (pending approval)

Item	Where	Why
PruneReason::{NoIngestionPipeline, QueryTimeResultsNotRetained, AcrossEvaluationsWithoutWindowLayout}	types/src/stages/pruning.rs	The doc's level-4 and level-5 prune rules and its "not offered yet" rule need a reason in pruned_candidates; the doc defines no PruneReason list.
generate_physical_asap_memo_candidates(logical, data, query_time_results_retained, raw_data_retained) -> PhysicalASAPMemoCandidates (PR 0: also entries, returned Result<_, ErrorGenerating…>)	stage2-physical-optimization/src/catalog/mod.rs	The doc names no Stage 2 entry signature. The unused argument and the error that never happens are deleted.
PhysicalASAPMemoCandidates::new sets pruned_candidates empty; Stage 2 assigns the pub field	types/src/stages/physical/memo_candidates.rs	new (PR 0) has no pruned-list argument.
Private: Deployment { has_ingestion_pipeline, query_time_results_retained, raw_data_retained }, enumerate_candidate_configurations (named in the doc), executor_can_run, pruned, logical_edges_into_input, prune_reasons, read_by_query_output	stage2-physical-optimization	The implementation.
Test fixtures in catalog/tests.rs (example, example_with_root, candidate, group, generate, stage2, offered, pruned, id_of)	tests	Build Pass 2's output through Memo methods.

Spec questions (conservative choice taken)

1. Which operators run at ingestion time. The doc's level 4 prunes IngestionTime only when there is no ingestion pipeline, so every operator (readouts included) gets ingestion-time alternatives. Those that cannot feed a query output are pruned as ReachesNoQueryOutput (an unmaterialized ingestion-time result is read only at ingestion time, and only windowed results are materialized across evaluations). Main forced readouts and population operators to query time.
2. Counters and filters at ingestion time. Main refused an ingestion-time counter state (rate/increase) over data not known to arrive in time order (Mixed) and an ingestion-time filtered build. Neither is a doc rule, so neither is ported. Should they become level-4 prune rules (two more PruneReasons)?
3. Query output at query time. ReachesNoQueryOutput treats each query as a query-time consumer that is not materialized: a query may read its root group's alternatives with τ = QueryTime or (IngestionTime, AcrossEvaluations). The doc does not state how a query reads its output.
4. Level 1 in Stage 2. The doc puts executor support at level 1 (Pass 1/2). Pass 1/2 record no pruned list, so the per-entity-aggregate rule stays in Stage 2 as on main.
5. OneEvaluation for every candidate. Offered for every query-time candidate, including those one query reaches; main offered it only to summaries or to nodes two or more queries reach. The cost model chooses (no constraint (J)).

How

generate_physical_asap_memo_candidates:

1. For each logical candidate (group order, then index), enumerate_candidate_configurations (the doc's name) tries every (τ, ℓ) in TIMINGS × LIFETIMES order through PhysicalProperties::new. It offers the valid ones and records each pruned one with its reason.
2. For each offered alternative $c$ and input $i$, each logical edge into $(c_{\log}, i)$ and each offered alternative $u$ of its producer gives one PhysicalEdge with both $\alpha$s. Edges use provisional IDs.
3. prune_reasons: buildable is the least fixed point of "every input has an admissible edge from a buildable producer" (a cycle without a buildable way out is not). Reached is the closure from each query root group's buildable alternatives that a query-time consumer may read, over admissible edges into buildable producers.
4. The usable alternatives are numbered densely (PhysicalCandidateId); edges with a pruned end are dropped and the rest renumbered. PhysicalASAPMemoCandidates::new builds the admissible-edge indexes.

Key code interfaces added / modified

types/src/stages/physical/memo_candidates.rs

/// Stage 2's output, which Stage 3 selects from. The type stays (PR 0).
impl PhysicalASAPMemoCandidates {
    /// Builds the output and its admissible-edge indexes; `pruned_candidates` starts empty.
    pub fn new(
        logical_candidates: LogicalASAPMemoCandidates,
        physical_candidates: Vec<PhysicalCandidate>,
        query_roots: Vec<QueryRoot<MemoGroupId>>,
        edges: Vec<PhysicalEdge>,
    ) -> Self; // was todo!()
    /// P^adm_{c,i}: the admissible edges into input `input_index_of_consumer` of `consumer`;
    /// Stage 3 allocates y on them.
    pub fn admissible_edges_into(&self, consumer: PhysicalCandidateId, input_index_of_consumer: usize)
        -> &[PhysicalEdgeId]; // was todo!()
    /// The admissible edges out of `producer`: every consumer that may share it.
    pub fn admissible_edges_out_of(&self, producer: PhysicalCandidateId) -> &[PhysicalEdgeId]; // was todo!()
}

types/src/stages/pruning.rs

/// Why a stage removed a candidate. The type stays.
pub enum PruneReason {
    /// Stage 2: the executor has no kernel for this exact per-entity Aggregate.
    PerEntityAggregateNotExecutable, // unchanged
    /// Stage 2, level 4: ingestion time, and the data is at rest or of unknown arrival.
    NoIngestionPipeline, // new
    /// Stage 2: a query-time scan of raw data the deployment does not retain.
    RawDataNotRetained, // unchanged
    /// Stage 2, level 5: query time, materialized across evaluations, and the deployment
    /// keeps no query-time results.
    QueryTimeResultsNotRetained, // new
    /// Stage 2, level 5: materialized across evaluations without a window layout (not offered yet).
    AcrossEvaluationsWithoutWindowLayout, // new
    /// Stage 2: an input has no buildable admissible producer.
    InputHasNoBuildableProducer, // unchanged
    /// Stage 2: no query output reads it over admissible edges.
    ReachesNoQueryOutput, // unchanged
    /// Stage 2: shared within one evaluation with fewer than two reading queries.
    TooFewQueryReaders, // removed (constraint (J) deleted)
    /// Stage 3: dominated by the named candidate.
    DominatedBy(PhysicalCandidateId), // unchanged
    /// Stage 3: asap-cost could not cost it.
    UsageNotCostable, // unchanged
}

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

/// Stage 2's entry, called by the planner facade and `planning_stages` after Pass 2.
pub fn generate_physical_asap_memo_candidates(
    logical: &LogicalASAPMemoCandidates,
    data: &DataWorkload,                 // its arrival decides the ingestion pipeline
    query_time_results_retained: bool,   // DeploymentCapabilities
    raw_data_retained: bool,             // DeploymentCapabilities
) -> PhysicalASAPMemoCandidates; // was (logical, entries, data, …) -> Result<_, ErrorGeneratingPhysicalASAPMemoCandidates>

/// Removed.
pub enum ErrorGeneratingPhysicalASAPMemoCandidates { Pruning(String) } // removed

planner/src/run.rs

/// Why the facade selected no plan. The type stays.
pub enum SelectionError {
    /// Bad input: capabilities contradict the data workload.
    InvalidDeploymentCapabilities(DeploymentCapabilitiesError), // unchanged
    /// Pass 1 or Pass 2 failed.
    LogicalASAPCandidates(String), // unchanged
    /// Stage 2 failed.
    PhysicalASAPCandidates(String), // removed
    /// Stage 3 selected no plan.
    Selection(ErrorSelectingPlan), // unchanged
    /// The selected plan could not be built.
    Realization(PhysicalCandidateError), // unchanged
    /// `--run-until` stopped before plan selection.
    NotRun, // unchanged
}

Usage (the running example's query A, from catalog/tests.rs; end-to-end planning stops at Stage 3, PR 6):

// What Pass 2 hands Stage 2 for `quantile_over_time(0.95, latency[…])`:
// G0 = Scan latency; G1 = KLL states with an unwindowed build and a 10 s tumbling build;
// G2 = SummaryEstimate[q=0.95](G1), the query's root. Logical edges G0→G1 (both builds)
// and G1→G2 (both builds), all α^log = 1.
let example = example(true);
// Stage 2 as the facade calls it: data ingesting at 1000 rows/s, the deployment keeps
// query-time results and raw data.
let physical = generate_physical_asap_memo_candidates(&example.logical, &ingesting(), true, true);
// Offered: Scan {QT/None, QT/OneEvaluation, IT/None}; unwindowed build {QT/None, QT/OneEvaluation};
// 10 s build {QT/None, QT/OneEvaluation, QT/AcrossEvaluations, IT/AcrossEvaluations};
// readout {QT/None, QT/OneEvaluation}. Pruned: every AcrossEvaluations without ω
// (AcrossEvaluationsWithoutWindowLayout), IT/None of both builds and of the readout
// (ReachesNoQueryOutput).
// 30 physical edges; the IT/None scan has one admissible consumer, the 10 s IT/AcrossEvaluations build:
let into_ingestion_build = physical.admissible_edges_into(id_of(&physical, example.windowed_state, IT_ACROSS), 0);
// Stage 3 (PR 6) checks alignment per context: the 10 s materialized build delivers to a
// query every 10 s at phase 0, not at phase 5 s.

Major algorithm changes

Stage 2 expansion and pruning. Problem: main expanded operator nodes into materialization modes, filtered by per-query eligibility, with consumer-count constraints and timing-only producer sets per input, and no edge objects. PR 0 deleted all of it. Before (main): modes per node kind; per-query eligibility; interned producer sets per input; pruning with consumer-count constraints. After: the product of timing × lifetime per logical candidate less the prune rules above; explicit edges with both $\alpha$; buildable/reached fixed points over admissible edges. Correctness: the invalid combination cannot be built (PhysicalProperties::new); pruning removes only alternatives no feasible selection can use: a candidate without a buildable producer for some input violates (A) when selected, and one no query output reaches violates (F). Tests below. Complexity: alternatives ≤ 5 per logical candidate; edges ≤ 25 per logical edge before pruning; both fixed points are linear in edges per round (rounds ≤ DAG depth). Limits: spec questions 1–5.

Before / After

• Before: planning stopped at Stage 2's todo!().
• After: Stage 2 returns $\mathcal C$, $\mathcal E$ with $\alpha^{\log}$ and $\alpha^{\mathrm{phys}}$, the admissible-edge indexes, and every pruned alternative with its reason. Planning stops at Stage 3 (PR 6).

Representation, rules, enumeration

• (a) Representation: unchanged types (PR 0). PhysicalASAPMemoCandidates is now filled.
• (b) Rules: decision-tree levels 4–5 with their prune rules; buildable/reached pruning.
• (c) Enumeration: none. Admissible edges are the domain of Stage 3's $y_{u,c,i}$.

Notation	Code
$\tau_v$, $\ell_v$, $p^{\mathrm{phys}}_v$	PhysicalCandidate::physical_properties (PhysicalProperties { timing, lifetime })
$\mathcal C$	PhysicalASAPMemoCandidates::physical_candidates
$(u, c, i)$, $\mathcal E$	PhysicalEdge, PhysicalASAPMemoCandidates::edges()
$\alpha^{\log}_{u,c,i}$	PhysicalEdge::satisfies_logical_requirement (from LogicalEdge)
$r^{\mathrm{phys}}{c,i}$, $\alpha^{\mathrm{phys}}{u,c,i}$	producer_satisfies_physical_requirements(producer, consumer) → PhysicalEdge::satisfies_physical_requirement
$P^{\mathrm{adm}}_{c,i}$	admissible_edges_into(c, i)
alignment with κ	PhysicalCandidate::delivers_on(memo, context) (Stage 3)
decision-tree prunes	PrunedCandidate<PhysicalCandidateKey> with PruneReason

Concepts not in the formulation: the pruned list and its reasons (diagnostics, #1205).

Verification

• cargo test -p asap-stage2-physical-optimization: 12 passed. Every test checks that each emitted candidate has an admissible edge into each input and that edges are grouped by (consumer, input) over emitted candidates (generate). Coverage: expansion per level over the running example (every_logical_candidate_expands_into_its_valid_timings_and_lifetimes); the invalid combination and Disk never appear (the_invalid_combination_and_disk_are_never_alternatives); level 4 (ingestion_time_needs_an_ingestion_pipeline, without_raw_data_only_ingestion_time_builds_read_the_scan, which also covers InputHasNoBuildableProducer); level 5 (query_time_across_evaluations_needs_retained_query_time_results); executor support (a_per_entity_aggregate_without_a_kernel_is_pruned_whole); edges with both $\alpha$ and the admissible indexes (physical_edges_carry_both_compatibilities, an_unsatisfied_logical_requirement_is_not_admissible); alignment on emitted candidates (alignment_is_checked_per_context_with_delivers_on); pruning (a_consumer_without_a_producer_for_one_input_cannot_be_built, a_scalar_root_reaches_the_groups_it_reads, a_root_group_read_at_ingestion_time_keeps_its_ingestion_time_alternative).
• cargo fmt --check; cargo clippy -p asap-types -p asap-stage2-physical-optimization -p asap-planner -p asap-devtools --all-targets --all-features -- -D warnings: clean. crate_graph: 6 passed.
• An independent review (fresh agent, diff only) found no correctness bugs. Fixed from it: the doc's function name enumerate_candidate_configurations, stale "operator node" text in planner-layering.md, "kept" wording, one builder for pruned candidates, and three tests (scalar root, an input group without candidates, a root group read at ingestion time). Not taken: visiting groups in topological order for the buildable fixed point (rounds are bounded by DAG depth), HashMaps replaced by nested Vecs, a debug_assert! on logical-edge input indexes.

Test status

cargo test -p <crate> --all-features --no-fail-fast, per crate, passed / failed / ignored. PR 4 is #1234 at 72a4037; PR 5 is this PR (all crates at 36f4208; Stage 2 re-run at d11f12e, which changes only Stage 2 and docs).

Crate	PR 4	PR 5	Where failures stop now
asap-stage2-physical-optimization	0 / 0 / 0	12 / 0 / 0	—
asap-planner	9 / 19 / 1	9 / 19 / 1	19 at PR 6 (select_plan; were PR 5)
asap-devtools	40 / 54 / 3	40 / 54 / 3	42 at PR 6 (40, plus 2 asap_planner subprocesses; were PR 5); 12 at PR 9 (10 labels as before, 2 now at stage2_pruned_candidates_json, #1205)
asap-integration-tests	101 / 69 / 4	101 / 69 / 4	68 at PR 6 (49 at select_plan, were PR 5; 19 as before); 1 as before (materialized_views)
asap-executor	210 / 63 / 1	210 / 63 / 1	all at PR 6 (the one at PR 5 now stops at select_plan)
asap-cost	31 / 1 / 0	31 / 1 / 0	unchanged (PR 6)
asap-types	197 / 0 / 0	197 / 0 / 0	—
Pass 2, Pass 1, Stage 0, accuracy, frontends (promql/sql/common/metricsql), sql-function-catalog, metricsql_common, Stage 3	20, 33, 13, 54, 218/175/49/15, 10, 3, 3	same	—
Total	1181 / 206 / 9	1193 / 206 / 9	No failure stops at PR 5; planning stops at Stage 3 (PR 6)