diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index ae1f0cd1f..cd400e912 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -1510,11 +1510,9 @@ fn display_cap_shows_the_selected_then_valid_cheapest_then_each_cause() { assert!(costs.len() >= 2, "{costs:?}"); } -/// With no supported summaries declared, `stage_pipeline` plans with the -/// executor's families, as `plan_installable` does: a count's CountSketch, -/// which the Planner's executor does not run, is pruned in Pass 1 instead of -/// offered (before, `Q1 CountSketch` was a candidate), and the document -/// shows the families used. A declared list is kept as declared. +/// With no supported summaries declared, `stage_pipeline` uses the executor's +/// families, as `plan_installable` does. CountSketch cannot answer a total, so +/// it is neither generated for count nor reported as deployment-pruned. #[test] fn undeclared_supported_summaries_default_to_the_executors_families() { let document = document_of(&[ @@ -1540,8 +1538,8 @@ fn undeclared_supported_summaries_default_to_the_executors_families() { } assert!(labels.iter().all(|label| !label.contains("CountSketch"))); assert_eq!( - document["stage1_logical_asap_optimization"]["pass1_pruned"][0]["pruned"], - serde_json::json!([{ "candidate": "CountSketch", "reason": "UnsupportedByDeployment" }]) + document["stage1_logical_asap_optimization"]["pass1_pruned"], + serde_json::json!([]) ); assert_eq!( document["deployment_capabilities"]["supported_summaries"], diff --git a/crates/executor/src/capability.rs b/crates/executor/src/capability.rs index 8794899a1..9a19b9286 100644 --- a/crates/executor/src/capability.rs +++ b/crates/executor/src/capability.rs @@ -12,8 +12,8 @@ use crate::summary_kernels::hll_sketch::HLL_PRECISIONS; use crate::summary_kernels::weighted_frequency::{self, FrequencyAlgorithm}; use crate::Error; use planner_types::ir::schema::{ - ExactKind, ExactParams, FieldDataType as SummaryFamilyType, GroupingStrategy, SketchAlgorithm, - SketchParams, SketchStatistic, SummaryUpdate, + ExactKind, ExactParams, FamilyCapability, FieldDataType as SummaryFamilyType, GroupingStrategy, + SketchAlgorithm, SketchParams, SketchStatistic, StatisticKind, SummaryUpdate, }; /// Check the same contract used by `create_planner_accumulator` before a plan @@ -341,67 +341,90 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> { .map_err(Error::Invalid) } -/// A sketch evaluation is native only for the families Planner can read directly. +/// The families the executor builds per instance and reads, in +/// `SketchAlgorithm` order. A family it builds but reads nothing from +/// (plain Count Sketch) and one it does not build (Theta) are absent. +pub(crate) const EVALUATED_FAMILIES: [SketchAlgorithm; 8] = [ + SketchAlgorithm::UnivMon, + SketchAlgorithm::Kll, + SketchAlgorithm::Cms, + SketchAlgorithm::Hll, + SketchAlgorithm::DDSketch, + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::Kmv, + SketchAlgorithm::CountSketchWithHeap, +]; + +/// The readouts the family-capability table lists for an evaluated family +/// that the executor does not evaluate yet: a per-instance Count-Min's item +/// frequency (the executor answers an item's frequency only from a Hydra +/// grid, which Stage 2 does not produce yet; `SupportedSummary` gains a +/// grouping dimension when it does). +pub(crate) const UNEVALUATED_READOUTS: [(SketchAlgorithm, StatisticKind); 1] = + [(SketchAlgorithm::Cms, StatisticKind::ItemFrequency)]; + +/// Whether the executor reads `readout` from a state of `family` whose +/// family-capability table row answers it. +fn evaluates(family: &SummaryFamilyType, readout: StatisticKind) -> bool { + use planner_types::ir::schema::HydraKind; + match family { + SummaryFamilyType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => { + let algorithm = kind.algorithm(); + EVALUATED_FAMILIES.contains(algorithm) + && !UNEVALUATED_READOUTS.contains(&(algorithm.clone(), readout)) + } + SummaryFamilyType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => { + matches!(kind, HydraKind::HydraCms | HydraKind::HydraKll) + } + _ => false, + } +} + +/// A sketch evaluation is native when the family-capability table says the +/// family answers the statistic and the executor evaluates that readout. +/// Top-k is a keyed evaluation (`Operator::keyed_evaluation`), not a sketch +/// evaluation. pub fn validate_sketch_evaluation( family: &SummaryFamilyType, query: &SketchStatistic, ) -> Result<(), Error> { validate_native_family(family)?; - use planner_types::ir::schema::SketchAlgorithm as A; - // A point count without an item value reads the total count. - let bare_count = matches!(query, SketchStatistic::PointCount { value: None, .. }); - let supported = match family { - SummaryFamilyType::Sketch(kind, _) => match (kind.algorithm(), query) { - (A::Kll, SketchStatistic::Quantile { q }) - | (A::DDSketch, SketchStatistic::Quantile { q }) => { - if !(0.0..=1.0).contains(q) { - return Err(Error::Invalid( - "quantile evaluation requires quantile in [0,1]".into(), - )); - } - true - } - (A::DDSketch, _) => bare_count, - (A::Hll, SketchStatistic::Cardinality) => true, - (A::Hll, _) => bare_count, - // HydraCms answers a group's item frequency. A group's total - // needs a group-mass query asap_sketchlib's Hydra lacks. - (A::Cms, SketchStatistic::PointCount { value, .. }) - if matches!(family, SummaryFamilyType::Sketch(_, grouping) - if grouping != &GroupingStrategy::PerSubpopulationInstance) => - { - if value.is_none() { - return Err(Error::Invalid( - "HydraCms does not answer a group total: asap_sketchlib's Hydra has no \ - group-mass query; only an item's point count within a group is answered" - .into(), - )); + let readout = query.kind(); + // HydraCms answers a group's item frequency. A group's total needs a + // group-mass query asap_sketchlib's Hydra lacks. + if readout == StatisticKind::Total + && matches!( + family, + SummaryFamilyType::Sketch( + _, + GroupingStrategy::SharedMultiSubpopulation { + kind: planner_types::ir::schema::HydraKind::HydraCms, + .. } - true - } - (A::Kmv, SketchStatistic::Cardinality) => true, - (A::Kmv, _) => bare_count, - // One frequency state answers its total, distinct count, L2 norm - // and entropy. - ( - A::UnivMon, - SketchStatistic::Cardinality - | SketchStatistic::FrequencyL2 - | SketchStatistic::FrequencyEntropy, - ) => true, - (A::UnivMon, _) => bare_count, - // Only count intents read a Count-Min bare count, and their - // updates have unit weight; the evaluation is typed Int64 on that basis. - (A::Cms, _) => bare_count, - _ => false, - }, - _ => false, - }; - if !supported { + ) + ) + { + return Err(Error::Invalid( + "HydraCms does not answer a group total: asap_sketchlib's Hydra has no group-mass \ + query; only an item's point count within a group is answered" + .into(), + )); + } + if readout == StatisticKind::TopK + || !FamilyCapability::of(family).answers(query) + || !evaluates(family, readout) + { return Err(Error::Invalid( "evaluation is not implemented for this summary family".into(), )); } + if let SketchStatistic::Quantile { q } = query { + if !(0.0..=1.0).contains(q) { + return Err(Error::Invalid( + "quantile evaluation requires quantile in [0,1]".into(), + )); + } + } Ok(()) } diff --git a/crates/executor/src/summary_kernels/dd_sketch.rs b/crates/executor/src/summary_kernels/dd_sketch.rs index a57666a86..a3e31ac08 100644 --- a/crates/executor/src/summary_kernels/dd_sketch.rs +++ b/crates/executor/src/summary_kernels/dd_sketch.rs @@ -42,7 +42,7 @@ impl AggregateCore for DDSketchAccumulator { self.inner.get_count() == 0 } - /// Quantiles, and the total sample count as a bare `PointCount`. + /// Quantiles. fn estimate(&self, query: &SketchStatistic) -> Result { match query { // PromQL `quantile_over_time` and SQL `percentile_cont` @@ -55,7 +55,6 @@ impl AggregateCore for DDSketchAccumulator { .inner .quantile_interpolated(*q)? .ok_or_else(|| "DDSketch quantile of an empty population".into()), - SketchStatistic::PointCount { value: None, .. } => Ok(self.inner.get_count() as f64), other => Err(format!("DDSketch does not answer {other:?}").into()), } } @@ -73,9 +72,9 @@ mod tests { } } - // A bare point count reads the total sample count, and merge adds counts. + // A merged state's quantile is the union's. #[test] - fn count_and_quantile_survive_merge() { + fn quantile_survives_merge() { let (mut a, mut b) = ( DDSketchAccumulator::new(0.01), DDSketchAccumulator::new(0.01), @@ -85,19 +84,20 @@ mod tests { b.inner.add(&f64::from(v + 50)); } let merged = a.merge_with(&b).unwrap(); - assert_eq!(merged.estimate(&bare_count()).unwrap(), 100.0); let median = merged .estimate(&SketchStatistic::Quantile { q: 0.5 }) .unwrap(); assert!((median - 50.0).abs() <= 1.0, "{median}"); } - // An empty DDSketch has no quantile, and unsupported queries are errors. + // An empty DDSketch has no quantile, and unsupported queries (its + // total among them) are errors. #[test] fn empty_quantile_and_unsupported_queries_fail() { let dd = DDSketchAccumulator::new(0.01); assert!(dd.estimate(&SketchStatistic::Quantile { q: 0.5 }).is_err()); assert!(dd.estimate(&SketchStatistic::Cardinality).is_err()); + assert!(dd.estimate(&bare_count()).is_err()); } // The quantile at q interpolates linearly between the order statistics diff --git a/crates/executor/src/summary_kernels/hll_sketch.rs b/crates/executor/src/summary_kernels/hll_sketch.rs index e04a282a3..191cfe605 100644 --- a/crates/executor/src/summary_kernels/hll_sketch.rs +++ b/crates/executor/src/summary_kernels/hll_sketch.rs @@ -155,13 +155,10 @@ impl AggregateCore for HllSketchAccumulator { Ok(Box::new(merged)) } - /// Distinct count, rounded down to an integer. A bare `PointCount` over - /// an HLL also reads the distinct count. + /// Distinct count, rounded down to an integer. fn estimate(&self, query: &SketchStatistic) -> Result { match query { - SketchStatistic::Cardinality | SketchStatistic::PointCount { value: None, .. } => { - Ok(self.inner.estimate() as f64) - } + SketchStatistic::Cardinality => Ok(self.inner.estimate() as f64), other => Err(format!("HLL does not answer {other:?}").into()), } } diff --git a/crates/executor/src/summary_kernels/kmv.rs b/crates/executor/src/summary_kernels/kmv.rs index 7d61408c5..790083479 100644 --- a/crates/executor/src/summary_kernels/kmv.rs +++ b/crates/executor/src/summary_kernels/kmv.rs @@ -62,12 +62,10 @@ impl AggregateCore for KmvAccumulator { Ok(Box::new(merged)) } - /// Distinct count. A bare `PointCount` over a KMV also reads the distinct count. + /// Distinct count. fn estimate(&self, query: &SketchStatistic) -> Result { match query { - SketchStatistic::Cardinality | SketchStatistic::PointCount { value: None, .. } => { - Ok(self.inner.estimate()) - } + SketchStatistic::Cardinality => Ok(self.inner.estimate()), other => Err(format!("KMV does not answer {other:?}").into()), } } @@ -150,14 +148,6 @@ mod tests { fn counts_fewer_than_k_distinct_items_exactly() { let kmv = build(1024, (0..500).chain(0..500)); assert_eq!(cardinality(&kmv), 500.0); - assert_eq!( - kmv.estimate(&SketchStatistic::PointCount { - key: planner_types::ir::scalar::ColumnRef::SampleValue, - value: None, - }) - .unwrap(), - 500.0 - ); } // Above k the estimate is within ten times KMV's relative standard diff --git a/crates/executor/src/summary_kernels/traits.rs b/crates/executor/src/summary_kernels/traits.rs index 6705cbb99..4cde5ae1a 100644 --- a/crates/executor/src/summary_kernels/traits.rs +++ b/crates/executor/src/summary_kernels/traits.rs @@ -68,10 +68,9 @@ mod tests { .downcast_mut::() .unwrap(); dd.inner.add(&3.0); - let count = SketchStatistic::PointCount { - key: planner_types::ir::scalar::ColumnRef::SampleValue, - value: None, - }; - assert_eq!(state.estimate(&count).unwrap(), 1.0); + let median = state + .estimate(&SketchStatistic::Quantile { q: 0.5 }) + .unwrap(); + assert!((median - 3.0).abs() <= 0.03, "{median}"); } } diff --git a/crates/executor/src/supported_summaries.rs b/crates/executor/src/supported_summaries.rs index c67168abd..1d1e976e4 100644 --- a/crates/executor/src/supported_summaries.rs +++ b/crates/executor/src/supported_summaries.rs @@ -3,39 +3,24 @@ //! `DeploymentCapabilities::supported_summaries`, so the Planner offers no //! summary the executor would reject when it binds the plan. -use planner_types::ir::schema::{SketchAlgorithm, StatisticKind, SupportedSummary}; +use crate::capability::{EVALUATED_FAMILIES, UNEVALUATED_READOUTS}; +use planner_types::ir::schema::{FamilyCapability, SupportedSummary}; /// Every sketch family this executor builds from rows and the statistics it -/// answers from one state of it, in `SketchAlgorithm` order. A family the -/// executor builds but reads nothing from (plain Count Sketch) and one it -/// does not build (Theta) are absent. Exact accumulators are always -/// supported and are not listed. -/// -/// Per-instance states only: Count-Min answers its total (an item's -/// frequency only from a Hydra grid, which Stage 2 does not produce yet; -/// `SupportedSummary` gains a grouping dimension when it does), and the -/// heap families answer top-k only. +/// answers from one state of it: the family-capability table's row of each +/// family it evaluates, less the readouts it does not evaluate yet. Exact +/// accumulators are always supported and are not listed. pub fn supported_summary_families() -> Vec { - use SketchAlgorithm as A; - use StatisticKind::*; - let families: [(A, &[StatisticKind]); 8] = [ - ( - A::UnivMon, - &[Total, Distinct, FrequencyL2, FrequencyEntropy], - ), - (A::Kll, &[Quantile]), - (A::Cms, &[Total]), - (A::Hll, &[Total, Distinct]), - (A::DDSketch, &[Total, Quantile]), - (A::CmsWithHeap, &[TopK]), - (A::Kmv, &[Total, Distinct]), - (A::CountSketchWithHeap, &[TopK]), - ]; - families + EVALUATED_FAMILIES .into_iter() - .map(|(family, readouts)| SupportedSummary { + .map(|family| SupportedSummary { + readouts: FamilyCapability::of_per_instance_sketch(&family) + .readouts + .iter() + .map(|&(readout, _)| readout) + .filter(|readout| !UNEVALUATED_READOUTS.contains(&(family.clone(), *readout))) + .collect(), family, - readouts: readouts.to_vec(), }) .collect() } @@ -46,11 +31,13 @@ mod tests { use crate::capability::validate_sketch_evaluation; use crate::factory::create_planner_accumulator; use crate::summary_kernels::weighted_frequency::WeightedFrequency; + use planner_types::ir::operator::operator_properties::{Reduction, Source}; use planner_types::ir::scalar::ColumnRef; use planner_types::ir::schema::{ - FieldDataType, GroupingStrategy, SketchKind, SketchParams, SketchStatistic, - SummaryInputExpr, SummaryUpdate, + DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchKind, + SketchParams, SketchStatistic, StatisticKind, SummaryInputExpr, SummaryUpdate, }; + use planner_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode}; const ALGORITHMS: [SketchAlgorithm; 10] = [ SketchAlgorithm::UnivMon, @@ -132,7 +119,11 @@ mod tests { fn statistic(readout: StatisticKind) -> SketchStatistic { let key = ColumnRef::Named("item".into()); match readout { - StatisticKind::Total => SketchStatistic::PointCount { key, value: None }, + // Pass 1 reads a total keyed by the sample value. + StatisticKind::Total => SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }, StatisticKind::ItemFrequency => SketchStatistic::PointCount { key, value: Some("a".into()), @@ -162,18 +153,90 @@ mod tests { /// Whether the executor answers `readout` from `family`'s state: top-k /// through the keyed heap evaluation, every other readout through the - /// sketch evaluation. + /// sketch evaluation, whose kernel must then read it from a built state. fn evaluation_answers(family: &FieldDataType, readout: StatisticKind) -> bool { match (readout, family) { (StatisticKind::TopK, FieldDataType::Sketch(kind, _)) => { WeightedFrequency::configuration(kind).is_ok() } - _ => validate_sketch_evaluation(family, &statistic(readout)).is_ok(), + _ => { + let accepted = validate_sketch_evaluation(family, &statistic(readout)).is_ok(); + if accepted { + assert!(kernel_answers(family, readout), "{family:?} {readout:?}"); + } + accepted + } } } + /// Whether the kernel of a per-instance `family` state built from one + /// update reads `readout`. + fn kernel_answers(family: &FieldDataType, readout: StatisticKind) -> bool { + let FieldDataType::Sketch(_, grouping) = family else { + return false; + }; + let value = SummaryUpdate::column(ColumnRef::SampleValue); + let keyed = SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("item".into()))), + ..value.clone() + }; + let Ok(mut updater) = create_planner_accumulator(family, &value, grouping) + .or_else(|_| create_planner_accumulator(family, &keyed, grouping)) + else { + return false; + }; + let updated = if updater.is_keyed() { + updater.update_keyed( + &crate::KeyByLabelValues::new_with_labels(vec!["a".into()]), + 1.0, + 0, + ) + } else { + updater.update_single(1.0, 0) + }; + updated.is_ok() + && updater + .snapshot_accumulator() + .estimate(&statistic(readout)) + .is_ok() + } + + /// Whether the IR builds a structurally valid `SummaryEstimate` of + /// `readout` over a `family` state. + fn ir_builds_estimate(family: &FieldDataType, readout: StatisticKind) -> bool { + let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: "t".into(), + }, + predicates: vec![], + schema: Schema::new(vec![ + Field::plain("item", DataType::Utf8, false), + Field::plain("value", DataType::Float64, false), + ]), + })) + .unwrap(); + let state = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + child: scan, + family: family.clone(), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + filter: None, + })) + .unwrap(); + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: state, + query: statistic(readout), + })) + .is_ok_and(|estimate| estimate.validate_structure().is_ok()) + } + /// The list holds exactly the (family, readout) pairs the factory builds - /// per instance and an evaluation answers from that state. + /// per instance and an evaluation answers from that state, and the + /// IR's `SummaryEstimate` validation accepts exactly what the + /// family-capability table answers. A listed family's readouts are its + /// table row less the ones named not evaluated, each of which the table + /// answers and the executor does not. #[test] fn supported_summary_families_match_the_factory_and_evaluations() { let listed = supported_summary_families(); @@ -187,6 +250,17 @@ mod tests { .iter() .any(|s| &s.family == algorithm && s.readouts.contains(&readout)); assert_eq!(declared, executed, "{algorithm:?} {readout:?}"); + let answered = FamilyCapability::of(&family).answers(&statistic(readout)); + assert_eq!( + ir_builds_estimate(&family, readout), + answered, + "IR: {algorithm:?} {readout:?}" + ); + if UNEVALUATED_READOUTS.contains(&(algorithm.clone(), readout)) { + assert!(answered && !executed, "{algorithm:?} {readout:?}"); + } else if EVALUATED_FAMILIES.contains(algorithm) { + assert_eq!(answered, executed, "table: {algorithm:?} {readout:?}"); + } } } assert!(listed.iter().all(|s| !s.readouts.is_empty())); diff --git a/crates/frontend-promql/tests/count_planning.rs b/crates/frontend-promql/tests/count_planning.rs index 3fe459bc5..6b965876b 100644 --- a/crates/frontend-promql/tests/count_planning.rs +++ b/crates/frontend-promql/tests/count_planning.rs @@ -58,7 +58,8 @@ fn exact_counts_select_count_accumulators() { } } -// CMS/CountSketch count updates must stay +1 even for zero or negative samples. +// CMS/UnivMon count updates stay +1 even for zero or negative samples. +// CountSketch cannot answer totals and must not be offered for count. #[test] fn frequency_count_candidates_use_unit_weights() { for query in ["count_over_time(up[5m])", "count(up)"] { @@ -88,12 +89,15 @@ fn frequency_count_candidates_use_unit_weights() { continue; }; assert!( - !matches!(kind.algorithm(), SketchAlgorithm::Hll), + !matches!( + kind.algorithm(), + SketchAlgorithm::Hll | SketchAlgorithm::CountSketch + ), "{query}: {node:?}" ); if !matches!( kind.algorithm(), - SketchAlgorithm::Cms | SketchAlgorithm::CountSketch + SketchAlgorithm::Cms | SketchAlgorithm::UnivMon ) { continue; } @@ -124,7 +128,7 @@ fn frequency_count_candidates_use_unit_weights() { "{query}: {candidates:?}" ); assert!( - algorithms.contains(&SketchAlgorithm::CountSketch), + algorithms.contains(&SketchAlgorithm::UnivMon), "{query}: {candidates:?}" ); } diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index 24d7f9b95..09e98fc8a 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -3077,11 +3077,11 @@ mod tests { ); } - /// A count over a window, read from a CountSketch built in tumbling + /// A count over a window, read from a Count-Min built in tumbling /// panes, keeps the guarantee of one build over the window: Pass 1 - /// offers CountSketch for counts, and its merged counters add. + /// offers Count-Min for counts, and its merged counters add. #[test] - fn a_count_from_count_sketch_panes_keeps_its_guarantee() { + fn a_count_from_count_min_panes_keeps_its_guarantee() { use crate::per_computation_candidates::{ compose_logical_candidate, generate_per_computation_candidates, ComputationCandidate, }; @@ -3101,8 +3101,8 @@ mod tests { t.choices .candidates .iter() - .position(|c| *c == ComputationCandidate::Summary(SketchAlgorithm::CountSketch)) - .expect("CountSketch is offered for a count") + .position(|c| *c == ComputationCandidate::Summary(SketchAlgorithm::Cms)) + .expect("Count-Min is offered for a count") }) .collect(); let roots: Vec = compose_logical_candidate(&inventory, &choice) diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index be8d3b922..59723336a 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs @@ -3764,15 +3764,13 @@ fn summary_update( } } (_, Some(SketchAlgorithm::UnivMon)) - | (AggIntent::Count { .. }, Some(SketchAlgorithm::Cms | SketchAlgorithm::CountSketch)) => { - SummaryUpdate { - item: Some(weight), - weight: SummaryInputExpr::Constant(1.0), - weight_domain: WeightDomain::NonNegative { - proof: NonNegativeWeightProof::UnitCount, - }, - } - } + | (AggIntent::Count { .. }, Some(SketchAlgorithm::Cms)) => SummaryUpdate { + item: Some(weight), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + }, _ => SummaryUpdate { item: None, weight, @@ -3851,6 +3849,32 @@ mod tests { assert!(choices.contains(&ComputationCandidate::Summary(SketchAlgorithm::UnivMon))); } + /// Count Sketch's signed counters give no total, so Pass 1 offers no + /// Count Sketch for a count, and every count candidate it offers + /// answers the total. + #[test] + fn a_count_offers_no_count_sketch() { + let choices = computation_candidates_for_intent(&AggIntent::Count { + accuracy: AccuracyTarget::max_error(0.05).failure_probability(0.01), + }) + .unwrap() + .candidates; + assert!(!choices.contains(&ComputationCandidate::Summary(SketchAlgorithm::CountSketch))); + let total = statistic(&AggIntent::Count { + accuracy: AccuracyTarget::exact(), + }) + .unwrap(); + for choice in &choices { + if let ComputationCandidate::Summary(algorithm) = choice { + assert!( + asap_types::ir::schema::FamilyCapability::of_per_instance_sketch(algorithm) + .answers(&total), + "{algorithm:?}" + ); + } + } + } + /// Guarantee composition through scalar and vector arithmetic (#669). mod guarantee_composition { use super::*; @@ -5109,9 +5133,9 @@ mod tests { /// `combinations` counts each choice once: picking a fused /// whole-expression candidate with any but the first candidate of the - /// target it absorbs is the same candidate. An absorbed target with four - /// candidates (an approximate inner count) gives 1 exact · 4 + 2 - /// whole-expression = 6, not 3 · 4. + /// target it absorbs is the same candidate. An absorbed target with three + /// candidates (an approximate inner count: its accumulator, Count-Min and + /// UnivMon) gives 1 exact · 3 + 2 whole-expression = 5, not 3 · 3. #[test] fn combinations_skip_choices_a_whole_expression_candidate_absorbs() { let inventory = inventory("topk by (job) (10, count_over_time(m[1m]))"); @@ -5127,7 +5151,7 @@ mod tests { let enumerated = enumerated_choices(&inventory); assert!(whole > 0 && m > 1, "{n} {whole} {m}"); assert_eq!(inventory.combinations(), n * m + whole); - assert_eq!(inventory.combinations(), 6); + assert_eq!(inventory.combinations(), 5); assert_eq!(inventory.combinations(), enumerated); } @@ -5552,14 +5576,13 @@ mod tests { /// A family the deployment builds but cannot answer the measure's /// readout from is pruned: UnivMon, supported for distinct counts and - /// L2 only, is no count candidate; Count-Min and CountSketch, supported - /// for the total, stay. + /// L2 only, is no count candidate; Count-Min, supported for the total, + /// stays. #[test] fn a_readout_the_deployment_cannot_answer_prunes_its_family() { let restricted = inventory("count_over_time(m[1m])").without_unsupported_summaries(&supporting(&[ (SketchAlgorithm::Cms, &[StatisticKind::Total]), - (SketchAlgorithm::CountSketch, &[StatisticKind::Total]), ( SketchAlgorithm::UnivMon, &[StatisticKind::Distinct, StatisticKind::FrequencyL2], @@ -5571,7 +5594,6 @@ mod tests { [ ComputationCandidate::ExactAccumulator(vec![ExactKind::Count]), ComputationCandidate::Summary(SketchAlgorithm::Cms), - ComputationCandidate::Summary(SketchAlgorithm::CountSketch), ] ); assert_eq!( diff --git a/crates/logical-asap-optimization/src/realization.rs b/crates/logical-asap-optimization/src/realization.rs index 12cd66600..7fee1cba0 100644 --- a/crates/logical-asap-optimization/src/realization.rs +++ b/crates/logical-asap-optimization/src/realization.rs @@ -54,11 +54,9 @@ pub fn summary_candidates(intent: &AggIntent) -> &'static [SketchAlgorithm] { SketchAlgorithm::CmsWithHeap, SketchAlgorithm::CountSketchWithHeap, ], - AggIntent::Count { .. } => &[ - SketchAlgorithm::Cms, - SketchAlgorithm::CountSketch, - SketchAlgorithm::UnivMon, - ], + // Count Sketch's signed counters give no total, so it serves no + // count. + AggIntent::Count { .. } => &[SketchAlgorithm::Cms, SketchAlgorithm::UnivMon], _ => &[], } } diff --git a/crates/logical-asap-optimization/tests/per_computation_candidates.rs b/crates/logical-asap-optimization/tests/per_computation_candidates.rs index f1c11a864..54e0226f3 100644 --- a/crates/logical-asap-optimization/tests/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/tests/per_computation_candidates.rs @@ -406,11 +406,7 @@ fn multi_measure_aggregate_has_one_target_per_measure() { ); assert_eq!( algorithms(&candidates[1].choices.candidates), - vec![ - SketchAlgorithm::Cms, - SketchAlgorithm::CountSketch, - SketchAlgorithm::UnivMon - ] + vec![SketchAlgorithm::Cms, SketchAlgorithm::UnivMon] ); } diff --git a/crates/types/src/ir/operator/asap.rs b/crates/types/src/ir/operator/asap.rs index 1ad72fcdd..fe0200eab 100644 --- a/crates/types/src/ir/operator/asap.rs +++ b/crates/types/src/ir/operator/asap.rs @@ -9,7 +9,7 @@ use super::node::{OperatorNode, OperatorResultKind}; use crate::ir::operator::maintained_population::{MaintainedPopulation, PopulationStatistic}; use crate::ir::operator::operator_properties::Reduction; use crate::ir::schema::sketch::{GroupingStrategy, SketchStatistic, SummaryUpdate}; -use crate::ir::schema::{ColumnId, DataType, Field, FieldDataType, Schema}; +use crate::ir::schema::{ColumnId, DataType, FamilyCapability, Field, FieldDataType, Schema}; use crate::ir::SchemaDerivationError; /// Why an ASAP operator cannot be used yet. @@ -634,7 +634,8 @@ impl ASAPOp { query, } => { needs_state(summary_input, "SummaryEstimate")?; - use crate::ir::schema::sketch::SketchCategory as C; + // The family-capability table is the one source of which + // statistic a state's family answers. let states: Vec<_> = summary_input .schema .fields @@ -643,21 +644,9 @@ impl ASAPOp { .collect(); let valid = match states.as_slice() { [field] => match &field.dtype { - FieldDataType::Sketch(kind, _) => matches!( - (kind.category(), query), - (C::Quantile, SketchStatistic::Quantile { .. }) - | (C::Cardinality | C::Universal, SketchStatistic::Cardinality) - | ( - C::Frequency | C::Universal, - SketchStatistic::PointCount { .. } - ) - | ( - C::Universal, - SketchStatistic::FrequencyL2 - | SketchStatistic::FrequencyEntropy - ) - | (C::TopK | C::Universal, SketchStatistic::TopK { .. }) - ), + family @ FieldDataType::Sketch(..) => { + FamilyCapability::of(family).answers(query) + } _ => false, }, _ => false, diff --git a/crates/types/src/ir/schema/family_capability.rs b/crates/types/src/ir/schema/family_capability.rs index 1e974d857..75ddf6591 100644 --- a/crates/types/src/ir/schema/family_capability.rs +++ b/crates/types/src/ir/schema/family_capability.rs @@ -3,7 +3,10 @@ //! //! 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. One row per family says +//! table to decide whether a composition is sound. It is the one source of +//! which statistic a family answers: the IR's `SummaryEstimate` validation, +//! the executor's evaluation check and its supported-summary list all read +//! it. One row per family says //! how the inputs of a merge must relate ([`MergeInputRule`]) and which //! statistics ([`StatisticKind`]) one state answers, each with the error of //! that readout from a merged state ([`MergedError`]), so one state can @@ -126,30 +129,10 @@ impl FamilyCapability { } /// The error of `statistic` read from a merged state of this family; - /// `None` when the family has no sound merge. - /// - /// A statistic the family does not answer takes the family's default: - /// exact for an exact accumulator, approximate for a heavy-item readout - /// (top-k, entropy) of a family with heaps, and otherwise the guarantee - /// of one build over the union. Pass 1 offers CountSketch for a count, - /// which its row does not answer (the executor has no CountSketch - /// evaluation), and this default keeps that candidate's guarantee until - /// the realization and the table agree. + /// `None` when the family has no sound merge or does not answer + /// `statistic`. pub fn merged_error(&self, statistic: &SketchStatistic) -> Option { - self.merge_inputs?; - Some(self.readout(statistic).unwrap_or_else(|| { - let has_heaps = self - .readouts - .iter() - .any(|&(_, error)| error == MergedError::Approximate); - match statistic.kind() { - _ if self.readouts.is_empty() => MergedError::Exact, - StatisticKind::TopK | StatisticKind::FrequencyEntropy if has_heaps => { - MergedError::Approximate - } - _ => MergedError::AsIfBuiltOverUnion, - } - })) + self.merge_inputs.and(self.readout(statistic)) } /// The table entry of `statistic`'s kind: its merged error. @@ -161,13 +144,69 @@ impl FamilyCapability { .map(|&(_, error)| error) } + /// The capability of a per-subpopulation state of `algorithm`, whatever + /// its parameters: the table row the IR and a deployment's executor + /// read for that family. + pub fn of_per_instance_sketch(algorithm: &SketchAlgorithm) -> Self { + use MergeInputRule::{Disjoint, MayOverlap}; + use MergedError::{Approximate, AsIfBuiltOverUnion as Union}; + use SketchAlgorithm as S; + use StatisticKind::*; + let (merge_inputs, readouts): (_, &'static [(StatisticKind, MergedError)]) = match algorithm + { + // A quantile sketch answers every rank. + S::Kll | S::DDSketch => (Some(Disjoint), &[(Quantile, Union)]), + // Linear sketches: counters add, so a merged state's + // point counts are the union's. Count-Min's unsigned counters + // also give its total; Count Sketch's signed counters do not. + S::Cms => (Some(Disjoint), &[(ItemFrequency, Union), (Total, Union)]), + S::CountSketch => (Some(Disjoint), &[(ItemFrequency, Union)]), + // Top-k reads the heap, which keeps only the heavy + // items of its own part: an item heavy over the window + // but in no single part is lost. + S::CmsWithHeap | S::CountSketchWithHeap => (Some(Disjoint), &[(TopK, Approximate)]), + // One frequency vector serves its total and every + // moment. L2 reads layer 0's Count Sketch, whose + // counters add. The distinct count's bound needs each + // layer's heap to keep `min(|S_j|, k)` of the layer's + // keys, chosen without reading the next layer's + // selection bit; `UnivMon::merge` re-ranks the union of + // the parts' heap keys by the merged sketch, which + // keeps both, and its top-layer completeness is the + // union's. Entropy reads heavy items a part may have + // dropped. + S::UnivMon => ( + Some(Disjoint), + &[ + (Total, Union), + (Distinct, Union), + (FrequencyL2, Union), + (FrequencyEntropy, Approximate), + ], + ), + // HLL's merge keeps each register's maximum, which is + // idempotent: an item both inputs saw counts once, so + // overlapping inputs merge to the state of their union. + // `asap_sketchlib`'s KMV hashes every item with one + // shared seed (`CANONICAL_HASH_SEED`), and `KMV::merge` + // keeps the k smallest distinct hashes: again the state + // of the union. + S::Hll | S::Kmv => (Some(MayOverlap), &[(Distinct, Union)]), + // Theta is not offered, and has no merge. + S::Theta => (None, &[(Distinct, Approximate)]), + }; + Self { + merge_inputs, + readouts, + } + } + /// The capability of `family`. Families not listed have no capability /// (they fail closed). pub fn of(family: &FieldDataType) -> Self { use ExactKind as E; use MergeInputRule::{Disjoint, DisjointInTime, MayOverlap}; use MergedError::{Approximate, AsIfBuiltOverUnion as Union}; - use SketchAlgorithm as S; use StatisticKind::*; let (merge_inputs, readouts): (_, &'static [(StatisticKind, MergedError)]) = match family { FieldDataType::ExactAggregate(kind, _) => match kind { @@ -193,55 +232,7 @@ impl FamilyCapability { } } FieldDataType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => { - match kind.algorithm() { - // A quantile sketch answers every rank. - S::Kll => (Some(Disjoint), &[(Quantile, Union)]), - S::DDSketch => (Some(Disjoint), &[(Quantile, Union), (Total, Union)]), - // Linear sketches: counters add. Each Count Sketch row's - // squared counters estimate F2 (AMS). - S::Cms => (Some(Disjoint), &[(ItemFrequency, Union), (Total, Union)]), - S::CountSketch => ( - Some(Disjoint), - &[(ItemFrequency, Union), (FrequencyL2, Union)], - ), - // Point counts read the linear counters; top-k reads the - // heap, which keeps only the heavy items of its own - // part: an item heavy over the window but in no single - // part is lost. - S::CmsWithHeap | S::CountSketchWithHeap => ( - Some(Disjoint), - &[(TopK, Approximate), (ItemFrequency, Union)], - ), - // One frequency vector serves its total and every - // moment. L2 reads layer 0's Count Sketch, whose - // counters add. The distinct count's bound needs each - // layer's heap to keep `min(|S_j|, k)` of the layer's - // keys, chosen without reading the next layer's - // selection bit; `UnivMon::merge` re-ranks the union of - // the parts' heap keys by the merged sketch, which - // keeps both, and its top-layer completeness is the - // union's. Entropy reads heavy items a part may have - // dropped. - S::UnivMon => ( - Some(Disjoint), - &[ - (Total, Union), - (Distinct, Union), - (FrequencyL2, Union), - (FrequencyEntropy, Approximate), - ], - ), - // HLL's merge keeps each register's maximum, which is - // idempotent: an item both inputs saw counts once, so - // overlapping inputs merge to the state of their union. - // `asap_sketchlib`'s KMV hashes every item with one - // shared seed (`CANONICAL_HASH_SEED`), and `KMV::merge` - // keeps the k smallest distinct hashes: again the state - // of the union. - S::Hll | S::Kmv => (Some(MayOverlap), &[(Distinct, Union)]), - // Theta is not offered, and has no merge. - S::Theta => (None, &[(Distinct, Approximate)]), - } + return Self::of_per_instance_sketch(kind.algorithm()) } FieldDataType::Plain(_) | FieldDataType::SummarySample(..) @@ -312,8 +303,9 @@ mod tests { } } - /// A quantile sketch answers every rank; heap sketches answer top-k and - /// item frequencies; CountSketch answers item frequencies and L2. + /// A quantile sketch answers every rank, not its total; heap sketches + /// answer top-k; Count Sketch answers item frequencies, not its total; + /// HLL answers the distinct count, not its total. #[test] fn summary_capability_facts() { let kll = sketch( @@ -323,6 +315,10 @@ mod tests { failure_probability: None, }, ); + let ddsketch = sketch( + SketchAlgorithm::DDSketch, + SketchParams::DDSketch { alpha: 0.01 }, + ); let count_sketch = sketch( SketchAlgorithm::CountSketch, SketchParams::CountSketch { @@ -354,23 +350,30 @@ mod tests { ], vec![SketchStatistic::Cardinality, total()], ), + ( + &ddsketch, + vec![SketchStatistic::Quantile { q: 0.9 }], + vec![total()], + ), ( &count_sketch, - vec![item(), SketchStatistic::FrequencyL2], + vec![item()], vec![ + total(), + SketchStatistic::FrequencyL2, SketchStatistic::FrequencyEntropy, SketchStatistic::TopK { k: 3 }, ], ), ( &cms_heap, - vec![SketchStatistic::TopK { k: 10 }, item()], - vec![SketchStatistic::FrequencyL2], + vec![SketchStatistic::TopK { k: 10 }], + vec![SketchStatistic::FrequencyL2, item()], ), ( &hll, vec![SketchStatistic::Cardinality], - vec![SketchStatistic::FrequencyL2, item()], + vec![SketchStatistic::FrequencyL2, item(), total()], ), ] { let capability = FamilyCapability::of(family); diff --git a/crates/types/src/ir/schema/mod.rs b/crates/types/src/ir/schema/mod.rs index f7d35d98b..3d2eae47f 100644 --- a/crates/types/src/ir/schema/mod.rs +++ b/crates/types/src/ir/schema/mod.rs @@ -27,9 +27,8 @@ pub use family_capability::{ pub use sketch::{ default_hydra_params, hydra_kind_for, EntityIdentity, ExactKind, ExactParams, GroupingStrategy, HydraKind, HydraParams, NonNegativeWeightProof, SamplingKind, SamplingParams, SketchAlgorithm, - SketchCategory, SketchKind, SketchParams, SketchStatistic, StatModelKind, StatModelParams, - SummaryInputExpr, SummaryUpdate, WaveletKind, WaveletParams, WeightDomain, - EXACT_ACCUMULATOR_STATE_BYTES, + SketchKind, SketchParams, SketchStatistic, StatModelKind, StatModelParams, SummaryInputExpr, + SummaryUpdate, WaveletKind, WaveletParams, WeightDomain, EXACT_ACCUMULATOR_STATE_BYTES, }; use serde::{Deserialize, Serialize}; diff --git a/crates/types/src/ir/schema/sketch.rs b/crates/types/src/ir/schema/sketch.rs index 70a70b7d5..9943f552f 100644 --- a/crates/types/src/ir/schema/sketch.rs +++ b/crates/types/src/ir/schema/sketch.rs @@ -50,9 +50,8 @@ pub enum ExactParams { // ── Approximate sketches ───────────────────────────────────────────────────── /// A specific sketch algorithm — bounded error, sized by its -/// [`SketchParams`]. Each algorithm belongs to exactly one [`SketchKind`] -/// category (e.g. `Kll` and `DDSketch` both realize quantile sketches); -/// [`SketchKind::new`] is where that classification is made. +/// [`SketchParams`]. The statistics each answers are its row of the +/// family-capability table ([`FamilyCapability`](crate::ir::schema::FamilyCapability)). #[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] pub enum SketchAlgorithm { /// Universal frequency-vector summary with shared statistic evaluations. @@ -248,44 +247,27 @@ impl PartialEq for SketchParams { } } -/// The query category served by a committed sketch. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] -pub enum SketchCategory { - Universal, - Quantile, - Cardinality, - Frequency, - TopK, -} - -/// A committed sketch choice: which *category* of query shape it answers — -/// quantile-style, cardinality-style, frequency-style, or heavy-hitter/ -/// top-k-style estimation — together with the concrete [`SketchAlgorithm`] -/// and [`SketchParams`] realizing it. Sits between +/// A committed sketch choice: the concrete [`SketchAlgorithm`] and the +/// [`SketchParams`] realizing it. Sits between /// [`FieldDataType::Sketch`](crate::ir::schema::FieldDataType::Sketch) /// (the `Sketch` family as a whole, sibling to `SummarySample`/`Wavelet`/ -/// `StatModel`) and the bare algorithm: `Kll` vs. `DDSketch` is a choice -/// *within* `Quantile`, not a choice *of* `SketchKind` — every `Quantile` -/// value already carries which of the two (and its params) was picked. +/// `StatModel`) and the bare algorithm. Which statistics a state of the +/// kind answers is the family-capability table's +/// ([`FamilyCapability`](crate::ir::schema::FamilyCapability)). /// -/// [`SketchKind::new`] is the one place `(SketchAlgorithm, SketchParams)` -/// pairs get classified into a category; construct through it rather than -/// naming a variant directly, so a new algorithm can't drift out of sync -/// with its category. See `asap_logical_asap_optimization::summary_candidates` for the -/// `AggIntent -> [SketchAlgorithm]` candidate list this ultimately groups. +/// Construct through [`SketchKind::new`], which rejects parameters of a +/// different algorithm. See `asap_logical_asap_optimization::summary_candidates` for the +/// `AggIntent -> [SketchAlgorithm]` candidate list. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct SketchKind { - category: SketchCategory, algorithm: SketchAlgorithm, params: SketchParams, } impl SketchKind { - /// Classify `(algorithm, params)` into its `SketchKind` category. The - /// one place that mapping is made — every other piece of this crate - /// that needs to know an algorithm's category goes through this rather - /// than re-deriving it. Panics when `params` belongs to a different - /// algorithm; invalid committed sketch states cannot be constructed. + /// The kind of `algorithm` with `params`. Panics when `params` belongs + /// to a different algorithm; invalid committed sketch states cannot be + /// constructed. pub fn new(algorithm: SketchAlgorithm, params: SketchParams) -> Self { let valid_params = matches!( (&algorithm, ¶ms), @@ -313,34 +295,15 @@ impl SketchKind { valid_params, "SketchKind parameter mismatch: algorithm={algorithm:?}, params={params:?}" ); - let category = match algorithm { - SketchAlgorithm::UnivMon => SketchCategory::Universal, - SketchAlgorithm::Kll | SketchAlgorithm::DDSketch => SketchCategory::Quantile, - SketchAlgorithm::Hll | SketchAlgorithm::Theta | SketchAlgorithm::Kmv => { - SketchCategory::Cardinality - } - SketchAlgorithm::Cms | SketchAlgorithm::CountSketch => SketchCategory::Frequency, - SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap => { - SketchCategory::TopK - } - }; - Self { - category, - algorithm, - params, - } - } - - pub fn category(&self) -> SketchCategory { - self.category + Self { algorithm, params } } - /// The algorithm this kind committed to, regardless of category. + /// The algorithm this kind committed to. pub fn algorithm(&self) -> &SketchAlgorithm { &self.algorithm } - /// The parameters this kind committed to, regardless of category. + /// The parameters this kind committed to. pub fn params(&self) -> &SketchParams { &self.params } @@ -870,7 +833,7 @@ mod tests { } #[test] - fn sketch_kind_classifies_a_valid_algorithm_and_params_pair() { + fn sketch_kind_keeps_a_valid_algorithm_and_params_pair() { let kind = SketchKind::new( SketchAlgorithm::Kll, SketchParams::Kll { @@ -878,7 +841,6 @@ mod tests { failure_probability: None, }, ); - assert_eq!(kind.category(), SketchCategory::Quantile); assert_eq!(kind.algorithm(), &SketchAlgorithm::Kll); } diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index 34b1bbcfb..04bdfa89d 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -210,7 +210,7 @@ fn family_capability_table() { SketchParams::DDSketch { alpha: 0.01 }, ), Some(Disjoint), - vec![(Quantile, Union), (Total, Union)], + vec![(Quantile, Union)], ), (hll(), Some(MayOverlap), vec![(Distinct, Union)]), ( @@ -224,7 +224,7 @@ fn family_capability_table() { SketchParams::CountSketch { width, depth }, ), Some(Disjoint), - vec![(ItemFrequency, Union), (FrequencyL2, Union)], + vec![(ItemFrequency, Union)], ), ( sketch( @@ -263,11 +263,7 @@ fn family_capability_table() { None, vec![(Distinct, Approximate)], ), - ( - cms_with_heap(), - Some(Disjoint), - vec![(TopK, Approximate), (ItemFrequency, Union)], - ), + (cms_with_heap(), Some(Disjoint), vec![(TopK, Approximate)]), ( sketch( SketchAlgorithm::CountSketchWithHeap, @@ -278,7 +274,7 @@ fn family_capability_table() { }, ), Some(Disjoint), - vec![(TopK, Approximate), (ItemFrequency, Union)], + vec![(TopK, Approximate)], ), // Hydra over a linear sketch is a grid of linear sketches; Hydra // over KLL has no accuracy model and fails closed. @@ -383,15 +379,13 @@ fn family_capability_table() { row.is_some(), "{family:?} {statistic:?}" ); - // An answered statistic's merged error is its row's; the - // default for the others is pinned below. - if row.is_some() || merge_inputs.is_none() { - assert_eq!( - capability.merged_error(statistic), - merge_inputs.and(row), - "{family:?} {statistic:?}" - ); - } + // A merged state's error is its row's; none for a statistic + // the family does not answer. + assert_eq!( + capability.merged_error(statistic), + merge_inputs.and(row), + "{family:?} {statistic:?}" + ); } } } @@ -400,9 +394,8 @@ fn family_capability_table() { /// part's heaps keep only that part's heavy items. UnivMon's L2 reads layer /// 0's Count Sketch, which adds, and its distinct count's bound holds for /// any heap that keeps `min(|S_j|, k)` of a layer's keys, as the merge's -/// does, so both keep the guarantee of one build over the union. Counter -/// readouts of a heap family (item frequencies) keep their guarantee. A -/// statistic the family does not answer takes the family's default. +/// does, so both keep the guarantee of one build over the union. A +/// statistic the family does not answer has no merged error. #[test] fn merged_error_depends_on_the_readout() { use SketchStatistic::{Cardinality, FrequencyEntropy, FrequencyL2, TopK}; @@ -428,33 +421,15 @@ fn merged_error_depends_on_the_readout() { ] { assert_eq!(univmon.merged_error(&readout), Some(error), "{readout:?}"); } - // A statistic the family does not answer takes its default: a heavy - // item readout of a family with heaps is approximate. - assert_eq!( - univmon.merged_error(&TopK { k: 5 }), - Some(MergedError::Approximate) - ); + assert_eq!(univmon.merged_error(&TopK { k: 5 }), None); let heap = FamilyCapability::of(&cms_with_heap()); assert_eq!( heap.merged_error(&TopK { k: 5 }), Some(MergedError::Approximate) ); - let item = SketchStatistic::PointCount { - key: ColumnRef::Named("k".into()), - value: Some("a".into()), - }; - assert_eq!( - heap.merged_error(&item), - Some(MergedError::AsIfBuiltOverUnion) - ); - // A CMS with a heap answers item frequencies, not its total, whose - // default is the guarantee of one build over the union. - assert_eq!( - heap.merged_error(&point()), - Some(MergedError::AsIfBuiltOverUnion) - ); - // Pass 1 offers CountSketch for a count (a bare point count), which - // its row does not answer: merged panes keep one build's guarantee. + assert_eq!(heap.merged_error(&point()), None); + // Count Sketch's signed counters give no total: no merged error for a + // bare point count, and one build's guarantee for an item's frequency. let count_sketch = FamilyCapability::of(&sketch( SketchAlgorithm::CountSketch, SketchParams::CountSketch { @@ -463,16 +438,18 @@ fn merged_error_depends_on_the_readout() { }, )); assert!(!count_sketch.answers(&point())); + assert_eq!(count_sketch.merged_error(&point()), None); + let item = SketchStatistic::PointCount { + key: ColumnRef::Named("k".into()), + value: Some("a".into()), + }; assert_eq!( - count_sketch.merged_error(&point()), + count_sketch.merged_error(&item), Some(MergedError::AsIfBuiltOverUnion) ); - // An exact accumulator's merged state is exact. + // An exact accumulator is read as its finalized value, not a readout. let sum = exact(ExactKind::Sum, ExactParams::Sum); - assert_eq!( - FamilyCapability::of(&sum).merged_error(&point()), - Some(MergedError::Exact) - ); + assert_eq!(FamilyCapability::of(&sum).merged_error(&point()), None); assert_eq!( FamilyCapability::of(&hll()).merged_error(&Cardinality), Some(MergedError::AsIfBuiltOverUnion) @@ -485,6 +462,205 @@ fn merged_error_depends_on_the_readout() { assert_eq!(FamilyCapability::of(&irate).merged_error(&point()), None); } +/// A structurally validated `SummaryEstimate` of `query` over a `family` +/// state. +fn estimate(family: FieldDataType, query: SketchStatistic) -> Result, String> { + let node = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: family_state(family, "us"), + query, + })) + .map_err(|e| e.to_string())?; + node.validate_structure().map_err(|e| e.to_string())?; + Ok(node) +} + +/// A statistic the family does not answer is rejected: a DDSketch's or an +/// HLL's total, a quantile over a Count-Min, top-k over a KLL, and any +/// readout of an exact accumulator. What the table answers builds. +#[test] +fn a_statistic_the_family_does_not_answer_is_rejected() { + let cms = sketch( + SketchAlgorithm::Cms, + SketchParams::Cms { + width: 64, + depth: 4, + }, + ); + let ddsketch = sketch( + SketchAlgorithm::DDSketch, + SketchParams::DDSketch { alpha: 0.01 }, + ); + let total = || SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }; + for (family, query) in [ + (ddsketch.clone(), total()), + (hll(), total()), + (cms.clone(), SketchStatistic::Quantile { q: 0.9 }), + (kll(), SketchStatistic::TopK { k: 5 }), + (exact(ExactKind::Count, ExactParams::Count), total()), + ] { + let error = estimate(family.clone(), query.clone()).unwrap_err(); + assert!( + error.contains("evaluation does not match its summary family"), + "{family:?} {query:?}: {error}" + ); + } + estimate(cms, total()).unwrap(); + estimate(ddsketch, SketchStatistic::Quantile { q: 0.9 }).unwrap(); + estimate(hll(), SketchStatistic::Cardinality).unwrap(); +} + +/// What the `SummaryEstimate` validation accepted before it read the +/// family-capability table: a match on the algorithm's category, with +/// Hydra classified by its inner algorithm. +fn accepted_by_category(algorithm: &SketchAlgorithm, query: &SketchStatistic) -> bool { + use SketchAlgorithm as A; + use SketchStatistic as Q; + match algorithm { + A::Kll | A::DDSketch => matches!(query, Q::Quantile { .. }), + A::Hll | A::Theta | A::Kmv => matches!(query, Q::Cardinality), + A::Cms | A::CountSketch => matches!(query, Q::PointCount { .. }), + A::CmsWithHeap | A::CountSketchWithHeap => matches!(query, Q::TopK { .. }), + A::UnivMon => !matches!(query, Q::Quantile { .. }), + } +} + +/// Reading the family-capability table changes what the `SummaryEstimate` +/// validation accepts only where the old category match was wrong: a +/// UnivMon answers no item frequency or top-k, a Count Sketch's signed +/// counters give no total, and a Hydra grid answers an item's frequency +/// within a group, not the group's total. Every other (family, statistic) +/// pair keeps its answer. +#[test] +fn estimate_validation_matches_the_category_match_except_the_fixed_rows() { + let (width, depth) = (64, 4); + let mut families: Vec = [ + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + SketchParams::DDSketch { alpha: 0.01 }, + SketchParams::Hll { + precision: 12, + relative_error: None, + }, + SketchParams::Theta { k: 1024 }, + SketchParams::Kmv { + k: 1024, + relative_error: None, + }, + SketchParams::Cms { width, depth }, + SketchParams::CountSketch { width, depth }, + SketchParams::CmsWithHeap { + width, + depth, + heap_size: 10, + }, + SketchParams::CountSketchWithHeap { + width, + depth, + heap_size: 10, + }, + SketchParams::UnivMon { + heap_size: 10, + sketch_rows: depth, + sketch_cols: width, + layers: 8, + distinct_failure_probability: None, + }, + ] + .into_iter() + .map(|params| { + use SketchAlgorithm as A; + let algorithm = match ¶ms { + SketchParams::Kll { .. } => A::Kll, + SketchParams::DDSketch { .. } => A::DDSketch, + SketchParams::Hll { .. } => A::Hll, + SketchParams::Theta { .. } => A::Theta, + SketchParams::Kmv { .. } => A::Kmv, + SketchParams::Cms { .. } => A::Cms, + SketchParams::CountSketch { .. } => A::CountSketch, + SketchParams::CmsWithHeap { .. } => A::CmsWithHeap, + SketchParams::CountSketchWithHeap { .. } => A::CountSketchWithHeap, + SketchParams::UnivMon { .. } => A::UnivMon, + }; + sketch(algorithm, params) + }) + .collect(); + families.extend([ + hydra( + SketchAlgorithm::Cms, + SketchParams::Cms { width, depth }, + HydraKind::HydraCms, + HydraParams::HydraCms { + width, + depth, + shared_rows: depth, + shared_columns: width, + }, + ), + hydra( + SketchAlgorithm::CountSketch, + SketchParams::CountSketch { width, depth }, + HydraKind::HydraCountSketch, + HydraParams::HydraCountSketch { + width, + depth, + shared_rows: depth, + shared_columns: width, + }, + ), + hydra( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + HydraKind::HydraKll, + HydraParams::HydraKll { + k: 200, + shared_buckets: width, + }, + ), + ]); + let statistics = [ + SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }, + SketchStatistic::PointCount { + key: ColumnRef::Named("item".into()), + value: Some("a".into()), + }, + SketchStatistic::Quantile { q: 0.99 }, + SketchStatistic::Cardinality, + SketchStatistic::FrequencyL2, + SketchStatistic::FrequencyEntropy, + SketchStatistic::TopK { k: 5 }, + ]; + for family in families { + let FieldDataType::Sketch(kind, grouping) = &family else { + unreachable!() + }; + let hydra = *grouping != GroupingStrategy::PerSubpopulationInstance; + for query in &statistics { + let fixed = match (kind.algorithm(), query.kind()) { + (SketchAlgorithm::UnivMon, StatisticKind::ItemFrequency | StatisticKind::TopK) => { + true + } + (SketchAlgorithm::CountSketch, StatisticKind::Total) => true, + (SketchAlgorithm::Cms, StatisticKind::Total) => hydra, + _ => false, + }; + let before = accepted_by_category(kind.algorithm(), query); + let after = estimate(family.clone(), query.clone()).is_ok(); + assert_eq!(after, before != fixed, "{family:?} {query:?}"); + } + } +} + /// 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. diff --git a/docs/design_docs/proposals/asap-primitive-schema.md b/docs/design_docs/proposals/asap-primitive-schema.md index 6bc3a1b19..c58989aa6 100644 --- a/docs/design_docs/proposals/asap-primitive-schema.md +++ b/docs/design_docs/proposals/asap-primitive-schema.md @@ -68,17 +68,16 @@ Based on our requirement, each field should contain the following information. 1. **What type of the ASAP Primitive is.** The field's type is a [`FieldDataType`](#63-schema-and-field-types-cratestypessrcirschemamodrs) (`crates/types/src/ir/schema/mod.rs`). A column is either a raw value or a summary state: - **Raw value**: `Plain(DataType)`, e.g., a number or a string. - - **Summary state**: described from coarse to fine by four levels: + - **Summary state**: described from coarse to fine by three levels: | Level | Code | Values | KLL example | |---|---|---|---| | Family | `FieldDataType` variant | `ExactAggregate`, `Sketch`, `SummarySample`, `Wavelet`, `StatModel` | `Sketch` | - | Category (sketches only) | `SketchCategory` | `Quantile`, `Frequency`, `Cardinality`, `TopK`, `Universal` | `Quantile` | | Algorithm | `SketchAlgorithm` (other families: `ExactKind`, `SamplingKind`, …) | `Kll`, `Cms`, `Hll`, `DDSketch`, … | `Kll` | | Parameters | `SketchParams` (other families: `ExactParams`, `SamplingParams`, …) | per algorithm | `Kll { k: 200 }` | - - For a sketch, category, algorithm and parameters are bundled as one `SketchKind` ([§6.4](#64-summary-family-parameters-cratestypessrcirschemasketchrs)). A sketch also records its `GroupingStrategy`: one instance per group, or one shared structure (Hydra) for all groups. - - So a quantile KLL sketch with `k = 200`, one instance per group, has the type `Sketch(SketchKind { Quantile, Kll, Kll { k: 200 } }, PerSubpopulationInstance)`. + - For a sketch, algorithm and parameters are bundled as one `SketchKind` ([§6.4](#64-summary-family-parameters-cratestypessrcirschemasketchrs)). A sketch also records its `GroupingStrategy`: one instance per group, or one shared structure (Hydra) for all groups. + - So a quantile KLL sketch with `k = 200`, one instance per group, has the type `Sketch(SketchKind { Kll, Kll { k: 200 } }, PerSubpopulationInstance)`. Which statistics a family answers is the family-capability table's (`FamilyCapability`), not part of the type. 2. **What query intent the summarized ASAP Primitive can support.** This is not stored in the field: it follows from the type in item 1. There are two kinds of intent: @@ -88,7 +87,7 @@ Based on our requirement, each field should contain the following information. |---|---|---| | `Quantile` | `Sketch`: `Kll`, `DDSketch` | `SummaryEstimate(Quantile { q })` | | `Cardinality` | `Sketch`: `Hll`, `Kmv`, `UnivMon` (one column only) | `SummaryEstimate(Cardinality)` | - | `Count` (approximate) | `Sketch`: `Cms`, `CountSketch`, `UnivMon` | `SummaryEstimate(PointCount { .. })` | + | `Count` (approximate) | `Sketch`: `Cms`, `UnivMon` | `SummaryEstimate(PointCount { .. })` | | `TopK` | `Sketch`: `CmsWithHeap`, `CountSketchWithHeap` | `SummaryEstimate(TopK { k })` | | `FrequencyL2`, `FrequencyEntropy` | `Sketch`: `UnivMon` | `SummaryEstimate(FrequencyL2 \| FrequencyEntropy)` | | `Sum`, `Count`, `Min`, `Max`, `Rate`, `IRate`, `Increase` (exact) | `ExactAggregate(ExactKind, …)` | `FinalizeExactAccumulator` | @@ -834,12 +833,12 @@ input groups output groups | Exact `Sum`, `Count`, `Min`, `Max` | adds, or keeps the extremum | exact | | KMV | keeps the `k` smallest distinct hashes of the union (one shared hash seed) | the same state as a direct build: relative error of the distinct count | | HLL | keeps each register's maximum | the same state as a direct build: relative error of the distinct count | -| DDSketch | adds the bucket counts (one shared bucket mapping) | the same state as a direct build: relative error of every quantile; its total count | +| DDSketch | adds the bucket counts (one shared bucket mapping) | the same state as a direct build: relative error of every quantile | | KLL | concatenates and compacts the levels | not the same state (compactions are randomized), but the same rank error bound and failure probability as a direct build of that `k` | | Count-Min | adds the counters (shared hashes) | the same state as a direct build: item frequency and total, the error relative to the coarse group's total, `ε·‖group‖₁` | -| Count Sketch | adds the counters (shared hashes) | the same state as a direct build: item frequency, the error relative to the coarse group's L2 norm, `ε·‖group‖₂`; the L2 estimate | +| Count Sketch | adds the counters (shared hashes) | the same state as a direct build: item frequency, the error relative to the coarse group's L2 norm, `ε·‖group‖₂` | | UnivMon | adds the layer counters and re-ranks the union of the heap keys | total, L2 and distinct count as a direct build; entropy approximate | -| Count-Min or Count Sketch with heap | adds the counters | item frequencies as a direct build; top-k approximate (an item heavy in the coarse group may be in no finer group's heap) | +| Count-Min or Count Sketch with heap | adds the counters | top-k approximate (an item heavy in the coarse group may be in no finer group's heap) | Rejected: counter states (`Rate`, `Increase`), which merge consecutive windows of one series, not different series; `IRate` and Theta, which have no merge; Hydra, whose groups share one state, so they are no `group_by` columns. @@ -886,7 +885,7 @@ OperatorNode §6.1 ├── schema: Schema │ §6.3 │ └── fields[i].dtype: FieldDataType ◀───────┘ §6.3 (the same type: the output │ └── Sketch(SketchKind, GroupingStrategy) `state` field has type `family`) -│ └── category, algorithm, params §6.4 +│ └── algorithm, params §6.4 ├── guarantee: AccuracyGuarantee §6.5 (accuracy of an evaluation) └── coverage() → SummaryCoverage §6.6 └── definition: OperatorNode (again a node, so the same structure) @@ -1118,7 +1117,7 @@ pub enum FieldDataType { Plain(DataType), /// Exact accumulator state: which accumulator, and its parameters. ExactAggregate(ExactKind, ExactParams), - /// Sketch state: the chosen sketch (category, algorithm, parameters), and + /// Sketch state: the chosen sketch (algorithm, parameters), and /// whether each group has its own sketch or all groups share one. Sketch(SketchKind, GroupingStrategy), /// Summary sample state: the sampling method, and its parameters. @@ -1158,33 +1157,25 @@ pub enum DataType { pub enum ExactKind { Sum, Count, Min, Max, Increase, Rate, IRate } pub enum ExactParams { Sum, Count, Min, Max, Increase, Rate, IRate } -/// A chosen sketch. Built only through `new`, so the three fields always agree. +/// A chosen sketch. Built only through `new`, so the two fields always agree. +/// Which statistics it answers is its family's row of the family-capability +/// table (`FamilyCapability`). pub struct SketchKind { - /// Which aggregation intents the sketch answers: `Quantile` (quantiles), - /// `Cardinality` (distinct counts), `Frequency` (item counts), `TopK` - /// (heavy hitters), or `Universal` (frequency moments such as L2 and - /// entropy, plus counts, distinct counts and top-k). Derived from `algorithm`. - category: SketchCategory, /// Which sketch algorithm. algorithm: SketchAlgorithm, /// That algorithm's size parameters; must match `algorithm`. params: SketchParams, } impl SketchKind { - /// The only constructor. Derives the category from the algorithm, and - /// panics if `params` belong to a different algorithm. + /// The only constructor. Panics if `params` belong to a different + /// algorithm. pub fn new(algorithm: SketchAlgorithm, params: SketchParams) -> Self; - /// Which aggregation intents the sketch answers (see the `category` field). - pub fn category(&self) -> SketchCategory; /// The sketch algorithm. pub fn algorithm(&self) -> &SketchAlgorithm; /// The algorithm's size parameters. pub fn params(&self) -> &SketchParams; } -pub enum SketchCategory { Universal, Quantile, Cardinality, Frequency, TopK } -// Universal: UnivMon | Quantile: Kll, DDSketch | Cardinality: Hll, Theta, Kmv -// (Theta is not offered as a candidate until asap_sketchlib has a Theta sketch) -// Frequency: Cms, CountSketch | TopK: CmsWithHeap, CountSketchWithHeap +// Theta is not offered as a candidate until asap_sketchlib has a Theta sketch. pub enum SketchAlgorithm { UnivMon, Kll, Cms, Hll, DDSketch, CmsWithHeap, Kmv, Theta, CountSketch, CountSketchWithHeap } /// Size parameters of each algorithm. Larger values: more memory, less error.