From 2ce35c4bb2af53ab2c9931ec5ab8e98400b0aedf Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:21:32 -0600 Subject: [PATCH 1/6] fix: give ClickHouse backfill a typed table source --- data_plane/src/main.rs | 3 +-- .../accelerator.rs | 5 ++-- .../sketch_db/backfill/clickhouse_reader.rs | 24 ++++++++++++------- .../storage_engines/sketch_db/backfill/mod.rs | 7 ++++++ .../sketch_db/backfill/service.rs | 11 ++++----- .../tests/clickhouse_differential_e2e.rs | 2 +- .../tests/clickhouse_q05_process_e2e.rs | 2 +- .../query-engine/clickhouse-sql-support.md | 2 +- 8 files changed, 34 insertions(+), 22 deletions(-) diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 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, #[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..d489fbedb 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -792,8 +792,9 @@ mod tests { let job = registry.create( cfg.policy_fp_u64(), (0, 2_000), - crate::storage_engines::sketch_db::backfill::BackfillSource::Prometheus { - url: "clickhouse://configured".into(), + crate::storage_engines::sketch_db::backfill::BackfillSource::ClickHouse { + database: "asap_e2e".into(), + table: "samples".into(), }, 2, ); diff --git a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs index 63354546d..5ed8784bc 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs @@ -82,15 +82,15 @@ impl ClickHouseReader { } } -/// Adds ClickHouse to the existing backfill lifecycle without extending the -/// shared `BackfillSource` enum. Jobs opt in with the reserved -/// `Prometheus { url: "clickhouse://configured" }` source marker; all other -/// sources retain the default factory behavior. +/// Resolve a typed table source using deployment-local connection settings. pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactory { let fallback = super::service::default_reader_factory(); Arc::new(move |source| match source { - BackfillSource::Prometheus { url } if url == "clickhouse://configured" => { - Ok(Arc::new(ClickHouseReader::new(config.clone())?) as Arc) + BackfillSource::ClickHouse { database, table } => { + let mut source_config = config.clone(); + source_config.database = database.clone(); + source_config.table = table.clone(); + Ok(Arc::new(ClickHouseReader::new(source_config)?) as Arc) } source => fallback(source), }) @@ -187,12 +187,18 @@ mod tests { } #[test] - fn configured_source_marker_enters_clickhouse_backfill_lifecycle() { + fn typed_source_enters_clickhouse_backfill_lifecycle() { let factory = clickhouse_reader_factory(config("samples")); - let reader = factory(&BackfillSource::Prometheus { - url: "clickhouse://configured".into(), + let reader = factory(&BackfillSource::ClickHouse { + database: "another_database".into(), + table: "another_table".into(), }) .unwrap(); assert_eq!(reader.source_name(), "ClickHouseReader"); + assert!(factory(&BackfillSource::ClickHouse { + database: "default".into(), + table: "samples; DROP TABLE x".into(), + }) + .is_err()); } } diff --git a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs index 0c3a6c69a..d8d15b012 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs @@ -70,6 +70,9 @@ pub enum BackfillSource { /// Prometheus (or VictoriaMetrics / Thanos / Cortex) via the /// HTTP range-query API. Prometheus { url: String }, + /// ClickHouse table read through the deployment's configured connection. + /// The source identity belongs to the job; credentials remain local. + ClickHouse { database: String, table: String }, /// Rebuild from a different sketch already in the store. Used for /// lossless schema widenings (e.g. CMS(256) → CMS(2048)) where /// the source sketch is a strict subset of the target's @@ -1120,6 +1123,10 @@ mod tests { #[test] fn backfill_source_roundtrips_through_serde() { for src in [ + BackfillSource::ClickHouse { + database: "telemetry".into(), + table: "samples".into(), + }, BackfillSource::Prometheus { url: "u".to_string(), }, diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs index ede452244..c22beeaa6 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -301,23 +301,22 @@ pub fn noop_reader_factory() -> ReaderFactory { /// * [`BackfillSource::Prometheus`] — routed to /// [`super::prometheus_reader::PrometheusReader`]. /// -/// All other variants (`S3Gorilla`, `OtherSketch`) return a clear +/// Other variants return a clear /// "not yet implemented" error, which the worker surfaces on /// `BackfillJob::error_message` so the control plane / operator sees /// exactly which reader is missing. pub fn default_reader_factory() -> ReaderFactory { - Arc::new(|source| { - match source { + Arc::new(|source| match source { BackfillSource::Prometheus { url } => { let reader = super::prometheus_reader::PrometheusReader::new(url.clone()); Ok(Arc::new(reader) as Arc) } - BackfillSource::S3Gorilla { .. } + BackfillSource::ClickHouse { .. } + | BackfillSource::S3Gorilla { .. } | BackfillSource::OtherSketch { .. } => Err(format!( - "reader for {source:?} not yet implemented; only Prometheus is wired in-tree as of Phase 5h" + "no reader configured for {source:?}; ClickHouse requires its deployment reader factory" ) .into()), - } }) } diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index 88b3fb24b..58520316f 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -317,7 +317,7 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() { "agg_id": config.policy_fp_u64(), "start_ms": 0, "end_ms": 2000, - "source": {"Prometheus": {"url": "clickhouse://configured"}}, + "source": {"ClickHouse": {"database": "default", "table": "telemetry"}}, "windows_total": 1 })) .send() diff --git a/data_plane/tests/clickhouse_q05_process_e2e.rs b/data_plane/tests/clickhouse_q05_process_e2e.rs index 937d75b05..35b0d04b7 100644 --- a/data_plane/tests/clickhouse_q05_process_e2e.rs +++ b/data_plane/tests/clickhouse_q05_process_e2e.rs @@ -271,7 +271,7 @@ async fn q05_sql_is_planned_backfilled_and_served_warm_by_backend_process() { .post(format!("{base}/api/v1/db/backfill")) .json( &serde_json::json!({"agg_id":agg_id,"start_ms":start_ms,"end_ms":end_ms, - "source":{"Prometheus":{"url":"clickhouse://configured"}},"windows_total":1}), + "source":{"ClickHouse":{"database":"asap_q05_e2e","table":"q05_samples"}},"windows_total":1}), ) .send() .await diff --git a/docs/developer_docs/query-engine/clickhouse-sql-support.md b/docs/developer_docs/query-engine/clickhouse-sql-support.md index 4bfd87bbf..c74f803dd 100644 --- a/docs/developer_docs/query-engine/clickhouse-sql-support.md +++ b/docs/developer_docs/query-engine/clickhouse-sql-support.md @@ -74,7 +74,7 @@ pane coverage, and execution errors fail closed to exact ClickHouse. The proxy preserves the upstream status, safe headers, and response body. ClickHouse can also provide samples to the queued backfill service through the -explicit `clickhouse://configured` source marker. Backfill populates the same +typed `ClickHouse { database, table }` source. Backfill populates the same SummaryStore instances used by other ingest sources; it does not introduce a second storage or catalog lifecycle. From 2fe798c1093dad8c58a4ca64db81a57d131a9947 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:38:18 -0600 Subject: [PATCH 2/6] test: resolve ClickHouse backfill from the job table --- data_plane/tests/clickhouse_differential_e2e.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data_plane/tests/clickhouse_differential_e2e.rs b/data_plane/tests/clickhouse_differential_e2e.rs index 58520316f..7868c12b8 100644 --- a/data_plane/tests/clickhouse_differential_e2e.rs +++ b/data_plane/tests/clickhouse_differential_e2e.rs @@ -257,7 +257,7 @@ async fn compiled_publication_executes_mixed_dag_in_data_plane_process() { .arg("--clickhouse-url") .arg(&clickhouse_url) .arg("--clickhouse-backfill-table") - .arg("telemetry") + .arg("deployment_default_not_the_job_table") .arg("--clickhouse-backfill-database") .arg("default") .arg("--enable-backfill-worker") From b09a057fc443f988d474a9b23bee74e6714e4ee7 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:46:42 -0600 Subject: [PATCH 3/6] fix: validate ClickHouse backfill source before reading --- .../accelerator.rs | 4 +- .../sketch_db/backfill/clickhouse_reader.rs | 13 ++++- .../sketch_db/backfill/service.rs | 52 +++++++++++++++++++ 3 files changed, 66 insertions(+), 3 deletions(-) diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index d489fbedb..d9a72c3d4 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -441,8 +441,8 @@ mod tests { String::new(), "requests".into(), None, - None, - None, + Some("asap_e2e.samples".into()), + Some("value".into()), ); config.pane_origin_ms = Some(0); let sds = SummaryCatalog::from_materializations(41, 1, &[config.clone()]).unwrap(); diff --git a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs index 5ed8784bc..fc5d784e9 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs @@ -87,6 +87,12 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor let fallback = super::service::default_reader_factory(); Arc::new(move |source| match source { BackfillSource::ClickHouse { database, table } => { + if database != &config.database { + return Err(RawSampleReaderError::Other { + reason: "ClickHouse source database differs from the deployment database".into(), + } + .into()); + } let mut source_config = config.clone(); source_config.database = database.clone(); source_config.table = table.clone(); @@ -190,11 +196,16 @@ mod tests { fn typed_source_enters_clickhouse_backfill_lifecycle() { let factory = clickhouse_reader_factory(config("samples")); let reader = factory(&BackfillSource::ClickHouse { - database: "another_database".into(), + database: "metrics".into(), table: "another_table".into(), }) .unwrap(); assert_eq!(reader.source_name(), "ClickHouseReader"); + assert!(factory(&BackfillSource::ClickHouse { + database: "another_database".into(), + table: "samples".into(), + }) + .is_err()); assert!(factory(&BackfillSource::ClickHouse { database: "default".into(), table: "samples; DROP TABLE x".into(), diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs index c22beeaa6..b2644605a 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -199,6 +199,22 @@ impl BackfillService { "BackfillService picking up queued job" ); + if let BackfillSource::ClickHouse { database, table } = &job.source { + let snapshot = self.config_source.snapshot(); + let configured_table = snapshot + .get_aggregation_config(job.agg_id) + .and_then(|config| config.table_name.as_deref()); + let qualified_table = format!("{database}.{table}"); + if configured_table != Some(table.as_str()) + && configured_table != Some(qualified_table.as_str()) + { + self.registry.mark_failed( + job.job_id, + "ClickHouse source differs from the installed materialization table", + ); + continue; + } + } let reader = match (self.reader_factory)(&job.source) { Ok(r) => r, Err(e) => { @@ -383,6 +399,42 @@ mod tests { } } + #[tokio::test(flavor = "current_thread")] + async fn clickhouse_source_must_match_installed_table_before_reader_creation() { + let mut cfg = sum_config(1, "latency"); + cfg.table_name = Some("expected_table".into()); + let agg_fp = cfg.policy_fp_u64(); + let hot = HotReloadStreamingConfig::from_arc(streaming_with(cfg)); + let registry = Arc::new(BackfillRegistry::new()); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let called = calls.clone(); + let service = BackfillService::new( + registry.clone(), + hot, + Arc::new(move |_| { + called.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + Ok(Arc::new(MockRawSampleReader::new(vec![]))) + }), + BackfillServiceConfig { + poll_interval: Duration::from_millis(5), + }, + ); + let handle = service.spawn(); + let job_id = registry.create( + agg_fp, + (0, 20), + BackfillSource::ClickHouse { + database: "default".into(), + table: "wrong_table".into(), + }, + 1, + ); + let status = wait_for_status(®istry, job_id, BackfillStatus::Failed, 2000).await; + handle.shutdown().await; + assert_eq!(status, BackfillStatus::Failed); + assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 0); + } + #[tokio::test(flavor = "current_thread")] async fn service_drains_queued_job_to_complete() { let cfg = sum_config(1, "latency"); From 297607ce66003941f0d4bf203bfc2bcd54c2fc1f Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:48:33 -0600 Subject: [PATCH 4/6] test: bind ClickHouse reader fixture to its catalog table --- .../query_engines/asap_clickhouse_query_engine/accelerator.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index d9a72c3d4..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 @@ -756,6 +756,8 @@ mod tests { None, ); cfg.pane_origin_ms = Some(0); + cfg.table_name = Some("asap_e2e.samples".into()); + cfg.value_column = Some("value".into()); let hot = crate::storage_engines::types::HotReloadStreamingConfig::from_arc(Arc::new( crate::storage_engines::types::StreamingConfig::new(HashMap::from([( cfg.policy_fp_u64(), From 8e371ba40ae8ddb3242839047c5f2d545236ee07 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:50:25 -0600 Subject: [PATCH 5/6] test: isolate invalid ClickHouse table validation --- .../src/storage_engines/sketch_db/backfill/clickhouse_reader.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs b/data_plane/src/storage_engines/sketch_db/backfill/clickhouse_reader.rs index fc5d784e9..68fb332b7 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 @@ -207,7 +207,7 @@ mod tests { }) .is_err()); assert!(factory(&BackfillSource::ClickHouse { - database: "default".into(), + database: "metrics".into(), table: "samples; DROP TABLE x".into(), }) .is_err()); From 3cb771b88c2fe7dcfc647d81617b5db8d52e4ae3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 18:05:18 -0600 Subject: [PATCH 6/6] style: format ClickHouse source validation --- .../storage_engines/sketch_db/backfill/clickhouse_reader.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 68fb332b7..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 @@ -89,7 +89,8 @@ pub fn clickhouse_reader_factory(config: ClickHouseReaderConfig) -> ReaderFactor BackfillSource::ClickHouse { database, table } => { if database != &config.database { return Err(RawSampleReaderError::Other { - reason: "ClickHouse source database differs from the deployment database".into(), + reason: "ClickHouse source database differs from the deployment database" + .into(), } .into()); }