Share precompute installation contracts and catalog validation

control_plane/src/physical/mod.rs

@@ -31,7 +31,6 @@ pub mod plan;
 pub mod plan_cache;
 pub mod planner;
 pub mod post_asap;
-pub mod precompute_contract;
 pub mod runtime_capability;
 pub mod sketch_catalog;
 pub mod stage_split;

crates/asap_types/src/lib.rs

@@ -28,3 +28,5 @@ pub use policy_registry::PolicyRegistry;
 pub use query_requirements::*;
 pub use routing_index::RoutingIndex;
 pub use storage_backend::*;
+
+pub mod precompute_plan;

control_plane/src/physical/precompute_contract.rs → crates/asap_types/src/precompute_plan/catalog.rs

@@ -1,9 +1,9 @@
 //! Catalog consistency checks for the precompute execution plan.
-use super::compiler::*;
-use super::summary_catalog::SummaryCatalog;
-use asap_types::sds::{
+use super::*;
+use crate::sds::{
     DataSourceIdentity, SummaryDefinitionId, SummaryDescriptor, ValueProjectionIdentity,
 };
+use crate::summary_catalog::SummaryCatalog;
 use planner_types::pre_asap::{ColumnRef, Source};
 use std::collections::BTreeSet;
 fn invalid(reason: impl Into<String>) -> PrecomputePlanError {
@@ -89,7 +89,7 @@ impl PrecomputePlan {
             if data.source != expected_source
                 || data.value_projection != expected_projection
                 || data.population_filter_canonical
-                    != asap_types::utils::normalize_spatial_filter(&config.spatial_filter)
+                    != crate::utils::normalize_spatial_filter(&config.spatial_filter)
                 || data.group_by_keys != config.grouping_labels.labels.iter().cloned().collect()
             {
                 return Err(invalid("source/population/grouping differs from catalog"));
@@ -133,9 +133,9 @@ impl PrecomputePlan {
                 .filter(|v| *v > 0)
                 .ok_or_else(|| invalid("invalid slide interval"))?;
             let expected_slide = match config.window_type {
-                asap_types::WindowKind::Tumbling => None,
-                asap_types::WindowKind::Sliding => Some(slide),
-                asap_types::WindowKind::Session => {
+                crate::WindowKind::Tumbling => None,
+                crate::WindowKind::Sliding => Some(slide),
+                crate::WindowKind::Session => {
                     return Err(invalid("session lifecycle is not supported"))
                 }
             };

data_plane/src/drivers/ingest/otel.rs

@@ -2411,10 +2411,8 @@ fn take_summary_frame_identity(
     })
 }
 
-fn state_encoding_for_wire(
-    encoding: i32,
-) -> Option<control_plane::physical::compiler::StateEncoding> {
-    use control_plane::physical::compiler::StateEncoding;
+fn state_encoding_for_wire(encoding: i32) -> Option<asap_types::precompute_plan::StateEncoding> {
+    use asap_types::precompute_plan::StateEncoding;
     match encoding {
         ENCODING_PROTO | ENCODING_PROTO_DELTA => Some(StateEncoding::SketchlibProtobufV1),
         ENCODING_MSGPACK | ENCODING_MSGPACK_DELTA => Some(StateEncoding::SketchCoreMsgpackV1),

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -242,7 +242,7 @@ impl PrometheusRemoteWriteReceiver {
         if physical_plan.precompute_plan.envelope.plan_id == 0
             || !matches!(
                 physical_plan.precompute_plan.ingest.protocol,
-                control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1
+                asap_types::precompute_plan::IngestProtocol::PrometheusRemoteWriteV1
             )
             || physical_plan.precompute_plan.ingest.endpoint_path != "/api/v1/write"
         {
@@ -779,7 +779,7 @@ mod tests {
             generated_at_unix_ms: 1,
             activation_unix_ms: 1,
             expiry_unix_ms: None,
-            backend_compat: control_plane::physical::compiler::BACKEND_COMPAT.into(),
+            backend_compat: asap_types::precompute_plan::BACKEND_COMPAT.into(),
             planner_revision: PLANNER_REVISION.into(),
             capability_snapshot_id: "test".into(),
         };

data_plane/src/drivers/query/servers/http.rs

@@ -2690,7 +2690,7 @@ mod tests {
             generated_at_unix_ms: 0,
             activation_unix_ms: 1,
             expiry_unix_ms: None,
-            backend_compat: control_plane::physical::compiler::BACKEND_COMPAT.into(),
+            backend_compat: asap_types::precompute_plan::BACKEND_COMPAT.into(),
             planner_revision: PLANNER_REVISION.into(),
             capability_snapshot_id: "test".into(),
         };
@@ -5865,7 +5865,7 @@ async fn handle_health(State(state): State<AppState>) -> axum::response::Respons
             && active.expiry_unix_ms().is_none_or(|expiry| now < expiry);
         let ingest_ready = matches!(
             active.precompute_plan.ingest.protocol,
-            control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1
+            asap_types::precompute_plan::IngestProtocol::PrometheusRemoteWriteV1
         ) && active.precompute_plan.ingest.endpoint_path == "/api/v1/write";
         if !lifecycle_ready || !ingest_ready {
             return (
@@ -6187,7 +6187,7 @@ async fn handle_post_physical_plan(
         && (active.plan_id() == 0
             || !matches!(
                 active.precompute_plan.ingest.protocol,
-                control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1
+                asap_types::precompute_plan::IngestProtocol::PrometheusRemoteWriteV1
             )
             || active.precompute_plan.ingest.endpoint_path != "/api/v1/write")
     {

data_plane/src/main.rs

@@ -592,7 +592,7 @@ async fn main() -> Result<()> {
             && (active.plan_id() == 0
                 || !matches!(
                     active.precompute_plan.ingest.protocol,
-                    control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1
+                    asap_types::precompute_plan::IngestProtocol::PrometheusRemoteWriteV1
                 )
                 || active.precompute_plan.ingest.endpoint_path != "/api/v1/write")
         {
@@ -732,9 +732,9 @@ async fn main() -> Result<()> {
     };
 
     // Bootstrap projections share one immutable physical-plan envelope.
-    let initial_precompute_plan = control_plane::physical::compiler::PrecomputePlan {
+    let initial_precompute_plan = asap_types::precompute_plan::PrecomputePlan {
         summary_catalog: None,
-        envelope: control_plane::physical::compiler::PlanEnvelope {
+        envelope: asap_types::precompute_plan::PlanEnvelope {
             plan_id: 0,
             plan_version: 0,
             generated_at_unix_ms: 0,
@@ -744,10 +744,10 @@ async fn main() -> Result<()> {
             planner_revision: control_plane::physical::compiler::PLANNER_REVISION.into(),
             capability_snapshot_id: "bootstrap".into(),
         },
-        ingest: control_plane::physical::compiler::IngestContract {
-            protocol: control_plane::physical::compiler::IngestProtocol::ModifiedOtlpMetricsV1,
+        ingest: asap_types::precompute_plan::IngestContract {
+            protocol: asap_types::precompute_plan::IngestProtocol::ModifiedOtlpMetricsV1,
             endpoint_path: "/v1/metrics".into(),
-            timestamp_unit: control_plane::physical::compiler::TimestampUnit::UnixNanoseconds,
+            timestamp_unit: asap_types::precompute_plan::TimestampUnit::UnixNanoseconds,
             require_plan_identity: false,
             require_summary_definition_identity: false,
             require_registered_producer: false,

data_plane/src/precompute_engine/frame_lineage.rs

@@ -191,7 +191,7 @@ impl FrameLineageTracker {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use control_plane::physical::compiler::StateEncoding;
+    use asap_types::precompute_plan::StateEncoding;
 
     fn frame(sequence: u64, kind: SummaryFrameKind) -> SummaryFrameIdentity {
         SummaryFrameIdentity {

data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs

@@ -647,17 +647,17 @@ mod tests {
                 Box::new(SumAccumulator::with_sum(3.0)),
             );
         }
-        let envelope = control_plane::physical::compiler::PlanEnvelope {
+        let envelope = asap_types::precompute_plan::PlanEnvelope {
             plan_id: 41,
             plan_version: 1,
             generated_at_unix_ms: 0,
             activation_unix_ms: 0,
             expiry_unix_ms: None,
-            backend_compat: control_plane::physical::compiler::BACKEND_COMPAT.into(),
+            backend_compat: asap_types::precompute_plan::BACKEND_COMPAT.into(),
             planner_revision: control_plane::physical::compiler::PLANNER_REVISION.into(),
             capability_snapshot_id: "clickhouse-test".into(),
         };
-        let mut precompute = control_plane::physical::compiler::PrecomputePlan::build(
+        let mut precompute = asap_types::precompute_plan::PrecomputePlan::build(
             envelope.clone(),
             vec![config],
             &["fixture".into()],

data_plane/src/storage_engines/types/hot_reload_config.rs

@@ -88,10 +88,10 @@ use crate::storage_engines::types::StreamingConfig;
 pub struct ActivePhysicalPlan {
     /// Authoritative generation and lifecycle identity shared by every plan
     /// projection in this immutable snapshot.
-    pub envelope: control_plane::physical::compiler::PlanEnvelope,
+    pub envelope: asap_types::precompute_plan::PlanEnvelope,
     /// Present for authoritative installations; legacy bootstrap has no catalog.
     pub summary_catalog: Option<Arc<control_plane::physical::summary_catalog::SummaryCatalog>>,
-    pub precompute_plan: control_plane::physical::compiler::PrecomputePlan,
+    pub precompute_plan: asap_types::precompute_plan::PrecomputePlan,
     pub transmission_plan: control_plane::physical::compiler::TransmissionPlan,
     pub runtime_config: Arc<StreamingConfig>,
     pub query_plan: Arc<control_plane::query_plan::QueryPlan>,
@@ -655,7 +655,7 @@ mod tests {
         activation_unix_ms: u64,
         expiry_unix_ms: Option<u64>,
     ) -> ActivePhysicalPlan {
-        let envelope = control_plane::physical::compiler::PlanEnvelope {
+        let envelope = asap_types::precompute_plan::PlanEnvelope {
             plan_id,
             plan_version,
             generated_at_unix_ms: activation_unix_ms,
@@ -668,15 +668,15 @@ mod tests {
         ActivePhysicalPlan {
             envelope: envelope.clone(),
             summary_catalog: None,
-            precompute_plan: control_plane::physical::compiler::PrecomputePlan {
+            precompute_plan: asap_types::precompute_plan::PrecomputePlan {
                 summary_catalog: None,
                 envelope: envelope.clone(),
-                ingest: control_plane::physical::compiler::IngestContract {
+                ingest: asap_types::precompute_plan::IngestContract {
                     protocol:
-                        control_plane::physical::compiler::IngestProtocol::ModifiedOtlpMetricsV1,
+                        asap_types::precompute_plan::IngestProtocol::ModifiedOtlpMetricsV1,
                     endpoint_path: "/v1/metrics".into(),
                     timestamp_unit:
-                        control_plane::physical::compiler::TimestampUnit::UnixNanoseconds,
+                        asap_types::precompute_plan::TimestampUnit::UnixNanoseconds,
                     require_plan_identity: true,
                     require_summary_definition_identity: true,
                     require_registered_producer: true,
@@ -688,7 +688,7 @@ mod tests {
             },
             transmission_plan: control_plane::physical::compiler::TransmissionPlan {
                 summary_catalog: None,
-                envelope: control_plane::physical::compiler::PlanEnvelope {
+                envelope: asap_types::precompute_plan::PlanEnvelope {
                     plan_id,
                     plan_version,
                     generated_at_unix_ms: activation_unix_ms,

docs/design_docs/summary-catalog-sds-architecture.md

@@ -155,7 +155,10 @@ representation for shared runtime snapshots; it is not Planner's
 node IDs and typed operator tags while serializing Planner payloads that contain
 process-local `Rc` pointers. The control plane constructs it and checks its
 bindings against QueryPlan; precompute execution consumes the shared contract.
-`QueryPlan` and `PrecomputePlan` definitions still reside in the control-plane
+`PrecomputePlan`, its envelope, ingest, producer, state schema, and catalog
+consistency checks live in `asap_types::precompute_plan`. The compiler chooses
+materializations and placement; data-plane installation uses the shared
+contract. `QueryPlan` definitions still reside in the control-plane
 crate while their remaining compilation methods are separated from wire types.
 
 The implemented ownership split is: