feat(backfill): mirror replayed batches into SketchIndex

data_plane/src/main.rs

@@ -795,7 +795,9 @@ async fn main() -> Result<()> {
             hot_reload_config.clone(),
             data_plane::stores::sketch_db::default_reader_factory(),
             data_plane::stores::sketch_db::BackfillServiceConfig::default(),
-        );
+        )
+        // M2.3.6e — replayed batches mirror into SketchIndex too.
+        .with_sketch_index(sketch_index.clone());
         info!(
             "Spawning BackfillService drain loop (reader factory: default — Prometheus sources wired, S3/OtherSketch fail fast)"
         );

data_plane/src/precompute_engine/output_sink.rs

@@ -1,11 +1,7 @@
-use crate::stores::sketch_db::index::{
-    canonical_parameters, compute_sid, AccuracyBound, AggKind, SketchIndex,
-    SketchInstanceMetadata,
-};
+use crate::stores::sketch_db::index::SketchIndex;
 use crate::stores::types::hot_reload_config::HotReloadStreamingConfig;
 use crate::stores::types::{AggregateCore, PrecomputedOutput};
 use crate::stores::Store;
-use std::collections::BTreeSet;
 use std::sync::{Arc, Mutex};
 use tracing::{debug_span, warn};
 
@@ -81,64 +77,13 @@ impl SketchIndexSink {
         let Some(agg_cfg) = cfg.get_aggregation_config(output.aggregation_id) else {
             warn!(
                 agg_id = output.aggregation_id,
-                "DualWriteSink: agg_config missing from streaming snapshot; skipping SketchIndex write"
+                "SketchIndexSink: agg_config missing from streaming snapshot; skipping write"
             );
             return false;
         };
-
-        // Canonicalize the per-DP attrs. AggregationConfig's
-        // grouping_labels.labels is sorted at construction; align
-        // with KeyByLabelValues.labels positionally.
-        let label_values_vec = output
-            .key
-            .as_ref()
-            .map(|k| k.labels.clone())
-            .unwrap_or_default();
-        let key_names = &agg_cfg.grouping_labels.labels;
-        let mut attrs_fp = String::new();
-        let mut label_values_map: std::collections::BTreeMap<String, String> =
-            std::collections::BTreeMap::new();
-        for (k, v) in key_names.iter().zip(label_values_vec.iter()) {
-            attrs_fp.push_str(k);
-            attrs_fp.push('=');
-            attrs_fp.push_str(v);
-            attrs_fp.push(';');
-            label_values_map.insert(k.clone(), v.clone());
-        }
-
-        let agg_kind = AggKind::Precompute {
-            agg_type: agg_cfg.aggregation_type,
-            parameters_canonical: canonical_parameters(&agg_cfg.parameters),
-        };
-        let sid = compute_sid(&agg_cfg.metric, &attrs_fp, &agg_kind);
-
-        // Register the sid in SketchIndex on first sight. M2.3.3
-        // doesn't compute `Capability` / `AccuracyBound` for
-        // precomputes (they answer exact stats), so both stay `None`.
-        if self.sketch_index.instance(sid).is_none() {
-            let group_by_keys: BTreeSet<String> = key_names.iter().cloned().collect();
-            self.sketch_index.register(SketchInstanceMetadata {
-                sid,
-                metric_name: agg_cfg.metric.clone(),
-                group_by_keys,
-                capability: None,
-                agg_kind: agg_kind.clone(),
-                accuracy: None,
-                first_seen_unix_ms: output.start_timestamp as i64,
-                retired_at_ms: None,
-                expires_at_ms: None,
-            });
-            let _ = AccuracyBound::from_config; // silence unused if all paths skip
-        }
-
-        let window = (output.start_timestamp, output.end_timestamp);
-        self.sketch_index.append_precompute(
-            sid,
-            label_values_map,
-            window,
-            accumulator.clone_boxed_core(),
-        );
-        true
+        self.sketch_index
+            .ingest_precompute_for_agg_config(agg_cfg, output, accumulator)
+            .is_some()
     }
 }
 
