refactor: own SDS stored-state decoding and dataset-bound semantic identity

Backend now owns SDS frame decoding, delta reconstruction, sketch readouts
and native batch frames, and the stored-definition semantic fragment. The
logical dataset is recorded on the materialization and in
`SummarySemantics::Planner`, so equal fragments over different datasets get
different definition IDs. Planner stays deployment-agnostic.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: zzylol

control_plane/src/main.rs

@@ -199,7 +199,7 @@ struct CompileAndPublishPhysicalPlanRequest {
     workload_cost_evidence: Option<physical::workload_cost::WorkloadCostEvidence>,
     queries: Vec<PhysicalPlanQueryRequest>,
     data_workload: planner_types::workload::DataWorkload,
-    dataset_identity: planner_types::post_asap::LogicalDatasetIdentity,
+    dataset_identity: asap_types::semantic_fragment::LogicalDatasetIdentity,
     #[serde(rename = "collector_ids")]
     target_collector_ids: Vec<String>,
     capability_snapshot_id: String,

control_plane/src/physical/compiler.rs

@@ -200,7 +200,7 @@ pub struct TopKMembershipEvidence {
 #[serde(deny_unknown_fields)]
 pub struct PhysicalDeploymentContext {
     /// Semantic dataset served by this deployment's input channel; never an endpoint.
-    pub dataset_identity: planner_types::post_asap::LogicalDatasetIdentity,
+    pub dataset_identity: asap_types::semantic_fragment::LogicalDatasetIdentity,
     pub target: PhysicalDeploymentTarget,
     #[serde(rename = "collector_ids")]
     pub target_collector_ids: Vec<String>,
@@ -1493,11 +1493,12 @@ impl DeploymentPlanCompiler {
                             query_id: query.query_id.clone(),
                             reason: "persisted semantic root is absent".into(),
                         })?;
+                runtime_materialization.dataset_identity =
+                    Some(environment.dataset_identity.clone());
                 runtime_materialization.semantic_fragment = Some(
-                    asap_types::semantic_fragment::SemanticFragment::from_stored_output_in_dataset(
+                    asap_types::semantic_fragment::SemanticFragment::from_stored_output(
                         &compiled_dag.dag,
                         semantic_root,
-                        environment.dataset_identity.clone(),
                     )
                     .map_err(|reason| CompileError::Query {
                         query_id: query.query_id.clone(),
@@ -4690,6 +4691,7 @@ pub(crate) mod tests {
             first.summary_catalog.definitions,
             relocated.summary_catalog.definitions
         );
+        // An ingest binding forged to another dataset is rejected, as is a missing one.
         let mut forged = first.precompute_plan.clone();
         forged.ingest.dataset_identity.as_mut().unwrap().namespace = "other-tenant".into();
         assert!(forged
@@ -4699,11 +4701,28 @@ pub(crate) mod tests {
             .contains("dataset"));
         forged.ingest.dataset_identity = None;
         assert!(forged.validate().is_err());
+        // A materialization whose recorded dataset differs from the installed ingest binding is rejected.
+        let mut forged = first.precompute_plan.clone();
+        let materialization = forged
+            .materializations
+            .iter_mut()
+            .find(|m| m.semantic_fragment.is_some())
+            .expect("planner materialization");
+        assert_eq!(
+            materialization.dataset_identity,
+            forged.ingest.dataset_identity
+        );
+        materialization.dataset_identity.as_mut().unwrap().dataset = "other-metrics".into();
+        assert!(forged
+            .validate()
+            .unwrap_err()
+            .to_string()
+            .contains("dataset"));
     }
 
     fn environment(now: u64) -> PhysicalDeploymentContext {
         PhysicalDeploymentContext {
-            dataset_identity: planner_types::post_asap::LogicalDatasetIdentity {
+            dataset_identity: asap_types::semantic_fragment::LogicalDatasetIdentity {
                 namespace: "test".into(),
                 dataset: "metrics".into(),
             },

crates/asap_types/src/aggregation_config.rs

@@ -98,6 +98,9 @@ pub struct PrecomputeMaterialization {
     /// Planner-selected dependency closure ending at the persisted output.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub semantic_fragment: Option<crate::semantic_fragment::SemanticFragment>,
+    /// Logical dataset that the fragment's source names resolve in.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dataset_identity: Option<crate::semantic_fragment::LogicalDatasetIdentity>,
     pub aggregation_type: AggregationType,
     pub aggregation_sub_type: String,
     pub parameters: HashMap<String, Value>,
@@ -279,6 +282,7 @@ impl PrecomputeMaterialization {
         Self {
             stored_output_id: None,
             semantic_fragment: None,
+            dataset_identity: None,
             aggregation_type,
             aggregation_sub_type,
             parameters,

crates/asap_types/src/precompute_plan.rs

@@ -139,7 +139,7 @@ pub enum TimestampUnit {
 #[serde(deny_unknown_fields)]
 pub struct IngestContract {
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub dataset_identity: Option<planner_types::post_asap::LogicalDatasetIdentity>,
+    pub dataset_identity: Option<crate::semantic_fragment::LogicalDatasetIdentity>,
     pub protocol: IngestProtocol,
     pub endpoint_path: String,
     pub timestamp_unit: TimestampUnit,
@@ -366,7 +366,7 @@ impl PrecomputePlan {
             .collect();
         let dataset_identity = materializations
             .iter()
-            .filter_map(|m| m.semantic_fragment.as_ref()?.dataset_identity.clone())
+            .filter_map(|m| m.dataset_identity.clone())
             .next();
         let plan = Self {
             summary_catalog: None,
@@ -484,7 +484,7 @@ impl PrecomputePlan {
                 fragment
                     .validate()
                     .map_err(PrecomputePlanError::CatalogContract)?;
-                if fragment.dataset_identity != self.ingest.dataset_identity {
+                if materialization.dataset_identity != self.ingest.dataset_identity {
                     return Err(PrecomputePlanError::CatalogContract(
                         "semantic dataset differs from installed input binding".into(),
                     ));

crates/asap_types/src/precompute_plan/catalog.rs

@@ -70,10 +70,11 @@ impl PrecomputePlan {
                             if stored_output.fingerprint() == config.policy_fingerprint())
                         {
                             found = true;
-                            let actual = match &self.ingest.dataset_identity {
-                                Some(dataset) => crate::semantic_fragment::SemanticFragment::from_stored_output_in_dataset(&dag, *id, dataset.clone()),
-                                None => crate::semantic_fragment::SemanticFragment::from_stored_output(&dag, *id),
-                            }.map_err(invalid)?;
+                            let actual =
+                                crate::semantic_fragment::SemanticFragment::from_stored_output(
+                                    &dag, *id,
+                                )
+                                .map_err(invalid)?;
                             if &actual != expected {
                                 return Err(invalid(
                                     "semantic definition differs from Planner-selected producer",