[zzylol]

Stacked on #1214; part of #1200, tracked in #1207

Merge unit. This PR merges into main only together with #1207's steps 1a–6 (#1207, Order): the #1200 stack, #1203 2a and 2b, #1201 and #1204 PR 1.

Representation / rules / enumeration

• (a) Representation. OperatorNode has no timing and no lifetime: a logical candidate has only its logical properties, so frontends, Stage 0, Stage 1 (Pass 1 and Pass 2) and accuracy propagation cannot see physical properties. The physical properties $p^{\mathrm{phys}} = (\tau, \ell)$ of a node live only in physical types:
  • PhysicalPropertiesAssignment (node → PhysicalProperties): Stage 2's choice in, and the timing pass's completed result out;
  • Stage 2's PhysicalCandidate.properties (unchanged);
  • PhysicalWorkloadPlan.physical_properties, PlanOutput.physical_properties and WholePlanSelection.physical_properties (the selected plan's nodes);
  • the export, PhysicalASAPDAGNode { output_state.timing, lifetime } (unchanged wire fields).
• (b) Rules. No rule changes. Stage 0 CSE and the Pass 2 rules (reuse.rs, scan_labels.rs, window_candidates.rs, state_groups.rs, shared_configuration_readers.rs) no longer compare or copy the two fields; they were None on every logical node, so every comparison gives the same answer as before.
• (c) Enumeration. No change: Stage 2 still offers the same (timing, lifetime) alternatives and Stage 3 picks among them. The timing pass places every node by the same rules; it now writes each node's properties into the completed assignment instead of into a rebuilt copy of the node.

Why

#1200's formulation gives physical properties $(\tau, \ell)$ only to physical candidates; a logical candidate has $p^{\log}$ only. Before this PR every OperatorNode carried timing: Option<ExecutionTiming> and lifetime: Lifetime, so logical code had to compare, copy and reject them: Stage 0 CSE hashed them, five Pass 2 rules compared or copied them, Pass 1 rejected an input that had them (AssignedPhysicalAnnotation), and Stage 2 rejected a memo node that had them (AssignedExecution, "assigned execution timing or lifetime"). Physical code also used the fields as a back door: a plan Stage 3 selected left its properties on the nodes ("pins"), and the timing pass honored a pre-set timing, so the same pair was stored on the node and in the assignment and could disagree (PhysicalPropertiesConflictWithFixedTiming, InvalidPhysicalProperties on a node).

What

• OperatorNode loses timing, lifetime, with_timing, with_lifetime and physical_properties().
• PhysicalPropertiesAssignment is the one place a node's physical properties live in IR form. An entry may have any valid PhysicalProperties: a materialized lifetime keeps the node's output (as before); lifetime None fixes only its timing (what a pin did). get is public.
• apply_physical_properties(root, &assignment) / apply_physical_properties_to_roots(roots, &assignment) return the completed assignment: every reachable node with its PhysicalProperties. Nodes are no longer rebuilt, so node identity is kept and NodeToTimedNodeCache goes away.
• Export takes the completed assignment: compile_physical_asap_dag(root, &timed), compile_physical_asap_workload(roots, &timed); a node it does not list is UntimedNode, as an untimed node was.
• Rewrites after the timing pass carry entries instead of copying fields: tumbling-window chain expansion returns TumblingWindowChainExpansion::physical_properties; with_confirmed_weights[_at] give a rebuilt node its original's entry; split_shared_by_timing already did (copy_assignment).
• The selected plan's properties travel with it: PhysicalWorkloadPlan.physical_properties (Stage 2 / realizer), PlanOutput.physical_properties (facade), WholePlanSelection.physical_properties (reference pricer); installable_plan_of takes them. PlanOutput::physical_asap_dag and installation re-apply the timing pass with them, which gives every node exactly the properties Stage 3 selected (what the pins did).
• The facade carries the plan to run as one value, PlanningRun::selected_plan: Option<SelectedPlan> with SelectedPlan { roots, physical_properties } (the roots with Stage 3's weight proofs and their assignment), and leaves the RealizedPlan as Stage 2 built it. In debug builds Stage1ToStage3Pass asserts that the assignment lists every node of the roots (PhysicalPropertiesAssignment::lists_every_node_of), so a rewrite that drops an entry cannot silently fall back to the timing pass's default.
• Stage 2's realizer checks the produced timing in plan.physical_properties instead of on the node.
• The executor's physical_planner::logical::restore no longer writes timing / lifetime into the IR it rebuilds (nothing read them).
• Logical FlatDag nodes (the devtools viewer documents' logical DAGs, stage0, logical candidates, Stage 2 pruned operator labels) lose their timing: null and lifetime: "None" keys. The DAG viewer (stages.js) no longer reads them from logical nodes, and its two hand-written memo fixtures drop them.
• Docs: pre-asap-ir.md, post-asap-ir.md, library-api.md describe the assignment as the home of physical properties.

Side map, not a wrapper node type

A node's properties are kept in a side map keyed by node, PhysicalPropertiesAssignment, not in a physical node type wrapping OperatorNode. A wrapper would either need its own child references (a second DAG type with its own map_children, flattening, export and expansion code) or still look its children's properties up in a map. The side map already existed as Stage 2's assignment, keeps Rc identity (no node is rebuilt to be timed), and encodes each node's pair once: the timing pass's output is the same type as its input, with every node listed. Re-applying a completed assignment is the identity, which replaces the pins.

Concepts not in the formulation

• Completed assignment: a PhysicalPropertiesAssignment that lists every node reachable from the roots (the timing pass's output). The formulation's $p^{\mathrm{phys}}_v$ per candidate, extended to every operator node of a plan.
• An entry with lifetime None fixes a node's timing: $(\tau, \mathrm{None})$ chosen for a node by a caller (Stage 3's selection re-applied for export) instead of derived by the timing pass from its consumer.

Formulation notation → code

Notation	Code	Status
$p^{\mathrm{phys}}_v = (\tau_v, \ell_v)$ of an operator node of a plan	PhysicalPropertiesAssignment::get(node) -> Option<PhysicalProperties>	was OperatorNode::{timing, lifetime} / OperatorNode::physical_properties()
$(\tau, \mathrm{None})$	PhysicalProperties::not_materialized_at(timing)	new here; deleted in #1216 for PhysicalProperties::new(τ, Lifetime::None)
$p^{\mathrm{phys}}$ of every node of the selected plan	PhysicalWorkloadPlan::physical_properties, PlanOutput::physical_properties, WholePlanSelection::physical_properties	new; were the pinned node fields

New abstractions not in #1207 or the doc: decisions (2026-11-05)

Per #1207's interface policy; the user decided each row on 2026-11-05. #1207's steps 1a–6 are one merge unit, so an interim abstraction may stay only when a later PR of the unit deletes it.

Abstraction	What it is for	Decision
PhysicalWorkloadPlan::physical_properties, PlanOutput::physical_properties, WholePlanSelection::physical_properties, TumblingWindowChainExpansion::physical_properties	carry the side map PhysicalPropertiesAssignment with the plan	Interim, deleted with the side map by #1201 + #1204 PR 1
SelectedPlan { roots, physical_properties }, SelectedPlan::of	the plan to run with its assignment, as one value	Interim, deleted by #1201 + #1204 PR 1
PhysicalPropertiesAssignment::get (now pub), impl PartialEq, copy_assignment	read an entry; WholePlanSelection's derive; carry an entry to a rebuilt node	Interim, deleted with the side map by #1201 + #1204 PR 1
PhysicalPropertiesAssignment::lists_every_node_of	the facade's debug check	Inlined in #1216 (deleted with its assertion)
PhysicalProperties::not_materialized_at(τ)	(τ, None)	Deleted in #1216: PhysicalProperties::new(τ, Lifetime::None)
integration-tests default_physical_properties, physical_properties_at_ingestion	test fixtures	Inlined in #1216

Signatures that now take the assignment (not new names): data_state(node, &assignment), with_confirmed_weights(.., &mut assignment), compile_physical_asap_dag[_with_node_ids](.., &assignment), compile_physical_asap_workload(.., &assignment).

Divergences from the doc's section 3: none in what this PR changes; the doc's "logical stages carry no physical properties" now holds for OperatorNode.

How

The timing pass (crates/types/src/ir/properties/timing.rs) keeps its placement rules. placement now returns a PhysicalProperties: a node the assignment lists takes its entry (after the same filtered-build and upstream-at-query-time checks the materialized entries had; a SummaryEstimate / EvaluatePopulation entry at ingestion time is UpstreamNotAtIngestion), any other node is (QueryTime, None) for a summary build, SummaryEstimate or EvaluatePopulation, and (consumer, None) otherwise. forces_query_time reads entries the same way. write records each node in the output assignment (a node reached at two timings is still ConflictingTiming) and validate reads each child's timing from it. The edge rules are unchanged.

Every former pin is an assignment entry: Stage 3's selection reaches export as PlanOutput.physical_properties; tests that pinned a node at ingestion time (placed_at_ingestion, with_timing(Some(..))) now set PhysicalProperties::NOT_MATERIALIZED_AT_INGESTION for it. Tests that only checked the pins' conflicts or invalid pins are deleted (see Deleted): those states cannot be represented any more.

Key code interfaces added / modified

crates/types/src/ir/operator/node.rs

/// A node of the logical DAG; stays. It carries no physical properties.
pub struct OperatorNode {
    /// The operation.
    pub operator: Operator,                  // unchanged
    /// Output category, derived from the operator.
    pub result_kind: OperatorResultKind,     // unchanged
    /// Output schema.
    pub schema: Schema,                      // unchanged
    /// Accuracy guarantee, once assessed; None never means exact.
    pub guarantee: Option<AccuracyGuarantee>, // unchanged
    pub timing: Option<ExecutionTiming>,     // removed
    pub lifetime: Lifetime,                  // removed
}
impl OperatorNode {
    pub fn with_timing(self, timing: Option<ExecutionTiming>) -> Self;  // removed
    pub fn with_lifetime(self, lifetime: Lifetime) -> Self;             // removed
    pub fn physical_properties(&self) -> Option<Result<PhysicalProperties, InvalidPhysicalProperties>>; // removed
    pub fn validate_execution_timing(self: &Rc<Self>) -> Result<(), SchemaDerivationError>; // removed (test-support)
}

crates/types/src/ir/properties/timing.rs

/// Physical properties of operator nodes, by node identity; stays. Stage 2's
/// choice in, the timing pass's completed result out. Keeps every listed node alive.
pub struct PhysicalPropertiesAssignment { /* node → (Rc<OperatorNode>, PhysicalProperties) */ }
impl PhysicalPropertiesAssignment {
    /// Give `node` `properties`: a materialized lifetime keeps its output,
    /// lifetime None fixes only its timing.
    pub fn set(&mut self, node: &Rc<OperatorNode>, properties: PhysicalProperties); // was: materialized only
    /// `node`'s properties, if listed. Physical stages read a node's timing and lifetime here.
    pub fn get(&self, node: &OperatorNode) -> Option<PhysicalProperties>;           // was private
    /// Give `copy` the entry of `original`, for a rebuilt node.
    pub fn copy_assignment(&mut self, original: &OperatorNode, copy: &Rc<OperatorNode>); // unchanged
    /// Test support: every summary build materialized at ingestion where allowed.
    pub fn materialized_at_ingestion_where_allowed(roots: &[Rc<OperatorNode>]) -> Self;   // unchanged
}
/// Equal when they list the same nodes with the same properties.
impl PartialEq for PhysicalPropertiesAssignment;                                      // new

/// The completed assignment of the DAG below `root`, edges validated.
pub fn apply_physical_properties(
    root: &Rc<OperatorNode>,
    assignment: &PhysicalPropertiesAssignment,
) -> Result<PhysicalPropertiesAssignment, ExecutionDataStateError>; // was (root, assignment, &mut NodeToTimedNodeCache) -> Rc<OperatorNode>
/// The completed assignment of a workload; a node several roots share is timed once.
pub fn apply_physical_properties_to_roots(
    roots: &[QueryRoot],
    assignment: &PhysicalPropertiesAssignment,
) -> Result<PhysicalPropertiesAssignment, ExecutionDataStateError>; // was -> Vec<QueryRoot> (timed copies)
/// The data state of `node`'s output, if `timed` lists it.
pub fn data_state(node: &OperatorNode, timed: &PhysicalPropertiesAssignment) -> Option<ExecutionDataState>; // was data_state(node)

pub struct NodeToTimedNodeCache;                                    // removed (the completed assignment replaces it)

crates/types/src/ir/properties/execution_data_state.rs

/// Why the timing pass rejects an assignment; stays.
pub enum ExecutionDataStateError {
    /// A population maintained or evaluated against its contract.
    InvalidMaintainedPopulation,                                                     // unchanged
    /// A query-time value under an ingestion-time summary.
    QueryTimeValueUnderIngestionSummary { edge: &'static str, child: ExecutionDataState }, // unchanged
    /// A query-time build over unstored ingestion-time input.
    UnstoredIngestionInputUnderQueryTimeSummary { edge: &'static str, operator: &'static str }, // unchanged
    /// Any other edge whose input data state the consumer does not accept.
    IllegalChildDataState { edge: &'static str, child: ExecutionDataState },          // unchanged
    /// Opaque summary state composed into another summary.
    UnsupportedStateComposition { family: String },                                  // unchanged
    /// One node reached at two timings.
    ConflictingTiming { first: ExecutionDataState, second: ExecutionDataState },      // unchanged
    /// A root of rows at ingestion time that is not stored there (e.g. an entry fixing it there).
    UnstoredIngestionRoot { operator: &'static str },                                 // unchanged (doc)
    /// A filtered summary build placed at ingestion time.
    FilteredSummaryAtIngestion { operator: &'static str },                            // unchanged
    /// An ingestion-time BinaryOp the update path cannot evaluate.
    InvalidIngestionTimeBinary,                                                       // unchanged
    /// A checked division not at query time or with two guards.
    InvalidCheckedDivision,                                                           // unchanged
    /// An exact operator over a non-plain column.
    NonPlainOperand { column: String, dtype: String },                               // unchanged
    /// A reserved ASAP operator with no timing contract.
    UnimplementedOperator { operator: &'static str },                                 // unchanged
    /// An ingestion-time entry over something that runs only at query time.
    UpstreamNotAtIngestion { operator: &'static str },                                // unchanged
    /// Export reached a node the assignment does not list (bad input: not timed first).
    UntimedNode { operator: &'static str },                                           // unchanged (doc)
    /// The exported DAG failed its own validation.
    InvalidPhysicalDAG(PhysicalASAPDAGValidationError),                               // unchanged
    PhysicalPropertiesConflictWithFixedTiming { operator: &'static str, fixed: ExecutionTiming, assigned: PhysicalProperties }, // removed: one entry per node
    InvalidPhysicalProperties { operator: &'static str, reason: InvalidPhysicalProperties },                                   // removed: an entry is a valid PhysicalProperties
}

crates/types/src/ir/physical_export.rs

/// Export the DAG below `root`, each node with its properties from `timed` (test support).
pub fn compile_physical_asap_dag(root: &Rc<OperatorNode>, timed: &PhysicalPropertiesAssignment)
    -> Result<PhysicalASAPDAG, ExecutionDataStateError>;            // was (root)
/// Export a batch, each node with its properties from `timed`; a node it does not list is UntimedNode.
pub fn compile_physical_asap_workload(roots: &[QueryRoot], timed: &PhysicalPropertiesAssignment)
    -> Result<PhysicalASAPDAG, ExecutionDataStateError>;            // was (roots)
/// The same, with each exported node's IR node.
pub fn compile_physical_asap_workload_with_node_ids(roots: &[QueryRoot], timed: &PhysicalPropertiesAssignment)
    -> Result<PhysicalASAPDAGCompilation, ExecutionDataStateError>; // was (roots)

crates/types/src/ir/flat.rs

/// An OperatorNode whose children are node ids; stays.
pub struct FlatNode {
    /// The operation over child ids.
    pub operator: Operator<NodeId>,               // unchanged
    pub(crate) result_kind: OperatorResultKind,   // unchanged
    pub(crate) schema: Schema,                    // unchanged
    pub(crate) guarantee: Option<AccuracyGuarantee>, // unchanged
    pub(crate) timing: Option<ExecutionTiming>,   // removed
    pub(crate) lifetime: Lifetime,                // removed
}

crates/types/src/ir/operator/tumbling_window_chain.rs

/// A workload with its chains expanded; stays.
pub struct TumblingWindowChainExpansion {
    /// The roots with every chain expanded.
    pub roots: Vec<QueryRoot>,                                   // unchanged
    /// Each expanded chain with its tumbling windows.
    pub chains: Vec<ExpandedTumblingWindowChain>,                // unchanged
    /// Each reached node with the node replacing it.
    pub expanded_nodes: Vec<(Rc<OperatorNode>, Rc<OperatorNode>)>, // unchanged
    /// The expanded workload's physical properties: a replacing node takes its original's,
    /// each tumbling window its chain's, the range and shift below it (chain timing, None).
    pub physical_properties: PhysicalPropertiesAssignment,       // new; was written into the nodes
}
/// Expand every chain of `roots`, whose properties `timed` lists (empty before timing).
pub fn expand_tumbling_window_chains_in_workload(roots: &[QueryRoot], timed: &PhysicalPropertiesAssignment)
    -> Result<TumblingWindowChainExpansion, SchemaDerivationError>; // was (roots)

crates/types/src/physical/properties.rs, crates/types/src/physical/workload_plan.rs, crates/types/src/workload/bounds.rs

impl PhysicalProperties {
    /// (timing, None): not materialized, at `timing`. Used by the timing pass and the chain expansion.
    pub fn not_materialized_at(timing: ExecutionTiming) -> Self;   // new
}

/// One Stage 2 physical plan; stays.
pub struct PhysicalWorkloadPlan {
    /// `P<n>` or `P<n>-m<k>`.
    pub id: String,                                   // unchanged
    /// The logical candidate it implements.
    pub from_logical: String,                         // unchanged
    /// The query roots `dag` exports (no longer timed copies).
    pub roots: Vec<QueryRoot>,                        // unchanged type
    /// The completed assignment `dag` exports `roots` with.
    pub physical_properties: PhysicalPropertiesAssignment, // new
    /// The exported plan.
    pub dag: PhysicalASAPDAG,                         // unchanged
}

/// `roots` with weight requirements confirmed; a rebuilt node takes its original's entry.
pub fn with_confirmed_weights(roots: &[QueryRoot], data: &DataWorkload,
    physical_properties: &mut PhysicalPropertiesAssignment) -> Vec<QueryRoot>;   // gains physical_properties
pub fn with_confirmed_weights_at(roots: &[QueryRoot], data: &DataWorkload,
    confirm_at: &dyn Fn(&Rc<OperatorNode>) -> bool,
    physical_properties: &mut PhysicalPropertiesAssignment) -> Vec<QueryRoot>;   // gains physical_properties

crates/planner/src/pass.rs, crates/planner/src/installable.rs, crates/plan-selection/src/report.rs

/// One workload's selected plans; stays.
pub struct PlanOutput {
    /// One plan per operator query.
    pub plans: Vec<QueryPlan>,                                    // unchanged
    /// Standalone scalar queries by batch position.
    pub scalar_roots: Vec<(usize, ScalarExpr)>,                   // unchanged
    /// One plan per operator a scalar reads.
    pub scalar_plans: Vec<QueryPlan>,                             // unchanged
    /// Stage 3's selection, when the pass prices candidates.
    pub selection: Option<Selection>,                             // unchanged
    /// Kept chains' tumbling-window retention, by node id.
    pub tumbling_window_retention_ms: BTreeMap<PhysicalASAPNodeId, u64>, // unchanged
    /// The physical properties the pass selected, by node; empty = timing-pass default.
    pub physical_properties: PhysicalPropertiesAssignment,        // new; were pins on the nodes
}

/// The installable plan of `roots`, each node with `physical_properties`' entry.
pub fn installable_plan_of(
    roots: &[QueryRoot],
    physical_properties: &PhysicalPropertiesAssignment,           // new
    selection: Option<&Selection>,
    tumbling_window_retention_ms: &BTreeMap<PhysicalASAPNodeId, u64>,
    entries: &[QueryWorkloadEntry],
) -> Result<(InstallablePlan, Option<SelectedPlanReport>), InstallablePlanError>;

/// The reference pricer's selection (feature `reference-pricer`); stays.
pub struct WholePlanSelection {
    /// The selected candidate's id.
    pub selected: String,                                         // unchanged
    /// The plan to run.
    pub plan: Vec<QueryRoot>,                                     // unchanged
    /// The physical properties of `plan`'s nodes.
    pub physical_properties: PhysicalPropertiesAssignment,        // new
    /// Each valid candidate's cost.
    pub costs: BTreeMap<String, PlanCost>,                        // unchanged
    /// Per valid candidate, the accuracy accepted per query.
    pub accepted: BTreeMap<String, Vec<AcceptedAccuracy>>,        // unchanged
    /// The rejected candidates.
    pub rejected: Vec<Rejection>,                                 // unchanged
    /// The selected plan's tumbling-window retention by node id.
    pub tumbling_window_retention_ms: BTreeMap<PhysicalASAPNodeId, u64>, // unchanged
    /// Candidates checked.
    pub candidates_checked: u64,                                  // unchanged
}

Logical-stage errors removed: PerComputationCandidateError::AssignedPhysicalAnnotation (Pass 1), ConfigurationError::AssignedExecution and the inventory's "assigned execution timing or lifetime" rejection (Stage 2): a logical node cannot carry physical properties any more.

Usage (the facade, unchanged for a caller; what changed is where Stage 3's choice travels):

// The user's workload: one PromQL query evaluated every 10 s over arriving data.
let input = UserInput::new(&workload, FrontendInput::Promql, PlanningModels::builtin());
// Stages 0–3 run; Stage 2's realizer builds the selected plan and returns its
// completed assignment in `RealizedPlan.plan.physical_properties`, which
// `Stage1ToStage3Pass` hands on as `output.physical_properties`.
let output: PlanOutput = e2e_plan(input).await?;
// The selected nodes carry no timing; Stage 3's (timing, lifetime) per node is here,
// e.g. a tumbling-window chain at (IngestionTime, AcrossEvaluations(Memory)).
let chain_properties = output.physical_properties.get(&chain_node);
// Export re-applies the timing pass with that assignment: every node keeps what
// Stage 3 selected, the same document a deployment runtime installs.
let dag: PhysicalASAPDAG = output.physical_asap_dag()?;

crates/integration-tests/tests/runtime_handoff.rs checks this flow: the facade's DAG equals compile_physical_asap_workload(&output.roots(), &output.physical_properties).

Major algorithm changes

None. The timing pass's placement and edge rules are unchanged; it records each node's properties in the completed assignment instead of a rebuilt node, so it no longer allocates a copy of the DAG.

Deleted

• OperatorNode::{timing, lifetime}, OperatorNode::{with_timing, with_lifetime, physical_properties}, OperatorNode::validate_execution_timing (test-support) and its test execution_timing_checks_edges_not_function_names.
• FlatNode::{timing, lifetime} (the timing / lifetime keys of logical DAG documents).
• NodeToTimedNodeCache, NodeToTimedNodeCache::timed; asap_types::test_support::timed; the logical crate's test_support::{timed, at_ingestion} (the integration tests' fixtures are now default_physical_properties and physical_properties_at_ingestion in integration-tests/src/lib.rs).
• ExecutionDataStateError::PhysicalPropertiesConflictWithFixedTiming, ExecutionDataStateError::InvalidPhysicalProperties; timing.rs's timed_properties and PhysicalPropertiesAssignment::resolve; the child.timing.is_some() check under EvaluatePopulation (always true after timing).
• PerComputationCandidateError::AssignedPhysicalAnnotation and its test cases; ConfigurationError::AssignedExecution; the inventory's assigned-timing rejection; Stage 2's "a node fixed at ingestion time cannot be built at query time" filter on shared results.
• The timing/lifetime comparisons and copies in Stage 0 cse.rs (hash, equality, interning), pass2/memo/rules/reuse.rs, rules/scan_labels.rs, window_candidates.rs, state_groups.rs, rules/shared_configuration_readers.rs (including its skip of an "already assigned" build), catalog/factoring.rs, split_by_timing.rs, accuracy propagation.rs (bind_guarantees, with_resized_summaries), plan-selection accuracy_positions.rs, workload/bounds.rs, schema_support.rs, the executor's logical::restore; tumbling_window_chain.rs's timed_like.
• Tests of removed behavior: timing.rs fixed_timing_keeps_an_agreeing_assignment_and_rejects_a_conflict (its first half kept as an_entry_gives_a_node_its_timing_and_lifetime), ingestion_fixed_node_rejects_mat_at_query, an_invalid_timed_node_reports_its_error, a_summary_build_over_an_invalid_input_reports_its_error, a_non_asap_node_over_an_invalid_input_reports_its_error and helpers placed_at_ingestion, invalid_at_ingestion; split_by_timing a_fixed_timing_conflict_survives_the_split (and its helpers finalize, exact_sum); summary_coverage timing_does_not_block_merges; runtime_install's untime; weighted_topk_binding's unpinned.
• Renamed (test support): PhysicalASAPDAG::untimed_query_roots → query_roots.

Before this PR

Every operator node, logical or physical, carried timing and lifetime. Logical stages carried None / Lifetime::None through every rewrite and compared them in CSE and the sharing rules; Pass 1 and Stage 2 rejected inputs that had them. The timing pass rebuilt the whole DAG to write the pair into each node, and honored a pair already on a node, so Stage 3's selection reached export as pins on the nodes and a pin could conflict with an assignment entry.

After this PR

A logical node has no physical properties; only physical types hold them, each node's pair encoded once in a PhysicalPropertiesAssignment. The timing pass keeps the input nodes and returns the completed assignment; export, installation, cost and Stage 2's realizer read it. Stage 3's selection reaches export through PlanOutput.physical_properties.

Wire format: unchanged. INSTALLABLE_PLAN_WIRE_VERSION stays 12; PhysicalASAPDAGNode keeps output_state.timing and lifetime. Only the developer-tool viewer documents change: logical DAG nodes no longer have timing / lifetime keys (always null / "None" before); the viewer and its fixtures change with them.

Plans: unchanged. UPDATE_SNAPSHOTS=1 cargo test --release -p asap-devtools --test corpus_snapshots -- --include-ignored leaves git status without snapshot changes.

Verification

Speed mode: tests covering the changed code, plus fmt and clippy.

• Review round (head 5ef723d9c, rebased on refactor(types, stages): replace Materialization with PhysicalProperties (timing, lifetime) #1214 64e721439): cargo fmt --all -- --check and cargo clippy --workspace --all-targets -- -D warnings clean; cargo test -p asap-types (253), -p asap-cost (45), -p asap-accuracy (77), -p asap-logical-asap-optimization (203), -p asap-frontend-promql (228), -p asap-frontend-sql (176), -p asap-physical-asap-optimization (28), -p asap-executor (296), -p asap-planner (89, 8 wall-clock ignored), -p asap-plan-selection --lib (142), -p asap-devtools (108), -p asap-integration-tests (364): 0 failed. The first run's one failure, range_evaluation::stage3_plan_selection_prices_61_evaluations_per_dashboard_request, came from refactor(types, stages): replace Materialization with PhysicalProperties (timing, lifetime) #1214's detail-text rename and is fixed there (rerun after the rebase: ok). The review fixes also undo two bad renames from the timed → physical_properties sweep (a rebuilt node in accuracy propagation.rs, a boolean in runtime_install.rs).

Earlier round (at 6500daf19):

• cargo fmt --all -- --check: clean. cargo clippy --workspace --all-targets -- -D warnings: clean.
• cargo test -p asap-types (143 + 101 + 8 passed), -p asap-cost (36 + 1 + 8 passed), -p asap-accuracy (77 passed), -p asap-logical-asap-optimization --lib (201 passed), -p asap-physical-asap-optimization (28 passed), -p asap-plan-selection --lib (142 passed), -p asap-executor (all targets passed; 2 ignored), -p asap-planner (11 + 78 passed; 8 ignored), -p asap-devtools --lib (18 passed), -p asap-devtools --test devtools (82 passed; 1 ignored), -p asap-integration-tests (364 passed; 26 ignored). Ignored tests are the wall-clock / time-budget ones.
• DAG viewer: node --test stages.test.mjs memo.test.mjs enumeration.test.mjs (8 + 13 + 12 passed) with the updated fixtures.
• Snapshots: UPDATE_SNAPSHOTS=1 cargo test --release -p asap-devtools --test corpus_snapshots -- --include-ignored (9 passed); git status shows no snapshot change, so every corpus and sketch-plan snapshot selects the same plan.
• Behavior kept: the timing pass's tests now pin nodes through PhysicalPropertiesAssignment entries (fixed_at_ingestion, apply_fixing) and expect the same errors and timings; runtime_handoff checks that the facade's DAG equals the selected plan compiled with PlanOutput.physical_properties.

🤖 Generated with Claude Code

https://claude.ai/code/session_01K1M6br9PL4SDeVS1XpezKJ

[zzylol]

Closed: replaced by #1207's new order (interfaces first, then one PR per stage), starting with PR 0 #1226. The code on this branch is reused in PRs 1–8; the branch is kept.