From 0b2507f84a38b543065a7a634f441ba687cdfd77 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:35:48 -0600 Subject: [PATCH 1/3] refactor(sds): share typed grouping projections --- control_plane/src/clickhouse.rs | 4 +- control_plane/src/physical/compiler.rs | 6 +- crates/asap_types/src/accumulator_spec.rs | 2 +- crates/asap_types/src/aggregation_config.rs | 21 +- crates/asap_types/src/grouping_projection.rs | 202 ++++++++++++++++++ crates/asap_types/src/lib.rs | 2 + crates/asap_types/src/policy_fingerprint.rs | 12 +- crates/asap_types/src/precompute_plan.rs | 10 +- .../asap_types/src/precompute_plan/catalog.rs | 4 +- crates/asap_types/src/routing_index.rs | 2 +- crates/asap_types/src/sds.rs | 33 ++- crates/asap_types/src/summary_catalog.rs | 3 +- data_plane/src/drivers/ingest/otel.rs | 4 +- .../drivers/ingest/prometheus_remote_write.rs | 12 +- data_plane/src/drivers/query/servers/http.rs | 2 +- .../src/precompute_engine/ingest_handler.rs | 10 +- .../src/precompute_engine/output_sink.rs | 6 +- data_plane/src/precompute_engine/worker.rs | 2 +- .../sketch_db/backfill/clickhouse_reader.rs | 21 ++ .../sketch_db/backfill/processor.rs | 6 +- .../storage_engines/sketch_db/index/mod.rs | 6 +- .../sketch_db/lifecycle/eviction.rs | 2 +- .../sketch_db/lifecycle/reconcile.rs | 2 +- .../tests/test_utilities/engine_factories.rs | 16 +- .../summary-catalog-sds-architecture.md | 14 +- 25 files changed, 332 insertions(+), 72 deletions(-) create mode 100644 crates/asap_types/src/grouping_projection.rs diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index a89dee86b..f674399e1 100644 --- a/control_plane/src/clickhouse.rs +++ b/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)), diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index a914e40df..7ae535234 100644 --- a/control_plane/src/physical/compiler.rs +++ b/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::>(); 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); diff --git a/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs index 2e56219d2..23732563a 100644 --- a/crates/asap_types/src/accumulator_spec.rs +++ b/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 }; diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index 5edbe279d..3e9f5cab3 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/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, - 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, pub aggregated_labels: KeyByLabelNames, @@ -231,7 +232,7 @@ impl PrecomputeMaterialization { aggregation_type: AggregationType, aggregation_sub_type: String, parameters: HashMap, - grouping_labels: KeyByLabelNames, + grouping_labels: impl Into, 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() diff --git a/crates/asap_types/src/grouping_projection.rs b/crates/asap_types/src/grouping_projection.rs new file mode 100644 index 000000000..0b34cb894 --- /dev/null +++ b/crates/asap_types/src/grouping_projection.rs @@ -0,0 +1,202 @@ +//! 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); + +impl GroupingProjection { + pub fn new(mut columns: Vec) -> 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 { + self.0.iter().map(|c| &c.name) + } + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + pub fn names(&self) -> Vec { + 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 { + serde_json::from_value(value.clone()) + } +} +impl From 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 for GroupingProjection { + fn from_iter>(iter: T) -> Self { + KeyByLabelNames::new(iter.into_iter().collect()).into() + } +} +impl<'de> Deserialize<'de> for GroupingProjection { + fn deserialize>(deserializer: D) -> Result { + #[derive(Deserialize)] + #[serde(untagged)] + enum Wire { + Columns(Vec), + Names(Vec), + 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); + numeric.validate().unwrap(); + nullable.validate().unwrap(); + } +} + +impl Serialize for GroupingProjection { + fn serialize(&self, serializer: S) -> Result { + 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( + grouping: &GroupingProjection, + serializer: S, +) -> Result { + if grouping.is_legacy_labels() { + grouping.label_names().serialize(serializer) + } else { + grouping.serialize(serializer) + } +} diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index ccf828726..3c0a23997 100644 --- a/crates/asap_types/src/lib.rs +++ b/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; diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 3daae2f94..19daa348d 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/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()); diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 270e262ed..a3017d16d 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -144,7 +144,7 @@ pub struct StateSchemaContract { deserialize_with = "crate::sds::deserialize_state_value_projection" )] pub value_projection: crate::sds::ValueProjectionIdentity, - pub group_by: Vec, + pub group_by: crate::GroupingProjection, pub window: StateWindowContract, pub encodings: Vec, } @@ -234,7 +234,7 @@ impl PrecomputePlan { family, source, value_projection, - group_by: materialization.grouping_labels.labels.clone(), + group_by: materialization.grouping_labels.clone(), window: StateWindowContract { kind: materialization.window_type, size_ms: materialization.window_size.saturating_mul(1_000), @@ -389,6 +389,10 @@ impl PrecomputePlan { } let mut materializations = BTreeSet::new(); for materialization in &self.materializations { + materialization + .grouping_labels + .validate() + .map_err(PrecomputePlanError::CatalogContract)?; materialization .window_layout .validate(materialization.window_size, materialization.slide_interval) @@ -474,7 +478,7 @@ impl PrecomputePlan { || schema.family != family || schema.source != source || schema.value_projection != value_projection - || schema.group_by != materialization.grouping_labels.labels + || schema.group_by != materialization.grouping_labels || schema.window.kind != materialization.window_type || schema.window.size_ms != materialization.window_size.saturating_mul(1_000) || schema.window.slide_ms diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index 80d682ab4..b6845a2df 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -85,7 +85,7 @@ impl PrecomputePlan { || &data.value_projection != expected_projection || data.population_filter_canonical != config.population_filter_canonical().map_err(invalid)? - || data.group_by_keys != config.grouping_labels.labels.iter().cloned().collect() + || data.group_by_keys != config.grouping_labels { return Err(invalid("source/population/grouping differs from catalog")); } @@ -136,7 +136,7 @@ impl PrecomputePlan { || schema.family != expected_family || schema.source != source || &schema.value_projection != projection - || schema.group_by != config.grouping_labels.labels + || schema.group_by != config.grouping_labels || schema.window.kind != config.window_type || schema.window.size_ms != size || schema.window.slide_ms != expected_slide diff --git a/crates/asap_types/src/routing_index.rs b/crates/asap_types/src/routing_index.rs index 2a21f2379..86bd497cf 100644 --- a/crates/asap_types/src/routing_index.rs +++ b/crates/asap_types/src/routing_index.rs @@ -102,7 +102,7 @@ impl RoutingIndex { let mut hit = None; for fp in self.candidates_for_metric(metric) { let cfg = self.get(*fp)?; - let policy_keys: BTreeSet<_> = cfg.grouping_labels.labels.iter().cloned().collect(); + let policy_keys: BTreeSet<_> = cfg.grouping_labels.iter().cloned().collect(); if cfg.aggregation_type != agg_type || &policy_keys != group_by_keys || !cfg.spatial_filter_normalized.is_empty() diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index 7dc648cde..13a801817 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -4,7 +4,7 @@ use crate::{AggregationType, PrecomputeMaterialization}; use planner_types::post_asap::{SketchAlgorithm, SketchParams, SummaryFamilyType}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::BTreeMap; #[derive(Debug, Clone, PartialEq, Eq)] pub struct SdsError(pub String); @@ -775,7 +775,7 @@ pub struct DataDescriptor { #[serde(default, skip_serializing_if = "Option::is_none")] pub timestamp_column: Option, pub population_filter_canonical: String, - pub group_by_keys: BTreeSet, + pub group_by_keys: crate::GroupingProjection, #[serde(default, skip_serializing_if = "Option::is_none")] pub partitioning: Option, /// Versioned contract for timestamp interpretation and @@ -851,6 +851,19 @@ impl DataDescriptor { observation_semantics, } } + pub fn with_grouping_projection(mut self, grouping: crate::GroupingProjection) -> Self { + self.group_by_keys = grouping; + self.id = data_descriptor_id( + &self.source, + &self.value_projection, + &self.population_filter_canonical, + &self.group_by_keys, + &self.observation_semantics, + self.partitioning, + self.timestamp_column.as_deref(), + ); + self + } pub fn with_partitioning(mut self, partitioning: Option) -> Self { self.partitioning = partitioning; self.id = data_descriptor_id( @@ -874,6 +887,12 @@ impl DataDescriptor { } pub fn validate(&self) -> Result<(), SdsError> { self.value_projection.validate().map_err(SdsError)?; + self.group_by_keys.validate().map_err(SdsError)?; + if matches!(self.source, DataSourceIdentity::Table { .. }) { + for column in self.group_by_keys.columns() { + crate::table_population::validate_column_name(&column.name).map_err(SdsError)?; + } + } if let Some(column) = &self.timestamp_column { if !matches!(self.source, DataSourceIdentity::Table { .. }) || column.is_empty() { return Err(SdsError( @@ -902,7 +921,7 @@ fn data_descriptor_id( source: &DataSourceIdentity, value_projection: &ValueProjectionIdentity, filter: &str, - group_by: &BTreeSet, + group_by: &crate::GroupingProjection, observation_semantics: &str, partitioning: Option, timestamp_column: Option<&str>, @@ -924,9 +943,14 @@ fn data_descriptor_id( if let Some(column) = timestamp_column { key.push_str(&format!("|timestamp-ms:{}:{column}", column.len())); } - for name in group_by { + for name in group_by.names() { key.push_str(&format!("|{}:{name}", name.len())); } + if !group_by.is_legacy_labels() { + let typed = + canonical(&serde_json::to_value(group_by).expect("group projection serializes")); + key.push_str(&format!("|group-types:{}:{typed}", typed.len())); + } key.push_str(&format!( "|{}:{observation_semantics}", observation_semantics.len() @@ -937,6 +961,7 @@ fn data_descriptor_id( #[cfg(test)] mod tests { use super::*; + use std::collections::BTreeSet; fn completion(epoch: u64) -> SummaryWindowCompletion { SummaryWindowCompletion { diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 42ce40c75..b7864c2c5 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -118,9 +118,10 @@ impl SummaryCatalog { config .population_filter_canonical() .map_err(SummaryCatalogError::Descriptor)?, - config.grouping_labels.labels.clone(), + config.grouping_labels.names(), "asap.timestamped-observations.v2", ) + .with_grouping_projection(config.grouping_labels.clone()) .with_partitioning(config.partitioning) .with_timestamp_column(config.table_timestamp_column.clone()); Ok(( diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index b6d9da82c..2dcad0930 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -606,9 +606,7 @@ fn resolve_bucket_sid_for_agg_config( point_labels: &HashMap, ) -> (u64, asap_types::PolicyFingerprint) { let grouping_pairs: Vec<(&str, &str)> = config - .grouping_labels - .labels - .iter() + .grouping_labels.iter() .map(|name| { let v = point_labels.get(name).map(|s| s.as_str()).unwrap_or(""); (name.as_str(), v) diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index d557031d9..985079ac5 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -363,9 +363,7 @@ impl PrometheusRemoteWriteReceiver { let mut labels = group_key.as_population_labels(); if labels.is_empty() { labels = config - .grouping_labels - .labels - .iter() + .grouping_labels.iter() .cloned() .zip(group_key.values().labels) .collect(); @@ -669,9 +667,7 @@ fn route_messages( Vec::new() } else { config - .grouping_labels - .labels - .iter() + .grouping_labels.iter() .map(|name| { ( name.as_str(), @@ -957,7 +953,7 @@ mod tests { aggregation_type: AggregationType::Sum, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(vec!["job".into()]), + grouping_labels: KeyByLabelNames::new(vec!["job".into()]).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -1028,7 +1024,7 @@ mod tests { } _ => HashMap::new(), }, - grouping_labels: KeyByLabelNames::new(grouping), + grouping_labels: KeyByLabelNames::new(grouping).into(), aggregated_labels: KeyByLabelNames::new(aggregated), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 8b9444ac4..6a390db68 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3658,7 +3658,7 @@ aggregations: aggregation_type: AggregationType::Sum, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::empty(), + grouping_labels: KeyByLabelNames::empty().into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), diff --git a/data_plane/src/precompute_engine/ingest_handler.rs b/data_plane/src/precompute_engine/ingest_handler.rs index c6a7c2af0..00958f313 100644 --- a/data_plane/src/precompute_engine/ingest_handler.rs +++ b/data_plane/src/precompute_engine/ingest_handler.rs @@ -278,14 +278,14 @@ impl IngestState { labels: &std::collections::HashMap, config: &AggregationConfig, ) -> Arc { - crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.labels.iter().map( - |name| { + crate::precompute_engine::group_key::intern_pairs( + config.grouping_labels.iter().map(|name| { ( name.as_str(), labels.get(name).map(String::as_str).unwrap_or(""), ) - }, - )) + }), + ) } } @@ -296,7 +296,7 @@ fn extract_group_key( config: &AggregationConfig, ) -> Arc { let labels = parse_labels_from_series_key(series_key); - crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.labels.iter().map( + crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.iter().map( |name| { ( name.as_str(), diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 6bf61842e..8bbef5316 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -174,9 +174,7 @@ impl SketchStoreSink { if let Some(revision) = &output.input_revision { let group_values = output.population_labels.clone().unwrap_or_else(|| { agg_cfg - .grouping_labels - .labels - .iter() + .grouping_labels.iter() .cloned() .zip(output.key.clone().unwrap_or_default().labels) .collect() @@ -372,7 +370,7 @@ mod tests { parameters: HashMap::new(), grouping_labels: KeyByLabelNames::new( grouping_keys.iter().map(|s| s.to_string()).collect(), - ), + ).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 6fdbeed50..73e8b9a10 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -1273,7 +1273,7 @@ pub fn extract_key_from_series(series_key: &str, config: &AggregationConfig) -> let labels = parse_labels_from_series_key(series_key); let mut values = Vec::new(); - for label_name in &config.grouping_labels.labels { + for label_name in &config.grouping_labels.names() { if let Some(val) = labels.get(label_name.as_str()) { values.push(val.to_string()); } else { diff --git a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs index 33190ad3a..596c2a7ff 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs @@ -153,6 +153,9 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor } .into()); } + if !materialization.grouping_labels.is_legacy_labels() { + return Err("typed table grouping requires a typed grouping reader".into()); + } let mut source_config = config.clone(); source_config.database = database.clone(); source_config.table = table.clone(); @@ -395,6 +398,24 @@ mod tests { ) .unwrap(); assert_eq!(reader.source_name(), "ClickHouseReader"); + let mut typed = materialization.clone(); + typed.grouping_labels = + asap_types::GroupingProjection::new(vec![planner_types::pre_asap::Column::new( + "tenant", + planner_types::pre_asap::DataType::Int64, + false, + )]); + let rejected = factory( + &BackfillSource::ClickHouse { + database: "metrics".into(), + table: "another_table".into(), + }, + &typed, + ); + assert!( + matches!(rejected, Err(error) if error.to_string().contains("typed grouping reader")) + ); + assert!(factory( &BackfillSource::ClickHouse { database: "another_database".into(), diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs index ad01a9070..70b6d7c4f 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs @@ -93,7 +93,7 @@ use super::BackfillRegistry; fn extract_group_key(series_key: &str, config: &AggregationConfig) -> String { let labels = parse_labels_from_series_key(series_key); let mut values = Vec::new(); - for label_name in &config.grouping_labels.labels { + for label_name in &config.grouping_labels.names() { if let Some(val) = labels.get(label_name.as_str()) { values.push(*val); } else { @@ -127,9 +127,7 @@ fn resolve_backfill_bucket_sid( ) -> u64 { let labels = parse_labels_from_series_key(series_key); let grouping_pairs: Vec<(&str, &str)> = config - .grouping_labels - .labels - .iter() + .grouping_labels.iter() .map(|name| (name.as_str(), *labels.get(name.as_str()).unwrap_or(&""))) .collect(); let attrs_fp = canonical_attrs_fingerprint(&grouping_pairs); diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index f480e00ab..930ca6a9a 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -155,7 +155,7 @@ fn build_attrs_fp_and_label_map( .as_ref() .map(|k| k.labels.clone()) .unwrap_or_default(); - let key_names = &agg_cfg.grouping_labels.labels; + let key_names = &agg_cfg.grouping_labels.names(); let mut attrs_fp = String::new(); let mut label_values_map: BTreeMap = BTreeMap::new(); for (k, v) in key_names.iter().zip(label_values_vec.iter()) { @@ -2572,7 +2572,7 @@ impl SketchStore { accumulator: &dyn crate::storage_engines::types::AggregateCore, ) -> Option { let (_attrs_fp, label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output); - let key_names = &agg_cfg.grouping_labels.labels; + let key_names = &agg_cfg.grouping_labels.names(); let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg); let (capability, accuracy) = agg_kind.capability_and_accuracy(); @@ -2667,7 +2667,7 @@ impl SketchStore { let target_agg_type = agg_cfg.aggregation_type; let target_params = canonical_parameters(&agg_cfg.parameters); let target_group_keys: BTreeSet = - agg_cfg.grouping_labels.labels.iter().cloned().collect(); + agg_cfg.grouping_labels.iter().cloned().collect(); // Collect the matching sids under a short read lock; then call // `remove_instance` per sid (which takes its own write lock). diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs index 7cdaf6e92..174c4ae5c 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -260,7 +260,7 @@ mod tests { aggregation_type: AggregationType::Sum, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::empty(), + grouping_labels: KeyByLabelNames::empty().into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs index 128c82622..25b085e13 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs @@ -197,7 +197,7 @@ fn signature_from_agg_config(cfg: &AggregationConfig) -> Vec { parameters_canonical: canonical_parameters(&cfg.parameters), spatial_filter_canonical: cfg.spatial_filter_normalized.clone(), }; - let group_by_keys: BTreeSet = cfg.grouping_labels.labels.iter().cloned().collect(); + let group_by_keys: BTreeSet = cfg.grouping_labels.iter().cloned().collect(); signature_bytes(&cfg.metric, &agg_kind, &group_by_keys) } diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs index 12af96e44..3fce2d44b 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -93,7 +93,7 @@ pub fn create_engine_single_pop_with_aggregated( aggregation_type, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()), + grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()).into(), aggregated_labels: KeyByLabelNames::new(aggregated_label_strings), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -190,7 +190,7 @@ pub fn create_engine_dual_input( aggregation_type: value_agg_type, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()), + grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -217,7 +217,7 @@ pub fn create_engine_dual_input( aggregation_type: key_agg_type, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()), + grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()).into(), aggregated_labels: KeyByLabelNames::new(aggregated_label_strings), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -309,7 +309,7 @@ pub fn create_engine_two_metrics( aggregation_type: aggregation_type_a, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(labels_a.clone()), + grouping_labels: KeyByLabelNames::new(labels_a.clone()).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -335,7 +335,7 @@ pub fn create_engine_two_metrics( aggregation_type: aggregation_type_b, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(labels_b.clone()), + grouping_labels: KeyByLabelNames::new(labels_b.clone()).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -437,7 +437,7 @@ pub fn create_engine_three_metrics( aggregation_type: agg_type, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(labels.clone()), + grouping_labels: KeyByLabelNames::new(labels.clone()).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -516,7 +516,7 @@ pub fn create_engine_multi_timestamp( aggregation_type, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()), + grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), @@ -587,7 +587,7 @@ pub fn create_engine_multi_timestamp_with_window( aggregation_type, aggregation_sub_type: String::new(), parameters: HashMap::new(), - grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()), + grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()).into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 82fd1c4a8..5d5ec5ed7 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -184,14 +184,24 @@ compatibility DTO while older sidecars are read. The implemented `SummaryDescriptor` currently contains one `SummaryOperator`, one derived `FidelityGuarantee`, and a numeric state-schema version. The -implemented `DataDescriptor` contains metric name, canonical population filter, -grouping keys and versioned observation semantics. The shared contract now also +implemented `DataDescriptor` contains typed source and value projections, a +canonical population filter, typed grouping columns and versioned observation +semantics. The shared contract now also defines `SummaryInstance`, `ObservedSummaryInventory`, placement, completeness, state references, catalog generation and ephemeral leases. The control-plane reconciler emits create, update, recover, retire, garbage-collect, promote and expire actions. Summary payloads and the application of those actions remain in the SummaryStore runtime. +The same `GroupingProjection` supplies source columns to precompute configuration, +`DataDescriptor` and the state-schema contract. Each column retains the Planner's +name, type and nullability; routing derives names without storing a second list. +Legacy label lists decode as non-null UTF-8 columns and keep their existing +identities. A changed type or nullability changes catalog and policy identity. +A SQL map column is one grouping value, not a set of PromQL labels. Typed +ClickHouse group transport remains a separate execution capability: the current +reader rejects non-label projections until that transport is implemented. + `DataDescriptor`, precompute configuration and state-schema validation share `ValueProjectionIdentity`: sample value, named column, or a finite numeric constant using the Planner's `ScalarValue`. A constant input such as `1` does From 47ef14798e99802f694a8e345dd04e43ba838ebb Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:39:46 -0600 Subject: [PATCH 2/3] fix(sds): reject typed groups on string-label sources --- crates/asap_types/src/grouping_projection.rs | 27 ++++++++++++++++++-- crates/asap_types/src/precompute_plan.rs | 8 ++++++ crates/asap_types/src/sds.rs | 8 ++++++ 3 files changed, 41 insertions(+), 2 deletions(-) diff --git a/crates/asap_types/src/grouping_projection.rs b/crates/asap_types/src/grouping_projection.rs index 0b34cb894..561b29f82 100644 --- a/crates/asap_types/src/grouping_projection.rs +++ b/crates/asap_types/src/grouping_projection.rs @@ -174,8 +174,31 @@ mod identity_tests { assert_ne!(legacy.id, numeric.id); assert_ne!(legacy.id, nullable.id); assert_ne!(numeric.id, nullable.id); - numeric.validate().unwrap(); - nullable.validate().unwrap(); + 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::::new(), + "table.samples.v1", + ) + .with_grouping_projection(grouping); + table.validate().unwrap(); + } } } diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index a3017d16d..7f5128631 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -389,6 +389,14 @@ impl PrecomputePlan { } let mut materializations = BTreeSet::new(); for materialization in &self.materializations { + if materialization.table_name.is_none() + && !materialization.grouping_labels.is_legacy_labels() + { + return Err(PrecomputePlanError::CatalogContract( + "time-series grouping requires non-null string labels".into(), + )); + } + materialization .grouping_labels .validate() diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index 13a801817..d61ba2f7f 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -888,6 +888,14 @@ impl DataDescriptor { pub fn validate(&self) -> Result<(), SdsError> { self.value_projection.validate().map_err(SdsError)?; self.group_by_keys.validate().map_err(SdsError)?; + if matches!(self.source, DataSourceIdentity::TimeSeries { .. }) + && !self.group_by_keys.is_legacy_labels() + { + return Err(SdsError( + "time-series grouping requires non-null string labels".into(), + )); + } + if matches!(self.source, DataSourceIdentity::Table { .. }) { for column in self.group_by_keys.columns() { crate::table_population::validate_column_name(&column.name).map_err(SdsError)?; From 5685e60a1da818a74d9b40fe1d5e804d1e2def4b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 20:54:45 -0600 Subject: [PATCH 3/3] style: format typed grouping consumers --- data_plane/src/drivers/ingest/otel.rs | 3 ++- .../drivers/ingest/prometheus_remote_write.rs | 6 +++-- .../src/precompute_engine/ingest_handler.rs | 22 +++++++++---------- .../src/precompute_engine/output_sink.rs | 6 +++-- .../sketch_db/backfill/processor.rs | 3 ++- .../storage_engines/sketch_db/index/mod.rs | 3 +-- 6 files changed, 23 insertions(+), 20 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 2dcad0930..c5a685af8 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -606,7 +606,8 @@ fn resolve_bucket_sid_for_agg_config( point_labels: &HashMap, ) -> (u64, asap_types::PolicyFingerprint) { let grouping_pairs: Vec<(&str, &str)> = config - .grouping_labels.iter() + .grouping_labels + .iter() .map(|name| { let v = point_labels.get(name).map(|s| s.as_str()).unwrap_or(""); (name.as_str(), v) diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 985079ac5..5593e66aa 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -363,7 +363,8 @@ impl PrometheusRemoteWriteReceiver { let mut labels = group_key.as_population_labels(); if labels.is_empty() { labels = config - .grouping_labels.iter() + .grouping_labels + .iter() .cloned() .zip(group_key.values().labels) .collect(); @@ -667,7 +668,8 @@ fn route_messages( Vec::new() } else { config - .grouping_labels.iter() + .grouping_labels + .iter() .map(|name| { ( name.as_str(), diff --git a/data_plane/src/precompute_engine/ingest_handler.rs b/data_plane/src/precompute_engine/ingest_handler.rs index 00958f313..e50323b37 100644 --- a/data_plane/src/precompute_engine/ingest_handler.rs +++ b/data_plane/src/precompute_engine/ingest_handler.rs @@ -278,14 +278,14 @@ impl IngestState { labels: &std::collections::HashMap, config: &AggregationConfig, ) -> Arc { - crate::precompute_engine::group_key::intern_pairs( - config.grouping_labels.iter().map(|name| { + crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.iter().map( + |name| { ( name.as_str(), labels.get(name).map(String::as_str).unwrap_or(""), ) - }), - ) + }, + )) } } @@ -296,14 +296,12 @@ fn extract_group_key( config: &AggregationConfig, ) -> Arc { let labels = parse_labels_from_series_key(series_key); - crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.iter().map( - |name| { - ( - name.as_str(), - labels.get(name.as_str()).copied().unwrap_or(""), - ) - }, - )) + crate::precompute_engine::group_key::intern_pairs(config.grouping_labels.iter().map(|name| { + ( + name.as_str(), + labels.get(name.as_str()).copied().unwrap_or(""), + ) + })) } #[cfg(test)] diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 8bbef5316..b7412abf4 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -174,7 +174,8 @@ impl SketchStoreSink { if let Some(revision) = &output.input_revision { let group_values = output.population_labels.clone().unwrap_or_else(|| { agg_cfg - .grouping_labels.iter() + .grouping_labels + .iter() .cloned() .zip(output.key.clone().unwrap_or_default().labels) .collect() @@ -370,7 +371,8 @@ mod tests { parameters: HashMap::new(), grouping_labels: KeyByLabelNames::new( grouping_keys.iter().map(|s| s.to_string()).collect(), - ).into(), + ) + .into(), aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs index 70b6d7c4f..21fc95ffe 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs @@ -127,7 +127,8 @@ fn resolve_backfill_bucket_sid( ) -> u64 { let labels = parse_labels_from_series_key(series_key); let grouping_pairs: Vec<(&str, &str)> = config - .grouping_labels.iter() + .grouping_labels + .iter() .map(|name| (name.as_str(), *labels.get(name.as_str()).unwrap_or(&""))) .collect(); let attrs_fp = canonical_attrs_fingerprint(&grouping_pairs); diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 930ca6a9a..4a0959a09 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -2666,8 +2666,7 @@ impl SketchStore { let target_metric = agg_cfg.metric.as_str(); let target_agg_type = agg_cfg.aggregation_type; let target_params = canonical_parameters(&agg_cfg.parameters); - let target_group_keys: BTreeSet = - agg_cfg.grouping_labels.iter().cloned().collect(); + let target_group_keys: BTreeSet = agg_cfg.grouping_labels.iter().cloned().collect(); // Collect the matching sids under a short read lock; then call // `remove_instance` per sid (which takes its own write lock).