refactor: consume Planner-owned physical DAG operators

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 = "cd7e9e0f710816d49190dabd6c789359067a208f" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cd7e9e0f710816d49190dabd6c789359067a208f" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cd7e9e0f710816d49190dabd6c789359067a208f" }
-asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cd7e9e0f710816d49190dabd6c789359067a208f" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "645cacb04a451e25147f0d23aaddd05da16b4746" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "645cacb04a451e25147f0d23aaddd05da16b4746" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "645cacb04a451e25147f0d23aaddd05da16b4746" }
+asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "645cacb04a451e25147f0d23aaddd05da16b4746" }
 
 # 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 = "645cacb04a451e25147f0d23aaddd05da16b4746" }
+asap_sketch_codec = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "645cacb04a451e25147f0d23aaddd05da16b4746" }
 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,17 +37,7 @@ fn planner_forest(queries: &[control_plane::physical::compiler::QueryCompilation
                 vec![lhs, rhs],
                 json!({"operator_debug":format!("{operator:?}"),"timing_debug":format!("{timing:?}")}),
             ),
-            SummaryExpr::CandidateTopK {
-                candidates,
-                values,
-                k,
-                grouping,
-                completeness,
-            } => (
-                "CandidateTopK",
-                vec![candidates, values],
-                json!({"k":k,"grouping_debug":format!("{grouping:?}"),"completeness_debug":format!("{completeness:?}")}),
-            ),
+
             SummaryExpr::ValueOperation {
                 child,
                 operation,
@@ -84,6 +74,7 @@ fn planner_forest(queries: &[control_plane::physical::compiler::QueryCompilation
                 right,
                 kind,
                 pred,
+                ..
             } => (
                 "RelationalJoin",
                 vec![left, right],
@@ -105,7 +96,7 @@ fn planner_forest(queries: &[control_plane::physical::compiler::QueryCompilation
                 vec![summary_input],
                 json!({"query_debug":format!("{query:?}")}),
             ),
-            SummaryExpr::SummaryMerge { children } => {
+            SummaryExpr::SummaryMerge { children, .. } => {
                 ("SummaryMerge", children.iter().collect(), json!({}))
             }
         };

control_plane/examples/offline_planner_replay.rs

@@ -59,7 +59,7 @@ fn inspect(
             inspect(summary_input, model, seen, states, raw)
         }
         SummaryExpr::ValueOperation { child, .. } => inspect(child, model, seen, states, raw),
-        SummaryExpr::SummaryMerge { children } => {
+        SummaryExpr::SummaryMerge { children, .. } => {
             for child in children {
                 inspect(child, model, seen, states, raw);
             }
@@ -82,12 +82,6 @@ fn inspect(
             inspect(lhs, model, seen, states, raw);
             inspect(rhs, model, seen, states, raw);
         }
-        SummaryExpr::CandidateTopK {
-            candidates, values, ..
-        } => {
-            inspect(candidates, model, seen, states, raw);
-            inspect(values, model, seen, states, raw);
-        }
     }
 }
 

control_plane/src/emit/mod.rs

@@ -54,11 +54,9 @@ fn extract_from_node(node: &Rc<SummaryNode>) -> Option<SketchAlgorithm> {
         // `ExactAgg` case.
         SummaryExpr::SummaryAgg { .. } => None,
         SummaryExpr::SummaryEstimate { summary_input, .. } => extract_from_node(summary_input),
-        SummaryExpr::SummaryMerge { children } => children.iter().find_map(extract_from_node),
+        SummaryExpr::SummaryMerge { children, .. } => children.iter().find_map(extract_from_node),
         SummaryExpr::ValueOperation { child, .. } => extract_from_node(child),
-        SummaryExpr::CandidateTopK {
-            candidates, values, ..
-        } => extract_from_node(candidates).or_else(|| extract_from_node(values)),
+
         // Not surfaced by any `Bind*` path yet (gated on rules that
         // haven't landed — see `deployment_expr.rs`'s module docs).
         SummaryExpr::BinaryOp { .. }