@@ -233,7 +178,7 @@ impl OutputSink for NoopOutputSink {
 mod tests {
     use super::*;
     use crate::precompute_engine::operators::SumAccumulator;
-    use crate::stores::sketch_db::index::SidLookup;
+    use crate::stores::sketch_db::index::{AggKind, SidLookup};
     use crate::stores::types::{KeyByLabelValues, StreamingConfig};
     use asap_types::aggregation_config::AggregationConfig;
     use asap_types::enums::WindowType;

data_plane/src/stores/sketch_db/backfill/processor.rs

@@ -129,6 +129,11 @@ pub struct BackfillWindowProcessor {
     /// Where per-window writes land. Same trait the live output
     /// sink uses; different call site.
     store: Arc<dyn Store>,
+    /// Phase 5 M2.3.6e — mirror every batch into `SketchIndex` so the
+    /// new sid-keyed query path sees backfilled data the same way it
+    /// sees live precompute output. Optional so test fixtures that
+    /// pre-date M2.3 stay compiling.
+    sketch_index: Option<Arc<crate::stores::sketch_db::index::SketchIndex>>,
     /// Registry where we record which `(agg_id, window_range)`
     /// tuples this job wrote. Phase 5f's coverage tracker reads
     /// this list.
@@ -150,11 +155,23 @@ impl BackfillWindowProcessor {
             config,
             schemas,
             store,
+            sketch_index: None,
             registry,
             job_id,
         }
     }
 
+    /// Attach a `SketchIndex` so each batch is mirrored there in
+    /// addition to the legacy store. Builder-style so existing call
+    /// sites opt in with one chained call.
+    pub fn with_sketch_index(
+        mut self,
+        sketch_index: Arc<crate::stores::sketch_db::index::SketchIndex>,
+    ) -> Self {
+        self.sketch_index = Some(sketch_index);
+        self
+    }
+
     /// Look up the `AggregationConfig` for `agg_id` in the current
     /// `StreamingConfig` snapshot. Returns an error string if the
     /// agg has been removed from the config since the job was
@@ -230,6 +247,18 @@ impl WindowProcessor for BackfillWindowProcessor {
             batch.push((output, accumulator));
         }
 
+        // Phase 5 M2.3.6e — mirror the batch into the SketchIndex
+        // BEFORE handing it to the legacy store. The store
+        // `insert_precomputed_output_batch` consumes the batch by
+        // value, so we mirror first while the borrows are still
+        // live. Best-effort: a missing index just means the legacy
+        // store remains the source of truth for this window.
+        if let Some(idx) = self.sketch_index.as_ref() {
+            for (output, accumulator) in &batch {
+                idx.ingest_precompute_for_agg_config(&config, output, accumulator.as_ref());
+            }
+        }
+
         // Single atomic batch write — mirrors live worker's emit_batch
         // approach. The store is responsible for per-key atomicity;
         // we don't need cross-key transactions.

data_plane/src/stores/sketch_db/backfill/service.rs

@@ -97,6 +97,9 @@ pub struct BackfillService {
     registry: Arc<BackfillRegistry>,
     schemas: Arc<SchemaRegistry>,
     store: Arc<dyn Store>,
+    /// Phase 5 M2.3.6e — when set, every BackfillWindowProcessor the
+    /// service constructs mirrors its writes into this SketchIndex.
+    sketch_index: Option<Arc<crate::stores::sketch_db::index::SketchIndex>>,
     config_source: HotReloadStreamingConfig,
     reader_factory: ReaderFactory,
     service_config: BackfillServiceConfig,
@@ -115,12 +118,23 @@ impl BackfillService {
             registry,
             schemas,
             store,
+            sketch_index: None,
             config_source,
             reader_factory,
             service_config,
         }
     }
 
+    /// Attach a `SketchIndex` so each replayed batch is also mirrored
+    /// there. Builder-style; safe to omit (legacy tests).
+    pub fn with_sketch_index(
+        mut self,
+        sketch_index: Arc<crate::stores::sketch_db::index::SketchIndex>,
+    ) -> Self {
+        self.sketch_index = Some(sketch_index);
+        self
+    }
+
     /// Spawn the service as a tokio task. Returns a `BackfillServiceHandle`
     /// with a `shutdown` oneshot so `main.rs` can stop it cleanly on
     /// ctrl-c.
@@ -185,13 +199,16 @@ impl BackfillService {
             // Build the per-job processor with the current config
             // snapshot. The processor snapshots again per window so
             // mid-job config swaps stay visible.
-            let processor = BackfillWindowProcessor::new(
+            let mut processor = BackfillWindowProcessor::new(
                 self.config_source.clone(),
                 self.schemas.clone(),
                 self.store.clone(),
                 self.registry.clone(),
                 job.job_id,
             );
+            if let Some(idx) = self.sketch_index.as_ref() {
+                processor = processor.with_sketch_index(idx.clone());
+            }
             let worker = BackfillWorker::new(self.registry.clone());
 
             let filter = LabelFilter::for_metric(

data_plane/src/stores/sketch_db/index/mod.rs

@@ -903,6 +903,66 @@ impl SketchIndex {
 }
 
 impl SketchIndex {
+    /// Phase 5 M2.3.6e — write-side helper. Given an
+    /// `AggregationConfig` and one `(PrecomputedOutput, AggregateCore)`
+    /// pair (the shape both the live worker AND the backfill processor
+    /// emit), compute the precompute sid, register a metadata entry on
+    /// first sight, and append the payload window. Used by
+    /// `SketchIndexSink` (live ingest) and `BackfillWindowProcessor`
+    /// (archive replay) so they share one canonical sid-derivation
+    /// path.
+    ///
+    /// Returns the sid the entry landed under (or `None` when the
+    /// agg_config / output combination doesn't fit the precompute
+    /// model — caller logs and skips).
+    pub fn ingest_precompute_for_agg_config(
+        &self,
+        agg_cfg: &asap_types::aggregation_config::AggregationConfig,
+        output: &crate::stores::types::PrecomputedOutput,
+        accumulator: &dyn crate::stores::types::AggregateCore,
+    ) -> Option<u64> {
+        let label_values_vec = output
+            .key
+            .as_ref()
+            .map(|k| k.labels.clone())
+            .unwrap_or_default();
+        let key_names = &agg_cfg.grouping_labels.labels;
+        let mut attrs_fp = String::new();
+        let mut label_values_map: BTreeMap<String, String> = BTreeMap::new();
+        for (k, v) in key_names.iter().zip(label_values_vec.iter()) {
+            attrs_fp.push_str(k);
+            attrs_fp.push('=');
+            attrs_fp.push_str(v);
+            attrs_fp.push(';');
+            label_values_map.insert(k.clone(), v.clone());
+        }
+
+        let agg_kind = AggKind::Precompute {
+            agg_type: agg_cfg.aggregation_type,
+            parameters_canonical: canonical_parameters(&agg_cfg.parameters),
+        };
+        let sid = compute_sid(&agg_cfg.metric, &attrs_fp, &agg_kind);
+
+        if self.instance(sid).is_none() {
+            let group_by_keys: BTreeSet<String> = key_names.iter().cloned().collect();
+            self.register(SketchInstanceMetadata {
+                sid,
+                metric_name: agg_cfg.metric.clone(),
+                group_by_keys,
+                capability: None,
+                agg_kind: agg_kind.clone(),
+                accuracy: None,
+                first_seen_unix_ms: output.start_timestamp as i64,
+                retired_at_ms: None,
+                expires_at_ms: None,
+            });
+        }
+
+        let window = (output.start_timestamp, output.end_timestamp);
+        self.append_precompute(sid, label_values_map, window, accumulator.clone_boxed_core());
+        Some(sid)
+    }
+
     /// Phase 5 M2.3.6d — eviction-side helper. Removes every sid in the
     /// index whose metadata was registered against `agg_cfg`, i.e.
     /// shares the same metric, agg_type, parameters canonicalization,