diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 56a4a84..be14d90 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -114,6 +114,28 @@ pub struct RangeQueryExecutionContext { pub lookback_bucket_count: usize, /// Tumbling window size in ms pub tumbling_window_ms: u64, + /// The value aggregation's `WindowType`. Picks how the per-step loop + /// composes a step's window from `bucket_map`: Sliding buckets are each + /// already a complete `window_size_ms`-wide merged window (see + /// `worker.rs::merge_panes_for_window`), so a step takes exactly the one + /// bucket at `current_time - lookback_ms` (`lookback_ms` == + /// `window_size_ms` here); Tumbling buckets are genuinely disjoint, so a + /// step sums every bucket `scan_window` finds across the lookback span + /// (#608). + pub window_type: WindowType, + /// The value aggregation's actual `window_size_ms`, independent of + /// `tumbling_window_ms` (which is `bucket_step_ms`, not the window + /// size). Used only to assert `lookback_ms == window_size_ms` for + /// Sliding before `single_window` relies on that equality (#608 review). + pub window_size_ms: u64, + /// Same as `window_type`, for the keys aggregation -- `None` when + /// there's no separate `keys_query`. Can legitimately differ from + /// `window_type` (e.g. a Sliding SetAggregator keys aggregation paired + /// with a Tumbling value aggregation, or vice versa). + pub keys_window_type: Option, + /// Same as `window_size_ms`, for the keys aggregation. `None` under the + /// same condition as `keys_window_type`. + pub keys_window_size_ms: Option, /// 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 @@ -1478,6 +1500,43 @@ impl SimpleEngine { window_buckets } + /// Returns whatever bucket(s) `bucket_map` has at exactly + /// `window_start`, or empty if none. Unlike `scan_window`, does not walk + /// or sum multiple grid positions: for a Sliding aggregation, the bucket + /// at `window_start` is already the complete, correctly-merged answer + /// for its window (`worker.rs::merge_panes_for_window` pre-merges before + /// storing), so summing it with neighboring positions would double-count + /// overlapping data (#608). Used identically by + /// `execute_range_query_pipeline` for both the value side and the keys + /// side. + fn single_window( + bucket_map: &HashMap>, + window_start: u64, + ) -> Vec> { + bucket_map + .get(&window_start) + .map(|buckets| buckets.iter().map(|b| b.clone_boxed_core()).collect()) + .unwrap_or_default() + } + + /// Picks how a step's window is composed from `bucket_map`: Sliding -> + /// `single_window` (one lookup); Tumbling -> `scan_window` + /// (scan-and-sum). Used identically by `execute_range_query_pipeline` + /// for both the value side and the keys side (#608). + fn window_buckets_for_step( + bucket_map: &HashMap>, + window_start: u64, + window_end: u64, + step_increment: u64, + window_type: WindowType, + ) -> Vec> { + if window_type == WindowType::Sliding { + Self::single_window(bucket_map, window_start) + } else { + Self::scan_window(bucket_map, window_start, window_end, step_increment) + } + } + /// Execute the range query pipeline fn execute_range_query_pipeline( &self, @@ -1527,13 +1586,29 @@ 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 window_type = context.window_type; + // single_window's correctness for Sliding depends on this equality + // holding -- it looks up exactly one bucket at + // `current_time - lookback_ms` and trusts that position to be the + // step's whole window. Active assert (not debug_assert!): a broken + // equality here means silently wrong data, the same failure mode + // #608 fixed, not just a debug-time nicety (#608 review). + assert!( + window_type != WindowType::Sliding || lookback_ms == context.window_size_ms, + "Sliding range query: lookback_ms ({lookback_ms}) must equal window_size_ms \ + ({}) -- single_window's per-step lookup is only correct under this invariant", + context.window_size_ms + ); let keys_lookback_ms = context.keys_lookback_ms; let keys_tumbling_window_ms = context.keys_tumbling_window_ms; - - // Named distinctly from `WindowType` (Sliding/Tumbling, picks the store - // fetch call) -- this describes step-to-step overlap in the OUTPUT - // iteration, an unrelated concept that happens to reuse the words - // "sliding"/"hopping". See #581. + let keys_window_type = context.keys_window_type; + let keys_window_size_ms = context.keys_window_size_ms; + + // Named distinctly from `WindowType` (Sliding/Tumbling, picks how a + // step's window is composed from `bucket_map` below -- one lookup vs. + // a scan-and-sum, see #608) -- this describes step-to-step overlap in + // the OUTPUT iteration, an unrelated concept that happens to reuse + // the words "sliding"/"hopping". See #581. let step_overlap_mode = if buckets_per_step <= lookback_bucket_count { "sliding (slide <= size)" } else { @@ -1586,6 +1661,7 @@ impl SimpleEngine { bucket_map: HashMap>, lookback_ms: u64, tumbling_window_ms: u64, + window_type: WindowType, }, } @@ -1608,6 +1684,19 @@ impl SimpleEngine { 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"); + let keys_window_type = + keys_window_type.expect("keys_raw_data implies keys_window_type is Some"); + let keys_window_size_ms = + keys_window_size_ms.expect("keys_raw_data implies keys_window_size_ms is Some"); + // Same invariant as the value side's assert above, for the + // keys aggregation (#608 review). + assert!( + keys_window_type != WindowType::Sliding + || keys_lookback_ms == keys_window_size_ms, + "Sliding range query: keys_lookback_ms ({keys_lookback_ms}) must equal \ + keys_window_size_ms ({keys_window_size_ms}) -- single_window's per-step \ + keys lookup is only correct under this invariant" + ); keys_map .iter() .filter_map( @@ -1618,6 +1707,7 @@ impl SimpleEngine { bucket_map: Self::build_bucket_map(raw_keys_buckets), lookback_ms: keys_lookback_ms, tumbling_window_ms: keys_tumbling_window_ms, + window_type: keys_window_type, }, )), None => { @@ -1673,13 +1763,15 @@ impl SimpleEngine { bucket_map: keys_bucket_map, lookback_ms: keys_lookback_ms, tumbling_window_ms: keys_tumbling_window_ms, + window_type: keys_window_type, } => { let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); - let keys_window_buckets = Self::scan_window( + let keys_window_buckets = Self::window_buckets_for_step( keys_bucket_map, keys_window_start, current_time, *keys_tumbling_window_ms, + *keys_window_type, ); if keys_window_buckets.is_empty() { @@ -1709,9 +1801,13 @@ impl SimpleEngine { // This means we look at buckets that START within this range let window_start = current_time.saturating_sub(lookback_ms); - // Collect all AVAILABLE buckets in this window (skip missing ones) - let window_buckets = - Self::scan_window(&bucket_map, window_start, current_time, tumbling_window_ms); + let window_buckets = Self::window_buckets_for_step( + &bucket_map, + window_start, + current_time, + tumbling_window_ms, + window_type, + ); if window_buckets.is_empty() { // No data at all for this window - skip sample diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 22f843a..0ad0e28 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -576,12 +576,16 @@ impl SimpleEngine { let end_ms = Self::convert_query_time_to_data_time(end); let step_ms = (step * 1000.0) as u64; - let tumbling_window_ms = self - .streaming_config - .read() - .unwrap() - .get_aggregation_config(base_context.agg_info.aggregation_id_for_value) - .map(Self::bucket_step_ms)?; + let (tumbling_window_ms, window_type, window_size_ms) = { + let sc = self.streaming_config.read().unwrap(); + let config = + sc.get_aggregation_config(base_context.agg_info.aggregation_id_for_value)?; + ( + Self::bucket_step_ms(config), + config.window_type, + config.window_size_ms, + ) + }; self.validate_range_query_params(start_ms, end_ms, step_ms, tumbling_window_ms) .map_err(|e| { @@ -611,16 +615,20 @@ impl SimpleEngine { .keys_query .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 - .read() - .unwrap() - .get_aggregation_config(base_context.agg_info.aggregation_id_for_key) - .map(Self::bucket_step_ms)?, - ), - None => None, - }; + let (keys_tumbling_window_ms, keys_window_type, keys_window_size_ms) = + match keys_lookback_ms { + Some(_) => { + let sc = self.streaming_config.read().unwrap(); + let config = + sc.get_aggregation_config(base_context.agg_info.aggregation_id_for_key)?; + ( + Some(Self::bucket_step_ms(config)), + Some(config.window_type), + Some(config.window_size_ms), + ) + } + None => (None, 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 @@ -645,6 +653,10 @@ impl SimpleEngine { buckets_per_step, lookback_bucket_count, tumbling_window_ms, + window_type, + window_size_ms, + keys_window_type, + keys_window_size_ms, keys_lookback_ms, keys_tumbling_window_ms, }) diff --git a/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs b/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs index c80818b..a2bc49c 100644 --- a/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs +++ b/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs @@ -503,12 +503,18 @@ mod tests { /// Step 4000: window [1000,4000) -> panes {1000,2000,3000} -> 10+100+1000=1110 /// Step 5000: window [2000,5000) -> panes {2000,3000,4000} -> 100+1000+10000=11100 /// - /// Currently fails: pins the pre-existing bug tracked in - /// https://github.com/ProjectASAP/ASAPQuery/issues/608 (range queries - /// over Sliding-window aggregations use the overlap-scan fetch, not - /// exact-window fetch, then get merged downstream as if Tumbling). - /// Un-ignore once #608 lands. - #[ignore = "known bug, see #608"] + /// Pins the bug tracked in + /// https://github.com/ProjectASAP/ASAPQuery/issues/608: the store holds + /// one pre-merged, `window_size_ms`-wide bucket per grid position (see + /// `create_engine_multi_timestamp_with_window`, mirroring + /// `worker.rs::merge_panes_for_window`'s real output shape) -- each + /// bucket is already a complete answer for its own window. The range + /// pipeline's per-step `scan_window` doesn't know that: it walks every + /// grid position in `[current_time - window_size_ms, current_time)` and + /// sums whatever it finds there, which is correct for genuinely disjoint + /// Tumbling buckets but over-counts for Sliding, where every position in + /// that span holds a distinct, overlapping full-window bucket. Here that + /// over-count is `111 + 1110 + 11100 = 12321` instead of `111`. #[tokio::test(flavor = "multi_thread")] async fn range_query_sliding_multi_step_off_grid_windows_no_double_count_or_drop() { let data = vec![ @@ -625,10 +631,10 @@ mod tests { /// Step 3000: window [1000,3000) -> panes {1000,2000} -> 10+100=110 /// Step 4000: window [2000,4000) -> panes {2000,3000} -> 100+1000=1100 /// - /// Currently fails: same pre-existing bug as the test above, tracked in - /// https://github.com/ProjectASAP/ASAPQuery/issues/608. Un-ignore once - /// #608 lands. - #[ignore = "known bug, see #608"] + /// Same root cause as the test above (#608): each grid position holds a + /// complete, already-merged window, and scan_window sums every position + /// in the lookback span instead of taking the single one at + /// `current_time - window_size_ms`. #[tokio::test(flavor = "multi_thread")] async fn range_query_sliding_overlap_shifts_correctly_across_steps() { let data = vec![ 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 f22a264..4cfa642 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -456,6 +456,98 @@ mod tests { ); } + /// #608's keys-side counterpart: SetAggregator is a real Sliding-capable + /// keys aggregation (unlike DeltaSetAggregator, restricted to Tumbling + /// by #606), so the keys-side per-step composition has the same + /// double-count risk as the value side. Three keys buckets, each + /// window_size_ms(=2000)-wide and already fully merged (matching + /// worker.rs's real output shape), starting 1000ms (slide_interval_ms) + /// apart, each with a DISTINCT key so a leaked neighbor is directly + /// observable rather than masked by set-union idempotence: + /// bucket [1000,3000) -> {host-a,evt-1} + /// bucket [2000,4000) -> {host-a,evt-2} + /// bucket [3000,5000) -> {host-a,evt-3} + /// Step 3000: lookback window [1000,3000) -> exactly bucket [1000,3000) + /// -> only evt-1. A scan-and-sum over every grid position in + /// [1000,3000) would also visit t=2000 and wrongly pull in evt-2 (a + /// window that only starts becoming valid data at t=4000). + /// Step 4000: lookback window [2000,4000) -> exactly bucket [2000,4000) + /// -> only evt-2 (evt-1 must have rolled off, evt-3 must not leak in). + #[tokio::test(flavor = "multi_thread")] + async fn range_query_sliding_keys_no_double_count_across_steps() { + 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-a".to_string(), "evt-2".to_string()], + }); + let mut keys_3 = SetAggregatorAccumulator::new(); + keys_3.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-3".to_string()], + }); + + let engine = create_range_engine_dual_input_sliding_keys( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + ( + 3000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ( + 4000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ], + vec![ + (3000, None, Box::new(keys_1) as Box), + (4000, None, Box::new(keys_2) as Box), + (5000, None, Box::new(keys_3) as Box), + ], + "count(event_frequency) by (host, event)", + 1000, // value_window_ms (Tumbling, unaffected by #608) + 2000, // key_window_size_ms + 1000, // key_slide_interval_ms + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 3.0, 4.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert!( + labels_have_sample_at(&elements, &["host-a", "evt-1"], 3000), + "step 3000 must include evt-1 (its own window)" + ); + assert!( + !labels_have_sample_at(&elements, &["host-a", "evt-2"], 3000), + "#608: step 3000 must NOT include evt-2 -- that key only becomes valid \ + at t=4000; a scan-and-sum over [1000,3000) wrongly visits evt-2's grid \ + position too and leaks it in one step early" + ); + assert!( + labels_have_sample_at(&elements, &["host-a", "evt-2"], 4000), + "step 4000 must include evt-2 (its own window)" + ); + assert!( + !labels_have_sample_at(&elements, &["host-a", "evt-1"], 4000), + "step 4000 must NOT still include evt-1 -- it rolled off" + ); + assert!( + !labels_have_sample_at(&elements, &["host-a", "evt-3"], 4000), + "#608: step 4000 must NOT include evt-3 -- that key only becomes valid \ + at t=5000; a scan-and-sum over [2000,4000) wrongly visits evt-3's grid \ + position too and leaks it in one step early" + ); + } + #[tokio::test(flavor = "multi_thread")] async fn range_query_dual_population_returns_key_expansion() { // Same dual-population shape as native_binary_instant_tests::binary_expr_vector_vector_dual_population,