feat(ir): record summarized columns in SummaryCoverage

SummaryCoverage now records which columns a state summarizes as well as
which rows: `input` (the SummaryAgg update expression) and `group_by`
(its reduction). with_coverage rejects a SummaryAgg declaration whose
columns differ from the node's own (ColumnMismatch), and SummaryMerge
requires coverage on every input (UnknownInput) with identical columns.
merge_disjoint checks columns too. summary_input_data is removed. A
nested SummaryMerge carries no coverage until #646, so it is rejected.

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

Author: zzylol

crates/types/src/ir/asap.rs

@@ -6,6 +6,7 @@ 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};
@@ -487,17 +488,13 @@ impl ASAPOp {
                         ));
                     }
                 }
-                // Equal schemas do not prove the states summarize the same
-                // expression with the same grouping; the producers do.
-                let input_data = first.summary_input_data();
-                if input_data.is_none()
-                    || children
-                        .iter()
-                        .any(|c| c.summary_input_data() != input_data)
-                {
-                    return Err(SchemaDerivationError::InvalidScalarSignature(
-                        "summary merge inputs must share update expression and reduction".into(),
-                    ));
+                // 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(())
             }

crates/types/src/ir/node.rs

@@ -10,12 +10,10 @@ use serde::{Deserialize, Serialize};
 
 use super::asap::ASAPOp;
 use super::non_asap::NonASAPOp;
-use super::operator_properties::Reduction;
 use super::summary_coverage::{CoverageError, SummaryCoverage};
 use crate::ir::SchemaDerivationError;
 use crate::post_asap::execution_data_state::ExecutionTiming;
 use crate::post_asap::guarantee::ResultGuarantee;
-use crate::post_asap::SummaryUpdate;
 use crate::pre_asap::schema::Schema;
 
 /// The output category of an operator, derived from the operation and its
@@ -148,7 +146,8 @@ impl OperatorNode {
     }
 
     /// Attach caller-established coverage. Required on summary nodes; see
-    /// [`Self::requires_coverage`].
+    /// [`Self::requires_coverage`]. On a `SummaryAgg` the coverage columns
+    /// must equal the node's `input` and `reduction`.
     pub fn with_coverage(
         mut self,
         coverage: SummaryCoverage,
@@ -157,28 +156,18 @@ impl OperatorNode {
         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);
         Ok(self)
     }
 
