Migrate backend to renamed ASAPPlanner APIs

Cargo.toml

@@ -20,10 +20,10 @@ asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", branch
 [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 = "f46cbf6c5738db2f4460d419baa8af5572f5276a" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f46cbf6c5738db2f4460d419baa8af5572f5276a" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f46cbf6c5738db2f4460d419baa8af5572f5276a" }
-asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f46cbf6c5738db2f4460d419baa8af5572f5276a" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" }
+asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "10d93846f0ef9f06ab19e599701d699d9cdd15af" }
 
 # Shared external deps (used by 2+ crates)
 serde = { version = "1.0", features = ["derive"] }

control_plane/src/physical/compiler/windows.rs

@@ -305,20 +305,32 @@ mod tests {
             0
         )
         .is_err());
-        quote.framework =
-            SummaryWindowFramework::Extension("backend.exact-hierarchical-rollup.v1".into());
+        quote.framework = SummaryWindowFramework::ExponentialHistogram;
         quote.layout = WindowMaterializationLayout::HierarchicalRollup {
             base_pane_secs: 10,
             levels_secs: vec![30],
         };
+        assert!(quote
+            .layout
+            .validate(quote.window_secs, quote.slide_secs)
+            .is_ok());
+        assert!(!supported(
+            &quote,
+            PhysicalDeploymentTarget::BackendLocalRemoteWrite
+        ));
         model.quotes = vec![quote];
-        assert!(prepare_window_implementations(
+        let error = prepare_window_implementations(
             &mut query,
             &model,
             PhysicalDeploymentTarget::BackendLocalRemoteWrite,
-            0
+            0,
         )
-        .is_err());
+        .unwrap_err();
+        assert!(matches!(
+            error,
+            CompileError::Lifecycle { reason, .. }
+                if reason.contains("does not match an executable state layout")
+        ));
     }
 
     // Conflicting identities and repeated shape evidence must fail rather than silently lose an offer.

control_plane/src/physical/plan_dot.rs

@@ -46,7 +46,15 @@ pub fn render(plan: &CompiledPhysicalPlan) -> String {
             emit_node(
                 &mut dot,
                 &id,
-                &format!("{:?} #{}{}", node.operator, node.id.0, binding),
+                &format!(
+                    "{} #{}{}",
+                    node.payload
+                        .get("kind")
+                        .and_then(serde_json::Value::as_str)
+                        .unwrap_or("unknown"),
+                    node.id.0,
+                    binding
+                ),
                 "",
             );
         }

control_plane/src/planner_selection.rs

