From f90936e8b7483033c2f7c602e9c051694137fd4f Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:39:45 -0600 Subject: [PATCH] refactor: share precompute installation contracts --- control_plane/src/physical/compiler.rs | 512 +---------------- control_plane/src/physical/mod.rs | 1 - crates/asap_types/src/lib.rs | 2 + crates/asap_types/src/precompute_plan.rs | 517 ++++++++++++++++++ .../asap_types/src/precompute_plan/catalog.rs | 14 +- data_plane/src/drivers/ingest/otel.rs | 6 +- .../drivers/ingest/prometheus_remote_write.rs | 4 +- data_plane/src/drivers/query/servers/http.rs | 6 +- data_plane/src/main.rs | 12 +- .../src/precompute_engine/frame_lineage.rs | 2 +- .../accelerator.rs | 6 +- .../types/hot_reload_config.rs | 16 +- .../summary-catalog-sds-architecture.md | 5 +- 13 files changed, 561 insertions(+), 542 deletions(-) create mode 100644 crates/asap_types/src/precompute_plan.rs rename control_plane/src/physical/precompute_contract.rs => crates/asap_types/src/precompute_plan/catalog.rs (95%) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index fdca6ad8..83b60d83 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -39,7 +39,7 @@ use planner_types::pre_asap::Source; pub const PLANNER_REVISION: &str = "0deceda3e776216c5542d638d958b159f22e27ce"; pub const BACKEND_REVISION: &str = env!("ASAPQUERY_BACKEND_REVISION"); -pub const BACKEND_COMPAT: &str = "asap-query-backend.v1"; +pub use asap_types::precompute_plan::BACKEND_COMPAT; /// Matches the data plane's default persistence memory limit. A backend-local /// summary candidate must fit its complete retained state inside this budget. pub const DEFAULT_RETAINED_SUMMARY_MEMORY_BUDGET_BYTES: u64 = 2 * 1024 * 1024 * 1024; @@ -252,17 +252,11 @@ fn u64_is_zero(value: &u64) -> bool { *value == 0 } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct PlanEnvelope { - pub plan_id: u64, - pub plan_version: u64, - pub generated_at_unix_ms: u64, - pub activation_unix_ms: u64, - pub expiry_unix_ms: Option, - pub backend_compat: String, - pub planner_revision: String, - pub capability_snapshot_id: String, -} +pub use asap_types::precompute_plan::{ + ExactStateKind, IngestContract, IngestProtocol, PlanEnvelope, PrecomputePlan, + PrecomputePlanError, ProducerContract, StateEncoding, StateFamilyContract, StateSchemaContract, + StateWindowContract, TimestampUnit, +}; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct CollectorMaterialization { @@ -307,472 +301,6 @@ pub struct CollectorPlan { pub transmission_rules: Vec, } -/// Backend-side materialization projection consumed by the streaming -/// precompute engine. This is deliberately config-driven: it contains no -/// PromQL string or ad-hoc scheduler job. The aggregation definitions are -/// emitted to `/api/v1/streaming-config`, where the runtime matches incoming -/// series, maintains windows, and writes content-addressed materializations. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct PrecomputePlan { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub summary_catalog: Option, - pub envelope: PlanEnvelope, - pub ingest: IngestContract, - pub schemas: Vec, - pub producers: Vec, - pub materializations: Vec, - /// Planner semantic DAGs and backend-owned placement for this generation. - /// Empty only for legacy/config-only construction paths. - #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] - pub executable_dags: - BTreeMap, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum IngestProtocol { - ModifiedOtlpMetricsV1, - PrometheusRemoteWriteV1, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum TimestampUnit { - UnixNanoseconds, - UnixMilliseconds, -} - -/// Backend ingress semantics installed with the precompute projection. This -/// replaces implicit knowledge formerly hidden in the streaming-config path. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct IngestContract { - pub protocol: IngestProtocol, - pub endpoint_path: String, - pub timestamp_unit: TimestampUnit, - pub require_plan_identity: bool, - #[serde(alias = "require_materialization_identity")] - pub require_summary_definition_identity: bool, - pub require_registered_producer: bool, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] -#[serde(rename_all = "snake_case")] -pub enum StateEncoding { - SketchlibProtobufV1, - SketchCoreMsgpackV1, - ExactAccumulatorV1, - ExactCounterAccumulatorV2, -} - -/// Serializable physical state identity derived from Planner's canonical -/// summary family. This is a wire DTO, not a second planning algebra. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(tag = "family", rename_all = "snake_case", deny_unknown_fields)] -pub enum StateFamilyContract { - Exact { - kind: ExactStateKind, - }, - Sketch { - algorithm: SketchAlgorithm, - parameters: SketchParams, - }, -} - -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum ExactStateKind { - Sum, - Count, - MinMax, - Increase, - Rate, - IRate, -} - -impl TryFrom<&SummaryFamilyType> for StateFamilyContract { - type Error = (); - - fn try_from(family: &SummaryFamilyType) -> Result { - use planner_types::post_asap::ExactKind; - Ok(match family { - SummaryFamilyType::ExactAggregate(kind, _) => Self::Exact { - kind: match kind { - ExactKind::Sum => ExactStateKind::Sum, - ExactKind::Count => ExactStateKind::Count, - ExactKind::MinMax => ExactStateKind::MinMax, - ExactKind::Increase => ExactStateKind::Increase, - ExactKind::Rate => ExactStateKind::Rate, - ExactKind::IRate => ExactStateKind::IRate, - }, - }, - SummaryFamilyType::Sketch(kind, _) => Self::Sketch { - algorithm: kind.algorithm().clone(), - parameters: kind.params().clone(), - }, - _ => return Err(()), - }) - } -} - -/// Decoder/schema contract for one content-addressed materialization. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(deny_unknown_fields)] -pub struct StateSchemaContract { - pub schema_id: String, - pub schema_version: u32, - pub materialization: asap_types::sds::SummaryDefinitionId, - pub family: StateFamilyContract, - pub source: Source, - pub value_column: planner_types::pre_asap::ColumnRef, - pub group_by: Vec, - pub window: StateWindowContract, - pub encodings: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct StateWindowContract { - pub kind: asap_types::WindowKind, - pub size_ms: u64, - pub slide_ms: Option, - #[serde( - default, - alias = "paneOriginMs", - skip_serializing_if = "Option::is_none" - )] - pub pane_origin_ms: Option, -} - -/// A collector authorized to produce state for one materialization. Runtime -/// producer epochs and frame sequences belong to TransmissionPlan. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] -#[serde(deny_unknown_fields)] -pub struct ProducerContract { - pub producer_id: String, - pub collector_id: String, - pub materialization: asap_types::sds::SummaryDefinitionId, - pub schema_id: String, -} - -#[derive(Debug, Error, PartialEq, Eq)] -pub enum PrecomputePlanError { - #[error("invalid precompute catalog contract: {0}")] - CatalogContract(String), - #[error("PrecomputePlan envelope does not match its SummaryCatalog identity/lifecycle")] - PlanIdentityMismatch, - #[error("unsupported precompute ingest protocol/endpoint/identity contract")] - UnsupportedIngestEndpoint, - #[error("duplicate materialization {0}")] - DuplicateMaterialization(u64), - #[error("schema set does not exactly match the materialization set")] - SchemaSetMismatch, - #[error("schema {schema_id} has invalid version or no encoding")] - InvalidSchema { schema_id: String }, - #[error("materialization {0} uses a summary family unsupported by the runtime schema")] - UnsupportedFamily(u64), - #[error("materialization {materialization} has an invalid window layout: {reason}")] - InvalidWindowLayout { - materialization: u64, - reason: String, - }, - #[error("producer {producer_id} references an unknown materialization or schema")] - InvalidProducer { producer_id: String }, - #[error("materialization {0} has no registered producer")] - MissingProducer(u64), - #[error("duplicate producer binding {0}")] - DuplicateProducer(String), -} - -impl PrecomputePlan { - pub fn build( - envelope: PlanEnvelope, - materializations: Vec, - producer_ids: &[String], - ) -> Result { - let schemas = materializations - .iter() - .map(|materialization| { - let fingerprint = materialization.policy_fingerprint(); - let accumulator = materialization - .accumulator_spec() - .map_err(|_| PrecomputePlanError::UnsupportedFamily(fingerprint.0))?; - let family = StateFamilyContract::try_from(&accumulator.family) - .map_err(|_| PrecomputePlanError::UnsupportedFamily(fingerprint.0))?; - let source = materialization.table_name.as_ref().map_or_else( - || Source::TimeSeries { - metric: materialization.metric.clone(), - }, - |table_ref| Source::Table { - table_ref: table_ref.clone(), - }, - ); - let value_column = materialization - .value_column - .clone() - .map(planner_types::pre_asap::ColumnRef::Named) - .unwrap_or(planner_types::pre_asap::ColumnRef::SampleValue); - Ok(StateSchemaContract { - schema_id: state_schema_id(fingerprint), - schema_version: 1, - materialization: fingerprint.into(), - family, - source, - value_column, - group_by: materialization.grouping_labels.labels.clone(), - window: StateWindowContract { - kind: materialization.window_type, - size_ms: materialization.window_size.saturating_mul(1_000), - slide_ms: match materialization.window_type { - asap_types::WindowKind::Tumbling => None, - asap_types::WindowKind::Sliding => { - Some(materialization.slide_interval.saturating_mul(1_000)) - } - asap_types::WindowKind::Session => None, - }, - pane_origin_ms: materialization.pane_origin_ms, - }, - encodings: state_encodings(&accumulator.family), - }) - }) - .collect::, PrecomputePlanError>>()?; - let producers = producer_ids - .iter() - .flat_map(|producer_id| { - schemas.iter().map(move |schema| ProducerContract { - producer_id: producer_id.clone(), - collector_id: producer_id.clone(), - materialization: schema.materialization, - schema_id: schema.schema_id.clone(), - }) - }) - .collect(); - let plan = Self { - summary_catalog: None, - envelope, - ingest: IngestContract { - protocol: IngestProtocol::ModifiedOtlpMetricsV1, - endpoint_path: "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/v1/metrics".into(), - timestamp_unit: TimestampUnit::UnixNanoseconds, - require_plan_identity: true, - require_summary_definition_identity: true, - require_registered_producer: true, - }, - schemas, - producers, - materializations, - executable_dags: BTreeMap::new(), - }; - plan.validate()?; - Ok(plan) - } - - /// Build the backend-local projection used when raw Prometheus samples - /// are precomputed inside ASAPQuery rather than by ASAPCollector. - pub fn build_backend_local( - envelope: PlanEnvelope, - materializations: Vec, - ) -> Result { - let mut plan = Self::build(envelope, materializations, &["backend-local".into()])?; - plan.ingest = IngestContract { - protocol: IngestProtocol::PrometheusRemoteWriteV1, - endpoint_path: "/api/v1/write".into(), - timestamp_unit: TimestampUnit::UnixMilliseconds, - require_plan_identity: false, - require_summary_definition_identity: false, - require_registered_producer: false, - }; - plan.producers.clear(); - plan.validate()?; - Ok(plan) - } - - pub fn runtime_materializations( - &self, - ) -> Result, PrecomputePlanError> { - self.validate()?; - Ok(self - .materializations - .iter() - .cloned() - .map(|materialization| (materialization.policy_fp_u64(), materialization)) - .collect()) - } - - pub fn validate(&self) -> Result<(), PrecomputePlanError> { - let valid_ingest = match self.ingest.protocol { - IngestProtocol::ModifiedOtlpMetricsV1 => { - self.ingest.endpoint_path == "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/v1/metrics" - && self.ingest.timestamp_unit == TimestampUnit::UnixNanoseconds - && self.ingest.require_plan_identity - && self.ingest.require_summary_definition_identity - && self.ingest.require_registered_producer - } - IngestProtocol::PrometheusRemoteWriteV1 => { - self.ingest.endpoint_path == "/api/v1/write" - && self.ingest.timestamp_unit == TimestampUnit::UnixMilliseconds - && !self.ingest.require_plan_identity - && !self.ingest.require_summary_definition_identity - && !self.ingest.require_registered_producer - } - }; - if !valid_ingest { - return Err(PrecomputePlanError::UnsupportedIngestEndpoint); - } - for (query_id, installed) in &self.executable_dags { - if query_id != &installed.document.query_id { - return Err(PrecomputePlanError::CatalogContract( - "post-ASAP DAG map key differs from document query ID".into(), - )); - } - installed - .validate() - .map_err(PrecomputePlanError::CatalogContract)?; - } - let mut materializations = BTreeSet::new(); - for materialization in &self.materializations { - materialization - .window_layout - .validate(materialization.window_size, materialization.slide_interval) - .map_err(|reason| PrecomputePlanError::InvalidWindowLayout { - materialization: materialization.policy_fp_u64(), - reason, - })?; - let expected_kind = if materialization.slide_interval == materialization.window_size { - asap_types::WindowKind::Tumbling - } else { - asap_types::WindowKind::Sliding - }; - if materialization.window_type != expected_kind { - return Err(PrecomputePlanError::InvalidWindowLayout { - materialization: materialization.policy_fp_u64(), - reason: "window kind disagrees with size and slide".into(), - }); - } - // HLL is supported as an ingested sketch envelope, not as a raw - // accumulator. Validate here so external installs cannot bypass it. - if self.ingest.protocol == IngestProtocol::PrometheusRemoteWriteV1 - && materialization.aggregation_type == asap_types::AggregationType::HLL - { - return Err(PrecomputePlanError::UnsupportedFamily( - materialization.policy_fp_u64(), - )); - } - if !materializations.insert(materialization.policy_fingerprint().into()) { - return Err(PrecomputePlanError::DuplicateMaterialization( - materialization.policy_fp_u64(), - )); - } - } - let mut schema_ids = BTreeSet::new(); - for schema in &self.schemas { - if schema.schema_id.trim().is_empty() - || !schema_ids.insert(schema.schema_id.as_str()) - || schema.schema_version == 0 - || schema.encodings.is_empty() - { - return Err(PrecomputePlanError::InvalidSchema { - schema_id: schema.schema_id.clone(), - }); - } - } - let schemas: BTreeSet<_> = self - .schemas - .iter() - .map(|schema| schema.materialization) - .collect(); - if schemas != materializations || schemas.len() != self.schemas.len() { - return Err(PrecomputePlanError::SchemaSetMismatch); - } - let schema_by_materialization: BTreeMap<_, _> = self - .schemas - .iter() - .map(|schema| (schema.materialization, schema.schema_id.as_str())) - .collect(); - for schema in &self.schemas { - let materialization = self - .materializations - .iter() - .find(|candidate| { - candidate.policy_fingerprint() == schema.materialization.fingerprint() - }) - .ok_or(PrecomputePlanError::SchemaSetMismatch)?; - let accumulator = materialization.accumulator_spec().map_err(|_| { - PrecomputePlanError::UnsupportedFamily(schema.materialization.as_u64()) - })?; - let family = StateFamilyContract::try_from(&accumulator.family).map_err(|_| { - PrecomputePlanError::UnsupportedFamily(schema.materialization.as_u64()) - })?; - let source = materialization.table_name.as_ref().map_or_else( - || Source::TimeSeries { - metric: materialization.metric.clone(), - }, - |table_ref| Source::Table { - table_ref: table_ref.clone(), - }, - ); - let value_column = materialization - .value_column - .clone() - .map(planner_types::pre_asap::ColumnRef::Named) - .unwrap_or(planner_types::pre_asap::ColumnRef::SampleValue); - if schema.schema_id != state_schema_id(schema.materialization.fingerprint()) - || schema.family != family - || schema.source != source - || schema.value_column != value_column - || schema.group_by != materialization.grouping_labels.labels - || schema.window.kind != materialization.window_type - || schema.window.size_ms != materialization.window_size.saturating_mul(1_000) - || schema.window.slide_ms - != match materialization.window_type { - asap_types::WindowKind::Tumbling => None, - asap_types::WindowKind::Sliding => { - Some(materialization.slide_interval.saturating_mul(1_000)) - } - asap_types::WindowKind::Session => None, - } - || schema.window.pane_origin_ms != materialization.pane_origin_ms - || schema.encodings != state_encodings(&accumulator.family) - { - return Err(PrecomputePlanError::InvalidSchema { - schema_id: schema.schema_id.clone(), - }); - } - } - let mut producers = BTreeSet::new(); - let mut produced = BTreeSet::new(); - for producer in &self.producers { - if !materializations.contains(&producer.materialization) - || schema_by_materialization - .get(&producer.materialization) - .copied() - != Some(producer.schema_id.as_str()) - { - return Err(PrecomputePlanError::InvalidProducer { - producer_id: producer.producer_id.clone(), - }); - } - let key = ( - producer.producer_id.as_str(), - producer.materialization, - producer.schema_id.as_str(), - ); - if !producers.insert(key) { - return Err(PrecomputePlanError::DuplicateProducer( - producer.producer_id.clone(), - )); - } - produced.insert(producer.materialization); - } - if self.ingest.require_registered_producer { - if let Some(missing) = materializations.difference(&produced).next() { - return Err(PrecomputePlanError::MissingProducer(missing.as_u64())); - } - } - Ok(()) - } -} - /// Complete physical projection of one post-ASAP planning decision. /// All three child plans share the same envelope and are compiled together. #[derive(Debug, Clone)] @@ -3126,34 +2654,6 @@ pub fn select_post_asap( ) } -pub(super) fn state_schema_id(fingerprint: asap_types::PolicyFingerprint) -> String { - format!("{}:summary-state:v1:{}", BACKEND_COMPAT, fingerprint.0) -} - -pub(super) fn state_encodings(family: &SummaryFamilyType) -> Vec { - match family { - SummaryFamilyType::ExactAggregate( - planner_types::post_asap::ExactKind::Increase - | planner_types::post_asap::ExactKind::Rate, - _, - ) => vec![StateEncoding::ExactCounterAccumulatorV2], - SummaryFamilyType::ExactAggregate(..) => vec![StateEncoding::ExactAccumulatorV1], - SummaryFamilyType::Sketch(kind, _) - if matches!( - kind.algorithm(), - SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap - ) => - { - vec![StateEncoding::SketchCoreMsgpackV1] - } - SummaryFamilyType::Sketch(..) => vec![ - StateEncoding::SketchlibProtobufV1, - StateEncoding::SketchCoreMsgpackV1, - ], - _ => Vec::new(), - } -} - fn validate_evidence( query_id: &str, evidence: &TopKMembershipEvidence, diff --git a/control_plane/src/physical/mod.rs b/control_plane/src/physical/mod.rs index fbf7ac72..fde19b25 100644 --- a/control_plane/src/physical/mod.rs +++ b/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; diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index f1d9f3dd..ccfb16a2 100644 --- a/crates/asap_types/src/lib.rs +++ b/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; diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs new file mode 100644 index 00000000..8cff067d --- /dev/null +++ b/crates/asap_types/src/precompute_plan.rs @@ -0,0 +1,517 @@ +//! Shared precompute installation contract and catalog consistency checks. +//! Compilation chooses these values; runtime consumers validate the same DTO. + +mod catalog; + +use planner_types::post_asap::{SketchAlgorithm, SketchParams, SummaryFamilyType}; +use planner_types::pre_asap::Source; +use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; +use thiserror::Error; + +pub const BACKEND_COMPAT: &str = "asap-query-backend.v1"; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct PlanEnvelope { + pub plan_id: u64, + pub plan_version: u64, + pub generated_at_unix_ms: u64, + pub activation_unix_ms: u64, + pub expiry_unix_ms: Option, + pub backend_compat: String, + pub planner_revision: String, + pub capability_snapshot_id: String, +} + +/// Backend-side materialization projection consumed by the streaming +/// precompute engine. This is deliberately config-driven: it contains no +/// PromQL string or ad-hoc scheduler job. The aggregation definitions are +/// emitted to `/api/v1/streaming-config`, where the runtime matches incoming +/// series, maintains windows, and writes content-addressed materializations. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PrecomputePlan { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub summary_catalog: Option, + pub envelope: PlanEnvelope, + pub ingest: IngestContract, + pub schemas: Vec, + pub producers: Vec, + pub materializations: Vec, + /// Planner semantic DAGs and backend-owned placement for this generation. + /// Empty only for legacy/config-only construction paths. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub executable_dags: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum IngestProtocol { + ModifiedOtlpMetricsV1, + PrometheusRemoteWriteV1, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TimestampUnit { + UnixNanoseconds, + UnixMilliseconds, +} + +/// Backend ingress semantics installed with the precompute projection. This +/// replaces implicit knowledge formerly hidden in the streaming-config path. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct IngestContract { + pub protocol: IngestProtocol, + pub endpoint_path: String, + pub timestamp_unit: TimestampUnit, + pub require_plan_identity: bool, + #[serde(alias = "require_materialization_identity")] + pub require_summary_definition_identity: bool, + pub require_registered_producer: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(rename_all = "snake_case")] +pub enum StateEncoding { + SketchlibProtobufV1, + SketchCoreMsgpackV1, + ExactAccumulatorV1, + ExactCounterAccumulatorV2, +} + +/// Serializable physical state identity derived from Planner's canonical +/// summary family. This is a wire DTO, not a second planning algebra. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "family", rename_all = "snake_case", deny_unknown_fields)] +pub enum StateFamilyContract { + Exact { + kind: ExactStateKind, + }, + Sketch { + algorithm: SketchAlgorithm, + parameters: SketchParams, + }, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ExactStateKind { + Sum, + Count, + MinMax, + Increase, + Rate, + IRate, +} + +impl TryFrom<&SummaryFamilyType> for StateFamilyContract { + type Error = (); + + fn try_from(family: &SummaryFamilyType) -> Result { + use planner_types::post_asap::ExactKind; + Ok(match family { + SummaryFamilyType::ExactAggregate(kind, _) => Self::Exact { + kind: match kind { + ExactKind::Sum => ExactStateKind::Sum, + ExactKind::Count => ExactStateKind::Count, + ExactKind::MinMax => ExactStateKind::MinMax, + ExactKind::Increase => ExactStateKind::Increase, + ExactKind::Rate => ExactStateKind::Rate, + ExactKind::IRate => ExactStateKind::IRate, + }, + }, + SummaryFamilyType::Sketch(kind, _) => Self::Sketch { + algorithm: kind.algorithm().clone(), + parameters: kind.params().clone(), + }, + _ => return Err(()), + }) + } +} + +/// Decoder/schema contract for one content-addressed materialization. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct StateSchemaContract { + pub schema_id: String, + pub schema_version: u32, + pub materialization: crate::sds::SummaryDefinitionId, + pub family: StateFamilyContract, + pub source: Source, + pub value_column: planner_types::pre_asap::ColumnRef, + pub group_by: Vec, + pub window: StateWindowContract, + pub encodings: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct StateWindowContract { + pub kind: crate::WindowKind, + pub size_ms: u64, + pub slide_ms: Option, + #[serde( + default, + alias = "paneOriginMs", + skip_serializing_if = "Option::is_none" + )] + pub pane_origin_ms: Option, +} + +/// A collector authorized to produce state for one materialization. Runtime +/// producer epochs and frame sequences belong to TransmissionPlan. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(deny_unknown_fields)] +pub struct ProducerContract { + pub producer_id: String, + pub collector_id: String, + pub materialization: crate::sds::SummaryDefinitionId, + pub schema_id: String, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum PrecomputePlanError { + #[error("invalid precompute catalog contract: {0}")] + CatalogContract(String), + #[error("PrecomputePlan envelope does not match its SummaryCatalog identity/lifecycle")] + PlanIdentityMismatch, + #[error("unsupported precompute ingest protocol/endpoint/identity contract")] + UnsupportedIngestEndpoint, + #[error("duplicate materialization {0}")] + DuplicateMaterialization(u64), + #[error("schema set does not exactly match the materialization set")] + SchemaSetMismatch, + #[error("schema {schema_id} has invalid version or no encoding")] + InvalidSchema { schema_id: String }, + #[error("materialization {0} uses a summary family unsupported by the runtime schema")] + UnsupportedFamily(u64), + #[error("materialization {materialization} has an invalid window layout: {reason}")] + InvalidWindowLayout { + materialization: u64, + reason: String, + }, + #[error("producer {producer_id} references an unknown materialization or schema")] + InvalidProducer { producer_id: String }, + #[error("materialization {0} has no registered producer")] + MissingProducer(u64), + #[error("duplicate producer binding {0}")] + DuplicateProducer(String), +} + +impl PrecomputePlan { + pub fn build( + envelope: PlanEnvelope, + materializations: Vec, + producer_ids: &[String], + ) -> Result { + let schemas = materializations + .iter() + .map(|materialization| { + let fingerprint = materialization.policy_fingerprint(); + let accumulator = materialization + .accumulator_spec() + .map_err(|_| PrecomputePlanError::UnsupportedFamily(fingerprint.0))?; + let family = StateFamilyContract::try_from(&accumulator.family) + .map_err(|_| PrecomputePlanError::UnsupportedFamily(fingerprint.0))?; + let source = materialization.table_name.as_ref().map_or_else( + || Source::TimeSeries { + metric: materialization.metric.clone(), + }, + |table_ref| Source::Table { + table_ref: table_ref.clone(), + }, + ); + let value_column = materialization + .value_column + .clone() + .map(planner_types::pre_asap::ColumnRef::Named) + .unwrap_or(planner_types::pre_asap::ColumnRef::SampleValue); + Ok(StateSchemaContract { + schema_id: state_schema_id(fingerprint), + schema_version: 1, + materialization: fingerprint.into(), + family, + source, + value_column, + group_by: materialization.grouping_labels.labels.clone(), + window: StateWindowContract { + kind: materialization.window_type, + size_ms: materialization.window_size.saturating_mul(1_000), + slide_ms: match materialization.window_type { + crate::WindowKind::Tumbling => None, + crate::WindowKind::Sliding => { + Some(materialization.slide_interval.saturating_mul(1_000)) + } + crate::WindowKind::Session => None, + }, + pane_origin_ms: materialization.pane_origin_ms, + }, + encodings: state_encodings(&accumulator.family), + }) + }) + .collect::, PrecomputePlanError>>()?; + let producers = producer_ids + .iter() + .flat_map(|producer_id| { + schemas.iter().map(move |schema| ProducerContract { + producer_id: producer_id.clone(), + collector_id: producer_id.clone(), + materialization: schema.materialization, + schema_id: schema.schema_id.clone(), + }) + }) + .collect(); + let plan = Self { + summary_catalog: None, + envelope, + ingest: IngestContract { + protocol: IngestProtocol::ModifiedOtlpMetricsV1, + endpoint_path: "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/v1/metrics".into(), + timestamp_unit: TimestampUnit::UnixNanoseconds, + require_plan_identity: true, + require_summary_definition_identity: true, + require_registered_producer: true, + }, + schemas, + producers, + materializations, + executable_dags: BTreeMap::new(), + }; + plan.validate()?; + Ok(plan) + } + + /// Build the backend-local projection used when raw Prometheus samples + /// are precomputed inside ASAPQuery rather than by ASAPCollector. + pub fn build_backend_local( + envelope: PlanEnvelope, + materializations: Vec, + ) -> Result { + let mut plan = Self::build(envelope, materializations, &["backend-local".into()])?; + plan.ingest = IngestContract { + protocol: IngestProtocol::PrometheusRemoteWriteV1, + endpoint_path: "/api/v1/write".into(), + timestamp_unit: TimestampUnit::UnixMilliseconds, + require_plan_identity: false, + require_summary_definition_identity: false, + require_registered_producer: false, + }; + plan.producers.clear(); + plan.validate()?; + Ok(plan) + } + + pub fn runtime_materializations( + &self, + ) -> Result, PrecomputePlanError> { + self.validate()?; + Ok(self + .materializations + .iter() + .cloned() + .map(|materialization| (materialization.policy_fp_u64(), materialization)) + .collect()) + } + + pub fn validate(&self) -> Result<(), PrecomputePlanError> { + let valid_ingest = match self.ingest.protocol { + IngestProtocol::ModifiedOtlpMetricsV1 => { + self.ingest.endpoint_path == "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/v1/metrics" + && self.ingest.timestamp_unit == TimestampUnit::UnixNanoseconds + && self.ingest.require_plan_identity + && self.ingest.require_summary_definition_identity + && self.ingest.require_registered_producer + } + IngestProtocol::PrometheusRemoteWriteV1 => { + self.ingest.endpoint_path == "/api/v1/write" + && self.ingest.timestamp_unit == TimestampUnit::UnixMilliseconds + && !self.ingest.require_plan_identity + && !self.ingest.require_summary_definition_identity + && !self.ingest.require_registered_producer + } + }; + if !valid_ingest { + return Err(PrecomputePlanError::UnsupportedIngestEndpoint); + } + for (query_id, installed) in &self.executable_dags { + if query_id != &installed.document.query_id { + return Err(PrecomputePlanError::CatalogContract( + "post-ASAP DAG map key differs from document query ID".into(), + )); + } + installed + .validate() + .map_err(PrecomputePlanError::CatalogContract)?; + } + let mut materializations = BTreeSet::new(); + for materialization in &self.materializations { + materialization + .window_layout + .validate(materialization.window_size, materialization.slide_interval) + .map_err(|reason| PrecomputePlanError::InvalidWindowLayout { + materialization: materialization.policy_fp_u64(), + reason, + })?; + let expected_kind = if materialization.slide_interval == materialization.window_size { + crate::WindowKind::Tumbling + } else { + crate::WindowKind::Sliding + }; + if materialization.window_type != expected_kind { + return Err(PrecomputePlanError::InvalidWindowLayout { + materialization: materialization.policy_fp_u64(), + reason: "window kind disagrees with size and slide".into(), + }); + } + // HLL is supported as an ingested sketch envelope, not as a raw + // accumulator. Validate here so external installs cannot bypass it. + if self.ingest.protocol == IngestProtocol::PrometheusRemoteWriteV1 + && materialization.aggregation_type == crate::AggregationType::HLL + { + return Err(PrecomputePlanError::UnsupportedFamily( + materialization.policy_fp_u64(), + )); + } + if !materializations.insert(materialization.policy_fingerprint().into()) { + return Err(PrecomputePlanError::DuplicateMaterialization( + materialization.policy_fp_u64(), + )); + } + } + let mut schema_ids = BTreeSet::new(); + for schema in &self.schemas { + if schema.schema_id.trim().is_empty() + || !schema_ids.insert(schema.schema_id.as_str()) + || schema.schema_version == 0 + || schema.encodings.is_empty() + { + return Err(PrecomputePlanError::InvalidSchema { + schema_id: schema.schema_id.clone(), + }); + } + } + let schemas: BTreeSet<_> = self + .schemas + .iter() + .map(|schema| schema.materialization) + .collect(); + if schemas != materializations || schemas.len() != self.schemas.len() { + return Err(PrecomputePlanError::SchemaSetMismatch); + } + let schema_by_materialization: BTreeMap<_, _> = self + .schemas + .iter() + .map(|schema| (schema.materialization, schema.schema_id.as_str())) + .collect(); + for schema in &self.schemas { + let materialization = self + .materializations + .iter() + .find(|candidate| { + candidate.policy_fingerprint() == schema.materialization.fingerprint() + }) + .ok_or(PrecomputePlanError::SchemaSetMismatch)?; + let accumulator = materialization.accumulator_spec().map_err(|_| { + PrecomputePlanError::UnsupportedFamily(schema.materialization.as_u64()) + })?; + let family = StateFamilyContract::try_from(&accumulator.family).map_err(|_| { + PrecomputePlanError::UnsupportedFamily(schema.materialization.as_u64()) + })?; + let source = materialization.table_name.as_ref().map_or_else( + || Source::TimeSeries { + metric: materialization.metric.clone(), + }, + |table_ref| Source::Table { + table_ref: table_ref.clone(), + }, + ); + let value_column = materialization + .value_column + .clone() + .map(planner_types::pre_asap::ColumnRef::Named) + .unwrap_or(planner_types::pre_asap::ColumnRef::SampleValue); + if schema.schema_id != state_schema_id(schema.materialization.fingerprint()) + || schema.family != family + || schema.source != source + || schema.value_column != value_column + || schema.group_by != materialization.grouping_labels.labels + || schema.window.kind != materialization.window_type + || schema.window.size_ms != materialization.window_size.saturating_mul(1_000) + || schema.window.slide_ms + != match materialization.window_type { + crate::WindowKind::Tumbling => None, + crate::WindowKind::Sliding => { + Some(materialization.slide_interval.saturating_mul(1_000)) + } + crate::WindowKind::Session => None, + } + || schema.window.pane_origin_ms != materialization.pane_origin_ms + || schema.encodings != state_encodings(&accumulator.family) + { + return Err(PrecomputePlanError::InvalidSchema { + schema_id: schema.schema_id.clone(), + }); + } + } + let mut producers = BTreeSet::new(); + let mut produced = BTreeSet::new(); + for producer in &self.producers { + if !materializations.contains(&producer.materialization) + || schema_by_materialization + .get(&producer.materialization) + .copied() + != Some(producer.schema_id.as_str()) + { + return Err(PrecomputePlanError::InvalidProducer { + producer_id: producer.producer_id.clone(), + }); + } + let key = ( + producer.producer_id.as_str(), + producer.materialization, + producer.schema_id.as_str(), + ); + if !producers.insert(key) { + return Err(PrecomputePlanError::DuplicateProducer( + producer.producer_id.clone(), + )); + } + produced.insert(producer.materialization); + } + if self.ingest.require_registered_producer { + if let Some(missing) = materializations.difference(&produced).next() { + return Err(PrecomputePlanError::MissingProducer(missing.as_u64())); + } + } + Ok(()) + } +} + +pub(crate) fn state_schema_id(fingerprint: crate::PolicyFingerprint) -> String { + format!("{}:summary-state:v1:{}", BACKEND_COMPAT, fingerprint.0) +} + +pub(crate) fn state_encodings(family: &SummaryFamilyType) -> Vec { + match family { + SummaryFamilyType::ExactAggregate( + planner_types::post_asap::ExactKind::Increase + | planner_types::post_asap::ExactKind::Rate, + _, + ) => vec![StateEncoding::ExactCounterAccumulatorV2], + SummaryFamilyType::ExactAggregate(..) => vec![StateEncoding::ExactAccumulatorV1], + SummaryFamilyType::Sketch(kind, _) + if matches!( + kind.algorithm(), + SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap + ) => + { + vec![StateEncoding::SketchCoreMsgpackV1] + } + SummaryFamilyType::Sketch(..) => vec![ + StateEncoding::SketchlibProtobufV1, + StateEncoding::SketchCoreMsgpackV1, + ], + _ => Vec::new(), + } +} diff --git a/control_plane/src/physical/precompute_contract.rs b/crates/asap_types/src/precompute_plan/catalog.rs similarity index 95% rename from control_plane/src/physical/precompute_contract.rs rename to crates/asap_types/src/precompute_plan/catalog.rs index bac0b497..48fd1c41 100644 --- a/control_plane/src/physical/precompute_contract.rs +++ b/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) -> 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")) } }; diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 73a7056d..b6d9da82 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -2411,10 +2411,8 @@ fn take_summary_frame_identity( }) } -fn state_encoding_for_wire( - encoding: i32, -) -> Option { - use control_plane::physical::compiler::StateEncoding; +fn state_encoding_for_wire(encoding: i32) -> Option { + use asap_types::precompute_plan::StateEncoding; match encoding { ENCODING_PROTO | ENCODING_PROTO_DELTA => Some(StateEncoding::SketchlibProtobufV1), ENCODING_MSGPACK | ENCODING_MSGPACK_DELTA => Some(StateEncoding::SketchCoreMsgpackV1), diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 05c47a2c..ad7ad5ac 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/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(), }; diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index b3f916fa..f7756cfd 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/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) -> 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") { diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 0bc72d2c..9a3ffee2 100644 --- a/data_plane/src/main.rs +++ b/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: "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/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, diff --git a/data_plane/src/precompute_engine/frame_lineage.rs b/data_plane/src/precompute_engine/frame_lineage.rs index c283ef41..6598a138 100644 --- a/data_plane/src/precompute_engine/frame_lineage.rs +++ b/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 { diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index 32e9720c..f721742a 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/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()], diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index b3ed50b9..c007f8f5 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/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>, - 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, pub query_plan: Arc, @@ -655,7 +655,7 @@ mod tests { activation_unix_ms: u64, expiry_unix_ms: Option, ) -> 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: "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/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, diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 44131f62..83b8d829 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/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: