From cb0c6bf427787842565226f725ace83643000fb2 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 21 Oct 2026 05:00:38 +0000 Subject: [PATCH] fix(executor): a DDSketch takes zero and negative samples The DDSketch updater refused every sample <= 0, so a DDSketch build over a gauge that ever reports exactly 0 (or a negative value) failed at run time: #795's P5-m1 over a gauge declared non-negative, and every DDSketch selected under ignore_accuracy_guarantees. asap_sketchlib's DdSketch keeps a zero bucket and a negative store, so the updater now accepts zero and any finite sample whose magnitude is within ddsketch_indexable_bounds, positive or negative, and adds it with try_add. NaN, +-Inf, magnitudes above the range, and nonzero magnitudes below it (which the zero bucket would take without the relative error) are still refused. The DDSketch guarantee (relative error alpha per quantile) does not assume positive values; whether it carries through sum and avg still depends on the samples' declared sign, unchanged. Tests: the updater's domain and its quantiles over 0, 0, 3, 5 and over signed samples; through plan_installable and a stand-in runtime, a DDSketch-pane plan over a gauge reporting zero (avg of p99s, declared non-negative) and over signed samples including all-negative series (max of p99s), each equal to query_time_reference and within 0.05 of the exact interpolated values. Two executor tests that locked the old positive-only domain now check the new one. --- .../executor/src/summary_kernels/factory.rs | 77 ++++++++- .../executor/tests/precompute_population.rs | 33 ++-- crates/executor/tests/state_column_updater.rs | 6 +- crates/planner/tests/runtime_install.rs | 159 ++++++++++++++++++ 4 files changed, 260 insertions(+), 15 deletions(-) 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); +}