feat(ir): require disjoint coverage for summary merges

Author: zzylol

crates/types/src/ir/asap.rs

@@ -213,6 +213,29 @@ impl ASAPOp {
         }
     }
 
+    /// Derive joint observation coverage; unknown or overlapping inputs fail closed.
+    pub fn merged_extent(
+        &self,
+    ) -> Result<super::observation_extent::ObservationExtent, SchemaDerivationError> {
+        let ASAPOp::SummaryMerge { children } = self else {
+            return Err(SchemaDerivationError::InvalidScalarSignature(
+                "coverage merge requires SummaryMerge".into(),
+            ));
+        };
+        let inputs = children
+            .iter()
+            .map(|child| {
+                child.observation_extent.clone().ok_or_else(|| {
+                    SchemaDerivationError::InvalidScalarSignature(
+                        "summary merge requires known observation coverage".into(),
+                    )
+                })
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        super::observation_extent::ObservationExtent::merge_disjoint(&inputs)
+            .map_err(|error| SchemaDerivationError::InvalidScalarSignature(error.to_string()))
+    }
+
     /// Output schema derived from the operator and its children. Summary
     /// planning may retain a more specific schema (evaluation column naming)
     /// through [`OperatorNode::with_schema`]; all structural metadata must
@@ -483,6 +506,7 @@ impl ASAPOp {
                         ));
                     }
                 }
+                self.merged_extent()?;
                 Ok(())
             }
             SummaryEstimate {

crates/types/src/ir/node.rs

@@ -110,7 +110,11 @@ 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);
+        if let Some(op @ ASAPOp::SummaryMerge { .. }) = node.asap() {
+            node.observation_extent = Some(op.merged_extent()?);
+        }
+        Ok(node)
     }
 
     /// Build a node with caller-supplied output names and qualifiers. For
@@ -320,6 +324,13 @@ impl OperatorNode {
                 None if node.requires_coverage() => return Err(CoverageError::Missing.into()),
                 None => {}
             }
+            if let Some(op @ ASAPOp::SummaryMerge { .. }) = node.asap() {
+                if node.observation_extent.as_ref() != Some(&op.merged_extent()?) {
+                    return Err(SchemaDerivationError::InvalidScalarSignature(
+                        "retained merge coverage disagrees with input union".into(),
+                    ));
+                }
+            }
             node.operator.validate_inputs()?;
             if node.result_kind != node.operator.output_kind() {
                 return Err(SchemaDerivationError::InvalidScalarSignature(

crates/types/tests/summary_merge_structure.rs

@@ -15,7 +15,7 @@ fn state(k: u32) -> Rc<OperatorNode> {
         schema: Schema::new(vec![Field::plain("value", DataType::Float64, false)]),
     }))
     .unwrap();
-    OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
+    let summary = OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg {
         child: scan,
         family: FieldDataType::Sketch(
             SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }),
@@ -26,17 +26,38 @@ fn state(k: u32) -> Rc<OperatorNode> {
         grouping: GroupingStrategy::default(),
         filter: None,
     }))
-    .unwrap()
+    .unwrap();
+    std::rc::Rc::new(
+        summary
+            .with_observation_extent(asap_types::ir::observation_extent::ObservationExtent {
+                source: "latencies:timestamp".into(),
+                revision: "fixture-1".into(),
+                input: SummaryUpdate::column(ColumnRef::SampleValue),
+                grouping: Reduction::by(vec![]),
+                multiplicity:
+                    asap_types::ir::observation_extent::ObservationMultiplicity::OncePerObservation,
+                regions: vec![asap_types::ir::observation_extent::ExtentRegion {
+                    start_ms: 0,
+                    end_ms: 1,
+                    population: Default::default(),
+                }],
+            })
+            .unwrap(),
+    )
 }
 /// Two KLL panes compose into one typed logical state without timing assignment.
 #[test]
 fn compatible_panes_merge_structurally() {
     let root = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
-        children: vec![state(200), state(200)],
+        children: vec![state(200), shifted_state(200, 1, 2)],
     }))
     .unwrap();
     root.validate_structure().unwrap();
     assert_eq!(root.schema.fields.len(), 1);
+    assert_eq!(
+        root.observation_extent.as_ref().unwrap().regions[0].end_ms,
+        2
+    );
 }
 /// An empty merge, raw rows and differently sized state cannot masquerade as compatible panes.
 #[test]
@@ -51,3 +72,38 @@ fn incompatible_merge_inputs_fail() {
         );
     }
 }
+
+fn shifted_state(k: u32, start: i64, end: i64) -> Rc<OperatorNode> {
+    let mut node = (*state(k)).clone();
+    let region = &mut node.observation_extent.as_mut().unwrap().regions[0];
+    region.start_ms = start;
+    region.end_ms = end;
+    Rc::new(node)
+}
+/// Schema equality cannot authorize overlapping or unknown observation coverage.
+#[test]
+fn unsafe_coverage_merge_is_rejected() {
+    let mut unknown = (*state(200)).clone();
+    unknown.observation_extent = None;
+    for children in [
+        vec![state(200), state(200)],
+        vec![state(200), Rc::new(unknown)],
+    ] {
+        assert!(
+            OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children })).is_err()
+        );
+    }
+}
+
+/// Gapped time coverage remains disconnected, and forged output metadata is rejected.
+#[test]
+fn merge_derives_coverage_and_validates_retained_metadata() {
+    let root = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
+        children: vec![state(200), shifted_state(200, 2, 3)],
+    }))
+    .unwrap();
+    assert_eq!(root.observation_extent.as_ref().unwrap().regions.len(), 2);
+    let mut forged = (*root).clone();
+    forged.observation_extent.as_mut().unwrap().regions[0].end_ms = 2;
+    assert!(Rc::new(forged).validate_structure().is_err());
+}