Fix sliding window aggregation: feed samples into all overlapping windows

Previously each sample was assigned to only one window via
window_start_for(), which is incorrect for sliding windows where
window_size > slide_interval. Added window_starts_containing() that
returns all window starts whose range covers the timestamp, and use
it in the worker aggregation loop.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: zzylol

asap-query-engine/src/precompute_engine/window_manager.rs

@@ -73,6 +73,20 @@ impl WindowManager {
         closed
     }
 
+    /// Return all window starts whose window `[start, start + window_size_ms)`
+    /// contains the given timestamp. For tumbling windows this returns exactly
+    /// one start; for sliding windows it returns `ceil(window_size / slide)`
+    /// starts.
+    pub fn window_starts_containing(&self, timestamp_ms: i64) -> Vec<i64> {
+        let mut starts = Vec::new();
+        let mut start = self.window_start_for(timestamp_ms);
+        while start + self.window_size_ms > timestamp_ms {
+            starts.push(start);
+            start -= self.slide_interval_ms;
+        }
+        starts
+    }
+
     /// Return the window `[start, end)` boundaries for a given window start.
     pub fn window_bounds(&self, window_start: i64) -> (i64, i64) {
         (window_start, window_start + self.window_size_ms)
@@ -151,4 +165,34 @@ mod tests {
         let closed = wm.closed_windows(15_000, 35_000);
         assert_eq!(closed, vec![0]);
     }
+
+    #[test]
+    fn test_window_starts_containing_tumbling() {
+        // 60s tumbling windows — each sample belongs to exactly one window
+        let wm = WindowManager::new(60, 0);
+        let mut starts = wm.window_starts_containing(15_000);
+        starts.sort();
+        assert_eq!(starts, vec![0]);
+
+        let mut starts = wm.window_starts_containing(60_000);
+        starts.sort();
+        assert_eq!(starts, vec![60_000]);
+    }
+
+    #[test]
+    fn test_window_starts_containing_sliding() {
+        // 30s window, 10s slide — each sample belongs to 3 windows
+        let wm = WindowManager::new(30, 10);
+
+        // t=15_000 belongs to [0, 30_000), [10_000, 40_000)
+        // and [-10_000, 20_000) which starts negative — still returned
+        let mut starts = wm.window_starts_containing(15_000);
+        starts.sort();
+        assert_eq!(starts, vec![-10_000, 0, 10_000]);
+
+        // t=30_000 belongs to [10_000, 40_000), [20_000, 50_000), [30_000, 60_000)
+        let mut starts = wm.window_starts_containing(30_000);
+        starts.sort();
+        assert_eq!(starts, vec![10_000, 20_000, 30_000]);
+    }
 }

asap-query-engine/src/precompute_engine/worker.rs

@@ -221,18 +221,20 @@ impl Worker {
                     continue; // already dropped
                 }
 
-                let window_start = agg_state.window_manager.window_start_for(ts);
-
-                let updater = agg_state
-                    .active_windows
-                    .entry(window_start)
-                    .or_insert_with(|| create_accumulator_updater(&agg_state.config));
-
-                if updater.is_keyed() {
-                    let key = extract_key_from_series(series_key, &agg_state.config);
-                    updater.update_keyed(&key, val, ts);
-                } else {
-                    updater.update_single(val, ts);
+                let window_starts = agg_state.window_manager.window_starts_containing(ts);
+
+                for window_start in window_starts {
+                    let updater = agg_state
+                        .active_windows
+                        .entry(window_start)
+                        .or_insert_with(|| create_accumulator_updater(&agg_state.config));
+
+                    if updater.is_keyed() {
+                        let key = extract_key_from_series(series_key, &agg_state.config);
+                        updater.update_keyed(&key, val, ts);
+                    } else {
+                        updater.update_single(val, ts);
+                    }
                 }
             }