feat: execute TopK exact values through candidate-filtered Prometheus

control_plane/src/physical/compiler.rs

@@ -3655,7 +3655,12 @@ fn collect_selected_materializations(
                 candidates, values, ..
             } => {
                 walk(candidates, readout, composable, grouping.clone(), selected)?;
-                walk(values, readout, composable, grouping.clone(), selected)?;
+                // In a hybrid TopK, the sketch is only a candidate-membership
+                // sidecar. Prometheus owns the authoritative value subtree;
+                // provisioning local exact state here duplicates that work.
+                if !composable {
+                    walk(values, readout, composable, grouping.clone(), selected)?;
+                }
             }
             SummaryExpr::ValueOperation { child, .. } => {
                 walk(child, readout, composable, grouping.clone(), selected)?;
@@ -4125,6 +4130,53 @@ mod tests {
         assert_eq!(retained_partition_count(counter, Some(5)), 5);
     }
 
+    #[test]
+    fn hybrid_weighted_topk_installs_only_candidates_and_delegates_filtered_exact_values() {
+        use crate::query_plan::{logical::LogicalOperator, QueryPlanNode};
+        let query = "topk(2, sum by (job) (rate(m[1m])))";
+        let evidence = TopKMembershipEvidence {
+            selected_lower_bound: 101.0,
+            excluded_upper_bound: 100.0,
+            interval_failure_probability: 0.001,
+            observed_at_unix_ms: 9_500,
+            source: "unit-fixture".into(),
+        };
+        let mut request = request_with_evidence("topk-rate", query, Some(evidence)).unwrap();
+        request.hybrid_execution = true;
+        let mut environment = environment(10_000);
+        environment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        environment.collector_ids.clear();
+
+        let plan = PhysicalCompiler.compile(request, environment).unwrap();
+
+        assert_eq!(plan.precompute_plan.materializations.len(), 1);
+        assert_eq!(
+            plan.precompute_plan.materializations[0].aggregation_type,
+            asap_types::AggregationType::CountMinSketchWithHeap
+        );
+        let entry = plan.query_plan.lookup(query).unwrap();
+        let QueryPlanNode::CandidateTopK { inputs, .. } = &entry.nodes[&entry.root] else {
+            panic!("expected candidate TopK: {entry:#?}");
+        };
+        assert!(matches!(
+            &entry.nodes[&inputs[1]],
+            QueryPlanNode::Logical {
+                operator: LogicalOperator::CandidateExactSubquery { query, item_label },
+                inputs: exact_inputs,
+            } if query == "sum by (job) (rate(m[1m]))"
+                && item_label == "job"
+                && exact_inputs == &vec![inputs[0]]
+        ));
+        assert!(entry.nodes.values().all(|node| !matches!(
+            node,
+            QueryPlanNode::ExactReadout { .. }
+                | QueryPlanNode::Logical {
+                    operator: LogicalOperator::Scan { .. },
+                    ..
+                }
+        )));
+    }
+
     #[test]
     fn retained_footprint_rejects_eval_sized_cms_across_all_panes() {
         let evidence = TopKMembershipEvidence {

control_plane/src/physical/workload_cost.rs

@@ -234,7 +234,11 @@ pub fn manifest(
                 ));
             }
             if let crate::query_plan::QueryPlanNode::Logical {
-                operator: crate::query_plan::logical::LogicalOperator::ExactSubquery { query },
+                operator:
+                    crate::query_plan::logical::LogicalOperator::ExactSubquery { query }
+                    | crate::query_plan::logical::LogicalOperator::CandidateExactSubquery {
+                        query, ..
+                    },
                 ..
             } = node
             {

control_plane/src/query_plan.rs

@@ -756,8 +756,60 @@ where
                             })
                     })
                     .collect::<Result<Vec<_>, _>>()?;
