From c327e4eac5fd27fd623ebb6ae32ffa5a7ef600d7 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:48:33 -0600 Subject: [PATCH 01/14] feat: bind SQL populations through the shared catalog --- control_plane/src/clickhouse.rs | 97 +++++++++++++- crates/asap_types/src/aggregation_config.rs | 42 ++++++ crates/asap_types/src/lib.rs | 1 + crates/asap_types/src/policy_fingerprint.rs | 7 + .../asap_types/src/precompute_plan/catalog.rs | 6 +- crates/asap_types/src/summary_catalog.rs | 27 +++- crates/asap_types/src/table_population.rs | 126 ++++++++++++++++++ 7 files changed, 297 insertions(+), 9 deletions(-) create mode 100644 crates/asap_types/src/table_population.rs diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 34669ca8b..390387d90 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -215,7 +215,7 @@ fn bind_selected_node( request: &ClickHouseSqlWorkload, ) -> Result { let (table_ref, value_column, source_window, spatial_filter) = - clickhouse_materialization_leaf_contract(node) + clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms) .map_err(crate::query_plan::QueryPlanError::Invalid)?; let expected = crate::physical::compiler::physical_materialization_family(family); let selected = select_materialization( @@ -238,6 +238,8 @@ fn bind_selected_node( fn clickhouse_materialization_leaf_contract( node: &planner_types::post_asap::SummaryNode, + evaluation_start_ms: u64, + evaluation_end_ms: u64, ) -> Result<(String, String, Option, String), String> { use planner_types::{ post_asap::SummaryExpr, @@ -298,6 +300,7 @@ fn clickhouse_materialization_leaf_contract( let mut lower_ms = None; let mut upper_ms = None; let mut leaves = Vec::new(); + let mut population = asap_types::table_population::TablePopulation::default(); for predicate in predicates { comparisons(&predicate.0, &mut leaves); } @@ -324,7 +327,14 @@ fn clickhouse_materialization_leaf_contract( .time_index .is_some_and(|index| schema.columns[index].name == name) => { - lower_ms = Some(*value); + let bound = if matches!(op, CompareOpKind::Gt) { + value + .checked_add(1) + .ok_or("SQL exclusive lower timestamp overflows")? + } else { + *value + }; + lower_ms = Some(lower_ms.map_or(bound, |previous: i64| previous.max(bound))); } ( name, @@ -334,7 +344,27 @@ fn clickhouse_materialization_leaf_contract( .time_index .is_some_and(|index| schema.columns[index].name == name) => { - upper_ms = Some(*value); + let bound = if matches!(op, CompareOpKind::Le) { + value + .checked_add(1) + .ok_or("SQL inclusive upper timestamp overflows")? + } else { + *value + }; + upper_ms = Some(upper_ms.map_or(bound, |previous: i64| previous.min(bound))); + } + (_, _, QueryExpr::Literal(value)) + if !schema + .time_index + .is_some_and(|index| schema.columns[index].name == name) => + { + population + .predicates + .push(asap_types::table_population::TableColumnPredicate { + column: name.into(), + operator: op.clone(), + value: value.clone(), + }); } _ => { return Err(format!( @@ -343,6 +373,16 @@ fn clickhouse_materialization_leaf_contract( } } } + population.validate()?; + if let (Some(lower), Some(upper)) = (lower_ms, upper_ms) { + if u64::try_from(lower).ok() != Some(evaluation_start_ms) + || u64::try_from(upper).ok() != Some(evaluation_end_ms) + { + return Err( + "SQL source timestamp bounds differ from the fixed evaluation range".into(), + ); + } + } let inferred_window = match (lower_ms, upper_ms) { (Some(lower), Some(upper)) if upper > lower && (upper - lower) % 1_000 == 0 => { Some((upper - lower) as u64 / 1_000) @@ -359,7 +399,7 @@ fn clickhouse_materialization_leaf_contract( table_ref.to_owned(), value_column, Some(window_secs), - String::new(), + population.canonical(), )) } @@ -374,7 +414,7 @@ fn select_materialization<'a>( let mut matches = materializations.iter().filter(|candidate| { candidate.table_name.as_deref() == Some(table_ref) && candidate.value_column.as_deref() == Some(value_column) - && candidate.spatial_filter_normalized == spatial_filter + && candidate.population_filter_canonical().ok().as_deref() == Some(spatial_filter) && candidate .accumulator_spec() .ok() @@ -553,7 +593,7 @@ mod tests { vec![], ) }; - let request = ClickHouseSqlWorkload { + let mut request = ClickHouseSqlWorkload { sds, precompute_plan: precompute, transmission_plan: transmission, @@ -604,5 +644,50 @@ mod tests { Ok(planner_types::post_asap::ValueOperation::Project { .. }) ) })); + let original = request.queries[0].sql.clone(); + request.queries[0].sql = original.replace("timestamp_ms < 2000", "timestamp_ms <= 1999"); + assert!(compile_clickhouse_workload(&request).await.is_ok()); + request.queries[0].sql = original.replace("timestamp_ms < 2000", "timestamp_ms <= 2000"); + assert!(compile_clickhouse_workload(&request).await.is_err()); + request.queries[0].sql = original + .replace("timestamp_ms >= 0", "timestamp_ms >= 1000") + .replace("timestamp_ms < 2000", "timestamp_ms < 3000"); + assert!(compile_clickhouse_workload(&request).await.is_err()); + + request + .tables + .get_mut("telemetry") + .unwrap() + .columns + .push(Column::new("metric", DataType::Utf8, false)); + request.queries[0].sql = original.replace( + "WHERE timestamp_ms", + "WHERE metric = 'requests' AND timestamp_ms", + ); + assert!( + compile_clickhouse_workload(&request).await.is_err(), + "an unfiltered summary cannot satisfy a filtered query" + ); + let mut config = request.precompute_plan.materializations[0].clone(); + config.table_population = Some(asap_types::table_population::TablePopulation { + predicates: vec![asap_types::table_population::TableColumnPredicate { + column: "metric".into(), + operator: planner_types::pre_asap::CompareOpKind::Eq, + value: planner_types::pre_asap::ScalarValue::Utf8("requests".into()), + }], + }); + request.sds = SummaryCatalog::from_materializations(71, 1, &[config.clone()]).unwrap(); + let envelope = request.precompute_plan.envelope.clone(); + request.precompute_plan = + PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap(); + request.precompute_plan.summary_catalog = Some(request.sds.reference().unwrap()); + request.transmission_plan = TransmissionPlan::build( + envelope, + &request.precompute_plan, + &std::collections::BTreeMap::new(), + ) + .unwrap(); + request.transmission_plan.summary_catalog = Some(request.sds.reference().unwrap()); + assert!(compile_clickhouse_workload(&request).await.is_ok()); } } diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index 9f16833df..89be1899e 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -126,6 +126,12 @@ pub struct PrecomputeMaterialization { // 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 + #[serde( + default, + alias = "tablePopulation", + skip_serializing_if = "Option::is_none" + )] + pub table_population: Option, } /// Policy-match handles for both the key and value dimensions of a @@ -162,6 +168,20 @@ impl AggregationIdInfo { pub type AggregationConfig = PrecomputeMaterialization; impl PrecomputeMaterialization { + pub fn population_filter_canonical(&self) -> Result { + if let Some(population) = &self.table_population { + if self.table_name.is_none() || !self.spatial_filter.is_empty() { + return Err( + "typed table population requires a table and no PromQL label filter".into(), + ); + } + population.validate()?; + Ok(population.canonical()) + } else { + Ok(normalize_spatial_filter(&self.spatial_filter)) + } + } + #[allow(clippy::too_many_arguments)] pub fn new( aggregation_type: AggregationType, @@ -209,6 +229,7 @@ impl PrecomputeMaterialization { num_aggregates_to_retain, table_name, value_column, + table_population: None, } } @@ -318,6 +339,14 @@ impl PrecomputeMaterialization { value_column, ); config.pane_origin_ms = pane_origin_ms; + config.table_population = data + .get("tablePopulation") + .or_else(|| data.get("table_population")) + .filter(|value| !value.is_null()) + .cloned() + .map(serde_json::from_value) + .transpose()?; + config.population_filter_canonical()?; Ok(config) } @@ -457,6 +486,16 @@ impl PrecomputeMaterialization { value_column, ); config.pane_origin_ms = pane_origin_ms; + config.table_population = aggregation_data + .get("tablePopulation") + .or_else(|| aggregation_data.get("table_population")) + .filter(|value| !value.is_null()) + .cloned() + .map(serde_yaml::from_value) + .transpose()?; + config + .population_filter_canonical() + .map_err(anyhow::Error::msg)?; Ok(config) } } @@ -492,6 +531,9 @@ impl SerializableToSink for PrecomputeMaterialization { if let Some(ref value_column) = self.value_column { json["valueColumn"] = serde_json::json!(value_column); } + if let Some(ref population) = self.table_population { + json["tablePopulation"] = serde_json::json!(population); + } json } diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index ccfb16a20..ccf828726 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -13,6 +13,7 @@ pub mod routing_index; pub mod sds; pub mod storage_backend; pub mod summary_catalog; +pub mod table_population; pub mod traits; pub mod utils; diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index eea167c87..2f5d2e573 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -171,6 +171,13 @@ impl PolicyFingerprint { // 10. spatial_filter_normalized — canonicalized predicate buf.extend_from_slice(cfg.spatial_filter_normalized.as_bytes()); + if let Some(population) = &cfg.table_population { + let canonical = population.canonical(); + if !canonical.is_empty() { + buf.push(0); + buf.extend_from_slice(canonical.as_bytes()); + } + } Self(xxh64(&buf, 0)) } diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index 48fd1c411..55ba93e8b 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -89,12 +89,14 @@ impl PrecomputePlan { if data.source != expected_source || data.value_projection != expected_projection || data.population_filter_canonical - != crate::utils::normalize_spatial_filter(&config.spatial_filter) + != config.population_filter_canonical().map_err(invalid)? || data.group_by_keys != config.grouping_labels.labels.iter().cloned().collect() { return Err(invalid("source/population/grouping differs from catalog")); } - if config.spatial_filter_normalized != data.population_filter_canonical { + if config.spatial_filter_normalized + != crate::utils::normalize_spatial_filter(&config.spatial_filter) + { return Err(invalid("normalized population predicate drift")); } let schema = self diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 90095a330..e5474ef3f 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -120,7 +120,9 @@ impl SummaryCatalog { let data = DataDescriptor::new_typed( source, value_projection, - crate::utils::normalize_spatial_filter(&config.spatial_filter), + config + .population_filter_canonical() + .map_err(SummaryCatalogError::Descriptor)?, config.grouping_labels.labels.clone(), "asap.timestamped-observations.v2", ); @@ -279,6 +281,29 @@ mod tests { } // Content changes invalidate references even when plan/version are reused. + #[test] + fn table_populations_have_distinct_materialization_and_data_identities() { + use crate::table_population::{TableColumnPredicate, TablePopulation}; + 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.table_population = Some(TablePopulation { + predicates: vec![TableColumnPredicate { + column: "metric".into(), + operator: CompareOpKind::Eq, + value: ScalarValue::Utf8("requests".into()), + }], + }); + let mut errors = requests.clone(); + errors.table_population.as_mut().unwrap().predicates[0].value = + ScalarValue::Utf8("errors".into()); + assert_ne!(requests.policy_fingerprint(), errors.policy_fingerprint()); + let catalog = SummaryCatalog::from_materializations(1, 1, &[requests, errors]).unwrap(); + assert_eq!(catalog.data_descriptors.len(), 2); + assert_eq!(catalog.materializations.len(), 2); + } + #[test] fn snapshot_reference_is_deterministic_and_content_sensitive() { let a = config("requests", "", 60); diff --git a/crates/asap_types/src/table_population.rs b/crates/asap_types/src/table_population.rs new file mode 100644 index 000000000..63566fa78 --- /dev/null +++ b/crates/asap_types/src/table_population.rs @@ -0,0 +1,126 @@ +//! Typed table predicates shared by catalog identity and source readers. + +use planner_types::pre_asap::{CompareOpKind, ScalarValue}; +use serde::{Deserialize, Serialize}; + +/// A conjunction of column/literal comparisons. Column names are schema names, +/// not SQL fragments; readers must bind literal values as parameters. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct TablePopulation { + pub predicates: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct TableColumnPredicate { + pub column: String, + pub operator: CompareOpKind, + pub value: ScalarValue, +} + +impl TablePopulation { + pub fn validate(&self) -> Result<(), String> { + for predicate in &self.predicates { + if predicate.column.is_empty() + || !predicate + .column + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_') + { + return Err("table population column is not an unqualified identifier".into()); + } + if !matches!( + predicate.operator, + CompareOpKind::Eq + | CompareOpKind::Ne + | CompareOpKind::Lt + | CompareOpKind::Le + | CompareOpKind::Gt + | CompareOpKind::Ge + ) { + return Err("table population comparison is unsupported".into()); + } + if matches!(predicate.value, ScalarValue::Null) + || matches!(predicate.value, ScalarValue::Float64(value) if !value.is_finite()) + { + return Err("table population requires a finite non-null literal".into()); + } + } + Ok(()) + } + + /// Reordered or repeated conjuncts identify the same population. + pub fn canonical(&self) -> String { + if self.predicates.is_empty() { + return String::new(); + } + let mut predicates: Vec<_> = self + .predicates + .iter() + .map(|predicate| serde_json::to_string(predicate).expect("predicate serialization")) + .collect(); + predicates.sort(); + predicates.dedup(); + format!("sql.and.v1:[{}]", predicates.join(",")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn metric(value: &str) -> TableColumnPredicate { + TableColumnPredicate { + column: "metric".into(), + operator: CompareOpKind::Eq, + value: ScalarValue::Utf8(value.into()), + } + } + + #[test] + fn population_identity_distinguishes_values_and_ignores_conjunct_order() { + let a = metric("requests"); + let b = TableColumnPredicate { + column: "status".into(), + operator: CompareOpKind::Ge, + value: ScalarValue::Int64(500), + }; + assert_eq!( + TablePopulation { + predicates: vec![a.clone(), b.clone(), a.clone()] + } + .canonical(), + TablePopulation { + predicates: vec![b, a.clone()] + } + .canonical() + ); + assert_ne!( + TablePopulation { + predicates: vec![a] + } + .canonical(), + TablePopulation { + predicates: vec![metric("errors")] + } + .canonical() + ); + } + + #[test] + fn rejects_sql_fragments_and_nonfinite_literals() { + let mut predicate = metric("requests"); + predicate.column = "metric OR 1=1".into(); + assert!(TablePopulation { + predicates: vec![predicate] + } + .validate() + .is_err()); + let mut predicate = metric("requests"); + predicate.value = ScalarValue::Float64(f64::NAN); + assert!(TablePopulation { + predicates: vec![predicate] + } + .validate() + .is_err()); + } +} From 921d5a482e943758e366946c390ee268d8b0367d Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:47:24 -0600 Subject: [PATCH 02/14] feat(sds): identify entity-preserving population partitions --- crates/asap_types/src/aggregation_config.rs | 14 +++++ crates/asap_types/src/policy_fingerprint.rs | 4 ++ .../asap_types/src/precompute_plan/catalog.rs | 3 +- crates/asap_types/src/sds.rs | 56 +++++++++++++++++++ crates/asap_types/src/summary_catalog.rs | 3 +- 5 files changed, 78 insertions(+), 2 deletions(-) diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index 89be1899e..a7d6d166b 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -100,6 +100,8 @@ pub struct PrecomputeMaterialization { pub aggregation_sub_type: String, pub parameters: HashMap, pub grouping_labels: KeyByLabelNames, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub partitioning: Option, pub aggregated_labels: KeyByLabelNames, pub rollup_labels: KeyByLabelNames, pub original_yaml: String, @@ -209,6 +211,7 @@ impl PrecomputeMaterialization { aggregation_sub_type, parameters, grouping_labels, + partitioning: None, aggregated_labels, rollup_labels, original_yaml, @@ -338,6 +341,11 @@ impl PrecomputeMaterialization { table_name, value_column, ); + config.partitioning = data + .get("partitioning") + .filter(|value| !value.is_null()) + .map(|value| serde_json::from_value(value.clone())) + .transpose()?; config.pane_origin_ms = pane_origin_ms; config.table_population = data .get("tablePopulation") @@ -485,6 +493,11 @@ impl PrecomputeMaterialization { table_name, value_column, ); + config.partitioning = aggregation_data + .get("partitioning") + .filter(|value| !value.is_null()) + .map(|value| serde_yaml::from_value(value.clone())) + .transpose()?; config.pane_origin_ms = pane_origin_ms; config.table_population = aggregation_data .get("tablePopulation") @@ -508,6 +521,7 @@ impl SerializableToSink for PrecomputeMaterialization { "aggregationType": self.aggregation_type, "aggregationSubType": self.aggregation_sub_type, "parameters": self.parameters, + "partitioning": self.partitioning, "originalYaml": self.original_yaml, "windowSize": self.window_size, "slideInterval": self.slide_interval, diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 2f5d2e573..8475d463d 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -114,6 +114,10 @@ impl PolicyFingerprint { } buf.push(0); + if let Some(partitioning) = cfg.partitioning { + buf.extend_from_slice(format!("partition:{partitioning:?}\0").as_bytes()); + } + // 5. grouping_labels (already sorted at construction per // KeyByLabelNames invariant; encode as `,`-joined list) for l in &cfg.grouping_labels.labels { diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index 55ba93e8b..eb8a5408f 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -86,7 +86,8 @@ impl PrecomputePlan { .map_or(ValueProjectionIdentity::SampleValue, |name| { ValueProjectionIdentity::Column { name: name.clone() } }); - if data.source != expected_source + if data.partitioning != config.partitioning + || data.source != expected_source || data.value_projection != expected_projection || data.population_filter_canonical != config.population_filter_canonical().map_err(invalid)? diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index b54c50299..a85422bc9 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -698,6 +698,15 @@ pub enum ValueProjectionIdentity { Column { name: String }, } +/// Whether a materialization preserves source entities or pools a population. +/// Grouped label names remain in `DataDescriptor::group_by_keys`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PopulationPartitioning { + PerEntity, + Grouped, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct DataDescriptor { @@ -706,6 +715,8 @@ pub struct DataDescriptor { pub value_projection: ValueProjectionIdentity, pub population_filter_canonical: String, pub group_by_keys: BTreeSet, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub partitioning: Option, /// Versioned contract for timestamp interpretation and /// missing/duplicate/invalid observation handling. pub observation_semantics: String, @@ -765,6 +776,7 @@ impl DataDescriptor { &population_filter_canonical, &group_by_keys, &observation_semantics, + None, ); Self { id, @@ -772,9 +784,22 @@ impl DataDescriptor { value_projection, population_filter_canonical, group_by_keys, + partitioning: None, observation_semantics, } } + pub fn with_partitioning(mut self, partitioning: Option) -> Self { + self.partitioning = partitioning; + self.id = data_descriptor_id( + &self.source, + &self.value_projection, + &self.population_filter_canonical, + &self.group_by_keys, + &self.observation_semantics, + partitioning, + ); + self + } pub fn id(&self) -> &DataDescriptorId { &self.id } @@ -786,6 +811,7 @@ impl DataDescriptor { &self.population_filter_canonical, &self.group_by_keys, &self.observation_semantics, + self.partitioning, ) { return Err(SdsError("data descriptor ID/content mismatch".into())); @@ -799,6 +825,7 @@ fn data_descriptor_id( filter: &str, group_by: &BTreeSet, observation_semantics: &str, + partitioning: Option, ) -> DataDescriptorId { // Length framing keeps distinct typed sources, projections, predicates, // and grouping keys collision-free in the content identity. @@ -811,6 +838,9 @@ fn data_descriptor_id( projection.len(), filter.len() ); + if let Some(partitioning) = partitioning { + key.push_str(&format!("|partition:{partitioning:?}")); + } for name in group_by { key.push_str(&format!("|{}:{name}", name.len())); } @@ -1286,3 +1316,29 @@ mod tests { ); } } + +#[cfg(test)] +mod partition_identity_tests { + use super::*; + #[test] + fn entity_and_global_population_have_distinct_identity() { + let legacy = DataDescriptor::new_typed( + DataSourceIdentity::TimeSeries { metric: "m".into() }, + ValueProjectionIdentity::SampleValue, + "", + Vec::::new(), + "v1", + ); + let entity = legacy + .clone() + .with_partitioning(Some(PopulationPartitioning::PerEntity)); + let grouped = legacy + .clone() + .with_partitioning(Some(PopulationPartitioning::Grouped)); + assert_ne!(entity.id, grouped.id); + assert_ne!(entity.id, legacy.id); + assert_eq!(legacy.clone().with_partitioning(None).id, legacy.id); + entity.validate().unwrap(); + grouped.validate().unwrap(); + } +} diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index e5474ef3f..92eb58259 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -125,7 +125,8 @@ impl SummaryCatalog { .map_err(SummaryCatalogError::Descriptor)?, config.grouping_labels.labels.clone(), "asap.timestamped-observations.v2", - ); + ) + .with_partitioning(config.partitioning); Ok(( config.policy_fingerprint(), summary, From 5f7434ef1559643f233cea560c5fd78b6d72f086 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:21:32 -0600 Subject: [PATCH 03/14] fix: give ClickHouse backfill a typed table source --- data_plane/src/main.rs | 3 +-- .../accelerator.rs | 5 ++-- .../sketch_db/backfill/clickhouse_reader.rs | 24 ++++++++++++------- .../storage_engines/sketch_db/backfill/mod.rs | 7 ++++++ .../sketch_db/backfill/service.rs | 11 ++++----- .../tests/clickhouse_differential_e2e.rs | 2 +- .../tests/clickhouse_q05_process_e2e.rs | 2 +- .../query-engine/clickhouse-sql-support.md | 2 +- 8 files changed, 34 insertions(+), 22 deletions(-) diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 9a3ffee22..c751ddf9f 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -110,8 +110,7 @@ struct Args { #[arg(long, env = "ASAP_CLICKHOUSE_DATABASE", default_value = "default")] clickhouse_database: String, - /// Enable ClickHouse as a source for queued backfill jobs whose source URL - /// is `clickhouse://configured`. + /// Enable the configured ClickHouse connection for typed table backfill jobs. #[arg(long, env = "ASAP_CLICKHOUSE_BACKFILL_TABLE")] clickhouse_backfill_table: Option, #[arg( 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 f721742ad..3ab3a1b60 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 @@ -792,8 +792,9 @@ mod tests { let job = registry.create( cfg.policy_fp_u64(), (0, 2_000), - crate::storage_engines::sketch_db::backfill::BackfillSource::Prometheus { - url: "clickhouse://configured".into(), + crate::storage_engines::sketch_db::backfill::BackfillSource::ClickHouse { + database: "asap_e2e".into(), + table: "samples".into(), }, 2, ); 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 63354546d..5ed8784bc 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 @@ -82,15 +82,15 @@ impl ClickHouseReader { } } -/// Adds ClickHouse to the existing backfill lifecycle without extending the -/// shared `BackfillSource` enum. Jobs opt in with the reserved -/// `Prometheus { url: "clickhouse://configured" }` source marker; all other -/// sources retain the default factory behavior. +/// Resolve a typed table source using deployment-local connection settings. pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactory { let fallback = super::service::default_reader_factory(); Arc::new(move |source| match source { - BackfillSource::Prometheus { url } if url == "clickhouse://configured" => { - Ok(Arc::new(ClickHouseReader::new(config.clone())?) as Arc) + BackfillSource::ClickHouse { database, table } => { + let mut source_config = config.clone(); + source_config.database = database.clone(); + source_config.table = table.clone(); + Ok(Arc::new(ClickHouseReader::new(source_config)?) as Arc) } source => fallback(source), }) @@ -187,12 +187,18 @@ mod tests { } #[test] - fn configured_source_marker_enters_clickhouse_backfill_lifecycle() { + fn typed_source_enters_clickhouse_backfill_lifecycle() { let factory = clickhouse_reader_factory(config("samples")); - let reader = factory(&BackfillSource::Prometheus { - url: "clickhouse://configured".into(), + let reader = factory(&BackfillSource::ClickHouse { + database: "another_database".into(), + table: "another_table".into(), }) .unwrap(); assert_eq!(reader.source_name(), "ClickHouseReader"); + assert!(factory(&BackfillSource::ClickHouse { + database: "default".into(), + table: "samples; DROP TABLE x".into(), + }) + .is_err()); } } diff --git a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs index 0c3a6c69a..d8d15b012 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs @@ -70,6 +70,9 @@ pub enum BackfillSource { /// Prometheus (or VictoriaMetrics / Thanos / Cortex) via the /// HTTP range-query API. Prometheus { url: String }, + /// ClickHouse table read through the deployment's configured connection. + /// The source identity belongs to the job; credentials remain local. + ClickHouse { database: String, table: String }, /// Rebuild from a different sketch already in the store. Used for /// lossless schema widenings (e.g. CMS(256) → CMS(2048)) where /// the source sketch is a strict subset of the target's @@ -1120,6 +1123,10 @@ mod tests { #[test] fn backfill_source_roundtrips_through_serde() { for src in [ + BackfillSource::ClickHouse { + database: "telemetry".into(), + table: "samples".into(), + }, BackfillSource::Prometheus { url: "u".to_string(), }, diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs index ede452244..c22beeaa6 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -301,23 +301,22 @@ pub fn noop_reader_factory() -> ReaderFactory { /// * [`BackfillSource::Prometheus`] — routed to /// [`super::prometheus_reader::PrometheusReader`]. /// -/// All other variants (`S3Gorilla`, `OtherSketch`) return a clear +/// Other variants return a clear /// "not yet implemented" error, which the worker surfaces on /// `BackfillJob::error_message` so the control plane / operator sees /// exactly which reader is missing. pub fn default_reader_factory() -> ReaderFactory { - Arc::new(|source| { - match source { + Arc::new(|source| match source { BackfillSource::Prometheus { url } => { let reader = super::prometheus_reader::PrometheusReader::new(url.clone()); Ok(Arc::new(reader) as Arc) } - BackfillSource::S3Gorilla { .. } + BackfillSource::ClickHouse { .. } + | BackfillSource::S3Gorilla { .. } | BackfillSource::OtherSketch { .. } => Err(format!( - "reader for {source:?} not yet implemented; only Prometheus is wired in-tree as of Phase 5h" + "no reader configured for {source:?}; ClickHouse requires its deployment reader factory" ) .into()), - } }) } diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index 88b3fb24b..58520316f 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -317,7 +317,7 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() { "agg_id": config.policy_fp_u64(), "start_ms": 0, "end_ms": 2000, - "source": {"Prometheus": {"url": "clickhouse://configured"}}, + "source": {"ClickHouse": {"database": "default", "table": "telemetry"}}, "windows_total": 1 })) .send() diff --git a/data_plane/tests/clickhouse_q05_process_e2e.rs b/data_plane/tests/clickhouse_q05_process_e2e.rs index 937d75b05..35b0d04b7 100644 --- a/data_plane/tests/clickhouse_q05_process_e2e.rs +++ b/data_plane/tests/clickhouse_q05_process_e2e.rs @@ -271,7 +271,7 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { .post(format!("{base}/api/v1/db/backfill")) .json( &serde_json::json!({"agg_id":agg_id,"start_ms":start_ms,"end_ms":end_ms, - "source":{"Prometheus":{"url":"clickhouse://configured"}},"windows_total":1}), + "source":{"ClickHouse":{"database":"asap_q05_e2e","table":"q05_samples"}},"windows_total":1}), ) .send() .await diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 4bfd87bbf..c74f803dd 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -74,7 +74,7 @@ pane coverage, and execution errors fail closed to exact ClickHouse. The proxy preserves the upstream status, safe headers, and response body. ClickHouse can also provide samples to the queued backfill service through the -explicit `clickhouse://configured` source marker. Backfill populates the same +typed `ClickHouse { database, table }` source. Backfill populates the same SummaryStore instances used by other ingest sources; it does not introduce a second storage or catalog lifecycle. From 004c4991668a776ebf7db0dbdc9ea225303020ee Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:38:18 -0600 Subject: [PATCH 04/14] test: resolve ClickHouse backfill from the job table --- data_plane/tests/clickhouse_differential_e2e.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index 58520316f..7868c12b8 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -257,7 +257,7 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() { .arg("--clickhouse-url") .arg(&clickhouse_url) .arg("--clickhouse-backfill-table") - .arg("telemetry") + .arg("deployment_default_not_the_job_table") .arg("--clickhouse-backfill-database") .arg("default") .arg("--enable-backfill-worker") From 6d8a4ae418bfd71c715e18b2ab8eb542a96202bf Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:46:42 -0600 Subject: [PATCH 05/14] fix: validate ClickHouse backfill source before reading --- .../accelerator.rs | 4 +- .../sketch_db/backfill/clickhouse_reader.rs | 13 ++++- .../sketch_db/backfill/service.rs | 52 +++++++++++++++++++ 3 files changed, 66 insertions(+), 3 deletions(-) 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 3ab3a1b60..65d0e065c 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 @@ -441,8 +441,8 @@ mod tests { String::new(), "requests".into(), None, - None, - None, + Some("asap_e2e.samples".into()), + Some("value".into()), ); config.pane_origin_ms = Some(0); let sds = SummaryCatalog::from_materializations(41, 1, &[config.clone()]).unwrap(); 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 5ed8784bc..fc5d784e9 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 @@ -87,6 +87,12 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor let fallback = super::service::default_reader_factory(); Arc::new(move |source| match source { BackfillSource::ClickHouse { database, table } => { + if database != &config.database { + return Err(RawSampleReaderError::Other { + reason: "ClickHouse source database differs from the deployment database".into(), + } + .into()); + } let mut source_config = config.clone(); source_config.database = database.clone(); source_config.table = table.clone(); @@ -190,11 +196,16 @@ mod tests { fn typed_source_enters_clickhouse_backfill_lifecycle() { let factory = clickhouse_reader_factory(config("samples")); let reader = factory(&BackfillSource::ClickHouse { - database: "another_database".into(), + database: "metrics".into(), table: "another_table".into(), }) .unwrap(); assert_eq!(reader.source_name(), "ClickHouseReader"); + assert!(factory(&BackfillSource::ClickHouse { + database: "another_database".into(), + table: "samples".into(), + }) + .is_err()); assert!(factory(&BackfillSource::ClickHouse { database: "default".into(), table: "samples; DROP TABLE x".into(), diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs index c22beeaa6..b2644605a 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -199,6 +199,22 @@ impl BackfillService { "BackfillService picking up queued job" ); + if let BackfillSource::ClickHouse { database, table } = &job.source { + let snapshot = self.config_source.snapshot(); + let configured_table = snapshot + .get_aggregation_config(job.agg_id) + .and_then(|config| config.table_name.as_deref()); + let qualified_table = format!("{database}.{table}"); + if configured_table != Some(table.as_str()) + && configured_table != Some(qualified_table.as_str()) + { + self.registry.mark_failed( + job.job_id, + "ClickHouse source differs from the installed materialization table", + ); + continue; + } + } let reader = match (self.reader_factory)(&job.source) { Ok(r) => r, Err(e) => { @@ -383,6 +399,42 @@ mod tests { } } + #[tokio::test(flavor = "current_thread")] + async fn clickhouse_source_must_match_installed_table_before_reader_creation() { + let mut cfg = sum_config(1, "latency"); + cfg.table_name = Some("expected_table".into()); + let agg_fp = cfg.policy_fp_u64(); + let hot = HotReloadStreamingConfig::from_arc(streaming_with(cfg)); + let registry = Arc::new(BackfillRegistry::new()); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let called = calls.clone(); + let service = BackfillService::new( + registry.clone(), + hot, + Arc::new(move |_| { + called.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + Ok(Arc::new(MockRawSampleReader::new(vec![]))) + }), + BackfillServiceConfig { + poll_interval: Duration::from_millis(5), + }, + ); + let handle = service.spawn(); + let job_id = registry.create( + agg_fp, + (0, 20), + BackfillSource::ClickHouse { + database: "default".into(), + table: "wrong_table".into(), + }, + 1, + ); + let status = wait_for_status(®istry, job_id, BackfillStatus::Failed, 2000).await; + handle.shutdown().await; + assert_eq!(status, BackfillStatus::Failed); + assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 0); + } + #[tokio::test(flavor = "current_thread")] async fn service_drains_queued_job_to_complete() { let cfg = sum_config(1, "latency"); From 4a516073a8d9eb6caa7d9516ab54a92abc419c24 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:48:33 -0600 Subject: [PATCH 06/14] test: bind ClickHouse reader fixture to its catalog table --- .../query_engines/asap_clickhouse_query_engine/accelerator.rs | 2 ++ 1 file changed, 2 insertions(+) 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 65d0e065c..b264c82e8 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 @@ -756,6 +756,8 @@ mod tests { None, ); cfg.pane_origin_ms = Some(0); + cfg.table_name = Some("asap_e2e.samples".into()); + cfg.value_column = Some("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(), From 8b50aa53c2b98ba89b6902eef3fe2692095fe0de Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:55:57 -0600 Subject: [PATCH 07/14] test: declare legacy fixture population contracts --- .../drivers/ingest/prometheus_remote_write.rs | 4 ++++ data_plane/src/drivers/query/servers/http.rs | 2 ++ data_plane/src/precompute_engine/output_sink.rs | 2 ++ .../sketch_db/lifecycle/eviction.rs | 2 ++ .../src/tests/test_utilities/engine_factories.rs | 16 ++++++++++++++++ 5 files changed, 26 insertions(+) diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index ad7ad5ac6..a800d6ad4 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -858,6 +858,8 @@ mod tests { num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let policy_fp = aggregation.policy_fp_u64(); let streaming = StreamingConfig::new(HashMap::from([(policy_fp, aggregation)])); @@ -911,6 +913,8 @@ mod tests { num_aggregates_to_retain: Some(80), table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let cms = config( AggregationType::CountMinSketchWithHeap, diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index f7756cfda..20855ba0b 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3673,6 +3673,8 @@ aggregations: num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; // PR 5: streaming-config is keyed on the policy // fingerprint. Build a marker→fingerprint map so the test diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index bbc112c0a..dcde51a6e 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -344,6 +344,8 @@ mod tests { num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, } } 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 a5feb2f29..e9af55412 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -275,6 +275,8 @@ mod tests { num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: 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 b820bdc63..1d53ffd37 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -108,6 +108,8 @@ pub fn create_engine_single_pop_with_aggregated( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); @@ -202,6 +204,8 @@ pub fn create_engine_dual_input( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let value_id = value_agg_config.policy_fp_u64(); aggregation_configs.insert(value_id, value_agg_config); @@ -226,6 +230,8 @@ pub fn create_engine_dual_input( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let keys_id = keys_agg_config.policy_fp_u64(); aggregation_configs.insert(keys_id, keys_agg_config); @@ -315,6 +321,8 @@ pub fn create_engine_two_metrics( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let id_a = agg_config_a.policy_fp_u64(); aggregation_configs.insert(id_a, agg_config_a); @@ -338,6 +346,8 @@ pub fn create_engine_two_metrics( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let id_b = agg_config_b.policy_fp_u64(); aggregation_configs.insert(id_b, agg_config_b); @@ -437,6 +447,8 @@ pub fn create_engine_three_metrics( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let id = cfg.policy_fp_u64(); ids.push(id); @@ -513,6 +525,8 @@ pub fn create_engine_multi_timestamp( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); @@ -581,6 +595,8 @@ pub fn create_engine_multi_timestamp_with_window( num_aggregates_to_retain: None, table_name: None, value_column: None, + table_population: None, + partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); From fd372e3b112b15dc036c3733757eeb93101d8f17 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:59:52 -0600 Subject: [PATCH 08/14] feat: apply catalog population and projection in ClickHouse backfill --- crates/asap_types/src/policy_fingerprint.rs | 8 + crates/asap_types/src/summary_catalog.rs | 12 ++ crates/asap_types/src/table_population.rs | 2 + .../sketch_db/backfill/clickhouse_reader.rs | 180 ++++++++++++++++-- .../sketch_db/backfill/raw_sample_reader.rs | 28 +-- .../sketch_db/backfill/service.rs | 49 +++-- .../tests/clickhouse_differential_e2e.rs | 16 +- .../query-engine/clickhouse-sql-support.md | 20 ++ 8 files changed, 250 insertions(+), 65 deletions(-) diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 8475d463d..fabe0fbdc 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -175,6 +175,14 @@ impl PolicyFingerprint { // 10. spatial_filter_normalized — canonicalized predicate buf.extend_from_slice(cfg.spatial_filter_normalized.as_bytes()); + if let Some(table) = &cfg.table_name { + 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 { + buf.extend_from_slice(column.as_bytes()); + } + } if let Some(population) = &cfg.table_population { let canonical = population.canonical(); if !canonical.is_empty() { diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 92eb58259..da2975bff 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -300,6 +300,18 @@ mod tests { errors.table_population.as_mut().unwrap().predicates[0].value = ScalarValue::Utf8("errors".into()); assert_ne!(requests.policy_fingerprint(), errors.policy_fingerprint()); + let mut other_table = requests.clone(); + other_table.table_name = Some("other_samples".into()); + assert_ne!( + requests.policy_fingerprint(), + other_table.policy_fingerprint() + ); + let mut other_value = requests.clone(); + other_value.value_column = Some("other_value".into()); + assert_ne!( + requests.policy_fingerprint(), + other_value.policy_fingerprint() + ); let catalog = SummaryCatalog::from_materializations(1, 1, &[requests, errors]).unwrap(); assert_eq!(catalog.data_descriptors.len(), 2); assert_eq!(catalog.materializations.len(), 2); diff --git a/crates/asap_types/src/table_population.rs b/crates/asap_types/src/table_population.rs index 63566fa78..dc91a3366 100644 --- a/crates/asap_types/src/table_population.rs +++ b/crates/asap_types/src/table_population.rs @@ -21,6 +21,8 @@ impl TablePopulation { pub fn validate(&self) -> Result<(), String> { for predicate in &self.predicates { if predicate.column.is_empty() + || !predicate.column.as_bytes()[0].is_ascii_alphabetic() + && !predicate.column.starts_with('_') || !predicate .column .bytes() 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 fc5d784e9..a4b0b30b0 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 @@ -31,6 +31,7 @@ impl ClickHouseReaderConfig { ("value column", self.value_column.as_str()), ] { if value.is_empty() + || !value.as_bytes()[0].is_ascii_alphabetic() && !value.starts_with('_') || !value .bytes() .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_') @@ -47,11 +48,20 @@ impl ClickHouseReaderConfig { pub struct ClickHouseReader { config: ClickHouseReaderConfig, http: reqwest::Client, + population: Option, + output_metric: Option, +} + +#[derive(Deserialize)] +#[serde(untagged)] +enum ClickHouseLabels { + Series(String), + Map(std::collections::BTreeMap), } #[derive(Deserialize)] struct ClickHouseSampleRow { - labels: String, + labels: ClickHouseLabels, timestamp_ms: i64, value: f64, } @@ -62,14 +72,53 @@ impl ClickHouseReader { Ok(Self { config, http: reqwest::Client::new(), + population: None, + output_metric: None, }) } fn sql(&self) -> String { let c = &self.config; + let population = self.population.as_ref().map_or_else( + || format!("{} = {{metric:String}}", c.metric_column), + |population| { + if population.predicates.is_empty() { + return "1".into(); + } + population + .predicates + .iter() + .enumerate() + .map(|(index, predicate)| { + use planner_types::pre_asap::{CompareOpKind, ScalarValue}; + let operator = match predicate.operator { + CompareOpKind::Eq => "=", + CompareOpKind::Ne => "!=", + CompareOpKind::Lt => "<", + CompareOpKind::Le => "<=", + CompareOpKind::Gt => ">", + CompareOpKind::Ge => ">=", + _ => unreachable!("validated table predicate"), + }; + let kind = match predicate.value { + ScalarValue::Utf8(_) => "String", + ScalarValue::Int64(_) => "Int64", + ScalarValue::Float64(_) => "Float64", + ScalarValue::Boolean(_) => "Bool", + ScalarValue::Null => unreachable!("validated table literal"), + }; + format!( + "{} {operator} {{population_{index}:{kind}}}", + predicate.column + ) + }) + .collect::>() + .join(" AND ") + }, + ); format!( "SELECT {labels} AS labels, {timestamp} AS timestamp_ms, {value} AS value \ - FROM {database}.{table} WHERE {metric} = {{metric:String}} \ + FROM {database}.{table} WHERE {population} \ AND {timestamp} >= {{start_ms:Int64}} AND {timestamp} < {{end_ms:Int64}} \ ORDER BY labels, timestamp_ms FORMAT JSONEachRow", labels = c.labels_column, @@ -77,7 +126,6 @@ impl ClickHouseReader { value = c.value_column, database = c.database, table = c.table, - metric = c.metric_column, ) } } @@ -85,20 +133,29 @@ impl ClickHouseReader { /// Resolve a typed table source using deployment-local connection settings. pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactory { let fallback = super::service::default_reader_factory(); - Arc::new(move |source| match source { + Arc::new(move |source, materialization| match source { BackfillSource::ClickHouse { database, table } => { if database != &config.database { return Err(RawSampleReaderError::Other { - reason: "ClickHouse source database differs from the deployment database".into(), + reason: "ClickHouse source database differs from the deployment database" + .into(), } .into()); } let mut source_config = config.clone(); source_config.database = database.clone(); source_config.table = table.clone(); - Ok(Arc::new(ClickHouseReader::new(source_config)?) as Arc) + source_config.value_column = materialization + .value_column + .clone() + .ok_or("table materialization has no value projection")?; + materialization.population_filter_canonical()?; + let mut reader = ClickHouseReader::new(source_config)?; + reader.population = materialization.table_population.clone(); + reader.output_metric = Some(materialization.metric.clone()); + Ok(Arc::new(reader) as Arc) } - source => fallback(source), + source => fallback(source, materialization), }) } @@ -122,6 +179,19 @@ impl RawSampleReader for ClickHouseReader { ("param_start_ms", start_ms.as_str()), ("param_end_ms", end_ms.as_str()), ]); + if let Some(population) = &self.population { + for (index, predicate) in population.predicates.iter().enumerate() { + use planner_types::pre_asap::ScalarValue; + let value = match &predicate.value { + ScalarValue::Utf8(value) => value.clone(), + ScalarValue::Int64(value) => value.to_string(), + ScalarValue::Float64(value) => value.to_string(), + ScalarValue::Boolean(value) => value.to_string(), + ScalarValue::Null => unreachable!("validated table literal"), + }; + request = request.query(&[(format!("param_population_{index}"), value)]); + } + } if let Some(user) = &self.config.user { request = request.basic_auth(user, self.config.password.as_ref()); } @@ -149,7 +219,35 @@ impl RawSampleReader for ClickHouseReader { reason: error.to_string(), })?; let sample = RawSample { - labels: row.labels, + labels: match row.labels { + ClickHouseLabels::Series(series) => { + if self.population.is_some() { + let metric = self.output_metric.as_deref().unwrap_or(&filter.metric); + let suffix = series.find('{').map_or("", |start| &series[start..]); + format!("{metric}{suffix}") + } else { + series + } + } + ClickHouseLabels::Map(labels) => { + let metric = self.output_metric.as_deref().unwrap_or(&filter.metric); + let labels = labels + .iter() + .map(|(key, value)| { + format!( + "{key}={}", + serde_json::to_string(value).expect("label serialization") + ) + }) + .collect::>() + .join(","); + if labels.is_empty() { + metric.into() + } else { + format!("{metric}{{{labels}}}") + } + } + }, timestamp_ms: row.timestamp_ms, value: row.value, }; @@ -192,24 +290,66 @@ mod tests { assert!(sql.contains("FORMAT JSONEachRow")); } + #[test] + fn table_population_values_are_parameters_not_sql_fragments() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.population = Some(asap_types::table_population::TablePopulation { + predicates: vec![asap_types::table_population::TableColumnPredicate { + column: "metric".into(), + operator: planner_types::pre_asap::CompareOpKind::Eq, + value: planner_types::pre_asap::ScalarValue::Utf8("requests' OR 1=1 --".into()), + }], + }); + let sql = reader.sql(); + assert!(sql.contains("metric = {population_0:String}")); + assert!(!sql.contains("requests")); + assert!(!sql.contains("metric = {metric:String}")); + } + #[test] fn typed_source_enters_clickhouse_backfill_lifecycle() { + let materialization = asap_types::PrecomputeMaterialization::new( + asap_types::AggregationType::Sum, + String::new(), + Default::default(), + asap_types::KeyByLabelNames::empty(), + asap_types::KeyByLabelNames::empty(), + asap_types::KeyByLabelNames::empty(), + String::new(), + 1, + 1, + asap_types::WindowKind::Tumbling, + String::new(), + "samples.value".into(), + None, + Some("another_table".into()), + Some("value".into()), + ); let factory = clickhouse_reader_factory(config("samples")); - let reader = factory(&BackfillSource::ClickHouse { - database: "metrics".into(), - table: "another_table".into(), - }) + let reader = factory( + &BackfillSource::ClickHouse { + database: "metrics".into(), + table: "another_table".into(), + }, + &materialization, + ) .unwrap(); assert_eq!(reader.source_name(), "ClickHouseReader"); - assert!(factory(&BackfillSource::ClickHouse { - database: "another_database".into(), - table: "samples".into(), - }) + assert!(factory( + &BackfillSource::ClickHouse { + database: "another_database".into(), + table: "samples".into(), + }, + &materialization + ) .is_err()); - assert!(factory(&BackfillSource::ClickHouse { - database: "default".into(), - table: "samples; DROP TABLE x".into(), - }) + assert!(factory( + &BackfillSource::ClickHouse { + database: "metrics".into(), + table: "samples; DROP TABLE x".into(), + }, + &materialization + ) .is_err()); } } diff --git a/data_plane/src/storage_engines/sketch_db/backfill/raw_sample_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/raw_sample_reader.rs index 4e6e4f8de..c49f27a37 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/raw_sample_reader.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/raw_sample_reader.rs @@ -1,37 +1,23 @@ //! `RawSampleReader` — trait + mock implementation for reading raw //! samples from the exact DB during a [`BackfillJob`] run. //! -//! Supports the future backfill scope ([`future-storage-and-compression.md`](../../../../../docs/design_docs/future-storage-and-compression.md)). -//! and specifically §10.2's `BackfillSource` dispatch: every -//! concrete source (S3+Gorilla, Prometheus, ClickHouse, OtherSketch) -//! will eventually implement this trait so the worker pool (Phase 5c) -//! and the rebuild logic (Phase 5e) are source-agnostic. +//! Prometheus and ClickHouse implement the same worker-facing sample interface. +//! S3/Gorilla and rebuilding from another stored sketch remain unsupported. //! -//! ## Phase 5b scope (what this file covers) +//! ## Contracts //! //! * `RawSample` struct — the decoded-sample shape used end-to-end //! from exact-DB read through sketch replay. //! * `LabelFilter` struct — the subset of the series selector the //! reader needs to apply (metric + grouping label equality). -//! Deliberately narrow: the full PromQL matcher language isn't -//! needed for backfill, and a narrow type simplifies every reader -//! implementation. +//! SQL populations and projections come from the installed materialization +//! when constructing the reader, rather than being interpreted as PromQL. //! * `RawSampleReader` trait — async `read_samples(range, filter)` -//! returning a `Vec`. Plain Vec (not a stream) so the -//! trait stays object-safe and easy to mock; Phase 5e can revisit -//! streaming if large ranges become a memory pressure. +//! returning a `Vec`. Each requested window is buffered; large +//! windows still require a future streaming processor interface. //! * `MockRawSampleReader` — in-memory implementation used by //! Phase 5c's worker tests and Phase 5e's rebuild tests. //! -//! ## Out of scope for 5b (future phases) -//! -//! * `PrometheusReader` / `S3GorillaReader` / `ClickHouseReader` — -//! real network-backed implementations, deferred until Phase 5e -//! needs them. -//! * Streaming variant returning an `impl Stream` -//! — Phase 5e decides based on observed memory behaviour. -//! * `OtherSketch` variant lookup (reads from an existing -//! precompute rather than raw samples) — Phase 5e. use std::collections::HashMap; diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs index b2644605a..3fd4347ac 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -26,14 +26,14 @@ //! //! The service doesn't know how to talk to Prometheus / S3 //! — those are network-backed and deployment-specific. Instead, the -//! caller provides a [`ReaderFactory`] that takes a `BackfillSource` +//! caller provides a [`ReaderFactory`] that takes a `BackfillSource` and the +//! installed materialization's population and projection contract, //! and returns an `Arc`. When no factory is //! registered for a given source variant, the job fails with a clear //! "no reader configured" message (better than hanging in Queued). //! -//! Phase 5e-v1 doesn't ship a real PrometheusReader — that's Phase -//! 5h. Tests use `MockRawSampleReader`. Production deployments can -//! register a factory once the HTTP readers exist. +//! Production factories implement Prometheus and ClickHouse. Tests can +//! substitute `MockRawSampleReader` without changing the worker lifecycle. //! //! ## Time-disjoint enforcement //! @@ -68,6 +68,7 @@ use crate::storage_engines::types::HotReloadStreamingConfig; pub type ReaderFactory = Arc< dyn Fn( &BackfillSource, + &asap_types::PrecomputeMaterialization, ) -> Result, Box> + Send + Sync, @@ -199,11 +200,23 @@ impl BackfillService { "BackfillService picking up queued job" ); + let snapshot = self.config_source.snapshot(); + let Some(materialization) = snapshot.get_aggregation_config(job.agg_id) else { + self.registry + .mark_failed(job.job_id, "backfill materialization is not installed"); + continue; + }; + if materialization.table_population.is_some() + && !matches!(job.source, BackfillSource::ClickHouse { .. }) + { + self.registry.mark_failed( + job.job_id, + "typed table population requires a ClickHouse reader", + ); + continue; + } if let BackfillSource::ClickHouse { database, table } = &job.source { - let snapshot = self.config_source.snapshot(); - let configured_table = snapshot - .get_aggregation_config(job.agg_id) - .and_then(|config| config.table_name.as_deref()); + let configured_table = materialization.table_name.as_deref(); let qualified_table = format!("{database}.{table}"); if configured_table != Some(table.as_str()) && configured_table != Some(qualified_table.as_str()) @@ -215,7 +228,7 @@ impl BackfillService { continue; } } - let reader = match (self.reader_factory)(&job.source) { + let reader = match (self.reader_factory)(&job.source, materialization) { Ok(r) => r, Err(e) => { let msg = format!("reader factory failed: {e}"); @@ -241,13 +254,7 @@ impl BackfillService { } let worker = BackfillWorker::new(self.registry.clone()); - let filter = LabelFilter::for_metric( - self.config_source - .snapshot() - .get_aggregation_config(job.agg_id) - .map(|c| c.metric.clone()) - .unwrap_or_default(), - ); + let filter = LabelFilter::for_metric(materialization.metric.clone()); match worker .run_job(job.job_id, &filter, reader.as_ref(), &processor) .await @@ -306,7 +313,7 @@ impl Drop for BackfillServiceHandle { /// in `main.rs` when no real readers are wired yet: posted jobs /// get picked up, attempted, and fail fast with a clear reason. pub fn noop_reader_factory() -> ReaderFactory { - Arc::new(|source| { + Arc::new(|source, _materialization| { Err(format!("no RawSampleReader registered for source variant {source:?}").into()) }) } @@ -322,7 +329,7 @@ pub fn noop_reader_factory() -> ReaderFactory { /// `BackfillJob::error_message` so the control plane / operator sees /// exactly which reader is missing. pub fn default_reader_factory() -> ReaderFactory { - Arc::new(|source| match source { + Arc::new(|source, _materialization| match source { BackfillSource::Prometheus { url } => { let reader = super::prometheus_reader::PrometheusReader::new(url.clone()); Ok(Arc::new(reader) as Arc) @@ -411,7 +418,7 @@ mod tests { let service = BackfillService::new( registry.clone(), hot, - Arc::new(move |_| { + Arc::new(move |_, _| { called.fetch_add(1, std::sync::atomic::Ordering::Relaxed); Ok(Arc::new(MockRawSampleReader::new(vec![]))) }), @@ -445,7 +452,7 @@ mod tests { // Factory returns a fresh mock reader per call — seeded with a // handful of samples that cover the job's range. - let reader_factory: ReaderFactory = Arc::new(|_src| { + let reader_factory: ReaderFactory = Arc::new(|_src, _materialization| { Ok(Arc::new(MockRawSampleReader::new(vec![ RawSample { labels: "latency".into(), @@ -527,7 +534,7 @@ mod tests { // Factory records the order in which it's invoked. let invocations: Arc>> = Arc::new(Mutex::new(Vec::new())); let invocations_clone = invocations.clone(); - let reader_factory: ReaderFactory = Arc::new(move |src| { + let reader_factory: ReaderFactory = Arc::new(move |src, _materialization| { invocations_clone.lock().unwrap().push(format!("{src:?}")); Ok(Arc::new(MockRawSampleReader::new(vec![])) as Arc) }); diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index 7868c12b8..ddbd02995 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -81,6 +81,13 @@ fn mixed_workload( Some("value".into()), ); config.pane_origin_ms = Some(0); + config.table_population = Some(asap_types::table_population::TablePopulation { + predicates: vec![asap_types::table_population::TableColumnPredicate { + column: "metric".into(), + operator: planner_types::pre_asap::CompareOpKind::Eq, + value: planner_types::pre_asap::ScalarValue::Utf8("requests".into()), + }], + }); let sds = asap_types::summary_catalog::SummaryCatalog::from_materializations( 72, 1, @@ -108,6 +115,7 @@ fn mixed_workload( vec![ Column::new(time, DataType::Timestamp, false), Column::new(value, DataType::Float64, false), + Column::new("metric", DataType::Utf8, false), ], 0, vec![], @@ -202,13 +210,13 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() { let user = std::env::var("CLICKHOUSE_USER").ok(); let password = std::env::var("CLICKHOUSE_PASSWORD").ok(); let client = reqwest::Client::new(); - let sql = "SELECT sums.timestamp, sums.total / divisors.divisor AS ratio FROM (SELECT 2000 AS timestamp, sum(value) AS total FROM telemetry WHERE timestamp_ms >= 0 AND timestamp_ms < 2000) AS sums INNER JOIN divisors ON sums.timestamp = divisors.timestamp"; + let sql = "SELECT sums.timestamp, sums.total / divisors.divisor AS ratio FROM (SELECT 2000 AS timestamp, sum(value) AS total FROM telemetry WHERE metric = 'requests' AND timestamp_ms >= 0 AND timestamp_ms < 2000) AS sums INNER JOIN divisors ON sums.timestamp = divisors.timestamp"; for statement in [ "DROP TABLE IF EXISTS default.telemetry", "DROP TABLE IF EXISTS default.divisors", - "CREATE TABLE default.telemetry(metric String, labels String, timestamp_ms Int64, value Float64) ENGINE=Memory", + "CREATE TABLE default.telemetry(metric String, labels Map(String,String), timestamp_ms Int64, value Float64, wrong_value Float64) ENGINE=Memory", "CREATE TABLE default.divisors(timestamp Int64, divisor Float64) ENGINE=Memory", - "INSERT INTO default.telemetry VALUES ('telemetry.value','telemetry.value',100,2),('telemetry.value','telemetry.value',1100,3)", + "INSERT INTO default.telemetry VALUES ('requests',map('member','a'),0,2,10000),('requests',map('member','b'),1100,3,10000),('errors',map('member','c'),1100,99999,10000),('requests',map('member','a'),2000,88888,10000)", "INSERT INTO default.divisors VALUES (2000,10)", ] { let mut request = client.post(&clickhouse_url).body(statement); @@ -260,6 +268,8 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() { .arg("deployment_default_not_the_job_table") .arg("--clickhouse-backfill-database") .arg("default") + .arg("--clickhouse-backfill-value-column") + .arg("wrong_value") .arg("--enable-backfill-worker") .arg("--precompute-allowed-lateness-ms") .arg("0") diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index c74f803dd..8695dc7a3 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -78,6 +78,26 @@ typed `ClickHouse { database, table }` source. Backfill populates the same SummaryStore instances used by other ingest sources; it does not introduce a second storage or catalog lifecycle. +SQL materializations can carry a shared `TablePopulation` conjunction of typed +column/literal comparisons. Its canonical identity is stored in the catalog and +included in the materialization fingerprint together with table and value-column +identity. These predicates do not use the PromQL label-filter normalizer. The +reader takes its value projection and population from the installed materialization, +binds literal values as ClickHouse parameters, and accepts either encoded series +labels or a `Map(String,String)` label column. The requested database/table must +match the deployment and installed source respectively. +SQL fingerprints now include the explicit table/value source, so existing SQL +materializations must be republished and rebuilt; their old state is not reused +under the new identity. Legacy PromQL fingerprints remain unchanged. + +SQL timestamp comparisons retain their exact integer-millisecond inclusivity. +For example, `t <= 1999` and `t < 2000` identify the same half-open interval; +`t <= 2000` does not. A source range differing from the installed fixed evaluation +is rejected. Whole-second panes cannot yet cover an arbitrary inclusive boundary +fragment; those queries require exact fallback until the compiler can compose +boundary exact reads with summary interiors. Backfill still buffers a requested +window and has not demonstrated bounded memory at large window/cardinality scale. + ## Verification Focused tests cover language-isolated lookup, atomic install rejection for an From d04e6e3d50038bbfc562ff5cf98c94f0723718a3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 18:19:27 -0600 Subject: [PATCH 09/14] Bind SQL timestamp projection and preserve unfiltered populations --- control_plane/src/clickhouse.rs | 25 ++++++++++++++-- crates/asap_types/src/aggregation_config.rs | 29 +++++++++++++++++++ crates/asap_types/src/policy_fingerprint.rs | 4 +++ .../asap_types/src/precompute_plan/catalog.rs | 1 + crates/asap_types/src/sds.rs | 23 +++++++++++++++ crates/asap_types/src/summary_catalog.rs | 20 ++++++++++--- .../drivers/ingest/prometheus_remote_write.rs | 2 ++ data_plane/src/drivers/query/servers/http.rs | 1 + .../src/precompute_engine/output_sink.rs | 1 + .../accelerator.rs | 12 ++++---- .../sketch_db/backfill/clickhouse_reader.rs | 18 ++++++++++-- .../sketch_db/lifecycle/eviction.rs | 1 + .../tests/test_utilities/engine_factories.rs | 8 +++++ .../tests/clickhouse_differential_e2e.rs | 1 + .../query-engine/clickhouse-sql-support.md | 11 +++++++ 15 files changed, 144 insertions(+), 13 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 390387d90..66e8aa186 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -214,7 +214,7 @@ fn bind_selected_node( query: &ClickHouseSqlWorkloadEntry, request: &ClickHouseSqlWorkload, ) -> Result { - let (table_ref, value_column, source_window, spatial_filter) = + let (table_ref, value_column, source_window, spatial_filter, timestamp_column) = clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms) .map_err(crate::query_plan::QueryPlanError::Invalid)?; let expected = crate::physical::compiler::physical_materialization_family(family); @@ -226,6 +226,11 @@ fn bind_selected_node( &expected, source_window.unwrap_or((query.end_ms.saturating_sub(query.start_ms)) / 1000), )?; + if selected.table_timestamp_column.as_deref() != Some(timestamp_column.as_str()) { + return Err(crate::query_plan::QueryPlanError::Invalid( + "SQL timestamp projection differs from the installed materialization".into(), + )); + } Ok(MaterializationBinding { materialization: selected.policy_fingerprint().into(), output_grouping: PhysicalGrouping::Reduce(selected.grouping_labels.labels.clone()), @@ -240,7 +245,7 @@ fn clickhouse_materialization_leaf_contract( node: &planner_types::post_asap::SummaryNode, evaluation_start_ms: u64, evaluation_end_ms: u64, -) -> Result<(String, String, Option, String), String> { +) -> Result<(String, String, Option, String, String), String> { use planner_types::{ post_asap::SummaryExpr, pre_asap::{CompareOpKind, QueryExpr, ScalarValue, Source}, @@ -400,6 +405,12 @@ fn clickhouse_materialization_leaf_contract( value_column, Some(window_secs), population.canonical(), + schema + .time_index + .and_then(|index| schema.columns.get(index)) + .ok_or("SQL summary source has no timestamp projection")? + .name + .clone(), )) } @@ -467,6 +478,7 @@ mod tests { Some(value_column.into()), ); value.pane_origin_ms = Some(0); + value.table_timestamp_column = Some("timestamp_ms".into()); value } @@ -645,6 +657,15 @@ mod tests { ) })); let original = request.queries[0].sql.clone(); + let schema = request.tables.get_mut("telemetry").unwrap(); + schema.columns[schema.time_index.unwrap()].name = "other_timestamp".into(); + request.queries[0].sql = original.replace("timestamp_ms", "other_timestamp"); + assert!( + compile_clickhouse_workload(&request).await.is_err(), + "a summary cannot bind a different timestamp projection" + ); + let schema = request.tables.get_mut("telemetry").unwrap(); + schema.columns[schema.time_index.unwrap()].name = "timestamp_ms".into(); request.queries[0].sql = original.replace("timestamp_ms < 2000", "timestamp_ms <= 1999"); assert!(compile_clickhouse_workload(&request).await.is_ok()); request.queries[0].sql = original.replace("timestamp_ms < 2000", "timestamp_ms <= 2000"); diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index a7d6d166b..e22482f2d 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -128,6 +128,13 @@ pub struct PrecomputeMaterialization { // 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 + /// Table timestamp projection, in Unix milliseconds. + #[serde( + default, + alias = "tableTimestampColumn", + skip_serializing_if = "Option::is_none" + )] + pub table_timestamp_column: Option, #[serde( default, alias = "tablePopulation", @@ -171,6 +178,14 @@ pub type AggregationConfig = PrecomputeMaterialization; impl PrecomputeMaterialization { pub fn population_filter_canonical(&self) -> Result { + 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()); + } + } + if self.table_name.is_some() && !self.spatial_filter.is_empty() { + return Err("table populations cannot use a PromQL label filter".into()); + } if let Some(population) = &self.table_population { if self.table_name.is_none() || !self.spatial_filter.is_empty() { return Err( @@ -233,6 +248,7 @@ impl PrecomputeMaterialization { table_name, value_column, table_population: None, + table_timestamp_column: None, } } @@ -347,6 +363,11 @@ impl PrecomputeMaterialization { .map(|value| serde_json::from_value(value.clone())) .transpose()?; config.pane_origin_ms = pane_origin_ms; + config.table_timestamp_column = data + .get("tableTimestampColumn") + .or_else(|| data.get("table_timestamp_column")) + .and_then(Value::as_str) + .map(str::to_owned); config.table_population = data .get("tablePopulation") .or_else(|| data.get("table_population")) @@ -499,6 +520,11 @@ impl PrecomputeMaterialization { .map(|value| serde_yaml::from_value(value.clone())) .transpose()?; config.pane_origin_ms = pane_origin_ms; + config.table_timestamp_column = aggregation_data + .get("tableTimestampColumn") + .or_else(|| aggregation_data.get("table_timestamp_column")) + .and_then(serde_yaml::Value::as_str) + .map(str::to_owned); config.table_population = aggregation_data .get("tablePopulation") .or_else(|| aggregation_data.get("table_population")) @@ -548,6 +574,9 @@ impl SerializableToSink for PrecomputeMaterialization { if let Some(ref population) = self.table_population { json["tablePopulation"] = serde_json::json!(population); } + if let Some(ref column) = self.table_timestamp_column { + json["tableTimestampColumn"] = serde_json::json!(column); + } json } diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index fabe0fbdc..53acfc573 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -190,6 +190,10 @@ impl PolicyFingerprint { buf.extend_from_slice(canonical.as_bytes()); } } + if let Some(column) = &cfg.table_timestamp_column { + buf.extend_from_slice(b"\0timestamp-ms\0"); + buf.extend_from_slice(column.as_bytes()); + } Self(xxh64(&buf, 0)) } diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index eb8a5408f..26c5e1e9a 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -87,6 +87,7 @@ impl PrecomputePlan { ValueProjectionIdentity::Column { name: name.clone() } }); if data.partitioning != config.partitioning + || data.timestamp_column != config.table_timestamp_column || data.source != expected_source || data.value_projection != expected_projection || data.population_filter_canonical diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index a85422bc9..c5566a609 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -713,6 +713,9 @@ pub struct DataDescriptor { pub id: DataDescriptorId, pub source: DataSourceIdentity, pub value_projection: ValueProjectionIdentity, + /// Table column containing Unix milliseconds. Absent for time-series sources. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timestamp_column: Option, pub population_filter_canonical: String, pub group_by_keys: BTreeSet, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -777,11 +780,13 @@ impl DataDescriptor { &group_by_keys, &observation_semantics, None, + None, ); Self { id, source, value_projection, + timestamp_column: None, population_filter_canonical, group_by_keys, partitioning: None, @@ -797,13 +802,26 @@ impl DataDescriptor { &self.group_by_keys, &self.observation_semantics, partitioning, + self.timestamp_column.as_deref(), ); self } + pub fn with_timestamp_column(mut self, column: Option) -> Self { + self.timestamp_column = column; + let partitioning = self.partitioning; + self.with_partitioning(partitioning) + } pub fn id(&self) -> &DataDescriptorId { &self.id } pub fn validate(&self) -> Result<(), SdsError> { + if let Some(column) = &self.timestamp_column { + if !matches!(self.source, DataSourceIdentity::Table { .. }) || column.is_empty() { + return Err(SdsError( + "table timestamp projection requires a table and a column".into(), + )); + } + } if self.id != data_descriptor_id( &self.source, @@ -812,6 +830,7 @@ impl DataDescriptor { &self.group_by_keys, &self.observation_semantics, self.partitioning, + self.timestamp_column.as_deref(), ) { return Err(SdsError("data descriptor ID/content mismatch".into())); @@ -826,6 +845,7 @@ fn data_descriptor_id( group_by: &BTreeSet, observation_semantics: &str, partitioning: Option, + timestamp_column: Option<&str>, ) -> DataDescriptorId { // Length framing keeps distinct typed sources, projections, predicates, // and grouping keys collision-free in the content identity. @@ -841,6 +861,9 @@ fn data_descriptor_id( if let Some(partitioning) = partitioning { key.push_str(&format!("|partition:{partitioning:?}")); } + if let Some(column) = timestamp_column { + key.push_str(&format!("|timestamp-ms:{}:{column}", column.len())); + } for name in group_by { key.push_str(&format!("|{}:{name}", name.len())); } diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index da2975bff..571384571 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -126,7 +126,8 @@ impl SummaryCatalog { config.grouping_labels.labels.clone(), "asap.timestamped-observations.v2", ) - .with_partitioning(config.partitioning); + .with_partitioning(config.partitioning) + .with_timestamp_column(config.table_timestamp_column.clone()); Ok(( config.policy_fingerprint(), summary, @@ -312,9 +313,20 @@ mod tests { requests.policy_fingerprint(), other_value.policy_fingerprint() ); - let catalog = SummaryCatalog::from_materializations(1, 1, &[requests, errors]).unwrap(); - assert_eq!(catalog.data_descriptors.len(), 2); - assert_eq!(catalog.materializations.len(), 2); + let mut other_time = requests.clone(); + other_time.table_timestamp_column = Some("event_time_ms".into()); + assert_ne!( + requests.policy_fingerprint(), + other_time.policy_fingerprint() + ); + let mut invalid_labels = requests.clone(); + invalid_labels.table_population = None; + invalid_labels.spatial_filter = "job=\"requests\"".into(); + assert!(SummaryCatalog::from_materializations(1, 1, &[invalid_labels]).is_err()); + let catalog = + SummaryCatalog::from_materializations(1, 1, &[requests, errors, other_time]).unwrap(); + assert_eq!(catalog.data_descriptors.len(), 3); + assert_eq!(catalog.materializations.len(), 3); } #[test] diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index a800d6ad4..f3da96a76 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -859,6 +859,7 @@ mod tests { table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let policy_fp = aggregation.policy_fp_u64(); @@ -914,6 +915,7 @@ mod tests { table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let cms = config( diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 20855ba0b..897c25a1a 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3674,6 +3674,7 @@ aggregations: table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; // PR 5: streaming-config is keyed on the policy diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index dcde51a6e..768a08291 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -345,6 +345,7 @@ mod tests { table_name: None, value_column: 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 b264c82e8..a4a023a4f 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 @@ -445,6 +445,7 @@ mod tests { Some("value".into()), ); config.pane_origin_ms = Some(0); + config.table_timestamp_column = Some("timestamp_ms".into()); let sds = SummaryCatalog::from_materializations(41, 1, &[config.clone()]).unwrap(); let materialization = *sds.materializations.keys().next().unwrap(); let read = QueryNodeId(0); @@ -732,7 +733,7 @@ mod tests { "CREATE DATABASE IF NOT EXISTS asap_e2e", "DROP TABLE IF EXISTS asap_e2e.samples", "CREATE TABLE asap_e2e.samples(metric String, labels String, timestamp_ms Int64, value Float64) ENGINE=Memory", - "INSERT INTO asap_e2e.samples VALUES ('requests','requests',100,2),('requests','requests',1100,3)", + "INSERT INTO asap_e2e.samples VALUES ('requests','requests',100,2),('errors','errors',1100,3)", ] { let mut request = client.post(&base_url).body(sql); if let Some(user) = &user { request = request.basic_auth(user, password.as_ref()); } @@ -757,6 +758,7 @@ 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()); let hot = crate::storage_engines::types::HotReloadStreamingConfig::from_arc(Arc::new( crate::storage_engines::types::StreamingConfig::new(HashMap::from([( @@ -772,7 +774,7 @@ mod tests { table: "samples".into(), metric_column: "metric".into(), labels_column: "labels".into(), - timestamp_ms_column: "timestamp_ms".into(), + timestamp_ms_column: "wrong_deployment_timestamp".into(), value_column: "value".into(), user, password, @@ -825,9 +827,9 @@ mod tests { ), }; assert_eq!(response.body, "1970-01-01T00:00:02\t50.0\n"); - let mut exact = client.post(std::env::var("CLICKHOUSE_URL").unwrap()).body( - "SELECT sum(value) * 10 FROM asap_e2e.samples WHERE metric='requests' FORMAT TabSeparated", - ); + let mut exact = client + .post(std::env::var("CLICKHOUSE_URL").unwrap()) + .body("SELECT sum(value) * 10 FROM asap_e2e.samples FORMAT TabSeparated"); if let Some(user) = std::env::var("CLICKHOUSE_USER").ok() { exact = exact.basic_auth(user, std::env::var("CLICKHOUSE_PASSWORD").ok()); } 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 a4b0b30b0..8663988bf 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 @@ -145,13 +145,17 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor let mut source_config = config.clone(); source_config.database = database.clone(); source_config.table = table.clone(); + source_config.timestamp_ms_column = materialization + .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")?; materialization.population_filter_canonical()?; let mut reader = ClickHouseReader::new(source_config)?; - reader.population = materialization.table_population.clone(); + reader.population = Some(materialization.table_population.clone().unwrap_or_default()); reader.output_metric = Some(materialization.metric.clone()); Ok(Arc::new(reader) as Arc) } @@ -306,9 +310,18 @@ mod tests { assert!(!sql.contains("metric = {metric:String}")); } + #[test] + fn unfiltered_table_population_does_not_filter_by_output_metric() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.population = Some(Default::default()); + let sql = reader.sql(); + assert!(sql.contains("WHERE 1 AND")); + assert!(!sql.contains("{metric:String}")); + } + #[test] fn typed_source_enters_clickhouse_backfill_lifecycle() { - let materialization = asap_types::PrecomputeMaterialization::new( + let mut materialization = asap_types::PrecomputeMaterialization::new( asap_types::AggregationType::Sum, String::new(), Default::default(), @@ -326,6 +339,7 @@ mod tests { Some("value".into()), ); let factory = clickhouse_reader_factory(config("samples")); + materialization.table_timestamp_column = Some("timestamp_ms".into()); let reader = factory( &BackfillSource::ClickHouse { database: "metrics".into(), 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 e9af55412..557b8326a 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -276,6 +276,7 @@ mod tests { table_name: None, value_column: 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 1d53ffd37..6f742df94 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -109,6 +109,7 @@ pub fn create_engine_single_pop_with_aggregated( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); @@ -205,6 +206,7 @@ pub fn create_engine_dual_input( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let value_id = value_agg_config.policy_fp_u64(); @@ -231,6 +233,7 @@ pub fn create_engine_dual_input( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let keys_id = keys_agg_config.policy_fp_u64(); @@ -322,6 +325,7 @@ pub fn create_engine_two_metrics( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let id_a = agg_config_a.policy_fp_u64(); @@ -347,6 +351,7 @@ pub fn create_engine_two_metrics( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let id_b = agg_config_b.policy_fp_u64(); @@ -448,6 +453,7 @@ pub fn create_engine_three_metrics( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let id = cfg.policy_fp_u64(); @@ -526,6 +532,7 @@ pub fn create_engine_multi_timestamp( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); @@ -596,6 +603,7 @@ pub fn create_engine_multi_timestamp_with_window( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index ddbd02995..84de688ee 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -81,6 +81,7 @@ fn mixed_workload( Some("value".into()), ); config.pane_origin_ms = Some(0); + config.table_timestamp_column = Some("timestamp_ms".into()); config.table_population = Some(asap_types::table_population::TablePopulation { predicates: vec![asap_types::table_population::TableColumnPredicate { column: "metric".into(), diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 8695dc7a3..3668aca7b 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -86,6 +86,17 @@ reader takes its value projection and population from the installed materializat binds literal values as ClickHouse parameters, and accepts either encoded series labels or a `Map(String,String)` label column. The requested database/table must match the deployment and installed source respectively. +An absent or empty table population means every row in the table's time interval; +the output metric name never becomes an implicit SQL predicate. Table sources +reject legacy PromQL spatial filters. + +The installed `table_timestamp_column` names a Unix-millisecond column and is +shared as `DataDescriptor.timestamp_column`. It enters both identities and is +checked against the Planner source schema's `time_index`. Backfill uses this +installed projection even when the deployment default names a different column. +Legacy table definitions without a timestamp projection cannot be bound or +backfilled; republish them with the explicit column. Time-series definitions +retain their existing timestamp semantics and identity. SQL fingerprints now include the explicit table/value source, so existing SQL materializations must be republished and rebuilt; their old state is not reused under the new identity. Legacy PromQL fingerprints remain unchanged. From 3f169a178876604fe3739907e7d10037fa1c59fc Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 18:19:27 -0600 Subject: [PATCH 10/14] Bind SQL timestamp projection and preserve unfiltered populations --- control_plane/src/clickhouse.rs | 25 ++++++++++++++-- crates/asap_types/src/aggregation_config.rs | 29 +++++++++++++++++++ crates/asap_types/src/policy_fingerprint.rs | 4 +++ .../asap_types/src/precompute_plan/catalog.rs | 1 + crates/asap_types/src/sds.rs | 23 +++++++++++++++ crates/asap_types/src/summary_catalog.rs | 20 ++++++++++--- .../drivers/ingest/prometheus_remote_write.rs | 2 ++ data_plane/src/drivers/query/servers/http.rs | 1 + .../src/precompute_engine/output_sink.rs | 1 + .../accelerator.rs | 12 ++++---- .../sketch_db/backfill/clickhouse_reader.rs | 18 ++++++++++-- .../sketch_db/lifecycle/eviction.rs | 1 + .../tests/test_utilities/engine_factories.rs | 8 +++++ .../tests/clickhouse_differential_e2e.rs | 1 + .../query-engine/clickhouse-sql-support.md | 11 +++++++ 15 files changed, 144 insertions(+), 13 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 390387d90..66e8aa186 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -214,7 +214,7 @@ fn bind_selected_node( query: &ClickHouseSqlWorkloadEntry, request: &ClickHouseSqlWorkload, ) -> Result { - let (table_ref, value_column, source_window, spatial_filter) = + let (table_ref, value_column, source_window, spatial_filter, timestamp_column) = clickhouse_materialization_leaf_contract(node, query.start_ms, query.end_ms) .map_err(crate::query_plan::QueryPlanError::Invalid)?; let expected = crate::physical::compiler::physical_materialization_family(family); @@ -226,6 +226,11 @@ fn bind_selected_node( &expected, source_window.unwrap_or((query.end_ms.saturating_sub(query.start_ms)) / 1000), )?; + if selected.table_timestamp_column.as_deref() != Some(timestamp_column.as_str()) { + return Err(crate::query_plan::QueryPlanError::Invalid( + "SQL timestamp projection differs from the installed materialization".into(), + )); + } Ok(MaterializationBinding { materialization: selected.policy_fingerprint().into(), output_grouping: PhysicalGrouping::Reduce(selected.grouping_labels.labels.clone()), @@ -240,7 +245,7 @@ fn clickhouse_materialization_leaf_contract( node: &planner_types::post_asap::SummaryNode, evaluation_start_ms: u64, evaluation_end_ms: u64, -) -> Result<(String, String, Option, String), String> { +) -> Result<(String, String, Option, String, String), String> { use planner_types::{ post_asap::SummaryExpr, pre_asap::{CompareOpKind, QueryExpr, ScalarValue, Source}, @@ -400,6 +405,12 @@ fn clickhouse_materialization_leaf_contract( value_column, Some(window_secs), population.canonical(), + schema + .time_index + .and_then(|index| schema.columns.get(index)) + .ok_or("SQL summary source has no timestamp projection")? + .name + .clone(), )) } @@ -467,6 +478,7 @@ mod tests { Some(value_column.into()), ); value.pane_origin_ms = Some(0); + value.table_timestamp_column = Some("timestamp_ms".into()); value } @@ -645,6 +657,15 @@ mod tests { ) })); let original = request.queries[0].sql.clone(); + let schema = request.tables.get_mut("telemetry").unwrap(); + schema.columns[schema.time_index.unwrap()].name = "other_timestamp".into(); + request.queries[0].sql = original.replace("timestamp_ms", "other_timestamp"); + assert!( + compile_clickhouse_workload(&request).await.is_err(), + "a summary cannot bind a different timestamp projection" + ); + let schema = request.tables.get_mut("telemetry").unwrap(); + schema.columns[schema.time_index.unwrap()].name = "timestamp_ms".into(); request.queries[0].sql = original.replace("timestamp_ms < 2000", "timestamp_ms <= 1999"); assert!(compile_clickhouse_workload(&request).await.is_ok()); request.queries[0].sql = original.replace("timestamp_ms < 2000", "timestamp_ms <= 2000"); diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index a7d6d166b..e22482f2d 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -128,6 +128,13 @@ pub struct PrecomputeMaterialization { // 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 + /// Table timestamp projection, in Unix milliseconds. + #[serde( + default, + alias = "tableTimestampColumn", + skip_serializing_if = "Option::is_none" + )] + pub table_timestamp_column: Option, #[serde( default, alias = "tablePopulation", @@ -171,6 +178,14 @@ pub type AggregationConfig = PrecomputeMaterialization; impl PrecomputeMaterialization { pub fn population_filter_canonical(&self) -> Result { + 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()); + } + } + if self.table_name.is_some() && !self.spatial_filter.is_empty() { + return Err("table populations cannot use a PromQL label filter".into()); + } if let Some(population) = &self.table_population { if self.table_name.is_none() || !self.spatial_filter.is_empty() { return Err( @@ -233,6 +248,7 @@ impl PrecomputeMaterialization { table_name, value_column, table_population: None, + table_timestamp_column: None, } } @@ -347,6 +363,11 @@ impl PrecomputeMaterialization { .map(|value| serde_json::from_value(value.clone())) .transpose()?; config.pane_origin_ms = pane_origin_ms; + config.table_timestamp_column = data + .get("tableTimestampColumn") + .or_else(|| data.get("table_timestamp_column")) + .and_then(Value::as_str) + .map(str::to_owned); config.table_population = data .get("tablePopulation") .or_else(|| data.get("table_population")) @@ -499,6 +520,11 @@ impl PrecomputeMaterialization { .map(|value| serde_yaml::from_value(value.clone())) .transpose()?; config.pane_origin_ms = pane_origin_ms; + config.table_timestamp_column = aggregation_data + .get("tableTimestampColumn") + .or_else(|| aggregation_data.get("table_timestamp_column")) + .and_then(serde_yaml::Value::as_str) + .map(str::to_owned); config.table_population = aggregation_data .get("tablePopulation") .or_else(|| aggregation_data.get("table_population")) @@ -548,6 +574,9 @@ impl SerializableToSink for PrecomputeMaterialization { if let Some(ref population) = self.table_population { json["tablePopulation"] = serde_json::json!(population); } + if let Some(ref column) = self.table_timestamp_column { + json["tableTimestampColumn"] = serde_json::json!(column); + } json } diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index fabe0fbdc..53acfc573 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -190,6 +190,10 @@ impl PolicyFingerprint { buf.extend_from_slice(canonical.as_bytes()); } } + if let Some(column) = &cfg.table_timestamp_column { + buf.extend_from_slice(b"\0timestamp-ms\0"); + buf.extend_from_slice(column.as_bytes()); + } Self(xxh64(&buf, 0)) } diff --git a/crates/asap_types/src/precompute_plan/catalog.rs b/crates/asap_types/src/precompute_plan/catalog.rs index eb8a5408f..26c5e1e9a 100644 --- a/crates/asap_types/src/precompute_plan/catalog.rs +++ b/crates/asap_types/src/precompute_plan/catalog.rs @@ -87,6 +87,7 @@ impl PrecomputePlan { ValueProjectionIdentity::Column { name: name.clone() } }); if data.partitioning != config.partitioning + || data.timestamp_column != config.table_timestamp_column || data.source != expected_source || data.value_projection != expected_projection || data.population_filter_canonical diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index a85422bc9..c5566a609 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -713,6 +713,9 @@ pub struct DataDescriptor { pub id: DataDescriptorId, pub source: DataSourceIdentity, pub value_projection: ValueProjectionIdentity, + /// Table column containing Unix milliseconds. Absent for time-series sources. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timestamp_column: Option, pub population_filter_canonical: String, pub group_by_keys: BTreeSet, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -777,11 +780,13 @@ impl DataDescriptor { &group_by_keys, &observation_semantics, None, + None, ); Self { id, source, value_projection, + timestamp_column: None, population_filter_canonical, group_by_keys, partitioning: None, @@ -797,13 +802,26 @@ impl DataDescriptor { &self.group_by_keys, &self.observation_semantics, partitioning, + self.timestamp_column.as_deref(), ); self } + pub fn with_timestamp_column(mut self, column: Option) -> Self { + self.timestamp_column = column; + let partitioning = self.partitioning; + self.with_partitioning(partitioning) + } pub fn id(&self) -> &DataDescriptorId { &self.id } pub fn validate(&self) -> Result<(), SdsError> { + if let Some(column) = &self.timestamp_column { + if !matches!(self.source, DataSourceIdentity::Table { .. }) || column.is_empty() { + return Err(SdsError( + "table timestamp projection requires a table and a column".into(), + )); + } + } if self.id != data_descriptor_id( &self.source, @@ -812,6 +830,7 @@ impl DataDescriptor { &self.group_by_keys, &self.observation_semantics, self.partitioning, + self.timestamp_column.as_deref(), ) { return Err(SdsError("data descriptor ID/content mismatch".into())); @@ -826,6 +845,7 @@ fn data_descriptor_id( group_by: &BTreeSet, observation_semantics: &str, partitioning: Option, + timestamp_column: Option<&str>, ) -> DataDescriptorId { // Length framing keeps distinct typed sources, projections, predicates, // and grouping keys collision-free in the content identity. @@ -841,6 +861,9 @@ fn data_descriptor_id( if let Some(partitioning) = partitioning { key.push_str(&format!("|partition:{partitioning:?}")); } + if let Some(column) = timestamp_column { + key.push_str(&format!("|timestamp-ms:{}:{column}", column.len())); + } for name in group_by { key.push_str(&format!("|{}:{name}", name.len())); } diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index da2975bff..571384571 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -126,7 +126,8 @@ impl SummaryCatalog { config.grouping_labels.labels.clone(), "asap.timestamped-observations.v2", ) - .with_partitioning(config.partitioning); + .with_partitioning(config.partitioning) + .with_timestamp_column(config.table_timestamp_column.clone()); Ok(( config.policy_fingerprint(), summary, @@ -312,9 +313,20 @@ mod tests { requests.policy_fingerprint(), other_value.policy_fingerprint() ); - let catalog = SummaryCatalog::from_materializations(1, 1, &[requests, errors]).unwrap(); - assert_eq!(catalog.data_descriptors.len(), 2); - assert_eq!(catalog.materializations.len(), 2); + let mut other_time = requests.clone(); + other_time.table_timestamp_column = Some("event_time_ms".into()); + assert_ne!( + requests.policy_fingerprint(), + other_time.policy_fingerprint() + ); + let mut invalid_labels = requests.clone(); + invalid_labels.table_population = None; + invalid_labels.spatial_filter = "job=\"requests\"".into(); + assert!(SummaryCatalog::from_materializations(1, 1, &[invalid_labels]).is_err()); + let catalog = + SummaryCatalog::from_materializations(1, 1, &[requests, errors, other_time]).unwrap(); + assert_eq!(catalog.data_descriptors.len(), 3); + assert_eq!(catalog.materializations.len(), 3); } #[test] diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index a800d6ad4..f3da96a76 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -859,6 +859,7 @@ mod tests { table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let policy_fp = aggregation.policy_fp_u64(); @@ -914,6 +915,7 @@ mod tests { table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let cms = config( diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 20855ba0b..897c25a1a 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -3674,6 +3674,7 @@ aggregations: table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; // PR 5: streaming-config is keyed on the policy diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index dcde51a6e..768a08291 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -345,6 +345,7 @@ mod tests { table_name: None, value_column: 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 b264c82e8..a4a023a4f 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 @@ -445,6 +445,7 @@ mod tests { Some("value".into()), ); config.pane_origin_ms = Some(0); + config.table_timestamp_column = Some("timestamp_ms".into()); let sds = SummaryCatalog::from_materializations(41, 1, &[config.clone()]).unwrap(); let materialization = *sds.materializations.keys().next().unwrap(); let read = QueryNodeId(0); @@ -732,7 +733,7 @@ mod tests { "CREATE DATABASE IF NOT EXISTS asap_e2e", "DROP TABLE IF EXISTS asap_e2e.samples", "CREATE TABLE asap_e2e.samples(metric String, labels String, timestamp_ms Int64, value Float64) ENGINE=Memory", - "INSERT INTO asap_e2e.samples VALUES ('requests','requests',100,2),('requests','requests',1100,3)", + "INSERT INTO asap_e2e.samples VALUES ('requests','requests',100,2),('errors','errors',1100,3)", ] { let mut request = client.post(&base_url).body(sql); if let Some(user) = &user { request = request.basic_auth(user, password.as_ref()); } @@ -757,6 +758,7 @@ 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()); let hot = crate::storage_engines::types::HotReloadStreamingConfig::from_arc(Arc::new( crate::storage_engines::types::StreamingConfig::new(HashMap::from([( @@ -772,7 +774,7 @@ mod tests { table: "samples".into(), metric_column: "metric".into(), labels_column: "labels".into(), - timestamp_ms_column: "timestamp_ms".into(), + timestamp_ms_column: "wrong_deployment_timestamp".into(), value_column: "value".into(), user, password, @@ -825,9 +827,9 @@ mod tests { ), }; assert_eq!(response.body, "1970-01-01T00:00:02\t50.0\n"); - let mut exact = client.post(std::env::var("CLICKHOUSE_URL").unwrap()).body( - "SELECT sum(value) * 10 FROM asap_e2e.samples WHERE metric='requests' FORMAT TabSeparated", - ); + let mut exact = client + .post(std::env::var("CLICKHOUSE_URL").unwrap()) + .body("SELECT sum(value) * 10 FROM asap_e2e.samples FORMAT TabSeparated"); if let Some(user) = std::env::var("CLICKHOUSE_USER").ok() { exact = exact.basic_auth(user, std::env::var("CLICKHOUSE_PASSWORD").ok()); } 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 a4b0b30b0..8663988bf 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 @@ -145,13 +145,17 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor let mut source_config = config.clone(); source_config.database = database.clone(); source_config.table = table.clone(); + source_config.timestamp_ms_column = materialization + .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")?; materialization.population_filter_canonical()?; let mut reader = ClickHouseReader::new(source_config)?; - reader.population = materialization.table_population.clone(); + reader.population = Some(materialization.table_population.clone().unwrap_or_default()); reader.output_metric = Some(materialization.metric.clone()); Ok(Arc::new(reader) as Arc) } @@ -306,9 +310,18 @@ mod tests { assert!(!sql.contains("metric = {metric:String}")); } + #[test] + fn unfiltered_table_population_does_not_filter_by_output_metric() { + let mut reader = ClickHouseReader::new(config("samples")).unwrap(); + reader.population = Some(Default::default()); + let sql = reader.sql(); + assert!(sql.contains("WHERE 1 AND")); + assert!(!sql.contains("{metric:String}")); + } + #[test] fn typed_source_enters_clickhouse_backfill_lifecycle() { - let materialization = asap_types::PrecomputeMaterialization::new( + let mut materialization = asap_types::PrecomputeMaterialization::new( asap_types::AggregationType::Sum, String::new(), Default::default(), @@ -326,6 +339,7 @@ mod tests { Some("value".into()), ); let factory = clickhouse_reader_factory(config("samples")); + materialization.table_timestamp_column = Some("timestamp_ms".into()); let reader = factory( &BackfillSource::ClickHouse { database: "metrics".into(), 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 e9af55412..557b8326a 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -276,6 +276,7 @@ mod tests { table_name: None, value_column: 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 1d53ffd37..6f742df94 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -109,6 +109,7 @@ pub fn create_engine_single_pop_with_aggregated( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); @@ -205,6 +206,7 @@ pub fn create_engine_dual_input( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let value_id = value_agg_config.policy_fp_u64(); @@ -231,6 +233,7 @@ pub fn create_engine_dual_input( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let keys_id = keys_agg_config.policy_fp_u64(); @@ -322,6 +325,7 @@ pub fn create_engine_two_metrics( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let id_a = agg_config_a.policy_fp_u64(); @@ -347,6 +351,7 @@ pub fn create_engine_two_metrics( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let id_b = agg_config_b.policy_fp_u64(); @@ -448,6 +453,7 @@ pub fn create_engine_three_metrics( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let id = cfg.policy_fp_u64(); @@ -526,6 +532,7 @@ pub fn create_engine_multi_timestamp( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); @@ -596,6 +603,7 @@ pub fn create_engine_multi_timestamp_with_window( table_name: None, value_column: None, table_population: None, + table_timestamp_column: None, partitioning: None, }; let agg_id = agg_config.policy_fp_u64(); diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index ddbd02995..84de688ee 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -81,6 +81,7 @@ fn mixed_workload( Some("value".into()), ); config.pane_origin_ms = Some(0); + config.table_timestamp_column = Some("timestamp_ms".into()); config.table_population = Some(asap_types::table_population::TablePopulation { predicates: vec![asap_types::table_population::TableColumnPredicate { column: "metric".into(), diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 8695dc7a3..3668aca7b 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -86,6 +86,17 @@ reader takes its value projection and population from the installed materializat binds literal values as ClickHouse parameters, and accepts either encoded series labels or a `Map(String,String)` label column. The requested database/table must match the deployment and installed source respectively. +An absent or empty table population means every row in the table's time interval; +the output metric name never becomes an implicit SQL predicate. Table sources +reject legacy PromQL spatial filters. + +The installed `table_timestamp_column` names a Unix-millisecond column and is +shared as `DataDescriptor.timestamp_column`. It enters both identities and is +checked against the Planner source schema's `time_index`. Backfill uses this +installed projection even when the deployment default names a different column. +Legacy table definitions without a timestamp projection cannot be bound or +backfilled; republish them with the explicit column. Time-series definitions +retain their existing timestamp semantics and identity. SQL fingerprints now include the explicit table/value source, so existing SQL materializations must be republished and rebuilt; their old state is not reused under the new identity. Legacy PromQL fingerprints remain unchanged. From 12666e6707b99e2c606d7dfa5a6eec81b22b7154 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 18:24:34 -0600 Subject: [PATCH 11/14] Validate SQL timestamp identifiers before catalog publication --- crates/asap_types/src/aggregation_config.rs | 1 + crates/asap_types/src/sds.rs | 1 + crates/asap_types/src/summary_catalog.rs | 3 +++ crates/asap_types/src/table_population.rs | 23 ++++++++++++--------- 4 files changed, 18 insertions(+), 10 deletions(-) diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index e22482f2d..e2411a565 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -182,6 +182,7 @@ impl PrecomputeMaterialization { if self.table_name.is_none() || column.is_empty() { return Err("table timestamp projection requires a table and a column".into()); } + crate::table_population::validate_column_name(column)?; } if self.table_name.is_some() && !self.spatial_filter.is_empty() { return Err("table populations cannot use a PromQL label filter".into()); diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index c5566a609..1a137e360 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -821,6 +821,7 @@ impl DataDescriptor { "table timestamp projection requires a table and a column".into(), )); } + crate::table_population::validate_column_name(column).map_err(SdsError)?; } if self.id != data_descriptor_id( diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 571384571..4febef66b 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -323,6 +323,9 @@ mod tests { invalid_labels.table_population = None; invalid_labels.spatial_filter = "job=\"requests\"".into(); assert!(SummaryCatalog::from_materializations(1, 1, &[invalid_labels]).is_err()); + let mut invalid_time = requests.clone(); + invalid_time.table_timestamp_column = Some("time; DROP TABLE samples".into()); + assert!(SummaryCatalog::from_materializations(1, 1, &[invalid_time]).is_err()); let catalog = SummaryCatalog::from_materializations(1, 1, &[requests, errors, other_time]).unwrap(); assert_eq!(catalog.data_descriptors.len(), 3); diff --git a/crates/asap_types/src/table_population.rs b/crates/asap_types/src/table_population.rs index dc91a3366..946813cf4 100644 --- a/crates/asap_types/src/table_population.rs +++ b/crates/asap_types/src/table_population.rs @@ -20,16 +20,7 @@ pub struct TableColumnPredicate { impl TablePopulation { pub fn validate(&self) -> Result<(), String> { for predicate in &self.predicates { - if predicate.column.is_empty() - || !predicate.column.as_bytes()[0].is_ascii_alphabetic() - && !predicate.column.starts_with('_') - || !predicate - .column - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_') - { - return Err("table population column is not an unqualified identifier".into()); - } + validate_column_name(&predicate.column)?; if !matches!( predicate.operator, CompareOpKind::Eq @@ -66,6 +57,18 @@ impl TablePopulation { } } +pub(crate) fn validate_column_name(column: &str) -> Result<(), String> { + if column.is_empty() + || !column.as_bytes()[0].is_ascii_alphabetic() && !column.starts_with('_') + || !column + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_') + { + return Err("table column is not an unqualified identifier".into()); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; From 56778985f6841c054d1bb6979c38b3843cb0b65f Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 18:24:34 -0600 Subject: [PATCH 12/14] Validate SQL timestamp identifiers before catalog publication --- crates/asap_types/src/aggregation_config.rs | 1 + crates/asap_types/src/sds.rs | 1 + crates/asap_types/src/summary_catalog.rs | 3 +++ crates/asap_types/src/table_population.rs | 23 ++++++++++++--------- 4 files changed, 18 insertions(+), 10 deletions(-) diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index e22482f2d..e2411a565 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -182,6 +182,7 @@ impl PrecomputeMaterialization { if self.table_name.is_none() || column.is_empty() { return Err("table timestamp projection requires a table and a column".into()); } + crate::table_population::validate_column_name(column)?; } if self.table_name.is_some() && !self.spatial_filter.is_empty() { return Err("table populations cannot use a PromQL label filter".into()); diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index c5566a609..1a137e360 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -821,6 +821,7 @@ impl DataDescriptor { "table timestamp projection requires a table and a column".into(), )); } + crate::table_population::validate_column_name(column).map_err(SdsError)?; } if self.id != data_descriptor_id( diff --git a/crates/asap_types/src/summary_catalog.rs b/crates/asap_types/src/summary_catalog.rs index 571384571..4febef66b 100644 --- a/crates/asap_types/src/summary_catalog.rs +++ b/crates/asap_types/src/summary_catalog.rs @@ -323,6 +323,9 @@ mod tests { invalid_labels.table_population = None; invalid_labels.spatial_filter = "job=\"requests\"".into(); assert!(SummaryCatalog::from_materializations(1, 1, &[invalid_labels]).is_err()); + let mut invalid_time = requests.clone(); + invalid_time.table_timestamp_column = Some("time; DROP TABLE samples".into()); + assert!(SummaryCatalog::from_materializations(1, 1, &[invalid_time]).is_err()); let catalog = SummaryCatalog::from_materializations(1, 1, &[requests, errors, other_time]).unwrap(); assert_eq!(catalog.data_descriptors.len(), 3); diff --git a/crates/asap_types/src/table_population.rs b/crates/asap_types/src/table_population.rs index dc91a3366..946813cf4 100644 --- a/crates/asap_types/src/table_population.rs +++ b/crates/asap_types/src/table_population.rs @@ -20,16 +20,7 @@ pub struct TableColumnPredicate { impl TablePopulation { pub fn validate(&self) -> Result<(), String> { for predicate in &self.predicates { - if predicate.column.is_empty() - || !predicate.column.as_bytes()[0].is_ascii_alphabetic() - && !predicate.column.starts_with('_') - || !predicate - .column - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_') - { - return Err("table population column is not an unqualified identifier".into()); - } + validate_column_name(&predicate.column)?; if !matches!( predicate.operator, CompareOpKind::Eq @@ -66,6 +57,18 @@ impl TablePopulation { } } +pub(crate) fn validate_column_name(column: &str) -> Result<(), String> { + if column.is_empty() + || !column.as_bytes()[0].is_ascii_alphabetic() && !column.starts_with('_') + || !column + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_') + { + return Err("table column is not an unqualified identifier".into()); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; From bbe7505abb4af9dc8aeb51d29021a1ca53be0d96 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 18:25:21 -0600 Subject: [PATCH 13/14] Document SQL source fields in policy identity --- crates/asap_types/src/policy_fingerprint.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 53acfc573..44d50669e 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -28,8 +28,9 @@ //! semantics — two policies with the same shape but different //! retention are *the same policy* for ingest/query routing //! purposes; retention is a separate concern). -//! - `table_name` / `value_column` (SQL-mode wire shape; folded into -//! `metric` upstream for time-series mode). +//! SQL source table, value projection, timestamp projection, and typed +//! population are included explicitly; the output metric is not a substitute +//! for these source semantics. //! //! ## Hash function //! From 4d4fbdcc62282c5dd022ca472d139806b49c96be Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 18:28:18 -0600 Subject: [PATCH 14/14] Persist selected SQL DAGs with shared backend placement bindings --- control_plane/src/clickhouse.rs | 79 ++++++++++++------- control_plane/src/physical/compiler.rs | 63 +++++---------- .../src/physical/executable_binding.rs | 41 ++++++++++ control_plane/src/query_plan.rs | 32 +++++++- .../query-engine/clickhouse-sql-support.md | 7 ++ 5 files changed, 147 insertions(+), 75 deletions(-) diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 66e8aa186..6486063e6 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -118,6 +118,7 @@ pub async fn compile_clickhouse_workload( tables: request.tables.clone(), }; let mut entries = std::collections::BTreeMap::new(); + let mut installed_dags = std::collections::BTreeMap::new(); for query in &request.queries { let planned = plan_clickhouse_sql(&query.sql, &catalog, request.accuracy.clone()).await?; let PhysicalExpr::Committed(crate::physical::post_asap::PostAsapPlan::Summary(root)) = @@ -127,7 +128,11 @@ pub async fn compile_clickhouse_workload( "SQL did not produce a summary DAG".into(), )); }; - let executable = QueryPlanEntry::compile_bound_relational( + let semantic = planner_types::post_asap::compile_executable_dag_with_node_ids(&root) + .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + let mut materialization_nodes = std::collections::BTreeMap::new(); + let mut query_nodes = std::collections::BTreeMap::new(); + let executable = QueryPlanEntry::compile_bound_relational_mapped( query.sql.clone(), planned.canonical_sql.clone(), &root, @@ -142,9 +147,34 @@ pub async fn compile_clickhouse_workload( cumulative_readout: query.cumulative, }, FallbackPolicy::ExactBackend, - |node, family| bind_selected_node(node, family, query, request), + |node, family| { + let binding = bind_selected_node(node, family, query, request)?; + let id = semantic.node_ids.node_id(node).ok_or_else(|| { + crate::query_plan::QueryPlanError::Invalid( + "selected SQL node is absent from semantic DAG".into(), + ) + })?; + materialization_nodes.insert(id, binding.materialization); + Ok(binding) + }, + |node, query_node| { + if let Some(id) = semantic.node_ids.node_id(node) { + query_nodes.insert(id, query_node); + } + }, ) .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?; + let installed = crate::physical::executable_binding::install_selected_dag( + query.sql.clone(), + &semantic.dag, + executable.root, + |id| materialization_nodes.get(&id).copied(), + |id| query_nodes.get(&id).copied(), + ) + .map_err(ClickHousePlanningError::Lower)?; + crate::physical::executable_binding::validate_query_plan(&installed, &executable) + .map_err(ClickHousePlanningError::Lower)?; + installed_dags.insert(query.sql.clone(), installed); if executable .nodes .values() @@ -154,32 +184,6 @@ pub async fn compile_clickhouse_workload( "compiled SQL contains an unsupported operator; publication refused".into(), )); } - let bindings = executable.materialization_bindings(); - let identities = bindings - .iter() - .map(|binding| { - request - .sds - .materializations - .get(&binding.materialization) - .ok_or_else(|| { - ClickHousePlanningError::Lower( - "compiled SQL binding is absent from SDS".into(), - ) - }) - }) - .collect::, _>>()?; - // Descriptor references are already represented by each DAG's - // MaterializationBinding and validated through SummaryCatalog. - let _descriptor_ids = identities - .iter() - .map(|identity| { - ( - &identity.summary_descriptor_id, - &identity.data_descriptor_id, - ) - }) - .collect::>(); let identity = QueryPlan::catalog_key(QueryLanguage::ClickHouseSql, &planned.canonical_sql); if entries.insert(identity.clone(), executable).is_some() { return Err(ClickHousePlanningError::Lower(format!( @@ -187,9 +191,11 @@ pub async fn compile_clickhouse_workload( ))); } } + let mut precompute_plan = request.precompute_plan.clone(); + precompute_plan.executable_dags = installed_dags; let publication = crate::physical::publication::PhysicalPlanPublication { summary_catalog: request.sds.clone(), - precompute_plan: request.precompute_plan.clone(), + precompute_plan, collector_plans: Vec::new(), transmission_plan: request.transmission_plan.clone(), query_plan: QueryPlan { @@ -632,6 +638,21 @@ mod tests { }], }; let publication = compile_clickhouse_workload(&request).await.unwrap(); + let installed = publication + .precompute_plan + .executable_dags + .get(&request.queries[0].sql) + .unwrap(); + installed.validate().unwrap(); + assert_eq!(installed.binding.precompute_sinks.len(), 1); + assert_eq!( + installed.binding.nodes.len(), + installed.document.nodes.len() + ); + assert!(installed.binding.nodes.values().any(|binding| matches!( + binding, + crate::physical::executable_binding::BackendNodeBinding::Query { .. } + ))); let entry = publication.query_plan.entries.values().next().unwrap(); assert!(entry .nodes diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 83b60d836..5de2565fc 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -2264,52 +2264,25 @@ impl PhysicalCompiler { for (query_index, compiled) in executable_dags.iter().enumerate() { let Some(compiled) = compiled else { continue }; let query_id = request.queries[query_index].query_id.clone(); - let mut placements = BTreeMap::new(); - let mut precompute_sinks = Vec::new(); - for node in &compiled.dag.nodes { - let placement = - if let Some(definition) = node_bindings.get(&(query_index, node.id)).copied() { - precompute_sinks.push(node.id); - crate::physical::executable_binding::BackendNodeBinding::Materialization { - summary_definition: definition.into(), - } - } else if node.output_state.timing - == planner_types::post_asap::ExecutionTiming::MaintenanceTime - { - super::executable_binding::BackendNodeBinding::MaintenanceInput - } else { - match query_node_bindings.get(&(query_index, node.id)).copied() { - Some(query_node) => { - super::executable_binding::BackendNodeBinding::Query { query_node } - } - None => super::executable_binding::BackendNodeBinding::QueryInput, - } - }; - placements.insert(node.id, placement); - } - precompute_sinks.sort(); - let installed = crate::physical::executable_binding::InstalledPostAsapDag { - document: super::executable_binding::OwnedPostAsapDag::from_executable( - query_id.clone(), - &compiled.dag, - ) - .map_err(|reason| CompileError::Query { - query_id: query_id.clone(), - reason, - })?, - binding: super::executable_binding::BackendExecutableBinding { - nodes: placements, - query_sink: compiled.dag.root, - query_plan_sink: query_plan - .entries - .values() - .find(|entry| entry.query_id == query_id) - .expect("compiled query entry exists") - .root, - precompute_sinks, + let query_plan_sink = query_plan + .entries + .values() + .find(|entry| entry.query_id == query_id) + .expect("compiled query entry exists") + .root; + let installed = super::executable_binding::install_selected_dag( + query_id.clone(), + &compiled.dag, + query_plan_sink, + |id| { + node_bindings + .get(&(query_index, id)) + .copied() + .map(Into::into) }, - }; - installed.validate().map_err(|reason| CompileError::Query { + |id| query_node_bindings.get(&(query_index, id)).copied(), + ) + .map_err(|reason| CompileError::Query { query_id: query_id.clone(), reason, })?; diff --git a/control_plane/src/physical/executable_binding.rs b/control_plane/src/physical/executable_binding.rs index 6aff751f5..28f82a57c 100644 --- a/control_plane/src/physical/executable_binding.rs +++ b/control_plane/src/physical/executable_binding.rs @@ -2,6 +2,47 @@ pub use asap_types::executable_plan::*; +/// Assign backend phases to a selected semantic DAG without changing its nodes. +pub fn install_selected_dag( + query_id: String, + dag: &planner_types::post_asap::ExecutableDag, + query_plan_sink: QueryNodeId, + materialization: impl Fn( + planner_types::post_asap::PostAsapNodeId, + ) -> Option, + query_node: impl Fn(planner_types::post_asap::PostAsapNodeId) -> Option, +) -> Result { + let mut nodes = std::collections::BTreeMap::new(); + let mut precompute_sinks = Vec::new(); + for node in &dag.nodes { + let binding = if let Some(summary_definition) = materialization(node.id) { + precompute_sinks.push(node.id); + BackendNodeBinding::Materialization { summary_definition } + } else if node.output_state.timing + == planner_types::post_asap::ExecutionTiming::MaintenanceTime + { + BackendNodeBinding::MaintenanceInput + } else { + query_node(node.id).map_or(BackendNodeBinding::QueryInput, |query_node| { + BackendNodeBinding::Query { query_node } + }) + }; + nodes.insert(node.id, binding); + } + precompute_sinks.sort(); + let installed = InstalledPostAsapDag { + document: OwnedPostAsapDag::from_executable(query_id, dag)?, + binding: BackendExecutableBinding { + nodes, + query_sink: dag.root, + query_plan_sink, + precompute_sinks, + }, + }; + installed.validate()?; + Ok(installed) +} + pub fn validate_query_plan( installed: &InstalledPostAsapDag, query: &crate::query_plan::QueryPlanEntry, diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index ff9d0e78b..1fbf7827f 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -402,6 +402,34 @@ impl QueryPlanEntry { } pub fn compile_bound_relational( + query_id: String, + canonical_query: String, + root: &Rc, + fixed_evaluation: FixedEvaluationRange, + instant: InstantExecution, + fallback: FallbackPolicy, + bind: F, + ) -> Result + where + F: FnMut( + &Rc, + &SummaryFamilyType, + ) -> Result, + { + Self::compile_bound_relational_mapped( + query_id, + canonical_query, + root, + fixed_evaluation, + instant, + fallback, + bind, + |_, _| {}, + ) + } + + /// Preserve Planner-to-runtime node identities for installed SQL DAGs. + pub fn compile_bound_relational_mapped( query_id: String, canonical_query: String, root: &Rc, @@ -409,12 +437,14 @@ impl QueryPlanEntry { instant: InstantExecution, fallback: FallbackPolicy, mut bind: F, + mut lowered: G, ) -> Result where F: FnMut( &Rc, &SummaryFamilyType, ) -> Result, + G: FnMut(&Rc, QueryNodeId), { let mut compiler = DagCompiler { next_id: 0, @@ -423,7 +453,7 @@ impl QueryPlanEntry { bind: &mut bind, logical_source: None, preserve_relational: true, - lowered: None, + lowered: Some(&mut lowered), }; let root = compiler.lower(root)?; Ok(Self { diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 3668aca7b..e2d62f534 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -41,6 +41,13 @@ pane duration, and `pane_origin_ms` through the authoritative SummaryCatalog. Any invalid SQL entry rejects the complete candidate snapshot before activation; the active generation remains unchanged. +SQL compilation also retains the selected Planner semantic DAG in +`PrecomputePlan.executable_dags`. The compiler records materialization and query +node bindings during lowering and assigns phases with the same placement builder +as PromQL. Planner node IDs remain distinct from SummaryDefinitionId and +QueryNodeId. This preserves the actual selected DAG across publication instead +of reconstructing it from materialization configs later. + The query listener snapshots `HotReloadActivePhysicalPlan` once per request. It uses the SQL parsing context and the matching `QueryPlanEntry` from that same snapshot, reads SummaryStore state, executes relational operators, and