feat(types): family-capability table and executor tumbling pane merges

crates/asap-physical-operators/src/operators/summary/mod.rs

@@ -1,4 +1,5 @@
 use super::*;
+use planner_types::post_asap::{FamilyCapability, SelectionRelation};
 /// A summary evaluation: a sketch query, or an exact evaluation with typed parameters.
 #[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
 pub enum SummaryEvaluation {
@@ -474,11 +475,26 @@ async fn build_summary(
         .collect())
 }
 async fn merge_summary(
-    rows: Vec<Vec<Value>>,
+    mut rows: Vec<Vec<Value>>,
     state_column: usize,
     groups: &[usize],
     context: &RunContext,
 ) -> Result<Vec<Vec<Value>>, Error> {
+    // Counter states join consecutive windows pairwise, which is exact only
+    // in time order: order the inputs by their window start.
+    if let Some(Value::Summary { family, .. }) = rows.first().map(|row| &row[state_column]) {
+        if FamilyCapability::of(family).merge.map(|m| m.inputs)
+            == Some(SelectionRelation::DisjointInTime)
+        {
+            rows.sort_by_cached_key(|row| match &row[state_column] {
+                Value::Summary { state, .. } => state
+                    .as_any()
+                    .downcast_ref::<crate::summary_kernels::exact::ExactAccumulator>()
+                    .and_then(|state| state.counter_window_start()),
+                _ => None,
+            });
+        }
+    }
     type GroupState = (Vec<Value>, SummaryFamilyType, Arc<dyn crate::AggregateCore>);
     let mut states: BTreeMap<Vec<Vec<u8>>, GroupState> = BTreeMap::new();
     let mut work = Cooperative::new(context);

crates/asap-physical-operators/src/physical_planner/mod.rs

@@ -16,7 +16,7 @@ use planner_types::ir::physical_export::{
 };
 use planner_types::ir::{ASAPOp, NonASAPOp, Operator as LogicalOperator, OperatorNode, ScalarExpr};
 use planner_types::{
-    post_asap::{FieldDataType, SketchStatistic, SummaryInputExpr},
+    post_asap::{FamilyCapability, FieldDataType, SketchStatistic, SummaryInputExpr},
     pre_asap::{
         AggIntent, ColumnRef, CompareOpKind, DataType, GroupKeys, Reduction as PlannerReduction,
     },
@@ -334,7 +334,13 @@ fn compile_internal(
                         "per-entity summary requires a resolved raw time range",
                     ));
                 };
