fix(query-engine): judge a Sliding range step's cover by distinct windows (#847)

* fix(query-engine): judge a Sliding range step's cover by distinct windows

Two workers can each emit a bucket for the same (key, window), so counting
buckets let a duplicate stand in for a missing cover window and return a
partial value. Count distinct window starts instead, on both the value and
the key side.

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

* fix(query-engine): check Sliding covers before cloning

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: milindsrivastava1997

asap-query-engine/src/engines/simple_engine/mod.rs

@@ -2342,11 +2342,9 @@ impl SimpleEngine {
     }
 
     /// 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.
+    /// `[window_start, window_end)`. Sliding walks by its stored-window
+    /// stride; Tumbling walks by `step_increment`. An incomplete Sliding
+    /// cover is rejected before accumulator sketches are cloned.
     ///
     /// Cost is proportional to `(window_end - window_start) / step_increment`
     /// -- the nominal window width -- not to how many buckets actually exist
@@ -2356,32 +2354,59 @@ impl SimpleEngine {
     /// `[0, current_time)` ("replay from the beginning") -- callers on that
     /// path MUST use `collect_bucket_map_entries_before` instead, never this
     /// (#581 stage E.4 review; see that function's doc for why).
-    fn sum_window(
+    fn window_buckets_for_step(
         bucket_map: &BucketMap,
         window_start: u64,
         window_end: u64,
         step_increment: u64,
-    ) -> Vec<Box<dyn AggregateCore>> {
+        window_type: WindowType,
+        stored_window_size_ms: u64,
+        expected_sliding_window_count: Option<usize>,
+    ) -> Option<Vec<Box<dyn AggregateCore>>> {
         // 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.
+        let grid_step_ms = if window_type == WindowType::Sliding {
+            stored_window_size_ms
+        } else {
+            step_increment
+        };
         assert!(
-            step_increment > 0,
-            "sum_window: step_increment must be nonzero, or this loop never terminates"
+            grid_step_ms > 0,
+            "window_buckets_for_step: grid step must be nonzero"
         );
-        let mut window_buckets: Vec<Box<dyn AggregateCore>> = Vec::new();
+        assert_eq!(
+            expected_sliding_window_count.is_some(),
+            window_type == WindowType::Sliding,
+            "only Sliding windows have a completeness requirement"
+        );
+
+        let mut selected_buckets = Vec::new();
+        let mut covered_window_count = 0;
         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()));
+            if let Some(window_buckets) = bucket_map.get(&t) {
+                if !window_buckets.is_empty() {
+                    covered_window_count += 1;
+                    selected_buckets.extend(window_buckets.iter());
+                }
             }
-            t += step_increment;
+            t += grid_step_ms;
+        }
+        if expected_sliding_window_count.is_some_and(|expected| covered_window_count < expected) {
+            return None;
         }
-        window_buckets
+
+        Some(
+            selected_buckets
+                .into_iter()
+                .map(|bucket| bucket.clone_boxed_core())
+                .collect(),
+        )
     }
 
     /// Collects every bucket in `bucket_map` with a start timestamp strictly
@@ -2444,30 +2469,6 @@ impl SimpleEngine {
             .collect()
     }
 
-    /// Picks how a step's window is composed from `bucket_map`: Sliding walks
-    /// the non-overlapping stored-window stride `W`; Tumbling walks the
-    /// bucket grid stride. Used identically for values and SetAggregator
-    /// keys. Walking Sliding by its slide `S` would double-count overlapping
-    /// full-window aggregates (#608/#621).
-    ///
-    /// NOT used for `AggregationType::DeltaSetAggregator` keys -- callers on
-    /// that path must call `collect_bucket_map_entries_before` directly
-    /// instead (see its doc comment).
-    fn window_buckets_for_step(
-        bucket_map: &BucketMap,
-        window_start: u64,
-        window_end: u64,
-        step_increment: u64,
-        window_type: WindowType,
-        stored_window_size_ms: u64,
-    ) -> Vec<Box<dyn AggregateCore>> {
-        if window_type == WindowType::Sliding {
-            Self::sum_window(bucket_map, window_start, window_end, stored_window_size_ms)
-        } else {
-            Self::sum_window(bucket_map, window_start, window_end, step_increment)
-        }
-    }
-
     /// Execute the range query pipeline.
     ///
     /// `enable_topk_limiting`/`enable_topk_formatting` mirror
@@ -2987,28 +2988,28 @@ impl SimpleEngine {
                                 };
                                 let keys_window_start =
                                     keys_window_end.saturating_sub(*keys_lookback_ms);
