From fbff58dce2e70f5892ea4db360adbd4e652f95af Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:15:37 +0000 Subject: [PATCH 1/6] feat(types): family-capability table with mergeability facts Add a generic family-capability table keyed by the state family: whether the family merges, how the merged inputs' selections must relate (disjoint, or overlap allowed for idempotent families), how a merged state's error relates to its parts', and whether it subtracts. Hydra over Count-Min or Count-Sketch merges; Hydra over KLL fails closed. The IR stays permissive: SummaryMerge accepts any family and consumers consult the table. SummaryCoverage skips the disjointness check for families whose merge allows overlap, and the union of identical selections is that selection. Execution timing validates SummaryMerge: it reads summary state at either phase, and a merge at ingestion time needs every input at ingestion time. --- crates/types/src/ir/summary_coverage.rs | 31 +- crates/types/src/ir/timing.rs | 18 +- .../types/src/post_asap/family_capability.rs | 119 +++++++ crates/types/src/post_asap/mod.rs | 5 +- crates/types/tests/summary_merge_structure.rs | 311 +++++++++++++++++- 5 files changed, 462 insertions(+), 22 deletions(-) create mode 100644 crates/types/src/post_asap/family_capability.rs diff --git a/crates/types/src/ir/summary_coverage.rs b/crates/types/src/ir/summary_coverage.rs index 53cb6b1e2..ab5e90f90 100644 --- a/crates/types/src/ir/summary_coverage.rs +++ b/crates/types/src/ir/summary_coverage.rs @@ -21,6 +21,7 @@ use super::asap::ASAPOp; use super::node::{Operator, OperatorNode}; use super::non_asap::{NonASAPOp, TimeRangeKind}; use super::scalar::{Predicate, ScalarExpr}; +use crate::post_asap::family_capability::{FamilyCapability, SelectionRelation}; use crate::pre_asap::expr_ir::{CompareOpKind, ScalarValue}; use crate::pre_asap::schema::{ColumnId, Schema}; @@ -71,14 +72,18 @@ pub enum CoverageError { EmptyMerge, #[error("summary merge inputs compute different things")] DefinitionMismatch, - #[error("summary merge inputs are not proven disjoint")] + #[error( + "summary merge inputs are not proven disjoint and the family counts shared rows twice" + )] PossibleOverlap, } impl SummaryCoverage { /// Coverage of a `SummaryAgg` or `SummaryMerge`. A merge fails unless /// every input has the same definition and their selections are - /// pairwise disjoint. + /// pairwise disjoint; a family whose merge is idempotent + /// ([`SelectionRelation::OverlapAllowed`] in the family-capability + /// table) may also merge overlapping selections. pub fn derive(node: &OperatorNode) -> Result { match node.asap() { Some(ASAPOp::SummaryAgg { .. }) => Ok(of_summary_agg(node)), @@ -88,17 +93,24 @@ impl SummaryCoverage { .map(|child| child.coverage().ok_or(CoverageError::NotSummary)) .collect::, _>>()?; let first = inputs.first().ok_or(CoverageError::EmptyMerge)?; + let may_overlap = matches!( + first.definition.asap(), + Some(ASAPOp::SummaryAgg { family, .. }) + if FamilyCapability::of(family).merge.map(|m| m.inputs) + == Some(SelectionRelation::OverlapAllowed) + ); let mut proven = HashSet::new(); for (i, input) in inputs.iter().enumerate() { if !same_definition(&first.definition, &input.definition, &mut proven) { return Err(CoverageError::DefinitionMismatch); } - let overlaps = inputs[..i].iter().any(|other| { - input - .selection - .iter() - .any(|a| other.selection.iter().any(|b| !a.disjoint(b))) - }); + let overlaps = !may_overlap + && inputs[..i].iter().any(|other| { + input + .selection + .iter() + .any(|a| other.selection.iter().any(|b| !a.disjoint(b))) + }); if overlaps { return Err(CoverageError::PossibleOverlap); } @@ -669,6 +681,9 @@ fn union(mut boxes: Vec) -> Vec { } fn join(a: &SelectionBox, b: &SelectionBox) -> Option { + if a == b { + return Some(a.clone()); + } if a.columns == b.columns { let ((al, au), (bl, bu)) = (a.relative_time.as_ref()?, b.relative_time.as_ref()?); let time = touching((al, au), (bl, bu), |x: &i64, y: &i64| Some(x.cmp(y)))?; diff --git a/crates/types/src/ir/timing.rs b/crates/types/src/ir/timing.rs index 25dc4bfad..d42725eac 100644 --- a/crates/types/src/ir/timing.rs +++ b/crates/types/src/ir/timing.rs @@ -44,6 +44,7 @@ //! |---|---| //! | `SummaryAgg.child` | Rows, or exact-accumulator state, never a query-time value when the state is maintained | //! | `SummaryEstimate.summary_input` | Summary state at either phase | +//! | `SummaryMerge.children` | Summary state at either phase; at ingestion time only for a merge at ingestion time | //! | `FinalizeExactAccumulator.child` | Exact-accumulator state | //! | `EvaluatePopulation.child` | A timed `MaintainPopulation` | //! | `MaintainPopulation.child` | Rows at its own timing matching the population's input | @@ -535,8 +536,21 @@ fn validate_asap( } Ok(()) } - ASAPOp::SummaryMerge { .. } - | ASAPOp::SummarySubtract { .. } + ASAPOp::SummaryMerge { children } => { + for child in children { + let s = state_of(child); + if s.primitive != DataPrimitive::SummaryState + || (timing == ExecutionTiming::IngestionTime && s.timing != timing) + { + return Err(ExecutionDataStateError::IllegalChildDataState { + edge: "SummaryMerge.children", + child: s, + }); + } + } + Ok(()) + } + ASAPOp::SummarySubtract { .. } | ASAPOp::SummaryDelete { .. } | ASAPOp::SummaryJoin { .. } | ASAPOp::Extension { .. } => Err(ExecutionDataStateError::UnimplementedOperator { diff --git a/crates/types/src/post_asap/family_capability.rs b/crates/types/src/post_asap/family_capability.rs new file mode 100644 index 000000000..e082cb05b --- /dev/null +++ b/crates/types/src/post_asap/family_capability.rs @@ -0,0 +1,119 @@ +//! The generic family-capability table: what each summary family can do, +//! keyed by its [`FieldDataType`]. +//! +//! The IR stays permissive: a `SummaryMerge` over any state family is +//! structurally valid. The planner rules and the executor consult this +//! table to decide whether a composition is sound. Today the table holds +//! the mergeability facts; the summary-capability facts (which statistics +//! a family answers) are added as further fields of [`FamilyCapability`]. + +use super::sketch::{ExactKind, GroupingStrategy, HydraKind, SketchAlgorithm}; +use crate::pre_asap::schema::FieldDataType; + +/// What one summary family can do. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FamilyCapability { + /// How states of the family merge; `None` when the family has no sound + /// merge. + pub merge: Option, + /// Whether a state subtracts out of another of the same family + /// (`SummarySubtract`, reserved): the subtrahend's selection must lie + /// inside the minuend's, and the result is the state of the difference. + pub subtractable: bool, +} + +/// How a family's states merge. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct MergeCapability { + /// How the selections of the merged states must relate. + pub inputs: SelectionRelation, + /// How the merged state's error relates to its parts' errors. + pub error: MergedError, +} + +/// How the selections of a merge's inputs must relate for the result to +/// keep the family's guarantee. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SelectionRelation { + /// No row in two inputs: the family counts a shared row twice. + Disjoint, + /// Inputs may share rows: the merge is idempotent. + OverlapAllowed, +} + +/// How a merged state's error relates to its parts. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MergedError { + /// The merged state is exact, as its parts are. + Exact, + /// Parts built with the same parameters merge into a state with the + /// guarantee of one build over the union of their rows: the error does + /// not grow with the number of parts, so a state sized for the whole + /// window may be built in parts. + SameAsUnionBuild, +} + +impl FamilyCapability { + /// The capability of `family`. Families not listed have no capability + /// (they fail closed). + pub fn of(family: &FieldDataType) -> Self { + use ExactKind as E; + use SketchAlgorithm as S; + let merge = |inputs, error| Some(MergeCapability { inputs, error }); + let (merge, subtractable) = match family { + // Rate/Increase/IRate read the samples at the window edges, so + // their parts do not compose into the window's value. + FieldDataType::ExactAggregate(kind, _) => match kind { + E::Sum | E::Count => (merge(SelectionRelation::Disjoint, MergedError::Exact), true), + E::Min | E::Max => ( + merge(SelectionRelation::OverlapAllowed, MergedError::Exact), + false, + ), + E::Rate | E::Increase | E::IRate => (None, false), + }, + // Hydra over a linear sketch is a shared grid of linear + // sketches: cells add and subtract. Hydra over KLL has no + // accuracy model. + FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => { + match kind { + HydraKind::HydraCms | HydraKind::HydraCountSketch => ( + merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), + true, + ), + HydraKind::HydraKll => (None, false), + } + } + FieldDataType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => { + match kind.algorithm() { + // Linear sketches: counters add and subtract. + S::Cms | S::CountSketch => ( + merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), + true, + ), + S::Kll | S::DDSketch | S::UnivMon => ( + merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), + false, + ), + S::Hll => ( + merge( + SelectionRelation::OverlapAllowed, + MergedError::SameAsUnionBuild, + ), + false, + ), + // Merged heap top-k states have no accuracy model; KMV and + // Theta are not offered. + S::CmsWithHeap | S::CountSketchWithHeap | S::Kmv | S::Theta => (None, false), + } + } + FieldDataType::Plain(_) + | FieldDataType::Sample(..) + | FieldDataType::Wavelet(..) + | FieldDataType::StatModel(..) => (None, false), + }; + Self { + merge, + subtractable, + } + } +} diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 7234d148b..93dba180e 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -19,11 +19,13 @@ //! `asap_aware_mapping::grouping`). It rides on `ASAPOp::SummaryAgg` and on //! sketch-valued edge types. //! -//! The rest: accuracy guarantees ([`guarantee`]), maintained populations, +//! The rest: what each family can do ([`family_capability`]), accuracy +//! guarantees ([`guarantee`]), maintained populations, //! summary-window panes, and the execution timing / //! data-state vocabulary ([`execution_data_state`]). pub mod execution_data_state; +pub mod family_capability; pub mod guarantee; pub mod maintained_population; pub mod query_time; @@ -35,6 +37,7 @@ pub use execution_data_state::{ lift_plain, DataPrimitive, ExactOperationSchemaError, ExecutionDataState, ExecutionDataStateError, ExecutionTiming, Materialization, Medium, }; +pub use family_capability::{FamilyCapability, MergeCapability, MergedError, SelectionRelation}; pub use guarantee::{ AccuracyError, AccuracyGuarantee, BoundExpr, CompositionOperator, ErrorMetric, GuaranteeSource, ProbabilityExpr, diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index d2b4194d9..47e010582 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -1,8 +1,13 @@ //! Window composition merges compatible summary states without consuming raw rows. use asap_types::{ ir::operator_properties::{Reduction, Source}, + ir::{apply_materialization_timings, MaterializationAssignment, TimingMemo}, ir::{ASAPOp, ExprSemantics, NonASAPOp, Operator, OperatorNode, Predicate, ScalarExpr}, - post_asap::{GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryUpdate}, + post_asap::{ + ExactKind, ExactParams, ExecutionTiming, FamilyCapability, GroupingStrategy, HydraKind, + HydraParams, Materialization, Medium, MergedError, SelectionRelation, SketchAlgorithm, + SketchKind, SketchParams, SummaryUpdate, + }, pre_asap::expr_ir::{CompareOpKind, ScalarValue}, pre_asap::{ColumnRef, DataType, Field, FieldDataType, Schema}, }; @@ -10,6 +15,22 @@ use std::rc::Rc; /// KLL over `value` for one `region`, so states of different regions are /// disjoint. fn state(k: u32, region: &str) -> Rc { + family_state( + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k, + failure_probability: None, + }, + ), + Default::default(), + ), + region, + ) +} +/// A `family` state over `value` for one `region`. +fn family_state(family: FieldDataType, region: &str) -> Rc { let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { source: Source::Table { table_ref: "latencies".into(), @@ -33,16 +54,7 @@ fn state(k: u32, region: &str) -> Rc { .unwrap(); OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child: only_region, - family: FieldDataType::Sketch( - SketchKind::new( - SketchAlgorithm::Kll, - SketchParams::Kll { - k, - failure_probability: None, - }, - ), - Default::default(), - ), + family, input: SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::by(vec![]), grouping: GroupingStrategy::default(), @@ -73,3 +85,280 @@ fn incompatible_merge_inputs_fail() { ); } } + +fn sketch(algorithm: SketchAlgorithm, params: SketchParams) -> FieldDataType { + FieldDataType::Sketch(SketchKind::new(algorithm, params), Default::default()) +} +fn hydra( + algorithm: SketchAlgorithm, + params: SketchParams, + kind: HydraKind, + hydra_params: HydraParams, +) -> FieldDataType { + FieldDataType::Sketch( + SketchKind::new(algorithm, params), + GroupingStrategy::SharedMultiSubpopulation { + kind, + params: hydra_params, + }, + ) +} +fn exact(kind: ExactKind, params: ExactParams) -> FieldDataType { + FieldDataType::ExactAggregate(kind, params) +} +fn kll() -> FieldDataType { + sketch( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ) +} +fn hll() -> FieldDataType { + sketch(SketchAlgorithm::Hll, SketchParams::Hll { precision: 12 }) +} +fn cms_with_heap() -> FieldDataType { + sketch( + SketchAlgorithm::CmsWithHeap, + SketchParams::CmsWithHeap { + width: 64, + depth: 4, + heap_size: 10, + }, + ) +} +fn merge(children: Vec>) -> Result, String> { + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children })) + .map_err(|e| e.to_string()) +} + +/// The family-capability table: which families merge, how their inputs' +/// selections must relate, how the merged error relates to the parts', and +/// which subtract. +#[test] +fn family_capability_table() { + use MergedError::{Exact, SameAsUnionBuild}; + use SelectionRelation::{Disjoint, OverlapAllowed}; + let (width, depth, heap_size) = (64, 4, 10); + for (family, merge, subtractable) in [ + ( + exact(ExactKind::Sum, ExactParams::Sum), + Some((Disjoint, Exact)), + true, + ), + ( + exact(ExactKind::Count, ExactParams::Count), + Some((Disjoint, Exact)), + true, + ), + ( + exact(ExactKind::Min, ExactParams::Min), + Some((OverlapAllowed, Exact)), + false, + ), + ( + exact(ExactKind::Max, ExactParams::Max), + Some((OverlapAllowed, Exact)), + false, + ), + (exact(ExactKind::Rate, ExactParams::Rate), None, false), + ( + exact(ExactKind::Increase, ExactParams::Increase), + None, + false, + ), + (exact(ExactKind::IRate, ExactParams::IRate), None, false), + (kll(), Some((Disjoint, SameAsUnionBuild)), false), + ( + sketch( + SketchAlgorithm::DDSketch, + SketchParams::DDSketch { alpha: 0.01 }, + ), + Some((Disjoint, SameAsUnionBuild)), + false, + ), + (hll(), Some((OverlapAllowed, SameAsUnionBuild)), false), + ( + sketch(SketchAlgorithm::Cms, SketchParams::Cms { width, depth }), + Some((Disjoint, SameAsUnionBuild)), + true, + ), + ( + sketch( + SketchAlgorithm::CountSketch, + SketchParams::CountSketch { width, depth }, + ), + Some((Disjoint, SameAsUnionBuild)), + true, + ), + ( + sketch( + SketchAlgorithm::UnivMon, + SketchParams::UnivMon { + heap_size, + sketch_rows: depth, + sketch_cols: width, + layers: 8, + }, + ), + Some((Disjoint, SameAsUnionBuild)), + false, + ), + (cms_with_heap(), None, false), + ( + sketch( + SketchAlgorithm::CountSketchWithHeap, + SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + }, + ), + None, + false, + ), + // Hydra over a linear sketch is a grid of linear sketches; Hydra + // over KLL has no accuracy model and fails closed. + ( + hydra( + SketchAlgorithm::Cms, + SketchParams::Cms { width, depth }, + HydraKind::HydraCms, + HydraParams::HydraCms { + width, + depth, + shared_rows: depth, + shared_columns: width, + }, + ), + Some((Disjoint, SameAsUnionBuild)), + true, + ), + ( + hydra( + SketchAlgorithm::CountSketch, + SketchParams::CountSketch { width, depth }, + HydraKind::HydraCountSketch, + HydraParams::HydraCountSketch { + width, + depth, + shared_rows: depth, + shared_columns: width, + }, + ), + Some((Disjoint, SameAsUnionBuild)), + true, + ), + ( + hydra( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + HydraKind::HydraKll, + HydraParams::HydraKll { + k: 200, + shared_buckets: width, + }, + ), + None, + false, + ), + (FieldDataType::Plain(DataType::Float64), None, false), + ] { + let capability = FamilyCapability::of(&family); + assert_eq!( + capability.merge.map(|m| (m.inputs, m.error)), + merge, + "{family:?}" + ); + assert_eq!(capability.subtractable, subtractable, "{family:?}"); + } +} + +/// The IR does not reject a merge by family: a heap state has no sound +/// merge, but only the table's consumers (the planner rules and the +/// executor) decide that. +#[test] +fn summary_merge_is_permissive_about_the_family() { + let merged = merge(vec![ + family_state(cms_with_heap(), "us"), + family_state(cms_with_heap(), "eu"), + ]) + .unwrap(); + merged.validate_structure().unwrap(); + assert_eq!(FamilyCapability::of(&cms_with_heap()).merge, None); +} + +/// Idempotent families (HLL, exact Min/Max) merge states that may share +/// rows, and the union of identical selections is that selection; counting +/// families (exact Sum, KLL) still need disjoint selections. +#[test] +fn idempotent_families_merge_overlapping_states() { + let same_region = |family: FieldDataType| { + merge(vec![ + family_state(family.clone(), "us"), + family_state(family, "us"), + ]) + }; + for family in [hll(), exact(ExactKind::Max, ExactParams::Max)] { + let merged = same_region(family.clone()).unwrap_or_else(|e| panic!("{family:?}: {e}")); + merged.validate_structure().unwrap(); + assert_eq!( + merged.coverage().unwrap().selection, + family_state(family, "us").coverage().unwrap().selection + ); + } + for family in [exact(ExactKind::Sum, ExactParams::Sum), kll()] { + let error = same_region(family).unwrap_err(); + assert!(error.contains("not proven disjoint"), "{error}"); + } +} + +/// `SummaryMerge` takes the timing of its consumer, and reads pane states +/// stored at ingestion time (#509 Example 3/4 Pattern B1) or built at query +/// time (B2). Merging at ingestion time needs every pane at ingestion time. +#[test] +fn summary_merge_timing() { + let panes = [state(200, "us"), state(200, "eu")]; + let root = merge(panes.to_vec()).unwrap(); + let timings = |assignment: &MaterializationAssignment| { + let timed = + apply_materialization_timings(&root, assignment, &mut TimingMemo::new()).unwrap(); + let children = timed.children(); + ( + timed.timing.unwrap(), + children + .iter() + .map(|c| c.timing.unwrap()) + .collect::>(), + ) + }; + use ExecutionTiming::{IngestionTime, QueryTime}; + assert_eq!( + timings(&MaterializationAssignment::default()), + (QueryTime, vec![QueryTime, QueryTime]) + ); + let at_ingestion = Materialization::MatAtIngestion(Medium::Memory); + let mut stored_panes = MaterializationAssignment::default(); + for pane in &panes { + stored_panes.set(pane, at_ingestion); + } + assert_eq!( + timings(&stored_panes), + (QueryTime, vec![IngestionTime, IngestionTime]) + ); + // A `SummaryAgg` without an assignment runs at query time, so a merge + // stored at ingestion time needs its panes there too. + let mut only_merge = MaterializationAssignment::default(); + only_merge.set(&root, at_ingestion); + assert!(apply_materialization_timings(&root, &only_merge, &mut TimingMemo::new()).is_err()); + let mut stored_merge = MaterializationAssignment::with_default(at_ingestion); + stored_merge.set(&root, at_ingestion); + assert_eq!( + timings(&stored_merge), + (IngestionTime, vec![IngestionTime, IngestionTime]) + ); +} From b7c939132140ea0e1aa48d1dfbf3c44ace763629 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:15:37 +0000 Subject: [PATCH 2/6] feat(physical-operators): merge tumbling panes of mergeable families A per-entity SummaryAgg reads its raw rows through one TimeShift, so a tumbling pane builds over its shifted window. A temporal SummaryMerge groups pane states by series and takes the run scope's timestamp, as a per-series SummaryAgg does. Both the query-time and the precompute lowering run a merge only when the family-capability table gives the family a sound merge. --- .../src/physical_planner/mod.rs | 40 +- .../src/physical_planner/precompute.rs | 1 + .../tests/precompute_population.rs | 19 +- .../tests/tumbling_pane_merge.rs | 378 ++++++++++++++++++ 4 files changed, 435 insertions(+), 3 deletions(-) create mode 100644 crates/integration-tests/tests/tumbling_pane_merge.rs diff --git a/crates/asap-physical-operators/src/physical_planner/mod.rs b/crates/asap-physical-operators/src/physical_planner/mod.rs index 7ba0074d9..a46340fcf 100644 --- a/crates/asap-physical-operators/src/physical_planner/mod.rs +++ b/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,24 @@ fn compile_internal( continue; } } + if let Payload::ASAP(ASAPOp::SummaryMerge { .. }) = &node.payload { + require_sound_merge(&schemas[0]).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 +535,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 { for schema in inputs { crate::values::validate_schema(schema)?; diff --git a/crates/asap-physical-operators/src/physical_planner/precompute.rs b/crates/asap-physical-operators/src/physical_planner/precompute.rs index ef2d00350..672ea9750 100644 --- a/crates/asap-physical-operators/src/physical_planner/precompute.rs +++ b/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())?, diff --git a/crates/asap-physical-operators/tests/precompute_population.rs b/crates/asap-physical-operators/tests/precompute_population.rs index dc69f45b8..dda368e15 100644 --- a/crates/asap-physical-operators/tests/precompute_population.rs +++ b/crates/asap-physical-operators/tests/precompute_population.rs @@ -260,6 +260,13 @@ fn state_dag( target: Option, merge: bool, ) -> CompiledPhysicalDAG { + try_state_dag(family, target, merge).unwrap() +} +fn try_state_dag( + family: FieldDataType, + target: Option, + merge: bool, +) -> Result { 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 { } // Only an explicit merge may collapse distinct pane updates before finalization. +#[test] +fn precompute_merge_requires_a_family_with_a_sound_merge() { + // Increase reads the samples at the window edges, so the + // family-capability table gives it no merge. + let increase = FieldDataType::ExactAggregate(ExactKind::Increase, ExactParams::Increase); + let error = try_state_dag(increase, 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); diff --git a/crates/integration-tests/tests/tumbling_pane_merge.rs b/crates/integration-tests/tests/tumbling_pane_merge.rs new file mode 100644 index 000000000..cfa04c15d --- /dev/null +++ b/crates/integration-tests/tests/tumbling_pane_merge.rs @@ -0,0 +1,378 @@ +//! Tumbling panes merged by `SummaryMerge` answer the same per-series query +//! as one build over the whole window (#509 Example 3/4, Pattern B). +//! The panes are written by hand here; the planner's window-composition +//! rule emits them later. +mod physical_common; + +use std::collections::BTreeMap; +use std::rc::Rc; +use std::sync::Arc; + +use asap_integration_tests::fixtures::lower_promql; +use asap_integration_tests::planning::selected_dag; +use asap_physical_operators::physical_planner::{ + compile, promql_rows, CompiledPhysicalDAG, InputContract, +}; +use asap_physical_operators::runtime::Scope; +use asap_physical_operators::values::{Batch, Value}; +use asap_types::ir::operator_properties::TimeShift; +use asap_types::ir::physical_export::{compile_physical_asap_dag_with_node_ids, PhysicalASAPDAG}; +use asap_types::ir::{ + apply_materialization_timings, ASAPOp, Materialization, MaterializationAssignment, Medium, + NonASAPOp, Operator, OperatorNode, TimingMemo, +}; +use asap_types::post_asap::{ExactKind, ExactParams, FieldDataType, Schema}; +use asap_types::types::AccuracyTarget; + +const EVALUATION_MS: i64 = 300_000; +const PANE_MS: i64 = 60_000; +const PANES: i64 = 5; + +/// The selected single-build plan for a 5-minute per-series PromQL query. +fn single_build(query: &str, accuracy: AccuracyTarget) -> Rc { + let root = lower_promql(query, accuracy.clone()).unwrap(); + let root = promql_rows::with_series_identity(&root).unwrap(); + selected_dag(root, accuracy) +} + +/// The per-entity `SummaryAgg` of `root` over its raw `TimeRange`. +fn summary_agg(root: &Rc) -> Rc { + OperatorNode::reachable(root) + .into_iter() + .find(|node| matches!(node.asap(), Some(ASAPOp::SummaryAgg { .. }))) + .expect("a summary build") +} + +/// Rewrite `root` so its summary is built in tumbling panes: pane `i` is +/// `TimeRange(1m)` over `TimeShift(i·1m)` over the same `Scan`, and a +/// `SummaryMerge` combines the panes, with `family` in place of the +/// selected one when given. Returns the root and the panes. +fn tumbling_panes( + root: &Rc, + family: Option, +) -> (Rc, Vec>) { + let agg = summary_agg(root); + let Some(NonASAPOp::TimeRange { kind, child, .. }) = agg.children()[0].non_asap() else { + panic!("expected a raw time range under the summary"); + }; + let panes: Vec<_> = (0..PANES) + .map(|i| { + let shifted = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift { + shift: TimeShift { + offset_ms: i * PANE_MS, + at: None, + }, + child: child.clone(), + })) + .unwrap(); + let range = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { + range: std::time::Duration::from_millis(PANE_MS as u64), + kind: *kind, + child: shifted, + })) + .unwrap(); + let mut operator = agg.operator.map_children(|_| range.clone()); + if let (Some(family), Operator::ASAP(ASAPOp::SummaryAgg { family: f, .. })) = + (&family, &mut operator) + { + *f = family.clone(); + } + OperatorNode::new_shared(operator).unwrap() + }) + .collect(); + let merged = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { + children: panes.clone(), + })) + .unwrap(); + let root = replace(root, &agg, &merged); + root.validate_structure().unwrap(); + (root, panes) +} + +/// `node` with `from` replaced by `to`. +fn replace( + node: &Rc, + from: &Rc, + to: &Rc, +) -> Rc { + if Rc::ptr_eq(node, from) { + return to.clone(); + } + Rc::new( + node.with_new_children(|child| replace(child, from, to)) + .unwrap() + .with_guarantee(node.guarantee.clone()), + ) +} + +/// Two series, four positive samples per minute in [0, 5m). +fn samples() -> Vec<(&'static str, i64, f64)> { + let mut rows = vec![]; + for (s, series) in ["a", "b"].into_iter().enumerate() { + for minute in 0..PANES { + for j in 0..4 { + let ts = minute * PANE_MS + j * 15_000; + rows.push((series, ts, (s * 100) as f64 + (minute * 4 + j + 1) as f64)); + } + } + } + rows +} + +struct Exported { + dag: PhysicalASAPDAG, + /// Each raw `TimeRange` input with the pane it reads (0 = most recent). + ranges: Vec<(u64, i64)>, + /// Each `SummaryAgg` node with its pane. + builds: Vec<(u64, i64)>, +} + +fn export(root: &Rc, assignment: &MaterializationAssignment) -> Exported { + let timed = apply_materialization_timings(root, assignment, &mut TimingMemo::new()).unwrap(); + let compiled = compile_physical_asap_dag_with_node_ids(&timed).unwrap(); + let pane = |node: &OperatorNode| -> i64 { + match node.children()[0].non_asap() { + Some(NonASAPOp::TimeShift { shift, .. }) => shift.offset_ms / PANE_MS, + _ => 0, + } + }; + let (mut ranges, mut builds) = (vec![], vec![]); + for node in &compiled.dag.nodes { + let logical = compiled.node_ids.operator_node(node.id).unwrap(); + match &logical.operator { + Operator::NonASAP(NonASAPOp::TimeRange { .. }) => { + ranges.push((node.id as u64, pane(logical))) + } + Operator::ASAP(ASAPOp::SummaryAgg { child, .. }) => { + builds.push((node.id as u64, pane(child))) + } + _ => {} + } + } + Exported { + dag: compiled.dag, + ranges, + builds, + } +} + +fn schema_of(dag: &PhysicalASAPDAG, id: u64) -> Arc { + Arc::new( + dag.nodes + .iter() + .find(|node| node.id as u64 == id) + .unwrap() + .output_schema + .clone(), + ) +} + +/// Raw rows for each `TimeRange` input. A deployment supplies pane `i` with +/// the samples in `[T - (i+1)·w, T - i·w)`; a single build reads all samples. +fn raw_inputs(exported: &Exported) -> BTreeMap { + exported + .ranges + .iter() + .map(|&(id, pane)| { + let schema = schema_of(&exported.dag, id); + let end = EVALUATION_MS - pane * PANE_MS; + let start = if exported.ranges.len() == 1 { + 0 + } else { + end - PANE_MS + }; + let rows = samples() + .into_iter() + .filter(|(_, ts, _)| (start..end).contains(ts)) + .map(|(series, ts, value)| { + let labels = BTreeMap::from([ + ("__name__".to_string(), "m".to_string()), + ("series".to_string(), series.to_string()), + ]); + promql_rows::series_row(&schema, &labels, ts, value).unwrap() + }) + .collect(); + (id, Batch::try_new(schema, rows).unwrap()) + }) + .collect() +} + +fn run(plan: &CompiledPhysicalDAG, inputs: BTreeMap, scope: Scope) -> Vec> { + let mut rows: Vec> = physical_common::execute(plan, inputs, scope) + .remove(0) + .iter() + .flat_map(|batch| batch.rows().to_vec()) + .collect(); + rows.sort_by_key(|row| format!("{row:?}")); + rows +} + +/// `Value` has no equality; its debug form identifies every scalar exactly. +fn render(rows: &[Vec]) -> Vec { + rows.iter().map(|row| format!("{row:?}")).collect() +} + +fn has_timestamp(row: &[Value], ms: i64) -> bool { + row.iter() + .any(|v| matches!(v, Value::Timestamp(t) if *t == ms)) +} + +fn query_scope() -> Scope { + Scope::Query { + evaluation_time_ms: EVALUATION_MS, + revision: 1, + } +} + +fn contracts(inputs: &BTreeMap) -> BTreeMap { + inputs + .iter() + .map(|(&id, batch)| (id, InputContract::bounded(batch.schema().clone()))) + .collect() +} + +/// Compile `root` with everything at query time (Pattern B2: panes rebuilt +/// from raw rows at query time). +fn compile_at_query_time(root: &Rc) -> Result { + let exported = export(root, &MaterializationAssignment::default()); + let inputs = raw_inputs(&exported); + compile( + &exported.dag, + contracts(&inputs), + &[exported.dag.roots[0] as u64], + ) + .map_err(|e| e.to_string())?; + Ok(exported) +} + +fn execute(root: &Rc) -> Vec> { + let exported = compile_at_query_time(root).unwrap(); + let inputs = raw_inputs(&exported); + let plan = compile( + &exported.dag, + contracts(&inputs), + &[exported.dag.roots[0] as u64], + ) + .unwrap(); + run(&plan, inputs, query_scope()) +} + +/// Build the panes in ingestion runs (stored `MatAtIngestion`, each with its +/// own build timestamp), then merge them in the query run (Pattern B1). +fn execute_retained(root: &Rc, panes: &[Rc]) -> Vec> { + let mut assignment = MaterializationAssignment::default(); + for pane in panes { + assignment.set(pane, Materialization::MatAtIngestion(Medium::Memory)); + } + let exported = export(root, &assignment); + let inputs = raw_inputs(&exported); + let contracts = contracts(&inputs); + let mut retained = BTreeMap::new(); + for &(build, pane) in &exported.builds { + let range = exported.ranges.iter().find(|r| r.1 == pane).unwrap().0; + let plan = compile( + &exported.dag, + BTreeMap::from([(range, contracts[&range].clone())]), + &[build], + ) + .unwrap(); + let end = EVALUATION_MS - pane * PANE_MS; + let rows = run( + &plan, + BTreeMap::from([(range, inputs[&range].clone())]), + Scope::Ingestion { + window_start_ms: end - PANE_MS, + window_end_ms: end, + revision: 1, + }, + ); + assert!(rows.iter().all(|row| has_timestamp(row, end))); + retained.insert( + build, + Batch::try_new(schema_of(&exported.dag, build), rows).unwrap(), + ); + } + let plan = compile( + &exported.dag, + retained + .iter() + .map(|(&id, batch)| (id, InputContract::bounded(batch.schema().clone()))) + .collect(), + &[exported.dag.roots[0] as u64], + ) + .unwrap(); + run(&plan, retained, query_scope()) +} + +fn values(rows: &[Vec]) -> Vec { + rows.iter() + .flat_map(|row| { + row.iter().filter_map(|v| match v { + Value::Float64(v) => Some(*v), + _ => None, + }) + }) + .collect() +} + +/// Returns the per-series values, which every execution path agrees on. +fn assert_pane_merge_matches_single_build(query: &str, accuracy: AccuracyTarget) -> Vec { + let single = single_build(query, accuracy); + let (panes, pane_builds) = tumbling_panes(&single, None); + let expected = execute(&single); + assert_eq!(expected.len(), 2, "{expected:?}"); + assert!(expected.iter().all(|row| has_timestamp(row, EVALUATION_MS))); + let result = values(&expected); + let expected = render(&expected); + assert_eq!( + render(&execute(&panes)), + expected, + "panes rebuilt at query time" + ); + assert_eq!( + render(&execute_retained(&panes, &pane_builds)), + expected, + "retained panes" + ); + result +} + +/// Exact per-series Sum over 5 merged 1-minute panes equals one 5-minute +/// build, timestamped at the evaluation time. +#[test] +fn exact_sum_over_tumbling_panes_matches_single_build() { + let sums = + assert_pane_merge_matches_single_build("sum_over_time(m[5m])", AccuracyTarget::Exact); + assert_eq!(sums, vec![210., 2210.]); +} + +/// The selected quantile sketch over 5 merged 1-minute panes equals one +/// 5-minute build, within the target of the exact p99 (20 and 120). +#[test] +fn quantile_over_tumbling_panes_matches_single_build() { + let p99 = assert_pane_merge_matches_single_build( + "quantile_over_time(0.99, m[5m])", + AccuracyTarget::Epsilon(0.05), + ); + let mut p99 = p99; + p99.sort_by(f64::total_cmp); + assert_eq!(p99.len(), 2); + for (estimate, exact) in p99.into_iter().zip([20., 120.]) { + assert!( + (estimate - exact).abs() <= 0.05 * exact, + "{estimate} vs {exact}" + ); + } +} + +/// The executor consults the family-capability table: panes of a family +/// with no sound merge (here exact Increase, which reads the window edges) +/// do not compile, although the IR accepts the merge. +#[test] +fn panes_of_a_family_without_a_sound_merge_do_not_compile() { + let single = single_build("sum_over_time(m[5m])", AccuracyTarget::Exact); + let increase = FieldDataType::ExactAggregate(ExactKind::Increase, ExactParams::Increase); + let (panes, _) = tumbling_panes(&single, Some(increase)); + let error = compile_at_query_time(&panes).err().expect("rejected"); + assert!(error.contains("no sound merge"), "{error}"); +} From 0a5292755d2f7162621bb2f07c8cfba6680eefc9 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 23:04:09 +0000 Subject: [PATCH 3/6] feat(types): per-readout merged error; KMV merges A merged UnivMon keeps the guarantee of readouts from its additive layer-0 counters (L2, point counts), while readouts from its per-layer heaps (distinct count, entropy, top-k) are uncertified after a merge. MergeCapability records the heap readouts' merged error, and FamilyCapability::merged_error answers it per readout. KMV merges with the guarantee of one build over the union: asap_sketchlib's KMV hashes with one shared seed and KMV::merge keeps the k smallest distinct hashes, so overlapping inputs are allowed. --- .../types/src/post_asap/family_capability.rs | 60 +++++++++++++++++-- crates/types/tests/summary_merge_structure.rs | 57 +++++++++++++++++- 2 files changed, 110 insertions(+), 7 deletions(-) diff --git a/crates/types/src/post_asap/family_capability.rs b/crates/types/src/post_asap/family_capability.rs index e082cb05b..81f8ab751 100644 --- a/crates/types/src/post_asap/family_capability.rs +++ b/crates/types/src/post_asap/family_capability.rs @@ -7,7 +7,7 @@ //! the mergeability facts; the summary-capability facts (which statistics //! a family answers) are added as further fields of [`FamilyCapability`]. -use super::sketch::{ExactKind, GroupingStrategy, HydraKind, SketchAlgorithm}; +use super::sketch::{ExactKind, GroupingStrategy, HydraKind, SketchAlgorithm, SketchStatistic}; use crate::pre_asap::schema::FieldDataType; /// What one summary family can do. @@ -29,6 +29,10 @@ pub struct MergeCapability { pub inputs: SelectionRelation, /// How the merged state's error relates to its parts' errors. pub error: MergedError, + /// The merged error of readouts that read the family's heavy-hitter + /// heaps (distinct count, entropy, top-k), when it differs from + /// `error`: each part's heaps keep only that part's heavy items. + pub heap_readouts: Option, } /// How the selections of a merge's inputs must relate for the result to @@ -51,6 +55,8 @@ pub enum MergedError { /// not grow with the number of parts, so a state sized for the whole /// window may be built in parts. SameAsUnionBuild, + /// The merged state's readout has no guarantee; it is evidence only. + Uncertified, } impl FamilyCapability { @@ -59,7 +65,13 @@ impl FamilyCapability { pub fn of(family: &FieldDataType) -> Self { use ExactKind as E; use SketchAlgorithm as S; - let merge = |inputs, error| Some(MergeCapability { inputs, error }); + let merge = |inputs, error| { + Some(MergeCapability { + inputs, + error, + heap_readouts: None, + }) + }; let (merge, subtractable) = match family { // Rate/Increase/IRate read the samples at the window edges, so // their parts do not compose into the window's value. @@ -90,10 +102,31 @@ impl FamilyCapability { merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), true, ), - S::Kll | S::DDSketch | S::UnivMon => ( + S::Kll | S::DDSketch => ( merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), false, ), + // Layer-0 counters add exactly, so L2 and point counts + // keep their guarantee; the per-layer heaps do not. + S::UnivMon => ( + Some(MergeCapability { + inputs: SelectionRelation::Disjoint, + error: MergedError::SameAsUnionBuild, + heap_readouts: Some(MergedError::Uncertified), + }), + false, + ), + // `asap_sketchlib`'s KMV hashes every item with one + // shared seed (`CANONICAL_HASH_SEED`), and `KMV::merge` + // keeps the k smallest distinct hashes: the state of + // the union. + S::Kmv => ( + merge( + SelectionRelation::OverlapAllowed, + MergedError::SameAsUnionBuild, + ), + false, + ), S::Hll => ( merge( SelectionRelation::OverlapAllowed, @@ -101,9 +134,9 @@ impl FamilyCapability { ), false, ), - // Merged heap top-k states have no accuracy model; KMV and - // Theta are not offered. - S::CmsWithHeap | S::CountSketchWithHeap | S::Kmv | S::Theta => (None, false), + // Merged heap top-k states have no accuracy model; + // Theta is not offered. + S::CmsWithHeap | S::CountSketchWithHeap | S::Theta => (None, false), } } FieldDataType::Plain(_) @@ -116,4 +149,19 @@ impl FamilyCapability { subtractable, } } + + /// The merged error of `readout` from a merged state of this family; + /// `None` when the family has no sound merge. + pub fn merged_error(&self, readout: &SketchStatistic) -> Option { + let merge = self.merge?; + Some(match (merge.heap_readouts, readout) { + ( + Some(heap), + SketchStatistic::Cardinality + | SketchStatistic::FrequencyEntropy + | SketchStatistic::TopK { .. }, + ) => heap, + _ => merge.error, + }) + } } diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index 47e010582..27625050a 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -6,7 +6,7 @@ use asap_types::{ post_asap::{ ExactKind, ExactParams, ExecutionTiming, FamilyCapability, GroupingStrategy, HydraKind, HydraParams, Materialization, Medium, MergedError, SelectionRelation, SketchAlgorithm, - SketchKind, SketchParams, SummaryUpdate, + SketchKind, SketchParams, SketchStatistic, SummaryUpdate, }, pre_asap::expr_ir::{CompareOpKind, ScalarValue}, pre_asap::{ColumnRef, DataType, Field, FieldDataType, Schema}, @@ -205,6 +205,18 @@ fn family_capability_table() { Some((Disjoint, SameAsUnionBuild)), false, ), + // KMV keeps the k smallest hashes under one shared seed: the merge + // is the k smallest of the union and ignores repeated hashes. + ( + sketch(SketchAlgorithm::Kmv, SketchParams::Kmv { k: 1024 }), + Some((OverlapAllowed, SameAsUnionBuild)), + false, + ), + ( + sketch(SketchAlgorithm::Theta, SketchParams::Theta { k: 1024 }), + None, + false, + ), (cms_with_heap(), None, false), ( sketch( @@ -278,6 +290,49 @@ fn family_capability_table() { } } +/// A merged UnivMon keeps the guarantee of readouts from its additive +/// layer-0 counters (L2, point counts); readouts from its per-layer heaps +/// (distinct count, entropy, top-k) are uncertified after a merge. Other +/// families' readouts share the family's merged error. +#[test] +fn merged_error_depends_on_the_readout() { + use SketchStatistic::{Cardinality, FrequencyEntropy, FrequencyL2, TopK}; + let univmon = FamilyCapability::of(&sketch( + SketchAlgorithm::UnivMon, + SketchParams::UnivMon { + heap_size: 10, + sketch_rows: 4, + sketch_cols: 64, + layers: 8, + }, + )); + let point = SketchStatistic::PointCount { + key: ColumnRef::Named("k".into()), + value: None, + }; + for (readout, error) in [ + (FrequencyL2, MergedError::SameAsUnionBuild), + (point, MergedError::SameAsUnionBuild), + (Cardinality, MergedError::Uncertified), + (FrequencyEntropy, MergedError::Uncertified), + (TopK { k: 5 }, MergedError::Uncertified), + ] { + assert_eq!(univmon.merged_error(&readout), Some(error), "{readout:?}"); + } + assert_eq!( + FamilyCapability::of(&hll()).merged_error(&Cardinality), + Some(MergedError::SameAsUnionBuild) + ); + assert_eq!( + FamilyCapability::of(&kll()).merged_error(&SketchStatistic::Quantile { q: 0.99 }), + Some(MergedError::SameAsUnionBuild) + ); + assert_eq!( + FamilyCapability::of(&cms_with_heap()).merged_error(&TopK { k: 5 }), + None + ); +} + /// The IR does not reject a merge by family: a heap state has no sound /// merge, but only the table's consumers (the planner rules and the /// executor) decide that. From e7c6860fa196e92df51a1c70d9a8c493d19593a7 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 23:40:10 +0000 Subject: [PATCH 4/6] fix(types): heap merges are approximate; counter panes merge disjoint in time Add MergedError::Approximate: the merge is allowed but the merged readout is evidence only. Count-Min and Count-Sketch with heap now merge, with approximate top-k and point counts that keep their guarantee. At the pinned asap_sketchlib, UnivMon::merge rebuilds each layer's heap from the parts' heap keys and L2 and distinct count read those heaps, so every UnivMon readout is approximate after a merge; a TODO marks where the pin move past upstream #221 restores L2. Rate and Increase merge exactly (IncreaseAccumulator::merge_pair joins time-ordered windows with reset handling) when their inputs are disjoint in time: SelectionRelation::DisjointInTime, checked by coverage as CoverageError::NotDisjointInTime. IRate stays without a merge. The UnimplementedOperator doc no longer lists SummaryMerge. --- crates/types/src/ir/summary_coverage.rs | 57 ++++++---- .../src/post_asap/execution_data_state.rs | 5 +- .../types/src/post_asap/family_capability.rs | 103 +++++++++--------- crates/types/tests/summary_coverage.rs | 53 ++++++++- crates/types/tests/summary_merge_structure.rs | 78 ++++++++----- 5 files changed, 188 insertions(+), 108 deletions(-) diff --git a/crates/types/src/ir/summary_coverage.rs b/crates/types/src/ir/summary_coverage.rs index ab5e90f90..6bb9c9080 100644 --- a/crates/types/src/ir/summary_coverage.rs +++ b/crates/types/src/ir/summary_coverage.rs @@ -76,6 +76,8 @@ pub enum CoverageError { "summary merge inputs are not proven disjoint and the family counts shared rows twice" )] PossibleOverlap, + #[error("summary merge inputs are not proven disjoint in time, which counter states need")] + NotDisjointInTime, } impl SummaryCoverage { @@ -83,7 +85,8 @@ impl SummaryCoverage { /// every input has the same definition and their selections are /// pairwise disjoint; a family whose merge is idempotent /// ([`SelectionRelation::OverlapAllowed`] in the family-capability - /// table) may also merge overlapping selections. + /// table) may also merge overlapping selections, and a counter family + /// ([`SelectionRelation::DisjointInTime`]) needs disjoint time windows. pub fn derive(node: &OperatorNode) -> Result { match node.asap() { Some(ASAPOp::SummaryAgg { .. }) => Ok(of_summary_agg(node)), @@ -93,26 +96,37 @@ impl SummaryCoverage { .map(|child| child.coverage().ok_or(CoverageError::NotSummary)) .collect::, _>>()?; let first = inputs.first().ok_or(CoverageError::EmptyMerge)?; - let may_overlap = matches!( - first.definition.asap(), - Some(ASAPOp::SummaryAgg { family, .. }) - if FamilyCapability::of(family).merge.map(|m| m.inputs) - == Some(SelectionRelation::OverlapAllowed) - ); + let relation = match first.definition.asap() { + Some(ASAPOp::SummaryAgg { family, .. }) => { + FamilyCapability::of(family).merge.map(|m| m.inputs) + } + _ => None, + }; let mut proven = HashSet::new(); for (i, input) in inputs.iter().enumerate() { if !same_definition(&first.definition, &input.definition, &mut proven) { return Err(CoverageError::DefinitionMismatch); } - let overlaps = !may_overlap - && inputs[..i].iter().any(|other| { + let any_pair = |separated: fn(&SelectionBox, &SelectionBox) -> bool| { + inputs[..i].iter().any(|other| { input .selection .iter() - .any(|a| other.selection.iter().any(|b| !a.disjoint(b))) - }); - if overlaps { - return Err(CoverageError::PossibleOverlap); + .any(|a| other.selection.iter().any(|b| !separated(a, b))) + }) + }; + match relation { + Some(SelectionRelation::OverlapAllowed) => {} + Some(SelectionRelation::DisjointInTime) => { + if any_pair(SelectionBox::disjoint_in_time) { + return Err(CoverageError::NotDisjointInTime); + } + } + Some(SelectionRelation::Disjoint) | None => { + if any_pair(SelectionBox::disjoint) { + return Err(CoverageError::PossibleOverlap); + } + } } } Ok(Self { @@ -648,16 +662,21 @@ impl SelectionBox { /// Whether no row lies in both boxes: some column, or the time window, /// is restricted by both and the restrictions are disjoint. fn disjoint(&self, other: &Self) -> bool { - let time = match (&self.relative_time, &other.relative_time) { + self.disjoint_in_time(other) + || self + .columns + .iter() + .any(|(column, a)| other.columns.get(column).is_some_and(|b| a.disjoint(b))) + } + + /// Whether the boxes' time windows do not overlap. + fn disjoint_in_time(&self, other: &Self) -> bool { + match (&self.relative_time, &other.relative_time) { (Some((al, au)), Some((bl, bu))) => { intervals_disjoint((al, au), (bl, bu), |a: &i64, b: &i64| Some(a.cmp(b))) } _ => false, - }; - time || self - .columns - .iter() - .any(|(column, a)| other.columns.get(column).is_some_and(|b| a.disjoint(b))) + } } } diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index 7e73c9752..3a5e13b7c 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -185,9 +185,8 @@ pub enum ExecutionDataStateError { #[error("exact operator consumes non-plain column {column:?} ({dtype})")] NonPlainOperand { column: String, dtype: String }, /// An ASAP operator with no execution timing contract yet in an - /// executable plan: `SummaryMerge` (schema and coverage are implemented) - /// and the reserved `SummarySubtract`, `SummaryDelete`, `SummaryJoin` - /// and `Extension`. + /// executable plan: the reserved `SummarySubtract`, `SummaryDelete`, + /// `SummaryJoin` and `Extension`. #[error("{operator} has no execution timing contract yet")] UnimplementedOperator { operator: &'static str }, /// A node assigned `MatAtIngestion` over a sub-DAG that can only run at diff --git a/crates/types/src/post_asap/family_capability.rs b/crates/types/src/post_asap/family_capability.rs index 81f8ab751..5142a9c78 100644 --- a/crates/types/src/post_asap/family_capability.rs +++ b/crates/types/src/post_asap/family_capability.rs @@ -31,7 +31,7 @@ pub struct MergeCapability { pub error: MergedError, /// The merged error of readouts that read the family's heavy-hitter /// heaps (distinct count, entropy, top-k), when it differs from - /// `error`: each part's heaps keep only that part's heavy items. + /// `error`. pub heap_readouts: Option, } @@ -41,6 +41,10 @@ pub struct MergeCapability { pub enum SelectionRelation { /// No row in two inputs: the family counts a shared row twice. Disjoint, + /// The inputs' time windows do not overlap. Counter states (Rate, + /// Increase) join consecutive windows in time order, so inputs that are + /// disjoint only by a value predicate would interleave in time. + DisjointInTime, /// Inputs may share rows: the merge is idempotent. OverlapAllowed, } @@ -55,8 +59,9 @@ pub enum MergedError { /// not grow with the number of parts, so a state sized for the whole /// window may be built in parts. SameAsUnionBuild, - /// The merged state's readout has no guarantee; it is evidence only. - Uncertified, + /// The merge is allowed, but the merged readout has no guarantee; it is + /// evidence only, for plan selection to judge. + Approximate, } impl FamilyCapability { @@ -64,6 +69,7 @@ impl FamilyCapability { /// (they fail closed). pub fn of(family: &FieldDataType) -> Self { use ExactKind as E; + use SelectionRelation::{Disjoint, DisjointInTime, OverlapAllowed}; use SketchAlgorithm as S; let merge = |inputs, error| { Some(MergeCapability { @@ -72,71 +78,68 @@ impl FamilyCapability { heap_readouts: None, }) }; + // A heavy-hitter heap keeps only the heavy items of its own part: + // an item heavy over the window but in no single part is lost. + let with_heaps = |error| { + Some(MergeCapability { + inputs: Disjoint, + error, + heap_readouts: Some(MergedError::Approximate), + }) + }; let (merge, subtractable) = match family { - // Rate/Increase/IRate read the samples at the window edges, so - // their parts do not compose into the window's value. FieldDataType::ExactAggregate(kind, _) => match kind { - E::Sum | E::Count => (merge(SelectionRelation::Disjoint, MergedError::Exact), true), - E::Min | E::Max => ( - merge(SelectionRelation::OverlapAllowed, MergedError::Exact), - false, - ), - E::Rate | E::Increase | E::IRate => (None, false), + E::Sum | E::Count => (merge(Disjoint, MergedError::Exact), true), + E::Min | E::Max => (merge(OverlapAllowed, MergedError::Exact), false), + // The counter accumulator joins time-ordered windows with + // reset handling (`IncreaseAccumulator::merge_pair`). + E::Rate | E::Increase => (merge(DisjointInTime, MergedError::Exact), false), + // IRate reads the last two samples of the window, which a + // pane does not keep across its boundary. + E::IRate => (None, false), }, // Hydra over a linear sketch is a shared grid of linear // sketches: cells add and subtract. Hydra over KLL has no // accuracy model. FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => { match kind { - HydraKind::HydraCms | HydraKind::HydraCountSketch => ( - merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), - true, - ), + HydraKind::HydraCms | HydraKind::HydraCountSketch => { + (merge(Disjoint, MergedError::SameAsUnionBuild), true) + } HydraKind::HydraKll => (None, false), } } FieldDataType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => { match kind.algorithm() { // Linear sketches: counters add and subtract. - S::Cms | S::CountSketch => ( - merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), - true, - ), - S::Kll | S::DDSketch => ( - merge(SelectionRelation::Disjoint, MergedError::SameAsUnionBuild), - false, - ), - // Layer-0 counters add exactly, so L2 and point counts - // keep their guarantee; the per-layer heaps do not. - S::UnivMon => ( - Some(MergeCapability { - inputs: SelectionRelation::Disjoint, - error: MergedError::SameAsUnionBuild, - heap_readouts: Some(MergedError::Uncertified), - }), - false, - ), + S::Cms | S::CountSketch => { + (merge(Disjoint, MergedError::SameAsUnionBuild), true) + } + S::Kll | S::DDSketch => (merge(Disjoint, MergedError::SameAsUnionBuild), false), + // Point counts read the linear counters; top-k reads the + // heap. + S::CmsWithHeap | S::CountSketchWithHeap => { + (with_heaps(MergedError::SameAsUnionBuild), false) + } + // At the pinned `asap_sketchlib`, `UnivMon::merge` + // rebuilds each layer's heap from the parts' heap keys, + // and L2 and distinct count read those heaps, so every + // readout is approximate after a merge. + // TODO(univmon-certification): when the pin moves past + // upstream asap_sketchlib #221, L2 reads layer 0's + // Count-Sketch directly: set `error` to + // `SameAsUnionBuild` (and revisit distinct count); + // entropy stays approximate. + S::UnivMon => (with_heaps(MergedError::Approximate), false), // `asap_sketchlib`'s KMV hashes every item with one // shared seed (`CANONICAL_HASH_SEED`), and `KMV::merge` // keeps the k smallest distinct hashes: the state of - // the union. - S::Kmv => ( - merge( - SelectionRelation::OverlapAllowed, - MergedError::SameAsUnionBuild, - ), - false, - ), - S::Hll => ( - merge( - SelectionRelation::OverlapAllowed, - MergedError::SameAsUnionBuild, - ), - false, - ), - // Merged heap top-k states have no accuracy model; + // the union. The executor has no KMV kernel yet (#697). + S::Kmv | S::Hll => { + (merge(OverlapAllowed, MergedError::SameAsUnionBuild), false) + } // Theta is not offered. - S::CmsWithHeap | S::CountSketchWithHeap | S::Theta => (None, false), + S::Theta => (None, false), } } FieldDataType::Plain(_) diff --git a/crates/types/tests/summary_coverage.rs b/crates/types/tests/summary_coverage.rs index 4d489edb2..37fed3a0a 100644 --- a/crates/types/tests/summary_coverage.rs +++ b/crates/types/tests/summary_coverage.rs @@ -12,7 +12,8 @@ use asap_types::ir::{ SchemaDerivationError, TimeRangeKind, }; use asap_types::post_asap::{ - ExecutionTiming, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryUpdate, + ExactKind, ExactParams, ExecutionTiming, GroupingStrategy, SketchAlgorithm, SketchKind, + SketchParams, SummaryUpdate, }; use asap_types::pre_asap::expr_ir::{ArithmeticOpKind, CompareOpKind, ScalarValue}; use asap_types::pre_asap::{ColumnRef, DataType, Field, FieldDataType, Schema, TimeShift}; @@ -81,9 +82,10 @@ fn filter(child: Rc, pred: ScalarExpr) -> Rc { } fn kll(child: Rc, column: &str) -> Rc { - OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + summary( child, - family: FieldDataType::Sketch( + column, + FieldDataType::Sketch( SketchKind::new( SketchAlgorithm::Kll, SketchParams::Kll { @@ -93,6 +95,13 @@ fn kll(child: Rc, column: &str) -> Rc { ), Default::default(), ), + ) +} + +fn summary(child: Rc, column: &str, family: FieldDataType) -> Rc { + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + child, + family, input: SummaryUpdate::column(ColumnRef::Named(column.into())), reduction: Reduction::by(vec![JOB]), grouping: GroupingStrategy::default(), @@ -108,6 +117,19 @@ fn latency_where(pred: ScalarExpr) -> Rc { /// Tumbling pane: `TimeRange(1m)` over `TimeShift(shift)` over `m`. fn pane(shift_ms: i64, predicates: Vec) -> Rc { + kll(pane_rows(shift_ms, predicates), "value") +} + +/// Exact counter-increase pane over the same rows as [`pane`]. +fn increase_pane(shift_ms: i64, predicates: Vec) -> Rc { + summary( + pane_rows(shift_ms, predicates), + "value", + FieldDataType::ExactAggregate(ExactKind::Increase, ExactParams::Increase), + ) +} + +fn pane_rows(shift_ms: i64, predicates: Vec) -> Rc { let shifted = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift { shift: TimeShift { offset_ms: shift_ms, @@ -116,13 +138,12 @@ fn pane(shift_ms: i64, predicates: Vec) -> Rc { child: series(predicates), })) .unwrap(); - let range = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { range: Duration::from_millis(60_000), kind: TimeRangeKind::Range, child: shifted, })) - .unwrap(); - kll(range, "value") + .unwrap() } fn merge(children: Vec>) -> Result, SchemaDerivationError> { @@ -280,6 +301,26 @@ fn time_panes_merge_and_keep_gaps() { ); } +/// Counter panes merge only when their time windows are disjoint: states +/// disjoint by a label but over the same window would interleave in time. +#[test] +fn counter_panes_merge_only_disjoint_in_time() { + let joined = merge(vec![ + increase_pane(0, vec![]), + increase_pane(60_000, vec![]), + ]) + .unwrap(); + assert_eq!( + joined.coverage().unwrap().selection[0].relative_time, + relative(-120_000, 0) + ); + let by_job = |job| increase_pane(0, vec![Predicate(eq(2, job))]); + assert_eq!( + rejected(vec![by_job("api"), by_job("web")]), + CoverageError::NotDisjointInTime + ); +} + /// Label matchers on the scan lift like any filter. #[test] fn scan_predicates_lift() { diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index 27625050a..40d568174 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -138,8 +138,8 @@ fn merge(children: Vec>) -> Result, String> { /// which subtract. #[test] fn family_capability_table() { - use MergedError::{Exact, SameAsUnionBuild}; - use SelectionRelation::{Disjoint, OverlapAllowed}; + use MergedError::{Approximate, Exact, SameAsUnionBuild}; + use SelectionRelation::{Disjoint, DisjointInTime, OverlapAllowed}; let (width, depth, heap_size) = (64, 4, 10); for (family, merge, subtractable) in [ ( @@ -162,10 +162,15 @@ fn family_capability_table() { Some((OverlapAllowed, Exact)), false, ), - (exact(ExactKind::Rate, ExactParams::Rate), None, false), + // Counter panes merge exactly when ordered and disjoint in time. + ( + exact(ExactKind::Rate, ExactParams::Rate), + Some((DisjointInTime, Exact)), + false, + ), ( exact(ExactKind::Increase, ExactParams::Increase), - None, + Some((DisjointInTime, Exact)), false, ), (exact(ExactKind::IRate, ExactParams::IRate), None, false), @@ -202,7 +207,7 @@ fn family_capability_table() { layers: 8, }, ), - Some((Disjoint, SameAsUnionBuild)), + Some((Disjoint, Approximate)), false, ), // KMV keeps the k smallest hashes under one shared seed: the merge @@ -217,7 +222,7 @@ fn family_capability_table() { None, false, ), - (cms_with_heap(), None, false), + (cms_with_heap(), Some((Disjoint, SameAsUnionBuild)), false), ( sketch( SketchAlgorithm::CountSketchWithHeap, @@ -227,7 +232,7 @@ fn family_capability_table() { heap_size, }, ), - None, + Some((Disjoint, SameAsUnionBuild)), false, ), // Hydra over a linear sketch is a grid of linear sketches; Hydra @@ -290,13 +295,18 @@ fn family_capability_table() { } } -/// A merged UnivMon keeps the guarantee of readouts from its additive -/// layer-0 counters (L2, point counts); readouts from its per-layer heaps -/// (distinct count, entropy, top-k) are uncertified after a merge. Other -/// families' readouts share the family's merged error. +/// Heap readouts (distinct count, entropy, top-k) of a merged state are +/// approximate: each part's heaps keep only that part's heavy items. At the +/// pinned `asap_sketchlib`, UnivMon's L2 and distinct count also read the +/// merged heaps, so every UnivMon readout is approximate after a merge. +/// Counter readouts of a heap family (point counts) keep their guarantee. #[test] fn merged_error_depends_on_the_readout() { use SketchStatistic::{Cardinality, FrequencyEntropy, FrequencyL2, TopK}; + let point = || SketchStatistic::PointCount { + key: ColumnRef::Named("k".into()), + value: None, + }; let univmon = FamilyCapability::of(&sketch( SketchAlgorithm::UnivMon, SketchParams::UnivMon { @@ -306,45 +316,53 @@ fn merged_error_depends_on_the_readout() { layers: 8, }, )); - let point = SketchStatistic::PointCount { - key: ColumnRef::Named("k".into()), - value: None, - }; - for (readout, error) in [ - (FrequencyL2, MergedError::SameAsUnionBuild), - (point, MergedError::SameAsUnionBuild), - (Cardinality, MergedError::Uncertified), - (FrequencyEntropy, MergedError::Uncertified), - (TopK { k: 5 }, MergedError::Uncertified), + for readout in [ + FrequencyL2, + point(), + Cardinality, + FrequencyEntropy, + TopK { k: 5 }, ] { - assert_eq!(univmon.merged_error(&readout), Some(error), "{readout:?}"); + assert_eq!( + univmon.merged_error(&readout), + Some(MergedError::Approximate), + "{readout:?}" + ); } + let heap = FamilyCapability::of(&cms_with_heap()); assert_eq!( - FamilyCapability::of(&hll()).merged_error(&Cardinality), + heap.merged_error(&TopK { k: 5 }), + Some(MergedError::Approximate) + ); + assert_eq!( + heap.merged_error(&point()), Some(MergedError::SameAsUnionBuild) ); assert_eq!( - FamilyCapability::of(&kll()).merged_error(&SketchStatistic::Quantile { q: 0.99 }), + FamilyCapability::of(&hll()).merged_error(&Cardinality), Some(MergedError::SameAsUnionBuild) ); assert_eq!( - FamilyCapability::of(&cms_with_heap()).merged_error(&TopK { k: 5 }), - None + FamilyCapability::of(&kll()).merged_error(&SketchStatistic::Quantile { q: 0.99 }), + Some(MergedError::SameAsUnionBuild) ); + let irate = exact(ExactKind::IRate, ExactParams::IRate); + assert_eq!(FamilyCapability::of(&irate).merged_error(&point()), None); } -/// The IR does not reject a merge by family: a heap state has no sound +/// The IR does not reject a merge by family: exact IRate has no sound /// merge, but only the table's consumers (the planner rules and the /// executor) decide that. #[test] fn summary_merge_is_permissive_about_the_family() { + let irate = || exact(ExactKind::IRate, ExactParams::IRate); let merged = merge(vec![ - family_state(cms_with_heap(), "us"), - family_state(cms_with_heap(), "eu"), + family_state(irate(), "us"), + family_state(irate(), "eu"), ]) .unwrap(); merged.validate_structure().unwrap(); - assert_eq!(FamilyCapability::of(&cms_with_heap()).merge, None); + assert_eq!(FamilyCapability::of(&irate()).merge, None); } /// Idempotent families (HLL, exact Min/Max) merge states that may share From 2aa833d2b3a50afc18915b3b34a0d1ef594a203e Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 23:40:10 +0000 Subject: [PATCH 5/6] fix(physical-operators): reject a summary merge without inputs; test pane merges A SummaryMerge with no inputs is an invalid DAG instead of an index panic. Tests cover an exact Increase pane merge against the single build, compacted KLL panes (k = 200, 60 x 1000 values) within the calibrated rank error, and IRate as the precompute family without a merge. --- .../src/physical_planner/mod.rs | 5 +- .../tests/physical_semantics.rs | 84 +++++++++++++++++++ .../tests/precompute_population.rs | 8 +- .../tests/summary_projection.rs | 37 ++++++++ .../tests/tumbling_pane_merge.rs | 35 +++----- 5 files changed, 140 insertions(+), 29 deletions(-) diff --git a/crates/asap-physical-operators/src/physical_planner/mod.rs b/crates/asap-physical-operators/src/physical_planner/mod.rs index a46340fcf..19e869417 100644 --- a/crates/asap-physical-operators/src/physical_planner/mod.rs +++ b/crates/asap-physical-operators/src/physical_planner/mod.rs @@ -455,7 +455,10 @@ fn compile_internal( } } if let Payload::ASAP(ASAPOp::SummaryMerge { .. }) = &node.payload { - require_sound_merge(&schemas[0]).map_err(|e| invalid(format!("node {id}: {e}")))?; + 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 diff --git a/crates/asap-physical-operators/tests/physical_semantics.rs b/crates/asap-physical-operators/tests/physical_semantics.rs index 266378368..a1b43889d 100644 --- a/crates/asap-physical-operators/tests/physical_semantics.rs +++ b/crates/asap-physical-operators/tests/physical_semantics.rs @@ -646,6 +646,90 @@ 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}"); + } +} + // Retained zero-column rows still own Vec headers and must consume the output budget. #[test] fn zero_column_output_obeys_memory_limit() { diff --git a/crates/asap-physical-operators/tests/precompute_population.rs b/crates/asap-physical-operators/tests/precompute_population.rs index dda368e15..2c56ae4d1 100644 --- a/crates/asap-physical-operators/tests/precompute_population.rs +++ b/crates/asap-physical-operators/tests/precompute_population.rs @@ -401,10 +401,10 @@ fn sum_state(value: f64) -> Arc { // Only an explicit merge may collapse distinct pane updates before finalization. #[test] fn precompute_merge_requires_a_family_with_a_sound_merge() { - // Increase reads the samples at the window edges, so the - // family-capability table gives it no merge. - let increase = FieldDataType::ExactAggregate(ExactKind::Increase, ExactParams::Increase); - let error = try_state_dag(increase, None, true).err().expect("rejected"); + // 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(); diff --git a/crates/asap-physical-operators/tests/summary_projection.rs b/crates/asap-physical-operators/tests/summary_projection.rs index 32f5b1d46..bdd2d92d7 100644 --- a/crates/asap-physical-operators/tests/summary_projection.rs +++ b/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}" + ); +} diff --git a/crates/integration-tests/tests/tumbling_pane_merge.rs b/crates/integration-tests/tests/tumbling_pane_merge.rs index cfa04c15d..f64285050 100644 --- a/crates/integration-tests/tests/tumbling_pane_merge.rs +++ b/crates/integration-tests/tests/tumbling_pane_merge.rs @@ -21,7 +21,7 @@ use asap_types::ir::{ apply_materialization_timings, ASAPOp, Materialization, MaterializationAssignment, Medium, NonASAPOp, Operator, OperatorNode, TimingMemo, }; -use asap_types::post_asap::{ExactKind, ExactParams, FieldDataType, Schema}; +use asap_types::post_asap::Schema; use asap_types::types::AccuracyTarget; const EVALUATION_MS: i64 = 300_000; @@ -45,12 +45,8 @@ fn summary_agg(root: &Rc) -> Rc { /// Rewrite `root` so its summary is built in tumbling panes: pane `i` is /// `TimeRange(1m)` over `TimeShift(i·1m)` over the same `Scan`, and a -/// `SummaryMerge` combines the panes, with `family` in place of the -/// selected one when given. Returns the root and the panes. -fn tumbling_panes( - root: &Rc, - family: Option, -) -> (Rc, Vec>) { +/// `SummaryMerge` combines the panes. Returns the root and the panes. +fn tumbling_panes(root: &Rc) -> (Rc, Vec>) { let agg = summary_agg(root); let Some(NonASAPOp::TimeRange { kind, child, .. }) = agg.children()[0].non_asap() else { panic!("expected a raw time range under the summary"); @@ -71,13 +67,7 @@ fn tumbling_panes( child: shifted, })) .unwrap(); - let mut operator = agg.operator.map_children(|_| range.clone()); - if let (Some(family), Operator::ASAP(ASAPOp::SummaryAgg { family: f, .. })) = - (&family, &mut operator) - { - *f = family.clone(); - } - OperatorNode::new_shared(operator).unwrap() + OperatorNode::new_shared(agg.operator.map_children(|_| range.clone())).unwrap() }) .collect(); let merged = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { @@ -318,7 +308,7 @@ fn values(rows: &[Vec]) -> Vec { /// Returns the per-series values, which every execution path agrees on. fn assert_pane_merge_matches_single_build(query: &str, accuracy: AccuracyTarget) -> Vec { let single = single_build(query, accuracy); - let (panes, pane_builds) = tumbling_panes(&single, None); + let (panes, pane_builds) = tumbling_panes(&single); let expected = execute(&single); assert_eq!(expected.len(), 2, "{expected:?}"); assert!(expected.iter().all(|row| has_timestamp(row, EVALUATION_MS))); @@ -365,14 +355,11 @@ fn quantile_over_tumbling_panes_matches_single_build() { } } -/// The executor consults the family-capability table: panes of a family -/// with no sound merge (here exact Increase, which reads the window edges) -/// do not compile, although the IR accepts the merge. +/// Exact counter Increase over 5 merged 1-minute panes equals one 5-minute +/// build: the merge joins consecutive panes across their boundaries. #[test] -fn panes_of_a_family_without_a_sound_merge_do_not_compile() { - let single = single_build("sum_over_time(m[5m])", AccuracyTarget::Exact); - let increase = FieldDataType::ExactAggregate(ExactKind::Increase, ExactParams::Increase); - let (panes, _) = tumbling_panes(&single, Some(increase)); - let error = compile_at_query_time(&panes).err().expect("rejected"); - assert!(error.contains("no sound merge"), "{error}"); +fn exact_increase_over_tumbling_panes_matches_single_build() { + let increases = + assert_pane_merge_matches_single_build("increase(m[5m])", AccuracyTarget::Exact); + assert_eq!(increases.len(), 2); } From 5b5fc9f673ec2770594b9a399260b2dcf3e03330 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 02:32:08 +0000 Subject: [PATCH 6/6] fix(physical-operators): merge counter panes in time order Counter states (Rate, Increase) join consecutive windows pairwise, which is exact only in time order. A merge over a family whose inputs must be disjoint in time now orders its input states by their earliest sample time before folding, so panes merged in any input order give the single-build result. --- .../src/operators/summary/mod.rs | 18 +++++- .../src/summary_kernels/exact.rs | 14 +++++ .../tests/physical_semantics.rs | 60 +++++++++++++++++++ 3 files changed, 91 insertions(+), 1 deletion(-) diff --git a/crates/asap-physical-operators/src/operators/summary/mod.rs b/crates/asap-physical-operators/src/operators/summary/mod.rs index 5dab8109f..46d82c6e5 100644 --- a/crates/asap-physical-operators/src/operators/summary/mod.rs +++ b/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>, + mut rows: Vec>, state_column: usize, groups: &[usize], context: &RunContext, ) -> Result>, 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::() + .and_then(|state| state.counter_window_start()), + _ => None, + }); + } + } type GroupState = (Vec, SummaryFamilyType, Arc); let mut states: BTreeMap>, GroupState> = BTreeMap::new(); let mut work = Cooperative::new(context); diff --git a/crates/asap-physical-operators/src/summary_kernels/exact.rs b/crates/asap-physical-operators/src/summary_kernels/exact.rs index d5375f9bf..7e2d0ec5b 100644 --- a/crates/asap-physical-operators/src/summary_kernels/exact.rs +++ b/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 { + 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, .. }) diff --git a/crates/asap-physical-operators/tests/physical_semantics.rs b/crates/asap-physical-operators/tests/physical_semantics.rs index a1b43889d..4173730fa 100644 --- a/crates/asap-physical-operators/tests/physical_semantics.rs +++ b/crates/asap-physical-operators/tests/physical_semantics.rs @@ -730,6 +730,66 @@ fn compacted_kll_panes_merge_within_the_calibrated_rank_error() { } } +// 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| -> Vec> { + 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::() + .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() {