feat(precompute_engine): add DualWriteSink mirroring precompute writes - #156
Merged
Merged
Conversation
Phase 5 M2.3.4a — infrastructure for the precompute ingest migration.
Adds `DualWriteSink`, a new `OutputSink` implementation that fans
each batch of `(PrecomputedOutput, AggregateCore)` rows out to BOTH:
- the legacy `SketchStore` (so the existing query path keeps
working), and
- the new sid-keyed `SketchIndex` via
`append_precompute` (so the M2.3.5 query path has data to read).
Per-row, the sink:
1. Snapshots `HotReloadStreamingConfig` to find the AggregationConfig
for `output.aggregation_id`.
2. Zips `agg_cfg.grouping_labels.labels` with
`output.key.labels` to build the canonical attrs fingerprint.
3. Computes `sid = compute_sid(metric, attrs_fp, AggKind::Precompute
{ agg_type, parameters_canonical })`.
4. Registers the sid in `SketchIndex` on first sight (with
`capability: None`, `accuracy: None` — precomputes answer exact
stats).
5. Appends the cloned accumulator via `append_precompute`.
Best-effort on the SketchIndex half: a missing AggregationConfig
(stale snapshot, retired config) logs at `warn!` and skips, so the
legacy SketchStore write isn't blocked.
Not wired into production `main.rs` yet — that's M2.3.4b. This PR is
pure additive infrastructure so the wire-in PR stays small.
2 new unit tests: full dual-write round-trip with one Sum-config +
PrecomputedOutput, plus graceful skip when agg_id is missing from
the streaming snapshot. 802/804 lib tests pass (2 pre-existing
`schema_timeline_dispatch` failures unrelated).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
3 tasks
zzylol
added a commit
that referenced
this pull request
May 13, 2026
Phase 5 M2.3.4b — replaces the production `StoreOutputSink` in `main.rs` with `DualWriteSink` (introduced in PR #156). Every precompute the worker emits now lands in BOTH the legacy `SketchStore` (which still serves queries today) AND the new sid-keyed `SketchIndex` via `append_precompute`. The legacy query path is unchanged — SketchIndex is being populated but no consumer reads precompute payloads from it yet. That's M2.3.5. Per-row overhead is one snapshot read + per-row agg-id lookup + sid hash + Box clone, all paid on the worker thread. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
3 tasks
zzylol
added a commit
that referenced
this pull request
May 13, 2026
…hIndex only (#160) Phase 5 M2.3.6a — first step of the SketchStore retirement. The production OutputSink is now `SketchIndexSink` (renamed and slimmed-down successor to `DualWriteSink` from PR #156); it writes ONLY to `SketchIndex` via `append_precompute`. The legacy `SketchStore` agg_id-keyed write path is gone from `main.rs`. Reads have already migrated to `SketchIndex` in PR #159 (M2.3.5b), so the legacy store no longer participates in either side of the ingest / query loop. The `Store` trait, `store/{global,per_key, common}.rs`, and the eviction service still exist; subsequent M2.3.6 sub-PRs delete them and rename `SketchIndex → SketchStore`. Test fixtures and the schema eviction service kept their existing `Store` plumbing untouched (so this PR's blast radius stays contained). 804/806 lib tests pass — the 2 pre-existing `schema_timeline_dispatch` failures predate this change. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
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.
Summary
Phase 5 M2.3.4a — infrastructure for the precompute ingest migration. Adds
DualWriteSink, a newOutputSinkimplementation that fans each batch of(PrecomputedOutput, AggregateCore)rows out to BOTH:SketchStore(existing query path keeps working), andSketchIndexviaappend_precompute(M2.3.5 query path will read from it).Not wired into production
main.rsyet — that's M2.3.4b. This PR is pure additive infrastructure so the wire-in PR stays small and reviewable.How it works
Per row in the batch, the sink:
HotReloadStreamingConfigto find theAggregationConfigforoutput.aggregation_id.agg_cfg.grouping_labels.labelswithoutput.key.labelsto build the canonical attrs fingerprint.sid = compute_sid(metric, attrs_fp, AggKind::Precompute { agg_type, parameters_canonical }).SketchIndexon first sight (capability: None,accuracy: None— precomputes are exact).append_precompute.Best-effort on the SketchIndex half: a missing
AggregationConfig(stale snapshot, retired config) logs atwarn!and skips, so a transient SketchIndex miss never blocks the legacy SketchStore write.Test plan
cargo test -p data_plane --lib precompute_engine::output_sink— 2/2 pass.cargo test -p data_plane --lib— 802/804 (2 pre-existingschema_timeline_dispatchfailures unrelated).cargo check --workspaceclean.🤖 Generated with Claude Code