fix(planner): heap top-k plans compile and answer per ranked item

crates/executor/src/physical_planner/mod.rs

@@ -711,7 +711,11 @@ fn temporal_evaluation_drops_name(node: &PhysicalASAPDAGNode) -> bool {
                         | SketchStatistic::Cardinality
                         | SketchStatistic::PointCount { .. }
                         | SketchStatistic::FrequencyL2
-                        | SketchStatistic::FrequencyEntropy,
+                        | SketchStatistic::FrequencyEntropy
+                        // A heap ranking series identities over raw samples
+                        // answers a `sum_over_time` or `count_over_time` per
+                        // ranked series.
+                        | SketchStatistic::TopK { .. },
                     ..
                 })
         )

crates/executor/tests/weighted_topk_binding.rs

@@ -14,10 +14,6 @@ use planner_types::ir::schema::{DataType, *};
 use planner_types::ir::ASAPOp;
 use std::{collections::BTreeMap, rc::Rc, sync::Arc};
 #[test]
-#[ignore = "the whole-expression heap keeps the outer top-k's keys indexed against the absorbed \
-            aggregate, so the physical planner rejects its keyed evaluation shape (#639); the \
-            unconfirmed Count-Min heap also has no non-negative weight proof until plan selection \
-            confirms one"]
 fn planner_weighted_topk_binds_at_either_deployment_phase() {
     assert_weighted_binding(SketchAlgorithm::CmsWithHeap);
     assert_weighted_binding(SketchAlgorithm::CountSketchWithHeap);
@@ -46,13 +42,38 @@ fn heap_candidate(
         .unwrap_or_else(|| panic!("missing {algorithm:?} candidate"))
 }
 
+/// `plan` as Stage 3 hands it to the executor when the data declares `m` a
+/// counter: its rate weights' open requirement confirmed non-negative.
+fn confirmed_as_counter(
+    plan: Rc<planner_types::ir::OperatorNode>,
+) -> Rc<planner_types::ir::OperatorNode> {
+    use planner_types::ir::QueryRoot;
+    use planner_types::workload::{MetricType, SourceFacts};
+    let data = DataWorkload {
+        sources: [(
+            "m".to_string(),
+            SourceFacts {
+                metric_type: Some(MetricType::Counter),
+                ..SourceFacts::default()
+            },
+        )]
+        .into(),
+        ..DataWorkload::default()
+    };
+    match asap_plan_selection::with_confirmed_weights(&[QueryRoot::Operator(plan)], &data).remove(0)
+    {
+        QueryRoot::Operator(node) => node,
+        QueryRoot::Scalar(_) => unreachable!("an operator root"),
+    }
+}
+
 fn assert_weighted_binding(algorithm: SketchAlgorithm) {
     let root = lower_promql(
         "topk by(job)(2, sum by(service, job)(rate(m[1m])))",
         AccuracyTarget::max_error(0.1),
     )
     .unwrap();
-    let plan = heap_candidate(root, &algorithm);
+    let plan = confirmed_as_counter(heap_candidate(root, &algorithm));
     let dag = compile_physical_asap_dag(&plan).unwrap();
     let rate_id_of = |dag: &PhysicalASAPDAG| {
         let build=dag.nodes.iter().find(|node|matches!(&node.payload,PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg {family:FieldDataType::Sketch(kind,_),..})if kind.algorithm()==&algorithm)).unwrap();
@@ -142,14 +163,17 @@ fn assert_weighted_binding(algorithm: SketchAlgorithm) {
             output
         });
         // Two items per `job`: the outer `by(job)` partitions the ranking.
+        // Each row is `[job, topk_2, service]`: the top-k's columns, then
+        // the ranked item's label.
         let schema = &dag
             .nodes
             .iter()
             .find(|node| node.id == dag.roots[0])
             .unwrap()
             .output_schema;
         let column = |name: &str| schema.fields.iter().position(|f| f.name == name).unwrap();
-        let (job, value) = (column("job"), column("value"));
+        let (job, value) = (column("job"), column("topk_2"));
+        assert_eq!(column("service"), 2);
         let mut ranked = output
             .iter()
             .map(|row| match (&row[job], &row[value]) {

crates/integration-tests/tests/stage1_logical_asap_optimization.rs

@@ -113,7 +113,21 @@ fn compose_all(label: &str, root: QueryRoot) -> usize {
         let root = &composed[0].1;
         root.validate_structure()
             .unwrap_or_else(|e| panic!("{label} {choice:?}: {e}"));
-        assert_eq!(output_columns(root), expected, "{label} {choice:?}");
+        let columns = output_columns(root);
+        // A heap top-k answers the declared `[keys.., topk_k]` with one row
+        // per ranked item: `topk_k` its Float64 estimate (the value the
+        // exact top-k's physical plan answers too), then the item labels.
+        let heap_topk = expected
+            .last()
+            .is_some_and(|(name, _, _)| name.starts_with("topk_"))
+            && columns.get(expected.len() - 1).map(|c| c.1.as_str()) == Some("Plain(Float64)");
+        if heap_topk {
+            let (ranked, keys) = expected.split_last().unwrap();
+            assert_eq!(&columns[..keys.len()], keys, "{label} {choice:?}");
+            assert_eq!(columns[keys.len()].0, ranked.0, "{label} {choice:?}");
+        } else {
+            assert_eq!(columns, expected, "{label} {choice:?}");
+        }
         for (digit, radix) in choice.iter_mut().zip(&radix).rev() {
             *digit += 1;
             if *digit < *radix {

crates/logical-asap-optimization/src/per_computation_candidates.rs

@@ -961,8 +961,14 @@ fn whole_expression_input(
         input.item = Some(SummaryInputExpr::Column(ColumnRef::Named(
             asap_types::ir::schema::PROMQL_SERIES_IDENTITY.into(),
         )));
-    } else if realized.child.schema.closed {
-        // An encoded label set needs an open PromQL schema.
+    } else if realized.child.schema.closed
+        && (matches!(input.item, Some(SummaryInputExpr::EntityIdentity(_)))
+            || !reads_only_time_series(&realized.child))
+    {
+        // An encoded label set needs an open PromQL schema. PromQL rows that
+        // carry the series identity are closed, but their inner group labels
+        // are item columns. (A SQL top-k keeps its heaps over its per-key
+        // result.)
         return None;
     }
     let partitions = partitions_over(reduction, &child.schema, &realized.child.schema)?;
@@ -2722,7 +2728,9 @@ fn with_names_of(
             .map(|field| field.name.clone())
             .collect()
     };
-    if names(&output) == names(target) {
+    // A top-k's ranked rows may carry their item columns after the
+    // aggregate's ([`ranked_as_topk`]).
+    if names(&output).starts_with(&names(target)) {
         return Ok(output);
     }
     if output.schema.fields.len() != target.schema.fields.len() {
@@ -2773,10 +2781,67 @@ fn whole_expression(
         filter: None,
     }))?;
     sizing.built.borrow_mut().push((state.clone(), key));
-    Ok(OperatorNode::new_shared(Operator::ASAP(
-        ASAPOp::SummaryEstimate {
-            summary_input: state,
-            query: statistic(intent)?,
+    let estimate = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate {
+        summary_input: state,
+        query: statistic(intent)?,
+    }))?;
+    ranked_as_topk(target, estimate)
+}
+
+/// A heap's top-k `estimate` (`[partitions.., items.., topk]`, one row per
+/// ranked item) in `target`'s columns: its `[keys.., topk_k]`, with
+/// `topk_k` the estimated Float64 weight, then, over PromQL rows, the item
+/// columns. This is the exact top-k's physical shape (its ranked rows'
+/// value, then their other labels; a SQL result keeps only its logical
+/// columns), so readers of `[keys.., topk_k]` find those columns where they
+/// were.
+fn ranked_as_topk(
+    target: &OperatorNode,
+    estimate: Rc<OperatorNode>,
+) -> Result<Rc<OperatorNode>, PerComputationCandidateError> {
+    let (ranked, keys) = target
+        .schema
+        .fields
+        .split_last()
+        .expect("an aggregate outputs its measure");
+    let fields = &estimate.schema.fields;
+    let estimated = fields.len() - 1;
+    let column = |index: usize, name: &str| ProjectItem {
+        alias: Some(name.to_string()),
+        expr: ScalarExpr::Column(index),
+    };
+    // Each key by name among the estimate's partition and item columns.
+    let key_columns = keys
+        .iter()
+        .map(|key| {
+            fields[..estimated]
+                .iter()
+                .position(|field| field.name == key.name)
+                .ok_or(PerComputationCandidateError::Unsupported(
+                    "top-k estimate without a partition column",
+                ))
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+    let promql = match target.non_asap() {
+        Some(NonASAPOp::Aggregate { child, .. }) => reads_only_time_series(child),
+        _ => false,
+    };
+    let cols = key_columns
+        .iter()
+        .zip(keys)
+        .map(|(&index, key)| column(index, &key.name))
+        .chain(std::iter::once(column(estimated, &ranked.name)))
+        .chain(
+            (0..estimated)
+                .filter(|index| promql && !key_columns.contains(index))
+                .map(|index| column(index, &fields[index].name)),
+        )
+        .collect();
+    Ok(OperatorNode::new_shared(Operator::NonASAP(
+        NonASAPOp::Project {
+            cols,
+            qualifier: None,
+            child: estimate,
         },
     ))?)
 }
@@ -2884,10 +2949,14 @@ fn substitute_measure(
                 measure_input(intent, input, filter.as_ref())?,
             )?;
             sizing.built.borrow_mut().push((summary_input.clone(), key));
-            OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate {
+            let estimate = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate {
                 summary_input,
                 query: statistic(intent)?,
-            }))?
+            }))?;
+            match intent {
+                AggIntent::TopK { .. } => ranked_as_topk(target, estimate)?,
+                _ => estimate,
+            }
         }
         ComputationCandidate::ExactAccumulator(kinds) => {
             let (AggIntent::Avg { col }, [ExactKind::Sum, ExactKind::Count]) =

crates/plan-selection/src/lib.rs

@@ -1293,9 +1293,11 @@ impl<'a> PlanSelector<'a> {
             }
         }
         // Only a candidate that may be selected is compiled: the check runs
-        // the executor's own compiler, the costliest check.
+        // the executor's own compiler, the costliest check. It compiles the
+        // plan the executor would run, with each weight requirement Stage 3
+        // confirmed made a proof, as `Selection::plan` holds it.
         if let Some(check) = self.models.executor_compile_check {
-            if let Err(reason) = check.compiles(&sized.roots) {
+            if let Err(reason) = check.compiles(&with_confirmed_weights(&sized.roots, self.data)) {
                 self.reject(&candidate.id, RejectionCause::ExecutorCannotCompile, reason);
                 return Ok(());
             }