From eb30dd3b5248e760e423f364270809518d7b849e Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 29 May 2026 08:32:14 -0600 Subject: [PATCH 1/3] control_plane: honor sketch_family_override for HLL/CMS/CountSketch in the agent emit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `bind_workload_typed` derived the statistic class purely from the query's AggType (Quantile/Cardinality/Frequency). When a workload entry pinned a non-quantile `sketch_family_override` (HLL / CountMinSketch / CountSketch) but its query classified as a different/incompatible class (e.g. `count(...)`, `count_over_time(...)`, `topk(...)` landing on Quantile), `is_valid_pair(override, statistic)` rejected the override and the binder fell back to the catalog default — DDSketch. The controller therefore emitted `family: ddsketch` to the fused asap_edge agent for every HLL/CMS/CountSketch metric, so those families never ran at the edge and their warm queries couldn't resolve. Fix: an explicit override is authoritative for the family AND the statistic it answers. When the query-derived statistic is incompatible with the override, re-derive the statistic from the override (HLL→Cardinality, CountMinSketch→Frequency, CountSketch→TopK, DDSketch/KLL→Quantile) so the override drives both. DDSketch/KLL quantile paths are unchanged. Verified live (controller-driven multinode, multi-sketch workload): the agent's effective config now carries family: hll / countminsketch / countsketch (+ item_label / emit_heap) for the respective metrics instead of all-ddsketch. Adds a regression unit test. Co-Authored-By: Claude Opus 4.8 (1M context) --- control_plane/src/optimizer/rules/mod.rs | 60 +++++++++++++++++++++++- 1 file changed, 59 insertions(+), 1 deletion(-) 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 = From c49e64de8b8860bfb547b5f265124755c628063b Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 29 May 2026 10:35:33 -0600 Subject: [PATCH 2/3] control_plane: keep item_label out of aggregate_by + pow2 CountSketch cols (fix topk explosion + crash) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fixes that make the heap-bearing CountSketch (warm topk) actually deployable end-to-end: 1. Exclude item_label from aggregate_by (emit_edge_yaml_asap_edge). A `topk(10, sum by (host) (m))` workload with grouping_labels:[zone] folds `host` into grouping_labels, so the emitter wrote aggregate_by:[host, zone]. But `host` is the item_label — the heavy-hitter dimension fed to the top-k heap, NOT a series grouping key. Leaving it in keyed the edge series PER host: ~5000 per-host sids + heaps instead of one heap per zone (cardinality explosion that also collapsed agent throughput ~100x). Now the emitted aggregate_by drops the item_label → 4 per-zone CountSketchWithHeap sids. Verified live: gct_ms_topk registers {['zone']: 4} (was 4598). 2. Round CountSketch cols up to the next power of two (BindCountSketchOnTopK). The ε-derived width `ceil(e/eps)` (e.g. 55 for eps=0.05) is not a power of two, but the agent's asapedgeprocessor config_validate REJECTS non-pow2 countsketch cols (sketchlib bit-slices the hash with a pow2 column mask) → the collector crashed on config apply ("cols=55 must be a power of two"), so the agent never came up. Rounding up only tightens the bound (ε ≤ e/w). Verified live: cols=64, agent starts clean. Adds regression tests: fused_emit_excludes_item_label_from_aggregate_by. Co-Authored-By: Claude Opus 4.8 (1M context) --- control_plane/src/emit/stage_config.rs | 70 ++++++++++++++++++- .../src/sketch_algebra/rules/bind_cms_topk.rs | 8 ++- 2 files changed, 75 insertions(+), 3 deletions(-) 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/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( From ab93137cb638ec9d453df9395405402fa51f0bff Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 29 May 2026 12:02:36 -0600 Subject: [PATCH 3/3] control_plane: match FrequencyTopk sids by metric+capability, not item dim in group_by_keys A topk query `topk(k, sum by (item) (m))` lowers to a FrequencyTopk candidate whose group_by_keys includes the inner `by (item)` dim. But for a heap-bearing CountSketch the item is the heap's ranked dimension (the sid's item_label, projected OUT of the series key); the sid is grouped by its own grouping_labels (e.g. zone). Requiring the item in the sid's group_by_keys never matches (item not subset of {zone}) so the topk query fell through to archive and returned empty. Clear group_by_keys for FrequencyTopk candidates so the metric's heap-bearing sids match by metric+capability; the reducer reads each heap. Co-Authored-By: Claude Opus 4.8 (1M context) --- control_plane/src/asap_tier_analysis.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) 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(),