diff --git a/control_plane/src/sketch_algebra/matcher.rs b/control_plane/src/sketch_algebra/matcher.rs index e3f988d0..c2ed7a73 100644 --- a/control_plane/src/sketch_algebra/matcher.rs +++ b/control_plane/src/sketch_algebra/matcher.rs @@ -22,12 +22,11 @@ //! on whatever node carries it, not inside the kind — see //! `crates/asap_types/src/key_by_label_names.rs`'s module doc for the same //! design call made on the data-plane side), so a two-`Implementation` -//! `Matcher` cannot correctly answer that question. That richer, -//! grouping-aware matching already exists as production code in -//! `asap_types::capability_matching::find_compatible_aggregation`, which -//! checks `AggregationType` compatibility (analogous to `SummaryFamily` -//! here) *and* `grouping_labels` subset-compatibility side by side — the -//! two checks compose at the caller, not inside a single `Matcher::is_satisfied_by`. +//! `Matcher` cannot correctly answer that question. A caller needing that +//! richer, grouping-aware answer must check `AggregationType` compatibility +//! (analogous to `SummaryFamily` here) *and* `grouping_labels` +//! subset-compatibility side by side, composed at the call site rather than +//! inside a single `Matcher::is_satisfied_by`. use asap_plan::{Implementation, Matcher}; use asap_sketch::SummaryKind; diff --git a/crates/asap_types/src/capability_matching.rs b/crates/asap_types/src/capability_matching.rs index ad2d75f3..91218ea8 100644 --- a/crates/asap_types/src/capability_matching.rs +++ b/crates/asap_types/src/capability_matching.rs @@ -1,15 +1,6 @@ -use std::cmp::Ordering; -use std::collections::HashMap; - -use crate::KeyByLabelNames; use crate::Statistic; use serde::{Deserialize, Serialize}; -use tracing::{debug, warn}; -use crate::aggregation_config::{AggregationConfig, AggregationIdInfo}; -use crate::enums::WindowType; -use crate::query_requirements::QueryRequirements; -use crate::utils::normalize_spatial_filter; use crate::AggregationType; pub const ENGINE_ID_ASAP_QUERY: &str = "asap_query"; @@ -20,12 +11,12 @@ pub const CANONICAL_QUERY_ENGINE_IDS: &[&str] = &[ENGINE_ID_ASAP_QUERY, ENGINE_I // --------------------------------------------------------------------------- // Phase-5: storage-backend capability axis // -// Today's `find_compatible_aggregation` matches on -// `(metric, statistic, sub_type, window_size, grouping_labels, spatial_filter)` -// — there is no axis for "which storage tier serves this query." The Phase-5 -// `GorillaQueryEngine` (PR #85) introduces a parallel exact tier; the planner / -// router needs to disambiguate between ASAP-tier sketches and Gorilla-S3 -// chunks. See `docs/design-gorilla-s3-cold-engine.md` §8. +// Matching on `(metric, statistic, sub_type, window_size, grouping_labels, +// spatial_filter)` alone has no axis for "which storage tier serves this +// query." The Phase-5 `GorillaQueryEngine` (PR #85) introduces a parallel +// exact tier; the planner / router needs to disambiguate between ASAP-tier +// sketches and Gorilla-S3 chunks. See `docs/design-gorilla-s3-cold-engine.md` +// §8. // --------------------------------------------------------------------------- /// Which physical storage tier a query (or a metric configuration) routes to. @@ -121,17 +112,14 @@ pub enum AccuracyTarget { /// `promql_utilities` retirement, the **single source of truth** for it — /// there used to be a second, independently-maintained table /// (`promql_utilities::query_logics::logics::map_statistic_to_precompute_operator`, -/// the planner's own canonical map) that this one had to agree with, -/// checked by a `capability_canonical_map_agreement` test. That table was -/// dead code (a Python-planner relic — nothing in Rust ever called it -/// except that one test) and was deleted; this is now the only table. +/// the planner's own canonical map) that this one had to agree with. That +/// table was dead code (a Python-planner relic) and was deleted; this is +/// now the only table. /// -/// The runtime caller (`find_compatible_aggregation`) has no -/// `QueryTreatmentType` to consult — `QueryRequirements` is treatment-agnostic -/// — so this list intentionally enumerates *every* type that could serve the -/// statistic. Selection between e.g. `Sum` (exact) and `CountMinSketch` -/// (approximate) for `Statistic::Sum` is made downstream via -/// `aggregation_priority` (largest window size wins). +/// `QueryTreatmentType` is not consulted here — this list intentionally +/// enumerates *every* type that could serve the statistic, treatment-agnostic. +/// Selection between e.g. `Sum` (exact) and `CountMinSketch` (approximate) +/// for `Statistic::Sum` is made downstream by the caller. pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] { match stat { // Sum: exact via Sum / MultipleSum; approximate via CountMinSketch. @@ -309,245 +297,6 @@ pub fn required_sub_type(stat: Statistic) -> Option<&'static str> { } } -/// Whether this value aggregation type requires a paired key aggregation -/// (`SetAggregator` or `DeltaSetAggregator`). -pub fn is_multi_population_value_type(agg_type: AggregationType) -> bool { - agg_type.is_multi_population_value_type() -} - -/// Whether this type is a key aggregation (tracks which label-value combinations exist). -fn is_key_agg_type(agg_type: AggregationType) -> bool { - agg_type.is_key_agg_type() -} - -/// Window compatibility: can `config` serve a query needing `data_range_ms`? -/// -/// - `None` (spatial-only): always compatible. -/// - Tumbling: `data_range_ms` must be a positive integer multiple of `window_size_ms`. -/// - Sliding: `data_range_ms` must equal `window_size_ms` exactly (a sliding window -/// precomputes one fixed range per timestamp; overlapping windows cannot be merged). -pub fn window_compatible(config: &AggregationConfig, data_range_ms: Option) -> bool { - let Some(range) = data_range_ms else { - return true; - }; - let window_ms = config.window_size * 1000; - if window_ms == 0 || range == 0 { - return false; - } - match config.window_type { - WindowType::Sliding => range == window_ms, - WindowType::Tumbling => range % window_ms == 0, - } -} - -/// 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 { - 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. -/// - Both empty → compatible. -/// - Config non-empty and matches query → compatible. -/// - Config non-empty and query differs (or is empty) → incompatible. -pub fn spatial_filter_compatible(config_filter: &str, req_filter: &str) -> bool { - let config_norm = normalize_spatial_filter(config_filter); - let req_norm = normalize_spatial_filter(req_filter); - if config_norm.is_empty() { - // Config has no filter — compatible with any query filter. - return true; - } - config_norm == req_norm -} - -/// Aggregation priority comparator: prefer larger `window_size` (descending). -/// This is a separate function so callers can swap the policy without touching matching logic. -/// -/// Sort keys (each `then_with`s the previous when equal): -/// 1. **Larger `window_size` wins.** Coarser windows can answer -/// finer-grained queries by re-aggregation. -/// 2. **Single-population variants beat multi-population.** Multi-pop -/// types (`CountMinSketch`, `MultipleSum`, etc.) require a paired -/// key-aggregation lookup downstream; single-pop types -/// (`Sum`, `Increase`, `MinMax`, …) don't. Preferring single-pop -/// avoids the key-aggregation hunt when both shapes serve the -/// statistic — which is the common case for `Statistic::Sum` -/// matching both `Sum` and `CountMinSketch`. -/// 3. **Tie-break on `policy_fp_u64()` (the policy fingerprint).** -/// Deterministic across runs and hosts; fixes the -/// HashMap-iteration-order flake on `avg_finds_sum_and_count`. -pub fn aggregation_priority(a: &AggregationConfig, b: &AggregationConfig) -> Ordering { - let a_multi = is_multi_population_value_type(a.aggregation_type); - let b_multi = is_multi_population_value_type(b.aggregation_type); - b.window_size - .cmp(&a.window_size) - // `false < true` in Rust's bool Ord → single-pop sorts FIRST. - .then_with(|| a_multi.cmp(&b_multi)) - .then_with(|| a.policy_fp_u64().cmp(&b.policy_fp_u64())) -} - -// --------------------------------------------------------------------------- -// Core matching function -// --------------------------------------------------------------------------- - -/// Find a compatible aggregation (or pair of aggregations for multi-population queries) -/// given all available aggregation configs and a set of query requirements. -/// -/// Returns `None` if no fully compatible match exists. -/// -/// Algorithm: -/// 1. For each statistic, collect and sort compatible candidates. -/// 2. For multi-statistic requirements (e.g. avg = [Sum, Count]), all must be -/// served by configs sharing the same `window_size` and `grouping_labels`. -/// 3. If the selected value aggregation type is multi-population, also find a -/// paired key aggregation (`SetAggregator` / `DeltaSetAggregator`) on the same metric. -pub fn find_compatible_aggregation( - configs: &HashMap, - requirements: &QueryRequirements, -) -> Option { - if requirements.statistics.is_empty() { - return None; - } - - debug!( - metric = %requirements.metric, - statistics = ?requirements.statistics, - data_range_ms = ?requirements.data_range_ms, - grouping_labels = ?requirements.grouping_labels.labels, - "capability matching: searching {} aggregation config(s)", - configs.len(), - ); - - // For each statistic, collect configs that pass all filters, sorted by priority. - let mut per_stat_candidates: Vec> = Vec::new(); - - for &stat in &requirements.statistics { - let types = compatible_agg_types(stat); - let sub_type = required_sub_type(stat); - - let mut candidates: Vec<&AggregationConfig> = configs - .values() - .filter(|c| { - let ok = c.metric == requirements.metric - && types.contains(&c.aggregation_type) - && sub_type.is_none_or(|st| c.aggregation_sub_type == st) - && window_compatible(c, requirements.data_range_ms) - && labels_compatible(&c.grouping_labels, &requirements.grouping_labels) - && spatial_filter_compatible( - &c.spatial_filter_normalized, - &requirements.spatial_filter_normalized, - ); - if !ok { - debug!( - policy_fp = c.policy_fp_u64(), - agg_type = %c.aggregation_type, - metric = %c.metric, - window_size_s = c.window_size, - "capability matching: rejected config for {:?}", - stat, - ); - } - ok - }) - .collect(); - - candidates.sort_by(|a, b| aggregation_priority(a, b)); - - if candidates.is_empty() { - warn!( - metric = %requirements.metric, - statistic = ?stat, - "capability matching: no compatible aggregation found for statistic", - ); - return None; - } - - debug!( - statistic = ?stat, - num_candidates = candidates.len(), - chosen_policy_fp = candidates[0].policy_fp_u64(), - chosen_agg_type = %candidates[0].aggregation_type, - chosen_window_size_s = candidates[0].window_size, - "capability matching: found candidates, chose best", - ); - - per_stat_candidates.push(candidates); - } - - // Pick the best candidate for the first statistic. - let value_agg = per_stat_candidates[0][0]; - - // For multi-statistic requirements, the remaining statistics must be served by a - // config that agrees on window_size and grouping_labels with the chosen value agg. - for (i, candidates) in per_stat_candidates.iter().enumerate().skip(1) { - let found = candidates.iter().any(|c| { - c.window_size == value_agg.window_size && c.grouping_labels == value_agg.grouping_labels - }); - if !found { - warn!( - metric = %requirements.metric, - statistic = ?requirements.statistics[i], - required_window_size_s = value_agg.window_size, - "capability matching: no matching window/labels for multi-statistic requirement", - ); - return None; - } - } - - // If value type is multi-population, find the paired key aggregation. - let key_agg: &AggregationConfig = if is_multi_population_value_type(value_agg.aggregation_type) - { - let ka = configs - .values() - .find(|c| c.metric == requirements.metric && is_key_agg_type(c.aggregation_type)); - if ka.is_none() { - warn!( - metric = %requirements.metric, - value_agg_type = %value_agg.aggregation_type, - "capability matching: multi-population value agg requires a key agg (SetAggregator/DeltaSetAggregator) but none found", - ); - } - ka? - } else { - value_agg - }; - - debug!( - metric = %requirements.metric, - value_policy_fp = value_agg.policy_fp_u64(), - value_agg_type = %value_agg.aggregation_type, - key_policy_fp = key_agg.policy_fp_u64(), - key_agg_type = %key_agg.aggregation_type, - "capability matching: resolved", - ); - - Some(AggregationIdInfo { - aggregation_id_for_value: value_agg.policy_fp_u64(), - aggregation_type_for_value: value_agg.aggregation_type, - aggregation_id_for_key: key_agg.policy_fp_u64(), - aggregation_type_for_key: key_agg.aggregation_type, - }) -} - // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- @@ -555,524 +304,6 @@ pub fn find_compatible_aggregation( #[cfg(test)] mod tests { use super::*; - use crate::utils::normalize_spatial_filter; - use crate::KeyByLabelNames; - use std::collections::HashMap; - - #[allow(clippy::too_many_arguments)] - fn make_config( - _id: u64, - metric: &str, - agg_type: &str, - sub_type: &str, - window_size_s: u64, - window_type: &str, - grouping: &[&str], - spatial_filter: &str, - ) -> AggregationConfig { - // `_id` is unused after PR 5 — identity is content-addressed - // via `PolicyFingerprint::from_config`. Kept as a parameter so - // the wide-coverage assertions in this module's test cases - // don't churn. - let grouping_labels = - KeyByLabelNames::new(grouping.iter().map(|s| s.to_string()).collect()); - let spatial_filter_normalized = normalize_spatial_filter(spatial_filter); - AggregationConfig { - aggregation_type: agg_type.parse::().expect("valid agg type"), - aggregation_sub_type: sub_type.to_string(), - parameters: HashMap::new(), - grouping_labels, - aggregated_labels: KeyByLabelNames::new(vec![]), - rollup_labels: KeyByLabelNames::new(vec![]), - original_yaml: String::new(), - window_size: window_size_s, - slide_interval: window_size_s, - window_type: window_type.parse::().unwrap_or_default(), - spatial_filter: spatial_filter.to_string(), - spatial_filter_normalized, - metric: metric.to_string(), - num_aggregates_to_retain: None, - table_name: None, - value_column: None, - } - } - - fn req( - metric: &str, - stats: &[Statistic], - data_range_ms: Option, - grouping: &[&str], - spatial_filter: &str, - ) -> QueryRequirements { - QueryRequirements { - metric: metric.to_string(), - statistics: stats.to_vec(), - data_range_ms, - grouping_labels: KeyByLabelNames::new(grouping.iter().map(|s| s.to_string()).collect()), - spatial_filter_normalized: normalize_spatial_filter(spatial_filter), - } - } - - fn single_config(config: AggregationConfig) -> HashMap { - let mut m = HashMap::new(); - m.insert(config.policy_fp_u64(), config); - m - } - - // --- basic type matching --- - - #[test] - fn basic_sum_match() { - let cfg = make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], ""); - let expected = cfg.policy_fp_u64(); - let configs = single_config(cfg); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); - assert!(result.is_some()); - assert_eq!(result.unwrap().aggregation_id_for_value, expected); - } - - #[test] - fn quantile_any_value_finds_kll() { - let cfg = make_config(2, "lat", "DatasketchesKLL", "", 300, "tumbling", &[], ""); - let expected = cfg.policy_fp_u64(); - let configs = single_config(cfg); - // quantile value (0.5 or 0.9) is NOT part of QueryRequirements — both should find the same config - let r1 = find_compatible_aggregation( - &configs, - &req("lat", &[Statistic::Quantile], Some(300_000), &[], ""), - ); - let r2 = find_compatible_aggregation( - &configs, - &req("lat", &[Statistic::Quantile], Some(300_000), &[], ""), - ); - assert_eq!(r1.unwrap().aggregation_id_for_value, expected); - assert_eq!(r2.unwrap().aggregation_id_for_value, expected); - } - - #[test] - fn quantile_matches_hydrarkll() { - let cfg = make_config(3, "lat", "HydraKLL", "", 300, "tumbling", &[], ""); - let expected = cfg.policy_fp_u64(); - let configs = single_config(cfg); - let result = find_compatible_aggregation( - &configs, - &req("lat", &[Statistic::Quantile], Some(300_000), &[], ""), - ); - assert_eq!(result.unwrap().aggregation_id_for_value, expected); - } - - #[test] - fn no_match_wrong_metric() { - let configs = single_config(make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], "")); - let result = find_compatible_aggregation( - &configs, - &req("mem", &[Statistic::Sum], Some(300_000), &[], ""), - ); - assert!(result.is_none()); - } - - #[test] - fn no_match_wrong_type() { - let configs = single_config(make_config( - 1, - "cpu", - "DatasketchesKLL", - "", - 300, - "tumbling", - &[], - "", - )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); - assert!(result.is_none()); - } - - // --- window compatibility --- - - #[test] - fn window_tumbling_exact() { - let configs = single_config(make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], "")); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); - assert!(result.is_some()); - } - - #[test] - fn window_tumbling_divisible() { - // 900_000 ms / 300 s = 3 buckets — valid merge - let configs = single_config(make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], "")); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(900_000), &[], ""), - ); - assert!(result.is_some()); - } - - #[test] - fn window_tumbling_not_divisible() { - // 600_000 ms / 900 s is not a whole number - let configs = single_config(make_config(1, "cpu", "Sum", "", 900, "tumbling", &[], "")); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(600_000), &[], ""), - ); - assert!(result.is_none()); - } - - #[test] - fn window_sliding_exact() { - let configs = single_config(make_config(1, "cpu", "Sum", "", 300, "sliding", &[], "")); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); - assert!(result.is_some()); - } - - #[test] - fn window_sliding_too_large() { - // Query range 600 s but sliding window only covers 300 s - let configs = single_config(make_config(1, "cpu", "Sum", "", 300, "sliding", &[], "")); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(600_000), &[], ""), - ); - assert!(result.is_none()); - } - - #[test] - fn window_priority_largest_wins() { - let small = make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], ""); - let large = make_config(2, "cpu", "Sum", "", 900, "tumbling", &[], ""); - let expected = large.policy_fp_u64(); - let mut configs = HashMap::new(); - configs.insert(small.policy_fp_u64(), small); - configs.insert(large.policy_fp_u64(), large); - // 900_000 ms is divisible by both 300 s and 900 s — prefer 900 s - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(900_000), &[], ""), - ); - assert_eq!(result.unwrap().aggregation_id_for_value, expected); - } - - #[test] - fn spatial_only_no_range() { - // data_range_ms = None → any window size is compatible - let configs = single_config(make_config(1, "cpu", "Sum", "", 900, "tumbling", &[], "")); - let result = - find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], None, &[], "")); - assert!(result.is_some()); - } - - // --- label compatibility --- - - #[test] - fn label_strict_exact() { - 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"], ""), - ); - assert!(result.is_some()); - } - - #[test] - 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", - "Sum", - "", - 300, - "tumbling", - &["job", "instance"], - "", - )); - let result = find_compatible_aggregation( - &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()); - } - - #[test] - fn label_mismatch_rejected() { - let configs = single_config(make_config( - 1, - "cpu", - "Sum", - "", - 300, - "tumbling", - &["region"], - "", - )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &["job"], ""), - ); - assert!(result.is_none()); - } - - // --- spatial filter compatibility --- - - #[test] - fn spatial_filter_empty_both() { - let configs = single_config(make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], "")); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); - assert!(result.is_some()); - } - - #[test] - fn spatial_filter_query_empty_config_has_filter() { - // Config scoped to env=prod, query has no filter → reject - let configs = single_config(make_config( - 1, - "cpu", - "Sum", - "", - 300, - "tumbling", - &[], - "env=prod", - )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); - assert!(result.is_none()); - } - - #[test] - fn spatial_filter_same() { - let configs = single_config(make_config( - 1, - "cpu", - "Sum", - "", - 300, - "tumbling", - &[], - "env=prod", - )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], "env=prod"), - ); - assert!(result.is_some()); - } - - #[test] - fn spatial_filter_different() { - let configs = single_config(make_config( - 1, - "cpu", - "Sum", - "", - 300, - "tumbling", - &[], - "env=prod", - )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], "env=staging"), - ); - assert!(result.is_none()); - } - - // --- sub-type --- - - #[test] - fn sub_type_min_matches_min() { - let configs = single_config(make_config( - 1, - "cpu", - "MinMax", - "min", - 300, - "tumbling", - &[], - "", - )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Min], Some(300_000), &[], ""), - ); - assert!(result.is_some()); - } - - #[test] - fn sub_type_max_rejects_min() { - // Max statistic requires sub_type == "max", but config has "min" - let configs = single_config(make_config( - 1, - "cpu", - "MinMax", - "min", - 300, - "tumbling", - &[], - "", - )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Max], Some(300_000), &[], ""), - ); - assert!(result.is_none()); - } - - // --- multi-population --- - - // `multi_pop_finds_key_agg` retired alongside the - // `SetAggregator` / `DeltaSetAggregator` family — it asserted - // that a CountMinSketchWithHeap value paired with a - // DeltaSetAggregator key resolved for Topk; the key half is - // no longer expressible. - - #[test] - fn multi_pop_no_key_agg_returns_none() { - // CountMinSketchWithHeap present but no SetAggregator/DeltaSetAggregator - let configs = single_config(make_config( - 10, - "req", - "CountMinSketchWithHeap", - "", - 300, - "tumbling", - &[], - "", - )); - let result = find_compatible_aggregation( - &configs, - &req("req", &[Statistic::Topk], Some(300_000), &[], ""), - ); - assert!(result.is_none()); - } - - // --- avg (Vec) --- - - #[test] - fn avg_finds_sum_and_count() { - let sum = make_config(1, "cpu", "Sum", "", 300, "tumbling", &["job"], ""); - let cnt = make_config( - 2, - "cpu", - "CountMinSketch", - "", - 300, - "tumbling", - &["job"], - "", - ); - let mut configs = HashMap::new(); - configs.insert(sum.policy_fp_u64(), sum); - configs.insert(cnt.policy_fp_u64(), cnt); - let result = find_compatible_aggregation( - &configs, - &req( - "cpu", - &[Statistic::Sum, Statistic::Count], - Some(300_000), - &["job"], - "", - ), - ); - assert!(result.is_some()); - } - - #[test] - fn avg_different_windows_rejected() { - let sum = make_config(1, "cpu", "Sum", "", 300, "tumbling", &["job"], ""); - // Count config has different window_size — must be rejected - let cnt = make_config( - 2, - "cpu", - "CountMinSketch", - "", - 900, - "tumbling", - &["job"], - "", - ); - let mut configs = HashMap::new(); - configs.insert(sum.policy_fp_u64(), sum); - configs.insert(cnt.policy_fp_u64(), cnt); - let result = find_compatible_aggregation( - &configs, - &req( - "cpu", - &[Statistic::Sum, Statistic::Count], - Some(300_000), - &["job"], - "", - ), - ); - assert!(result.is_none()); - } /// Pin the canonical-approximator picks driving the ASAP-tier query path /// (the "five sketch types" CMS / KLL / HLL / DDSketch / CountSketch @@ -1145,64 +376,6 @@ mod tests { "HLL must be a compatible type for Count (warm-engine-error fix)", ); } - - /// Phase-3.1 regression test for the canonical MVP-demo failure - /// described in `docs/spec-mvp-controller-driven-multi-stage-demo.md`: - /// the controller plans `http_requests_total_latency_ms` as a - /// `DDSketch` for the `quantile_over_time(0.99, - /// http_requests_total_latency_ms[1m])` query class. When the - /// inference YAML doesn't include an exact-string entry for the - /// query, `find_query_config` misses and the engine falls into - /// capability matching. Pre-fix, `compatible_agg_types(Quantile)` - /// listed only KLL types, so the DDSketch agg was filtered out - /// and the ASAP-tier engine returned a 404 / null; post-fix, - /// DDSketch is enumerated and capability matching resolves the - /// agg cleanly. - #[test] - fn ddsketch_resolves_quantile_query_post_fix() { - let cfg = make_config( - 42, - "http_requests_total_latency_ms", - "DDSketch", - "", - 60, - "tumbling", - &[], - "", - ); - let expected = cfg.policy_fp_u64(); - let mut configs = HashMap::new(); - configs.insert(cfg.policy_fp_u64(), cfg); - let result = find_compatible_aggregation( - &configs, - &req( - "http_requests_total_latency_ms", - &[Statistic::Quantile], - Some(60_000), - &[], - "", - ), - ); - let info = result.expect( - "post-fix: capability matching must resolve quantile_over_time against a DDSketch-only config", - ); - assert_eq!(info.aggregation_id_for_value, expected); - assert_eq!(info.aggregation_type_for_value, AggregationType::DDSketch); - // DDSketch is single-population (not is_multi_population_value_type), - // so the matcher pairs it with itself for the key agg. - assert_eq!(info.aggregation_id_for_key, expected); - } - - /// Regression test for the pre-fix bug: a query for `Statistic::Sum` - /// against a CMS-only configuration must now resolve via capability - /// matching, not fall through to the cold tier. Pre-fix, this returned - /// `None`; post-fix, it returns the CMS aggregation paired with the - /// `DeltaSetAggregator` key aggregation. - // `cms_resolves_sum_query_post_fix` retired alongside the - // `SetAggregator` / `DeltaSetAggregator` family — same gist as - // `multi_pop_finds_key_agg` above (CountMinSketch value + - // DeltaSetAggregator key for Sum). - // ----------------------------------------------------------------------- // Phase-5: storage-backend routing // diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 83d0c175..412369d7 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -13,9 +13,8 @@ pub mod utils; pub use aggregation_config::*; pub use aggregation_type::AggregationType; pub use capability_matching::{ - compatible_storage_backends, find_compatible_aggregation, parse_storage_backend_engine_id, - AccuracyTarget, StorageBackend, CANONICAL_QUERY_ENGINE_IDS, ENGINE_ID_ASAP_QUERY, - ENGINE_ID_THANOS_QUERY, + compatible_storage_backends, parse_storage_backend_engine_id, AccuracyTarget, StorageBackend, + CANONICAL_QUERY_ENGINE_IDS, ENGINE_ID_ASAP_QUERY, ENGINE_ID_THANOS_QUERY, }; pub use enums::*; pub use key_by_label_names::KeyByLabelNames; diff --git a/crates/asap_types/src/streaming_config.rs b/crates/asap_types/src/streaming_config.rs index dba4286c..9eb59b98 100644 --- a/crates/asap_types/src/streaming_config.rs +++ b/crates/asap_types/src/streaming_config.rs @@ -6,12 +6,10 @@ use std::fs::File; use std::io::BufReader; use std::ops::Index; -use crate::aggregation_config::{AggregationConfig, AggregationIdInfo}; -use crate::capability_matching::find_compatible_aggregation as common_find_compatible; +use crate::aggregation_config::AggregationConfig; use crate::capability_matching::StorageBackend; use crate::enums::QueryLanguage; use crate::policy_registry::PolicyRegistry; -use crate::query_requirements::QueryRequirements; /// One continuous-monitoring (CDM) threshold spec. The data-plane monitor /// coordinator owns the AUTHORITATIVE `tau`/`epsilon`/`window_ms` (the edge @@ -189,17 +187,6 @@ impl StreamingConfig { } } -impl StreamingConfig { - /// Find a compatible aggregation for the given requirements using capability-based matching. - /// Delegates to `asap_types::find_compatible_aggregation`. - pub fn find_compatible_aggregation( - &self, - requirements: &QueryRequirements, - ) -> Option { - common_find_compatible(&self.aggregation_configs, requirements) - } -} - impl Index for StreamingConfig { type Output = AggregationConfig;