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 cfb8be50..1332cdfe 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -410,23 +410,48 @@ impl ASAPQueryEngine { )); } - let result = reducer - .evaluate( - &hit_sids, - &candidate.function, - &candidate.function_args, - start_ms, - end_ms, - ) - .map_err(|e| { - crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), - format!( - "SketchStore reducer failed for `{query}` over \ - [{start_ms}, {end_ms}]: {e:?} — failing over to archive" - ), + // ExactAgg capability → dispatch the per-(group_by_keys) + // accumulator-merge path; sketch capabilities → the + // sketch-decode path. See `SketchReducer::evaluate_exact_agg` + // for the ExactAgg path's semantics. + let result = match &candidate.required_capability { + crate::storage_engines::sketch_db::index::Capability::ExactAgg(agg_type) => { + reducer + .evaluate_exact_agg( + &hit_sids, + *agg_type, + &candidate.group_by_keys, + start_ms, + end_ms, + ) + .map_err(|e| { + crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore exact-agg reducer failed for `{query}` over \ + [{start_ms}, {end_ms}]: {e:?} — failing over to archive" + ), + ) + })? + } + _ => reducer + .evaluate( + &hit_sids, + &candidate.function, + &candidate.function_args, + start_ms, + end_ms, ) - })?; + .map_err(|e| { + crate::query_engines::EngineError::capability_miss( + asap_types::StorageBackend::SketchStore.data_source_id(), + format!( + "SketchStore reducer failed for `{query}` over \ + [{start_ms}, {end_ms}]: {e:?} — failing over to archive" + ), + ) + })?, + }; combined_result = Some(result); } @@ -823,13 +848,33 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu combined_t0 = t0_ms; } - let result = match reducer.evaluate( - &hit_sids, - &candidate.function, - &candidate.function_args, - t0_ms, - now_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. + let reducer_result = match &candidate.required_capability { + crate::storage_engines::sketch_db::index::Capability::ExactAgg( + agg_type, + ) => reducer.evaluate_exact_agg( + &hit_sids, + *agg_type, + &candidate.group_by_keys, + t0_ms, + now_ms, + ), + _ => reducer.evaluate( + &hit_sids, + &candidate.function, + &candidate.function_args, + t0_ms, + now_ms, + ), + }; + + let result = match reducer_result { Ok(r) => r, Err( crate::storage_engines::sketch_db::query::ASAPTierError::UnsupportedFunction( @@ -1737,6 +1782,103 @@ mod asap_tier_classify_tests { ); } } + + /// `sum by (zone) (http_requests_total)` end-to-end via the + /// `execute(&str)` adapter. Mirrors the MVP smoke test's Axis-C + /// failure: the control plane's analyzer minted ExactAgg(Sum) + /// sids for `http_requests_total` (one per zone), the engine + /// resolved them via `instances_matching`, but the reducer + /// returned `UnsupportedFunction("sum")` because + /// `SketchReducer::evaluate` only knows sketch-backed query + /// families. This test pins the ExactAgg dispatch branch added + /// to `execute` so the new path emits per-zone instant-vector + /// results instead of a CapabilityMiss. + #[tokio::test] + async fn execute_sum_by_zone_dispatches_to_exact_agg_reducer() { + 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()); + // Mirror the smoke-test setup: four ExactAgg(Sum) sids, one + // per zone (z0..z3), registered with `group_by_keys=["zone"]` + // and carrying a `SumAccumulator` per window. + let zones = ["z0", "z1", "z2", "z3"]; + // Anchor windows so the engine's instant-query default + // lookback (5 min) reaches them. + 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 window_start = now_ms.saturating_sub(60_000); + let window_end = now_ms.saturating_sub(30_000); + + for (i, zone) in zones.iter().enumerate() { + let sid = 9000 + 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, + }); + let value = ((i + 1) * 100) as f64; + let mut lm = BTreeMap::new(); + lm.insert("zone".to_string(), zone.to_string()); + idx.append_precompute( + sid, + lm, + (window_start, window_end), + Box::new(SumAccumulator::with_sum(value)), + ); + } + + let engine = build_engine_with_index(idx); + let result = engine + .execute("sum by (zone) (http_requests_total)") + .await + .expect("sum by (zone) must dispatch to ExactAgg reducer, not capability-miss"); + + // Expect a Vector (instant) result with one entry per zone. + let vector = match result { + QueryResult::Vector(v) => v, + other => panic!("expected Vector, got {other:?}"), + }; + assert_eq!(vector.values.len(), 4, "one entry per zone"); + // Per-zone values match what each SumAccumulator carries. + // KeyByLabelValues stores values only; the override carries + // the corresponding keys. + let mut by_zone: std::collections::HashMap = + std::collections::HashMap::new(); + for el in &vector.values { + // The element's label keys override + label values together + // identify the zone. + let keys = el + .label_keys_override + .as_ref() + .expect("override populated for ExactAgg path"); + let vals = &el.labels.labels; + assert_eq!(keys.len(), vals.len()); + let zone_idx = keys + .iter() + .position(|k| k == "zone") + .expect("zone key present"); + by_zone.insert(vals[zone_idx].clone(), el.value); + } + assert_eq!(by_zone.get("z0").copied(), Some(100.0)); + assert_eq!(by_zone.get("z1").copied(), Some(200.0)); + assert_eq!(by_zone.get("z2").copied(), Some(300.0)); + assert_eq!(by_zone.get("z3").copied(), Some(400.0)); + } } // =========================================================================== 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 c61be423..1e8fbda7 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -424,6 +424,80 @@ impl SketchStore { .collect() } + /// Range-query the ExactAgg state for ONE sid. Sister of + /// [`Self::query_range`] for the exact-aggregation branch — same + /// `[start, end]` semantics, but yields `Box` + /// payloads keyed by their FULL `BTreeMap` label + /// map (label KEYS preserved, not just values). + /// + /// Used by the ASAP-tier `sum by (...)` dispatch path + /// (`SketchReducer::evaluate_exact_agg`) so the engine can + /// project label maps onto a query-time `group_by_keys` subset + /// (`{zone: z0, rack: r0}` → grouped by `zone` only). The + /// existing [`Self::query_precomputes_by_agg`] flattens labels + /// to a `KeyByLabelValues` (values only, no keys), which loses + /// the projection information the engine needs. + /// + /// Returns an empty Vec when the sid carries no ExactAgg state + /// in `[start, end]` (or is sketch-backed). Defensive — caller + /// is responsible for confirming the sid's `agg_kind` is + /// `AggKind::ExactAgg { .. }` before calling. + pub fn query_exact_agg_range( + &self, + sid: u64, + start_unix_ms: u64, + end_unix_ms: u64, + ) -> Vec<( + BTreeMap, + BTreeMap>, + )> { + let store = match self.series.get(&sid) { + Some(s) => s.clone(), + None => return Vec::new(), + }; + let guard = store.write().unwrap(); + let mut by_label_id: HashMap< + LabelValuesId, + BTreeMap>, + > = HashMap::new(); + + 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 let Some(p) = payload.as_exact_agg() { + by_label_id + .entry(*label_id) + .or_default() + .insert(win.1 as i64, Arc::from(p.clone_boxed_core())); + } + } + 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 let Some(p) = payload.as_exact_agg() { + by_label_id + .entry(*label_id) + .or_default() + .insert(win.1 as i64, Arc::from(p.clone_boxed_core())); + } + } + buf.clear(); + } + + by_label_id + .into_iter() + .map(|(label_id, samples)| { + let label_values_map = + guard.intern.resolve(label_id).cloned().unwrap_or_default(); + (label_values_map, samples) + }) + .collect() + } + /// 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 fad42d3c..08a0ca0b 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 @@ -50,6 +50,7 @@ //! as `UnsupportedCapability` for now and document the gap. use std::collections::BTreeMap; +use std::sync::Arc; use asap_sketchlib::sketches::countminsketch::CountMinSketch; use asap_sketchlib::sketches::countsketch::CountSketch; @@ -65,9 +66,10 @@ use crate::storage_engines::sketch_db::query::delta_apply::{ cumulative_evaluate, per_window_evaluate, DeltaSketchKind, }; use crate::storage_engines::sketch_db::index::{ - Capability, SketchEncoding, SketchStore, SketchInstanceMetadata, SketchKindHandle, - SketchSampleState, + AggregationType, Capability, SketchEncoding, SketchStore, SketchInstanceMetadata, + SketchKindHandle, SketchSampleState, }; +use promql_utilities::query_logics::enums::Statistic; /// Reducer wrapping a `&SketchStore`. Constructed per-query; cheap. pub struct SketchReducer<'a> { @@ -557,6 +559,199 @@ impl<'a> SketchReducer<'a> { }) } + /// ExactAgg dispatch — sister of [`Self::evaluate`] for sids whose + /// `Capability` is `ExactAgg(_)`. The sketch-backed `evaluate` + /// path can't answer these because they carry `Box` payloads (per-window `SumAccumulator` / + /// `IncreaseAccumulator` / `MinMaxAccumulator` etc.) rather than + /// opaque sketch bytes. + /// + /// PromQL `sum by (group_by_keys) (metric)` lowers (via the + /// control plane's `analyze_promql_for_asap_tier`) to a candidate + /// with `required_capability = ExactAgg(Sum)` and + /// `group_by_keys = {requested labels}`. This method walks every + /// hit sid's exact-aggregation state, projects each window's + /// label map onto `group_by_keys` (so a sid registered with + /// `[zone, rack]` answering a `by (zone)` query collapses across + /// rack values), and emits one `(label_values, [(window_end, + /// scalar)])` series per distinct projected group. + /// + /// For each (group, window_end) pair we MERGE all matching + /// accumulators via `AggregateCore::merge_with` and then read the + /// `Statistic` the agg_type implies — `Sum`/`Increase` → + /// `Statistic::Sum`, `MinMax` → currently UnsupportedCapability + /// (min vs max disambiguation needs the outer function name; deferred + /// to a follow-up). Both `SumAccumulator` and `IncreaseAccumulator` + /// answer `Statistic::Sum` from their `query_statistic` (the latter + /// returns the accumulated increase, which is what a PromQL `sum` + /// over rate/increase wants). + /// + /// `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. + pub fn evaluate_exact_agg( + &self, + sids: &[u64], + agg_type: AggregationType, + group_by_keys: &std::collections::BTreeSet, + t0_ms: u64, + t1_ms: u64, + ) -> Result { + // Pick the Statistic answer this agg_type implies. PromQL + // `sum by (...)` against an ExactAgg sid is the standard + // counter rollup — every additive type answers via Sum. + let stat = match agg_type { + AggregationType::Sum + | AggregationType::MultipleSum + | AggregationType::Increase + | AggregationType::MultipleIncrease => Statistic::Sum, + // MinMax disambiguation requires the outer PromQL function + // name (min vs max); deferred until the engine threads it + // through. Today min/max queries are not produced by the + // analyzer's ExactAgg(MinMax) capability path for `sum by` + // queries, so this branch is defensive. + other => { + return Err(ASAPTierError::UnsupportedCapability { + function: format!("sum_by_for_{other:?}"), + capability: Capability::ExactAgg(other), + }); + } + }; + + // (projected_group_map, window_end_ms) -> Vec + // BTreeMap so window_ends sort naturally for output and the + // group map key is a Vec<(k,v)> tuple sorted by key (BTreeMap + // iteration is key-sorted, so collecting yields a canonical + // order). + type GroupKey = Vec<(String, String)>; + let mut grouped: BTreeMap< + (GroupKey, i64), + Vec>, + > = BTreeMap::new(); + let mut metric_name_for_err = String::new(); + let mut cov_lo: u64 = u64::MAX; + let mut cov_hi: u64 = 0; + let mut any_window = false; + + for &sid in sids { + let meta = match self.index.instance(sid) { + Some(m) => m, + None => continue, + }; + metric_name_for_err = meta.metric_name.clone(); + + // Pull every (label_map, samples) tuple this sid carries + // in window. Sketch-backed sids (or sids with no in-window + // exact-agg state) return empty. + let series_list = self.index.query_exact_agg_range(sid, t0_ms, t1_ms); + for (label_map, samples) in series_list { + // Project label_map onto group_by_keys. Missing keys are + // dropped (the user didn't ask for them); requested keys + // absent from the sid's label_map become empty-string + // values so a sid registered with a subset of the + // requested keys still groups deterministically. + let projected: GroupKey = if group_by_keys.is_empty() { + Vec::new() + } else { + group_by_keys + .iter() + .map(|k| { + let v = label_map.get(k).cloned().unwrap_or_default(); + (k.clone(), v) + }) + .collect() + }; + + for (window_end, acc) in samples { + any_window = true; + let w = if window_end >= 0 { window_end as u64 } else { 0 }; + if w < cov_lo { + cov_lo = w; + } + if w > cov_hi { + cov_hi = w; + } + grouped + .entry((projected.clone(), window_end)) + .or_default() + .push(acc); + } + } + } + + if !any_window { + return Err(ASAPTierError::NoData { + metric_name: metric_name_for_err, + }); + } + + // Fold per-(group, window) accumulator lists into a single + // scalar via `merge_with` (additive across the list) and + // `query_statistic`. Re-bucket by group so each group emits + // ONE series with the full per-window timeseries. + let mut by_group: BTreeMap> = BTreeMap::new(); + for ((group, w_end), accs) in grouped { + // Merge all accumulators landing in (group, window). For + // a single ExactAgg sid covering one group there's + // typically one entry; multiple entries come from multiple + // sids that share the projected group (e.g. several + // (zone=z0, rack=*) sids collapsing to a single zone=z0 + // group). + let mut iter = accs.into_iter(); + let head = match iter.next() { + Some(h) => h, + None => continue, + }; + let mut merged: Box = + head.clone_boxed_core(); + for next in iter { + match merged.merge_with(next.as_ref()) { + Ok(m) => merged = m, + Err(e) => { + return Err(ASAPTierError::DeserializeFailure { + sid: 0, + encoding: SketchEncoding::ProtoFull, + reason: format!("exact-agg merge failed: {e}"), + }); + } + } + } + let value = match merged.query_statistic( + stat, + &None, + &std::collections::HashMap::new(), + ) { + Ok(v) => v, + Err(e) => { + return Err(ASAPTierError::DeserializeFailure { + sid: 0, + encoding: SketchEncoding::ProtoFull, + reason: format!("exact-agg query_statistic({stat:?}) failed: {e}"), + }); + } + }; + by_group.entry(group).or_default().push((w_end, value)); + } + + // Build the series. BTreeMap iteration is already sorted, so + // each series's samples vec is in window-end order. + 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)); + } + + let coverage = if cov_lo <= cov_hi { + Some((cov_lo, cov_hi)) + } else { + None + }; + Ok(ASAPTierResult { + series: out_series, + coverage, + }) + } + /// Decode one window's sketch state and run the family-appropriate /// reduction. /// 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 c31f813e..0b1a3831 100644 --- a/data_plane/src/storage_engines/sketch_db/query/tests.rs +++ b/data_plane/src/storage_engines/sketch_db/query/tests.rs @@ -735,6 +735,219 @@ fn hll_cumulative_full_plus_one_delta() { // information it needs. // --------------------------------------------------------------------------- +// --------------------------------------------------------------------------- +// ExactAgg dispatch — regression coverage for `sum by (...)` PromQL. +// Pins that `SketchReducer::evaluate_exact_agg`: +// 1. Walks ExactAgg sids (not sketch sids). +// 2. Groups per-window AggregateCore state by the projected +// `group_by_keys` (subset of each sid's full label map). +// 3. Merges accumulators inside a group via `AggregateCore::merge_with` +// and reads `Statistic::Sum` for additive types. +// 4. Surfaces `NoData` when no in-window state exists (so the engine +// routes the query to archive instead of returning a stale answer). +// --------------------------------------------------------------------------- + +fn exact_agg_meta( + sid: u64, + metric: &str, + group_by_keys: &[&str], + agg_type: crate::storage_engines::sketch_db::data::AggregationType, +) -> SketchInstanceMetadata { + SketchInstanceMetadata { + sid, + metric_name: metric.to_string(), + group_by_keys: group_by_keys.iter().map(|s| s.to_string()).collect(), + capability: Some(Capability::ExactAgg(agg_type)), + agg_kind: AggKind::ExactAgg { + agg_type, + 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, + } +} + +#[test] +fn evaluate_exact_agg_sums_per_group_across_zones() { + use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; + use crate::storage_engines::sketch_db::data::AggregationType; + + let idx = SketchStore::new(); + // Four sids, one per zone, mirroring the post-#290 startup-replan + // ExactAgg(Sum) sids the smoke test exercises. + let zones = ["z0", "z1", "z2", "z3"]; + for (i, zone) in zones.iter().enumerate() { + let sid = 1000 + i as u64; + idx.register(exact_agg_meta( + sid, + "http_requests_total", + &["zone"], + AggregationType::Sum, + )); + // Two windows of data per zone, the sum value distinct per zone + // (10, 20, 30, 40) so the test can assert per-group correctness. + let value = ((i + 1) * 10) as f64; + for (j, (ws, we)) in [(100u64, 200u64), (200, 300)].iter().enumerate() { + let mut lm = BTreeMap::new(); + lm.insert("zone".to_string(), zone.to_string()); + // The second window's accumulator carries the same value so + // the per-window per-zone scalar is constant; the engine + // chooses the last window for instant queries. + let _ = j; + idx.append_precompute( + sid, + lm, + (*ws, *we), + Box::new(SumAccumulator::with_sum(value)), + ); + } + } + + let reducer = SketchReducer::new(&idx); + let group_by: BTreeSet = ["zone".to_string()].into_iter().collect(); + let result = reducer + .evaluate_exact_agg( + &[1000, 1001, 1002, 1003], + AggregationType::Sum, + &group_by, + 0, + 400, + ) + .expect("exact-agg evaluate should succeed"); + + // One series per zone, each with two windows of samples. + assert_eq!(result.series.len(), 4, "one series per zone"); + let mut per_zone: BTreeMap = BTreeMap::new(); + for (label_map, samples) in &result.series { + let zone = label_map + .get("zone") + .cloned() + .expect("series carries `zone` label"); + // Each window emits one sample; both windows for one zone + // share the same value so the last sample is the canonical + // instant readout. + let last = samples.last().expect("at least one sample").1; + per_zone.insert(zone, last); + } + assert_eq!(per_zone.get("z0").copied(), Some(10.0)); + assert_eq!(per_zone.get("z1").copied(), Some(20.0)); + assert_eq!(per_zone.get("z2").copied(), Some(30.0)); + assert_eq!(per_zone.get("z3").copied(), Some(40.0)); + + // Coverage spans the entire window range. + let (cov_lo, cov_hi) = result.coverage.expect("coverage populated"); + assert_eq!(cov_lo, 200); + assert_eq!(cov_hi, 300); +} + +#[test] +fn evaluate_exact_agg_collapses_subgroups_into_requested_groups() { + // Two sids share a (zone, rack) label space: sid 5000 is + // (zone=z0, rack=r0), sid 5001 is (zone=z0, rack=r1). A + // `sum by (zone)` query MUST collapse both racks into one + // (zone=z0) group with their values added. + use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; + use crate::storage_engines::sketch_db::data::AggregationType; + + let idx = SketchStore::new(); + for (sid, rack, value) in [(5000u64, "r0", 7.0_f64), (5001, "r1", 13.0)] { + idx.register(exact_agg_meta( + sid, + "http_requests_total", + &["rack", "zone"], + AggregationType::Sum, + )); + 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)), + ); + } + + let reducer = SketchReducer::new(&idx); + let group_by: BTreeSet = ["zone".to_string()].into_iter().collect(); + let result = reducer + .evaluate_exact_agg( + &[5000, 5001], + AggregationType::Sum, + &group_by, + 0, + 300, + ) + .expect("evaluate ok"); + + assert_eq!(result.series.len(), 1, "rack values collapse into one zone group"); + let (label_map, samples) = &result.series[0]; + assert_eq!(label_map.get("zone").cloned(), Some("z0".to_string())); + assert!(!label_map.contains_key("rack"), "rack dropped (not in group_by)"); + let last = samples.last().expect("at least one sample").1; + assert!( + (last - 20.0).abs() < 1e-9, + "merged sum 7 + 13 = 20, got {last}" + ); +} + +#[test] +fn evaluate_exact_agg_unsupported_capability_for_minmax() { + use crate::storage_engines::sketch_db::data::AggregationType; + + let idx = SketchStore::new(); + idx.register(exact_agg_meta( + 7000, + "http_requests_total", + &["zone"], + AggregationType::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) + .expect_err("MinMax dispatch should surface as UnsupportedCapability"); + match err { + ASAPTierError::UnsupportedCapability { capability, .. } => { + assert!(matches!( + capability, + Capability::ExactAgg(AggregationType::MinMax) + )); + } + other => panic!("expected UnsupportedCapability, got {other:?}"), + } +} + +#[test] +fn evaluate_exact_agg_no_data_when_window_empty() { + use crate::storage_engines::sketch_db::data::AggregationType; + + let idx = SketchStore::new(); + idx.register(exact_agg_meta( + 8000, + "http_requests_total", + &["zone"], + AggregationType::Sum, + )); + + 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) + .expect_err("empty in-window state should surface as NoData"); + match err { + ASAPTierError::NoData { metric_name } => { + assert_eq!(metric_name, "http_requests_total"); + } + other => panic!("expected NoData, got {other:?}"), + } +} + #[test] fn coverage_reports_observed_window_range() { let idx = SketchStore::new();