feat(ir): derive summary coverage as definition + selection

crates/types/src/ir/cse.rs

@@ -112,7 +112,6 @@ pub fn structural_hash(node: &OperatorNode, cache: &mut HashCache) -> u64 {
         &node.schema,
         &node.guarantee,
         node.timing,
-        &node.coverage,
     );
     serde_json::to_string(&own)
         .unwrap_or_default()
@@ -168,7 +167,6 @@ fn same_node(left: &OperatorNode, right: &OperatorNode, memo: &mut EqMemo) -> bo
         && left.result_kind == right.result_kind
         && left.schema == right.schema
         && left.timing == right.timing
-        && left.coverage == right.coverage
         && same_value(&left.guarantee, &right.guarantee)
         && same_value(&own_fields(left), &own_fields(right))
 }
@@ -250,14 +248,9 @@ fn intern_bottom_up(
     let operator = node
         .operator
         .map_children(|child| intern_bottom_up(table, visited, child));
-    let rebuilt = OperatorNode {
-        operator,
-        result_kind: node.result_kind,
-        schema: node.schema.clone(),
-        guarantee: node.guarantee.clone(),
-        timing: node.timing,
-        coverage: node.coverage.clone(),
-    };
+    let rebuilt = OperatorNode::with_schema(operator, node.schema.clone())
+        .with_guarantee(node.guarantee.clone())
+        .with_timing(node.timing);
     let interned = table.intern(rebuilt);
     visited.insert(Rc::as_ptr(node), (Rc::clone(node), Rc::clone(&interned)));
     interned

crates/types/src/ir/flat.rs

@@ -15,7 +15,6 @@ use serde::{Deserialize, Serialize};
 
 use super::node::{Operator, OperatorNode, OperatorResultKind};
 use super::query::QueryRoot;
-use super::summary_coverage::SummaryCoverage;
 use crate::post_asap::execution_data_state::ExecutionTiming;
 use crate::post_asap::guarantee::ResultGuarantee;
 use crate::pre_asap::schema::Schema;
@@ -31,8 +30,6 @@ pub struct FlatNode {
     pub schema: Schema,
     pub guarantee: Option<ResultGuarantee>,
     pub timing: Option<ExecutionTiming>,
-    #[serde(default)]
-    pub coverage: Option<SummaryCoverage>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -80,7 +77,6 @@ fn visit(
         schema: node.schema.clone(),
         guarantee: node.guarantee.clone(),
         timing: node.timing,
-        coverage: node.coverage.clone(),
     });
     originals.push(Rc::clone(node));
     ids.insert(Rc::as_ptr(node), id);

crates/types/src/ir/node.rs

@@ -3,14 +3,15 @@
 //! every traversal needs (output category and schema, accuracy guarantee,
 //! execution timing).
 
+use std::cell::OnceCell;
 use std::collections::HashSet;
 use std::rc::Rc;
 
 use serde::{Deserialize, Serialize};
 
 use super::asap::ASAPOp;
 use super::non_asap::NonASAPOp;
-use super::summary_coverage::{CoverageError, SummaryCoverage};
+use super::summary_coverage::SummaryCoverage;
 use crate::ir::SchemaDerivationError;
 use crate::post_asap::execution_data_state::ExecutionTiming;
 use crate::post_asap::guarantee::ResultGuarantee;
@@ -100,8 +101,33 @@ pub struct OperatorNode {
     pub schema: Schema,
     pub guarantee: Option<ResultGuarantee>,
     pub timing: Option<ExecutionTiming>,
-    #[serde(default)]
-    pub coverage: Option<SummaryCoverage>,
+    /// Cache for [`Self::coverage`], derived from `operator`.
+    #[serde(skip)]
+    coverage_cache: CoverageCache,
+}
+
+/// A lazily filled [`SummaryCoverage`]. It is not part of a node's value:
+/// equality ignores it, serialization skips it, and a clone starts empty so
+/// a clone whose operator is then edited cannot read a stale entry.
+#[derive(Default)]
+struct CoverageCache(OnceCell<Option<SummaryCoverage>>);
+
+impl Clone for CoverageCache {
+    fn clone(&self) -> Self {
+        Self::default()
+    }
+}
+
+impl PartialEq for CoverageCache {
+    fn eq(&self, _: &Self) -> bool {
+        true
+    }
+}
+
+impl std::fmt::Debug for CoverageCache {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("CoverageCache")
+    }
 }
 
 impl OperatorNode {
@@ -110,7 +136,9 @@ impl OperatorNode {
     /// ASAP operator, ...).
     pub fn new(operator: Operator) -> Result<Self, SchemaDerivationError> {
         let schema = operator.output_schema()?;
-        Ok(Self::with_schema(operator, schema))
+        let node = Self::with_schema(operator, schema);
+        node.check_merge()?;
+        Ok(node)
     }
 
     /// Build a node with caller-supplied output names and qualifiers. For
@@ -124,7 +152,7 @@ impl OperatorNode {
             schema,
             guarantee: None,
             timing: None,
-            coverage: None,
+            coverage_cache: CoverageCache::default(),
         }
     }
 
@@ -145,23 +173,22 @@ impl OperatorNode {
         self
     }
 
-    /// Attach caller-established coverage. Required on summary nodes; see
-    /// [`Self::requires_coverage`].
-    pub fn with_coverage(
-        mut self,
-        coverage: SummaryCoverage,
-    ) -> Result<Self, SchemaDerivationError> {
-        coverage.validate()?;
-        if self.result_kind != OperatorResultKind::State {
-            return Err(CoverageError::NotState.into());
-        }
-        self.coverage = Some(coverage);
-        Ok(self)
+    /// What this summary state covers; `None` for a node that is not a
+    /// `SummaryAgg` or a valid `SummaryMerge`. Derived on first use.
+    pub fn coverage(&self) -> Option<&SummaryCoverage> {
+        self.coverage_cache
+            .0
+            .get_or_init(|| SummaryCoverage::derive(self).ok())
+            .as_ref()
     }
 
-    /// Summary nodes whose state can be composed must declare coverage.
-    pub fn requires_coverage(&self) -> bool {
-        matches!(self.asap(), Some(ASAPOp::SummaryAgg { .. }))
+    /// A `SummaryMerge` is valid only over inputs with the same definition
+    /// and disjoint selections.
+    fn check_merge(&self) -> Result<(), SchemaDerivationError> {
+        if matches!(self.asap(), Some(ASAPOp::SummaryMerge { .. })) {
+            SummaryCoverage::derive(self)?;
+        }
+        Ok(())
     }
 
     pub fn non_asap(&self) -> Option<&NonASAPOp> {
@@ -315,14 +342,8 @@ impl OperatorNode {
                     "invalid time or identity column in schema".into(),
                 ));
             }
-            match &node.coverage {
-                Some(coverage) => {
-                    (*node.as_ref()).clone().with_coverage(coverage.clone())?;
-                }
-                None if node.requires_coverage() => return Err(CoverageError::Missing.into()),
-                None => {}
-            }
             node.operator.validate_inputs()?;
+            node.check_merge()?;
             if node.result_kind != node.operator.output_kind() {
                 return Err(SchemaDerivationError::InvalidScalarSignature(
                     "retained result kind disagrees with operation".into(),