diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs
index 0bc72d2ca..bbb861bc1 100644
--- a/data_plane/src/main.rs
+++ b/data_plane/src/main.rs
@@ -110,8 +110,7 @@ struct Args {
     #[arg(long, env = "ASAP_CLICKHOUSE_DATABASE", default_value = "default")]
     clickhouse_database: String,
 
-    /// Enable ClickHouse as a source for queued backfill jobs whose source URL
-    /// is `clickhouse://configured`.
+    /// Enable the configured ClickHouse connection for typed table backfill jobs.
     #[arg(long, env = "ASAP_CLICKHOUSE_BACKFILL_TABLE")]
     clickhouse_backfill_table: Option<String>,
     #[arg(
diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs
index 32e9720c3..ee731976a 100644
--- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs
+++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs
@@ -441,8 +441,8 @@ mod tests {
             String::new(),
             "requests".into(),
             None,
-            None,
-            None,
+            Some("asap_e2e.samples".into()),
+            Some("value".into()),
         );
         config.pane_origin_ms = Some(0);
         let sds = SummaryCatalog::from_materializations(41, 1, &[config.clone()]).unwrap();
@@ -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,
         );
diff --git a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs
index 63354546d..95036c7f8 100644
--- a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs
+++ b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs
@@ -82,15 +82,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());
     }
 }
diff --git a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs
index 0c3a6c69a..d8d15b012 100644
--- a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs
@@ -70,6 +70,9 @@ pub enum BackfillSource {
     /// Prometheus (or VictoriaMetrics / Thanos / Cortex) via the
     /// HTTP range-query API.
     Prometheus { url: String },
+    /// ClickHouse table read through the deployment's configured connection.
+    /// The source identity belongs to the job; credentials remain local.
+    ClickHouse { database: String, table: String },
     /// Rebuild from a different sketch already in the store. Used for
     /// lossless schema widenings (e.g. CMS(256) → CMS(2048)) where
     /// the source sketch is a strict subset of the target's
@@ -1120,6 +1123,10 @@ mod tests {
     #[test]
     fn backfill_source_roundtrips_through_serde() {
         for src in [
+            BackfillSource::ClickHouse {
+                database: "telemetry".into(),
+                table: "samples".into(),
+            },
             BackfillSource::Prometheus {
                 url: "u".to_string(),
             },
diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs
index ede452244..b2644605a 100644
--- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs
+++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs
@@ -199,6 +199,22 @@ impl BackfillService {
                 "BackfillService picking up queued job"
             );
 
+            if let BackfillSource::ClickHouse { database, table } = &job.source {
+                let snapshot = self.config_source.snapshot();
+                let configured_table = snapshot
+                    .get_aggregation_config(job.agg_id)
+                    .and_then(|config| config.table_name.as_deref());
+                let qualified_table = format!("{database}.{table}");
+                if configured_table != Some(table.as_str())
+                    && configured_table != Some(qualified_table.as_str())
+                {
+                    self.registry.mark_failed(
+                        job.job_id,
+                        "ClickHouse source differs from the installed materialization table",
+                    );
+                    continue;
+                }
+            }
             let reader = match (self.reader_factory)(&job.source) {
                 Ok(r) => r,
                 Err(e) => {
@@ -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");
diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs
index 88b3fb24b..7868c12b8 100644
--- a/data_plane/tests/clickhouse_differential_e2e.rs
+++ b/data_plane/tests/clickhouse_differential_e2e.rs
@@ -257,7 +257,7 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() {
         .arg("--clickhouse-url")
         .arg(&clickhouse_url)
         .arg("--clickhouse-backfill-table")
-        .arg("telemetry")
+        .arg("deployment_default_not_the_job_table")
         .arg("--clickhouse-backfill-database")
         .arg("default")
         .arg("--enable-backfill-worker")
@@ -317,7 +317,7 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() {
             "agg_id": config.policy_fp_u64(),
             "start_ms": 0,
             "end_ms": 2000,
-            "source": {"Prometheus": {"url": "clickhouse://configured"}},
+            "source": {"ClickHouse": {"database": "default", "table": "telemetry"}},
             "windows_total": 1
         }))
         .send()
diff --git a/data_plane/tests/clickhouse_q05_process_e2e.rs b/data_plane/tests/clickhouse_q05_process_e2e.rs
index 937d75b05..35b0d04b7 100644
--- a/data_plane/tests/clickhouse_q05_process_e2e.rs
+++ b/data_plane/tests/clickhouse_q05_process_e2e.rs
@@ -271,7 +271,7 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() {
         .post(format!("{base}/api/v1/db/backfill"))
         .json(
             &serde_json::json!({"agg_id":agg_id,"start_ms":start_ms,"end_ms":end_ms,
-            "source":{"Prometheus":{"url":"clickhouse://configured"}},"windows_total":1}),
+            "source":{"ClickHouse":{"database":"asap_q05_e2e","table":"q05_samples"}},"windows_total":1}),
         )
         .send()
         .await
diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md
index 4bfd87bbf..c74f803dd 100644
--- a/docs/developer_docs/query-engine/clickhouse-sql-support.md
+++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md
@@ -74,7 +74,7 @@ pane coverage, and execution errors fail closed to exact ClickHouse. The proxy
 preserves the upstream status, safe headers, and response body.
 
 ClickHouse can also provide samples to the queued backfill service through the
-explicit `clickhouse://configured` source marker. Backfill populates the same
+typed `ClickHouse { database, table }` source. Backfill populates the same
 SummaryStore instances used by other ingest sources; it does not introduce a
 second storage or catalog lifecycle.