From 751095856313015b966bacc1790aecf6303f0ecb Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 23:46:00 -0400 Subject: [PATCH] refactor(query-engine): remove is_exact_query as a threaded bool StoreQueryParams::is_exact_query was a bool computed once and threaded through StoreQueryPlan, then consulted by execute_store_query to pick between store fetch strategies. Every StoreQueryParams builder had to independently derive it correctly, and past call sites already drifted out of sync more than once (#580, #582, #608). Since #616 replaced the tolerant-scan branch with scan_windows_via_exact (a grid-walk of exact lookups), the flag's only remaining job was choosing between that grid-walk and a single direct exact call -- but a range exactly one window wide already makes scan_windows_via_exact degenerate to a single exact lookup. So it can go away entirely: create_store_query_plan still narrows the Sliding-instant values query to one window's width (unchanged), and execute_store_query now unconditionally calls scan_windows_via_exact. The one other thing is_exact_query did -- telling execute_and_merge_store_queries whether to use Sliding or Tumbling merge semantics -- is unrelated to fetch mechanism and is now passed explicitly as a WindowType parameter, sourced from create_store_query_plan's (now three-element) return value and threaded onto QueryExecutionContext. Also adds window_semantics_consistency_tests.rs: hardening tests written against the observable PromQL query surface (not internal struct/function names), covering Sliding/Tumbling instant-vs-range agreement, keys queries over each WindowType, and window-grid boundary cases. Fixes #613. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/elastic.rs | 3 +- .../src/engines/simple_engine/mod.rs | 119 ++-- .../src/engines/simple_engine/promql.rs | 8 +- .../src/engines/simple_engine/sql.rs | 3 +- asap-query-engine/src/tests/mod.rs | 1 + .../src/tests/native_range_query_tests.rs | 11 +- .../src/tests/test_utilities/comparison.rs | 17 +- .../window_semantics_consistency_tests.rs | 638 ++++++++++++++++++ 8 files changed, 705 insertions(+), 95 deletions(-) create mode 100644 asap-query-engine/src/tests/window_semantics_consistency_tests.rs diff --git a/asap-query-engine/src/engines/simple_engine/elastic.rs b/asap-query-engine/src/engines/simple_engine/elastic.rs index c21d1eb1..b390b212 100644 --- a/asap-query-engine/src/engines/simple_engine/elastic.rs +++ b/asap-query-engine/src/engines/simple_engine/elastic.rs @@ -58,7 +58,7 @@ impl SimpleEngine { // Parse time range information from first query predicate if available, otherwise default to entire history up to query_time. let timestamps = self.resolve_query_time_range_elastic(query_time, query_info); - let (query_plan, do_merge) = self + let (query_plan, do_merge, value_window_type) = self .create_store_query_plan(&metric, ×tamps, &agg_info) .map_err(|e| { warn!("Failed to create store query plan: {}", e); @@ -82,6 +82,7 @@ impl SimpleEngine { metadata: query_metadata, store_plan: query_plan.clone(), agg_info: agg_info.clone(), + value_window_type, do_merge, spatial_filter, query_time, diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index be14d90c..fc0767d5 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -51,8 +51,6 @@ pub struct StoreQueryParams { pub start_timestamp: u64, /// Milliseconds since epoch. pub end_timestamp: u64, - /// true for sliding windows (exact match), false for tumbling (range) - pub is_exact_query: bool, } /// Complete plan for querying store (values + optional separate keys) @@ -79,6 +77,9 @@ pub struct QueryExecutionContext { pub metadata: QueryMetadata, pub store_plan: StoreQueryPlan, pub agg_info: AggregationIdInfo, + /// The value aggregation's WindowType -- Sliding fetches/merges a single + /// already-complete window; Tumbling sums the disjoint buckets in range. + pub value_window_type: WindowType, /// Whether to merge multiple precomputes (true for temporal queries) pub do_merge: bool, #[allow(dead_code)] @@ -416,24 +417,33 @@ impl SimpleEngine { } }; + // Keys always fetch via the window-grid walk (execute_store_query), + // never a single exact-window lookup -- this is an explicit, + // permanent choice, not a WindowType derivation: a keys query + // conceptually always needs to see the key's own bucket(s), not "the + // one window ending now." Ok(StoreQueryParams { metric: metric.to_string(), aggregation_id: agg_info.aggregation_id_for_key, start_timestamp, end_timestamp, - is_exact_query: false, // Keys always use range queries }) } /// 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`. + /// + /// Returns the value aggregation's `WindowType` alongside the plan -- + /// callers need it again later (e.g. to pick merge semantics) and it's + /// cheaper to hand back what was already looked up here than to + /// re-fetch the aggregation config. fn create_store_query_plan( &self, metric: &str, timestamps: &QueryTimestamps, agg_info: &AggregationIdInfo, - ) -> Result<(StoreQueryPlan, bool), String> { + ) -> Result<(StoreQueryPlan, bool, WindowType), String> { let sc = self.streaming_config.read().unwrap().clone(); // Get aggregation config for value to determine window type let aggregation_config_for_value = sc @@ -446,13 +456,15 @@ 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 + // Determine start/end for values query based on window type. For + // Sliding, narrow to exactly the one window ending "now" -- + // execute_store_query's window-grid walk degenerates to a single + // exact lookup when given a range exactly one window wide, so this + // narrowing (not a separate flag) is what makes it an "exact" fetch. + let (values_start, values_end) = if window_type == WindowType::Sliding { let exact_start = timestamps.end_timestamp - aggregation_config_for_value.window_size_ms; (exact_start, timestamps.end_timestamp) @@ -466,7 +478,6 @@ impl SimpleEngine { aggregation_id: agg_info.aggregation_id_for_value, start_timestamp: values_start, end_timestamp: values_end, - is_exact_query, }; // Determine if we need a separate keys query @@ -482,6 +493,7 @@ impl SimpleEngine { keys_query, }, do_merge, + window_type, )) } @@ -504,13 +516,14 @@ impl SimpleEngine { } } - /// Non-exact store query: walks the aggregation's window grid - /// (`bucket_step_ms` apart, each window `window_size_ms` wide, per - /// `WindowManager::window_start_for`) and looks up every grid position - /// in `[start_timestamp, end_timestamp)` with an exact match, merging - /// the sparse per-window results. Used for range queries, key queries, - /// and instant queries over tumbling windows — everywhere - /// `is_exact_query` is false. + /// Walks the aggregation's window grid (`bucket_step_ms` apart, each + /// window `window_size_ms` wide, per `WindowManager::window_start_for`) + /// and looks up every grid position in `[start_timestamp, end_timestamp)` + /// with an exact match, merging the sparse per-window results. A range + /// exactly one window wide degenerates to a single exact lookup -- an + /// instant Sliding-window fetch gets "the one window ending now" this + /// way, by being narrowed to one window's width before calling + /// (`create_store_query_plan`), not via a separate exact/scan flag. fn scan_windows_via_exact( &self, params: &StoreQueryParams, @@ -590,65 +603,20 @@ impl SimpleEngine { params: &StoreQueryParams, ) -> Result { debug!( - "Querying store: metric={}, agg_id={}, range=[{}, {}], exact={}", - params.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - params.is_exact_query + "Querying store: metric={}, agg_id={}, range=[{}, {}]", + params.metric, params.aggregation_id, params.start_timestamp, params.end_timestamp, ); let store_query_start_time = Instant::now(); - - let result = if params.is_exact_query { + let result = self.scan_windows_via_exact(params); + if let Ok(ref outputs) = result { + let store_query_duration = store_query_start_time.elapsed(); debug!( - "Sliding window query: Looking for exact window [{}, {}]", - params.start_timestamp, params.end_timestamp + "Window-grid query took: {:.2}ms, found {} unique keys", + store_query_duration.as_secs_f64() * 1000.0, + outputs.len() ); - let res = self - .store - .query_precomputed_output_exact( - ¶ms.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - ) - .map_err(|e| { - format!( - "Error querying store for metric {}, agg {}, range [{}, {}]: {}", - params.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - e - ) - }); - if let Ok(ref outputs) = res { - let store_query_duration = store_query_start_time.elapsed(); - debug!( - "Sliding window exact query took: {:.2}ms, found {} unique keys", - store_query_duration.as_secs_f64() * 1000.0, - outputs.len() - ); - } - res - } else { - debug!( - "Window-grid query: range [{}, {}]", - params.start_timestamp, params.end_timestamp - ); - let res = self.scan_windows_via_exact(params); - if let Ok(ref outputs) = res { - let store_query_duration = store_query_start_time.elapsed(); - debug!( - "Window-grid query took: {:.2}ms, found {} unique keys", - store_query_duration.as_secs_f64() * 1000.0, - outputs.len() - ); - } - res - }; - + } result } @@ -658,6 +626,7 @@ impl SimpleEngine { plan: &StoreQueryPlan, do_merge: bool, agg_info: &AggregationIdInfo, + value_window_type: WindowType, ) -> Result<(MergedOutputsMap, Option), String> { // Query and merge values let values_map = self.execute_store_query(&plan.values_query).map_err(|e| { @@ -675,13 +644,8 @@ 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 merged_values = if plan.values_query.is_exact_query { + let merged_values = if value_window_type == WindowType::Sliding { // Sliding window: expected exactly 1 precompute per key today // (ponytail: hardcoded, #554 will make >1 legitimate — don't // block on it). The store can legitimately return more than @@ -716,7 +680,7 @@ impl SimpleEngine { }; let merge_duration = merge_start_time.elapsed(); - let did_merge = window_type == WindowType::Sliding + let did_merge = value_window_type == WindowType::Sliding || do_merge || agg_info.aggregation_type_for_value == AggregationType::DeltaSetAggregator; debug!( @@ -897,6 +861,7 @@ impl SimpleEngine { &context.store_plan, context.do_merge, &context.agg_info, + context.value_window_type, )?; // Step 2: Collect results diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 0ad0e28b..8465b46f 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -360,7 +360,7 @@ impl SimpleEngine { query_kwargs, }; - let (query_plan, do_merge) = self + let (query_plan, do_merge, value_window_type) = self .create_store_query_plan(&metric, ×tamps, &agg_info) .map_err(|e| { warn!("Failed to create store query plan: {}", e); @@ -384,6 +384,7 @@ impl SimpleEngine { metadata, store_plan: query_plan, agg_info, + value_window_type, do_merge, spatial_filter, query_time, @@ -594,10 +595,13 @@ impl SimpleEngine { }) .ok()?; + // Widening the fetch range to cover the whole step span (rather than + // one window's width) is what makes this a window-grid walk instead + // of a single exact lookup -- there's no separate flag to set for + // that; it falls out of execute_store_query's range-driven behavior. let mut extended_store_plan = base_context.store_plan.clone(); let lookback_ms = Self::widen_query_window(&mut extended_store_plan.values_query, start_ms, end_ms); - extended_store_plan.values_query.is_exact_query = false; let buckets_per_step = (step_ms / tumbling_window_ms) as usize; let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; diff --git a/asap-query-engine/src/engines/simple_engine/sql.rs b/asap-query-engine/src/engines/simple_engine/sql.rs index 24c08c84..1e7f5d3e 100644 --- a/asap-query-engine/src/engines/simple_engine/sql.rs +++ b/asap-query-engine/src/engines/simple_engine/sql.rs @@ -382,7 +382,7 @@ impl SimpleEngine { spatial_filter: String, query_time: u64, ) -> Option { - let (query_plan, do_merge) = self + let (query_plan, do_merge, value_window_type) = self .create_store_query_plan(metric, timestamps, &agg_info) .map_err(|e| { warn!("Failed to create store query plan: {}", e); @@ -406,6 +406,7 @@ impl SimpleEngine { metadata, store_plan: query_plan, agg_info, + value_window_type, do_merge, spatial_filter, query_time, diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs index 2a2aa5c4..b4a5bbe3 100644 --- a/asap-query-engine/src/tests/mod.rs +++ b/asap-query-engine/src/tests/mod.rs @@ -15,6 +15,7 @@ pub mod sql_pattern_matching_tests; pub mod store_correctness_tests; pub mod structural_matching_tests; pub mod trait_design_tests; +pub mod window_semantics_consistency_tests; #[cfg(test)] pub mod test_utilities; 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 4cfa6428..88bfdcf6 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -8,10 +8,9 @@ //! value/key aggregations, values keyed `None`, grouping coming entirely //! from the keys aggregation's `get_keys()`) silently return an empty //! result over a range instead of the expanded key set. -//! 2. `finish_range_context` (`promql.rs`) unconditionally forces -//! `is_exact_query = false`, ignoring the aggregation's real `WindowType`, -//! so Sliding-window range queries don't fetch/merge the way the instant -//! path does. +//! 2. The range per-step merge logic didn't distinguish Sliding from +//! Tumbling, so Sliding-window range queries didn't fetch/merge the way +//! the instant path does (fixed by #608/#621). //! //! These tests are RED against current code: they mirror instant-query //! precedents that already pass (`native_binary_instant_tests.rs`'s @@ -790,8 +789,8 @@ mod tests { async fn range_query_sliding_window_single_bucket_regression() { // No-collision counterpart to the merge tests above: a single // Sliding bucket per output step must still return its value - // unchanged once is_exact_query correctly honors WindowType::Sliding - // for range queries. Mirrors + // unchanged now that the per-step merge logic correctly honors + // WindowType::Sliding for range queries. Mirrors // native_pipeline_merge_tests::sliding_single_bucket_returns_its_value. let data = vec![( 1_000_000, diff --git a/asap-query-engine/src/tests/test_utilities/comparison.rs b/asap-query-engine/src/tests/test_utilities/comparison.rs index 954d5107..133703a1 100644 --- a/asap-query-engine/src/tests/test_utilities/comparison.rs +++ b/asap-query-engine/src/tests/test_utilities/comparison.rs @@ -2,7 +2,7 @@ //! //! Provides assertion helpers for deep equality checking of query execution contexts. -use crate::data_model::{AggregationIdInfo, AggregationType}; +use crate::data_model::{AggregationIdInfo, AggregationType, WindowType}; use crate::engines::simple_engine::{ QueryExecutionContext, QueryMetadata, StoreQueryParams, StoreQueryPlan, }; @@ -30,6 +30,13 @@ pub fn assert_execution_context_equivalent( test_name ); + // Compare value_window_type + assert_eq!( + context1.value_window_type, context2.value_window_type, + "{}: value_window_type mismatch", + test_name + ); + // Compare metadata assert_metadata_equivalent(&context1.metadata, &context2.metadata, test_name); @@ -120,12 +127,6 @@ pub fn assert_store_params_equivalent( "{}: End timestamp mismatch - PromQL={}, SQL={}", test_name, params1.end_timestamp, params2.end_timestamp ); - - assert_eq!( - params1.is_exact_query, params2.is_exact_query, - "{}: Query type mismatch - PromQL={}, SQL={}", - test_name, params1.is_exact_query, params2.is_exact_query - ); } /// Assert that two KeyByLabelNames objects are equivalent @@ -193,7 +194,6 @@ mod tests { aggregation_id: 1, start_timestamp: 1000, end_timestamp: 2000, - is_exact_query: false, }, keys_query: None, }, @@ -203,6 +203,7 @@ mod tests { aggregation_type_for_key: AggregationType::Sum, aggregation_type_for_value: AggregationType::Sum, }, + value_window_type: WindowType::Tumbling, do_merge: true, // OnlyTemporal queries merge spatial_filter: String::new(), query_time: 2_000_000, // query timestamp in milliseconds diff --git a/asap-query-engine/src/tests/window_semantics_consistency_tests.rs b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs new file mode 100644 index 00000000..ad9c5bb1 --- /dev/null +++ b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs @@ -0,0 +1,638 @@ +//! Hardening tests for the store-query contract that governs how Sliding vs +//! Tumbling aggregations get fetched/combined, written ahead of an internal +//! refactor of that decision logic (see `exact_window_grid_adversarial_tests.rs` +//! for the original bug writeups this complements: #600, #606, #608). +//! +//! These tests are deliberately NOT shaped around any particular internal +//! function or struct -- every assertion here is made by driving the public +//! PromQL entry points (`handle_query_promql` / `handle_range_query_promql`) +//! and reading back `QueryResult` values, so they stay valid across an +//! internal reshape of how the tolerant-scan vs exact-lookup decision is +//! threaded through the code. +//! +//! Distinct from the existing coverage in `exact_window_grid_adversarial_tests.rs` +//! and `native_range_query_tests.rs`, this file focuses on: +//! +//! 1. Directly cross-checking the INSTANT and RANGE query paths against each +//! other (not just against a hand-computed literal) for both Sliding and +//! Tumbling, at the same timestamp -- the exact shape of bug that has +//! recurred three times (different query code paths drifting out of +//! sync). +//! 2. An instant-query keys lookup with a Sliding key aggregation (the +//! existing Sliding-keys coverage in `native_range_query_tests.rs` only +//! drives the range path). +//! 3. Grid-aligned vs off-grid instant query times for a Sliding aggregation. +//! 4. Query span exactly one bucket wide vs narrower than one bucket, +//! through the PromQL engine (the existing coverage for this shape lives +//! at the `Store` trait level in `exact_window_grid_adversarial_tests.rs`, +//! not through PromQL). + +#[cfg(test)] +mod tests { + use crate::data_model::{ + AggregationConfig, AggregationReference, AggregationType, CleanupPolicy, InferenceConfig, + KeyByLabelValues, PrecomputedOutput, PromQLSchema, QueryConfig, QueryLanguage, + SchemaConfig, StreamingConfig, WindowType, + }; + use crate::engines::query_result::{QueryResult, RangeVectorElement}; + use crate::engines::simple_engine::SimpleEngine; + use crate::precompute_operators::sum_accumulator::SumAccumulator; + use crate::precompute_operators::{CountMinSketchAccumulator, SetAggregatorAccumulator}; + use crate::stores::simple_map_store::SimpleMapStore; + use crate::stores::Store; + use crate::tests::test_utilities::engine_factories::create_engine_multi_timestamp_with_window; + use crate::AggregateCore; + use promql_utilities::data_model::KeyByLabelNames; + use std::collections::HashMap; + use std::sync::Arc; + + fn matrix_values(qr: QueryResult) -> Vec { + match qr { + QueryResult::Matrix(m) => m.values, + _ => panic!("expected matrix (range vector) result"), + } + } + + fn vector_values(qr: QueryResult) -> Vec<(Vec, f64)> { + match qr { + QueryResult::Vector(iv) => iv + .values + .into_iter() + .map(|e| (e.labels.labels, e.value)) + .collect(), + _ => panic!("expected vector (instant) result"), + } + } + + fn host_a_samples(elements: &[RangeVectorElement]) -> Vec<(u64, f64)> { + let mut samples: Vec<(u64, f64)> = elements + .iter() + .find(|e| e.labels.labels.contains(&"host-a".to_string())) + .map(|e| e.samples.iter().map(|s| (s.timestamp, s.value)).collect()) + .unwrap_or_default(); + samples.sort_by_key(|(ts, _)| *ts); + samples + } + + fn assert_close(actual: f64, expected: f64, ctx: &str) { + assert!( + (actual - expected).abs() < 1e-6, + "{ctx}: expected {expected}, got {actual}" + ); + } + + /// Single host-a instant value, or panics if absent/ambiguous -- used + /// where a test wants "the" value, not merely presence. + fn single_host_a_value(qr: QueryResult) -> f64 { + let values = vector_values(qr); + let matches: Vec = values + .into_iter() + .filter(|(labels, _)| labels.contains(&"host-a".to_string())) + .map(|(_, v)| v) + .collect(); + assert_eq!( + matches.len(), + 1, + "expected exactly one host-a series in instant result" + ); + matches[0] + } + + // ════════════════════════════════════════════════════════════════════ + // 1. Sliding: instant and range paths must AGREE, at every step, on + // exactly the correct (non-double-counted) value. + // ════════════════════════════════════════════════════════════════════ + + /// window_size=3000, slide=1000 -- classic overlapping-Sliding-window + /// shape (the historical 111-vs-12321 bug). Panes: 1000->1, 2000->10, + /// 3000->100, 4000->1000, 5000->10000. + /// step 3000: window [0,3000) -> 1+10+100=111 + /// step 4000: window [1000,4000) -> 10+100+1000=1110 + /// step 5000: window [2000,5000) -> 100+1000+10000=11100 + /// A naive "sum every overlapping window touching the range" substitute + /// would produce 111+1110+11100=12321 for a query collapsed to one + /// point -- this test's exact per-step values are far enough apart from + /// that failure mode (and from each other) that neither the instant nor + /// the range path can silently drift into it without failing here. + /// + /// The key property this test pins beyond `exact_window_grid_adversarial_tests` + /// is that BOTH paths are driven from the SAME engine/data and compared + /// directly against each other (`assert_eq!(instant_value, range_value)`), + /// not just each checked separately against a hand-computed literal -- + /// so a refactor that makes the two paths diverge fails immediately even + /// if it happens to leave one of them "coincidentally" correct-looking. + #[tokio::test(flavor = "multi_thread")] + async fn sliding_instant_and_range_agree_at_every_step_no_double_count() { + let data = vec![ + ( + 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(1.0)) as Box, + ), + ( + 2_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(10.0)) as Box, + ), + ( + 3_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(100.0)) as Box, + ), + ( + 4_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(1_000.0)) as Box, + ), + ( + 5_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(10_000.0)) as Box, + ), + ]; + let query = "sum_over_time(cpu_load[3s])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 3_000, + 1_000, + WindowType::Sliding, + ); + + let range_result = engine + .handle_range_query_promql(query.to_string(), 3.0, 5.0, 1.0) + .expect("range query failed"); + let range_samples = host_a_samples(&matrix_values(range_result.1)); + + let expected = vec![(3_000u64, 111.0), (4_000, 1_110.0), (5_000, 11_100.0)]; + assert_eq!( + range_samples, expected, + "range path must produce exactly the un-inflated per-step values" + ); + + for (query_time_sec, (ts, expected_value)) in + [3.0, 4.0, 5.0].into_iter().zip(expected.into_iter()) + { + let instant_result = engine + .handle_query_promql(query.to_string(), query_time_sec) + .unwrap_or_else(|| panic!("instant query at t={query_time_sec} failed")); + let instant_value = single_host_a_value(instant_result.1); + assert_close( + instant_value, + expected_value, + &format!("instant value at t={ts}"), + ); + + let range_value = range_samples + .iter() + .find(|(s_ts, _)| *s_ts == ts) + .map(|(_, v)| *v) + .unwrap_or_else(|| panic!("range path missing sample at t={ts}")); + assert_close( + instant_value, + range_value, + &format!( + "instant and range paths must agree exactly at t={ts} -- \ + a path that independently re-derives Sliding's window/merge \ + logic can silently drift out of sync with the other" + ), + ); + } + } + + // ════════════════════════════════════════════════════════════════════ + // 2. Tumbling: instant and range paths must AGREE, and both must + // correctly SUM every disjoint bucket in the query's window (proving + // a fix for Sliding's over-counting can't accidentally break + // Tumbling's legitimate summing). + // ════════════════════════════════════════════════════════════════════ + + /// 4 disjoint Tumbling buckets (1000ms each) with distinct values + /// 1,2,3,4 at ts=1000,2000,3000,4000. A `sum_over_time(metric[4s])` + /// query spans exactly all 4 -- correct answer is the full sum (10.0). + /// A buggy "exact-lookup-only" substitute (treating Tumbling like + /// Sliding: pick one bucket, don't merge) would return just the last + /// bucket's value (4.0) instead -- distinguishable from the correct sum + /// by more than 1e-6, so this test would catch that regression too. + #[tokio::test(flavor = "multi_thread")] + async fn tumbling_instant_and_range_agree_and_sum_all_n_windows() { + let data = vec![ + ( + 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(1.0)) as Box, + ), + ( + 2_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(2.0)) as Box, + ), + ( + 3_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(3.0)) as Box, + ), + ( + 4_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(4.0)) as Box, + ), + ]; + let query = "sum_over_time(cpu_load[4s])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 1_000, // window_size_ms: each ingest bucket is 1000ms, Tumbling + 1_000, // slide_interval_ms == window_size_ms for Tumbling + WindowType::Tumbling, + ); + + let instant_result = engine + .handle_query_promql(query.to_string(), 4.0) + .expect("instant query failed"); + let instant_value = single_host_a_value(instant_result.1); + + let range_result = engine + .handle_range_query_promql(query.to_string(), 4.0, 4.5, 1.0) + .expect("range query failed"); + let range_samples = host_a_samples(&matrix_values(range_result.1)); + + assert_eq!(range_samples.len(), 1, "expected exactly one output step"); + let (range_ts, range_value) = range_samples[0]; + assert_eq!(range_ts, 4_000); + + assert_close( + instant_value, + 10.0, + "instant value must be the sum of all 4 disjoint Tumbling buckets (1+2+3+4)", + ); + assert_close( + range_value, + 10.0, + "range value must be the sum of all 4 disjoint Tumbling buckets (1+2+3+4)", + ); + assert_close( + instant_value, + range_value, + "instant and range paths must agree exactly on the Tumbling sum", + ); + } + + // ════════════════════════════════════════════════════════════════════ + // 3. Keys queries: correct key set through the INSTANT path regardless + // of whether the key aggregation is Sliding or Tumbling. + // ════════════════════════════════════════════════════════════════════ + + /// Builds a dual-population engine (separate value/key aggregations) + /// with ONE key bucket, ending at `end_ts`, whose width and window type + /// are caller-controlled. For Sliding, the bucket is inserted already + /// pre-merged and `window_size_ms`-wide -- exactly the shape real + /// Sliding data takes when it reaches the store (see + /// `worker.rs::merge_panes_for_window`, and the same convention used by + /// `native_range_query_tests.rs::create_range_engine_dual_input_sliding_keys`). + /// SetAggregator is used for the key aggregation (not DeltaSetAggregator, + /// which #606 restricts to Tumbling only, so it can't express the + /// Sliding case here). + fn build_dual_engine_with_key_window( + end_ts: u64, + key_window_type: WindowType, + key_window_size_ms: u64, + key_slide_interval_ms: u64, + ) -> SimpleEngine { + let mut aggregation_configs = HashMap::new(); + aggregation_configs.insert( + 1u64, + AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::CountMinSketch, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::empty(), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: 1_000, + slide_interval_ms: 1_000, + window_type: WindowType::Tumbling, + 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, + }, + ); + aggregation_configs.insert( + 2u64, + AggregationConfig { + aggregation_id: 2, + aggregation_type: AggregationType::SetAggregator, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::empty(), + aggregated_labels: KeyByLabelNames::new(vec!["host".to_string()]), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: key_window_size_ms, + slide_interval_ms: key_slide_interval_ms, + window_type: key_window_type, + 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, + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + // Value bucket: plain Tumbling 1000ms, ending at end_ts. + let cms = CountMinSketchAccumulator::new(2, 3); + let value_output = PrecomputedOutput::new(end_ts - 1_000, end_ts, None, 1); + store + .insert_precomputed_output(value_output, Box::new(cms)) + .unwrap(); + + // Key bucket: window_size_ms-wide, ending at end_ts -- already + // pre-merged, matching real Sliding data's on-store shape. + let mut keys = SetAggregatorAccumulator::new(); + keys.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + let keys_output = PrecomputedOutput::new(end_ts - key_window_size_ms, end_ts, None, 2); + store + .insert_precomputed_output(keys_output, Box::new(keys)) + .unwrap(); + + let promql_schema = PromQLSchema::new().add_metric( + "event_frequency".to_string(), + KeyByLabelNames::new(vec!["host".to_string()]), + ); + let query_config = QueryConfig::new("count(event_frequency) by (host)".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, + 1_000, + QueryLanguage::promql, + ) + } + + /// Two engines, identical value data and identical logical key ("host-a" + /// valid over the window ending at t=5000), differing only in whether + /// the KEY aggregation is Tumbling (window=slide=1000) or Sliding + /// (window=2000, slide=1000). Both must resolve the SAME key set through + /// the instant query path -- keys queries conceptually always need to + /// see the key's own bucket correctly, independent of the value-side + /// double-counting concern that only applies to Sliding VALUE data. + #[tokio::test(flavor = "multi_thread")] + async fn instant_keys_query_correct_for_both_tumbling_and_sliding_key_aggregation() { + let tumbling_engine = + build_dual_engine_with_key_window(5_000, WindowType::Tumbling, 1_000, 1_000); + let sliding_engine = + build_dual_engine_with_key_window(5_000, WindowType::Sliding, 2_000, 1_000); + + let query = "count(event_frequency) by (host)"; + + let (_, tumbling_qr) = tumbling_engine + .handle_query_promql(query.to_string(), 5.0) + .expect("tumbling-keys instant query failed"); + let tumbling_values = vector_values(tumbling_qr); + + let (_, sliding_qr) = sliding_engine + .handle_query_promql(query.to_string(), 5.0) + .expect("sliding-keys instant query failed"); + let sliding_values = vector_values(sliding_qr); + + assert!( + tumbling_values + .iter() + .any(|(labels, _)| labels.contains(&"host-a".to_string())), + "Tumbling key aggregation must resolve host-a, got {tumbling_values:?}" + ); + assert!( + sliding_values + .iter() + .any(|(labels, _)| labels.contains(&"host-a".to_string())), + "Sliding key aggregation must resolve host-a exactly the same way \ + a Tumbling one does, got {sliding_values:?}" + ); + assert_eq!( + tumbling_values.len(), + sliding_values.len(), + "same logical key set must be returned regardless of the key \ + aggregation's WindowType" + ); + } + + // ════════════════════════════════════════════════════════════════════ + // 4. Sliding: a grid position with a genuine (fully-paned) window must + // resolve to its exact value; a grid position whose window is + // missing a pane (a gap) must resolve to NO data -- not a + // fabricated/wrong-window value borrowed from a neighboring window. + // ════════════════════════════════════════════════════════════════════ + + /// window_size=2000, slide=1000. Panes at 1000(v=5), 2000(v=7), + /// [3000 missing], 4000(v=9), 5000(v=11). Per + /// `create_engine_multi_timestamp_with_window`'s pane-merge logic, a + /// window only materializes in the store if ALL its panes are present: + /// window_start=0 -> panes {1000,2000} both present -> [0,2000)=12 + /// window_start=1000 -> panes {2000,3000} -- 3000 missing -> gap + /// window_start=2000 -> panes {3000,4000} -- 3000 missing -> gap + /// window_start=3000 -> panes {4000,5000} both present -> [3000,5000)=20 + /// query_time=2.0s -> window [0,2000) -> resolves to 12.0. + /// query_time=3.0s -> window [1000,3000) -> gap -> must be absent. + /// query_time=5.0s -> window [3000,5000) -> resolves to 20.0. + #[tokio::test(flavor = "multi_thread")] + async fn sliding_instant_query_resolves_present_window_and_returns_no_data_for_gap() { + let data = vec![ + ( + 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(5.0)) as Box, + ), + ( + 2_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(7.0)) as Box, + ), + // gap: no pane ending at 3000 + ( + 4_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(9.0)) as Box, + ), + ( + 5_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(11.0)) as Box, + ), + ]; + let query = "sum_over_time(cpu_load[2s])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 2_000, + 1_000, + WindowType::Sliding, + ); + + let (_, first_qr) = engine + .handle_query_promql(query.to_string(), 2.0) + .expect("instant query for the fully-paned window [0,2000) failed"); + assert_close( + single_host_a_value(first_qr), + 12.0, + "window [0,2000) has both required panes (5+7) and must resolve exactly", + ); + + let gap_result = engine.handle_query_promql(query.to_string(), 3.0); + let gap_values = match gap_result { + Some((_, qr)) => vector_values(qr), + None => Vec::new(), + }; + assert!( + !gap_values + .iter() + .any(|(labels, _)| labels.contains(&"host-a".to_string())), + "window [1000,3000) is missing its 3000-ending pane -- must yield no data \ + for host-a, not a value borrowed from a neighboring window: got {gap_values:?}" + ); + + let (_, third_qr) = engine + .handle_query_promql(query.to_string(), 5.0) + .expect("instant query for the fully-paned window [3000,5000) failed"); + assert_close( + single_host_a_value(third_qr), + 20.0, + "window [3000,5000) has both required panes (9+11) and must resolve exactly, \ + unaffected by the gap immediately before it", + ); + } + + // ════════════════════════════════════════════════════════════════════ + // 5/6. Query span exactly one bucket wide vs narrower than one bucket. + // ════════════════════════════════════════════════════════════════════ + + /// Single Tumbling bucket [0,1000)=42. A `[1s]` query's window is + /// exactly as wide as the stored bucket -- both instant and range paths + /// must return the bucket's exact value, not empty and not doubled. + #[tokio::test(flavor = "multi_thread")] + async fn tumbling_query_exactly_one_bucket_wide_returns_exact_value() { + let data = vec![( + 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(42.0)) as Box, + )]; + let query = "sum_over_time(cpu_load[1s])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 1_000, + 1_000, + WindowType::Tumbling, + ); + + let (_, instant_qr) = engine + .handle_query_promql(query.to_string(), 1.0) + .expect("instant query failed"); + let instant_value = single_host_a_value(instant_qr); + assert_close( + instant_value, + 42.0, + "query span exactly one bucket wide must return the bucket's exact value", + ); + + let (_, range_qr) = engine + .handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0) + .expect("range query failed"); + let range_samples = host_a_samples(&matrix_values(range_qr)); + assert_eq!( + range_samples, + vec![(1_000, 42.0)], + "range path must agree with the instant path for a query span exactly \ + one bucket wide" + ); + } + + /// Same single Tumbling bucket [0,1000)=42, but the query's window + /// (`[500ms]`) is NARROWER than the stored bucket. No stored bucket's + /// `[window_start, window_end)` fits inside the requested `[500,1000)` + /// span, so the correct behavior is to return NO data -- not a + /// fabricated partial value (e.g. neither 42.0 wrongly reused nor some + /// halved/interpolated 21.0). + #[tokio::test(flavor = "multi_thread")] + async fn tumbling_query_narrower_than_bucket_returns_no_data_not_partial() { + let data = vec![( + 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(42.0)) as Box, + )]; + let query = "sum_over_time(cpu_load[500ms])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 1_000, + 1_000, + WindowType::Tumbling, + ); + + let instant_result = engine.handle_query_promql(query.to_string(), 1.0); + let instant_values = match instant_result { + Some((_, qr)) => vector_values(qr), + None => Vec::new(), + }; + assert!( + !instant_values + .iter() + .any(|(labels, _)| labels.contains(&"host-a".to_string())), + "a query window narrower than the stored bucket must yield no data \ + for host-a, not a fabricated/partial value: got {instant_values:?}" + ); + + let range_result = engine.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0); + let range_has_sample_at_1000 = match range_result { + Some((_, qr)) => host_a_samples(&matrix_values(qr)) + .iter() + .any(|(ts, _)| *ts == 1_000), + None => false, + }; + assert!( + !range_has_sample_at_1000, + "range path must likewise yield no sample at t=1000 for a query window \ + narrower than the stored bucket" + ); + } +}