Execute Planner physical operators through typed deployment bindings

Cargo.toml

@@ -3,7 +3,6 @@ resolver = "2"
 members = [
     "crates/asap_otel_proto",
     "crates/asap_types",
-    "crates/asap_sketch_codec",
     "data_plane",
     "control_plane",
 ]
@@ -15,10 +14,10 @@ version = "0.1.0"
 [workspace.dependencies]
 # Keep Planner frontends, selection, and IR on the same immutable revision.
 # Alias upstream asap-types because this workspace also defines asap_types.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
-asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "bccc837f5caf21c888b2cce73c64a1be283b679a" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
+asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
 
 # Shared external deps (used by 2+ crates)
 serde = { version = "1.0", features = ["derive"] }
@@ -38,6 +37,8 @@ arc-swap = "1.7"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 
 # Internal crates
+asap-physical-operators = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
+asap_sketch_codec = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4b0839ce1733aab8231eb90944c876063a4551f9" }
 asap_types = { path = "crates/asap_types" }
 asap_otel_proto = { path = "crates/asap_otel_proto" }
 indexmap = { version = "2.0", features = ["serde"] }

control_plane/Cargo.toml

@@ -12,6 +12,7 @@ name = "control_plane"
 path = "src/main.rs"
 
 [dependencies]
+asap-physical-operators.workspace = true
 tokio       = { version = "1",    features = ["full"] }
 axum        = { version = "0.7",  features = ["ws"] }
 futures-util = "0.3"

control_plane/examples/calibration_candidates.rs

