feat: bind SQL populations and projections through SummaryCatalog

control_plane/src/clickhouse.rs

@@ -214,8 +214,8 @@ fn bind_selected_node(
     query: &ClickHouseSqlWorkloadEntry,
     request: &ClickHouseSqlWorkload,
 ) -> Result<MaterializationBinding, crate::query_plan::QueryPlanError> {
-    let (table_ref, value_column, source_window, spatial_filter) =
-        clickhouse_materialization_leaf_contract(node)
+    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);
     let selected = select_materialization(
@@ -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()),
@@ -238,7 +243,9 @@ fn bind_selected_node(
 
 fn clickhouse_materialization_leaf_contract(
     node: &planner_types::post_asap::SummaryNode,
-) -> Result<(String, String, Option<u64>, String), String> {
+    evaluation_start_ms: u64,
+    evaluation_end_ms: u64,
+) -> Result<(String, String, Option<u64>, String, String), String> {
     use planner_types::{
         post_asap::SummaryExpr,
         pre_asap::{CompareOpKind, QueryExpr, ScalarValue, Source},
@@ -298,6 +305,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 +332,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 +349,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 +378,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 +404,13 @@ fn clickhouse_materialization_leaf_contract(
         table_ref.to_owned(),
         value_column,
         Some(window_secs),
-        String::new(),
+        population.canonical(),
+        schema
+            .time_index
+            .and_then(|index| schema.columns.get(index))
+            .ok_or("SQL summary source has no timestamp projection")?
+            .name
+            .clone(),
     ))
 }
 
@@ -374,7 +425,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()
@@ -427,6 +478,7 @@ mod tests {
             Some(value_column.into()),
         );
         value.pane_origin_ms = Some(0);
+        value.table_timestamp_column = Some("timestamp_ms".into());
         value
     }
 
@@ -553,7 +605,7 @@ mod tests {
                 vec![],
             )
         };
-        let request = ClickHouseSqlWorkload {
+        let mut request = ClickHouseSqlWorkload {
             sds,
             precompute_plan: precompute,
             transmission_plan: transmission,
@@ -604,5 +656,59 @@ mod tests {
                 Ok(planner_types::post_asap::ValueOperation::Project { .. })
             )
         }));
+        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");
+        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());
     }
 }

crates/asap_types/src/aggregation_config.rs

@@ -100,6 +100,8 @@ pub struct PrecomputeMaterialization {
     pub aggregation_sub_type: String,
     pub parameters: HashMap<String, Value>,
     pub grouping_labels: KeyByLabelNames,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub partitioning: Option<crate::sds::PopulationPartitioning>,
     pub aggregated_labels: KeyByLabelNames,
     pub rollup_labels: KeyByLabelNames,
     pub original_yaml: String,
@@ -126,6 +128,19 @@ pub struct PrecomputeMaterialization {
     // SQL-specific fields (optional, used when query_language=sql)
     pub table_name: Option<String>,   // SQL mode: table name
     pub value_column: Option<String>, // 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<String>,
+    #[serde(
+        default,
+        alias = "tablePopulation",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub table_population: Option<crate::table_population::TablePopulation>,
 }
 
 /// Policy-match handles for both the key and value dimensions of a
@@ -162,6 +177,29 @@ impl AggregationIdInfo {
 pub type AggregationConfig = PrecomputeMaterialization;
 
 impl PrecomputeMaterialization {
+    pub fn population_filter_canonical(&self) -> Result<String, String> {
+        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());
+            }
+            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());
+        }
+        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,
@@ -189,6 +227,7 @@ impl PrecomputeMaterialization {
             aggregation_sub_type,
             parameters,
             grouping_labels,
+            partitioning: None,
             aggregated_labels,
             rollup_labels,
             original_yaml,
@@ -209,6 +248,8 @@ impl PrecomputeMaterialization {
             num_aggregates_to_retain,
             table_name,
             value_column,
+            table_population: None,
+            table_timestamp_column: None,
         }
     }
 
@@ -317,7 +358,25 @@ 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_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"))
+            .filter(|value| !value.is_null())
+            .cloned()
+            .map(serde_json::from_value)
+            .transpose()?;
+        config.population_filter_canonical()?;
         Ok(config)
     }
 
@@ -456,7 +515,27 @@ 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_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"))
+            .filter(|value| !value.is_null())
+            .cloned()
+            .map(serde_yaml::from_value)
+            .transpose()?;
+        config
+            .population_filter_canonical()
+            .map_err(anyhow::Error::msg)?;
         Ok(config)
     }
 }
@@ -469,6 +548,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,
@@ -492,6 +572,12 @@ 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);
+        }
+        if let Some(ref column) = self.table_timestamp_column {
+            json["tableTimestampColumn"] = serde_json::json!(column);
+        }
 
         json
     }

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;