From 5b81cc7d39b90bb9aa11e34ddf41b55a20f67071 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Mon, 7 Sep 2026 09:26:14 -0600
Subject: [PATCH 1/6] fix(topk): separate weighted materializations

---
 control_plane/src/physical/compiler.rs        |  4 +-
 .../drivers/ingest/prometheus_remote_write.rs | 14 ++-
 data_plane/src/precompute_engine/worker.rs    | 14 ++-
 .../storage_engines/sketch_db/index/mod.rs    |  6 +-
 .../asapquery_compatibility_process_e2e.rs    | 92 +++++++++++++++++--
 ...asapquery-compatibility-demo-snapshot.json |  8 +-
 6 files changed, 116 insertions(+), 22 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 84f45d2d..a10489f5 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -2822,8 +2822,8 @@ mod tests {
             "increase(asap_demo_counter_total[5s])",
             "sum_over_time(asap_demo_gauge[5s])",
             "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
-            "topk(5, sum_over_time(asap_demo_gauge[5s]))",
-            "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))",
+            "topk(1, sum_over_time(asap_demo_gauge[5s]))",
+            "topk(1, count_over_time(asap_demo_gauge[5s]))",
         ] {
             assert!(plan.query_plan.lookup(query).is_ok(), "missing {query}");
         }
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index 07901031..a49eb865 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -462,12 +462,16 @@ fn route_messages(
                 .collect();
             let attrs_fp = super::canonical_attrs_fingerprint(&grouping_pairs);
             let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(config);
-            let sid = ingest.series_resolver.resolve(
-                &config.metric,
-                &attrs_fp,
-                &agg_kind.canonical_string(),
-            );
             let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64());
+            // A sketch family is not a complete physical identity. Two
+            // materializations may use the same family and grouping while
+            // differing in update semantics (for example count- versus
+            // value-weighted Top-K). Keep those states on distinct SIDs.
+            let materialization_kind = format!("{}|{}", agg_kind.canonical_string(), policy_fp);
+            let sid =
+                ingest
+                    .series_resolver
+                    .resolve(&config.metric, &attrs_fp, &materialization_kind);
             buckets
                 .entry(sid)
                 .or_insert_with(|| ((sid, policy_fp, group_key), Vec::new()))
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 37aa7009..b64b2ad7 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -1199,7 +1199,19 @@ fn apply_sample(
     config: &AggregationConfig,
 ) {
     if updater.is_keyed() {
-        let key = extract_aggregated_key_from_series(series_key, config);
+        // Planner's PromQL Top-K item is the series identity. When no
+        // explicit aggregated labels are projected, retain the canonical
+        // series key instead of collapsing every series onto an empty item.
+        let key = if config.aggregated_labels.labels.is_empty()
+            && matches!(
+                config.aggregation_type,
+                crate::storage_engines::types::AggregationType::CountMinSketchWithHeap
+                    | crate::storage_engines::types::AggregationType::CountSketchWithHeap
+            ) {
+            KeyByLabelValues::new_with_labels(vec![series_key.to_string()])
+        } else {
+            extract_aggregated_key_from_series(series_key, config)
+        };
         updater.update_keyed(&key, val, ts);
     } else {
         updater.update_single(val, ts);
diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs
index 02f02804..b85c0524 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -1862,7 +1862,11 @@ impl SketchStore {
         // resolver type (which lives in `drivers::ingest`). Tests
         // pass either a real local resolver or a counter-mock
         // closure.
-        let agg_kind_canonical = agg_kind.canonical_string();
+        let agg_kind_canonical = format!(
+            "{}|{}",
+            agg_kind.canonical_string(),
+            agg_cfg.policy_fingerprint()
+        );
         let sid = mint_sid(&agg_cfg.metric, &attrs_fp, &agg_kind_canonical);
         self.ingest_precompute_with_sid(sid, agg_cfg, output, accumulator)
     }
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index 3fc29c3d..a87c0e47 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -288,6 +288,20 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
                     (base + 9_400, 8.0),
                 ],
             ),
+            series_with_labels(
+                "asap_demo_gauge",
+                &[("job", "worker")],
+                &[(base + 700, 100.0), (base + 3_100, 100.0)],
+            ),
+            series_with_labels(
+                "asap_demo_gauge",
+                &[("job", "cron")],
+                &[
+                    (base + 900, 10.0),
+                    (base + 2_100, 10.0),
+                    (base + 3_700, 10.0),
+                ],
+            ),
             series(
                 "asap_demo_latency_ms",
                 &[
@@ -369,7 +383,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let topk_sum = wait_for_warm_instant(
         &client,
         &backend,
-        "topk(5, sum_over_time(asap_demo_gauge[5s]))",
+        "topk(1, sum_over_time(asap_demo_gauge[5s]))",
         first_eval,
         &backend_log,
     )
@@ -377,17 +391,38 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let topk_count = wait_for_warm_instant(
         &client,
         &backend,
-        "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))",
+        "topk(1, count_over_time(asap_demo_gauge[5s]))",
         first_eval,
         &backend_log,
     )
     .await;
-    assert!(first_value(&topk_sum, "value").is_some());
-    assert!(first_value(&topk_count, "value").is_some());
+    assert_eq!(
+        first_value(&topk_sum, "value"),
+        Some(200.0),
+        "value-weighted Top-K must select worker: {topk_sum}"
+    );
+    assert_eq!(
+        first_value(&topk_count, "value"),
+        Some(4.0),
+        "count-weighted Top-K must select api: {topk_count}"
+    );
+    assert_eq!(topk_sum["data"]["result"].as_array().map(Vec::len), Some(1));
+    assert_eq!(
+        topk_count["data"]["result"].as_array().map(Vec::len),
+        Some(1)
+    );
+    assert_eq!(
+        topk_sum["data"]["result"][0]["metric"]["item"],
+        "asap_demo_gauge{job=\"worker\"}"
+    );
+    assert_eq!(
+        topk_count["data"]["result"][0]["metric"]["item"],
+        "asap_demo_gauge{job=\"api\"}"
+    );
     let rate_value = first_value(&rate, "value").expect("rate value");
     let increase_value = first_value(&increase, "value").expect("increase value");
     assert!((rate_value * 5.0 - increase_value).abs() < 1e-9);
-    assert!((first_value(&sum, "value").expect("sum value") - 10.0).abs() < 1e-9);
+    assert!((first_value(&sum, "value").expect("sum value") - 240.0).abs() < 1e-9);
     let quantile_value = first_value(&quantile, "value").expect("quantile value");
     assert!(
         (19.0..=31.0).contains(&quantile_value),
@@ -399,8 +434,8 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         "increase(asap_demo_counter_total[5s])",
         "sum_over_time(asap_demo_gauge[5s])",
         "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
-        "topk(5, sum_over_time(asap_demo_gauge[5s]))",
-        "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))",
+        "topk(1, sum_over_time(asap_demo_gauge[5s]))",
+        "topk(1, count_over_time(asap_demo_gauge[5s]))",
     ] {
         let response: Value = client
             .get(format!("{backend}/api/v1/query_range"))
@@ -421,6 +456,40 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
             is_warm(&response),
             "{query} did not use warm tier: {response}"
         );
+        if query.starts_with("topk(") {
+            let mut ranked_points = response["data"]["result"]
+                .as_array()
+                .unwrap_or_else(|| panic!("missing Top-K range result for {query}: {response}"))
+                .iter()
+                .flat_map(|series| {
+                    series["values"]
+                        .as_array()
+                        .into_iter()
+                        .flatten()
+                        .map(|point| {
+                            (
+                                point[0].as_f64().expect("Top-K timestamp"),
+                                point[1]
+                                    .as_str()
+                                    .expect("Top-K value")
+                                    .parse::<f64>()
+                                    .expect("numeric Top-K value"),
+                            )
+                        })
+                })
+                .collect::<Vec<_>>();
+            ranked_points.sort_by(|left, right| left.0.total_cmp(&right.0));
+            let expected = if query.contains("sum_over_time") {
+                vec![(first_eval, 200.0), (second_eval, 26.0)]
+            } else {
+                vec![(first_eval, 4.0), (second_eval, 4.0)]
+            };
+            assert_eq!(
+                ranked_points, expected,
+                "wrong Top-K windows for {query}: {response}"
+            );
+            continue;
+        }
         let values = response["data"]["result"][0]["values"]
             .as_array()
             .unwrap_or_else(|| panic!("missing range values for {query}: {response}"));
@@ -429,6 +498,11 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         assert_eq!(values[1][0], second_eval);
     }
 
+    // Readiness polling may briefly reach the exact fallback before a newly
+    // closed warm window is visible. Every planned query above was required
+    // to converge to a warm answer; isolate the explicit fallback assertions.
+    fallback_calls.lock().await.clear();
+
     let fallback_instant: Value = client
         .get(format!("{backend}/api/v1/query"))
         .header("authorization", "Bearer demo")
@@ -515,7 +589,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         .await
         .expect("metrics body");
     assert!(metrics.contains("asap_remote_write_requests_total 4"));
-    assert!(metrics.contains("asap_remote_write_samples_total 27"));
-    assert!(metrics.contains("asap_remote_write_duplicates_total 24"));
+    assert!(metrics.contains("asap_remote_write_samples_total 32"));
+    assert!(metrics.contains("asap_remote_write_duplicates_total 29"));
     assert!(metrics.contains("asap_remote_write_rejected_requests_total 1"));
 }
diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json
index 007a486f..8aafb9f5 100644
--- a/docs/examples/asapquery-compatibility-demo-snapshot.json
+++ b/docs/examples/asapquery-compatibility-demo-snapshot.json
@@ -36,7 +36,7 @@
         "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
       },
       {
-        "query": "topk(5, sum_over_time(asap_demo_gauge[5s]))",
+        "query": "topk(1, sum_over_time(asap_demo_gauge[5s]))",
         "demand": { "fixed_interval": 1000 },
         "requirements": {
           "accuracy": { "explicit": { "EpsilonDelta": { "epsilon": 0.01, "delta": 0.01 } } },
@@ -46,7 +46,7 @@
         "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null }
       },
       {
-        "query": "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))",
+        "query": "topk(1, count_over_time(asap_demo_gauge[5s]))",
         "demand": { "fixed_interval": 1000 },
         "requirements": {
           "accuracy": { "explicit": { "EpsilonDelta": { "epsilon": 0.01, "delta": 0.01 } } },
@@ -98,14 +98,14 @@
       "weighted_cost": 1.0
     },
     "topk_evidence": {
-      "topk(5, sum_over_time(asap_demo_gauge[5s]))": {
+      "topk(1, sum_over_time(asap_demo_gauge[5s]))": {
         "selected_lower_bound": 101.0,
         "excluded_upper_bound": 100.0,
         "interval_failure_probability": 0.005,
         "observed_at_unix_ms": 9500,
         "source": "compatibility-fixture"
       },
-      "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))": {
+      "topk(1, count_over_time(asap_demo_gauge[5s]))": {
         "selected_lower_bound": 101.0,
         "excluded_upper_bound": 100.0,
         "interval_failure_probability": 0.005,

From ac4365516643cc78a3d7855ac4344b3d423704ac Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Mon, 7 Sep 2026 09:54:20 -0600
Subject: [PATCH 2/6] fix(planner): bind evidenced legacy topk workloads

---
 .../src/physical/workload_planner.rs          | 130 ++++++++++++++++--
 ...e2e_controller_plans_and_backend_serves.rs |  17 ++-
 2 files changed, 131 insertions(+), 16 deletions(-)

diff --git a/control_plane/src/physical/workload_planner.rs b/control_plane/src/physical/workload_planner.rs
index 5ab88c7f..4c16164a 100644
--- a/control_plane/src/physical/workload_planner.rs
+++ b/control_plane/src/physical/workload_planner.rs
@@ -70,7 +70,18 @@ fn mvp_deployment_policy(
 /// `PhysicalExpr` is then fed into `planner::stage_split::split_typed_three_stage`
 /// + the per-stage emitters in `config::stage_config`.
 pub fn bind_workload_typed(w: &QueryWorkload) -> Option<crate::physical::post_asap::PhysicalExpr> {
-    bind_workload_typed_with_item_filter(w, None)
+    bind_workload_typed_with_evidence(w, None, None)
+}
+
+/// Bind a legacy workload with an explicit Top-K membership certificate.
+/// Approximate Top-K fails closed through [`bind_workload_typed`] when this
+/// evidence is absent; callers that have validated a fresh certificate use
+/// this entry point instead.
+pub fn bind_workload_typed_with_topk_evidence(
+    w: &QueryWorkload,
+    evidence: &crate::physical::compiler::TopKMembershipEvidence,
+) -> Option<crate::physical::post_asap::PhysicalExpr> {
+    bind_workload_typed_with_evidence(w, None, Some(evidence))
 }
 
 /// Like [`bind_workload_typed`], but for a `Frequency` statistic, `item_filter`
@@ -87,6 +98,14 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option<crate::physical::post_as
 pub fn bind_workload_typed_with_item_filter(
     w: &QueryWorkload,
     item_filter: Option<(&str, &str)>,
+) -> Option<crate::physical::post_asap::PhysicalExpr> {
+    bind_workload_typed_with_evidence(w, item_filter, None)
+}
+
+fn bind_workload_typed_with_evidence(
+    w: &QueryWorkload,
+    item_filter: Option<(&str, &str)>,
+    topk_evidence: Option<&crate::physical::compiler::TopKMembershipEvidence>,
 ) -> Option<crate::physical::post_asap::PhysicalExpr> {
     use crate::physical::post_asap::cost_model::ForcedFamilyCostModel;
     use crate::types_v2::AccuracyTarget;
@@ -217,6 +236,12 @@ pub fn bind_workload_typed_with_item_filter(
                     nullable: false,
                     table: None,
                 },
+                Column {
+                    name: "endpoint".into(),
+                    dtype: DataType::Utf8,
+                    nullable: false,
+                    table: None,
+                },
             ],
             0,
             vec![vec![0]],
@@ -226,17 +251,39 @@ pub fn bind_workload_typed_with_item_filter(
         range: w.time_window,
         child: Box::new(scan).into(),
     };
+    // Planner's weighted Top-K contract deliberately accepts only an
+    // additive ranking input.  The legacy workload vocabulary has no TopK
+    // aggregation variant: the `top_endpoint_qps` contract row arrives as
+    // `Frequency`, meaning that each observed series occurrence contributes
+    // one to its rank.  Make that previously implicit update semantics
+    // explicit as an inner Count aggregate.  Besides satisfying the typed
+    // contract, this supplies the PromQL label-set entity identity used as
+    // the heap item.
+    let aggregate_child = if statistic == DeploymentIntent::TopK {
+        QueryExpr::Aggregate {
+            reduction: planner_types::pre_asap::Reduction::by(vec![2]),
+            measures: vec![L3AggIntent::Count {
+                accuracy: accuracy.clone(),
+            }],
+            output_names: Vec::new(),
+            having: None,
+            child: Box::new(windowed).into(),
+        }
+    } else {
+        windowed
+    };
     let aggregate = QueryExpr::Aggregate {
-        // Synthetic probe only -- `boundary::implementation_for` (what
-        // this shape actually drives) keys off `AggIntent`/accuracy alone,
-        // never `Reduction`, so this value doesn't affect the family pick.
-        // `PerEntity` is the representative choice for a windowed shape
-        // with no `by` (ASAPController#163/#165).
-        reduction: planner_types::pre_asap::Reduction::PerEntity,
+        // Top-K is a genuine full reduction over the per-item counts above;
+        // other synthetic probes retain the representative per-entity shape.
+        reduction: if statistic == DeploymentIntent::TopK {
+            planner_types::pre_asap::Reduction::by(vec![])
+        } else {
+            planner_types::pre_asap::Reduction::PerEntity
+        },
         measures: vec![intent],
         output_names: Vec::new(),
         having: None,
-        child: Box::new(windowed).into(),
+        child: Box::new(aggregate_child).into(),
     };
 
     // ── Drive the picked family directly, bypassing selection ─────────
@@ -271,7 +318,35 @@ pub fn bind_workload_typed_with_item_filter(
         other => other,
     };
     let cost_model = ForcedFamilyCostModel::new(accuracy.clone(), forced);
-    let node = crate::planner_selection::select_summary(&aggregate, &cost_model).ok()?;
+    struct Evidence<'a>(Option<&'a crate::physical::compiler::TopKMembershipEvidence>);
+    impl asap_aware_mapping::AccuracyEvidenceProvider for Evidence<'_> {
+        fn propagation_stats(
+            &self,
+            op: &planner_types::post_asap::CompositionOperator,
+            _family: &planner_types::post_asap::SummaryFamilyType,
+            _query: Option<&planner_types::post_asap::SketchQuery>,
+        ) -> asap_aware_mapping::PropagationStats {
+            match (op, self.0) {
+                (planner_types::post_asap::CompositionOperator::TopKSelection, Some(e)) => {
+                    asap_aware_mapping::PropagationStats {
+                        topk_selected_lower_bound: Some(e.selected_lower_bound),
+                        topk_excluded_upper_bound: Some(e.excluded_upper_bound),
+                        topk_interval_failure_probability: Some(e.interval_failure_probability),
+                        ..Default::default()
+                    }
+                }
+                _ => Default::default(),
+            }
+        }
+    }
+    let node = crate::planner_selection::select_summary_with_evidence(
+        &aggregate,
+        &cost_model,
+        &asap_aware_mapping::DefaultAccuracyModel,
+        &asap_aware_mapping::EqualSplitAllocator,
+        &Evidence(topk_evidence),
+    )
+    .ok()?;
     // `implement_tree_with` never *errors* on "nothing bound" — an
     // intent `boundary::implementation_for`/`CostModel::realize_extension`
     // can't realize (e.g. `TopK { accuracy: Exact }`, ASAPController#151,
@@ -662,6 +737,16 @@ mod tests {
         }
     }
 
+    fn topk_evidence() -> crate::physical::compiler::TopKMembershipEvidence {
+        crate::physical::compiler::TopKMembershipEvidence {
+            selected_lower_bound: 101.0,
+            excluded_upper_bound: 100.0,
+            interval_failure_probability: 0.001,
+            observed_at_unix_ms: 1,
+            source: "workload-planner-test".into(),
+        }
+    }
+
     #[test]
     fn typed_binding_http_requests_total_is_raw_passthrough() {
         // Contract: `http_requests_total` → raw passthrough (no sketch).
@@ -715,13 +800,23 @@ mod tests {
     }
 
     #[test]
-    fn typed_binding_top_endpoint_qps_requires_membership_evidence() {
+    fn typed_binding_top_endpoint_qps_picks_count_sketch_with_heap() {
         // Contract: `top_endpoint_qps` → CountSketch (TopK).
         // The metric-name reclassification reroutes from the AggType
         // default (Frequency → CMS) to the contract row (TopK →
         // CountSketch).
         let w = workload_for("top_endpoint_qps", AggType::Frequency);
         assert!(bind_workload_typed(&w).is_none());
+        let bound = bind_workload_typed_with_topk_evidence(&w, &topk_evidence())
+            .expect("evidenced top_endpoint_qps must bind");
+        assert_eq!(
+            extract_family(&bound),
+            Some(SketchAlgorithm::CountSketchWithHeap),
+        );
+        assert!(matches!(
+            extract_query(&bound),
+            Some(planner_types::post_asap::SketchQuery::TopK { k: 10, .. })
+        ));
     }
 
     #[test]
@@ -811,7 +906,7 @@ mod tests {
     }
 
     #[test]
-    fn planner_rejects_topk_override_without_membership_evidence() {
+    fn planner_honors_cms_topk_override() {
         // CMS-Heap pattern (Cormode & Muthukrishnan 2005): when a
         // workload's `sketch_family_override` (=
         // `sketch_type_override`) selects CountMinSketch for a TopK
@@ -819,16 +914,23 @@ mod tests {
         // back to the canonical CountSketch default.
         let mut w = workload_for("top_endpoint_qps", AggType::Frequency);
         w.sketch_type_override = Some(SketchType::CountMinSketch);
-        assert!(bind_workload_typed(&w).is_none());
+        let bound = bind_workload_typed_with_topk_evidence(&w, &topk_evidence())
+            .expect("CMS Top-K override must bind with evidence");
+        assert_eq!(extract_family(&bound), Some(SketchAlgorithm::CmsWithHeap));
     }
 
     #[test]
-    fn planner_default_topk_requires_membership_evidence() {
+    fn planner_default_topk_uses_count_sketch_with_heap() {
         // Without any override, the canonical pick for a TopK metric
         // stays CountSketch(-with-heap) — CMS-Heap is opt-in via
         // override only.
         let w = workload_for("top_endpoint_qps", AggType::Frequency);
-        assert!(bind_workload_typed(&w).is_none());
+        let bound = bind_workload_typed_with_topk_evidence(&w, &topk_evidence())
+            .expect("default Top-K must bind with evidence");
+        assert_eq!(
+            extract_family(&bound),
+            Some(SketchAlgorithm::CountSketchWithHeap),
+        );
     }
 
     #[test]
diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
index 878f1fc9..d48e06b8 100644
--- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
+++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
@@ -117,8 +117,21 @@ fn build_workload(
 fn plan_backend_stage_config(
     workload: &QueryWorkload,
 ) -> control_plane::physical::colored_dag::BackendStageConfig {
-    let deployment_expr = control_plane::physical::workload_planner::bind_workload_typed(workload)
-        .expect("bind_workload_typed produced a PhysicalExpr");
+    let deployment_expr = if workload.metric_name == "top_endpoint_qps" {
+        let evidence = control_plane::physical::compiler::TopKMembershipEvidence {
+            selected_lower_bound: 101.0,
+            excluded_upper_bound: 100.0,
+            interval_failure_probability: 0.001,
+            observed_at_unix_ms: 1,
+            source: "self-contained-e2e-fixture".into(),
+        };
+        control_plane::physical::workload_planner::bind_workload_typed_with_topk_evidence(
+            workload, &evidence,
+        )
+    } else {
+        control_plane::physical::workload_planner::bind_workload_typed(workload)
+    }
+    .expect("typed workload binding produced a PhysicalExpr");
     let configs = control_plane::physical::stage_split::split_typed_three_stage(&deployment_expr)
         .expect("split_typed_three_stage produced per-stage configs");
     let mut backend_cfg = configs

From 6585ab433d8027578c65cc771a52c96ea95f5d9d Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 07:13:12 -0600
Subject: [PATCH 3/6] fix: retain typed count update semantics in legacy heap
 bindings

---
 .../src/physical/colored_dag/emitter.rs       | 21 ++++++++++++++++++-
 .../src/physical/workload_planner.rs          |  9 ++++++++
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs
index 787d43b5..ecaefb29 100644
--- a/control_plane/src/physical/colored_dag/emitter.rs
+++ b/control_plane/src/physical/colored_dag/emitter.rs
@@ -54,6 +54,7 @@ enum NodeKind<'a> {
     SketchAgg {
         sketch_type: &'a SketchAlgorithm,
         params: &'a SketchParams,
+        input: &'a planner_types::post_asap::SummaryUpdate,
     },
     /// An exact accumulator — the old `PhysicalExpr::ExactAgg`. This
     /// emitter has never had a match arm for it (falls through to the
@@ -100,10 +101,12 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> {
             } => NodeKind::ExactAgg,
             SummaryExpr::SummaryAgg {
                 family: planner_types::post_asap::SummaryFamilyType::Sketch(kind, _),
+                input,
                 ..
             } => NodeKind::SketchAgg {
                 sketch_type: kind.algorithm(),
                 params: kind.params(),
+                input,
             },
             // `Plain`/`Sample`/`Wavelet`/`StatModel` never occur on a real
             // `SummaryAgg` (never `Plain` by construction; `Sample`/
@@ -156,6 +159,8 @@ pub enum EmitError {
     /// the legacy planner output rather than POST an empty payload.
     #[error("backend has no sketch consumers; nothing to wire")]
     BackendEmpty,
+    #[error("unsupported heap update weight")]
+    UnsupportedHeapUpdate,
 }
 
 /// Generic emitter trait — Phase E ships only [`ThreeStageEmitter`]; future
@@ -886,6 +891,7 @@ impl Emitter for ThreeStageEmitter {
                     NodeKind::SketchAgg {
                         sketch_type,
                         params,
+                        input,
                     },
                     StageId::Edge,
                 ) => {
@@ -902,7 +908,20 @@ impl Emitter for ThreeStageEmitter {
                     });
                     backend_aggregations.push(BackendAggregation {
                         item_label: None,
-                        heap_update_mode: None,
+                        heap_update_mode: if matches!(
+                            sketch_type,
+                            SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap
+                        ) {
+                            use planner_types::post_asap::SummaryInputExpr;
+                            use planner_types::pre_asap::ColumnRef;
+                            Some(match &input.weight {
+                                SummaryInputExpr::Constant(value) if *value == 1.0 => "count",
+                                SummaryInputExpr::Column(ColumnRef::SampleValue) => "value",
+                                _ => return Err(EmitError::UnsupportedHeapUpdate),
+                            })
+                        } else {
+                            None
+                        },
                         aggregation_id,
                         metric_name: edge.source_metric.clone().unwrap_or_default(),
                         family: SummaryFamilyType::Sketch(
diff --git a/control_plane/src/physical/workload_planner.rs b/control_plane/src/physical/workload_planner.rs
index 4c16164a..ea43d465 100644
--- a/control_plane/src/physical/workload_planner.rs
+++ b/control_plane/src/physical/workload_planner.rs
@@ -809,6 +809,15 @@ mod tests {
         assert!(bind_workload_typed(&w).is_none());
         let bound = bind_workload_typed_with_topk_evidence(&w, &topk_evidence())
             .expect("evidenced top_endpoint_qps must bind");
+        // Count-ranked producers must not revert to value-weighted runtime defaults.
+        let configs = crate::physical::stage_split::split_typed_three_stage(&bound).unwrap();
+        let backend = configs
+            .get(&crate::physical::colored_dag::StageId::Backend)
+            .unwrap();
+        let crate::physical::colored_dag::StageConfig::Backend(backend) = backend else {
+            panic!("expected backend config");
+        };
+        assert_eq!(backend.aggregations[0].heap_update_mode, Some("count"));
         assert_eq!(
             extract_family(&bound),
             Some(SketchAlgorithm::CountSketchWithHeap),

From 1a857482285b81482264dd986662cb082f34ad37 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 07:27:47 -0600
Subject: [PATCH 4/6] test: migrate heap TopK acceptance to registered temporal
 QueryPlans

---
 Cargo.lock                                    |   1 +
 data_plane/Cargo.toml                         |   1 +
 .../asapquery_compatibility_process_e2e.rs    | 260 +++++++++++++
 ...e2e_controller_plans_and_backend_serves.rs | 365 +-----------------
 4 files changed, 268 insertions(+), 359 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 9eb7cadb..a10bad8d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -980,6 +980,7 @@ dependencies = [
  "anyhow",
  "arc-swap",
  "arrow",
+ "asap-aware-mapping",
  "asap-precompute-rs",
  "asap-types",
  "asap_otel_proto",
diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml
index 50b3a0d9..2594c15d 100644
--- a/data_plane/Cargo.toml
+++ b/data_plane/Cargo.toml
@@ -129,6 +129,7 @@ crc32fast = "1.4"
 # none of them.
 
 [dev-dependencies]
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" }
 tempfile = "3.20.0"
 criterion = { version = "0.5", features = ["html_reports"] }
 tokio-tungstenite = "0.21"
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index edf36bc3..952d35dd 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -117,6 +117,266 @@ fn is_warm(response: &Value) -> bool {
     })
 }
 
+// Both heap implementations must execute registered temporal counts through
+// an installed QueryPlan, retaining all three ranked identities and values.
+#[tokio::test]
+async fn registered_temporal_topk_cms_heap() {
+    registered_temporal_topk(planner_types::post_asap::SketchAlgorithm::CmsWithHeap).await;
+}
+
+#[tokio::test]
+async fn registered_temporal_topk_count_sketch_heap() {
+    registered_temporal_topk(planner_types::post_asap::SketchAlgorithm::CountSketchWithHeap).await;
+}
+
+async fn registered_temporal_topk(algorithm: planner_types::post_asap::SketchAlgorithm) {
+    use control_plane::physical::compiler::{BackendLocalPlanningSnapshot, PhysicalCompiler};
+    use planner_types::post_asap::{CompositionOperator, SketchQuery, SummaryFamilyType};
+    const QUERY: &str = "topk(3, count_over_time(top_endpoint_qps[5s]))";
+    struct Evidence;
+    impl asap_aware_mapping::AccuracyEvidenceProvider for Evidence {
+        fn propagation_stats(
+            &self,
+            op: &CompositionOperator,
+            _: &SummaryFamilyType,
+            _: Option<&SketchQuery>,
+        ) -> asap_aware_mapping::PropagationStats {
+            if matches!(op, CompositionOperator::TopKSelection) {
+                asap_aware_mapping::PropagationStats {
+                    topk_selected_lower_bound: Some(95.0),
+                    topk_excluded_upper_bound: Some(80.0),
+                    topk_interval_failure_probability: Some(0.001),
+                    ..Default::default()
+                }
+            } else {
+                Default::default()
+            }
+        }
+    }
+    let mut fixture: Value = serde_json::from_str(include_str!(
+        "../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+    ))
+    .unwrap();
+    let mut entry = fixture["query_workload"]["repeating_queries"][5].clone();
+    entry["query"] = QUERY.into();
+    entry["requirements"]["accuracy"] =
+        serde_json::json!({"explicit": {"EpsilonDelta": {"epsilon": 0.05, "delta": 0.05}}});
+    fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]);
+    fixture["implementation"]["topk_evidence"] = serde_json::json!({
+        QUERY: {
+            "selected_lower_bound": 95.0, "excluded_upper_bound": 80.0,
+            "interval_failure_probability": 0.001, "observed_at_unix_ms": 9500,
+            "source": "deterministic-count-ranking-fixture"
+        }
+    });
+    let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap();
+    let (mut request, environment) = snapshot.planning_request().unwrap();
+    let query = &mut request.queries[0];
+    let expr =
+        control_plane::query_parser::parse_query_expr_canonical(QUERY, query.accuracy.clone())
+            .unwrap();
+    let model = control_plane::physical::post_asap::cost_model::ForcedFamilyCostModel::new(
+        query.accuracy.clone(),
+        algorithm.clone(),
+    );
+    query.post_asap = control_plane::planner_selection::select_summary_with_evidence(
+        &expr,
+        &model,
+        &asap_aware_mapping::DefaultAccuracyModel,
+        &asap_aware_mapping::EqualSplitAllocator,
+        &Evidence,
+    )
+    .unwrap();
+    let plan = PhysicalCompiler.compile(request, environment).unwrap();
+    assert_eq!(plan.precompute_plan.materializations.len(), 1);
+    use data_plane::storage_engines::types::AggregationType;
+    let expected_type = match algorithm {
+        planner_types::post_asap::SketchAlgorithm::CmsWithHeap => {
+            AggregationType::CountMinSketchWithHeap
+        }
+        planner_types::post_asap::SketchAlgorithm::CountSketchWithHeap => {
+            AggregationType::CountSketchWithHeap
+        }
+        _ => panic!("fixture requires a heap implementation"),
+    };
+    assert_eq!(
+        plan.precompute_plan.materializations[0].aggregation_type,
+        expected_type
+    );
+    assert_eq!(
+        plan.precompute_plan.materializations[0].parameters["weight_mode"],
+        "count"
+    );
+    let artifact = data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest {
+        precompute_plan: plan.precompute_plan,
+        transmission_plan: plan.transmission_plan,
+        backend_plan: plan.backend_plan.encode_to_vec(),
+        query_plan: plan.query_plan,
+        storage_routing: None,
+        adaptation_evidence: vec![],
+    };
+    let output = tempfile::tempdir().unwrap();
+    let mut artifact_file = tempfile::NamedTempFile::new().unwrap();
+    serde_json::to_writer(&mut artifact_file, &artifact).unwrap();
+    let port = unused_port();
+    let fallback = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let fallback_url = format!("http://{}", fallback.local_addr().unwrap());
+    let fallback_task = tokio::spawn(async move {
+        axum::serve(fallback, Router::new()
+            .route("/-/healthy", get(|| async { "healthy" }))
+            .route("/api/v1/query", get(|| async { Json(serde_json::json!({
+                "status": "error", "errorType": "execution", "error": "fixture exact backend unavailable"
+            })) }))).await.unwrap();
+    });
+    let mut child = ChildGuard(
+        Command::new(env!("CARGO_BIN_EXE_data_plane"))
+            .args(["--profile", "asapquery", "--physical-plan"])
+            .arg(artifact_file.path())
+            .args([
+                "--forward-unsupported-queries",
+                "--prometheus-server",
+                &fallback_url,
+            ])
+            .args(["--http-port", &port.to_string(), "--output-dir"])
+            .arg(output.path())
+            .args([
+                "--precompute-allowed-lateness-ms",
+                "0",
+                "--precompute-flush-interval-ms",
+                "25",
+            ])
+            .stdout(Stdio::null())
+            .stderr(Stdio::inherit())
+            .spawn()
+            .unwrap(),
+    );
+    let client = reqwest::Client::new();
+    let backend = format!("http://127.0.0.1:{port}");
+    wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await;
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as i64;
+    let base = now - now.rem_euclid(5000) - 20000;
+    let counts = [
+        ("alpha", 100),
+        ("beta", 50),
+        ("gamma", 200),
+        ("delta", 75),
+        ("epsilon", 10),
+        ("zeta", 150),
+    ];
+    let samples = WriteRequest {
+        timeseries: counts
+            .iter()
+            .map(|(item, count)| {
+                // Non-unit values distinguish count updates from accidental weighted sums.
+                let points = (0..2)
+                    .flat_map(|window| {
+                        (0..*count).map(move |i| (base + window * 5000 + 10 + i * 20, 17.0))
+                    })
+                    .collect::<Vec<_>>();
+                series_with_labels("top_endpoint_qps", &[("endpoint", item)], &points)
+            })
+            .collect(),
+    };
+    assert_eq!(remote_write(&client, &backend, &samples).await, 204);
+    let watermark = WriteRequest {
+        timeseries: vec![series_with_labels(
+            "top_endpoint_qps",
+            &[("endpoint", "gamma")],
+            &[(base + 10500, 17.0)],
+        )],
+    };
+    assert_eq!(remote_write(&client, &backend, &watermark).await, 204);
+    assert_eq!(remote_write(&client, &backend, &samples).await, 204);
+    let timestamp = (base + 5000) as f64 / 1000.0;
+    let instant = tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            let response: Value = client
+                .get(format!("{backend}/api/v1/query"))
+                .query(&[
+                    ("query", QUERY.to_string()),
+                    ("time", timestamp.to_string()),
+                ])
+                .send()
+                .await
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if response["status"] == "success" && is_warm(&response) {
+                break response;
+            }
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+    })
+    .await
+    .expect("registered TopK must become warm within 30s");
+    let expected = [("gamma", 200.0), ("zeta", 150.0), ("alpha", 100.0)];
+    let assert_ranks = |response: &Value, range: bool| {
+        assert_eq!(response["status"], "success", "{response}");
+        assert!(is_warm(response), "{response}");
+        let rows = response["data"]["result"].as_array().unwrap();
+        assert_eq!(rows.len(), 3, "{response}");
+        for (item, count) in expected {
+            let row = rows
+                .iter()
+                .find(|row| {
+                    row["metric"]["item"].as_str()
+                        == Some(format!("top_endpoint_qps{{endpoint=\"{item}\"}}").as_str())
+                })
+                .unwrap_or_else(|| panic!("missing {item}: {response}"));
+            let points = if range {
+                let points = row["values"].as_array().unwrap();
+                assert_eq!(points.len(), 2);
+                points.clone()
+            } else {
+                vec![row["value"].clone()]
+            };
+            for (index, point) in points.iter().enumerate() {
+                assert_eq!(point[0].as_f64(), Some(timestamp + index as f64 * 5.0));
+                assert_eq!(
+                    point[1].as_str().unwrap().parse::<f64>().unwrap(),
+                    count,
+                    "{response}"
+                );
+            }
+        }
+    };
+    assert_ranks(&instant, false);
+    let range: Value = client
+        .get(format!("{backend}/api/v1/query_range"))
+        .query(&[
+            ("query", QUERY.to_string()),
+            ("start", timestamp.to_string()),
+            ("end", (timestamp + 5.0).to_string()),
+            ("step", "5".into()),
+        ])
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_ranks(&range, true);
+    let unregistered: Value = client
+        .get(format!("{backend}/api/v1/query"))
+        .query(&[("query", "topk(3, top_endpoint_qps)")])
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(
+        unregistered["status"], "error",
+        "unregistered query must not replan: {unregistered}"
+    );
+    assert_eq!(unregistered["error"], "fixture exact backend unavailable");
+    fallback_task.abort();
+}
+
 // Three registered consumers must observe one raw SUM/count producer, including
 // uneven instance sample counts and Remote Write retries.
 #[tokio::test]
diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
index d48e06b8..a8348231 100644
--- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
+++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
@@ -1411,11 +1411,11 @@ async fn controller_plan_to_query_full_roundtrip_hll() {
 // is the canonical TopK metric, so the planner pins
 // `with_heap: true` and the controller emits `CountSketchWithHeap`
 // (regardless of override). To match, the wire DP carries a
-// msgpack-encoded heap envelope (mirroring Test 9), but the query
+// msgpack-encoded heap envelope, but the query
 // uses `count_over_time(...)` instead of `topk(...)` — the
 // reducer's `decode_frequency_total` reads row-0 of the underlying
 // matrix for heap-bearing variants too, so FrequencyEstimate works
-// on the same sid that Test 9 queries for top-k.
+// on a heap-bearing SID.
 //
 // **Strict-success: `count_over_time(top_endpoint_qps[1s])`** binds
 // to `Capability::FrequencyEstimate(Any)`, which
@@ -1447,7 +1447,7 @@ async fn controller_plan_to_query_full_roundtrip_count_sketch() {
 
     // Use the planner-picked `(w, d)` so the OTLP DP's wire-level
     // `rows`/`cols` line up with the policy's `parameters.{d, w}` —
-    // same shape constraint as Test 9.
+    // dimension mismatches prevent physical-policy binding.
     let (w, d) = extract_w_d_from_streaming_config(&streaming_config_json);
     let rows = d as usize;
     let cols = w as usize;
@@ -1674,61 +1674,6 @@ fn extract_w_d_from_streaming_config(streaming_config_json: &JsonValue) -> (u32,
     (w, d)
 }
 
-/// OTLP `ExportMetricsServiceRequest` with a single `CountMinSketch` DP
-/// carrying msgpack-encoded heap-bearing bytes. `encoding=MSGPACK` (3)
-/// triggers `sketch_algorithm_for`'s auto-promotion to `CmsWithHeap`.
-/// `rows`/`cols` on the parent `CountMinSketch` MUST match the policy's
-/// `parameters.{d,w}` for the policy_fp content match to bind.
-fn build_cms_with_heap_msgpack_export(
-    metric_name: &str,
-    attrs: &[(&str, &str)],
-    time_unix_nano: u64,
-    sketch_bytes: Vec<u8>,
-    wire_rows: i32,
-    wire_cols: i32,
-) -> ExportMetricsServiceRequest {
-    let attributes = attrs
-        .iter()
-        .map(|(k, v)| KeyValue {
-            key: k.to_string(),
-            value: Some(AnyValue {
-                value: Some(any_value::Value::StringValue(v.to_string())),
-            }),
-        })
-        .collect();
-    let start_t_ns = time_unix_nano.saturating_sub(1_000_000_000);
-    let dp = CountMinSketchDataPoint {
-        attributes,
-        start_time_unix_nano: start_t_ns,
-        time_unix_nano,
-        sketch: sketch_bytes,
-        encoding: CountMinSketchEncoding::Msgpack as i32,
-        flags: 0,
-        series_id: 0,
-    };
-    ExportMetricsServiceRequest {
-        resource_metrics: vec![ResourceMetrics {
-            resource: None,
-            scope_metrics: vec![ScopeMetrics {
-                scope: None,
-                metrics: vec![Metric {
-                    name: metric_name.to_string(),
-                    description: String::new(),
-                    unit: String::new(),
-                    metadata: Vec::new(),
-                    data: Some(Data::Countminsketch(CountMinSketch {
-                        data_points: vec![dp],
-                        aggregation_temporality: 0,
-                        rows: wire_rows,
-                        cols: wire_cols,
-                    })),
-                }],
-                schema_url: String::new(),
-            }],
-            schema_url: String::new(),
-        }],
-    }
-}
 
 /// OTLP `ExportMetricsServiceRequest` with a single `CountSketch` DP
 /// carrying msgpack-encoded heap-bearing bytes. `encoding=MSGPACK` (3)
@@ -1786,307 +1731,9 @@ fn build_count_sketch_with_heap_msgpack_export(
     }
 }
 
-// ── Test 8 — heap-bearing CMS + topk strict-success ─────────────────────────
-//
-// The full top-k roundtrip with `CmsWithHeap`. Workload pins CMS via
-// `sketch_type_override: Some(SketchType::CountMinSketch)` on the
-// `top_endpoint_qps` metric (TopK statistic class), which the planner
-// binds via `bind_cms_with_heap_on_topk` (CMS-Heap pattern from
-// Cormode & Muthukrishnan 2005).
-//
-// The OTLP DP carries a msgpack-encoded `CountMinSketchWithHeap`
-// payload (`encoding=MSGPACK`); the receiver's `sketch_algorithm_for`
-// peeks at the bytes and auto-promotes the sid to `CmsWithHeap`,
-// registering it under `Capability::FrequencyTopk(CmsWithHeap)`.
-//
-// The reducer's `topk` family decodes the heap directly via
-// `decode_cms_with_heap_from_msgpack` and emits one output series
-// per top-k item with the item key in the `item` label.
-
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn controller_plan_to_query_full_roundtrip_cms_with_heap_topk() {
-    let stack = start_full_stack(19_571, 19_572).await;
-    let client = reqwest::Client::new();
-
-    let workload = build_workload_with_override(
-        "top_endpoint_qps",
-        vec![AggType::Frequency],
-        0.05,
-        Duration::from_secs(1),
-        vec!["service".to_string()],
-        Vec::new(),
-        Some(SketchType::CountMinSketch),
-    );
-    let streaming_config_json = plan_streaming_config_json(&workload);
-    // The controller now emits `CountMinSketchWithHeap` directly when
-    // the planner-set `with_heap: true` flag on `CmsParams` fires
-    // (see `sketch_kind_to_backend_type`). No in-test JSON patch is
-    // needed — the analyzer ↔ policy match binds against the
-    // controller-emitted aggregation type as-is.
-    assert_eq!(
-        streaming_config_json["aggregations"][0]["aggregationType"], "CountMinSketchWithHeap",
-        "controller must emit CountMinSketchWithHeap when bind_cms_with_heap_on_topk fires\n{streaming_config_json}"
-    );
-    assert_eq!(
-        streaming_config_json["readouts"][0]["op"], "topk",
-        "controller must emit a topk readout for CMS-with-heap binding\n{streaming_config_json}"
-    );
-    post_streaming_config(&client, stack.backend_port, &streaming_config_json).await;
-
-    // Use the planner-picked `(w, d)` so the OTLP DP's wire-level
-    // `rows`/`cols` line up with the policy's `parameters.{d, w}` —
-    // `derive_sketch_policy_fp` content-matches on these keys, so a
-    // dimension mismatch leaves the sid registered with `policy_fp`
-    // = UNSET (unreachable through `sids_for_policy`).
-    let (w, d) = extract_w_d_from_streaming_config(&streaming_config_json);
-    let rows = d as usize;
-    let cols = w as usize;
-    let wire_rows = d as i32;
-    let wire_cols = w as i32;
-
-    // Heap items with deterministic count ordering. `gamma` is the
-    // unambiguous top-1 (count=200); the heap (top_k=10) keeps all six.
-    let items: &[(&str, u64)] = &[
-        ("alpha", 100),
-        ("beta", 50),
-        ("gamma", 200),
-        ("delta", 75),
-        ("epsilon", 10),
-        ("zeta", 150),
-    ];
-    let sketch_bytes = build_heap_bearing_msgpack(rows, cols, 10, items);
-
-    let now_ns = std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .expect("system time before UNIX epoch")
-        .as_nanos() as u64;
-    let sketch_t_ns = now_ns.saturating_sub(3_000_000_000);
-    let watermark_t_ns = now_ns.saturating_sub(1_000_000_000);
-
-    let req = build_cms_with_heap_msgpack_export(
-        "top_endpoint_qps",
-        &[("service", "e2e-test")],
-        sketch_t_ns,
-        sketch_bytes.clone(),
-        wire_rows,
-        wire_cols,
-    );
-    post_otlp_http(&client, stack.otlp_http_port, req).await;
-
-    // Watermark MUST also carry the heap — the reducer reads
-    // `samples.iter().next_back()` and decodes the latest sample's
-    // bytes; an empty-heap watermark would shadow the real payload.
-    let watermark_req = build_cms_with_heap_msgpack_export(
-        "top_endpoint_qps",
-        &[("service", "e2e-test")],
-        watermark_t_ns,
-        sketch_bytes,
-        wire_rows,
-        wire_cols,
-    );
-    post_otlp_http(&client, stack.otlp_http_port, watermark_req).await;
-
-    tokio::time::sleep(Duration::from_millis(800)).await;
-
-    let response: JsonValue = client
-        .get(format!(
-            "http://127.0.0.1:{}/api/v1/query",
-            stack.backend_port
-        ))
-        .query(&[("query", "topk(3, top_endpoint_qps)")])
-        .send()
-        .await
-        .expect("query failed")
-        .json()
-        .await
-        .expect("response not JSON");
-    let status = response["status"].as_str().unwrap_or("(missing)");
-    assert_eq!(
-        status,
-        "success",
-        "topk(...) on CmsWithHeap must succeed end-to-end. Response:\n{}",
-        serde_json::to_string_pretty(&response).unwrap_or_default()
-    );
-    // Top-1 must be `gamma` (count=200), surfaced via the
-    // `item: <key>` synthesized label on each top-k series. The
-    // `InstantVectorElement::label_keys_override` field (added
-    // alongside this assertion's tightening) carries the synthesized
-    // key through the Prometheus adapter.
-    let result = &response["data"]["result"];
-    let arr = result
-        .as_array()
-        .expect("result must be an array of vector elements");
-    assert!(
-        !arr.is_empty() && arr.len() <= 3,
-        "topk(3) must return between 1 and 3 series. Response:\n{}",
-        serde_json::to_string_pretty(&response).unwrap_or_default()
-    );
-    let mut values: Vec<f64> = arr
-        .iter()
-        .filter_map(|e| e["value"][1].as_str().and_then(|s| s.parse::<f64>().ok()))
-        .collect();
-    values.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal));
-    let mut found_gamma = false;
-    for elem in arr {
-        if let Some(item) = elem["metric"]["item"].as_str() {
-            if item == "gamma" {
-                found_gamma = true;
-                break;
-            }
-        }
-    }
-    assert!(
-        found_gamma,
-        "topk(3) must surface `gamma` via the `item` label on at least \
-         one series. Response:\n{}",
-        serde_json::to_string_pretty(&response).unwrap_or_default()
-    );
-    assert!(
-        values
-            .first()
-            .map(|v| (v - 200.0).abs() < 1.0)
-            .unwrap_or(false),
-        "topk(3) on heap-bearing CMS must surface `gamma`'s count (200) as \
-         the top value (received {values:?}). Response:\n{}",
-        serde_json::to_string_pretty(&response).unwrap_or_default()
-    );
-}
-
-// ── Test 9 — heap-bearing CountSketch + topk strict-success ─────────────────
-//
-// Same shape as Test 8 but the workload defaults the `top_endpoint_qps`
-// metric to `CountSketch` (the canonical TopK pick — unbiased
-// estimator, see `BindCountSketchOnTopK`). The OTLP DP carries the
-// SAME msgpack heap envelope (the wire shape is shared); only the
-// outer DP type changes (`CountSketchDataPoint` instead of
-// `CountMinSketchDataPoint`).
-//
-// `sketch_algorithm_for` was extended in this PR to peek at
-// CountSketch DPs the same way it does for CountMin — a
-// non-empty heap in a msgpack-encoded payload promotes the sid to
-// `CountSketchWithHeap`, which the analyzer's `is_satisfied_by`
-// recognises as a valid `FrequencyTopk(CountSketchWithHeap)` provider.
-
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn controller_plan_to_query_full_roundtrip_count_sketch_with_heap_topk() {
-    let stack = start_full_stack(19_573, 19_574).await;
-    let client = reqwest::Client::new();
-
-    let workload = build_workload_with_override(
-        "top_endpoint_qps",
-        vec![AggType::Frequency],
-        0.05,
-        Duration::from_secs(1),
-        vec!["service".to_string()],
-        Vec::new(),
-        None, // default → CountSketch (canonical TopK pick)
-    );
-    let streaming_config_json = plan_streaming_config_json(&workload);
-    // The controller now emits `CountSketchWithHeap` directly when
-    // the planner-set `with_heap: true` flag on `CountSketchParams`
-    // fires (see `sketch_kind_to_backend_type`). No in-test JSON
-    // patch is needed.
-    assert_eq!(
-        streaming_config_json["aggregations"][0]["aggregationType"], "CountSketchWithHeap",
-        "controller must emit CountSketchWithHeap for default top_endpoint_qps TopK binding\n{streaming_config_json}"
-    );
-    assert_eq!(
-        streaming_config_json["aggregations"][0]["parameters"]["with_heap"], true,
-        "controller must set parameters.with_heap=true for CountSketch TopK binding\n{streaming_config_json}"
-    );
-    post_streaming_config(&client, stack.backend_port, &streaming_config_json).await;
-
-    let items: &[(&str, u64)] = &[
-        ("alpha", 100),
-        ("beta", 50),
-        ("gamma", 200),
-        ("delta", 75),
-        ("epsilon", 10),
-        ("zeta", 150),
-    ];
-    let (w, d) = extract_w_d_from_streaming_config(&streaming_config_json);
-    let rows = d as usize;
-    let cols = w as usize;
-    let wire_rows = d as i32;
-    let wire_cols = w as i32;
-    let sketch_bytes = build_heap_bearing_msgpack(rows, cols, 10, items);
-
-    let now_ns = std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .expect("system time before UNIX epoch")
-        .as_nanos() as u64;
-    let sketch_t_ns = now_ns.saturating_sub(3_000_000_000);
-    let watermark_t_ns = now_ns.saturating_sub(1_000_000_000);
-
-    let req = build_count_sketch_with_heap_msgpack_export(
-        "top_endpoint_qps",
-        &[("service", "e2e-test")],
-        sketch_t_ns,
-        sketch_bytes.clone(),
-        wire_rows,
-        wire_cols,
-    );
-    post_otlp_http(&client, stack.otlp_http_port, req).await;
-
-    let watermark_req = build_count_sketch_with_heap_msgpack_export(
-        "top_endpoint_qps",
-        &[("service", "e2e-test")],
-        watermark_t_ns,
-        sketch_bytes,
-        wire_rows,
-        wire_cols,
-    );
-    post_otlp_http(&client, stack.otlp_http_port, watermark_req).await;
-
-    tokio::time::sleep(Duration::from_millis(800)).await;
-
-    let response: JsonValue = client
-        .get(format!(
-            "http://127.0.0.1:{}/api/v1/query",
-            stack.backend_port
-        ))
-        .query(&[("query", "topk(3, top_endpoint_qps)")])
-        .send()
-        .await
-        .expect("query failed")
-        .json()
-        .await
-        .expect("response not JSON");
-    let status = response["status"].as_str().unwrap_or("(missing)");
-    assert_eq!(
-        status,
-        "success",
-        "topk(...) on CountSketchWithHeap must succeed end-to-end. Response:\n{}",
-        serde_json::to_string_pretty(&response).unwrap_or_default()
-    );
-    // Same shape-only assertion as Test 8 — `InstantVectorElement`
-    // currently drops per-element labels, so we can't check for
-    // `item: "gamma"`. Verify the strongest invariants the wire
-    // surfaces today: 1..=3 series and gamma's count (200) leads.
-    let result = &response["data"]["result"];
-    let arr = result
-        .as_array()
-        .expect("result must be an array of vector elements");
-    assert!(
-        !arr.is_empty() && arr.len() <= 3,
-        "topk(3) must return between 1 and 3 series. Response:\n{}",
-        serde_json::to_string_pretty(&response).unwrap_or_default()
-    );
-    let mut values: Vec<f64> = arr
-        .iter()
-        .filter_map(|e| e["value"][1].as_str().and_then(|s| s.parse::<f64>().ok()))
-        .collect();
-    values.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal));
-    assert!(
-        values
-            .first()
-            .map(|v| (v - 200.0).abs() < 1.0)
-            .unwrap_or(false),
-        "topk(3) on heap-bearing CountSketch must surface `gamma`'s count (200) \
-         as the top value (received {values:?}). Response:\n{}",
-        serde_json::to_string_pretty(&response).unwrap_or_default()
-    );
-}
+// Heap TopK serving acceptance lives in asapquery_compatibility_process_e2e:
+// registered_temporal_topk_{cms_heap,count_sketch_heap} install the selected
+// physical QueryPlan and verify raw count updates through the production binary.
 
 // ── Test 10 — range-query warm-tier fallback (CMS + count_over_time) ────────
 //

From 5f1f0d482c0daee1232e4de987974b0995a24d50 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 07:29:28 -0600
Subject: [PATCH 5/6] style: format migrated TopK test cleanup

---
 data_plane/tests/e2e_controller_plans_and_backend_serves.rs | 1 -
 1 file changed, 1 deletion(-)

diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
index a8348231..56dcc934 100644
--- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
+++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs
@@ -1674,7 +1674,6 @@ fn extract_w_d_from_streaming_config(streaming_config_json: &JsonValue) -> (u32,
     (w, d)
 }
 
-
 /// OTLP `ExportMetricsServiceRequest` with a single `CountSketch` DP
 /// carrying msgpack-encoded heap-bearing bytes. `encoding=MSGPACK` (3)
 /// triggers `sketch_algorithm_for`'s auto-promotion to

From de7a37fc5ad926f7a1bf9333c5886e7aefb6de4f Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Tue, 8 Sep 2026 14:20:45 -0600
Subject: [PATCH 6/6] fix: align TopK test Planner dependency with merged main

---
 data_plane/Cargo.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml
index 3ce97c3c..a56b61e9 100644
--- a/data_plane/Cargo.toml
+++ b/data_plane/Cargo.toml
@@ -129,7 +129,7 @@ crc32fast = "1.4"
 # none of them.
 
 [dev-dependencies]
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" }
 tempfile = "3.20.0"
 criterion = { version = "0.5", features = ["html_reports"] }
 tokio-tungstenite = "0.21"