merge: stack PR 749 on current PR 737 and main

Author: zzylol

Cargo.lock

@@ -3,7 +3,6 @@ resolver = "2"
 members = [
     "crates/asap_otel_proto",
     "crates/asap_types",
-    "crates/asap_sketch_codec",
     "data_plane",
     "control_plane",
 ]
@@ -12,13 +11,19 @@ members = [
 edition = "2021"
 version = "0.1.0"
 
+# ASAPCollector's `asap-precompute-rs` currently declares Sketchlib as a
+# relative path. When Collector is consumed from Git, resolve that dependency
+# to the same Git-sourced Sketchlib package as the backend.
+[patch."https://github.com/ProjectASAP/ASAPCollector"]
+asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", branch = "main" }
+
 [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 = "851493116d42674d09cf9646dde59532003025e4" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "851493116d42674d09cf9646dde59532003025e4" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "851493116d42674d09cf9646dde59532003025e4" }
-asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "851493116d42674d09cf9646dde59532003025e4" }
+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"] }

Cargo.toml

@@ -310,14 +310,27 @@ mod tests {
             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/compiler/windows.rs

@@ -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}")))

control_plane/src/planner_selection.rs

@@ -20,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;
 pub const MAINTENANCE_DAG_SCHEMA_VERSION: u32 = 2;
 
 /// Versioned, language-neutral Planner DAG persisted with an installed plan.