fix(query-engine): Sliding range queries no longer sum overlapping windows

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<WindowType>,
+    /// Same as `window_size_ms`, for the keys aggregation. `None` under the
+    /// same condition as `keys_window_type`.
+    pub keys_window_size_ms: Option<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
@@ -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<u64, Vec<&dyn AggregateCore>>,
+        window_start: u64,
+    ) -> Vec<Box<dyn AggregateCore>> {
+        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<u64, Vec<&dyn AggregateCore>>,
+        window_start: u64,
+        window_end: u64,
+        step_increment: u64,
+        window_type: WindowType,
+    ) -> Vec<Box<dyn AggregateCore>> {
+        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<u64, Vec<&'a dyn AggregateCore>>,
                 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

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,
         })

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![

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<dyn AggregateCore>,
+                ),
+                (
+                    4000,
+                    None,
+                    Box::new(CountMinSketchAccumulator::new(2, 3)) as Box<dyn AggregateCore>,
+                ),
+            ],
+            vec![
+                (3000, None, Box::new(keys_1) as Box<dyn AggregateCore>),
+                (4000, None, Box::new(keys_2) as Box<dyn AggregateCore>),
+                (5000, None, Box::new(keys_3) as Box<dyn AggregateCore>),
+            ],
+            "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,