fix: give ClickHouse backfill jobs a typed table source

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<String>,
     #[arg(

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();
@@ -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(),
@@ -792,8 +794,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,
         );

data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs

@@ -82,15 +82,22 @@ 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<dyn RawSampleReader>)
+        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();
+            Ok(Arc::new(ClickHouseReader::new(source_config)?) as Arc<dyn RawSampleReader>)
         }
         source => fallback(source),
     })
@@ -187,12 +194,23 @@ 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: "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: "metrics".into(),
+            table: "samples; DROP TABLE x".into(),
+        })
+        .is_err());
     }
 }

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(),
             },

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) => {
@@ -301,23 +317,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<dyn RawSampleReader>)
         }
-        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()),
-    }
     })
 }
 
@@ -384,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(&registry, 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");

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")
@@ -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()

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

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.