From 0163b27746fd4d495f45c5b26d0e91e4382aee98 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 08:23:01 -0600 Subject: [PATCH] Unify catalog generation and data identity contracts --- control_plane/src/physical/compiler.rs | 16 +-- .../src/physical/precompute_contract.rs | 21 ++- .../src/physical/summary_reconcile.rs | 2 +- crates/asap_types/src/sds.rs | 130 ++++++++++++++++-- crates/asap_types/src/summary_catalog.rs | 41 +++--- .../storage_engines/sketch_db/index/mod.rs | 11 +- .../src/storage_engines/sketch_db/sds.rs | 5 +- 7 files changed, 179 insertions(+), 47 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index bb92d9ab..52a80a55 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -296,7 +296,7 @@ pub struct CollectorLifecycle { pub struct CollectorPlan { /// Absent only in legacy artifacts; catalog-aware validation requires it. #[serde(default, skip_serializing_if = "Option::is_none")] - pub summary_catalog: Option, + pub summary_catalog: Option, pub collector_id: String, pub envelope: PlanEnvelope, pub materializations: Vec, @@ -311,7 +311,7 @@ pub struct CollectorPlan { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PrecomputePlan { #[serde(default, skip_serializing_if = "Option::is_none")] - pub summary_catalog: Option, + pub summary_catalog: Option, pub envelope: PlanEnvelope, pub ingest: IngestContract, pub schemas: Vec, @@ -997,7 +997,7 @@ pub struct RuntimeAdaptationEvidence { pub struct TransmissionPlan { /// Absent only in legacy artifacts; catalog-aware validation requires it. #[serde(default, skip_serializing_if = "Option::is_none")] - pub summary_catalog: Option, + pub summary_catalog: Option, pub envelope: PlanEnvelope, pub frame_identity: FrameIdentityContract, pub rules: Vec, @@ -1063,7 +1063,7 @@ pub enum TransmissionPlanError { } fn validate_catalog_projection( - reference: Option<&super::summary_catalog::SummaryCatalogReference>, + reference: Option<&asap_types::sds::CatalogGeneration>, envelope: &PlanEnvelope, materializations: impl IntoIterator, catalog: &super::summary_catalog::SummaryCatalog, @@ -5259,7 +5259,7 @@ mod tests { let identity = &plan.summary_catalog.materializations[&binding.materialization]; let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id]; ( - data.metric_name.as_str(), + data.time_series_metric().unwrap(), binding.window_ms, binding.readout_lookback_ms, ) @@ -5292,7 +5292,7 @@ mod tests { let identity = &plan.summary_catalog.materializations[&bindings[0].materialization]; let data = &plan.summary_catalog.data_descriptors[&identity.data_descriptor_id]; assert_eq!( - (data.metric_name.as_str(), bindings[0].window_ms), + (data.time_series_metric().unwrap(), bindings[0].window_ms), ("a", 60_000) ); assert!(!query @@ -6064,8 +6064,8 @@ mod tests { .summary_catalog .materializations[&binding.materialization] .data_descriptor_id] - .metric_name - .as_str(), + .time_series_metric() + .unwrap(), ), _ => None, }) diff --git a/control_plane/src/physical/precompute_contract.rs b/control_plane/src/physical/precompute_contract.rs index adcf6765..bac0b497 100644 --- a/control_plane/src/physical/precompute_contract.rs +++ b/control_plane/src/physical/precompute_contract.rs @@ -1,7 +1,9 @@ //! Catalog consistency checks for the precompute execution plan. use super::compiler::*; use super::summary_catalog::SummaryCatalog; -use asap_types::sds::{SummaryDefinitionId, SummaryDescriptor}; +use asap_types::sds::{ + DataSourceIdentity, SummaryDefinitionId, SummaryDescriptor, ValueProjectionIdentity, +}; use planner_types::pre_asap::{ColumnRef, Source}; use std::collections::BTreeSet; fn invalid(reason: impl Into) -> PrecomputePlanError { @@ -70,7 +72,22 @@ impl PrecomputePlan { return Err(invalid("pane origin differs from catalog definition")); } let data = &catalog.data_descriptors[&binding.data_descriptor_id]; - if data.metric_name != config.metric + let expected_source = config.table_name.as_ref().map_or_else( + || DataSourceIdentity::TimeSeries { + metric: config.metric.clone(), + }, + |table_ref| DataSourceIdentity::Table { + table_ref: table_ref.clone(), + }, + ); + let expected_projection = config + .value_column + .as_ref() + .map_or(ValueProjectionIdentity::SampleValue, |name| { + ValueProjectionIdentity::Column { name: name.clone() } + }); + if data.source != expected_source + || data.value_projection != expected_projection || data.population_filter_canonical != asap_types::utils::normalize_spatial_filter(&config.spatial_filter) || data.group_by_keys != config.grouping_labels.labels.iter().cloned().collect() diff --git a/control_plane/src/physical/summary_reconcile.rs b/control_plane/src/physical/summary_reconcile.rs index 5407e449..3a1f23c8 100644 --- a/control_plane/src/physical/summary_reconcile.rs +++ b/control_plane/src/physical/summary_reconcile.rs @@ -143,7 +143,7 @@ fn catalog_generation(catalog: &SummaryCatalog) -> Result, - /// Versioned contract for value projection, timestamp interpretation and + /// Versioned contract for timestamp interpretation and /// missing/duplicate/invalid observation handling. pub observation_semantics: String, } impl DataDescriptor { + pub fn time_series_metric(&self) -> Option<&str> { + match &self.source { + DataSourceIdentity::TimeSeries { metric } => Some(metric), + DataSourceIdentity::Table { .. } => None, + } + } + pub fn new( metric: impl Into, filter: impl Into, @@ -599,19 +624,39 @@ impl DataDescriptor { group_by: impl IntoIterator, observation_semantics: impl Into, ) -> Self { - let metric_name = metric.into(); + let source = DataSourceIdentity::TimeSeries { + metric: metric.into(), + }; + Self::new_typed( + source, + ValueProjectionIdentity::SampleValue, + filter, + group_by, + observation_semantics, + ) + } + + pub fn new_typed( + source: DataSourceIdentity, + value_projection: ValueProjectionIdentity, + filter: impl Into, + group_by: impl IntoIterator, + observation_semantics: impl Into, + ) -> Self { let population_filter_canonical = filter.into(); let group_by_keys = group_by.into_iter().collect(); let observation_semantics = observation_semantics.into(); let id = data_descriptor_id( - &metric_name, + &source, + &value_projection, &population_filter_canonical, &group_by_keys, &observation_semantics, ); Self { id, - metric_name, + source, + value_projection, population_filter_canonical, group_by_keys, observation_semantics, @@ -623,7 +668,8 @@ impl DataDescriptor { pub fn validate(&self) -> Result<(), SdsError> { if self.id != data_descriptor_id( - &self.metric_name, + &self.source, + &self.value_projection, &self.population_filter_canonical, &self.group_by_keys, &self.observation_semantics, @@ -635,16 +681,21 @@ impl DataDescriptor { } } fn data_descriptor_id( - metric: &str, + source: &DataSourceIdentity, + value_projection: &ValueProjectionIdentity, filter: &str, group_by: &BTreeSet, observation_semantics: &str, ) -> DataDescriptorId { - // Preserve the existing v1 length-framed data identity, now normalizing the - // grouping set at the shared contract boundary. + // Length framing keeps distinct typed sources, projections, predicates, + // and grouping keys collision-free in the content identity. + let source = canonical(&serde_json::to_value(source).expect("data source serializes")); + let projection = + canonical(&serde_json::to_value(value_projection).expect("value projection serializes")); let mut key = format!( - "data:v1|{}:{metric}|{}:{filter}", - metric.len(), + "data:v2|{}:{source}|{}:{projection}|{}:{filter}", + source.len(), + projection.len(), filter.len() ); for name in group_by { @@ -684,7 +735,7 @@ mod tests { schema_version: 1, plan_id: 1, plan_version: 2, - snapshot_digest: "abc".into(), + snapshot_sha256: "abc".into(), }, placement: SummaryPlacement { producer_id: "producer".into(), @@ -825,6 +876,53 @@ mod tests { DataDescriptor::new_with_semantics("cpu", "", [], "samples.v2").id ); } + + #[test] + fn source_and_value_projection_are_part_of_data_identity() { + let metric = DataDescriptor::new("events", "", []); + let table_value = DataDescriptor::new_typed( + DataSourceIdentity::Table { + table_ref: "events".into(), + }, + ValueProjectionIdentity::Column { + name: "value".into(), + }, + "", + [], + "asap.timestamped-observations.v2", + ); + let table_cost = DataDescriptor::new_typed( + DataSourceIdentity::Table { + table_ref: "events".into(), + }, + ValueProjectionIdentity::Column { + name: "cost".into(), + }, + "", + [], + "asap.timestamped-observations.v2", + ); + assert_ne!(metric.id, table_value.id); + assert_ne!(table_value.id, table_cost.id); + assert_eq!(metric.time_series_metric(), Some("events")); + assert_eq!(table_value.time_series_metric(), None); + } + + #[test] + fn catalog_generation_accepts_legacy_digest_name() { + let generation: CatalogGeneration = serde_json::from_value(json!({ + "schema_version": 1, + "plan_id": 2, + "plan_version": 3, + "snapshot_digest": "abc" + })) + .unwrap(); + assert_eq!(generation.snapshot_sha256, "abc"); + assert!(serde_json::to_value(generation) + .unwrap() + .get("snapshot_digest") + .is_none()); + } #[test] fn wire_roundtrip_and_tampered_id_validation() { let original = descriptor( @@ -842,7 +940,9 @@ mod tests { decoded.state_schema_version = 2; assert!(decoded.validate().is_err()); let mut data = DataDescriptor::new("cpu", "", []); - data.metric_name = "other".into(); + data.source = DataSourceIdentity::TimeSeries { + metric: "other".into(), + }; assert!(data.validate().is_err()); } #[test] diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 406c83a9..7274dc1b 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -6,22 +6,13 @@ use std::collections::BTreeMap; use crate::sds::{ - DataDescriptor, DataDescriptorId, SummaryDefinitionId, SummaryDescriptor, SummaryDescriptorId, + CatalogGeneration, DataDescriptor, DataDescriptorId, DataSourceIdentity, SummaryDefinitionId, + SummaryDescriptor, SummaryDescriptorId, ValueProjectionIdentity, }; use crate::PolicyFingerprint; use serde::{Deserialize, Serialize}; -pub const SUMMARY_CATALOG_SCHEMA_VERSION: u32 = 1; - -/// Identifies one immutable catalog snapshot without duplicating descriptors. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct SummaryCatalogReference { - pub schema_version: u32, - pub plan_id: u64, - pub plan_version: u64, - pub snapshot_sha256: String, -} +pub const SUMMARY_CATALOG_SCHEMA_VERSION: u32 = 2; /// Stable materialization identity binds operator and population descriptors. /// Concrete intervals, groups and completeness belong to runtime instances. @@ -65,7 +56,7 @@ pub enum SummaryCatalogError { ReferenceMismatch, } -impl SummaryCatalogReference { +impl CatalogGeneration { /// Validate an untrusted wire reference against the installed immutable /// snapshot and its enclosing plan generation. pub fn validate_snapshot( @@ -83,13 +74,13 @@ impl SummaryCatalogReference { } impl SummaryCatalog { - pub fn reference(&self) -> Result { + pub fn reference(&self) -> Result { use sha2::{Digest, Sha256}; self.validate()?; // BTreeMap tables and typed descriptor fields serialize deterministically. let bytes = serde_json::to_vec(self) .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; - Ok(SummaryCatalogReference { + Ok(CatalogGeneration { schema_version: self.schema_version, plan_id: self.plan_id, plan_version: self.plan_version, @@ -107,10 +98,26 @@ impl SummaryCatalog { .map(|config| { let summary = SummaryDescriptor::from_config(config) .map_err(|error| SummaryCatalogError::Descriptor(error.to_string()))?; - let data = DataDescriptor::new( - config.metric.clone(), + let source = config.table_name.as_ref().map_or_else( + || DataSourceIdentity::TimeSeries { + metric: config.metric.clone(), + }, + |table_ref| DataSourceIdentity::Table { + table_ref: table_ref.clone(), + }, + ); + let value_projection = config + .value_column + .as_ref() + .map_or(ValueProjectionIdentity::SampleValue, |name| { + ValueProjectionIdentity::Column { name: name.clone() } + }); + let data = DataDescriptor::new_typed( + source, + value_projection, crate::utils::normalize_spatial_filter(&config.spatial_filter), config.grouping_labels.labels.clone(), + "asap.timestamped-observations.v2", ); Ok(( config.policy_fingerprint(), diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index fb0ee47f..8353d646 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -724,7 +724,10 @@ impl SketchStore { return; } }; - let metric_name = instance.data_descriptor.metric_name.clone(); + let metric_name = instance + .data_descriptor + .time_series_metric() + .map(str::to_owned); // Fixed lock order: instances → policy_to_series_ids → metric_to_series_ids. let mut instances = self.instances.write().unwrap(); let mut policy_idx = self.policy_to_series_ids.write().unwrap(); @@ -733,7 +736,9 @@ impl SketchStore { if !policy_fp.is_unset() { policy_idx.entry(policy_fp).or_default().insert(sid); } - metric_idx.entry(metric_name).or_default().insert(sid); + if let Some(metric_name) = metric_name { + metric_idx.entry(metric_name).or_default().insert(sid); + } } /// Install one authoritative catalog snapshot for future registrations. @@ -844,7 +849,7 @@ impl SketchStore { schema_version: reference.schema_version, plan_id: reference.plan_id, plan_version: reference.plan_version, - snapshot_digest: reference.snapshot_sha256, + snapshot_sha256: reference.snapshot_sha256, }; let instances = self.instances.read().unwrap(); let durable = self.persistence_read.read().unwrap().clone(); diff --git a/data_plane/src/storage_engines/sketch_db/sds.rs b/data_plane/src/storage_engines/sketch_db/sds.rs index 9c6a768f..6cbdd0ff 100644 --- a/data_plane/src/storage_engines/sketch_db/sds.rs +++ b/data_plane/src/storage_engines/sketch_db/sds.rs @@ -235,7 +235,10 @@ impl SummaryDescriptorRegistry { + std::mem::size_of::>()); for descriptor in data.values().filter_map(Weak::upgrade) { total += std::mem::size_of::() + descriptor.id.canonical().len(); - total += descriptor.metric_name.len() + descriptor.population_filter_canonical.len(); + total += serde_json::to_string(&descriptor.source).map_or(0, |value| value.len()); + total += + serde_json::to_string(&descriptor.value_projection).map_or(0, |value| value.len()); + total += descriptor.population_filter_canonical.len(); total += descriptor .group_by_keys .iter()