feat(ir): reject summary merges of heap-based sketches

A merged top-k heap can miss an item that is heavy in only one input, so
CmsWithHeap, CountSketchWithHeap and UnivMon states do not merge. Also
correct the merge_disjoint doc: SummaryMerge checks schema, update,
reduction and heap families, not accuracy.

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

Author: zzylol

crates/types/src/ir/asap.rs

@@ -10,7 +10,7 @@ use super::summary_coverage::{CoverageError, SummaryCoverage};
 use crate::ir::operator_properties::Reduction;
 use crate::ir::SchemaDerivationError;
 use crate::post_asap::maintained_population::{MaintainedPopulation, PopulationStatistic};
-use crate::post_asap::sketch::{GroupingStrategy, SketchStatistic, SummaryUpdate};
+use crate::post_asap::sketch::{GroupingStrategy, SketchAlgorithm, SketchStatistic, SummaryUpdate};
 use crate::pre_asap::schema::{ColumnId, DataType, Field, FieldDataType, Schema};
 
 /// Why an ASAP operator cannot be used yet.
@@ -42,7 +42,9 @@ pub enum ASAPOp<C = Rc<OperatorNode>> {
     },
     /// Read an exact accumulator's state as its finalized value: the
     /// maintenance-to-read boundary before query-time operators.
-    FinalizeExactAccumulator { child: C },
+    FinalizeExactAccumulator {
+        child: C,
+    },
     /// Maintain the full declared population, including membership changes.
     MaintainPopulation {
         child: C,
@@ -54,7 +56,9 @@ pub enum ASAPOp<C = Rc<OperatorNode>> {
         evaluation: PopulationStatistic,
     },
     /// Merge compatible partial states for the same grouping and family.
-    SummaryMerge { children: Vec<C> },
+    SummaryMerge {
+        children: Vec<C>,
+    },
     // ── Reserved: migrated but unimplemented ──
     SummarySubtract {
         left: C,
@@ -498,6 +502,23 @@ impl ASAPOp {
                         ));
                     }
                 }
+                // A heap of top-k candidates does not merge exactly: an item heavy
+                // in only one input can be missing from the merged heap.
+                if first.schema.fields.iter().any(|field| {
+                    matches!(
+                        &field.dtype,
+                        FieldDataType::Sketch(kind, _) if matches!(
+                            kind.algorithm(),
+                            SketchAlgorithm::CmsWithHeap
+                                | SketchAlgorithm::CountSketchWithHeap
+                                | SketchAlgorithm::UnivMon
+                        )
+                    )
+                }) {
+                    return Err(SchemaDerivationError::InvalidScalarSignature(
+                        "summary merge does not support heap-based sketches".into(),
+                    ));
+                }
                 // Coverage records only time and population; what each state
                 // summarizes and how it is grouped come from the producers.
                 let update = first.summary_update();

crates/types/src/ir/summary_coverage.rs

@@ -75,8 +75,10 @@ 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.
+    /// provably disjoint. `SummaryMerge` checks the rest: equal schemas (so the
+    /// same family and parameters), equal update and reduction, and no
+    /// heap-based family. Accuracy of the merged state is not assessed; its
+    /// `guarantee` is `None`.
     pub fn merge_disjoint(inputs: &[Self]) -> Result<Self, CoverageError> {
         let first = inputs.first().ok_or(CoverageError::EmptyMerge)?;
         let mut merged = first.clone();

crates/types/tests/summary_merge_structure.rs

@@ -7,6 +7,13 @@ use asap_types::{
 };
 use std::rc::Rc;
 fn state(k: u32) -> Rc<OperatorNode> {
+    state_with(FieldDataType::Sketch(
+        SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }),
+        Default::default(),
+    ))
+}
+
+fn state_with(family: FieldDataType) -> Rc<OperatorNode> {
     let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
         source: Source::Table {
             table_ref: "latencies".into(),
@@ -17,10 +24,7 @@ fn state(k: u32) -> Rc<OperatorNode> {
     .unwrap();
     let summary = OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg {
         child: scan,
-        family: FieldDataType::Sketch(
-            SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }),
-            Default::default(),
-        ),
+        family,
         input: SummaryUpdate::column(ColumnRef::SampleValue),
         reduction: Reduction::by(vec![]),
         grouping: GroupingStrategy::default(),
@@ -70,11 +74,35 @@ fn incompatible_merge_inputs_fail() {
 }
 
 fn shifted_state(k: u32, start: i64, end: i64) -> Rc<OperatorNode> {
-    let mut node = (*state(k)).clone();
+    shifted(&state(k), start, end)
+}
+
+fn shifted(state: &Rc<OperatorNode>, start: i64, end: i64) -> Rc<OperatorNode> {
+    let mut node = (**state).clone();
     let region = &mut node.coverage.as_mut().unwrap().regions[0];
     region.time_ms = Some(start..end);
     Rc::new(node)
 }
+
+/// A heap of top-k candidates does not merge exactly, even over disjoint panes.
+#[test]
+fn heap_based_sketches_do_not_merge() {
+    let heap = state_with(FieldDataType::Sketch(
+        SketchKind::new(
+            SketchAlgorithm::CmsWithHeap,
+            SketchParams::CmsWithHeap {
+                width: 1024,
+                depth: 4,
+                heap_size: 10,
+            },
+        ),
+        Default::default(),
+    ));
+    let result = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
+        children: vec![Rc::clone(&heap), shifted(&heap, 1, 2)],
+    }));
+    assert!(result.is_err());
+}
 /// Schema equality cannot authorize overlapping or unknown observation coverage.
 #[test]
 fn unsafe_coverage_merge_is_rejected() {