From 69c60bb80b73f5a6d426066d13aaa7366d56eb9a Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Fri, 8 May 2026 14:33:10 -0400 Subject: [PATCH] fix: IncreaseAccumulator implements Statistic::Sum (sum-instant of counters now works) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Counters are ingested by the warm tier as `IncreaseAccumulator` (and `MultipleIncreaseAccumulator` for keyed variants). Pre-fix neither trait `query` answered `Statistic::Sum`, so an instant `sum by (zone) (http_requests_total)` capability-missed even though the matcher and OnlySpatial dispatch path were correct (PR #108 regression test pinned the matcher contract; the actual data-flow gap is here). Fix: * `IncreaseAccumulator::query(Sum, ..)` returns `last_seen_measurement.value` — the latest cumulative counter value of that series, matching Prometheus' `sum()` instant semantics. * `MultipleIncreaseAccumulator::query` already delegates to the inner `IncreaseAccumulator`, so per-key Sum follows automatically. * `compatible_agg_types(Statistic::Sum)` now lists `Increase` / `MultipleIncrease` so capability matching accepts counter-shaped configs. * Updated the two unit tests that previously asserted Sum errors. * Added a new warm-tier regression test (`sum_by_zone_instant_over_increase_accumulator_does_not_error`) that builds a `SimpleEngine` with two `IncreaseAccumulator` series under different zone labels and asserts `sum by (zone) (http_requests_total)` returns `QueryResult::Vector` with the per-zone latest cumulative values. `rate([5m])` and `increase([5m])` are unaffected — those statistics resolve to `Statistic::Rate` / `Statistic::Increase` which already worked, and the changed match arm only adds a new case for `Sum`. Refs: ProjectASAP/ASAPCollector#46 Refs: PR #108 (warm-tier replay regression diagnosis) Co-Authored-By: Claude Opus 4.7 (1M context) --- .../rs/asap_types/src/capability_matching.rs | 11 +++ .../increase_accumulator.rs | 44 ++++++++- .../multiple_increase_accumulator.rs | 45 ++++++++- .../warm_engine_replay_regression_tests.rs | 92 +++++++++++++++++++ 4 files changed, 189 insertions(+), 3 deletions(-) diff --git a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs index d5e1d4ab..1e1617d6 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -131,6 +131,17 @@ pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] { AggregationType::Sum, AggregationType::MultipleSum, AggregationType::CountMinSketch, + // Counters: warm-tier ingest stores counter metrics + // (OTel `Sum` / monotonic=true) as Increase / + // MultipleIncrease accumulators, whose `query` + // implementation answers `Statistic::Sum` with the + // latest cumulative value per series — matching + // Prometheus' instant `sum()` semantics. + // Without these here, `sum by (zone) (http_requests_total)` + // capability-misses (issue ProjectASAP/ASAPCollector#46; + // diagnosis in PR #108). + AggregationType::Increase, + AggregationType::MultipleIncrease, ], // Count: exact via MultipleSum (the planner's canonical pick for // Count-Exact uses `MultipleSum` with sub_type="count"); approximate diff --git a/asap-query-engine/src/precompute_operators/increase_accumulator.rs b/asap-query-engine/src/precompute_operators/increase_accumulator.rs index 7d804e8f..a2a88a0d 100644 --- a/asap-query-engine/src/precompute_operators/increase_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/increase_accumulator.rs @@ -293,6 +293,18 @@ impl SingleSubpopulationAggregate for IncreaseAccumulator { let value_diff = self.last_seen_measurement.value - self.starting_measurement.value; Ok(value_diff / time_diff * 1000.0) } + // For instant `sum [by (...)] (counter_metric)` Prometheus + // sums the latest cumulative value of each matching series. + // The IncreaseAccumulator already tracks that latest value + // in `last_seen_measurement`, so per-series Sum is just + // that scalar; the engine's outer aggregation groups by the + // `by` labels and adds the per-series totals across keys. + // + // See PR #108 audit conclusion (commit 4359e10) and issue + // ProjectASAP/ASAPCollector#46: pre-fix the warm tier ingested + // counters as IncreaseAccumulator and bare `sum by (...) ()` + // capability-missed because this trait did not answer Sum. + Statistic::Sum => Ok(self.last_seen_measurement.value), _ => Err(format!("Unsupported statistic in IncreaseAccumulator: {statistic:?}").into()), } } @@ -404,7 +416,37 @@ mod tests { 7.5 ); // 15.0 / 2.0 - assert!(crate::SingleSubpopulationAggregate::query(&acc, Statistic::Sum, None).is_err()); + // Statistic::Sum returns the latest cumulative counter value, + // matching Prometheus semantics for instant `sum()`. + // (Issue ProjectASAP/ASAPCollector#46, PR #108 diagnosis.) + assert_eq!( + crate::SingleSubpopulationAggregate::query(&acc, Statistic::Sum, None).unwrap(), + 25.0 + ); + + // Unsupported statistics still error. + assert!(crate::SingleSubpopulationAggregate::query(&acc, Statistic::Min, None).is_err()); + } + + #[test] + fn test_increase_accumulator_sum_is_latest_cumulative_value() { + // Instant `sum ()` semantics: the per-series summand is + // the latest cumulative counter value. Two series with latest + // values 100 and 50 (started at 10 and 5 respectively) should + // each report Sum = 100 and Sum = 50 — the engine's `sum by` + // outer aggregation does the cross-series total. + let acc_a = + IncreaseAccumulator::new(Measurement::new(10.0), 1000, Measurement::new(100.0), 2000); + let acc_b = + IncreaseAccumulator::new(Measurement::new(5.0), 1000, Measurement::new(50.0), 2000); + assert_eq!( + crate::SingleSubpopulationAggregate::query(&acc_a, Statistic::Sum, None).unwrap(), + 100.0 + ); + assert_eq!( + crate::SingleSubpopulationAggregate::query(&acc_b, Statistic::Sum, None).unwrap(), + 50.0 + ); } #[test] diff --git a/asap-query-engine/src/precompute_operators/multiple_increase_accumulator.rs b/asap-query-engine/src/precompute_operators/multiple_increase_accumulator.rs index 0cb8abd0..adabf4ae 100644 --- a/asap-query-engine/src/precompute_operators/multiple_increase_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/multiple_increase_accumulator.rs @@ -432,13 +432,54 @@ mod tests { // Test rate query (15.0 increase over 1 second = 15.0 per second) assert_eq!(acc.query(Statistic::Rate, &key, None).unwrap(), 15.0); - // Test error cases - assert!(acc.query(Statistic::Sum, &key, None).is_err()); + // Sum returns the latest cumulative counter value for the + // queried key (per-series Prometheus `sum()` semantics; + // see issue ProjectASAP/ASAPCollector#46 and PR #108 diagnosis). + // The series here was created with last_seen=25.0. + assert_eq!(acc.query(Statistic::Sum, &key, None).unwrap(), 25.0); + + // Unsupported statistic still errors. + assert!(acc.query(Statistic::Min, &key, None).is_err()); let unknown_key = KeyByLabelValues::new(); assert!(acc.query(Statistic::Increase, &unknown_key, None).is_err()); } + #[test] + fn test_multiple_increase_accumulator_sum_per_key() { + // `sum by (zone) (counter)` reaches MultipleIncreaseAccumulator + // only when the warm-tier ingest groups multiple series under + // a single accumulator (the `Multiple*` variant). In that case + // each per-key Sum should be the series' latest cumulative + // value; the engine's outer `by` aggregation does the cross-key + // grouping. (Issue ProjectASAP/ASAPCollector#46.) + let mut acc = MultipleIncreaseAccumulator::new(); + let east = KeyByLabelValues::new_with_labels(vec!["us-east-1".to_string()]); + let west = KeyByLabelValues::new_with_labels(vec!["us-west-2".to_string()]); + + acc.update( + east.clone(), + IncreaseAccumulator::new( + Measurement::new(10.0), + 1000, + Measurement::new(100.0), + 2000, + ), + ); + acc.update( + west.clone(), + IncreaseAccumulator::new( + Measurement::new(5.0), + 1000, + Measurement::new(50.0), + 2000, + ), + ); + + assert_eq!(acc.query(Statistic::Sum, &east, None).unwrap(), 100.0); + assert_eq!(acc.query(Statistic::Sum, &west, None).unwrap(), 50.0); + } + #[test] fn test_multiple_increase_accumulator_merge() { let mut acc1 = MultipleIncreaseAccumulator::new(); diff --git a/asap-query-engine/src/tests/datafusion/warm_engine_replay_regression_tests.rs b/asap-query-engine/src/tests/datafusion/warm_engine_replay_regression_tests.rs index 0614183d..ce694fd6 100644 --- a/asap-query-engine/src/tests/datafusion/warm_engine_replay_regression_tests.rs +++ b/asap-query-engine/src/tests/datafusion/warm_engine_replay_regression_tests.rs @@ -36,6 +36,8 @@ mod tests { }; use crate::engines::simple::engine::SimpleEngine; use crate::engines::QueryResult; + use crate::data_model::Measurement; + use crate::precompute_operators::increase_accumulator::IncreaseAccumulator; use crate::precompute_operators::sum_accumulator::SumAccumulator; use crate::precompute_operators::DDSketchAccumulator; use crate::stores::sketch_db::simple_map_store::SimpleMapStore; @@ -311,4 +313,94 @@ mod tests { other => panic!("expected instant vector, got {other:?}"), } } + + // ------------------------------------------------------------------ + // (4) Instant `sum by (zone) (counter)` backed by IncreaseAccumulator + // + // This is the actual demo failure shape from + // ProjectASAP/ASAPCollector#46: the warm-tier ingest path + // stores OTel-`Sum`/monotonic counters as + // `IncreaseAccumulator`, not `SumAccumulator`. Pre-fix, this + // query class capability-missed because `IncreaseAccumulator` + // did not implement `Statistic::Sum`. Post-fix: + // + // a) `compatible_agg_types(Statistic::Sum)` lists + // `Increase` / `MultipleIncrease`, so capability matching + // accepts the counter-shaped configs. + // b) `IncreaseAccumulator::query(Sum, ..)` returns the + // latest cumulative value of the series, matching + // Prometheus' `sum()` instant semantics. + // c) The engine's outer `sum by (zone) (...)` aggregation + // groups + sums those per-series totals across keys. + // ------------------------------------------------------------------ + + #[test] + fn sum_by_zone_instant_over_increase_accumulator_does_not_error() { + init_tracing_for_test(); + let query = "sum by (zone) (http_requests_total)"; + + // Two zones, two cumulative-counter series. Each + // IncreaseAccumulator's `last_seen_measurement` is the latest + // cumulative value the series has reported. + let east = IncreaseAccumulator::new( + Measurement::new(10.0), + (WINDOW_END_MS - WINDOW_LEN_MS) as i64, + Measurement::new(123.0), + WINDOW_END_MS as i64, + ); + let west = IncreaseAccumulator::new( + Measurement::new(0.0), + (WINDOW_END_MS - WINDOW_LEN_MS) as i64, + Measurement::new(45.0), + WINDOW_END_MS as i64, + ); + + let engine = build_engine_with_window( + "http_requests_total", + AggregationType::Increase, + vec!["zone"], + vec![ + (Some(vec!["us-east-1".to_string()]), Box::new(east)), + (Some(vec!["us-west-2".to_string()]), Box::new(west)), + ], + query, + ); + + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME_SEC); + let (_labels, qr) = result.expect( + "warm engine must answer instant `sum by (zone) (counter)` against IncreaseAccumulator", + ); + + match qr { + QueryResult::Vector(iv) => { + assert_eq!( + iv.values.len(), + 2, + "expected 2 zones, got {} values", + iv.values.len() + ); + let mut by_zone = std::collections::HashMap::new(); + for el in &iv.values { + let zone = el + .labels + .labels + .first() + .cloned() + .unwrap_or_else(|| "".to_string()); + by_zone.insert(zone, el.value); + } + // Per-zone Sum is the latest cumulative value of that + // series (Prometheus semantics for sum()). + assert!( + (by_zone.get("us-east-1").copied().unwrap_or(f64::NAN) - 123.0).abs() < 1e-9, + "us-east-1 should be 123.0 (latest cumulative), by_zone={by_zone:?}" + ); + assert!( + (by_zone.get("us-west-2").copied().unwrap_or(f64::NAN) - 45.0).abs() < 1e-9, + "us-west-2 should be 45.0 (latest cumulative), by_zone={by_zone:?}" + ); + } + other => panic!("expected instant vector, got {other:?}"), + } + } }