refactor(types, stages): physical properties off the logical IR (OperatorNode has no timing or lifetime)

crates/accuracy/src/propagation.rs

@@ -122,7 +122,7 @@ pub fn root_guarantee(root: &QueryRoot) -> Option<AccuracyGuarantee> {
 
 /// Bind known statistics in existing contracts, then recover compositions
 /// that previously required constant child bounds. Physical state contracts,
-/// schemas, timing, lifetime, and sharing are preserved. This does not
+/// schemas and sharing are preserved. This does not
 /// infer facts: `bind` must leave unsupported statistics unknown.
 pub fn bind_guarantees(
     root: &QueryRoot,
@@ -137,9 +137,7 @@ pub fn bind_guarantees(
             return done.clone();
         }
         let operator = node.operator.map_children(|child| walk(child, bind, cache));
-        let mut rebuilt = OperatorNode::with_schema(operator, node.schema.clone())
-            .with_timing(node.timing)
-            .with_lifetime(node.lifetime);
+        let mut rebuilt = OperatorNode::with_schema(operator, node.schema.clone());
         rebuilt.guarantee = node
             .guarantee
             .as_ref()
@@ -162,8 +160,8 @@ pub fn bind_guarantees(
 /// `roots` with every sketch summary state rebuilt with the parameters
 /// `resize` gives it (same algorithm; `None` keeps them), and every node
 /// above one rebuilt over its new inputs with its guarantee composed again
-/// ([`composed_guarantee`]; a node that had none keeps none). Timing,
-/// materialization and sharing are kept. Pass 1 sizes summaries
+/// ([`composed_guarantee`]; a node that had none keeps none). Sharing is
+/// kept. Pass 1 sizes summaries
 /// for a failure probability this way.
 pub fn with_resized_summaries(
     roots: &[QueryRoot],
@@ -212,11 +210,7 @@ pub fn with_resized_summaries(
                 .guarantee
                 .as_ref()
                 .and_then(|_| composed_guarantee(&rebuilt));
-            Rc::new(
-                rebuilt
-                    .with_timing(node.timing)
-                    .with_lifetime(node.lifetime),
-            )
+            Rc::new(rebuilt)
         } else {
             node.clone()
         };
@@ -653,10 +647,10 @@ fn merged_tumbling_windows_guarantee(node: &OperatorNode) -> Option<Option<Accur
         return Some(None);
     };
     if let Operator::ASAP(ASAPOp::SummaryAgg { child: range, .. }) = &mut window_build.operator {
-        let mut timed = (**range).clone();
-        timed.guarantee = composed_guarantee(&timed);
-        debug_assert!(Rc::ptr_eq(timed.children()[0], input));
-        *range = Rc::new(timed);
+        let mut composed = (**range).clone();
+        composed.guarantee = composed_guarantee(&composed);
+        debug_assert!(Rc::ptr_eq(composed.children()[0], input));
+        *range = Rc::new(composed);
     }
     let window_build = Rc::new(window_build);
     let reader = OperatorNode::with_schema(

crates/cost/tests/ingestion_refresh_runs.rs

@@ -96,8 +96,8 @@ fn dag_of(
             )
         })
         .collect();
-    let timed = apply_physical_properties_to_roots(&roots, &assignment).unwrap();
-    compile_physical_asap_workload(&timed).unwrap()
+    let physical_properties = apply_physical_properties_to_roots(&roots, &assignment).unwrap();
+    compile_physical_asap_workload(&roots, &physical_properties).unwrap()
 }
 
 /// `(multiplier, per_second)` of `dag`'s `Scan`, each query every 10 s.

crates/devtools/src/bin/asap_planner.rs

@@ -394,7 +394,7 @@ fn document(queries: Vec<Value>, entries: &[QueryWorkloadEntry], run: &PlanningS
 fn stage3_json(run: &PlanningStagesRun, entries: &[QueryWorkloadEntry]) -> Value {
     let mut stage3 = selection_json(&run.run.selection);
     match (&run.run.selection, &run.run.selected_plan) {
-        (Ok(selection), Some(roots)) => {
+        (Ok(selection), Some(plan)) => {
             let tumbling_window_retention_ms = run
                 .run
                 .selected_physical_plan
@@ -406,8 +406,13 @@ fn stage3_json(run: &PlanningStagesRun, entries: &[QueryWorkloadEntry]) -> Value
                 .map(|(node, ms)| (node.to_string(), json!(ms)))
                 .collect::<serde_json::Map<_, _>>()
                 .into();
-            match materialized_views_json(roots, selection, &tumbling_window_retention_ms, entries)
-            {
+            match materialized_views_json(
+                &plan.roots,
+                &plan.physical_properties,
+                selection,
+                &tumbling_window_retention_ms,
+                entries,
+            ) {
                 Ok(views) => stage3["materialized_views"] = Value::Array(views),
                 Err(error) => {
                     stage3["materialized_views"] = json!([]);
@@ -646,6 +651,7 @@ fn arrival_together(earlier: DataArrival, later: DataArrival) -> Option<DataArri
 /// tumbling-window chain's tumbling windows, newest first) and their number (`installed_count`).
 fn materialized_views_json(
     roots: &[QueryRoot],
+    physical_properties: &asap_types::ir::PhysicalPropertiesAssignment,
     selection: &Selection,
     tumbling_window_retention_ms: &std::collections::BTreeMap<
         asap_types::ir::physical_export::PhysicalASAPNodeId,
@@ -655,6 +661,7 @@ fn materialized_views_json(
 ) -> Result<Vec<Value>, String> {
     let (plan, report) = installable_plan_of(
         roots,
+        physical_properties,
         Some(selection),
         tumbling_window_retention_ms,
         entries,

crates/devtools/src/memo_pricing.rs

@@ -145,6 +145,7 @@ fn price_choice(
         Ok(selection) => {
             match installable_plan_of(
                 &selection.plan,
+                &selection.physical_properties,
                 None,
                 &selection.tumbling_window_retention_ms,
                 entries,

crates/devtools/src/planner_fuzz/mod.rs

@@ -503,7 +503,11 @@ impl Planner {
                     selected: "selected".into(),
                     cost_per_second,
                     stage0_roots: roots,
-                    logical_roots: run.run.selected_plan.clone(),
+                    logical_roots: run
+                        .run
+                        .selected_plan
+                        .as_ref()
+                        .map(|plan| plan.roots.clone()),
                 })
             }
         }

crates/devtools/src/planning_stages.rs

@@ -13,18 +13,16 @@ use asap_physical_asap_optimization::{
 };
 use asap_plan_selection::select_plan;
 use asap_planner::{
-    run_planning, PlanSearchSettings, PlanningModels, PlanningReport, PlanningRun, SelectionError,
+    run_planning, PlanSearchSettings, PlanningModels, PlanningReport, PlanningRun, SelectedPlan,
+    SelectionError,
 };
 use asap_types::ir::schema::schema_support::with_promql_series_identity;
 use asap_types::ir::schema::{DataType, Field, Schema};
 use asap_types::ir::QueryRoot;
 use asap_types::physical::PhysicalWorkloadPlan;
 use asap_types::stages::{AssignmentRealizer, DagOfLogicalMemoGroups, Memo};
-use asap_types::workload::bounds::with_confirmed_weights_at;
 use asap_types::workload::{DataWorkload, PlanningWorkload, QueryWorkloadEntry};
 use serde_json::Value;
-use std::collections::HashSet;
-use std::rc::Rc;
 
 use crate::memo_pricing::{reference_enumeration, ReferenceEnumeration};
 
@@ -109,9 +107,9 @@ impl PlanningStagesRun {
             .run
             .selected_plan
             .iter()
-            .map(|roots| ShownLogicalCandidate {
+            .map(|plan| ShownLogicalCandidate {
                 id: "L1".into(),
-                roots: Ok(roots.clone()),
+                roots: Ok(plan.roots.clone()),
             });
         let enumerated = self.reference.iter().flat_map(|reference| {
             reference
@@ -290,8 +288,8 @@ pub fn run_search_over(
     validate(&roots)?;
     let run = run_planning(roots.clone(), entries, data, models, settings)
         .map_err(|e| PlanningStagesError::Pass1(e.to_string()))?;
-    if let Some(roots) = &run.selected_plan {
-        validate(roots)?;
+    if let Some(plan) = &run.selected_plan {
+        validate(&plan.roots)?;
     }
     let reference = reference_choices
         .and_then(|max| reference_enumeration(&run, &roots, entries, data, models, max));
@@ -434,16 +432,8 @@ pub fn run_stages_until(
                 Ok(realized) => {
                     // As the facade does: the plan to run carries the weight
                     // proofs Stage 3 made.
-                    let proven: HashSet<*const asap_types::ir::OperatorNode> = selection
-                        .nodes_with_proven_non_negative_weights
-                        .iter()
-                        .filter_map(|node| realized.operator_node_of_selected_node.get(node.0))
-                        .map(Rc::as_ptr)
-                        .collect();
-                    let plan = with_confirmed_weights_at(&realized.plan.roots, data, &|node| {
-                        proven.contains(&Rc::as_ptr(node))
-                    });
-                    validate(&plan)?;
+                    let plan = SelectedPlan::of(&realized, &selection, data);
+                    validate(&plan.roots)?;
                     run.selected_plan = Some(plan);
                     run.selected_physical_plan = Some(realized);
                     run.selection = Ok(selection);

crates/devtools/tests/asap_planner.rs

@@ -1630,6 +1630,7 @@ fn scans_projecting_different_labels_are_shared() {
     let (_, reference) = plan("partial-sharing", &partial);
     let (installed, _) = asap_planner::installable_plan_of(
         &physical.roots,
+        &physical.physical_properties,
         None,
         &offered.tumbling_window_retention_ms,
         &entries,

crates/executor/src/physical_planner/logical.rs

@@ -36,8 +36,6 @@ pub(super) fn restore(dag: &PhysicalASAPDAG) -> Result<BTreeMap<NodeId, Rc<Opera
                 .map_children(|child| Rc::clone(&done[&(*child as u64)]));
             let mut rebuilt = OperatorNode::with_schema(operator, node.output_schema.clone());
             rebuilt.guarantee = node.guarantee.clone();
-            rebuilt.timing = Some(node.output_state.timing);
-            rebuilt.lifetime = node.lifetime;
             done.insert(node.id as u64, Rc::new(rebuilt));
         }
         if next.len() == before {

crates/executor/src/physical_planner/mod.rs

@@ -1607,8 +1607,7 @@ mod compilation_tests {
         use planner_types::ir::physical_export::compile_physical_asap_workload;
         use planner_types::ir::schema::{Field, Schema};
         use planner_types::ir::{
-            apply_physical_properties, NodeToTimedNodeCache, Operator,
-            PhysicalPropertiesAssignment, QueryRoot,
+            apply_physical_properties_to_roots, Operator, PhysicalPropertiesAssignment, QueryRoot,
         };
         let minute_ms = 60_000;
         let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
@@ -1644,22 +1643,11 @@ mod compilation_tests {
                 .unwrap()
             })
             .collect();
-        // One timing cache, so every window reads the one `Scan`. The cache is
-        // keyed by node address, so `windows` stays alive while it runs.
-        let mut cache = NodeToTimedNodeCache::new();
-        let roots: Vec<_> = windows
-            .iter()
-            .map(|window| {
-                let timed = apply_physical_properties(
-                    window,
-                    &PhysicalPropertiesAssignment::default(),
-                    &mut cache,
-                )
+        let roots: Vec<_> = windows.into_iter().map(QueryRoot::Operator).collect();
+        let physical_properties =
+            apply_physical_properties_to_roots(&roots, &PhysicalPropertiesAssignment::default())
                 .unwrap();
-                QueryRoot::Operator(timed)
-            })
-            .collect();
-        let dag = compile_physical_asap_workload(&roots).unwrap();
+        let dag = compile_physical_asap_workload(&roots, &physical_properties).unwrap();
         let roots = dag.roots.iter().map(|&id| id as NodeId).collect();
         (dag, roots)
     }

crates/executor/tests/common/mod.rs

@@ -1,20 +1,16 @@
 #![allow(dead_code)]
 use planner_types::ir::physical_export::PhysicalASAPDAG;
 use planner_types::ir::{
-    apply_physical_properties, NodeToTimedNodeCache, OperatorNode, PhysicalPropertiesAssignment,
-    QueryRoot,
+    apply_physical_properties, OperatorNode, PhysicalPropertiesAssignment, QueryRoot,
 };
 use std::rc::Rc;
 
 pub fn compile_physical_asap_dag(
     root: &Rc<OperatorNode>,
 ) -> Result<PhysicalASAPDAG, Box<dyn std::error::Error>> {
-    let root = apply_physical_properties(
-        root,
-        &PhysicalPropertiesAssignment::default(),
-        &mut NodeToTimedNodeCache::default(),
-    )?;
-    Ok(planner_types::ir::physical_export::compile_physical_asap_dag(&root)?)
+    let physical_properties =
+        apply_physical_properties(root, &PhysicalPropertiesAssignment::default())?;
+    Ok(planner_types::ir::physical_export::compile_physical_asap_dag(root, &physical_properties)?)
 }
 
 /// The logical DAG the planner pipeline (`asap_planner::run_planning`) selects for the
@@ -27,7 +23,8 @@ pub fn selected_dag(
     let run = run_all_stages(root, accuracy);
     let [QueryRoot::Operator(selected)] = run
         .selected_plan
-        .as_deref()
+        .as_ref()
+        .map(|plan| plan.roots.as_slice())
         .unwrap_or_else(|| panic!("a plan is selected: {:?}", run.selection))
     else {
         panic!("one operator root");

crates/executor/tests/time_series_window.rs

@@ -12,8 +12,8 @@ use planner_types::ir::physical_export::{compile_physical_asap_dag, PhysicalASAP
 use planner_types::ir::scalar::{CompareOpKind, ScalarValue};
 use planner_types::ir::schema::{DataType, Field, Schema, PROMQL_SERIES_IDENTITY};
 use planner_types::ir::{
-    apply_physical_properties, ExprSemantics, NodeToTimedNodeCache, NonASAPOp, Operator,
-    OperatorNode, PhysicalPropertiesAssignment, Predicate, ScalarExpr, TimeRangeKind,
+    apply_physical_properties, ExprSemantics, NonASAPOp, Operator, OperatorNode,
+    PhysicalPropertiesAssignment, Predicate, ScalarExpr, TimeRangeKind,
 };
 use std::collections::BTreeMap;
 use std::sync::Arc;
@@ -84,13 +84,9 @@ fn window_over_rows(
         child,
     }))
     .unwrap();
-    let timed = apply_physical_properties(
-        &range,
-        &PhysicalPropertiesAssignment::default(),
-        &mut NodeToTimedNodeCache::new(),
-    )
-    .unwrap();
-    compile_physical_asap_dag(&timed).unwrap()
+    let physical_properties =
+        apply_physical_properties(&range, &PhysicalPropertiesAssignment::default()).unwrap();
+    compile_physical_asap_dag(&range, &physical_properties).unwrap()
 }
 
 /// `(instance, ts, value)` samples of `latency_ms` as a store's raw data.

crates/executor/tests/weighted_topk_binding.rs

@@ -69,6 +69,7 @@ fn confirmed_as_counter(
     match planner_types::workload::bounds::with_confirmed_weights(
         &[QueryRoot::Operator(plan)],
         &data,
+        &mut Default::default(),
     )
     .remove(0)
     {
@@ -274,33 +275,20 @@ fn rate_updates_cannot_enter_integer_heap_factory() {
 /// it upstream of a query-time node, so the pin is lifted first.
 fn continuously_maintained_dag(candidate: &Rc<planner_types::ir::OperatorNode>) -> PhysicalASAPDAG {
     use planner_types::ir::{
-        apply_physical_properties, NodeToTimedNodeCache, Operator as IrOperator, OperatorNode,
+        apply_physical_properties, Operator as IrOperator, OperatorNode,
         PhysicalPropertiesAssignment,
     };
     use planner_types::physical::PhysicalProperties;
-    fn unpinned(node: &Rc<OperatorNode>) -> Rc<OperatorNode> {
-        let mut copy = (**node).clone();
-        copy.operator = node.operator.map_children(unpinned);
-        if matches!(
-            copy.operator,
-            IrOperator::ASAP(ASAPOp::FinalizeExactAccumulator { .. })
-        ) {
-            copy.timing = None;
-        }
-        Rc::new(copy)
-    }
-    let candidate = unpinned(candidate);
     let mut assignment = PhysicalPropertiesAssignment::default();
-    for node in OperatorNode::reachable(&candidate) {
+    for node in OperatorNode::reachable(candidate) {
         if matches!(node.operator, IrOperator::ASAP(ASAPOp::SummaryAgg { .. })) {
             assignment.set(
                 &node,
                 PhysicalProperties::MATERIALIZED_IN_MEMORY_ACROSS_EVALUATIONS_AT_INGESTION,
             );
         }
     }
-    let timed =
-        apply_physical_properties(&candidate, &assignment, &mut NodeToTimedNodeCache::new())
-            .unwrap();
-    planner_types::ir::physical_export::compile_physical_asap_dag(&timed).unwrap()
+    let physical_properties = apply_physical_properties(candidate, &assignment).unwrap();
+    planner_types::ir::physical_export::compile_physical_asap_dag(candidate, &physical_properties)
+        .unwrap()
 }

crates/frontend-promql/tests/planner_support.rs

@@ -37,7 +37,12 @@ pub fn run_all_stages(
 /// at `accuracy`.
 pub fn selected_dag(root: Rc<OperatorNode>, accuracy: AccuracyTarget) -> Rc<OperatorNode> {
     let run = run_all_stages(root, accuracy);
-    match run.selected_plan.as_deref().expect("a plan is selected") {
+    match run
+        .selected_plan
+        .as_ref()
+        .map(|plan| plan.roots.as_slice())
+        .expect("a plan is selected")
+    {
         [asap_types::ir::QueryRoot::Operator(node)] => Rc::clone(node),
         other => panic!("one operator root: {other:?}"),
     }

crates/frontend-promql/tests/support.rs

@@ -83,16 +83,11 @@ pub fn promql_scalar(node: &ScalarExpr) -> Option<f64> {
 pub fn physical_asap_dag(
     root: &Rc<OperatorNode>,
 ) -> asap_types::ir::physical_export::PhysicalASAPDAG {
-    use asap_types::ir::{
-        apply_physical_properties, NodeToTimedNodeCache, PhysicalPropertiesAssignment,
-    };
-    let timed = apply_physical_properties(
-        root,
-        &PhysicalPropertiesAssignment::default(),
-        &mut NodeToTimedNodeCache::new(),
-    )
-    .expect("default physical properties");
-    asap_types::ir::physical_export::compile_physical_asap_dag(&timed)
+    use asap_types::ir::{apply_physical_properties, PhysicalPropertiesAssignment};
+    let physical_properties =
+        apply_physical_properties(root, &PhysicalPropertiesAssignment::default())
+            .expect("default physical properties");
+    asap_types::ir::physical_export::compile_physical_asap_dag(root, &physical_properties)
         .expect("post-ASAP DAG export")
 }