diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs index 08038fd2..d2da4566 100644 --- a/control_plane/src/asap_tier_analysis.rs +++ b/control_plane/src/asap_tier_analysis.rs @@ -241,9 +241,26 @@ pub fn analyze_promql_for_asap_tier(metricsql: &str) -> ASAPTierAnalysis { for intent in &intents { match capability_for(intent) { Some(cap) => { + // For a heavy-hitter top-k (`FrequencyTopk`), the inner + // `by (item)` labels (e.g. `topk(k, sum by (host) (m))`) + // are the heap's RANKED dimension — recorded as the sid's + // item_label and projected OUT of the series key into the + // top-k heap, NOT a series grouping key. The sketch is + // grouped by its OWN grouping_labels (e.g. zone) with the + // item in the heap, so requiring the item dimension in the + // sid's group_by_keys would never match (item ∉ {zone}) and + // the topk query falls through to archive. Match the + // metric's FrequencyTopk sids by metric + capability instead + // (empty required keys ⊆ any grouping); the reducer reads + // each matched sid's heap. + let candidate_keys = if matches!(cap, Capability::FrequencyTopk(_)) { + BTreeSet::new() + } else { + group_by_keys.clone() + }; out.candidates.push(ASAPTierCandidate { metric_name: metric_name.clone(), - group_by_keys: group_by_keys.clone(), + group_by_keys: candidate_keys, required_capability: cap, function: trace.function.clone(), function_args: trace.function_args.clone(), diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 02b45bb4..567f8145 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -1864,8 +1864,29 @@ fn emit_edge_yaml_asap_edge( .cloned() .unwrap_or_default(); if !grouping.is_empty() { - let by: Vec = grouping.into_iter().map(Value::String).collect(); - e.insert("aggregate_by".into(), Value::Sequence(by)); + // Exclude the item_label (the inner heavy-hitter dimension + // for HLL / CMS / heap-bearing CountSketch) from + // aggregate_by: it is the sketch SUBJECT — hashed into the + // sketch / fed to the top-k heap — NOT a series grouping + // key. A query like `topk(10, sum by (host) (m))` lands + // `host` in grouping_labels, but for a heap-bearing + // CountSketch `host` is the item_label; leaving it in + // aggregate_by keys the edge series PER host (one series + + // heap per host — a cardinality explosion) instead of one + // heap per group. The agent observe path already projects + // item_label out of the series key for the item-keyed + // families, so the two layers must agree. No-op for metrics + // without an item_label or whose item_label isn't a + // grouping label. + let item_label = cfg.metric_to_item_label.get(*metric); + let by: Vec = grouping + .into_iter() + .filter(|k| item_label.map(|il| il != k).unwrap_or(true)) + .map(Value::String) + .collect(); + if !by.is_empty() { + e.insert("aggregate_by".into(), Value::Sequence(by)); + } } // Family-specific params — mirror the reads in // `build_edge_processor_block`. The fused processor's @@ -6594,4 +6615,49 @@ mod tests { "an enumerated CountSketch with with_heap=true must emit emit_heap:\n{yaml}" ); } + + /// Regression for the top-k cardinality explosion (#5): the item_label + /// (heavy-hitter dimension, e.g. `host` from `topk(.., sum by (host)(m))`) + /// must NOT appear in `aggregate_by` — it is the sketch/heap SUBJECT, not + /// a series grouping key. Leaving it in keyed the edge series per-host + /// (one series + heap per host) instead of one heap per group. + #[test] + fn fused_emit_excludes_item_label_from_aggregate_by() { + let _env = crate::test_support::EnvVarGuard::set("ASAP_EDGE_FUSED", "1"); + let mut cfg = fused_cfg_countsketch_no_processor(); + // grouping_labels carries BOTH the real grouping key (zone) AND the + // item_label (host) — as a `topk(10, sum by (host)(m))` workload with + // grouping_labels:[zone] produces after the query's `by (host)` is + // folded in. + cfg.metric_to_grouping_labels.insert( + "endpoint_request_freq".into(), + vec!["host".to_string(), "zone".to_string()], + ); + cfg.metric_to_item_label + .insert("endpoint_request_freq".into(), "host".to_string()); + let yaml = emit_edge_yaml_asap_edge(&cfg, "ws://c/", "agent-1") + .expect("fused emit ok"); + let doc: serde_yaml::Value = serde_yaml::from_str(&yaml) + .unwrap_or_else(|e| panic!("emitted YAML must parse: {e}\n{yaml}")); + let metrics = doc + .get("processors") + .and_then(|p| p.get("asap_edge")) + .and_then(|p| p.get("metrics")) + .and_then(|m| m.as_sequence()) + .expect("metrics list present"); + let entry = metrics + .iter() + .find(|e| e.get("metric").and_then(|v| v.as_str()) == Some("endpoint_request_freq")) + .expect("endpoint_request_freq entry present"); + let by: Vec<&str> = entry + .get("aggregate_by") + .and_then(|v| v.as_sequence()) + .map(|s| s.iter().filter_map(|v| v.as_str()).collect()) + .unwrap_or_default(); + assert_eq!( + by, + vec!["zone"], + "item_label `host` must be excluded from aggregate_by (got {by:?})\n{yaml}" + ); + } } diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs index aae2a5bf..884563d0 100644 --- a/control_plane/src/optimizer/rules/mod.rs +++ b/control_plane/src/optimizer/rules/mod.rs @@ -102,13 +102,40 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option (StatisticClass::Quantile, AccuracyPreference::RelativeError), AggType::Cardinality => (StatisticClass::Cardinality, AccuracyPreference::default()), AggType::Frequency => (StatisticClass::Frequency, AccuracyPreference::default()), }); + // An explicit `sketch_family_override` is authoritative for the FAMILY + // — and therefore for the STATISTIC CLASS it answers. The query-derived + // statistic above only covers Quantile/Cardinality/Frequency from + // `AggType`; a `count(...)` / `topk(...)` / `count_over_time(...)` + // query can classify as the wrong class, so without this an HLL / + // CountMinSketch / CountSketch override would mismatch the derived + // statistic, be rejected by `is_valid_pair` below, and silently fall + // back to the catalog default (DDSketch) — the controller would then + // emit `family: ddsketch` for an HLL/CMS/CountSketch metric. Re-derive + // the statistic from the override whenever the derived one is + // incompatible, so the override drives both family and statistic. + if let Some(st) = w.sketch_type_override.as_ref() { + let ov = SketchKind::from(st.clone()); + if !is_valid_pair(ov.clone(), statistic) { + let (s, ap) = match ov { + SketchKind::DDSketch | SketchKind::Kll => { + (StatisticClass::Quantile, AccuracyPreference::RelativeError) + } + SketchKind::Hll => (StatisticClass::Cardinality, AccuracyPreference::default()), + SketchKind::Cms => (StatisticClass::Frequency, AccuracyPreference::default()), + SketchKind::CountSketch => (StatisticClass::TopK, AccuracyPreference::default()), + }; + statistic = s; + accuracy_pref = ap; + } + } + // SumRateCount → no sketch (raw passthrough). Decline the typed // binding so the caller falls back to the legacy raw plan. if statistic == StatisticClass::SumRateCount { @@ -478,6 +505,37 @@ mod tests { assert_eq!(plan.agent_config.sketch_type, SketchType::CountSketch); } + /// Regression: an explicit `sketch_family_override` must pin the family + /// (and statistic) even when the query's `AggType` classifies as a + /// different/incompatible class. Without the override re-deriving the + /// statistic, `is_valid_pair` rejected HLL/CMS/CountSketch overrides + /// against a Quantile-classified query and fell back to DDSketch, so + /// the controller emitted `family: ddsketch` for those metrics. + #[test] + fn override_pins_nonquantile_family_over_misclassified_query() { + use crate::emit::extract_root_sketch_kind; + use crate::sketch_algebra::params::SketchKind; + for (ov, expect) in [ + (SketchType::DDSketch, SketchKind::DDSketch), + (SketchType::KLL, SketchKind::Kll), + (SketchType::HLL, SketchKind::Hll), + (SketchType::CountMinSketch, SketchKind::Cms), + (SketchType::CountSketch, SketchKind::CountSketch), + ] { + // Query classifies as Quantile (the mis-derived case observed + // live for count()/topk()/count_over_time()); the override must win. + let mut w = workload(vec![AggType::Quantile]); + w.sketch_type_override = Some(ov.clone()); + let pe = bind_workload_typed(&w) + .unwrap_or_else(|| panic!("bind declined for override {ov:?}")); + assert_eq!( + extract_root_sketch_kind(&pe), + Some(expect.clone()), + "override {ov:?} should pin family {expect:?}, not fall back to DDSketch", + ); + } + } + #[test] fn quantile_priority_wins() { let plan = diff --git a/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs b/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs index 64582b6b..579d68d1 100644 --- a/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs +++ b/control_plane/src/sketch_algebra/rules/bind_cms_topk.rs @@ -73,7 +73,13 @@ impl Rule for BindCountSketchOnTopK { let w = (std::f64::consts::E / eps).ceil() as u32; let d = (1.0 / delta).ln().ceil() as u32; - let w = w.max(2); + // CountSketch columns MUST be a power of two: the agent + // (asapedgeprocessor config_validate) rejects non-pow2 cols because + // sketchlib bit-slices the hash with a pow2 column mask. Round the + // ε-derived width UP to the next power of two — this only tightens + // the additive bound (ε ≤ e/w) and prevents an agent-side + // "cols must be a power of two" crash on config apply. + let w = w.max(2).next_power_of_two(); let d = d.max(1); Some(PhysicalExpr::estimate_over_agg(