diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs
index 34669ca8b..66e8aa186 100644
--- a/control_plane/src/clickhouse.rs
+++ b/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());
     }
 }
diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs
index 9f16833df..e2411a565 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<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
     }
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..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
 //!
@@ -114,6 +115,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 {
@@ -171,6 +176,25 @@ 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() {
+                buf.push(0);
+                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 48fd1c411..26c5e1e9a 100644
--- a/crates/asap_types/src/precompute_plan/catalog.rs
+++ b/crates/asap_types/src/precompute_plan/catalog.rs
@@ -86,15 +86,19 @@ impl PrecomputePlan {
                 .map_or(ValueProjectionIdentity::SampleValue, |name| {
                     ValueProjectionIdentity::Column { name: name.clone() }
                 });
-            if data.source != expected_source
+            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
-                    != 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/sds.rs b/crates/asap_types/src/sds.rs
index b54c50299..1a137e360 100644
--- a/crates/asap_types/src/sds.rs
+++ b/crates/asap_types/src/sds.rs
@@ -698,14 +698,28 @@ 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 {
     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<String>,
     pub population_filter_canonical: String,
     pub group_by_keys: BTreeSet<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub partitioning: Option<PopulationPartitioning>,
     /// Versioned contract for timestamp interpretation and
     /// missing/duplicate/invalid observation handling.
     pub observation_semantics: String,
@@ -765,20 +779,50 @@ impl DataDescriptor {
             &population_filter_canonical,
             &group_by_keys,
             &observation_semantics,
+            None,
+            None,
         );
         Self {
             id,
             source,
             value_projection,
+            timestamp_column: None,
             population_filter_canonical,
             group_by_keys,
+            partitioning: None,
             observation_semantics,
         }
     }
+    pub fn with_partitioning(mut self, partitioning: Option<PopulationPartitioning>) -> 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.timestamp_column.as_deref(),
+        );
+        self
+    }
+    pub fn with_timestamp_column(mut self, column: Option<String>) -> 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(),
+                ));
+            }
+            crate::table_population::validate_column_name(column).map_err(SdsError)?;
+        }
         if self.id
             != data_descriptor_id(
                 &self.source,
@@ -786,6 +830,8 @@ impl DataDescriptor {
                 &self.population_filter_canonical,
                 &self.group_by_keys,
                 &self.observation_semantics,
+                self.partitioning,
+                self.timestamp_column.as_deref(),
             )
         {
             return Err(SdsError("data descriptor ID/content mismatch".into()));
@@ -799,6 +845,8 @@ fn data_descriptor_id(
     filter: &str,
     group_by: &BTreeSet<String>,
     observation_semantics: &str,
+    partitioning: Option<PopulationPartitioning>,
+    timestamp_column: Option<&str>,
 ) -> DataDescriptorId {
     // Length framing keeps distinct typed sources, projections, predicates,
     // and grouping keys collision-free in the content identity.
@@ -811,6 +859,12 @@ fn data_descriptor_id(
         projection.len(),
         filter.len()
     );
+    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()));
     }
@@ -1286,3 +1340,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::<String>::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 90095a330..4febef66b 100644
--- a/crates/asap_types/src/summary_catalog.rs
+++ b/crates/asap_types/src/summary_catalog.rs
@@ -120,10 +120,14 @@ 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",
-                );
+                )
+                .with_partitioning(config.partitioning)
+                .with_timestamp_column(config.table_timestamp_column.clone());
                 Ok((
                     config.policy_fingerprint(),
                     summary,
@@ -279,6 +283,55 @@ 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 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 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 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);
+        assert_eq!(catalog.materializations.len(), 3);
+    }
+
     #[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..946813cf4
--- /dev/null
+++ b/crates/asap_types/src/table_population.rs
@@ -0,0 +1,131 @@
+//! 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<TableColumnPredicate>,
+}
+
+#[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 {
+            validate_column_name(&predicate.column)?;
+            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(","))
+    }
+}
+
+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::*;
+
+    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());
+    }
+}
diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
index ad7ad5ac6..f3da96a76 100644
--- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs
+++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs
@@ -858,6 +858,9 @@ mod tests {
             num_aggregates_to_retain: None,
             table_name: None,
             value_column: None,
+            table_population: None,
+            table_timestamp_column: None,
+            partitioning: None,
         };
         let policy_fp = aggregation.policy_fp_u64();
         let streaming = StreamingConfig::new(HashMap::from([(policy_fp, aggregation)]));
@@ -911,6 +914,9 @@ mod tests {
                 num_aggregates_to_retain: Some(80),
                 table_name: None,
                 value_column: None,
+                table_population: None,
+                table_timestamp_column: 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..897c25a1a 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -3673,6 +3673,9 @@ aggregations:
                 num_aggregates_to_retain: None,
                 table_name: None,
                 value_column: None,
+                table_population: None,
+                table_timestamp_column: 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..768a08291 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/data_plane/src/precompute_engine/output_sink.rs
@@ -344,6 +344,9 @@ mod tests {
             num_aggregates_to_retain: None,
             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 95036c7f8..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
@@ -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<asap_types::table_population::TablePopulation>,
+    output_metric: Option<String>,
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum ClickHouseLabels {
+    Series(String),
+    Map(std::collections::BTreeMap<String, String>),
 }
 
 #[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::<Vec<_>>()
+                    .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,7 +133,7 @@ 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 {
@@ -97,9 +145,21 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor
             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<dyn RawSampleReader>)
+            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 = Some(materialization.table_population.clone().unwrap_or_default());
+            reader.output_metric = Some(materialization.metric.clone());
+            Ok(Arc::new(reader) as Arc<dyn RawSampleReader>)
         }
-        source => fallback(source),
+        source => fallback(source, materialization),
     })
 }
 
@@ -123,6 +183,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());
         }
@@ -150,7 +223,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::<Vec<_>>()
+                            .join(",");
+                        if labels.is_empty() {
+                            metric.into()
+                        } else {
+                            format!("{metric}{{{labels}}}")
+                        }
+                    }
+                },
                 timestamp_ms: row.timestamp_ms,
                 value: row.value,
             };
@@ -193,24 +294,76 @@ 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 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 mut 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(),
-        })
+        materialization.table_timestamp_column = Some("timestamp_ms".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: "metrics".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<RawSample>`. 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<RawSample>`. 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<Item = RawSample>`
-//!   — 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<dyn RawSampleReader>`. 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<Arc<dyn RawSampleReader>, Box<dyn std::error::Error + Send + Sync>>
         + 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<dyn RawSampleReader>)
@@ -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<Mutex<Vec<String>>> = 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<dyn RawSampleReader>)
         });
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..557b8326a 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,9 @@ mod tests {
             num_aggregates_to_retain: None,
             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 b820bdc63..6f742df94 100644
--- a/data_plane/src/tests/test_utilities/engine_factories.rs
+++ b/data_plane/src/tests/test_utilities/engine_factories.rs
@@ -108,6 +108,9 @@ pub fn create_engine_single_pop_with_aggregated(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        table_population: None,
+        table_timestamp_column: None,
+        partitioning: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
@@ -202,6 +205,9 @@ pub fn create_engine_dual_input(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        table_population: None,
+        table_timestamp_column: None,
+        partitioning: None,
     };
     let value_id = value_agg_config.policy_fp_u64();
     aggregation_configs.insert(value_id, value_agg_config);
@@ -226,6 +232,9 @@ pub fn create_engine_dual_input(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        table_population: None,
+        table_timestamp_column: None,
+        partitioning: None,
     };
     let keys_id = keys_agg_config.policy_fp_u64();
     aggregation_configs.insert(keys_id, keys_agg_config);
@@ -315,6 +324,9 @@ pub fn create_engine_two_metrics(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        table_population: None,
+        table_timestamp_column: None,
+        partitioning: None,
     };
     let id_a = agg_config_a.policy_fp_u64();
     aggregation_configs.insert(id_a, agg_config_a);
@@ -338,6 +350,9 @@ pub fn create_engine_two_metrics(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        table_population: None,
+        table_timestamp_column: None,
+        partitioning: None,
     };
     let id_b = agg_config_b.policy_fp_u64();
     aggregation_configs.insert(id_b, agg_config_b);
@@ -437,6 +452,9 @@ pub fn create_engine_three_metrics(
             num_aggregates_to_retain: None,
             table_name: None,
             value_column: None,
+            table_population: None,
+            table_timestamp_column: None,
+            partitioning: None,
         };
         let id = cfg.policy_fp_u64();
         ids.push(id);
@@ -513,6 +531,9 @@ pub fn create_engine_multi_timestamp(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        table_population: None,
+        table_timestamp_column: None,
+        partitioning: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
@@ -581,6 +602,9 @@ pub fn create_engine_multi_timestamp_with_window(
         num_aggregates_to_retain: None,
         table_name: None,
         value_column: None,
+        table_population: None,
+        table_timestamp_column: None,
+        partitioning: None,
     };
     let agg_id = agg_config.policy_fp_u64();
     aggregation_configs.insert(agg_id, agg_config);
diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs
index 7868c12b8..84de688ee 100644
--- a/data_plane/tests/clickhouse_differential_e2e.rs
+++ b/data_plane/tests/clickhouse_differential_e2e.rs
@@ -81,6 +81,14 @@ 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(),
+            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 +116,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 +211,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 +269,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..3668aca7b 100644
--- a/docs/developer_docs/query-engine/clickhouse-sql-support.md
+++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md
@@ -78,6 +78,37 @@ 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.
+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.
+
+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