refactor: split backend plans and remove Collector dependency

Cargo.toml

@@ -3,6 +3,7 @@ resolver = "2"
 members = [
     "crates/asap_otel_proto",
     "crates/asap_types",
+    "crates/asap_sketch_codec",
     "data_plane",
     "control_plane",
 ]
@@ -11,12 +12,6 @@ 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."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/ProjectASAP/ASAPCollector"]
-asap_sketchlib = { git = "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/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.

control_plane/src/clickhouse.rs

@@ -247,6 +247,7 @@ pub async fn compile_automatic_clickhouse_workload(
     let mut entries = std::collections::BTreeMap::new();
     let mut window_templates = std::collections::BTreeMap::<String, Vec<String>>::new();
     let mut installed_dags = std::collections::BTreeMap::new();
+    let mut selected_dags = std::collections::BTreeMap::new();
     let mut materializations = std::collections::BTreeMap::new();
     let mut selection_traces = std::collections::BTreeMap::new();
     for query in &request.queries {
@@ -286,7 +287,13 @@ pub async fn compile_automatic_clickhouse_workload(
                 "duplicate canonical SQL query identity".into(),
             ));
         }
-        installed_dags.insert(query.sql.clone(), installed);
+        selected_dags.insert(query.sql.clone(), installed.document.clone());
+        installed_dags.insert(
+            query.sql.clone(),
+            installed
+                .maintenance_projection()
+                .map_err(ClickHousePlanningError::Lower)?,
+        );
     }
     let configs: Vec<_> = materializations.into_values().collect();
     let sds = SummaryCatalog::from_materializations(
@@ -322,6 +329,7 @@ pub async fn compile_automatic_clickhouse_workload(
                 tables: request.tables.clone(),
                 accuracy: request.accuracy.clone(),
             }),
+            selected_dags,
             entries,
         },
     };
@@ -423,14 +431,21 @@ pub async fn compile_clickhouse_workload(
     let mut entries = std::collections::BTreeMap::new();
     let mut window_templates = std::collections::BTreeMap::<String, Vec<String>>::new();
     let mut installed_dags = std::collections::BTreeMap::new();
+    let mut selected_dags = std::collections::BTreeMap::new();
     for query in &request.queries {
         let planned = plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?;
         let template = planned.canonical_sql.clone();
         let (executable, installed) = compile_selected_sql(query, planned, |node, family| {
             bind_selected_node(node, family, query, request)
         })?;
         index_sql_template(&mut window_templates, template, &executable);
-        installed_dags.insert(query.sql.clone(), installed);
+        selected_dags.insert(query.sql.clone(), installed.document.clone());
+        installed_dags.insert(
+            query.sql.clone(),
+            installed
+                .maintenance_projection()
+                .map_err(ClickHousePlanningError::Lower)?,
+        );
         let identity =
             QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &executable.canonical_query);
         if entries.insert(identity.clone(), executable).is_some() {
@@ -454,6 +469,7 @@ pub async fn compile_clickhouse_workload(
                 tables: request.tables.clone(),
                 accuracy: request.accuracy.clone(),
             }),
+            selected_dags,
             entries,
         },
     };
@@ -1641,10 +1657,16 @@ mod tests {
             installed.binding.nodes.len(),
             installed.document.nodes.len()
         );
-        assert!(installed.binding.nodes.values().any(|binding| matches!(
+        assert!(!installed.binding.nodes.values().any(|binding| matches!(
             binding,
             crate::physical::executable_binding::BackendNodeBinding::Query { .. }
         )));
+        assert!(
+            publication.query_plan.selected_dags[&request.queries[0].sql]
+                .nodes
+                .len()
+                > installed.document.nodes.len()
+        );
         let entry = publication.query_plan.entries.values().next().unwrap();
         // External SQL retains its literal time range until it can be bound.
         assert!(!entry.canonical_query.starts_with("moving-window-v1:"));

control_plane/src/physical/compiler.rs

@@ -1852,10 +1852,11 @@ impl PhysicalPlanCompiler {
                 });
             }
         }
-        let query_plan = QueryPlan {
+        let mut query_plan = QueryPlan {
             plan_id,
             plan_version: envelope.plan_version,
             clickhouse_context: None,
+            selected_dags: BTreeMap::new(),
             entries: query_entries,
         };
         let mut installed_dags = BTreeMap::new();
@@ -1884,6 +1885,9 @@ impl PhysicalPlanCompiler {
                 query_id: query_id.clone(),
                 reason,
             })?;
