From b7b440eb7ae5d3ac6e374d955e0dfa4974ed4e22 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 24 Aug 2026 21:54:32 -0400 Subject: [PATCH] fix(query-engine): range query bucket scan steps by slide_interval_ms, not window_size_ms (#600) finish_range_context derived the range-query bucket-scan step from window_size_ms for both the value and keys sides, but precompute buckets are always persisted on the slide_interval_ms grid (window_manager.rs's panes_for_window). Tumbling windows set the two equal, masking this; Sliding windows with slide_interval_ms < window_size_ms had real buckets fall on timestamps the scan never visited and got silently dropped from the merge. Fixes #600. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/promql.rs | 23 +- .../src/tests/native_range_query_tests.rs | 218 ++++++++++++++++++ 2 files changed, 239 insertions(+), 2 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 10e072e..41a859a 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -557,6 +557,25 @@ impl SimpleEngine { lookback_ms } + /// The bucket-map grid width for scanning an aggregation's stored + /// buckets in a range query: `slide_interval_ms`, not `window_size_ms`. + /// `precompute_engine/window_manager.rs` persists buckets on the + /// `slide_interval_ms` grid unconditionally (its `panes_for_window` + /// steps by `slide_interval_ms`, regardless of `WindowType`) — for + /// Tumbling aggregations the two are equal by construction, so this is + /// a no-op there, but for Sliding aggregations with + /// `slide_interval_ms < window_size_ms`, stepping by `window_size_ms` + /// walks straight past real buckets and silently drops them (#600). + /// Mirrors `WindowManager::new`'s `slide_interval_ms == 0` fallback so a + /// config that leaves the field unset is still treated as Tumbling. + fn bucket_step_ms(config: &asap_types::AggregationConfig) -> u64 { + if config.slide_interval_ms == 0 { + config.window_size_ms + } else { + config.slide_interval_ms + } + } + /// Extends an instant `QueryExecutionContext` into a `RangeQueryExecutionContext`: /// computes the lookback window from the aggregation's tumbling window size, /// validates the range params, and widens the store plan to cover @@ -581,7 +600,7 @@ impl SimpleEngine { .read() .unwrap() .get_aggregation_config(base_context.agg_info.aggregation_id_for_value) - .map(|c| c.window_size_ms)?; + .map(Self::bucket_step_ms)?; self.validate_range_query_params(start_ms, end_ms, step_ms, tumbling_window_ms) .map_err(|e| { @@ -617,7 +636,7 @@ impl SimpleEngine { .read() .unwrap() .get_aggregation_config(base_context.agg_info.aggregation_id_for_key) - .map(|c| c.window_size_ms)?, + .map(Self::bucket_step_ms)?, ), None => None, }; 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 70899fa..3f72ad9 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -254,6 +254,178 @@ mod tests { ) } + /// Same dual-population shape as `create_range_engine_dual_input_with_windows`, + /// but the KEY aggregation is a Sliding window with + /// `key_slide_interval_ms < key_window_size_ms` (#600). Real Sliding + /// buckets are persisted on the slide_interval_ms grid, not the + /// window_size_ms grid (`precompute_engine/window_manager.rs`), so the + /// keys bucket span here is `key_slide_interval_ms`, not + /// `key_window_size_ms` -- unlike the value side, which stays Tumbling + /// (span == window) exactly as `create_range_engine_dual_input_with_windows` + /// already does. + #[allow(clippy::too_many_arguments)] + fn create_range_engine_dual_input_sliding_keys( + metric: &str, + value_agg_type: AggregationType, + key_agg_type: AggregationType, + grouping_labels: Vec<&str>, + aggregated_labels: Vec<&str>, + value_data: TimeSeriesData, + keys_data: TimeSeriesData, + promql_query: &str, + value_window_ms: u64, + key_window_size_ms: u64, + key_slide_interval_ms: u64, + ) -> SimpleEngine { + let grouping_label_strings: Vec = + grouping_labels.iter().map(|s| s.to_string()).collect(); + let aggregated_label_strings: Vec = + aggregated_labels.iter().map(|s| s.to_string()).collect(); + let all_labels: Vec = grouping_label_strings + .iter() + .chain(aggregated_label_strings.iter()) + .cloned() + .collect(); + + let mut aggregation_configs = HashMap::new(); + aggregation_configs.insert( + 1u64, + AggregationConfig { + aggregation_id: 1, + aggregation_type: value_agg_type, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_label_strings.clone()), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: value_window_ms, + slide_interval_ms: value_window_ms, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: metric.to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + aggregation_configs.insert( + 2u64, + AggregationConfig { + aggregation_id: 2, + aggregation_type: key_agg_type, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_label_strings), + aggregated_labels: KeyByLabelNames::new(aggregated_label_strings), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: key_window_size_ms, + slide_interval_ms: key_slide_interval_ms, + window_type: WindowType::Sliding, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: metric.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, + }); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + for (agg_id, bucket_span_ms, data) in [ + (1u64, value_window_ms, value_data), + (2u64, key_slide_interval_ms, keys_data), + ] { + for (timestamp, label_values_opt, acc) in data { + let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); + let output = + PrecomputedOutput::new(timestamp - bucket_span_ms, timestamp, key, agg_id); + store.insert_precomputed_output(output, acc).unwrap(); + } + } + + let promql_schema = + PromQLSchema::new().add_metric(metric.to_string(), KeyByLabelNames::new(all_labels)); + + let query_config = QueryConfig::new(promql_query.to_string()) + .add_aggregation(AggregationReference::new(1, None)) + .add_aggregation(AggregationReference::new(2, None)); + + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(promql_schema), + query_configs: vec![query_config], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + + SimpleEngine::new( + store, + inference_config, + streaming_config, + WINDOW_MS, + QueryLanguage::promql, + ) + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_sliding_keys_bucket_found_on_slide_interval_grid() { + // #600: the keys-side scan_window must step by the KEY aggregation's + // own slide_interval_ms, not its window_size_ms. Key aggregation: + // window_size_ms=2000, slide_interval_ms=1000 (Sliding) -- real + // buckets land on the 1000ms grid (start=1000), which isn't on the + // 2000ms window_size_ms grid ({0, 2000, 4000, ...}) at all. A scan + // that steps by window_size_ms never visits t=1000 and silently + // drops host-a's key. + let mut keys_add = SetAggregatorAccumulator::new(); + keys_add.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input_sliding_keys( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![( + 2000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + )], + // Keys bucket spans [1000, 2000) -- on the slide_interval_ms=1000 + // grid, but not the window_size_ms=2000 grid. + vec![(2000, None, Box::new(keys_add) as Box)], + "count(event_frequency) by (host, event)", + 1000, // value_window_ms (Tumbling, unaffected by #600) + 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(), 2.0, 2.5, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert!( + key_has_sample_at(&elements, "host-a", 2000), + "BUG #600: host-a's keys delta bucket (start=1000, on the \ + slide_interval_ms=1000 grid but not the window_size_ms=2000 \ + grid) was not found -- the keys-side scan_window is stepping \ + by window_size_ms instead of slide_interval_ms" + ); + } + #[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, @@ -525,6 +697,52 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread")] + async fn range_query_sliding_window_slide_lt_size_merges_bucket_off_window_size_grid() { + // #600: value-side counterpart of + // range_query_sliding_keys_bucket_found_on_slide_interval_grid. + // window_size_ms=2000, but create_engine_multi_timestamp_with_window + // fixes the bucket span at slide_interval_ms=1000, so the two + // buckets below land at start=0 and start=1000 -- only one of which + // is on the window_size_ms=2000 grid ({0, 2000, ...}). A scan_window + // that steps by window_size_ms instead of slide_interval_ms never + // visits start=1000 and silently drops that bucket from the merge. + let data = vec![ + ( + 1000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(10.0)) as Box, + ), + ( + 2000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(5.0)) as Box, + ), + ]; + let query = "sum_over_time(http_requests[1s])"; + let engine = create_engine_multi_timestamp_with_window( + "http_requests", + AggregationType::Sum, + vec!["host"], + data, + query, + 2000, // window_size_ms + WindowType::Sliding, + ); + + let result = engine.handle_range_query_promql(query.to_string(), 2.0, 2.5, 2.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + assert_eq!(elements.len(), 1, "expected one series for host-a"); + assert!( + (elements[0].samples[0].value - 15.0).abs() < 1e-9, + "BUG #600: expected both buckets (10.0 + 5.0 = 15.0) merged, got {} \ + -- tumbling_window_ms is stepping by window_size_ms=2000 instead \ + of slide_interval_ms=1000, so the bucket at start=1000 is dropped", + elements[0].samples[0].value + ); + } + #[tokio::test(flavor = "multi_thread")] async fn range_query_dual_population_expands_keys_across_multiple_steps() { // Extends range_query_dual_population_returns_key_expansion across