Execute typed SQL grouping through SummaryStore

control_plane/Cargo.toml

@@ -76,17 +76,17 @@ asap_types.workspace = true
 # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this*
 # repo is a real one. Vendored locally instead of chased upstream -- see
 # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`.
-planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "e17a53715b445ab490f9c5f07d25fda730d252ec" }
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "e17a53715b445ab490f9c5f07d25fda730d252ec" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "7261dab97f8e43070e6b9e33c85493b12b9a351c" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "7261dab97f8e43070e6b9e33c85493b12b9a351c" }
 
 # L1 adoption (design-target-architecture.md Part B): the PromQL front
 # end itself, replacing control_plane's own query_parser/promql.rs.
 # Pinned via `rev`, not a floating branch reference. Same rev as
 # `planner-types`/`asap-aware-mapping` above -- these three MUST move
 # together (two revs of the same upstream repo's types in one workspace
 # resolve to distinct Rust types that won't unify).
-asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "e17a53715b445ab490f9c5f07d25fda730d252ec" }
-asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "e17a53715b445ab490f9c5f07d25fda730d252ec" }
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "7261dab97f8e43070e6b9e33c85493b12b9a351c" }
+asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "7261dab97f8e43070e6b9e33c85493b12b9a351c" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/src/clickhouse.rs

@@ -220,14 +220,31 @@ fn materialize_selected_sql(
     use planner_types::{post_asap::SummaryExpr, pre_asap::Reduction};
     let SummaryExpr::SummaryAgg {
         reduction: Reduction::Reduce(keys),
+        child,
         ..
     } = &node.expr
     else {
         return Err("SQL materialization requires a supported reduction".into());
     };
-    if keys.is_without() || !keys.keys().is_empty() {
-        return Err("SQL grouped source projection requires a typed grouping reader".into());
+    if keys.is_without() {
+        return Err("SQL grouping exclusion requires a resolved projection".into());
+    }
+    let SummaryExpr::KeepPreAsap(source) = &child.expr else {
+        return Err("SQL grouping requires a typed source subtree".into());
+    };
+    let source_schema = source.output_schema().map_err(|error| error.to_string())?;
+    let mut columns = Vec::new();
+    for key in keys.keys() {
+        let mut column = source_schema
+            .columns
+            .get(*key)
+            .cloned()
+            .ok_or("SQL grouping column is absent from source schema")?;
+        column.table = None;
+        columns.push(column);
     }
+    let grouping = asap_types::GroupingProjection::new(columns);
+    grouping.validate_table_group_codec()?;
     let (table, value, window, population, timestamp) =
         clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms)?;
     let window_secs = window.ok_or("SQL materialization requires a bounded window")?;
@@ -237,7 +254,7 @@ fn materialize_selected_sql(
         family: crate::physical::compiler::physical_materialization_family(family),
         window_secs,
         spatial_filter: String::new(),
-        grouping: Vec::new(),
+        grouping: grouping.names(),
         item_label: None,
         heap_update_mode: None,
         aggregation_input: AggregationInput::Raw,
@@ -248,6 +265,7 @@ fn materialize_selected_sql(
     )
     .map_err(|error| error.to_string())?;
     config.table_name = Some(table);
+    config.grouping_labels = grouping;
     config.value_projection = Some(value);
     config.table_timestamp_column = Some(timestamp);
     config.table_population = Some(population);
@@ -440,6 +458,26 @@ fn bind_selected_node(
     })
 }
 
