diff --git a/Cargo.lock b/Cargo.lock index 99de8350c..a49ca3f33 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -364,7 +364,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f46cbf6c5738db2f4460d419baa8af5572f5276a#f46cbf6c5738db2f4460d419baa8af5572f5276a" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" dependencies = [ "asap-types", "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib)", @@ -376,7 +376,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f46cbf6c5738db2f4460d419baa8af5572f5276a#f46cbf6c5738db2f4460d419baa8af5572f5276a" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" dependencies = [ "asap-types", "promql-parser 0.10.0 (git+https://github.com/ProjectASAP/promql-parser?rev=9fede7eecca923c9882fe256484d00d37f8706cb)", @@ -385,7 +385,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f46cbf6c5738db2f4460d419baa8af5572f5276a#f46cbf6c5738db2f4460d419baa8af5572f5276a" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -408,12 +408,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f46cbf6c5738db2f4460d419baa8af5572f5276a#f46cbf6c5738db2f4460d419baa8af5572f5276a" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f46cbf6c5738db2f4460d419baa8af5572f5276a#f46cbf6c5738db2f4460d419baa8af5572f5276a" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=10d93846f0ef9f06ab19e599701d699d9cdd15af#10d93846f0ef9f06ab19e599701d699d9cdd15af" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index df62d75f4..f1da29dae 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/control_plane/src/physical/compiler/windows.rs b/control_plane/src/physical/compiler/windows.rs index d135ab17d..072ee7c2e 100644 --- a/control_plane/src/physical/compiler/windows.rs +++ b/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( + "e, + 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. diff --git a/control_plane/src/physical/plan_dot.rs b/control_plane/src/physical/plan_dot.rs index d6ced46ec..67bad1b79 100644 --- a/control_plane/src/physical/plan_dot.rs +++ b/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 + ), "", ); } diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index bedc1e83b..b536460cc 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -100,7 +100,7 @@ fn summary_identity(node: &SummaryNode) -> Option { 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::>(); - 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::>(); @@ -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}"))) diff --git a/crates/asap_types/src/derived_input.rs b/crates/asap_types/src/derived_input.rs index 964ee8d3f..018635c1a 100644 --- a/crates/asap_types/src/derived_input.rs +++ b/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)), }) diff --git a/crates/asap_types/src/executable_plan.rs b/crates/asap_types/src/executable_plan.rs index 04647cd44..ed40942aa 100644 --- a/crates/asap_types/src/executable_plan.rs +++ b/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() {} send_sync::(); 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); diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index a219ff03b..746649fc1 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/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 { diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 965dab547..e8f416e55 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -305,7 +305,7 @@ impl PrecomputeOperatorRegistry 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; diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index 918f84779..4cbd57a66 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/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 {