refactor(data_plane): rekey ingest bucketing from (agg_id, group_key) to sid (B7.6) - #284
Merged
Merged
Conversation
… to sid (B7.6) Schema retirement #5 step 6: the precompute engine's per-bucket state used to be keyed by `(agg_id, group_key)`. The grouping label values already fold into the sid via `SeriesIdResolver`'s `(metric, attrs_fingerprint, agg_kind_canonical)` identity contract, so the tuple collapses to a single u64 — `sid` is now the bucket key throughout the ingest → router → worker path. Scope: - `WorkerMessage::GroupSamples` / `AccumulatorInput` carry `sid: u64, policy_fp: PolicyFingerprint, group_key: String` instead of `agg_id + group_key`. `policy_fp` is the source-config handle the worker uses to look up `AggregationConfig` from the hot-reload snapshot; `group_key` still travels for emit-time `KeyByLabelValues` rendering. This is a BREAKING change to `WorkerMessage`, but the enum is private to the data_plane crate. - `SeriesRouter::route_group_batch` hashes by `sid` alone (`worker_for_sid`). Same sid always lands on the same worker; bucket state stays single-owner. - `Worker::group_states: HashMap<u64, GroupState>` (was `HashMap<(u64, String), GroupState>`). `GroupState` gains `policy_fp` and `group_key` fields so `evict_orphaned_groups` can check policy liveness and the emit path can render labels without re-keying the bucket. - `process_group_samples` / `process_accumulator_input` take `(sid, policy_fp, group_key, ...)`. All in-crate call sites updated. - OTLP ingest helper `resolve_bucket_sid_for_agg_config(state, config, point_labels) → (sid, policy_fp)`: derives the bucket sid for one `(config, DP)` pair by resolving against `(metric, grouping-label- values, ExactAgg-of-config)`. Used by all three OTLP dispatch paths (raw points, opaque SketchEnvelope, modified-OTLP first-class sketches). Crucially, "attrs" for sid purposes is the GROUPING-LABEL projection of wire labels — not the full label set — so distinct `(rack, node, pod)` tuples under a `grouping_labels=[zone]` policy still roll up into one bucket per zone (the GROUP-BY semantic). - Regression test added: `drivers::ingest::otel::sid_bucketing_tests:: raw_otlp_buckets_by_sid_with_distinct_group_keys` drives `route_otlp_to_precompute` end-to-end with two `zone` values × two DPs each, asserts exactly two `GroupSamples` are emitted with distinct non-zero sids that round-trip through `SeriesIdResolver::lookup`. The test docstring documents a pre-existing `format_series_key`/`parse_labels_from_series_key` inconsistency that makes `extract_group_key_for` return "" for OTLP inputs; B7.6 bucketing is unaffected because it reads `point.labels` directly (HashMap lookup), not the joined series_key. Files touched (3): - `data_plane/src/precompute_engine/series_router.rs` — message shape + routing hash + test rename. - `data_plane/src/precompute_engine/worker.rs` — `GroupState` / `Worker.group_states` retyped, `get_or_create_group_state` / `process_group_samples` / `process_accumulator_input` / `evict_orphaned_groups` / `flush_all` reworked, 30+ test call sites updated to pass `(sid, PolicyFingerprint, group_key)`. - `data_plane/src/drivers/ingest/otel.rs` — three OTLP dispatch paths switched to sid-bucketing via new `resolve_bucket_sid_for_agg_config` helper; added `sid_bucketing_tests` mod with the regression test. Test plan: - `cargo build -p data_plane` — clean. - `cargo test -p data_plane --lib` — 713 passed / 2 ignored, no new failures vs. main. - `cargo test -p data_plane` integration suite — same 2 pre-existing failures as origin/main (`controller_plan_to_query_full_roundtrip_ ddsketch` / `_kll`); verified by re-running on origin/main HEAD. Unrelated to B7.6. NOT in scope (left for B7.7): - `output_sink.rs` already reads `output.policy_fp` (no `agg_id`); no changes needed there. - `backfill/processor.rs` still uses `(agg_id, group_key)` internally — retired by B7.7. - `AggregationConfig::aggregation_id()` accessor remains; retiring it is deferred until B7.6 + B7.7 both land (per task brief). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Merged
4 tasks
zzylol
added a commit
that referenced
this pull request
May 18, 2026
…y roundtrip cleanly (#285) `format_series_key` (in `drivers/ingest/otel.rs`) emitted unquoted `key=value` pairs, but `parse_labels_from_series_key` (in `precompute_engine/worker.rs`) required quoted `key="value"`. The mismatch meant `IngestState::extract_group_key_for` returned `""` for every OTLP wire-format input, and any other consumer that roundtripped a freshly-formatted series key through the parser got an empty label map back. Discovery: PR #284 (B7.6 ingest sid rekey) noticed the bug while implementing the bucketing rekey, side-stepped it by having the new helper read `point.labels` directly via HashMap lookup, and left a TODO-style note in the test docstring. Bucketing was unaffected because B7.6 routes by sid; emit-time `KeyByLabelValues` content elsewhere was silently broken. Decision: option (B) — bend `format_series_key`. The quoted PromQL- style `metric{k="v",...}` form is the canonical shape every other producer / consumer in the data plane already uses: * `render_series_key` in `storage_engines/sketch_db/backfill/prometheus_reader.rs` emits quoted-with-escapes (matches the Prometheus wire format) * `RawSample.labels`' rustdoc documents the quoted form * `sample_matches` in `storage_engines/sketch_db/backfill/raw_sample_reader.rs` strips `"` from values when parsing — expects quoted * Every existing parser test passes the quoted form * Nothing persists `format_series_key`'s output to disk (it flows into in-memory `WorkerMessage::GroupSamples` payloads and debug log lines only) `format_series_key` now escapes embedded `"`, `\`, `\n` per the PromQL lexer rules (matching `render_series_key`'s `escape_label_value`). The parser walks past `\<x>` escape pairs when scanning for the closing quote so values containing embedded `"` no longer terminate early. The returned `&str` is still the raw (un-decoded) slice — a new `decode_label_value(&str) -> Cow<str>` helper unescapes when needed. Returning the un-decoded slice keeps the existing `HashMap<&str, &str>` API (and its `processor.rs` caller, which is off-limits this PR for the B7.7 parallel agent) working without change; most live callers compare against literal config values that never contain escapable characters, so the borrow is fine. Regression coverage (new `series_key_roundtrip_tests` module in `otel.rs` + new unit tests in `worker.rs`): * canonical PromQL quoted shape pinned * roundtrip with commas in value * roundtrip with equals in value * roundtrip with embedded `"` (exercises the escape pair scan + `decode_label_value`) * roundtrip with `\` in value * roundtrip with `\n` in value * roundtrip with all metacharacters in one value * empty-labels bare-braces case * `decode_label_value` borrows when no escapes, unescapes when present, passes unknown escapes through verbatim All 725 `cargo test -p data_plane --lib` tests pass (2 pre-existing ignored, unchanged). Updated the stale "this returns empty" note in B7.6's `raw_otlp_buckets_by_sid_with_distinct_group_keys` test docstring to point at this fix. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
4 tasks
zzylol
added a commit
that referenced
this pull request
May 18, 2026
…_id to sid (B7.7) (#286) 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<u64, SidBucket>` (sid-keyed) instead of `HashMap<String, Vec<RawSample>>` (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) <noreply@anthropic.com>
This was referenced May 18, 2026
Closed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Scope
Schema retirement #5 step 6 (issue #272). The precompute engine's per-bucket state used to be keyed by
(agg_id, group_key). The grouping label values already fold into the sid viaSeriesIdResolver's(metric, attrs_fingerprint, agg_kind_canonical)identity contract, so the tuple collapses to a singleu64—sidis now the bucket key throughout the ingest → router → worker path.WorkerMessage shape change (BREAKING within the data_plane crate)
WorkerMessage::GroupSamples/AccumulatorInputcarrysid: u64,policy_fp: PolicyFingerprint,group_key: Stringin place of the oldagg_id+group_keypair.sid— registry-allocated bucket identity; folds in(metric, attrs_fingerprint, agg_kind_canonical). Worker keysgroup_stateson this.policy_fp— sourceAggregationConfigfingerprint. Worker looks up itsAggregationConfigviasnap.get_aggregation_config(policy_fp.as_u64()).group_key— kept for emit-timeKeyByLabelValuesrendering.WorkerMessageispubbut only used inside the data_plane crate (the precompute engine's private dispatch contract); no external crate consumes it.Files touched
data_plane/src/precompute_engine/series_router.rs— message shape + routing hash (worker_for_sidreplacesworker_for_group) + test renamed totest_consistent_sid_routing.data_plane/src/precompute_engine/worker.rs—GroupState/Worker.group_statesretyped toHashMap<u64, _>;get_or_create_group_state/process_group_samples/process_accumulator_input/evict_orphaned_groups/flush_allreworked; 30+ test call sites updated to pass(sid, PolicyFingerprint, group_key).data_plane/src/drivers/ingest/otel.rs— all three OTLP dispatch paths switched to sid-bucketing via new helperresolve_bucket_sid_for_agg_config(state, config, point_labels). Crucially, "attrs" for sid resolution is the GROUPING-LABEL projection of wire labels (not the full label set), so distinct(rack, node, pod)tuples under agrouping_labels=[zone]policy still roll up into one bucket per zone (the GROUP-BY semantic).Test added
drivers::ingest::otel::sid_bucketing_tests::raw_otlp_buckets_by_sid_with_distinct_group_keys— drivesroute_otlp_to_precomputeend-to-end with twozonevalues × two DPs each, asserts:WorkerMessage::GroupSamplesare emittedSeriesIdResolver::lookuprecords for(metric, "zone=<zv>;", ExactAgg-canonical)policy_fp = config.aggregation_id()on every messageThe test docstring documents a pre-existing
format_series_key/parse_labels_from_series_keyinconsistency that makesextract_group_key_forreturn "" for OTLP inputs; B7.6 bucketing is unaffected because it readspoint.labelsdirectly (HashMap lookup), not the joined series_key.Test plan
cargo build -p data_plane— cleancargo test -p data_plane --lib— 713 passed / 2 ignored, no regressions vs. maincargo test -p data_planeintegration suite — same 2 pre-existing failures as origin/main (controller_plan_to_query_full_roundtrip_ddsketch/_kll, both reproducible on a freshorigin/mainworktree). Unrelated to B7.6.NOT in scope (B7.7's domain)
output_sink.rsalready readsoutput.policy_fp(noagg_idinPrecomputedOutput); no changes needed there.backfill/processor.rsstill uses(agg_id, group_key)internally — B7.7.AggregationConfig::aggregation_id()accessor remains; retiring it is deferred until B7.6 + B7.7 both land (per task brief).🤖 Generated with Claude Code