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/physical_planner/mod.rs b/crates/asap-physical-operators/src/physical_planner/mod.rs index 7ba0074d9..19e869417 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,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 { 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/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 266378368..4173730fa 100644 --- a/crates/asap-physical-operators/tests/physical_semantics.rs +++ b/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| -> 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() { diff --git a/crates/asap-physical-operators/tests/precompute_population.rs b/crates/asap-physical-operators/tests/precompute_population.rs index dc69f45b8..2c56ae4d1 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() { + // 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); 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 new file mode 100644 index 000000000..f64285050 --- /dev/null +++ b/crates/integration-tests/tests/tumbling_pane_merge.rs @@ -0,0 +1,365 @@ +//! 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::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. 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"); + }; + 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(); + OperatorNode::new_shared(agg.operator.map_children(|_| range.clone())).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); + 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}" + ); + } +} + +/// Exact counter Increase over 5 merged 1-minute panes equals one 5-minute +/// build: the merge joins consecutive panes across their boundaries. +#[test] +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); +} diff --git a/crates/types/src/ir/summary_coverage.rs b/crates/types/src/ir/summary_coverage.rs index 53cb6b1e2..6bb9c9080 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,21 @@ 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, + #[error("summary merge inputs are not proven disjoint in time, which counter states need")] + NotDisjointInTime, } 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, 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)), @@ -88,19 +96,37 @@ impl SummaryCoverage { .map(|child| child.coverage().ok_or(CoverageError::NotSummary)) .collect::, _>>()?; let first = inputs.first().ok_or(CoverageError::EmptyMerge)?; + 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 = inputs[..i].iter().any(|other| { - input - .selection - .iter() - .any(|a| other.selection.iter().any(|b| !a.disjoint(b))) - }); - if overlaps { - return Err(CoverageError::PossibleOverlap); + let any_pair = |separated: fn(&SelectionBox, &SelectionBox) -> bool| { + inputs[..i].iter().any(|other| { + input + .selection + .iter() + .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 { @@ -636,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))) + } } } @@ -669,6 +700,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/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 new file mode 100644 index 000000000..5142a9c78 --- /dev/null +++ b/crates/types/src/post_asap/family_capability.rs @@ -0,0 +1,170 @@ +//! 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, SketchStatistic}; +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, + /// The merged error of readouts that read the family's heavy-hitter + /// heaps (distinct count, entropy, top-k), when it differs from + /// `error`. + pub heap_readouts: Option, +} + +/// 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, + /// 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, +} + +/// 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, + /// The merge is allowed, but the merged readout has no guarantee; it is + /// evidence only, for plan selection to judge. + Approximate, +} + +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 SelectionRelation::{Disjoint, DisjointInTime, OverlapAllowed}; + use SketchAlgorithm as S; + let merge = |inputs, error| { + Some(MergeCapability { + inputs, + error, + 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 { + FieldDataType::ExactAggregate(kind, _) => match kind { + 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(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(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. The executor has no KMV kernel yet (#697). + S::Kmv | S::Hll => { + (merge(OverlapAllowed, MergedError::SameAsUnionBuild), false) + } + // Theta is not offered. + S::Theta => (None, false), + } + } + FieldDataType::Plain(_) + | FieldDataType::Sample(..) + | FieldDataType::Wavelet(..) + | FieldDataType::StatModel(..) => (None, false), + }; + Self { + merge, + 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/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_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 d2b4194d9..40d568174 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, SketchStatistic, 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,353 @@ 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::{Approximate, Exact, SameAsUnionBuild}; + use SelectionRelation::{Disjoint, DisjointInTime, 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, + ), + // Counter panes merge exactly when ordered and disjoint in time. + ( + exact(ExactKind::Rate, ExactParams::Rate), + Some((DisjointInTime, Exact)), + false, + ), + ( + exact(ExactKind::Increase, ExactParams::Increase), + Some((DisjointInTime, Exact)), + 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, Approximate)), + 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(), Some((Disjoint, SameAsUnionBuild)), false), + ( + sketch( + SketchAlgorithm::CountSketchWithHeap, + SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + }, + ), + Some((Disjoint, SameAsUnionBuild)), + 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:?}"); + } +} + +/// 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 { + heap_size: 10, + sketch_rows: 4, + sketch_cols: 64, + layers: 8, + }, + )); + for readout in [ + FrequencyL2, + point(), + Cardinality, + FrequencyEntropy, + TopK { k: 5 }, + ] { + assert_eq!( + univmon.merged_error(&readout), + Some(MergedError::Approximate), + "{readout:?}" + ); + } + let heap = FamilyCapability::of(&cms_with_heap()); + assert_eq!( + heap.merged_error(&TopK { k: 5 }), + Some(MergedError::Approximate) + ); + assert_eq!( + heap.merged_error(&point()), + Some(MergedError::SameAsUnionBuild) + ); + 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) + ); + 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: 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(irate(), "us"), + family_state(irate(), "eu"), + ]) + .unwrap(); + merged.validate_structure().unwrap(); + assert_eq!(FamilyCapability::of(&irate()).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]) + ); +}