From 8336e9b29fb65785867214e42bcaf1d9f4e0f168 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 20 Aug 2026 17:46:32 -0400 Subject: [PATCH 1/3] refactor(query-engine): extract range-query walk/merge into a shared helper execute_range_query_pipeline's per-step "walk by window_size_ms, exact- timestamp lookup, merge found buckets, tolerate misses" logic is pulled into merge_window_at_timestamp, called once per step in place of the duplicated inline code. Pure extraction, verified as a no-op via the existing end-to-end range-query arithmetic tests (handle_range_query_promql -> execute_range_query_pipeline) plus 4 new direct unit tests, including one asserting the walk ignores denser intermediate buckets rather than merging everything in range. This is checkpoint 1 of PR B (design doc's #557 stack): the helper is not wired to anything new yet, so Tumbling behavior is unchanged and Sliding is still unreachable in production. Next: wire a Sliding branch into create_store_query_plan/execute_and_merge_store_queries so instant queries reuse this same helper (single-step case), plus the alignment fix. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 230 +++++++++++++----- 1 file changed, 174 insertions(+), 56 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 481a32d..6bae1d1 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1438,13 +1438,51 @@ impl SimpleEngine { // Some((output_labels, QueryResult::matrix(range_elements))) // } + /// Extracted from `execute_range_query_pipeline`'s per-step loop + /// (unchanged logic) so it can also be called once, not per-step, from + /// Sliding's instant-query path (#557 design doc §2). Walking by `W` + /// against `S`-grid keys only ever touches the non-overlapping + /// `W`-strided subset of the store's denser `S`-spaced buckets, since + /// `S | W` always holds for a valid sliding config — that's what makes + /// merging them safe without separate overlap accounting. + /// `Ok(None)` means no buckets were found at all (distinct from `Err`, + /// which means buckets were found but merging them failed) — callers + /// log these two cases differently. + fn merge_window_at_timestamp( + bucket_map: &HashMap, + current_time: u64, + lookback_ms: u64, + window_size_ms: u64, + accumulator_type: AggregationType, + ) -> Result>, String> { + use crate::engines::window_merger::create_window_merger; + + let window_start = current_time.saturating_sub(lookback_ms); + let mut window_buckets: Vec> = Vec::new(); + + let mut t = window_start; + while t < current_time { + if let Some(bucket) = bucket_map.get(&t) { + window_buckets.push((*bucket).clone_boxed_core()); + } + t += window_size_ms; + } + + if window_buckets.is_empty() { + return Ok(None); + } + + let mut merger = create_window_merger(accumulator_type); + merger.initialize(window_buckets); + merger.get_merged().map(Some) + } + /// Execute the range query pipeline fn execute_range_query_pipeline( &self, context: &RangeQueryExecutionContext, ) -> Result, String> { use crate::engines::query_result::RangeVectorElement; - use crate::engines::window_merger::create_window_merger; // Step 1: Fetch all data needed for the entire range let all_data = self.execute_store_query(&context.base.store_plan.values_query)?; @@ -1523,64 +1561,49 @@ impl SimpleEngine { // Iterate by OUTPUT timestamp, not by bucket index let mut current_time = start_ms; while current_time <= end_ms { - // Window covers [current_time - lookback_ms, current_time) - // 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 mut window_buckets: Vec> = Vec::new(); - - let mut t = window_start; - while t < current_time { - if let Some(bucket) = bucket_map.get(&t) { - window_buckets.push((*bucket).clone_boxed_core()); - } - // If bucket missing at timestamp t, just skip it (partial data is okay) - t += tumbling_window_ms; - } - - if !window_buckets.is_empty() { - // Merge available buckets - let mut merger = create_window_merger(*accumulator_type); - merger.initialize(window_buckets); - - match merger.get_merged() { - Ok(merged) => { - // Query statistic and emit sample at current_time - match self.query_precompute_for_statistic( - merged.as_ref(), - &context.base.metadata.statistic_to_compute, - &Some(key.clone()), - &context.base.metadata.query_kwargs, - ) { - Ok(value) => { - debug!( - "Key {:?}: emitting sample (t={}, value={})", - key, current_time, value - ); - element.add_sample(current_time, value); - } - Err(e) => { - debug!( - "Failed to query statistic at t={} for key {:?}: {}", - current_time, key, e - ); - } + match Self::merge_window_at_timestamp( + &bucket_map, + current_time, + lookback_ms, + tumbling_window_ms, + *accumulator_type, + ) { + Ok(Some(merged)) => { + // Query statistic and emit sample at current_time + match self.query_precompute_for_statistic( + merged.as_ref(), + &context.base.metadata.statistic_to_compute, + &Some(key.clone()), + &context.base.metadata.query_kwargs, + ) { + Ok(value) => { + debug!( + "Key {:?}: emitting sample (t={}, value={})", + key, current_time, value + ); + element.add_sample(current_time, value); + } + Err(e) => { + debug!( + "Failed to query statistic at t={} for key {:?}: {}", + current_time, key, e + ); } - } - Err(e) => { - debug!( - "Failed to get merged result at t={} for key {:?}: {}", - current_time, key, e - ); } } - } else { - // No data at all for this window - skip sample - debug!( - "Key {:?}: skipping sample at {} - no data in window [{}, {})", - key, current_time, window_start, current_time - ); + Ok(None) => { + // No data at all for this window - skip sample + debug!( + "Key {:?}: skipping sample at {} - no data in window", + key, current_time + ); + } + Err(e) => { + debug!( + "Failed to get merged result at t={} for key {:?}: {}", + current_time, key, e + ); + } } current_time += step_ms; @@ -1790,6 +1813,101 @@ mod range_query_tests { } } + mod merge_window_at_timestamp_tests { + use super::*; + use crate::engines::simple_engine::SimpleEngine; + use std::collections::HashMap; + + /// Builds owned `(start_timestamp, bucket)` pairs; the caller turns + /// these into a `bucket_map` (a separate step, since the map borrows + /// from these and both must live in the caller's own scope). + fn owned_buckets(entries: &[(u64, f64)]) -> Vec<(u64, Box)> { + entries + .iter() + .map(|(ts, v)| { + ( + *ts, + Box::new(MockBucketAccumulator::new(*ts, *v)) as Box, + ) + }) + .collect() + } + + fn bucket_map(owned: &[(u64, Box)]) -> HashMap { + owned.iter().map(|(ts, b)| (*ts, b.as_ref())).collect() + } + + fn merged_value(result: &Option>) -> f64 { + result + .as_ref() + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .value + } + + #[test] + fn single_bucket_exact_match_k1() { + // W=300_000, one bucket at exactly current_time - W. + let owned = owned_buckets(&[(0, 10.0)]); + let map = bucket_map(&owned); + let result = SimpleEngine::merge_window_at_timestamp( + &map, + 300_000, + 300_000, + 300_000, + AggregationType::Sum, + ) + .unwrap(); + assert_eq!(merged_value(&result), 10.0); + } + + #[test] + fn strided_merge_ignores_denser_intermediate_buckets() { + // W=300_000, S=60_000. Store holds every 60_000 slot (S-grid), + // but only the 300_000-strided ones (0, 300_000, 600_000) should + // be picked up for a k=3 query ending at 900_000. If this test + // ever starts asserting 150.0 (sum of all 15 S-spaced buckets) + // instead of 30.0, the walk stopped ignoring the denser grid and + // is double-counting. + let entries: Vec<(u64, f64)> = (0..15).map(|i| (i * 60_000, 10.0)).collect(); + let owned = owned_buckets(&entries); + let map = bucket_map(&owned); + + let result = SimpleEngine::merge_window_at_timestamp( + &map, + 900_000, + 900_000, + 300_000, + AggregationType::Sum, + ) + .unwrap(); + assert_eq!(merged_value(&result), 30.0); // 3 strided buckets, not 15 + } + + #[test] + fn tolerates_missing_buckets_within_the_window() { + // W=100. Positions 0, 100, 200 expected; 100 is missing. + let owned = owned_buckets(&[(0, 1.0), (200, 4.0)]); + let map = bucket_map(&owned); + let result = + SimpleEngine::merge_window_at_timestamp(&map, 300, 300, 100, AggregationType::Sum) + .unwrap(); + assert_eq!(merged_value(&result), 5.0); // 1.0 + 4.0, 100 silently skipped + } + + #[test] + fn no_buckets_found_returns_ok_none() { + let owned = owned_buckets(&[(999_999, 1.0)]); // nowhere near the window + let map = bucket_map(&owned); + let result = + SimpleEngine::merge_window_at_timestamp(&map, 300, 300, 100, AggregationType::Sum) + .unwrap(); + assert!(result.is_none()); + } + } + /// Simulates the sliding window loop from execute_range_query_pipeline /// Returns: Vec of (timestamp, merged_value, max_bucket_id_in_window) fn simulate_sliding_window( From 0021b93587e2306c516bf3f34d099ca4f306dd7b Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 20 Aug 2026 20:47:20 -0400 Subject: [PATCH 2/3] feat(query-engine): serve sliding-window instant queries via the merge walk create_store_query_plan and execute_and_merge_store_queries now branch on WindowType instead of the (now-always-false) is_exact_query flag. Sliding does a plain range fetch over [end - range_ms, end) -- same primitive Tumbling already uses -- with the end timestamp floor-aligned to the aggregation's slide_interval_ms grid first (align_to_slide_interval), then merges via merge_window_at_timestamp (checkpoint 1's extracted helper), called once instead of per-step. query_precomputed_output_exact / is_exact_query: true are now unreachable (tracked in #559 for removal). Tumbling's existing merge_precomputed_outputs path is untouched. 9 new end-to-end tests (tests/sliding_window_execution_tests.rs) drive a hand-wired Sliding AggregationConfig through the real handle_query_promql path (query_configs bypasses capability matching, which still has the strict pre-#557-PR-D Sliding rule) -- k=1,2,3,6 merge correctness (each asserting the stride-selected sum, not the sum of every S-spaced bucket in range), the alignment fix, and 4 negative cases (empty store, no data near the window, partial data tolerance, and all-strided-positions-missing resolving to zero series rather than None). Checkpoint 2 of PR B (design doc's #557 stack). Instant queries only -- the range-query side (aligning start once in promql.rs's range-context builders) is still to come. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 131 ++++++--- asap-query-engine/src/tests/mod.rs | 1 + .../tests/sliding_window_execution_tests.rs | 267 ++++++++++++++++++ 3 files changed, 359 insertions(+), 40 deletions(-) create mode 100644 asap-query-engine/src/tests/sliding_window_execution_tests.rs diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 6bae1d1..610ca68 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -390,6 +390,24 @@ impl SimpleEngine { }) } + /// Floor-aligns `timestamp` down to the nearest `slide_interval_ms` + /// boundary from Unix epoch 0 -- the grid sliding-window worker buckets + /// sit on (`precompute_engine/window_manager.rs`). Silent + warn on + /// misalignment, same precedent as `align_end_timestamp_promql`'s + /// ingestion-interval alignment. + fn align_to_slide_interval(timestamp: u64, slide_interval_ms: u64) -> u64 { + if timestamp.is_multiple_of(slide_interval_ms) { + return timestamp; + } + let aligned = (timestamp / slide_interval_ms) * slide_interval_ms; + warn!( + "Sliding-window query timestamp {} is not aligned with slide interval of {} ms; \ + aligning down to {}.", + timestamp, slide_interval_ms, aligned + ); + aligned + } + /// Creates a plan for querying the store based on aggregation configuration. /// Also derives `do_merge`: true when the requested time range spans more /// than one stored window, i.e. `range_ms > window_size_ms`. @@ -411,19 +429,23 @@ impl SimpleEngine { })?; let window_type = aggregation_config_for_value.window_type; - let is_exact_query = window_type == WindowType::Sliding; let range_ms = timestamps.end_timestamp - timestamps.start_timestamp; let do_merge = range_ms > aggregation_config_for_value.window_size_ms; - // Determine start/end for values query based on window type - let (values_start, values_end) = if is_exact_query { - // Sliding window: exact window match - let exact_start = - timestamps.end_timestamp - aggregation_config_for_value.window_size_ms; - (exact_start, timestamps.end_timestamp) - } else { - // Tumbling window: range query - (timestamps.start_timestamp, timestamps.end_timestamp) + // Determine start/end for values query based on window type. Sliding + // floor-aligns its end timestamp to the worker's slide_interval_ms + // grid first (design doc §4, #557) -- store lookups are exact- + // timestamp keyed, and misalignment silently looks like missing + // data rather than erroring. + let (values_start, values_end) = match window_type { + WindowType::Sliding => { + let aligned_end = Self::align_to_slide_interval( + timestamps.end_timestamp, + aggregation_config_for_value.slide_interval_ms, + ); + (aligned_end.saturating_sub(range_ms), aligned_end) + } + WindowType::Tumbling => (timestamps.start_timestamp, timestamps.end_timestamp), }; let values_query = StoreQueryParams { @@ -431,7 +453,7 @@ impl SimpleEngine { aggregation_id: agg_info.aggregation_id_for_value, start_timestamp: values_start, end_timestamp: values_end, - is_exact_query, + is_exact_query: false, }; // Determine if we need a separate keys query @@ -542,44 +564,73 @@ impl SimpleEngine { debug!("Store query returned {} unique keys", values_map.len()); let merge_start_time = Instant::now(); - let window_type = if plan.values_query.is_exact_query { - WindowType::Sliding - } else { - WindowType::Tumbling + let (window_type, window_size_ms) = { + let sc = self.streaming_config.read().unwrap(); + let config = sc + .get_aggregation_config(agg_info.aggregation_id_for_value) + .ok_or_else(|| { + format!( + "Aggregation config not found for aggregation_id: {}", + agg_info.aggregation_id_for_value + ) + })?; + (config.window_type, config.window_size_ms) }; - let merged_values = if plan.values_query.is_exact_query { - // Sliding window: no merge needed, extract buckets from timestamped data - debug!("Sliding window mode: Skipping merge (expecting 1 precompute per key)"); - values_map - .into_iter() - .map(|(key, timestamped_buckets)| { - if timestamped_buckets.len() != 1 { - warn!( - "Sliding window expected 1 precompute per key, found {}. Using first.", - timestamped_buckets.len() - ); + let merged_values = match window_type { + WindowType::Sliding => { + // Reuses the range query's stride-window_size_ms walk, + // called once instead of per-step (design doc §2/§3, #557). + // Keys with no data in the window, or where merging failed, + // are skipped rather than failing the whole query -- same + // tolerance merge_precomputed_outputs already has for keys + // with an empty bucket list. + let lookback_ms = + plan.values_query.end_timestamp - plan.values_query.start_timestamp; + let mut merged = HashMap::with_capacity(values_map.len()); + for (key, timestamped_buckets) in &values_map { + let bucket_map: HashMap = timestamped_buckets + .iter() + .map(|((start, _), bucket)| (*start, bucket.as_ref())) + .collect(); + match Self::merge_window_at_timestamp( + &bucket_map, + plan.values_query.end_timestamp, + lookback_ms, + window_size_ms, + agg_info.aggregation_type_for_value, + ) { + Ok(Some(merged_bucket)) => { + merged.insert(key.clone(), merged_bucket); + } + Ok(None) => { + debug!("Sliding window: no data in window for key {:?}", key); + } + Err(e) => { + warn!( + "Sliding window: failed to merge window for key {:?}: {}", + key, e + ); + } } - // Extract bucket from timestamped tuple - let (_, bucket) = timestamped_buckets.into_iter().next().unwrap(); - (key, bucket.as_ref().clone_boxed_core()) - }) - .collect() - } else { - // Tumbling window: merge needed - debug!("Tumbling window mode: Merging {} outputs", values_map.len()); - self.merge_precomputed_outputs( - &values_map, - do_merge, - agg_info.aggregation_type_for_value, - ) + } + merged + } + WindowType::Tumbling => { + debug!("Tumbling window mode: Merging {} outputs", values_map.len()); + self.merge_precomputed_outputs( + &values_map, + do_merge, + agg_info.aggregation_type_for_value, + ) + } }; let merge_duration = merge_start_time.elapsed(); debug!( "[LATENCY] Precomputed output processing ({}): {:.2}ms, resulted in {} merged outputs", if window_type == WindowType::Sliding { - "no merge" + "sliding walk" } else { "merge" }, diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs index 9fbe6fc..b84099f 100644 --- a/asap-query-engine/src/tests/mod.rs +++ b/asap-query-engine/src/tests/mod.rs @@ -5,6 +5,7 @@ pub mod elastic_dsl_query_tests; pub mod elastic_forwarding_tests; pub mod prometheus_forwarding_tests; pub mod query_equivalence_tests; +pub mod sliding_window_execution_tests; pub mod sql_pattern_matching_tests; pub mod store_correctness_tests; pub mod trait_design_tests; diff --git a/asap-query-engine/src/tests/sliding_window_execution_tests.rs b/asap-query-engine/src/tests/sliding_window_execution_tests.rs new file mode 100644 index 0000000..9af517a --- /dev/null +++ b/asap-query-engine/src/tests/sliding_window_execution_tests.rs @@ -0,0 +1,267 @@ +//! End-to-end correctness tests for sliding-window instant-query execution +//! (issue #557). `should_use_sliding_window()` is hardcoded `false`, so the +//! live planner can't emit a `Sliding` config yet — these hand-construct one +//! directly and register it via `query_configs` (which `promql.rs` checks +//! before falling back to capability matching), the same bypass +//! `capability_matching_tests.rs` uses, so `window_compatible`'s still-strict +//! Sliding rule (unrelaxed until #557's PR D) never gets in the way. +//! +//! Every bucket carries the same single grouping-label key (`pod="a"`) +//! rather than no key at all: `format_final_results` (pre-existing, +//! unrelated to #557) drops `None`-keyed entries, and every other test in +//! this codebase that asserts on actual result *values* uses a real key for +//! the same reason — a `None` key is for genuinely unresolvable results, +//! not "no grouping requested". + +use crate::data_model::{ + AggregationConfig, AggregationReference, AggregationType, CleanupPolicy, InferenceConfig, + KeyByLabelValues, PrecomputedOutput, PromQLSchema, QueryConfig, QueryLanguage, SchemaConfig, + StreamingConfig, WindowType, +}; +use crate::engines::query_result::QueryResult; +use crate::engines::simple_engine::SimpleEngine; +use crate::precompute_operators::sum_accumulator::SumAccumulator; +use crate::stores::simple_map_store::SimpleMapStore; +use crate::stores::traits::Store; +use promql_utilities::data_model::KeyByLabelNames; +use std::collections::HashMap; +use std::sync::Arc; + +const METRIC: &str = "reqs"; +const INGESTION_INTERVAL_MS: u64 = 1000; + +fn key() -> KeyByLabelValues { + KeyByLabelValues::new_with_labels(vec!["a".to_string()]) +} + +/// Builds a `SimpleEngine` with one `Sliding` aggregation (`window_size_ms`, +/// `slide_interval_ms`) and one `Sum` bucket inserted at every +/// `slide_interval_ms` from 0 up to (not including) `bucket_count * +/// slide_interval_ms`. Bucket `i`'s value is `(i + 1) as f64`, distinct per +/// bucket so a test can tell exactly which ones got merged. +fn engine_with_sliding_buckets( + window_size_ms: u64, + slide_interval_ms: u64, + bucket_count: u64, + promql_query: &str, +) -> SimpleEngine { + engine_with_sliding_buckets_missing( + window_size_ms, + slide_interval_ms, + bucket_count, + &[], + promql_query, + ) +} + +/// Same as `engine_with_sliding_buckets`, but skips inserting the buckets +/// at the given indices - for testing "partial data is okay" tolerance. +fn engine_with_sliding_buckets_missing( + window_size_ms: u64, + slide_interval_ms: u64, + bucket_count: u64, + missing: &[u64], + promql_query: &str, +) -> SimpleEngine { + let agg_config = AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(vec!["pod".to_string()]), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms, + 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 mut aggregation_configs = HashMap::new(); + aggregation_configs.insert(1u64, agg_config); + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + for i in 0..bucket_count { + if missing.contains(&i) { + continue; + } + let start = i * slide_interval_ms; + let output = PrecomputedOutput::new(start, start + window_size_ms, Some(key()), 1); + let value = (i + 1) as f64; + store + .insert_precomputed_output(output, Box::new(SumAccumulator::with_sum(value))) + .unwrap(); + } + + let promql_schema = PromQLSchema::new().add_metric( + METRIC.to_string(), + KeyByLabelNames::new(vec!["pod".to_string()]), + ); + let query_config = QueryConfig::new(promql_query.to_string()) + .add_aggregation(AggregationReference::new(1, 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, + INGESTION_INTERVAL_MS, + QueryLanguage::promql, + ) +} + +/// Runs `sum_over_time(reqs[range_seconds])` at `query_time_seconds` and +/// returns the single `pod="a"` result value. +fn query_sum(engine: &SimpleEngine, range_seconds: u64, query_time_seconds: f64) -> f64 { + let query = format!("sum_over_time({METRIC}[{range_seconds}s])"); + let (_labels, result) = engine + .handle_query_promql(query, query_time_seconds) + .expect("query should resolve via the registered query_config"); + match result { + QueryResult::Vector(vector) => { + assert_eq!(vector.values.len(), 1, "expected exactly one series"); + vector.values[0].value + } + other => panic!("expected an instant vector, got {other:?}"), + } +} + +/// For the case where query dispatch/resolution itself fails (unmatched +/// pattern, no compatible aggregation, or the store has literally nothing), +/// distinct from a resolved query that matched zero series (see +/// `query_returns_empty_vector`). +fn query_returns_none(engine: &SimpleEngine, range_seconds: u64, query_time_seconds: f64) -> bool { + let query = format!("sum_over_time({METRIC}[{range_seconds}s])"); + engine + .handle_query_promql(query, query_time_seconds) + .is_none() +} + +/// For the case where the query resolves and executes, but no data was +/// usable for it (e.g. the store has buckets, just not at any position the +/// merge walk needed): `handle_query_promql` returns `Some` with an empty +/// vector here, not `None`. +fn query_returns_empty_vector(engine: &SimpleEngine, range_seconds: u64, query_time_seconds: f64) { + let query = format!("sum_over_time({METRIC}[{range_seconds}s])"); + let (_labels, result) = engine + .handle_query_promql(query, query_time_seconds) + .expect("query should resolve (Some) even though no series matched"); + match result { + QueryResult::Vector(vector) => { + assert!( + vector.values.is_empty(), + "expected zero series, got {:?}", + vector.values + ); + } + other => panic!("expected an instant vector, got {other:?}"), + } +} + +// W=300_000ms (5m), S=60_000ms (1m). Buckets 0..15 cover [0, 900_000). +// Bucket i's start_timestamp is i*60_000 and its value is i+1. +const W: u64 = 300_000; +const S: u64 = 60_000; + +#[test] +fn k1_exact_match_uses_single_bucket() { + // range = W: query at t=300_000 (on-grid) should merge exactly bucket + // i=0 (start_timestamp=0), value 1 - the pre-#557 exact-match case, + // now served through the unified walk instead of query_precomputed_output_exact. + let engine = engine_with_sliding_buckets(W, S, 15, "sum_over_time(reqs[300s])"); + assert_eq!(query_sum(&engine, 300, 300.0), 1.0); +} + +#[test] +fn k2_merges_two_strided_buckets_not_all_intermediate_ones() { + // range = 2*W = 600_000, query at t=600_000. Strided buckets are at + // start_timestamp 0 (i=0, value 1) and 300_000 (i=5, value 6) = 7. + // Summing all 10 S-spaced buckets in [0, 600_000) would wrongly give 55. + let engine = engine_with_sliding_buckets(W, S, 15, "sum_over_time(reqs[600s])"); + assert_eq!(query_sum(&engine, 600, 600.0), 7.0); +} + +#[test] +fn k3_merges_three_strided_buckets_not_all_fifteen() { + // range = 3*W = 900_000, query at t=900_000. Strided buckets: i=0 (1), + // i=5 (6), i=10 (11) = 18. Summing all 15 would wrongly give 120. + let engine = engine_with_sliding_buckets(W, S, 15, "sum_over_time(reqs[900s])"); + assert_eq!(query_sum(&engine, 900, 900.0), 18.0); +} + +#[test] +fn k6_merges_six_strided_buckets() { + // range = 6*W = 1_800_000, query at t=1_800_000. Strided buckets: + // i=0,5,10,15,20,25 -> values 1,6,11,16,21,26 = 81. + let engine = engine_with_sliding_buckets(W, S, 30, "sum_over_time(reqs[1800s])"); + assert_eq!(query_sum(&engine, 1800, 1800.0), 81.0); +} + +#[test] +fn misaligned_query_timestamp_floor_aligns_instead_of_missing_data() { + // Query at t=305_000 (5s past the 300_000 slide-interval boundary) for + // range=W=300_000: aligned_end floor-aligns 305_000 -> 300_000, so + // window_start = 300_000 - 300_000 = 0, and the walk finds bucket i=0 + // (start_timestamp=0, value 1) - the latest complete W-wide window as + // of the aligned query time. Without the fix, aligned_end would stay + // 305_000, window_start would be 5_000, and the walk would look up + // bucket_map[5_000] (nothing there, since buckets only exist on the + // 60_000 grid) - silently resolving to zero series despite bucket i=0 + // existing. + let engine = engine_with_sliding_buckets(W, S, 15, "sum_over_time(reqs[300s])"); + assert_eq!(query_sum(&engine, 300, 305.0), 1.0); +} + +// --- Negative cases --- + +#[test] +fn no_data_anywhere_near_the_window_returns_none() { + // Buckets only exist in [0, 900_000). Querying far outside that range + // hits the store's own empty-fetch guard - the query never resolves. + let engine = engine_with_sliding_buckets(W, S, 15, "sum_over_time(reqs[300s])"); + assert!(query_returns_none(&engine, 300, 90_000.0)); +} + +#[test] +fn empty_store_returns_none() { + let engine = engine_with_sliding_buckets(W, S, 0, "sum_over_time(reqs[300s])"); + assert!(query_returns_none(&engine, 300, 300.0)); +} + +#[test] +fn partial_data_merges_only_the_buckets_that_exist() { + // k=3 query (strided positions i=0, 5, 10), but i=5 is missing from the + // store. "Partial data is okay" (design doc §2/§4): the walk should + // merge just i=0 and i=10 (1 + 11 = 12) instead of erroring outright. + let engine = engine_with_sliding_buckets_missing(W, S, 15, &[5], "sum_over_time(reqs[900s])"); + assert_eq!(query_sum(&engine, 900, 900.0), 12.0); +} + +#[test] +fn all_strided_positions_missing_resolves_to_zero_series() { + // Only intermediate S-spaced buckets exist (i=1..4), none of the + // strided positions (i=0, 5, 10) a k=3 query actually needs. The store + // has data (so this resolves, unlike the empty-store case above), but + // none of it is usable for this query - must match zero series, not a + // wrong answer built from buckets outside the stride. + let engine = + engine_with_sliding_buckets_missing(W, S, 15, &[0, 5, 10], "sum_over_time(reqs[900s])"); + query_returns_empty_vector(&engine, 900, 900.0); +} From 38db488a7e2782d1b15ac3c0ada0ded4f009075d Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 20 Aug 2026 21:54:49 -0400 Subject: [PATCH 3/3] feat(query-engine): align sliding-window range-query lookups without shifting reported timestamps promql.rs's two range-context builders now floor-align a Sliding query's lookup anchor (RangeQueryExecutionContext::aligned_start_ms) instead of the client-requested start_ms, once per query rather than per step -- PR A's step % slide_interval_ms validation is what guarantees the offset between the two stays constant across every step. execute_range_query_pipeline computes that fixed offset once and subtracts it only when calling merge_window_at_timestamp; the timestamp reported back via element.add_sample stays exactly what the client asked for. Misalignment still logs via align_to_slide_interval's existing warn! (no new logging needed -- it already fires whenever a real shift happens). An earlier version of this change shadowed start_ms directly, which also shifted every reported sample timestamp to the aligned grid -- caught before landing and reworked to decouple "what we tell the client" from "what we used internally to look up data". 2 new end-to-end range-query tests: strided-merge correctness across 4 output steps, and a misaligned-start case proving the response's timestamps match the request even though the underlying lookups are grid-aligned a few seconds earlier. Completes PR B (design doc's #557 stack) -- both the instant-query (previous commit) and range-query fetch paths now serve Sliding correctly. Still to push/open as an actual PR (base=557-sliding-window-query-engine). Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 21 +++++- .../src/engines/simple_engine/promql.rs | 39 +++++++++- .../tests/sliding_window_execution_tests.rs | 74 ++++++++++++++++++- 3 files changed, 128 insertions(+), 6 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 610ca68..f7576f6 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -106,8 +106,16 @@ pub struct RangeQueryParams { pub struct RangeQueryExecutionContext { /// Base context (metric, metadata, store_plan, etc.) pub base: QueryExecutionContext, - /// Range-specific parameters + /// Range-specific parameters. `range_params.start` is the client's + /// requested start, unmodified -- reported sample timestamps are always + /// derived from this, never from `aligned_start_ms`. pub range_params: RangeQueryParams, + /// Sliding's floor-aligned version of `range_params.start` (equal to + /// it for Tumbling). The walk uses this -- via a fixed offset computed + /// once in `execute_range_query_pipeline` -- to stay grid-aligned + /// without shifting what gets reported back to the client. See design + /// doc §4 for #557. + pub aligned_start_ms: u64, /// Number of buckets per step (step / tumbling_window) pub buckets_per_step: usize, /// Number of buckets in lookback window @@ -1560,6 +1568,14 @@ impl SimpleEngine { let buckets_per_step = context.buckets_per_step; let lookback_bucket_count = context.lookback_bucket_count; + // Fixed for the whole query: how far below the client's requested + // `start_ms` the grid-aligned lookup anchor sits (0 for Tumbling, + // or Sliding already on-grid). Subtracting this from each step's + // `current_time` keeps the walk grid-aligned without changing what + // gets reported back as each sample's timestamp -- see + // `RangeQueryExecutionContext::aligned_start_ms`. + let lookup_offset_ms = start_ms.saturating_sub(context.aligned_start_ms); + let window_mode = if buckets_per_step <= lookback_bucket_count { "sliding (slide <= size)" } else { @@ -1612,9 +1628,10 @@ impl SimpleEngine { // Iterate by OUTPUT timestamp, not by bucket index let mut current_time = start_ms; while current_time <= end_ms { + let lookup_time = current_time.saturating_sub(lookup_offset_ms); match Self::merge_window_at_timestamp( &bucket_map, - current_time, + lookup_time, lookback_ms, tumbling_window_ms, *accumulator_type, diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index d0212e8..170cb9f 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -10,6 +10,7 @@ use super::{ }; use crate::data_model::{AggregationIdInfo, KeyByLabelValues, QueryConfig, SchemaConfig}; use crate::engines::query_result::{QueryResult, RangeVectorElement}; +use asap_types::enums::WindowType; use asap_types::query_requirements::build_query_requirements_promql; use asap_types::PromQLSchema; use promql_utilities::ast_matching::PromQLMatchResult; @@ -518,6 +519,21 @@ impl SimpleEngine { }) .ok()?; + // Sliding floor-aligns the loop's lookup anchor once, here -- the + // validation above (step % slide_interval_ms == 0) is what + // guarantees every later step stays grid-aligned without + // re-aligning per-step. `start_ms` itself (what the client asked + // for) is untouched -- it still seeds `range_params.start`, so + // reported sample timestamps never shift. Only the fetch bound and + // `aligned_start_ms` (which the walk anchors on) use the aligned + // value. See design doc §4/§5 for #557. + let aligned_start_ms = match window_type { + WindowType::Sliding => { + SimpleEngine::align_to_slide_interval(start_ms, slide_interval_ms) + } + WindowType::Tumbling => start_ms, + }; + let lookback_ms = base_context.store_plan.values_query.end_timestamp - base_context.store_plan.values_query.start_timestamp; @@ -525,7 +541,8 @@ impl SimpleEngine { let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; let mut extended_store_plan = base_context.store_plan.clone(); - extended_store_plan.values_query.start_timestamp = start_ms.saturating_sub(lookback_ms); + extended_store_plan.values_query.start_timestamp = + aligned_start_ms.saturating_sub(lookback_ms); extended_store_plan.values_query.end_timestamp = end_ms; extended_store_plan.values_query.is_exact_query = false; @@ -539,6 +556,7 @@ impl SimpleEngine { end: end_ms, step: step_ms, }, + aligned_start_ms, buckets_per_step, lookback_bucket_count, tumbling_window_ms, @@ -1149,6 +1167,21 @@ impl SimpleEngine { }) .ok()?; + // Sliding floor-aligns the loop's lookup anchor once, here -- the + // validation above (step % slide_interval_ms == 0) is what + // guarantees every later step stays grid-aligned without + // re-aligning per-step. `start_ms` itself (what the client asked + // for) is untouched -- it still seeds `range_params.start`, so + // reported sample timestamps never shift. Only the fetch bound and + // `aligned_start_ms` (which the walk anchors on) use the aligned + // value. See design doc §4/§5 for #557. + let aligned_start_ms = match window_type { + WindowType::Sliding => { + SimpleEngine::align_to_slide_interval(start_ms, slide_interval_ms) + } + WindowType::Tumbling => start_ms, + }; + // Calculate lookback from the base context's store plan let lookback_ms = base_context.store_plan.values_query.end_timestamp - base_context.store_plan.values_query.start_timestamp; @@ -1158,7 +1191,8 @@ impl SimpleEngine { // Modify the store plan to cover the entire range let mut extended_store_plan = base_context.store_plan.clone(); - extended_store_plan.values_query.start_timestamp = start_ms.saturating_sub(lookback_ms); + extended_store_plan.values_query.start_timestamp = + aligned_start_ms.saturating_sub(lookback_ms); extended_store_plan.values_query.end_timestamp = end_ms; // Range queries always use range fetch, not exact extended_store_plan.values_query.is_exact_query = false; @@ -1173,6 +1207,7 @@ impl SimpleEngine { end: end_ms, step: step_ms, }, + aligned_start_ms, buckets_per_step, lookback_bucket_count, tumbling_window_ms, diff --git a/asap-query-engine/src/tests/sliding_window_execution_tests.rs b/asap-query-engine/src/tests/sliding_window_execution_tests.rs index 9af517a..1d27a5c 100644 --- a/asap-query-engine/src/tests/sliding_window_execution_tests.rs +++ b/asap-query-engine/src/tests/sliding_window_execution_tests.rs @@ -1,5 +1,5 @@ -//! End-to-end correctness tests for sliding-window instant-query execution -//! (issue #557). `should_use_sliding_window()` is hardcoded `false`, so the +//! End-to-end correctness tests for sliding-window instant- and range-query +//! execution (issue #557). `should_use_sliding_window()` is hardcoded `false`, so the //! live planner can't emit a `Sliding` config yet — these hand-construct one //! directly and register it via `query_configs` (which `promql.rs` checks //! before falling back to capability matching), the same bypass @@ -214,6 +214,76 @@ fn k6_merges_six_strided_buckets() { assert_eq!(query_sum(&engine, 1800, 1800.0), 81.0); } +// --- Range queries --- + +/// Runs `sum_over_time(reqs[range_seconds])` as a range query and returns +/// `(timestamp, value)` samples for the `pod="a"` series. +fn range_query_samples( + engine: &SimpleEngine, + range_seconds: u64, + start_seconds: f64, + end_seconds: f64, + step_seconds: f64, +) -> Vec<(u64, f64)> { + let query = format!("sum_over_time({METRIC}[{range_seconds}s])"); + let (_labels, result) = engine + .handle_range_query_promql(query, start_seconds, end_seconds, step_seconds) + .expect("range query should resolve via the registered query_config"); + match result { + QueryResult::Matrix(matrix) => { + assert_eq!(matrix.values.len(), 1, "expected exactly one series"); + matrix.values[0] + .samples + .iter() + .map(|s| (s.timestamp, s.value)) + .collect() + } + other => panic!("expected a range vector, got {other:?}"), + } +} + +#[test] +fn range_query_merges_strided_buckets_at_each_step() { + // k=3 query (range=900s=3*W), step=300s (=W, a multiple of S=60s per + // PR A's validation). Buckets 0..30 cover [0, 1_800_000). Output points + // at 900s, 1200s, 1500s, 1800s, each summing its own 3 strided buckets + // -- not the same "sum everything" answer at every step. + let engine = engine_with_sliding_buckets(W, S, 30, "sum_over_time(reqs[900s])"); + let samples = range_query_samples(&engine, 900, 900.0, 1800.0, 300.0); + assert_eq!( + samples, + vec![ + (900_000, 18.0), // i=0,5,10 -> 1+6+11 + (1_200_000, 33.0), // i=5,10,15 -> 6+11+16 + (1_500_000, 48.0), // i=10,15,20 -> 11+16+21 + (1_800_000, 63.0), // i=15,20,25 -> 16+21+26 + ] + ); +} + +#[test] +fn range_query_misaligned_start_reports_requested_timestamps_not_aligned_ones() { + // start=905s is 5s off the 60s slide grid; step=300s is a multiple of + // S so PR A's validation passes regardless of start's alignment. + // §4 keeps the *reported* timestamps exactly as requested (905, 1205, + // ...) even though the lookup anchor internally floor-aligns to 900s + // for every step (a fixed 5s offset, since step is S-aligned) - the + // data is a few seconds stale, but the response's time axis matches + // what the client asked for. Same values as the aligned test above, + // just relabeled. + let engine = engine_with_sliding_buckets(W, S, 30, "sum_over_time(reqs[900s])"); + let samples = range_query_samples(&engine, 900, 905.0, 1805.0, 300.0); + assert_eq!( + samples, + vec![ + (905_000, 18.0), + (1_205_000, 33.0), + (1_505_000, 48.0), + (1_805_000, 63.0), + ] + ); +} + #[test] fn misaligned_query_timestamp_floor_aligns_instead_of_missing_data() { // Query at t=305_000 (5s past the 300_000 slide-interval boundary) for