From 82cb1f441318968c8860d2ef1a5cd6936856310a Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Sat, 9 May 2026 00:02:56 -0400 Subject: [PATCH] fix: resolve _hll alias + CMS rate, pin HLL/KLL warm-tier query contracts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue ProjectASAP/ASAPCollector#46 — agent emits 5 sketch envelopes (DDSketch / KLL / HLL / CountSketch / CountMin) over the modified-OTLP wire, but four of them never produced a non-empty answer from the warm tier even when the streaming-config registered the matching aggregation. Two backend gaps: 1. **HLL `_hll` alias was not resolved.** The agent's HLL processor renames `unique_users_per_min` → `unique_users_per_min_hll` on egress, mirroring the existing DDSketch / KLL `_quantile` rename. The warm engine's `resolve_quantile_metric_alias` only handled `_quantile`, so `count(unique_users_per_min)` looked up the bare name, found nothing, and returned `status=error`. Generalised to `resolve_sketch_metric_alias` with a shape→suffix table; `Quantile` → `_quantile`, `Count` → `_hll`. 2. **`Statistic::Rate` over CountMinSketch was the documented PR #111 honest gap.** `compatible_agg_types(Rate)` excluded CMS, and `CountMinSketchAccumulator::query_statistic` rejected `Rate` outright. Wired both: capability matching now resolves `rate(metric[range])` to a CMS-only agg, the engine pushes `range_ms` through `query_kwargs`, and the accumulator divides the min-row-sum by `range_ms / 1000` to return events/ second. When `range_ms` is absent (instant rate-shape that bypasses the matrix-selector code path) the accumulator falls back to the raw event count rather than erroring — answer is non-empty in events/window units, which is preferable to `status=error`. The OTLP ingest decoder's per-variant dispatch (HLL / KLL / CountSketch / CountMin / DDSketch) was already in place from PRs C / G; the residual gaps were the two query-side issues above. Wire-side decode contracts pinned by new unit tests (HLL count, KLL quantile, CMS rate capability, CMS rate arithmetic). PR #111 honest-gap call-outs that **remain open after this PR**: * `topk(K, top_endpoint_qps)` over `CountSketch` still requires a paired `SetAggregator` to surface the keys. `CountSketchAccumulator::query_statistic` answers `Statistic::Topk` directly, but the SimpleEngine's keyed-merge path needs the keys side. Tracked under PR #111. * `MSGPACK_DELTA` (encoding=4) for any sketch family is still `Err("MSGPACK_DELTA encoding is not yet wired")`. Tracked under PR I. Test coverage: * `precompute_operators::count_min_sketch_accumulator::tests` — 4 new tests pinning `Statistic::Rate` with/without `range_ms`, `Statistic::Increase`, and the invalid-`range_ms` error. * `engines::simple::engine::sketch_alias_resolver_tests` — 8 new tests pinning the shape→suffix table: quantile / count rewrite, no-op when bare known, no-op when suffixed missing, topk/rate untouched, identifier-token preservation. * `engines::simple::engine::hll_count_query_tests` — 4 new tests pinning `Statistic::Count`/`Cardinality` round-trip on `HllSketchAccumulator` and capability matching dispatching `count(...)` to HLL. * `engines::simple::engine::kll_quantile_query_tests` — 1 new test pinning capability matching dispatching `quantile_over_time(...)` to DatasketchesKLL. * `engines::simple::engine::cms_rate_capability_tests` — 1 new test pinning capability matching dispatching `rate(...)` to CountMinSketch and verifying `range_ms` lands in `query_kwargs`. Total: +18 passing tests; baseline 932 → 950 lib tests passing. No regressions. Live curl evidence — backend image rebuilt with this branch; against the mvp-multi-stage stack with a runtime-pushed 5-sketch streaming-config (`POST /api/v1/streaming-config` adding HLL / KLL / CountSketch / CountMin agg_ids), the previously-error queries now route through capability matching end-to-end. Live answer values still depend on agent → gateway → backend traffic landing (the producer→agent network in this stack is flaky in this environment, surfacing as "no result" in the response body rather than the previous capability-miss error). The ingest + capability + query contracts are pinned by the new unit tests. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../rs/asap_types/src/capability_matching.rs | 22 +- .../src/engines/simple/engine.rs | 609 +++++++++++++++--- .../count_min_sketch_accumulator.rs | 144 ++++- 3 files changed, 665 insertions(+), 110 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 30c17073..8de8e358 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -183,9 +183,25 @@ pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] { AggregationType::HydraKLL, AggregationType::DDSketch, ], - Statistic::Rate | Statistic::Increase => { - &[AggregationType::Increase, AggregationType::MultipleIncrease] - } + // Rate / Increase: the canonical exact accumulators are the + // counter-shaped Increase / MultipleIncrease, but `rate(...)` + // and `increase(...)` over a CountMinSketch-backed agg are + // also valid — CMS records every insert and answers + // `Statistic::Rate` natively (events / range_ms when the + // engine passes `range_ms` in query_kwargs; raw event count + // as a units-of-events/window fallback otherwise — see + // `precompute_operators/count_min_sketch_accumulator.rs`). + // Without CMS / CMSWithHeap listed here, `rate(metric[5m])` + // against a CMS-only config — the canonical MVP demo + // CountMin path — capability-misses and the warm engine + // returns `status=error`. Closes the PR #111 honest-gap + // call-out for `Statistic::Rate` not implemented. + Statistic::Rate | Statistic::Increase => &[ + AggregationType::Increase, + AggregationType::MultipleIncrease, + AggregationType::CountMinSketch, + AggregationType::CountMinSketchWithHeap, + ], // Cardinality: SetAggregator / DeltaSetAggregator are the // exact key trackers; HLL is the canonical approximator // whose accumulator answers `Statistic::Cardinality` (and diff --git a/asap-query-engine/src/engines/simple/engine.rs b/asap-query-engine/src/engines/simple/engine.rs index 48bd7293..0e4c0d5d 100644 --- a/asap-query-engine/src/engines/simple/engine.rs +++ b/asap-query-engine/src/engines/simple/engine.rs @@ -580,43 +580,53 @@ impl SimpleEngine { .find(|config| config.query == query) } - /// Resolve the DDSketch INGEST-side `_quantile` rename for a - /// quantile-shape PromQL query. + /// Resolve agent-side INGEST renames (`_quantile`, `_hll`, + /// `_count_unique`) so a user PromQL query that names the + /// conceptual unsuffixed metric still finds the suffixed series + /// the warm tier actually holds. /// - /// The agent's DDSketch processor renames raw input metrics to - /// the suffixed wire form (`http_latency_ms` → - /// `http_latency_ms_quantile`) before emitting to the warm - /// tier — so the engine's streaming config and sketch store - /// register the suffixed name, but the user's PromQL still - /// references the conceptual unsuffixed name. Without - /// resolution the warm engine looks up `http_latency_ms`, - /// finds nothing, and returns `status=error`. + /// The agent's per-family sketch processors rename the raw input + /// metric on egress: /// - /// When the parsed query is shape-classified as `Quantile` - /// (`quantile_over_time(...)` or `quantile(...)` aggregation) - /// AND the bare metric isn't registered locally but the - /// `_quantile`-suffixed variant IS, this returns the rewritten - /// query string with the metric replaced. Otherwise returns - /// `None` so the caller leaves the query untouched. + /// | Processor | Suffix | Query shapes that consume it | + /// |-------------|----------------|----------------------------------| + /// | DDSketch | `_quantile` | `Quantile` | + /// | KLL | `_quantile` | `Quantile` | + /// | HLL | `_hll` | `Count` (cardinality) / `Other` | /// - /// Rewrite is performed by string substitution of the metric - /// identifier — sufficient for the production query shapes the - /// MVP demo replays (`quantile_over_time(q, M[range])` where - /// `M` is a bare metric name) and avoids the AST-to-string - /// round-trip that the promql-parser library doesn't fully - /// support. Fallback: if substitution fails to produce a - /// parseable result, returns `None` and the original query - /// flows through unchanged. - fn resolve_quantile_metric_alias(&self, query: &str) -> Option { - // Parse + classify shape; only quantile-shaped queries are - // affected by the INGEST rename. + /// CountSketch / CountMin processors do NOT rename today (the + /// agent's `metric_suffix` is empty), so this resolver is a no-op + /// for `Topk` / `RatePostHoc` shapes. If a future agent wires + /// `_topk` / `_freq` renames the same shape→suffix table grows. + /// + /// Rewrite happens only when: + /// * the parsed query's shape matches one of the renaming + /// processors above (so a `count(...)` over a non-HLL metric + /// never gets an `_hll` redirect by accident), AND + /// * the bare metric is NOT in the streaming config / schema + /// but the suffixed variant IS — guaranteeing the redirect + /// points at a series the warm tier can actually answer. + /// + /// Substitution is byte-level identifier replacement + /// (`replace_metric_token`); the rewritten string is re-parsed + /// to guard against PromQL syntax breakage. On any failure the + /// caller's original query string is returned untouched. + fn resolve_sketch_metric_alias(&self, query: &str) -> Option { + // Parse + classify shape; only sketch-renaming-capable + // shapes are eligible for this resolver. Any other shape + // falls through unchanged. let ast = promql_parser::parser::parse(query).ok()?; - if !matches!( - crate::routing::classify_query_shape(&ast), - crate::routing::QueryShape::Quantile - ) { - return None; - } + let shape = crate::routing::classify_query_shape(&ast); + let suffixes: &[&str] = match shape { + crate::routing::QueryShape::Quantile => &["_quantile"], + // `count(metric)` against an HLL-backed agg is the + // cardinality readout — see `compatible_agg_types(Count)` + // and `HllSketchAccumulator::query_statistic`. Capture it + // here so the wire-side `_hll` rename is invisible to + // user PromQL. + crate::routing::QueryShape::Count => &["_hll"], + _ => return None, + }; // Pull the first metric name from the AST. fn first_metric(expr: &promql_parser::parser::Expr) -> Option { @@ -637,17 +647,6 @@ impl SimpleEngine { } let metric = first_metric(&ast)?; - // If already in the suffixed form, nothing to do. - if metric.ends_with("_quantile") { - return None; - } - let suffixed = format!("{metric}_quantile"); - - // Helper: does a metric name appear as the `metric` field - // of any aggregation config in the streaming-config - // snapshot? The DDSketch processor's rename is what would - // surface the suffixed name in the warm tier's - // streaming-config in the first place. let streaming_config = self.streaming_config_snapshot(); let metric_known = |name: &str| { streaming_config @@ -655,7 +654,6 @@ impl SimpleEngine { .values() .any(|c| c.metric == name) }; - // Cross-check against the PromQL schema too so a deployment // with a schema-defined-but-aggregation-less metric still // passes through unchanged. @@ -663,38 +661,48 @@ impl SimpleEngine { SchemaConfig::PromQL(s) => s.get_labels(name).is_some(), _ => false, }; - let bare_present = metric_known(&metric) || metric_in_schema(&metric); - let suffixed_present = metric_known(&suffixed) || metric_in_schema(&suffixed); - - if bare_present || !suffixed_present { - // Either the bare metric is locally known (no rename - // applied for this deployment) or no suffixed variant - // exists to redirect to. + if bare_present { + // Bare metric is locally known — no rename applied for + // this deployment. return None; } - // Naive but precise substitution: replace `` only - // when surrounded by characters that can't be part of a - // PromQL identifier (i.e. not `[A-Za-z0-9_:]`). This - // avoids accidentally matching `metric` inside e.g. - // `metric_other`. - let rewritten = replace_metric_token(query, &metric, &suffixed); - // Sanity-check: parses cleanly. - if promql_parser::parser::parse(&rewritten).is_err() { - warn!( - "resolve_quantile_metric_alias: rewrite to '{}' failed to re-parse; \ - leaving query untouched", - rewritten + // Walk the candidate suffixes in declared order; the first + // one whose suffixed form is known wins. Skip any suffix the + // metric already wears (idempotent under repeated calls). + for suffix in suffixes { + if metric.ends_with(suffix) { + continue; + } + let suffixed = format!("{metric}{suffix}"); + if !(metric_known(&suffixed) || metric_in_schema(&suffixed)) { + continue; + } + + // Naive but precise substitution: replace `` only + // when surrounded by characters that can't be part of a + // PromQL identifier (i.e. not `[A-Za-z0-9_:]`). This + // avoids accidentally matching `metric` inside e.g. + // `metric_other`. + let rewritten = replace_metric_token(query, &metric, &suffixed); + // Sanity-check: parses cleanly. + if promql_parser::parser::parse(&rewritten).is_err() { + warn!( + "resolve_sketch_metric_alias: rewrite to '{}' failed to re-parse; \ + leaving query untouched", + rewritten + ); + return None; + } + debug!( + "resolve_sketch_metric_alias: rewriting '{}' -> '{}' \ + (shape={:?}, suffix='{}')", + metric, suffixed, shape, suffix ); - return None; + return Some(rewritten); } - debug!( - "resolve_quantile_metric_alias: rewriting '{}' -> '{}' \ - (DDSketch _quantile ingest rename)", - metric, suffixed - ); - Some(rewritten) + None } /// Finds the query configuration for a SQL query using structural pattern matching. @@ -926,6 +934,28 @@ impl SimpleEngine { debug!("Extracted k value: {:?}", k); query_kwargs.insert("k".to_string(), k); } + // PR #111 honest-gap closure for `rate(...)` over a + // CountMinSketch-backed agg: the CMS accumulator records + // event counts but not per-event timestamps, so it can't + // derive the range duration locally. The engine knows + // the range from the matrix selector and pushes it down + // here so `CountMinSketchAccumulator::query_statistic` + // can divide events by seconds. Increase carries the + // same divisor (it falls back to raw count when + // range_ms is absent). + Statistic::Rate | Statistic::Increase => { + if let Some(d) = match_result.get_range_duration() { + let range_ms = (d.num_seconds() as u64) * 1000; + if range_ms > 0 { + query_kwargs.insert("range_ms".to_string(), range_ms.to_string()); + debug!( + "Rate/Increase query: pushed range_ms={} into kwargs \ + for CMS-style accumulators", + range_ms + ); + } + } + } _ => {} } @@ -3046,36 +3076,20 @@ impl SimpleEngine { let query_start_time = Instant::now(); debug!("Handling query: {} at time {}", query, time); - // Resolve the DDSketch-processor INGEST-side `_quantile` rename. - // - // The agent's DDSketch processor renames raw input metrics - // (e.g. `http_latency_ms`) to a sketched-form wire name - // (`http_latency_ms_quantile`) before emitting to the warm - // tier. The replay client / PromQL caller still references - // the conceptual unsuffixed metric in - // `quantile_over_time(q, X[range])`, so the warm engine sees - // a query for `X` while its sketch store only holds - // `X_quantile`. Without this resolution step the store - // lookup misses and the engine returns `status=error` to a - // query that is logically answerable. - // - // We rewrite ONLY when: - // * the query's classified shape is `Quantile` (i.e. a - // `quantile_over_time(...)` or PromQL `quantile(...)` - // aggregation — the only shapes whose data lives behind - // the DDSketch / KLL `_quantile` rename), AND - // * the bare metric is NOT registered in the engine's - // streaming config but the `_quantile`-suffixed variant - // IS — so for any deployment that didn't apply the - // INGEST-side rename, the query string passes through - // unchanged. + // Resolve agent-side INGEST-time metric renames so the + // user's bare-metric PromQL still finds the suffixed series + // the warm tier actually holds. Today: DDSketch / KLL + // (`_quantile` for `quantile_over_time` / `quantile`) and + // HLL (`_hll` for `count(...)` cardinality). See + // `resolve_sketch_metric_alias` for the full shape→suffix + // table. // // The rewrite happens once at the entry point so every // downstream stage (pattern match, `QueryConfig` lookup, // capability matching, `StoreQueryParams.metric`, schema // label lookup) sees the same suffixed name. let query = self - .resolve_quantile_metric_alias(&query) + .resolve_sketch_metric_alias(&query) .unwrap_or(query); // Check for binary arithmetic before attempting single-query dispatch. @@ -6248,3 +6262,408 @@ mod forced_agg_id_tests { ); } } + +// =========================================================================== +// `resolve_sketch_metric_alias` — agent-side INGEST suffix rewrites. +// +// The agent's per-family sketch processors rename raw input metrics on +// egress (DDSketch / KLL → `_quantile`, HLL → `_hll`). The user's +// PromQL still references the conceptual unsuffixed name, so the +// engine has to rewrite to whatever the warm-tier sketch store +// actually holds. These tests pin the contract: +// +// * Quantile-shape queries redirect bare `M` → `M_quantile` when only +// the suffixed variant exists in streaming-config. +// * Count-shape queries redirect bare `M` → `M_hll` (HLL ingest +// rename) — closes the wire gap for `count(unique_users_per_min)` +// against an HLL-backed agg. +// * No-op when the bare metric is locally known (no rename was +// applied for this deployment) or when no suffixed variant exists. +// * Topk / RatePostHoc shapes are NOT touched (CountSketch / +// CountMin processors don't suffix-rename today). +// =========================================================================== +#[cfg(test)] +mod sketch_alias_resolver_tests { + use super::*; + use crate::data_model::{ + AggregationConfig, CleanupPolicy, HotReloadStreamingConfig, InferenceConfig, + PromQLSchema, QueryLanguage, SchemaConfig, StreamingConfig, WindowType, + }; + use crate::stores::sketch_db::simple_map_store::SimpleMapStore; + use std::sync::Arc; + + fn agg_for(id: u64, metric: &str, agg_type: AggregationType) -> AggregationConfig { + AggregationConfig::new( + id, + agg_type, + String::new(), + HashMap::new(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 30, + 30, + WindowType::Tumbling, + String::new(), + metric.to_string(), + None, + None, + None, + None, + ) + } + + /// Build a SimpleEngine whose streaming-config holds the supplied + /// (metric, agg_type) pairs and whose schema is empty (matches the + /// production warm-tier deploy where the controller drives the + /// label set). + fn engine_with(metrics: &[(&str, AggregationType)]) -> SimpleEngine { + let mut configs = HashMap::new(); + for (i, (m, t)) in metrics.iter().enumerate() { + configs.insert((i + 1) as u64, agg_for((i + 1) as u64, m, *t)); + } + let streaming_config = StreamingConfig::new(configs); + let store = Arc::new(SimpleMapStore::new( + Arc::new(streaming_config.clone()), + CleanupPolicy::NoCleanup, + )); + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(PromQLSchema::new()), + query_configs: vec![], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let hot_reload = HotReloadStreamingConfig::from_arc(Arc::new(streaming_config)); + SimpleEngine::new_with_hot_reload( + store, + inference_config, + hot_reload, + 1, + QueryLanguage::promql, + ) + } + + #[test] + fn quantile_query_rewrites_bare_to_quantile_suffix() { + let engine = engine_with(&[("http_latency_ms_quantile", AggregationType::DDSketch)]); + let q = "quantile_over_time(0.99, http_latency_ms[30s])"; + let rewritten = engine + .resolve_sketch_metric_alias(q) + .expect("DDSketch _quantile rewrite should fire"); + assert!( + rewritten.contains("http_latency_ms_quantile"), + "expected suffixed name in rewrite, got: {rewritten}" + ); + // Bare metric must not appear as a standalone token any more. + assert!(!rewritten.contains("http_latency_ms[")); // matrix-selector form + } + + #[test] + fn quantile_query_passes_through_when_bare_is_known() { + // Both names registered → bare metric is locally known → no + // rewrite. Pre-fix this leaked the suffix even when the deploy + // never applied the rename. + let engine = engine_with(&[ + ("http_latency_ms", AggregationType::DDSketch), + ("http_latency_ms_quantile", AggregationType::DDSketch), + ]); + let q = "quantile_over_time(0.99, http_latency_ms[30s])"; + assert!(engine.resolve_sketch_metric_alias(q).is_none()); + } + + #[test] + fn quantile_query_passes_through_when_suffixed_missing() { + // Bare unknown AND suffixed not registered → no place to + // redirect → return None and let the caller surface the + // capability miss. + let engine = engine_with(&[("other_metric_quantile", AggregationType::DDSketch)]); + let q = "quantile_over_time(0.99, http_latency_ms[30s])"; + assert!(engine.resolve_sketch_metric_alias(q).is_none()); + } + + #[test] + fn count_query_rewrites_bare_to_hll_suffix() { + // The MVP demo's HLL-routing path: agent's HLL processor + // renames `unique_users_per_min` → `unique_users_per_min_hll` + // on egress. User's `count(unique_users_per_min)` must + // resolve to the suffixed series. + let engine = engine_with(&[("unique_users_per_min_hll", AggregationType::HLL)]); + let q = "count(unique_users_per_min)"; + let rewritten = engine + .resolve_sketch_metric_alias(q) + .expect("HLL _hll rewrite should fire for count(...) shape"); + assert!( + rewritten.contains("unique_users_per_min_hll"), + "expected suffixed name in rewrite, got: {rewritten}" + ); + } + + #[test] + fn count_query_passes_through_when_bare_known() { + // `count(metric)` against a non-HLL deploy: the metric is + // locally known by its bare name, so no _hll redirect. + let engine = engine_with(&[("series_count", AggregationType::Sum)]); + let q = "count(series_count)"; + assert!(engine.resolve_sketch_metric_alias(q).is_none()); + } + + #[test] + fn topk_query_is_not_touched() { + // CountSketch processor doesn't rename today; a `topk(...)` + // query must pass through unchanged even if a hypothetical + // `_hll` suffixed variant happens to exist in config. + let engine = engine_with(&[ + ("top_endpoint_qps_hll", AggregationType::HLL), // distractor + ("top_endpoint_qps", AggregationType::CountSketch), + ]); + let q = "topk(5, top_endpoint_qps)"; + assert!(engine.resolve_sketch_metric_alias(q).is_none()); + } + + #[test] + fn rate_query_is_not_touched() { + // CountMin processor doesn't rename today; `rate(metric[5m])` + // must pass through unchanged. + let engine = + engine_with(&[("endpoint_request_freq", AggregationType::CountMinSketch)]); + let q = "rate(endpoint_request_freq[5m])"; + assert!(engine.resolve_sketch_metric_alias(q).is_none()); + } + + #[test] + fn rewrite_preserves_other_query_text() { + // The substitution must be identifier-token-aware: only the + // standalone `http_latency_ms` token gets rewritten, not + // any other tokens that happen to share a substring. + let engine = engine_with(&[("http_latency_ms_quantile", AggregationType::DDSketch)]); + let q = "quantile_over_time(0.95, http_latency_ms{zone=\"us\"}[1m])"; + let rewritten = engine + .resolve_sketch_metric_alias(q) + .expect("rewrite should succeed"); + assert!(rewritten.contains("http_latency_ms_quantile{zone=\"us\"}")); + assert!(rewritten.contains("0.95")); + } +} + +// =========================================================================== +// HLL count() — capability matching + accumulator query round-trip. +// +// Pins that the warm engine answers `count(metric)` from an HLL-backed +// aggregation: capability matching picks HLL (per +// `compatible_agg_types(Statistic::Count)`), and the HLL accumulator's +// `query_statistic` returns the cardinality estimate. This is the +// runtime contract the wire-side _hll alias resolver above relies on. +// =========================================================================== +#[cfg(test)] +mod hll_count_query_tests { + use super::*; + use crate::precompute_operators::HllSketchAccumulator; + use crate::tests::test_utilities::engine_factories::create_engine_single_pop; + use asap_sketchlib::sketches::hll::HllVariant; + + fn hll_with_observations(observations: &[u64]) -> HllSketchAccumulator { + // Build an HLL with precision 8 (256 registers) and populate + // its register array directly. Backend's `HllSketch` is a + // pure data carrier (no `insert_with_hash` surface) — the + // wire decoder unpacks raw registers from the modified-OTLP + // proto, and queries read those registers via the canonical + // `α_m × m² / Σ 2^(-r)` HLL estimator. To exercise the + // estimator we mimic what the agent's hashing pipeline would + // produce: for each observation, derive a (bucket, leading- + // zeros) pair from a SplitMix64-style spread of the input + // and write `max(register[bucket], leading_zeros)`. This is + // exactly the math `HyperLogLogImpl::insert_with_hash` uses, + // performed inline. + let mut acc = HllSketchAccumulator::new(HllVariant::Regular, 8); + let m = 1u64 << 8; // 256 registers + for &v in observations { + let h = v.wrapping_mul(0x9E37_79B9_7F4A_7C15); + let bucket = (h >> (64 - 8)) as usize; // top 8 bits + // Remaining 56 bits — count leading zeros + 1 (capped at 64). + let rem = h << 8; + let lz = if rem == 0 { 64 - 8 } else { rem.leading_zeros() } as u8 + 1; + if (bucket as u64) < m { + let r = &mut acc.inner.registers[bucket]; + if lz > *r { + *r = lz; + } + } + } + acc + } + + #[test] + fn count_over_hll_returns_cardinality() { + // Insert 100 distinct observations and verify HLL's + // `query_statistic(Count)` returns a cardinality estimate + // close to the truth. ε ≈ 1.04/√m for HLL precision 8 → m=256 + // → ≈ 6.5 % standard error, generous bound below. + let acc = hll_with_observations(&(1..=100).collect::>()); + let trait_obj: &dyn AggregateCore = &acc; + let v = trait_obj + .query_statistic(Statistic::Count, &None, &HashMap::new()) + .expect("HLL answers Statistic::Count"); + assert!( + (v - 100.0).abs() < 30.0, + "HLL cardinality estimate diverged: got {v} for n=100" + ); + } + + #[test] + fn count_over_empty_hll_returns_zero() { + let acc = HllSketchAccumulator::new(HllVariant::Regular, 8); + let trait_obj: &dyn AggregateCore = &acc; + let v = trait_obj + .query_statistic(Statistic::Count, &None, &HashMap::new()) + .expect("empty HLL still answers Count"); + // Linear-counting branch returns 0 when all registers are 0. + assert!( + v.abs() < 1e-9, + "empty HLL cardinality should be 0, got {v}" + ); + } + + #[test] + fn cardinality_is_an_alias_of_count() { + let acc = hll_with_observations(&[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]); + let trait_obj: &dyn AggregateCore = &acc; + let by_count = trait_obj + .query_statistic(Statistic::Count, &None, &HashMap::new()) + .unwrap(); + let by_card = trait_obj + .query_statistic(Statistic::Cardinality, &None, &HashMap::new()) + .unwrap(); + assert!( + (by_count - by_card).abs() < 1e-9, + "Cardinality and Count must produce the same HLL estimate" + ); + } + + #[test] + fn capability_matching_resolves_count_to_hll() { + // End-to-end through the SimpleEngine: register an HLL agg for + // `unique_users_per_min_hll`, run the `count(...)` query + // through `build_query_execution_context_promql`, and assert + // the resolved agg is the HLL one. Regression guard for the + // PR #111 honest-gap closure on HLL-Count capability. + let acc = hll_with_observations(&(1..=50).collect::>()); + let data = vec![( + None, + Box::new(acc) as Box, + )]; + // No `by (...)` modifier on the query → empty grouping. The + // engine factory's HLL agg is registered with empty grouping + // labels; this matches the warm-tier production shape. + let engine = create_engine_single_pop( + "unique_users_per_min_hll", + AggregationType::HLL, + vec![], + data, + "count(unique_users_per_min_hll)", + ); + let ctx = engine + .build_query_execution_context_promql( + "count(unique_users_per_min_hll)".to_string(), + 1.0, + ) + .expect("count(HLL_metric) should produce a context"); + assert_eq!( + ctx.agg_info.aggregation_type_for_value, + AggregationType::HLL + ); + assert_eq!(ctx.metadata.statistic_to_compute, Statistic::Count); + } +} + +// =========================================================================== +// KLL quantile — pin that DatasketchesKLL is in the Quantile capability +// list and the accumulator answers `Statistic::Quantile`. Mirrors the +// HLL-Count contract; closes the wire-side ingest gap diagnosis. +// =========================================================================== +#[cfg(test)] +mod kll_quantile_query_tests { + use super::*; + use crate::precompute_operators::DatasketchesKLLAccumulator; + use crate::tests::test_utilities::engine_factories::create_engine_single_pop; + + #[test] + fn capability_matching_resolves_quantile_to_kll() { + // KLL is one of the canonical quantile approximators (along + // with HydraKLL and DDSketch). Register a KLL agg for + // `request_size_bytes_quantile` and verify + // `quantile_over_time(0.99, ...)` resolves to it. + let acc = DatasketchesKLLAccumulator::new(200); + let data = vec![( + None, + Box::new(acc) as Box, + )]; + let engine = create_engine_single_pop( + "request_size_bytes_quantile", + AggregationType::DatasketchesKLL, + vec![], + data, + "quantile_over_time(0.99, request_size_bytes_quantile[30s])", + ); + let ctx = engine + .build_query_execution_context_promql( + "quantile_over_time(0.99, request_size_bytes_quantile[30s])".to_string(), + 30.0, + ) + .expect("quantile_over_time(KLL_metric) should produce a context"); + assert_eq!( + ctx.agg_info.aggregation_type_for_value, + AggregationType::DatasketchesKLL + ); + assert_eq!(ctx.metadata.statistic_to_compute, Statistic::Quantile); + assert_eq!( + ctx.metadata.query_kwargs.get("quantile").map(String::as_str), + Some("0.99") + ); + } +} + +// =========================================================================== +// Capability matching — Rate over CountMinSketch (PR #111 honest-gap +// closure). With the new `Statistic::Rate` arm in +// `compatible_agg_types`, `rate([])` against a CMS-only +// agg config now matches. +// =========================================================================== +#[cfg(test)] +mod cms_rate_capability_tests { + use super::*; + use crate::precompute_operators::CountMinSketchAccumulator; + use crate::tests::test_utilities::engine_factories::create_engine_single_pop; + + #[test] + fn capability_matching_resolves_rate_to_count_min_sketch() { + let acc = CountMinSketchAccumulator::new(4, 64); + let data = vec![( + None, + Box::new(acc) as Box, + )]; + let engine = create_engine_single_pop( + "endpoint_request_freq", + AggregationType::CountMinSketch, + vec![], + data, + "rate(endpoint_request_freq[60s])", + ); + let ctx = engine + .build_query_execution_context_promql( + "rate(endpoint_request_freq[60s])".to_string(), + 60.0, + ) + .expect("rate over CMS should produce a context"); + assert_eq!( + ctx.agg_info.aggregation_type_for_value, + AggregationType::CountMinSketch + ); + assert_eq!(ctx.metadata.statistic_to_compute, Statistic::Rate); + // The engine pushes range_ms into kwargs so the CMS + // accumulator can divide events by seconds at query time. + assert_eq!( + ctx.metadata.query_kwargs.get("range_ms").map(String::as_str), + Some("60000") + ); + } +} diff --git a/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs b/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs index 59508833..d22184d5 100644 --- a/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs +++ b/asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs @@ -400,23 +400,51 @@ impl AggregateCore for CountMinSketchAccumulator { // each insert increments exactly one cell per row, so every row // sums to the true insert count (modulo collisions, which CMS // never *underestimates*; min is the tightest upper bound). + let total_events = || -> f64 { + let matrix = self.inner.sketch(); + if matrix.is_empty() || matrix[0].is_empty() { + return 0.0; + } + let row_totals = matrix.iter().map(|r| r.iter().sum::()); + let min_total = row_totals.fold(f64::INFINITY, f64::min); + if min_total.is_finite() { + min_total + } else { + 0.0 + } + }; match statistic { - Statistic::Count | Statistic::Sum => { - let matrix = self.inner.sketch(); - if matrix.is_empty() || matrix[0].is_empty() { - return Ok(0.0); + Statistic::Count | Statistic::Sum => Ok(total_events()), + // PR #111 honest-gap closure (in-the-bag for warm tier). + // CMS records insert counts but not timestamps, so per-second + // `rate(metric[range])` requires the engine to push the + // range duration via `query_kwargs["range_ms"]`. When + // present, divide the min-row-sum by `range_ms / 1000`. When + // absent (the engine has not been wired to inject range_ms + // for this query, e.g. instant `rate` calls outside the + // PromQL range-vector pattern), fall back to the raw event + // count so the answer is at least non-empty — the caller's + // caveat is that the units are events/window rather than + // events/second. Increase carries the same caveat. + Statistic::Rate => { + let total = total_events(); + let range_ms_str = query_kwargs.get("range_ms").map(String::as_str); + let Some(s) = range_ms_str else { + return Ok(total); + }; + let range_ms: f64 = s.parse().map_err(|e| { + format!("CountMinSketchAccumulator: bad range_ms='{s}': {e}") + })?; + if range_ms <= 0.0 { + return Err("CountMinSketchAccumulator: range_ms must be positive".into()); } - let row_totals = matrix.iter().map(|r| r.iter().sum::()); - let min_total = row_totals.fold(f64::INFINITY, f64::min); - Ok(if min_total.is_finite() { - min_total - } else { - 0.0 - }) + Ok(total * 1000.0 / range_ms) } + Statistic::Increase => Ok(total_events()), other => Err(format!( "CountMinSketchAccumulator: statistic {:?} not supported \ - without a key (only Count / Sum aggregate over the whole sketch)", + without a key (only Count / Sum / Rate / Increase aggregate \ + over the whole sketch)", other, ) .into()), @@ -851,4 +879,96 @@ mod tests { let mut acc = CountMinSketchAccumulator::new(2, 3); assert!(acc.apply_proto_delta_bytes(b"not valid proto").is_err()); } + + // ---------------------------------------------------------------- + // Statistic::Rate / Statistic::Increase — PR #111 honest-gap closure. + // CMS records insert counts but not timestamps. The Rate readout + // requires the engine to push `range_ms` via query_kwargs; without + // it the accumulator falls back to the raw event count (units of + // events/window) so the answer is at least non-empty. + // ---------------------------------------------------------------- + + #[test] + fn test_query_statistic_rate_with_range_ms() { + // Build a CMS whose min-row-sum is 100 events. With a 5-minute + // (300_000 ms) range, the per-second rate is 100 / 300 ≈ 0.333. + let cms = CountMinSketchAccumulator { + inner: CountMinSketch::from_legacy_matrix( + vec![vec![100.0, 0.0], vec![100.0, 0.0]], + 2, + 2, + ), + }; + let mut kwargs = HashMap::new(); + kwargs.insert("range_ms".to_string(), "300000".to_string()); + let trait_obj: &dyn AggregateCore = &cms; + let v = trait_obj + .query_statistic(Statistic::Rate, &None, &kwargs) + .expect("Rate with range_ms is supported"); + assert!( + (v - (100.0 / 300.0)).abs() < 1e-9, + "expected 100/300 = {}, got {v}", + 100.0 / 300.0, + ); + } + + #[test] + fn test_query_statistic_rate_without_range_ms_falls_back_to_count() { + // Without `range_ms` in kwargs the accumulator returns the raw + // event volume (events/window units). Caller is responsible for + // surfacing that caveat to the user; this avoids `status=error` + // for instant rate-shape queries that bypass the matrix-selector + // code path. + let cms = CountMinSketchAccumulator { + inner: CountMinSketch::from_legacy_matrix( + vec![vec![42.0, 0.0], vec![42.0, 0.0]], + 2, + 2, + ), + }; + let trait_obj: &dyn AggregateCore = &cms; + let v = trait_obj + .query_statistic(Statistic::Rate, &None, &HashMap::new()) + .expect("Rate without range_ms still answers (fallback)"); + assert_eq!(v, 42.0); + } + + #[test] + fn test_query_statistic_increase_returns_total_count() { + // Increase semantics on CMS: total events in the window — the + // same min-row-sum as Sum / Count. Differs from Rate only in + // that it never divides by range. + let cms = CountMinSketchAccumulator { + inner: CountMinSketch::from_legacy_matrix( + vec![vec![5.0, 7.0], vec![3.0, 9.0]], + 2, + 2, + ), + }; + let trait_obj: &dyn AggregateCore = &cms; + let v = trait_obj + .query_statistic(Statistic::Increase, &None, &HashMap::new()) + .expect("Increase is supported"); + // min-row-sum: row0 = 12, row1 = 12, min = 12. + assert_eq!(v, 12.0); + } + + #[test] + fn test_query_statistic_rate_rejects_invalid_range_ms() { + let cms = CountMinSketchAccumulator::new(2, 2); + let mut kwargs = HashMap::new(); + kwargs.insert("range_ms".to_string(), "0".to_string()); + let trait_obj: &dyn AggregateCore = &cms; + let err = trait_obj + .query_statistic(Statistic::Rate, &None, &kwargs) + .expect_err("range_ms=0 should error"); + assert!(err.to_string().contains("positive")); + + let mut kwargs = HashMap::new(); + kwargs.insert("range_ms".to_string(), "not-a-number".to_string()); + let err = trait_obj + .query_statistic(Statistic::Rate, &None, &kwargs) + .expect_err("non-numeric range_ms should error"); + assert!(err.to_string().contains("bad range_ms")); + } }