feat(sketch_algebra): unify SketchKind onto asap_sketch::SummaryKind

control_plane/Cargo.toml

@@ -40,13 +40,14 @@ asap_types.workspace = true
 # tagged releases yet. Re-pin as ASAPController's IR evolves; move to a tag
 # once one exists.
 #
-# Bumped to 283b6ab (merge of PR #141, "convert Implementation::is_satisfied_by
-# into a Matcher trait") to pick up asap-plan::Matcher/Implementation and
-# asap-sketch::{SummaryKind, SummaryParams} for the sketch-identity
-# unification work (see scratchpad/artifacts/enum-unification-plan.md).
-asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "283b6ab97159233e2dfc221e575415973c73aaa7" }
-asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "283b6ab97159233e2dfc221e575415973c73aaa7" }
-asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "283b6ab97159233e2dfc221e575415973c73aaa7" }
+# Bumped to 150ef7d (merge of PR #142, "derive PartialOrd/Ord for
+# SummaryKind") for the sketch-identity unification work (see
+# scratchpad/artifacts/enum-unification-plan.md) -- SummaryKind needs Ord
+# for the BTreeSet<SummaryKind> deterministic-emission-order contract in
+# control_plane::physical::colored_dag::emitter::EdgeStageConfig::metric_to_family.
+asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "150ef7d0786d24286b578dfae9dbbcefc8fbac3e" }
+asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "150ef7d0786d24286b578dfae9dbbcefc8fbac3e" }
+asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "150ef7d0786d24286b578dfae9dbbcefc8fbac3e" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/src/emit/backend_push.rs

