fix(query-engine): expand self-keyed accumulators in range queries - #587
Merged
Conversation
…k) without keys_query execute_range_query_pipeline's single-population branch used each value group's own store-level group_key directly and never called get_keys() on the merged value accumulator, so self-keyed accumulators like CountMinSketchWithHeap (top-k) returned empty over a range instead of expanding into their top-k keys — the instant path already does this via collect_results_same_aggregation. Now every group tries merged.get_keys() first (mirroring the instant path exactly) and only falls back to the precomputed key list — the store's group_key for single-population metrics, or the separate keys aggregation's expansion for dual-population metrics — when get_keys() returns None. This also covers self-keyed accumulators stored under a non-None outer key (real Arroyo/worker.rs ingestion always wraps Some(key), even for empty grouping), not just the None-keyed case from the original report. Fixes #584. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Contributor
Author
|
This is reachable in production: The new tests ( Suggest gating the |
…eys_query in range queries Review on #587 caught a regression: the previous fix called merged.get_keys() unconditionally on every group's merged value accumulator, including dual-population groups (separate keys_query present). collect_all_results (instant path) branches globally on query shape instead — dual-population always goes through collect_results_separate_keys, which never consults the value accumulator's own get_keys() at all. A real, tested capability-matched config (sql.rs) pairs a self-keyed CountMinSketchWithHeap value aggregation with a separate DeltaSetAggregator keys aggregation; the previous fix let the heap's own (window-shifting) top-k keys silently override the keys aggregation's expansion for that config. Range queries now mirror collect_all_results exactly: dual-population groups always use the keys aggregation's expansion, full stop; only single-population groups let the value accumulator's own get_keys() take priority (falling back to the store-level group key), which is what #584 actually needed. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
milindsrivastava1997
marked this pull request as ready for review
August 24, 2026 12:23
milindsrivastava1997
added a commit
that referenced
this pull request
Aug 24, 2026
…-keyed top-k #587 (fixes #584, top-k self-keyed accumulator expansion for single-population range queries) landed on main while this branch was in flight, independently rewriting the same region of execute_range_query_pipeline that #583's fix rewrote -- both branches diverged from the same base commit (e8d4d3b). The two fixes are orthogonal by design, confirmed while reconciling: dual-population groups (#583's KeysSource::PerStep) never consult the value accumulator's own get_keys() at all, in either version -- #587's own PR review explicitly established that invariant (a self-keyed CountMinSketchWithHeap value paired with a separate DeltaSetAggregator keys aggregation is a real capability-matched config; the keys aggregation's expansion must always win for dual-population). So merging didn't require picking a side, just composing both: - KeysSource::PerStep (dual-population, #583): resolves expansion keys from the keys aggregation, per output step, before ever touching the value merge -- unchanged from #583's own commit. - KeysSource::Fixed (single-population, #584/#587): resolution moves to AFTER the value merge -- try the merged value accumulator's own get_keys() first (self-keyed, e.g. top-k), falling back to the store-level group key otherwise. - Adopted #587's fix to the None => all_data... branch: changed from filter_map (dropping every group_key=None group entirely -- the original, pre-#587 bug, which #583 had inherited unmodified since it never touched this branch) to a plain map that keeps group_key=None groups with an empty fallback list, letting the per-step self-keyed check populate the real keys. git's own 3-way auto-merge produced code that wouldn't compile (referenced fallback_keys/is_dual_population that don't exist in #583's KeysSource-based structure) and, separately, silently mis-aligned two different new tests' closing braces as shared context in native_range_query_tests.rs -- both caught and fixed by hand rather than trusted, per the resolving-merge-conflicts skill. Also extends #587's own range_query_dual_population_self_keyed_value_still_uses_keys_query test (previously a single-timestamp check) to add a mid-range keys change in the same test, so #583's per-step resolution and #587's self-keyed-override protection are both pinned holding simultaneously, not just verified as separately-provable-orthogonal. Verified: 19/19 native_range_query_tests pass (up from 16 pre-merge, +3 from #587), full workspace suite (cargo test --workspace) 0 failures, cargo clippy --lib --tests clean. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
milindsrivastava1997
added a commit
that referenced
this pull request
Aug 25, 2026
Two fixes from a code review of the #583/#587 merge (PR #595): - promql.rs: guard keys_tumbling_window_ms against 0. A zero window_size_ms on the key aggregation's config would make execute_range_query_pipeline's per-step scan_window (`while t < window_end { ...; t += step_increment }`) loop forever, since t would never advance. The value side is accidentally protected from this by validate_range_query_params's `step.is_multiple_of(tumbling_window_ms)` check (only 0 is a multiple of 0); keys had no equivalent, so added an explicit check. - mod.rs: distinguish "keys merge succeeded but get_keys() returned None" (e.g. a DeltaSetAggregator remove with no matching add resolved in this window) from the routine, expected "no buckets in this window at all" case. The former now warn!s -- it means a merge DID happen but couldn't resolve a key set, which is worth visibility on -- while the latter (which fires routinely, e.g. before a key first exists, in nearly every dual-population test in this file) stays debug! to avoid making normal usage noisy. Two other findings from the same review were investigated (spawned a subagent to write confirming/refuting tests, no production changes) and confirmed real, but are out of scope for this PR -- filed separately with their proving tests rather than fixed here: - #600: execute_range_query_pipeline's scan_window steps the bucket map by window_size_ms, but Sliding-aggregation buckets are actually persisted at slide_interval_ms (confirmed via precompute_engine/window_manager.rs). Affects both the keys side (introduced by #583) and the value side (pre-existing, predates #583/#587 entirely). - #601: build_bucket_map doesn't sort same-start-timestamp buckets before NaiveMerger's sequential fold, which is order-sensitive for DeltaSetAggregatorAccumulator with 3+ colliding deltas (2-bucket collisions are order-independent via conflict-cancellation, which is why this wasn't caught by the earlier 2-bucket collision test in this file). Confirmed: same 3 logical deltas, different insertion order, different final answer. Also filed #598 (finish_range_context reads streaming_config twice, separate momentary read locks -- a hot-reload landing between them could give the value and key sides inconsistent config generations) and #599 (DeltaSetAggregator's per-step keys replay is O(range^2) by design, tracking the already-documented tradeoff outside the design doc). Verified: 19/19 native_range_query_tests pass, full workspace suite (cargo test --workspace) 0 failures, cargo clippy --lib --tests clean. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
milindsrivastava1997
added a commit
that referenced
this pull request
Aug 25, 2026
) * test(query-engine): add RED tests for per-step keys_query snapshot bug (#583) execute_range_query_pipeline fetches/merges keys_query once, anchored at the range's end, and reuses that single snapshot for every output step. Two new tests reproduce the two failure modes this causes: - DeltaSetAggregator: a key added partway through the range gets a phantom sample at earlier steps, before it actually existed. - SetAggregator: a key present only in an earlier window is excluded from the final (end-anchored) keys fetch entirely, so its whole series silently vanishes from the output instead of just its later samples. Both fail against current code (confirmed via targeted `cargo test --lib native_range_query_tests`); skipping the full-suite pre-commit hook for this commit since it's expected to fail on these intentionally-RED tests. * test(query-engine): check all 4 host-a/host-b x t=1000/2000 conditions Extend both #583 RED tests to assert presence/absence at every (key, timestamp) combination instead of just the one that first demonstrates the bug, via a shared key_has_sample_at helper: - DeltaSetAggregator (cumulative deltas): host-a present at 1000 and 2000; host-b absent at 1000, present at 2000. - SetAggregator (per-window snapshot, no accumulation): host-a present at 1000, absent at 2000; host-b absent at 1000, present at 2000. Still RED against current code. * test(query-engine): expand #583 RED coverage to 16 cases + NaiveMerger pin Extends the RED test suite for the per-step keys_query snapshot bug (#583) well beyond the original 2 cases: - SetAggregator/DeltaSetAggregator through the binary-expr arm path (build_arm_range_context), not just the plain range dispatch - A 5-window oscillating add/remove/add/remove/add sequence for DeltaSetAggregator, asserted at every intermediate step - A key change landing on an interior step, not just a range boundary - Keys and values bucket widths differing (mismatched tumbling granularities), which the existing fixtures couldn't expose since both aggregations shared one window size - Multiple independent groups (real grouping_labels): one group's key change must not leak into another's per-step output, including a sharper simultaneous-cross-add variant - A group with keys but zero value data anywhere, which today hard-fails the entire range query instead of being skipped - SetAggregator merging multiple colliding same-timestamp buckets within one window (previously only ever exercised with exactly one bucket per window) Also adds `assert_all_at`, a shared mismatch-collector so these multi-assertion tests report every divergence in one panic instead of stopping at the first failing assert, and `create_range_engine_dual_input_with_windows`, a superset of the existing fixture helper that lets value/key aggregations use different bucket widths. Separately pins (GREEN, not RED) that NaiveMerger's sequential pairwise fold -- not a flat merge_accumulators call -- is what makes DeltaSetAggregator's add/remove/add/remove/add replay chronologically correct; the fix this test suite is driving toward depends on that distinction and it's easy to get backwards. 16 tests total: 5 green, 11 RED. Design rationale for each case recorded in docs/583-range-keys-per-step-design.md. Skipping the full test/lint pre-commit hooks for this commit since it's expected to fail on these intentionally-RED tests (same as the precedent commit 15bb61e on this branch). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * docs(query-engine): record #583 fix design discussion Recap of the grilling session that produced the RED test suite in 13d945e and the implementation plan it's driving toward: bug recap, 8 numbered design decisions with reasoning (architecture, the generic keys_query widening formula, recompute-vs-incremental tradeoff, NaiveMerger ordering correctness, do_merge, non-fatal missing-group handling, cross-group isolation), and the full RED test inventory mapped to which decision each one pins. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * fix(query-engine): widen keys_query per range step, stage 1 of 583 fix Stage 1 of the #583 fix (per-step keys_query in range queries): plumbing only, behavior-inert on its own -- confirmed via `cargo test --lib native_range_query_tests` showing the exact same 5 passed / 11 failed split before and after this change. RangeQueryExecutionContext gains keys_lookback_ms and keys_tumbling_window_ms (both None when there's no separate keys_query), populated in finish_range_context by widening keys_query the same way values_query already is: lookback is derived from the instant window create_keys_query_params already computed (end - start), then start_ms.saturating_sub(lookback) re-anchors it across the whole range. This needs no AggregationType branching -- for SetAggregator the instant window is [end-window_size, end], so this produces a normal sliding window; for DeltaSetAggregator the instant window is [0, end], so the lookback equals end_ms and saturating_sub gives 0 for every current_time in the per-step loop (current_time <= end_ms always holds), i.e. "replay from the beginning," for free. Nothing consumes these two new fields yet -- that's stage 2, which actually replaces execute_range_query_pipeline's single global keys merge with the per-step one these fields make possible. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * style(query-engine): cargo fmt native_range_query_tests.rs Pre-existing drift from the RED-test commit (13d945e), which used --no-verify to skip the (then-failing-for-unrelated-reasons) pre-commit test hook and never actually ran through cargo fmt. No semantic change. Using --no-verify here too: the cargo-test hook stashes unstaged changes before running, so with stage 2's mod.rs changes still unstaged at this point it would (correctly) see only stage 1's code and report the still-expected 11 RED tests -- not a real failure, just a false negative from splitting a pure-formatting commit ahead of the logic commit that depends on it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * fix(query-engine): per-step keys merge in execute_range_query_pipeline, stage 2 of 583 fix Stage 2 of the #583 fix: the actual per-step keys merge. All 16 native_range_query_tests pass (5 that were already green, 11 that were RED), plus the full workspace suite (cargo test --workspace, 0 failures across every crate). What follows is exactly what was explained before making this change: 1. Replace the single keys fetch+merge with a raw fetch. Delete the fetch_and_merge_keys(...) call; replace with execute_store_query(&context.base.store_plan.keys_query) when Some -- same call the values side already uses, giving raw (unmerged) buckets, same shape as all_data. 2. Change what `groups` carries. Today each group carries a fixed Vec<KeyByLabelValues> (the one-time merged snapshot). Replaced with a small KeysSource enum: Fixed(Vec<KeyByLabelValues>) for single-population groups (unchanged -- they never had a per-step keys concern), or PerStep(&Vec<TimestampedBucket>) for dual-population groups -- a reference to that group's raw keys buckets, not yet merged. 3. Q6, at group-construction time. Where the code today does all_data.get(group_key).ok_or_else(|| "No value for key")? (hard- fails the whole query), changed to warn! + skip that one group via filter_map. 4. Inside the per-group loop, mirror the value side's own pattern exactly. The value side already builds a bucket_map once per group, then re-derives window_buckets fresh every current_time iteration. Added the identical second copy of that pattern for keys: build a keys_bucket_map once per group (only for PerStep groups), then inside the current_time loop, do the same windowed scan-and-merge -- using context.keys_lookback_ms/ context.keys_tumbling_window_ms from stage 1 instead of the value side's fields -- to get expansion_keys fresh at every step, instead of reusing one fixed set. 5. One new call flagged rather than snuck in: if a step's keys-window merge comes back empty or get_keys() returns None (e.g. DeltaSetAggregatorAccumulator with unresolved removals), that's treated as "skip this step's sample for this group" -- not a hard error. That's a natural extension of Q6's "non-fatal" philosophy to a narrower case (a specific step's keys, not a whole group's), but it wasn't explicitly one of the 8 grilled design decisions, so it was flagged and confirmed before implementing rather than decided silently. Two follow-up questions were asked and answered before implementing: Why step 1 (raw fetch instead of fetch_and_merge_keys)? -- fetch_and_merge_keys does two things: raw fetch, then merge_precomputed_outputs collapses all the fetched buckets into one merged accumulator per group. That collapse is literally Bug 1: once buckets are merged together, there's no way left to ask what the key set looked like at t=1000 specifically -- that information is gone. To merge per-step, the loop needs the raw, unmerged buckets still available when it reaches current_time, so it can merge only the subset whose start < current_time at each step. Calling fetch_and_merge_keys throws that away before the loop even starts. This mirrors how values already work: execute_store_query (raw fetch, no merge) happens once up front; merging happens later, per-step, via bucket_map + NaiveMerger. Keys need that same split -- step 1 is what makes step 2 possible at all, not an independent cleanup. Why the KeysSource enum in step 2? -- Two genuinely different cases exist for a group's expansion keys, carrying different data: single-population has no separate keys_query, so the value's own key IS the output key at every timestamp, unconditionally -- nothing to look up or merge (today's existing None branch, unchanged). Dual-population's key set has to be recomputed from raw keys buckets, per-step -- that's the whole fix. `groups` needs one uniform element type, but "how to get this group's expansion_keys" is fundamentally different shaped data for the two cases -- an already-final Vec<KeyByLabelValues> vs. a &Vec<TimestampedBucket> still needing per-step work. Two separate Option fields (one per case, "exactly one is ever Some" by convention) would allow invalid states (both Some, both None) that would just have to be trusted not to happen. The enum makes "it's one or the other, never both" a compile-time guarantee instead of a convention -- for mission- critical logic, the type system should rule out the invalid state rather than the author having to. A clarifying question was also asked and answered: does single-population need any new per-step logic too? No -- single-population doesn't need any new per-step logic. The #583 bug is specifically about a separate keys aggregation whose key set can drift independently of the value data over time (a DeltaSetAggregator/SetAggregator snapshot merged once and reused). Single-population has no such thing -- there's only one aggregation, and group_key (the value bucket's own stored key) IS the output key, permanently, by construction. expansion_keys = vec![group_key.clone()] isn't a merge result that could go stale -- it's a tautology, so there's nothing for Bug 1 to reuse-across-steps incorrectly. The one thing that does need to vary per step -- does this key actually have a sample at this particular timestamp -- is already handled, correctly, by the existing value-side per-step windowing (bucket_map/window_buckets, the `if !window_buckets.is_empty()` check). That logic predates #583, isn't part of the bug, and doesn't change in this fix; it's exactly the same mechanism dual-population's key side needs to newly mirror -- single-population already gets it for free because key and value are the same data. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * refactor(query-engine): extract widen_query_window, dedup values/keys widening finish_range_context's keys_query widening block (added for #583) duplicated the exact formula the existing values_query widening block used: lookback = end - start of the query's current window, then start = start_ms.saturating_sub(lookback), end = end_ms. Written inline at the time to keep that stage's diff small and easy to verify in isolation; noted as a follow-up rather than done then. Extracts widen_query_window(query: &mut StoreQueryParams, start_ms, end_ms) -> u64, used for both values_query and keys_query. Behavior- preserving: full lib suite (564 tests) and the 16 native_range_query tests unchanged before and after. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * refactor(query-engine): fix stale doc comment, dedup bucket_map/scan_window Follow-up cleanup from the #583 duplication survey (items 1 and 3; items 2 and 4 filed as #596 and #597 -- both need a real design decision, not a mechanical refactor, so left out of this commit). 1. fetch_and_merge_keys's doc comment still claimed it was shared by the instant and range paths. False since #583's fix replaced the range path's call site with a raw execute_store_query fetch -- corrected to say so. 2. execute_range_query_pipeline had the same ~10-line pattern written twice, once for values and once for keys (#583 introduced the second copy): build a bucket_map from (start,end)->bucket tuples, then scan a window range collecting matching buckets. Extracted build_bucket_map and scan_window as private helpers, used by both the value and key sides. Behavior-preserving: full lib suite (564 tests) and the 16 native_range_query tests unchanged before and after. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * docs(query-engine): update #583 design doc status now that the fix landed Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * fix(query-engine): warn on unresolved keys merge, guard zero keys window Two fixes from a code review of the #583/#587 merge (PR #595): - promql.rs: guard keys_tumbling_window_ms against 0. A zero window_size_ms on the key aggregation's config would make execute_range_query_pipeline's per-step scan_window (`while t < window_end { ...; t += step_increment }`) loop forever, since t would never advance. The value side is accidentally protected from this by validate_range_query_params's `step.is_multiple_of(tumbling_window_ms)` check (only 0 is a multiple of 0); keys had no equivalent, so added an explicit check. - mod.rs: distinguish "keys merge succeeded but get_keys() returned None" (e.g. a DeltaSetAggregator remove with no matching add resolved in this window) from the routine, expected "no buckets in this window at all" case. The former now warn!s -- it means a merge DID happen but couldn't resolve a key set, which is worth visibility on -- while the latter (which fires routinely, e.g. before a key first exists, in nearly every dual-population test in this file) stays debug! to avoid making normal usage noisy. Two other findings from the same review were investigated (spawned a subagent to write confirming/refuting tests, no production changes) and confirmed real, but are out of scope for this PR -- filed separately with their proving tests rather than fixed here: - #600: execute_range_query_pipeline's scan_window steps the bucket map by window_size_ms, but Sliding-aggregation buckets are actually persisted at slide_interval_ms (confirmed via precompute_engine/window_manager.rs). Affects both the keys side (introduced by #583) and the value side (pre-existing, predates #583/#587 entirely). - #601: build_bucket_map doesn't sort same-start-timestamp buckets before NaiveMerger's sequential fold, which is order-sensitive for DeltaSetAggregatorAccumulator with 3+ colliding deltas (2-bucket collisions are order-independent via conflict-cancellation, which is why this wasn't caught by the earlier 2-bucket collision test in this file). Confirmed: same 3 logical deltas, different insertion order, different final answer. Also filed #598 (finish_range_context reads streaming_config twice, separate momentary read locks -- a hot-reload landing between them could give the value and key sides inconsistent config generations) and #599 (DeltaSetAggregator's per-step keys replay is O(range^2) by design, tracking the already-documented tradeoff outside the design doc). Verified: 19/19 native_range_query_tests pass, full workspace suite (cargo test --workspace) 0 failures, cargo clippy --lib --tests clean. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> * fix(query-engine): guard scan_window against zero step, embed KeysSource's per-step data Two fixes from a second PR #595 review pass: 1. scan_window itself now asserts step_increment > 0 instead of relying on callers to never pass 0. The earlier fix (7312593) only guarded keys_tumbling_window_ms at its one call site in finish_range_context; scan_window has two callers (values, keys) and shouldn't depend on either having validated its own step source -- the value side's protection is itself just incidental (validate_range_query_params's step-is-multiple-of check). Kept as a release-mode assert!, not debug_assert!: a hung query is a production incident. 2. KeysSource::PerStep now carries its bucket_map, lookback_ms, and tumbling_window_ms directly as struct fields, built once at groups-construction time, instead of three separate Option fields at function scope that only stayed in sync by convention -- each re-unwrapped via .expect() on every iteration of the per-step loop. Same reasoning that motivated choosing this enum over two raw Option fields in the first place (see 583-range-keys-per-step design doc, "why the KeysSource enum"), just carried all the way through instead of partway: make the invalid state (PerStep present but a companion value missing) unrepresentable, not merely panic-guarded. Verified: 19/19 native_range_query_tests pass, full workspace suite (cargo test --workspace) 0 failures, cargo clippy --lib clean. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5 <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
execute_range_query_pipeline's single-population branch never calledget_keys()on the merged value accumulator, so self-keyed accumulators likeCountMinSketchWithHeap(top-k) returned empty over a range instead of expanding into their top-k keys — the instant path already does this viacollect_results_same_aggregation.merged.get_keys()first (mirroring the instant path exactly) and only falls back to the precomputed key list when it returnsNone. This also covers self-keyed accumulators stored under a non-Noneouter key (real Arroyo/worker.rs ingestion always wrapsSome(key), even for empty grouping), not just theNone-keyed case from the original report.Fixes #584.
Test plan
cargo test -p query_engine_rust --lib native_range_query_tests— 7 passed, including two new regression tests for Range queries drop self-keyed accumulator expansion (top-k) for single-population metrics #584 (range_query_self_keyed_topk_expands_without_keys_query,range_query_self_keyed_topk_expands_with_non_none_outer_key)cargo test -p query_engine_rust --lib— 554 passed, 0 failed🤖 Generated with Claude Code