refactor(sds): share typed grouping projections

control_plane/src/clickhouse.rs

@@ -150,7 +150,7 @@ pub async fn compile_automatic_clickhouse_workload(
                 .map_err(crate::query_plan::QueryPlanError::Invalid)?;
             let binding = MaterializationBinding {
                 materialization: config.policy_fingerprint().into(),
-                output_grouping: PhysicalGrouping::Reduce(config.grouping_labels.labels.clone()),
+                output_grouping: PhysicalGrouping::Reduce(config.grouping_labels.names()),
                 window_ms: config.slide_interval * 1000,
                 pane_origin_ms: config.pane_origin_ms,
                 readout_lookback_ms: Some(query.end_ms - query.start_ms),
@@ -433,7 +433,7 @@ fn bind_selected_node(
     }
     Ok(MaterializationBinding {
         materialization: selected.policy_fingerprint().into(),
-        output_grouping: PhysicalGrouping::Reduce(selected.grouping_labels.labels.clone()),
+        output_grouping: PhysicalGrouping::Reduce(selected.grouping_labels.names()),
         window_ms: selected.slide_interval.saturating_mul(1000),
         pane_origin_ms: selected.pane_origin_ms,
         readout_lookback_ms: source_window.map(|seconds| seconds.saturating_mul(1000)),

control_plane/src/physical/compiler.rs

@@ -2189,7 +2189,7 @@ impl PhysicalCompiler {
                         readout_lookback_ms: source_window.map(|seconds| seconds.saturating_mul(1_000)),
                         materialization: fingerprint.into(),
                         output_grouping: PhysicalGrouping::Reduce(
-                            materialization.grouping_labels.labels.clone(),
+                            materialization.grouping_labels.names(),
                         ),
                         item_labels: materialization.aggregated_labels.labels.clone(),
                         window_ms: stored_interval_ms,
@@ -2835,7 +2835,7 @@ fn retained_partition_count(
             materialization.aggregation_type,
             A::Increase | A::MultipleIncrease | A::MinMax | A::MultipleMinMax
         )
-        || !materialization.grouping_labels.labels.is_empty()
+        || !materialization.grouping_labels.names().is_empty()
     {
         u128::from(input_cardinality.unwrap_or(1).max(1))
     } else {
@@ -3869,7 +3869,7 @@ mod tests {
             })
             .collect::<Vec<_>>();
         assert_eq!(heaps.len(), 1, "unpartitioned TopK owns one global CMS");
-        assert!(heaps[0].grouping_labels.labels.is_empty());
+        assert!(heaps[0].grouping_labels.names().is_empty());
         assert_eq!(heaps[0].aggregated_labels.labels, vec!["job"]);
         assert_eq!(heaps[0].parameters["weight_scale"], 1_000_000);
         assert_eq!(retained_partition_count(heaps[0], Some(5)), 1);

crates/asap_types/src/accumulator_spec.rs

@@ -403,7 +403,7 @@ impl AggregationConfig {
         };
 
         let grouping = if keyed {
-            Some(self.grouping_labels.clone())
+            Some(self.grouping_labels.label_names())
         } else {
             None
         };

crates/asap_types/src/aggregation_config.rs

@@ -99,7 +99,8 @@ pub struct PrecomputeMaterialization {
     pub aggregation_type: AggregationType,
     pub aggregation_sub_type: String,
     pub parameters: HashMap<String, Value>,
-    pub grouping_labels: KeyByLabelNames,
+    #[serde(serialize_with = "crate::grouping_projection::serialize_config_grouping")]
+    pub grouping_labels: crate::GroupingProjection,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub partitioning: Option<crate::sds::PopulationPartitioning>,
     pub aggregated_labels: KeyByLabelNames,
@@ -231,7 +232,7 @@ impl PrecomputeMaterialization {
         aggregation_type: AggregationType,
         aggregation_sub_type: String,
         parameters: HashMap<String, Value>,
-        grouping_labels: KeyByLabelNames,
+        grouping_labels: impl Into<crate::GroupingProjection>,
         aggregated_labels: KeyByLabelNames,
         rollup_labels: KeyByLabelNames,
         original_yaml: String,
@@ -252,7 +253,7 @@ impl PrecomputeMaterialization {
             aggregation_type,
             aggregation_sub_type,
             parameters,
-            grouping_labels,
+            grouping_labels: grouping_labels.into(),
             partitioning: None,
             aggregated_labels,
             rollup_labels,
@@ -342,7 +343,8 @@ impl PrecomputeMaterialization {
         let original_yaml = data["originalYaml"].as_str().unwrap_or("").to_string();
 
         // Deserialize KeyByLabelNames - assuming they have deserialize_from_json methods
-        let grouping_labels = KeyByLabelNames::deserialize_from_json(&data["groupingLabels"])?;
+        let grouping_labels =
+            crate::GroupingProjection::deserialize_from_json(&data["groupingLabels"])?;
         let aggregated_labels = KeyByLabelNames::deserialize_from_json(&data["aggregatedLabels"])?;
         let rollup_labels = KeyByLabelNames::deserialize_from_json(&data["rollupLabels"])?;
 
@@ -445,15 +447,8 @@ impl PrecomputeMaterialization {
         // fixtures that still spell out the field parse cleanly.
 
         let labels = &aggregation_data["labels"];
-        let grouping_labels = KeyByLabelNames::new(
-            labels["grouping"]
-                .as_sequence()
-                .ok_or_else(|| anyhow::anyhow!("Missing grouping labels"))?
-                .iter()
-                .filter_map(|v| v.as_str())
-                .map(|s| s.to_string())
-                .collect(),
-        );
+        let grouping_labels: crate::GroupingProjection =
+            serde_yaml::from_value(labels["grouping"].clone())?;
         let aggregated_labels = KeyByLabelNames::new(
             labels["aggregated"]
                 .as_sequence()

crates/asap_types/src/grouping_projection.rs

@@ -0,0 +1,225 @@
+//! Typed source columns defining one summary population key.
+use crate::KeyByLabelNames;
+use planner_types::pre_asap::{Column, DataType};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GroupingProjection(Vec<Column>);
+
+impl GroupingProjection {
+    pub fn new(mut columns: Vec<Column>) -> Self {
+        columns.sort_by(|a, b| a.name.cmp(&b.name));
+        Self(columns)
+    }
+    pub fn validate(&self) -> Result<(), String> {
+        let mut previous: Option<&str> = None;
+        for column in &self.0 {
+            if column.name.is_empty()
+                || column.table.is_some()
+                || previous == Some(column.name.as_str())
+            {
+                return Err("grouping columns must be unqualified and unique".into());
+            }
+            previous = Some(&column.name);
+        }
+        Ok(())
+    }
+    pub fn push(&mut self, name: String) {
+        self.0.push(Column::new(name, DataType::Utf8, false));
+        self.0.sort_by(|a, b| a.name.cmp(&b.name));
+        self.0.dedup();
+    }
+    pub fn columns(&self) -> &[Column] {
+        &self.0
+    }
+    pub fn iter(&self) -> impl Iterator<Item = &String> {
+        self.0.iter().map(|c| &c.name)
+    }
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+    pub fn names(&self) -> Vec<String> {
+        self.0.iter().map(|c| c.name.clone()).collect()
+    }
+    pub fn label_names(&self) -> KeyByLabelNames {
+        KeyByLabelNames::new(self.names())
+    }
+    pub fn is_legacy_labels(&self) -> bool {
+        self.0
+            .iter()
+            .all(|c| c.dtype == DataType::Utf8 && !c.nullable && c.table.is_none())
+    }
+    pub fn serialize_to_json(&self) -> serde_json::Value {
+        if self.is_legacy_labels() {
+            serde_json::json!(self.names())
+        } else {
+            serde_json::to_value(self).expect("grouping projection serializes")
+        }
+    }
+    pub fn deserialize_from_json(value: &serde_json::Value) -> Result<Self, serde_json::Error> {
+        serde_json::from_value(value.clone())
+    }
+}
+impl From<KeyByLabelNames> for GroupingProjection {
+    fn from(mut names: KeyByLabelNames) -> Self {
+        names.labels.sort();
+        names.labels.dedup();
+        Self::new(
+            names
+                .labels
+                .into_iter()
+                .map(|name| Column::new(name, DataType::Utf8, false))
+                .collect(),
+        )
+    }
+}
+impl FromIterator<String> for GroupingProjection {
+    fn from_iter<T: IntoIterator<Item = String>>(iter: T) -> Self {
+        KeyByLabelNames::new(iter.into_iter().collect()).into()
+    }
+}
+impl<'de> Deserialize<'de> for GroupingProjection {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        #[derive(Deserialize)]
+        #[serde(untagged)]
+        enum Wire {
+            Columns(Vec<Column>),
+            Names(Vec<String>),
+            Legacy(KeyByLabelNames),
+        }
+        Ok(match Wire::deserialize(deserializer)? {
+            Wire::Columns(columns) => Self::new(columns),
+            Wire::Names(names) => KeyByLabelNames::new(names).into(),
+            Wire::Legacy(names) => names.into(),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    /// Legacy label lists become one non-null string column per name.
+    #[test]
+    fn legacy_and_typed_groups_share_one_projection() {
+        let legacy: GroupingProjection =
+            serde_json::from_value(serde_json::json!({"labels":["job"]})).unwrap();
+        assert_eq!(
+            legacy.columns(),
+            &[Column::new("job", DataType::Utf8, false)]
+        );
+        let typed: GroupingProjection = serde_json::from_value(
+            serde_json::json!([{"name":"job","dtype":"utf8","nullable":false}]),
+        )
+        .unwrap();
+        assert_eq!(typed, legacy);
+        assert_eq!(typed.serialize_to_json(), serde_json::json!(["job"]));
+        assert_eq!(
+            serde_json::to_value(&typed).unwrap(),
+            serde_json::json!(["job"])
+        );
+        #[derive(Serialize)]
+        struct LegacyConfig {
+            #[serde(serialize_with = "serialize_config_grouping")]
+            grouping_labels: GroupingProjection,
+        }
+        assert_eq!(
+            serde_json::to_value(LegacyConfig {
+                grouping_labels: typed
+            })
+            .unwrap(),
+            serde_json::json!({"grouping_labels":{"labels":["job"]}})
+        );
+    }
+    /// Numeric grouping types must survive wire transport rather than become labels.
+    #[test]
+    fn typed_group_retains_type_and_rejects_duplicate_columns() {
+        let typed = GroupingProjection::new(vec![Column::new("tenant", DataType::Int64, true)]);
+        let decoded: GroupingProjection =
+            serde_json::from_value(typed.serialize_to_json()).unwrap();
+        assert_eq!(decoded, typed);
+        assert!(!decoded.is_legacy_labels());
+        let duplicate = GroupingProjection::new(vec![
+            Column::new("tenant", DataType::Int64, true),
+            Column::new("tenant", DataType::Utf8, false),
+        ]);
+        assert!(duplicate.validate().is_err());
+    }
+}
+
+#[cfg(test)]
+mod identity_tests {
+    use super::*;
+    /// A type or nullability change cannot reuse an incompatible summary population.
+    #[test]
+    fn descriptor_identity_includes_group_type_and_nullability() {
+        let legacy = crate::sds::DataDescriptor::new("m", "", vec!["tenant".into()]);
+        let same = legacy
+            .clone()
+            .with_grouping_projection(KeyByLabelNames::new(vec!["tenant".into()]).into());
+        assert_eq!(legacy.id, same.id);
+        let numeric = legacy
+            .clone()
+            .with_grouping_projection(GroupingProjection::new(vec![Column::new(
+                "tenant",
+                DataType::Int64,
+                false,
+            )]));
+        let nullable = legacy
+            .clone()
+            .with_grouping_projection(GroupingProjection::new(vec![Column::new(
+                "tenant",
+                DataType::Utf8,
+                true,
+            )]));
+        assert_ne!(legacy.id, numeric.id);
+        assert_ne!(legacy.id, nullable.id);
+        assert_ne!(numeric.id, nullable.id);
+        assert!(numeric
+            .validate()
+            .unwrap_err()
+            .to_string()
+            .contains("string labels"));
+        assert!(nullable
+            .validate()
+            .unwrap_err()
+            .to_string()
+            .contains("string labels"));
+        for grouping in [numeric.group_by_keys, nullable.group_by_keys] {
+            let table = crate::sds::DataDescriptor::new_typed(
+                crate::sds::DataSourceIdentity::Table {
+                    table_ref: "samples".into(),
+                },
+                crate::sds::ValueProjectionIdentity::Column {
+                    name: "value".into(),
+                },
+                "",
+                Vec::<String>::new(),
+                "table.samples.v1",
+            )
+            .with_grouping_projection(grouping);
+            table.validate().unwrap();
+        }
+    }
+}
+
+impl Serialize for GroupingProjection {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        if self.is_legacy_labels() {
+            self.names().serialize(serializer)
+        } else {
+            self.0.serialize(serializer)
+        }
+    }
+}
+
+/// Preserve the older precompute-config object shape for ordinary label groups.
+pub(crate) fn serialize_config_grouping<S: Serializer>(
+    grouping: &GroupingProjection,
+    serializer: S,
+) -> Result<S::Ok, S::Error> {
+    if grouping.is_legacy_labels() {
+        grouping.label_names().serialize(serializer)
+    } else {
+        grouping.serialize(serializer)
+    }
+}

crates/asap_types/src/lib.rs

@@ -4,6 +4,7 @@ pub mod aggregation_config;
 pub mod aggregation_type;
 pub mod enums;
 pub mod executable_plan;
+pub mod grouping_projection;
 pub mod key_by_label_names;
 pub mod monitor_spec;
 pub mod policy_fingerprint;
@@ -22,6 +23,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 key_by_label_names::KeyByLabelNames;
 pub use monitor_spec::{MonitorFunctional, MonitorSpec};
 pub use policy_fingerprint::PolicyFingerprint;

crates/asap_types/src/policy_fingerprint.rs

@@ -121,12 +121,22 @@ impl PolicyFingerprint {
 
         // 5. grouping_labels (already sorted at construction per
         //    KeyByLabelNames invariant; encode as `,`-joined list)
-        for l in &cfg.grouping_labels.labels {
+        for l in &cfg.grouping_labels.names() {
             buf.extend_from_slice(l.as_bytes());
             buf.push(b',');
         }
         buf.push(0);
 
+        if !cfg.grouping_labels.is_legacy_labels() {
+            buf.extend_from_slice(b"typed-grouping:");
+            buf.extend_from_slice(
+                serde_json::to_string(&cfg.grouping_labels)
+                    .expect("group projection serializes")
+                    .as_bytes(),
+            );
+            buf.push(0);
+        }
+
         // 6. aggregated_labels
         for l in &cfg.aggregated_labels.labels {
             buf.extend_from_slice(l.as_bytes());