feat(types): an unmaterialized summary build runs at its consumer's timing; one α^phys error at export

crates/executor/tests/every_valid_state.rs

@@ -18,7 +18,7 @@ use planner_types::ir::{
 };
 use planner_types::physical::{Lifetime, PhysicalProperties};
 
-/// `FinalizeExactAccumulator(SummaryAgg(Sum)(TimeRange 5 min(Scan latency_ms)))`.
+/// `FinalizeExactAccumulator(SummaryMerge(SummaryAgg(Sum)(TimeRange 5 min(Scan latency_ms))))`.
 fn plan() -> Rc<OperatorNode> {
     let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
         source: Source::TimeSeries {
@@ -50,19 +50,30 @@ fn plan() -> Rc<OperatorNode> {
         filter: None,
     }))
     .unwrap();
+    let merge = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
+        children: vec![state],
+        newest_tumbling_windows: None,
+        group_by: None,
+    }))
+    .unwrap();
     OperatorNode::new_shared(Operator::ASAP(ASAPOp::FinalizeExactAccumulator {
-        child: state,
+        child: merge,
     }))
     .unwrap()
 }
 
+/// The build under the merge under `root`.
+fn build_of(root: &Rc<OperatorNode>) -> &Rc<OperatorNode> {
+    root.children()[0].children()[0]
+}
+
 /// What the physical planner makes of the build in each valid state: a
 /// state materialized across evaluations (either timing, either medium) is
