From d4f5fc0a485e50ec88f69b1697515eaba333d1da Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 18 May 2026 05:53:47 -0600 Subject: [PATCH] refactor(data_plane): rekey backfill processor + SketchStore from agg_id to sid (B7.7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third sub-step of schema-retirement #5 (issue #272), following B7.6's ingest-side rekey. The backfill processor's per-window grouping is now keyed by `sid: u64` instead of `group_key: String`, matching B7.6's sid-bucketed worker state. The SketchStore exposes a new sid-direct write path (`ingest_precompute_with_sid`) so callers that already hold the bucket sid skip the resolver round-trip inside the mint-driven `ingest_precompute_for_agg_config` wrapper. ## Sites rekeyed - `data_plane/src/storage_engines/sketch_db/backfill/processor.rs` - `process_window` now groups raw samples into a `HashMap` (sid-keyed) instead of `HashMap>` (group_key-keyed). - New helper `resolve_backfill_bucket_sid` mirrors `resolve_bucket_sid_for_agg_config` from `drivers/ingest/otel.rs` so backfill and live ingest mint the SAME sid for the same `(metric, grouping-values, agg_kind)` tuple. This is the invariant that lets backfill writes land in the same store row live ingest already populated for `[created_at, ∞)`. - Per-bucket writes go through the new `ingest_precompute_with_sid` path; the mint-driven sibling is no longer called from this file. - Resolver-less fallback (legacy / registry-only test setups) keeps a stable per-`group_key` bucket id so accumulator builds still preserve sample ordering — but the write itself is skipped in that branch anyway (no resolver ⇒ no precompute write, matching pre-B7.7 behaviour). - `data_plane/src/storage_engines/sketch_db/index/mod.rs` - New `pub fn ingest_precompute_with_sid(sid, agg_cfg, output, accumulator)` takes the bucket sid directly. The existing `ingest_precompute_for_agg_config` is refactored into a thin mint-driven wrapper that delegates to the new entry point — callers that don't yet hold the sid (the live `SketchStoreSink`) keep working unchanged. - Extracted `build_attrs_fp_and_label_map` shared by both methods so the mint-driven path (B7.6) and the sid-direct path (B7.7) stay byte-identical on the values they hand to the index. ## Tests added - `process_window_buckets_by_sid_via_resolver` — drives `process_window` end-to-end with two distinct svc values × two samples each, asserts exactly two sids land in the SketchStore, both `classify()` as `Hit`, and registry provenance is one entry per window. - `backfill_sid_matches_live_ingest_sid_for_same_grouping_values` — locks the live-vs-backfill sid namespace invariant: the sid the backfill helper computes for `(cfg, "latency{svc=a,zone=z0}")` must equal what the live ingest path's `resolve_bucket_sid_for_agg_config` mirror computes for the same `(metric, grouping-values, agg_kind)` tuple via the SAME resolver. ## Not in scope (deferred follow-ups) - `output_sink.rs` production code already consumes `output.policy_fp` (PR #284's report: "no changes needed there"). Its only `aggregation_id()` site is in a test that builds a `StreamingConfig` map keyed by policy_fp.as_u64() — the legitimate policy-registry use, not a bucket key. - `worker.rs` / `series_router.rs` / `drivers/ingest/otel.rs` are B7.6's domain (already merged) — not touched. - Remaining `aggregation_id()` accessor sites are all test-side `StreamingConfig` map-key uses (the map IS keyed by policy_fp.as_u64()) — those stay until the accessor itself is retired after #4 (re-enable ignored tests). ## Test plan - `cargo build -p data_plane` — clean - `cargo test -p data_plane --lib` — 715 passed / 2 ignored, no regressions vs origin/main - `cargo test -p data_plane` integration suite — same 2 pre-existing failures `controller_plan_to_query_full_roundtrip_ddsketch` / `_kll` PR #284 confirmed are pre-existing - Both new regression tests pass Co-Authored-By: Claude Opus 4.7 (1M context) --- .../sketch_db/backfill/processor.rs | 284 ++++++++++++++++-- .../storage_engines/sketch_db/index/mod.rs | 91 ++++-- 2 files changed, 336 insertions(+), 39 deletions(-) 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 5119ac75..55bd7bbb 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs @@ -69,15 +69,18 @@ use std::sync::Arc; use async_trait::async_trait; use tracing::debug; -use crate::storage_engines::types::{AggregateCore, HotReloadStreamingConfig, KeyByLabelValues}; +use crate::drivers::ingest::canonical_attrs_fingerprint; +use crate::drivers::ingest::series_resolver::SeriesIdResolver; use crate::precompute_engine::worker::parse_labels_from_series_key; +use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; +use crate::storage_engines::types::{AggregateCore, HotReloadStreamingConfig, KeyByLabelValues}; use asap_types::aggregation_config::AggregationConfig; use asap_types::PolicyFingerprint; -use super::BackfillRegistry; +use super::raw_sample_reader::RawSample; use super::window_builder::build_backfilled_accumulator; use super::worker::WindowProcessor; -use super::raw_sample_reader::RawSample; +use super::BackfillRegistry; /// Turn a series key into the `group_key` string the /// grouping_labels-based partitioning produces in live ingest. @@ -109,6 +112,54 @@ fn build_group_key_label_values(group_key: &str) -> KeyByLabelValues { KeyByLabelValues::new_with_labels(labels) } +/// B7.7 — resolve the bucket sid for a backfill sample's +/// `(config, series_key)` pair. Mirrors +/// `resolve_bucket_sid_for_agg_config` in `drivers/ingest/otel.rs` +/// but adapted to the backfill side: the raw sample carries its labels +/// embedded in `series_key` (the `metric{k1="v1",k2="v2"}` text shape +/// `RawSample::labels` holds), so we parse them out first. +/// +/// The sid identity tuple is `(metric, attrs_fp, agg_kind_canonical)` +/// — identical to what the live ingest path computes, so the same +/// `(metric, grouping-values, agg_kind)` produces the SAME sid no +/// matter which path (live or backfill) saw the sample first. That +/// invariant is what lets backfill writes land in the same store +/// row the live ingest already populated for `[created_at, ∞)`. +fn resolve_backfill_bucket_sid( + resolver: &SeriesIdResolver, + config: &AggregationConfig, + series_key: &str, +) -> u64 { + let labels = parse_labels_from_series_key(series_key); + let grouping_pairs: Vec<(&str, &str)> = config + .grouping_labels + .labels + .iter() + .map(|name| (name.as_str(), *labels.get(name.as_str()).unwrap_or(&""))) + .collect(); + let attrs_fp = canonical_attrs_fingerprint(&grouping_pairs); + let agg_kind = AggKind::ExactAgg { + agg_type: config.aggregation_type, + parameters_canonical: canonical_parameters(&config.parameters), + spatial_filter_canonical: config.spatial_filter_normalized.clone(), + }; + let agg_kind_canonical = agg_kind.canonical_string(); + resolver.resolve(&config.metric, &attrs_fp, &agg_kind_canonical) +} + +/// Fallback bucket id for the resolver-less code path (registry-only +/// processors / legacy tests). Stable per `group_key` so all samples +/// in one group still aggregate into one bucket; the value never +/// reaches `SketchStore` because the resolver-less branch in +/// `process_window` skips the precompute write entirely. +fn fallback_bucket_id(group_key: &str) -> u64 { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + let mut h = DefaultHasher::new(); + group_key.hash(&mut h); + h.finish() +} + /// `WindowProcessor` that rebuilds sketches from raw samples and /// writes them to the store. Holds references to the shared /// registries / config so each window can look up its own config @@ -195,6 +246,14 @@ impl BackfillWindowProcessor { } } +/// One per-sid bucket assembled by [`BackfillWindowProcessor::process_window`]. +/// Carries the grouping-label string alongside the samples so emit-time +/// `KeyByLabelValues` rendering matches what the live ingest path produces. +struct SidBucket { + group_key: String, + samples: Vec, +} + #[async_trait] impl WindowProcessor for BackfillWindowProcessor { async fn process_window( @@ -205,17 +264,44 @@ impl WindowProcessor for BackfillWindowProcessor { ) -> Result<(), Box> { let config = self.config_for_agg(agg_id)?; - // Group samples by `group_key` — the same partitioning - // live ingest does. Uses insertion-order preserving Vec - // per group so §10.5 ordering is preserved within each - // group's sample stream. - let mut by_group: HashMap> = HashMap::new(); + // B7.7 — sid-keyed bucketing. Per the schema-retirement #5 + // step 6 plan, the backfill processor's per-window grouping is + // rekeyed from `group_key: String` to `sid: u64`. The grouping + // label values fold into the sid via + // `SeriesIdResolver::resolve(metric, attrs_fp, agg_kind)` — + // the same identity contract the live ingest path uses, so a + // backfilled bucket lands on the SAME sid that live ingest + // would mint for the same `(metric, grouping-values, agg_kind)` + // tuple. `group_key` is kept alongside the sid so emit-time + // `KeyByLabelValues` rendering on `PrecomputedOutput` matches + // what live ingest produces. Uses insertion-order preserving + // Vec per bucket so §10.5 ordering is preserved within each + // bucket's sample stream. + // + // When no SeriesIdResolver is attached (legacy / registry-only + // tests), we fall back to a stable per-`group_key` bucket id — + // the per-window write is skipped anyway in that path, so the + // bucket identity doesn't matter beyond preserving sample + // ordering for the (unused) accumulator builds. + let resolver_opt = self.series_resolver.as_ref(); + let mut by_bucket: HashMap = HashMap::new(); for sample in samples { let group_key = extract_group_key(&sample.labels, &config); - by_group.entry(group_key).or_default().push(sample); + let sid = match resolver_opt { + Some(r) => resolve_backfill_bucket_sid(r.as_ref(), &config, &sample.labels), + None => fallback_bucket_id(&group_key), + }; + by_bucket + .entry(sid) + .or_insert_with(|| SidBucket { + group_key: group_key.clone(), + samples: Vec::new(), + }) + .samples + .push(sample); } - if by_group.is_empty() { + if by_bucket.is_empty() { // Empty window — no samples, no writes. Still count as // "processed" since the worker's tick_progress will // increment. @@ -228,11 +314,19 @@ impl WindowProcessor for BackfillWindowProcessor { return Ok(()); } - let mut batch: Vec<(crate::storage_engines::types::PrecomputedOutput, Box)> = - Vec::with_capacity(by_group.len()); - - for (group_key, group_samples) in by_group { - let accumulator = build_backfilled_accumulator(&config, &group_samples); + // Build per-sid `(sid, PrecomputedOutput, accumulator)` triples. + // Carrying the sid alongside the pair lets the write loop hand + // it straight to `ingest_precompute_with_sid` instead of + // re-resolving inside the mint-driven path. + let mut batch: Vec<( + u64, + crate::storage_engines::types::PrecomputedOutput, + Box, + )> = Vec::with_capacity(by_bucket.len()); + + for (sid, bucket) in by_bucket { + let SidBucket { group_key, samples } = bucket; + let accumulator = build_backfilled_accumulator(&config, &samples); // Keyed accumulators (MultipleSubpopulation) carry their // subpopulation keys internally; the PrecomputedOutput's // `key` represents the *group* key (grouping_labels @@ -250,8 +344,7 @@ impl WindowProcessor for BackfillWindowProcessor { self.job_id, PolicyFingerprint::from_config(&config), ); - let _ = agg_id; - batch.push((output, accumulator)); + batch.push((sid, output, accumulator)); } // Phase 5 M2.3.6g — replayed batches land in SketchStore only. @@ -263,11 +356,17 @@ impl WindowProcessor for BackfillWindowProcessor { // with a warn — sid minting requires the shared resolver. if let Some(idx) = self.sketch_index.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), + Some(_resolver) => { + // B7.7 — sid is pre-resolved per bucket above; hand + // it directly to the index's sid-direct ingest + // path. The mint-driven sibling + // `ingest_precompute_for_agg_config` would resolve + // to the same sid (the resolver is idempotent), + // but the round-trip is redundant now that we hold + // the value. + for (sid, output, accumulator) in &batch { + idx.ingest_precompute_with_sid( + *sid, &config, output, accumulator.as_ref(), @@ -720,4 +819,145 @@ mod tests { .expect("within-retention start should be accepted"); assert!(registry.get(job_id).is_some()); } + + // ─── B7.7: sid-keyed bucketing tests ───────────────────────────────── + + /// Regression for B7.7 (schema-retirement #5 step 6): the + /// backfill processor groups raw samples by `sid: u64` and writes + /// each bucket via `SketchStore::ingest_precompute_with_sid`. The + /// sids it allocates match what the shared `SeriesIdResolver` + /// would mint for the same `(metric, grouping-values, agg_kind)` + /// tuple — i.e. live ingest and backfill share one sid namespace. + /// + /// Drives `process_window` end-to-end with samples spanning two + /// distinct `svc` values × two samples each. Asserts: + /// - exactly two `SketchInstanceMetadata` entries land in the + /// `SketchStore` (one per distinct sid bucket) + /// - their sids equal what the shared `SeriesIdResolver` would + /// mint for the same `(metric, grouping-values, agg_kind)` + /// tuple (so live-vs-backfill sid identity holds) + /// - both sids `classify()` as `Hit` (i.e. the per-sid append + /// paths landed under the same sids the metadata was + /// registered with) + /// - the registry recorded one provenance entry for the window + #[tokio::test] + async fn process_window_buckets_by_sid_via_resolver() { + use crate::drivers::ingest::series_resolver::SeriesIdResolver; + use crate::storage_engines::sketch_db::index::{SidLookup, SketchStore}; + + let cfg = sum_config(1, "latency", vec!["svc"]); + let fp = cfg.aggregation_id(); + let streaming = streaming_config_with(cfg.clone()); + let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); + let registry = Arc::new(BackfillRegistry::new()); + let sketch_index = Arc::new(SketchStore::new()); + let resolver = Arc::new(SeriesIdResolver::new()); + + let job_id = registry.create( + fp, + (0, 100), + BackfillSource::Prometheus { url: "x".into() }, + 1, + ); + + let processor = BackfillWindowProcessor::new(hot, registry.clone(), job_id) + .with_sketch_index(sketch_index.clone()) + .with_series_resolver(resolver.clone()); + + // Two distinct svc values × two samples each. Same window + // (0, 100). After the processor runs, we expect exactly two + // sids in the SketchStore — one per distinct svc value. + let samples = vec![ + RawSample { + labels: "latency{svc=\"a\"}".into(), + timestamp_ms: 10, + value: 1.0, + }, + RawSample { + labels: "latency{svc=\"a\"}".into(), + timestamp_ms: 20, + value: 2.0, + }, + RawSample { + labels: "latency{svc=\"b\"}".into(), + timestamp_ms: 15, + value: 3.0, + }, + RawSample { + labels: "latency{svc=\"b\"}".into(), + timestamp_ms: 25, + value: 4.0, + }, + ]; + processor + .process_window(fp, (0, 100), samples) + .await + .expect("happy path"); + + // Two distinct sids landed in the index. + assert_eq!( + sketch_index.instance_count(), + 2, + "one sid per distinct svc bucket" + ); + + // The sids the processor allocated equal what + // `SeriesIdResolver::lookup` would return for the same + // `(metric, grouping-values, agg_kind)` tuple — i.e. live + // ingest and backfill share one sid namespace. + let sid_a = resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"a\"}"); + let sid_b = resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"b\"}"); + assert_ne!(sid_a, sid_b, "distinct svc values mint distinct sids"); + assert_eq!(sketch_index.classify(sid_a), SidLookup::Hit); + assert_eq!(sketch_index.classify(sid_b), SidLookup::Hit); + + // Provenance was recorded once per window (not once per + // bucket) — same shape as the pre-rekey path. + let written = registry.windows_written_by(job_id); + assert_eq!(written, vec![(fp, (0u64, 100u64))]); + } + + /// B7.7 invariant: the sid the backfill processor mints for a + /// `(config, grouping-values)` tuple is bit-equal to the sid the + /// live ingest path's `resolve_bucket_sid_for_agg_config` would + /// mint via the SAME `SeriesIdResolver`. Locks the "live and + /// backfill share one sid namespace" contract — without it, the + /// `[created_at, ∞)` and `[0, created_at)` halves of the agg's + /// timeline would live under DIFFERENT sids and the query path + /// would only see half the history. + #[test] + fn backfill_sid_matches_live_ingest_sid_for_same_grouping_values() { + use crate::drivers::ingest::canonical_attrs_fingerprint; + use crate::drivers::ingest::series_resolver::SeriesIdResolver; + use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; + + let cfg = sum_config(1, "latency", vec!["svc", "zone"]); + let resolver = SeriesIdResolver::new(); + + // Backfill side: derive sid via the new helper. + let backfill_sid = resolve_backfill_bucket_sid( + &resolver, + &cfg, + "latency{svc=\"a\",zone=\"z0\"}", + ); + + // Live side: mirror what `resolve_bucket_sid_for_agg_config` + // in drivers/ingest/otel.rs does, manually here so the test + // doesn't need to drive the OTLP pipeline. + let live_attrs_fp = + canonical_attrs_fingerprint(&[("svc", "a"), ("zone", "z0")]); + let live_agg_kind = AggKind::ExactAgg { + agg_type: cfg.aggregation_type, + parameters_canonical: canonical_parameters(&cfg.parameters), + spatial_filter_canonical: cfg.spatial_filter_normalized.clone(), + }; + let live_sid = + resolver.resolve(&cfg.metric, &live_attrs_fp, &live_agg_kind.canonical_string()); + + assert_eq!( + backfill_sid, live_sid, + "backfill and live ingest MUST mint the same sid for the same \ + (metric, grouping-values, agg_kind) tuple" + ); + } } 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 ab7bdec5..c61be423 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -45,6 +45,41 @@ fn now_ms() -> u64 { .unwrap_or(0) } +/// Joint helper shared by [`SketchStore::ingest_precompute_for_agg_config`] +/// and [`SketchStore::ingest_precompute_with_sid`] — folds the +/// grouping-label values on `output` against the +/// `agg_cfg.grouping_labels` ordering into: +/// +/// 1. `attrs_fp`: the `key=value;` canonical attrs-fingerprint string +/// `SeriesIdResolver` uses as one third of the sid identity tuple. +/// 2. `label_values_map`: the `BTreeMap` shape +/// `append_precompute` records on the per-window precompute row. +/// +/// Both are pure functions of `(agg_cfg, output.key)` — kept together +/// so the mint-driven path (B7.6) and the sid-direct path (B7.7) stay +/// byte-identical on the values they hand to the index. +fn build_attrs_fp_and_label_map( + agg_cfg: &asap_types::aggregation_config::AggregationConfig, + output: &crate::storage_engines::types::PrecomputedOutput, +) -> (String, BTreeMap) { + 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()); + } + (attrs_fp, label_values_map) +} + /// Metadata for one logical sketch instance, keyed by `series_id`. /// Populated at ingest time when a sketch DataPoint with a fresh sid /// arrives (or `(metric, attrs)` produces a fresh sid via the @@ -690,22 +725,13 @@ impl SketchStore { output: &crate::storage_engines::types::PrecomputedOutput, accumulator: &dyn crate::storage_engines::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()); - } - + // B7.7 — this wrapper now derives the sid via `mint_sid` and + // delegates to `ingest_precompute_with_sid`. Callers that + // already hold the bucket sid (B7.6's worker passes it on the + // `WorkerMessage`; B7.7's backfill processor groups raw + // samples by sid up-front) skip the resolver round-trip by + // invoking the sid-direct sibling. + let (attrs_fp, _label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output); let agg_kind = AggKind::ExactAgg { agg_type: agg_cfg.aggregation_type, parameters_canonical: canonical_parameters(&agg_cfg.parameters), @@ -724,6 +750,37 @@ impl SketchStore { // closure. let agg_kind_canonical = agg_kind.canonical_string(); let sid = mint_sid(&agg_cfg.metric, &attrs_fp, &agg_kind_canonical); + self.ingest_precompute_with_sid(sid, agg_cfg, output, accumulator) + } + + /// B7.7 sid-direct sibling of [`Self::ingest_precompute_for_agg_config`]. + /// + /// Callers that already hold the bucket sid (the live worker after + /// B7.6 reshaped its `WorkerMessage`, and the backfill processor + /// after B7.7 rekeyed its per-window grouping from group_key to + /// sid) skip the mint round-trip by handing the sid in directly. + /// The mint-driven [`Self::ingest_precompute_for_agg_config`] is + /// content-addressed and idempotent with this method — passing the + /// resolver-minted sid here yields the same state under the same + /// sid — so both methods can coexist while migration finishes. + /// + /// The §6.3 ingest barrier (`Retired` / `Expired` sids reject + /// writes) and first-sight metadata registration are identical to + /// the mint-driven path. + pub fn ingest_precompute_with_sid( + &self, + sid: u64, + agg_cfg: &asap_types::aggregation_config::AggregationConfig, + output: &crate::storage_engines::types::PrecomputedOutput, + accumulator: &dyn crate::storage_engines::types::AggregateCore, + ) -> Option { + let (_attrs_fp, label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output); + let key_names = &agg_cfg.grouping_labels.labels; + let agg_kind = AggKind::ExactAgg { + agg_type: agg_cfg.aggregation_type, + parameters_canonical: canonical_parameters(&agg_cfg.parameters), + spatial_filter_canonical: agg_cfg.spatial_filter_normalized.clone(), + }; match self.instance(sid) { None => { @@ -741,7 +798,7 @@ impl SketchStore { metric_name: agg_cfg.metric.clone(), group_by_keys, capability: Some(Capability::ExactAgg(agg_cfg.aggregation_type)), - agg_kind: agg_kind.clone(), + agg_kind, accuracy: None, first_seen_unix_ms: output.start_timestamp as i64, retired_at_ms: None,