+                let candidate_input = self.lower(candidates)?;
+                let value_input = if let Some(original) = &self.logical_source {
+                    let parsed = promql_parser::parser::parse(original)
+                        .map_err(|error| QueryPlanError::Invalid(error.to_string()))?;
+                    let promql_parser::parser::Expr::Aggregate(aggregate) = parsed else {
+                        return Err(QueryPlanError::Invalid(
+                            "CandidateTopK requires a top-level PromQL aggregate".into(),
+                        ));
+                    };
+                    if aggregate.op.to_string() != "topk" {
+                        return Err(QueryPlanError::Invalid(
+                            "CandidateTopK requires a topk source expression".into(),
+                        ));
+                    }
+                    fn item_label(node: &SummaryNode) -> Option<String> {
+                        match &node.expr {
+                            SummaryExpr::SummaryEstimate { summary_input, .. } => {
+                                item_label(summary_input)
+                            }
+                            SummaryExpr::SummaryAgg { input, .. } => match &input.item {
+                                Some(planner_types::post_asap::SummaryInputExpr::Column(
+                                    planner_types::pre_asap::ColumnRef::Named(label),
+                                )) => Some(label.clone()),
+                                Some(planner_types::post_asap::SummaryInputExpr::Column(
+                                    planner_types::pre_asap::ColumnRef::Qualified { name, .. },
+                                )) => Some(name.clone()),
+                                _ => None,
+                            },
+                            _ => None,
+                        }
+                    }
+                    let item_label = item_label(candidates).ok_or_else(|| {
+                        QueryPlanError::Invalid(
+                            "CandidateTopK membership has no named item label".into(),
+                        )
+                    })?;
+                    let value_id = QueryNodeId(self.next_id);
+                    self.next_id += 1;
+                    self.nodes.insert(
+                        value_id,
+                        QueryPlanNode::Logical {
+                            operator: logical::LogicalOperator::CandidateExactSubquery {
+                                query: aggregate.expr.to_string(),
+                                item_label,
+                            },
+                            inputs: vec![candidate_input],
+                        },
+                    );
+                    value_id
+                } else {
+                    self.lower(values)?
+                };
                 QueryPlanNode::CandidateTopK {
-                    inputs: [self.lower(candidates)?, self.lower(values)?],
+                    inputs: [candidate_input, value_input],
                     k: u64::try_from(*k).map_err(|_| {
                         QueryPlanError::Invalid("CandidateTopK k exceeds u64".into())
                     })?,

control_plane/src/query_plan/logical.rs

@@ -16,6 +16,12 @@ pub enum LogicalOperator {
     ExactSubquery {
         query: String,
     },
+    /// Prometheus exact subtree whose selectors are restricted at runtime by
+    /// the candidate vector produced by its single input.
+    CandidateExactSubquery {
+        query: String,
+        item_label: String,
+    },
     Scan {
         metric: Option<String>,
         matchers: Vec<LabelMatcher>,
@@ -141,6 +147,7 @@ impl LogicalOperator {
     pub fn validate(&self, inputs: usize) -> Result<(), QueryPlanError> {
         let expected = match self {
             Self::Scan { .. } | Self::ExactSubquery { .. } => 0,
+            Self::CandidateExactSubquery { .. } => 1,
             Self::Binary { .. } | Self::HistogramQuantile => 2,
             _ => 1,
         };
@@ -156,7 +163,7 @@ impl LogicalOperator {
         ) {
             return Err(invalid("zero range"));
         }
-        if let Self::ExactSubquery { query } = self {
+        if let Self::ExactSubquery { query } | Self::CandidateExactSubquery { query, .. } = self {
             let parsed = parser::parse(query).map_err(|e| invalid(e.to_string()))?;
             if matches!(parsed, Expr::MatrixSelector(_) | Expr::Subquery(_)) {
                 return Err(invalid(
@@ -1407,7 +1414,9 @@ pub fn externalize_residuals(entry: &mut QueryPlanEntry) -> Result<(), QueryPlan
         if let QueryPlanNode::Logical { operator, .. } = node {
             exact = matches!(
                 operator,
-                LogicalOperator::Scan { .. } | LogicalOperator::ExactSubquery { .. }
+                LogicalOperator::Scan { .. }
+                    | LogicalOperator::ExactSubquery { .. }
+                    | LogicalOperator::CandidateExactSubquery { .. }
             );
         }
         for child in node.inputs() {
@@ -1422,7 +1431,8 @@ pub fn externalize_residuals(entry: &mut QueryPlanEntry) -> Result<(), QueryPlan
         if matches!(
             entry.nodes.get(&id),
             Some(QueryPlanNode::Logical {
-                operator: LogicalOperator::ExactSubquery { .. },
+                operator: LogicalOperator::ExactSubquery { .. }
+                    | LogicalOperator::CandidateExactSubquery { .. },
                 ..
             })
         ) {

data_plane/src/drivers/query/servers/http.rs

@@ -1642,12 +1642,12 @@ async fn annotate_data_source(response: Response, data_source_id: &'static str)
                     }
                     if raw > 0 {
                         ("failed", "invalid_provenance")
+                    } else if remote > 0 && summary > 0 {
+                        ("hybrid", "hybrid")
                     } else if remote > 0 || summary == 0 {
                         (
                             "exact_fallback",
-                            if summary > 0 {
-                                "hybrid"
-                            } else if remote > 0 {
+                            if remote > 0 {
                                 "external_exact"
                             } else {
                                 "invalid_provenance"
@@ -6795,14 +6795,14 @@ mod logical_provenance_tests {
     use super::*;
 
     #[tokio::test]
-    async fn hybrid_execution_is_fallback_with_measured_branch_counts() {
+    async fn hybrid_execution_has_its_own_route_with_measured_branch_counts() {
         // A successful mixed graph must never inherit the pure-ASAP route from its engine name.
         let response = Json(serde_json::json!({"status":"success", "warnings":[
             "asap_execution:hybrid", "asap_logical_stats:raw=0,summary=1,memo_hits=3,remote=2,remote_rpcs=1,remote_branches=2"
         ], "data":{"resultType":"vector", "result":[]}}))
         .into_response();
         let response = annotate_data_source(response, "asap_query").await;
-        assert_eq!(response.headers()["x-asap-execution"], "exact_fallback");
+        assert_eq!(response.headers()["x-asap-execution"], "hybrid");
         assert_eq!(response.headers()["x-asap-execution-detail"], "hybrid");
         assert_eq!(
             response.headers()["x-asap-summary-readout-evaluations"],

data_plane/src/query_engines/asap_query_engine/engine.rs

@@ -176,11 +176,54 @@ impl ASAPQueryEngine {
             physical.query_plan.plan_id,
             physical.query_plan.plan_version,
         )?;
-        super::exact_subqueries::prepare(
+        // Candidate-filtered exact cuts have a data dependency: read the
+        // installed membership subtree once, then use that vector to build the
+        // Prometheus selector. Keeping the result as a prepared leaf also means
+        // CandidateTopK reuses the same membership readout during composition.
+        let dependencies = super::exact_subqueries::candidate_dependencies(entry, times)?;
+        let mut prepared = super::logical_dag::PreparedLeaves::new();
+        let unique_inputs = dependencies
+            .into_iter()
+            .map(|(_, input, at, _)| (input, at))
+            .collect::<std::collections::BTreeSet<_>>();
+        for (input, at) in unique_inputs {
+            let evaluation_ms = u64::try_from(at).map_err(|_| {
+                crate::query_engines::EngineError::capability_miss(
+                    "exact_subquery",
+                    "candidate evaluation predates epoch",
+                )
+            })?;
+            let mut subtree = entry.clone();
+            subtree.root = input;
+            let reachable = subtree.topological_order().map_err(|error| {
+                crate::query_engines::EngineError::capability_miss(
+                    "installed_logical_dag",
+                    error.to_string(),
+                )
+            })?;
+            subtree.nodes.retain(|id, _| reachable.contains(id));
+            let (result, _) = self.execute_logical_entry(
+                physical,
+                &subtree,
+                &super::logical_dag::PreparedLeaves::new(),
+                evaluation_ms,
+            )?;
+            prepared.insert(
+                (input, at),
+                super::logical_dag::PreparedLeaf {
+                    value: super::logical_dag::from_result(result)?,
+                    remote: false,
+                    remote_evaluations: 0,
+                    remote_rpcs: 0,
+                },
+            );
+        }
+        super::exact_subqueries::prepare_with_candidates(
             entry,
             times,
             self.exact_subquery_endpoint.as_deref(),
             &self.exact_subquery_client,
+            prepared,
         )
         .await
     }