@@ -37,6 +37,7 @@ fn planner_forest(queries: &[control_plane::physical::compiler::QueryCompilation
                 vec![lhs, rhs],
                 json!({"operator_debug":format!("{operator:?}"),"timing_debug":format!("{timing:?}")}),
             ),
+
             SummaryExpr::ValueOperation {
                 child,
                 operation,
@@ -73,11 +74,11 @@ fn planner_forest(queries: &[control_plane::physical::compiler::QueryCompilation
                 right,
                 kind,
                 pred,
-                pruning,
+                ..
             } => (
                 "RelationalJoin",
                 vec![left, right],
-                json!({"kind_debug":format!("{kind:?}"),"predicate_debug":format!("{pred:?}"), "pruning_debug":format!("{pruning:?}")}),
+                json!({"kind_debug":format!("{kind:?}"),"predicate_debug":format!("{pred:?}")}),
             ),
             SummaryExpr::SummarySubtract { left, right } => {
                 ("SummarySubtract", vec![left, right], json!({}))

control_plane/src/emit/mod.rs

@@ -56,6 +56,7 @@ fn extract_from_node(node: &Rc<SummaryNode>) -> Option<SketchAlgorithm> {
         SummaryExpr::SummaryEstimate { summary_input, .. } => extract_from_node(summary_input),
         SummaryExpr::SummaryMerge { children, .. } => children.iter().find_map(extract_from_node),
         SummaryExpr::ValueOperation { child, .. } => extract_from_node(child),
+
         // Not surfaced by any `Bind*` path yet (gated on rules that
         // haven't landed — see `deployment_expr.rs`'s module docs).
         SummaryExpr::BinaryOp { .. }

control_plane/src/physical/compiler.rs

@@ -2094,6 +2094,7 @@ fn summary_agg_metric(node: &SummaryNode) -> Option<String> {
                 }
             }
             SummaryExpr::SummaryAgg { child, .. } => walk(child, metrics),
+
             SummaryExpr::ValueOperation { child, .. } => walk(child, metrics),
             SummaryExpr::SummaryEstimate { summary_input, .. } => walk(summary_input, metrics),
             SummaryExpr::SummaryMerge { children, .. } => {
@@ -2162,11 +2163,12 @@ fn observed_population_matches_root(
     };
     if measures.is_empty()
         || !measures.iter().all(|intent| {
-            matches!(
-                intent,
-                AggIntent::FrequencyL2 { col: None, .. }
-                    | AggIntent::FrequencyEntropy { col: None, .. }
-            ) || matches!(intent, AggIntent::Cardinality { cols, .. } if cols.is_empty())
+            matches!(intent, AggIntent::Cardinality { cols, .. } if cols.is_empty())
+                || matches!(
+                    intent,
+                    AggIntent::FrequencyL2 { col: None, .. }
+                        | AggIntent::FrequencyEntropy { col: None, .. }
+                )
         })
     {
         return false;
@@ -2369,6 +2371,7 @@ fn requires_exact_erp_fallback(
                 walk(left, out);
                 walk(right, out);
             }
+
             SummaryExpr::KeepPreAsap(_) => {}
         }
     }
@@ -3601,16 +3604,14 @@ fn collect_selected_materializations(
         }
         match &node.expr {
             SummaryExpr::RelationalJoin {
-                right: candidates,
                 left: values,
+                right: candidates,
                 kind: planner_types::pre_asap::JoinKind::Semi,
                 pruning: Some(_),
                 ..
             } => {
                 walk(candidates, 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.
+                // Explicit external authoritative values do not need duplicate local state.
                 if !composable {
                     walk(values, readout, composable, grouping.clone(), selected)?;
                 }
@@ -4479,6 +4480,7 @@ pub(crate) mod tests {
         }
     }
 
+    // The explicit rate-value frontier cannot be rebound as raw counter deltas.
     #[test]
     fn unsupported_rate_heap_uses_explicit_exact_route() {
         let query = "topk(2, sum by (job) (rate(m[1m])))";
@@ -4506,6 +4508,7 @@ pub(crate) mod tests {
         assert!(plan.precompute_plan.materializations.is_empty());
     }
 
+    // Native exact values remain explicit; unsupported heaps publish no stored state.
     #[test]
     fn hybrid_rate_topk_preserves_the_original_exact_subquery() {
         let query = "topk(2, sum by (job) (rate(m[1m])))";
@@ -4530,7 +4533,7 @@ pub(crate) mod tests {
         assert!(matches!(
             &entry.nodes[&entry.root],
             QueryPlanNode::Logical {
-                operator: ResidualQueryOperator::TopKSelection { k: 2, .. },
+                operator: ResidualQueryOperator::Limit { n: 2, .. },
                 ..
             }
         ));
@@ -5753,6 +5756,11 @@ pub(crate) mod tests {
         let selected = match &selected_root.expr {
             SummaryExpr::SummaryEstimate { summary_input, .. } => summary_input.clone(),
             SummaryExpr::SummaryAgg { .. } => selected_root.clone(),
+            SummaryExpr::ValueOperation {
+                child,
+                operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator,
+                ..
+            } => child.clone(),
             _ => panic!("expected maintained aggregate fixture"),
         };
         request.queries[0].selected_plan_root = Rc::new(SummaryNode {
@@ -5956,15 +5964,6 @@ pub(crate) mod tests {
         let (right, _) = right.into_physical_compilation_request().unwrap();
         let left = request.queries[0].selected_plan_root.clone();
         let right = right.queries[0].selected_plan_root.clone();
-        let right = Rc::new(SummaryNode {
-            expr: SummaryExpr::ValueOperation {
-                timing: planner_types::post_asap::ExecutionTiming::QueryTime,
-                operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator,
-                child: right.clone(),
-            },
-            schema: right.schema.clone(),
-            guarantee: None,
-        });
         request.queries[0].selected_plan_root = Rc::new(SummaryNode {
             expr: SummaryExpr::BinaryOp {
                 timing: planner_types::post_asap::ExecutionTiming::QueryTime,

control_plane/src/physical/plan_dot.rs

@@ -139,6 +139,16 @@ fn escape(value: &str) -> String {
 
 fn query_node_label(node: &QueryPlanNode) -> String {
     match node {
+        QueryPlanNode::PhysicalFragment { dag, .. } => {
+            asap_physical_operators::physical_planner::CompiledPhysicalDag::decode(dag)
+                .map(|plan| {
+                    format!(
+                        "Physical\n{}",
+                        plan.operator_name(plan.roots()[0]).unwrap_or("Input")
+                    )
+                })
+                .unwrap_or_else(|_| "Invalid physical DAG".into())
+        }
         QueryPlanNode::RelationalJoin { .. } => "RelationalJoin".into(),
         QueryPlanNode::Relational { .. } => "Relational".into(),
         QueryPlanNode::Logical { operator, .. } => format!("Logical\n{}", residual_label(operator)),
@@ -154,7 +164,6 @@ fn query_node_label(node: &QueryPlanNode) -> String {
         QueryPlanNode::SummaryEstimate { query, .. } => format!("SummaryEstimate\n{query:?}"),
         QueryPlanNode::ExactReadout { readout, .. } => format!("ExactReadout\n{readout:?}"),
         QueryPlanNode::SummaryMerge { .. } => "SummaryMerge".into(),
-        QueryPlanNode::MembershipFilter { .. } => "MembershipFilter".into(),
         QueryPlanNode::ExternalExact { .. } => "ExternalExact".into(),
         QueryPlanNode::ExactFallback { reason } => format!("ExactFallback\n{reason}"),
     }
@@ -169,7 +178,7 @@ fn residual_label(operator: &ResidualQueryOperator) -> &'static str {
         ResidualQueryOperator::UnaryNegate => "UnaryNegate",
         ResidualQueryOperator::VectorToScalar => "VectorToScalar",
         ResidualQueryOperator::Aggregate { .. } => "Aggregate",
-        ResidualQueryOperator::TopKSelection { .. } => "TopKSelection",
+        ResidualQueryOperator::Limit { .. } => "Limit",
         ResidualQueryOperator::Binary { .. } => "Binary",
         ResidualQueryOperator::Temporal { .. } => "Temporal",
         ResidualQueryOperator::Sort { .. } => "Sort",

control_plane/src/physical/post_asap/cost_model.rs

@@ -174,9 +174,20 @@ impl ControlPlaneCostModel {
         &self,
         candidate: &ReplacementSubDAG,
     ) -> Option<CandidateCostEstimate> {
+        if let Replacement::Rewrite(root) = &candidate.replacement {
+            // Price a semantic rewrite through its executable summary candidates,
+            // in the same retained-state units as a direct summary candidate.
+            // Unknown realizations remain uncosted rather than receiving zero.
+            use asap_aware_mapping::{ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG};
+            return SketchAlgorithmStrategy::new(self)
+                .replacements(&TargetSubDAG::new(root))
+                .iter()
+                .filter(|candidate| matches!(candidate.replacement, Replacement::Summary(_)))
+                .filter_map(|candidate| self.candidate_cost_estimate(candidate))
+                .min_by(|a, b| a.value.total_cmp(&b.value));
+        }
         let Replacement::Summary(root) = &candidate.replacement else {
-            // Exact compositions have a separate measured rate model. Raw
-            // rewrites have no retained-state estimate in this model.
+            // Exact compositions have a separate measured rate model.
             return None;
         };
         let dag = planner_types::post_asap::compile_executable_dag(root).ok()?;
@@ -510,6 +521,23 @@ fn intent_accuracy(intent: &AggIntent) -> AccuracyTarget {
 }
 
 impl CostModel for ControlPlaneCostModel {
+    fn summary_support_evidence(
+        &self,
+        summary: &planner_types::post_asap::SummaryNode,
+    ) -> Option<bool> {
+        use planner_types::post_asap::{NonNegativeWeightProof, WeightDomain};
+        let dag =
+            planner_types::post_asap::compile_executable_dag(&std::rc::Rc::new(summary.clone()))
+                .ok()?;
+        // Counter-weighted heaps now consume explicit rate values. The raw
+        // ingestion adapter cannot bind that frontier as counter deltas.
+        let requires_rate_values = dag.nodes.iter().any(|node| matches!(&node.payload,
+            ExecutableOperatorPayload::SummaryAgg { input, family: SummaryFamilyType::Sketch(kind, _), .. }
+                if matches!(kind.algorithm(), SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap)
+                && matches!(input.weight_domain, WeightDomain::NonNegative { proof: NonNegativeWeightProof::ResetAwareCounterDerivative })));
+        requires_rate_values.then_some(false)
+    }
+
     fn candidate_cost(
         &self,
         candidate: &ReplacementSubDAG,
@@ -869,6 +897,13 @@ impl ForcedFamilyCostModel {
 }
 
 impl CostModel for ForcedFamilyCostModel {
+    fn summary_support_evidence(
+        &self,
+        summary: &planner_types::post_asap::SummaryNode,
+    ) -> Option<bool> {
+        self.inner.summary_support_evidence(summary)
+    }
+
     fn candidate_cost(
         &self,
         candidate: &ReplacementSubDAG,