diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 71fe6ce..56a4a84 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -463,6 +463,105 @@ impl SimpleEngine { )) } + /// The bucket-map grid width for scanning an aggregation's stored + /// buckets: `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 + } + } + + /// 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. + fn scan_windows_via_exact( + &self, + params: &StoreQueryParams, + ) -> Result { + let sc = self.streaming_config.read().unwrap().clone(); + let config = sc + .get_aggregation_config(params.aggregation_id) + .ok_or_else(|| { + format!( + "Aggregation config not found for aggregation_id: {}", + params.aggregation_id + ) + })?; + // DeltaSetAggregator keys queries span [0, end_timestamp] -- + // "all keys ever seen" (see create_keys_query_params) -- a nominal + // range the grid-walk below can't cover cheaply. The tolerant scan + // stays fast here regardless of nominal range width (it short- + // circuits per-epoch via time_bounds() and binary-searches within + // surviving epochs), so keep using it for this one case. + if config.aggregation_type == AggregationType::DeltaSetAggregator { + return self + .store + .query_precomputed_output( + ¶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 + ) + }); + } + + let window_size_ms = config.window_size_ms; + let step_ms = Self::bucket_step_ms(config); + + let mut merged: TimestampedBucketsMap = HashMap::new(); + if window_size_ms == 0 || step_ms == 0 || params.start_timestamp > params.end_timestamp { + return Ok(merged); + } + + let mut window_start = params.start_timestamp.div_ceil(step_ms) * step_ms; + while window_start + window_size_ms <= params.end_timestamp { + let window_end = window_start + window_size_ms; + let partial = self + .store + .query_precomputed_output_exact( + ¶ms.metric, + params.aggregation_id, + window_start, + window_end, + ) + .map_err(|e| { + format!( + "Error querying store for metric {}, agg {}, window [{}, {}]: {}", + params.metric, params.aggregation_id, window_start, window_end, e + ) + })?; + for (key, buckets) in partial { + merged.entry(key).or_default().extend(buckets); + } + window_start += step_ms; + } + Ok(merged) + } + /// Executes a single store query based on parameters fn execute_store_query( &self, @@ -484,12 +583,24 @@ impl SimpleEngine { "Sliding window query: Looking for exact window [{}, {}]", params.start_timestamp, params.end_timestamp ); - let res = self.store.query_precomputed_output_exact( - ¶ms.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - ); + 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!( @@ -501,35 +612,22 @@ impl SimpleEngine { res } else { debug!( - "Tumbling window query: range [{}, {}]", + "Window-grid query: range [{}, {}]", params.start_timestamp, params.end_timestamp ); - let res = self.store.query_precomputed_output( - ¶ms.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - ); - if res.is_ok() { + let res = self.scan_windows_via_exact(params); + if let Ok(ref outputs) = res { let store_query_duration = store_query_start_time.elapsed(); debug!( - "Tumbling window range query took: {:.2}ms", - store_query_duration.as_secs_f64() * 1000.0 + "Window-grid query took: {:.2}ms, found {} unique keys", + store_query_duration.as_secs_f64() * 1000.0, + outputs.len() ); } res }; - result.map_err(|e| { - format!( - "Error querying store for metric {}, agg {}, range [{}, {}]: {}", - params.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - e - ) - }) + result } /// Executes the full store query plan and returns merged results diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 41a859a..22f843a 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -557,25 +557,6 @@ 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 diff --git a/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs b/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs new file mode 100644 index 0000000..c80818b --- /dev/null +++ b/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs @@ -0,0 +1,905 @@ +//! Adversarial tests for the store-query contract, written directly from the +//! spec in `stores/traits.rs` (`Store::query_precomputed_output` / +//! `query_precomputed_output_exact`) and the window/bucket alignment model in +//! `precompute_engine/window_manager.rs`. +//! +//! These are NOT shaped around any particular implementation. They target the +//! class of bug that shows up when a tolerant range scan +//! (`query_precomputed_output`-shaped: "give me every stored window whose +//! `[window_start, window_end)` fits inside `[start, end]`") gets reimplemented +//! as a series of strict per-window exact lookups +//! (`query_precomputed_output_exact`-shaped): stepping by the wrong grid +//! (`window_size_ms` instead of `slide_interval_ms`) silently drops sliding +//! windows off that grid (issue #600), a naive "enumerate every grid position" +//! substitute for a sparse tolerant scan can go pathological on a wide range +//! with sparse data (the `DeltaSetAggregator` keys query, which spans +//! `[0, end_timestamp]`), and boundary/gap handling can silently differ from +//! the tolerant scan's containment semantics. +//! +//! Two levels of test live here: +//! +//! - **Store-level** (`SimpleMapStore` directly): pin the `Store` trait +//! contract itself -- empty-range edge cases, boundary containment, +//! overlapping sliding windows, and sealed/mutable epoch spans. These +//! exercise the store's own (unchanged) `query_precomputed_output`, so they +//! document the ground truth the rest of the pipeline must reproduce. +//! - **Engine-level** (`SimpleEngine` + PromQL): exercise the actual query +//! pipeline (`handle_query_promql` / `handle_range_query_promql`), which is +//! where a tolerant-scan-to-exact-lookups rewrite would actually live. Each +//! expected result below is computed by hand from `window_manager.rs`'s +//! grid math (`window_start_for`, `window_starts_containing`, +//! `panes_for_window`), not from reading the implementation under test. + +#[cfg(test)] +mod tests { + use crate::data_model::{ + AggregationConfig, AggregationReference, AggregationType, CleanupPolicy, InferenceConfig, + KeyByLabelValues, LockStrategy, PrecomputedOutput, PromQLSchema, QueryConfig, + QueryLanguage, SchemaConfig, SerializableToSink, 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, DeltaSetAggregatorAccumulator}; + use crate::stores::simple_map_store::SimpleMapStore; + use crate::stores::{Store, TimestampedBucketsMap}; + 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; + use std::time::{Duration, Instant}; + + // ════════════════════════════════════════════════════════════════════ + // ── Store-level tests: pin the `Store` trait contract directly ────── + // ════════════════════════════════════════════════════════════════════ + + fn make_agg_config( + agg_id: u64, + aggregation_type: AggregationType, + window_size_ms: u64, + slide_interval_ms: u64, + window_type: WindowType, + num_aggregates_to_retain: Option, + ) -> AggregationConfig { + AggregationConfig { + aggregation_id: agg_id, + aggregation_type, + 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, + slide_interval_ms, + window_type, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "cpu_usage".to_string(), + num_aggregates_to_retain, + read_count_threshold: None, + table_name: None, + value_column: None, + } + } + + fn make_store(configs: Vec<(u64, AggregationConfig)>, policy: CleanupPolicy) -> SimpleMapStore { + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs: configs.into_iter().collect(), + }); + SimpleMapStore::new_with_strategy(streaming_config, policy, LockStrategy::PerKey) + } + + fn sum_entry( + agg_id: u64, + start: u64, + end: u64, + value: f64, + ) -> (PrecomputedOutput, Box) { + ( + PrecomputedOutput::new(start, end, None, agg_id), + Box::new(SumAccumulator::with_sum(value)), + ) + } + + fn keyed_delta_entry( + agg_id: u64, + start: u64, + end: u64, + key_label: &str, + ) -> (PrecomputedOutput, Box) { + let mut acc = DeltaSetAggregatorAccumulator::new(); + acc.add_key(KeyByLabelValues { + labels: vec![key_label.to_string()], + }); + ( + PrecomputedOutput::new(start, end, None, agg_id), + Box::new(acc), + ) + } + + fn total_bucket_count(result: &TimestampedBucketsMap) -> usize { + result.values().map(|v| v.len()).sum() + } + + fn timestamps_for_none_key(result: &TimestampedBucketsMap) -> Vec<(u64, u64)> { + let mut ts: Vec<(u64, u64)> = result + .get(&None) + .map(|buckets| buckets.iter().map(|(range, _)| *range).collect()) + .unwrap_or_default(); + ts.sort_unstable(); + ts + } + + /// Spec: `window_start >= start && window_start <= end && window_end <= end`. + /// With `start == end == window_start`, the third condition + /// (`window_end <= end`) fails for any window with positive width -- a + /// degenerate `[t, t)` query range must never partial-match a real window + /// just because its start lines up. + #[test] + fn store_query_start_equals_end_excludes_any_real_window() { + let store = make_store( + vec![( + 1, + make_agg_config( + 1, + AggregationType::Sum, + 1000, + 1000, + WindowType::Tumbling, + None, + ), + )], + CleanupPolicy::NoCleanup, + ); + let (out, acc) = sum_entry(1, 5_000, 6_000, 7.0); + store.insert_precomputed_output(out, acc).unwrap(); + + let result = store + .query_precomputed_output("cpu_usage", 1, 5_000, 5_000) + .unwrap(); + assert!( + result.is_empty(), + "degenerate [t, t) range must not partial-match the window starting at t: {:?}", + timestamps_for_none_key(&result) + ); + } + + /// Spec conditions are `window_start >= start && window_start <= end && + /// window_end <= end`. With `start > end`, no real window can satisfy + /// `window_start <= end` while also being `>= start` unless start<=end -- + /// this pins that an inverted range never accidentally matches via a + /// buggy min/max swap or an unchecked subtraction. + #[test] + fn store_query_start_greater_than_end_returns_empty_even_with_window_start_between() { + let store = make_store( + vec![( + 1, + make_agg_config( + 1, + AggregationType::Sum, + 1000, + 1000, + WindowType::Tumbling, + None, + ), + )], + CleanupPolicy::NoCleanup, + ); + // window_start = 5_000 sits numerically "between" end=3_000 and + // start=7_000, which is exactly the kind of case a start/end swap bug + // would wrongly match. + let (out, acc) = sum_entry(1, 5_000, 6_000, 7.0); + store.insert_precomputed_output(out, acc).unwrap(); + + let result = store + .query_precomputed_output("cpu_usage", 1, 7_000, 3_000) + .unwrap(); + assert!( + result.is_empty(), + "start > end must return empty regardless of window placement: {:?}", + timestamps_for_none_key(&result) + ); + } + + /// A window whose `window_start` precedes the query's `start` must be + /// excluded entirely (not partially matched) even though it overlaps the + /// query range -- this is the "partial window at the start of the range" + /// case from the spec. + #[test] + fn store_query_excludes_window_whose_start_precedes_query_start() { + let store = make_store( + vec![( + 1, + make_agg_config( + 1, + AggregationType::Sum, + 1000, + 1000, + WindowType::Tumbling, + None, + ), + )], + CleanupPolicy::NoCleanup, + ); + let (out1, acc1) = sum_entry(1, 1_000, 2_000, 11.0); // starts before query_start + let (out2, acc2) = sum_entry(1, 2_000, 3_000, 22.0); // fully inside + store.insert_precomputed_output(out1, acc1).unwrap(); + store.insert_precomputed_output(out2, acc2).unwrap(); + + let result = store + .query_precomputed_output("cpu_usage", 1, 1_500, 3_000) + .unwrap(); + assert_eq!( + timestamps_for_none_key(&result), + vec![(2_000, 3_000)], + "window [1000,2000) starts before query_start=1500 and must be excluded entirely, \ + not truncated/partially matched" + ); + } + + /// Symmetric case: a window whose `window_end` exceeds the query's `end` + /// must be excluded entirely -- the "partial window at the end of the + /// range" case. + #[test] + fn store_query_excludes_window_whose_end_exceeds_query_end() { + let store = make_store( + vec![( + 1, + make_agg_config( + 1, + AggregationType::Sum, + 1000, + 1000, + WindowType::Tumbling, + None, + ), + )], + CleanupPolicy::NoCleanup, + ); + let (out1, acc1) = sum_entry(1, 3_000, 4_000, 33.0); // fully inside + let (out2, acc2) = sum_entry(1, 4_000, 5_000, 44.0); // ends after query_end + store.insert_precomputed_output(out1, acc1).unwrap(); + store.insert_precomputed_output(out2, acc2).unwrap(); + + let result = store + .query_precomputed_output("cpu_usage", 1, 3_000, 4_500) + .unwrap(); + assert_eq!( + timestamps_for_none_key(&result), + vec![(3_000, 4_000)], + "window [4000,5000) ends after query_end=4500 and must be excluded entirely, \ + not truncated/partially matched" + ); + } + + /// Three overlapping sliding windows (slide=1000, size=3000) inserted + /// directly -- a tolerant range scan covering all three must return all + /// three as DISTINCT entries, not merge/dedup/drop any of them, even + /// though their ranges heavily overlap. + #[test] + fn store_query_overlapping_sliding_windows_all_returned_distinctly() { + let store = make_store( + vec![( + 1, + make_agg_config( + 1, + AggregationType::Sum, + 3000, + 1000, + WindowType::Sliding, + None, + ), + )], + CleanupPolicy::NoCleanup, + ); + let (o1, a1) = sum_entry(1, 0, 3_000, 1.0); + let (o2, a2) = sum_entry(1, 1_000, 4_000, 2.0); + let (o3, a3) = sum_entry(1, 2_000, 5_000, 3.0); + store.insert_precomputed_output(o1, a1).unwrap(); + store.insert_precomputed_output(o2, a2).unwrap(); + store.insert_precomputed_output(o3, a3).unwrap(); + + let result = store + .query_precomputed_output("cpu_usage", 1, 0, 5_000) + .unwrap(); + assert_eq!( + total_bucket_count(&result), + 3, + "all three overlapping windows must be returned distinctly" + ); + assert_eq!( + timestamps_for_none_key(&result), + vec![(0, 3_000), (1_000, 4_000), (2_000, 5_000)], + ); + + // Confirm values weren't merged/overwritten across the overlapping + // entries -- each bucket's accumulator must carry its own distinct + // value (1.0, 2.0, 3.0), not the same one returned three times (which + // a merge/dedup bug would produce as identical serialized JSON). + let buckets = result.get(&None).unwrap(); + let json_for = |range: (u64, u64)| -> serde_json::Value { + buckets + .iter() + .find(|(r, _)| *r == range) + .unwrap_or_else(|| panic!("missing bucket {range:?}")) + .1 + .serialize_to_json() + }; + let j1 = json_for((0, 3_000)); + let j2 = json_for((1_000, 4_000)); + let j3 = json_for((2_000, 5_000)); + assert_eq!(j1, SumAccumulator::with_sum(1.0).serialize_to_json()); + assert_eq!(j2, SumAccumulator::with_sum(2.0).serialize_to_json()); + assert_eq!(j3, SumAccumulator::with_sum(3.0).serialize_to_json()); + } + + /// Range query spanning both sealed epochs and the still-open (mutable) + /// current epoch, in one call. Mirrors + /// `store_correctness_tests::test_exact_query_correct_after_epoch_rotation` + /// but drives the tolerant range scan (`query_precomputed_output`) + /// instead of the exact lookup, and checks correctness across the whole + /// retained span in a single query rather than probing one window at a + /// time. + /// + /// capacity=2, 10 inserts => retention_limit=8, evicting windows 0 and 1 + /// (oldest sealed epoch). Windows 2..10 remain, spread across multiple + /// sealed epochs plus the current (newest, still "mutable") epoch. + #[test] + fn store_query_range_spans_sealed_and_mutable_epochs() { + let store = make_store( + vec![( + 1, + make_agg_config( + 1, + AggregationType::Sum, + 60_000, + 60_000, + WindowType::Tumbling, + Some(2), + ), + )], + CleanupPolicy::CircularBuffer, + ); + let n = 10u64; + for i in 0..n { + let (out, acc) = sum_entry(1, i * 60_000, (i + 1) * 60_000, i as f64); + store.insert_precomputed_output(out, acc).unwrap(); + } + + let result = store + .query_precomputed_output("cpu_usage", 1, 0, n * 60_000) + .unwrap(); + let ts = timestamps_for_none_key(&result); + let expected: Vec<(u64, u64)> = (2..n).map(|i| (i * 60_000, (i + 1) * 60_000)).collect(); + assert_eq!( + ts, expected, + "range query spanning sealed + mutable epochs must return exactly windows 2..10, \ + with windows 0,1 evicted" + ); + + let buckets = result.get(&None).unwrap(); + for i in 2..n { + let range = (i * 60_000, (i + 1) * 60_000); + let acc = &buckets.iter().find(|(r, _)| *r == range).unwrap().1; + let expected_json = SumAccumulator::with_sum(i as f64).serialize_to_json(); + assert_eq!( + acc.serialize_to_json(), + expected_json, + "window {i} must return its own value across the sealed/mutable epoch boundary" + ); + } + } + + /// Baseline/regression pin for `SimpleMapStore` itself (not the engine): + /// a `DeltaSetAggregator`-shaped keys query spans `[0, end_timestamp]` + /// per `create_keys_query_params`. With real data confined to a narrow + /// sliver near a huge `end_timestamp`, the underlying store's tolerant + /// scan must both (a) return the correct sparse set and (b) complete + /// quickly -- it must not be a function of the nominal range width. This + /// documents that the store layer itself was never the risk; the risk is + /// in whatever composes calls to it (see the engine-level counterpart + /// below). + #[test] + fn store_query_delta_set_wide_range_from_zero_stays_fast_and_correct() { + let store = make_store( + vec![( + 1, + make_agg_config( + 1, + AggregationType::DeltaSetAggregator, + 1_000, + 1_000, + WindowType::Tumbling, + None, + ), + )], + CleanupPolicy::NoCleanup, + ); + // 1e11 ms (~3170 years) nominal end, with 5 real windows clustered in + // a 5000ms sliver near it -- deliberately chosen so that a hypothetical + // "enumerate every grid position from 0" scan (1e8 iterations at + // window_size_ms=1000 steps) would be obviously, measurably slow, + // while a real sparse-store scan is not. + let base: u64 = 100_000_000_000; + for (i, label) in ["a", "b", "c", "d", "e"].iter().enumerate() { + let start = base + i as u64 * 1_000; + let (out, acc) = keyed_delta_entry(1, start, start + 1_000, label); + store.insert_precomputed_output(out, acc).unwrap(); + } + + let query_start = Instant::now(); + let result = store + .query_precomputed_output("cpu_usage", 1, 0, base + 5_000) + .unwrap(); + let elapsed = query_start.elapsed(); + + assert_eq!( + total_bucket_count(&result), + 5, + "must return exactly the 5 real windows, sparse, not padded" + ); + assert!( + elapsed < Duration::from_secs(2), + "store-level wide-range keys query took {:?}, expected near-instant \ + (proportional to real data, not nominal range width)", + elapsed + ); + } + + // ════════════════════════════════════════════════════════════════════ + // ── Engine-level tests: drive the actual PromQL query pipeline ────── + // ════════════════════════════════════════════════════════════════════ + + 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}" + ); + } + + /// #600-style off-grid bug, but with THREE panes spanning a wider + /// window_size (3000ms, slide fixed at 1000ms by the factory) and THREE + /// output steps, none of whose windows sit on the window_size_ms grid + /// ({0, 3000, 6000, ...}) except the very first. A `bucket_step_ms` that + /// steps by `window_size_ms` instead of `slide_interval_ms` would miss + /// panes at 1000/2000/4000, silently under-summing every step after the + /// first -- and a broken step could also double count if it re-visits a + /// pane from an earlier step's window. Each step's expected sum is + /// distinct, so both failure modes (drop vs double-count) are caught. + /// + /// Panes (bucket = [end-1000, end)): 1000->1, 2000->10, 3000->100, + /// 4000->1000, 5000->10000. + /// Step 3000: window [0,3000) -> panes {0,1000,2000} -> 1+10+100=111 + /// Step 4000: window [1000,4000) -> panes {1000,2000,3000} -> 10+100+1000=1110 + /// Step 5000: window [2000,5000) -> panes {2000,3000,4000} -> 100+1000+10000=11100 + /// + /// Currently fails: pins the pre-existing bug tracked in + /// https://github.com/ProjectASAP/ASAPQuery/issues/608 (range queries + /// over Sliding-window aggregations use the overlap-scan fetch, not + /// exact-window fetch, then get merged downstream as if Tumbling). + /// Un-ignore once #608 lands. + #[ignore = "known bug, see #608"] + #[tokio::test(flavor = "multi_thread")] + async fn range_query_sliding_multi_step_off_grid_windows_no_double_count_or_drop() { + 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, // slide_interval_ms + WindowType::Sliding, + ); + + let result = engine.handle_range_query_promql(query.to_string(), 3.0, 5.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + let samples = host_a_samples(&elements); + + assert_eq!( + samples, + vec![(3_000, 111.0), (4_000, 1_110.0), (5_000, 11_100.0)], + "BUG-#600-class: off-window_size_ms-grid panes must be found at every step, \ + each step summing exactly its own 3 panes, not dropped or double-counted" + ); + } + + /// Instant-query counterpart of the range test above: the off-grid pane + /// scan must work identically through the single-timestamp instant path + /// (`handle_query_promql`), not just the multi-step range path. + /// query_time=4.0s -> window [1000,4000) -> panes {1000,2000,3000} = 1110. + #[tokio::test(flavor = "multi_thread")] + async fn instant_query_sliding_off_grid_window_merges_correctly() { + 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, + ), + ]; + 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, // slide_interval_ms + WindowType::Sliding, + ); + + let (_, qr) = engine + .handle_query_promql(query.to_string(), 4.0) + .expect("instant query failed"); + let values = vector_values(qr); + assert_eq!(values.len(), 1, "expected exactly one series for host-a"); + assert_close( + values[0].1, + 1_110.0, + "instant query at t=4.0 must merge panes {1000,2000,3000} = 10+100+1000", + ); + } + + /// Generalizes the off-grid case to check the window genuinely SHIFTS + /// with each step (window_size=2000, slide=1000) -- i.e. no step + /// accidentally reuses a neighboring step's pane set (which would show up + /// as either a dropped pane at the new edge or a stale pane retained past + /// where it should have rolled off). + /// Panes: 1000->1, 2000->10, 3000->100, 4000->1000. + /// Step 2000: window [0,2000) -> panes {0,1000} -> 1+10=11 + /// Step 3000: window [1000,3000) -> panes {1000,2000} -> 10+100=110 + /// Step 4000: window [2000,4000) -> panes {2000,3000} -> 100+1000=1100 + /// + /// Currently fails: same pre-existing bug as the test above, tracked in + /// https://github.com/ProjectASAP/ASAPQuery/issues/608. Un-ignore once + /// #608 lands. + #[ignore = "known bug, see #608"] + #[tokio::test(flavor = "multi_thread")] + async fn range_query_sliding_overlap_shifts_correctly_across_steps() { + 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, + ), + ]; + 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, // slide_interval_ms + WindowType::Sliding, + ); + + let result = engine.handle_range_query_promql(query.to_string(), 2.0, 4.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + let samples = host_a_samples(&elements); + + assert_eq!( + samples, + vec![(2_000, 11.0), (3_000, 110.0), (4_000, 1_100.0)], + "sliding window must shift correctly at every step, without leaking a stale \ + pane forward or dropping the newly-included one" + ); + } + + /// Tumbling range query spanning 5 windows with a GAP in the middle + /// (window ending at 3000 was never inserted). Per the tolerant-scan + /// contract, results are sparse -- the output series must have samples at + /// every step EXCEPT the missing one, not fail the whole query and not + /// silently pad the gap with a zero or a stale value. + #[tokio::test(flavor = "multi_thread")] + async fn range_query_tumbling_multi_window_gap_returns_sparse_not_padded() { + 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, + ), + // gap: no window ending at 3000 + ( + 4_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(4.0)) as Box, + ), + ( + 5_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(5.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, // slide_interval_ms == window_size_ms for Tumbling + WindowType::Tumbling, + ); + + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 5.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + let samples = host_a_samples(&elements); + + assert_eq!( + samples, + vec![(1_000, 1.0), (2_000, 2.0), (4_000, 4.0), (5_000, 5.0)], + "gap at t=3000 (never inserted) must be absent from the output -- sparse, \ + not padded with a fabricated/zero sample, and must not fail the whole query" + ); + } + + /// Custom dual-population engine builder parametrized by a caller-chosen + /// base timestamp (unlike `create_engine_dual_input`, which hardcodes + /// 1_000_000). Needed to place real data far from t=0 while exercising + /// `create_keys_query_params`'s `DeltaSetAggregator` span of + /// `[0, end_timestamp]` against a genuinely huge nominal range. + #[allow(clippy::too_many_arguments)] + #[allow(clippy::type_complexity)] + fn make_dual_engine_at_timestamp( + metric: &str, + value_agg_type: AggregationType, + key_agg_type: AggregationType, + grouping_labels: Vec<&str>, + aggregated_labels: Vec<&str>, + timestamp: u64, + value_data: Vec<(Option>, Box)>, + keys_data: Vec<(Option>, Box)>, + promql_query: &str, + ) -> 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: 1_000, + slide_interval_ms: 1_000, + 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: 1_000, + slide_interval_ms: 1_000, + 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, + }, + ); + + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + for (label_values_opt, acc) in value_data { + let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); + let output = PrecomputedOutput::new(timestamp - 1_000, timestamp, key, 1); + store.insert_precomputed_output(output, acc).unwrap(); + } + for (label_values_opt, acc) in keys_data { + let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); + let output = PrecomputedOutput::new(timestamp - 1_000, timestamp, key, 2); + 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, + 1000, + QueryLanguage::promql, + ) + } + + /// Engine-level counterpart to `store_query_delta_set_wide_range_from_zero_stays_fast_and_correct`: + /// this is the test that actually stresses `create_keys_query_params`'s + /// `[0, end_timestamp]` span for `DeltaSetAggregator` through the real + /// query pipeline. Real data lives in a single 1000ms window ~1e11 ms + /// (~3170 years) after t=0; the nominal keys-query range is therefore + /// ~1e11 ms wide with only that one sliver of real data in it. A + /// tolerant-scan-shaped composition must still resolve this correctly and + /// quickly; an "enumerate every grid position from 0" substitute (1e8 + /// iterations at 1000ms steps) would not be quick. + #[tokio::test(flavor = "multi_thread")] + async fn instant_query_delta_set_keys_wide_range_from_zero_completes_quickly_and_correctly() { + let base_ts: u64 = 100_000_000_000; // 1e11 ms + let cms = CountMinSketchAccumulator::new(2, 3); + let mut keys = DeltaSetAggregatorAccumulator::new(); + keys.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + + let engine = make_dual_engine_at_timestamp( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + base_ts, + vec![(None, Box::new(cms) as Box)], + vec![(None, Box::new(keys) as Box)], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let query_time_sec = base_ts as f64 / 1000.0; + + let call_start = Instant::now(); + let result = engine.handle_query_promql(query.to_string(), query_time_sec); + let elapsed = call_start.elapsed(); + + let (_, qr) = result.expect("query failed to resolve real data near a huge timestamp"); + let values = vector_values(qr); + assert!( + values + .iter() + .any(|(labels, _)| labels.contains(&"host-a".to_string())), + "expected host-a's key to be resolved via the DeltaSetAggregator keys query \ + spanning [0, {base_ts}], got {values:?}" + ); + assert!( + elapsed < Duration::from_secs(5), + "instant query with a DeltaSetAggregator keys span of [0, {base_ts}] (~1e11ms) \ + took {elapsed:?} -- a per-grid-position enumeration substitute for the tolerant \ + scan would be expected to blow well past this on a range this wide" + ); + } +} diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs index dcf0427..2a2aa5c 100644 --- a/asap-query-engine/src/tests/mod.rs +++ b/asap-query-engine/src/tests/mod.rs @@ -3,6 +3,7 @@ pub mod clickhouse_forwarding_tests; pub mod dispatch_arithmetic_tests; pub mod elastic_dsl_query_tests; pub mod elastic_forwarding_tests; +pub mod exact_window_grid_adversarial_tests; pub mod native_binary_arithmetic_plan_tests; pub mod native_binary_instant_tests; pub mod native_pipeline_merge_tests; diff --git a/asap-query-engine/src/tests/native_binary_instant_tests.rs b/asap-query-engine/src/tests/native_binary_instant_tests.rs index 1967068..9aef669 100644 --- a/asap-query-engine/src/tests/native_binary_instant_tests.rs +++ b/asap-query-engine/src/tests/native_binary_instant_tests.rs @@ -500,6 +500,7 @@ mod tests { data, leaf_query, 1_000, // window_size_ms, matches the fixed 1000ms bucket width + 1_000, // slide_interval_ms WindowType::Sliding, ); diff --git a/asap-query-engine/src/tests/native_pipeline_merge_tests.rs b/asap-query-engine/src/tests/native_pipeline_merge_tests.rs index 044b11a..77ef267 100644 --- a/asap-query-engine/src/tests/native_pipeline_merge_tests.rs +++ b/asap-query-engine/src/tests/native_pipeline_merge_tests.rs @@ -48,6 +48,7 @@ async fn sliding_single_bucket_returns_its_value() { data, query, SLIDING_WINDOW_MS, + SLIDING_WINDOW_MS, WindowType::Sliding, ); @@ -81,6 +82,7 @@ async fn sliding_two_buckets_for_same_key_are_merged_not_dropped() { data, query, SLIDING_WINDOW_MS, + SLIDING_WINDOW_MS, WindowType::Sliding, ); @@ -123,6 +125,7 @@ async fn sliding_bucket_count_mismatch_still_returns_merged_result() { data, query, SLIDING_WINDOW_MS, + SLIDING_WINDOW_MS, WindowType::Sliding, ); @@ -161,6 +164,7 @@ async fn tumbling_multi_bucket_merge_unaffected_by_sliding_fix() { // window_size_ms < query range so do_merge=true. Equal (5s window, // 5s range) hits a separate, pre-existing panic — see #569, not this stage. 1_000, + 1_000, // slide_interval_ms == window_size_ms for Tumbling WindowType::Tumbling, ); 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 0e83815..f22a264 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -346,7 +346,11 @@ mod tests { for (agg_id, bucket_span_ms, data) in [ (1u64, value_window_ms, value_data), - (2u64, key_slide_interval_ms, keys_data), + // Keys buckets are window_size_ms wide, same as values -- real + // Sliding data (any aggregation type) reaches the store already + // pre-merged by worker.rs::merge_panes_for_window, one entry per + // window_size_ms-wide window on the slide_interval_ms grid. + (2u64, key_window_size_ms, keys_data), ] { for (timestamp, label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); @@ -383,10 +387,34 @@ mod tests { // #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 + // buckets land on the 1000ms grid (start=3000), 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 + // that steps by window_size_ms never visits t=3000 and silently // drops host-a's key. + // + // Where the numbers come from: + // - key_window_size_ms=2000, key_slide_interval_ms=1000 are the + // scenario's fixed inputs (the Sliding shape #600 is about). + // - The bucket must be genuinely window_size_ms wide (2000) -- real + // Sliding data reaches the store already pre-merged by + // worker.rs::merge_panes_for_window, one window_size_ms-wide entry + // per window, never a raw slide-width pane -- while its *start* + // must be off the window_size_ms grid ({0,2000,4000,...}) but on + // the slide_interval_ms grid ({0,1000,2000,...}). The smallest + // such start is 1000, but 3000 is used instead because a bucket + // starting at 1000 with width 2000 ends at 3000, which also works + // -- 3000 was simply picked without checking whether 1000 (with a + // smaller timestamp/query shift) would have worked equally well. + // - keys_data/value_data timestamp=5000 makes the factory (which + // computes each bucket as [timestamp - width, timestamp)) place + // the keys bucket at exactly [3000,5000). + // - The query (5.0s-5.5s, step=1.0s) produces one output step at + // t=5000, whose keys lookback window is + // [5000 - key_window_size_ms, 5000) = [3000,5000) -- lining up + // exactly with the inserted bucket. + // - value_data is also placed at timestamp=5000 (Tumbling, 1000ms + // wide -> bucket [4000,5000)) purely so the CountMinSketch value + // side resolves at the same t=5000 step; it's unrelated to #600. let mut keys_add = SetAggregatorAccumulator::new(); keys_add.add_key(KeyByLabelValues { labels: vec!["host-a".to_string(), "evt-1".to_string()], @@ -399,13 +427,15 @@ mod tests { vec![], vec!["host", "event"], vec![( - 2000, + 5000, 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)], + // Keys bucket spans [3000, 5000) -- window_size_ms=2000 wide (a + // real pre-merged window, matching worker.rs's output shape), + // starting at 3000: on the slide_interval_ms=1000 grid, but not + // on the window_size_ms=2000 grid ({0, 2000, 4000, ...}). + vec![(5000, 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 @@ -413,13 +443,13 @@ mod tests { ); 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 result = engine.handle_range_query_promql(query.to_string(), 5.0, 5.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 \ + key_has_sample_at(&elements, "host-a", 5000), + "BUG #600: host-a's keys delta bucket (start=3000, 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" @@ -602,6 +632,7 @@ mod tests { data, query, 1_000, // window_size_ms, matches the fixed 1000ms bucket width + 1_000, // slide_interval_ms WindowType::Sliding, ); @@ -647,6 +678,7 @@ mod tests { data, query, 1_000, + 1_000, // slide_interval_ms WindowType::Sliding, ); @@ -682,6 +714,7 @@ mod tests { data, query, 1_000, + 1_000, // slide_interval_ms WindowType::Sliding, ); @@ -727,6 +760,7 @@ mod tests { data, query, 2000, // window_size_ms + 1000, // slide_interval_ms WindowType::Sliding, ); diff --git a/asap-query-engine/src/tests/test_utilities/engine_factories.rs b/asap-query-engine/src/tests/test_utilities/engine_factories.rs index b3c54f5..6433dc0 100644 --- a/asap-query-engine/src/tests/test_utilities/engine_factories.rs +++ b/asap-query-engine/src/tests/test_utilities/engine_factories.rs @@ -21,6 +21,13 @@ use std::sync::Arc; /// Data to insert into a store: (label_values, accumulator) pub type AccumulatorData = Vec<(Option>, Box)>; +/// Window size (ms) for the fixed-window (Tumbling-only) engine builders in +/// this file. Not used by `create_engine_multi_timestamp_with_window`, which +/// takes its own `window_size_ms`/`slide_interval_ms` -- Sliding needs those +/// independently configurable, but every Tumbling-only builder here always +/// wants slide == window_size == this one value. +const TEST_TUMBLING_WINDOW_SIZE_MS: u64 = 1000; + /// Creates a SimpleEngine with a single aggregation populated with given data. /// /// # Arguments @@ -79,8 +86,8 @@ pub fn create_engine_single_pop_with_aggregated( aggregated_labels: KeyByLabelNames::new(aggregated_label_strings), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, + window_size_ms: TEST_TUMBLING_WINDOW_SIZE_MS, + slide_interval_ms: TEST_TUMBLING_WINDOW_SIZE_MS, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -105,7 +112,8 @@ pub fn create_engine_single_pop_with_aggregated( let timestamp = 1_000_000_u64; for (label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp, timestamp, key, 1); + let output = + PrecomputedOutput::new(timestamp - TEST_TUMBLING_WINDOW_SIZE_MS, timestamp, key, 1); store.insert_precomputed_output(output, acc).unwrap(); } @@ -127,7 +135,7 @@ pub fn create_engine_single_pop_with_aggregated( // None, inference_config, streaming_config, - 1000, + TEST_TUMBLING_WINDOW_SIZE_MS, QueryLanguage::promql, ) } @@ -176,8 +184,8 @@ pub fn create_engine_dual_input( aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, + window_size_ms: TEST_TUMBLING_WINDOW_SIZE_MS, + slide_interval_ms: TEST_TUMBLING_WINDOW_SIZE_MS, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -199,8 +207,8 @@ pub fn create_engine_dual_input( aggregated_labels: KeyByLabelNames::new(aggregated_label_strings), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, + window_size_ms: TEST_TUMBLING_WINDOW_SIZE_MS, + slide_interval_ms: TEST_TUMBLING_WINDOW_SIZE_MS, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -225,14 +233,16 @@ pub fn create_engine_dual_input( let timestamp = 1_000_000_u64; for (label_values_opt, acc) in value_data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp, timestamp, key, 1); + let output = + PrecomputedOutput::new(timestamp - TEST_TUMBLING_WINDOW_SIZE_MS, timestamp, key, 1); store.insert_precomputed_output(output, acc).unwrap(); } // Insert keys data for (label_values_opt, acc) in keys_data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp, timestamp, key, 2); + let output = + PrecomputedOutput::new(timestamp - TEST_TUMBLING_WINDOW_SIZE_MS, timestamp, key, 2); store.insert_precomputed_output(output, acc).unwrap(); } @@ -255,7 +265,7 @@ pub fn create_engine_dual_input( // None, inference_config, streaming_config, - 1000, + TEST_TUMBLING_WINDOW_SIZE_MS, QueryLanguage::promql, ) } @@ -292,8 +302,8 @@ pub fn create_engine_two_metrics( aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, + window_size_ms: TEST_TUMBLING_WINDOW_SIZE_MS, + slide_interval_ms: TEST_TUMBLING_WINDOW_SIZE_MS, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -314,8 +324,8 @@ pub fn create_engine_two_metrics( aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, + window_size_ms: TEST_TUMBLING_WINDOW_SIZE_MS, + slide_interval_ms: TEST_TUMBLING_WINDOW_SIZE_MS, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -339,12 +349,14 @@ pub fn create_engine_two_metrics( let timestamp = 1_000_000_u64; for (label_values_opt, acc) in data_a { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp, timestamp, key, 1); + let output = + PrecomputedOutput::new(timestamp - TEST_TUMBLING_WINDOW_SIZE_MS, timestamp, key, 1); store.insert_precomputed_output(output, acc).unwrap(); } for (label_values_opt, acc) in data_b { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp, timestamp, key, 2); + let output = + PrecomputedOutput::new(timestamp - TEST_TUMBLING_WINDOW_SIZE_MS, timestamp, key, 2); store.insert_precomputed_output(output, acc).unwrap(); } @@ -368,7 +380,7 @@ pub fn create_engine_two_metrics( store, inference_config, streaming_config, - 1000, + TEST_TUMBLING_WINDOW_SIZE_MS, QueryLanguage::promql, ) } @@ -417,8 +429,8 @@ pub fn create_engine_three_metrics( aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, + window_size_ms: TEST_TUMBLING_WINDOW_SIZE_MS, + slide_interval_ms: TEST_TUMBLING_WINDOW_SIZE_MS, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -444,7 +456,12 @@ pub fn create_engine_three_metrics( for (agg_id, data) in [(1u64, data_a), (2u64, data_b), (3u64, data_c)] { for (label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp, timestamp, key, agg_id); + let output = PrecomputedOutput::new( + timestamp - TEST_TUMBLING_WINDOW_SIZE_MS, + timestamp, + key, + agg_id, + ); store.insert_precomputed_output(output, acc).unwrap(); } } @@ -471,7 +488,7 @@ pub fn create_engine_three_metrics( store, inference_config, streaming_config, - 1000, + TEST_TUMBLING_WINDOW_SIZE_MS, QueryLanguage::promql, ) } @@ -498,8 +515,8 @@ pub fn create_engine_multi_timestamp( aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, + window_size_ms: TEST_TUMBLING_WINDOW_SIZE_MS, + slide_interval_ms: TEST_TUMBLING_WINDOW_SIZE_MS, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -522,7 +539,8 @@ pub fn create_engine_multi_timestamp( for (timestamp, label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp - 1000, timestamp, key, 1); + let output = + PrecomputedOutput::new(timestamp - TEST_TUMBLING_WINDOW_SIZE_MS, timestamp, key, 1); store.insert_precomputed_output(output, acc).unwrap(); } @@ -545,7 +563,7 @@ pub fn create_engine_multi_timestamp( // None, inference_config, streaming_config, - 1000, + TEST_TUMBLING_WINDOW_SIZE_MS, QueryLanguage::promql, ) } @@ -563,6 +581,7 @@ pub fn create_engine_multi_timestamp_with_window( data: Vec<(u64, Option>, Box)>, promql_query: &str, window_size_ms: u64, + slide_interval_ms: u64, window_type: WindowType, ) -> SimpleEngine { let grouping_label_strings: Vec = @@ -579,7 +598,7 @@ pub fn create_engine_multi_timestamp_with_window( rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), window_size_ms, - slide_interval_ms: 1000, + slide_interval_ms, window_type, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -600,10 +619,62 @@ pub fn create_engine_multi_timestamp_with_window( CleanupPolicy::NoCleanup, )); + // `data` entries are panes (width = slide_interval_ms, keyed by their end + // timestamp), not pre-merged windows. For Tumbling (window_size == slide) + // each pane already IS a full window. For Sliding (window_size > slide) a + // window is `window_size_ms / slide_interval_ms` consecutive panes merged + // together -- mirroring `worker.rs::merge_panes_for_window`, which is how + // real Sliding-window data reaches the store (always pre-merged; the + // store never holds raw sub-window panes). Windows missing a pane are + // skipped (sparse), not padded. + let slide_ms = slide_interval_ms; + let num_panes = window_size_ms / slide_ms; + + let mut per_key: HashMap>, Vec<(u64, Box)>> = + HashMap::new(); for (timestamp, label_values_opt, acc) in data { + per_key + .entry(label_values_opt) + .or_default() + .push((timestamp, acc)); + } + + for (label_values_opt, mut panes) in per_key { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp - 1000, timestamp, key, 1); - store.insert_precomputed_output(output, acc).unwrap(); + panes.sort_by_key(|(ts, _)| *ts); + + if num_panes <= 1 { + for (ts, acc) in panes { + let output = PrecomputedOutput::new(ts - window_size_ms, ts, key.clone(), 1); + store.insert_precomputed_output(output, acc).unwrap(); + } + continue; + } + + let pane_map: HashMap> = + panes.iter().map(|(ts, acc)| (*ts, acc)).collect(); + let (min_ts, max_ts) = (panes[0].0, panes[panes.len() - 1].0); + + let mut window_start = min_ts.saturating_sub(window_size_ms); + while window_start + window_size_ms <= max_ts { + let pane_ends: Vec = (1..=num_panes) + .map(|i| window_start + i * slide_ms) + .collect(); + if pane_ends.iter().all(|t| pane_map.contains_key(t)) { + let mut merged = pane_map[&pane_ends[0]].clone_boxed_core(); + for t in &pane_ends[1..] { + merged = merged.merge_with(pane_map[t].as_ref()).unwrap(); + } + let output = PrecomputedOutput::new( + window_start, + window_start + window_size_ms, + key.clone(), + 1, + ); + store.insert_precomputed_output(output, merged).unwrap(); + } + window_start += slide_ms; + } } let promql_schema = PromQLSchema::new().add_metric( @@ -625,7 +696,7 @@ pub fn create_engine_multi_timestamp_with_window( // None, inference_config, streaming_config, - 1000, + slide_interval_ms, QueryLanguage::promql, ) }