diff --git a/crates/executor/src/summary_kernels/factory.rs b/crates/executor/src/summary_kernels/factory.rs index a5808cbf4..7277fb620 100644 --- a/crates/executor/src/summary_kernels/factory.rs +++ b/crates/executor/src/summary_kernels/factory.rs @@ -194,19 +194,26 @@ impl DDSketchAccumulatorUpdater { } impl AccumulatorUpdater for DDSketchAccumulatorUpdater { + /// Zero (the sketch's zero bucket) and any finite sample whose + /// magnitude is within the indexable range, positive or negative (its + /// negative store). Refused: NaN, ±Inf, a magnitude above the range, + /// and a nonzero magnitude below it, which the zero bucket would take + /// without the relative error. fn validate_single_input(&self, value: f64) -> Result<(), String> { let (minimum, maximum) = asap_sketchlib::sketches::ddsketch::ddsketch_indexable_bounds(self.alpha); - if value.is_finite() && value > 0.0 && value >= minimum && value <= maximum { + let magnitude = value.abs(); + if value == 0.0 || (magnitude >= minimum && magnitude <= maximum) { Ok(()) } else { - Err("DDS maintenance input is outside its positive representable domain".into()) + Err(format!( + "DDSketch input {value} is outside its domain: zero, or a finite magnitude in [{minimum:e}, {maximum:e}]" + )) } } fn update_single(&mut self, value: f64, _timestamp_ms: i64) -> Result<(), String> { - self.acc.inner.add(&value); - Ok(()) + self.acc.inner.try_add(&value).map_err(str::to_string) } fn update_keyed( @@ -771,3 +778,65 @@ mod planner_parameter_regression { assert!(create_planner_accumulator(&family(128), &update, &grouping(128)).is_err()); } } + +#[cfg(test)] +mod ddsketch_domain { + use super::*; + + /// A DDSketch takes zero and negative samples (its zero bucket and + /// negative store) as well as positive ones; it refuses NaN, ±Inf, and + /// magnitudes outside its indexable range, including a nonzero magnitude + /// below it, which would lose its relative error in the zero bucket. + #[test] + fn accepts_zero_and_signed_samples_within_its_indexable_range() { + let updater = DDSketchAccumulatorUpdater::new(0.01); + let (minimum, maximum) = + asap_sketchlib::sketches::ddsketch::ddsketch_indexable_bounds(0.01); + for value in [0.0, -0.0, 3.0, -3.0, minimum, -minimum, maximum, -maximum] { + assert!(updater.validate_single_input(value).is_ok(), "{value}"); + } + for value in [ + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + minimum / 2.0, + -minimum / 2.0, + maximum * 2.0, + -maximum * 2.0, + ] { + assert!(updater.validate_single_input(value).is_err(), "{value}"); + } + } + + /// Quantiles over 0, 0, 3, 5 and over signed samples are within alpha of + /// the sample at rank ⌈q·n⌉. + #[test] + fn quantiles_over_zero_and_signed_samples() { + for (samples, expected) in [ + ( + vec![0.0, 0.0, 3.0, 5.0], + [(0.25, 0.0), (0.5, 0.0), (0.75, 3.0), (1.0, 5.0)], + ), + ( + vec![-7.0, -3.0, 0.0, 4.0], + [(0.25, -7.0), (0.5, -3.0), (0.75, 0.0), (1.0, 4.0)], + ), + ] { + let mut updater = DDSketchAccumulatorUpdater::new(0.01); + for value in &samples { + updater.validate_single_input(*value).unwrap(); + updater.update_single(*value, 0).unwrap(); + } + for (q, exact) in expected { + let estimate = updater + .acc + .estimate(&planner_types::ir::schema::SketchStatistic::Quantile { q }) + .unwrap(); + assert!( + (estimate - exact).abs() <= 0.01 * f64::abs(exact), + "{samples:?} q={q}: {estimate} vs {exact}" + ); + } + } + } +} diff --git a/crates/executor/tests/precompute_population.rs b/crates/executor/tests/precompute_population.rs index fb400640d..93ef59797 100644 --- a/crates/executor/tests/precompute_population.rs +++ b/crates/executor/tests/precompute_population.rs @@ -447,9 +447,11 @@ fn explicit_merge_changes_pane_cardinality() { } } -// Typed updates reject invalid domains before publishing any target state. +// Typed updates reject invalid domains before publishing any target state: +// a DDSketch takes zero and negative values (its zero bucket and negative +// store), and refuses NaN, ±Inf and magnitudes beyond its indexable range. #[test] -fn precompute_rejects_nonfinite_and_nonpositive_dds_updates() { +fn precompute_rejects_nonfinite_and_out_of_range_dds_updates() { let source = FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum); let target = FieldDataType::Sketch( SketchKind::new( @@ -459,14 +461,25 @@ fn precompute_rejects_nonfinite_and_nonpositive_dds_updates() { GroupingStrategy::default(), ); let program = state_dag(source.clone(), Some(target), false); - assert!(native_run( - &program, - source.clone(), - vec![sum_state(20.)], - ingestion_context(Limits::default()) - ) - .is_ok()); - for value in [-20., 0., f64::MAX, f64::NAN, f64::INFINITY] { + for value in [20., -20., 0.] { + assert!( + native_run( + &program, + source.clone(), + vec![sum_state(value)], + ingestion_context(Limits::default()) + ) + .is_ok(), + "{value}" + ); + } + for value in [ + f64::MAX, + -f64::MAX, + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + ] { assert!(native_run( &program, source.clone(), diff --git a/crates/executor/tests/state_column_updater.rs b/crates/executor/tests/state_column_updater.rs index 4cfcad8e6..fdf1b6717 100644 --- a/crates/executor/tests/state_column_updater.rs +++ b/crates/executor/tests/state_column_updater.rs @@ -516,7 +516,11 @@ fn per_sample_updates_keep_each_kernels_input_contract() { )) .unwrap(); let group = dd.add_group().unwrap(); - assert!(dd.update(group, -1.0, 0).is_err()); + // A DDSketch takes zero and negative values, not NaN or ±Inf. + dd.update(group, -1.0, 0).unwrap(); + dd.update(group, 0.0, 0).unwrap(); + assert!(dd.update(group, f64::NAN, 0).is_err()); + assert!(dd.update(group, f64::INFINITY, 0).is_err()); // A keyed family takes items, a value family takes values. let mut cms = StateColumnUpdater::new(&keyed_families()[0]).unwrap(); let group = cms.add_group().unwrap(); diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 59b336cc9..83d974498 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -1649,3 +1649,162 @@ async fn ignoring_accuracy_guarantees_plans_the_undeclared_p99_query() { "no raw samples read" ); } + +/// [`p99_avg_workload`]'s query with `outer` in place of `avg`, planned +/// for a deployment that keeps no raw samples. +async fn plan_p99(outer: &str, values_non_negative: bool) -> InstallablePlan { + let mut json = serde_json::to_value(p99_avg_workload(values_non_negative)).unwrap(); + json["query_workload"]["repeating_queries"][0]["query"] = + format!("{outer} by (region, service, job) (quantile_over_time(0.99, data[100s]))").into(); + let workload: PlanningWorkload = serde_json::from_value(json).unwrap(); + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + plan_installable(UserInput::new( + &workload, + FrontendInput::Promql { histograms: None }, + PlanningModels::builtin().with_deployment_capabilities(&capabilities), + )) + .await + .expect("a valid plan") +} + +/// `data` samples of series `0..series`, `value(series, step)` every +/// `interval_ms` over the `steps` steps ending just before `EVALUATION_MS`, +/// two series per (region, service, job) group; and each group's series' +/// windows `(EVALUATION_MS - 100 s, EVALUATION_MS]`. +#[allow(clippy::type_complexity)] +fn data_samples( + series: usize, + steps: i64, + interval_ms: i64, + value: impl Fn(usize, i64) -> f64, +) -> ( + Vec, + BTreeMap<(String, String, String), Vec>>, +) { + let mut samples = Vec::new(); + let mut per_group: BTreeMap<(String, String, String), Vec>> = BTreeMap::new(); + for s in 0..series { + let (region, service, job) = ( + format!("r{}", s % 2), + format!("s{}", s / 4), + "j".to_string(), + ); + let mut window = Vec::new(); + for step in 0..steps { + let timestamp_ms = EVALUATION_MS - steps * interval_ms + 50 + step * interval_ms; + let v = value(s, step); + samples.push(Sample { + labels: BTreeMap::from([ + ("__name__".to_string(), "data".to_string()), + ("region".to_string(), region.clone()), + ("service".to_string(), service.clone()), + ("job".to_string(), job.clone()), + ("instance".to_string(), format!("i{s}")), + ]), + timestamp_ms, + value: v, + }); + if timestamp_ms > EVALUATION_MS - 100_000 { + window.push(v); + } + } + per_group + .entry((region, service, job)) + .or_default() + .push(window); + } + (samples, per_group) +} + +/// PromQL's interpolated `quantile_over_time(q, ...)` of `window`. +fn interpolated_quantile(q: f64, window: &[f64]) -> f64 { + let mut sorted = window.to_vec(); + sorted.sort_by(f64::total_cmp); + let rank = q * (sorted.len() - 1) as f64; + let (low, high) = (rank.floor() as usize, rank.ceil() as usize); + sorted[low] + (sorted[high] - sorted[low]) * (rank - low as f64) +} + +/// Runs `plan` over `samples` and checks every group's answer against +/// `combine` of its series' exact p99s, within the relative error 0.05, and +/// against the query-time reference. +fn assert_p99s_within_alpha( + plan: &InstallablePlan, + samples: &[Sample], + per_group: &BTreeMap<(String, String, String), Vec>>, + combine: impl Fn(&[f64]) -> f64, +) { + let run = run_data_plane( + &serde_json::to_string(plan).unwrap(), + samples, + EVALUATION_MS, + &mut SummaryStore::new(), + ); + assert_eq!(run.answers_per_query, run.reference_answers_per_query); + let answers = &run.answers_per_query[0]; + assert_eq!(answers.len(), per_group.len()); + for (labels, value) in answers { + let key = ( + labels["region"].clone(), + labels["service"].clone(), + labels["job"].clone(), + ); + let p99s: Vec = per_group[&key] + .iter() + .map(|window| interpolated_quantile(0.99, window)) + .collect(); + let exact = combine(&p99s); + assert!( + (value - exact).abs() <= 0.05 * exact.abs(), + "{labels:?}: {value} vs exact {exact}" + ); + } +} + +/// A non-negative gauge that reports exactly 0 is still summarized by the +/// selected DDSketch panes (zero has its own bucket): each series' window +/// holds the samples 0, 0, 3 and 5 scaled by the series, and every answer +/// equals the query-time reference's and is within the relative error 0.05 +/// of the exact average of PromQL's interpolated p99s. +#[tokio::test] +async fn a_ddsketch_over_a_gauge_reporting_zero_runs_within_alpha() { + let plan = plan_p99("avg", true).await; + assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion"); + let (samples, per_group) = data_samples(8, 4, 20_000, |s, step| { + (s + 1) as f64 * [0.0, 0.0, 3.0, 5.0][step as usize] + }); + assert!(per_group.values().flatten().all(|window| window.len() == 4)); + let average = |p99s: &[f64]| p99s.iter().sum::() / p99s.len() as f64; + assert_p99s_within_alpha(&plan, &samples, &per_group, average); + // Many samples, a third of them zero. + let (samples, per_group) = data_samples(8, 1_100, 100, |s, step| { + (s + 1) as f64 * [0.0, 3.0, 5.0][(step % 3) as usize] + }); + assert_p99s_within_alpha(&plan, &samples, &per_group, average); +} + +/// Signed samples (negative, zero and positive) are summarized by the +/// selected DDSketch panes, through their negative store and zero bucket: +/// `max` of the per-series p99s carries the relative bound whatever the +/// samples' sign, and every answer equals the query-time reference's and is +/// within 0.05 of the exact value, including series whose samples are all +/// negative. +#[tokio::test] +async fn a_ddsketch_over_signed_samples_runs_within_alpha() { + let plan = plan_p99("max", false).await; + assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion"); + let max = |p99s: &[f64]| p99s.iter().copied().fold(f64::NEG_INFINITY, f64::max); + // Series 0..4 cross zero; series 4..8 are all negative. + let (samples, per_group) = data_samples(8, 1_100, 100, |s, step| { + let scale = (s + 1) as f64; + if s < 4 { + scale * [-7.0, -3.0, 0.0, 2.0, 4.0][(step % 5) as usize] + } else { + scale * [-9.0, -8.0, -5.0, -1.0][(step % 4) as usize] + } + }); + assert_p99s_within_alpha(&plan, &samples, &per_group, max); +}