feat(ir): define compatible logical summary merges

Author: zzylol

crates/types/src/ir/asap.rs

@@ -39,9 +39,7 @@ pub enum ASAPOp {
     },
     /// Read an exact accumulator's state as its finalized value: the
     /// maintenance-to-read boundary before query-time operators.
-    FinalizeExactAccumulator {
-        child: Rc<OperatorNode>,
-    },
+    FinalizeExactAccumulator { child: Rc<OperatorNode> },
     /// Maintain the full declared population, including membership changes.
     MaintainPopulation {
         child: Rc<OperatorNode>,
@@ -52,10 +50,9 @@ pub enum ASAPOp {
         child: Rc<OperatorNode>,
         evaluation: PopulationStatistic,
     },
-    // ── Reserved: migrated but unimplemented (§1.3 of the proposal) ──
-    SummaryMerge {
-        children: Vec<Rc<OperatorNode>>,
-    },
+    /// Merge compatible partial states for the same grouping and family.
+    SummaryMerge { children: Vec<Rc<OperatorNode>> },
+    // ── Reserved: migrated but unimplemented ──
     SummarySubtract {
         left: Rc<OperatorNode>,
         right: Rc<OperatorNode>,
@@ -186,11 +183,7 @@ impl ASAPOp {
         use ASAPOp::*;
         matches!(
             self,
-            SummaryMerge { .. }
-                | SummarySubtract { .. }
-                | SummaryDelete { .. }
-                | SummaryJoin { .. }
-                | Extension { .. }
+            SummarySubtract { .. } | SummaryDelete { .. } | SummaryJoin { .. } | Extension { .. }
         )
     }
 
@@ -202,6 +195,14 @@ impl ASAPOp {
     pub fn produced_state(&self) -> Option<&FieldDataType> {
         match self {
             ASAPOp::SummaryAgg { family, .. } | ASAPOp::SummaryJoin { family, .. } => Some(family),
+            ASAPOp::SummaryMerge { children } => children.first().and_then(|child| {
+                child
+                    .schema
+                    .fields
+                    .iter()
+                    .find(|field| !field.is_plain())
+                    .map(|field| &field.dtype)
+            }),
             _ => None,
         }
     }
@@ -405,8 +406,11 @@ impl ASAPOp {
                     .output_schema()?
                 }
             }
-            SummaryMerge { .. }
-            | SummarySubtract { .. }
+            SummaryMerge { children } => {
+                self.validate_inputs()?;
+                children[0].schema.clone()
+            }
+            SummarySubtract { .. }
             | SummaryDelete { .. }
             | SummaryJoin { .. }
             | Extension { .. } => return Err(Self::unimplemented()),
@@ -444,6 +448,37 @@ impl ASAPOp {
             }
         };
         match self {
+            SummaryMerge { children } => {
+                let Some(first) = children.first() else {
+                    return Err(SchemaDerivationError::InvalidScalarSignature(
+                        "summary merge requires at least one state input".into(),
+                    ));
+                };
+                // Matching state parameters and grouping positions are necessary;
+                // matching names alone cannot prove two states compatible.
+                if first
+                    .schema
+                    .fields
+                    .iter()
+                    .filter(|field| !field.is_plain())
+                    .count()
+                    != 1
+                {
+                    return Err(SchemaDerivationError::InvalidScalarSignature(
+                        "summary merge requires exactly one state column".into(),
+                    ));
+                }
+                for child in children {
+                    needs_state(child, "SummaryMerge")?;
+                    if child.schema != first.schema {
+                        return Err(SchemaDerivationError::InvalidScalarSignature(
+                            "summary merge inputs must have identical state and grouping schemas"
+                                .into(),
+                        ));
+                    }
+                }
+                Ok(())
+            }
             SummaryEstimate {
                 summary_input,
                 query,

crates/types/tests/summary_merge_structure.rs

@@ -0,0 +1,53 @@
+//! Window composition merges compatible summary states without consuming raw rows.
+use asap_types::{
+    ir::operator_properties::{Reduction, Source},
+    ir::{ASAPOp, NonASAPOp, Operator, OperatorNode},
+    post_asap::{GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryUpdate},
+    pre_asap::{ColumnRef, DataType, Field, FieldDataType, Schema},
+};
+use std::rc::Rc;
+fn state(k: u32) -> Rc<OperatorNode> {
+    let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
+        source: Source::Table {
+            table_ref: "latencies".into(),
+        },
+        predicates: vec![],
+        schema: Schema::new(vec![Field::plain("value", DataType::Float64, false)]),
+    }))
+    .unwrap();
+    OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
+        child: scan,
+        family: FieldDataType::Sketch(
+            SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }),
+            Default::default(),
+        ),
+        input: SummaryUpdate::column(ColumnRef::SampleValue),
+        reduction: Reduction::by(vec![]),
+        grouping: GroupingStrategy::default(),
+        filter: None,
+    }))
+    .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)],
+    }))
+    .unwrap();
+    root.validate_structure().unwrap();
+    assert_eq!(root.schema.fields.len(), 1);
+}
+/// An empty merge, raw rows and differently sized state cannot masquerade as compatible panes.
+#[test]
+fn incompatible_merge_inputs_fail() {
+    for children in [
+        vec![],
+        vec![state(200), state(300)],
+        vec![state(200).children()[0].clone()],
+    ] {
+        assert!(
+            OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children })).is_err()
+        );
+    }
+}