Version population key encoding in shared materialization identity

crates/asap_types/src/aggregation_config.rs

@@ -101,6 +101,11 @@ pub struct PrecomputeMaterialization {
     pub parameters: HashMap<String, Value>,
     #[serde(serialize_with = "crate::grouping_projection::serialize_config_grouping")]
     pub grouping_labels: crate::GroupingProjection,
+    #[serde(
+        default,
+        skip_serializing_if = "crate::grouping_projection::PopulationKeyEncoding::is_legacy"
+    )]
+    pub population_key_encoding: crate::grouping_projection::PopulationKeyEncoding,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub partitioning: Option<crate::sds::PopulationPartitioning>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -286,6 +291,7 @@ impl PrecomputeMaterialization {
             aggregation_sub_type,
             parameters,
             grouping_labels: grouping_labels.into(),
+            population_key_encoding: Default::default(),
             partitioning: None,
             derived_input: None,
             aggregated_labels,
@@ -433,6 +439,11 @@ impl PrecomputeMaterialization {
             table_name,
             value_column,
         );
+        config.population_key_encoding = data
+            .get("population_key_encoding")
+            .map(|value| serde_json::from_value(value.clone()))
+            .transpose()?
+            .unwrap_or_default();
         config.derived_input = data
             .get("derived_input")
             .filter(|v| !v.is_null())
@@ -633,6 +644,11 @@ impl PrecomputeMaterialization {
             table_name,
             value_column,
         );
+        config.population_key_encoding = aggregation_data
+            .get("population_key_encoding")
+            .map(|value| serde_yaml::from_value(value.clone()))
+            .transpose()?
+            .unwrap_or_default();
         config.derived_input = aggregation_data
             .get("derived_input")
             .filter(|v| !v.is_null())
@@ -683,6 +699,9 @@ impl SerializableToSink for PrecomputeMaterialization {
             "metric": self.metric,
         });
 
+        if !self.population_key_encoding.is_legacy() {
+            json["population_key_encoding"] = serde_json::json!(self.population_key_encoding);
+        }
         if let Some(input) = &self.derived_input {
             json["derived_input"] = serde_json::json!(input);
         }

crates/asap_types/src/grouping_projection.rs

@@ -3,6 +3,21 @@ use crate::KeyByLabelNames;
 use planner_types::pre_asap::{Column, DataType};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 
