From 15bb61e3734bb9f3c6ecee81335f8fa2b4698c31 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 09:05:42 -0400 Subject: [PATCH 01/12] 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. --- .../src/tests/native_range_query_tests.rs | 155 +++++++++++++++++- 1 file changed, 154 insertions(+), 1 deletion(-) diff --git a/asap-query-engine/src/tests/native_range_query_tests.rs b/asap-query-engine/src/tests/native_range_query_tests.rs index 891e3c7..1cfbf60 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -29,7 +29,9 @@ mod tests { use crate::engines::query_result::{QueryResult, RangeVectorElement}; use crate::engines::simple_engine::SimpleEngine; use crate::precompute_operators::sum_accumulator::SumAccumulator; - use crate::precompute_operators::{CountMinSketchAccumulator, DeltaSetAggregatorAccumulator}; + use crate::precompute_operators::{ + CountMinSketchAccumulator, DeltaSetAggregatorAccumulator, SetAggregatorAccumulator, + }; use crate::stores::simple_map_store::SimpleMapStore; use crate::stores::Store; use crate::tests::test_utilities::engine_factories::create_engine_multi_timestamp_with_window; @@ -371,4 +373,155 @@ mod tests { timestamps ); } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_key_appearing_midrange_has_no_phantom_earlier_sample() { + // Issue #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. host-b's DeltaSetAggregator "added" delta + // only appears in the bucket at t=2000, so the correct per-step key + // set at t=1000 must not include host-b yet — but the current + // single-snapshot merge folds host-b's later add into the whole + // range, giving it a phantom sample at t=1000 before it existed. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = DeltaSetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = DeltaSetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + let host_b_series = elements + .iter() + .find(|e| e.labels.labels.contains(&"host-b".to_string())); + let host_b_timestamps: std::collections::HashSet = host_b_series + .map(|e| e.samples.iter().map(|s| s.timestamp).collect()) + .unwrap_or_default(); + + assert!( + !host_b_timestamps.contains(&1000), + "host-b's key delta only appears at t=2000, so it must not have a \ + phantom sample at t=1000 (before it existed), got samples at {:?}", + host_b_timestamps + ); + assert!( + host_b_timestamps.contains(&2000), + "host-b should have a sample at t=2000, once its key delta appears, \ + got samples at {:?}", + host_b_timestamps + ); + + // DeltaSetAggregator accumulates: once added, a key stays in the + // reconstructed set for every later step too (no removal here), so + // host-a — added at t=1000 — must still be present at t=2000. + let host_a_series = elements + .iter() + .find(|e| e.labels.labels.contains(&"host-a".to_string())); + let host_a_timestamps: std::collections::HashSet = host_a_series + .map(|e| e.samples.iter().map(|s| s.timestamp).collect()) + .unwrap_or_default(); + assert!( + host_a_timestamps.contains(&2000), + "host-a was added at t=1000 and DeltaSetAggregator never removes it, \ + so it should still have a sample at t=2000, got samples at {:?}", + host_a_timestamps + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_set_aggregator_earlier_key_not_silently_dropped() { + // Issue #583 (second half): for SetAggregator ("latest window only"), + // create_keys_query_params scopes keys_query to [end-window_size, end] + // — a single instant-anchored window at the *range's* end, not each + // step's own window. host-a's full-snapshot bucket only exists at + // t=1000; with range end=2000 and window_size=1000, the keys fetch + // window is [1000, 2000], which excludes host-a's bucket (0..1000) + // entirely — its start (0) falls before the query window's start (1000). + // Since the range loop iterates merged_keys (not all_data), host-a + // never gets iterated at all — its whole series silently vanishes + // from the output, even though it had a real sample at t=1000. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = SetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = SetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + // SetAggregator is a full snapshot per window, not a delta: host-a is + // the live set only for the window ending at t=1000, and is replaced + // by host-b in the window ending at t=2000. So the correct output has + // host-a present at t=1000 but absent at t=2000. + let host_a_series = elements + .iter() + .find(|e| e.labels.labels.contains(&"host-a".to_string())); + let host_a_timestamps: std::collections::HashSet = host_a_series + .map(|e| e.samples.iter().map(|s| s.timestamp).collect()) + .unwrap_or_default(); + + assert!( + host_a_timestamps.contains(&1000), + "host-a existed at t=1000 (its own window) but was dropped entirely \ + from the range output because the final keys snapshot (scoped to \ + the range's end window) no longer contains it; got samples at {:?}", + host_a_timestamps + ); + assert!( + !host_a_timestamps.contains(&2000), + "host-a's SetAggregator snapshot at t=2000 no longer contains it \ + (host-b replaced it), so it must not have a sample at t=2000, \ + got samples at {:?}", + host_a_timestamps + ); + } } From 2e259ee2c7375acd9dd35a0dc7df9f8f24081fb0 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 09:45:54 -0400 Subject: [PATCH 02/12] 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. --- .../src/tests/native_range_query_tests.rs | 88 +++++++++---------- 1 file changed, 43 insertions(+), 45 deletions(-) diff --git a/asap-query-engine/src/tests/native_range_query_tests.rs b/asap-query-engine/src/tests/native_range_query_tests.rs index 1cfbf60..3881dd0 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -49,6 +49,16 @@ mod tests { } } + /// True if the output series for `host_label` (matched by label value) has + /// a sample at `ts`. False if the series is absent entirely, or present + /// but without a sample at that timestamp. + fn key_has_sample_at(elements: &[RangeVectorElement], host_label: &str, ts: u64) -> bool { + elements + .iter() + .find(|e| e.labels.labels.contains(&host_label.to_string())) + .is_some_and(|e| e.samples.iter().any(|s| s.timestamp == ts)) + } + /// One tumbling-window bucket: (bucket end timestamp ms, label values, accumulator). type TimeSeriesData = Vec<(u64, Option>, Box)>; @@ -416,40 +426,28 @@ mod tests { let (_, qr) = result.expect("range query failed"); let elements = matrix_values(qr); - let host_b_series = elements - .iter() - .find(|e| e.labels.labels.contains(&"host-b".to_string())); - let host_b_timestamps: std::collections::HashSet = host_b_series - .map(|e| e.samples.iter().map(|s| s.timestamp).collect()) - .unwrap_or_default(); - + // DeltaSetAggregator accumulates: once added, a key stays in the + // reconstructed set at every later step too (no removal here). + // host-a is added at t=1000, so it should be present at both steps. + // host-b is added only at t=2000, so it must be absent at t=1000 + // (not yet added) and present at t=2000. assert!( - !host_b_timestamps.contains(&1000), - "host-b's key delta only appears at t=2000, so it must not have a \ - phantom sample at t=1000 (before it existed), got samples at {:?}", - host_b_timestamps + key_has_sample_at(&elements, "host-a", 1000), + "host-a was added at t=1000, so it should have a sample there" ); assert!( - host_b_timestamps.contains(&2000), - "host-b should have a sample at t=2000, once its key delta appears, \ - got samples at {:?}", - host_b_timestamps + key_has_sample_at(&elements, "host-a", 2000), + "host-a was added at t=1000 and DeltaSetAggregator never removes it, \ + so it should still have a sample at t=2000" ); - - // DeltaSetAggregator accumulates: once added, a key stays in the - // reconstructed set for every later step too (no removal here), so - // host-a — added at t=1000 — must still be present at t=2000. - let host_a_series = elements - .iter() - .find(|e| e.labels.labels.contains(&"host-a".to_string())); - let host_a_timestamps: std::collections::HashSet = host_a_series - .map(|e| e.samples.iter().map(|s| s.timestamp).collect()) - .unwrap_or_default(); assert!( - host_a_timestamps.contains(&2000), - "host-a was added at t=1000 and DeltaSetAggregator never removes it, \ - so it should still have a sample at t=2000, got samples at {:?}", - host_a_timestamps + !key_has_sample_at(&elements, "host-b", 1000), + "host-b's key delta only appears at t=2000, so it must not have a \ + phantom sample at t=1000 (before it existed)" + ); + assert!( + key_has_sample_at(&elements, "host-b", 2000), + "host-b should have a sample at t=2000, once its key delta appears" ); } @@ -500,28 +498,28 @@ mod tests { // SetAggregator is a full snapshot per window, not a delta: host-a is // the live set only for the window ending at t=1000, and is replaced - // by host-b in the window ending at t=2000. So the correct output has - // host-a present at t=1000 but absent at t=2000. - let host_a_series = elements - .iter() - .find(|e| e.labels.labels.contains(&"host-a".to_string())); - let host_a_timestamps: std::collections::HashSet = host_a_series - .map(|e| e.samples.iter().map(|s| s.timestamp).collect()) - .unwrap_or_default(); - + // by host-b in the window ending at t=2000 — each step's own snapshot + // is disjoint from the other's, unlike DeltaSetAggregator's accumulation. assert!( - host_a_timestamps.contains(&1000), + key_has_sample_at(&elements, "host-a", 1000), "host-a existed at t=1000 (its own window) but was dropped entirely \ from the range output because the final keys snapshot (scoped to \ - the range's end window) no longer contains it; got samples at {:?}", - host_a_timestamps + the range's end window) no longer contains it" ); assert!( - !host_a_timestamps.contains(&2000), + !key_has_sample_at(&elements, "host-a", 2000), "host-a's SetAggregator snapshot at t=2000 no longer contains it \ - (host-b replaced it), so it must not have a sample at t=2000, \ - got samples at {:?}", - host_a_timestamps + (host-b replaced it), so it must not have a sample at t=2000" + ); + assert!( + !key_has_sample_at(&elements, "host-b", 1000), + "host-b doesn't appear until the window ending at t=2000, so it \ + must not have a sample at t=1000" + ); + assert!( + key_has_sample_at(&elements, "host-b", 2000), + "host-b is the live set for the window ending at t=2000, so it \ + should have a sample there" ); } } From 13d945ee69612e04c6f058576a1ddfd7b9d7f59e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 15:47:17 -0400 Subject: [PATCH 03/12] 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 --- .../src/engines/window_merger.rs | 85 ++ .../src/tests/native_range_query_tests.rs | 952 +++++++++++++++++- 2 files changed, 1001 insertions(+), 36 deletions(-) diff --git a/asap-query-engine/src/engines/window_merger.rs b/asap-query-engine/src/engines/window_merger.rs index 552249e..bb57632 100644 --- a/asap-query-engine/src/engines/window_merger.rs +++ b/asap-query-engine/src/engines/window_merger.rs @@ -401,4 +401,89 @@ mod tests { .unwrap(); assert_eq!(mock3.value, 120.0); } + + /// Pins a subtle, easy-to-miss requirement that `execute_range_query_pipeline`'s + /// per-step keys merge (#583) depends on: `NaiveMerger` folds buckets + /// *pairwise, left-to-right* (each `get_merged` walks `buckets[0].merge_with(buckets[1])`, + /// then that result `.merge_with(buckets[2])`, and so on) rather than + /// passing the whole slice to a single flat N-way merge. For most + /// accumulators the distinction is invisible (merging is associative and + /// commutative). It is NOT invisible for `DeltaSetAggregatorAccumulator`: + /// its merge treats a key present in both the added set and the removed + /// set as a cancelling conflict (see `merge_accumulators`), so a key that + /// toggles more than once within the merged buckets only nets out to the + /// chronologically correct state if the deltas are folded in order. + /// + /// Grows the window one bucket at a time (add, remove, add, remove, add) + /// via `slide(0, ..)` -- mirroring how the range pipeline's per-step + /// window for `DeltaSetAggregator` only ever grows, never expires -- and + /// checks `get_merged()` after *every* addition, not just the final one, + /// so a fold that's only correct at the boundary (e.g. an implementation + /// that happens to get the last step right by luck) can't hide. + /// + /// If a future change to the range-query pipeline (or to `WindowMerger` + /// itself) ever collects a `DeltaSetAggregator` window's buckets and + /// merges them with one flat call instead of `NaiveMerger`'s sequential + /// fold, this test is the tripwire that catches it. + #[test] + fn naive_merger_sequential_fold_replays_delta_set_toggles_at_every_window() { + use crate::data_model::traits::MergeableAccumulator; + use crate::precompute_operators::DeltaSetAggregatorAccumulator; + + let key = KeyByLabelValues::new_with_labels(vec!["host-a".to_string()]); + + let mut add = DeltaSetAggregatorAccumulator::new(); + add.add_key(key.clone()); + let mut remove = DeltaSetAggregatorAccumulator::new(); + remove.remove_key(key.clone()); + + // add, remove, add, remove, add -> present, absent, present, absent, present + let deltas = [ + add.clone(), + remove.clone(), + add.clone(), + remove.clone(), + add.clone(), + ]; + let expected_present = [true, false, true, false, true]; + + let mut merger = NaiveMerger::new(); + merger.initialize(vec![Box::new(deltas[0].clone())]); + let mismatches: Vec = expected_present + .iter() + .enumerate() + .filter_map(|(i, &expected)| { + if i > 0 { + merger.slide(0, vec![Box::new(deltas[i].clone())]); + } + let merged = merger.get_merged().unwrap(); + let actual = merged + .get_keys() + .expect("no unresolved removals after a sequential fold") + .contains(&key); + (actual != expected) + .then(|| format!("window {}: expected {expected}, got {actual}", i + 1)) + }) + .collect(); + assert!( + mismatches.is_empty(), + "NaiveMerger's sequential left-fold must replay each toggle chronologically \ + at every window, not just the final one -- diverged at: {mismatches:?}" + ); + + // Contrast: the same 5 buckets merged in one flat call (not a + // sequential fold) lose the key entirely, proving these are NOT + // interchangeable for DeltaSetAggregator. + let flat_result = DeltaSetAggregatorAccumulator::merge_accumulators(deltas.to_vec()) + .expect("flat merge should still succeed, just give the wrong answer"); + assert!( + !flat_result + .get_keys() + .expect("no unresolved removals after the flat merge either") + .contains(&key), + "a flat (non-sequential) merge_accumulators call over the same 5 buckets \ + must NOT net the key to present -- this is exactly the mistake the \ + sequential fold above avoids" + ); + } } diff --git a/asap-query-engine/src/tests/native_range_query_tests.rs b/asap-query-engine/src/tests/native_range_query_tests.rs index 3881dd0..c932018 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -59,6 +59,44 @@ mod tests { .is_some_and(|e| e.samples.iter().any(|s| s.timestamp == ts)) } + /// Like `key_has_sample_at`, but matches a series only if ALL given label + /// values are present -- needed once there's more than one grouping + /// label (e.g. region + host), since a bare host value alone can't tell + /// "host-b under region=eu" apart from "host-b under region=us". + fn labels_have_sample_at(elements: &[RangeVectorElement], label_values: &[&str], ts: u64) -> bool { + elements + .iter() + .find(|e| { + label_values + .iter() + .all(|lv| e.labels.labels.contains(&lv.to_string())) + }) + .is_some_and(|e| e.samples.iter().any(|s| s.timestamp == ts)) + } + + /// Checks every `(label_values, ts, expected_present, reason)` case + /// against `elements` and reports ALL mismatches in one panic, instead of + /// stopping at the first failing `assert!` -- each of these tests makes + /// several independent claims (about different keys/timestamps), and a + /// fix attempt that gets some right and some wrong is much easier to + /// debug when every divergence is visible at once. + fn assert_all_at(elements: &[RangeVectorElement], cases: &[(&[&str], u64, bool, &str)]) { + let mismatches: Vec = cases + .iter() + .filter_map(|(labels, ts, expected, reason)| { + let actual = labels_have_sample_at(elements, labels, *ts); + (actual != *expected).then(|| { + format!("{labels:?}@{ts}: expected present={expected}, got {actual} -- {reason}") + }) + }) + .collect(); + assert!( + mismatches.is_empty(), + "diverged from expectations at:\n{}", + mismatches.join("\n") + ); + } + /// One tumbling-window bucket: (bucket end timestamp ms, label values, accumulator). type TimeSeriesData = Vec<(u64, Option>, Box)>; @@ -77,6 +115,37 @@ mod tests { value_data: TimeSeriesData, keys_data: TimeSeriesData, promql_query: &str, + ) -> SimpleEngine { + create_range_engine_dual_input_with_windows( + metric, + value_agg_type, + key_agg_type, + grouping_labels, + aggregated_labels, + value_data, + keys_data, + promql_query, + WINDOW_MS, + WINDOW_MS, + ) + } + + /// Same as `create_range_engine_dual_input`, but lets the value and key + /// aggregations use different bucket widths — the value aggregation's + /// `tumbling_window_ms` must not be assumed to also be the key + /// aggregation's bucket width when scanning the keys `bucket_map` (#583). + #[allow(clippy::too_many_arguments)] + fn create_range_engine_dual_input_with_windows( + metric: &str, + value_agg_type: AggregationType, + key_agg_type: AggregationType, + grouping_labels: Vec<&str>, + aggregated_labels: Vec<&str>, + value_data: TimeSeriesData, + keys_data: TimeSeriesData, + promql_query: &str, + value_window_ms: u64, + key_window_ms: u64, ) -> SimpleEngine { let grouping_label_strings: Vec = grouping_labels.iter().map(|s| s.to_string()).collect(); @@ -100,8 +169,8 @@ mod tests { aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: WINDOW_MS, - slide_interval_ms: WINDOW_MS, + window_size_ms: value_window_ms, + slide_interval_ms: value_window_ms, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -123,8 +192,8 @@ mod tests { aggregated_labels: KeyByLabelNames::new(aggregated_label_strings), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: WINDOW_MS, - slide_interval_ms: WINDOW_MS, + window_size_ms: key_window_ms, + slide_interval_ms: key_window_ms, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -145,10 +214,13 @@ mod tests { CleanupPolicy::NoCleanup, )); - for (agg_id, data) in [(1u64, value_data), (2u64, keys_data)] { + for (agg_id, window_ms, data) in [ + (1u64, value_window_ms, value_data), + (2u64, key_window_ms, keys_data), + ] { for (timestamp, label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp - WINDOW_MS, timestamp, key, agg_id); + let output = PrecomputedOutput::new(timestamp - window_ms, timestamp, key, agg_id); store.insert_precomputed_output(output, acc).unwrap(); } } @@ -431,23 +503,30 @@ mod tests { // host-a is added at t=1000, so it should be present at both steps. // host-b is added only at t=2000, so it must be absent at t=1000 // (not yet added) and present at t=2000. - assert!( - key_has_sample_at(&elements, "host-a", 1000), - "host-a was added at t=1000, so it should have a sample there" - ); - assert!( - key_has_sample_at(&elements, "host-a", 2000), - "host-a was added at t=1000 and DeltaSetAggregator never removes it, \ - so it should still have a sample at t=2000" - ); - assert!( - !key_has_sample_at(&elements, "host-b", 1000), - "host-b's key delta only appears at t=2000, so it must not have a \ - phantom sample at t=1000 (before it existed)" - ); - assert!( - key_has_sample_at(&elements, "host-b", 2000), - "host-b should have a sample at t=2000, once its key delta appears" + assert_all_at( + &elements, + &[ + (&["host-a"], 1000, true, "host-a was added at t=1000"), + ( + &["host-a"], + 2000, + true, + "host-a was added at t=1000 and DeltaSetAggregator never removes it", + ), + ( + &["host-b"], + 1000, + false, + "host-b's key delta only appears at t=2000, so it must not have a \ + phantom sample at t=1000 (before it existed)", + ), + ( + &["host-b"], + 2000, + true, + "host-b should have a sample at t=2000, once its key delta appears", + ), + ], ); } @@ -500,26 +579,827 @@ mod tests { // the live set only for the window ending at t=1000, and is replaced // by host-b in the window ending at t=2000 — each step's own snapshot // is disjoint from the other's, unlike DeltaSetAggregator's accumulation. + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a existed at t=1000 (its own window) but was dropped entirely \ + from the range output because the final keys snapshot (scoped to \ + the range's end window) no longer contains it", + ), + ( + &["host-a"], + 2000, + false, + "host-a's SetAggregator snapshot at t=2000 no longer contains it \ + (host-b replaced it)", + ), + ( + &["host-b"], + 1000, + false, + "host-b doesn't appear until the window ending at t=2000", + ), + ( + &["host-b"], + 2000, + true, + "host-b is the live set for the window ending at t=2000", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_binary_expr_arm_key_appearing_midrange_has_no_phantom_earlier_sample() { + // Issue #583's "check other candidates" section names + // build_arm_range_context specifically: handle_binary_expr_range_promql + // builds each arm's RangeQueryExecutionContext via the same + // finish_range_context/execute_range_query_pipeline used by the plain + // range path, so the same single-snapshot keys bug should reproduce + // through a binary expression's scalar arm. Same fixture as + // range_query_dual_population_key_appearing_midrange_has_no_phantom_earlier_sample, + // wrapped in `* 1` so it takes the detect_scalar_arm path in + // handle_binary_expr_range_promql instead of the plain-query dispatch. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = DeltaSetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = DeltaSetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event) * 1"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + (&["host-a"], 1000, true, "host-a was added at t=1000"), + ( + &["host-a"], + 2000, + true, + "host-a was added at t=1000 and DeltaSetAggregator never removes it", + ), + ( + &["host-b"], + 1000, + false, + "host-b's key delta only appears at t=2000, so it must not have a \ + phantom sample at t=1000 (before it existed) — through the binary \ + expr arm path (build_arm_range_context) this time, not the plain range path", + ), + ( + &["host-b"], + 2000, + true, + "host-b should have a sample at t=2000, once its key delta appears", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_delta_set_aggregator_oscillating_add_remove_across_five_windows() { + // Generalizes the phantom-sample tests beyond a single add: host-a + // toggles membership every window (add, remove, add, remove, add) + // across 5 tumbling windows. A correct per-step fix must replay + // deltas only up to *each step's own* end, so presence should + // alternate present/absent/present/absent/present across the 5 + // output steps. The current single end-anchored snapshot instead + // merges all 5 deltas into one net state (present, since the last + // delta is an add) and reuses it for every step — so it would wrongly + // show host-a present at every step, including the two "removed" + // windows (t=2000, t=4000). + let value_data: TimeSeriesData = (1..=5) + .map(|i| { + ( + i * 1000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ) + }) + .collect(); + + let mut keys_add = DeltaSetAggregatorAccumulator::new(); + keys_add.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_remove = DeltaSetAggregatorAccumulator::new(); + keys_remove.remove_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let keys_data: TimeSeriesData = vec![ + (1000, None, Box::new(keys_add.clone()) as Box), + (2000, None, Box::new(keys_remove.clone()) as Box), + (3000, None, Box::new(keys_add.clone()) as Box), + (4000, None, Box::new(keys_remove.clone()) as Box), + (5000, None, Box::new(keys_add) as Box), + ]; + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + value_data, + keys_data, + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 5.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + let expected_present = [ + (1000, true), + (2000, false), + (3000, true), + (4000, false), + (5000, true), + ]; + let mismatches: Vec = expected_present + .iter() + .filter_map(|&(ts, expected)| { + let actual = key_has_sample_at(&elements, "host-a", ts); + (actual != expected).then(|| format!("t={ts}: expected {expected}, got {actual}")) + }) + .collect(); assert!( - key_has_sample_at(&elements, "host-a", 1000), - "host-a existed at t=1000 (its own window) but was dropped entirely \ - from the range output because the final keys snapshot (scoped to \ - the range's end window) no longer contains it" + mismatches.is_empty(), + "host-a's net membership diverged from the per-step expectation (deltas \ + replayed only up to each step's own end) at: {mismatches:?}" ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_binary_expr_arm_set_aggregator_earlier_key_not_silently_dropped() { + // SetAggregator counterpart to + // range_query_binary_expr_arm_key_appearing_midrange_has_no_phantom_earlier_sample: + // closes the matrix by driving + // range_query_set_aggregator_earlier_key_not_silently_dropped's fixture + // through the binary-expr arm path (build_arm_range_context) too, + // instead of only the plain range dispatch. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = SetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = SetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event) * 1"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a existed at t=1000 (its own window) but was dropped entirely \ + from the range output because the final keys snapshot (scoped to \ + the range's end window) no longer contains it — through the binary \ + expr arm path (build_arm_range_context) this time", + ), + ( + &["host-a"], + 2000, + false, + "host-a's SetAggregator snapshot at t=2000 no longer contains it \ + (host-b replaced it)", + ), + ( + &["host-b"], + 1000, + false, + "host-b doesn't appear until the window ending at t=2000", + ), + ( + &["host-b"], + 2000, + true, + "host-b is the live set for the window ending at t=2000", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_delta_set_aggregator_key_change_on_middle_step_not_just_boundary() { + // All prior #583 tests query exactly 2 output steps, where the key + // change lands on one of the range's own boundaries (start or end). + // A fix that special-cases the first/last iteration of the per-step + // loop (rather than genuinely scoping every iteration) could pass + // those while still getting an interior step wrong. This test spans + // 3 output steps (1000, 2000, 3000) with the key added only on the + // *middle* one, so t=1000's snapshot must differ from t=2000's and + // t=3000's purely by virtue of being an interior loop iteration. + let value_data: TimeSeriesData = vec![ + ( + 1000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ( + 2000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ( + 3000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ]; + + let mut keys_2000 = DeltaSetAggregatorAccumulator::new(); + keys_2000.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let keys_data: TimeSeriesData = vec![ + ( + 1000, + None, + Box::new(DeltaSetAggregatorAccumulator::new()) as Box, + ), + (2000, None, Box::new(keys_2000) as Box), + ( + 3000, + None, + Box::new(DeltaSetAggregatorAccumulator::new()) as Box, + ), + ]; + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + value_data, + keys_data, + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 3.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + let expected_present = [(1000, false), (2000, true), (3000, true)]; + let mismatches: Vec = expected_present + .iter() + .filter_map(|&(ts, expected)| { + let actual = key_has_sample_at(&elements, "host-a", ts); + (actual != expected).then(|| format!("t={ts}: expected {expected}, got {actual}")) + }) + .collect(); assert!( - !key_has_sample_at(&elements, "host-a", 2000), - "host-a's SetAggregator snapshot at t=2000 no longer contains it \ - (host-b replaced it), so it must not have a sample at t=2000" + mismatches.is_empty(), + "host-a's key is added only on the middle step (t=2000), so t=1000 must not \ + see it yet while t=2000 and t=3000 must — diverged at: {mismatches:?}" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_delta_set_aggregator_key_bucket_width_differs_from_value_bucket_width() { + // Issue #583's fix scans the keys bucket_map using a per-step + // increment. That increment must come from the KEY aggregation's own + // window_size_ms, not the VALUE aggregation's tumbling_window_ms — an + // implementation that (incorrectly) reuses the value side's bucket + // width to walk the keys bucket_map would silently skip every keys + // bucket whose start isn't a multiple of that width. All other + // #583 tests use the same window size (WINDOW_MS) for both + // aggregations, so none of them would catch that mistake — this one + // deliberately sets them apart: value buckets are 1000ms wide, key + // (DeltaSetAggregator) buckets are 500ms wide, and both key delta + // buckets are placed off the 1000ms grid (starts at 500 and 1500). + let value_window_ms = 1000; + let key_window_ms = 500; + + let value_data: TimeSeriesData = vec![ + ( + 1000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ( + 2000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ]; + + // host-a's delta bucket: start=500, end=1000 (off the 1000ms grid). + let mut keys_host_a = DeltaSetAggregatorAccumulator::new(); + keys_host_a.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + // host-b's delta bucket: start=1500, end=2000 (also off the grid). + let mut keys_host_b = DeltaSetAggregatorAccumulator::new(); + keys_host_b.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + let keys_data: TimeSeriesData = vec![ + (1000, None, Box::new(keys_host_a) as Box), + (2000, None, Box::new(keys_host_b) as Box), + ]; + + let engine = create_range_engine_dual_input_with_windows( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + value_data, + keys_data, + "count(event_frequency) by (host, event)", + value_window_ms, + key_window_ms, ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + // host-a's bucket (start=500) falls inside [0,1000), so it must be + // visible at t=1000. host-b's bucket (start=1500) doesn't exist yet + // at t=1000, but is inside [0,2000) by t=2000, and DeltaSetAggregator + // accumulates, so host-a stays present at t=2000 too. + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a's delta bucket starts at t=500, which is inside [0,1000) -- it \ + must be found even though 500 isn't a multiple of the value \ + aggregation's 1000ms bucket width", + ), + ( + &["host-a"], + 2000, + true, + "host-a was added by t=1000 and DeltaSetAggregator never removes it", + ), + ( + &["host-b"], + 1000, + false, + "host-b's delta bucket starts at t=1500, which is not yet inside [0,1000)", + ), + ( + &["host-b"], + 2000, + true, + "host-b's delta bucket starts at t=1500, which is inside [0,2000) -- it \ + must be found even though 1500 isn't a multiple of the value \ + aggregation's 1000ms bucket width", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_per_step_key_change_does_not_leak_across_groups() { + // All prior #583 tests use grouping_labels: vec![] -- a single + // implicit group (group_key = None) -- so "the per-step key set" and + // "the per-step key set for THIS group" are the same computation, + // and a bug that pools state across groups would be invisible. + // Here the value aggregation genuinely groups by `region`, giving + // two independent groups (region=us, region=eu), each with its own + // keys aggregation state. region=us gains host-b mid-range (added at + // t=2000); region=eu's host set (host-x) never changes. A fix whose + // per-step keys merge isn't correctly scoped per group_key -- e.g. + // one that merges all groups' DeltaSetAggregator buckets together + // before re-splitting by group -- would leak host-b into region=eu's + // output at t=2000, since nothing about merge_with prevents merging + // two different regions' delta-sets together. + let cms_us_1 = CountMinSketchAccumulator::new(2, 3); + let cms_us_2 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_1 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_us_1000 = DeltaSetAggregatorAccumulator::new(); + keys_us_1000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_us_2000 = DeltaSetAggregatorAccumulator::new(); + keys_us_2000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-b".to_string(), "evt-1".to_string()], + }); + let mut keys_eu_1000 = DeltaSetAggregatorAccumulator::new(); + keys_eu_1000.add_key(KeyByLabelValues { + labels: vec!["eu".to_string(), "host-x".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec!["region"], + vec!["host", "event"], + vec![ + (1000, Some(vec!["us".to_string()]), Box::new(cms_us_1) as Box), + (2000, Some(vec!["us".to_string()]), Box::new(cms_us_2) as Box), + (1000, Some(vec!["eu".to_string()]), Box::new(cms_eu_1) as Box), + (2000, Some(vec!["eu".to_string()]), Box::new(cms_eu_2) as Box), + ], + vec![ + (1000, Some(vec!["us".to_string()]), Box::new(keys_us_1000) as Box), + (2000, Some(vec!["us".to_string()]), Box::new(keys_us_2000) as Box), + (1000, Some(vec!["eu".to_string()]), Box::new(keys_eu_1000) as Box), + ], + "count(event_frequency) by (region, host, event)", + ); + + let query = "count(event_frequency) by (region, host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + (&["us", "host-a"], 1000, true, "region=us has host-a from the start"), + ( + &["us", "host-b"], + 1000, + false, + "region=us's host-b is only added at t=2000", + ), + (&["us", "host-b"], 2000, true, "region=us's host-b was added by t=2000"), + (&["eu", "host-x"], 1000, true, "region=eu has host-x from the start"), + ( + &["eu", "host-x"], + 2000, + true, + "region=eu's host-x is unaffected by region=us's mid-range change", + ), + ( + &["eu", "host-b"], + 2000, + false, + "region=us's host-b addition must not leak into region=eu's output -- \ + each group's per-step key expansion must be scoped to its own \ + group_key, not pooled across groups", + ), + ( + &["us", "host-x"], + 1000, + false, + "region=eu's host-x must not leak into region=us's output either", + ), + ( + &["us", "host-x"], + 2000, + false, + "region=eu's host-x must not leak into region=us's output either", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_simultaneous_cross_group_adds_stay_isolated() { + // Sharper variant of ...does_not_leak_across_groups: there, region=eu + // never changes, so a failure could plausibly be misread as plain + // Bug 1 (single-snapshot-per-group) rather than cross-group bleed. + // Here BOTH groups add a DIFFERENT new host at the SAME timestamp + // (t=2000), so any implementation that pools keys across group_keys + // before re-splitting would show up as *both* regions gaining *both* + // new hosts -- a decisive, unambiguous signature distinct from + // Bug 1's "one group's own key is phantom-early" symptom. + let cms_us_1 = CountMinSketchAccumulator::new(2, 3); + let cms_us_2 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_1 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_us_1000 = DeltaSetAggregatorAccumulator::new(); + keys_us_1000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_us_2000 = DeltaSetAggregatorAccumulator::new(); + keys_us_2000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-b".to_string(), "evt-1".to_string()], + }); + let mut keys_eu_1000 = DeltaSetAggregatorAccumulator::new(); + keys_eu_1000.add_key(KeyByLabelValues { + labels: vec!["eu".to_string(), "host-p".to_string(), "evt-1".to_string()], + }); + let mut keys_eu_2000 = DeltaSetAggregatorAccumulator::new(); + keys_eu_2000.add_key(KeyByLabelValues { + labels: vec!["eu".to_string(), "host-q".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec!["region"], + vec!["host", "event"], + vec![ + (1000, Some(vec!["us".to_string()]), Box::new(cms_us_1) as Box), + (2000, Some(vec!["us".to_string()]), Box::new(cms_us_2) as Box), + (1000, Some(vec!["eu".to_string()]), Box::new(cms_eu_1) as Box), + (2000, Some(vec!["eu".to_string()]), Box::new(cms_eu_2) as Box), + ], + vec![ + (1000, Some(vec!["us".to_string()]), Box::new(keys_us_1000) as Box), + (2000, Some(vec!["us".to_string()]), Box::new(keys_us_2000) as Box), + (1000, Some(vec!["eu".to_string()]), Box::new(keys_eu_1000) as Box), + (2000, Some(vec!["eu".to_string()]), Box::new(keys_eu_2000) as Box), + ], + "count(event_frequency) by (region, host, event)", + ); + + let query = "count(event_frequency) by (region, host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + let expected = [ + (["us", "host-a"], 1000, true), + (["us", "host-b"], 1000, false), + (["us", "host-a"], 2000, true), + (["us", "host-b"], 2000, true), + (["eu", "host-p"], 1000, true), + (["eu", "host-q"], 1000, false), + (["eu", "host-p"], 2000, true), + (["eu", "host-q"], 2000, true), + // Cross-group: neither region's host should ever appear under the other. + (["us", "host-p"], 1000, false), + (["us", "host-q"], 1000, false), + (["us", "host-p"], 2000, false), + (["us", "host-q"], 2000, false), + (["eu", "host-a"], 1000, false), + (["eu", "host-b"], 1000, false), + (["eu", "host-a"], 2000, false), + (["eu", "host-b"], 2000, false), + ]; + let mismatches: Vec = expected + .iter() + .filter_map(|(labels, ts, expected_present)| { + let actual = labels_have_sample_at(&elements, labels, *ts); + (actual != *expected_present) + .then(|| format!("{labels:?}@{ts}: expected {expected_present}, got {actual}")) + }) + .collect(); assert!( - !key_has_sample_at(&elements, "host-b", 1000), - "host-b doesn't appear until the window ending at t=2000, so it \ - must not have a sample at t=1000" + mismatches.is_empty(), + "simultaneous cross-group adds must stay isolated per group_key -- \ + diverged at: {mismatches:?}" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_group_with_no_value_data_is_skipped_not_fatal() { + // Today, execute_range_query_pipeline hard-fails the ENTIRE range + // query if any group resolved from merged_keys has no matching entry + // in all_data: `all_data.get(group_key).ok_or_else(|| "No value for + // key")?`. region=orphan has keys data (a real DeltaSetAggregator + // delta) but NEVER has any value/CMS data at all -- under today's + // code this poisons the WHOLE query, so even region=normal's + // perfectly good data disappears. + // + // Per #583's design discussion: this should become non-fatal -- + // skip the orphaned group (with a loud warning, not asserted here + // since this is a unit test, not a log-capture test) and still + // return the rest of the query's results. This test pins the + // EXPECTED (fixed) behavior, so it's RED today: today the whole + // `handle_range_query_promql` call returns None and the `.expect()` + // below panics before any of the assertions run. + let cms_normal_1 = CountMinSketchAccumulator::new(2, 3); + let cms_normal_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_normal = DeltaSetAggregatorAccumulator::new(); + keys_normal.add_key(KeyByLabelValues { + labels: vec!["normal".to_string(), "host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_orphan = DeltaSetAggregatorAccumulator::new(); + keys_orphan.add_key(KeyByLabelValues { + labels: vec!["orphan".to_string(), "host-z".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec!["region"], + vec!["host", "event"], + vec![ + ( + 1000, + Some(vec!["normal".to_string()]), + Box::new(cms_normal_1) as Box, + ), + ( + 2000, + Some(vec!["normal".to_string()]), + Box::new(cms_normal_2) as Box, + ), + // Deliberately NO value data for region=orphan, at any timestamp. + ], + vec![ + ( + 1000, + Some(vec!["normal".to_string()]), + Box::new(keys_normal) as Box, + ), + ( + 1000, + Some(vec!["orphan".to_string()]), + Box::new(keys_orphan) as Box, + ), + ], + "count(event_frequency) by (region, host, event)", + ); + + let query = "count(event_frequency) by (region, host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect( + "range query should succeed by skipping the value-less region=orphan group, \ + not fail the entire query because of it", + ); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + ( + &["normal", "host-a"], + 1000, + true, + "region=normal has real value data and should be unaffected by \ + region=orphan having none", + ), + ( + &["normal", "host-a"], + 2000, + true, + "region=normal's host-a persists (DeltaSetAggregator never removes it)", + ), + ], ); assert!( - key_has_sample_at(&elements, "host-b", 2000), - "host-b is the live set for the window ending at t=2000, so it \ - should have a sample there" + !elements + .iter() + .any(|e| e.labels.labels.contains(&"orphan".to_string())), + "region=orphan has keys data but no value data anywhere -- it must be \ + silently skipped, not appear as an (empty or otherwise) series" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_set_aggregator_merges_multiple_buckets_within_one_window() { + // Every other SetAggregator test uses window_size_ms == bucket + // width, so each window only ever contains exactly ONE bucket -- + // DeltaSetAggregator gets multi-bucket-per-window coverage "for + // free" via its always-widened [0,t] window, but SetAggregator's + // bounded sliding window has never been exercised with more than one + // bucket in play. Mirrors the values-side sliding-collision fixture + // (range_query_sliding_window_merges_both_buckets: two buckets + // sharing the same (start,end)) but for keys: two SetAggregator + // buckets both at t=1000 (start=0,end=1000) must UNION into + // {host-a, host-b}, exactly like the existing bucket_map handling + // for values (`bucket_map.entry(*start).or_default().push(..)`, + // #567/#570) -- a fix that naively does `.insert()` instead of + // accumulating for the keys side would silently drop one of them. + // + // t=2000 adds a third key (host-c) via a single non-colliding + // bucket; since SetAggregator is "latest window only" (not + // cumulative), it replaces the t=1000 pair entirely -- this also + // keeps the test genuinely RED against today's code (Bug 2: the + // single end-anchored keys fetch window [1000,2000] excludes the + // t=1000 collision pair's bucket, whose start=0 falls outside it). + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_1_a = SetAggregatorAccumulator::new(); + keys_1_a.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_1_b = SetAggregatorAccumulator::new(); + keys_1_b.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = SetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-c".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + // Both buckets share (start=0, end=1000): a genuine collision. + (1000, None, Box::new(keys_1_a) as Box), + (1000, None, Box::new(keys_1_b) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a's bucket collides with host-b's at (start=0,end=1000) -- \ + both must be merged (unioned), not one dropped in favor of the other", + ), + ( + &["host-b"], + 1000, + true, + "host-b's bucket collides with host-a's at (start=0,end=1000) -- \ + both must be merged (unioned), not one dropped in favor of the other", + ), + (&["host-c"], 1000, false, "host-c's bucket doesn't exist until t=2000"), + ( + &["host-a"], + 2000, + false, + "SetAggregator is latest-window-only: host-c's window replaces \ + host-a/host-b's, it doesn't accumulate alongside them", + ), + ( + &["host-b"], + 2000, + false, + "SetAggregator is latest-window-only: host-c's window replaces \ + host-a/host-b's, it doesn't accumulate alongside them", + ), + (&["host-c"], 2000, true, "host-c is the live set for the window ending at t=2000"), + ], ); } } From 8f5d315fbdff1c4c0a852a455fccbc877b938b13 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 15:47:34 -0400 Subject: [PATCH 04/12] 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 --- docs/583-range-keys-per-step-design.md | 243 +++++++++++++++++++++++++ 1 file changed, 243 insertions(+) create mode 100644 docs/583-range-keys-per-step-design.md diff --git a/docs/583-range-keys-per-step-design.md b/docs/583-range-keys-per-step-design.md new file mode 100644 index 0000000..2a89d8a --- /dev/null +++ b/docs/583-range-keys-per-step-design.md @@ -0,0 +1,243 @@ +# #583 range-query per-step keys — grilling session recap + +Design discussion for the fix to [#583](https://github.com/ProjectASAP/ASAPQuery/issues/583) +("Range query key expansion uses one snapshot instead of per-step keys"). +No fix code has been written yet — this session was scoped to (1) confirming +understanding of the bug, (2) working out the fix design question-by-question, +and (3) building out a RED test suite that pins every failure mode discussed. +Recorded here so the eventual implementation turn doesn't have to re-derive +any of this. + +## Bug recap + +`execute_range_query_pipeline` (`asap-query-engine/src/engines/simple_engine/mod.rs`) +fetches and merges `keys_query` **once**, anchored at the range's `end`, and +reuses that single snapshot for every output timestamp in the range. Two +distinct problems fall out of this: + +1. **Stale snapshot reused across steps** — a key that starts or stops + existing partway through the range gets phantom samples before it existed, + or is silently dropped even at timestamps where it legitimately existed. + Confirmed to also affect the binary-expr arm path + (`handle_binary_expr_range_promql` → `build_arm_range_context` → + `finish_range_context`), which the issue only flagged as "worth checking." +2. **`keys_query`'s window is instant-anchored, not range-aware** — + `create_keys_query_params` computes the keys window purely from + `end_timestamp`. `finish_range_context` widens `values_query` to + `[start-lookback, end]` but clones `keys_query` unchanged. For + `SetAggregator` ("latest window only") this drops labels that existed + earlier in the range but fell out of the final instant's window. + `DeltaSetAggregator`'s `[0, end]` window happens to already be correct + (it's a full replay-from-start aggregator), so this half of the bug is + `SetAggregator`-specific. + +Both collapse to one fix: fetch/merge keys **per output step**, scoped to +that step's own window, not once at `end`. + +## Q&A + +### Q1 — architecture: per-step store queries, or one fetch + per-step in-memory merge? + +**Answer: one fetch + per-step in-memory merge.** The values side of +`execute_range_query_pipeline` already solves this exact problem: one +`execute_store_query` fetches all raw buckets across the whole range once, +then the `current_time` loop does an in-memory windowed merge per step +(`bucket_map` lookup + `create_window_merger`). The keys side should mirror +this instead of issuing N separate store queries (one per output timestamp). + +### Q2 — how does `finish_range_context` compute each side's widened window? + +**Answer: generalize the existing `values_query` widening formula to +`keys_query`, with zero `AggregationType` branching in the fix.** + +`finish_range_context` already does, for values: +```rust +let lookback_ms = base_context.store_plan.values_query.end_timestamp + - base_context.store_plan.values_query.start_timestamp; +extended_store_plan.values_query.start_timestamp = start_ms.saturating_sub(lookback_ms); +extended_store_plan.values_query.end_timestamp = end_ms; +``` +It doesn't know or care *why* that window is that width. The same trick +works for `keys_query` and happens to handle both aggregation types +correctly for free: + +- `create_keys_query_params` already computed the instant keys window: + `[end-window_size, end]` for `SetAggregator`, `[0, end]` for + `DeltaSetAggregator`. +- `keys_lookback_ms = keys_query.end - keys_query.start` → `window_size` for + `SetAggregator`, `end_ms` (the range's end) for `DeltaSetAggregator`. +- Widen the same way: `start_ms.saturating_sub(keys_lookback_ms)`. For + `SetAggregator` that's a normal sliding window. For `DeltaSetAggregator`, + since `keys_lookback_ms == end_ms` and the per-step loop invariant is + `current_time <= end_ms`, `current_time.saturating_sub(keys_lookback_ms)` + saturates to `0` at **every** step — "replay from the beginning," for free, + provably (not by luck): the `while current_time <= end_ms` loop condition + guarantees `current_time <= end_ms` before every iteration body runs. + +This also means "always merge from t=0 for DeltaSetAggregator" (a specific +follow-up question raised) is already guaranteed by this derivation — no +separate `if key_agg_type == DeltaSetAggregator { start = 0 }` branch is +needed anywhere. + +### Q3 — recompute-from-scratch per step, or incremental carry-forward? + +**Answer: recompute-from-scratch first; optimize later if measured.** + +For `DeltaSetAggregator`, each step's window is `[0, t]`, growing every +step. Recomputing fresh each step (mirroring the existing values loop +exactly) costs `O(N²)` bucket-merges total across `N` output steps (step 1 +processes 1 bucket, step 2 processes 2, ..., step N processes N). An +incremental version — carry a running merged accumulator forward across the +loop, only folding in new buckets since the last step — gives the same +result (verified, see below) in `O(N)` total, but requires restructuring the +loop to persist per-key-group merger state across iterations. + +Decision: ship recompute-from-scratch (least new code, easiest to verify, +matches the existing values-loop pattern exactly). `SetAggregator`'s window +is bounded by `window_size` regardless of range length, so it never has this +blowup — only `DeltaSetAggregator` does. Leave a comment flagging the +`O(N²)` replay cost as a known ceiling; revisit only if it's an actual +measured problem. + +**Correctness check performed before trusting this:** `NaiveMerger.merge_all()` +folds buckets **pairwise, left-to-right** (`buckets[0].merge_with(buckets[1])`, +then that result `.merge_with(buckets[2])`, ...) rather than passing the +whole slice to `DeltaSetAggregatorAccumulator::merge_accumulators` in one +flat N-way call. This distinction matters: `merge_accumulators`'s "key in +both added and removed → cancel both" conflict logic is symmetric and +order-blind when given the whole batch at once, but `NaiveMerger`'s +sequential fold means each binary step only ever resolves conflicts between +"the running state so far" and "the next bucket" — which correctly threads +chronological order through, including for a key that toggles 3+ times +within one merge window. Hand-traced against a 5-step add/remove/add/remove/add +sequence and confirmed correct at every intermediate step, not just the +final one. Pinned as an explicit, readable regression test (see below) since +this is easy to miss if `WindowMerger`'s implementation ever changes. + +### Q4 — does the keys bucket_map walk need its own step increment? + +**Answer: yes.** The per-step window-building loop scans `bucket_map` in +increments of the tumbling bucket width (`t += tumbling_window_ms` in the +existing values loop). That `tumbling_window_ms` is the **value** +aggregation's bucket width. The **key** aggregation (`aggregation_id_for_key`) +can have a different `window_size_ms`. The fix needs a separate +`keys_tumbling_window_ms`, fetched from the key aggregation's own config +(the same way `create_keys_query_params` already does internally), used to +walk the keys `bucket_map` — reusing the value side's width would silently +skip every keys bucket whose start isn't a multiple of that width. + +### Q5 — does `do_merge` still matter for keys once per-step merging is in place? + +**Answer: no, drop it for the range path.** Today `fetch_and_merge_keys` +takes a `do_merge` flag derived from the **value** aggregation's window +(`create_store_query_plan`: `do_merge = range_ms > value_aggregation.window_size_ms`) +and reuses it wholesale for the keys merge. Once keys mirror the values loop +(Q1), `do_merge` becomes moot the same way it already is for values (the +values `current_time` loop never consults it — it always does the +bucket-map-window-merge dance regardless of window count). The range +pipeline's keys fetch should become a raw `execute_store_query(keys_params)` +call (no merge), replacing `fetch_and_merge_keys`, for the range path only — +the instant-query path is unaffected. + +Related: [#581](https://github.com/ProjectASAP/ASAPQuery/issues/581) +"Unify PromQL instant and range query fetch/merge paths" independently names +this exact gap (range path's "fresh `WindowMerger` re-merged from scratch +per step" vs instant's `do_merge` short-circuit) as one of three drift bugs +(#570, #582, #587) between the two paths. This fix is a partial step toward +#581, not a full closer. + +### Q6 — a group with keys but no value data anywhere: fatal, or skip? + +**Answer: non-fatal skip + loud warning, per (step, group).** Today, +`execute_range_query_pipeline` hard-fails the **entire** range query if any +group resolved from `merged_keys` has no matching entry in `all_data`: +`all_data.get(group_key).ok_or_else(|| "No value for key: ...")?`. This is a +one-time check against a single global `groups` list today. Once key +expansion is per-step, a group having zero value data isn't an anomaly — a +key can legitimately exist per the keys aggregation before/after the value +aggregation ever has data for it (e.g. ingestion boundaries). This should +downgrade from hard-error (poisoning every other group's results in the same +query) to skip-this-group + warn loudly. + +### Q7 — multiple independent groups (real `grouping_labels`): does per-step scoping leak across groups? + +Explored via a concrete example (region=us gains host-b mid-range, +region=eu's host-x never changes — does host-b leak into region=eu's +output?). **Finding: not a real distinct risk in today's code.** Store +query results already come back partitioned by `group_key` +(`MergedOutputsMap = HashMap, ...>`), and +`merge_precomputed_outputs` merges *within* each group_key's own buckets, +not across groups. The RED tests written for this (see below) confirmed it: +failures were always "this group's own key is phantom-early" (Bug 1), never +actual cross-group bleed. Kept as regression guards for once Bug 1 is fixed, +not because they proved a second bug. + +A related candidate — "a group that only appears partway through the +range" — was traced through by hand and found to **not** actually be RED +against today's code as originally described (a `DeltaSetAggregator`'s +`keys_query` already spans `[0, end]` today, so a brand-new group is picked +up correctly by coincidence when its only key never changes). It was +replaced with the Q6 scenario instead (a group with keys but zero value +data, which **does** genuinely hard-fail today). + +### Q8 — a toggle within a single output step's window, at the full-pipeline level + +Raised as "no existing test has two key-delta buckets landing inside the +same step's window" — **this claim was checked and found wrong**: since +`DeltaSetAggregator`'s window always starts at `0` (Q2), the oscillating +5-window test's own `t=2000` checkpoint already merges 2 toggle-buckets +(`[0,1000)` add, `[1000,2000)` remove) within one window. No new test needed +for that framing. + +Re-checking turned up a real, different gap: every `SetAggregator` test uses +`window_size_ms == bucket_width_ms`, so each window only ever contains +**exactly one** bucket. `DeltaSetAggregator` gets multi-bucket-per-window +coverage "for free" via its always-widened `[0,t]` window; `SetAggregator`'s +bounded sliding window has never been exercised merging more than one +bucket. Added `range_query_set_aggregator_merges_multiple_buckets_within_one_window`: +two `SetAggregator` buckets colliding at the same `(start=0,end=1000)` (same +construction as `range_query_sliding_window_merges_both_buckets`, applied to +keys instead of values) must union into `{host-a, host-b}`, guarding against +a per-step `bucket_map` implementation that does `.insert()` instead of +`.entry().or_default().push()` for the keys side and silently drops one of +the colliding buckets. + +## RED test inventory + +All in `asap-query-engine/src/tests/native_range_query_tests.rs` unless noted. +16 tests total: 5 green (regression guards / already-correct behavior), 11 RED +(pin the fix's target behavior). + +| Test | Pins | +|---|---| +| `range_query_dual_population_key_appearing_midrange_has_no_phantom_earlier_sample` | Bug 1, `DeltaSetAggregator`, plain range path | +| `range_query_set_aggregator_earlier_key_not_silently_dropped` | Bug 2, `SetAggregator`, plain range path | +| `range_query_binary_expr_arm_key_appearing_midrange_has_no_phantom_earlier_sample` | Bug 1 through `build_arm_range_context` (binary-expr scalar arm) | +| `range_query_binary_expr_arm_set_aggregator_earlier_key_not_silently_dropped` | Bug 2 through `build_arm_range_context` | +| `range_query_delta_set_aggregator_oscillating_add_remove_across_five_windows` | Multi-toggle replay correctness (add/remove ×5), asserted at every step | +| `range_query_delta_set_aggregator_key_change_on_middle_step_not_just_boundary` | 3-step range, change lands on the interior step, not a boundary | +| `range_query_delta_set_aggregator_key_bucket_width_differs_from_value_bucket_width` | Q4 — keys bucket_map must step by the key aggregation's own width | +| `range_query_dual_population_per_step_key_change_does_not_leak_across_groups` | Q7 — one group changes, one doesn't | +| `range_query_dual_population_simultaneous_cross_group_adds_stay_isolated` | Q7 — both groups change at the same timestamp (sharper signature) | +| `range_query_dual_population_group_with_no_value_data_is_skipped_not_fatal` | Q6 — orphaned group must not poison the whole query | +| `range_query_set_aggregator_merges_multiple_buckets_within_one_window` | Q8 — `SetAggregator` must union, not drop, colliding same-timestamp buckets within one window | +| `naive_merger_sequential_fold_replays_delta_set_toggles_at_every_window` (`window_merger.rs`, **GREEN**) | Explicit, readable pin of the NaiveMerger-sequential-fold requirement Q3 depends on; contrasts against a flat `merge_accumulators` call to make the distinction undeniable | + +`assert_all_at` (test-module helper) collects every mismatched +`(labels, timestamp, expected)` case into one panic message instead of +stopping at the first failing `assert!`, per explicit request — applied to +all multi-assertion tests in the file. + +`create_range_engine_dual_input_with_windows` (thin superset of +`create_range_engine_dual_input`) lets tests set different bucket widths for +the value vs. key aggregation configs, needed for the Q4 test; existing call +sites untouched. + +## Not yet done + +- The actual fix implementation (this session was design + tests only, by + explicit request). +- Incremental carry-forward merging for `DeltaSetAggregator` (Q3) — deferred + until proven necessary. +- Broader unification with #581 — out of scope for #583, but this fix moves + in that direction. From d8a2a1304527ac3dba90d371a48f50994ff333fc Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 15:47:49 -0400 Subject: [PATCH 05/12] 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 --- .../src/engines/simple_engine/mod.rs | 13 ++++++++ .../src/engines/simple_engine/promql.rs | 33 +++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index e7f610c..19a09e3 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -114,6 +114,19 @@ pub struct RangeQueryExecutionContext { pub lookback_bucket_count: usize, /// Tumbling window size in ms pub tumbling_window_ms: u64, + /// Per-step lookback for the keys aggregation (#583): `keys_query.end - + /// keys_query.start` from the instant window `create_keys_query_params` + /// computed before widening. `None` when there's no separate + /// `keys_query`. Reused, unmodified, as the per-step + /// `current_time.saturating_sub(keys_lookback_ms)` window start for + /// every output step -- this single value is what makes `SetAggregator` + /// a normal sliding window and `DeltaSetAggregator` always replay from + /// `0`, with no `AggregationType` branching needed here. + pub keys_lookback_ms: Option, + /// Bucket width for the keys aggregation, separate from + /// `tumbling_window_ms` (which is the *value* aggregation's width) -- + /// the two can legitimately differ. + pub keys_tumbling_window_ms: Option, } // /// Parsed components of a sketch query, extracted either via the PromQL AST diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index bd51a88..570f371 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -584,6 +584,37 @@ impl SimpleEngine { extended_store_plan.values_query.end_timestamp = end_ms; extended_store_plan.values_query.is_exact_query = false; + // #583: widen keys_query the same way, using the instant window + // create_keys_query_params already computed for it as the source of + // truth for "how far back does this aggregation type look." This + // needs no AggregationType branching: for SetAggregator the instant + // window is [end-window_size, end], so keys_lookback_ms == window_size + // and this widens to a normal sliding window; for DeltaSetAggregator + // the instant window is [0, end], so keys_lookback_ms == end_ms, + // which saturating_sub's to 0 for every current_time <= end_ms in + // the per-step loop -- i.e. "replay from the beginning," for free. + let keys_lookback_ms = base_context + .store_plan + .keys_query + .as_ref() + .map(|kq| kq.end_timestamp - kq.start_timestamp); + let keys_tumbling_window_ms = match keys_lookback_ms { + Some(_) => Some( + self.streaming_config + .read() + .unwrap() + .get_aggregation_config(base_context.agg_info.aggregation_id_for_key) + .map(|c| c.window_size_ms)?, + ), + None => None, + }; + if let (Some(keys_query), Some(lookback)) = + (extended_store_plan.keys_query.as_mut(), keys_lookback_ms) + { + keys_query.start_timestamp = start_ms.saturating_sub(lookback); + keys_query.end_timestamp = end_ms; + } + Some(RangeQueryExecutionContext { base: QueryExecutionContext { store_plan: extended_store_plan, @@ -597,6 +628,8 @@ impl SimpleEngine { buckets_per_step, lookback_bucket_count, tumbling_window_ms, + keys_lookback_ms, + keys_tumbling_window_ms, }) } From a755448ec9db321957b9aad903da966b4fefc9b3 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 16:19:40 -0400 Subject: [PATCH 06/12] 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 --- .../src/tests/native_range_query_tests.rs | 171 +++++++++++++++--- 1 file changed, 143 insertions(+), 28 deletions(-) diff --git a/asap-query-engine/src/tests/native_range_query_tests.rs b/asap-query-engine/src/tests/native_range_query_tests.rs index c932018..fde1326 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -63,7 +63,11 @@ mod tests { /// values are present -- needed once there's more than one grouping /// label (e.g. region + host), since a bare host value alone can't tell /// "host-b under region=eu" apart from "host-b under region=us". - fn labels_have_sample_at(elements: &[RangeVectorElement], label_values: &[&str], ts: u64) -> bool { + fn labels_have_sample_at( + elements: &[RangeVectorElement], + label_values: &[&str], + ts: u64, + ) -> bool { elements .iter() .find(|e| { @@ -86,7 +90,9 @@ mod tests { .filter_map(|(labels, ts, expected, reason)| { let actual = labels_have_sample_at(elements, labels, *ts); (actual != *expected).then(|| { - format!("{labels:?}@{ts}: expected present={expected}, got {actual} -- {reason}") + format!( + "{labels:?}@{ts}: expected present={expected}, got {actual} -- {reason}" + ) }) }) .collect(); @@ -716,10 +722,26 @@ mod tests { labels: vec!["host-a".to_string(), "evt-1".to_string()], }); let keys_data: TimeSeriesData = vec![ - (1000, None, Box::new(keys_add.clone()) as Box), - (2000, None, Box::new(keys_remove.clone()) as Box), - (3000, None, Box::new(keys_add.clone()) as Box), - (4000, None, Box::new(keys_remove.clone()) as Box), + ( + 1000, + None, + Box::new(keys_add.clone()) as Box, + ), + ( + 2000, + None, + Box::new(keys_remove.clone()) as Box, + ), + ( + 3000, + None, + Box::new(keys_add.clone()) as Box, + ), + ( + 4000, + None, + Box::new(keys_remove.clone()) as Box, + ), (5000, None, Box::new(keys_add) as Box), ]; @@ -1054,15 +1076,43 @@ mod tests { vec!["region"], vec!["host", "event"], vec![ - (1000, Some(vec!["us".to_string()]), Box::new(cms_us_1) as Box), - (2000, Some(vec!["us".to_string()]), Box::new(cms_us_2) as Box), - (1000, Some(vec!["eu".to_string()]), Box::new(cms_eu_1) as Box), - (2000, Some(vec!["eu".to_string()]), Box::new(cms_eu_2) as Box), + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(cms_us_1) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(cms_us_2) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_1) as Box, + ), + ( + 2000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_2) as Box, + ), ], vec![ - (1000, Some(vec!["us".to_string()]), Box::new(keys_us_1000) as Box), - (2000, Some(vec!["us".to_string()]), Box::new(keys_us_2000) as Box), - (1000, Some(vec!["eu".to_string()]), Box::new(keys_eu_1000) as Box), + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(keys_us_1000) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(keys_us_2000) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(keys_eu_1000) as Box, + ), ], "count(event_frequency) by (region, host, event)", ); @@ -1075,15 +1125,30 @@ mod tests { assert_all_at( &elements, &[ - (&["us", "host-a"], 1000, true, "region=us has host-a from the start"), + ( + &["us", "host-a"], + 1000, + true, + "region=us has host-a from the start", + ), ( &["us", "host-b"], 1000, false, "region=us's host-b is only added at t=2000", ), - (&["us", "host-b"], 2000, true, "region=us's host-b was added by t=2000"), - (&["eu", "host-x"], 1000, true, "region=eu has host-x from the start"), + ( + &["us", "host-b"], + 2000, + true, + "region=us's host-b was added by t=2000", + ), + ( + &["eu", "host-x"], + 1000, + true, + "region=eu has host-x from the start", + ), ( &["eu", "host-x"], 2000, @@ -1153,16 +1218,48 @@ mod tests { vec!["region"], vec!["host", "event"], vec![ - (1000, Some(vec!["us".to_string()]), Box::new(cms_us_1) as Box), - (2000, Some(vec!["us".to_string()]), Box::new(cms_us_2) as Box), - (1000, Some(vec!["eu".to_string()]), Box::new(cms_eu_1) as Box), - (2000, Some(vec!["eu".to_string()]), Box::new(cms_eu_2) as Box), + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(cms_us_1) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(cms_us_2) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_1) as Box, + ), + ( + 2000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_2) as Box, + ), ], vec![ - (1000, Some(vec!["us".to_string()]), Box::new(keys_us_1000) as Box), - (2000, Some(vec!["us".to_string()]), Box::new(keys_us_2000) as Box), - (1000, Some(vec!["eu".to_string()]), Box::new(keys_eu_1000) as Box), - (2000, Some(vec!["eu".to_string()]), Box::new(keys_eu_2000) as Box), + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(keys_us_1000) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(keys_us_2000) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(keys_eu_1000) as Box, + ), + ( + 2000, + Some(vec!["eu".to_string()]), + Box::new(keys_eu_2000) as Box, + ), ], "count(event_frequency) by (region, host, event)", ); @@ -1228,11 +1325,19 @@ mod tests { let mut keys_normal = DeltaSetAggregatorAccumulator::new(); keys_normal.add_key(KeyByLabelValues { - labels: vec!["normal".to_string(), "host-a".to_string(), "evt-1".to_string()], + labels: vec![ + "normal".to_string(), + "host-a".to_string(), + "evt-1".to_string(), + ], }); let mut keys_orphan = DeltaSetAggregatorAccumulator::new(); keys_orphan.add_key(KeyByLabelValues { - labels: vec!["orphan".to_string(), "host-z".to_string(), "evt-1".to_string()], + labels: vec![ + "orphan".to_string(), + "host-z".to_string(), + "evt-1".to_string(), + ], }); let engine = create_range_engine_dual_input( @@ -1383,7 +1488,12 @@ mod tests { "host-b's bucket collides with host-a's at (start=0,end=1000) -- \ both must be merged (unioned), not one dropped in favor of the other", ), - (&["host-c"], 1000, false, "host-c's bucket doesn't exist until t=2000"), + ( + &["host-c"], + 1000, + false, + "host-c's bucket doesn't exist until t=2000", + ), ( &["host-a"], 2000, @@ -1398,7 +1508,12 @@ mod tests { "SetAggregator is latest-window-only: host-c's window replaces \ host-a/host-b's, it doesn't accumulate alongside them", ), - (&["host-c"], 2000, true, "host-c is the live set for the window ending at t=2000"), + ( + &["host-c"], + 2000, + true, + "host-c is the live set for the window ending at t=2000", + ), ], ); } From b9610c6e1ab76b339dc4d15f2bf4f99bf9a5fa9c Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 16:20:12 -0400 Subject: [PATCH 07/12] 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 (the one-time merged snapshot). Replaced with a small KeysSource enum: Fixed(Vec) for single-population groups (unchanged -- they never had a per-step keys concern), or PerStep(&Vec) 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 vs. a &Vec 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 --- .../src/engines/simple_engine/mod.rs | 156 ++++++++++++++---- 1 file changed, 122 insertions(+), 34 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 19a09e3..a0b0a20 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1497,20 +1497,25 @@ impl SimpleEngine { all_data.values().map(|v| v.len()).sum::() ); - // Dual-population metrics (separate key/value aggregations) need the - // keys side merged once up front, then expanded per value group below - // — mirrors execute_and_merge_store_queries/collect_results_separate_keys - // (#580: this pipeline previously never read keys_query at all). - let merged_keys = self.fetch_and_merge_keys( - &context.base.store_plan.keys_query, - context.base.agg_info.aggregation_type_for_key, - context.base.do_merge, - )?; + // #583: fetch keys raw (no merge). Unlike keys, values have always + // been fetched raw here and merged per-step below (see the loop); + // keys used to go through fetch_and_merge_keys, which collapses + // every fetched bucket into ONE snapshot before this function ever + // sees it. That collapse is the bug: once buckets are merged + // together there's no way to ask what the key set looked like at + // any specific earlier timestamp. Fetching raw and merging per-step, + // mirroring the values loop, is the fix. + let keys_raw_data: Option = match &context.base.store_plan.keys_query + { + Some(keys_query) => Some(self.execute_store_query(keys_query)?), + None => None, + }; let mut results: HashMap = HashMap::new(); // Determine accumulator type for merger selection let accumulator_type = &context.base.agg_info.aggregation_type_for_value; + let key_accumulator_type = context.base.agg_info.aggregation_type_for_key; // Calculate step parameters let step_ms = context.range_params.step; @@ -1520,6 +1525,8 @@ impl SimpleEngine { let lookback_bucket_count = context.lookback_bucket_count; let tumbling_window_ms = context.tumbling_window_ms; let lookback_ms = (lookback_bucket_count as u64) * tumbling_window_ms; + let keys_lookback_ms = context.keys_lookback_ms; + let keys_tumbling_window_ms = context.keys_tumbling_window_ms; let window_mode = if buckets_per_step <= lookback_bucket_count { "sliding (slide <= size)" @@ -1538,39 +1545,57 @@ impl SimpleEngine { window_mode ); - // Resolve, for every value group, which output label-keys it serves: - // its own key for single-population metrics, or every key the merged - // keys aggregation expands it to for dual-population metrics. Mirrors - // collect_results_separate_keys exactly, including its error - // semantics — an unresolvable key set fails the whole range query - // (so callers fall back to Prometheus) instead of silently returning - // a partial result. See #582 review. - let groups: Vec<( - &Vec, - Vec, - )> = match &merged_keys { + // #583: where a group's output label-keys come from. `Fixed` for + // single-population metrics — the value's own key IS the output + // key at every timestamp, nothing to merge. `PerStep` for + // dual-population metrics — a reference to that group's raw + // (unmerged) keys buckets, windowed and merged fresh at every + // output timestamp in the loop below, instead of once up front. + enum KeysSource<'a> { + Fixed(Vec), + PerStep(&'a Vec), + } + + // Resolve, for every value group, which groups exist at all (a + // one-time operation — see design doc) and where their expansion + // keys come from. A group with keys data but no value data + // anywhere in the queried range is skipped with a warning instead + // of failing the whole range query (#583; previously + // `.ok_or_else(...)?` here hard-failed everything for one missing + // group). See #582 review for collect_results_separate_keys parity. + let groups: Vec<(&Vec, KeysSource)> = match &keys_raw_data + { Some(keys_map) => keys_map .iter() - .map(|(group_key, keys_precompute)| { - let timestamped_buckets = all_data - .get(group_key) - .ok_or_else(|| format!("No value for key: {:?}", group_key))?; - let expansion_keys = keys_precompute - .get_keys() - .ok_or_else(|| "Keys required for separate aggregation".to_string())?; - Ok((timestamped_buckets, expansion_keys)) - }) - .collect::, String>>()?, + .filter_map( + |(group_key, raw_keys_buckets)| match all_data.get(group_key) { + Some(timestamped_buckets) => { + Some((timestamped_buckets, KeysSource::PerStep(raw_keys_buckets))) + } + None => { + warn!( + "Range query: group {:?} has keys data but no value data \ + anywhere in the queried range — skipping this group instead \ + of failing the whole query (#583)", + group_key + ); + None + } + }, + ) + .collect(), None => all_data .iter() .filter_map(|(group_key, buckets)| { - group_key.as_ref().map(|k| (buckets, vec![k.clone()])) + group_key + .as_ref() + .map(|k| (buckets, KeysSource::Fixed(vec![k.clone()]))) }) .collect(), }; // Process each value group independently - for (timestamped_buckets, expansion_keys) in groups { + for (timestamped_buckets, keys_source) in groups { // Build lookup: bucket_start_timestamp -> all buckets sharing // that start. A Sliding aggregation can legitimately return more // than one bucket per start timestamp (#567/#570) — every one of @@ -1580,15 +1605,78 @@ impl SimpleEngine { bucket_map.entry(*start).or_default().push(bucket.as_ref()); } + // Same idea, for keys (#583) — only built for dual-population + // groups; Fixed groups have nothing to window. + let keys_bucket_map: Option>> = match &keys_source + { + KeysSource::PerStep(raw_keys_buckets) => { + let mut m: HashMap> = HashMap::new(); + for ((start, _), bucket) in raw_keys_buckets.iter() { + m.entry(*start).or_default().push(bucket.as_ref()); + } + Some(m) + } + KeysSource::Fixed(_) => None, + }; + debug!( - "Group with {} start-timestamps, expands to keys: {:?}", + "Group with {} start-timestamps ({} keys start-timestamps)", bucket_map.len(), - expansion_keys + keys_bucket_map.as_ref().map(|m| m.len()).unwrap_or(0) ); // Iterate by OUTPUT timestamp, not by bucket index let mut current_time = start_ms; while current_time <= end_ms { + // #583: resolve THIS step's own expansion keys — not a + // single snapshot reused for every step. + let expansion_keys: Vec = match &keys_source { + KeysSource::Fixed(keys) => keys.clone(), + KeysSource::PerStep(_) => { + let keys_bucket_map = keys_bucket_map + .as_ref() + .expect("PerStep implies keys_bucket_map is Some"); + let keys_lookback_ms = + keys_lookback_ms.expect("PerStep implies keys_lookback_ms is Some"); + let keys_tumbling_window_ms = keys_tumbling_window_ms + .expect("PerStep implies keys_tumbling_window_ms is Some"); + + let keys_window_start = current_time.saturating_sub(keys_lookback_ms); + let mut keys_window_buckets: Vec> = Vec::new(); + let mut kt = keys_window_start; + while kt < current_time { + if let Some(buckets) = keys_bucket_map.get(&kt) { + keys_window_buckets + .extend(buckets.iter().map(|b| b.clone_boxed_core())); + } + kt += keys_tumbling_window_ms; + } + + if keys_window_buckets.is_empty() { + Vec::new() + } else { + let mut key_merger = create_window_merger(key_accumulator_type); + key_merger.initialize(keys_window_buckets); + match key_merger.get_merged() { + Ok(merged) => merged.get_keys().unwrap_or_default(), + Err(e) => { + debug!("Failed to merge keys at t={}: {}", current_time, e); + Vec::new() + } + } + } + } + }; + + if expansion_keys.is_empty() { + debug!( + "No expansion keys resolved at t={} — skipping this step for this group", + current_time + ); + current_time += step_ms; + continue; + } + // Window covers [current_time - lookback_ms, current_time) // This means we look at buckets that START within this range let window_start = current_time.saturating_sub(lookback_ms); From bb951e1475bc5c7ce88adba87acf2e51a49167bb Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 16:42:32 -0400 Subject: [PATCH 08/12] 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 --- .../src/engines/simple_engine/promql.rs | 43 +++++++++++-------- 1 file changed, 25 insertions(+), 18 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 570f371..8e9682b 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -6,7 +6,7 @@ use super::SimpleEngine; use super::{ QueryExecutionContext, QueryMetadata, QueryTimestamps, RangeQueryExecutionContext, - RangeQueryParams, + RangeQueryParams, StoreQueryParams, }; use crate::data_model::{AggregationIdInfo, KeyByLabelValues, QueryConfig, SchemaConfig}; use crate::engines::query_result::{InstantVectorElement, QueryResult, RangeVectorElement}; @@ -540,6 +540,23 @@ impl SimpleEngine { } } + /// Widens `query`'s window to `[start_ms - lookback, end_ms]`, where + /// `lookback` is the width `query` already had (`end_timestamp - + /// start_timestamp`) before this call. Re-anchors whatever window an + /// aggregation's instant fetch already computed so it slides across the + /// whole range, without needing to know *why* that window is that width + /// — e.g. it's what lets the same call widen a Tumbling `values_query`, + /// a Sliding-window instant fetch, or (per #583) a `SetAggregator`'s + /// `[end-window_size, end]` keys window or a `DeltaSetAggregator`'s + /// `[0, end]` one, identically. Returns the lookback so callers can + /// derive bucket-count/step metadata from it. + fn widen_query_window(query: &mut StoreQueryParams, start_ms: u64, end_ms: u64) -> u64 { + let lookback_ms = query.end_timestamp - query.start_timestamp; + query.start_timestamp = start_ms.saturating_sub(lookback_ms); + query.end_timestamp = end_ms; + lookback_ms + } + /// Extends an instant `QueryExecutionContext` into a `RangeQueryExecutionContext`: /// computes the lookback window from the aggregation's tumbling window size, /// validates the range params, and widens the store plan to cover @@ -573,17 +590,14 @@ impl SimpleEngine { }) .ok()?; - let lookback_ms = base_context.store_plan.values_query.end_timestamp - - base_context.store_plan.values_query.start_timestamp; + let mut extended_store_plan = base_context.store_plan.clone(); + let lookback_ms = + Self::widen_query_window(&mut extended_store_plan.values_query, start_ms, end_ms); + extended_store_plan.values_query.is_exact_query = false; let buckets_per_step = (step_ms / tumbling_window_ms) as usize; let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; - let mut extended_store_plan = base_context.store_plan.clone(); - extended_store_plan.values_query.start_timestamp = start_ms.saturating_sub(lookback_ms); - extended_store_plan.values_query.end_timestamp = end_ms; - extended_store_plan.values_query.is_exact_query = false; - // #583: widen keys_query the same way, using the instant window // create_keys_query_params already computed for it as the source of // truth for "how far back does this aggregation type look." This @@ -593,11 +607,10 @@ impl SimpleEngine { // the instant window is [0, end], so keys_lookback_ms == end_ms, // which saturating_sub's to 0 for every current_time <= end_ms in // the per-step loop -- i.e. "replay from the beginning," for free. - let keys_lookback_ms = base_context - .store_plan + let keys_lookback_ms = extended_store_plan .keys_query - .as_ref() - .map(|kq| kq.end_timestamp - kq.start_timestamp); + .as_mut() + .map(|keys_query| Self::widen_query_window(keys_query, start_ms, end_ms)); let keys_tumbling_window_ms = match keys_lookback_ms { Some(_) => Some( self.streaming_config @@ -608,12 +621,6 @@ impl SimpleEngine { ), None => None, }; - if let (Some(keys_query), Some(lookback)) = - (extended_store_plan.keys_query.as_mut(), keys_lookback_ms) - { - keys_query.start_timestamp = start_ms.saturating_sub(lookback); - keys_query.end_timestamp = end_ms; - } Some(RangeQueryExecutionContext { base: QueryExecutionContext { From dca8b5f837953a14ea9e5a1c1a037a0c53b394ff Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 17:40:30 -0400 Subject: [PATCH 09/12] 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 --- .../src/engines/simple_engine/mod.rs | 89 +++++++++++-------- 1 file changed, 54 insertions(+), 35 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index a0b0a20..2ab2d3e 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -617,9 +617,11 @@ impl SimpleEngine { } /// Fetches and merges the keys side of a dual-population query plan, if - /// present. Shared by `execute_and_merge_store_queries` (instant) and - /// `execute_range_query_pipeline` (range) — both build a `StoreQueryPlan` - /// that may carry a separate `keys_query` and need it merged the same way. + /// present. Used by `execute_and_merge_store_queries` (instant) only — + /// `execute_range_query_pipeline` (range) fetches keys raw via + /// `execute_store_query` and merges them per output step instead (#583), + /// since a single merged snapshot can't answer "what did the key set + /// look like at an earlier timestamp." fn fetch_and_merge_keys( &self, keys_query: &Option, @@ -1476,6 +1478,45 @@ impl SimpleEngine { // Some((output_labels, QueryResult::matrix(range_elements))) // } + /// Builds a lookup from bucket start-timestamp to every bucket sharing + /// that start. A Sliding aggregation can legitimately return more than + /// one bucket per start timestamp (#567/#570) — every one of them must + /// be merged, not just the last one collected here. Used identically by + /// `execute_range_query_pipeline` for both the value side and (#583) + /// the keys side. + fn build_bucket_map( + buckets: &[crate::stores::TimestampedBucket], + ) -> HashMap> { + let mut bucket_map: HashMap> = HashMap::new(); + for ((start, _), bucket) in buckets { + bucket_map.entry(*start).or_default().push(bucket.as_ref()); + } + bucket_map + } + + /// Collects every bucket in `bucket_map` whose start falls in + /// `[window_start, window_end)`, stepping by `step_increment`. Missing + /// buckets at a given start are skipped (partial data is okay). Used + /// identically by `execute_range_query_pipeline` for both the value + /// side and (#583) the keys side — the only difference between the two + /// call sites is which map/lookback/bucket-width they pass in. + fn scan_window( + bucket_map: &HashMap>, + window_start: u64, + window_end: u64, + step_increment: u64, + ) -> Vec> { + let mut window_buckets: Vec> = Vec::new(); + let mut t = window_start; + while t < window_end { + if let Some(buckets) = bucket_map.get(&t) { + window_buckets.extend(buckets.iter().map(|b| b.clone_boxed_core())); + } + t += step_increment; + } + window_buckets + } + /// Execute the range query pipeline fn execute_range_query_pipeline( &self, @@ -1596,25 +1637,14 @@ impl SimpleEngine { // Process each value group independently for (timestamped_buckets, keys_source) in groups { - // Build lookup: bucket_start_timestamp -> all buckets sharing - // that start. A Sliding aggregation can legitimately return more - // than one bucket per start timestamp (#567/#570) — every one of - // them must be merged, not just the last one collected here. - let mut bucket_map: HashMap> = HashMap::new(); - for ((start, _), bucket) in timestamped_buckets { - bucket_map.entry(*start).or_default().push(bucket.as_ref()); - } + let bucket_map = Self::build_bucket_map(timestamped_buckets); // Same idea, for keys (#583) — only built for dual-population // groups; Fixed groups have nothing to window. let keys_bucket_map: Option>> = match &keys_source { KeysSource::PerStep(raw_keys_buckets) => { - let mut m: HashMap> = HashMap::new(); - for ((start, _), bucket) in raw_keys_buckets.iter() { - m.entry(*start).or_default().push(bucket.as_ref()); - } - Some(m) + Some(Self::build_bucket_map(raw_keys_buckets)) } KeysSource::Fixed(_) => None, }; @@ -1642,15 +1672,12 @@ impl SimpleEngine { .expect("PerStep implies keys_tumbling_window_ms is Some"); let keys_window_start = current_time.saturating_sub(keys_lookback_ms); - let mut keys_window_buckets: Vec> = Vec::new(); - let mut kt = keys_window_start; - while kt < current_time { - if let Some(buckets) = keys_bucket_map.get(&kt) { - keys_window_buckets - .extend(buckets.iter().map(|b| b.clone_boxed_core())); - } - kt += keys_tumbling_window_ms; - } + let keys_window_buckets = Self::scan_window( + keys_bucket_map, + keys_window_start, + current_time, + keys_tumbling_window_ms, + ); if keys_window_buckets.is_empty() { Vec::new() @@ -1682,16 +1709,8 @@ impl SimpleEngine { let window_start = current_time.saturating_sub(lookback_ms); // Collect all AVAILABLE buckets in this window (skip missing ones) - let mut window_buckets: Vec> = Vec::new(); - - let mut t = window_start; - while t < current_time { - if let Some(buckets) = bucket_map.get(&t) { - window_buckets.extend(buckets.iter().map(|b| b.clone_boxed_core())); - } - // If no bucket at timestamp t, just skip it (partial data is okay) - t += tumbling_window_ms; - } + let window_buckets = + Self::scan_window(&bucket_map, window_start, current_time, tumbling_window_ms); if !window_buckets.is_empty() { // Merge available buckets From 1c7b7a28eacfb7f63ecade24c91711c1bf0b0839 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 17:49:13 -0400 Subject: [PATCH 10/12] docs(query-engine): update #583 design doc status now that the fix landed Co-Authored-By: Claude Sonnet 5 --- docs/583-range-keys-per-step-design.md | 38 ++++++++++++++++++++++++-- 1 file changed, 35 insertions(+), 3 deletions(-) diff --git a/docs/583-range-keys-per-step-design.md b/docs/583-range-keys-per-step-design.md index 2a89d8a..6412456 100644 --- a/docs/583-range-keys-per-step-design.md +++ b/docs/583-range-keys-per-step-design.md @@ -233,10 +233,42 @@ all multi-assertion tests in the file. the value vs. key aggregation configs, needed for the Q4 test; existing call sites untouched. -## Not yet done +## Status + +The fix landed (PR [#595](https://github.com/ProjectASAP/ASAPQuery/pull/595), +draft), staged as planned: + +- Stage 1 — widened `keys_query`'s window in `finish_range_context` + (`keys_lookback_ms`/`keys_tumbling_window_ms` on + `RangeQueryExecutionContext`), behavior-inert on its own. +- Stage 2 — the actual per-step keys merge in + `execute_range_query_pipeline` (raw fetch, `KeysSource` enum, per-step + windowed merge, Q6's non-fatal group skip). +- Stage 3 — confirmed the binary-expr arm path needed no separate code + change, as predicted. +- Stage 4 — full regression pass, 0 failures. + +All 16 `native_range_query_tests` pass; full workspace suite (`cargo test +--workspace`) passes with 0 failures. + +Follow-up refactor (`widen_query_window`, dedupping the values/keys window +formula) and a broader duplication survey against the instant-query path +were also done post-fix. Two items came out of that survey as genuine +design decisions rather than mechanical refactors, and were filed as +separate issues rather than folded into #583: + +- [#596](https://github.com/ProjectASAP/ASAPQuery/issues/596) — range + queries never got the CMS/KLL batch-merge fast path the instant path has + (`NaiveMerger` only does the sequential fallback); the two fallback + implementations also differ in error-handling policy, so unifying them + isn't free. +- [#597](https://github.com/ProjectASAP/ASAPQuery/issues/597) — #583's Q6 + (skip a group with keys-but-no-value-data instead of hard-failing) was + only applied to the range path; the instant path + (`collect_results_separate_keys`) still hard-fails on the identical case. + +## Explicitly not done here -- The actual fix implementation (this session was design + tests only, by - explicit request). - Incremental carry-forward merging for `DeltaSetAggregator` (Q3) — deferred until proven necessary. - Broader unification with #581 — out of scope for #583, but this fix moves From 73125931025e59001397c55a31dd9e1d7327ed61 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 20:32:09 -0400 Subject: [PATCH 11/12] 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 --- .../src/engines/simple_engine/mod.rs | 19 ++++++++++++++++++- .../src/engines/simple_engine/promql.rs | 10 ++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index affd730..32edcf1 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1709,7 +1709,24 @@ impl SimpleEngine { let mut key_merger = create_window_merger(key_accumulator_type); key_merger.initialize(keys_window_buckets); match key_merger.get_merged() { - Ok(merged) => merged.get_keys().unwrap_or_default(), + Ok(merged) => match merged.get_keys() { + Some(keys) => keys, + None => { + // e.g. a DeltaSetAggregator "remove" with no + // matching "add" resolved in this window -- + // distinct from (and louder than) the routine, + // expected "no buckets in this window at all" + // case below, since it means a merge DID + // happen but couldn't resolve a key set. + warn!( + "Keys merge at t={} produced an unresolved key \ + set (get_keys() returned None) -- skipping \ + this step for this group", + current_time + ); + Vec::new() + } + }, Err(e) => { debug!("Failed to merge keys at t={}: {}", current_time, e); Vec::new() diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 8e9682b..10e072e 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -621,6 +621,16 @@ impl SimpleEngine { ), None => None, }; + // A zero window_size_ms 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 has no equivalent check, so guard explicitly. + if keys_tumbling_window_ms == Some(0) { + warn!("Range query validation failed: key aggregation window_size_ms is 0"); + return None; + } Some(RangeQueryExecutionContext { base: QueryExecutionContext { From 320a2e84abb24678ad064c909d57ffd35b2ccd8b Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 20:53:17 -0400 Subject: [PATCH 12/12] 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 --- .../src/engines/simple_engine/mod.rs | 112 +++++++++++------- 1 file changed, 71 insertions(+), 41 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 32edcf1..410b810 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1506,6 +1506,17 @@ impl SimpleEngine { window_end: u64, step_increment: u64, ) -> Vec> { + // A zero step never advances `t`, so the loop below would never + // terminate. This is the single caller-facing guard for that hazard + // -- callers upstream may also validate their own step sources, but + // this function has two callers (values, keys) and should not rely + // on either of them to have done so. Kept active in release builds + // (assert!, not debug_assert!): a hung query is a production + // incident, not just a debug-time nicety. + assert!( + step_increment > 0, + "scan_window: step_increment must be nonzero, or this loop never terminates" + ); let mut window_buckets: Vec> = Vec::new(); let mut t = window_start; while t < window_end { @@ -1605,9 +1616,23 @@ impl SimpleEngine { // top-k), evaluated AFTER merging the window's value buckets // (a top-k heap's keys can depend on that window's data), // falling back to the store-level group key otherwise. + // PerStep bundles everything a dual-population group's per-step + // resolution needs (bucket_map, lookback_ms, tumbling_window_ms) in + // one place, built once at group-construction time — rather than + // three separate Option fields at function scope that only + // happened to be Some together by convention, each re-unwrapped via + // .expect() on every iteration of the per-step loop. Making the + // invalid state (PerStep present but one companion value missing) + // unrepresentable is the same reasoning that motivated this enum + // over two raw Option fields in the first place — just applied all + // the way through instead of partway. enum KeysSource<'a> { Fixed(Vec), - PerStep(&'a Vec), + PerStep { + bucket_map: HashMap>, + lookback_ms: u64, + tumbling_window_ms: u64, + }, } // Resolve, for every value group, which groups exist at all (a @@ -1619,25 +1644,41 @@ impl SimpleEngine { // group). See #582 review for collect_results_separate_keys parity. let groups: Vec<(&Vec, KeysSource)> = match &keys_raw_data { - Some(keys_map) => keys_map - .iter() - .filter_map( - |(group_key, raw_keys_buckets)| match all_data.get(group_key) { - Some(timestamped_buckets) => { - Some((timestamped_buckets, KeysSource::PerStep(raw_keys_buckets))) - } - None => { - warn!( - "Range query: group {:?} has keys data but no value data \ + Some(keys_map) => { + // keys_raw_data is Some, so context.keys_lookback_ms / + // context.keys_tumbling_window_ms are guaranteed Some too + // (both derived from the same keys_query.is_some() check in + // finish_range_context) -- resolved once here instead of + // re-unwrapped per group per step. + let keys_lookback_ms = + keys_lookback_ms.expect("keys_raw_data implies keys_lookback_ms is Some"); + let keys_tumbling_window_ms = keys_tumbling_window_ms + .expect("keys_raw_data implies keys_tumbling_window_ms is Some"); + keys_map + .iter() + .filter_map( + |(group_key, raw_keys_buckets)| match all_data.get(group_key) { + Some(timestamped_buckets) => Some(( + timestamped_buckets, + KeysSource::PerStep { + bucket_map: Self::build_bucket_map(raw_keys_buckets), + lookback_ms: keys_lookback_ms, + tumbling_window_ms: keys_tumbling_window_ms, + }, + )), + None => { + warn!( + "Range query: group {:?} has keys data but no value data \ anywhere in the queried range — skipping this group instead \ of failing the whole query (#583)", - group_key - ); - None - } - }, - ) - .collect(), + group_key + ); + None + } + }, + ) + .collect() + } // #584/#587: keep every group, including group_key=None — that's // exactly where a self-keyed single-population accumulator // (e.g. top-k) is typically stored. An empty fallback list here @@ -1659,20 +1700,13 @@ impl SimpleEngine { for (timestamped_buckets, keys_source) in groups { let bucket_map = Self::build_bucket_map(timestamped_buckets); - // Same idea, for keys (#583) — only built for dual-population - // groups; Fixed groups have nothing to window. - let keys_bucket_map: Option>> = match &keys_source - { - KeysSource::PerStep(raw_keys_buckets) => { - Some(Self::build_bucket_map(raw_keys_buckets)) - } - KeysSource::Fixed(_) => None, - }; - debug!( "Group with {} start-timestamps ({} keys start-timestamps)", bucket_map.len(), - keys_bucket_map.as_ref().map(|m| m.len()).unwrap_or(0) + match &keys_source { + KeysSource::PerStep { bucket_map, .. } => bucket_map.len(), + KeysSource::Fixed(_) => 0, + } ); // Iterate by OUTPUT timestamp, not by bucket index @@ -1686,21 +1720,17 @@ impl SimpleEngine { // value merge below. Fixed (single-population) groups defer // key resolution until after the value merge (#584/#587). let per_step_keys: Option> = match &keys_source { - KeysSource::PerStep(_) => { - let keys_bucket_map = keys_bucket_map - .as_ref() - .expect("PerStep implies keys_bucket_map is Some"); - let keys_lookback_ms = - keys_lookback_ms.expect("PerStep implies keys_lookback_ms is Some"); - let keys_tumbling_window_ms = keys_tumbling_window_ms - .expect("PerStep implies keys_tumbling_window_ms is Some"); - - let keys_window_start = current_time.saturating_sub(keys_lookback_ms); + KeysSource::PerStep { + bucket_map: keys_bucket_map, + lookback_ms: keys_lookback_ms, + tumbling_window_ms: keys_tumbling_window_ms, + } => { + let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); let keys_window_buckets = Self::scan_window( keys_bucket_map, keys_window_start, current_time, - keys_tumbling_window_ms, + *keys_tumbling_window_ms, ); let expansion_keys = if keys_window_buckets.is_empty() { @@ -1771,7 +1801,7 @@ impl SimpleEngine { // priority, falling back to the store-level // group key otherwise. let resolved_keys = match &keys_source { - KeysSource::PerStep(_) => per_step_keys.expect( + KeysSource::PerStep { .. } => per_step_keys.expect( "PerStep always sets per_step_keys above, or continues", ), KeysSource::Fixed(fallback_keys) => {