-                                Self::window_buckets_for_step(
+                                let expected_sliding_window_count =
+                                    (*keys_window_type == WindowType::Sliding).then_some(
+                                        (*keys_lookback_ms / *keys_stored_window_size_ms) as usize,
+                                    );
+                                let Some(keys_window_buckets) = Self::window_buckets_for_step(
                                     keys_bucket_map,
                                     keys_window_start,
                                     keys_window_end,
                                     *keys_tumbling_window_ms,
                                     *keys_window_type,
                                     *keys_stored_window_size_ms,
-                                )
+                                    expected_sliding_window_count,
+                                ) else {
+                                    debug!(
+                                        "Skipping incomplete Sliding key cover at t={}",
+                                        current_time
+                                    );
+                                    continue;
+                                };
+                                keys_window_buckets
                             };
 
-                        if *keys_window_type == WindowType::Sliding {
-                            let expected =
-                                (*keys_lookback_ms / *keys_stored_window_size_ms) as usize;
-                            if keys_window_buckets.len() < expected {
-                                debug!(
-                                    "Skipping incomplete Sliding key cover at t={}",
-                                    current_time
-                                );
-                                continue;
-                            }
-                        }
-
                         if keys_window_buckets.is_empty() {
                             debug!(
                                 "No keys data in window at t={} — skipping this step for this group",
@@ -3043,14 +3044,23 @@ impl SimpleEngine {
                 };
                 let window_start = window_end.saturating_sub(lookback_ms);
 
-                let window_buckets = Self::window_buckets_for_step(
+                let expected_sliding_window_count = (window_type == WindowType::Sliding)
+                    .then_some((lookback_ms / context.window_size_ms) as usize);
+                let Some(window_buckets) = Self::window_buckets_for_step(
                     bucket_map,
                     window_start,
                     window_end,
                     tumbling_window_ms,
                     window_type,
                     context.window_size_ms,
-                );
+                    expected_sliding_window_count,
+                ) else {
+                    debug!(
+                        "Skipping incomplete Sliding value cover at t={}",
+                        current_time
+                    );
+                    continue;
+                };
 
                 trace!(
                     current_time,
@@ -3063,17 +3073,6 @@ impl SimpleEngine {
                     "Composed query output window from stored buckets"
                 );
 
-                if window_type == WindowType::Sliding {
-                    let expected = (lookback_ms / context.window_size_ms) as usize;
-                    if window_buckets.len() < expected {
-                        debug!(
-                            "Skipping incomplete Sliding value cover at t={}",
-                            current_time
-                        );
-                        continue;
-                    }
-                }
-
                 if window_buckets.is_empty() {
                     // No data at all for this window - skip sample
                     debug!(

asap-query-engine/src/tests/window_semantics_consistency_tests.rs

@@ -331,6 +331,184 @@ mod tests {
         );
     }
 
+    /// Two workers can each emit a bucket for the same (key, window), so a
+    /// range step's Sliding cover must be judged by distinct window starts,
+    /// not bucket count: two buckets for [3s, 6s) must not stand in for the
+    /// missing [0s, 3s).
+    #[tokio::test(flavor = "multi_thread")]
+    async fn sliding_range_step_with_duplicate_bucket_and_missing_window_is_skipped() {
+        let query = "sum_over_time(cpu_load[6s])";
+        let streaming_config = Arc::new(StreamingConfig {
+            aggregation_configs: HashMap::from([(
+                1,
+                AggregationConfig {
+                    aggregation_id: 1,
+                    aggregation_type: AggregationType::Sum,
+                    aggregation_sub_type: String::new(),
+                    parameters: HashMap::new(),
+                    grouping_labels: KeyByLabelNames::new(vec!["host".to_string()]),
+                    aggregated_labels: KeyByLabelNames::empty(),
+                    rollup_labels: KeyByLabelNames::empty(),
+                    original_yaml: String::new(),
+                    window_size_ms: 3_000,
+                    slide_interval_ms: 1_000,
+                    window_type: WindowType::Sliding,
+                    spatial_filter: String::new(),
+                    spatial_filter_normalized: String::new(),
+                    metric: "cpu_load".to_string(),
+                    num_aggregates_to_retain: None,
+                    read_count_threshold: None,
+                    table_name: None,
+                    value_column: None,
+                },
+            )]),
+        });
+        let store = Arc::new(SimpleMapStore::new(
+            streaming_config.clone(),
+            CleanupPolicy::NoCleanup,
+        ));
+        let host_a = Some(KeyByLabelValues {
+            labels: vec!["host-a".to_string()],
+        });
+        for (start, value) in [
+            (3_000, 10.0),
+            (3_000, 10.0),
+            (1_000, 100.0),
+            (4_000, 1_000.0),
+        ] {
+            store
+                .insert_precomputed_output(
+                    PrecomputedOutput::new(start, start + 3_000, host_a.clone(), 1),
+                    Box::new(SumAccumulator::with_sum(value)),
+                )
+                .unwrap();
+        }
+        let inference_config = InferenceConfig {
+            schema: SchemaConfig::PromQL(PromQLSchema::new().add_metric(
+                "cpu_load".to_string(),
+                KeyByLabelNames::new(vec!["host".to_string()]),
+            )),
+            query_configs: vec![QueryConfig::new(query.to_string())
+                .add_aggregation(AggregationReference::new(1, None))],
+            cleanup_policy: CleanupPolicy::NoCleanup,
+        };
+        let engine = SimpleEngine::new(
+            store,
+            inference_config,
+            streaming_config,
+            1_000,
+            QueryLanguage::promql,
+        );
+
+        let result = engine
+            .handle_range_query_promql(query.to_string(), 6.0, 7.0, 1.0)
+            .expect("native query execution should not fail")
+            .expect("the step at 7s has a complete cover");
+
+        assert_eq!(
+            host_a_samples(&matrix_values(result.1)),
+            vec![(7_000, 1_100.0)]
+        );
+    }
+
+    /// The same cover rule applies to per-step keys: worker partials at one
+    /// window start cannot make a missing key window look complete.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn sliding_key_cover_with_duplicate_bucket_and_missing_window_is_skipped() {
+        let query = "sum by (host) (count_over_time(event_frequency[6s]))";
+        let config = |aggregation_id, aggregation_type, aggregated_labels| AggregationConfig {
+            aggregation_id,
+            aggregation_type,
+            aggregation_sub_type: String::new(),
+            parameters: HashMap::new(),
+            grouping_labels: KeyByLabelNames::empty(),
+            aggregated_labels,
+            rollup_labels: KeyByLabelNames::empty(),
+            original_yaml: String::new(),
+            window_size_ms: 3_000,
+            slide_interval_ms: 1_000,
+            window_type: WindowType::Sliding,
+            spatial_filter: String::new(),
+            spatial_filter_normalized: String::new(),
+            metric: "event_frequency".to_string(),
+            num_aggregates_to_retain: None,
+            read_count_threshold: None,
+            table_name: None,
+            value_column: None,
+        };
+        let streaming_config = Arc::new(StreamingConfig {
+            aggregation_configs: HashMap::from([
+                (
+                    1,
+                    config(1, AggregationType::CountMinSketch, KeyByLabelNames::empty()),
+                ),
+                (
+                    2,
+                    config(
+                        2,
+                        AggregationType::SetAggregator,
+                        KeyByLabelNames::new(vec!["host".to_string()]),
+                    ),
+                ),
+            ]),
+        });
+        let store = Arc::new(SimpleMapStore::new(
+            streaming_config.clone(),
+            CleanupPolicy::NoCleanup,
+        ));
+        let host_a = KeyByLabelValues {
+            labels: vec!["host-a".to_string()],
+        };
+
+        // Values fully cover both steps. Keys have two worker partials at
+        // [3s, 6s), no [0s, 3s) bucket, then complete coverage at 7s.
+        for start in [0, 1_000, 3_000, 4_000] {
+            let mut values = CountMinSketchAccumulator::new(4, 128);
+            values.inner.update(&host_a.to_semicolon_str(), 1.0);
+            store
+                .insert_precomputed_output(
+                    PrecomputedOutput::new(start, start + 3_000, None, 1),
+                    Box::new(values),
+                )
+                .unwrap();
+        }
+        for start in [3_000, 3_000, 1_000, 4_000] {
+            let mut keys = SetAggregatorAccumulator::new();
+            keys.add_key(host_a.clone());
+            store
+                .insert_precomputed_output(
+                    PrecomputedOutput::new(start, start + 3_000, None, 2),
+                    Box::new(keys),
+                )
+                .unwrap();
+        }
+
+        let inference_config = InferenceConfig {
+            schema: SchemaConfig::PromQL(PromQLSchema::new().add_metric(
+                "event_frequency".to_string(),
+                KeyByLabelNames::new(vec!["host".to_string()]),
+            )),
+            query_configs: vec![QueryConfig::new(query.to_string())
+                .add_aggregation(AggregationReference::new(1, None))
+                .add_aggregation(AggregationReference::new(2, None))],
+            cleanup_policy: CleanupPolicy::NoCleanup,
+        };
+        let engine = SimpleEngine::new(
+            store,
+            inference_config,
+            streaming_config,
+            1_000,
+            QueryLanguage::promql,
+        );
+
+        let result = engine
+            .handle_range_query_promql(query.to_string(), 6.0, 7.0, 1.0)
+            .expect("native query execution should not fail")
+            .expect("the step at 7s has a complete key cover");
+
+        assert_eq!(host_a_samples(&matrix_values(result.1)), vec![(7_000, 2.0)]);
+    }
+
     // ════════════════════════════════════════════════════════════════════
     // 2. Tumbling: instant and range paths must AGREE, and both must
     //    correctly SUM every disjoint bucket in the query's window (proving