@@ -100,7 +100,7 @@ fn summary_identity(node: &SummaryNode) -> Option<String> {
         let hash = explain_identity(
             "summary_node",
             &serde_json::json!({
-            "operator": node.operator, "payload": node.payload, "state": node.output_state,
+            "payload": node.payload, "state": node.output_state,
             "schema": node.output_schema, "guarantee": node.guarantee, "inputs": inputs}),
         );
         memo.insert(id, hash.clone());
@@ -399,7 +399,7 @@ fn select_workload_impl(
     let selection = space.global_selection(cost_model);
     if let Some(trace) = trace.as_deref_mut() {
         let groups = space.cost_sorted(cost_model).iter().enumerate().map(|(index, group)| {
-            let chosen = selection.groups().find(|selected| Rc::ptr_eq(selected.target, group.target))
+            let chosen = selection.target_selections().find(|selected| Rc::ptr_eq(selected.target, group.target))
                 .and_then(|selected| selected.chosen);
             let candidates = group.candidates.iter().zip(&group.costs).enumerate()
                 .map(|(rank, (candidate, cost))| serde_json::json!({
@@ -418,8 +418,8 @@ fn select_workload_impl(
                     "estimated_cost_status": if cost.is_finite() { "available" } else { "not_reported_by_cost_model" },
                     "selected": chosen.is_some_and(|chosen| std::ptr::eq(chosen, *candidate)),
                 })).collect::<Vec<_>>();
-            let rejected = space.groups().find(|memo| Rc::ptr_eq(&memo.target, group.target))
-                .into_iter().flat_map(|memo| &memo.rejected).map(|candidate| serde_json::json!({
+            let rejected = space.target_subdag_candidates().find(|candidates| Rc::ptr_eq(&candidates.target, group.target))
+                .into_iter().flat_map(|candidates| &candidates.rejected).map(|candidate| serde_json::json!({
                     "status": "rejected", "strategy": candidate.strategy,
                     "description": candidate.description, "reason": candidate.error.to_string()
                 })).collect::<Vec<_>>();
@@ -434,7 +434,7 @@ fn select_workload_impl(
         .iter()
         .map(|(id, root)| {
             selection
-                .materialize(root)
+                .assemble_selected_dag(root)
                 .map_err(|error| SelectionError::Workload(error.to_string()))?
                 .map(|node| (*id, node))
                 .ok_or_else(|| SelectionError::Workload(format!("missing query root {id}")))

crates/asap_types/src/derived_input.rs

@@ -106,7 +106,7 @@ impl DerivedInputIdentity {
             }
             edges.sort();
             let bytes = serde_json::to_vec(&serde_json::json!({
-                "version": 1, "operator": node.operator, "payload": node.payload,
+                "version": 1, "payload": node.payload,
                 "state": node.output_state, "schema": node.output_schema,
                 "guarantee": node.guarantee, "inputs": edges,
             }))
@@ -208,19 +208,17 @@ mod tests {
     fn program(source: u32, root: u32) -> OwnedPostAsapDag {
         use crate::executable_plan::{OwnedPostAsapEdge, OwnedPostAsapNode};
         use planner_types::post_asap::{
-            EdgeRole, ExecutableOperator, ExecutionDataState, GroupingEdgeCompatibility,
-            WindowEdgeCompatibility,
+            EdgeRole, ExecutionDataState, GroupingEdgeCompatibility, WindowEdgeCompatibility,
         };
         let state = ExecutionDataState::MAINTENANCE_SUMMARY;
         OwnedPostAsapDag {
-            schema_version: 1,
+            schema_version: crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION,
             query_id: "query-a".into(),
             root: PostAsapNodeId(root),
             nodes: [source, root]
                 .into_iter()
                 .map(|id| OwnedPostAsapNode {
                     id: PostAsapNodeId(id),
-                    operator: ExecutableOperator::SummaryMerge,
                     payload: serde_json::json!({"kind":"summary_merge"}),
                     output_state: state,
                     output_schema: serde_json::json!({"fields":[],"time_index":null}),
@@ -353,13 +351,12 @@ mod tests {
     #[test]
     fn literal_leaves_are_hashed_without_inventing_materialization_references() {
         use planner_types::{
-            post_asap::{ExecutableOperator, ExecutableOperatorPayload},
+            post_asap::ExecutableOperatorPayload,
             pre_asap::{QueryExpr, ScalarValue},
         };
         let mut dag = program(1, 2);
         let mut literal = dag.nodes[0].clone();
         literal.id = PostAsapNodeId(3);
-        literal.operator = ExecutableOperator::Fallback;
         literal.payload = serde_json::to_value(ExecutableOperatorPayload::Fallback {
             expression: QueryExpr::Literal(ScalarValue::Int64(2)),
         })

crates/asap_types/src/executable_plan.rs

@@ -10,9 +10,8 @@ use std::collections::{BTreeMap, BTreeSet};
 
 use crate::sds::SummaryDefinitionId;
 use planner_types::post_asap::{
-    EdgeRole, ExecutableDag, ExecutableDagEdge, ExecutableDagNode, ExecutableOperator,
-    ExecutionDataState, ExecutionTiming, GroupingEdgeCompatibility, PostAsapNodeId,
-    WindowEdgeCompatibility,
+    EdgeRole, ExecutableDag, ExecutableDagEdge, ExecutableDagNode, ExecutionDataState,
+    ExecutionTiming, GroupingEdgeCompatibility, PostAsapNodeId, WindowEdgeCompatibility,
 };
 use serde::{Deserialize, Serialize};
 
@@ -21,7 +20,7 @@ use serde::{Deserialize, Serialize};
 #[serde(transparent)]
 pub struct QueryNodeId(pub u64);
 
-pub const OWNED_POST_ASAP_DAG_SCHEMA_VERSION: u32 = 1;
+pub const OWNED_POST_ASAP_DAG_SCHEMA_VERSION: u32 = 2;
 
 /// Versioned, language-neutral Planner DAG persisted with an installed plan.
 /// Plan lifecycle belongs to the enclosing `PrecomputePlan`; this document
@@ -43,7 +42,6 @@ pub struct OwnedPostAsapDag {
 #[serde(deny_unknown_fields)]
 pub struct OwnedPostAsapNode {
     pub id: PostAsapNodeId,
-    pub operator: ExecutableOperator,
     pub payload: serde_json::Value,
     pub output_state: ExecutionDataState,
     pub output_schema: serde_json::Value,
@@ -70,7 +68,6 @@ impl OwnedPostAsapDag {
             .map(|node| {
                 Ok(OwnedPostAsapNode {
                     id: node.id,
-                    operator: node.operator,
                     payload: serde_json::to_value(&node.payload).map_err(|e| e.to_string())?,
                     output_state: node.output_state,
                     output_schema: serde_json::to_value(&node.output_schema)
@@ -138,15 +135,8 @@ impl OwnedPostAsapDag {
             .map(|node| {
                 let payload: planner_types::post_asap::ExecutableOperatorPayload =
                     serde_json::from_value(node.payload.clone()).map_err(|e| e.to_string())?;
-                if payload.operator() != node.operator {
-                    return Err(format!(
-                        "post-ASAP node {} operator disagrees with payload",
-                        node.id.0
-                    ));
-                }
                 Ok(ExecutableDagNode {
                     id: node.id,
-                    operator: node.operator,
                     payload,
                     output_state: node.output_state,
                     output_schema: serde_json::from_value(node.output_schema.clone())
@@ -287,7 +277,7 @@ mod tests {
         fn send_sync<T: Send + Sync>() {}
         send_sync::<InstalledPostAsapDag>();
         let wire = serde_json::json!({
-            "schema_version": 1, "query_id": "q", "nodes": [], "edges": [], "root": 0
+            "schema_version": 2, "query_id": "q", "nodes": [], "edges": [], "root": 0
         });
         let document: OwnedPostAsapDag = serde_json::from_value(wire.clone()).unwrap();
         assert_eq!(serde_json::to_value(document).unwrap(), wire);

crates/asap_types/src/precompute_plan.rs

@@ -1031,7 +1031,6 @@ mod source_window_cohort_tests {
         config.partitioning = Some(crate::sds::PopulationPartitioning::Grouped);
         let mut node = ExecutableDagNode {
             id: PostAsapNodeId(1),
-            operator: ExecutableOperator::SummaryAgg,
             payload: ExecutableOperatorPayload::SummaryAgg {
                 family: SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum),
                 input: SummaryUpdate {

data_plane/src/precompute_engine/maintenance_runtime.rs

@@ -305,7 +305,7 @@ impl PrecomputeOperatorRegistry<MaintenanceValue> for OperatorAdapter<'_> {
             }
             payload => Err(format!(
                 "maintenance operator {:?} has no summary-state implementation",
-                payload.operator()
+                payload
             )),
         }
     }
@@ -2149,8 +2149,8 @@ mod tests {
     use super::*;
     use crate::precompute_engine::operators::SumAccumulator;
     use planner_types::post_asap::{
-        EdgeRole, ExecutableDag, ExecutableDagEdge, ExecutableOperator, GroupingEdgeCompatibility,
-        SummarySchema, WindowEdgeCompatibility,
+        EdgeRole, ExecutableDag, ExecutableDagEdge, GroupingEdgeCompatibility, SummarySchema,
+        WindowEdgeCompatibility,
     };
 
     fn definition(value: u64) -> asap_types::sds::SummaryDefinitionId {
@@ -2222,7 +2222,6 @@ mod tests {
     fn node(id: u32) -> ExecutableDagNode {
         ExecutableDagNode {
             id: PostAsapNodeId(id),
-            operator: ExecutableOperator::SummaryMerge,
             payload: ExecutableOperatorPayload::SummaryMerge,
             output_state: planner_types::post_asap::ExecutionDataState::MAINTENANCE_SUMMARY,
             output_schema: SummarySchema {
@@ -2463,9 +2462,7 @@ mod tests {
             dtype: SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum),
             nullable: false,
         }];
-        read.operator = read.payload.operator();
         read.output_state = planner_types::post_asap::ExecutionDataState::MAINTENANCE_ROWS;
-        aggregate.operator = aggregate.payload.operator();
         aggregate.output_schema.fields = vec![SummaryField {
             name: "state".into(),
             dtype: configs[1].accumulator_spec().unwrap().family,
@@ -2880,7 +2877,6 @@ mod tests {
         second_node.id = PostAsapNodeId(5);
         let mut merge = second_node.clone();
         merge.id = PostAsapNodeId(6);
-        merge.operator = ExecutableOperator::SummaryMerge;
         merge.payload = ExecutableOperatorPayload::SummaryMerge;
         dag.nodes.extend([second_node, merge]);
         let original = dag
@@ -3559,7 +3555,6 @@ mod tests {
             configs: &[],
         };
         let mut aggregate = node(1);
-        aggregate.operator = ExecutableOperator::SummaryAgg;
         aggregate.payload = ExecutableOperatorPayload::SummaryAgg {
             family: SummaryFamilyType::ExactAggregate(ExactKind::Count, ExactParams::Count),
             input: SummaryUpdate::column(ColumnRef::SampleValue),
@@ -4046,7 +4041,6 @@ mod tests {
     #[test]
     fn unsupported_maintenance_operator_propagates_failure_without_commit() {
         let mut unsupported = node(1);
-        unsupported.operator = ExecutableOperator::SummarySubtract;
         unsupported.payload = ExecutableOperatorPayload::SummarySubtract;
         let mut query = node(2);
         query.output_state = planner_types::post_asap::ExecutionDataState::READ_ROWS;

data_plane/src/precompute_engine/subdag_scheduler.rs

@@ -196,8 +196,8 @@ mod tests {
     use super::*;
     use planner_types::post_asap::{EdgeRole, ExecutionDataState};
     use planner_types::post_asap::{
-        ExecutableDagEdge, ExecutableOperator, ExecutableOperatorPayload,
-        GroupingEdgeCompatibility, SummarySchema, WindowEdgeCompatibility,
+        ExecutableDagEdge, ExecutableOperatorPayload, GroupingEdgeCompatibility, SummarySchema,
+        WindowEdgeCompatibility,
     };
     use std::sync::Mutex;
 
@@ -229,7 +229,6 @@ mod tests {
     fn node(id: u32) -> ExecutableDagNode {
         ExecutableDagNode {
             id: PostAsapNodeId(id),
-            operator: ExecutableOperator::SummarySubtract,
             payload: ExecutableOperatorPayload::SummarySubtract,
             output_state: ExecutionDataState::MAINTENANCE_SUMMARY,
             output_schema: SummarySchema {
@@ -324,7 +323,6 @@ mod tests {
             }
         }
         let mut binary = node(3);
-        binary.operator = ExecutableOperator::Binary;
         binary.payload = ExecutableOperatorPayload::Binary {
             timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime,
             operator: BinaryOperator {