diff --git a/Cargo.lock b/Cargo.lock index 910e48b62..3b4c0a73f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -443,6 +443,7 @@ dependencies = [ name = "asap-plan-selection" version = "0.1.0" dependencies = [ + "asap-executor", "asap-frontend-promql", "asap-logical-asap-optimization", "asap-physical-asap-optimization", @@ -485,6 +486,7 @@ version = "0.1.0" name = "asap-types" version = "0.1.0" dependencies = [ + "asap_sketchlib", "serde", "serde_json", "thiserror 2.0.18", @@ -493,7 +495,7 @@ dependencies = [ [[package]] name = "asap_sketchlib" version = "0.4.0" -source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=63e565506426020c505ef0b84e0c91f450d09e7c#63e565506426020c505ef0b84e0c91f450d09e7c" +source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=21b63578c2e612e153f5aaa4d3848f1035a13c60#21b63578c2e612e153f5aaa4d3848f1035a13c60" dependencies = [ "rand 0.9.5", "rmp-serde", diff --git a/Cargo.toml b/Cargo.toml index e69bed7fd..12c98166a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,8 +23,10 @@ resolver = "2" arrow = { version = "58.4", default-features = false } # The hasher Arrow already uses, for the executor's hot interning maps. ahash = { version = "0.8", default-features = false, features = ["std", "runtime-rng"] } -# `planner-pin/0.4.0-plus`: release 0.4.0 (eb04e7f) plus the open sketchlib -# PRs #245 (checked f64 updates, heap rescoring), #246 (KLL lazy-buffer -# follow-ups), #247 (Count Sketch stack medians) and #248 -# (`DdSketch::quantile_interpolated`). Moves to a release once they merge. -asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "63e565506426020c505ef0b84e0c91f450d09e7c" } +# main after release 0.4.0 (eb04e7f): #245 (checked f64 updates, heap +# rescoring), #246 (KLL lazy-buffer follow-ups), #247 (Count Sketch stack +# medians), #248 (`DdSketch::quantile_interpolated`), #249 (`memory_bytes` and +# `estimated_memory_bytes`; KMV without `experimental`), #251 (the public +# capability table) and #254 (Planner names in that table). Moves to 0.4.1 +# once that is released. +asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "21b63578c2e612e153f5aaa4d3848f1035a13c60" } diff --git a/crates/executor/Cargo.toml b/crates/executor/Cargo.toml index db84de459..b577fa4d6 100644 --- a/crates/executor/Cargo.toml +++ b/crates/executor/Cargo.toml @@ -9,7 +9,7 @@ ahash.workspace = true chrono = { version = "0.4.39", default-features = false, features = ["std"] } futures = "0.3" planner-types = { package = "asap-types", path = "../types" } -asap_sketchlib = { workspace = true, features = ["experimental"] } +asap_sketchlib = { workspace = true } serde = { version = "1", features = ["derive", "rc"] } serde_json = "1" thiserror = "2" diff --git a/crates/executor/src/capability.rs b/crates/executor/src/capability.rs index e4e4a4d95..f7a2e3afd 100644 --- a/crates/executor/src/capability.rs +++ b/crates/executor/src/capability.rs @@ -325,37 +325,41 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> { .map_err(Error::Invalid) } -/// 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, +/// What the executor evaluates from a per-instance state of each family it +/// builds and reads: its supported-summary declaration. Every readout here is +/// in the family's row of the family-capability table; a readout the row +/// lists and this omits is not evaluated yet (Count-Min's and the heap +/// families' item frequency, which the executor answers only from a Hydra +/// grid; KLL's rank; UnivMon's G-sum). Plain Count Sketch is built but read +/// from nothing. +pub(crate) const EVALUATED_READOUTS: &[(SketchAlgorithm, &[StatisticKind])] = &[ + (SketchAlgorithm::Cms, &[StatisticKind::Total]), + (SketchAlgorithm::CmsWithHeap, &[StatisticKind::TopK]), + (SketchAlgorithm::CountSketchWithHeap, &[StatisticKind::TopK]), + (SketchAlgorithm::Hll, &[StatisticKind::Distinct]), + (SketchAlgorithm::Kmv, &[StatisticKind::Distinct]), + (SketchAlgorithm::Kll, &[StatisticKind::Quantile]), + (SketchAlgorithm::DDSketch, &[StatisticKind::Quantile]), + ( + SketchAlgorithm::UnivMon, + &[ + StatisticKind::Total, + StatisticKind::FrequencyL2, + StatisticKind::Distinct, + StatisticKind::FrequencyEntropy, + ], + ), ]; -/// 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)) + EVALUATED_READOUTS + .iter() + .any(|(family, readouts)| family == kind.algorithm() && readouts.contains(&readout)) } SummaryFamilyType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => { matches!(kind, HydraKind::HydraCms | HydraKind::HydraKll) diff --git a/crates/executor/src/summary_kernels/datasketches_kll.rs b/crates/executor/src/summary_kernels/datasketches_kll.rs index e5dcf0848..80c40c1a9 100644 --- a/crates/executor/src/summary_kernels/datasketches_kll.rs +++ b/crates/executor/src/summary_kernels/datasketches_kll.rs @@ -54,11 +54,13 @@ impl AggregateCore for DatasketchesKLLAccumulator { .as_any() .downcast_ref::() .ok_or("KLL merges only with KLL")?; - if self.inner.k() != other.inner.k() { + // sketchlib's merge panics on a different k or m. + let (k, m) = (self.inner.k(), self.inner.m()); + if (k, m) != (other.inner.k(), other.inner.m()) { return Err(format!( - "KLL merges only with the same k ({} vs {})", - self.inner.k(), - other.inner.k() + "KLL merges only with the same k and m ({k}, {m} vs {}, {})", + other.inner.k(), + other.inner.m() ) .into()); } @@ -136,13 +138,18 @@ mod tests { assert!(kll.approx_memory_bytes() > empty); } - // States of different k do not merge. + // States of different k or m do not merge: an error, not sketchlib's panic. #[test] - fn merge_rejects_a_different_k() { + fn merge_rejects_a_different_k_or_m() { let kll = DatasketchesKLLAccumulator::new(200); assert!(kll .merge_with(&DatasketchesKLLAccumulator::new(100)) .is_err()); + assert!(kll + .merge_with(&DatasketchesKLLAccumulator { + inner: Kll::new(200, 4), + }) + .is_err()); assert!(kll .merge_with(&DatasketchesKLLAccumulator::new(200)) .is_ok()); diff --git a/crates/executor/src/summary_kernels/dd_sketch.rs b/crates/executor/src/summary_kernels/dd_sketch.rs index a3e31ac08..295eb8e5f 100644 --- a/crates/executor/src/summary_kernels/dd_sketch.rs +++ b/crates/executor/src/summary_kernels/dd_sketch.rs @@ -42,6 +42,12 @@ impl AggregateCore for DDSketchAccumulator { self.inner.get_count() == 0 } + /// sketchlib's count of what the sketch holds now (its struct + /// included); the bucket stores grow with the values' range. + fn approx_memory_bytes(&self) -> usize { + self.inner.memory_bytes() + } + /// Quantiles. fn estimate(&self, query: &SketchStatistic) -> Result { match query { diff --git a/crates/executor/src/summary_kernels/factory.rs b/crates/executor/src/summary_kernels/factory.rs index 344bec1a0..b650f9bd4 100644 --- a/crates/executor/src/summary_kernels/factory.rs +++ b/crates/executor/src/summary_kernels/factory.rs @@ -263,11 +263,7 @@ impl AccumulatorUpdater for DDSketchAccumulatorUpdater { } fn memory_usage_bytes(&self) -> usize { - // The struct and the 8-byte counts both bucket stores allocate, - // which grow with the values' range. - let sketch = &self.acc.inner; - let bins = sketch.positive_store_counts().len() + sketch.negative_store_counts().len(); - std::mem::size_of::() + bins * std::mem::size_of::() + self.acc.approx_memory_bytes() } } @@ -716,10 +712,8 @@ mod planner_parameter_regression { }, ), ] { - let family = SummaryFamilyType::Sketch( - SketchKind::new(algorithm.clone(), params), - Default::default(), - ); + let family = + SummaryFamilyType::Sketch(SketchKind::new(algorithm, params), Default::default()); let update = SummaryUpdate { item: Some(SummaryInputExpr::Column( planner_types::ir::scalar::ColumnRef::Named("host".into()), @@ -872,9 +866,9 @@ mod ddsketch_domain { mod tests { use super::*; - /// A DDSketch updater reports the bins its stores allocate: none - /// before a sample, one 128-bin chunk of the positive store after a - /// few positive samples. + /// A DDSketch updater reports sketchlib's count of the bins its stores + /// allocate: none before a sample, one 128-bin chunk of the positive + /// store after a few positive samples. #[test] fn ddsketch_memory_is_its_allocated_bins() { let mut updater = DDSketchAccumulatorUpdater::new(0.05); @@ -884,6 +878,10 @@ mod tests { updater.update_single(value, 0).unwrap(); } assert_eq!(updater.memory_usage_bytes(), empty + 128 * 8); + assert_eq!( + updater.memory_usage_bytes(), + updater.acc.inner.memory_bytes() + ); } /// A numeric sketch takes an Int64 as its exact Float64, and refuses diff --git a/crates/executor/src/summary_kernels/univmon.rs b/crates/executor/src/summary_kernels/univmon.rs index 2ca679eff..508e535cc 100644 --- a/crates/executor/src/summary_kernels/univmon.rs +++ b/crates/executor/src/summary_kernels/univmon.rs @@ -160,29 +160,30 @@ impl UnivMonAccumulator { } impl AggregateCore for UnivMonAccumulator { - /// Counters and heap slots of every layer, plus an upper bound on the - /// variable-length keys the heaps retain: the smaller of every slot - /// holding the longest key seen and every layer holding a copy of every - /// key inserted. Constant time, since a build charges memory after - /// every row; the library exposes no eviction hook for an exact running - /// total, so a stream of long keys is still overcounted. + /// This accumulator's own fields, sketchlib's bytes for the sketch's + /// shape with keys that own no bytes ([`UnivMon::estimated_memory_bytes`]: + /// at least what it holds), and an upper bound on the variable-length + /// keys the heaps retain: the smaller of every slot holding the longest + /// key seen and every layer holding a copy of every key inserted. + /// Constant time, since a build charges memory after every row and + /// [`UnivMon::memory_bytes`] walks the heaps; the library exposes no + /// eviction hook for an exact running total, so a stream of long keys is + /// still overcounted. fn approx_memory_bytes(&self) -> usize { - let layers = self.inner.layers; + let (heap_size, rows, cols, layers) = self.dimensions(); let key_bytes = layers - .saturating_mul(self.inner.heap_size) + .saturating_mul(heap_size) .saturating_mul(self.max_key_bytes) .min(layers.saturating_mul(self.inserted_key_bytes)); - key_bytes - .saturating_add(std::mem::size_of::()) - .saturating_add( - self.inner.layers.saturating_mul( - self.inner - .rows - .saturating_mul(self.inner.cols) - .saturating_mul(16) - .saturating_add(self.inner.heap_size.saturating_mul(256)), - ), - ) + (std::mem::size_of::() - std::mem::size_of::()) + .saturating_add(UnivMon::estimated_memory_bytes( + heap_size, + rows, + cols, + layers, + &DataInput::U64(0), + )) + .saturating_add(key_bytes) } fn clone_boxed_core(&self) -> Box { Box::new(self.clone()) diff --git a/crates/executor/src/supported_summaries.rs b/crates/executor/src/supported_summaries.rs index 6cd9afc82..6ad48c3e1 100644 --- a/crates/executor/src/supported_summaries.rs +++ b/crates/executor/src/supported_summaries.rs @@ -3,24 +3,18 @@ //! `DeploymentCapabilities::supported_summaries`, so the Planner offers no //! summary the executor would reject when it binds the plan. -use crate::capability::{EVALUATED_FAMILIES, UNEVALUATED_READOUTS}; -use planner_types::ir::schema::{FamilyCapability, SupportedSummary}; +use crate::capability::EVALUATED_READOUTS; +use planner_types::ir::schema::SupportedSummary; /// Every sketch family this executor builds from rows and the statistics it -/// 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. +/// answers from one state of it (`EVALUATED_READOUTS`). Exact accumulators +/// are always supported and are not listed. pub fn supported_summary_families() -> Vec { - EVALUATED_FAMILIES - .into_iter() - .map(|family| SupportedSummary { - readouts: FamilyCapability::of_per_instance_sketch(&family) - .readouts - .iter() - .map(|&(readout, _)| readout) - .filter(|readout| !UNEVALUATED_READOUTS.contains(&(family.clone(), *readout))) - .collect(), + EVALUATED_READOUTS + .iter() + .map(|&(family, readouts)| SupportedSummary { family, + readouts: readouts.to_vec(), }) .collect() } @@ -34,12 +28,13 @@ mod tests { use planner_types::ir::operator::operator_properties::{Reduction, Source}; use planner_types::ir::scalar::ColumnRef; use planner_types::ir::schema::{ - DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchKind, - SketchParams, SketchStatistic, StatisticKind, SummaryInputExpr, SummaryUpdate, + DataType, FamilyCapability, Field, FieldDataType, GroupingStrategy, Schema, + SketchAlgorithm, SketchKind, SketchParams, SketchStatistic, StatisticKind, + SummaryInputExpr, SummaryUpdate, }; use planner_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode}; - const ALGORITHMS: [SketchAlgorithm; 10] = [ + const ALGORITHMS: [SketchAlgorithm; 9] = [ SketchAlgorithm::UnivMon, SketchAlgorithm::Kll, SketchAlgorithm::Cms, @@ -47,7 +42,6 @@ mod tests { SketchAlgorithm::DDSketch, SketchAlgorithm::CmsWithHeap, SketchAlgorithm::Kmv, - SketchAlgorithm::Theta, SketchAlgorithm::CountSketch, SketchAlgorithm::CountSketchWithHeap, ]; @@ -95,7 +89,6 @@ mod tests { k: 1024, relative_error: None, }, - A::Theta => SketchParams::Theta { k: 1024 }, A::CountSketch => SketchParams::CountSketch { width: 64, depth: 3, @@ -105,12 +98,13 @@ mod tests { depth: 3, heap_size: 10, }, + other => unreachable!("{other:?} is not probed"), } } fn per_instance(algorithm: &SketchAlgorithm) -> FieldDataType { FieldDataType::Sketch( - SketchKind::new(algorithm.clone(), params(algorithm)), + SketchKind::new(*algorithm, params(algorithm)), GroupingStrategy::PerSubpopulationInstance, ) } @@ -133,6 +127,7 @@ mod tests { StatisticKind::FrequencyL2 => SketchStatistic::FrequencyL2, StatisticKind::FrequencyEntropy => SketchStatistic::FrequencyEntropy, StatisticKind::TopK => SketchStatistic::TopK { k: 5 }, + other => unreachable!("{other:?} is not probed"), } } @@ -230,9 +225,8 @@ mod tests { /// The list holds exactly the (family, readout) pairs the factory builds /// 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. + /// family-capability table answers. Every declared readout is in the + /// family's table row. #[test] fn supported_summary_families_match_the_factory_and_evaluations() { let listed = supported_summary_families(); @@ -252,11 +246,10 @@ mod tests { 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!( + !declared || answered, + "declared but not in the table: {algorithm:?} {readout:?}" + ); } } assert!(listed.iter().all(|s| !s.readouts.is_empty())); diff --git a/crates/executor/tests/deployment.rs b/crates/executor/tests/deployment.rs index dd61f4c5b..c9e2da35f 100644 --- a/crates/executor/tests/deployment.rs +++ b/crates/executor/tests/deployment.rs @@ -311,7 +311,7 @@ fn factory_and_dag_heap_states_are_interchangeable() { FrequencyAlgorithm::CountSketch, ), ] { - let family = heap_sketch(algorithm.clone(), 1_024, 5, 4); + let family = heap_sketch(algorithm, 1_024, 5, 4); let factory = build_weighted_frequency(&family, [(0, 1.0), (1, 2.0)]); let mut dag = WeightedFrequency::new(frequency, 1_024, 5, 4).unwrap(); dag.update(&[Value::Utf8("host-0".into())], 3.0).unwrap(); diff --git a/crates/frontend-promql/tests/count_planning.rs b/crates/frontend-promql/tests/count_planning.rs index 6b965876b..491cadff5 100644 --- a/crates/frontend-promql/tests/count_planning.rs +++ b/crates/frontend-promql/tests/count_planning.rs @@ -109,7 +109,7 @@ fn frequency_count_candidates_use_unit_weights() { )), "post-ASAP DAG must preserve the count update contract" ); - algorithms.push(kind.algorithm().clone()); + algorithms.push(*kind.algorithm()); assert!(input.item.is_some(), "frequency keys must be explicit"); assert_eq!( input.weight, diff --git a/crates/integration-tests/tests/deployment_capabilities.rs b/crates/integration-tests/tests/deployment_capabilities.rs index 12ab1dcc2..8063c00e0 100644 --- a/crates/integration-tests/tests/deployment_capabilities.rs +++ b/crates/integration-tests/tests/deployment_capabilities.rs @@ -21,7 +21,7 @@ fn supporting(supported: &[(SketchAlgorithm, &[StatisticKind])]) -> DeploymentCa supported .iter() .map(|(family, readouts)| SupportedSummary { - family: family.clone(), + family: *family, readouts: readouts.to_vec(), }) .collect(), @@ -38,7 +38,7 @@ fn summaries_used(dag: &FlatDag) -> Vec<(SketchAlgorithm, StatisticKind)> { .flat_map(|(build, family, _)| { estimates_of(dag, build) .into_iter() - .map(move |(_, statistic)| (family.clone(), statistic.kind())) + .map(move |(_, statistic)| (family, statistic.kind())) }) .collect() } diff --git a/crates/integration-tests/tests/planner_layering_common/mod.rs b/crates/integration-tests/tests/planner_layering_common/mod.rs index 691f9d2e3..745b3fcea 100644 --- a/crates/integration-tests/tests/planner_layering_common/mod.rs +++ b/crates/integration-tests/tests/planner_layering_common/mod.rs @@ -562,7 +562,7 @@ pub fn sketch_builds(dag: &impl ExportedDag) -> Vec<(NodeId, SketchAlgorithm, Sk family: FieldDataType::Sketch(kind, _), .. }, - ) => Some((id, kind.algorithm().clone(), kind.params().clone())), + ) => Some((id, *kind.algorithm(), kind.params().clone())), _ => None, }) .collect() diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index 364338c7c..81d5c665c 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -258,7 +258,7 @@ fn planner_only_e2e_temporal_topk_preserves_query_update_and_evaluation_contract let candidates = candidates_using( &pre, AccuracyTarget::max_error(0.01).failure_probability(0.01), - expected_family.clone(), + expected_family, ); assert!( !candidates.is_empty(), @@ -771,7 +771,7 @@ fn ddsketch_ratio_selection( PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), .. - }) => Some(kind.algorithm().clone()), + }) => Some(*kind.algorithm()), _ => None, }) .collect(); diff --git a/crates/integration-tests/tests/sketch_rollup.rs b/crates/integration-tests/tests/sketch_rollup.rs index 9431a70a4..293847e94 100644 --- a/crates/integration-tests/tests/sketch_rollup.rs +++ b/crates/integration-tests/tests/sketch_rollup.rs @@ -249,7 +249,7 @@ async fn direct_and_rolled_up( ); let rolled = replace(&candidate, &build, &merge); rolled.validate_structure().unwrap(); - out.push((kind.algorithm().clone(), candidate, rolled)); + out.push((*kind.algorithm(), candidate, rolled)); } out } @@ -268,7 +268,7 @@ async fn distinct_counts_rolled_up_equal_the_direct_build() { accuracy, ) .await; - let families: Vec<_> = pairs.iter().map(|(family, ..)| family.clone()).collect(); + let families: Vec<_> = pairs.iter().map(|(family, ..)| *family).collect(); assert!(families.contains(&SketchAlgorithm::Kmv), "{families:?}"); for (family, direct, rolled) in pairs { let (direct, rolled) = (execute(&direct), execute(&rolled)); diff --git a/crates/integration-tests/tests/sql_two_percentiles.rs b/crates/integration-tests/tests/sql_two_percentiles.rs index d65ca8d36..76e0b2d00 100644 --- a/crates/integration-tests/tests/sql_two_percentiles.rs +++ b/crates/integration-tests/tests/sql_two_percentiles.rs @@ -212,7 +212,7 @@ fn sketch_of(root: &Rc) -> Option { family: FieldDataType::Sketch(kind, _), .. }, - ) => Some(kind.algorithm().clone()), + ) => Some(*kind.algorithm()), _ => None, }) } @@ -271,7 +271,7 @@ async fn q1_with_one_exact_and_one_sketched_percentile_compiles_and_matches() { for algorithm in [SketchAlgorithm::Kll, SketchAlgorithm::DDSketch] { let candidate = mixed .iter() - .find(|c| sketch_of(operator(&c.roots[0])) == Some(algorithm.clone())) + .find(|c| sketch_of(operator(&c.roots[0])) == Some(algorithm)) .unwrap_or_else(|| panic!("no mixed {algorithm:?} plan")); let root = operator(&candidate.roots[0]); // The percentile the exact aggregate computes. @@ -494,8 +494,7 @@ async fn q2_pane_chain_serves_q1_through_newest_panes_and_a_rollup() { let (id, roots) = candidates .iter() .find(|(_, roots)| { - newest_panes_rolled_up(roots) - && sketch_of(operator(&roots[0])) == Some(algorithm.clone()) + newest_panes_rolled_up(roots) && sketch_of(operator(&roots[0])) == Some(algorithm) }) .unwrap_or_else(|| panic!("no {algorithm:?} newest-panes rollup candidate")); let q1 = answers(operator(&roots[0])); diff --git a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs index 0ad4ae47e..c89aaf9c4 100644 --- a/crates/integration-tests/tests/stage1_logical_asap_optimization.rs +++ b/crates/integration-tests/tests/stage1_logical_asap_optimization.rs @@ -213,7 +213,7 @@ fn update_of(estimate: &OperatorNode) -> (SketchAlgorithm, SummaryUpdate) { else { panic!("sketch state expected") }; - (kind.algorithm().clone(), input.clone()) + (*kind.algorithm(), input.clone()) } /// A heap sketch's bound is per item frequency, not Top-K membership, and diff --git a/crates/logical-asap-optimization/src/accuracy/estimators/cardinality.rs b/crates/logical-asap-optimization/src/accuracy/estimators/cardinality.rs index e9cd9848f..8d6217911 100644 --- a/crates/logical-asap-optimization/src/accuracy/estimators/cardinality.rs +++ b/crates/logical-asap-optimization/src/accuracy/estimators/cardinality.rs @@ -1,4 +1,4 @@ -//! KMV's provable distinct-count bound and Theta's Chebyshev calibration. +//! KMV's provable distinct-count bound. //! //! # KMV (#694) //! @@ -57,15 +57,6 @@ pub(super) fn guarantee( "kmv_exact_below_k_chernoff_v1", )) } - SketchParams::Theta { k } => Some(super::bounded_guarantee( - algorithm, - params, - query, - AccuracyMetric::Relative, - 10.0 / f64::from(k.saturating_sub(2).max(1)).sqrt(), - ProbabilityExpr::Constant { value: 0.01 }, - "theta_variance_chebyshev_99_v1", - )), _ => None, } } @@ -119,12 +110,6 @@ pub(crate) fn kmv_k(eps: f64, delta: f64) -> u32 { hi } -/// 99%-confidence Theta relative bound via Chebyshev, using -/// `RSE <= 1/sqrt(k-2)` and a ten-standard-deviation interval. -pub(crate) fn theta_k_99(eps: f64) -> u32 { - saturating_ceil(100.0 / (eps * eps) + 2.0, 16, 1 << 26) -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/logical-asap-optimization/src/accuracy/estimators/cms.rs b/crates/logical-asap-optimization/src/accuracy/estimators/cms.rs index e7a85f364..df5dcb3d8 100644 --- a/crates/logical-asap-optimization/src/accuracy/estimators/cms.rs +++ b/crates/logical-asap-optimization/src/accuracy/estimators/cms.rs @@ -50,7 +50,10 @@ mod tests { SketchKind::new(SketchAlgorithm::Cms, params), GroupingStrategy::default(), ), - &SketchStatistic::Cardinality, + &SketchStatistic::PointCount { + key: asap_types::ir::scalar::expr_ir::ColumnRef::SampleValue, + value: None, + }, ) .unwrap(); assert_eq!(g.spec.accuracy_metric, AccuracyMetric::ABSOLUTE_STREAM_L1); diff --git a/crates/logical-asap-optimization/src/accuracy/estimators/count_sketch.rs b/crates/logical-asap-optimization/src/accuracy/estimators/count_sketch.rs index 9a2522cd6..4dffa5d43 100644 --- a/crates/logical-asap-optimization/src/accuracy/estimators/count_sketch.rs +++ b/crates/logical-asap-optimization/src/accuracy/estimators/count_sketch.rs @@ -167,7 +167,7 @@ mod tests { ), &SketchStatistic::PointCount { key: asap_types::ir::scalar::expr_ir::ColumnRef::SampleValue, - value: None, + value: Some("1".into()), }, ) .expect("CountSketch has a parameter-derived L2 guarantee"); diff --git a/crates/logical-asap-optimization/src/accuracy/estimators/mod.rs b/crates/logical-asap-optimization/src/accuracy/estimators/mod.rs index 21209c851..e9b44e084 100644 --- a/crates/logical-asap-optimization/src/accuracy/estimators/mod.rs +++ b/crates/logical-asap-optimization/src/accuracy/estimators/mod.rs @@ -1,6 +1,7 @@ //! Dispatch committed estimator parameters to their accuracy models. use super::*; use asap_types::ir::operator::AggIntent; +use asap_types::ir::schema::FamilyCapability; pub mod cardinality; pub mod cms; @@ -25,9 +26,7 @@ pub(super) fn sketch_guarantee( SketchParams::CountSketch { .. } | SketchParams::CountSketchWithHeap { .. } => { count_sketch::guarantee(algorithm, params, query) } - SketchParams::Kmv { .. } | SketchParams::Theta { .. } => { - cardinality::guarantee(algorithm, params, query) - } + SketchParams::Kmv { .. } => cardinality::guarantee(algorithm, params, query), SketchParams::UnivMon { .. } => univmon::guarantee(algorithm, params, query), } } @@ -58,7 +57,8 @@ fn bounded_guarantee( /// `family` built over an **exact** input, derived from the family's /// committed parameters by inverting the same sizing formulas /// [`crate::realization::default_size_params`] uses. Exact for a plain value -/// or an exact aggregate. `None` when the family has no error model +/// or an exact aggregate. `None` when a sketch does not answer `query` (its +/// family-capability table row), and when the family has no error model /// (`SummarySample`, `Wavelet`, `StatModel`, and sketch evaluations without a /// certified bound): Pass 1 then leaves the guarantee `None` and keeps the /// candidate for plan selection to decide. @@ -71,7 +71,10 @@ pub fn local_guarantee( FieldDataType::ExactAggregate(kind, _) => Some(AccuracyGuarantee::exact(format!( "ExactAggregate({kind:?})" ))), - FieldDataType::Sketch(kind, _) => sketch_guarantee(kind.algorithm(), kind.params(), query), + FieldDataType::Sketch(kind, _) if FamilyCapability::of(family).answers(query) => { + sketch_guarantee(kind.algorithm(), kind.params(), query) + } + FieldDataType::Sketch(..) => None, // No error model is registered for these families. FieldDataType::SummarySample(..) | FieldDataType::Wavelet(..) @@ -86,7 +89,7 @@ pub fn local_guarantee( /// UnivMon's heap, and KMV's `k` at the relative error it was sized for /// (#694). Other parameters are returned as they are: DDSketch never fails; /// classic HLL's failure probability comes from its source's distinct bound -/// ([`hll::min_precision`]); Theta is calibrated at a fixed 99%; a KMV not +/// ([`hll::min_precision`]); a KMV not /// sized for a relative error and an L2 or entropy UnivMon are not /// enlarged. pub fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams { @@ -161,13 +164,10 @@ pub(crate) fn size_params( heap_size: topk_capacity(k, eps), } } - // Non-preferred candidates (DDSketch / Theta / Kmv / CountSketch / + // Non-preferred candidates (DDSketch / Kmv / CountSketch / // CountSketchWithHeap) are only reachable once a cost model picks // them; sized here so that wiring is local. SketchAlgorithm::DDSketch => ddsketch::size_params(eps), - SketchAlgorithm::Theta => SketchParams::Theta { - k: cardinality::theta_k_99(eps), - }, SketchAlgorithm::Kmv => SketchParams::Kmv { k: cardinality::kmv_k(eps, delta), relative_error: Some(eps), @@ -191,6 +191,29 @@ pub(crate) fn size_params( heap_size: topk_capacity(k, eps), } } + // Listed, not `_`: a family `asap_sketchlib` adds fails to compile + // here until the planner sizes it or lists it as not offered. + SketchAlgorithm::CountSketchWithL2Norm + | SketchAlgorithm::CmsWithSpaceSaving + | SketchAlgorithm::CountSketchWithSpaceSaving + | SketchAlgorithm::CmsFold + | SketchAlgorithm::CountSketchFold + | SketchAlgorithm::HllHip + | SketchAlgorithm::KllDynamic + | SketchAlgorithm::SpaceSaving + | SketchAlgorithm::Elastic + | SketchAlgorithm::Coco + | SketchAlgorithm::Bloom + | SketchAlgorithm::SetAggregator + | SketchAlgorithm::UnivMonTwoTier + | SketchAlgorithm::UnivMonQ + | SketchAlgorithm::HydraCms + | SketchAlgorithm::HydraCountSketch + | SketchAlgorithm::HydraKll + | SketchAlgorithm::HydraHll + | SketchAlgorithm::HydraUnivMon => { + unreachable!("Pass 1 offers only the families in FAMILY_PREFERENCE, not {kind:?}") + } } } @@ -348,7 +371,6 @@ pub fn size_for_all_readers(params: &[&SketchParams]) -> Option { k: k.max(*k2), relative_error: min_recorded(relative_error, *e), }, - (P::Theta { k }, P::Theta { k: k2 }) => P::Theta { k: k.max(*k2) }, _ => return None, }) })?; @@ -672,4 +694,37 @@ mod tests { assert!(hll::classic_failure_probability(p - 1, 0.05, 64) > 0.01); assert_eq!(hll::min_precision(0.05, 1e-30, 4096, 12), 12); } + + /// A sketch read for a statistic its table row does not answer (an HLL + /// for a total) has no guarantee; one it answers keeps its guarantee, + /// and a plain value or exact aggregate stays exact. + #[test] + fn local_guarantee_needs_a_statistic_the_family_answers() { + use asap_types::ir::scalar::expr_ir::ColumnRef; + use asap_types::ir::schema::{ + DataType, ExactKind, ExactParams, GroupingStrategy, SketchKind, + }; + let hll = FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::Hll, + SketchParams::Hll { + precision: 12, + relative_error: None, + }, + ), + GroupingStrategy::default(), + ); + let total = SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }; + assert!(local_guarantee(&hll, &total).is_none()); + assert!(local_guarantee(&hll, &SketchStatistic::Cardinality).is_some()); + for exact in [ + FieldDataType::Plain(DataType::Float64), + FieldDataType::ExactAggregate(ExactKind::Count, ExactParams::Count), + ] { + assert!(local_guarantee(&exact, &total).is_some(), "{exact:?}"); + } + } } diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index 063516d65..bf130f5bc 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -1081,7 +1081,7 @@ fn sized_for_all( let operator = Operator::ASAP(ASAPOp::SummaryAgg { child: Rc::clone(child), family: FieldDataType::Sketch( - SketchKind::new(kind.algorithm().clone(), params), + SketchKind::new(*kind.algorithm(), params), grouping_strategy.clone(), ), input: input.clone(), @@ -3296,9 +3296,7 @@ mod tests { /// in which every target chose `family`. fn chose(queries: &[(&str, f64)], family: SketchAlgorithm) -> Vec { use crate::per_computation_candidates::ComputationCandidate; - chose_where(queries, |c| { - *c == ComputationCandidate::Summary(family.clone()) - }) + chose_where(queries, |c| *c == ComputationCandidate::Summary(family)) } /// The composed candidate of `queries` (each at its own ε, δ = 0.01) diff --git a/crates/logical-asap-optimization/src/deployment_capabilities.rs b/crates/logical-asap-optimization/src/deployment_capabilities.rs index d992f34b0..cd361f7ef 100644 --- a/crates/logical-asap-optimization/src/deployment_capabilities.rs +++ b/crates/logical-asap-optimization/src/deployment_capabilities.rs @@ -116,7 +116,7 @@ mod tests { #[test] fn default_supports_everything_and_json_fills_the_rest() { let default = DeploymentCapabilities::default(); - assert!(default.supports(&SketchAlgorithm::Theta, StatisticKind::TopK)); + assert!(default.supports(&SketchAlgorithm::Bloom, StatisticKind::TopK)); assert!(default.raw_data_retained && !default.query_time_results_retained); let parsed: DeploymentCapabilities = serde_json::from_str( r#"{"supported_summaries": [{"family": "Kll", "readouts": ["Quantile"]}]}"#, diff --git a/crates/logical-asap-optimization/src/memo/rules/state_configurations.rs b/crates/logical-asap-optimization/src/memo/rules/state_configurations.rs index 7cf6bc00b..a92644ba2 100644 --- a/crates/logical-asap-optimization/src/memo/rules/state_configurations.rs +++ b/crates/logical-asap-optimization/src/memo/rules/state_configurations.rs @@ -227,7 +227,7 @@ impl SharedProducer for SummaryStates { StateCovering::Size(match (first, configuration) { (FieldDataType::Sketch(kind, grouping), Some(params)) => { FieldDataType::Sketch( - SketchKind::new(kind.algorithm().clone(), params), + SketchKind::new(*kind.algorithm(), params), grouping.clone(), ) } diff --git a/crates/logical-asap-optimization/src/memo/state_configurations.rs b/crates/logical-asap-optimization/src/memo/state_configurations.rs index f3b000971..bdd38c96b 100644 --- a/crates/logical-asap-optimization/src/memo/state_configurations.rs +++ b/crates/logical-asap-optimization/src/memo/state_configurations.rs @@ -365,7 +365,7 @@ pub fn shared_size(a: &FieldDataType, b: &FieldDataType) -> Option PerComputationCandidates { let Some(NonASAPOp::Aggregate { measures, .. }) = target.aggregate.non_asap() else { continue; }; - let Ok(readout) = statistic(&measures[target.measure]).map(|s| s.kind()) else { + let Some(readout) = sketch_statistic(&measures[target.measure]).map(|s| s.kind()) + else { continue; }; let unsupported = |candidate: &ComputationCandidate| { @@ -526,7 +529,7 @@ pub fn computation_candidates_for_intent( }); } } - let summaries = summary_candidates(intent).iter().cloned(); + let summaries = summary_candidates(intent).into_iter(); if matches!(intent, AggIntent::Quantile { q, .. } if !(0.0..=1.0).contains(q)) { choices.pruned.extend(summaries.map(|algorithm| { ( @@ -981,12 +984,8 @@ fn add_whole_expression_targets( .iter() .cloned() .partition(|candidate| { - matches!( - candidate, - ComputationCandidate::Summary( - SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap - ) - ) + matches!(candidate, ComputationCandidate::Summary(family) + if family.capability().answers(StatisticKind::TopK)) }); if heaps.is_empty() { continue; @@ -1023,12 +1022,8 @@ fn prune_topk_heaps_over_one_row_per_item(targets: &mut [ComputationTarget]) { target .choices .prune(PruneReason::TopKHeapOverOneRowPerItem, |candidate| { - matches!( - candidate, - ComputationCandidate::Summary( - SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap - ) - ) + matches!(candidate, ComputationCandidate::Summary(family) + if family.capability().answers(StatisticKind::TopK)) }); } } @@ -1457,7 +1452,7 @@ pub fn with_resized_summaries( ) = &mut operator { if let Some(params) = resize(node, kind.params()).filter(|p| p != kind.params()) { - *kind = SketchKind::new(kind.algorithm().clone(), params); + *kind = SketchKind::new(*kind.algorithm(), params); changed = true; } } @@ -3223,7 +3218,9 @@ fn whole_expression( sizing.built.borrow_mut().push((state.clone(), key)); let estimate = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { summary_input: state, - query: statistic(intent)?, + query: sketch_statistic(intent).ok_or(PerComputationCandidateError::Unsupported( + "sketch evaluation", + ))?, }))?; ranked_as_topk(target, estimate) } @@ -3389,7 +3386,9 @@ fn substitute_measure( sizing.built.borrow_mut().push((summary_input.clone(), key)); let estimate = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { summary_input, - query: statistic(intent)?, + query: sketch_statistic(intent).ok_or( + PerComputationCandidateError::Unsupported("sketch evaluation"), + )?, }))?; match intent { AggIntent::TopK { .. } => ranked_as_topk(target, estimate)?, @@ -3675,16 +3674,15 @@ fn sized(algorithm: &SketchAlgorithm, intent: &AggIntent, share: usize) -> Sketc let size = |target: &AccuracyTarget| { let (epsilon, delta) = accuracy_budget(target); SketchKind::new( - algorithm.clone(), - default_size_params(algorithm.clone(), intent, epsilon, delta), + *algorithm, + default_size_params(*algorithm, intent, epsilon, delta), ) }; let whole = size(&target); if share <= 1 { return whole; } - let metric = statistic(intent) - .ok() + let metric = sketch_statistic(intent) .and_then(|query| { local_guarantee( &FieldDataType::Sketch(whole.clone(), GroupingStrategy::default()), @@ -3872,25 +3870,6 @@ fn summary_update( }) } -fn statistic(intent: &AggIntent) -> Result { - Ok(match intent { - AggIntent::Quantile { q, .. } => SketchStatistic::Quantile { q: *q }, - AggIntent::Cardinality { .. } => SketchStatistic::Cardinality, - AggIntent::FrequencyL2 { .. } => SketchStatistic::FrequencyL2, - AggIntent::FrequencyEntropy { .. } => SketchStatistic::FrequencyEntropy, - AggIntent::TopK { k, .. } => SketchStatistic::TopK { k: *k }, - AggIntent::Count { .. } => SketchStatistic::PointCount { - key: ColumnRef::SampleValue, - value: None, - }, - _ => { - return Err(PerComputationCandidateError::Unsupported( - "sketch evaluation", - )) - } - }) -} - #[cfg(test)] mod tests { /// Aggregate replacement qualifiers survive both renaming and later input rebinding. @@ -3924,7 +3903,6 @@ mod tests { use super::*; use asap_types::ir::properties::AccuracyTarget; - use asap_types::ir::schema::StatisticKind; /// Approximate count keeps one exact option, its accumulator, beside the /// universal summary; plain exact execution is not offered. #[test] @@ -3953,7 +3931,7 @@ mod tests { .unwrap() .candidates; assert!(!choices.contains(&ComputationCandidate::Summary(SketchAlgorithm::CountSketch))); - let total = statistic(&AggIntent::Count { + let total = sketch_statistic(&AggIntent::Count { accuracy: AccuracyTarget::exact(), }) .unwrap(); @@ -4703,10 +4681,7 @@ mod tests { .filter(|node| { matches!(node.asap(), Some(ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), .. - }) if matches!( - kind.algorithm(), - SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap - )) + }) if kind.algorithm().capability().answers(StatisticKind::TopK)) }) .collect() } @@ -5627,7 +5602,7 @@ mod tests { supported .iter() .map(|(family, readouts)| crate::SupportedSummary { - family: family.clone(), + family: *family, readouts: readouts.to_vec(), }) .collect(), @@ -5791,7 +5766,7 @@ mod tests { .iter() .map(|a| { ( - ComputationCandidate::Summary(a.clone()), + ComputationCandidate::Summary(*a), PruneReason::RequiresExact, ) }) @@ -5912,7 +5887,7 @@ mod tests { let (_, root) = composed_with(inventory, topk) .into_iter() .skip(own) - .find(|(candidate, _)| *candidate == ComputationCandidate::Summary(algorithm.clone())) + .find(|(candidate, _)| *candidate == ComputationCandidate::Summary(algorithm)) .unwrap_or_else(|| panic!("a fused {algorithm:?} candidate")); let [state] = &heap_states(&root)[..] else { panic!("one heap: {root:#?}") diff --git a/crates/logical-asap-optimization/src/realization.rs b/crates/logical-asap-optimization/src/realization.rs index 7fee1cba0..05f6e0280 100644 --- a/crates/logical-asap-optimization/src/realization.rs +++ b/crates/logical-asap-optimization/src/realization.rs @@ -11,8 +11,8 @@ use asap_types::ir::scalar::column_resolution::{resolve_column_ref, ResolveError use asap_types::ir::scalar::expr_ir::ColumnRef; use asap_types::ir::schema::ColumnId; use asap_types::ir::schema::{ - EntityIdentity, Field, FieldDataType, NonNegativeWeightProof, Schema, SketchAlgorithm, - SketchParams, SummaryInputExpr, SummaryUpdate, WeightDomain, + EntityIdentity, FamilyCapability, Field, FieldDataType, NonNegativeWeightProof, Schema, + SketchAlgorithm, SketchParams, SketchStatistic, SummaryInputExpr, SummaryUpdate, WeightDomain, }; use asap_types::ir::{NonASAPOp, OperatorNode}; #[cfg(test)] @@ -24,41 +24,67 @@ use std::rc::Rc; /// one. `ln(1/0.01) → depth 5`, matching the conventional CMS sizing. pub const DEFAULT_DELTA: f64 = 0.01; -/// The sketch kinds that can serve an intent, most-preferred first. -/// This is the `AggIntent → SketchAlgorithm` map of issue #98, read by -/// Pass 1's summary choices. Listed here so the candidate set has one home. -pub fn summary_candidates(intent: &AggIntent) -> &'static [SketchAlgorithm] { - match intent { - AggIntent::Quantile { .. } => &[SketchAlgorithm::Kll, SketchAlgorithm::DDSketch], - // A distinct-tuple count hashes the whole tuple as one item - // (`SummaryInputExpr::Tuple`), which the distinct-count sketches take - // unchanged. UnivMon is dropped there: it estimates frequency moments - // over a single value stream, and `realize_value_frequency_summary_input` - // would feed it one column of the tuple. Theta is not offered: - // `asap_sketchlib` has no Theta sketch to execute it; add it back - // once the library has one. - AggIntent::Cardinality { cols, .. } if cols.len() > 1 => { - &[SketchAlgorithm::Hll, SketchAlgorithm::Kmv] - } - AggIntent::Cardinality { .. } => &[ - SketchAlgorithm::Hll, - SketchAlgorithm::Kmv, - SketchAlgorithm::UnivMon, - ], - AggIntent::FrequencyL2 { .. } | AggIntent::FrequencyEntropy { .. } => { - &[SketchAlgorithm::UnivMon] - } - // Count-Sketch-with-heap is CMS-with-heap's balanced/zero-mean-error - // alternative for the same heavy-hitter shape. - AggIntent::TopK { .. } => &[ - SketchAlgorithm::CmsWithHeap, - SketchAlgorithm::CountSketchWithHeap, - ], - // Count Sketch's signed counters give no total, so it serves no - // count. - AggIntent::Count { .. } => &[SketchAlgorithm::Cms, SketchAlgorithm::UnivMon], - _ => &[], - } +/// The sketch families the planner sizes and offers, in the order Pass 1 +/// offers those that answer an intent's statistic: the specialized families +/// before the universal UnivMon. +const FAMILY_PREFERENCE: &[SketchAlgorithm] = &[ + SketchAlgorithm::Kll, + SketchAlgorithm::DDSketch, + SketchAlgorithm::Hll, + SketchAlgorithm::Kmv, + SketchAlgorithm::Cms, + SketchAlgorithm::CountSketch, + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + SketchAlgorithm::UnivMon, +]; + +/// The sketch kinds that can serve an intent, most-preferred first: the +/// families whose family-capability table row answers the intent's +/// statistic, in preference order, less UnivMon for a distinct-tuple +/// count. Empty for an intent no sketch +/// statistic serves. +pub fn summary_candidates(intent: &AggIntent) -> Vec { + candidates_by(intent, FamilyCapability::of_per_instance_sketch) +} + +/// [`summary_candidates`] with each family's table row read from `capability`. +fn candidates_by( + intent: &AggIntent, + capability: impl Fn(&SketchAlgorithm) -> FamilyCapability, +) -> Vec { + let Some(statistic) = sketch_statistic(intent) else { + return Vec::new(); + }; + // A distinct-tuple count hashes the whole tuple as one item + // (`SummaryInputExpr::Tuple`), which the distinct-count sketches take + // unchanged. UnivMon is dropped there: it estimates frequency moments + // over a single value stream, and `realize_value_frequency_summary_input` + // would feed it one column of the tuple. + let tuple = matches!(intent, AggIntent::Cardinality { cols, .. } if cols.len() > 1); + FAMILY_PREFERENCE + .iter() + .filter(|family| capability(family).answers(&statistic)) + .filter(|family| !(tuple && **family == SketchAlgorithm::UnivMon)) + .cloned() + .collect() +} + +/// The statistic a summary is read for to answer `intent`, or `None` when +/// no sketch statistic serves the intent. +pub(crate) fn sketch_statistic(intent: &AggIntent) -> Option { + Some(match intent { + AggIntent::Quantile { q, .. } => SketchStatistic::Quantile { q: *q }, + AggIntent::Cardinality { .. } => SketchStatistic::Cardinality, + AggIntent::FrequencyL2 { .. } => SketchStatistic::FrequencyL2, + AggIntent::FrequencyEntropy { .. } => SketchStatistic::FrequencyEntropy, + AggIntent::TopK { k, .. } => SketchStatistic::TopK { k: *k }, + AggIntent::Count { .. } => SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }, + _ => return None, + }) } /// The [`AccuracyTarget`] an approximate-capable intent's summary is sized @@ -522,4 +548,91 @@ mod tests { ); } } + + /// Each approximable intent is offered the families it was offered when + /// the list was written by hand, in the same order; a non-sketch intent + /// is offered none. + #[test] + fn summary_candidates_keep_the_hand_written_lists() { + use SketchAlgorithm as S; + let accuracy = AccuracyTarget::max_error(0.01); + let cardinality = |cols: Vec| AggIntent::Cardinality { + cols, + accuracy: accuracy.clone(), + }; + let cases = [ + ( + AggIntent::Quantile { + col: None, + q: 0.99, + accuracy: accuracy.clone(), + }, + vec![S::Kll, S::DDSketch], + ), + (cardinality(vec![]), vec![S::Hll, S::Kmv, S::UnivMon]), + (cardinality(vec![1]), vec![S::Hll, S::Kmv, S::UnivMon]), + (cardinality(vec![1, 2]), vec![S::Hll, S::Kmv]), + ( + AggIntent::FrequencyL2 { + col: None, + accuracy: accuracy.clone(), + }, + vec![S::UnivMon], + ), + ( + AggIntent::FrequencyEntropy { + col: None, + accuracy: accuracy.clone(), + }, + vec![S::UnivMon], + ), + ( + AggIntent::TopK { + k: 10, + accuracy: accuracy.clone(), + }, + vec![S::CmsWithHeap, S::CountSketchWithHeap], + ), + ( + AggIntent::Count { + accuracy: accuracy.clone(), + }, + vec![S::Cms, S::UnivMon], + ), + (AggIntent::Sum { col: None }, vec![]), + ]; + for (intent, want) in cases { + assert_eq!(summary_candidates(&intent), want, "{intent:?}"); + } + } + + /// Dropping a readout from a family's table row drops the family from + /// the candidates of the intents that read it, with no other edit. + #[test] + fn a_dropped_readout_drops_the_family() { + // Count-Min-with-heap answered from Count-Min's row: no top-k readout. + let heap_without_top_k = |family: &SketchAlgorithm| { + FamilyCapability::of_per_instance_sketch(match family { + SketchAlgorithm::CmsWithHeap => &SketchAlgorithm::Cms, + other => other, + }) + }; + let top_k = AggIntent::TopK { + k: 10, + accuracy: AccuracyTarget::max_error(0.01), + }; + assert_eq!( + candidates_by(&top_k, heap_without_top_k), + vec![SketchAlgorithm::CountSketchWithHeap] + ); + } + + /// The preference order names each family it offers once. + #[test] + fn preference_order_names_each_family_once() { + let mut seen = std::collections::HashSet::new(); + for family in FAMILY_PREFERENCE { + assert!(seen.insert(family), "{family:?} twice"); + } + } } diff --git a/crates/logical-asap-optimization/tests/per_computation_candidates.rs b/crates/logical-asap-optimization/tests/per_computation_candidates.rs index 38dda4e88..7108b643f 100644 --- a/crates/logical-asap-optimization/tests/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/tests/per_computation_candidates.rs @@ -28,7 +28,7 @@ fn algorithms(choices: &[ComputationCandidate]) -> Vec { choices .iter() .filter_map(|choice| match choice { - ComputationCandidate::Summary(algorithm) => Some(algorithm.clone()), + ComputationCandidate::Summary(algorithm) => Some(*algorithm), _ => None, }) .collect() diff --git a/crates/plan-selection/Cargo.toml b/crates/plan-selection/Cargo.toml index 6e14b4d23..526e33fb7 100644 --- a/crates/plan-selection/Cargo.toml +++ b/crates/plan-selection/Cargo.toml @@ -22,3 +22,6 @@ highs = "=2.4.0" [dev-dependencies] asap-logical-asap-optimization = { path = "../logical-asap-optimization", features = ["test-support"] } asap-frontend-promql = { path = "../frontend-promql" } +# The executor's charge for each summary state, checked against the bytes +# plan selection prices for it plus the accumulator's own fields (#947). +asap-executor = { path = "../executor" } diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 3f7ff498c..a22ebf3d0 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -4972,21 +4972,22 @@ fn keeps_every_value(measure: &AggIntent) -> bool { } /// Update operations per input row and bytes per state, from the sketch's -/// parameters and `input_rows` (the rows one state folds), modelled on what -/// `asap_sketchlib` allocates: -/// - Count-Min / Count-Sketch: `depth × width` 8-byte counters, plus a -/// 24-byte entry per heap slot for the top-k variants; -/// - KLL: its fixed bytes ([`kll_fixed_bytes`]: the struct and its level -/// array) plus the 8-byte item array as `asap_sketchlib` grows it for the -/// rows the state folds, at most its full capacity ([`kll_buffer_items`]); -/// - DDSketch: its struct and, per bucket store it fills, 8-byte counts in -/// the 128-bin chunks the library grows a store by -/// ([`ddsketch_state_bytes`]); -/// - HLL: one 1-byte register per bucket, `2^precision`; -/// - KMV: its `k` smallest 64-bit hashes (Theta, which candidate generation -/// does not offer because `asap_sketchlib` has no Theta sketch, likewise); -/// - UnivMon: per layer, a `rows × cols` Count-Sketch of 8-byte counters and -/// a heap of `heap_size` 24-byte entries. +/// parameters and `input_rows` (the rows one state folds). The bytes are +/// `asap_sketchlib`'s `estimated_memory_bytes` for the type the executor +/// builds (Count-Min and Count Sketch over `f64` counters), given what the +/// state is assumed to hold: +/// - a Count-Min / Count Sketch top-k heap full of keys of +/// [`ASSUMED_HEAP_KEY_BYTES`]; each UnivMon layer's heap full of keys of +/// [`ASSUMED_VALUE_KEY_BYTES`]; +/// - DDSketch and KMV: the rows the state folds, at least one (any number +/// when unknown), each a distinct item for KMV; a DDSketch's values span +/// [`DDSKETCH_ASSUMED_VALUE_RANGE`] and fill its negative store too +/// unless proven non-negative. +/// +/// KLL is priced without the library's estimate: its fixed bytes +/// ([`kll_fixed_bytes`]: the struct and its level array) plus the 8-byte +/// item array as `asap_sketchlib` grows it for the rows the state folds, at +/// most its full capacity ([`kll_buffer_items`]). /// /// An exact accumulator's state is the executor's per-group state, /// [`EXACT_ACCUMULATOR_STATE_BYTES`] for every kind. @@ -5004,7 +5005,8 @@ fn summary_shape(family: &FieldDataType, input_rows: StateInputRows) -> (f64, u6 } /// What one summary state's bytes depend on besides its parameters (only -/// a DDSketch's and a KLL's do): the input rows it folds and the sign of their values. +/// a KLL's, a DDSketch's and a KMV's do): the input rows it folds and the +/// sign of their values. /// The rows are whatever its producer emits (raw data samples, rates, sums), /// not a summary sample. #[derive(Debug, Clone, Copy, PartialEq)] @@ -5031,30 +5033,76 @@ impl StateInputRows { /// size the measured table interpolates between /// ([`cost::measured`]). fn sketch_shape(params: &SketchParams, input_rows: StateInputRows) -> (f64, u64) { + use asap_sketchlib::{ + Classic, CountMinF64, CountMinSketchWithHeap, CountSketch, CountSketchWithHeap, DataInput, + DdSketch, HyperLogLog, Kmv, RegularPath, UnivMon, Vector2D, + }; + let heap_key = DataInput::Bytes(&[0; ASSUMED_HEAP_KEY_BYTES]); + let rows = input_rows.rows_per_state.map_or(usize::MAX, |rows| { + usize::try_from(rows.max(1)).unwrap_or(usize::MAX) + }); match params { - SketchParams::Cms { width, depth } | SketchParams::CountSketch { width, depth } => { - (f64::from(*depth), 8 * u64::from(*width) * u64::from(*depth)) - } + SketchParams::Cms { width, depth } => ( + f64::from(*depth), + CountMinF64::estimated_memory_bytes(*depth as usize, *width as usize) as u64, + ), + SketchParams::CountSketch { width, depth } => ( + f64::from(*depth), + CountSketch::, RegularPath>::estimated_memory_bytes( + *depth as usize, + *width as usize, + ) as u64, + ), SketchParams::CmsWithHeap { width, depth, heap_size, - } - | SketchParams::CountSketchWithHeap { + } => ( + f64::from(*depth) + 1.0, + CountMinSketchWithHeap::, RegularPath>::estimated_memory_bytes( + *depth as usize, + *width as usize, + *heap_size as usize, + &heap_key, + ) as u64, + ), + SketchParams::CountSketchWithHeap { width, depth, heap_size, } => ( f64::from(*depth) + 1.0, - 8 * u64::from(*width) * u64::from(*depth) + 24 * u64::from(*heap_size), + CountSketchWithHeap::, RegularPath>::estimated_memory_bytes( + *depth as usize, + *width as usize, + *heap_size as usize, + &heap_key, + ) as u64, ), SketchParams::Kll { k, .. } => ( 1.0, kll_fixed_bytes() + 8 * kll_buffer_items(*k, input_rows), ), - SketchParams::DDSketch { alpha } => (1.0, ddsketch_state_bytes(*alpha, input_rows)), - SketchParams::Hll { precision, .. } => (1.0, 1u64 << (*precision).min(63)), - SketchParams::Kmv { k, .. } | SketchParams::Theta { k } => (1.0, 8 * u64::from(*k)), + // The executor builds no DDSketch outside α in (0, 1), where + // sketchlib's estimate panics: priced as an empty one. + SketchParams::DDSketch { alpha } if !(*alpha > 0.0 && *alpha < 1.0) => { + (1.0, std::mem::size_of::() as u64) + } + SketchParams::DDSketch { alpha } => ( + 1.0, + DdSketch::estimated_memory_bytes( + *alpha, + rows, + 1.0, + DDSKETCH_ASSUMED_VALUE_RANGE, + !input_rows.non_negative, + ) as u64, + ), + SketchParams::Hll { precision, .. } => ( + 1.0, + HyperLogLog::::estimated_memory_bytes(usize::from(*precision)) as u64, + ), + SketchParams::Kmv { k, .. } => (1.0, Kmv::estimated_memory_bytes(*k as usize, rows) as u64), SketchParams::UnivMon { heap_size, sketch_rows, @@ -5063,9 +5111,13 @@ fn sketch_shape(params: &SketchParams, input_rows: StateInputRows) -> (f64, u64) .. } => ( univmon_expected_layers(*layers) * (f64::from(*sketch_rows) + 1.0), - u64::from(*layers) - * (8 * u64::from(*sketch_rows) * u64::from(*sketch_cols) - + 24 * u64::from(*heap_size)), + UnivMon::estimated_memory_bytes( + *heap_size as usize, + *sketch_rows as usize, + *sketch_cols as usize, + usize::from(*layers), + &DataInput::Bytes(&[0; ASSUMED_VALUE_KEY_BYTES]), + ) as u64, ), } } @@ -5123,6 +5175,20 @@ fn kll_buffer_items(k: u32, input_rows: StateInputRows) -> u64 { } } +/// Bytes of the key one Count-Min / Count Sketch top-k heap slot holds: an +/// assumption, since the workload declares no label sizes. Such a key is a +/// label set in the executor's tuple-key encoding, a tag byte and a 4-byte +/// length before each value; this assumes three labels of 16 bytes, the +/// size [`row_bytes`] gives a string. +const ASSUMED_HEAP_KEY_BYTES: usize = 3 * (1 + 4 + 16); + +/// Bytes of the key one UnivMon heap slot holds: an assumption, since the +/// workload declares no label sizes. UnivMon counts the values of one +/// column, the executor keys each by its type tag and bytes, and a label +/// value is assumed to be 16 bytes, the size [`row_bytes`] gives a string. +/// A sample value's key owns no bytes, so this bounds it too. +const ASSUMED_VALUE_KEY_BYTES: usize = 1 + 16; + /// Layers a UnivMon row updates in expectation. `asap_sketchlib` updates /// layers `0..=b`, with bottom layer `b` the index of the first zero among /// hash bits `1..layers`, capped at `layers - 1`; so it reaches layer `j` @@ -5132,64 +5198,13 @@ fn univmon_expected_layers(layers: u8) -> f64 { 2.0 - 2f64.powi(1 - i32::from(layers.max(1))) } -/// Ratio of the largest to the smallest positive value a DDSketch is assumed -/// to see: nine decades, e.g. latencies from 1 ns to 1 s. An assumption, not -/// workload evidence: the data workload declares no value range. +/// Ratio of the largest to the smallest value magnitude a DDSketch is +/// assumed to see: nine decades, e.g. latencies from 1 ns to 1 s, priced as +/// magnitudes in `[1, 1e9]` (its bins depend on the ratio alone). An +/// assumption, not workload evidence: the data workload declares no value +/// range. const DDSKETCH_ASSUMED_VALUE_RANGE: f64 = 1e9; -/// Bins `asap_sketchlib`'s DDSketch grows a bucket store by -/// (`GROW_CHUNK`): a store's first value allocates this many around its -/// bin, and a value past either end adds at least this many more. -const DDSKETCH_GROW_CHUNK_BINS: u64 = 128; - -/// Bytes of one `asap_sketchlib` DDSketch with relative accuracy `alpha` -/// over `input_rows`: its struct (144 bytes), plus 8 bytes per allocated bin -/// of each bucket store its values can fill. The library allocates a store -/// at the first value of its sign (a value too small to index goes to a -/// zero count, in no store), so one store is charged when the values are -/// proven non-negative or a state folds at most one row, and two -/// otherwise. A store allocates one 128-bin chunk around its -/// first value, then 128-bin chunks on each side as values reach past -/// it: `128 · (1 + 2 · ⌈(spanned − 128)⁺ / 256⌉)` bins, where `spanned` -/// is the bins its values can touch, at most one per value and at most -/// [`ddsketch_bins`] over the assumed value range. E.g. 1,168 bytes for ten -/// non-negative values at α = 0.05, and 3,216 for many over nine decades. -fn ddsketch_state_bytes(alpha: f64, input_rows: StateInputRows) -> u64 { - let range_bins = ddsketch_bins(alpha); - let spanned = input_rows - .rows_per_state - .map_or(range_bins, |rows| range_bins.min(rows.max(1))); - let side_chunks = spanned - .saturating_sub(DDSKETCH_GROW_CHUNK_BINS) - .div_ceil(2 * DDSKETCH_GROW_CHUNK_BINS); - let store_bytes = 8 * DDSKETCH_GROW_CHUNK_BINS * (1 + 2 * side_chunks); - let at_most_one_row = input_rows.rows_per_state.is_some_and(|rows| rows <= 1); - let stores = if input_rows.non_negative || at_most_one_row { - 1 - } else { - 2 - }; - std::mem::size_of::() as u64 + stores * store_bytes -} - -/// Bins a DDSketch with relative accuracy `alpha` touches, assuming its -/// values span nine decades ([`DDSKETCH_ASSUMED_VALUE_RANGE`] = 1e9): its -/// bins grow geometrically by `gamma = (1 + alpha) / (1 - alpha)`, so -/// `ceil(ln(1e9) / ln(gamma))`, e.g. 1,037 bins at α = 0.01. The count is -/// proportional to the decades spanned, so values over three decades would -/// need a third of these bins and over twelve decades four thirds. -/// `asap_sketchlib`'s store has no bin limit; the cap is the bins spanning -/// every value it can index. -fn ddsketch_bins(alpha: f64) -> u64 { - let log_gamma = ((1.0 + alpha) / (1.0 - alpha)).ln(); - let (min, max) = asap_sketchlib::sketches::ddsketch::ddsketch_indexable_bounds(alpha); - let indexable = (max.ln() - min.ln()) / log_gamma; - (DDSKETCH_ASSUMED_VALUE_RANGE.ln() / log_gamma) - .min(indexable) - .ceil() - .max(1.0) as u64 -} - #[cfg(test)] mod tests { use super::*; @@ -5557,7 +5572,7 @@ mod tests { SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap ) => { - Some(kind.algorithm().clone()) + Some(*kind.algorithm()) } _ => None, }) @@ -7648,101 +7663,63 @@ mod tests { assert_eq!(asap_types::ir::schema::EXACT_ACCUMULATOR_STATE_BYTES, 56); } - /// A DDSketch is sized for values over nine decades: 1,037 bins at - /// α = 0.01, allocated as 1,152 (nine 128-bin chunks) of 8 bytes per - /// store, both stores for samples of unknown sign. - #[test] - fn ddsketch_bins_span_nine_decades() { - assert_eq!(ddsketch_bins(0.01), 1_037); - assert_eq!( - summary_shape( - &FieldDataType::Sketch( - SketchKind::new( - SketchAlgorithm::DDSketch, - SketchParams::DDSketch { alpha: 0.01 } - ), - GroupingStrategy::default(), - ), - StateInputRows::UNKNOWN - ) - .1, - 144 + 2 * 8 * 1_152 - ); - } - - /// The bytes `asap_sketchlib`'s DDSketch holds after adding `values`: - /// its struct and the 8-byte counts of both bucket stores. + /// The bytes `asap_sketchlib`'s DDSketch holds after adding `values`. fn library_ddsketch_bytes(alpha: f64, values: &[f64]) -> u64 { let mut sketch = asap_sketchlib::sketches::DdSketch::new(alpha); for value in values { sketch.add(value); } - let bins = sketch.positive_store_counts().len() + sketch.negative_store_counts().len(); - (std::mem::size_of::() + 8 * bins) as u64 + sketch.memory_bytes() as u64 } - /// The DDSketch state the model prices is what `asap_sketchlib` - /// allocates: one 128-bin chunk per store for a few samples, more for - /// samples spanning more bins, no negative store for values declared - /// non-negative, and one store for one row of either sign. A state - /// assumed to fold no row is charged one store, as for one row. + /// A DDSketch is priced for what `asap_sketchlib` can hold after its + /// rows: at least what it holds for a few samples, for samples of either + /// sign, and for many samples over nine decades; a store less for values + /// declared non-negative. Many rows are priced as unknown ones. #[test] - fn ddsketch_state_bytes_match_the_library() { + fn ddsketch_state_bytes_bound_the_library() { let alpha = 0.05; + let priced = |rows_per_state, non_negative| { + sketch_shape( + &SketchParams::DDSketch { alpha }, + StateInputRows { + rows_per_state, + non_negative, + }, + ) + .1 + }; // Ten Pareto(2) samples (scale 1), one 1 s pane of a 10 Hz series. let pareto: Vec = (0..10) .map(|i| (1.0 - (f64::from(i) + 0.5) / 10.0).powf(-0.5)) .collect(); - let non_negative = |rows_per_state| StateInputRows { - rows_per_state, - non_negative: true, - }; - assert_eq!(library_ddsketch_bytes(alpha, &pareto), 1_168); - assert_eq!(ddsketch_state_bytes(alpha, non_negative(Some(10))), 1_168); - // One row fills at most one store, whatever its sign: the library - // allocates a store at the first value of its sign. - let signed_rows = |rows_per_state| StateInputRows { - rows_per_state: Some(rows_per_state), - non_negative: false, - }; - for value in [-2.0, 2.0] { - assert_eq!( - ddsketch_state_bytes(alpha, signed_rows(1)), - library_ddsketch_bytes(alpha, &[value]) - ); - } - // Two rows, one of each sign, fill both stores. - assert_eq!( - ddsketch_state_bytes(alpha, signed_rows(2)), - library_ddsketch_bytes(alpha, &[-2.0, 2.0]) - ); - // Signed values fill both stores. let signed: Vec = pareto.iter().map(|v| -v).chain(pareto.clone()).collect(); - let signed_rows = StateInputRows { - rows_per_state: Some(20), - non_negative: false, - }; - assert_eq!( - library_ddsketch_bytes(alpha, &signed), - ddsketch_state_bytes(alpha, signed_rows) - ); - // Many samples over the assumed nine decades, the first at the - // bottom or at the middle of the range: the range's 208 bins, in - // three 128-bin chunks. let decades: Vec = (0..10_000) - .map(|i| 10f64.powf(-4.5 + 9.0 * f64::from((i * 7_919) % 10_000) / 10_000.0)) + .map(|i| 10f64.powf(9.0 * f64::from((i * 7_919) % 10_000) / 10_000.0)) .collect(); - let middle_first: Vec = [1.0].into_iter().chain(decades.clone()).collect(); - assert_eq!( - ddsketch_state_bytes(alpha, non_negative(Some(10_000))), - 3_216 - ); - assert_eq!(library_ddsketch_bytes(alpha, &decades), 3_216); - assert_eq!(library_ddsketch_bytes(alpha, &middle_first), 3_216); - assert_eq!( - ddsketch_state_bytes(alpha, non_negative(None)), - ddsketch_state_bytes(alpha, non_negative(Some(10_000))) - ); + let signed_decades: Vec = decades + .iter() + .enumerate() + .map(|(i, v)| if i % 2 == 0 { -v } else { *v }) + .collect(); + for (values, non_negative) in [ + (&pareto[..], true), + (&[-2.0][..], false), + (&[2.0][..], false), + (&[-2.0, 2.0][..], false), + (&signed[..], false), + (&decades[..], true), + (&signed_decades[..], false), + ] { + let rows = Some(values.len() as u64); + assert!( + library_ddsketch_bytes(alpha, values) <= priced(rows, non_negative), + "{} values, non-negative: {non_negative}", + values.len() + ); + } + assert!(priced(Some(10_000), true) < priced(Some(10_000), false)); + assert_eq!(priced(Some(10_000), false), priced(None, false)); } /// DDSketch state grows as its relative accuracy α tightens. @@ -7773,9 +7750,12 @@ mod tests { } /// Every sketch family's state is sized from its parameters, as - /// `asap_sketchlib` allocates it. + /// `asap_sketchlib` allocates it: an HLL's registers, a KMV's `k` + /// hashes once it has seen `k` distinct items, and a UnivMon's layers + /// once keys of the assumed size have churned through every heap. #[test] fn summary_state_is_sized_from_sketch_params() { + use asap_sketchlib::DataInput; let bytes = |algorithm, params| { summary_shape( &FieldDataType::Sketch( @@ -7794,8 +7774,12 @@ mod tests { relative_error: None } ), - 1 << 14 + asap_sketchlib::HyperLogLog::::new().memory_bytes() as u64 ); + let mut kmv: asap_sketchlib::Kmv = asap_sketchlib::Kmv::new(1_000); + for i in 0..2_000u64 { + kmv.update(&DataInput::U64(i)); + } assert_eq!( bytes( SketchAlgorithm::Kmv, @@ -7804,20 +7788,26 @@ mod tests { relative_error: None } ), - 8 * 1_000 + kmv.memory_bytes() as u64 ); + let mut univmon = asap_sketchlib::UnivMon::new(8, 3, 64, 2); + for i in 0..10_000u64 { + let mut key = [0; ASSUMED_VALUE_KEY_BYTES]; + key[..8].copy_from_slice(&i.to_le_bytes()); + univmon.update_weighted(&DataInput::Bytes(&key), 1); + } assert_eq!( bytes( SketchAlgorithm::UnivMon, SketchParams::UnivMon { - heap_size: 256, - sketch_rows: 5, - sketch_cols: 1_024, - layers: 16, + heap_size: 8, + sketch_rows: 3, + sketch_cols: 64, + layers: 2, distinct_failure_probability: None, } ), - 16 * (8 * 5 * 1_024 + 24 * 256) + univmon.memory_bytes() as u64 ); } @@ -7899,6 +7889,289 @@ mod tests { assert_eq!(model_bytes(None), 744 + 8 * 1_206); } + /// Plan selection prices each state the executor builds per group at + /// what the executor charges for it (#947). Each family is built + /// through the executor's factory and fed `n` rows shaped as the + /// pricing assumes: signed values over [`DDSKETCH_ASSUMED_VALUE_RANGE`], + /// label values of [`ASSUMED_VALUE_KEY_BYTES`] keys for UnivMon, or + /// keyed rows whose tuple keys are [`ASSUMED_HEAP_KEY_BYTES`] long. The + /// heap families' shapes are dominated by their heaps. The charge is + /// the sketch's bytes plus the accumulator's own fields, named per + /// family. It equals the priced bytes where `asap_sketchlib`'s estimate + /// is exact (Count-Min, Count Sketch, HLL, KMV at `n` distinct items) and + /// for KLL up to `k` rows. For DDSketch, the heap sketches and UnivMon + /// the price bounds it, by at most 1.5 times once the state is full; + /// UnivMon's price also bounds the bytes its sketch holds. + /// KLL past `k` rows is checked only loosely: the planner's formula + /// leaves out the compaction scratch space, a known gap until a + /// sketchlib estimate without the CDF cache can price KLL. + /// + /// HydraCms is left out: plan selection prices an independent Count-Min + /// per group, the executor one shared grid, so the two agree only at + /// about 79 groups for a 5 × 32 grid of 64 × 3 cells (a follow-up). + #[test] + fn state_bytes_bound_the_executors_charge() { + use asap_executor::factory::create_planner_accumulator; + use asap_executor::summary_kernels::univmon::UnivMonAccumulator; + use asap_executor::summary_kernels::weighted_frequency::WeightedFrequency; + use asap_executor::summary_kernels::HllSketchAccumulator; + use asap_executor::values::Value; + use asap_types::ir::scalar::expr_ir::ColumnRef; + use asap_types::ir::schema::{ + NonNegativeWeightProof, SummaryInputExpr, SummaryUpdate, WeightDomain, + }; + use std::mem::{align_of, size_of}; + use SketchAlgorithm as A; + /// How the priced bytes relate to the sketch's. + enum Priced { + Exact, + /// Exact up to `k` rows; past them at most the formula's + /// missing compaction scratch space over it. + Kll { + k: u32, + }, + /// At least, and at most 1.5 times once the state is full. + Bound, + } + /// The rows a family is fed. + enum Rows { + /// Signed values over the assumed range. + Values, + /// One label value a row, as Pass 1 feeds UnivMon a column. + Labels, + /// A label set and a weight a row. + Keyed, + } + // (family, parameters, rows fed, how it is priced, the + // accumulator's own fields, rows at which the state is full: four + // times the heap for the heap sketches, so each heap churns). + // + // The accumulators' own fields, beside the sketch each wraps: + // Count-Min / Count Sketch's width, depth and capacity, padded to + // the struct's alignment; HLL's precision tag; UnivMon's two key + // bounds. + let weighted_frequency_shape = + (3 * size_of::()).next_multiple_of(align_of::()); + let hll_precision_tag = size_of::() + - size_of::>(); + let univmon_key_bounds = 2 * size_of::(); + let families = [ + ( + A::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + Rows::Values, + Priced::Kll { k: 200 }, + 0, + 10_000, + ), + ( + A::DDSketch, + SketchParams::DDSketch { alpha: 0.01 }, + Rows::Values, + Priced::Bound, + 0, + 10_000, + ), + ( + A::Cms, + SketchParams::Cms { + width: 1_024, + depth: 5, + }, + Rows::Keyed, + Priced::Exact, + 0, + 10_000, + ), + ( + A::CountSketch, + SketchParams::CountSketch { + width: 1_024, + depth: 5, + }, + Rows::Keyed, + Priced::Exact, + weighted_frequency_shape, + 10_000, + ), + ( + A::CmsWithHeap, + SketchParams::CmsWithHeap { + width: 16, + depth: 1, + heap_size: 256, + }, + Rows::Keyed, + Priced::Bound, + weighted_frequency_shape, + 1_024, + ), + ( + A::CountSketchWithHeap, + SketchParams::CountSketchWithHeap { + width: 16, + depth: 1, + heap_size: 256, + }, + Rows::Keyed, + Priced::Bound, + weighted_frequency_shape, + 1_024, + ), + ( + A::Hll, + SketchParams::Hll { + precision: 12, + relative_error: None, + }, + Rows::Values, + Priced::Exact, + hll_precision_tag, + 10_000, + ), + ( + A::Kmv, + SketchParams::Kmv { + k: 1_024, + relative_error: None, + }, + Rows::Values, + Priced::Exact, + 0, + 10_000, + ), + // Its top layer sees one row in eight: 4 096 labels fill and + // churn every layer's heap. + ( + A::UnivMon, + SketchParams::UnivMon { + heap_size: 256, + sketch_rows: 1, + sketch_cols: 16, + layers: 4, + distinct_failure_probability: None, + }, + Rows::Labels, + Priced::Bound, + univmon_key_bounds, + 4_096, + ), + ]; + let value = SummaryUpdate::column(ColumnRef::SampleValue); + let item = Some(SummaryInputExpr::Column(ColumnRef::Named("item".into()))); + let keyed = SummaryUpdate { + item: item.clone(), + ..value.clone() + }; + let labeled = SummaryUpdate { + item, + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + }; + // A 16-byte label value: its key has the assumed size. + let label = |i: u32| format!("{i:016}"); + assert_eq!( + Value::Utf8(label(0).into()).key().unwrap().len(), + ASSUMED_VALUE_KEY_BYTES + ); + // Three 16-byte labels: a tuple key of the assumed size. + let labels = |i: u32| { + [ + label(i), + "service-frontend".to_string(), + "region-us-east-1".to_string(), + ] + }; + let key = asap_sketchlib::TupleKey::new( + labels(0) + .into_iter() + .map(asap_sketchlib::KeyColumn::String) + .collect(), + ); + assert_eq!(key.to_bytes().unwrap().len(), ASSUMED_HEAP_KEY_BYTES); + for (algorithm, params, rows, priced_as, fields, full) in &families { + for n in [1, 100, *full] { + let family = FieldDataType::Sketch( + SketchKind::new(*algorithm, params.clone()), + GroupingStrategy::default(), + ); + let input = match rows { + Rows::Values => &value, + Rows::Labels => &labeled, + Rows::Keyed => &keyed, + }; + let mut updater = + create_planner_accumulator(&family, input, &GroupingStrategy::default()) + .unwrap(); + for i in 0..n { + match rows { + // Distinct magnitudes over the assumed range, in a + // scrambled order, alternating in sign. + Rows::Values => { + let at = f64::from(i * 7_919 % n) / f64::from(n); + let magnitude = DDSKETCH_ASSUMED_VALUE_RANGE.powf(at); + let sign = if i % 2 == 0 { 1.0 } else { -1.0 }; + updater.update_single(sign * magnitude, 0).unwrap(); + } + Rows::Labels => updater + .update_value(&Value::Utf8(label(i).into()), 0) + .unwrap(), + Rows::Keyed => updater.update_keyed(&labels(i), 1.0, 0).unwrap(), + } + } + let state = updater.into_accumulator(); + let charged = state.approx_memory_bytes(); + let priced = fields + + sketch_shape( + params, + StateInputRows { + rows_per_state: Some(u64::from(n)), + non_negative: false, + }, + ) + .1 as usize; + let context = format!("{algorithm:?} after {n} rows: {priced} priced"); + match priced_as { + Priced::Exact => assert_eq!(charged, priced, "{context}"), + Priced::Kll { k } if n <= *k => assert_eq!(charged, priced, "{context}"), + // sketchlib bounds KLL's compaction scratch space at + // 2 (k - 1) items. + Priced::Kll { k } => assert!( + charged <= priced + 8 * 2 * (*k as usize - 1), + "{context}, {charged} charged" + ), + Priced::Bound => { + assert!(charged <= priced, "{context}, {charged} charged"); + if n == *full { + assert!( + priced as f64 <= 1.5 * charged as f64, + "{context}, {charged} charged" + ); + } + } + } + // UnivMon charges sketchlib's estimate itself, since its + // `memory_bytes` walks the heaps: the sketch holds no more. + // The estimate is a heap whose digest index has grown once + // more under churn, which the test's labels do not reach: a + // full heap holds about two thirds of it (1.52 times less + // here), so this bound is 1.6 times rather than 1.5. + if let Some(univmon) = state.as_any().downcast_ref::() { + let held = fields + univmon.sketch().memory_bytes(); + assert!(held <= charged, "{context}, {held} held"); + if n == *full { + assert!(priced as f64 <= 1.6 * held as f64, "{context}, {held} held"); + } + } + } + } + } + /// A UnivMon row updates layers `0..=b`, where its bottom layer `b` is /// geometric (`asap_sketchlib`'s `bottom_layer_for_hash`), so /// `2 - 2^(1 - layers)` layers in expectation; each updates one counter @@ -8230,7 +8503,7 @@ mod tests { Payload::ASAP(ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), .. - }) => Some(kind.algorithm().clone()), + }) => Some(*kind.algorithm()), _ => None, }) .collect() diff --git a/crates/plan-selection/src/search/costing/direct_profile.rs b/crates/plan-selection/src/search/costing/direct_profile.rs index 1650fa54f..40e2b23b3 100644 --- a/crates/plan-selection/src/search/costing/direct_profile.rs +++ b/crates/plan-selection/src/search/costing/direct_profile.rs @@ -921,7 +921,6 @@ fn stable_size(node: &asap_types::ir::OperatorNode, delta: Option) -> bool | SketchParams::DDSketch { .. } | SketchParams::Cms { .. } | SketchParams::CountSketch { .. } - | SketchParams::Theta { .. } ) { return false; } diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs index 580ef6fb8..f570d2027 100644 --- a/crates/planner/src/installable.rs +++ b/crates/planner/src/installable.rs @@ -1308,7 +1308,7 @@ mod tests { Some(asap_executor::supported_summary_families()) ); assert!(!defaulted.raw_data_retained); - assert!(!defaulted.supports(&SketchAlgorithm::Theta, StatisticKind::Distinct)); + assert!(!defaulted.supports(&SketchAlgorithm::CountSketch, StatisticKind::ItemFrequency)); let kll_only = DeploymentCapabilities { supported_summaries: Some(vec![SupportedSummary { family: SketchAlgorithm::Kll, diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 98860df6f..4fa96c6d9 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -2774,7 +2774,7 @@ fn heap_of(plan: &InstallablePlan) -> Option<(SketchAlgorithm, u32)> { FieldDataType::Sketch(kind, _) => match kind.params() { SketchParams::CmsWithHeap { width, .. } | SketchParams::CountSketchWithHeap { width, .. } => { - Some((kind.algorithm().clone(), *width)) + Some((*kind.algorithm(), *width)) } _ => None, }, @@ -2899,7 +2899,7 @@ async fn assert_forced_heap_answers( (samples, truth): (&[RawDataSample], &BTreeMap>), labels: (Option<&str>, &str, usize), ) { - let plan = plan_ignoring_guarantees(workload, Some(algorithm.clone())) + let plan = plan_ignoring_guarantees(workload, Some(algorithm)) .await .unwrap_or_else(|error| panic!("{algorithm:?}: {error}")); let (built, width) = heap_of(&plan).expect("a heap plan"); @@ -3130,7 +3130,7 @@ async fn a_forced_heap_grouped_topk_sum_plan_compiles_and_answers_within_its_bou let query = format!("topk(3, sum by (region) ({function}(data[1m])))"); let workload = heap_topk_workload(&query, "data", facts); let truth = BTreeMap::from([(String::new(), truth.clone())]); - let plan = plan_ignoring_guarantees(&workload, Some(algorithm.clone())) + let plan = plan_ignoring_guarantees(&workload, Some(algorithm)) .await .unwrap_or_else(|error| panic!("{query} {algorithm:?}: {error}")); let (built, width) = heap_of(&plan).expect("a heap plan"); diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs index 98a015f4d..7ff8e8edb 100644 --- a/crates/planner/tests/summary_sharing.rs +++ b/crates/planner/tests/summary_sharing.rs @@ -118,9 +118,7 @@ fn composed_summary_candidates( .choices .candidates .iter() - .position(|candidate| { - *candidate == ComputationCandidate::Summary(family.clone()) - }) + .position(|candidate| *candidate == ComputationCandidate::Summary(family)) .unwrap_or(0) }) .collect(); diff --git a/crates/types/Cargo.toml b/crates/types/Cargo.toml index 13b892319..2a9a0f742 100644 --- a/crates/types/Cargo.toml +++ b/crates/types/Cargo.toml @@ -15,3 +15,5 @@ edition = "2021" serde = { version = "1", features = ["derive", "rc"] } serde_json = "1" thiserror = "2" +# The sketch rows of the family-capability table (`SketchFamily::capability`). +asap_sketchlib = { workspace = true } diff --git a/crates/types/src/ir/schema/family_capability.rs b/crates/types/src/ir/schema/family_capability.rs index 75ddf6591..de675423d 100644 --- a/crates/types/src/ir/schema/family_capability.rs +++ b/crates/types/src/ir/schema/family_capability.rs @@ -11,9 +11,17 @@ //! statistics ([`StatisticKind`]) one state answers, each with the error of //! that readout from a merged state ([`MergedError`]), so one state can //! serve every consumer whose statistic it answers. +//! +//! A sketch family's row is `asap_sketchlib`'s +//! ([`SketchAlgorithm::capability`]), read unchanged: the family, +//! statistic and merged-error vocabulary is the library's. The planner +//! writes the rows of the exact accumulators, which the library does not +//! have. Which readouts an executor evaluates is its own declaration +//! (`DeploymentCapabilities::supported_summaries`), not this table's. use super::sketch::{ExactKind, GroupingStrategy, HydraKind, SketchAlgorithm, SketchStatistic}; use crate::ir::schema::FieldDataType; +pub use asap_sketchlib::{MergedError, StatisticKind}; /// What one summary family can do: its row of the family-capability table. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -29,27 +37,6 @@ pub struct FamilyCapability { pub readouts: &'static [(StatisticKind, MergedError)], } -/// The kind of a `SummaryEstimate` statistic, without its arguments: the -/// readout vocabulary of this table and of a deployment's supported -/// summaries. -#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub enum StatisticKind { - /// The total of the updates: a bare `PointCount` (no item value). - Total, - /// One item's frequency: a `PointCount` with an item value. - ItemFrequency, - /// The value at any quantile rank. - Quantile, - /// The number of distinct items. - Distinct, - /// The L2 norm of the item frequencies. - FrequencyL2, - /// The entropy of the item frequencies. - FrequencyEntropy, - /// The heaviest items, any `k` up to the state's heap. - TopK, -} - /// One sketch family a deployment's executor can build, with the readouts /// it can answer from the family's state: an entry of /// `DeploymentCapabilities::supported_summaries`. @@ -78,7 +65,8 @@ impl SketchStatistic { } /// How the inputs of a merge must relate for the result to keep the -/// family's guarantee. +/// family's guarantee: `asap_sketchlib`'s [`asap_sketchlib::MergeInputRule`] +/// plus [`Self::DisjointInTime`] for the exact counter accumulators. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum MergeInputRule { /// No row in two inputs: the family counts a shared row twice. @@ -91,19 +79,13 @@ pub enum MergeInputRule { MayOverlap, } -/// 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. - AsIfBuiltOverUnion, - /// The merge is allowed, but the merged readout has no guarantee; it is - /// evidence only, for plan selection to judge. - Approximate, +impl From for MergeInputRule { + fn from(rule: asap_sketchlib::MergeInputRule) -> Self { + match rule { + asap_sketchlib::MergeInputRule::Disjoint => Self::Disjoint, + asap_sketchlib::MergeInputRule::MayOverlap => Self::MayOverlap, + } + } } impl FamilyCapability { @@ -145,59 +127,13 @@ impl FamilyCapability { } /// 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. + /// its parameters: `asap_sketchlib`'s row of the family, which the IR and + /// a deployment's executor read. 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)]), - }; + let row = algorithm.capability(); Self { - merge_inputs, - readouts, + merge_inputs: row.merge_inputs.map(MergeInputRule::from), + readouts: row.readouts, } } @@ -206,8 +142,6 @@ impl FamilyCapability { pub fn of(family: &FieldDataType) -> Self { use ExactKind as E; use MergeInputRule::{Disjoint, DisjointInTime, MayOverlap}; - use MergedError::{Approximate, AsIfBuiltOverUnion as Union}; - use StatisticKind::*; let (merge_inputs, readouts): (_, &'static [(StatisticKind, MergedError)]) = match family { FieldDataType::ExactAggregate(kind, _) => match kind { E::Sum | E::Count => (Some(Disjoint), &[]), @@ -219,17 +153,18 @@ impl FamilyCapability { // pane does not keep across its boundary. E::IRate => (None, &[]), }, - // Hydra answers per subpopulation, as the inner family does. - // Over a linear sketch it is a shared grid of linear sketches - // whose cells add; over KLL it has no accuracy model, so no - // merge. + // A Hydra answers per subpopulation from a shared grid of the + // inner family's cells. HydraKll merges as sketchlib's row says: + // `Hydra::merge` requires the same k and m and otherwise returns + // an error (asap_sketchlib #251). Nothing builds one yet: Pass 1 + // offers no Hydra, and the executor builds only HydraCms + // (`create_hydra_cms`). FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => { - match kind { - HydraKind::HydraCms | HydraKind::HydraCountSketch => { - (Some(Disjoint), &[(ItemFrequency, Union)]) - } - HydraKind::HydraKll => (None, &[(Quantile, Approximate)]), - } + return Self::of_per_instance_sketch(&match kind { + HydraKind::HydraCms => SketchAlgorithm::HydraCms, + HydraKind::HydraCountSketch => SketchAlgorithm::HydraCountSketch, + HydraKind::HydraKll => SketchAlgorithm::HydraKll, + }) } FieldDataType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => { return Self::of_per_instance_sketch(kind.algorithm()) @@ -304,7 +239,7 @@ mod tests { } /// A quantile sketch answers every rank, not its total; heap sketches - /// answer top-k; Count Sketch answers item frequencies, not its total; + /// answer top-k and an item's frequency; Count Sketch answers item frequencies, not its total; /// HLL answers the distinct count, not its total. #[test] fn summary_capability_facts() { @@ -367,8 +302,8 @@ mod tests { ), ( &cms_heap, - vec![SketchStatistic::TopK { k: 10 }], - vec![SketchStatistic::FrequencyL2, item()], + vec![SketchStatistic::TopK { k: 10 }, item()], + vec![SketchStatistic::FrequencyL2, total()], ), ( &hll, diff --git a/crates/types/src/ir/schema/sketch.rs b/crates/types/src/ir/schema/sketch.rs index ef60be504..857c841a6 100644 --- a/crates/types/src/ir/schema/sketch.rs +++ b/crates/types/src/ir/schema/sketch.rs @@ -50,34 +50,11 @@ pub enum ExactParams { // ── Approximate sketches ───────────────────────────────────────────────────── /// A specific sketch algorithm — bounded error, sized by its -/// [`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. - UnivMon, - /// KLL quantile sketch (mergeable, ε-accurate rank queries). - Kll, - /// Count-Min Sketch (mergeable, (ε,δ)-accurate frequency queries). - Cms, - /// HyperLogLog (mergeable, (ε,δ)-accurate cardinality). - Hll, - /// DDSketch (mergeable, relative-error quantile queries). - DDSketch, - /// CMS augmented with a min-heap for top-k / heavy-hitter queries. - CmsWithHeap, - /// K-Minimum Values sketch (mergeable, join-cardinality estimation). - Kmv, - /// Theta sketch (mergeable, set operations + cardinality). - Theta, - /// Count-Sketch (mergeable, balanced/zero-mean-error frequency - /// queries — an alternative to CMS's one-sided bias). - CountSketch, - /// Count-Sketch augmented with a min-heap for top-k / heavy-hitter - /// queries — an alternative to `CmsWithHeap` on the Count-Sketch - /// substrate. - CountSketchWithHeap, -} +/// [`SketchParams`]. It is `asap_sketchlib`'s family enum; the statistics +/// each answers are its row of the family-capability table +/// ([`FamilyCapability`](crate::ir::schema::FamilyCapability)). The planner +/// sizes and offers only the families [`SketchParams`] has a variant for. +pub use asap_sketchlib::SketchAlgorithm; /// The largest KLL `k` `asap_sketchlib` builds: it silently caps a larger /// `k` at this (`MAX_CACHEABLE_K` in its `sketches::kll`). @@ -147,9 +124,6 @@ pub enum SketchParams { #[serde(default, skip_serializing_if = "Option::is_none")] relative_error: Option, }, - Theta { - k: u32, - }, CountSketch { width: u32, depth: u32, @@ -228,9 +202,7 @@ impl PartialEq for SketchParams { ) => (a0, a1, a2) == (b0, b1, b2), (P::Hll { precision: a, .. }, P::Hll { precision: b, .. }) => a == b, (P::DDSketch { alpha: a }, P::DDSketch { alpha: b }) => a == b, - (P::Kmv { k: a, .. }, P::Kmv { k: b, .. }) | (P::Theta { k: a }, P::Theta { k: b }) => { - a == b - } + (P::Kmv { k: a, .. }, P::Kmv { k: b, .. }) => a == b, // Different variants. Listed by `self`'s variant, not `_`, so a // new variant fails to compile here until it is compared above. (P::UnivMon { .. }, _) @@ -240,7 +212,6 @@ impl PartialEq for SketchParams { | (P::DDSketch { .. }, _) | (P::CmsWithHeap { .. }, _) | (P::Kmv { .. }, _) - | (P::Theta { .. }, _) | (P::CountSketch { .. }, _) | (P::CountSketchWithHeap { .. }, _) => false, } @@ -275,7 +246,6 @@ impl SketchKind { | (SketchAlgorithm::UnivMon, SketchParams::UnivMon { .. }) | (SketchAlgorithm::DDSketch, SketchParams::DDSketch { .. }) | (SketchAlgorithm::Hll, SketchParams::Hll { .. }) - | (SketchAlgorithm::Theta, SketchParams::Theta { .. }) | (SketchAlgorithm::Kmv, SketchParams::Kmv { .. }) | (SketchAlgorithm::Cms, SketchParams::Cms { .. }) | ( @@ -911,7 +881,6 @@ mod tests { SketchAlgorithm::DDSketch, SketchAlgorithm::CmsWithHeap, SketchAlgorithm::Kmv, - SketchAlgorithm::Theta, SketchAlgorithm::CountSketchWithHeap, ] { assert_eq!(hydra_kind_for(&algorithm), None, "{algorithm:?}"); diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index 04bdfa89d..22eddf661 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -161,8 +161,6 @@ fn merge(children: Vec>) -> Result, String> { #[test] fn family_capability_table() { use MergeInputRule::{Disjoint, DisjointInTime, MayOverlap}; - use MergedError::{Approximate, AsIfBuiltOverUnion as Union}; - use StatisticKind::*; let (width, depth, heap_size) = (64, 4, 10); // (family, how its merge inputs must relate, its readouts). type Row = ( @@ -203,28 +201,60 @@ fn family_capability_table() { vec![], ), (exact(ExactKind::IRate, ExactParams::IRate), None, vec![]), - (kll(), Some(Disjoint), vec![(Quantile, Union)]), + (FieldDataType::Plain(DataType::Float64), None, vec![]), + ( + FieldDataType::SummarySample( + SamplingKind::Reservoir, + SamplingParams::Reservoir { size: 64 }, + ), + None, + vec![], + ), + ( + FieldDataType::Wavelet(WaveletKind::Haar, WaveletParams::Haar { coefficients: 16 }), + None, + vec![], + ), + ( + FieldDataType::StatModel( + StatModelKind::Parametric, + StatModelParams::Parametric { + family: "gaussian_mixture".into(), + }, + ), + None, + vec![], + ), + ]; + // A sketch family's row is asap_sketchlib's, unchanged; a Hydra reads + // the row of its Hydra family. + let lib = |algorithm: SketchAlgorithm| { + let row = algorithm.capability(); + ( + row.merge_inputs.map(MergeInputRule::from), + row.readouts.to_vec(), + ) + }; + let sketches = [ + (kll(), SketchAlgorithm::Kll), ( sketch( SketchAlgorithm::DDSketch, SketchParams::DDSketch { alpha: 0.01 }, ), - Some(Disjoint), - vec![(Quantile, Union)], + SketchAlgorithm::DDSketch, ), - (hll(), Some(MayOverlap), vec![(Distinct, Union)]), + (hll(), SketchAlgorithm::Hll), ( sketch(SketchAlgorithm::Cms, SketchParams::Cms { width, depth }), - Some(Disjoint), - vec![(ItemFrequency, Union), (Total, Union)], + SketchAlgorithm::Cms, ), ( sketch( SketchAlgorithm::CountSketch, SketchParams::CountSketch { width, depth }, ), - Some(Disjoint), - vec![(ItemFrequency, Union)], + SketchAlgorithm::CountSketch, ), ( sketch( @@ -237,16 +267,8 @@ fn family_capability_table() { distinct_failure_probability: None, }, ), - Some(Disjoint), - vec![ - (Total, Union), - (Distinct, Union), - (FrequencyL2, Union), - (FrequencyEntropy, Approximate), - ], + SketchAlgorithm::UnivMon, ), - // 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, @@ -255,15 +277,9 @@ fn family_capability_table() { relative_error: None, }, ), - Some(MayOverlap), - vec![(Distinct, Union)], + SketchAlgorithm::Kmv, ), - ( - sketch(SketchAlgorithm::Theta, SketchParams::Theta { k: 1024 }), - None, - vec![(Distinct, Approximate)], - ), - (cms_with_heap(), Some(Disjoint), vec![(TopK, Approximate)]), + (cms_with_heap(), SketchAlgorithm::CmsWithHeap), ( sketch( SketchAlgorithm::CountSketchWithHeap, @@ -273,11 +289,8 @@ fn family_capability_table() { heap_size, }, ), - Some(Disjoint), - vec![(TopK, Approximate)], + SketchAlgorithm::CountSketchWithHeap, ), - // Hydra over a linear sketch is a grid of linear sketches; Hydra - // over KLL has no accuracy model and fails closed. ( hydra( SketchAlgorithm::Cms, @@ -290,8 +303,7 @@ fn family_capability_table() { shared_columns: width, }, ), - Some(Disjoint), - vec![(ItemFrequency, Union)], + SketchAlgorithm::HydraCms, ), ( hydra( @@ -305,8 +317,7 @@ fn family_capability_table() { shared_columns: width, }, ), - Some(Disjoint), - vec![(ItemFrequency, Union)], + SketchAlgorithm::HydraCountSketch, ), ( hydra( @@ -321,34 +332,15 @@ fn family_capability_table() { shared_buckets: width, }, ), - None, - vec![(Quantile, Approximate)], - ), - (FieldDataType::Plain(DataType::Float64), None, vec![]), - ( - FieldDataType::SummarySample( - SamplingKind::Reservoir, - SamplingParams::Reservoir { size: 64 }, - ), - None, - vec![], - ), - ( - FieldDataType::Wavelet(WaveletKind::Haar, WaveletParams::Haar { coefficients: 16 }), - None, - vec![], - ), - ( - FieldDataType::StatModel( - StatModelKind::Parametric, - StatModelParams::Parametric { - family: "gaussian_mixture".into(), - }, - ), - None, - vec![], + SketchAlgorithm::HydraKll, ), ]; + let rows = rows + .into_iter() + .chain(sketches.into_iter().map(|(family, algorithm)| { + let (merge_inputs, readouts) = lib(algorithm); + (family, merge_inputs, readouts) + })); // One statistic of every kind. let statistics = [ SketchStatistic::PointCount { @@ -368,7 +360,7 @@ fn family_capability_table() { for (family, merge_inputs, readouts) in rows { let capability = FamilyCapability::of(&family); assert_eq!(capability.merge_inputs, merge_inputs, "{family:?}"); - assert_eq!(capability.readouts, readouts.as_slice(), "{family:?}"); + assert_eq!(capability.readouts.to_vec(), readouts, "{family:?}"); for statistic in &statistics { let row = readouts .iter() @@ -391,11 +383,12 @@ fn family_capability_table() { } /// Heap readouts (entropy, top-k) of a merged state are approximate: each -/// 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. A -/// statistic the family does not answer has no merged error. +/// part's heaps keep only that part's heavy items. UnivMon's total adds +/// exactly. Its 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. 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}; @@ -415,7 +408,7 @@ fn merged_error_depends_on_the_readout() { )); for (readout, error) in [ (FrequencyL2, MergedError::AsIfBuiltOverUnion), - (point(), MergedError::AsIfBuiltOverUnion), + (point(), MergedError::Exact), (Cardinality, MergedError::AsIfBuiltOverUnion), (FrequencyEntropy, MergedError::Approximate), ] { @@ -520,19 +513,20 @@ fn accepted_by_category(algorithm: &SketchAlgorithm, query: &SketchStatistic) -> 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::Hll | 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 { .. }), + _ => unreachable!("{algorithm:?} is not probed"), } } /// 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. +/// counters give no total, and a heap family's counters answer an item's +/// frequency. 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); @@ -546,7 +540,6 @@ fn estimate_validation_matches_the_category_match_except_the_fixed_rows() { precision: 12, relative_error: None, }, - SketchParams::Theta { k: 1024 }, SketchParams::Kmv { k: 1024, relative_error: None, @@ -578,7 +571,6 @@ fn estimate_validation_matches_the_category_match_except_the_fixed_rows() { 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, @@ -641,19 +633,21 @@ fn estimate_validation_matches_the_category_match_except_the_fixed_rows() { SketchStatistic::TopK { k: 5 }, ]; for family in families { - let FieldDataType::Sketch(kind, grouping) = &family else { + let FieldDataType::Sketch(kind, _) = &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 fixed = matches!( + (kind.algorithm(), query.kind()), + ( + SketchAlgorithm::UnivMon, + StatisticKind::ItemFrequency | StatisticKind::TopK + ) | (SketchAlgorithm::CountSketch, StatisticKind::Total) + | ( + SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap, + StatisticKind::ItemFrequency, + ) + ); let before = accepted_by_category(kind.algorithm(), query); let after = estimate(family.clone(), query.clone()).is_ok(); assert_eq!(after, before != fixed, "{family:?} {query:?}"); diff --git a/docs/design_docs/concepts/accuracy-models.md b/docs/design_docs/concepts/accuracy-models.md index c28ec395f..b0e6c846c 100644 --- a/docs/design_docs/concepts/accuracy-models.md +++ b/docs/design_docs/concepts/accuracy-models.md @@ -176,7 +176,7 @@ The built-in models currently include: | Bounded Classic HLL | Source-conditioned relative-error/failure-probability bound and precision sizing for the linear-counting branch | | CMS | L1-normalized frequency bound from width and depth; does not by itself certify TopK membership | | CountSketch | L2-normalized frequency bound and median concentration bound, requiring valid odd depth | -| KMV / Theta | Parameter-derived cardinality bounds using the registered variance/Chebyshev model at 99% confidence | +| KMV | Provable distinct-count bound from the `k` smallest hashes | | UnivMon | Exact unit-update total for the supported evaluation; no universal guarantee for all its statistics | | Other families/evaluations | No default certificate where no accuracy model is registered | @@ -385,7 +385,7 @@ accuracy/ ├── hll.rs # Generic HLL and bounded Classic HLL ├── cms.rs ├── count_sketch.rs - ├── cardinality.rs # Shared KMV / Theta models + ├── cardinality.rs # KMV model └── univmon.rs ``` diff --git a/docs/develop_docs/extend-asap-aware-mapping.md b/docs/develop_docs/extend-asap-aware-mapping.md index e67ae847e..3bda592b1 100644 --- a/docs/develop_docs/extend-asap-aware-mapping.md +++ b/docs/develop_docs/extend-asap-aware-mapping.md @@ -10,26 +10,33 @@ owns the decision. A new sketch that Pass 1 can offer for an intent. -1. **Type.** Add the variant to `SketchAlgorithm` and its parameters to - `SketchParams` (`crates/types/src/ir/schema/sketch.rs`). -2. **Capability.** Add its row to `FamilyCapability::of` - (`crates/types/src/ir/schema/family_capability.rs`): which inputs it can - merge (`MergeInputRule`) and which statistics it answers (`StatisticKind`, - with the merged error). Pass 2's sharing, pane and rollup rules read only - this table. -3. **Offer and size.** List it for its intents in - `realization::summary_candidates` (`crates/logical-asap-optimization`), - most preferred first, and size it in - `realization::default_size_params`. +1. **Type.** `SketchAlgorithm` is `asap_sketchlib`'s family enum, so the + family must exist in the library first. Add its parameters to + `SketchParams` (`crates/types/src/ir/schema/sketch.rs`); a Hydra kind + maps to its library family in `FamilyCapability::of`. +2. **Capability.** Its row, which inputs it can merge and which statistics + it answers with the merged error, is the library's family-capability + table (`SketchAlgorithm::capability`), read unchanged by + `FamilyCapability::of_per_instance_sketch` + (`crates/types/src/ir/schema/family_capability.rs`). Pass 2's sharing, + pane and rollup rules read only this table. Which readouts the executor + evaluates is its own declaration, `EVALUATED_READOUTS` + (`crates/executor/src/capability.rs`); add the family's readouts there + once the executor answers them. +3. **Offer and size.** Place it in `realization::FAMILY_PREFERENCE` + (`crates/logical-asap-optimization`): `summary_candidates` offers it, in + that order, for every intent whose statistic its capability row answers. + Size it in `realization::default_size_params`. 4. **Accuracy.** Add its error model under `crates/logical-asap-optimization/src/accuracy/estimators/` and return it from `DefaultAccuracyModel`'s `local_guarantee`, by inverting the same sizing formulas. Without a guarantee, every candidate that uses it is rejected. -5. **Cost.** Give it a state size and per-operation CPU in - `crates/plan-selection/src/lib.rs` (and accept its parameters in - `crates/plan-selection/src/cost/measured.rs` if deployments - measure it). +5. **Cost.** Give it a state size (the library's `estimated_memory_bytes` + for the type the executor builds; KLL keeps the planner's own formula) + and per-operation CPU in `sketch_shape` + (`crates/plan-selection/src/lib.rs`), and accept its parameters in + `crates/plan-selection/src/cost/measured.rs` if deployments measure it. 6. **Executor.** Wrap the `asap_sketchlib` implementation in `crates/executor/src/summary_kernels/` and register it in `factory.rs` and `capability.rs`. Gaps in the library are fixed upstream,