control_plane: honor sketch_family_override for HLL/CMS/CountSketch in agent emit

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(),

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<Value> = 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<Value> = 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}"
+        );
+    }
 }

control_plane/src/optimizer/rules/mod.rs

@@ -102,13 +102,40 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option<crate::sketch_algebra::P
     // Priority: workload-spec metric-name match → AggType-driven
     // default. The metric-name match owns the demo contract rows; the
     // AggType fallback covers everything else.
-    let (statistic, accuracy_pref) =
+    let (mut statistic, mut accuracy_pref) =
         classify_demo_metric(&w.metric_name).unwrap_or_else(|| match w.aggregations[0] {
             AggType::Quantile => (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 =

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(