Unify catalog generation and typed data identity

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<super::summary_catalog::SummaryCatalogReference>,
+    pub summary_catalog: Option<asap_types::sds::CatalogGeneration>,
     pub collector_id: String,
     pub envelope: PlanEnvelope,
     pub materializations: Vec<CollectorMaterialization>,
@@ -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<super::summary_catalog::SummaryCatalogReference>,
+    pub summary_catalog: Option<asap_types::sds::CatalogGeneration>,
     pub envelope: PlanEnvelope,
     pub ingest: IngestContract,
     pub schemas: Vec<StateSchemaContract>,
@@ -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<super::summary_catalog::SummaryCatalogReference>,
+    pub summary_catalog: Option<asap_types::sds::CatalogGeneration>,
     pub envelope: PlanEnvelope,
     pub frame_identity: FrameIdentityContract,
     pub rules: Vec<TransmissionRule>,
@@ -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<Item = asap_types::sds::SummaryDefinitionId>,
     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,
             })

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<String>) -> 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()

control_plane/src/physical/summary_reconcile.rs

@@ -143,7 +143,7 @@ fn catalog_generation(catalog: &SummaryCatalog) -> Result<CatalogGeneration, Sum
         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,
     })
 }
 

crates/asap_types/src/sds.rs

@@ -73,7 +73,8 @@ pub struct CatalogGeneration {
     pub schema_version: u32,
     pub plan_id: u64,
     pub plan_version: u64,
-    pub snapshot_digest: String,
+    #[serde(alias = "snapshot_digest")]
+    pub snapshot_sha256: String,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -168,7 +169,7 @@ impl SummaryInstance {
             ));
         }
         if self.catalog_generation.schema_version == 0
-            || self.catalog_generation.snapshot_digest.is_empty()
+            || self.catalog_generation.snapshot_sha256.is_empty()
         {
             return Err(SdsError(
                 "summary instance has invalid catalog generation".into(),
@@ -568,18 +569,42 @@ impl FidelityGuarantee {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+#[serde(rename_all = "snake_case")]
+pub enum DataSourceIdentity {
+    TimeSeries { metric: String },
+    Table { table_ref: String },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+#[serde(rename_all = "snake_case")]
+pub enum ValueProjectionIdentity {
+    SampleValue,
+    Column { name: String },
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct DataDescriptor {
     pub id: DataDescriptorId,
-    pub metric_name: String,
+    pub source: DataSourceIdentity,
+    pub value_projection: ValueProjectionIdentity,
     pub population_filter_canonical: String,
     pub group_by_keys: BTreeSet<String>,
-    /// 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<String>,
         filter: impl Into<String>,
@@ -599,19 +624,39 @@ impl DataDescriptor {
         group_by: impl IntoIterator<Item = String>,
         observation_semantics: impl Into<String>,
     ) -> 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<String>,
+        group_by: impl IntoIterator<Item = String>,
+        observation_semantics: impl Into<String>,
+    ) -> 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<String>,
     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]