+/// Evaluate only exact integer constant arithmetic at the installation boundary.
+/// No SQL text rewrite, floating coercion, or runtime-column evaluation is allowed.
+fn constant_int64(expr: &QueryExpr) -> Option<i64> {
+    use planner_types::pre_asap::{ArithmeticOpKind, ScalarValue};
+    match expr {
+        QueryExpr::Literal(ScalarValue::Int64(value)) => Some(*value),
+        QueryExpr::Arithmetic { op, left, right } => {
+            let left = constant_int64(left)?;
+            let right = constant_int64(right)?;
+            match op {
+                ArithmeticOpKind::Add => left.checked_add(right),
+                ArithmeticOpKind::Sub => left.checked_sub(right),
+                ArithmeticOpKind::Mul => left.checked_mul(right),
+                _ => None,
+            }
+        }
+        _ => None,
+    }
+}
+
 fn clickhouse_materialization_leaf_contract(
     node: &planner_types::post_asap::SummaryNode,
     evaluation_start_ms: u64,
@@ -573,7 +611,10 @@ fn clickhouse_materialization_leaf_contract(
             .ok_or_else(|| {
                 "SQL materialization predicate references an unknown column".to_string()
             })?;
-        match (name, op, right.as_ref()) {
+        let folded =
+            constant_int64(right).map(|value| QueryExpr::Literal(ScalarValue::Int64(value)));
+        let right = folded.as_ref().unwrap_or(right.as_ref());
+        match (name, op, right) {
             (
                 name,
                 CompareOpKind::Gt | CompareOpKind::Ge,
@@ -906,6 +947,33 @@ mod tests {
         .contains("ambiguous"));
     }
 
+    #[test]
+    fn constant_integer_boundaries_reject_overflow_and_dynamic_values() {
+        use planner_types::pre_asap::{ArithmeticOpKind, ScalarValue};
+        let literal = |value| QueryExpr::Literal(ScalarValue::Int64(value));
+        let subtract = |left, right| QueryExpr::Arithmetic {
+            op: ArithmeticOpKind::Sub,
+            left: std::rc::Rc::new(left),
+            right: std::rc::Rc::new(right),
+        };
+        assert_eq!(
+            constant_int64(&subtract(literal(1_788_891_296_000), literal(43_200_000))),
+            Some(1_788_848_096_000)
+        );
+        assert_eq!(
+            constant_int64(&subtract(literal(i64::MIN), literal(1))),
+            None
+        );
+        assert_eq!(
+            constant_int64(&subtract(QueryExpr::Column(0), literal(1))),
+            None
+        );
+        assert_eq!(
+            constant_int64(&QueryExpr::Literal(ScalarValue::Float64(1.0))),
+            None
+        );
+    }
+
     #[test]
     fn sql_evaluation_rejects_empty_reversed_and_unrepresentable_ranges() {
         for (start_ms, end_ms) in [(2, 1), (1, 1), (0, u64::MAX)] {

crates/asap_types/Cargo.toml

@@ -4,6 +4,7 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
+base64 = "0.21"
 tracing.workspace = true
 serde.workspace = true
 serde_json.workspace = true
@@ -32,4 +33,4 @@ sha2 = "0.10"
 # exactly (`control_plane/Cargo.toml`) -- two different revs of the same
 # git dependency in one workspace resolve to two distinct Rust types that
 # won't unify.
-planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "e17a53715b445ab490f9c5f07d25fda730d252ec" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "7261dab97f8e43070e6b9e33c85493b12b9a351c" }

crates/asap_types/src/grouping_projection.rs

@@ -24,6 +24,24 @@ impl GroupingProjection {
         }
         Ok(())
     }
+    pub fn validate_table_group_codec(&self) -> Result<(), String> {
+        self.validate()?;
+        for column in self.columns() {
+            crate::table_population::validate_column_name(&column.name)?;
+            if column.nullable {
+                return Err(
+                    "nullable table grouping requires an explicit null-key encoding".into(),
+                );
+            }
+            if !table_group_codec_type(&column.dtype) {
+                return Err(
+                    "table group codec requires integer, string, boolean, or supported Map values"
+                        .into(),
+                );
+            }
+        }
+        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));
@@ -60,6 +78,21 @@ impl GroupingProjection {
         serde_json::from_value(value.clone())
     }
 }
+// Float grouping requires SQL equality canonicalization (+0/-0 and NaNs),
+// and timestamps need explicit unit/timezone semantics. Neither is claimed by v1.
+fn table_group_codec_type(dtype: &DataType) -> bool {
+    match dtype {
+        DataType::Int64 | DataType::Utf8 | DataType::Bool => true,
+        DataType::Map { key, value, .. } => {
+            matches!(
+                key.as_ref(),
+                DataType::Int64 | DataType::Utf8 | DataType::Bool
+            ) && table_group_codec_type(value)
+        }
+        _ => false,
+    }
+}
+
 impl From<KeyByLabelNames> for GroupingProjection {
     fn from(mut names: KeyByLabelNames) -> Self {
         names.labels.sort();
@@ -223,3 +256,84 @@ pub(crate) fn serialize_config_grouping<S: Serializer>(
         grouping.serialize(serializer)
     }
 }
+
+/// Table group values travel through the string-key index as lossless JSON bytes.
+/// Map values use ordered key/value pairs, preserving duplicate keys.
+pub const TABLE_GROUP_OBSERVATION_SEMANTICS: &str = "asap.table-column-groups.base64-json.v1";
+
+pub fn encode_table_group_value(value: &serde_json::Value) -> Result<String, String> {
+    use base64::Engine;
+    let bytes = serde_json::to_vec(value).map_err(|error| error.to_string())?;
+    Ok(base64::engine::general_purpose::STANDARD.encode(bytes))
+}
+
+pub fn decode_table_group_value(encoded: &str) -> Result<serde_json::Value, String> {
+    use base64::Engine;
+    let bytes = base64::engine::general_purpose::STANDARD
+        .decode(encoded)
+        .map_err(|error| error.to_string())?;
+    serde_json::from_slice(&bytes).map_err(|error| error.to_string())
+}
+
+#[cfg(test)]
+mod codec_tests {
+    use super::*;
+    #[test]
+    fn table_group_codec_rejects_noncanonical_float_and_timestamp_keys() {
+        for dtype in [
+            DataType::Float64,
+            DataType::Timestamp,
+            DataType::Map {
+                key: Box::new(DataType::Utf8),
+                value: Box::new(DataType::Float64),
+                value_nullable: false,
+            },
+        ] {
+            assert!(
+                GroupingProjection::new(vec![Column::new("g", dtype, false)])
+                    .validate_table_group_codec()
+                    .is_err()
+            );
+        }
+        assert!(GroupingProjection::new(vec![Column::new(
+            "g",
+            DataType::Map {
+                key: Box::new(DataType::Utf8),
+                value: Box::new(DataType::Int64),
+                value_nullable: true,
+            },
+            false
+        )])
+        .validate_table_group_codec()
+        .is_ok());
+    }
+
+    #[test]
+    fn equivalent_json_spelling_has_one_shared_group_encoding() {
+        use base64::Engine;
+        let first = base64::engine::general_purpose::STANDARD.encode(br#""a\/b""#);
+        let second = base64::engine::general_purpose::STANDARD.encode(br#""a/b""#);
+        assert_ne!(first, second);
+        let normalize = |input: &str| {
+            encode_table_group_value(&decode_table_group_value(input).unwrap()).unwrap()
+        };
+        assert_eq!(normalize(&first), normalize(&second));
+    }
+
+    /// The grouping codec preserves null, escaping, duplicate map keys and exact integers.
+    #[test]
+    fn typed_group_codec_is_lossless_at_numeric_and_string_boundaries() {
+        for value in [
+            serde_json::Value::Null,
+            serde_json::json!("a,\"b\\c\n"),
+            serde_json::json!(i64::MAX),
+            serde_json::json!(i64::MIN),
+            serde_json::json!([["k", i64::MAX], ["k", null]]),
+        ] {
+            let encoded = encode_table_group_value(&value).unwrap();
+            assert!(!encoded.contains(['\"', ',', '\\']));
+            assert_eq!(decode_table_group_value(&encoded).unwrap(), value);
+        }
+        assert!(decode_table_group_value("not base64").is_err());
+    }
+}

crates/asap_types/src/policy_fingerprint.rs

@@ -127,6 +127,12 @@ impl PolicyFingerprint {
         }
         buf.push(0);
 
+        if cfg.table_name.is_some() && !cfg.grouping_labels.is_empty() {
+            buf.extend_from_slice(
+                crate::grouping_projection::TABLE_GROUP_OBSERVATION_SEMANTICS.as_bytes(),
+            );
+            buf.push(0);
+        }
         if !cfg.grouping_labels.is_legacy_labels() {
             buf.extend_from_slice(b"typed-grouping:");
             buf.extend_from_slice(

crates/asap_types/src/precompute_plan/catalog.rs

@@ -78,6 +78,21 @@ impl PrecomputePlan {
                     table_ref: table_ref.clone(),
                 },
             );
+            if config.table_name.is_some()
+                && !config.grouping_labels.is_empty()
+                && data.observation_semantics
+                    != crate::grouping_projection::TABLE_GROUP_OBSERVATION_SEMANTICS
+            {
+                return Err(invalid(
+                    "table grouping codec differs from installed contract",
+                ));
+            }
+            if config.table_name.is_some() {
+                config
+                    .grouping_labels
+                    .validate_table_group_codec()
+                    .map_err(invalid)?;
+            }
             let expected_projection = config.effective_value_projection();
             if data.partitioning != config.partitioning
                 || data.timestamp_column != config.table_timestamp_column

crates/asap_types/src/summary_catalog.rs

@@ -119,7 +119,11 @@ impl SummaryCatalog {
                         .population_filter_canonical()
                         .map_err(SummaryCatalogError::Descriptor)?,
                     config.grouping_labels.names(),
-                    "asap.timestamped-observations.v2",
+                    if config.table_name.is_some() && !config.grouping_labels.is_empty() {
+                        crate::grouping_projection::TABLE_GROUP_OBSERVATION_SEMANTICS
+                    } else {
+                        "asap.timestamped-observations.v2"
+                    },
                 )
                 .with_grouping_projection(config.grouping_labels.clone())
                 .with_partitioning(config.partitioning)