-                let Some(NonASAPOp::Scan { schema, .. }) = child.non_asap() else {
+                // A tumbling pane reads its window through a `TimeShift`; the
+                // deployment supplies the shifted raw rows.
+                let source = match child.non_asap() {
+                    Some(NonASAPOp::TimeShift { child, .. }) => child,
+                    _ => child,
+                };
+                let Some(NonASAPOp::Scan { schema, .. }) = source.non_asap() else {
                     return Err(invalid("per-entity summary requires a resolved source"));
                 };
                 if !schema.closed || update.item.is_some() {
@@ -448,6 +454,27 @@ fn compile_internal(
                     continue;
                 }
             }
+            if let Payload::ASAP(ASAPOp::SummaryMerge { .. }) = &node.payload {
+                let input = schemas
+                    .first()
+                    .ok_or_else(|| invalid(format!("node {id}: summary merge requires inputs")))?;
+                require_sound_merge(input).map_err(|e| invalid(format!("node {id}: {e}")))?;
+                if output.time_index.is_some() {
+                    // Pane timestamps describe their own builds. The merged
+                    // state covers this evaluation's window, so merge by
+                    // series identity and attach the run scope's timestamp,
+                    // as a per-series `SummaryAgg` does.
+                    let merged = bind_operation(node, &schemas)?;
+                    let compact = merged.schema();
+                    physical_dag.add(helper_id(id, 1)?, inputs, merged)?;
+                    physical_dag.add(
+                        id,
+                        vec![helper_id(id, 1)?],
+                        Operator::scope_timestamp(compact, output)?,
+                    )?;
+                    continue;
+                }
+            }
             let mut operator = compile_node(node, &schemas)
                 .map_err(|error| invalid(format!("node {id}: {error}")))?;
             if operator.is_counter_evaluation() {
@@ -511,6 +538,18 @@ fn temporal_evaluation_drops_name(node: &PhysicalASAPDAGNode) -> bool {
 
 /// Bind a Planner node against the schemas supplied by its deployment edges.
 /// This is the same checked path used by complete DAG binding.
+/// The IR accepts a `SummaryMerge` over any state family; the executor runs
+/// one only when the family-capability table gives the family a sound merge.
+pub(crate) fn require_sound_merge(input: &SchemaRef) -> Result<(), Error> {
+    let state = &input.fields[summary_column(input)?].dtype;
+    if FamilyCapability::of(state).merge.is_none() {
+        return Err(invalid(format!(
+            "summary merge over {state:?} is unsupported: the family has no sound merge"
+        )));
+    }
+    Ok(())
+}
+
 pub fn compile_node(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result<Operator, Error> {
     for schema in inputs {
         crate::values::validate_schema(schema)?;

crates/asap-physical-operators/src/physical_planner/precompute.rs

@@ -567,6 +567,7 @@ fn fragment(
             if schemas.iter().any(|s| s != input) {
                 return Err(invalid("summary merge inputs differ"));
             }
+            super::require_sound_merge(input)?;
             let union = add(
                 (0..schemas.len() as u64).collect(),
                 Operator::union(input.clone(), schemas.len())?,

crates/asap-physical-operators/src/summary_kernels/exact.rs

@@ -106,6 +106,20 @@ impl ExactAccumulator {
         }
     }
 
+    /// The earliest sample time of a counter (Rate/Increase) state, over
+    /// all its populations; `None` for an empty or non-counter state.
+    /// Counter states merge pairwise in time order, so a merge of panes
+    /// orders them by this first.
+    pub fn counter_window_start(&self) -> Option<i64> {
+        std::iter::once(&self.scalar)
+            .chain(self.keyed.iter().flat_map(HashMap::values))
+            .filter_map(|state| match state {
+                ScalarState::Counter(Some(counter)) => Some(counter.starting_timestamp),
+                _ => None,
+            })
+            .min()
+    }
+
     /// SQL SUM distinguishes an empty/all-NULL input from an observed zero.
     pub(crate) fn is_empty_sum(&self) -> bool {
         self.keyed.is_none() && matches!(self.scalar, ScalarState::Sum { seen: false, .. })

crates/asap-physical-operators/tests/physical_semantics.rs

@@ -646,6 +646,150 @@ fn kll_partial_merge_and_multiple_evaluations_preserve_population() {
     }
 }
 
+// KLL panes that each compact merge into a state within the calibrated
+// single-build rank error: 60 panes of 1000 values at k = 200.
+#[test]
+fn compacted_kll_panes_merge_within_the_calibrated_rank_error() {
+    use planner_types::post_asap::{SketchAlgorithm, SketchKind, SketchParams, SketchStatistic};
+    const PANES: u64 = 60;
+    const PER_PANE: u64 = 1000;
+    const K: u32 = 200;
+    // The planner's empirical 99% KLL rank error, `2.296 / k^0.9723`
+    // (`asap_aware_mapping`'s `kll_rank_error_99`).
+    let bound = 2.296 / f64::from(K).powf(0.9723);
+    let n = (PANES * PER_PANE) as f64;
+    let input = schema(&[("v", DataType::Float64, false)]);
+    let family = FieldDataType::Sketch(
+        SketchKind::new(
+            SketchAlgorithm::Kll,
+            SketchParams::Kll {
+                k: K,
+                failure_probability: None,
+            },
+        ),
+        Default::default(),
+    );
+    let build = || Operator::summary_build(input.clone(), family.clone(), 0, None, vec![]).unwrap();
+    let state = build().schema();
+    let mut dag = PhysicalDAG::default();
+    for pane in 0..PANES {
+        // Pane `p` holds `[p, p + 1) * 1000`, as tumbling panes of a
+        // rising value would, so a lost pane shifts every rank.
+        let rows = (0..PER_PANE)
+            .map(|j| vec![Value::Float64((pane * PER_PANE + j) as f64)])
+            .collect();
+        dag.add(
+            pane,
+            vec![],
+            Operator::source(
+                input.clone(),
+                vec![Batch::try_new(input.clone(), rows).unwrap()],
+            )
+            .unwrap(),
+        )
+        .unwrap();
+        dag.add(PANES + pane, vec![pane], build()).unwrap();
+    }
+    let union = 2 * PANES;
+    dag.add(
+        union,
+        (PANES..2 * PANES).collect(),
+        Operator::union(state.clone(), PANES as usize).unwrap(),
+    )
+    .unwrap();
+    dag.add(
+        union + 1,
+        vec![union],
+        Operator::summary_merge(state.clone(), 0, vec![]).unwrap(),
+    )
+    .unwrap();
+    for (i, q) in [0.01, 0.1, 0.25, 0.5, 0.75, 0.9, 0.99]
+        .into_iter()
+        .enumerate()
+    {
+        let id = union + 2 + i as u64;
+        dag.add(
+            id,
+            vec![union + 1],
+            Operator::evaluation(
+                state.clone(),
+                0,
+                asap_physical_operators::operators::SummaryEvaluation::Sketch(
+                    SketchStatistic::Quantile { q },
+                ),
+            )
+            .unwrap(),
+        )
+        .unwrap();
+        let Value::Float64(estimate) = collect(&dag, id)[0][0] else {
+            panic!("quantile must be Float64");
+        };
+        // Value `v` has rank `v / n`.
+        let error = (estimate / n - q).abs();
+        assert!(error <= bound, "q={q}: rank error {error} > {bound}");
+    }
+}
+
+// Counter panes merged in any input order (here A, C, B) give the increase
+// of one build over the whole window, including the jumps between panes
+// and a counter reset inside B.
+#[test]
+fn counter_panes_merge_in_any_order() {
+    use asap_physical_operators::summary_kernels::exact::ExactAccumulator;
+    use planner_types::post_asap::{ExactKind, ExactParams};
+    let input = schema(&[
+        ("v", DataType::Float64, false),
+        ("t", DataType::Timestamp, false),
+    ]);
+    let family = FieldDataType::ExactAggregate(ExactKind::Increase, ExactParams::Increase);
+    let build =
+        || Operator::summary_build(input.clone(), family.clone(), 0, Some(1), vec![]).unwrap();
+    let state = build().schema();
+    let samples = |range: std::ops::Range<i64>| -> Vec<Vec<Value>> {
+        range
+            .map(|i| {
+                // Rises by 3 per sample and resets once, at sample 7.
+                let value = if i < 7 { 3 * i } else { 3 * (i - 7) + 1 };
+                vec![Value::Float64(value as f64), Value::Timestamp(i * 1_000)]
+            })
+            .collect()
+    };
+    let increase = |dag: &PhysicalDAG<'_, Batch, SchemaRef>, root: u64| -> f64 {
+        let rows = collect(dag, root);
+        let Value::Summary { state, .. } = &rows[0][0] else {
+            panic!("summary state expected");
+        };
+        state
+            .as_any()
+            .downcast_ref::<ExactAccumulator>()
+            .unwrap()
+            .evaluation(asap_physical_operators::Statistic::Increase, None, None)
+            .unwrap()
+            .unwrap()
+    };
+    let mut dag = PhysicalDAG::default();
+    // Panes A = [0, 4), B = [4, 8), C = [8, 12), added as A, C, B.
+    for (id, range) in [(0, 0..4), (1, 8..12), (2, 4..8), (3, 0..12)] {
+        let batch = Batch::try_new(input.clone(), samples(range)).unwrap();
+        dag.add(
+            id,
+            vec![],
+            Operator::source(input.clone(), vec![batch]).unwrap(),
+        )
+        .unwrap();
+        dag.add(id + 4, vec![id], build()).unwrap();
+    }
+    dag.add(8, vec![4, 5, 6], Operator::union(state.clone(), 3).unwrap())
+        .unwrap();
+    dag.add(
+        9,
+        vec![8],
+        Operator::summary_merge(state, 0, vec![]).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(increase(&dag, 9), increase(&dag, 7));
+}
+
 // Retained zero-column rows still own Vec headers and must consume the output budget.
 #[test]
 fn zero_column_output_obeys_memory_limit() {

crates/asap-physical-operators/tests/precompute_population.rs

@@ -260,6 +260,13 @@ fn state_dag(
     target: Option<FieldDataType>,
     merge: bool,
 ) -> CompiledPhysicalDAG {
+    try_state_dag(family, target, merge).unwrap()
+}
+fn try_state_dag(
+    family: FieldDataType,
+    target: Option<FieldDataType>,
+    merge: bool,
+) -> Result<CompiledPhysicalDAG, asap_physical_operators::Error> {
     let mut nodes = vec![PhysicalASAPDAGNode {
         id: 0,
         payload: PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryMerge { children: vec![] }),
@@ -329,7 +336,6 @@ fn state_dag(
         &[0],
         &[root as u64],
     )
-    .unwrap()
 }
 fn native_run(
     program: &CompiledPhysicalDAG,
@@ -393,6 +399,17 @@ fn sum_state(value: f64) -> Arc<dyn asap_physical_operators::AggregateCore> {
 }
 
 // Only an explicit merge may collapse distinct pane updates before finalization.
+#[test]
+fn precompute_merge_requires_a_family_with_a_sound_merge() {
+    // IRate reads the window's last two samples, so the family-capability
+    // table gives it no merge.
+    let irate = FieldDataType::ExactAggregate(ExactKind::IRate, ExactParams::IRate);
+    let error = try_state_dag(irate, None, true).err().expect("rejected");
+    assert!(error.to_string().contains("no sound merge"), "{error}");
+    let sum = FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum);
+    try_state_dag(sum, None, true).unwrap();
+}
+
 #[test]
 fn explicit_merge_changes_pane_cardinality() {
     let family = FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum);

crates/asap-physical-operators/tests/summary_projection.rs

@@ -179,3 +179,40 @@ fn post_asap_summary_projection_survives_recovery() {
         assert!(output.next().await.is_none());
     });
 }
+
+// A `SummaryMerge` with no inputs is an invalid DAG, not a panic.
+#[test]
+fn summary_merge_without_inputs_is_rejected() {
+    let family = FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum);
+    let schema = planner_types::pre_asap::Schema {
+        unique_keys: vec![],
+        closed: false,
+        fields: vec![Field {
+            table: None,
+            name: "state".into(),
+            dtype: family,
+            nullable: false,
+        }],
+        time_index: None,
+    };
+    let dag = PhysicalASAPDAG {
+        nodes: vec![PhysicalASAPDAGNode {
+            id: 0,
+            payload: PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryMerge { children: vec![] }),
+            output_schema: schema,
+            output_state: ExecutionDataState::INGESTION_SUMMARY,
+            materialization: Default::default(),
+            guarantee: None,
+        }],
+        edges: vec![],
+        roots: vec![0],
+        scalar_roots: vec![],
+    };
+    let error = compile(&dag, BTreeMap::new(), &[0])
+        .err()
+        .expect("rejected");
+    assert!(
+        error.to_string().contains("summary merge requires inputs"),
+        "{error}"
+    );
+}