diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index dc5fbea2..be6f8c29 100644 --- a/data_plane/src/main.rs +++ b/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)" ); diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 05e5a248..ca6a8066 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/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 = - 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 = 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; diff --git a/data_plane/src/stores/sketch_db/backfill/processor.rs b/data_plane/src/stores/sketch_db/backfill/processor.rs index f3afefa9..51810cfe 100644 --- a/data_plane/src/stores/sketch_db/backfill/processor.rs +++ b/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, + /// 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>, /// 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, + ) -> 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. diff --git a/data_plane/src/stores/sketch_db/backfill/service.rs b/data_plane/src/stores/sketch_db/backfill/service.rs index 736077be..7c605c2a 100644 --- a/data_plane/src/stores/sketch_db/backfill/service.rs +++ b/data_plane/src/stores/sketch_db/backfill/service.rs @@ -97,6 +97,9 @@ pub struct BackfillService { registry: Arc, schemas: Arc, store: Arc, + /// Phase 5 M2.3.6e — when set, every BackfillWindowProcessor the + /// service constructs mirrors its writes into this SketchIndex. + sketch_index: Option>, 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, + ) -> 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( diff --git a/data_plane/src/stores/sketch_db/index/mod.rs b/data_plane/src/stores/sketch_db/index/mod.rs index a943319a..225185e2 100644 --- a/data_plane/src/stores/sketch_db/index/mod.rs +++ b/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 { + 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 = 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 = 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,