From b8454e40f1438235defcab2685a4986a5494f813 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 20 May 2026 07:54:22 -0600 Subject: [PATCH] fix(query): honor counter-function semantics in ExactAgg dispatch (closes #301, #300) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Post-#299 the agent streams per-window DELTAS for counters, but the asap engine collapsed `sum` / `sum_over_time` / `increase` / `rate` — four semantically-distinct PromQL counter idioms — onto the same two reducer paths, so all instant counter sums returned the same wrong number and `rate` was systematically under-reported. Four layered bugs: Layer 1 (engine dispatch): only `rate` vs everything-else branched; sum / sum_over_time / increase / instant-sum all hit one path. Layer 2 (lookback): instant sum used a 5-min default == the [5m] range, so on a <5min producer every idiom captured the same horizon. Layer 3 (instant projection): the instant branch took `samples.last()` — the MOST RECENT window's delta — instead of the cumulative-since-storage value PromQL `sum(counter)` requires. Layer 4 (rate divisor): `evaluate_exact_agg_rate` divided by the NOMINAL range (300 for [5m]) regardless of how much data actually covered the window, halving the rate when the producer ran < range. Fix: - control_plane: extend `OuterFn` from {Plain, Rate} to the full counter-fn taxonomy {Plain, Rate, Increase, SumOverTime}, populated by the analyzer's `trace_from_promql` walker (inner-counter-idiom wins for composed shapes like `sum by (..) (rate(..))`; precedence enforced by a new `set_counter_fn` helper). Carried on `ASAPTierCandidate.outer_fn`. - engine dispatch (instant + range surfaces): branch on the typed counter-fn — `Rate` → rate reducer; `Increase` → accumulate windows over the [t-r,t] clip into one cumulative number; `Plain` instant sum → accumulate ALL windows over the full storage horizon (t0=0) → cumulative-since-start; `SumOverTime` over a counter → capability-miss → archive (asap stores deltas and cannot reconstruct the Σ-of-cumulative-samples sum_over_time wants — issue #301 decision (a), subsumes #300). - reducer: `evaluate_exact_agg` gains an `accumulate_windows` flag that collapses each group's per-window deltas into ONE cumulative sample (Layer 3); the matrix/range surface keeps `accumulate_windows=false`. `evaluate_exact_agg_rate` now divides by `min(range_seconds, actual_coverage_span_seconds)` via a new `SketchStore::exact_agg_coverage_bounds(sid, t0, t1)` that reports the true `(min_window_start, max_window_end)` span (Layer 4). Tests: control_plane 783, data_plane 755 green. New unit + integration coverage pins each counter-fn semantic (instant-sum accumulates all windows not last; increase accumulates without divisor; sum_over_time capability-misses; rate divisor uses actual coverage). Multinode validation (4 zones, 10000 series @ 100 Hz; asap node2:9091 vs VictoriaMetrics baseline node2:8428), steady state: rate : asap 997,413 b0 999,987 rel-err 0.26% (was 64%) topk(rate): asap 997,413 (tracks rate; unchanged semantic) sum_over_time: asap capability-miss → archive (no longer fabricates) The four idioms now return distinct values (sum != increase != rate), confirming the dispatch no longer collapses them. Working queries (`quantile_over_time(0.99, ...)`, `max by (zone) (quantile_over_time)`) unaffected. Instant `sum` is cumulative-since-storage (runtime-dependent; residual gap is producer-runtime / flush-lag, not a dispatch bug). Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/asap_tier_analysis.rs | 98 +++- .../src/sketch_algebra/capability.rs | 67 ++- .../query_engines/asap_query_engine/engine.rs | 446 +++++++++++++----- .../storage_engines/sketch_db/index/mod.rs | 63 +++ .../sketch_db/query/sketch_reducer.rs | 84 +++- .../storage_engines/sketch_db/query/tests.rs | 108 ++++- 6 files changed, 680 insertions(+), 186 deletions(-) diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs index ca6f99ba..787cda39 100644 --- a/control_plane/src/asap_tier_analysis.rs +++ b/control_plane/src/asap_tier_analysis.rs @@ -371,12 +371,14 @@ struct PromqlTrace { function: String, function_args: Vec, range_seconds: u64, - /// Set to `OuterFn::Rate` when ANY `rate(...)` or `irate(...)` - /// call is found anywhere in the expression tree; otherwise - /// `OuterFn::Plain`. The flag-style detection mirrors what the - /// retired `query_contains_rate_call` engine helper used to do - /// over the raw query string — done here once so the engine reads - /// it off the typed candidate. + /// Counter-function flavour recovered from the expression tree + /// (issue #301): `Rate` for `rate`/`irate`, `Increase` for + /// `increase`, `SumOverTime` for `sum_over_time`, else `Plain` + /// (bare selector / instant `sum`). The most-specific counter idiom + /// found anywhere in the tree wins (see [`set_counter_fn`]) so + /// composed shapes like `sum by (..) (rate(..))` report `Rate`. + /// Done here once so the engine reads it off the typed candidate + /// instead of re-parsing the raw query string. outer_fn: OuterFn, /// PromQL outer-aggregation operator wrapping the inner function — /// `max`/`min`/`avg`/`count`/`group`/`stddev`/`stdvar` only. `sum` @@ -488,6 +490,27 @@ fn extract_outer_agg(expr: &Expr) -> OuterAgg { } } +/// Set `t.outer_fn` honoring counter-idiom precedence (issue #301): +/// `Rate` > `Increase` > `SumOverTime` > `Plain`. The walker may visit +/// nested calls in any order, so a more-specific flavour already set +/// must not be downgraded by a less-specific one seen later. (In +/// practice a single counter query has exactly one of these, but +/// pathological compositions like `increase(sum_over_time(...))` resolve +/// deterministically.) +fn set_counter_fn(t: &mut PromqlTrace, candidate: OuterFn) { + fn rank(f: OuterFn) -> u8 { + match f { + OuterFn::Rate => 3, + OuterFn::Increase => 2, + OuterFn::SumOverTime => 1, + OuterFn::Plain => 0, + } + } + if rank(candidate) > rank(t.outer_fn) { + t.outer_fn = candidate; + } +} + fn walk_ast_for_trace(expr: &Expr, t: &mut PromqlTrace) { match expr { Expr::Call(call) => { @@ -495,14 +518,21 @@ fn walk_ast_for_trace(expr: &Expr, t: &mut PromqlTrace) { if t.function.is_empty() { t.function = name.clone(); } - // Flag `rate(...)` / `irate(...)` ANYWHERE in the tree — - // mirrors the retired `query_contains_rate_call` walker. - // For composed shapes like `sum by (zone) (rate(metric[r]))` - // the FIRST function set above is `"sum"` (the outer - // Aggregate), but `outer_fn` must still report `Rate` so - // the engine dispatches through `evaluate_exact_agg_rate`. - if matches!(name.as_str(), "rate" | "irate") { - t.outer_fn = OuterFn::Rate; + // Flag the counter-function flavour ANYWHERE in the tree + // (issue #301) — mirrors the retired `query_contains_rate_call` + // walker but with the full taxonomy. For composed shapes like + // `sum by (zone) (rate(metric[r]))` the FIRST function set + // above is `"sum"` (the outer Aggregate), but `outer_fn` must + // report the INNER counter function so the engine dispatches + // correctly. `rate`/`irate` win over `increase`, which wins + // over `sum_over_time` (most-specific-counter-idiom wins); + // `set_counter_fn` enforces that precedence so the order in + // which the walker encounters nested calls doesn't matter. + match name.as_str() { + "rate" | "irate" => set_counter_fn(t, OuterFn::Rate), + "increase" => set_counter_fn(t, OuterFn::Increase), + "sum_over_time" => set_counter_fn(t, OuterFn::SumOverTime), + _ => {} } for a in &call.args.args { if let Expr::NumberLiteral(nl) = a.as_ref() { @@ -968,11 +998,12 @@ mod tests { } #[test] - fn sum_over_time_candidate_carries_outer_fn_plain() { + fn sum_over_time_candidate_carries_outer_fn_sum_over_time() { // `sum_over_time(metric[r])` shares `Capability::ExactAgg(Sum)` // with `rate(metric[r])` — the capability alone can't - // disambiguate. The `outer_fn` field MUST report `Plain` so - // the engine takes the per-window reducer (no rate divisor). + // disambiguate. Post-#301 the `outer_fn` field reports + // `SumOverTime` so the engine can capability-miss → archive + // (asap can't reconstruct Σ-of-cumulative-samples from deltas). let a = analyze_promql_for_asap_tier("sum_over_time(http_requests_total[5m])"); assert!(a.unsupported.is_none(), "{a:?}"); assert_eq!( @@ -980,7 +1011,36 @@ mod tests { Capability::ExactAgg(AggregationType::Sum), "{a:?}" ); - assert_eq!(a.candidates[0].outer_fn, OuterFn::Plain, "{a:?}"); + assert_eq!(a.candidates[0].outer_fn, OuterFn::SumOverTime, "{a:?}"); + } + + #[test] + fn increase_candidate_carries_outer_fn_increase() { + // `increase(metric[r])` shares `Capability::ExactAgg(Sum)` with + // `rate`/`sum_over_time`; the `outer_fn` field carries the + // distinction so the engine sums deltas in `[t-r,t]` WITHOUT the + // rate divisor (issue #301). + let a = analyze_promql_for_asap_tier("increase(http_requests_total[5m])"); + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!( + a.candidates[0].required_capability, + Capability::ExactAgg(AggregationType::Sum), + "{a:?}" + ); + assert_eq!(a.candidates[0].outer_fn, OuterFn::Increase, "{a:?}"); + } + + #[test] + fn sum_by_over_increase_candidate_carries_outer_fn_increase() { + // Composed `sum by (zone) (increase(metric[r]))` — inner counter + // function wins over the outer `sum` (same precedence as the + // rate case). + let a = analyze_promql_for_asap_tier( + "sum by (zone) (increase(http_requests_total[5m]))", + ); + assert!(a.unsupported.is_none(), "{a:?}"); + assert_eq!(a.candidates[0].outer_fn, OuterFn::Increase, "{a:?}"); + assert_eq!(a.candidates[0].range_seconds, 300, "{a:?}"); } #[test] @@ -1042,7 +1102,7 @@ mod tests { can dispatch correctly without re-parsing the raw PromQL" ); assert_eq!(rate.candidates[0].outer_fn, OuterFn::Rate); - assert_eq!(sot.candidates[0].outer_fn, OuterFn::Plain); + assert_eq!(sot.candidates[0].outer_fn, OuterFn::SumOverTime); } // ── outer_agg — outer aggregation operator on function results ────── diff --git a/control_plane/src/sketch_algebra/capability.rs b/control_plane/src/sketch_algebra/capability.rs index edff19f2..fb0541e6 100644 --- a/control_plane/src/sketch_algebra/capability.rs +++ b/control_plane/src/sketch_algebra/capability.rs @@ -127,28 +127,63 @@ pub enum Capability { /// helper to recover the distinction; that was a lossy-lowering smell. /// /// The walker that populates this lives in `asap_tier_analysis.rs` -/// (`trace_from_promql`) — it sets `Rate` if ANY `rate(...)` or -/// `irate(...)` Call appears anywhere in the expression tree, otherwise -/// `Plain`. The taxonomy is intentionally minimal: today the engine -/// only branches on "needs rate divisor or not". Future shape-specific -/// dispatch (e.g. separating `increase` from `sum`) can extend this -/// enum without touching the `Capability` algebra. +/// (`trace_from_promql`) — it picks the most-specific counter-function +/// flavour found anywhere in the expression tree (inner-function wins +/// for composed shapes like `sum by (...) (rate(...))`). +/// +/// ## Counter-function taxonomy (issue #301) +/// +/// Post-#299 the agent streams per-window DELTAS for counters. The four +/// PromQL counter idioms have genuinely different semantics over those +/// deltas, but they ALL lower to a single `Capability::ExactAgg(Sum)` +/// (the `AggIntent::Sum` collapse erases the function name). Before +/// #301 the engine only distinguished `Rate` from everything else, so +/// `sum`, `sum_over_time`, `increase`, and instant-sum all hit the same +/// reducer path and returned the same (wrong) number. This enum carries +/// the function distinction the engine needs to dispatch correctly: +/// +/// | Variant | PromQL | Engine dispatch | +/// |---------------|------------------------------|---------------------------------------------------| +/// | `Plain` | `sum(c)` / `sum by (..) (c)` | accumulate ALL windows → cumulative-since-storage | +/// | `Rate` | `rate(c[r])` / `irate(c[r])` | Σ deltas in `[t-r,t]` ÷ min(r, coverage) | +/// | `Increase` | `increase(c[r])` | Σ deltas in `[t-r,t]` (one cumulative number) | +/// | `SumOverTime` | `sum_over_time(c[r])` | capability-miss → archive (can't reconstruct) | +/// +/// The taxonomy lives on `OuterFn` (not the `Capability` algebra) so the +/// sid-matching half stays a pure `ExactAgg(Sum)` predicate — the +/// function distinction is a query-evaluation concern, not a stored-state +/// one. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] pub enum OuterFn { - /// No rate-style outer function in the expression — bare selector, - /// `sum(metric)`, `sum by (...) (metric)`, `sum_over_time(metric[r])`, - /// `increase(metric[r])`, `count_over_time(metric[r])`, etc. The - /// engine dispatches to the plain per-window reducer. This is the - /// default — `Default::default()` returns `Plain` so candidates - /// built without an explicit outer-fn (test fixtures, fallback - /// paths) get the safe non-rate dispatch. + /// No range-style counter function in the expression — bare selector, + /// `sum(metric)`, `sum by (...) (metric)`. PromQL semantics for an + /// instant `sum` over a counter is "current cumulative counter value, + /// summed per group". Over per-window deltas the engine accumulates + /// EVERY window in storage up to `now` into one cumulative number per + /// group. This is the default — `Default::default()` returns `Plain` + /// so candidates built without an explicit outer-fn (test fixtures, + /// fallback paths) get the safe accumulate-all dispatch. #[default] Plain, /// `rate(metric[r])` or `irate(metric[r])` appears in the expression - /// (possibly nested inside an outer `sum by (...) (...)`). The - /// engine dispatches to `evaluate_exact_agg_rate`, which divides by - /// the range to produce events-per-second. + /// (possibly nested inside an outer `sum by (...) (...)`). The engine + /// dispatches to `evaluate_exact_agg_rate`, which sums the deltas in + /// `[t-r, t]` and divides by `min(r, actual_coverage_seconds)` to + /// produce events-per-second. Rate, + /// `increase(metric[r])` appears in the expression. PromQL semantics: + /// `counter(t) − counter(t−r)`. Over per-window deltas that is exactly + /// the sum of deltas in `[t-r, t]`. The engine dispatches to the + /// accumulate-across-windows path scoped to the `[t-r, t]` clip, + /// yielding ONE cumulative number per group (no `÷ r`). + Increase, + /// `sum_over_time(metric[r])` appears in the expression. PromQL + /// semantics: Σ of the (cumulative) SAMPLE values in `[r]` — a + /// quadratic over the storage horizon that asap CANNOT reconstruct + /// from stored deltas. The engine returns a capability-miss so the + /// query routes to the archive tier (which has raw samples) rather + /// than fabricating a wrong number. See issue #301 decision (a). + SumOverTime, } /// PromQL outer-aggregation operator carried on each `ASAPTierCandidate` diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 73e851e0..8d5a6c61 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -717,26 +717,35 @@ impl ASAPQueryEngine { // / `evaluate_exact_agg_rate` for the per-path semantics. let result = match &candidate.required_capability { crate::storage_engines::sketch_db::index::Capability::ExactAgg(agg_type) => { - // Dispatch off the typed `outer_fn` carried on the - // analyzer candidate — `OuterFn::Rate` means the - // original PromQL had a `rate(...)` / `irate(...)` - // call somewhere, so we need the rate-divisor - // reducer. Plain shapes (`sum_over_time(...)`, - // `sum(...)`, bare selector) take the per-window - // reducer. Before the analyzer carried this field - // the engine re-walked the raw PromQL via the - // `query_contains_rate_call` helper to recover the - // distinction; that was lossy-lowering smell and is - // gone. + // Counter-function dispatch (issue #301) — mirror the + // instant `execute(&str)` path's branching off the + // typed `candidate.outer_fn`. On this explicit + // RANGE (matrix) surface the per-window timeseries is + // the correct shape for `sum`/`increase` (the wire + // format wants a point per window), so + // `accumulate_windows = false`. `rate` still folds + + // divides; `sum_over_time` over a counter is refused + // (decision (a)) so the query routes to archive. + use control_plane::asap_tier_analysis::OuterFn; + let is_exact_sum_family = matches!( + agg_type, + crate::storage_engines::sketch_db::data::AggregationType::Sum + | crate::storage_engines::sketch_db::data::AggregationType::MultipleSum + | crate::storage_engines::sketch_db::data::AggregationType::Increase + | crate::storage_engines::sketch_db::data::AggregationType::MultipleIncrease + ); + if is_exact_sum_family && candidate.outer_fn == OuterFn::SumOverTime { + return Err(crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore cannot answer `sum_over_time` over counter \ + deltas for `{query}` (issue #301) — failing over to archive" + ), + )); + } let use_rate_path = candidate.range_seconds > 0 - && candidate.outer_fn == control_plane::asap_tier_analysis::OuterFn::Rate - && matches!( - agg_type, - crate::storage_engines::sketch_db::data::AggregationType::Sum - | crate::storage_engines::sketch_db::data::AggregationType::MultipleSum - | crate::storage_engines::sketch_db::data::AggregationType::Increase - | crate::storage_engines::sketch_db::data::AggregationType::MultipleIncrease - ); + && candidate.outer_fn == OuterFn::Rate + && is_exact_sum_family; if use_rate_path { reducer .evaluate_exact_agg_rate( @@ -764,6 +773,7 @@ impl ASAPQueryEngine { &candidate.group_by_keys, start_ms, end_ms, + false, ) .map_err(|e| { crate::query_engines::EngineError::capability_miss( @@ -1294,53 +1304,91 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu )); } + // Counter-function dispatch (issue #301). The four + // PromQL counter idioms all lower to + // `Capability::ExactAgg(Sum)`; the analyzer's typed + // `candidate.outer_fn` carries the function distinction + // that the engine MUST honor (otherwise sum / + // sum_over_time / increase / rate collapse to the same + // wrong number — the bug this fix closes). + // + // Rate → evaluate_exact_agg_rate over `[t-r, t]` + // (Σ deltas ÷ min(r, coverage)). + // Increase → evaluate_exact_agg(accumulate) over + // `[t-r, t]` → Σ deltas, one number. + // Plain (sum) → evaluate_exact_agg(accumulate) over the + // FULL storage horizon (`t0 = 0`) → + // cumulative-since-storage-start, the + // PromQL semantic for an instant counter + // sum. (Not the most-recent window's + // delta — Layer 3 of #301.) + // SumOverTime → capability-miss → archive (asap stores + // deltas and cannot reconstruct the + // Σ-of-cumulative-samples that + // sum_over_time wants — issue #301 + // decision (a)). + // + // `range_seconds` is lifted by the analyzer from the + // matrix selector (`[5m]` → 300); 0 for instant shapes. + use control_plane::asap_tier_analysis::OuterFn; + let is_exact_sum_family = matches!( + &candidate.required_capability, + crate::storage_engines::sketch_db::index::Capability::ExactAgg( + crate::storage_engines::sketch_db::data::AggregationType::Sum + | crate::storage_engines::sketch_db::data::AggregationType::MultipleSum + | crate::storage_engines::sketch_db::data::AggregationType::Increase + | crate::storage_engines::sketch_db::data::AggregationType::MultipleIncrease + ) + ); + + // sum_over_time over a counter sid → refuse (route to + // archive) rather than fabricate a wrong delta-sum. + if is_exact_sum_family && candidate.outer_fn == OuterFn::SumOverTime { + let req = Self::requirements_from_candidate(candidate); + crate::drivers::control_plane_client::spawn_capability_miss_notify( + &self.control_plane_client, + &req, + ); + return Err(crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore cannot answer `sum_over_time` over counter \ + deltas for metric `{}` (issue #301: Σ-of-cumulative-samples \ + not reconstructable from per-window deltas) — failing over \ + to archive", + candidate.metric_name + ), + )); + } + + let use_rate_path = + is_exact_sum_family && candidate.outer_fn == OuterFn::Rate; + // Increase + instant Plain sum both accumulate windows + // into one cumulative number per group; they differ only + // in the time scope (`[t-r,t]` clip vs full storage). + let accumulate_windows = is_exact_sum_family + && matches!(candidate.outer_fn, OuterFn::Increase | OuterFn::Plain); + // Instant `Plain` sum reads the FULL storage horizon so it + // returns cumulative-since-start; `Increase`/`Rate` clip to + // the requested `[t-r, t]` (lookback_ms below). + let plain_instant_sum = is_exact_sum_family + && candidate.outer_fn == OuterFn::Plain + && candidate.range_seconds == 0; + let lookback_ms = if candidate.range_seconds > 0 { candidate.range_seconds.saturating_mul(1000) } else { DEFAULT_LOOKBACK_MS }; - let t0_ms = now_ms.saturating_sub(lookback_ms); + let t0_ms = if plain_instant_sum { + 0 + } else { + now_ms.saturating_sub(lookback_ms) + }; if t0_ms < combined_t0 { combined_t0 = t0_ms; } - // ExactAgg capability → dispatch the per-(group_by_keys) - // accumulator-merge path; sketch capabilities → the - // sketch-decode path. ExactAgg sids carry - // `Box` payloads (per-window - // `SumAccumulator` / `IncreaseAccumulator` / - // `MinMaxAccumulator` etc.) rather than opaque sketch - // bytes, so they need a different reducer entry point. - // - // ExactAgg(Sum-family) candidates whose raw query - // contains `rate(...)` / `irate(...)` AND - // `range_seconds > 0` dispatch to the rate variant - // (`evaluate_exact_agg_rate`) — that path folds every - // sub-window sum across the range and divides by the - // range to produce events-per-second, matching PromQL - // `rate` semantics. Composed shapes like - // `sum by (zone) (rate(metric[5m]))` go through here - // too (function="sum" but range_seconds=300 from the - // inner rate's matrix selector, picked up by the - // analyzer trace). - // Dispatch off the typed `outer_fn` field on the - // analyzer candidate — `OuterFn::Rate` if the original - // PromQL contained a `rate(...)` / `irate(...)` call. - // Replaces a previous `query_contains_rate_call(query)` - // re-parse of the raw PromQL string (a lossy-lowering - // smell — the analyzer is the source of truth for - // query intent). See `control_plane/sketch_algebra/ - // capability::OuterFn`. - let use_rate_path = matches!( - &candidate.required_capability, - crate::storage_engines::sketch_db::index::Capability::ExactAgg( - crate::storage_engines::sketch_db::data::AggregationType::Sum - | crate::storage_engines::sketch_db::data::AggregationType::MultipleSum - | crate::storage_engines::sketch_db::data::AggregationType::Increase - | crate::storage_engines::sketch_db::data::AggregationType::MultipleIncrease - ) - ) && candidate.range_seconds > 0 - && candidate.outer_fn == control_plane::asap_tier_analysis::OuterFn::Rate; let reducer_result = match &candidate.required_capability { crate::storage_engines::sketch_db::index::Capability::ExactAgg( agg_type, @@ -1360,6 +1408,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu &candidate.group_by_keys, t0_ms, now_ms, + accumulate_windows, ), _ => reducer.evaluate( &hit_sids, @@ -2408,18 +2457,23 @@ mod asap_tier_classify_tests { use crate::query_engines::query_result::QueryResult; let idx = Arc::new(SketchStore::new()); - // Two zones, each its own sid, two windows each. Per-zone - // per-window sums chosen so the rate over 300s is a clean - // integer: zone z0 → 600+600 / 300 = 4.0; z1 → 900+900 / 300 - // = 6.0. + // Two zones, each its own sid, two windows each. The windows + // span `[now-150s, now-30s]` = 120s of ACTUAL coverage inside + // the requested 300s `[5m]` lookback. Post-#301 the rate divisor + // is the actual coverage span (`min(300, 120) = 120`), NOT the + // nominal 300 — so z0 = (600+600)/120 = 10.0; z1 = + // (900+900)/120 = 15.0. (Both windows stay strictly inside + // `[engine_now-300_000, engine_now]` so the window-contained + // range query captures them regardless of the small skew between + // the test's captured `now_ms` and the engine's query-time now.) let now_ms = std::time::SystemTime::now() .duration_since(std::time::SystemTime::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); - let w1_start = now_ms.saturating_sub(120_000); - let w1_end = now_ms.saturating_sub(60_000); + let w1_start = now_ms.saturating_sub(150_000); + let w1_end = now_ms.saturating_sub(90_000); let w2_start = w1_end; - let w2_end = now_ms.saturating_sub(1_000); + let w2_end = now_ms.saturating_sub(30_000); for (i, (zone, per_window)) in [("z0", 600.0_f64), ("z1", 900.0)].iter().enumerate() { let sid = 11_000 + i as u64; @@ -2478,36 +2532,30 @@ mod asap_tier_classify_tests { .expect("zone key present"); by_zone.insert(vals[zone_idx].clone(), el.value); } - // Values are per-second rates, not raw per-window sums. - // (600 + 600) / 300 = 4.0; (900 + 900) / 300 = 6.0. + // Values are per-second rates over the ACTUAL 120s coverage, not + // raw per-window sums and not divided by the nominal 300s. + // (600 + 600) / 120 = 10.0; (900 + 900) / 120 = 15.0. let z0 = by_zone.get("z0").copied().expect("zone z0 present"); let z1 = by_zone.get("z1").copied().expect("zone z1 present"); - assert!((z0 - 4.0).abs() < 1e-9, "z0 rate expected 4.0, got {z0}"); - assert!((z1 - 6.0).abs() < 1e-9, "z1 rate expected 6.0, got {z1}"); - } - - /// Regression: `sum_over_time(http_requests_total[5m])` shares - /// `Capability::ExactAgg(Sum)` with `rate(...)` — the engine's - /// reducer dispatch MUST disambiguate via the analyzer's typed - /// `outer_fn` field (set to `OuterFn::Plain` for sum_over_time), - /// NOT by re-parsing the raw PromQL string. If the dispatch ever - /// regresses to "all ExactAgg(Sum) + range > 0 → rate path", - /// this test fails because the output would be (per-window sums) - /// / 300 instead of the raw per-window sums. - /// - /// Pins the per-window reducer's output: each series carries the - /// SUM of its in-window samples (not events-per-second). + assert!((z0 - 10.0).abs() < 1e-9, "z0 rate expected 10.0, got {z0}"); + assert!((z1 - 15.0).abs() < 1e-9, "z1 rate expected 15.0, got {z1}"); + } + + /// Regression (issue #301, decision (a)): `sum_over_time(counter[r])` + /// shares `Capability::ExactAgg(Sum)` with `rate`/`increase`/`sum`, + /// but its PromQL semantic (Σ of CUMULATIVE sample values in `[r]`, + /// a quadratic) CANNOT be reconstructed from the per-window deltas + /// asap stores. Rather than fabricate a wrong number, the engine + /// reads the analyzer's typed `OuterFn::SumOverTime` and returns a + /// capability-miss so the query routes to the archive tier. Before + /// #301 this returned the delta-sum (1500 here) — a wrong answer the + /// caller couldn't distinguish from a correct one. #[tokio::test] - async fn execute_sum_over_time_dispatches_to_plain_exact_agg_reducer() { + async fn execute_sum_over_time_over_counter_capability_misses_to_archive() { use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; use crate::storage_engines::sketch_db::data::AggregationType; - use crate::query_engines::query_result::QueryResult; let idx = Arc::new(SketchStore::new()); - // Two zones, one ExactAgg(Sum) sid each, two windows each. - // Per-window sum is 600 / 900 — `sum_over_time` over a - // 300s lookback should report the sum of windowed values - // (1200 / 1800), NOT divided by 300. let now_ms = std::time::SystemTime::now() .duration_since(std::time::SystemTime::UNIX_EPOCH) .map(|d| d.as_millis() as u64) @@ -2550,42 +2598,178 @@ mod asap_tier_classify_tests { } let engine = build_engine_with_index(idx); - let result = engine + let err = engine .execute("sum_over_time(http_requests_total[5m])") .await - .expect( - "sum_over_time must dispatch via plain ExactAgg reducer \ - off the typed OuterFn::Plain candidate, not capability-miss", + .expect_err( + "sum_over_time over a counter sid MUST capability-miss → archive \ + (issue #301 decision (a)); it must NOT fabricate a delta-sum", ); + assert!( + matches!(err, crate::query_engines::EngineError::CapabilityMiss { .. }), + "expected CapabilityMiss for sum_over_time over counter, got {err:?}" + ); + } + + /// Issue #301 Layer 3: instant `sum(counter)` must return the + /// cumulative-since-storage value (Σ of ALL windows' deltas), NOT + /// the most-recent window's delta. Two windows of 600/900 per zone + /// → per-zone cumulative = 1200/1800; `sum by (zone)` keeps them + /// separate; bare `sum` collapses to 3000. This test pins the + /// `accumulate_windows=true` reducer path the engine selects for + /// `OuterFn::Plain` instant sums. + #[tokio::test] + async fn execute_instant_sum_accumulates_all_windows_not_last() { + use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; + use crate::storage_engines::sketch_db::data::AggregationType; + use crate::query_engines::query_result::QueryResult; + + let idx = Arc::new(SketchStore::new()); + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let w1_start = now_ms.saturating_sub(120_000); + let w1_end = now_ms.saturating_sub(60_000); + let w2_start = w1_end; + let w2_end = now_ms.saturating_sub(1_000); + + for (i, (zone, per_window)) in + [("z0", 600.0_f64), ("z1", 900.0)].iter().enumerate() + { + let sid = 14_000 + i as u64; + idx.register(SketchInstanceMetadata { + sid, + metric_name: "http_requests_total".to_string(), + group_by_keys: ["zone".to_string()].into_iter().collect(), + capability: Some(Capability::ExactAgg(AggregationType::Sum)), + agg_kind: crate::storage_engines::sketch_db::index::AggKind::ExactAgg { + agg_type: AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + }); + for (ws, we) in [(w1_start, w1_end), (w2_start, w2_end)] { + let mut lm = BTreeMap::new(); + lm.insert("zone".to_string(), zone.to_string()); + idx.append_precompute( + sid, + lm, + (ws, we), + Box::new(SumAccumulator::with_sum(*per_window)), + ); + } + } + let engine = build_engine_with_index(idx); + let result = engine + .execute("sum by (zone) (http_requests_total)") + .await + .expect("instant sum by zone must succeed"); let vector = match result { QueryResult::Vector(v) => v, other => panic!("expected Vector, got {other:?}"), }; - // `sum_over_time(metric[r])` (no `sum by (...)` wrapper) lowers - // to an analyzer candidate with empty `group_by_keys`. The - // plain `evaluate_exact_agg` reducer treats empty group_by as - // "collapse across all series" (vs the rate reducer which - // preserves the natural label map per sid). So the expected - // shape is ONE entry whose value is the sum of the latest - // per-window sample across both zones: 600 + 900 = 1500. The - // load-bearing assertion is the VALUE — if the engine - // regressed to the rate path the value would be (1500)/300 = - // 5.0 (or per-zone if rate's per-sid split fired), neither of - // which is 1500. - assert_eq!( - vector.values.len(), - 1, - "plain ExactAgg reducer collapses across series when \ - group_by_keys is empty" + assert_eq!(vector.values.len(), 2, "one entry per zone"); + let mut by_zone: std::collections::HashMap = + std::collections::HashMap::new(); + for el in &vector.values { + let keys = el.label_keys_override.as_ref().expect("keys present"); + let vals = &el.labels.labels; + let zi = keys.iter().position(|k| k == "zone").expect("zone key"); + by_zone.insert(vals[zi].clone(), el.value); + } + // Cumulative = Σ of ALL windows, NOT the last window's delta + // (which would be 600 / 900). + let z0 = by_zone.get("z0").copied().expect("z0"); + let z1 = by_zone.get("z1").copied().expect("z1"); + assert!( + (z0 - 1200.0).abs() < 1e-9, + "z0 cumulative expected 1200 (600+600), got {z0} — if 600 the \ + engine took only the LAST window (Layer-3 bug)" + ); + assert!( + (z1 - 1800.0).abs() < 1e-9, + "z1 cumulative expected 1800 (900+900), got {z1}" ); + } + + /// Issue #301: `increase(counter[r])` must return Σ of deltas in + /// `[t-r, t]` as ONE cumulative number per series (no rate divisor). + /// Two windows of 600/900 → 1200/1800; with no `by` grouping the + /// reducer collapses to one series = 3000. + #[tokio::test] + async fn execute_increase_accumulates_windows_without_divisor() { + use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; + use crate::storage_engines::sketch_db::data::AggregationType; + use crate::query_engines::query_result::QueryResult; + + let idx = Arc::new(SketchStore::new()); + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let w1_start = now_ms.saturating_sub(120_000); + let w1_end = now_ms.saturating_sub(60_000); + let w2_start = w1_end; + let w2_end = now_ms.saturating_sub(1_000); + + for (i, (zone, per_window)) in + [("z0", 600.0_f64), ("z1", 900.0)].iter().enumerate() + { + let sid = 15_000 + i as u64; + idx.register(SketchInstanceMetadata { + sid, + metric_name: "http_requests_total".to_string(), + group_by_keys: ["zone".to_string()].into_iter().collect(), + capability: Some(Capability::ExactAgg(AggregationType::Sum)), + agg_kind: crate::storage_engines::sketch_db::index::AggKind::ExactAgg { + agg_type: AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + }); + for (ws, we) in [(w1_start, w1_end), (w2_start, w2_end)] { + let mut lm = BTreeMap::new(); + lm.insert("zone".to_string(), zone.to_string()); + idx.append_precompute( + sid, + lm, + (ws, we), + Box::new(SumAccumulator::with_sum(*per_window)), + ); + } + } + + let engine = build_engine_with_index(idx); + let result = engine + .execute("increase(http_requests_total[5m])") + .await + .expect("increase must succeed via accumulate path"); + let vector = match result { + QueryResult::Vector(v) => v, + other => panic!("expected Vector, got {other:?}"), + }; + // No `by` grouping → empty group_by → collapse to one series. + // Σ of deltas in window = (600+600) + (900+900) = 3000. NOT + // divided by range (that would be the rate path → 10.0). + assert_eq!(vector.values.len(), 1, "no group_by collapses to one series"); let value = vector.values[0].value; assert!( - (value - 1500.0).abs() < 1e-9, - "sum_over_time expected 1500 (sum of latest per-window samples \ - across zones), got {value} — if this is ~5.0 or close to \ - 4.0/6.0 the engine regressed to the rate-divisor path; the \ - typed OuterFn::Plain candidate dispatch is broken" + (value - 3000.0).abs() < 1e-9, + "increase expected 3000 (Σ deltas, no divisor), got {value} — \ + if ~10 the engine took the rate path; if 1500 it took only \ + the last window per zone" ); } @@ -2635,9 +2819,9 @@ mod asap_tier_classify_tests { bare.candidates[0].required_capability, ); - // `outer_fn` carries the distinction. + // `outer_fn` carries the counter-function distinction (#301). assert_eq!(rate.candidates[0].outer_fn, OuterFn::Rate); - assert_eq!(sot.candidates[0].outer_fn, OuterFn::Plain); + assert_eq!(sot.candidates[0].outer_fn, OuterFn::SumOverTime); assert_eq!( sum_by_rate.candidates[0].outer_fn, OuterFn::Rate, @@ -2662,16 +2846,18 @@ mod asap_tier_classify_tests { use crate::query_engines::query_result::QueryResult; let idx = Arc::new(SketchStore::new()); - // Four zones. Two windows each; per-zone sums chosen so the - // per-zone rate over 300s is a clean integer. + // Four zones. Two windows each spanning `[now-150s, now-30s]` = + // 120s of actual coverage inside the 300s `[5m]` lookback. + // Post-#301 the rate divisor is the actual coverage span + // (`min(300, 120) = 120`), not the nominal 300. let now_ms = std::time::SystemTime::now() .duration_since(std::time::SystemTime::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); - let w1_start = now_ms.saturating_sub(120_000); - let w1_end = now_ms.saturating_sub(60_000); + let w1_start = now_ms.saturating_sub(150_000); + let w1_end = now_ms.saturating_sub(90_000); let w2_start = w1_end; - let w2_end = now_ms.saturating_sub(1_000); + let w2_end = now_ms.saturating_sub(30_000); let zones = ["z0", "z1", "z2", "z3"]; for (i, zone) in zones.iter().enumerate() { @@ -2692,8 +2878,9 @@ mod asap_tier_classify_tests { expires_at_ms: None, policy_fp: asap_types::PolicyFingerprint::UNSET, }); - // per_window: 300, 600, 900, 1200 → per-zone rates over - // 300s are 2, 4, 6, 8. + // per_window: 300, 600, 900, 1200 → per-zone totals 600, + // 1200, 1800, 2400 → rates over the 120s coverage are + // 5, 10, 15, 20. let per_window = ((i + 1) * 300) as f64; for (ws, we) in [(w1_start, w1_end), (w2_start, w2_end)] { let mut lm = BTreeMap::new(); @@ -2729,7 +2916,7 @@ mod asap_tier_classify_tests { let zone_idx = keys.iter().position(|k| k == "zone").expect("zone key present"); by_zone.insert(vals[zone_idx].clone(), el.value); } - for (zone, expected) in [("z0", 2.0_f64), ("z1", 4.0), ("z2", 6.0), ("z3", 8.0)] { + for (zone, expected) in [("z0", 5.0_f64), ("z1", 10.0), ("z2", 15.0), ("z3", 20.0)] { let got = by_zone.get(zone).copied().unwrap_or(f64::NAN); assert!((got - expected).abs() < 1e-9, "{zone} expected {expected}, got {got}"); } @@ -2756,8 +2943,11 @@ mod asap_tier_classify_tests { .duration_since(std::time::SystemTime::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); - let w_start = now_ms.saturating_sub(60_000); - let w_end = now_ms.saturating_sub(1_000); + // One window per zone spanning `[now-150s, now-30s]` = 120s of + // actual coverage inside the 300s `[5m]` lookback → coverage-aware + // rate divisor is `min(300, 120) = 120` (#301). + let w_start = now_ms.saturating_sub(150_000); + let w_end = now_ms.saturating_sub(30_000); // Four zones with distinct per-window sums → distinct rates. let zones = ["z0", "z1", "z2", "z3"]; @@ -2820,8 +3010,8 @@ mod asap_tier_classify_tests { .expect("zone key present"); ordered.push((vals[zone_idx].clone(), el.value)); } - // Per-window sums 300,600,900,1200 / 300s = 1, 2, 3, 4 → topk - // descending = z3, z2, z1, z0. + // Per-window sums 300,600,900,1200 / 120s coverage = 2.5, 5, + // 7.5, 10 → topk descending = z3, z2, z1, z0. let labels_in_order: Vec<&str> = ordered.iter().map(|(z, _)| z.as_str()).collect(); assert_eq!( @@ -2829,8 +3019,8 @@ mod asap_tier_classify_tests { vec!["z3", "z2", "z1", "z0"], "topk emits zones in descending rate order: {ordered:?}" ); - assert!((ordered[0].1 - 4.0).abs() < 1e-9); - assert!((ordered[3].1 - 1.0).abs() < 1e-9); + assert!((ordered[0].1 - 10.0).abs() < 1e-9, "got {}", ordered[0].1); + assert!((ordered[3].1 - 2.5).abs() < 1e-9, "got {}", ordered[3].1); } /// `topk(2, sum by (zone) (rate(...)))` — same shape but K < n, diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 1e8fbda7..bf35daf1 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -498,6 +498,69 @@ impl SketchStore { .collect() } + /// Actual coverage bounds `(min_window_start_ms, max_window_end_ms)` + /// of the exact-agg windows this sid holds within `[start_unix_ms, + /// end_unix_ms]`. `None` when no in-range exact-agg window exists. + /// + /// Issue #301 (Layer 4): `evaluate_exact_agg_rate` must divide by the + /// ACTUAL data span — not the nominal `[r]` — when the producer has + /// run for less than the requested range. `query_exact_agg_range` + /// keys samples by `window_end` only, dropping `window_start`; this + /// companion preserves the full `(start, end)` so the rate reducer + /// can compute a coverage-aware divisor. Cheap (one epoch scan); the + /// rate reducer already walks the same windows. + pub fn exact_agg_coverage_bounds( + &self, + sid: u64, + start_unix_ms: u64, + end_unix_ms: u64, + ) -> Option<(u64, u64)> { + let store = self.series.get(&sid)?.clone(); + let guard = store.read().unwrap(); + let mut min_start: u64 = u64::MAX; + let mut max_end: u64 = 0; + let mut any = false; + + let mut buf: Vec<(TimestampRange, LabelValuesId, &AggPayload)> = Vec::new(); + guard + .current_epoch + .range_query_into(start_unix_ms, end_unix_ms, &mut buf); + for (win, _label_id, payload) in &buf { + if payload.as_exact_agg().is_some() { + any = true; + if win.0 < min_start { + min_start = win.0; + } + if win.1 > max_end { + max_end = win.1; + } + } + } + buf.clear(); + + for sealed in guard.sealed_epochs.values() { + sealed.range_query_into(start_unix_ms, end_unix_ms, &mut buf); + for (win, _label_id, payload) in &buf { + if payload.as_exact_agg().is_some() { + any = true; + if win.0 < min_start { + min_start = win.0; + } + if win.1 > max_end { + max_end = win.1; + } + } + } + buf.clear(); + } + + if any { + Some((min_start, max_end)) + } else { + None + } + } + /// Phase 5 M2.3.5 — query the precompute payloads across every sid /// belonging to one `AggregationConfig` (identified by `metric` + /// `agg_cfg.aggregation_type`), shaped as the legacy `Store` diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs index 8ad31332..c75f1e45 100644 --- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs +++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs @@ -589,6 +589,20 @@ impl<'a> SketchReducer<'a> { /// `group_by_keys` empty (i.e. `sum(metric)` without `by (...)`) /// collapses every series to a single grouping with empty label /// map — the natural PromQL semantics. + /// + /// `accumulate_windows` (issue #301) controls the per-group output + /// shape: + /// - `false` (range-query / matrix surface): emit ONE sample per + /// `(group, window_end)` — the per-window delta timeseries. The + /// range-query wire format wants a matrix with a point per window. + /// - `true` (instant `sum(counter)` / `increase(counter[r])`): SUM + /// every in-range window's value into ONE cumulative number per + /// group, timestamped at `t1_ms`. This is the PromQL-correct + /// semantic for instant counter sums (cumulative-since-storage) + /// and `increase` (Σ deltas in `[t-r,t]`). Without this the + /// instant path's `.last()` projection (engine + /// `asap_tier_result_to_query_result`) returned only the MOST + /// RECENT window's delta — the Layer-3 bug from #301. pub fn evaluate_exact_agg( &self, sids: &[u64], @@ -596,6 +610,7 @@ impl<'a> SketchReducer<'a> { group_by_keys: &std::collections::BTreeSet, t0_ms: u64, t1_ms: u64, + accumulate_windows: bool, ) -> Result { // Pick the Statistic answer this agg_type implies. PromQL // `sum by (...)` against an ExactAgg sid is the standard @@ -735,10 +750,28 @@ impl<'a> SketchReducer<'a> { // Build the series. BTreeMap iteration is already sorted, so // each series's samples vec is in window-end order. + // + // When `accumulate_windows` is set, collapse each group's + // per-window deltas into ONE cumulative sample (Σ of values), + // timestamped at `t1_ms`. This is the PromQL semantic for an + // instant counter `sum` (cumulative-since-storage) and for + // `increase(counter[r])` (Σ deltas in the `[t0,t1]` clip). + // Otherwise keep the per-window timeseries for the matrix + // (range-query) surface. + let sample_ts = if t1_ms <= i64::MAX as u64 { + t1_ms as i64 + } else { + i64::MAX + }; let mut out_series: Vec<(BTreeMap, Vec<(i64, f64)>)> = Vec::new(); for (group, samples) in by_group { let label_map: BTreeMap = group.into_iter().collect(); - out_series.push((label_map, samples)); + if accumulate_windows { + let total: f64 = samples.iter().map(|(_, v)| *v).sum(); + out_series.push((label_map, vec![(sample_ts, total)])); + } else { + out_series.push((label_map, samples)); + } } let coverage = if cov_lo <= cov_hi { @@ -763,7 +796,8 @@ impl<'a> SketchReducer<'a> { /// `[t-r, t]`" — for our sub-window-sized sids that's: /// /// ```text - /// rate(t) = (Σ over windows w ⊆ [t-r, t] of Sum[w]) / r_seconds + /// rate(t) = (Σ over windows w ⊆ [t-r, t] of Sum[w]) / divisor + /// divisor = min(r_seconds, actual_coverage_span_seconds) /// ``` /// /// Differs from `evaluate_exact_agg` in two ways: @@ -772,8 +806,16 @@ impl<'a> SketchReducer<'a> { /// samples. For an instant rate query that's the correct shape: /// one number per series, where the number is "events per second /// in the lookback". - /// 2. Divides the merged `Statistic::Sum` by `range_seconds` to - /// produce the rate (events/sec). `range_seconds == 0` would + /// 2. Divides the merged `Statistic::Sum` by `min(range_seconds, + /// coverage_span)` to produce the rate (events/sec). Issue #301 + /// Layer 4: dividing by the NOMINAL `range_seconds` (300 for + /// `[5m]`) when the producer has only run for a fraction of that + /// span systematically UNDER-reports the rate (the smoke test's + /// 64% rate rel-err). `coverage_span` = `(max_window_end − + /// min_window_start)/1000` across the contributing windows, read + /// from `SketchStore::exact_agg_coverage_bounds`. Clamped to + /// `range_seconds` so a query whose window genuinely spans the + /// full `[r]` still divides by `r`. `range_seconds == 0` would /// indicate a non-range query routing through this path by /// mistake — defensive, surface as `UnsupportedCapability` so /// the engine falls over rather than divide-by-zero. @@ -818,7 +860,39 @@ impl<'a> SketchReducer<'a> { capability: Capability::ExactAgg(agg_type), }); } - let divisor = range_seconds as f64; + + // Coverage-aware divisor (issue #301 Layer 4). The merged Sum is + // "events in `[t0,t1] ∩ stored windows`". Dividing by the + // NOMINAL `range_seconds` (e.g. 300 for `[5m]`) when the producer + // has only run for part of that span under-reports the rate. + // Use the ACTUAL covered span = (max_window_end − + // min_window_start)/1000 across all contributing sids, clamped to + // `[1, range_seconds]`. Clamping to `range_seconds` keeps a + // full-window query dividing by `r`; the lower bound of 1s guards + // against divide-by-zero when only a single sub-second window + // exists. When no bounds are available (no in-range exact-agg + // windows on any sid) the per-sid loop below produces NoData + // anyway, so the divisor fallback to `range_seconds` is moot. + let mut span_lo: u64 = u64::MAX; + let mut span_hi: u64 = 0; + for &sid in sids { + if let Some((start, end)) = + self.index.exact_agg_coverage_bounds(sid, t0_ms, t1_ms) + { + if start < span_lo { + span_lo = start; + } + if end > span_hi { + span_hi = end; + } + } + } + let coverage_seconds: u64 = if span_lo <= span_hi { + span_hi.saturating_sub(span_lo) / 1000 + } else { + range_seconds + }; + let divisor = range_seconds.min(coverage_seconds).max(1) as f64; // Choice of grouping mirrors `evaluate_exact_agg`: // * `group_by_keys` empty → preserve each sid's own full diff --git a/data_plane/src/storage_engines/sketch_db/query/tests.rs b/data_plane/src/storage_engines/sketch_db/query/tests.rs index 97420609..15fdf028 100644 --- a/data_plane/src/storage_engines/sketch_db/query/tests.rs +++ b/data_plane/src/storage_engines/sketch_db/query/tests.rs @@ -816,6 +816,7 @@ fn evaluate_exact_agg_sums_per_group_across_zones() { &group_by, 0, 400, + false, // per-window (matrix) shape — no accumulate ) .expect("exact-agg evaluate should succeed"); @@ -881,6 +882,7 @@ fn evaluate_exact_agg_collapses_subgroups_into_requested_groups() { &group_by, 0, 300, + false, // per-window (matrix) shape — no accumulate ) .expect("evaluate ok"); @@ -910,7 +912,7 @@ fn evaluate_exact_agg_unsupported_capability_for_minmax() { let reducer = SketchReducer::new(&idx); let group_by: BTreeSet = ["zone".to_string()].into_iter().collect(); let err = reducer - .evaluate_exact_agg(&[7000], AggregationType::MinMax, &group_by, 0, 1000) + .evaluate_exact_agg(&[7000], AggregationType::MinMax, &group_by, 0, 1000, false) .expect_err("MinMax dispatch should surface as UnsupportedCapability"); match err { ASAPTierError::UnsupportedCapability { capability, .. } => { @@ -938,7 +940,7 @@ fn evaluate_exact_agg_no_data_when_window_empty() { let reducer = SketchReducer::new(&idx); let group_by: BTreeSet = ["zone".to_string()].into_iter().collect(); let err = reducer - .evaluate_exact_agg(&[8000], AggregationType::Sum, &group_by, 0, 1000) + .evaluate_exact_agg(&[8000], AggregationType::Sum, &group_by, 0, 1000, false) .expect_err("empty in-window state should surface as NoData"); match err { ASAPTierError::NoData { metric_name } => { @@ -1002,10 +1004,14 @@ fn coverage_reports_observed_window_range() { #[test] fn evaluate_exact_agg_rate_divides_total_events_by_range() { - // One zone, two windows. Each window's SumAccumulator carries - // 600 events (a steady 10 req/sec over a 60s window). Over a - // 300s rate range we'd want (600 + 600) / 300 = 4 events/sec at - // instant readout — and ONLY one sample (not per-window). + // One zone, two windows that TOGETHER span the full 300s rate range + // (`[0, 300_000]`). Each window carries 600 events → (600 + 600) / + // 300 = 4 events/sec. Because the data coverage (300s) equals the + // nominal range, the coverage-aware divisor (issue #301 Layer 4) is + // `min(300, 300) = 300` — same as the nominal divisor — so this + // test pins both the fold-to-one-sample behavior AND the + // full-coverage divisor case. (The partial-coverage case is pinned + // separately in `evaluate_exact_agg_rate_divisor_uses_actual_coverage`.) use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; use crate::storage_engines::sketch_db::data::AggregationType; @@ -1021,13 +1027,13 @@ fn evaluate_exact_agg_rate_divides_total_events_by_range() { idx.append_precompute( 2100, lm.clone(), - (0, 60_000), + (0, 150_000), Box::new(SumAccumulator::with_sum(600.0)), ); idx.append_precompute( 2100, lm, - (60_000, 120_000), + (150_000, 300_000), Box::new(SumAccumulator::with_sum(600.0)), ); @@ -1040,7 +1046,7 @@ fn evaluate_exact_agg_rate_divides_total_events_by_range() { &group_by, 300, // range_seconds 0, - 120_000, + 300_000, ) .expect("rate evaluate ok"); assert_eq!(result.series.len(), 1, "one series for the lone zone"); @@ -1048,14 +1054,71 @@ fn evaluate_exact_agg_rate_divides_total_events_by_range() { assert_eq!(labels.get("zone").cloned(), Some("z0".to_string())); assert_eq!(samples.len(), 1, "rate emits ONE sample per group"); let (ts, value) = samples[0]; - assert_eq!(ts, 120_000, "sample timestamped at t1"); - // (600 + 600) / 300 = 4.0 + assert_eq!(ts, 300_000, "sample timestamped at t1"); + // (600 + 600) / min(300, 300) = 4.0 assert!( (value - 4.0).abs() < 1e-9, "expected 4.0 events/sec, got {value}" ); } +#[test] +fn evaluate_exact_agg_rate_divisor_uses_actual_coverage() { + // Issue #301 Layer 4: when the producer has only run for part of the + // requested `[r]` window, the rate divisor must be the ACTUAL covered + // span — not the nominal `range_seconds` — otherwise the rate is + // systematically under-reported (the smoke test's 64% rate rel-err). + // + // Data spans `[0, 120_000]` = 120s of the requested 300s `[5m]` + // window. Total events = 1200. With the OLD nominal divisor the rate + // would be 1200/300 = 4.0 (too low); the coverage-aware divisor is + // `min(300, 120) = 120`, giving the correct 1200/120 = 10.0. + use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; + use crate::storage_engines::sketch_db::data::AggregationType; + + let idx = SketchStore::new(); + idx.register(exact_agg_meta( + 2150, + "http_requests_total", + &["zone"], + AggregationType::Sum, + )); + let mut lm = BTreeMap::new(); + lm.insert("zone".to_string(), "z0".to_string()); + idx.append_precompute( + 2150, + lm.clone(), + (0, 60_000), + Box::new(SumAccumulator::with_sum(600.0)), + ); + idx.append_precompute( + 2150, + lm, + (60_000, 120_000), + Box::new(SumAccumulator::with_sum(600.0)), + ); + + let reducer = SketchReducer::new(&idx); + let group_by: BTreeSet = ["zone".to_string()].into_iter().collect(); + let result = reducer + .evaluate_exact_agg_rate( + &[2150], + AggregationType::Sum, + &group_by, + 300, // nominal [5m] range + 0, + 300_000, + ) + .expect("rate evaluate ok"); + let (_labels, samples) = &result.series[0]; + let value = samples[0].1; + assert!( + (value - 10.0).abs() < 1e-9, + "coverage-aware divisor: 1200 / min(300, 120) = 10.0, got {value} \ + (if ~4.0 the divisor regressed to the nominal range)" + ); +} + #[test] fn evaluate_exact_agg_rate_per_group_across_zones() { // Multinode-demo shape: 4 zones, each its own sid, two windows @@ -1080,7 +1143,9 @@ fn evaluate_exact_agg_rate_per_group_across_zones() { AggregationType::Sum, )); let per_window = ((i + 1) * 300) as f64; - for (ws, we) in [(0u64, 60_000u64), (60_000, 120_000)] { + // Windows span the full 300s range so coverage == nominal range + // and the coverage-aware divisor (#301) is `min(300, 300) = 300`. + for (ws, we) in [(0u64, 150_000u64), (150_000, 300_000)] { let mut lm = BTreeMap::new(); lm.insert("zone".to_string(), zone.to_string()); idx.append_precompute( @@ -1101,7 +1166,7 @@ fn evaluate_exact_agg_rate_per_group_across_zones() { &group_by, 300, 0, - 120_000, + 300_000, ) .expect("rate evaluate ok"); assert_eq!(result.series.len(), 4, "one series per zone"); @@ -1121,7 +1186,8 @@ fn evaluate_exact_agg_rate_per_group_across_zones() { fn evaluate_exact_agg_rate_collapses_subgroups_into_requested_groups() { // Two sids share (zone, rack); a `sum by (zone) (rate(...))` // collapses both racks' sub-window sums into one zone's rate. - // 100 + 200 = 300 over 100s = 3.0 events/sec. + // 100 + 200 = 300 over a window spanning the full 100s range + // (coverage-aware divisor min(100, 100) = 100) = 3.0 events/sec. use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; use crate::storage_engines::sketch_db::data::AggregationType; @@ -1136,7 +1202,12 @@ fn evaluate_exact_agg_rate_collapses_subgroups_into_requested_groups() { let mut lm = BTreeMap::new(); lm.insert("zone".to_string(), "z0".to_string()); lm.insert("rack".to_string(), rack.to_string()); - idx.append_precompute(sid, lm, (100, 200), Box::new(SumAccumulator::with_sum(value))); + idx.append_precompute( + sid, + lm, + (0, 100_000), + Box::new(SumAccumulator::with_sum(value)), + ); } let reducer = SketchReducer::new(&idx); @@ -1148,7 +1219,7 @@ fn evaluate_exact_agg_rate_collapses_subgroups_into_requested_groups() { &group_by, 100, 0, - 300, + 300_000, ) .expect("rate evaluate ok"); assert_eq!(result.series.len(), 1, "racks collapse into one zone group"); @@ -1177,13 +1248,14 @@ fn evaluate_exact_agg_rate_no_group_by_keeps_per_sid_series() { )); let mut lm = BTreeMap::new(); lm.insert("zone".to_string(), zone.to_string()); - idx.append_precompute(sid, lm, (0, 60_000), Box::new(SumAccumulator::with_sum(value))); + // Window spans the full 150s range so coverage == nominal range. + idx.append_precompute(sid, lm, (0, 150_000), Box::new(SumAccumulator::with_sum(value))); } let reducer = SketchReducer::new(&idx); let empty: BTreeSet = BTreeSet::new(); let result = reducer - .evaluate_exact_agg_rate(&[6100, 6101], AggregationType::Sum, &empty, 150, 0, 60_000) + .evaluate_exact_agg_rate(&[6100, 6101], AggregationType::Sum, &empty, 150, 0, 300_000) .expect("rate evaluate ok"); assert_eq!(result.series.len(), 2, "two distinct series preserved"); let mut by_zone: BTreeMap = BTreeMap::new();