feat(ir): derive summary coverage from the subtree with one SummaryCoverage::derive

SummaryCoverage::derive(node, time_range) is the only way coverage is
computed, for every summary node:

- SummaryAgg: the one source scanned below it, the population read from
  its filters, its input and reduction (grouping keys named from its input
  schema) as columns, over the caller's time range.
- SummaryMerge: the disjoint union of its inputs' coverage, which must read
  the same source and columns; its time range comes from the inputs.

OperatorNode::new derives coverage (a SummaryAgg starts with no time
bounds), with_time_range replaces with_coverage and sets a SummaryAgg's
time range, validate_structure requires the retained coverage to equal the
derived one (DerivedMismatch), and with_new_children derives it again,
keeping the time range. A SummaryAgg whose source or population cannot be
read has no coverage and cannot be merged (NoSingleSource,
UnprovenPopulation); an unreadable population is never treated as
unrestricted. Merges now nest. CoverageRegion.time_ms is renamed
time_range.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: zzylol

crates/types/src/ir/asap.rs

@@ -6,7 +6,6 @@ use std::rc::Rc;
 use serde::{Deserialize, Serialize};
 
 use super::node::{OperatorNode, OperatorResultKind};
-use super::summary_coverage::CoverageError;
 use crate::ir::operator_properties::Reduction;
 use crate::ir::SchemaDerivationError;
 use crate::post_asap::maintained_population::{MaintainedPopulation, PopulationStatistic};
@@ -488,14 +487,6 @@ impl ASAPOp {
                         ));
                     }
                 }
-                // Equal schemas cannot tell a KLL over `latency` from one over
-                // `size`; the columns in each input's coverage can. A nested
-                // `SummaryMerge` carries no coverage yet (#646), so it is rejected.
-                let columns = first.coverage.as_ref().ok_or(CoverageError::UnknownInput)?;
-                for child in children {
-                    let coverage = child.coverage.as_ref().ok_or(CoverageError::UnknownInput)?;
-                    coverage.check_columns(columns)?;
-                }
                 Ok(())
             }
             SummaryEstimate {

crates/types/src/ir/node.rs

@@ -4,6 +4,7 @@
 //! execution timing).
 
 use std::collections::HashSet;
+use std::ops::Range;
 use std::rc::Rc;
 
 use serde::{Deserialize, Serialize};
@@ -110,7 +111,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 mut node = Self::with_schema(operator, schema);
+        node.coverage = node.derived_coverage(None)?;
+        Ok(node)
     }
 
     /// Build a node with caller-supplied output names and qualifiers. For
@@ -145,32 +148,41 @@ impl OperatorNode {
         self
     }
 
-    /// Attach caller-established coverage. Required on summary nodes; see
-    /// [`Self::requires_coverage`]. On a `SummaryAgg` the coverage columns
-    /// must equal the node's `input` and `reduction`.
-    pub fn with_coverage(
+    /// Set the time range a `SummaryAgg`'s state covers (for example an
+    /// ingested pane's bounds) and derive its coverage with
+    /// [`SummaryCoverage::derive`]. Fails on any other node, or when the
+    /// coverage cannot be derived.
+    pub fn with_time_range(
         mut self,
-        coverage: SummaryCoverage,
+        time_range: Option<Range<i64>>,
     ) -> Result<Self, SchemaDerivationError> {
-        coverage.validate()?;
-        if self.result_kind != OperatorResultKind::State {
-            return Err(CoverageError::NotState.into());
-        }
-        if let Some(ASAPOp::SummaryAgg {
-            input, reduction, ..
-        }) = self.asap()
-        {
-            if &coverage.input != input || &coverage.group_by != reduction {
-                return Err(CoverageError::ColumnMismatch.into());
-            }
-        }
-        self.coverage = Some(coverage);
+        self.coverage = Some(SummaryCoverage::derive(&self, time_range)?);
         Ok(self)
     }
 
-    /// Summary nodes whose state can be composed must declare coverage.
-    pub fn requires_coverage(&self) -> bool {
-        matches!(self.asap(), Some(ASAPOp::SummaryAgg { .. }))
+    /// The coverage this node should carry: [`SummaryCoverage::derive`] for a
+    /// `SummaryMerge`; for a `SummaryAgg` the same over `time_range`, or none
+    /// when its source or population cannot be read (it then cannot be
+    /// merged); none for any other node.
+    fn derived_coverage(
+        &self,
+        time_range: Option<Range<i64>>,
+    ) -> Result<Option<SummaryCoverage>, CoverageError> {
+        match self.asap() {
+            Some(ASAPOp::SummaryAgg { .. }) => Ok(SummaryCoverage::derive(self, time_range).ok()),
+            Some(ASAPOp::SummaryMerge { .. }) => SummaryCoverage::derive(self, None).map(Some),
+            _ => Ok(None),
+        }
+    }
+
+    /// The time range of a `SummaryAgg`'s coverage; `None` otherwise.
+    fn coverage_time_range(&self) -> Option<Range<i64>> {
+        match self.asap() {
+            Some(ASAPOp::SummaryAgg { .. }) => {
+                self.coverage.as_ref()?.regions.first()?.time_range.clone()
+            }
+            _ => None,
+        }
     }
 
     pub fn non_asap(&self) -> Option<&NonASAPOp> {
@@ -213,14 +225,18 @@ impl OperatorNode {
     /// structural schema metadata. Only names and qualifiers that override
     /// the old derived schema are retained, for either operator category. A
     /// change in output arity with such overrides needs an explicit new
-    /// naming assignment.
-    /// `guarantee` and `timing` depend on the inputs and are cleared.
+    /// naming assignment. `guarantee` and `timing` depend on the inputs and
+    /// are cleared. Coverage is derived again from the new inputs, keeping a
+    /// `SummaryAgg`'s time range; this fails when it can no longer be derived.
     pub fn with_new_children(
         &self,
         f: impl FnMut(&Rc<OperatorNode>) -> Rc<OperatorNode>,
     ) -> Result<Self, SchemaDerivationError> {
         let previous = self.operator.output_schema()?;
         let mut rebuilt = Self::new(self.operator.map_children(f))?;
+        if let (Some(ASAPOp::SummaryAgg { .. }), Some(_)) = (rebuilt.asap(), &self.coverage) {
+            rebuilt = rebuilt.with_time_range(self.coverage_time_range())?;
+        }
         for (i, (derived, retained)) in previous.fields.iter().zip(&self.schema.fields).enumerate()
         {
             let renamed = retained.name != derived.name;
@@ -324,14 +340,15 @@ impl OperatorNode {
                     "invalid time or identity column in schema".into(),
                 ));
             }
-            match &node.coverage {
-                Some(coverage) => {
-                    (*node.as_ref()).clone().with_coverage(coverage.clone())?;
+            node.operator.validate_inputs()?;
+            let derived = node.derived_coverage(node.coverage_time_range())?;
+            if node.coverage != derived {
+                return Err(match node.coverage {
+                    None => CoverageError::Missing,
+                    Some(_) => CoverageError::DerivedMismatch,
                 }
-                None if node.requires_coverage() => return Err(CoverageError::Missing.into()),
-                None => {}
+                .into());
             }
-            node.operator.validate_inputs()?;
             if node.result_kind != node.operator.output_kind() {
                 return Err(SchemaDerivationError::InvalidScalarSignature(
                     "retained result kind disagrees with operation".into(),