From ae2e4960dcc6b7b8242a26c15f1f709341f89477 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 22:43:16 -0400 Subject: [PATCH 1/2] 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 --- .../src/engines/simple_engine/mod.rs | 52 +++++++++--- .../window_semantics_consistency_tests.rs | 80 +++++++++++++++++++ 2 files changed, 119 insertions(+), 13 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index c0bb98bd..ac14419d 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -2444,6 +2444,21 @@ impl SimpleEngine { .collect() } + /// Number of distinct stored windows in a step's Sliding cover. A window + /// can hold several buckets (one per worker that saw the key), so the + /// bucket count can't tell a complete cover from a partial one. + fn sliding_cover_window_count( + bucket_map: &BucketMap, + window_start: u64, + window_end: u64, + stored_window_size_ms: u64, + ) -> usize { + (window_start..window_end) + .step_by(stored_window_size_ms as usize) + .filter(|t| bucket_map.get(t).is_some_and(|buckets| !buckets.is_empty())) + .count() + } + /// 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 @@ -2987,6 +3002,23 @@ impl SimpleEngine { }; let keys_window_start = keys_window_end.saturating_sub(*keys_lookback_ms); + if *keys_window_type == WindowType::Sliding { + let expected = + (*keys_lookback_ms / *keys_stored_window_size_ms) as usize; + let covered = Self::sliding_cover_window_count( + keys_bucket_map, + keys_window_start, + keys_window_end, + *keys_stored_window_size_ms, + ); + if covered < expected { + debug!( + "Skipping incomplete Sliding key cover at t={}", + current_time + ); + continue; + } + } Self::window_buckets_for_step( keys_bucket_map, keys_window_start, @@ -2997,18 +3029,6 @@ impl SimpleEngine { ) }; - 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", @@ -3065,7 +3085,13 @@ impl SimpleEngine { if window_type == WindowType::Sliding { let expected = (lookback_ms / context.window_size_ms) as usize; - if window_buckets.len() < expected { + let covered = Self::sliding_cover_window_count( + bucket_map, + window_start, + window_end, + context.window_size_ms, + ); + if covered < expected { debug!( "Skipping incomplete Sliding value cover at t={}", current_time diff --git a/asap-query-engine/src/tests/window_semantics_consistency_tests.rs b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs index 93e06013..e0307f1f 100644 --- a/asap-query-engine/src/tests/window_semantics_consistency_tests.rs +++ b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs @@ -331,6 +331,86 @@ 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)] + ); + } + // ════════════════════════════════════════════════════════════════════ // 2. Tumbling: instant and range paths must AGREE, and both must // correctly SUM every disjoint bucket in the query's window (proving From e2e04773292f2420d07f5bae709c6b3e70d9d0f9 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 23:30:31 -0400 Subject: [PATCH 2/2] fix(query-engine): check Sliding covers before cloning --- .../src/engines/simple_engine/mod.rs | 153 ++++++++---------- .../window_semantics_consistency_tests.rs | 98 +++++++++++ 2 files changed, 161 insertions(+), 90 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index ac14419d..b2e2e14d 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/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,12 +2354,15 @@ 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> { + window_type: WindowType, + stored_window_size_ms: u64, + expected_sliding_window_count: Option, + ) -> Option>> { // 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 @@ -2369,19 +2370,43 @@ impl SimpleEngine { // 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> = 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,45 +2469,6 @@ impl SimpleEngine { .collect() } - /// Number of distinct stored windows in a step's Sliding cover. A window - /// can hold several buckets (one per worker that saw the key), so the - /// bucket count can't tell a complete cover from a partial one. - fn sliding_cover_window_count( - bucket_map: &BucketMap, - window_start: u64, - window_end: u64, - stored_window_size_ms: u64, - ) -> usize { - (window_start..window_end) - .step_by(stored_window_size_ms as usize) - .filter(|t| bucket_map.get(t).is_some_and(|buckets| !buckets.is_empty())) - .count() - } - - /// 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> { - 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 @@ -3002,31 +2988,26 @@ impl SimpleEngine { }; let keys_window_start = keys_window_end.saturating_sub(*keys_lookback_ms); - if *keys_window_type == WindowType::Sliding { - let expected = - (*keys_lookback_ms / *keys_stored_window_size_ms) as usize; - let covered = Self::sliding_cover_window_count( - keys_bucket_map, - keys_window_start, - keys_window_end, - *keys_stored_window_size_ms, + let expected_sliding_window_count = + (*keys_window_type == WindowType::Sliding).then_some( + (*keys_lookback_ms / *keys_stored_window_size_ms) as usize, ); - if covered < expected { - debug!( - "Skipping incomplete Sliding key cover at t={}", - current_time - ); - continue; - } - } - Self::window_buckets_for_step( + 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_buckets.is_empty() { @@ -3063,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, @@ -3083,23 +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; - let covered = Self::sliding_cover_window_count( - bucket_map, - window_start, - window_end, - context.window_size_ms, - ); - if covered < 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!( diff --git a/asap-query-engine/src/tests/window_semantics_consistency_tests.rs b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs index e0307f1f..8df32eb3 100644 --- a/asap-query-engine/src/tests/window_semantics_consistency_tests.rs +++ b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs @@ -411,6 +411,104 @@ mod tests { ); } + /// 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