refactor(types, stages): replace Materialization with PhysicalProperties (timing, lifetime)

.agents/skills/asap-code-review/SKILL.md

@@ -72,7 +72,7 @@ a trade-off, give the author choices here too.
 For example:
 
 - `price` recomputes `query_reach(dag)`, a walk over the whole DAG, for every
-  MatAtQuery node: O(n · (n + e)) per candidate, times every candidate. At
+  `OneEvaluation` node: O(n · (n + e)) per candidate, times every candidate. At
   Example 3's 972 candidates, if each DAG has about 50 nodes, that is about 2.5·10⁶
   node visits instead of 5·10⁴. Alternative: compute it once per candidate
   before the loop and pass it in.

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, materialization, and sharing are preserved. This does not
+/// schemas, timing, lifetime, and sharing are preserved. This does not
 /// infer facts: `bind` must leave unsupported statistics unknown.
 pub fn bind_guarantees(
     root: &QueryRoot,
@@ -139,7 +139,7 @@ pub fn bind_guarantees(
         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_materialization(node.materialization);
+            .with_lifetime(node.lifetime);
         rebuilt.guarantee = node
             .guarantee
             .as_ref()
@@ -215,7 +215,7 @@ pub fn with_resized_summaries(
             Rc::new(
                 rebuilt
                     .with_timing(node.timing)
-                    .with_materialization(node.materialization),
+                    .with_lifetime(node.lifetime),
             )
         } else {
             node.clone()

crates/cost/src/analytical.rs

@@ -1489,6 +1489,13 @@ pub enum AnalyticalCostError {
     UnsupportedWithoutDistribution(&'static str),
     #[error("analytical arithmetic overflowed")]
     Overflow,
+    /// A node whose timing and lifetime cannot go together: a planner bug.
+    #[error("node {node:?}: {reason}")]
+    InvalidPhysicalProperties {
+        node: asap_types::ir::physical_export::PhysicalASAPNodeId,
+        #[source]
+        reason: asap_types::physical::InvalidPhysicalProperties,
+    },
     #[error("candidate has no supported exact or sketch state")]
     UnsupportedCandidate,
     #[error("inconsistent operator statistics: {0}")]

crates/cost/src/execution_demand.rs

@@ -9,7 +9,8 @@
 
 use super::analytical::AnalyticalCostError;
 use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPNodeId};
-use asap_types::physical::Materialization;
+use asap_types::ir::properties::ExecutionTiming;
+use asap_types::physical::PhysicalProperties;
 use asap_types::workload::{QueryRecurrence, RangeEvaluation, RepeatedDemand};
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 
@@ -280,7 +281,7 @@ pub fn query_starts(dag: &PhysicalASAPDAG) -> Vec<Vec<PhysicalASAPNodeId>> {
         .collect()
 }
 
-/// What computes a `NotMaterialized` node once per its own execution
+/// What computes a not-materialized node once per its own execution
 /// ([`consumers`]).
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Consumer {
@@ -290,19 +291,19 @@ pub enum Consumer {
     Materialized(PhysicalASAPNodeId),
 }
 
-/// The consumers of each `NotMaterialized` node of `dag` that any consumer
+/// The consumers of each not-materialized node of `dag` that any consumer
 /// reaches ([`computations`]): the queries and the materialized nodes whose
 /// walk down their inputs reaches it before meeting a materialized node,
 /// each once; the queries first, in order, then the materialized nodes in
 /// `dag`'s node order.
 pub fn consumers(dag: &PhysicalASAPDAG) -> HashMap<PhysicalASAPNodeId, Vec<Consumer>> {
-    let materialized: Vec<PhysicalASAPNodeId> = dag
+    let materialized_nodes: Vec<PhysicalASAPNodeId> = dag
         .nodes
         .iter()
-        .filter(|n| n.materialization != Materialization::NotMaterialized)
+        .filter(|n| n.lifetime.is_materialized())
         .map(|n| n.id)
         .collect();
-    let is_materialized: HashSet<PhysicalASAPNodeId> = materialized.iter().copied().collect();
+    let is_materialized: HashSet<PhysicalASAPNodeId> = materialized_nodes.iter().copied().collect();
     let inputs = |id: PhysicalASAPNodeId| {
         dag.edges
             .iter()
@@ -314,7 +315,7 @@ pub fn consumers(dag: &PhysicalASAPDAG) -> HashMap<PhysicalASAPNodeId, Vec<Consu
         .enumerate()
         .map(|(query, start)| (Consumer::Query(query), start))
         .chain(
-            materialized
+            materialized_nodes
                 .iter()
                 .map(|&id| (Consumer::Materialized(id), inputs(id).collect())),
         );
@@ -335,17 +336,17 @@ pub fn consumers(dag: &PhysicalASAPDAG) -> HashMap<PhysicalASAPNodeId, Vec<Consu
 
 /// How often each node of `dag` is computed: `(multiplier, per_second)`
 /// (module doc, "Cost model"). A materialized node is computed once for all
-/// its readers: once per second at ingestion time (`MatAtIngestion`), at
-/// the combined rate of the queries reaching it (`MatAtQuery`,
+/// its readers: once per second at ingestion time (materialized at ingestion time), at
+/// the combined rate of the queries reaching it (`OneEvaluation`,
 /// [`evaluation_rate`]), or once per request of those queries for a chain
 /// kept across evaluations, whose request builds only its newest tumbling window
 /// however many range steps it evaluates. `range_evaluations[q]` is query
-/// `q`'s range evaluation (none when missing). A `NotMaterialized` node is computed once per
+/// `q`'s range evaluation (none when missing). A not-materialized node is computed once per
 /// consumer that reaches it ([`consumers`]), at that consumer's rate. A consumer is a query
 /// (operator or scalar root), at its own rate, or a materialized node, at
 /// that node's rate; a walk stops at the first materialized node it meets,
 /// and a node read several times within one consumer's walk counts once for
-/// it. The `MatAtIngestion` consumers of one refresh window
+/// it. The consumers materialized at ingestion time of one refresh window
 /// ([`PhysicalASAPDAG::refresh_window`]) are refreshed in one ingestion run,
 /// which computes the node once for all of them: they count as one
 /// consumer (a consumer with an unbounded window counts alone). Its
@@ -358,12 +359,18 @@ pub fn computations(
     root_rates: Option<&[f64]>,
 ) -> Result<HashMap<PhysicalASAPNodeId, (u64, f64)>, AnalyticalCostError> {
     let range_of = |query: usize| range_evaluations.get(query).copied().flatten();
-    let materialized: HashMap<_, _> = dag
-        .nodes
-        .iter()
-        .filter(|n| n.materialization != Materialization::NotMaterialized)
-        .map(|n| (n.id, n.materialization))
-        .collect();
+    let mut materialized_nodes: HashMap<PhysicalASAPNodeId, PhysicalProperties> = HashMap::new();
+    for node in &dag.nodes {
+        let properties = node.physical_properties().map_err(|reason| {
+            AnalyticalCostError::InvalidPhysicalProperties {
+                node: node.id,
+                reason,
+            }
+        })?;
+        if properties.lifetime().is_materialized() {
+            materialized_nodes.insert(node.id, properties);
+        }
+    }
     let inputs = |id: PhysicalASAPNodeId| {
         dag.edges
             .iter()
@@ -385,14 +392,12 @@ pub fn computations(
         }
     }
     let mut rate_of_materialized = HashMap::new();
-    for (&id, materialization) in &materialized {
-        // A chain kept across evaluations is built once per request.
-        let per_request = matches!(
-            materialization,
-            Materialization::MatAtQueryKeptAcrossEvaluations(_)
-        );
-        let rate = match materialization {
-            Materialization::MatAtIngestion(_) => 1.0,
+    for (&id, &properties) in &materialized_nodes {
+        let at_ingestion = properties.timing() == ExecutionTiming::IngestionTime;
+        // A query-time chain materialized across evaluations is built once per request.
+        let per_request = !at_ingestion && properties.lifetime().is_across_evaluations();
+        let rate = match at_ingestion {
+            true => 1.0,
             _ => evaluation_rate(
                 reached.get(&id).into_iter().flatten().filter_map(|&q| {
                     let range = if per_request { None } else { range_of(q) };
@@ -416,9 +421,9 @@ pub fn computations(
     }
     // The refresh window of each node refreshed at ingestion.
     let refresh_windows = dag.refresh_windows(
-        materialized
+        materialized_nodes
             .iter()
-            .filter(|(_, m)| matches!(m, Materialization::MatAtIngestion(_)))
+            .filter(|(_, properties)| properties.timing() == ExecutionTiming::IngestionTime)
             .map(|(&id, _)| id),
     );
     let mut count: HashMap<_, (u64, f64)> = HashMap::new();

crates/cost/src/lib.rs

@@ -41,7 +41,7 @@ pub use measured::{
 pub use operator_statistics::EdgeStatistics;
 pub use ranking::{cheaper, worst_latency_ms, PlanRanking};
 pub use usage::{
-    family_name, kept_tumbling_window_chains, maintained_window_ms, node_map, node_usage,
+    family_name, maintained_window_ms, materialized_tumbling_window_chains, node_map, node_usage,
     plan_cost, query_reach, retained_tumbling_windows, trailing_window,
     tumbling_window_retention_ms, tumbling_window_retention_ms_of_plan, tumbling_windows_of,
     NodeMap, NodeUsage, ProducerOutput,
@@ -80,7 +80,7 @@ pub struct ResourceUsage {
     pub average_memory_bytes: f64,
     /// The most memory live at once, assuming every evaluation overlaps:
     /// in-memory kept state plus each computation's working memory, once
-    /// per consumer that recomputes it, with the `MatAtIngestion` consumers
+    /// per consumer that recomputes it, with the consumers materialized at ingestion time
     /// of one refresh window counted once.
     pub peak_memory_bytes: u64,
     /// Bytes kept on disk: state materialized on `Medium::Disk`, plus the
@@ -375,7 +375,7 @@ pub struct NodeCost {
     /// node the query reads first.
     pub cpu_core_seconds_per_second: f64,
     /// The number of consumers it is computed for, with the
-    /// `MatAtIngestion` consumers of one refresh window counted once (1
+    /// consumers materialized at ingestion time of one refresh window counted once (1
     /// when materialized, [`execution_demand::computations`]).
     pub multiplier: u64,
     /// How it was measured, e.g. "scan 1000 samples; x 1.0000/s; keeps