diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 3ce89e8ae..824a42843 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -459,6 +459,7 @@ async fn main() -> Result<()> { let output_sink = Arc::new(SketchStoreSink::new( sketch_index.clone(), hot_reload_config.clone(), + series_resolver.clone(), )); let engine = PrecomputeEngine::new( precompute_config, @@ -745,8 +746,11 @@ async fn main() -> Result<()> { data_plane::storage_engines::sketch_db::BackfillServiceConfig::default(), ) // M2.3.6e — replayed batches land in SketchStore (the only - // destination after the M2.3.6g store retirement). - .with_sketch_index(sketch_index.clone()); + // destination after the M2.3.6g store retirement). Resolver + // is the same shared mint authority as live ingest, so + // backfilled precompute sids share the OTel namespace. + .with_sketch_index(sketch_index.clone()) + .with_series_resolver(series_resolver.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 6349344eb..e1e89c4f6 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -1,3 +1,4 @@ +use crate::drivers::ingest::series_resolver::SeriesIdResolver; use crate::storage_engines::sketch_db::index::SketchStore; use crate::storage_engines::types::hot_reload_config::HotReloadStreamingConfig; use crate::storage_engines::types::{AggregateCore, PrecomputedOutput}; @@ -28,13 +29,25 @@ pub trait OutputSink: Send + Sync { pub struct SketchStoreSink { sketch_index: Arc, hot_reload: HotReloadStreamingConfig, + /// Single shared resolver across the ingest + precompute paths. Under + /// the registry-allocated sid model (PR-1..3), this is the canonical + /// mint authority — precompute sids share the same `next_sid` counter + /// as OTel-sketch sids, so the two paths can never collide on identity + /// even when the same `(metric, attrs)` carries both a sketch and an + /// exact precompute. + series_resolver: Arc, } impl SketchStoreSink { - pub fn new(sketch_index: Arc, hot_reload: HotReloadStreamingConfig) -> Self { + pub fn new( + sketch_index: Arc, + hot_reload: HotReloadStreamingConfig, + series_resolver: Arc, + ) -> Self { Self { sketch_index, hot_reload, + series_resolver, } } @@ -56,8 +69,14 @@ impl SketchStoreSink { ); return false; }; + let resolver = self.series_resolver.clone(); self.sketch_index - .ingest_precompute_for_agg_config(agg_cfg, output, accumulator) + .ingest_precompute_for_agg_config( + |metric, fp, ak| resolver.resolve(metric, fp, ak), + agg_cfg, + output, + accumulator, + ) .is_some() } } @@ -195,7 +214,11 @@ mod tests { let hot_reload = HotReloadStreamingConfig::new(streaming.clone()); let sketch_index = Arc::new(SketchStore::new()); - let sink = SketchStoreSink::new(sketch_index.clone(), hot_reload); + let sink = SketchStoreSink::new( + sketch_index.clone(), + hot_reload, + Arc::new(SeriesIdResolver::new()), + ); let key = KeyByLabelValues::new_with_labels(vec!["z0".to_string()]); let output = PrecomputedOutput::new(1000, 2000, Some(key), agg_id); @@ -234,7 +257,11 @@ mod tests { let streaming = StreamingConfig::new(HashMap::new()); let hot_reload = HotReloadStreamingConfig::new(streaming.clone()); let sketch_index = Arc::new(SketchStore::new()); - let sink = SketchStoreSink::new(sketch_index.clone(), hot_reload); + let sink = SketchStoreSink::new( + sketch_index.clone(), + hot_reload, + Arc::new(SeriesIdResolver::new()), + ); let output = PrecomputedOutput::new(1000, 2000, None, 99); let acc: Box = Box::new(SumAccumulator::with_sum(1.0)); diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs index 80fe7108c..56cdc2445 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs @@ -124,6 +124,12 @@ pub struct BackfillWindowProcessor { /// effects can skip attaching one (the processor becomes a /// registry-only logger in that case). sketch_index: Option>, + /// Shared sid mint authority. Same `SeriesIdResolver` the OTel + /// ingest path uses, so backfilled precompute sids land in the + /// same unified namespace as live precompute / sketch sids. Only + /// consulted when `sketch_index` is also set (no sketch_index ⇒ + /// no precompute write ⇒ no sid mint). + series_resolver: Option>, /// Registry where we record which `(agg_id, window_range)` /// tuples this job wrote. Phase 5f's coverage tracker reads /// this list. @@ -142,6 +148,7 @@ impl BackfillWindowProcessor { Self { config, sketch_index: None, + series_resolver: None, registry, job_id, } @@ -157,6 +164,19 @@ impl BackfillWindowProcessor { self } + /// Attach the shared `SeriesIdResolver` so precompute writes mint + /// sids via the same registry the OTel ingest path uses. Required + /// alongside [`Self::with_sketch_index`] — without a resolver the + /// processor still runs but skips the precompute write (registry + /// provenance still recorded, with a warn-log per missing call). + pub fn with_series_resolver( + mut self, + series_resolver: Arc, + ) -> Self { + self.series_resolver = Some(series_resolver); + 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 @@ -236,10 +256,32 @@ impl WindowProcessor for BackfillWindowProcessor { // No legacy SketchStore write path remains. When no // sketch_index is attached (tests), the writes are simply // dropped — the registry still records the (agg_id, range) - // provenance below. + // provenance below. When sketch_index is attached but no + // resolver was provided, the precompute write is skipped + // with a warn — sid minting requires the shared resolver. 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()); + match self.series_resolver.as_ref() { + Some(resolver) => { + for (output, accumulator) in &batch { + let resolver = resolver.clone(); + idx.ingest_precompute_for_agg_config( + |metric, fp, ak| resolver.resolve(metric, fp, ak), + &config, + output, + accumulator.as_ref(), + ); + } + } + None => { + tracing::warn!( + job_id = self.job_id, + agg_id, + batch_size = batch.len(), + "BackfillWindowProcessor has sketch_index but no \ + series_resolver attached; skipping precompute writes \ + for this window", + ); + } } } // Hold `batch` alive until after the registry record below, 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 bcd1e8f9f..88a416889 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -96,6 +96,9 @@ pub struct BackfillService { /// Phase 5 M2.3.6g — replayed batches land in `SketchStore` only; /// the legacy `Arc` field is gone. sketch_index: Option>, + /// Shared sid mint authority — wired alongside `sketch_index` so + /// backfilled precompute sids share the namespace with live ingest. + series_resolver: Option>, config_source: HotReloadStreamingConfig, reader_factory: ReaderFactory, service_config: BackfillServiceConfig, @@ -111,6 +114,7 @@ impl BackfillService { Self { registry, sketch_index: None, + series_resolver: None, config_source, reader_factory, service_config, @@ -127,6 +131,19 @@ impl BackfillService { self } + /// Attach the shared `SeriesIdResolver` so the per-job + /// `BackfillWindowProcessor` mints sids via the same registry the + /// OTel ingest path uses. Builder-style; should be paired with + /// [`Self::with_sketch_index`] in production (without it, + /// processor writes are skipped with a warn). + pub fn with_series_resolver( + mut self, + series_resolver: Arc, + ) -> Self { + self.series_resolver = Some(series_resolver); + 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. @@ -199,6 +216,9 @@ impl BackfillService { if let Some(idx) = self.sketch_index.as_ref() { processor = processor.with_sketch_index(idx.clone()); } + if let Some(resolver) = self.series_resolver.as_ref() { + processor = processor.with_series_resolver(resolver.clone()); + } let worker = BackfillWorker::new(self.registry.clone()); let filter = LabelFilter::for_metric( diff --git a/data_plane/src/storage_engines/sketch_db/data/mod.rs b/data_plane/src/storage_engines/sketch_db/data/mod.rs index 27e6c2002..7ca707305 100644 --- a/data_plane/src/storage_engines/sketch_db/data/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/data/mod.rs @@ -26,10 +26,10 @@ //! the read path (sid + label values + per-window samples). //! - [`AccuracyBound`] — `(epsilon, confidence)` derived from a //! `SketchConfig`. Surfaces in HTTP response headers. -//! - [`compute_sid`] — the canonical sid hash used today only by the -//! precompute ingest path (`SketchStore::ingest_precompute_for_agg_config`). -//! The OTel sketch ingest path moved to `SeriesIdResolver` (Option B -//! — registry-allocated sids); the precompute path follows in PR-4. +//! - sid minting is no longer in this module — `SeriesIdResolver` +//! (in `drivers::ingest::series_resolver`) is the single mint +//! authority for every sid in the system. `AggKind::canonical_string()` +//! here produces the third element of the resolver's cache key. //! - [`canonical_parameters`] — helper that renders a parameters //! `HashMap` into the canonical string form //! `AggKind::Precompute::parameters_canonical` expects. @@ -41,7 +41,9 @@ //! - [`AggregationType`] — the agg-type enum that //! `AggKind::Precompute` carries. -use xxhash_rust::xxh64::xxh64; +// `xxhash_rust::xxh64` import retired alongside `compute_sid` (PR-4). +// Sid minting is now registry-allocated via `SeriesIdResolver` — no +// content-addressed hash is computed at this layer. // ── Capability re-exports ──────────────────────────────────────────────────── // @@ -196,102 +198,13 @@ fn sketch_config_canonical(cfg: &SketchConfig) -> String { // ── sid hash ──────────────────────────────────────────────────────────────── // -// `compute_sketch_sid` was retired alongside the OTel ingest path's -// migration to `SeriesIdResolver` (Option B — registry-allocated sids). -// The remaining `compute_sid` function is still called by -// `SketchStore::ingest_precompute_for_agg_config` for PRECOMPUTE -// aggregations; that path migrates to the resolver in PR-4, at which -// point `compute_sid` and its `sketch_kind_tag` / `encode_sketch_config` -// helpers will go away too. Sketch sids today come exclusively from the -// resolver — same authority as precompute sids will after PR-4. - -/// Generalized content-addressed sid hash. Same `(metric, attrs, agg_kind)` -/// tuple always yields the same sid. -/// -/// The two branches encode disjointly: a `Sketch` payload starts with -/// `sketch_kind_tag` (1..=7), while a `Precompute` payload starts with -/// the byte `b'P'` (ASCII 80), which no sketch tag will ever produce. -/// So an attacker (or a colliding hash input) can't force a sketch sid -/// to overlap a precompute sid at the encoding level. -/// -/// Critically, the `Sketch` branch is bit-identical to the historical -/// `compute_sketch_sid` byte layout — existing sketch sids the M2 -/// wire format introduced stay stable across this generalization. -pub fn compute_sid( - metric_name: &str, - attrs_fingerprint: &str, - agg_kind: &AggKind, -) -> u64 { - let mut buf: Vec = - Vec::with_capacity(metric_name.len() + attrs_fingerprint.len() + 32); - buf.extend_from_slice(metric_name.as_bytes()); - buf.push(0); - buf.extend_from_slice(attrs_fingerprint.as_bytes()); - buf.push(0); - match agg_kind { - AggKind::Sketch { kind, config } => { - buf.push(sketch_kind_tag(*kind)); - buf.push(0); - encode_sketch_config(config, &mut buf); - } - AggKind::Precompute { - agg_type, - parameters_canonical, - } => { - buf.push(b'P'); - buf.push(0); - buf.extend_from_slice(agg_type.as_str().as_bytes()); - buf.push(0); - buf.extend_from_slice(parameters_canonical.as_bytes()); - } - } - let h = xxh64(&buf, 0); - if h == 0 { - 1 - } else { - h - } -} - -fn sketch_kind_tag(k: SketchKindHandle) -> u8 { - match k { - SketchKindHandle::DDSketch => 1, - SketchKindHandle::Kll => 2, - SketchKindHandle::Hll => 3, - SketchKindHandle::CountSketch => 4, - SketchKindHandle::CountMin => 5, - SketchKindHandle::CmsWithHeap => 6, - SketchKindHandle::CountSketchWithHeap => 7, - SketchKindHandle::Any => 0, - } -} - -fn encode_sketch_config(cfg: &SketchConfig, buf: &mut Vec) { - match cfg { - SketchConfig::DDSketch { relative_accuracy } => { - buf.push(b'D'); - buf.extend_from_slice(&relative_accuracy.to_le_bytes()); - } - SketchConfig::Kll { k } => { - buf.push(b'K'); - buf.extend_from_slice(&k.to_le_bytes()); - } - SketchConfig::Hll { precision } => { - buf.push(b'H'); - buf.extend_from_slice(&precision.to_le_bytes()); - } - SketchConfig::CountSketch { rows, cols } => { - buf.push(b'S'); - buf.extend_from_slice(&rows.to_le_bytes()); - buf.extend_from_slice(&cols.to_le_bytes()); - } - SketchConfig::CountMin { rows, cols } => { - buf.push(b'M'); - buf.extend_from_slice(&rows.to_le_bytes()); - buf.extend_from_slice(&cols.to_le_bytes()); - } - } -} +// `compute_sid` was retired alongside `compute_sketch_sid` (PR-3) and +// the precompute ingest migration (PR-4). All sid minting now flows +// through `SeriesIdResolver` — one registry-allocated u64 per +// `(metric, attrs_fingerprint, agg_kind_canonical)` triple, shared +// across the OTel sketch ingest path and the precompute output path. +// See `AggKind::canonical_string` for the agg-kind canonicalization +// that replaces the byte-layout this hash used to produce. // ── Accuracy ──────────────────────────────────────────────────────────────── diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index dc6cd31e2..bb95c3e85 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -33,9 +33,8 @@ use crate::storage_engines::sketch_db::lifecycle::AggStatus; // (`crate::storage_engines::sketch_db::index::*`) keep compiling // during the reorg. pub use crate::storage_engines::sketch_db::data::{ - canonical_parameters, compute_sid, AccuracyBound, AggKind, AggPayload, AggregationType, - Capability, SketchConfig, SketchEncoding, SketchKindHandle, SketchSampleState, - SketchTimeSeries, + canonical_parameters, AccuracyBound, AggKind, AggPayload, AggregationType, Capability, + SketchConfig, SketchEncoding, SketchKindHandle, SketchSampleState, SketchTimeSeries, }; fn now_ms() -> u64 { @@ -594,6 +593,7 @@ impl SketchStore { /// drop residual state cleanly. pub fn ingest_precompute_for_agg_config( &self, + mint_sid: impl FnOnce(&str, &str, &str) -> u64, agg_cfg: &asap_types::aggregation_config::AggregationConfig, output: &crate::storage_engines::types::PrecomputedOutput, accumulator: &dyn crate::storage_engines::types::AggregateCore, @@ -618,7 +618,14 @@ impl SketchStore { agg_type: agg_cfg.aggregation_type, parameters_canonical: canonical_parameters(&agg_cfg.parameters), }; - let sid = compute_sid(&agg_cfg.metric, &attrs_fp, &agg_kind); + // Sid mint delegated to the caller's closure — typically + // `|m, fp, ak| series_resolver.resolve(m, fp, ak)`. Keeps the + // SketchStore free of any layer-inverted dependency on the + // resolver type (which lives in `drivers::ingest`). Tests + // pass either a real local resolver or a counter-mock + // closure. + let agg_kind_canonical = agg_kind.canonical_string(); + let sid = mint_sid(&agg_cfg.metric, &attrs_fp, &agg_kind_canonical); match self.instance(sid) { None => { @@ -1126,72 +1133,18 @@ mod tests { assert_eq!(series[0].samples.len(), 4); } - // `compute_sketch_sid_*` tests removed alongside the function they - // exercised. The same identity properties (metric/attrs/kind/config - // disambiguate sketch sids) are now covered by the resolver's own - // `distinct_agg_kinds_same_series_distinct_sids` test plus the - // round-trip parity captured at the OTel ingest layer. - - #[test] - fn compute_sid_precompute_is_deterministic() { - let kind = AggKind::Precompute { - agg_type: AggregationType::Sum, - parameters_canonical: String::new(), - }; - let a = compute_sid("cpu_seconds", "zone=z0;", &kind); - let b = compute_sid("cpu_seconds", "zone=z0;", &kind); - assert_eq!(a, b); - assert_ne!(a, 0); - } - - #[test] - fn compute_sid_sketch_vs_precompute_never_collide() { - // Same metric + attrs; one is a sketch, one is a precompute. - // The 'S'/'P' discriminator byte must make the hashes differ. - let sketch = AggKind::Sketch { - kind: SketchKindHandle::DDSketch, - config: SketchConfig::DDSketch { - relative_accuracy: 0.01, - }, - }; - let precompute = AggKind::Precompute { - agg_type: AggregationType::Sum, - parameters_canonical: String::new(), - }; - let a = compute_sid("m", "zone=z0;", &sketch); - let b = compute_sid("m", "zone=z0;", &precompute); - assert_ne!(a, b, "sketch and precompute sids must not collide"); - } - - #[test] - fn compute_sid_precompute_distinguishes_agg_type() { - let sum = AggKind::Precompute { - agg_type: AggregationType::Sum, - parameters_canonical: String::new(), - }; - let count = AggKind::Precompute { - agg_type: AggregationType::Increase, - parameters_canonical: String::new(), - }; - let a = compute_sid("m", "zone=z0;", &sum); - let b = compute_sid("m", "zone=z0;", &count); - assert_ne!(a, b); - } - - #[test] - fn compute_sid_precompute_distinguishes_parameters() { - let p1 = AggKind::Precompute { - agg_type: AggregationType::DatasketchesKLL, - parameters_canonical: "k=200;".to_string(), - }; - let p2 = AggKind::Precompute { - agg_type: AggregationType::DatasketchesKLL, - parameters_canonical: "k=400;".to_string(), - }; - let a = compute_sid("m", "zone=z0;", &p1); - let b = compute_sid("m", "zone=z0;", &p2); - assert_ne!(a, b); - } + // sid-hash unit tests removed alongside `compute_sid` (PR-4) and + // `compute_sketch_sid` (PR-3). The identity properties they + // exercised — `(metric, attrs, agg_kind)` discriminates sids, + // sketch and precompute never collide, agg_type and parameters + // each contribute to identity — are now covered by: + // + // - `series_resolver::tests::distinct_agg_kinds_same_series_distinct_sids` + // (the resolver's identity contract under Interpretation B) + // - `AggKind::canonical_string` is exhaustive over the AggKind + // enum, so two variants whose fields differ produce different + // canonical strings → different resolver cache keys → different + // sids by construction. #[test] fn canonical_parameters_is_insertion_order_independent() { diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs index aaee850e5..2416db297 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -291,11 +291,27 @@ mod tests { agg_id: u64, ts: u64, ) -> u64 { + use crate::drivers::ingest::series_resolver::SeriesIdResolver; + use std::sync::Arc; let acc = SumAccumulator::with_sum(1.0); let output = crate::storage_engines::types::PrecomputedOutput::new(ts, ts + 1000, None, agg_id); let agg_cfg = streaming_config.get_aggregation_config(agg_id).unwrap(); + // Test-scoped resolver — each call mints fresh. Production + // shares one resolver across all sinks; tests don't need that + // because each fixture is isolated. Static-lifetime so multiple + // `write_one` calls in the same test share `next_sid` (matches + // the production single-resolver model). + thread_local! { + static RESOLVER: Arc = Arc::new(SeriesIdResolver::new()); + } + let resolver = RESOLVER.with(|r| r.clone()); sketch_index - .ingest_precompute_for_agg_config(agg_cfg, &output, &acc) + .ingest_precompute_for_agg_config( + |m, fp, ak| resolver.resolve(m, fp, ak), + agg_cfg, + &output, + &acc, + ) .expect("registered sid") } diff --git a/data_plane/src/tests/capability_matching_tests.rs b/data_plane/src/tests/capability_matching_tests.rs index 1b1c750cf..9f766441e 100644 --- a/data_plane/src/tests/capability_matching_tests.rs +++ b/data_plane/src/tests/capability_matching_tests.rs @@ -4,6 +4,7 @@ //! the engine falls back to searching StreamingConfig by capability, and that //! the existing query_config path still takes priority when an entry is present. +use crate::drivers::ingest::series_resolver::SeriesIdResolver; use crate::storage_engines::types::{ AggregationConfig, AggregationType, PrecomputedOutput, StreamingConfig, WindowType}; use crate::query_engines::asap_query_engine::engine::ASAPQueryEngine; @@ -82,7 +83,13 @@ fn engine_no_query_configs( } "DeltaSetAggregator" => Box::new(DeltaSetAggregatorAccumulator::new()), _ => Box::new(SumAccumulator::with_sum(42.0))}; - sketch_index.ingest_precompute_for_agg_config(c, &output, acc.as_ref()); + let resolver = Arc::new(SeriesIdResolver::new()); + sketch_index.ingest_precompute_for_agg_config( + |m, fp, ak| resolver.resolve(m, fp, ak), + c, + &output, + acc.as_ref(), + ); } let schema_label_names = @@ -110,7 +117,13 @@ fn engine_with_query_config( let window_ms = agg_config.window_size * 1000; let output = PrecomputedOutput::new(ts - window_ms, ts, None, agg_id); let acc = SumAccumulator::with_sum(99.0); - sketch_index.ingest_precompute_for_agg_config(&agg_config, &output, &acc); + let resolver = Arc::new(SeriesIdResolver::new()); + sketch_index.ingest_precompute_for_agg_config( + |m, fp, ak| resolver.resolve(m, fp, ak), + &agg_config, + &output, + &acc, + ); let schema_label_names = KeyByLabelNames::new(schema_labels.iter().map(|s| s.to_string()).collect()); diff --git a/data_plane/src/tests/schema_timeline_dispatch_tests.rs b/data_plane/src/tests/schema_timeline_dispatch_tests.rs index e4da841eb..0f8425087 100644 --- a/data_plane/src/tests/schema_timeline_dispatch_tests.rs +++ b/data_plane/src/tests/schema_timeline_dispatch_tests.rs @@ -88,10 +88,21 @@ fn seed_sum_at( ) { let key = Some(KeyByLabelValues { labels: vec![host.to_string()]}); + use crate::drivers::ingest::series_resolver::SeriesIdResolver; + use std::sync::Arc as StdArc; let output = PrecomputedOutput::new(ts, ts, key, agg_id); let acc = SumAccumulator::with_sum(sum); if let Some(agg_cfg) = streaming_config.get_aggregation_config(agg_id) { - sketch_index.ingest_precompute_for_agg_config(agg_cfg, &output, &acc); + thread_local! { + static RESOLVER: StdArc = StdArc::new(SeriesIdResolver::new()); + } + let resolver = RESOLVER.with(|r| r.clone()); + sketch_index.ingest_precompute_for_agg_config( + |m, fp, ak| resolver.resolve(m, fp, ak), + agg_cfg, + &output, + &acc, + ); } let _ = (ts, host); } diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs index b3b515cee..6ac68e512 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -5,6 +5,7 @@ //! hardcodes "SumAccumulator", these helpers build AggregationConfig with //! the correct aggregation_type string. +use crate::drivers::ingest::series_resolver::SeriesIdResolver; use crate::storage_engines::types::{ AggregationConfig, AggregationType, KeyByLabelValues, PrecomputedOutput, QueryLanguage, StreamingConfig, WindowType}; @@ -13,6 +14,26 @@ use crate::query_engines::asap_query_engine::engine::ASAPQueryEngine; use crate::AggregateCore; use promql_utilities::data_model::KeyByLabelNames; use std::collections::HashMap; + +/// Helper for test factories — wraps the closure-mint call with a +/// fresh resolver and forwards to `SketchStore::ingest_precompute_for_agg_config`. +/// Each factory gets its own resolver instance; tests are isolated so +/// the `next_sid = 1, 2, ...` counter doesn't bleed between fixtures. +fn ingest_with_fresh_resolver( + sketch_index: &crate::storage_engines::sketch_db::index::SketchStore, + resolver: &std::sync::Arc, + agg_cfg: &AggregationConfig, + output: &PrecomputedOutput, + accumulator: &dyn AggregateCore, +) -> Option { + let resolver = resolver.clone(); + sketch_index.ingest_precompute_for_agg_config( + |m, fp, ak| resolver.resolve(m, fp, ak), + agg_cfg, + output, + accumulator, + ) +} use std::sync::Arc; /// Data to insert into a store: (label_values, accumulator) @@ -92,6 +113,7 @@ pub fn create_engine_single_pop_with_aggregated( storage_backend: Default::default()}); let sketch_index = std::sync::Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()); + let resolver = std::sync::Arc::new(SeriesIdResolver::new()); // Insert data into SketchStore via the canonical helper (M2.3.6e). let agg_cfg = streaming_config @@ -102,7 +124,7 @@ pub fn create_engine_single_pop_with_aggregated( for (label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp, timestamp, key, 1); - sketch_index.ingest_precompute_for_agg_config(&agg_cfg, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, &agg_cfg, &output, acc.as_ref()); } ASAPQueryEngine::new(streaming_config, 1).with_sketch_index(sketch_index) @@ -189,6 +211,7 @@ pub fn create_engine_dual_input( storage_backend: Default::default()}); let sketch_index = std::sync::Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()); + let resolver = std::sync::Arc::new(SeriesIdResolver::new()); let agg_cfg_1 = streaming_config .get_aggregation_config(1) @@ -202,12 +225,12 @@ pub fn create_engine_dual_input( for (label_values_opt, acc) in value_data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp, timestamp, key, 1); - sketch_index.ingest_precompute_for_agg_config(&agg_cfg_1, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, &agg_cfg_1, &output, acc.as_ref()); } for (label_values_opt, acc) in keys_data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp, timestamp, key, 2); - sketch_index.ingest_precompute_for_agg_config(&agg_cfg_2, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, &agg_cfg_2, &output, acc.as_ref()); } ASAPQueryEngine::new(streaming_config, 1).with_sketch_index(sketch_index) @@ -281,18 +304,19 @@ pub fn create_engine_two_metrics( storage_backend: Default::default()}); let sketch_index = std::sync::Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()); + let resolver = std::sync::Arc::new(SeriesIdResolver::new()); let agg_cfg_1 = streaming_config.get_aggregation_config(1).cloned().expect("agg 1"); let agg_cfg_2 = streaming_config.get_aggregation_config(2).cloned().expect("agg 2"); let timestamp = 1_000_000_u64; for (label_values_opt, acc) in data_a { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp, timestamp, key, 1); - sketch_index.ingest_precompute_for_agg_config(&agg_cfg_1, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, &agg_cfg_1, &output, acc.as_ref()); } for (label_values_opt, acc) in data_b { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp, timestamp, key, 2); - sketch_index.ingest_precompute_for_agg_config(&agg_cfg_2, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, &agg_cfg_2, &output, acc.as_ref()); } let _ = (query_a, query_b); ASAPQueryEngine::new(streaming_config, 1).with_sketch_index(sketch_index) @@ -359,6 +383,7 @@ pub fn create_engine_three_metrics( storage_backend: Default::default()}); let sketch_index = std::sync::Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()); + let resolver = std::sync::Arc::new(SeriesIdResolver::new()); let agg_cfgs: Vec<_> = (1..=3) .map(|id| streaming_config.get_aggregation_config(id).cloned().expect("agg present")) .collect(); @@ -369,7 +394,7 @@ pub fn create_engine_three_metrics( for (label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp, timestamp, key, agg_id); - sketch_index.ingest_precompute_for_agg_config(agg_cfg, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, agg_cfg, &output, acc.as_ref()); } } @@ -415,11 +440,12 @@ pub fn create_engine_multi_timestamp( storage_backend: Default::default()}); let sketch_index = std::sync::Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()); + let resolver = std::sync::Arc::new(SeriesIdResolver::new()); let agg_cfg = streaming_config.get_aggregation_config(1).cloned().expect("agg 1"); for (timestamp, label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp - 1000, timestamp, key, 1); - sketch_index.ingest_precompute_for_agg_config(&agg_cfg, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, &agg_cfg, &output, acc.as_ref()); } ASAPQueryEngine::new(streaming_config, 1).with_sketch_index(sketch_index) } @@ -468,11 +494,12 @@ pub fn create_engine_multi_timestamp_with_window( storage_backend: Default::default()}); let sketch_index = std::sync::Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()); + let resolver = std::sync::Arc::new(SeriesIdResolver::new()); let agg_cfg = streaming_config.get_aggregation_config(1).cloned().expect("agg 1"); for (timestamp, label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); let output = PrecomputedOutput::new(timestamp - 1000, timestamp, key, 1); - sketch_index.ingest_precompute_for_agg_config(&agg_cfg, &output, acc.as_ref()); + ingest_with_fresh_resolver(&sketch_index, &resolver, &agg_cfg, &output, acc.as_ref()); } ASAPQueryEngine::new(streaming_config, 1).with_sketch_index(sketch_index) }