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 1e1617d63..30c170732 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -146,10 +146,20 @@ pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] { // Count: exact via MultipleSum (the planner's canonical pick for // Count-Exact uses `MultipleSum` with sub_type="count"); approximate // via CountMinSketch / CountMinSketchWithHeap. + // + // HLL is also valid here: the warm-tier MVP demo + // (ProjectASAP/ASAPCollector#46) plans `unique_users_per_min` + // as an HLL agg and the replay client queries it with + // `count(unique_users_per_min)`. `HllSketchAccumulator` + // answers `Statistic::Count` as a cardinality alias — + // see `precompute_operators/hll_sketch_accumulator.rs:220`. + // Without HLL listed here the warm engine returns `status=error` + // for every count-of-HLL replay row. Statistic::Count => &[ AggregationType::MultipleSum, AggregationType::CountMinSketch, AggregationType::CountMinSketchWithHeap, + AggregationType::HLL, ], Statistic::Min | Statistic::Max => { &[AggregationType::MinMax, AggregationType::MultipleMinMax] @@ -190,7 +200,19 @@ pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] { AggregationType::DeltaSetAggregator, AggregationType::HLL, ], - Statistic::Topk => &[AggregationType::CountMinSketchWithHeap], + // Topk: `CountMinSketchWithHeap` is the canonical CMS-Heap + // pattern. CountSketch is the second-tier reservoir-style + // approximator the MVP demo's controller plans for + // `top_endpoint_qps` (median-of-row estimator over a + // signed-counter matrix). `CountSketchAccumulator` answers + // `Statistic::Topk` directly — see + // `precompute_operators/count_sketch_accumulator.rs:284`. + // Without CountSketch listed here, `topk(K, top_endpoint_qps)` + // capability-misses and the warm engine returns `status=error`. + Statistic::Topk => &[ + AggregationType::CountMinSketchWithHeap, + AggregationType::CountSketch, + ], } } @@ -295,11 +317,29 @@ pub fn window_compatible(config: &AggregationConfig, data_range_ms: Option) } } -/// Label compatibility: strict exact match. -/// TODO: relax to superset (config.grouping_labels ⊇ req.grouping_labels) for -/// simple accumulators (Sum, MinMax, Increase). +/// Label compatibility: config can serve a query whose grouping is a +/// **subset** (including equality) of the config's grouping_labels. +/// +/// Pre-fix this was strict-exact: `config_labels == req_labels`. The +/// MVP demo (ProjectASAP/ASAPCollector#46) replays +/// `count(unique_users_per_min)` / `topk(5, top_endpoint_qps)` with +/// no `by (...)` modifier, which translates to `req.grouping_labels = +/// []`. The corresponding agg configs are per-zone (`[zone]` grouping). +/// Pre-fix every such replay row capability-missed and the warm engine +/// returned `status=error`. Post-fix the engine accepts the agg, runs +/// the per-zone accumulators through the merge path +/// (`execute_and_merge_store_queries` produces a per-key map; the +/// downstream merge collapses them to the requested `[]` grouping — +/// HLL/CMS/CountSketch all support natural across-key merge, and +/// scalar accumulators like Sum / Increase reduce by addition). +/// +/// Direction is asymmetric: `config ⊇ req` is OK (engine merges away +/// the extra labels), but `req ⊃ config` is NOT — the engine cannot +/// invent a label that the materialised agg never partitioned by. pub fn labels_compatible(config_labels: &KeyByLabelNames, req_labels: &KeyByLabelNames) -> bool { - config_labels == req_labels + let req: std::collections::HashSet<&String> = req_labels.labels.iter().collect(); + let cfg: std::collections::HashSet<&String> = config_labels.labels.iter().collect(); + req.is_subset(&cfg) } /// Spatial filter compatibility. @@ -725,8 +765,17 @@ mod tests { } #[test] - fn label_strict_superset_rejected() { - // Config has {job, instance}, query wants only {job} — strict mode rejects + fn label_superset_config_accepts_subset_query() { + // Config has `{job, instance}`, query wants only `{job}`. + // + // Pre-fix `labels_compatible` did strict-eq and rejected this, + // which broke the MVP demo (ProjectASAP/ASAPCollector#46): the + // agent's per-zone HLL agg has `grouping_labels = [zone]`, the + // replay client's `count(unique_users_per_min)` has no `by` + // modifier (req grouping = `[]`). Post-fix the agg can serve + // the broader-aggregation query — the engine's merge path + // collapses the extra label dimension before the result + // surface. See `labels_compatible` rustdoc. let configs = single_config(make_config( 1, "cpu", @@ -741,6 +790,38 @@ mod tests { &configs, &req("cpu", &[Statistic::Sum], Some(300_000), &["job"], ""), ); + assert!( + result.is_some(), + "post-fix: a config with `[job, instance]` grouping must serve a `[job]`-only req \ + via the merge path", + ); + } + + #[test] + fn label_subset_config_rejects_superset_query() { + // Config has only `[job]`, query wants `[job, instance]`. + // The engine cannot invent a partition the agg never + // materialised, so this remains incompatible. + let configs = single_config(make_config( + 1, + "cpu", + "Sum", + "", + 300, + "tumbling", + &["job"], + "", + )); + let result = find_compatible_aggregation( + &configs, + &req( + "cpu", + &[Statistic::Sum], + Some(300_000), + &["job", "instance"], + "", + ), + ); assert!(result.is_none()); } @@ -1113,6 +1194,29 @@ mod tests { .contains(&AggregationType::CountMinSketchWithHeap), "CountMinSketchWithHeap must be a compatible type for Topk", ); + // CountSketch → Topk (warm-engine-error-on-replay-queries fix). + // Required so the MVP demo's `topk(5, top_endpoint_qps)` — + // which routes through the agent's `countsketchprocessor` + // and lands as a CountSketch-only config — resolves + // through capability matching. Without this, the warm + // engine returned `status=error` for every topk replay row. + assert!( + compatible_agg_types(Statistic::Topk).contains(&AggregationType::CountSketch), + "CountSketch must be a compatible type for Topk (warm-engine-error fix)", + ); + // HLL → Count (warm-engine-error-on-replay-queries fix). The + // MVP demo's `count(unique_users_per_min)` is structurally a + // PromQL `Statistic::Count` (the AggregationOperator::Count + // → Statistic::Count mapping in + // `promql_utilities::query_logics::enums`); the + // `HllSketchAccumulator` answers it as a cardinality alias + // (`hll_sketch_accumulator.rs:220`). Without HLL listed + // here, capability matching missed and the warm engine + // returned `status=error` for every count-of-HLL replay row. + assert!( + compatible_agg_types(Statistic::Count).contains(&AggregationType::HLL), + "HLL must be a compatible type for Count (warm-engine-error fix)", + ); } /// Phase-3.1 regression test for the canonical MVP-demo failure diff --git a/asap-query-engine/src/engines/simple/engine.rs b/asap-query-engine/src/engines/simple/engine.rs index 85bbd8a82..48bd7293a 100644 --- a/asap-query-engine/src/engines/simple/engine.rs +++ b/asap-query-engine/src/engines/simple/engine.rs @@ -495,6 +495,78 @@ impl SimpleEngine { result } + /// Resolve the canonical "all labels" set for a metric, with a + /// streaming-config fallback for schema-empty deploys. + /// + /// The user-facing `inference_config.schema` is the source of truth + /// for "what labels does this metric carry" — but the production + /// warm-tier deploy launches with `--streaming-config` only and no + /// `--config`, so the schema is empty. Pre-fix every query lookup + /// in `build_promql_execution_context_tail` and + /// `build_query_requirements_promql` returned `None` / + /// `KeyByLabelNames::empty()` for that metric, killing capability + /// matching (`req.grouping_labels = []` strict-mismatches every + /// agg config's `[zone]`) and the downstream context build (the + /// `None` short-circuits the whole query). See + /// `tests::warm_engine_replay_regression_tests::production_conditions_*`. + /// + /// Fallback rules: + /// 1. Look up the metric in `inference_config.schema`. Return its + /// labels if present. + /// 2. Otherwise scan the current `StreamingConfig` snapshot for + /// every agg config whose `metric == name`. Union their + /// `grouping_labels` (the per-series partition the agg + /// materialises) and return that. The union preserves order + /// of first-appearance and de-dupes — `KeyByLabelNames` + /// equality is strict, so we have to keep insertion order + /// deterministic across config swaps. + /// 3. Returns `None` only if no agg config references the metric + /// AND the schema is empty. Callers translate that into the + /// same "metric unknown" outcome as before this helper landed. + fn resolve_metric_labels(&self, metric: &str) -> Option { + // (1) schema lookup — user-supplied source of truth. + if let SchemaConfig::PromQL(schema) = &self.inference_config.schema { + if let Some(labels) = schema.get_labels(metric).cloned() { + return Some(labels); + } + } + + // (2) streaming-config fallback — derived from whatever agg + // configs the controller / static YAML registered for the + // metric. Produces the union of `grouping_labels` across all + // matching aggs in deterministic insertion order. + let snap = self.streaming_config_snapshot(); + let mut seen = std::collections::HashSet::new(); + let mut union: Vec = Vec::new(); + // Sort by aggregation_id so the resulting label vector is + // stable across re-runs even though `aggregation_configs` is + // a `HashMap`. Without this ordering, two engines holding + // bit-identical configs could produce different + // `KeyByLabelNames` instances and `labels_compatible`'s + // strict-eq would flake intermittently. + let mut agg_ids: Vec = snap.aggregation_configs.keys().copied().collect(); + agg_ids.sort_unstable(); + for id in agg_ids { + let cfg = match snap.get_aggregation_config(id) { + Some(c) => c, + None => continue, + }; + if cfg.metric != metric { + continue; + } + for label in &cfg.grouping_labels.labels { + if seen.insert(label.clone()) { + union.push(label.clone()); + } + } + } + if union.is_empty() { + None + } else { + Some(KeyByLabelNames::new(union)) + } + } + /// Convert query timestamp (seconds) to data timestamp (milliseconds) pub fn convert_query_time_to_data_time(query_time: f64) -> u64 { (query_time * 1000.0) as u64 @@ -1502,11 +1574,15 @@ impl SimpleEngine { ) -> Option { let (metric, spatial_filter) = get_metric_and_spatial_filter(match_result); - let promql_schema = match &self.inference_config.schema { - SchemaConfig::PromQL(schema) => schema, - _ => return None, - }; - let all_labels = match promql_schema.get_labels(&metric).cloned() { + // Resolve the metric's "all labels" set. Falls back to a + // streaming-config-derived label union when the schema is + // empty for this metric — the production warm-tier deploy + // launches with `--streaming-config` only and an empty + // schema, and pre-fix every query for a streaming-config- + // registered metric blew up here on the schema lookup. See + // `Self::resolve_metric_labels` and the + // `production_conditions_*` regression tests for context. + let all_labels = match self.resolve_metric_labels(&metric) { Some(labels) => labels, None => { warn!("No metric configuration found for '{}'", metric); @@ -1952,13 +2028,16 @@ impl SimpleEngine { .map(|d| d.num_seconds() as u64 * 1000), }; - let all_labels = match &self.inference_config.schema { - SchemaConfig::PromQL(schema) => schema - .get_labels(&metric) - .cloned() - .unwrap_or_else(KeyByLabelNames::empty), - _ => KeyByLabelNames::empty(), - }; + // Resolve the metric's "all labels" set with the same + // schema-empty fallback used by + // `build_promql_execution_context_tail`. Without this + // fallback the schema-empty production deploy returns + // `KeyByLabelNames::empty()` for every metric, and + // `labels_compatible`'s strict-eq mismatches every agg + // config's `[zone]` → capability-miss → `status=error`. + let all_labels = self + .resolve_metric_labels(&metric) + .unwrap_or_else(KeyByLabelNames::empty); let grouping_labels = match query_pattern_type { QueryPatternType::OnlyTemporal => all_labels, 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 2683cae4e..5e20dab01 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 @@ -37,6 +37,9 @@ mod tests { use crate::engines::simple::engine::SimpleEngine; use crate::engines::QueryResult; use crate::data_model::Measurement; + use crate::precompute_operators::count_min_sketch_accumulator::CountMinSketchAccumulator; + use crate::precompute_operators::count_sketch_accumulator::CountSketchAccumulator; + use crate::precompute_operators::hll_sketch_accumulator::HllSketchAccumulator; use crate::precompute_operators::increase_accumulator::IncreaseAccumulator; use crate::precompute_operators::sum_accumulator::SumAccumulator; use crate::precompute_operators::DDSketchAccumulator; @@ -44,6 +47,7 @@ mod tests { use crate::stores::Store; use crate::AggregateCore; use asap_sketchlib::sketches::ddsketch::DdSketch; + use asap_sketchlib::sketches::{CountMinSketch, CountSketch, HllSketch}; use promql_utilities::data_model::KeyByLabelNames; use std::collections::HashMap; use std::sync::Arc; @@ -498,6 +502,328 @@ mod tests { ); } + // ------------------------------------------------------------------ + // (5) **Production-conditions** suite — schema-empty deploy. + // + // The deployed warm-tier backend (`base.yml`'s + // `--streaming-config=/etc/asap/streaming.yaml` + no + // `--config=…`) starts with `inference_config.schema` set to an + // empty `PromQLSchema` and no `query_configs`. The streaming + // config DOES carry agg configs, but they declare a non-empty + // `grouping_labels` (e.g. `[zone]`) — derived from the + // controller's planner output. The bug: capability matching's + // `labels_compatible` is strict-exact, and with an empty schema + // the engine builds `req.grouping_labels = []`, which fails to + // match any agg config's `[zone]`. Every replay query lands on + // `format_unsupported_query_response` → `status=error`, + // exactly the failure mode `replay.jsonl` shows for the MVP + // demo. + // + // `build_engine_production_conditions` mirrors that exact + // deploy shape so the regression suite pins both the resolver + // fix AND the labels-superset fix. + // ------------------------------------------------------------------ + + /// Build a `SimpleEngine` with the **production warm-tier deploy + /// shape** — the one ASAPCollector's `base.yml` produces: + /// + /// * `streaming_config` carries one agg config with a non-empty + /// `grouping_labels` (e.g. `[zone]`), keyed by the metric the + /// agent's processor emits (suffixed `_quantile` for DDSketch + /// metrics, plain name for HLL/CountSketch/CountMinSketch). + /// * `inference_config.schema = PromQLSchema::new()` (empty) — + /// the warm-tier binary is launched with `--streaming-config` + /// only, no `--config`. + /// * `inference_config.query_configs = []` — no exact-string + /// QueryConfig templates. + /// + /// Replay queries reach `find_compatible_aggregation` via the + /// capability-match fallback path. Pre-fix this fails because + /// `req.grouping_labels = []` can't strict-equal `[zone]`. + #[allow(clippy::too_many_arguments)] + fn build_engine_production_conditions( + agg_metric: &str, + agg_type: AggregationType, + agg_grouping_labels: Vec<&str>, + data: Vec<(Option>, Box)>, + ) -> SimpleEngine { + let label_strings: Vec = agg_grouping_labels + .iter() + .map(|s| s.to_string()) + .collect(); + + let mut aggregation_configs = HashMap::new(); + aggregation_configs.insert( + 1u64, + AggregationConfig { + aggregation_id: 1, + aggregation_type: agg_type, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(label_strings), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size: WINDOW_LEN_MS / 1000, + slide_interval: WINDOW_LEN_MS / 1000, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: agg_metric.to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + storage_backend: Default::default(), + }); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + for (label_values_opt, acc) in data { + let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); + let output = PrecomputedOutput::new( + WINDOW_END_MS - WINDOW_LEN_MS, + WINDOW_END_MS, + key, + 1, + ); + store.insert_precomputed_output(output, acc).unwrap(); + } + + // The crucial bit: schema is EMPTY, mirroring the + // `--streaming-config`-only deploy. The pre-fix engine fails + // here because `build_query_requirements_promql` resolves + // `all_labels` to `KeyByLabelNames::empty()`. + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(PromQLSchema::new()), + query_configs: vec![], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + + SimpleEngine::new( + store, + inference_config, + streaming_config, + SCRAPE_INTERVAL_S, + QueryLanguage::promql, + ) + } + + /// (5a) `replay.jsonl` 686/686 failing rows: replays the unsuffixed + /// metric name against a DDSketch agg keyed by the suffixed wire + /// name, with the production deploy's empty schema. + #[test] + fn production_conditions_quantile_over_time_does_not_error() { + init_tracing_for_test(); + let acc = dd_sketch_with_1_to_100(); + let engine = build_engine_production_conditions( + // Agent's DDSketch processor renames to `_quantile` before emit. + "http_requests_total_latency_ms_quantile", + AggregationType::DDSketch, + vec!["zone"], + vec![(Some(vec!["us-east-1".to_string()]), Box::new(acc))], + ); + + // Replay client uses the conceptual unsuffixed name. + let query = "quantile_over_time(0.99, http_requests_total_latency_ms[1m])"; + let (_labels, qr) = engine + .handle_query_promql(query.to_string(), QUERY_TIME_SEC) + .expect( + "production warm engine must answer quantile_over_time over DDSketch even when \ + inference_config has an empty PromQLSchema (replay.jsonl 686/686 errors)", + ); + + match qr { + QueryResult::Vector(iv) => { + assert!(!iv.values.is_empty(), "expected at least one quantile value"); + let v = iv.values[0].value; + assert!( + (v - 99.0).abs() < 5.0, + "expected ~99.0 from DDSketch.quantile(0.99), got {v}" + ); + } + other => panic!("expected instant vector, got {other:?}"), + } + } + + /// (5b) `replay.jsonl` 343/343 failing rows: instant + /// `sum by (zone) (http_requests_total)` against an + /// IncreaseAccumulator-backed counter. + #[test] + fn production_conditions_sum_by_zone_instant_does_not_error() { + init_tracing_for_test(); + 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_production_conditions( + "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)), + ], + ); + + let query = "sum by (zone) (http_requests_total)"; + let (_labels, qr) = engine + .handle_query_promql(query.to_string(), QUERY_TIME_SEC) + .expect( + "production warm engine must answer instant `sum by (zone) (counter)` against \ + IncreaseAccumulator under empty PromQLSchema", + ); + + match qr { + QueryResult::Vector(iv) => { + assert_eq!(iv.values.len(), 2, "expected 2 zones"); + } + other => panic!("expected instant vector, got {other:?}"), + } + } + + /// (5c) `replay.jsonl` 343/343 failing rows: `count(unique_users_per_min)` + /// against an HLL agg. HLL accumulator answers `Statistic::Count` as a + /// cardinality alias (`hll_sketch_accumulator.rs:220`), but + /// pre-fix `compatible_agg_types(Count)` did not list HLL — capability + /// match misses → engine returns None → `status=error`. + #[test] + fn production_conditions_count_against_hll_does_not_error() { + init_tracing_for_test(); + // HLL with a few "registers set" — actual cardinality value + // is irrelevant; the test only asserts the engine resolves + // the agg and runs the accumulator's query path without + // erroring. + let mut hll = HllSketch::new(asap_sketchlib::sketches::hll::HllVariant::Regular, 8); + for i in 0..1000u32 { + hll.update(i.to_string().as_bytes()); + } + let acc = HllSketchAccumulator { inner: hll }; + + let engine = build_engine_production_conditions( + "unique_users_per_min", + AggregationType::HLL, + vec!["zone"], + vec![(Some(vec!["us-east-1".to_string()]), Box::new(acc))], + ); + + let query = "count(unique_users_per_min)"; + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME_SEC); + assert!( + result.is_some(), + "production warm engine must answer `count()` under empty PromQLSchema; \ + got None → wire status=error", + ); + } + + /// (5d) `replay.jsonl` 342/342 failing rows: `topk(5, top_endpoint_qps)` + /// against a CountSketch agg. `CountSketchAccumulator` answers + /// `Statistic::Topk` (`count_sketch_accumulator.rs:284`), but pre-fix + /// `compatible_agg_types(Topk)` only listed `CountMinSketchWithHeap`; + /// CountSketch wasn't reachable through capability matching. + /// + /// **Follow-up note**: `CountSketch` is classified as + /// `is_multi_population_value_type`, so even after adding it to + /// the Topk compat list the matcher still requires a paired + /// `SetAggregator` / `DeltaSetAggregator` on the same metric. + /// The production deploy doesn't ship one — the right structural + /// fix is for the controller to plan `top_endpoint_qps` as + /// `CountMinSketchWithHeap` (the integrated CMS+heap accumulator + /// that answers `topk` without an external key tracker). PR #344 + /// declares that capability on the controller side; the matching + /// engine-side accumulator wiring is out of scope for the + /// warm-engine-error PR. Marked `#[ignore]` until the controller + /// switches family. + #[test] + #[ignore = "follow-up: standalone CountSketch agg requires a paired SetAggregator under \ + is_multi_population_value_type semantics; the right fix is for the controller \ + to plan top_endpoint_qps as CountMinSketchWithHeap (PR #344)."] + fn production_conditions_topk_against_count_sketch_does_not_error() { + init_tracing_for_test(); + let mut cs = CountSketch::new(4, 4096); + for i in 0..100u32 { + cs.update(&format!("endpoint-{i}"), 1.0); + } + let acc = CountSketchAccumulator { inner: cs }; + + let engine = build_engine_production_conditions( + "top_endpoint_qps", + AggregationType::CountSketch, + vec!["zone"], + vec![(Some(vec!["us-east-1".to_string()]), Box::new(acc))], + ); + + let query = "topk(5, top_endpoint_qps)"; + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME_SEC); + assert!( + result.is_some(), + "production warm engine must answer `topk(K, )` under empty \ + PromQLSchema; got None → wire status=error", + ); + } + + /// (5e) `replay.jsonl` 342/342 failing rows: `rate(endpoint_request_freq[5m])` + /// against a CountMinSketch agg. `CountMinSketchAccumulator` doesn't + /// directly answer `Statistic::Rate`, but the production demo's + /// frequency probe is structurally a per-series count from a CMS; + /// the engine should at minimum resolve the agg and surface a + /// `Some(...)` rather than `status=error`. (The accumulator may + /// fail at the inner `query_statistic(Rate, …)` step today; this + /// test pins that the surface stays answerable so the replay row + /// is non-empty.) + /// + /// Until CMS gets a `Statistic::Rate` answer, the realistic + /// production fallback is `Statistic::Count` — `rate` is the + /// per-second view of the count. We assert the engine resolves + /// the agg via capability matching; the value is allowed to be + /// any finite number. + #[test] + #[ignore = "follow-up: CountMinSketchAccumulator does not yet implement \ + Statistic::Rate; see TODO.md for tracking. The engine SHOULD resolve \ + the agg through Statistic::Count compat list, but capability \ + matching for `rate(...)` requests Statistic::Rate, which today \ + only lists Increase/MultipleIncrease."] + fn production_conditions_rate_against_cms_does_not_error() { + init_tracing_for_test(); + let mut cms = CountMinSketch::new(4, 4096); + for i in 0..100u32 { + cms.update(&format!("endpoint-{i}"), 1.0); + } + let acc = CountMinSketchAccumulator { inner: cms }; + + let engine = build_engine_production_conditions( + "endpoint_request_freq", + AggregationType::CountMinSketch, + vec!["zone"], + vec![(Some(vec!["us-east-1".to_string()]), Box::new(acc))], + ); + + let query = "rate(endpoint_request_freq[5m])"; + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME_SEC); + assert!( + result.is_some(), + "production warm engine must answer `rate([5m])` under empty \ + PromQLSchema; got None → wire status=error", + ); + } + /// Sanity: when a deployment registers the bare metric name /// (no DDSketch INGEST rename applied), the alias resolver /// must leave the query unchanged — both forms might coexist