-/// a materialized view slot the deployment binds; every other state is
-/// computed in the query run (a `OneEvaluation` result once for the run's
-/// readers). A summary build not materialized at ingestion time is not
-/// built by rule: the timing pass runs an unmaterialized build at query
-/// time.
+/// a materialized view slot the deployment binds; a query-time state that
+/// is not materialized across evaluations is computed in the query run (a
+/// `OneEvaluation` result once for the run's consumers). Not materialized at
+/// ingestion time, the build sits under the merge materialized at ingestion
+/// time and is computed in that view's ingestion run.
 #[test]
 fn physical_planner_over_every_valid_state() {
     use ExecutionTiming::{IngestionTime, QueryTime};
@@ -83,17 +94,38 @@ fn physical_planner_over_every_valid_state() {
     {
         let root = plan();
         let mut assignment = PhysicalPropertiesAssignment::default();
-        let unmaterialized_at_ingestion =
-            (properties.timing(), properties.lifetime()) == (IngestionTime, Lifetime::None);
-        if !unmaterialized_at_ingestion {
-            assignment.set(root.children()[0], properties);
+        match (properties.timing(), properties.lifetime()) {
+            // The build gets no entry: it runs in the ingestion run of the
+            // merge above it, which is materialized at ingestion time.
+            (IngestionTime, Lifetime::None) => assignment.set(
+                root.children()[0],
+                PhysicalProperties::new(
+                    ExecutionTiming::IngestionTime,
+                    Lifetime::AcrossEvaluations(Medium::Memory),
+                )
+                .expect("a valid timing and lifetime"),
+            ),
+            _ => assignment.set(build_of(&root), properties),
         }
         let timed = apply_physical_properties(&root, &assignment).unwrap();
         let mut dag = compile_physical_asap_dag(&root, &timed).unwrap();
         let views = dag.set_retention(&HashMap::new(), &HashMap::new()).unwrap();
-        let compiled = asap_executor::physical_planner::compile(&dag, &views, &[3])
-            .unwrap_or_else(|error| panic!("{properties:?}: {error}"));
-        let build = compiled.operator_name(2);
+        let id_of = |summary_merge: bool| {
+            dag.nodes
+                .iter()
+                .find(|node| match &node.payload {
+                    Operator::ASAP(ASAPOp::SummaryAgg { .. }) => !summary_merge,
+                    Operator::ASAP(ASAPOp::SummaryMerge { .. }) => summary_merge,
+                    _ => false,
+                })
+                .expect("the plan has a summary build and a merge")
+                .id
+        };
+        let (build_id, merge_id) = (id_of(false), id_of(true));
+        let compiled =
+            asap_executor::physical_planner::compile(&dag, &views, &[dag.roots[0] as u64])
+                .unwrap_or_else(|error| panic!("{properties:?}: {error}"));
+        let build = compiled.operator_name(build_id as u64);
         let computed_in_the_query_run = || {
             assert_ne!(build, Some("MaterializedView"), "{properties:?}");
             assert!(build.is_some(), "{properties:?}");
@@ -108,16 +140,24 @@ fn physical_planner_over_every_valid_state() {
             (QueryTime, Lifetime::None) => computed_in_the_query_run(),
             (QueryTime, Lifetime::OneEvaluation(_)) => computed_in_the_query_run(),
             (IngestionTime, Lifetime::None) => {
-                // Left unassigned: an unmaterialized build lands at query time.
+                let node = dag.nodes.iter().find(|node| node.id == build_id).unwrap();
+                assert_eq!(
+                    PhysicalProperties::new(node.output_state.timing, node.lifetime),
+                    Ok(properties)
+                );
+                // The query reads the merge's view; its ingestion run
+                // computes the build.
                 assert_eq!(
-                    PhysicalProperties::new(
-                        dag.nodes[2].output_state.timing,
-                        dag.nodes[2].lifetime
-                    ),
-                    Ok(
-                        PhysicalProperties::new(ExecutionTiming::QueryTime, Lifetime::None)
-                            .expect("a valid timing and lifetime")
-                    )
+                    compiled.operator_name(merge_id as u64),
+                    Some("MaterializedView")
+                );
+                let refresh =
+                    asap_executor::physical_planner::compile(&dag, &[], &[merge_id as u64])
+                        .unwrap();
+                let build = refresh.operator_name(build_id as u64);
+                assert!(
+                    build.is_some_and(|name| name != "MaterializedView"),
+                    "{build:?}"
                 );
             }
             (IngestionTime, Lifetime::OneEvaluation(_)) => {

crates/integration-tests/tests/tumbling_window_merge.rs

@@ -911,3 +911,94 @@ fn queries_at_60_and_90_seconds_read_one_chain_of_30_second_tumbling_windows() {
         );
     }
 }
+
+/// End to end, a summary build not materialized at ingestion time: a merge
+/// materialized at ingestion time reads the 5-minute Sum build, which is not
+/// materialized and so runs inside the merge's ingestion run. The query
+/// reads the merge's view and answers as the whole build at query time.
+#[test]
+fn an_unmaterialized_build_runs_in_the_ingestion_run_of_the_state_reading_it() {
+    let samples = samples(1);
+    let single = single_build("sum_over_time(m[5m])", AccuracyTarget::exact());
+    let build = summary_agg(&single);
+    let merged = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
+        children: vec![Rc::clone(&build)],
+        newest_tumbling_windows: None,
+        group_by: None,
+    }))
+    .unwrap();
+    let root = replace(&single, &build, &merged);
+    let mut assignment = PhysicalPropertiesAssignment::default();
+    assignment.set(
+        &merged,
+        PhysicalProperties::new(
+            ExecutionTiming::IngestionTime,
+            Lifetime::AcrossEvaluations(Medium::Memory),
+        )
+        .expect("a valid timing and lifetime"),
+    );
+    let mut exported = export(&root, &assignment);
+    let id_of = |summary_merge: bool| {
+        exported
+            .dag
+            .nodes
+            .iter()
+            .find(|node| match &node.payload {
+                Operator::ASAP(ASAPOp::SummaryAgg { .. }) => !summary_merge,
+                Operator::ASAP(ASAPOp::SummaryMerge { .. }) => summary_merge,
+                _ => false,
+            })
+            .expect("the plan has a summary build and a merge")
+            .id
+    };
+    let (build_id, merge_id) = (id_of(false), id_of(true));
+    let (build_id, merge_id) = (build_id as u64, merge_id as u64);
+    let properties = |id: u64| {
+        let node = exported
+            .dag
+            .nodes
+            .iter()
+            .find(|node| node.id as u64 == id)
+            .unwrap();
+        PhysicalProperties::new(node.output_state.timing, node.lifetime)
+    };
+    assert_eq!(
+        properties(build_id),
+        Ok(
+            PhysicalProperties::new(ExecutionTiming::IngestionTime, Lifetime::None)
+                .expect("a valid timing and lifetime")
+        )
+    );
+    assert_eq!(
+        properties(merge_id),
+        Ok(PhysicalProperties::new(
+            ExecutionTiming::IngestionTime,
+            Lifetime::AcrossEvaluations(Medium::Memory)
+        )
+        .expect("a valid timing and lifetime"))
+    );
+    exported.dag.validate().unwrap();
+    let views = views_at(&mut exported.dag, &[merge_id]).unwrap();
+    // The merge's ingestion run builds the Sum from the raw samples and
+    // materializes only the merge.
+    let ingestion = compile(&exported.dag, &[], &[merge_id]).unwrap();
+    let rows = run(
+        &ingestion,
+        raw_store(&exported.dag, &samples),
+        Scope::Ingestion {
+            window_start_ms: 0,
+            window_end_ms: EVALUATION_MS,
+        },
+    );
+    let store = InMemoryDataStore::default()
+        .with_materialized_view(
+            merge_id as usize,
+            vec![Batch::try_new(schema_of(&exported.dag, merge_id), rows).unwrap()],
+        )
+        .unwrap();
+    let query = compile(&exported.dag, &views, &[exported.dag.roots[0] as u64]).unwrap();
+    assert_eq!(
+        render(&run(&query, store, query_scope())),
+        render(&execute(&single, &samples))
+    );
+}

crates/physical-asap-optimization/src/catalog/configurations.rs

@@ -151,8 +151,12 @@ pub fn enumerate_candidate_configurations(
         //   `Disk` state given to it executes in memory (a view slot bound to
         //   the deployment's in-memory store, a one-evaluation result materialized
         //   within the query run), while cost prices it as disk;
-        // - (IngestionTime, None) for a summary build: an unmaterialized build
-        //   runs at query time (`apply_physical_properties`);
+        // - (IngestionTime, None) for a summary build: every consumer
+        //   handles it (the build runs inside the ingestion run of the
+        //   materialized state reading it), but offering it adds candidates
+        //   and can change selected plans, which is a separate decision;
+        //   `down_closed` requires every build below an ingestion-time build
+        //   to be materialized there;
         // - (IngestionTime, AcrossEvaluations) for a non-summary node: rows
         //   are not maintained at ingestion time, only summary states;
         // - (QueryTime, AcrossEvaluations) for anything but a tumbling-window

crates/plan-selection/src/lib.rs

@@ -5063,8 +5063,7 @@ mod tests {
 
     /// The KLL DAG whose node has `properties`: the build itself, except
     /// for (IngestionTime, None), which is the scan below the build
-    /// materialized at ingestion time (a summary build that is not
-    /// materialized runs at query time).
+    /// materialized at ingestion time.
     fn kll_dag_with(properties: PhysicalProperties) -> PhysicalASAPDAG {
         match (properties.timing(), properties.lifetime()) {
             (ExecutionTiming::IngestionTime, Lifetime::None) => kll_dag(at_ingestion()),

crates/types/src/ir/physical_export.rs

@@ -26,11 +26,12 @@ use crate::ir::properties::execution_data_state::{
 use crate::ir::properties::guarantee::AccuracyGuarantee;
 #[cfg(any(test, feature = "test-support"))]
 use crate::ir::properties::timing::apply_physical_properties_to_roots;
-use crate::ir::properties::timing::{
-    data_state, is_materializable_state_output, PhysicalPropertiesAssignment,
-};
+use crate::ir::properties::timing::{data_state, PhysicalPropertiesAssignment};
 use crate::ir::schema::{FieldDataType, Schema};
-use crate::physical::{InvalidPhysicalProperties, Lifetime, PhysicalProperties};
+use crate::physical::{
+    satisfies_physical, Evaluation, InvalidPhysicalProperties, Lifetime, PhysicalProperties,
+    PhysicalRequirement,
+};
 
 /// A node's operator, with each child replaced by its node id.
 pub type PhysicalASAPOperatorPayload = Operator<NodeId>;
@@ -179,14 +180,12 @@ pub enum PhysicalASAPDAGValidationError {
         #[source]
         reason: InvalidPhysicalProperties,
     },
-    /// A summary build (or maintained population) that is not materialized but does
-    /// not run at query time: a planner bug.
-    #[error(
-        "node {node:?} is a summary build that is not materialized but does not run at query time"
-    )]
-    UnmaterializedSummaryBuildNotAtQueryTime { node: PhysicalASAPNodeId },
-    #[error("ingestion node {consumer:?} depends on query node {producer:?}")]
-    QueryDependencyInIngestion {
+    /// An edge where α^phys fails (`satisfies_physical` is false): ingestion
+    /// work reads a query-time output, or a query-time node reads an
+    /// ingestion-time output that is not materialized (it runs only inside
+    /// the ingestion run that reads it). A planner bug.
+    #[error("node {consumer:?} cannot read node {producer:?} at its timing")]
+    PhysicalRequirementNotSatisfied {
         producer: PhysicalASAPNodeId,
         consumer: PhysicalASAPNodeId,
     },
@@ -604,31 +603,21 @@ impl PhysicalASAPDAG {
 
     pub fn validate(&self) -> Result<(), PhysicalASAPDAGValidationError> {
         let mut nodes = HashMap::new();
+        let mut properties_of = HashMap::new();
         for node in &self.nodes {
             if nodes.insert(node.id, node).is_some() {
                 return Err(PhysicalASAPDAGValidationError::DuplicateNodeId(node.id));
             }
-            // A node's timing and lifetime must go together; a summary
-            // build that is not materialized runs at query time. Any other node that
-            // is not materialized runs when its consumer runs.
+            // A node's timing and lifetime must go together. A node that is
+            // not materialized runs when its consumer runs (the edge checks below).
             let properties = PhysicalProperties::new(node.output_state.timing, node.lifetime)
                 .map_err(
                     |reason| PhysicalASAPDAGValidationError::InvalidPhysicalProperties {
                         node: node.id,
                         reason,
                     },
                 )?;
-            if is_materializable_state_output(&node.payload)
-                && properties
-                    == PhysicalProperties::new(ExecutionTiming::IngestionTime, Lifetime::None)
-                        .expect("a valid timing and lifetime")
-            {
-                return Err(
-                    PhysicalASAPDAGValidationError::UnmaterializedSummaryBuildNotAtQueryTime {
-                        node: node.id,
-                    },
-                );
-            }
+            properties_of.insert(node.id, properties);
             let materialized_across_evaluations = match (properties.timing(), properties.lifetime())
             {
                 (
@@ -750,13 +739,22 @@ impl PhysicalASAPDAG {
                     edge.consumer,
                 ));
             }
-            if producer.output_state.timing == ExecutionTiming::QueryTime
-                && nodes[&edge.consumer].output_state.timing == ExecutionTiming::IngestionTime
-            {
-                return Err(PhysicalASAPDAGValidationError::QueryDependencyInIngestion {
-                    producer: edge.producer,
-                    consumer: edge.consumer,
-                });
+            // α^phys: a node reads, in the same evaluation, only a producer
+            // that satisfies its timing (`satisfies_physical`).
+            let requirement = PhysicalRequirement::new(
+                nodes[&edge.consumer].output_state.timing,
+                Evaluation::Same,
+                None,
+            )
+            .expect("a same-evaluation read is valid at either timing");
+            let properties = properties_of[&edge.producer];
+            if !satisfies_physical(&properties, &requirement) {
+                return Err(
+                    PhysicalASAPDAGValidationError::PhysicalRequirementNotSatisfied {
+                        producer: edge.producer,
+                        consumer: edge.consumer,
+                    },
+                );
             }
             if edge.intermediate_schema != producer.output_schema {
                 return Err(PhysicalASAPDAGValidationError::EdgeSchemaMismatch {