+/// Version of the population-key routing contract. Choosing a new version
+/// changes materialization and data identity; it never migrates old SIDs.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum PopulationKeyEncoding {
+    #[default]
+    LegacyDelimited,
+    CanonicalLabelsV1,
+}
+impl PopulationKeyEncoding {
+    pub fn is_legacy(&self) -> bool {
+        *self == Self::LegacyDelimited
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct GroupingProjection(Vec<Column>);
 

crates/asap_types/src/lib.rs

@@ -27,7 +27,7 @@ pub use accuracy::{AccuracyKind, AccuracyProfile};
 pub use aggregation_config::*;
 pub use aggregation_type::AggregationType;
 pub use enums::*;
-pub use grouping_projection::GroupingProjection;
+pub use grouping_projection::{GroupingProjection, PopulationKeyEncoding};
 pub use key_by_label_names::KeyByLabelNames;
 pub use monitor_spec::{MonitorFunctional, MonitorSpec};
 pub use policy_fingerprint::PolicyFingerprint;

crates/asap_types/src/policy_fingerprint.rs

@@ -89,6 +89,10 @@ impl PolicyFingerprint {
     pub fn from_config(cfg: &AggregationConfig) -> Self {
         let mut buf: Vec<u8> = Vec::with_capacity(512);
 
+        if !cfg.population_key_encoding.is_legacy() {
+            // UTF-8 raw metrics cannot alias this nonlegacy domain prefix.
+            buf.extend_from_slice(b"\xffpopulation-key-canonical-labels-v1\0");
+        }
         // 1. metric name
         if cfg.derived_input.is_some() {
             // Raw policies start with UTF-8 metric bytes; 0xff is impossible
@@ -285,6 +289,45 @@ mod tests {
         )
     }
 
+    #[test]
+    fn population_encoding_preserves_legacy_wire_and_separates_identity() {
+        use crate::grouping_projection::PopulationKeyEncoding;
+        let legacy = cfg(
+            "m",
+            AggregationType::Sum,
+            HashMap::new(),
+            vec!["host"],
+            60,
+            "",
+        );
+        let wire = serde_json::to_value(&legacy).unwrap();
+        assert!(wire.get("population_key_encoding").is_none());
+        let decoded: AggregationConfig = serde_json::from_value(wire).unwrap();
+        assert!(decoded.population_key_encoding.is_legacy());
+        assert_eq!(legacy.policy_fingerprint(), decoded.policy_fingerprint());
+        let mut canonical = legacy.clone();
+        canonical.population_key_encoding = PopulationKeyEncoding::CanonicalLabelsV1;
+        assert_ne!(legacy.policy_fingerprint(), canonical.policy_fingerprint());
+        let wire = serde_json::to_value(&canonical).unwrap();
+        assert_eq!(wire["population_key_encoding"], "canonical_labels_v1");
+        let decoded: AggregationConfig = serde_json::from_value(wire).unwrap();
+        assert_eq!(decoded.policy_fingerprint(), canonical.policy_fingerprint());
+        use crate::traits::SerializableToSink;
+        let mut sink = canonical.serialize_to_json();
+        // Transport wrappers supply the three label projections separately.
+        sink["groupingLabels"] = serde_json::to_value(&canonical.grouping_labels).unwrap();
+        sink["aggregatedLabels"] =
+            serde_json::to_value(&canonical.aggregated_labels.labels).unwrap();
+        sink["rollupLabels"] = serde_json::to_value(&canonical.rollup_labels.labels).unwrap();
+        let decoded = AggregationConfig::deserialize_from_json(&sink).unwrap();
+        assert_eq!(
+            decoded.population_key_encoding,
+            canonical.population_key_encoding
+        );
+
+        assert!(serde_json::from_str::<PopulationKeyEncoding>("\"canonical_labels_v2\"").is_err());
+    }
+
     #[test]
     fn same_config_yields_same_fingerprint() {
         let a = cfg(

crates/asap_types/src/precompute_plan.rs

@@ -410,6 +410,11 @@ impl PrecomputePlan {
             return Err(PrecomputePlanError::UnsupportedIngestEndpoint);
         }
         for config in &self.materializations {
+            if !config.population_key_encoding.is_legacy() {
+                return Err(PrecomputePlanError::CatalogContract(
+                    "population key encoding is not supported by the installed runtime".into(),
+                ));
+            }
             let Some(derived) = &config.derived_input else {
                 continue;
             };
@@ -824,6 +829,29 @@ mod source_window_cohort_tests {
         }))
         .unwrap()
     }
+    #[test]
+    fn canonical_population_encoding_is_not_yet_installable() {
+        let mut config = full_window();
+        config.slide_interval = config.window_size;
+        config.population_key_encoding =
+            crate::grouping_projection::PopulationKeyEncoding::CanonicalLabelsV1;
+        let envelope = PlanEnvelope {
+            plan_id: 1,
+            plan_version: 1,
+            generated_at_unix_ms: 0,
+            activation_unix_ms: 0,
+            expiry_unix_ms: None,
+            backend_compat: "test".into(),
+            planner_revision: "test".into(),
+            capability_snapshot_id: "test".into(),
+        };
+        let error = PrecomputePlan::build_backend_local(envelope, vec![config]).unwrap_err();
+        assert!(
+            error.to_string().contains("population key encoding"),
+            "{error}"
+        );
+    }
+
     #[test]
     fn full_sliding_cohort_preserves_explicit_windows_and_identity() {
         let target = full_window();

crates/asap_types/src/precompute_plan/catalog.rs

@@ -87,7 +87,8 @@ impl PrecomputePlan {
                     .map_err(invalid)?;
             }
             let expected_projection = config.effective_value_projection();
-            if data.partitioning != config.partitioning
+            if data.population_key_encoding != config.population_key_encoding
+                || data.partitioning != config.partitioning
                 || data.timestamp_column != config.table_timestamp_column
                 || data.source != expected_source
                 || &data.value_projection != expected_projection

crates/asap_types/src/sds.rs

@@ -842,6 +842,11 @@ pub struct DataDescriptor {
     pub timestamp_column: Option<String>,
     pub population_filter_canonical: String,
     pub group_by_keys: crate::GroupingProjection,
+    #[serde(
+        default,
+        skip_serializing_if = "crate::grouping_projection::PopulationKeyEncoding::is_legacy"
+    )]
+    pub population_key_encoding: crate::grouping_projection::PopulationKeyEncoding,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub partitioning: Option<PopulationPartitioning>,
     /// Versioned contract for timestamp interpretation and
@@ -905,6 +910,7 @@ impl DataDescriptor {
             &observation_semantics,
             None,
             None,
+            Default::default(),
         );
         Self {
             id,
@@ -914,6 +920,7 @@ impl DataDescriptor {
             population_filter_canonical,
             group_by_keys,
             partitioning: None,
+            population_key_encoding: Default::default(),
             observation_semantics,
         }
     }
@@ -927,6 +934,7 @@ impl DataDescriptor {
             &self.observation_semantics,
             self.partitioning,
             self.timestamp_column.as_deref(),
+            self.population_key_encoding,
         );
         self
     }
@@ -940,9 +948,18 @@ impl DataDescriptor {
             &self.observation_semantics,
             partitioning,
             self.timestamp_column.as_deref(),
+            self.population_key_encoding,
         );
         self
     }
+    pub fn with_population_key_encoding(
+        mut self,
+        encoding: crate::grouping_projection::PopulationKeyEncoding,
+    ) -> Self {
+        self.population_key_encoding = encoding;
+        let partitioning = self.partitioning;
+        self.with_partitioning(partitioning)
+    }
     pub fn with_timestamp_column(mut self, column: Option<String>) -> Self {
         self.timestamp_column = column;
         let partitioning = self.partitioning;
@@ -987,13 +1004,15 @@ impl DataDescriptor {
                 &self.observation_semantics,
                 self.partitioning,
                 self.timestamp_column.as_deref(),
+                self.population_key_encoding,
             )
         {
             return Err(SdsError("data descriptor ID/content mismatch".into()));
         }
         Ok(())
     }
 }
+#[allow(clippy::too_many_arguments)]
 fn data_descriptor_id(
     source: &DataSourceIdentity,
     value_projection: &ValueProjectionIdentity,
@@ -1002,6 +1021,7 @@ fn data_descriptor_id(
     observation_semantics: &str,
     partitioning: Option<PopulationPartitioning>,
     timestamp_column: Option<&str>,
+    population_key_encoding: crate::grouping_projection::PopulationKeyEncoding,
 ) -> DataDescriptorId {
     // Length framing keeps distinct typed sources, projections, predicates,
     // and grouping keys collision-free in the content identity.
@@ -1032,6 +1052,9 @@ fn data_descriptor_id(
         "|{}:{observation_semantics}",
         observation_semantics.len()
     ));
+    if !population_key_encoding.is_legacy() {
+        key = format!("data:population-key:canonical-labels-v1|{key}");
+    }
     DataDescriptorId(key)
 }
 
@@ -1040,6 +1063,24 @@ mod tests {
     use super::*;
     use std::collections::BTreeSet;
 
+    #[test]
+    fn population_encoding_is_part_of_data_identity_with_legacy_default() {
+        use crate::grouping_projection::PopulationKeyEncoding;
+        let legacy = DataDescriptor::new("m", "", ["host".to_string()]);
+        let wire = serde_json::to_value(&legacy).unwrap();
+        assert!(wire.get("population_key_encoding").is_none());
+        let decoded: DataDescriptor = serde_json::from_value(wire).unwrap();
+        assert_eq!(legacy.id(), decoded.id());
+        let canonical = legacy
+            .clone()
+            .with_population_key_encoding(PopulationKeyEncoding::CanonicalLabelsV1);
+        assert_ne!(legacy.id(), canonical.id());
+        canonical.validate().unwrap();
+        let reverted =
+            canonical.with_population_key_encoding(PopulationKeyEncoding::LegacyDelimited);
+        assert_eq!(legacy.id(), reverted.id());
+    }
+
     fn completion(epoch: u64) -> SummaryWindowCompletion {
         SummaryWindowCompletion {
             catalog_generation: CatalogGeneration {

crates/asap_types/src/summary_catalog.rs

@@ -120,6 +120,7 @@ impl SummaryCatalog {
                 )
                 .with_grouping_projection(config.grouping_labels.clone())
                 .with_partitioning(config.partitioning)
+                .with_population_key_encoding(config.population_key_encoding)
                 .with_timestamp_column(config.table_timestamp_column.clone());
                 Ok((
                     config.policy_fingerprint(),

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -1013,6 +1013,7 @@ mod tests {
         use asap_types::enums::WindowKind;
         use asap_types::{AggregationConfig, AggregationType, KeyByLabelNames};
         let aggregation = AggregationConfig {
+            population_key_encoding: Default::default(),
             aggregation_type: AggregationType::Sum,
             aggregation_sub_type: String::new(),
             parameters: HashMap::new(),
@@ -1075,6 +1076,7 @@ mod tests {
 
         let config =
             |aggregation_type, grouping: Vec<String>, aggregated: Vec<String>| AggregationConfig {
+                population_key_encoding: Default::default(),
                 aggregation_type,
                 aggregation_sub_type: String::new(),
                 parameters: match aggregation_type {

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

@@ -3704,6 +3704,7 @@ aggregations:
         for marker in active_agg_ids {
             let metric = format!("metric_{marker}");
             let cfg = AggregationConfig {
+                population_key_encoding: Default::default(),
                 aggregation_type: AggregationType::Sum,
                 aggregation_sub_type: String::new(),
                 parameters: HashMap::new(),

data_plane/src/precompute_engine/output_sink.rs

@@ -404,6 +404,7 @@ mod tests {
         // via `PolicyFingerprint::from_config`. Callers obtain the id
         // via `config.policy_fp_u64()`.
         AggregationConfig {
+            population_key_encoding: Default::default(),
             aggregation_type: AggregationType::Sum,
             aggregation_sub_type: String::new(),
             parameters: HashMap::new(),

data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs

@@ -257,6 +257,7 @@ mod tests {
 
     fn sum_agg_config(id: u64) -> AggregationConfig {
         AggregationConfig {
+            population_key_encoding: Default::default(),
             aggregation_type: AggregationType::Sum,
             aggregation_sub_type: String::new(),
             parameters: HashMap::new(),