fix(planner): sync Python subpopulation logic, drop redundant heap override

does_precompute_operator_support_subpopulations in the Python copy of
this function still returned False for CountMinSketchWithHeap + TOPK
with a stale comment, the same bug the Rust copy fixed for planner/SQL
in a prior commit. Update it to return True, matching Rust (the heap
is self-keyed but still tracks many keys internally, like MultipleSum).

Also remove the manual grouping/aggregated-label override for SQL
top-k in planner/sql.rs: build_agg_configs_for_statistics already
produces the same split via set_subpopulation_labels now that
does_precompute_operator_support_subpopulations returns true for this
case, so the override was dead code.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V66Z7Fge2EYW2N1CDJzrxk

Author: milindsrivastava1997

asap-common/dependencies/py/promql_utilities/promql_utilities/query_logics/logics.py

@@ -97,9 +97,12 @@ def does_precompute_operator_support_subpopulations(
     elif (
         precompute_operator == "CountMinSketchWithHeap" and statistic == Statistic.TOPK
     ):
-        # topk and bottomk do not support subpopulations!
-        # other usages of CountMinSketchWithHeap will fall through.
-        return False
+        # CountMinSketchWithHeap (topk) is a heavy-hitters sketch: one heap
+        # instance tracks many keys internally (like MultipleSum), it just
+        # doesn't need an *external* paired key aggregation to enumerate them
+        # -- it discovers its own top-k keys. So it supports subpopulations
+        # the same way MultipleSum does.
+        return True
     # elif precompute_operator == "UnivMon":
     #     return statistic in ["sum", "count", "avg"]
     else:

asap-planner-rs/src/planner/sql.rs

@@ -141,7 +141,7 @@ impl SQLSingleQueryProcessor {
         let topk_k = sql_topk.map(|t| t.k);
         let topk_count_events = sql_topk.map(|t| t.count_events());
 
-        let mut configs = build_agg_configs_for_statistics(
+        let configs = build_agg_configs_for_statistics(
             &statistics,
             treatment_type,
             &spatial_output,
@@ -163,17 +163,6 @@ impl SQLSingleQueryProcessor {
         )
         .map_err(ControllerError::SqlParse)?;
 
-        if sql_topk.is_some() {
-            for cfg in &mut configs {
-                if cfg.aggregation_type == AggregationType::CountMinSketchWithHeap {
-                    // Heap-only self-keyed layout: the GROUP BY column is tracked
-                    // inside the sketch's aggregated dimension, not as a partition key.
-                    cfg.grouping_labels = KeyByLabelNames::empty();
-                    cfg.aggregated_labels = spatial_output.clone();
-                }
-            }
-        }
-
         // SQLPatternParser always produces second-based durations; convert to ms.
         // For a single-scrape-interval query this equals data_ingestion_interval_ms
         // by construction (the matcher's classification boundary), so this is a