+            query_plan
+                .selected_dags
+                .insert(query_id.clone(), installed.document.clone());
             installed_dags.insert(query_id, installed);
         }
         for materialization in &mut materializations {
@@ -1935,6 +1939,18 @@ impl PhysicalPlanCompiler {
                 }
             })?;
         }
+        // QueryPlan already owns executable readout nodes. Persist only
+        // maintenance ancestors under PrecomputePlan.
+        installed_dags = installed_dags
+            .into_iter()
+            .filter(|(_, installed)| !installed.binding.precompute_sinks.is_empty())
+            .map(|(query_id, installed)| {
+                installed
+                    .maintenance_projection()
+                    .map(|projected| (query_id.clone(), projected))
+                    .map_err(|reason| CompileError::Query { query_id, reason })
+            })
+            .collect::<Result<_, _>>()?;
         let mut precompute_plan = match environment.target {
             PhysicalDeploymentTarget::DistributedCollectors => {
                 PrecomputePlan::build(envelope.clone(), materializations, &producer_ids).and_then(
@@ -3900,8 +3916,8 @@ pub(crate) mod tests {
         assert_eq!(populations.len(), 1);
         let installed = serde_json::to_string(&plan.precompute_plan.executable_dags).unwrap();
         assert!(
-            installed.contains("MaintainPopulation"),
-            "shared state must originate in the installed Planner DAG"
+            !installed.contains("MaintainPopulation"),
+            "query-only population readout must not be executable maintenance"
         );
     }
 
@@ -4601,11 +4617,27 @@ pub(crate) mod tests {
             .precompute_plan
             .executable_dags
             .get(&entry.query_id)
-            .expect("compiled query retains its Planner DAG and backend placement");
+            .expect("compiled query retains its maintenance projection");
         installed.validate().expect("typed DAG document");
-        crate::physical::executable_binding::validate_query_plan(installed, entry)
-            .expect("query node bindings");
+        assert_eq!(
+            installed.document.schema_version,
+            asap_types::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION
+        );
+        assert!(installed
+            .document
+            .nodes
+            .iter()
+            .all(|node| node.output_state.timing
+                == planner_types::post_asap::ExecutionTiming::MaintenanceTime));
         assert_eq!(installed.binding.query_plan_sink, entry.root);
+        let mut mismatched = plan.to_publication_artifact().unwrap();
+        let projected = mismatched
+            .precompute_plan
+            .executable_dags
+            .get_mut(&entry.query_id)
+            .unwrap();
+        projected.binding.query_plan_sink = asap_types::executable_plan::QueryNodeId(u64::MAX);
+        assert!(mismatched.validate().is_err());
         assert!(installed.binding.nodes.values().any(|placement| matches!(
             placement,
             crate::physical::executable_binding::BackendNodeBinding::Materialization { .. }

control_plane/src/query_plan.rs

@@ -1050,6 +1050,7 @@ mod catalog_binding_tests {
                 plan_id: 7,
                 plan_version: 2,
                 clickhouse_context: None,
+                selected_dags: Default::default(),
                 entries: BTreeMap::from([(entry.canonical_query.clone(), entry)]),
             },
             catalog,
@@ -1328,6 +1329,7 @@ mod tests {
                 tables: Default::default(),
                 accuracy: planner_types::types::AccuracyTarget::Exact,
             }),
+            selected_dags: Default::default(),
             entries: [base, metricsql, clickhouse]
                 .into_iter()
                 .map(|entry| (QueryPlan::catalog_key(entry.language, "shared"), entry))

crates/asap_sketch_codec/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "asap_sketch_codec"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+asap_sketchlib = { git = "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/ProjectASAP/asap_sketchlib", branch = "main" }
+prost = "0.13"

crates/asap_sketch_codec/src/lib.rs

@@ -0,0 +1,93 @@
+//! Runtime-independent decoding of the sketchlib protobuf envelope.
+
+use asap_sketchlib::proto::sketchlib::{
+    sketch_envelope::SketchState, DdSketchState, KllState, SketchEnvelope,
+};
+use asap_sketchlib::DdSketch;
+use prost::Message;
+
+pub fn envelope_state(bytes: &[u8]) -> Result<Option<SketchState>, String> {
+    SketchEnvelope::decode(bytes)
+        .map(|envelope| envelope.sketch_state)
+        .map_err(|error| format!("decode SketchEnvelope: {error}"))
+}
+
+/// Accept the current full envelope and the supported legacy bare state.
+pub fn ddsketch_state(bytes: &[u8]) -> Result<(DdSketchState, f64), String> {
+    match SketchEnvelope::decode(bytes) {
+        Ok(envelope) => match envelope.sketch_state {
+            Some(SketchState::Ddsketch(state)) => Ok((state, envelope.sample_p)),
+            Some(_) => Err("SketchEnvelope contains a non-DDSketch state".into()),
+            None => DdSketchState::decode(bytes)
+                .map(|state| (state, 1.0))
+                .map_err(|error| format!("decode DdSketchState: {error}")),
+        },
+        Err(_) => DdSketchState::decode(bytes)
+            .map(|state| (state, 1.0))
+            .map_err(|error| format!("decode DdSketchState: {error}")),
+    }
+}
+
+pub fn reconstruct_ddsketch(bytes: &[u8]) -> Result<(DdSketch, f64), String> {
+    let (state, sample_p) = ddsketch_state(bytes)?;
+    if !state.alpha.is_finite() || !(0.0..1.0).contains(&state.alpha) || state.alpha == 0.0 {
+        return Err("DDSketch alpha must be finite and between zero and one".into());
+    }
+    Ok((
+        DdSketch::from_raw(state.alpha, state.store_counts, state.store_offset),
+        sample_p,
+    ))
+}
+
+pub fn kll_state(bytes: &[u8]) -> Result<KllState, String> {
+    match SketchEnvelope::decode(bytes) {
+        Ok(envelope) => match envelope.sketch_state {
+            Some(SketchState::Kll(state)) => Ok(state),
+            Some(_) => Err("SketchEnvelope contains a non-KLL state".into()),
+            None => KllState::decode(bytes).map_err(|error| format!("decode KllState: {error}")),
+        },
+        Err(_) => KllState::decode(bytes).map_err(|error| format!("decode KllState: {error}")),
+    }
+}
+
+pub fn encode_ddsketch(sketch: &DdSketch) -> Vec<u8> {
+    let envelope = SketchEnvelope {
+        format_version: 1,
+        producer: None,
+        hash_spec: None,
+        sample_p: 0.0,
+        sketch_state: Some(SketchState::Ddsketch(DdSketchState {
+            alpha: sketch.wire_alpha(),
+            store_counts: sketch.store_counts.clone(),
+            store_offset: sketch.store_offset,
+        })),
+    };
+    envelope.encode_to_vec()
+}
+
+pub fn encode_kll(sketch: &asap_sketchlib::sketches::kll::KLL<f64>) -> Vec<u8> {
+    use asap_sketchlib::proto::sketchlib::CoinState;
+    let (state, bit_cache, remaining_bits) = sketch.wire_coin();
+    SketchEnvelope {
+        format_version: 1,
+        producer: None,
+        hash_spec: None,
+        sample_p: 0.0,
+        sketch_state: Some(SketchState::Kll(KllState {
+            k: sketch.wire_k(),
+            m: sketch.wire_m(),
+            num_levels: sketch.wire_num_levels(),
+            levels: sketch.wire_levels(),
+            items: sketch.wire_items(),
+            coin: Some(CoinState {
+                state,
+                bit_cache,
+                remaining_bits,
+            }),
+            offset: 0.0,
+            value_scale: 0,
+            residuals: Vec::new(),
+        })),
+    }
+    .encode_to_vec()
+}

crates/asap_types/src/derived_input.rs

@@ -37,7 +37,11 @@ impl DerivedInputIdentity {
         root: PostAsapNodeId,
         frontiers: &BTreeMap<PostAsapNodeId, SummaryDefinitionId>,
     ) -> Result<Self, String> {
-        if document.schema_version != crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION {
+        if !matches!(
+            document.schema_version,
+            crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION
+                | crate::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION
+        ) {
             return Err("unsupported derived program document version".into());
         }
         let decoded = document.decode()?;