@@ -658,14 +658,14 @@ mod tests {
         use crate::physical::colored_dag::emitter::{
             AggregationInput, BackendAggregation, BackendReadout,
         };
-        use crate::sketch_algebra::params::{DDSketchParams, SketchKind, SketchParams};
         use crate::sketch_algebra::physical_expr::EstimateOp;
+        use asap_sketch::{SummaryKind, SummaryParams};
         BackendStageConfig {
             aggregations: vec![BackendAggregation {
                 aggregation_id: agg_id.to_string(),
                 metric_name: metric.to_string(),
-                sketch_kind: SketchKind::DDSketch,
-                sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
+                sketch_kind: SummaryKind::DDSketch,
+                sketch_params: SummaryParams::DDSketch { alpha: 0.01 },
                 grouping: vec![],
                 item_label: None,
                 spatial_filter: String::new(),

control_plane/src/emit/mod.rs

@@ -51,10 +51,10 @@ pub use trait_def::{
 pub use crate::workload::WorkloadRegistry;
 
 use crate::physical::colored_dag::emitter::EdgeStageConfig;
-use crate::sketch_algebra::params::SketchKind;
 use crate::sketch_algebra::PhysicalExpr;
 use crate::store::WorkloadStore;
 use anyhow::Result;
+use asap_sketch::SummaryKind;
 
 /// Phase ε.1.5 — which edge runtime an agent identifies as.
 ///
@@ -290,7 +290,7 @@ fn apply_cold_format_from_env(edge_cfg: &mut EdgeStageConfig) {
 /// (`Logical`-only, unresolved `Ref`, raw Mode-3 archive). These map
 /// onto the raw-passthrough default pipeline in the routing emitter,
 /// which is correct.
-pub fn extract_root_sketch_kind(expr: &PhysicalExpr) -> Option<SketchKind> {
+pub fn extract_root_sketch_kind(expr: &PhysicalExpr) -> Option<SummaryKind> {
     match expr {
         PhysicalExpr::SketchAgg { sketch_type, .. } => Some(sketch_type.clone()),
         PhysicalExpr::RawAtEdgeSketchAtBackend { family, .. } => Some(family.clone()),
@@ -324,7 +324,7 @@ pub fn extract_root_sketch_kind(expr: &PhysicalExpr) -> Option<SketchKind> {
 /// (`quantile_over_time` → DDSketch, `count`-distinct → HLL, `topk` →
 /// CountSketch, …). We therefore collect the UNION of every workload
 /// entry's committed sketch family per metric into a
-/// `BTreeSet<SketchKind>` (deterministic order). The emitter routes the
+/// `BTreeSet<SummaryKind>` (deterministic order). The emitter routes the
 /// metric to EACH family in its set and prunes pipelines/processors to
 /// the union of all sets — eliminating the prior all-5 fan-out that
 /// shipped sketch state through every family regardless of need.
@@ -344,8 +344,8 @@ pub fn extract_root_sketch_kind(expr: &PhysicalExpr) -> Option<SketchKind> {
 pub fn collect_metric_to_family(
     registry: &WorkloadRegistry,
     workload_store: &WorkloadStore,
-) -> std::collections::HashMap<String, std::collections::BTreeSet<SketchKind>> {
-    let mut out: std::collections::HashMap<String, std::collections::BTreeSet<SketchKind>> =
+) -> std::collections::HashMap<String, std::collections::BTreeSet<SummaryKind>> {
+    let mut out: std::collections::HashMap<String, std::collections::BTreeSet<SummaryKind>> =
         std::collections::HashMap::new();
     for entry in registry.entries() {
         // B2 (metric, role) restructure: walk EVERY role registered for
@@ -733,17 +733,16 @@ mod runtime_tests {
         let _env = crate::test_support::env_lock();
         use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, ExportTarget};
         use crate::physical::colored_dag::stage_id::StageId;
-        use crate::sketch_algebra::params::SketchKind;
-        use crate::sketch_algebra::params::{DDSketchParams, SketchParams};
+        use asap_sketch::SummaryParams;
 
         let cfg = EdgeStageConfig {
             source_metric: Some("m".to_string()),
             label_filters: Vec::new(),
             window_secs: Some(60),
             sketch_processors: vec![EdgeSketchProcessor {
                 processor_name: "ddsketch".to_string(),
-                sketch_kind: SketchKind::DDSketch,
-                sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
+                sketch_kind: SummaryKind::DDSketch,
+                sketch_params: SummaryParams::DDSketch { alpha: 0.01 },
                 aggregation_id: "agg0".to_string(),
             }],
             exporter_target: ExportTarget::Stage(StageId::Gateway),
@@ -919,7 +918,7 @@ mod runtime_tests {
 
     #[test]
     fn collect_metric_to_family_binds_all_six_contract_metrics_from_live_yaml() {
-        use crate::sketch_algebra::params::SketchKind;
+        use asap_sketch::SummaryKind;
 
         // The 6 contract metrics reproduced inline (mirrors
         // deploy/configs/mvp-workload.yaml entries 1, 5, 6, 7, 8 plus the
@@ -973,27 +972,27 @@ mod runtime_tests {
         // metric needs. For THIS workload every sketched metric is
         // queried by exactly one capability, so each set has size 1.
         use std::collections::BTreeSet;
-        let expected: Vec<(&str, Option<BTreeSet<SketchKind>>)> = vec![
+        let expected: Vec<(&str, Option<BTreeSet<SummaryKind>>)> = vec![
             (
                 "http_latency_ms",
-                Some(BTreeSet::from([SketchKind::DDSketch])),
+                Some(BTreeSet::from([SummaryKind::DDSketch])),
             ),
             ("http_requests_total", None), // raw passthrough
             (
                 "request_size_bytes",
-                Some(BTreeSet::from([SketchKind::Kll])),
+                Some(BTreeSet::from([SummaryKind::Kll])),
             ),
             (
                 "unique_users_per_min",
-                Some(BTreeSet::from([SketchKind::Hll])),
+                Some(BTreeSet::from([SummaryKind::Hll])),
             ),
             (
                 "top_endpoint_qps",
-                Some(BTreeSet::from([SketchKind::CountSketch])),
+                Some(BTreeSet::from([SummaryKind::CountSketchWithHeap])),
             ),
             (
                 "endpoint_request_freq",
-                Some(BTreeSet::from([SketchKind::Cms])),
+                Some(BTreeSet::from([SummaryKind::Cms])),
             ),
         ];
         for (metric, want) in &expected {
@@ -1122,9 +1121,9 @@ mod runtime_tests {
     /// analyzer's query-string → AggType parsing.
     #[test]
     fn collect_metric_to_family_unions_multiple_capabilities_per_metric() {
-        use crate::sketch_algebra::params::SketchKind;
         use crate::types::{AggType, QueryWorkload, SketchType, WorkloadCharacteristics};
         use crate::workload::AggRole;
+        use asap_sketch::SummaryKind;
         use std::collections::BTreeSet;
         use std::time::Duration;
 
@@ -1193,7 +1192,7 @@ mod runtime_tests {
             .unwrap_or_else(|| panic!("http_requests must be in the map\nmap: {map:?}"));
         assert_eq!(
             got,
-            BTreeSet::from([SketchKind::DDSketch, SketchKind::Hll, SketchKind::Cms]),
+            BTreeSet::from([SummaryKind::DDSketch, SummaryKind::Hll, SummaryKind::Cms]),
             "a metric queried by 3 capabilities must accumulate 3 families (UNION, not first-wins)\nmap: {map:?}"
         );
     }
@@ -1260,7 +1259,7 @@ mod runtime_tests {
         let _env = crate::test_support::env_lock();
         use crate::physical::colored_dag::emitter::{EdgeStageConfig, ExportTarget};
         use crate::physical::colored_dag::stage_id::StageId;
-        use crate::sketch_algebra::params::SketchKind;
+        use asap_sketch::SummaryKind;
 
         let yaml = r#"
 - metric_name: http_requests_total_latency_ms
@@ -1286,7 +1285,7 @@ mod runtime_tests {
             warm_passthrough_metrics: Vec::new(),
             metric_to_family: std::collections::HashMap::from([(
                 "http_requests_total_latency_ms".to_string(),
-                std::collections::BTreeSet::from([SketchKind::DDSketch]),
+                std::collections::BTreeSet::from([SummaryKind::DDSketch]),
             )]),
             metric_to_grouping_labels: std::collections::HashMap::new(),
             cumulative_counter_metrics: Vec::new(),

control_plane/src/emit/otap.rs

@@ -40,7 +40,7 @@ use std::collections::BTreeMap;
 
 use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, EdgeStageConfig, ExportTarget};
 use crate::physical::colored_dag::stage_id::StageId;
-use crate::sketch_algebra::params::{SketchKind, SketchParams};
+use asap_sketch::{SummaryKind, SummaryParams};
 
 /// Default URL for Prometheus's native OTLP HTTP receiver.
 /// Matches `super::stage_config::emit_edge_yaml`'s placeholder so the
@@ -296,39 +296,68 @@ fn build_asap_sketches_config(sp: &EdgeSketchProcessor, window_secs: Option<u64>
         Value::String(sketch_kind_tag(&sp.sketch_kind).into()),
     );
     match &sp.sketch_params {
-        SketchParams::Kll(p) => {
-            m.insert("k".into(), Value::Number((p.k as u64).into()));
+        SummaryParams::Kll { k } => {
+            m.insert("k".into(), Value::Number((*k as u64).into()));
         }
-        SketchParams::DDSketch(p) => {
-            m.insert("relative_accuracy".into(), Value::Number(p.alpha.into()));
+        SummaryParams::DDSketch { alpha } => {
+            m.insert("relative_accuracy".into(), Value::Number((*alpha).into()));
             m.insert("delta_transmission".into(), Value::Bool(true));
         }
-        SketchParams::Hll(_p) => {
+        SummaryParams::Hll { .. } => {
             m.insert("delta_transmission".into(), Value::Bool(true));
         }
-        SketchParams::Cms(p) => {
-            m.insert("rows".into(), Value::Number((p.d as u64).into()));
-            m.insert("columns".into(), Value::Number((p.w as u64).into()));
+        // Heap-bearing width/depth extraction is identical to the bare
+        // kind — this path never distinguished `with_heap` even before
+        // `SummaryKind` split it into its own variant (heap_size wasn't
+        // emitted here either way).
+        SummaryParams::Cms { width, depth } | SummaryParams::CmsWithHeap { width, depth, .. } => {
+            m.insert("rows".into(), Value::Number((*depth as u64).into()));
+            m.insert("columns".into(), Value::Number((*width as u64).into()));
             m.insert("delta_transmission".into(), Value::Bool(true));
         }
-        SketchParams::CountSketch(p) => {
-            let epsilon = std::f64::consts::E / (p.w as f64);
-            let delta = 2f64.powi(-(p.d as i32));
+        SummaryParams::CountSketch { width, depth }
+        | SummaryParams::CountSketchWithHeap { width, depth, .. } => {
+            let epsilon = std::f64::consts::E / (*width as f64);
+            let delta = 2f64.powi(-(*depth as i32));
             m.insert("epsilon".into(), Value::Number(epsilon.into()));
             m.insert("delta".into(), Value::Number(delta.into()));
             m.insert("delta_transmission".into(), Value::Bool(true));
         }
+        SummaryParams::Sum
+        | SummaryParams::Count
+        | SummaryParams::MinMax
+        | SummaryParams::Increase
+        | SummaryParams::Rate
+        | SummaryParams::Kmv { .. }
+        | SummaryParams::Theta { .. } => {
+            unreachable!(
+                "edge sketch processor config requested for a non-sketch or unsupported \
+                 SummaryKind; no Bind* rule in this repo produces one"
+            )
+        }
     }
     Value::Mapping(m)
 }
 
-fn sketch_kind_tag(kind: &SketchKind) -> &'static str {
+fn sketch_kind_tag(kind: &SummaryKind) -> &'static str {
     match kind {
-        SketchKind::Kll => "kll",
-        SketchKind::DDSketch => "ddsketch",
-        SketchKind::Hll => "hll",
-        SketchKind::Cms => "cms",
-        SketchKind::CountSketch => "count_sketch",
+        SummaryKind::Kll => "kll",
+        SummaryKind::DDSketch => "ddsketch",
+        SummaryKind::Hll => "hll",
+        SummaryKind::Cms | SummaryKind::CmsWithHeap => "cms",
+        SummaryKind::CountSketch | SummaryKind::CountSketchWithHeap => "count_sketch",
+        SummaryKind::Sum
+        | SummaryKind::Count
+        | SummaryKind::MinMax
+        | SummaryKind::Increase
+        | SummaryKind::Rate
+        | SummaryKind::Kmv
+        | SummaryKind::Theta => {
+            unreachable!(
+                "edge sketch processor config requested for a non-sketch or unsupported \
+                 SummaryKind; no Bind* rule in this repo produces one"
+            )
+        }
     }
 }
 
@@ -338,7 +367,7 @@ fn sketch_kind_tag(kind: &SketchKind) -> &'static str {
 mod tests {
     use super::*;
     use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric};
-    use crate::sketch_algebra::params::DDSketchParams;
+    use asap_sketch::{SummaryKind, SummaryParams};
 
     /// Minimal struct-stub used to validate the emitted DAG parses as the
     /// otap-dataflow schema. We don't pull in the otap-df-config crate
@@ -385,8 +414,8 @@ mod tests {
             window_secs: Some(60),
             sketch_processors: vec![EdgeSketchProcessor {
                 processor_name: "ddsketch".to_string(),
-                sketch_kind: SketchKind::DDSketch,
-                sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }),
+                sketch_kind: SummaryKind::DDSketch,
+                sketch_params: SummaryParams::DDSketch { alpha: 0.01 },
                 aggregation_id: "agg0".to_string(),
             }],
             exporter_target: ExportTarget::Stage(StageId::Gateway),