From 5d6017d3360073fffbfbbbf6e9de4891e061b481 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 22:17:29 -0400 Subject: [PATCH 1/2] fix(query-engine): Sliding range queries no longer sum overlapping windows execute_range_query_pipeline's per-step composition (scan_window) walked every grid position in a step's lookback span and summed whatever it found there. That's correct for Tumbling, where each stored bucket is a genuinely disjoint slice -- but wrong for Sliding, where each stored bucket is already a complete, pre-merged window (worker.rs pre-merges before storing). Summing several of those together double/triple-counted overlapping data. Both value-side and keys-side per-step composition now branch on the aggregation's WindowType: Sliding takes a single lookup at the step's exact window position (the data was already fetched by the existing wide fetch, so no extra store round-trips); Tumbling is unchanged. Fixes #608. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 82 ++++++++++++++--- .../src/engines/simple_engine/promql.rs | 31 ++++--- .../exact_window_grid_adversarial_tests.rs | 26 ++++-- .../src/tests/native_range_query_tests.rs | 92 +++++++++++++++++++ 4 files changed, 193 insertions(+), 38 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 56a4a84..29f9c1e 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -114,6 +114,20 @@ 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, + /// 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, /// 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 +1492,25 @@ 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() + } + /// Execute the range query pipeline fn execute_range_query_pipeline( &self, @@ -1527,13 +1560,16 @@ 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; let keys_lookback_ms = context.keys_lookback_ms; let keys_tumbling_window_ms = context.keys_tumbling_window_ms; + let keys_window_type = context.keys_window_type; - // 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. + // 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 +1622,7 @@ impl SimpleEngine { bucket_map: HashMap>, lookback_ms: u64, tumbling_window_ms: u64, + window_type: WindowType, }, } @@ -1608,6 +1645,8 @@ 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"); keys_map .iter() .filter_map( @@ -1618,6 +1657,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,14 +1713,23 @@ 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( - keys_bucket_map, - keys_window_start, - current_time, - *keys_tumbling_window_ms, - ); + // Same Sliding-vs-Tumbling split as the value side + // (#608): a Sliding keys aggregation (e.g. + // SetAggregator) stores complete pre-merged windows + // too, so one lookup, not a scan-and-sum. + let keys_window_buckets = if *keys_window_type == WindowType::Sliding { + Self::single_window(keys_bucket_map, keys_window_start) + } else { + Self::scan_window( + keys_bucket_map, + keys_window_start, + current_time, + *keys_tumbling_window_ms, + ) + }; if keys_window_buckets.is_empty() { debug!( @@ -1709,9 +1758,16 @@ 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); + // Sliding: `lookback_ms` == `window_size_ms`, so the bucket + // at `window_start` is already this step's complete answer -- + // one lookup, not a scan-and-sum (#608). Tumbling: buckets + // are genuinely disjoint, so sum every one found across the + // lookback span, as before. + let window_buckets = if window_type == WindowType::Sliding { + Self::single_window(&bucket_map, window_start) + } else { + Self::scan_window(&bucket_map, window_start, current_time, tumbling_window_ms) + }; 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..c039dd8 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -576,12 +576,12 @@ 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) = { + 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) + }; self.validate_range_query_params(start_ms, end_ms, step_ms, tumbling_window_ms) .map_err(|e| { @@ -611,15 +611,14 @@ 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) = 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)) + } + 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 }`) @@ -645,6 +644,8 @@ impl SimpleEngine { buckets_per_step, lookback_bucket_count, tumbling_window_ms, + window_type, + keys_window_type, 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, From 850efa1acf3473c65a33378254fdf0d3f8cae044 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 22:42:04 -0400 Subject: [PATCH 2/2] refactor(query-engine): dedupe Sliding/Tumbling branch, assert lookback invariant Code review on #621 flagged two follow-ups: - The Sliding-vs-Tumbling branch (single_window vs scan_window) was duplicated near-verbatim between the value-side and keys-side loops. Extracted into one window_buckets_for_step helper, used by both. - single_window's correctness for Sliding depends on lookback_ms == window_size_ms, previously documented but not checked anywhere. Threaded window_size_ms/keys_window_size_ms through RangeQueryExecutionContext and added active asserts (not debug_assert!) guarding the equality, matching scan_window's existing precedent of asserting its own precondition -- a broken invariant here means silently wrong data, the same failure mode #608 fixed. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 88 ++++++++++++++----- .../src/engines/simple_engine/promql.rs | 33 ++++--- 2 files changed, 86 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 29f9c1e..be14d90 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -123,11 +123,19 @@ pub struct RangeQueryExecutionContext { /// 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 @@ -1511,6 +1519,24 @@ impl SimpleEngine { .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, @@ -1561,9 +1587,22 @@ impl SimpleEngine { 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; 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. @@ -1647,6 +1686,17 @@ impl SimpleEngine { .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( @@ -1716,20 +1766,13 @@ impl SimpleEngine { window_type: keys_window_type, } => { let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); - // Same Sliding-vs-Tumbling split as the value side - // (#608): a Sliding keys aggregation (e.g. - // SetAggregator) stores complete pre-merged windows - // too, so one lookup, not a scan-and-sum. - let keys_window_buckets = if *keys_window_type == WindowType::Sliding { - Self::single_window(keys_bucket_map, keys_window_start) - } else { - Self::scan_window( - keys_bucket_map, - keys_window_start, - current_time, - *keys_tumbling_window_ms, - ) - }; + 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() { debug!( @@ -1758,16 +1801,13 @@ impl SimpleEngine { // This means we look at buckets that START within this range let window_start = current_time.saturating_sub(lookback_ms); - // Sliding: `lookback_ms` == `window_size_ms`, so the bucket - // at `window_start` is already this step's complete answer -- - // one lookup, not a scan-and-sum (#608). Tumbling: buckets - // are genuinely disjoint, so sum every one found across the - // lookback span, as before. - let window_buckets = if window_type == WindowType::Sliding { - Self::single_window(&bucket_map, window_start) - } else { - 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 c039dd8..0ad0e28 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -576,11 +576,15 @@ 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, window_type) = { + 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) + ( + 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) @@ -611,15 +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, keys_window_type) = 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)) - } - None => (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,7 +654,9 @@ impl SimpleEngine { 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, })