fix: IncreaseAccumulator implements Statistic::Sum (sum-instant of counters now works)

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(<counter>)` 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

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 (...) (<counter>)`
+            // 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(<counter>)`.
+        // (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 (<counter>)` 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]

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(<counter>)` 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();

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(<counter>)` 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(|| "<missing>".to_string());
+                    by_zone.insert(zone, el.value);
+                }
+                // Per-zone Sum is the latest cumulative value of that
+                // series (Prometheus semantics for sum(<counter>)).
+                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:?}"),
+        }
+    }
 }