feat(sketch-db): Phase 5h — PrometheusReader (real RawSampleReader for backfill)

asap-query-engine/src/main.rs

@@ -687,11 +687,11 @@ async fn main() -> Result<()> {
             schemas,
             store.clone(),
             hot_reload_config.clone(),
-            query_engine_rust::stores::sketch_db::noop_reader_factory(),
+            query_engine_rust::stores::sketch_db::default_reader_factory(),
             query_engine_rust::stores::sketch_db::BackfillServiceConfig::default(),
         );
         info!(
-            "Spawning BackfillService drain loop (reader factory: noop — jobs will fail fast until a real factory is wired)"
+            "Spawning BackfillService drain loop (reader factory: default — Prometheus sources wired, S3/ClickHouse/OtherSketch fail fast)"
         );
         Some(service.spawn())
     } else {

asap-query-engine/src/stores/sketch_db/backfill_service.rs

@@ -264,6 +264,33 @@ pub fn noop_reader_factory() -> ReaderFactory {
     })
 }
 
+/// Production-ready `ReaderFactory` covering the source variants
+/// whose readers ship in-tree as of Phase 5h:
+///
+/// * [`BackfillSource::Prometheus`] — routed to
+///   [`super::prometheus_reader::PrometheusReader`].
+///
+/// All other variants (`S3Gorilla`, `ClickHouse`, `OtherSketch`)
+/// return a clear "not yet implemented" error, which the worker
+/// surfaces on `BackfillJob::error_message` so the controller
+/// / operator sees exactly which reader is missing.
+pub fn default_reader_factory() -> ReaderFactory {
+    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::OtherSketch { .. } => Err(format!(
+            "reader for {source:?} not yet implemented; only Prometheus is wired in-tree as of Phase 5h"
+        )
+        .into()),
+    }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

asap-query-engine/src/stores/sketch_db/mod.rs

@@ -33,6 +33,7 @@ pub mod backfill_processor;
 pub mod backfill_service;
 pub mod backfill_window_builder;
 pub mod backfill_worker;
+pub mod prometheus_reader;
 pub mod raw_sample_reader;
 pub mod schema;
 pub mod simple_map_store;
@@ -42,11 +43,12 @@ pub use backfill::{
 };
 pub use backfill_processor::BackfillWindowProcessor;
 pub use backfill_service::{
-    noop_reader_factory, BackfillService, BackfillServiceConfig, BackfillServiceHandle,
-    ReaderFactory,
+    default_reader_factory, noop_reader_factory, BackfillService, BackfillServiceConfig,
+    BackfillServiceHandle, ReaderFactory,
 };
 pub use backfill_window_builder::build_backfilled_accumulator;
 pub use backfill_worker::{BackfillWorker, BackfillWorkerError, WindowProcessor};
+pub use prometheus_reader::PrometheusReader;
 pub use raw_sample_reader::{
     LabelFilter, MockRawSampleReader, RawSample, RawSampleReader, RawSampleReaderError,
 };