From d3c6cfe1209c3afba634ececa026d816b8b2b445 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 19:11:13 -0600 Subject: [PATCH 1/2] refactor(sds): share typed value projections with precompute plans --- control_plane/src/clickhouse.rs | 5 +- control_plane/src/physical/compiler.rs | 11 +- crates/asap_types/src/aggregation_config.rs | 117 ++++++++++++++++-- crates/asap_types/src/policy_fingerprint.rs | 13 +- crates/asap_types/src/precompute_plan.rs | 22 ++-- .../asap_types/src/precompute_plan/catalog.rs | 23 +--- crates/asap_types/src/sds.rs | 65 +++++++++- crates/asap_types/src/summary_catalog.rs | 18 +-- .../drivers/ingest/prometheus_remote_write.rs | 4 +- data_plane/src/drivers/query/servers/http.rs | 2 +- .../src/precompute_engine/output_sink.rs | 2 +- .../accelerator.rs | 4 +- .../sketch_db/backfill/clickhouse_reader.rs | 56 ++++++++- .../sketch_db/lifecycle/eviction.rs | 2 +- .../tests/test_utilities/engine_factories.rs | 16 +-- .../summary-catalog-sds-architecture.md | 12 ++ 16 files changed, 299 insertions(+), 73 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 7cd49accb..a89dee86b 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -248,7 +248,8 @@ fn materialize_selected_sql( ) .map_err(|error| error.to_string())?; config.table_name = Some(table); - config.value_column = Some(value); + config.value_projection = + Some(asap_types::sds::ValueProjectionIdentity::Column { name: value }); config.table_timestamp_column = Some(timestamp); config.table_population = Some(population); config.partitioning = Some(asap_types::sds::PopulationPartitioning::Grouped); @@ -632,7 +633,7 @@ fn select_materialization<'a>( ) -> Result<&'a asap_types::PrecomputeMaterialization, crate::query_plan::QueryPlanError> { let mut matches = materializations.iter().filter(|candidate| { candidate.table_name.as_deref() == Some(table_ref) - && candidate.value_column.as_deref() == Some(value_column) + && candidate.effective_value_projection().column() == Some(value_column) && candidate.population_filter_canonical().ok().as_deref() == Some(spatial_filter) && candidate .accumulator_spec() diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index d37872e8c..e0f7f5806 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -5141,6 +5141,13 @@ mod tests { let roundtrip: PrecomputePlan = serde_json::from_slice(&serde_json::to_vec(original).unwrap()).unwrap(); roundtrip.validate_against_catalog(catalog).unwrap(); + let mut legacy = serde_json::to_value(original).unwrap(); + for schema in legacy["schemas"].as_array_mut().unwrap() { + schema.as_object_mut().unwrap().remove("value_projection"); + schema["value_column"] = serde_json::json!("SampleValue"); + } + let decoded: PrecomputePlan = serde_json::from_value(legacy).unwrap(); + decoded.validate_against_catalog(catalog).unwrap(); let reject = |mutated: PrecomputePlan| assert!(mutated.validate_against_catalog(catalog).is_err()); let mut bad = original.clone(); @@ -5149,7 +5156,9 @@ mod tests { }; reject(bad); let mut bad = original.clone(); - bad.schemas[0].value_column = planner_types::pre_asap::ColumnRef::Named("other".into()); + bad.schemas[0].value_projection = asap_types::sds::ValueProjectionIdentity::Column { + name: "other".into(), + }; reject(bad); let mut bad = original.clone(); bad.schemas[0].group_by.push("other".into()); diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index e2411a565..f53e25713 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -126,8 +126,15 @@ pub struct PrecomputeMaterialization { pub num_aggregates_to_retain: Option, // SQL-specific fields (optional, used when query_language=sql) - pub table_name: Option, // SQL mode: table name - pub value_column: Option, // SQL mode: which value column to aggregate + pub table_name: Option, // SQL mode: table name + #[serde( + default, + alias = "value_column", + alias = "valueColumn", + alias = "valueProjection", + deserialize_with = "crate::sds::deserialize_optional_value_projection" + )] + pub value_projection: Option, /// Table timestamp projection, in Unix milliseconds. #[serde( default, @@ -177,7 +184,17 @@ impl AggregationIdInfo { pub type AggregationConfig = PrecomputeMaterialization; impl PrecomputeMaterialization { + pub fn effective_value_projection(&self) -> &crate::sds::ValueProjectionIdentity { + self.value_projection + .as_ref() + .unwrap_or(&crate::sds::ValueProjectionIdentity::SampleValue) + } + pub fn population_filter_canonical(&self) -> Result { + self.effective_value_projection().validate()?; + if self.value_projection.is_some() && self.table_name.is_none() { + return Err("explicit table value projection requires a table source".into()); + } if let Some(column) = &self.table_timestamp_column { if self.table_name.is_none() || column.is_empty() { return Err("table timestamp projection requires a table and a column".into()); @@ -247,7 +264,8 @@ impl PrecomputeMaterialization { metric, num_aggregates_to_retain, table_name, - value_column, + value_projection: value_column + .map(|name| crate::sds::ValueProjectionIdentity::Column { name }), table_population: None, table_timestamp_column: None, } @@ -363,6 +381,13 @@ impl PrecomputeMaterialization { .filter(|value| !value.is_null()) .map(|value| serde_json::from_value(value.clone())) .transpose()?; + if let Some(projection) = data + .get("valueProjection") + .or_else(|| data.get("value_projection")) + .filter(|value| !value.is_null()) + { + config.value_projection = Some(serde_json::from_value(projection.clone())?); + } config.pane_origin_ms = pane_origin_ms; config.table_timestamp_column = data .get("tableTimestampColumn") @@ -475,6 +500,12 @@ impl PrecomputeMaterialization { .unwrap_or("") .to_string(); + let typed_projection: Option = aggregation_data + .get("valueProjection") + .or_else(|| aggregation_data.get("value_projection")) + .filter(|value| !value.is_null()) + .map(|value| serde_json::to_value(value).and_then(serde_json::from_value)) + .transpose()?; let (metric, table_name, value_column) = match query_language { QueryLanguage::PromQl | QueryLanguage::MetricsQl => { let metric = aggregation_data["metric"] @@ -492,9 +523,22 @@ impl PrecomputeMaterialization { .to_string(); let column = aggregation_data["valueColumn"] .as_str() - .ok_or_else(|| anyhow::anyhow!("Missing valueColumn for ClickHouse SQL"))? - .to_string(); - (format!("{table}.{column}"), Some(table), Some(column)) + .or_else(|| { + typed_projection + .as_ref() + .and_then(|projection| projection.column()) + }) + .map(str::to_owned); + if column.is_none() && typed_projection.is_none() { + return Err(anyhow::anyhow!( + "Missing value projection for ClickHouse SQL" + )); + } + ( + format!("{table}.{}", column.as_deref().unwrap_or("constant")), + Some(table), + column, + ) } }; @@ -520,6 +564,9 @@ impl PrecomputeMaterialization { .filter(|value| !value.is_null()) .map(|value| serde_yaml::from_value(value.clone())) .transpose()?; + if let Some(projection) = typed_projection { + config.value_projection = Some(projection); + } config.pane_origin_ms = pane_origin_ms; config.table_timestamp_column = aggregation_data .get("tableTimestampColumn") @@ -569,8 +616,8 @@ impl SerializableToSink for PrecomputeMaterialization { if let Some(ref table_name) = self.table_name { json["tableName"] = serde_json::json!(table_name); } - if let Some(ref value_column) = self.value_column { - json["valueColumn"] = serde_json::json!(value_column); + if let Some(ref projection) = self.value_projection { + json["valueProjection"] = serde_json::json!(projection); } if let Some(ref population) = self.table_population { json["tablePopulation"] = serde_json::json!(population); @@ -721,4 +768,58 @@ mod tests { "PR 5: aggregationId must not appear on the wire — readers derive it from content" ); } + + #[test] + fn typed_projection_roundtrips_and_legacy_column_keeps_identity() { + use crate::sds::ValueProjectionIdentity; + use planner_types::pre_asap::ScalarValue; + let mut config = + AggregationConfig::from_yaml_data(&sample_yaml(false), None, QueryLanguage::PromQl) + .unwrap(); + config.table_name = Some("telemetry".into()); + config.value_projection = Some(ValueProjectionIdentity::Column { + name: "value".into(), + }); + let column_identity = config.policy_fingerprint(); + let mut legacy = serde_json::to_value(&config).unwrap(); + legacy.as_object_mut().unwrap().remove("value_projection"); + legacy["value_column"] = serde_json::json!("value"); + let decoded: AggregationConfig = serde_json::from_value(legacy).unwrap(); + assert_eq!(decoded.policy_fingerprint(), column_identity); + config.value_projection = Some(ValueProjectionIdentity::Constant { + value: ScalarValue::Int64(1), + }); + let mut wire = config.serialize_to_json(); + // The legacy JSON and YAML readers receive their labels from the + // enclosing streaming config, in their respective wire shapes. + wire["groupingLabels"] = config.grouping_labels.serialize_to_json(); + wire["aggregatedLabels"] = config.aggregated_labels.serialize_to_json(); + wire["rollupLabels"] = config.rollup_labels.serialize_to_json(); + wire["labels"] = serde_json::json!({ + "grouping": config.grouping_labels.serialize_to_json(), + "aggregated": config.aggregated_labels.serialize_to_json(), + "rollup": config.rollup_labels.serialize_to_json(), + }); + assert!(wire.get("valueColumn").is_none()); + let json = AggregationConfig::deserialize_from_json(&wire).unwrap(); + let yaml = AggregationConfig::from_yaml_data( + &serde_yaml::to_value(wire).unwrap(), + None, + QueryLanguage::ClickHouseSql, + ) + .unwrap(); + assert_eq!( + json.effective_value_projection(), + config.effective_value_projection() + ); + assert_eq!( + yaml.effective_value_projection(), + config.effective_value_projection() + ); + assert_ne!(config.policy_fingerprint(), column_identity); + config.value_projection = Some(ValueProjectionIdentity::Constant { + value: ScalarValue::Float64(f64::NAN), + }); + assert!(config.population_filter_canonical().is_err()); + } } diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 44d50669e..3daae2f94 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -180,9 +180,20 @@ impl PolicyFingerprint { buf.extend_from_slice(b"\0sql-source-v1\0"); buf.extend_from_slice(table.as_bytes()); buf.push(0); - if let Some(column) = &cfg.value_column { + if let Some(column) = cfg.effective_value_projection().column() { buf.extend_from_slice(column.as_bytes()); } + if matches!( + cfg.effective_value_projection(), + crate::sds::ValueProjectionIdentity::Constant { .. } + ) { + buf.extend_from_slice(b"\0constant-projection-v1\0"); + buf.extend_from_slice( + serde_json::to_string(cfg.effective_value_projection()) + .expect("finite validated projection serializes") + .as_bytes(), + ); + } } if let Some(population) = &cfg.table_population { let canonical = population.canonical(); diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 8cff067df..7cff5de13 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -139,7 +139,11 @@ pub struct StateSchemaContract { pub materialization: crate::sds::SummaryDefinitionId, pub family: StateFamilyContract, pub source: Source, - pub value_column: planner_types::pre_asap::ColumnRef, + #[serde( + alias = "value_column", + deserialize_with = "crate::sds::deserialize_state_value_projection" + )] + pub value_projection: crate::sds::ValueProjectionIdentity, pub group_by: Vec, pub window: StateWindowContract, pub encodings: Vec, @@ -222,18 +226,14 @@ impl PrecomputePlan { table_ref: table_ref.clone(), }, ); - let value_column = materialization - .value_column - .clone() - .map(planner_types::pre_asap::ColumnRef::Named) - .unwrap_or(planner_types::pre_asap::ColumnRef::SampleValue); + let value_projection = materialization.effective_value_projection().clone(); Ok(StateSchemaContract { schema_id: state_schema_id(fingerprint), schema_version: 1, materialization: fingerprint.into(), family, source, - value_column, + value_projection, group_by: materialization.grouping_labels.labels.clone(), window: StateWindowContract { kind: materialization.window_type, @@ -426,15 +426,11 @@ impl PrecomputePlan { table_ref: table_ref.clone(), }, ); - let value_column = materialization - .value_column - .clone() - .map(planner_types::pre_asap::ColumnRef::Named) - .unwrap_or(planner_types::pre_asap::ColumnRef::SampleValue); + let value_projection = materialization.effective_value_projection().clone(); if schema.schema_id != state_schema_id(schema.materialization.fingerprint()) || schema.family != family || schema.source != source - || schema.value_column != value_column + || schema.value_projection != value_projection || schema.group_by != materialization.grouping_labels.labels || schema.window.kind != materialization.window_type || schema.window.size_ms != materialization.window_size.saturating_mul(1_000) diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index 26c5e1e9a..80d682ab4 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -1,10 +1,8 @@ //! Catalog consistency checks for the precompute execution plan. use super::*; -use crate::sds::{ - DataSourceIdentity, SummaryDefinitionId, SummaryDescriptor, ValueProjectionIdentity, -}; +use crate::sds::{DataSourceIdentity, SummaryDefinitionId, SummaryDescriptor}; use crate::summary_catalog::SummaryCatalog; -use planner_types::pre_asap::{ColumnRef, Source}; +use planner_types::pre_asap::Source; use std::collections::BTreeSet; fn invalid(reason: impl Into) -> PrecomputePlanError { PrecomputePlanError::CatalogContract(reason.into()) @@ -80,16 +78,11 @@ impl PrecomputePlan { table_ref: table_ref.clone(), }, ); - let expected_projection = config - .value_column - .as_ref() - .map_or(ValueProjectionIdentity::SampleValue, |name| { - ValueProjectionIdentity::Column { name: name.clone() } - }); + let expected_projection = config.effective_value_projection(); if data.partitioning != config.partitioning || data.timestamp_column != config.table_timestamp_column || data.source != expected_source - || data.value_projection != expected_projection + || &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() @@ -121,11 +114,7 @@ impl PrecomputePlan { metric: config.metric.clone(), } }; - let col = if let Some(column) = &config.value_column { - ColumnRef::Named(column.clone()) - } else { - ColumnRef::SampleValue - }; + let projection = config.effective_value_projection(); let size = config .window_size .checked_mul(1000) @@ -146,7 +135,7 @@ impl PrecomputePlan { if schema.schema_id != state_schema_id(id.fingerprint()) || schema.family != expected_family || schema.source != source - || schema.value_column != col + || &schema.value_projection != projection || schema.group_by != config.grouping_labels.labels || schema.window.kind != config.window_type || schema.window.size_ms != size diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index 1a137e360..b838c9c1a 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -690,12 +690,70 @@ pub enum DataSourceIdentity { Table { table_ref: String }, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] #[serde(rename_all = "snake_case")] pub enum ValueProjectionIdentity { SampleValue, - Column { name: String }, + Column { + name: String, + }, + Constant { + value: planner_types::pre_asap::ScalarValue, + }, +} + +impl ValueProjectionIdentity { + pub fn column(&self) -> Option<&str> { + match self { + Self::Column { name } => Some(name), + _ => None, + } + } + + pub fn validate(&self) -> Result<(), String> { + use planner_types::pre_asap::ScalarValue; + match self { + Self::Column { name } => crate::table_population::validate_column_name(name), + Self::Constant { + value: ScalarValue::Int64(_), + } + | Self::SampleValue => Ok(()), + Self::Constant { + value: ScalarValue::Float64(value), + } if value.is_finite() => Ok(()), + Self::Constant { .. } => { + Err("summary value projection requires a finite numeric literal".into()) + } + } + } +} + +/// Compatibility adapter for old config column strings; storage is always typed. +pub(crate) fn deserialize_optional_value_projection<'de, D: serde::Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + let value = Option::::deserialize(deserializer)?; + value + .map(|value| match value { + Value::String(name) => Ok(ValueProjectionIdentity::Column { name }), + value => serde_json::from_value(value).map_err(serde::de::Error::custom), + }) + .transpose() +} + +/// Read legacy StateSchema ColumnRef values without retaining a parallel field. +pub(crate) fn deserialize_state_value_projection<'de, D: serde::Deserializer<'de>>( + deserializer: D, +) -> Result { + let value = Value::deserialize(deserializer)?; + if value == "SampleValue" { + return Ok(ValueProjectionIdentity::SampleValue); + } + if let Some(name) = value.get("Named").and_then(Value::as_str) { + return Ok(ValueProjectionIdentity::Column { name: name.into() }); + } + serde_json::from_value(value).map_err(serde::de::Error::custom) } /// Whether a materialization preserves source entities or pools a population. @@ -707,7 +765,7 @@ pub enum PopulationPartitioning { Grouped, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct DataDescriptor { pub id: DataDescriptorId, @@ -815,6 +873,7 @@ impl DataDescriptor { &self.id } pub fn validate(&self) -> Result<(), SdsError> { + self.value_projection.validate().map_err(SdsError)?; if let Some(column) = &self.timestamp_column { if !matches!(self.source, DataSourceIdentity::Table { .. }) || column.is_empty() { return Err(SdsError( diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 4febef66b..42ce40c75 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -7,7 +7,7 @@ use std::collections::BTreeMap; use crate::sds::{ CatalogGeneration, DataDescriptor, DataDescriptorId, DataSourceIdentity, SummaryDefinitionId, - SummaryDescriptor, SummaryDescriptorId, ValueProjectionIdentity, + SummaryDescriptor, SummaryDescriptorId, }; use crate::PolicyFingerprint; use crate::WindowMaterializationLayout; @@ -111,12 +111,7 @@ impl SummaryCatalog { table_ref: table_ref.clone(), }, ); - let value_projection = config - .value_column - .as_ref() - .map_or(ValueProjectionIdentity::SampleValue, |name| { - ValueProjectionIdentity::Column { name: name.clone() } - }); + let value_projection = config.effective_value_projection().clone(); let data = DataDescriptor::new_typed( source, value_projection, @@ -260,6 +255,7 @@ impl SummaryCatalog { #[cfg(test)] mod tests { use super::*; + use crate::sds::ValueProjectionIdentity; use crate::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; fn config(metric: &str, filter: &str, window: u64) -> PrecomputeMaterialization { @@ -289,7 +285,9 @@ mod tests { use planner_types::pre_asap::{CompareOpKind, ScalarValue}; let mut requests = config("raw_samples.value", "", 60); requests.table_name = Some("raw_samples".into()); - requests.value_column = Some("value".into()); + requests.value_projection = Some(ValueProjectionIdentity::Column { + name: "value".into(), + }); requests.table_population = Some(TablePopulation { predicates: vec![TableColumnPredicate { column: "metric".into(), @@ -308,7 +306,9 @@ mod tests { other_table.policy_fingerprint() ); let mut other_value = requests.clone(); - other_value.value_column = Some("other_value".into()); + other_value.value_projection = Some(ValueProjectionIdentity::Column { + name: "other_value".into(), + }); assert_ne!( requests.policy_fingerprint(), other_value.policy_fingerprint() diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index f3da96a76..4f29103db 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -857,7 +857,7 @@ mod tests { metric: "requests_total".into(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -913,7 +913,7 @@ mod tests { metric: "cpu_seconds_total".into(), num_aggregates_to_retain: Some(80), table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 897c25a1a..9715ce20b 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3672,7 +3672,7 @@ aggregations: metric: metric.clone(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 768a08291..0fd760c72 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -343,7 +343,7 @@ mod tests { metric: metric.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index a4a023a4f..4cc33dd64 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -759,7 +759,9 @@ mod tests { cfg.pane_origin_ms = Some(0); cfg.table_name = Some("asap_e2e.samples".into()); cfg.table_timestamp_column = Some("timestamp_ms".into()); - cfg.value_column = Some("value".into()); + cfg.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Column { + name: "value".into(), + }); let hot = crate::storage_engines::types::HotReloadStreamingConfig::from_arc(Arc::new( crate::storage_engines::types::StreamingConfig::new(HashMap::from([( cfg.policy_fp_u64(), 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 8663988bf..33190ad3a 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 @@ -49,6 +49,7 @@ pub struct ClickHouseReader { config: ClickHouseReaderConfig, http: reqwest::Client, population: Option, + value_projection: Option, output_metric: Option, } @@ -73,6 +74,7 @@ impl ClickHouseReader { config, http: reqwest::Client::new(), population: None, + value_projection: None, output_metric: None, }) } @@ -116,6 +118,15 @@ impl ClickHouseReader { .join(" AND ") }, ); + let value = match &self.value_projection { + Some(asap_types::sds::ValueProjectionIdentity::Constant { + value: planner_types::pre_asap::ScalarValue::Int64(_), + }) => "{projected_value:Int64}", + Some(asap_types::sds::ValueProjectionIdentity::Constant { + value: planner_types::pre_asap::ScalarValue::Float64(_), + }) => "{projected_value:Float64}", + _ => c.value_column.as_str(), + }; format!( "SELECT {labels} AS labels, {timestamp} AS timestamp_ms, {value} AS value \ FROM {database}.{table} WHERE {population} \ @@ -123,7 +134,7 @@ impl ClickHouseReader { ORDER BY labels, timestamp_ms FORMAT JSONEachRow", labels = c.labels_column, timestamp = c.timestamp_ms_column, - value = c.value_column, + value = value, database = c.database, table = c.table, ) @@ -149,13 +160,24 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor .table_timestamp_column .clone() .ok_or("table materialization has no timestamp projection")?; - source_config.value_column = materialization - .value_column - .clone() - .ok_or("table materialization has no value projection")?; + match materialization.effective_value_projection() { + asap_types::sds::ValueProjectionIdentity::Column { name } => { + source_config.value_column = name.clone() + } + asap_types::sds::ValueProjectionIdentity::Constant { + value: planner_types::pre_asap::ScalarValue::Int64(value), + } if value.unsigned_abs() > (1_u64 << 53) => { + return Err("integer projection exceeds exact Float64 ingest range".into()) + } + asap_types::sds::ValueProjectionIdentity::Constant { .. } => {} + asap_types::sds::ValueProjectionIdentity::SampleValue => { + return Err("table materialization has no explicit value projection".into()) + } + } materialization.population_filter_canonical()?; let mut reader = ClickHouseReader::new(source_config)?; reader.population = Some(materialization.table_population.clone().unwrap_or_default()); + reader.value_projection = Some(materialization.effective_value_projection().clone()); reader.output_metric = Some(materialization.metric.clone()); Ok(Arc::new(reader) as Arc) } @@ -183,6 +205,20 @@ impl RawSampleReader for ClickHouseReader { ("param_start_ms", start_ms.as_str()), ("param_end_ms", end_ms.as_str()), ]); + if let Some(asap_types::sds::ValueProjectionIdentity::Constant { value }) = + &self.value_projection + { + let value = match value { + planner_types::pre_asap::ScalarValue::Int64(value) => value.to_string(), + planner_types::pre_asap::ScalarValue::Float64(value) => value.to_string(), + _ => { + return Err(RawSampleReaderError::Other { + reason: "unsupported constant projection".into(), + }) + } + }; + request = request.query(&[("param_projected_value", value)]); + } if let Some(population) = &self.population { for (index, predicate) in population.predicates.iter().enumerate() { use planner_types::pre_asap::ScalarValue; @@ -319,6 +355,16 @@ mod tests { assert!(!sql.contains("{metric:String}")); } + #[test] + fn constant_projection_uses_a_typed_parameter_without_a_fake_column() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.value_projection = Some(asap_types::sds::ValueProjectionIdentity::Constant { + value: planner_types::pre_asap::ScalarValue::Int64(1), + }); + assert!(reader.sql().contains("{projected_value:Int64} AS value")); + assert!(!reader.sql().contains(" value AS value")); + } + #[test] fn typed_source_enters_clickhouse_backfill_lifecycle() { let mut materialization = asap_types::PrecomputeMaterialization::new( 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 557b8326a..7cdaf6e92 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -274,7 +274,7 @@ mod tests { metric: format!("metric_{id}"), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs index 6f742df94..12af96e44 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -107,7 +107,7 @@ pub fn create_engine_single_pop_with_aggregated( metric: metric.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -204,7 +204,7 @@ pub fn create_engine_dual_input( metric: metric.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -231,7 +231,7 @@ pub fn create_engine_dual_input( metric: metric.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -323,7 +323,7 @@ pub fn create_engine_two_metrics( metric: metric_a.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -349,7 +349,7 @@ pub fn create_engine_two_metrics( metric: metric_b.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -451,7 +451,7 @@ pub fn create_engine_three_metrics( metric: metric.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -530,7 +530,7 @@ pub fn create_engine_multi_timestamp( metric: metric.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, @@ -601,7 +601,7 @@ pub fn create_engine_multi_timestamp_with_window( metric: metric.to_string(), num_aggregates_to_retain: None, table_name: None, - value_column: None, + value_projection: None, table_population: None, table_timestamp_column: None, partitioning: None, diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 83b8d829e..82fd1c4a8 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -192,6 +192,18 @@ 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. +`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 +not masquerade as a table column. Projection identity participates in catalog +and policy identity; existing column identities remain unchanged. Older +`value_column` config and state-schema fields are accepted only by wire adapters +and become the same typed projection in memory. ClickHouse backfill binds a +constant as a typed query parameter and applies the installed table population +and timestamp projection. Its Float64 ingest boundary rejects integer constants +outside the exactly representable range. This contract enables literal inputs; +query lowering must still establish each aggregate's null and row semantics. + The durable `sid_metadata.json` format is versioned independently. Version 2 contains `summary_descriptors`, `data_descriptors`, and `bindings` tables. A binding stores only both descriptor IDs plus SID-local timestamps. Version-1 From 15391b67db313634d47e04b9978452d1c8d5f204 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 19:19:40 -0600 Subject: [PATCH 2/2] fix(sds): reject duplicate projection wire fields --- crates/asap_types/src/aggregation_config.rs | 43 ++++++++++++++++++++- 1 file changed, 42 insertions(+), 1 deletion(-) diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index f53e25713..b9847d766 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -294,6 +294,19 @@ impl PrecomputeMaterialization { pub fn deserialize_from_json( data: &Value, ) -> Result> { + if [ + "valueColumn", + "value_column", + "valueProjection", + "value_projection", + ] + .iter() + .filter(|key| data.get(**key).is_some_and(|value| !value.is_null())) + .count() + > 1 + { + return Err("multiple value projection fields are not allowed".into()); + } // `aggregationId` is silently ignored — identity is // content-addressed via PolicyFingerprint (PR 5). @@ -500,6 +513,25 @@ impl PrecomputeMaterialization { .unwrap_or("") .to_string(); + if [ + "valueColumn", + "value_column", + "valueProjection", + "value_projection", + ] + .iter() + .filter(|key| { + aggregation_data + .get(**key) + .is_some_and(|value| !value.is_null()) + }) + .count() + > 1 + { + return Err(anyhow::anyhow!( + "multiple value projection fields are not allowed" + )); + } let typed_projection: Option = aggregation_data .get("valueProjection") .or_else(|| aggregation_data.get("value_projection")) @@ -803,7 +835,7 @@ mod tests { assert!(wire.get("valueColumn").is_none()); let json = AggregationConfig::deserialize_from_json(&wire).unwrap(); let yaml = AggregationConfig::from_yaml_data( - &serde_yaml::to_value(wire).unwrap(), + &serde_yaml::to_value(&wire).unwrap(), None, QueryLanguage::ClickHouseSql, ) @@ -816,6 +848,15 @@ mod tests { yaml.effective_value_projection(), config.effective_value_projection() ); + let mut conflicting = wire; + conflicting["valueColumn"] = serde_json::json!("other_column"); + assert!(AggregationConfig::deserialize_from_json(&conflicting).is_err()); + assert!(AggregationConfig::from_yaml_data( + &serde_yaml::to_value(conflicting).unwrap(), + None, + QueryLanguage::ClickHouseSql + ) + .is_err()); assert_ne!(config.policy_fingerprint(), column_identity); config.value_projection = Some(ValueProjectionIdentity::Constant { value: ScalarValue::Float64(f64::NAN),