-    /// What a summary state summarizes from each row: the update expression
-    /// fed into the state (`input`, e.g. the `latency` column) and the
-    /// grouping (`reduction`, e.g. by `job`) of the `SummaryAgg` that produced
-    /// it. Which rows were included is `coverage`, not this. For a
-    /// `SummaryMerge` these are its first input's, which merge validation
-    /// requires every input to share. `None` for any other node. Equal schemas
-    /// cannot tell a KLL over `latency` from one over `size`; merging compares
-    /// this instead.
-    pub fn summary_input_data(&self) -> Option<(&SummaryUpdate, &Reduction)> {
-        match self.asap()? {
-            ASAPOp::SummaryAgg {
-                input, reduction, ..
-            } => Some((input, reduction)),
-            ASAPOp::SummaryMerge { children } => children.first()?.summary_input_data(),
-            _ => None,
-        }
-    }
-
     /// Summary nodes whose state can be composed must declare coverage.
     pub fn requires_coverage(&self) -> bool {
         matches!(self.asap(), Some(ASAPOp::SummaryAgg { .. }))

crates/types/src/ir/summary_coverage.rs

@@ -1,6 +1,10 @@
-//! Joint time/population coverage for summary composition, independent of schema.
+//! What a summary state covers, independent of schema: which rows (source and
+//! joint time/population regions) and which columns (the update expression fed
+//! into the state and the grouping).
 //! Equality predicates are a deliberately narrow proof vocabulary. Unsupported
 //! predicates cannot be declared disjoint merely by giving them different names.
+use crate::ir::operator_properties::Reduction;
+use crate::post_asap::SummaryUpdate;
 use crate::pre_asap::Source;
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
@@ -13,8 +17,15 @@ pub struct SummaryCoverage {
     /// Observation data source, as named by `Scan`: a table or a time series.
     /// Region time bounds refer to its time column.
     pub source: Source,
-    /// Union of joint regions; never the Cartesian product of independent bounds.
+    /// Rows: union of joint regions; never the Cartesian product of
+    /// independent bounds.
     pub regions: Vec<CoverageRegion>,
+    /// Columns: the update expression fed into the state from each row, equal
+    /// to the producing `SummaryAgg`'s `input` (e.g. the `latency` column).
+    pub input: SummaryUpdate,
+    /// Columns: the grouping, equal to the producing `SummaryAgg`'s
+    /// `reduction` (e.g. by `job`). Compared by child-schema column index.
+    pub group_by: Reduction,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -48,6 +59,10 @@ pub enum CoverageError {
     NotState,
     #[error("summary node requires coverage")]
     Missing,
+    #[error("summary coverage columns differ from the summary's input or grouping")]
+    ColumnMismatch,
+    #[error("summary merge requires coverage on every input")]
+    UnknownInput,
 }
 
 impl SummaryCoverage {
@@ -70,9 +85,17 @@ impl SummaryCoverage {
     }
 
     /// Every observation in a region is assumed to contribute once to the state.
-    /// Compose once-per-observation summaries only when their joint regions are
-    /// provably disjoint. Update/reduction compatibility, family merge capability
-    /// and accuracy are checked by `SummaryMerge`, not here.
+    /// Compose once-per-observation summaries only when they read the same
+    /// columns and their joint regions are provably disjoint. Family merge
+    /// capability and accuracy are not checked here.
+    /// Rejects `other` unless it reads the same columns as `self`.
+    pub fn check_columns(&self, other: &Self) -> Result<(), CoverageError> {
+        if self.input != other.input || self.group_by != other.group_by {
+            return Err(CoverageError::ColumnMismatch);
+        }
+        Ok(())
+    }
+
     pub fn merge_disjoint(inputs: &[Self]) -> Result<Self, CoverageError> {
         let first = inputs.first().ok_or(CoverageError::EmptyMerge)?;
         let mut merged = first.clone();
@@ -82,6 +105,7 @@ impl SummaryCoverage {
             if input.source != first.source {
                 return Err(CoverageError::SourceMismatch);
             }
+            input.check_columns(first)?;
             merged.regions.extend(input.regions.iter().cloned());
         }
         merged.validate()?;

crates/types/tests/schema_rebuilding.rs

@@ -17,6 +17,8 @@ fn coverage() -> SummaryCoverage {
             time_ms: None,
             population: Default::default(),
         }],
+        input: SummaryUpdate::column(ColumnRef::Named("value".into())),
+        group_by: Reduction::by(vec![0]),
     }
 }
 /// Rewrites clear coverage; a rewriter must declare it again for summary nodes.

crates/types/tests/structure_contract.rs

@@ -24,6 +24,10 @@ fn whole_table() -> asap_types::ir::summary_coverage::SummaryCoverage {
             time_ms: None,
             population: Default::default(),
         }],
+        input: asap_types::post_asap::SummaryUpdate::column(
+            asap_types::pre_asap::ColumnRef::Named("x".into()),
+        ),
+        group_by: asap_types::pre_asap::Reduction::by(vec![]),
     }
 }
 /// Resolved filters cannot hide invalid scalar types or out-of-scope columns.

crates/types/tests/summary_coverage.rs

@@ -20,6 +20,8 @@ fn coverage(start: i64, end: i64, population: &[(&str, &str)]) -> SummaryCoverag
                 .map(|(k, v)| (k.to_string(), v.to_string()))
                 .collect(),
         }],
+        input: SummaryUpdate::column(ColumnRef::Named("latency".into())),
+        group_by: Reduction::by(vec![]),
     }
 }
 /// Adjacent panes coalesce; gaps remain disconnected rather than becoming a hull.
@@ -148,3 +150,14 @@ fn regions_without_time_bounds() {
         Err(CoverageError::PossibleOverlap)
     );
 }
+
+/// Disjoint rows do not make coverage of different columns composable.
+#[test]
+fn different_columns_do_not_compose() {
+    let mut by_job = coverage(1, 2, &[]);
+    by_job.group_by = Reduction::by(vec![0]);
+    assert_eq!(
+        SummaryCoverage::merge_disjoint(&[coverage(0, 1, &[]), by_job]),
+        Err(CoverageError::ColumnMismatch)
+    );
+}

