diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs b/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs index 7e06ef63c..99ebbd894 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs @@ -1,41 +1,155 @@ -//! KLL normalized rank error at the registered empirical 99% calibration. +//! KLL normalized rank error at the registered empirical 99% calibration +//! and, below a 0.01 failure probability, that calibration scaled by the +//! Gaussian quantile ratio `z(1 − δ/2) / z(1 − 0.01/2)`. use super::*; +use asap_types::post_asap::sketch::KLL_MAX_K; pub(super) fn guarantee( algorithm: &SketchAlgorithm, params: &SketchParams, query: &SketchStatistic, ) -> Option { - let SketchParams::Kll { k } = params else { + let SketchParams::Kll { + k, + failure_probability, + } = params + else { return None; }; + if *k > KLL_MAX_K { + return None; + } + let delta = failure_probability.unwrap_or(KLL_CALIBRATED_FAILURE_PROBABILITY); + let scale = delta_scale(delta)?; + let contract = if delta < KLL_CALIBRATED_FAILURE_PROBABILITY { + KLL_DELTA_SCALED_CONTRACT + } else { + KLL_EMPIRICAL_99_CONTRACT + }; Some(super::bounded_guarantee( algorithm, params, query, ErrorMetric::Rank, - kll_rank_error_99(*k), - ProbabilityExpr::Constant { value: 0.01 }, - "apache_datasketches_kll_empirical_99_a9b42755072b", + kll_rank_error_99(*k) * scale, + ProbabilityExpr::Constant { value: delta }, + contract, )) } +/// The empirical 99% calibration of the pinned Apache DataSketches KLL. +const KLL_EMPIRICAL_99_CONTRACT: &str = "apache_datasketches_kll_empirical_99_a9b42755072b"; +/// That calibration scaled to a failure probability below 0.01 by the +/// Gaussian quantile ratio ([`delta_scale`]): a model, not a measurement, so +/// it is named apart from the calibration. +const KLL_DELTA_SCALED_CONTRACT: &str = + "apache_datasketches_kll_empirical_99_delta_scaled_a9b42755072b"; pub(crate) const KLL_RANK_ERROR_COEFFICIENT_99: f64 = 2.296; pub(crate) const KLL_RANK_ERROR_EXPONENT_99: f64 = 0.9723; +/// The failure probability of the empirical 99% calibration. +pub(crate) const KLL_CALIBRATED_FAILURE_PROBABILITY: f64 = 0.01; pub(crate) fn kll_rank_error_99(k: u32) -> f64 { KLL_RANK_ERROR_COEFFICIENT_99 / f64::from(k).powf(KLL_RANK_ERROR_EXPONENT_99) } /// Invert Apache DataSketches' empirical 99th-percentile, single-sided KLL -/// normalized rank-error fit: `epsilon = 2.296 / k^0.9723`. +/// normalized rank-error fit: `epsilon = 2.296 / k^0.9723`. Capped at +/// [`KLL_MAX_K`], the largest `k` `asap_sketchlib` builds; an `eps` that +/// needs more gets that `k`, whose guarantee then misses the target. pub(crate) fn kll_k(eps: f64) -> u32 { saturating_ceil( (KLL_RANK_ERROR_COEFFICIENT_99 / eps).powf(1.0 / KLL_RANK_ERROR_EXPONENT_99), 8, - 65_535, + KLL_MAX_K, ) } +/// How much the rank error at failure probability `delta` exceeds the error +/// at the calibrated 0.01, modelling the error as Gaussian: the ratio of the +/// two-sided standard normal quantiles, `z(1 − delta/2) / z(1 − 0.01/2)`. +/// It grows faster as `delta` shrinks than the KLL bound's +/// `sqrt(ln(1/delta))` (1.277 vs 1.225 at `delta` = 0.001), so it is the +/// conservative choice. 1 for `delta >= 0.01`; `None` unless `0 < delta < 1`. +fn delta_scale(delta: f64) -> Option { + if !(delta > 0.0 && delta < 1.0) { + None + } else if delta >= KLL_CALIBRATED_FAILURE_PROBABILITY { + Some(1.0) + } else { + // `z(1 − p) = −z(p)`, computed from the lower tail: `1 − delta/2` + // loses `delta` to rounding once it nears f64's epsilon. + Some( + normal_quantile(delta / 2.0) + / normal_quantile(KLL_CALIBRATED_FAILURE_PROBABILITY / 2.0), + ) + } +} + +/// The standard normal quantile `z(p)` for `0 < p < 1`, by Acklam's rational +/// approximation (relative error below 1.2e-9). +fn normal_quantile(p: f64) -> f64 { + const A: [f64; 6] = [ + -3.969_683_028_665_376e1, + 2.209_460_984_245_205e2, + -2.759_285_104_469_687e2, + 1.383_577_518_672_69e2, + -3.066_479_806_614_716e1, + 2.506_628_277_459_239, + ]; + const B: [f64; 5] = [ + -5.447_609_879_822_406e1, + 1.615_858_368_580_409e2, + -1.556_989_798_598_866e2, + 6.680_131_188_771_972e1, + -1.328_068_155_288_572e1, + ]; + const C: [f64; 6] = [ + -7.784_894_002_430_293e-3, + -3.223_964_580_411_365e-1, + -2.400_758_277_161_838, + -2.549_732_539_343_734, + 4.374_664_141_464_968, + 2.938_163_982_698_783, + ]; + const D: [f64; 4] = [ + 7.784_695_709_041_462e-3, + 3.224_671_290_700_398e-1, + 2.445_134_137_142_996, + 3.754_408_661_907_416, + ]; + let tail = |q: f64| { + (((((C[0] * q + C[1]) * q + C[2]) * q + C[3]) * q + C[4]) * q + C[5]) + / ((((D[0] * q + D[1]) * q + D[2]) * q + D[3]) * q + 1.0) + }; + const P_LOW: f64 = 0.024_25; + if p < P_LOW { + tail((-2.0 * p.ln()).sqrt()) + } else if p > 1.0 - P_LOW { + -tail((-2.0 * (1.0 - p).ln()).sqrt()) + } else { + let q = p - 0.5; + let r = q * q; + (((((A[0] * r + A[1]) * r + A[2]) * r + A[3]) * r + A[4]) * r + A[5]) * q + / (((((B[0] * r + B[1]) * r + B[2]) * r + B[3]) * r + B[4]) * r + 1.0) + } +} + +/// `k` for rank error `eps` at failure probability `delta`: the empirical +/// 99% calibration for `delta >= 0.01`; below, the calibration at +/// `eps / delta_scale(delta)`, recording `delta` so the guarantee reports it. +pub(crate) fn size_params(eps: f64, delta: f64) -> SketchParams { + if delta.is_nan() || delta <= 0.0 || delta >= KLL_CALIBRATED_FAILURE_PROBABILITY { + return SketchParams::Kll { + k: kll_k(eps), + failure_probability: None, + }; + } + SketchParams::Kll { + k: kll_k(eps / delta_scale(delta).unwrap_or(1.0)), + failure_probability: Some(delta), + } +} + #[cfg(test)] mod tests { use super::*; @@ -73,4 +187,132 @@ mod tests { if contract == "apache_datasketches_kll_empirical_99_a9b42755072b" ))); } + + /// Below δ = 0.01, KLL is sized by the KLL bound's δ dependence: k is + /// larger than at δ = 0.01, and the guarantee meets ε = 0.01, δ = 0.001. + #[test] + fn sizing_meets_a_failure_probability_below_one_percent() { + use crate::realization::default_size_params; + use asap_types::pre_asap::agg_intent::default_quantile; + let q = default_quantile(0.99); + let k = |delta| match default_size_params(SketchAlgorithm::Kll, &q, 0.01, delta) { + SketchParams::Kll { k, .. } => k, + other => panic!("{other:?}"), + }; + assert!(k(0.001) > k(0.01), "{} vs {}", k(0.001), k(0.01)); + let params = default_size_params(SketchAlgorithm::Kll, &q, 0.01, 0.001); + let g = guarantee( + &SketchAlgorithm::Kll, + ¶ms, + &SketchStatistic::Quantile { q: 0.99 }, + ) + .unwrap(); + assert!(DefaultAccuracyModel.satisfies( + &g, + &AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.001, + } + )); + assert_eq!(g.failure_probability.evaluate(), Some(0.001)); + } + + /// A KLL sized below δ = 0.01 names its own contract, not the empirical + /// 99% calibration's. + #[test] + fn sizing_below_one_percent_names_a_distinct_contract() { + let contract = |delta| { + let g = guarantee( + &SketchAlgorithm::Kll, + &size_params(0.01, delta), + &SketchStatistic::Quantile { q: 0.99 }, + ) + .unwrap(); + match &g.provenance[..] { + [GuaranteeSource::SketchEvaluation { contract, .. }] => contract.clone(), + other => panic!("{other:?}"), + } + }; + assert_eq!( + contract(0.01), + "apache_datasketches_kll_empirical_99_a9b42755072b" + ); + assert_eq!( + contract(0.001), + "apache_datasketches_kll_empirical_99_delta_scaled_a9b42755072b" + ); + } + + /// `k` never exceeds the 26 602 `asap_sketchlib` silently caps it at, so + /// a guarantee never claims a larger `k` than runs. + #[test] + fn kll_k_is_capped_at_the_library_maximum() { + assert_eq!(kll_k(1e-6), KLL_MAX_K); + assert_eq!(KLL_MAX_K, 26_602); + } + + /// A hand-built `k` above the library's cap gives no guarantee. + #[test] + fn a_k_above_the_library_cap_gives_no_guarantee() { + let g = |k| { + guarantee( + &SketchAlgorithm::Kll, + &SketchParams::Kll { + k, + failure_probability: None, + }, + &SketchStatistic::Quantile { q: 0.99 }, + ) + }; + assert!(g(KLL_MAX_K).is_some()); + assert!(g(KLL_MAX_K + 1).is_none()); + } + + /// Below δ = 0.01 the error scales by the Gaussian quantile ratio + /// `z(1 − δ/2) / z(1 − 0.01/2)`. + #[test] + fn delta_scale_is_the_gaussian_quantile_ratio() { + assert_eq!(delta_scale(0.01), Some(1.0)); + assert_eq!(delta_scale(0.5), Some(1.0)); + let ratio = delta_scale(0.001).unwrap(); + assert!((ratio - 1.277_463_039_557_138_7).abs() < 1e-8, "{ratio}"); + assert!((normal_quantile(0.975) - 1.959_963_984_540_054).abs() < 1e-8); + // Far in the tail, `1 − δ/2` rounds to 1 (or near it) in f64, so the + // scale must be computed from the lower tail, `−z(δ/2)`. + for (delta, expected) in [ + (1e-16, 3.224_121_029_197_093_4), + (1e-15, 3.116_223_140_825_122_4), + ] { + let scale = delta_scale(delta).unwrap(); + assert!((scale - expected).abs() < 1e-8, "{delta}: {scale}"); + } + assert_eq!( + size_params(0.01, 0.001), + SketchParams::Kll { + k: kll_k(0.01 / ratio), + failure_probability: Some(0.001), + } + ); + } + + /// A hand-built failure probability outside (0, 1) gives no guarantee + /// rather than an infinite or NaN error bound. + #[test] + fn an_invalid_failure_probability_gives_no_guarantee() { + for delta in [0.0, -0.1, 1.0, 1.5, f64::NAN, f64::INFINITY] { + let params = SketchParams::Kll { + k: 200, + failure_probability: Some(delta), + }; + assert!( + guarantee( + &SketchAlgorithm::Kll, + ¶ms, + &SketchStatistic::Quantile { q: 0.99 }, + ) + .is_none(), + "{delta}" + ); + } + } } diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs index 115f550c0..0e2c1b63c 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs @@ -81,7 +81,7 @@ pub(crate) fn size_params( // Baseline dimensions are candidates, not an inverted error bound. // Empirical models may size these; no theoretical guarantee is claimed. SketchAlgorithm::UnivMon => univmon::size_params(), - SketchAlgorithm::Kll => SketchParams::Kll { k: kll::kll_k(eps) }, + SketchAlgorithm::Kll => kll::size_params(eps, delta), SketchAlgorithm::Cms => SketchParams::Cms { width: cms::cms_width(eps), depth: cms::cms_depth(delta), diff --git a/crates/asap-aware-mapping/src/cost_model.rs b/crates/asap-aware-mapping/src/cost_model.rs index 86e7e6900..8d8191191 100644 --- a/crates/asap-aware-mapping/src/cost_model.rs +++ b/crates/asap-aware-mapping/src/cost_model.rs @@ -1120,7 +1120,10 @@ mod tests { match kind { SketchAlgorithm::Kll => { let k = if eps >= 0.01 { 200 } else { 2048 }; - SketchParams::Kll { k } + SketchParams::Kll { + k, + failure_probability: None, + } } other => crate::realization::default_size_params(other, intent, eps, delta), } @@ -1134,7 +1137,10 @@ mod tests { let intent = default_quantile(0.99); assert_eq!( DiscreteKllRungs.size_params(SketchAlgorithm::Kll, &intent, 0.001, 0.01), - SketchParams::Kll { k: 2048 }, + SketchParams::Kll { + k: 2048, + failure_probability: None + }, ); // Untouched kinds still fall through to the default formula. assert_eq!( diff --git a/crates/asap-aware-mapping/src/empirical_cost.rs b/crates/asap-aware-mapping/src/empirical_cost.rs index a63240355..7527ea8fb 100644 --- a/crates/asap-aware-mapping/src/empirical_cost.rs +++ b/crates/asap-aware-mapping/src/empirical_cost.rs @@ -357,7 +357,7 @@ fn valid_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> bool { layers, }, ) => *heap_size > 0 && *sketch_rows > 0 && *sketch_cols > 0 && (1..=64).contains(layers), - (SketchAlgorithm::Kll, SketchParams::Kll { k }) + (SketchAlgorithm::Kll, SketchParams::Kll { k, .. }) | (SketchAlgorithm::Kmv, SketchParams::Kmv { k }) | (SketchAlgorithm::Theta, SketchParams::Theta { k }) => *k > 0, (SketchAlgorithm::Hll, SketchParams::Hll { precision }) => (4..=18).contains(precision), @@ -524,6 +524,29 @@ mod tests { ) } + /// A KLL measurement matches a KLL of the same `k` whatever failure + /// probability either was sized for. + #[test] + fn lookup_ignores_the_kll_failure_probability() { + let (mut artifact, context, _) = fixture(); + artifact.records.truncate(1); + artifact.records[0].algorithm = SketchAlgorithm::Kll; + artifact.records[0].params = SketchParams::Kll { + k: 331, + failure_probability: None, + }; + let provider = EmpiricalEvidenceProvider::new(artifact, context).unwrap(); + assert!(provider + .lookup( + &SketchAlgorithm::Kll, + &SketchParams::Kll { + k: 331, + failure_probability: Some(0.001), + }, + ) + .is_ok()); + } + /// Real replacement generation follows measured update ranking while keeping /// every candidate and the same formally sized parameter configurations. #[test] diff --git a/crates/asap-aware-mapping/src/grouping.rs b/crates/asap-aware-mapping/src/grouping.rs index e866925ad..759697ab9 100644 --- a/crates/asap-aware-mapping/src/grouping.rs +++ b/crates/asap-aware-mapping/src/grouping.rs @@ -762,7 +762,7 @@ mod tests { #[test] fn cardinality_has_no_hydra_candidate_yet() { - // summary_candidates(Cardinality) = [Hll, Theta, Kmv] — none have a + // summary_candidates(Cardinality) = [Hll, Kmv, UnivMon] — none have a // modeled Hydra variant, so no candidate at all (not an error, just // an empty result, same conservatism as every other strategy here). let q = agg(vec![2], default_cardinality(), metric_scan(&["job"])); diff --git a/crates/asap-aware-mapping/src/realization.rs b/crates/asap-aware-mapping/src/realization.rs index 128625d77..bb7df9d15 100644 --- a/crates/asap-aware-mapping/src/realization.rs +++ b/crates/asap-aware-mapping/src/realization.rs @@ -27,15 +27,14 @@ pub fn summary_candidates(intent: &AggIntent) -> &'static [SketchAlgorithm] { // (`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. - AggIntent::Cardinality { cols, .. } if cols.len() > 1 => &[ - SketchAlgorithm::Hll, - SketchAlgorithm::Theta, - SketchAlgorithm::Kmv, - ], + // 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::Theta, SketchAlgorithm::Kmv, SketchAlgorithm::UnivMon, ], diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 9aa870f97..0d59d0dfa 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -7561,7 +7561,10 @@ mod tests { preferred(&approx), Realization::Sketch(SketchKind::new( SketchAlgorithm::Kll, - SketchParams::Kll { k: 269 }, + SketchParams::Kll { + k: 269, + failure_probability: None + }, )) ); @@ -7574,7 +7577,10 @@ mod tests { preferred(&looser), Realization::Sketch(SketchKind::new( SketchAlgorithm::Kll, - SketchParams::Kll { k: 52 }, + SketchParams::Kll { + k: 52, + failure_probability: None + }, )) ); } @@ -7675,7 +7681,6 @@ mod tests { summary_candidates(&default_cardinality()), &[ SketchAlgorithm::Hll, - SketchAlgorithm::Theta, SketchAlgorithm::Kmv, SketchAlgorithm::UnivMon ] @@ -7730,7 +7735,10 @@ mod tests { preferred(&intent), Realization::Sketch(SketchKind::new( SketchAlgorithm::Kll, - SketchParams::Kll { k: 65_535 }, + SketchParams::Kll { + k: 26_602, + failure_probability: None + }, )) ); } @@ -7990,7 +7998,6 @@ mod tests { kinds, vec![ SketchAlgorithm::Hll, - SketchAlgorithm::Theta, SketchAlgorithm::Kmv, SketchAlgorithm::UnivMon, ] @@ -8021,7 +8028,6 @@ mod tests { kinds, vec![ SketchAlgorithm::Hll, - SketchAlgorithm::Theta, SketchAlgorithm::Kmv, SketchAlgorithm::UnivMon, ] @@ -8379,14 +8385,14 @@ mod tests { } #[test] - fn cardinality_group_keeps_all_four_candidates() { + fn cardinality_group_keeps_all_three_candidates() { let root = agg(vec![2], default_cardinality(), metric_scan(&["job"])); let space = search_workload(vec![("q", root)]).unwrap(); let agg_group = space .target_subdag_candidates() .find(|g| matches!(g.target.non_asap(), Some(NonASAPOp::Aggregate { .. }))) .unwrap(); - assert_eq!(agg_group.candidates.len(), 4); + assert_eq!(agg_group.candidates.len(), 3); assert!(agg_group.candidates.iter().any(|candidate| matches!( &candidate.replacement, Replacement::SubDAG(node) if node.guarantee.is_none() @@ -9763,7 +9769,13 @@ mod tests { assert_eq!( family, &FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 269 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 269, + failure_probability: None + } + ), GroupingStrategy::default() ) ); @@ -9774,7 +9786,13 @@ mod tests { assert_eq!( field(&summary_input.schema, "value").dtype, FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 269 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 269, + failure_probability: None + } + ), GroupingStrategy::default() ) ); diff --git a/crates/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index d58cfc382..cfb31b349 100644 --- a/crates/asap-aware-mapping/src/stage3_plan_selection.rs +++ b/crates/asap-aware-mapping/src/stage3_plan_selection.rs @@ -31,8 +31,14 @@ //! Prices come from [`crate::analytical_cost::estimate_operator`] over edge //! statistics derived from the [`DataWorkload`] and a fixed default group //! count; summary build and estimation are priced as rows × sketch depth and -//! groups × k. These numbers are illustrative, not calibrated: a binary, -//! join or concat node is priced from its first input only (MVP; follow-up). +//! groups × k. State is priced by its bytes: a sketch by its parameters, a +//! fixed-size exact accumulator (sum, count, min, max, rate, ...) as one +//! value per group, and exact state that grows with the data (quantile, +//! top-k, distinct count, L2, entropy keep every value or item) as one value +//! per input row of its group, assuming rows spread evenly over groups and, +//! for distinct items, that every row may be distinct. These numbers are +//! illustrative, not calibrated: a binary, join or concat node is priced from +//! its first input only (MVP; follow-up). //! A cost error on any candidate aborts selection; it is reachable only from //! bad declared rates. Latency bounds and deployment capabilities are not //! checked yet. @@ -46,11 +52,12 @@ use asap_types::ir::physical_export::{ use asap_types::ir::{ ASAPOp, BinaryOperator, NonASAPOp, Operator, OperatorNode, QueryRoot, ScalarExpr, }; +use asap_types::post_asap::sketch::KLL_MAX_K; use asap_types::post_asap::{ CompositionOperator, ErrorMetric, FieldDataType, Materialization, ResultGuarantee, SketchAlgorithm, SketchParams, SketchStatistic, WeightDomain, }; -use asap_types::pre_asap::{DataType, Reduction, ScalarValue, Schema}; +use asap_types::pre_asap::{AggIntent, DataType, Reduction, ScalarValue, Schema}; use asap_types::types::AccuracyTarget; use asap_types::workload::DataWorkload; use thiserror::Error; @@ -103,7 +110,7 @@ pub struct CandidateCost { } /// A candidate that was not selected. `valid == false`: it failed a check; -/// `true`: it lost on cost. +/// `true`: it lost on cost, or tied and lost on enumeration order. #[derive(Debug, Clone, PartialEq)] pub struct Rejection { pub id: String, @@ -181,13 +188,19 @@ pub fn select_plan( }; for candidate in cands { if candidate.id != selected && costs.contains_key(&candidate.id) { + let total = costs[&candidate.id].total; + let reason = if total == best_total { + format!( + "tied with the selected candidate {selected} at {total:.3} {COST_UNIT}; \ + lost on enumeration order" + ) + } else { + format!("costlier: {total:.3} vs {best_total:.3} {COST_UNIT}") + }; rejected.push(Rejection { id: candidate.id.clone(), valid: true, - reason: format!( - "costlier: {:.3} vs {:.3} {COST_UNIT}", - costs[&candidate.id].total, best_total - ), + reason, }); } } @@ -617,7 +630,7 @@ fn price( let calibration = ResourceCalibration { cost_per_cpu_op: 1e-6, cost_per_scan_byte: 1e-7, - cost_per_retained_byte: 0.0, + cost_per_retained_byte: 1e-7, version: "illustrative-v1".into(), }; let nodes: HashMap<_, _> = dag.nodes.iter().map(|n| (n.id, n)).collect(); @@ -694,6 +707,7 @@ fn price( _ => 1, }; let out = edge(group_count); + let held = input.rows.div_ceil(group_count.max(1)); let estimate = estimate_operator( PhysicalOperator::HashAggregate { grouping_key_count: keys, @@ -703,7 +717,11 @@ fn price( edges: unary(out), group_count, key_bytes: 16 * keys, - accumulator_bytes_per_group: 8, + accumulator_bytes_per_group: measures + .iter() + .map(|m| if keeps_every_value(m) { 8 * held } else { 8 }) + .sum::() + .max(8), }, ); ( @@ -782,7 +800,7 @@ fn price( rows: group_count, bytes: group_count * state_bytes, }; - let ops = input.rows as f64 * depth as f64; + let ops = input.rows as f64 * depth; ( out, Ok(ResourceEstimate::new(ops, out.bytes, 0)), @@ -864,12 +882,41 @@ fn row_bytes(schema: &Schema) -> u64 { .max(1) } -/// Update operations per input row and bytes per state. -fn summary_shape(family: &FieldDataType) -> (u64, u64) { +/// Exact state that keeps every value or item of its group, so grows with +/// the data; other exact accumulators are fixed-size. +fn keeps_every_value(measure: &AggIntent) -> bool { + matches!( + measure, + AggIntent::Quantile { .. } + | AggIntent::TopK { .. } + | AggIntent::Cardinality { .. } + | AggIntent::FrequencyL2 { .. } + | AggIntent::FrequencyEntropy { .. } + ) +} + +/// Update operations per input row and bytes per state, from the sketch's +/// parameters as `asap_sketchlib` allocates it: +/// - Count-Min / Count-Sketch: `depth × width` 8-byte counters, plus a +/// 24-byte entry per heap slot for the top-k variants; +/// - KLL: the 8-byte item array it allocates up front ([`kll_items`]); +/// - DDSketch: one 8-byte count per bin it touches over an assumed nine +/// decades of values ([`ddsketch_bins`]); +/// - 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. +/// +/// Update operations per input row: one counter per row of depth for +/// Count-Min and Count-Sketch, plus one heap update for their top-k variants; +/// for UnivMon, the same per layer it updates ([`univmon_expected_layers`]); +/// 1 for every other sketch. +fn summary_shape(family: &FieldDataType) -> (f64, u64) { match family { FieldDataType::Sketch(kind, _) => match kind.params() { SketchParams::Cms { width, depth } | SketchParams::CountSketch { width, depth } => { - (u64::from(*depth), 8 * u64::from(*width) * u64::from(*depth)) + (f64::from(*depth), 8 * u64::from(*width) * u64::from(*depth)) } SketchParams::CmsWithHeap { width, @@ -881,13 +928,76 @@ fn summary_shape(family: &FieldDataType) -> (u64, u64) { depth, heap_size, } => ( - u64::from(*depth) + 1, + f64::from(*depth) + 1.0, 8 * u64::from(*width) * u64::from(*depth) + 24 * u64::from(*heap_size), ), - _ => (1, 1_024), + SketchParams::Kll { k, .. } => (1.0, 8 * kll_items(*k)), + SketchParams::DDSketch { alpha } => (1.0, ddsketch_bins(*alpha).saturating_mul(8)), + SketchParams::Hll { precision } => (1.0, 1u64 << (*precision).min(63)), + SketchParams::Kmv { k } | SketchParams::Theta { k } => (1.0, 8 * u64::from(*k)), + SketchParams::UnivMon { + heap_size, + sketch_rows, + sketch_cols, + layers, + } => ( + 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)), + ), }, - _ => (1, 8), + _ => (1.0, 8), + } +} + +/// Items `asap_sketchlib`'s KLL allocates up front for `k` +/// (`KLL::init_internal` → `compute_max_capacity(k, 8)`): over its 61 levels, +/// `max(ceil(k · (2/3)^i), 8)` each, with `k` capped at [`KLL_MAX_K`] as the +/// library caps it. 1,206 items at `k` = 269. +fn kll_items(k: u32) -> u64 { + const LEVELS: usize = 61; + const MIN_LEVEL_CAPACITY: f64 = 8.0; + let k = f64::from(k.min(KLL_MAX_K)); + let mut scale = 1.0_f64; + let mut total = 0; + for _ in 0..LEVELS { + total += (k * scale).ceil().max(MIN_LEVEL_CAPACITY) as u64; + scale *= 2.0 / 3.0; } + total +} + +/// 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` +/// with probability `2^-j`, and the expectation is +/// `sum_{j < layers} 2^-j = 2 - 2^(1 - layers)`. +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. +const DDSKETCH_ASSUMED_VALUE_RANGE: f64 = 1e9; + +/// 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)] @@ -897,7 +1007,9 @@ mod tests { use crate::stage2_physical_asap_optimization::optimize_physical_asap; use crate::test_support::lower_promql; use asap_types::ir::{ExprSemantics, ScalarExpr}; - use asap_types::post_asap::{BoundExpr, ErrorMetric, ProbabilityExpr}; + use asap_types::post_asap::{ + BoundExpr, ErrorMetric, GroupingStrategy, ProbabilityExpr, SketchKind, + }; use asap_types::pre_asap::expr_ir::ArithmeticOpKind; use asap_types::workload::{Evidence, Rate}; @@ -1211,6 +1323,269 @@ mod tests { } } + /// Exact quantile state keeps every value of its group: over one group of + /// 10 000 values it costs more than a quantile sketch sized for + /// ε = 0.01, δ = 0.01, which is selected; over 100 groups of 100 values it + /// is smaller than the sketch and is selected. + #[test] + fn exact_quantile_state_costs_more_than_a_sketch_over_large_groups() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + for (query, sketch_wins) in [ + ("quantile(0.99, rate(x[5m]))", true), + ("quantile by (job) (0.99, rate(x[5m]))", false), + ] { + let candidates = all_candidates(query, &target); + let selection = select_plan( + &candidates, + &[Some(target.clone())], + &data(), + PlanningModels::builtin(), + ) + .unwrap(); + let selected = candidates + .iter() + .find(|c| c.id == selection.selected) + .unwrap(); + assert_eq!( + has_summary_estimate(selected), + sketch_wins, + "{query}: {selection:?}" + ); + } + } + + /// Exact L2 state keeps one counter per distinct value of its group, so + /// it is priced like exact quantile state, which keeps every value. + #[test] + fn exact_l2_state_is_priced_like_exact_quantile_state() { + let aggregate_cost = |query: &str| { + let candidates = all_candidates(query, &AccuracyTarget::Exact); + let selection = + select_plan(&candidates, &[None], &data(), PlanningModels::builtin()).unwrap(); + let exact = candidates + .iter() + .find(|c| !has_summary_estimate(c)) + .unwrap(); + selection.costs[&exact.id] + .per_node + .values() + .find(|n| n.detail.starts_with("hash aggregate")) + .unwrap() + .cost + }; + assert_eq!( + aggregate_cost("l2_over_time(x[5m])"), + aggregate_cost("quantile_over_time(0.99, x[5m])") + ); + } + + /// A DDSketch is sized for values over nine decades: 1,037 bins at + /// α = 0.01, 8 bytes each. + #[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(), + )) + .1, + 8 * 1_037 + ); + } + + /// DDSketch state grows as its relative accuracy α tightens. + #[test] + fn ddsketch_state_grows_as_alpha_shrinks() { + let bytes = |alpha| { + summary_shape(&FieldDataType::Sketch( + SketchKind::new(SketchAlgorithm::DDSketch, SketchParams::DDSketch { alpha }), + GroupingStrategy::default(), + )) + .1 + }; + assert!( + bytes(0.1) < bytes(0.01), + "{} vs {}", + bytes(0.1), + bytes(0.01) + ); + assert!( + bytes(0.01) < bytes(0.001), + "{} vs {}", + bytes(0.01), + bytes(0.001) + ); + } + + /// Every sketch family's state is sized from its parameters, as + /// `asap_sketchlib` allocates it. + #[test] + fn summary_state_is_sized_from_sketch_params() { + let bytes = |algorithm, params| { + summary_shape(&FieldDataType::Sketch( + SketchKind::new(algorithm, params), + GroupingStrategy::default(), + )) + .1 + }; + assert_eq!( + bytes(SketchAlgorithm::Hll, SketchParams::Hll { precision: 14 }), + 1 << 14 + ); + assert_eq!( + bytes(SketchAlgorithm::Kmv, SketchParams::Kmv { k: 1_000 }), + 8 * 1_000 + ); + assert_eq!( + bytes( + SketchAlgorithm::UnivMon, + SketchParams::UnivMon { + heap_size: 256, + sketch_rows: 5, + sketch_cols: 1_024, + layers: 16, + } + ), + 16 * (8 * 5 * 1_024 + 24 * 256) + ); + } + + /// KLL state is the item array `asap_sketchlib` allocates up front: + /// `compute_max_capacity(k, 8)` 8-byte items, with `k` capped at 26 602. + #[test] + fn kll_state_is_sized_as_asap_sketchlib_allocates_it() { + let bytes = |k| { + summary_shape(&FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k, + failure_probability: None, + }, + ), + GroupingStrategy::default(), + )) + .1 + }; + assert_eq!(bytes(269), 8 * 1_206); + assert_eq!(bytes(26_602), 8 * 80_118); + assert_eq!(bytes(30_000), 8 * 80_118); + } + + /// 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 + /// per sketch row and its heap. + #[test] + fn univmon_update_cost_is_expected_layers_times_rows_and_heap() { + let ops = |layers| { + summary_shape(&FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::UnivMon, + SketchParams::UnivMon { + heap_size: 256, + sketch_rows: 5, + sketch_cols: 1_024, + layers, + }, + ), + GroupingStrategy::default(), + )) + .0 + }; + assert_eq!(ops(1), 6.0); + assert_eq!(ops(2), 1.5 * 6.0); + assert_eq!(ops(16), (2.0 - 2f64.powi(-15)) * 6.0); + } + + /// Exact distinct-count state keeps every item of its group: over 10 000 + /// groups of 300 rows it is smaller than a distinct-count sketch sized for + /// ε = 0.01 and is selected; over one group of 3 000 000 rows a sketch is. + #[test] + fn exact_distinct_count_beats_a_sketch_over_small_groups() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let one_series = DataWorkload { + input_cardinality: Evidence { + value: Some(1), + ..Default::default() + }, + ..data() + }; + for (data, sketch_wins) in [(data(), false), (one_series, true)] { + let candidates = all_candidates("distinct_over_time(x[5m])", &target); + let selection = select_plan( + &candidates, + &[Some(target.clone())], + &data, + PlanningModels::builtin(), + ) + .unwrap(); + let selected = candidates + .iter() + .find(|c| c.id == selection.selected) + .unwrap(); + assert_eq!(has_summary_estimate(selected), sketch_wins, "{selection:?}"); + } + } + + /// A candidate that ties the selected one is rejected for losing on + /// enumeration order, not as costlier. + #[test] + fn a_tie_is_reported_as_lost_on_enumeration_order() { + let mut candidates = candidates(); + candidates.truncate(1); + let mut twin = candidates[0].clone(); + twin.id = "P1-twin".into(); + candidates.push(twin); + let selection = + select_plan(&candidates, &[None], &data(), PlanningModels::builtin()).unwrap(); + assert_eq!(selection.selected, "P1"); + let reason = &selection.rejected[0].reason; + assert!( + reason.contains("tied with the selected candidate P1") + && reason.contains("enumeration order"), + "{reason}" + ); + } + + /// At δ = 0.001 a KLL candidate meets the target and is valid. + #[test] + fn kll_candidate_meets_a_failure_probability_below_one_percent() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.001, + }; + let candidates = all_candidates("quantile(0.99, rate(x[5m]))", &target); + let selection = select_plan( + &candidates, + &[Some(target)], + &data(), + PlanningModels::builtin(), + ) + .unwrap(); + let kll = candidates + .iter() + .find(|c| { + c.dag.nodes.iter().any(|n| { + matches!(&n.payload, Payload::ASAP(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), .. + }) if *kind.algorithm() == SketchAlgorithm::Kll) + }) + }) + .unwrap(); + assert!(selection.costs.contains_key(&kll.id), "{selection:?}"); + } + /// `scalar(v)` inherits `v`'s guarantee; arithmetic over it composes /// (`c · x` scales an absolute bound, `x − y` adds them); a rank error /// does not compose (a rank interval counts as none); one over exact diff --git a/crates/asap-aware-mapping/tests/per_computation_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs index d63889f3b..912ee6ed0 100644 --- a/crates/asap-aware-mapping/tests/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -90,7 +90,6 @@ fn cardinality_keeps_exact_specialized_and_universal_candidates() { algorithms(&choices), vec![ SketchAlgorithm::Hll, - SketchAlgorithm::Theta, SketchAlgorithm::Kmv, SketchAlgorithm::UnivMon ] @@ -286,7 +285,13 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { let summary = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child: scan(false), family: FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ), Default::default(), ), input: SummaryUpdate::column(ColumnRef::SampleValue), diff --git a/crates/asap-physical-operators/src/capability.rs b/crates/asap-physical-operators/src/capability.rs index 8bdb1891e..85a088f54 100644 --- a/crates/asap-physical-operators/src/capability.rs +++ b/crates/asap-physical-operators/src/capability.rs @@ -48,7 +48,12 @@ pub fn validate_summary_kernel( use SketchAlgorithm as A; use SketchParams as P; match (kind.algorithm(), kind.params()) { - (A::Kll, P::Kll { k }) if (8..=u16::MAX as u32).contains(k) => false, + // `asap_sketchlib` silently caps a larger `k`. + (A::Kll, P::Kll { k, .. }) + if (8..=planner_types::post_asap::sketch::KLL_MAX_K).contains(k) => + { + false + } (A::DDSketch, P::DDSketch { alpha }) if alpha.is_finite() && *alpha > 0.0 && *alpha < 1.0 => { diff --git a/crates/asap-physical-operators/src/summary_kernels/factory.rs b/crates/asap-physical-operators/src/summary_kernels/factory.rs index be819f8a7..2ec2f6a1a 100644 --- a/crates/asap-physical-operators/src/summary_kernels/factory.rs +++ b/crates/asap-physical-operators/src/summary_kernels/factory.rs @@ -563,9 +563,14 @@ pub fn create_planner_accumulator( return Err("Planner family and operator grouping disagree".into()); } let updater: Box = match (kind.algorithm(), kind.params()) { - (SketchAlgorithm::Kll, SketchParams::Kll { k }) => Box::new(KllAccumulatorUpdater::new( - u16::try_from(*k).map_err(|_| "KLL k exceeds runtime bound")?, - )), + (SketchAlgorithm::Kll, SketchParams::Kll { k, .. }) => { + if *k > planner_types::post_asap::sketch::KLL_MAX_K { + return Err("KLL k exceeds runtime bound".into()); + } + Box::new(KllAccumulatorUpdater::new( + u16::try_from(*k).map_err(|_| "KLL k exceeds runtime bound")?, + )) + } (SketchAlgorithm::DDSketch, SketchParams::DDSketch { alpha }) => { Box::new(DDSketchAccumulatorUpdater::new(*alpha)) } diff --git a/crates/asap-physical-operators/src/values.rs b/crates/asap-physical-operators/src/values.rs index 41640b7b6..9b22f0a2e 100644 --- a/crates/asap-physical-operators/src/values.rs +++ b/crates/asap-physical-operators/src/values.rs @@ -291,7 +291,7 @@ fn validate_state(family: &SummaryFamilyType, state: &dyn AggregateCore) -> Resu .downcast_ref::() .is_some_and(|s| s.family() == family && !s.is_keyed()), SummaryFamilyType::Sketch(kind, _) => match kind.params() { - SketchParams::Kll { k } => state + SketchParams::Kll { k, .. } => state .as_any() .downcast_ref::() .is_some_and(|s| u32::from(s.inner.k()) == *k), diff --git a/crates/asap-physical-operators/tests/common/mod.rs b/crates/asap-physical-operators/tests/common/mod.rs index a3b900287..4f79f7d83 100644 --- a/crates/asap-physical-operators/tests/common/mod.rs +++ b/crates/asap-physical-operators/tests/common/mod.rs @@ -32,8 +32,8 @@ pub fn selected_dag( } /// Every logical candidate (a composed CandidateLogicalASAPDAG) of the single -/// query `root` at `accuracy`, in enumeration order; a choice that does not -/// compose is skipped. +/// query `root` at `accuracy`, in enumeration order; a candidate that does not +/// compose panics, so none is silently skipped. pub fn logical_candidates( root: Rc, accuracy: planner_types::types::AccuracyTarget, @@ -41,9 +41,15 @@ pub fn logical_candidates( run_stage1_to_stage3(root, accuracy) .stage1_logical_asap_optimization .into_iter() - .filter_map(|candidate| match candidate.roots.ok()?.remove(0) { - QueryRoot::Operator(node) => Some(node), - QueryRoot::Scalar(_) => panic!("operator root"), + .map(|candidate| { + match candidate + .roots + .unwrap_or_else(|e| panic!("{} does not compose: {e}", candidate.id)) + .remove(0) + { + QueryRoot::Operator(node) => node, + QueryRoot::Scalar(_) => panic!("operator root"), + } }) .collect() } diff --git a/crates/asap-physical-operators/tests/deployment.rs b/crates/asap-physical-operators/tests/deployment.rs index 72e0503f1..42c85ca5b 100644 --- a/crates/asap-physical-operators/tests/deployment.rs +++ b/crates/asap-physical-operators/tests/deployment.rs @@ -9,7 +9,13 @@ use asap_physical_operators::{factory::create_planner_accumulator, AggregateCore fn family(k: u32) -> FieldDataType { FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k, + failure_probability: None, + }, + ), GroupingStrategy::PerSubpopulationInstance, ) } diff --git a/crates/asap-physical-operators/tests/physical_dag.rs b/crates/asap-physical-operators/tests/physical_dag.rs index 686eadb7e..e7a981f97 100644 --- a/crates/asap-physical-operators/tests/physical_dag.rs +++ b/crates/asap-physical-operators/tests/physical_dag.rs @@ -321,7 +321,13 @@ fn kll_raw_partial_and_precomputed_are_native_dags() { let input = schema(&[("value", DataType::Float64, false)]); let family = FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 512 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 512, + failure_probability: None, + }, + ), GroupingStrategy::PerSubpopulationInstance, ); let build = Operator::summary_build(input.clone(), family, 0, None, vec![]).unwrap(); diff --git a/crates/asap-physical-operators/tests/physical_semantics.rs b/crates/asap-physical-operators/tests/physical_semantics.rs index 042487f1c..266378368 100644 --- a/crates/asap-physical-operators/tests/physical_semantics.rs +++ b/crates/asap-physical-operators/tests/physical_semantics.rs @@ -559,7 +559,13 @@ fn kll_partial_merge_and_multiple_evaluations_preserve_population() { let input = schema(&[("v", DataType::Float64, false)]); let family = FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 512 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 512, + failure_probability: None, + }, + ), Default::default(), ); let mut dag = PhysicalDAG::default(); diff --git a/crates/asap-physical-operators/tests/plan_properties.rs b/crates/asap-physical-operators/tests/plan_properties.rs index ab488162d..231aecf93 100644 --- a/crates/asap-physical-operators/tests/plan_properties.rs +++ b/crates/asap-physical-operators/tests/plan_properties.rs @@ -121,6 +121,37 @@ fn blocking_inputs_require_an_explicit_finite_source() { } } +/// A KLL kernel accepts `k` only up to the 26 602 `asap_sketchlib` caps it at. +#[test] +fn kll_kernel_rejects_a_k_above_the_library_cap() { + use asap_physical_operators::capability::validate_summary_kernel; + use planner_types::post_asap::{ + GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryInputExpr, + SummaryUpdate, + }; + use planner_types::pre_asap::ColumnRef; + let grouping = GroupingStrategy::default(); + let update = SummaryUpdate { + item: None, + weight: SummaryInputExpr::Column(ColumnRef::SampleValue), + weight_domain: Default::default(), + }; + let kll = |k| { + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k, + failure_probability: None, + }, + ), + grouping.clone(), + ) + }; + assert!(validate_summary_kernel(&kll(26_602), &update, &grouping).is_ok()); + assert!(validate_summary_kernel(&kll(26_603), &update, &grouping).is_err()); +} + // Kernel support must not be mistaken for executable native state/evaluation support. #[test] fn summary_capability_levels_are_distinct() { @@ -167,7 +198,13 @@ fn summary_capability_levels_are_distinct() { ) .is_err()); let kll = FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 128 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 128, + failure_probability: None, + }, + ), grouping, ); assert!(validate_native_family(&kll).is_ok()); diff --git a/crates/asap-physical-operators/tests/precompute_candidates.rs b/crates/asap-physical-operators/tests/precompute_candidates.rs index 3b9b494af..752d4272f 100644 --- a/crates/asap-physical-operators/tests/precompute_candidates.rs +++ b/crates/asap-physical-operators/tests/precompute_candidates.rs @@ -20,11 +20,16 @@ use std::{collections::BTreeMap, sync::Arc}; /// `sum by(job)(rate(m[1m]))` with its rows carrying the series identity. fn grouped_rate_root() -> std::rc::Rc { + exact_promql_root("sum by(job)(rate(m[1m]))") +} + +/// `query`, exact, lowered with its rows carrying the series identity. +fn exact_promql_root(query: &str) -> std::rc::Rc { let workload = PlanningWorkload { query_workload: QueryWorkload { language: QueryLanguage::PromQL, query_batch: Some(vec![BatchEntry { - query: Query("sum by(job)(rate(m[1m]))".into()), + query: Query(query.into()), requirements: QueryRequirements { accuracy: AccuracyRequirement::Explicit(AccuracyTarget::Exact), ..Default::default() @@ -726,3 +731,17 @@ fn cut_candidate_rejects_invalid_frontiers() { assert!(compile_candidate(&dag, inputs.clone(), &[root], &frontier).is_err()); } } + +/// A candidate that does not compose fails `logical_candidates` instead of +/// being skipped, so an assertion over every candidate cannot pass on none. +/// An aggregate over a top-k output does not compose yet (#672). +/// This test relies on that bug: when #672 is fixed, the query composes and +/// this test must switch to another candidate that does not compose. +#[test] +#[should_panic(expected = "does not compose")] +fn logical_candidates_fails_on_a_candidate_that_does_not_compose() { + logical_candidates( + exact_promql_root("sum by (job) (topk by (job, instance) (1, rate(m[5m])))"), + AccuracyTarget::Exact, + ); +} diff --git a/crates/frontend-promql/tests/count_planning.rs b/crates/frontend-promql/tests/count_planning.rs index e88b483d4..40a07b8bd 100644 --- a/crates/frontend-promql/tests/count_planning.rs +++ b/crates/frontend-promql/tests/count_planning.rs @@ -275,3 +275,20 @@ fn cms_count_updates_total_ten_for_zero_positive_and_negative_samples() { assert_eq!(total, 10.0, "ten samples of {value}"); } } + +/// A candidate that does not compose fails `logical_candidates` instead of +/// being skipped, so an assertion over every candidate cannot pass on none. +/// An aggregate over a top-k output does not compose yet (#672). +/// This test relies on that bug: when #672 is fixed, the query composes and +/// this test must switch to another candidate that does not compose. +#[test] +#[should_panic(expected = "does not compose")] +fn logical_candidates_fails_on_a_candidate_that_does_not_compose() { + let accuracy = AccuracyTarget::Epsilon(0.01); + let root = lower_promql( + "sum by (job) (topk by (job, instance) (1, rate(m[5m])))", + accuracy.clone(), + ) + .unwrap(); + logical_candidates(root, accuracy); +} diff --git a/crates/frontend-promql/tests/support.rs b/crates/frontend-promql/tests/support.rs index 0d8f79927..f133b5887 100644 --- a/crates/frontend-promql/tests/support.rs +++ b/crates/frontend-promql/tests/support.rs @@ -138,8 +138,8 @@ pub fn selected_dag(root: Rc, accuracy: AccuracyTarget) -> Rc, @@ -148,9 +148,15 @@ pub fn logical_candidates( run_stage1_to_stage3(root, accuracy) .stage1_logical_asap_optimization .into_iter() - .filter_map(|candidate| match candidate.roots.ok()?.remove(0) { - asap_types::ir::QueryRoot::Operator(node) => Some(node), - asap_types::ir::QueryRoot::Scalar(_) => None, + .filter_map(|candidate| { + match candidate + .roots + .unwrap_or_else(|e| panic!("{} does not compose: {e}", candidate.id)) + .remove(0) + { + asap_types::ir::QueryRoot::Operator(node) => Some(node), + asap_types::ir::QueryRoot::Scalar(_) => None, + } }) .collect() } diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs index 28819a282..12593fd49 100644 --- a/crates/integration-tests/src/lib.rs +++ b/crates/integration-tests/src/lib.rs @@ -174,8 +174,8 @@ pub mod planning { } /// Every logical candidate (a composed CandidateLogicalASAPDAG) of the - /// single query `root`, in enumeration order; a choice that does not - /// compose is skipped. + /// single query `root`, in enumeration order; a candidate that does not + /// compose panics, so none is silently skipped. pub fn logical_candidates( root: Rc, accuracy: AccuracyTarget, @@ -183,9 +183,15 @@ pub mod planning { run_stage1_to_stage3(root, accuracy) .stage1_logical_asap_optimization .into_iter() - .filter_map(|candidate| match candidate.roots.ok()?.remove(0) { - QueryRoot::Operator(node) => Some(node), - QueryRoot::Scalar(_) => None, + .filter_map(|candidate| { + match candidate + .roots + .unwrap_or_else(|e| panic!("{} does not compose: {e}", candidate.id)) + .remove(0) + { + QueryRoot::Operator(node) => Some(node), + QueryRoot::Scalar(_) => None, + } }) .collect() } diff --git a/crates/integration-tests/tests/kll_pane_execution.rs b/crates/integration-tests/tests/kll_pane_execution.rs index 4869d681b..c7aa5194f 100644 --- a/crates/integration-tests/tests/kll_pane_execution.rs +++ b/crates/integration-tests/tests/kll_pane_execution.rs @@ -26,7 +26,13 @@ use std::{ fn family(k: u32) -> FieldDataType { FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k, + failure_probability: None, + }, + ), Default::default(), ) } diff --git a/crates/integration-tests/tests/planner_layering_example2.rs b/crates/integration-tests/tests/planner_layering_example2.rs index f2861fc90..44c71be52 100644 --- a/crates/integration-tests/tests/planner_layering_example2.rs +++ b/crates/integration-tests/tests/planner_layering_example2.rs @@ -317,9 +317,7 @@ fn stage1_logical_asap_optimization_pass1_offers_exact_and_univmon_for_each_stat fn stage1_logical_asap_optimization_pass1_offers_a_distinct_count_summary() { let found = options(&pipeline(), 0); assert!( - found - .iter() - .any(|o| ["Hll", "Theta", "Kmv"].contains(&o.as_str())), + found.iter().any(|o| ["Hll", "Kmv"].contains(&o.as_str())), "{found:?}" ); } @@ -495,6 +493,21 @@ fn stage3_plan_selection_charges_each_node_per_consuming_query() { assert_each_node_charged_per_consuming_query(&pipeline()); } +/// `asap_sketchlib` has no Theta sketch, so no candidate builds one. +#[test] +fn stage3_plan_selection_has_no_theta_candidate() { + let run = pipeline(); + for (id, cost) in &run.selection.costs { + assert!( + !cost + .per_node + .values() + .any(|n| n.detail.starts_with("build Theta")), + "{id} builds Theta" + ); + } +} + /// A UnivMon estimate has a guarantee, so plan selection judges it against its target instead of rejecting it for having none. #[test] #[ignore = "needs a UnivMon accuracy model: L2 in #621; distinct count and entropy have none yet"] diff --git a/crates/integration-tests/tests/planner_layering_example3.rs b/crates/integration-tests/tests/planner_layering_example3.rs index 96a75dab7..06450e771 100644 --- a/crates/integration-tests/tests/planner_layering_example3.rs +++ b/crates/integration-tests/tests/planner_layering_example3.rs @@ -150,7 +150,7 @@ fn stage1_logical_asap_optimization_a_keeps_five_independent_klls() { let sizes: BTreeSet<_> = builds .iter() .map(|(_, _, params)| match params { - SketchParams::Kll { k } => *k, + SketchParams::Kll { k, .. } => *k, other => panic!("{other:?}"), }) .collect(); diff --git a/crates/integration-tests/tests/promql_numeric_regressions.rs b/crates/integration-tests/tests/promql_numeric_regressions.rs index 2a5a8383c..d28144faa 100644 --- a/crates/integration-tests/tests/promql_numeric_regressions.rs +++ b/crates/integration-tests/tests/promql_numeric_regressions.rs @@ -265,3 +265,20 @@ fn sketch_counts_use_unit_weights_and_signed_sums_keep_value_weights() { post_asap_dag(node); } } + +/// A candidate that does not compose fails `logical_candidates` instead of +/// being skipped, so an assertion over every candidate cannot pass on none. +/// An aggregate over a top-k output does not compose yet (#672). +/// This test relies on that bug: when #672 is fixed, the query composes and +/// this test must switch to another candidate that does not compose. +#[test] +#[should_panic(expected = "does not compose")] +fn logical_candidates_fails_on_a_candidate_that_does_not_compose() { + let accuracy = AccuracyTarget::Epsilon(0.01); + let pre = lower_promql( + "sum by (job) (topk by (job, instance) (1, rate(m[5m])))", + accuracy.clone(), + ) + .unwrap(); + logical_candidates(pre, accuracy); +} diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index ee74ebb31..cb49b6aba 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -902,7 +902,13 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { assert_eq!( family, &FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 269 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 269, + failure_probability: None + } + ), GroupingStrategy::default() ) ); @@ -915,7 +921,13 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { assert_eq!( dtype(&summary_input.schema, "value"), &FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 269 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 269, + failure_probability: None + } + ), GroupingStrategy::default() ) ); diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index 27cda7c4a..3e4fcbe5d 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -677,7 +677,13 @@ async fn sql_quantile_binds_kll_sketch_over_named_column() { assert_eq!( family, &FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 269 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 269, + failure_probability: None + } + ), GroupingStrategy::default() ) ); @@ -697,7 +703,13 @@ async fn sql_quantile_binds_kll_sketch_over_named_column() { assert_eq!( summary_input.schema.fields[0].dtype, FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 269 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 269, + failure_probability: None + } + ), GroupingStrategy::default() ) ); diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs index b426bfed5..24206273c 100644 --- a/crates/planner/tests/summary_sharing.rs +++ b/crates/planner/tests/summary_sharing.rs @@ -259,7 +259,7 @@ fn kll_k(plan: &QueryPlan) -> u32 { else { panic!("sketch state: {:?}", deployment.operator); }; - let SketchParams::Kll { k } = kind.params() else { + let SketchParams::Kll { k, .. } = kind.params() else { panic!("KLL state: {kind:?}"); }; *k @@ -267,7 +267,7 @@ fn kll_k(plan: &QueryPlan) -> u32 { /// The KLL `k` sized for `epsilon`. fn kll_k_for(epsilon: f64) -> u32 { - let SketchParams::Kll { k } = default_size_params( + let SketchParams::Kll { k, .. } = default_size_params( SketchAlgorithm::Kll, &default_quantile(0.5), epsilon, diff --git a/crates/types/src/ir/timing.rs b/crates/types/src/ir/timing.rs index 9d09c172c..25dc4bfad 100644 --- a/crates/types/src/ir/timing.rs +++ b/crates/types/src/ir/timing.rs @@ -850,7 +850,13 @@ mod tests { fn kll() -> FieldDataType { FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ), GroupingStrategy::default(), ) } diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index 65e78d24f..b0a338c12 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -73,11 +73,19 @@ pub enum SketchAlgorithm { CountSketchWithHeap, } +/// The largest KLL `k` `asap_sketchlib` builds: it silently caps a larger +/// `k` at this (`MAX_CACHEABLE_K` in its `sketches::kll`). +pub const KLL_MAX_K: u32 = 26_602; + /// Concrete, catalog-validated parameters for a specific [`SketchAlgorithm`] /// instance. The variant must correspond to the associated `SketchAlgorithm`; /// mismatches are caught at post-ASAP bind time, before any later, /// deployment-specific stage ever sees the plan. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +/// +/// Equality compares the sketch that is built: a KLL's `failure_probability` +/// records what it was sized for, not its structure, so two KLLs with the +/// same `k` are equal (they share, and match the same offline evidence). +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum SketchParams { UnivMon { heap_size: u32, @@ -87,6 +95,13 @@ pub enum SketchParams { }, Kll { k: u32, + /// The failure probability `k` was sized for, when below the + /// empirical 99% calibration's 0.01; `None` is that calibration. + /// Sizing never records `Some(δ)` for δ ≥ 0.01; a hand-built one is + /// read as a claim at that δ with the calibration's error bound, and + /// one outside (0, 1) gives no guarantee. Ignored by equality. + #[serde(default, skip_serializing_if = "Option::is_none")] + failure_probability: Option, }, Cms { width: u32, @@ -120,6 +135,88 @@ pub enum SketchParams { }, } +impl PartialEq for SketchParams { + fn eq(&self, other: &Self) -> bool { + use SketchParams as P; + match (self, other) { + (P::Kll { k: a, .. }, P::Kll { k: b, .. }) => a == b, + ( + P::UnivMon { + heap_size: a0, + sketch_rows: a1, + sketch_cols: a2, + layers: a3, + }, + P::UnivMon { + heap_size: b0, + sketch_rows: b1, + sketch_cols: b2, + layers: b3, + }, + ) => (a0, a1, a2, a3) == (b0, b1, b2, b3), + ( + P::Cms { + width: a0, + depth: a1, + }, + P::Cms { + width: b0, + depth: b1, + }, + ) + | ( + P::CountSketch { + width: a0, + depth: a1, + }, + P::CountSketch { + width: b0, + depth: b1, + }, + ) => (a0, a1) == (b0, b1), + ( + P::CmsWithHeap { + width: a0, + depth: a1, + heap_size: a2, + }, + P::CmsWithHeap { + width: b0, + depth: b1, + heap_size: b2, + }, + ) + | ( + P::CountSketchWithHeap { + width: a0, + depth: a1, + heap_size: a2, + }, + P::CountSketchWithHeap { + width: b0, + depth: b1, + heap_size: b2, + }, + ) => (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, + // Different variants. Listed by `self`'s variant, not `_`, so a + // new variant fails to compile here until it is compared above. + (P::UnivMon { .. }, _) + | (P::Kll { .. }, _) + | (P::Cms { .. }, _) + | (P::Hll { .. }, _) + | (P::DDSketch { .. }, _) + | (P::CmsWithHeap { .. }, _) + | (P::Kmv { .. }, _) + | (P::Theta { .. }, _) + | (P::CountSketch { .. }, _) + | (P::CountSketchWithHeap { .. }, _) => false, + } + } +} + /// The query category served by a committed sketch. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum SketchCategory { @@ -470,7 +567,7 @@ pub fn default_hydra_params( per_subpopulation_params: &SketchParams, ) -> Option { match (kind, per_subpopulation_params) { - (HydraKind::HydraKll, SketchParams::Kll { k }) => Some(HydraParams::HydraKll { + (HydraKind::HydraKll, SketchParams::Kll { k, .. }) => Some(HydraParams::HydraKll { k: *k, shared_buckets: *k, }), @@ -658,11 +755,38 @@ mod tests { #[test] fn sketch_kind_classifies_a_valid_algorithm_and_params_pair() { - let kind = SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }); + let kind = SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ); assert_eq!(kind.category(), SketchCategory::Quantile); assert_eq!(kind.algorithm(), &SketchAlgorithm::Kll); } + /// Two KLLs with the same `k` are the same sketch, whatever failure + /// probability each was sized for, so they compare equal and can be shared. + #[test] + fn kll_params_compare_equal_whatever_their_failure_probability() { + let kll = |k, failure_probability| SketchParams::Kll { + k, + failure_probability, + }; + assert_eq!(kll(331, Some(0.001)), kll(331, None)); + assert_eq!(kll(331, Some(0.001)), kll(331, Some(0.0001))); + assert_ne!(kll(331, None), kll(269, None)); + assert_ne!(kll(331, None), SketchParams::Kmv { k: 331 }); + let family = |params| { + crate::post_asap::FieldDataType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, params), + GroupingStrategy::default(), + ) + }; + assert_eq!(family(kll(331, Some(0.001))), family(kll(331, None))); + } + #[test] #[should_panic(expected = "SketchKind parameter mismatch")] fn sketch_kind_rejects_params_from_another_algorithm() { @@ -707,7 +831,13 @@ mod tests { #[test] fn default_hydra_params_carries_over_per_subpopulation_params_by_kind() { assert_eq!( - default_hydra_params(HydraKind::HydraKll, &SketchParams::Kll { k: 200 }), + default_hydra_params( + HydraKind::HydraKll, + &SketchParams::Kll { + k: 200, + failure_probability: None + } + ), Some(HydraParams::HydraKll { k: 200, shared_buckets: 200, @@ -752,7 +882,13 @@ mod tests { // came from must agree. Degrades to `None` rather than panicking, // matching this module's conservative stance elsewhere. assert_eq!( - default_hydra_params(HydraKind::HydraCms, &SketchParams::Kll { k: 200 }), + default_hydra_params( + HydraKind::HydraCms, + &SketchParams::Kll { + k: 200, + failure_probability: None + } + ), None ); } diff --git a/crates/types/tests/structure_contract.rs b/crates/types/tests/structure_contract.rs index 181301237..ee502421e 100644 --- a/crates/types/tests/structure_contract.rs +++ b/crates/types/tests/structure_contract.rs @@ -91,7 +91,13 @@ fn state_evaluations_and_passthrough_keep_their_contracts() { }; use asap_types::pre_asap::{ColumnRef, Reduction}; let family = FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 100 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 100, + failure_probability: None, + }, + ), GroupingStrategy::default(), ); let state = Rc::new( diff --git a/crates/types/tests/summary_coverage.rs b/crates/types/tests/summary_coverage.rs index 9cbb6c68d..4d489edb2 100644 --- a/crates/types/tests/summary_coverage.rs +++ b/crates/types/tests/summary_coverage.rs @@ -84,7 +84,13 @@ fn kll(child: Rc, column: &str) -> Rc { OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child, family: FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ), Default::default(), ), input: SummaryUpdate::column(ColumnRef::Named(column.into())), @@ -584,7 +590,13 @@ fn design_doc_worked_example() { let state = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child: project, family: FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k: 200, + failure_probability: None, + }, + ), Default::default(), ), input: SummaryUpdate::column(ColumnRef::Named("latency".into())), diff --git a/crates/types/tests/summary_merge_structure.rs b/crates/types/tests/summary_merge_structure.rs index b621d8403..d2b4194d9 100644 --- a/crates/types/tests/summary_merge_structure.rs +++ b/crates/types/tests/summary_merge_structure.rs @@ -34,7 +34,13 @@ fn state(k: u32, region: &str) -> Rc { OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child: only_region, family: FieldDataType::Sketch( - SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }), + SketchKind::new( + SketchAlgorithm::Kll, + SketchParams::Kll { + k, + failure_probability: None, + }, + ), Default::default(), ), input: SummaryUpdate::column(ColumnRef::SampleValue), diff --git a/docs/design_docs/proposals/asap-primitive-schema.md b/docs/design_docs/proposals/asap-primitive-schema.md index 5f2856b49..bf901594f 100644 --- a/docs/design_docs/proposals/asap-primitive-schema.md +++ b/docs/design_docs/proposals/asap-primitive-schema.md @@ -87,7 +87,7 @@ Based on our requirement, each field should contain the following information. | Intent | Candidate types | Evaluation | |---|---|---| | `Quantile` | `Sketch`: `Kll`, `DDSketch` | `SummaryEstimate(Quantile { q })` | - | `Cardinality` | `Sketch`: `Hll`, `Theta`, `Kmv`, `UnivMon` (one column only) | `SummaryEstimate(Cardinality)` | + | `Cardinality` | `Sketch`: `Hll`, `Kmv`, `UnivMon` (one column only) | `SummaryEstimate(Cardinality)` | | `Count` (approximate) | `Sketch`: `Cms`, `CountSketch`, `UnivMon` | `SummaryEstimate(PointCount { .. })` | | `TopK` | `Sketch`: `CmsWithHeap`, `CountSketchWithHeap` | `SummaryEstimate(TopK { k })` | | `FrequencyL2`, `FrequencyEntropy` | `Sketch`: `UnivMon` | `SummaryEstimate(FrequencyL2 \| FrequencyEntropy)` | @@ -1158,6 +1158,7 @@ impl SketchKind { } pub enum SketchCategory { Universal, Quantile, Cardinality, Frequency, TopK } // Universal: UnivMon | Quantile: Kll, DDSketch | Cardinality: Hll, Theta, Kmv +// (Theta is not offered as a candidate until asap_sketchlib has a Theta sketch) // Frequency: Cms, CountSketch | TopK: CmsWithHeap, CountSketchWithHeap pub enum SketchAlgorithm { UnivMon, Kll, Cms, Hll, DDSketch, CmsWithHeap, Kmv, Theta, CountSketch, CountSketchWithHeap } diff --git a/docs/design_docs/proposals/planner-layering-example2-acceptance.md b/docs/design_docs/proposals/planner-layering-example2-acceptance.md index 26cd2b826..10be5a909 100644 --- a/docs/design_docs/proposals/planner-layering-example2-acceptance.md +++ b/docs/design_docs/proposals/planner-layering-example2-acceptance.md @@ -40,10 +40,10 @@ with Q1–Q3's targets and a 15-s ingestion interval (PromQL needs one). | Stage | Doc | Today (stand-in) | |---|---|---| | Query-language frontends | One summary-free DAG: `Distinct`, `Entropy`, `L2` over `flows.src_ip`, last 1 m | Same for the stand-in. SQL: Q1 is `aggregate:cardinality`; Q2 and Q3 are `count` per `src_ip` → window function / projection → `sum` (not recognized). | -| Logical ASAP-aware optimization, Pass 1 | Each computation: exact, a specialized summary, UnivMon. 3 × 3 × 3 = **27** | Q1: exact, HLL, Theta, KMV, UnivMon. Q2, Q3: exact, UnivMon (no specialized entropy or norm summary). 5 × 2 × 2 = **20** | +| Logical ASAP-aware optimization, Pass 1 | Each computation: exact, a specialized summary, UnivMon. 3 × 3 × 3 = **27** | Q1: exact, HLL, KMV, UnivMon (no Theta: `asap_sketchlib` has no Theta sketch). Q2, Q3: exact, UnivMon (no specialized entropy or norm summary). 4 × 2 × 2 = **16** | | Logical ASAP-aware optimization, Pass 2 | Summary-capability rule: + 1 (one UnivMon for all three, sized for ε = 0.01) + 9 (3 pairs share a UnivMon × 3 options of the third). Independent kept. **37**. Window composition left out. | Not implemented: **20**, no candidate shares a node. | | Physical ASAP-aware optimization | Same options as Example 1: no window form → rebuilt from raw samples each refresh. | One physical candidate per logical candidate, every sub-DAG `NotMaterialized`: **20**. | -| Plan selection | One UnivMon sized for ε = 0.01 vs three separate summaries; the shared one usually wins (one update per flow record instead of three). | HLL is invalid (its bound has no failure probability for the (ε, δ) target) and so is every UnivMon (its estimates have no guarantee). Q1 on Theta or KMV with Q2, Q3 exact is valid but costlier. The all-exact plan is selected. | +| Plan selection | One UnivMon sized for ε = 0.01 vs three separate summaries; the shared one usually wins (one update per flow record instead of three). | HLL is invalid (its bound has no failure probability for the (ε, δ) target) and so is every UnivMon (its estimates have no guarantee). Q1 on KMV with Q2, Q3 exact is valid but costlier. The all-exact plan is selected. | ## Invariants and tests @@ -63,7 +63,7 @@ with Q1–Q3's targets and a 15-s ingestion interval (PromQL needs one). |---|---|---| | Once recognized, the SQL workload has the stand-in's Pass 1 options | `stage1_logical_asap_optimization_sql_has_the_standin_options` | ignored: #598 (ambiguity 9) | | Each statistic has an exact and a UnivMon option | `stage1_logical_asap_optimization_pass1_offers_exact_and_univmon_for_each_statistic` | passes | -| Q1 has a specialized distinct-count summary (HLL, Theta or KMV) | `stage1_logical_asap_optimization_pass1_offers_a_distinct_count_summary` | passes | +| Q1 has a specialized distinct-count summary (HLL or KMV) | `stage1_logical_asap_optimization_pass1_offers_a_distinct_count_summary` | passes | | Q2 and Q3 have a specialized entropy and norm summary | `stage1_logical_asap_optimization_pass1_offers_specialized_entropy_and_l2_summaries` | ignored: no PR yet | | Every combination of the queries' own options is kept, with no shared summary | `stage1_logical_asap_optimization_keeps_every_independent_combination` | passes | | A candidate has one UnivMon build read by all three queries, feeding a distinct-count, an entropy and an L2 estimate | `stage1_logical_asap_optimization_summary_capability_adds_one_univmon_for_all_three` | ignored: #595, #596 | diff --git a/docs/design_docs/proposals/planner-layering-example3-acceptance.md b/docs/design_docs/proposals/planner-layering-example3-acceptance.md index 73cf4877c..6ef0e8a01 100644 --- a/docs/design_docs/proposals/planner-layering-example3-acceptance.md +++ b/docs/design_docs/proposals/planner-layering-example3-acceptance.md @@ -10,7 +10,7 @@ pipeline in `asap_devtools::{planner_layering_examples, planning_stages}`). Devtool: `stage_pipeline --example planner-layering-3a` and `planner-layering-3b`. Both run every candidate (243 for 3a) through plan selection; `--max-displayed` (default 64) limits only how many the document -shows, so 3a selects P122, KLL for all five queries. +shows, so 3a selects P109: KLL for q1 and q2, exact for q3–q5. This spec was written by a test designer who does not implement the stages. Tests run the stages through `planner_layering_common::run_stages`, which @@ -43,7 +43,7 @@ Shared data workload, except Pattern A's `arrival` is `mixed`. | Logical ASAP-aware optimization, Pass 1 | A: exact or KLL over its own interval per query, 2⁵ = 32. B: exact, KLL. | A: exact, KLL (k = 547), DDSketch per query, 3⁵ = **243**. B: exact, KLL (k = 269), DDSketch, **3**. | | Logical ASAP-aware optimization, Pass 2 | A: window composition adds one EH of KLLs over [T − 5 y, T], one merge + p99 estimate per query, plus a candidate for every grouping of two or more queries onto shared window summaries ("hundreds"). B: each Pass 1 option × {none, sliding L = 5 min s = 1 min (5 active windows), 1-min tumbling (merge 5)} = 6. Independent kept. | Not implemented: no shared input, no window summaries. | | Physical ASAP-aware optimization | See Example 4. | One physical candidate per logical candidate, every sub-DAG `NotMaterialized`. | -| Plan selection | Not stated for Example 3. | Every candidate is valid. A: five independent KLLs (DDSketch ties on cost). B: KLL (DDSketch ties). | +| Plan selection | Not stated for Example 3. | Every candidate is valid. A: KLL for q1 and q2, exact for q3–q5, whose rows are underestimated (#632); DDSketch costs slightly more than KLL. B: exact, since each window holds 20 samples per series; DDSketch and then KLL cost more. | ## Invariants and tests diff --git a/docs/develop_docs/extend-asap-aware-mapping.md b/docs/develop_docs/extend-asap-aware-mapping.md index b9a830140..06527ccf5 100644 --- a/docs/develop_docs/extend-asap-aware-mapping.md +++ b/docs/develop_docs/extend-asap-aware-mapping.md @@ -440,7 +440,7 @@ Check the identities or kinds of all candidates, not just the preferred one. The current sketch-family tests explicitly verify that: - quantile returns both KLL and DDSketch, -- cardinality returns HLL, Theta, and KMV. +- cardinality returns HLL and KMV (Theta is not offered until `asap_sketchlib` has a Theta sketch). This is the most important regression test for a strategy. diff --git a/docs/develop_docs/pre-asap-ir.md b/docs/develop_docs/pre-asap-ir.md index f73e28a30..60db2f5fc 100644 --- a/docs/develop_docs/pre-asap-ir.md +++ b/docs/develop_docs/pre-asap-ir.md @@ -293,7 +293,7 @@ before `Cardinality` took a list, SQL lowering dropped every argument after the first, reporting single-column cardinality as tuple cardinality. Realization is the single-column one with a wider item: a tuple becomes a -`SummaryInputExpr::Tuple`, which the distinct-count sketches (HLL, Theta, KMV) +`SummaryInputExpr::Tuple`, which the distinct-count sketches (HLL, KMV) hash as one value. UnivMon is withheld from a tuple — it estimates frequency moments over a single value stream. At `AccuracyTarget::Exact` the node stays a logical pass-through at any arity. Each SQL argument must be a bare column, diff --git a/docs/user_guide_docs/stage-viewer.md b/docs/user_guide_docs/stage-viewer.md index b13891cdc..debf3b8cf 100644 --- a/docs/user_guide_docs/stage-viewer.md +++ b/docs/user_guide_docs/stage-viewer.md @@ -69,9 +69,9 @@ plan wins. | Tab | Workload | What to look for | | --- | --- | --- | | 1 · dashboard panels | Two PromQL panels over 1M counter series every 10 s: an exact per-job rate, and the approximate top-10 series per job | The exact query stays exact; the CMS+heap and CountSketch+heap top-k plans are rejected with their reasons, so the all-exact plan wins; each query reads its own scan until cross-query sharing lands | -| 2 · SQL flow statistics | Distinct sources, entropy and L2 of per-source counts over the last minute of flows | All 80 plans compose and are priced; none is pruned. The SQL frontend does not yet recognize entropy and L2, so Q2 and Q3 are per-source counts with exact, Count-Min, CountSketch or UnivMon options. 68 plans are invalid: HLL for Q1 because its bound has "failure probability None", and UnivMon for Q1 or Count-Min or CountSketch for Q2 and Q3 because "the output has no guarantee, and the deployment has no evidence that it meets target". 12 are valid: Q1 exact, Theta or KMV with Q2 and Q3 each exact or UnivMon. P1 (all exact) is selected as the first of four plans tied at 52.80111 ms (with UnivMon on Q2, Q3 or both); Theta or KMV on Q1 costs 2·10⁻⁶ ms more | -| 3a · historical p99 batch | Five p99 reports over 1–5 years, run once | All 243 plans are priced. Nothing is shared yet: each query scans its own interval and builds its own sketch at query time. P122, KLL for all five queries, is selected: it is the first of 32 plans tied at the lowest cost, because KLL and DDSketch cost the same for every query; exact on any query costs more. The all-exact plan costs about 16% more | -| 3b · live p99 panel | p99 over the last 5 min every minute, 1M series | P2, KLL built from the window's raw samples at each evaluation, is selected; DDSketch costs the same and exact about 14% more. Everything runs at query time and nothing is materialized. KLL panes kept from ingestion time and the 200 ms latency check come with the window-summary PRs | +| 2 · SQL flow statistics | Distinct sources, entropy and L2 of per-source counts over the last minute of flows | All 64 plans compose and are priced; none is pruned. The SQL frontend does not yet recognize entropy and L2, so Q2 and Q3 are per-source counts with exact, Count-Min, CountSketch or UnivMon options. 56 plans are invalid: HLL for Q1 because its bound has "failure probability None", and UnivMon for Q1 or Count-Min or CountSketch for Q2 and Q3 because "the output has no guarantee, and the deployment has no evidence that it meets target". 8 are valid: Q1 exact or KMV with Q2 and Q3 each exact or UnivMon. Theta is not offered: `asap_sketchlib` has no Theta sketch. P1 (all exact) is selected as the first of four plans tied at 52.80111 ms (with UnivMon on Q2, Q3 or both); KMV on Q1 costs more, because its state is sized from its k | +| 3a · historical p99 batch | Five p99 reports over 1–5 years, run once | All 243 plans are priced. Nothing is shared yet: each query scans its own interval and builds its own sketch at query time. P109 is selected: KLL for the 5-year and 1-year queries, and exact for the three offset queries, whose rows are underestimated (#632). DDSketch in place of KLL costs slightly more, because its state is larger. The all-exact plan costs about 29% more | +| 3b · live p99 panel | p99 over the last 5 min every minute, 1M series | P1, exact over the window's raw samples at each evaluation, is selected: each series' 5-min window holds 20 samples, less state than any sketch. DDSketch costs about 5.9 times as much and KLL about 6.7 times. Everything runs at query time and nothing is materialized. KLL panes kept from ingestion time and the 200 ms latency check come with the window-summary PRs | | 4a · monthly p99 reports | Example 3a repeated monthly | The same plan, with its cost amortized over a month | | 4b · panes that pay off | p99 over the last hour every 10 min, 1k series sampled every second, raw data not kept by the deployment | Maintaining six 10-min KLL panes at ingestion time wins | diff --git a/tools/dag-viewer/examples/planner-layering-example1.json b/tools/dag-viewer/examples/planner-layering-example1.json index 3ee848f6a..2f79af830 100644 --- a/tools/dag-viewer/examples/planner-layering-example1.json +++ b/tools/dag-viewer/examples/planner-layering-example1.json @@ -8009,32 +8009,32 @@ "P1": { "per_node": { "0": { - "cost": 23.2, + "cost": 23.2000048, "detail": "scan 4000000 samples", "multiplier": 1 }, "1": { - "cost": 4.0, + "cost": 4.0000048, "detail": "time range 60s: pass 4000000 rows", "multiplier": 1 }, "10": { - "cost": 14.0, + "cost": 14.048, "detail": "sort 1000000 rows in 100 partitions", "multiplier": 1 }, "11": { - "cost": 0.001, + "cost": 0.0010048, "detail": "limit to 1000 rows", "multiplier": 1 }, "12": { - "cost": 0.001, + "cost": 0.0010048, "detail": "operator: pass 1000 rows", "multiplier": 1 }, "2": { - "cost": 4.0, + "cost": 4.8, "detail": "build exact Rate accumulator into 1000000 states: 4000000 rows x depth 1", "multiplier": 1 }, @@ -8044,7 +8044,7 @@ "multiplier": 1 }, "4": { - "cost": 1.0, + "cost": 1.00008, "detail": "build exact Sum accumulator into 100 states: 1000000 rows x depth 1", "multiplier": 1 }, @@ -8054,17 +8054,17 @@ "multiplier": 1 }, "6": { - "cost": 23.2, + "cost": 23.2000048, "detail": "scan 4000000 samples", "multiplier": 1 }, "7": { - "cost": 4.0, + "cost": 4.0000048, "detail": "time range 60s: pass 4000000 rows", "multiplier": 1 }, "8": { - "cost": 4.0, + "cost": 4.8, "detail": "build exact Sum accumulator into 1000000 states: 4000000 rows x depth 1", "multiplier": 1 }, @@ -8075,7 +8075,7 @@ } }, "source": "analytical-cost-v1 (illustrative statistics)", - "total": 79.40210000000002, + "total": 81.05020880000001, "unit": "cpu_ms_per_workload_evaluation" } }, diff --git a/tools/dag-viewer/examples/scalar-queries.json b/tools/dag-viewer/examples/scalar-queries.json index 05e7ef899..71fd1adf4 100644 --- a/tools/dag-viewer/examples/scalar-queries.json +++ b/tools/dag-viewer/examples/scalar-queries.json @@ -2577,17 +2577,17 @@ "P1": { "per_node": { "0": { - "cost": 0.039, + "cost": 0.0390016, "detail": "scan 15000 samples", "multiplier": 1 }, "1": { - "cost": 0.015, + "cost": 0.015001599999999999, "detail": "time range 15s: pass 15000 rows", "multiplier": 1 }, "10": { - "cost": 0.015, + "cost": 0.01508, "detail": "build exact Sum accumulator into 100 states: 15000 rows x depth 1", "multiplier": 1 }, @@ -2597,7 +2597,7 @@ "multiplier": 1 }, "2": { - "cost": 0.015, + "cost": 0.0150008, "detail": "build exact Sum accumulator into 1 states: 15000 rows x depth 1", "multiplier": 1 }, @@ -2607,17 +2607,17 @@ "multiplier": 1 }, "4": { - "cost": 0.063, + "cost": 0.0630032, "detail": "scan 15000 samples", "multiplier": 1 }, "5": { - "cost": 0.015, + "cost": 0.0150032, "detail": "time range 15s: pass 15000 rows", "multiplier": 1 }, "6": { - "cost": 0.015, + "cost": 0.0150008, "detail": "build exact Sum accumulator into 1 states: 15000 rows x depth 1", "multiplier": 1 }, @@ -2627,18 +2627,18 @@ "multiplier": 1 }, "8": { - "cost": 0.087, + "cost": 0.0870048, "detail": "scan 15000 samples", "multiplier": 1 }, "9": { - "cost": 0.015, + "cost": 0.015004799999999999, "detail": "time range 15s: pass 15000 rows", "multiplier": 1 } }, "source": "analytical-cost-v1 (illustrative statistics)", - "total": 0.279102, + "total": 0.2792028, "unit": "cpu_ms_per_workload_evaluation" } },