crates/types/tests/summary_merge_structure.rs

@@ -1,6 +1,7 @@
 //! Window composition merges compatible summary states without consuming raw rows.
 use asap_types::{
     ir::operator_properties::{Reduction, Source},
+    ir::summary_coverage::CoverageError,
     ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, SchemaDerivationError},
     post_asap::{GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryUpdate},
     pre_asap::{ColumnRef, DataType, Field, FieldDataType, Schema},
@@ -29,7 +30,7 @@ fn state_over(family: FieldDataType, input: SummaryUpdate) -> Rc<OperatorNode> {
     let summary = OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg {
         child: scan,
         family,
-        input,
+        input: input.clone(),
         reduction: Reduction::by(vec![]),
         grouping: GroupingStrategy::default(),
         filter: None,
@@ -45,6 +46,8 @@ fn state_over(family: FieldDataType, input: SummaryUpdate) -> Rc<OperatorNode> {
                     time_ms: Some(0..1),
                     population: Default::default(),
                 }],
+                input,
+                group_by: Reduction::by(vec![]),
             })
             .unwrap(),
     )
@@ -84,25 +87,66 @@ fn shifted(state: &Rc<OperatorNode>, start: i64, end: i64) -> Rc<OperatorNode> {
     Rc::new(node)
 }
 
-/// Equal schemas do not prove both states summarize the same expression.
+fn kll() -> FieldDataType {
+    FieldDataType::Sketch(
+        SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }),
+        Default::default(),
+    )
+}
+
+fn merge(children: Vec<Rc<OperatorNode>>) -> Result<Rc<OperatorNode>, SchemaDerivationError> {
+    OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children }))
+}
+
+/// Equal schemas cannot tell a KLL over one column from a KLL over another;
+/// the columns in their coverage can.
 #[test]
-fn different_update_expressions_do_not_merge() {
-    let kll = || {
-        FieldDataType::Sketch(
-            SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }),
-            Default::default(),
-        )
-    };
+fn different_coverage_columns_do_not_merge() {
     let named = state_over(
         kll(),
         SummaryUpdate::column(ColumnRef::Named("value".into())),
     );
-    let result = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
-        children: vec![state(200), shifted(&named, 1, 2)],
-    }));
     assert!(matches!(
-        result,
-        Err(SchemaDerivationError::InvalidScalarSignature(message))
-            if message.contains("update expression and reduction")
+        merge(vec![state(200), shifted(&named, 1, 2)]),
+        Err(SchemaDerivationError::Coverage(
+            CoverageError::ColumnMismatch
+        ))
     ));
 }
+
+/// A `SummaryAgg` cannot declare columns other than its own input and grouping.
+#[test]
+fn declared_columns_must_match_the_summary() {
+    let mut coverage = state(200).coverage.clone().unwrap();
+    coverage.input = SummaryUpdate::column(ColumnRef::Named("value".into()));
+    let fresh = (*state(200)).clone();
+    assert!(matches!(
+        fresh.clone().with_coverage(coverage.clone()),
+        Err(SchemaDerivationError::Coverage(
+            CoverageError::ColumnMismatch
+        ))
+    ));
+    let mut forged = fresh;
+    forged.coverage = Some(coverage);
+    assert!(matches!(
+        Rc::new(forged).validate_structure(),
+        Err(SchemaDerivationError::Coverage(
+            CoverageError::ColumnMismatch
+        ))
+    ));
+}
+
+/// Every merge input needs coverage. A nested merge has none until #646, so
+/// it is rejected for now.
+#[test]
+fn merge_inputs_without_coverage_are_rejected() {
+    let mut missing = (*state(200)).clone();
+    missing.coverage = None;
+    let nested = merge(vec![state(200)]).unwrap();
+    for input in [Rc::new(missing), nested] {
+        assert!(matches!(
+            merge(vec![state(200), input]),
+            Err(SchemaDerivationError::Coverage(CoverageError::UnknownInput))
+        ));
+    }
+}