diff --git a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs index 1c55e6c..2a6f3de 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -74,11 +74,47 @@ pub fn key_agg_window_valid(agg_type: AggregationType, window_type: WindowType) !(agg_type == AggregationType::DeltaSetAggregator && window_type == WindowType::Sliding) } +fn effective_grid_step(config: &AggregationConfig) -> u64 { + if config.slide_interval_ms == 0 { + config.window_size_ms + } else { + config.slide_interval_ms + } +} + +/// Whether a separate key aggregation can resolve the populations of a +/// value aggregation on the same epoch-zero window grid. +pub fn key_agg_compatible_with_value(value: &AggregationConfig, key: &AggregationConfig) -> bool { + if !key_agg_window_valid(key.aggregation_type, key.window_type) { + return false; + } + + match key.aggregation_type { + AggregationType::SetAggregator => { + key.window_type == value.window_type + && key.window_size_ms == value.window_size_ms + && effective_grid_step(key) == effective_grid_step(value) + } + AggregationType::DeltaSetAggregator if value.window_type == WindowType::Sliding => { + let value_slide_ms = value.slide_interval_ms; + let delta_window_ms = key.window_size_ms; + key.window_type == WindowType::Tumbling + && value_slide_ms > 0 + && delta_window_ms > 0 + && value_slide_ms.is_multiple_of(delta_window_ms) + && value.window_size_ms.is_multiple_of(delta_window_ms) + } + AggregationType::DeltaSetAggregator => key.window_type == WindowType::Tumbling, + _ => false, + } +} + /// Window compatibility: can `config` serve a query needing `data_range_ms`? /// -/// - Tumbling: `data_range_ms` must be a positive integer multiple of `window_size_ms`. -/// - Sliding: `data_range_ms` must equal `window_size_ms` exactly (a sliding window -/// precomputes one fixed range per timestamp; overlapping windows cannot be merged). +/// Both Tumbling and Sliding require `data_range_ms` to be a positive integer +/// multiple of `window_size_ms`. Sliding execution selects a non-overlapping, +/// `window_size_ms`-spaced subset from the denser slide grid (#554); it must +/// never merge every overlapping window on that grid. pub fn window_compatible(config: &AggregationConfig, data_range_ms: u64) -> bool { if !key_agg_window_valid(config.aggregation_type, config.window_type) { return false; @@ -88,7 +124,11 @@ pub fn window_compatible(config: &AggregationConfig, data_range_ms: u64) -> bool return false; } match config.window_type { - WindowType::Sliding => data_range_ms == window_ms, + WindowType::Sliding => { + config.slide_interval_ms > 0 + && window_ms.is_multiple_of(config.slide_interval_ms) + && data_range_ms.is_multiple_of(window_ms) + } WindowType::Tumbling => data_range_ms.is_multiple_of(window_ms), } } @@ -279,7 +319,7 @@ pub fn find_compatible_aggregation( if c.metric != requirements.metric || !is_key_agg_type(c.aggregation_type) { return false; } - if key_agg_window_valid(c.aggregation_type, c.window_type) { + if key_agg_compatible_with_value(value_agg, c) { true } else { invalid_window.get_or_insert(c); @@ -388,6 +428,18 @@ mod tests { } } + fn assert_key_compatibility_cases( + cases: &[(&str, &AggregationConfig, &AggregationConfig, bool)], + ) { + for (name, value, key, expected) in cases { + assert_eq!( + key_agg_compatible_with_value(value, key), + *expected, + "{name}" + ); + } + } + fn single_config(config: AggregationConfig) -> HashMap { let mut m = HashMap::new(); m.insert(config.aggregation_id, config); @@ -565,8 +617,8 @@ mod tests { } #[test] - fn window_sliding_too_large() { - // Query range 600_000 ms but sliding window only covers 300_000 ms + fn window_sliding_wider_exact_multiple_is_compatible() { + // Two non-overlapping stored 300_000ms windows exactly cover the query. let configs = single_config(make_config( 1, "cpu", @@ -579,7 +631,7 @@ mod tests { )); let result = find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 600_000, &[], "")); - assert!(result.is_none()); + assert!(result.is_some()); } #[test] @@ -900,6 +952,39 @@ mod tests { assert!(window_compatible(&config, 300_000)); } + #[test] + fn window_compatible_rejects_sliding_window_not_aligned_to_slide() { + let mut config = make_config(1, "req", "SetAggregator", "", 300_000, "sliding", &[], ""); + config.slide_interval_ms = 40_000; + assert!(!window_compatible(&config, 600_000)); + } + + #[test] + fn sliding_window_compatibility_boundary_matrix() { + let mut config = make_config(1, "req", "Sum", "", 5_000, "sliding", &[], ""); + config.slide_interval_ms = 1_000; + assert!(window_compatible(&config, 5_000)); + assert!(window_compatible(&config, 10_000)); + assert!(!window_compatible(&config, 6_000)); + + config.slide_interval_ms = 2_000; + assert!(!window_compatible(&config, 10_000)); + + config.window_size_ms = 1_000; + config.slide_interval_ms = 5_000; + assert!(!window_compatible(&config, 1_000)); + } + + #[test] + fn sliding_window_compatibility_rejects_zero_fields() { + let mut config = make_config(1, "req", "Sum", "", 5_000, "sliding", &[], ""); + config.slide_interval_ms = 0; + assert!(!window_compatible(&config, 5_000)); + config.slide_interval_ms = 1_000; + config.window_size_ms = 0; + assert!(!window_compatible(&config, 5_000)); + } + #[test] fn window_compatible_still_accepts_tumbling_delta_set_aggregator() { let config = make_config( @@ -971,7 +1056,7 @@ mod tests { "CountMinSketchWithHeap", "", 300_000, - "tumbling", + "sliding", &[], "", ), @@ -984,10 +1069,177 @@ mod tests { &configs, &req("req", &[Statistic::Topk], 300_000, &[], ""), ); - let info = result.expect("Sliding SetAggregator must still be accepted as a key agg"); + let info = + result.expect("Sliding SetAggregator on the value aggregation's grid must be accepted"); assert_eq!(info.aggregation_id_for_key, 11); } + #[test] + fn multi_pop_rejects_tumbling_delta_set_that_cannot_partition_sliding_value_window() { + let mut value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + value.slide_interval_ms = 1_000; + let delta_keys = make_config( + 11, + "req", + "DeltaSetAggregator", + "", + 2_000, + "tumbling", + &[], + "", + ); + let configs = HashMap::from([(10, value), (11, delta_keys)]); + + assert!( + find_compatible_aggregation( + &configs, + &req("req", &[Statistic::Count], 12_000, &[], ""), + ) + .is_none(), + "D=2s would include future events at an S=1s boundary" + ); + } + + #[test] + fn multi_pop_accepts_tumbling_delta_set_that_partitions_sliding_value_grid() { + let mut value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + value.slide_interval_ms = 1_000; + let delta_keys = make_config( + 11, + "req", + "DeltaSetAggregator", + "", + 1_000, + "tumbling", + &[], + "", + ); + let configs = HashMap::from([(10, value), (11, delta_keys)]); + + let result = find_compatible_aggregation( + &configs, + &req("req", &[Statistic::Count], 12_000, &[], ""), + ) + .expect("D=1s lies on S=1s and exactly partitions W=6s"); + + assert_eq!(result.aggregation_id_for_key, 11); + } + + #[test] + fn multi_pop_rejects_tumbling_set_key_on_mismatched_nonzero_grid_step() { + let value = make_config(10, "req", "CountMinSketch", "", 5_000, "tumbling", &[], ""); + let mut keys = make_config(11, "req", "SetAggregator", "", 5_000, "tumbling", &[], ""); + keys.slide_interval_ms = 1_000; + let configs = HashMap::from([(10, value), (11, keys)]); + + assert!(find_compatible_aggregation( + &configs, + &req("req", &[Statistic::Count], 5_000, &[], "") + ) + .is_none()); + } + + #[test] + fn tumbling_set_pairing_normalizes_zero_slide_to_window_size() { + let mut value = make_config(10, "req", "CountMinSketch", "", 5_000, "tumbling", &[], ""); + let mut key = make_config(11, "req", "SetAggregator", "", 5_000, "tumbling", &[], ""); + value.slide_interval_ms = 0; + key.slide_interval_ms = 0; + assert!(key_agg_compatible_with_value(&value, &key)); + key.slide_interval_ms = 5_000; + assert!(key_agg_compatible_with_value(&value, &key)); + } + + #[test] + fn set_pairing_rejects_each_grid_mismatch_dimension() { + let value = make_config(10, "req", "CountMinSketch", "", 5_000, "sliding", &[], ""); + let mut key = make_config(11, "req", "SetAggregator", "", 5_000, "sliding", &[], ""); + key.slide_interval_ms = 1_000; + assert!(!key_agg_compatible_with_value(&value, &key)); + key.slide_interval_ms = 5_000; + key.window_size_ms = 10_000; + assert!(!key_agg_compatible_with_value(&value, &key)); + key.window_size_ms = 5_000; + key.window_type = WindowType::Tumbling; + assert!(!key_agg_compatible_with_value(&value, &key)); + } + + #[test] + fn delta_set_pairing_truth_table_checks_both_divisors() { + let mut value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + value.slide_interval_ms = 2_000; + let key_valid = make_config( + 11, + "req", + "DeltaSetAggregator", + "", + 2_000, + "tumbling", + &[], + "", + ); + let key_bad_window = AggregationConfig { + window_size_ms: 3_000, + ..key_valid.clone() + }; + let key_bad_both = AggregationConfig { + window_size_ms: 4_000, + ..key_valid.clone() + }; + let mut value_bad_step = value.clone(); + value_bad_step.slide_interval_ms = 3_000; + let mut value_bad_window = value.clone(); + value_bad_window.slide_interval_ms = 4_000; + value_bad_window.window_size_ms = 6_000; + let key_divides_step_not_window = AggregationConfig { + window_size_ms: 4_000, + ..key_valid.clone() + }; + + assert_key_compatibility_cases(&[ + ("D divides S and W", &value, &key_valid, true), + ("D does not divide W", &value, &key_bad_window, false), + ("D divides neither S nor W", &value, &key_bad_both, false), + ("D does not divide S", &value_bad_step, &key_valid, false), + ( + "D divides S but not W", + &value_bad_window, + &key_divides_step_not_window, + false, + ), + ]); + } + + #[test] + fn delta_set_pairing_for_tumbling_values_does_not_apply_sliding_rules() { + let value = make_config(10, "req", "CountMinSketch", "", 6_000, "tumbling", &[], ""); + let key = make_config( + 11, + "req", + "DeltaSetAggregator", + "", + 1_000, + "tumbling", + &[], + "", + ); + assert!(key_agg_compatible_with_value(&value, &key)); + } + + #[test] + fn matching_skips_incompatible_key_candidate_and_selects_compatible_one() { + let value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + let mut incompatible = + make_config(11, "req", "SetAggregator", "", 6_000, "sliding", &[], ""); + incompatible.slide_interval_ms = 2_000; + let compatible = make_config(12, "req", "SetAggregator", "", 6_000, "sliding", &[], ""); + let configs = HashMap::from([(10, value), (11, incompatible), (12, compatible)]); + let result = + find_compatible_aggregation(&configs, &req("req", &[Statistic::Count], 6_000, &[], "")) + .expect("the compatible key candidate should be selected"); + assert_eq!(result.aggregation_id_for_key, 12); + } + // --- avg (Vec) --- #[test] diff --git a/asap-planner-rs/src/planner/cleanup.rs b/asap-planner-rs/src/planner/cleanup.rs index 4f07d75..5fab2de 100644 --- a/asap-planner-rs/src/planner/cleanup.rs +++ b/asap-planner-rs/src/planner/cleanup.rs @@ -9,11 +9,14 @@ use super::window::get_effective_repeat; /// (`data_range_ms >= t_repeat_ms >= data_ingestion_interval_ms`), this is /// exactly equivalent to the old pattern-type-gated `t_repeat_ms`-vs-range-duration /// split, with no shape check needed (see #508). +#[allow(clippy::too_many_arguments)] pub fn get_cleanup_param( cleanup_policy: CleanupPolicy, data_range_ms: u64, t_repeat_ms: u64, window_type: WindowType, + window_size_ms: u64, + slide_interval_ms: u64, range_duration_ms: u64, step_ms: u64, ) -> Result { @@ -29,12 +32,32 @@ pub fn get_cleanup_param( let t_lookback: u64 = data_range_ms; if window_type == WindowType::Sliding { - let result = if is_range_query { + if cleanup_policy == CleanupPolicy::NoCleanup { + return Err("NoCleanup policy should not call get_cleanup_param".to_string()); + } + if window_size_ms == 0 || slide_interval_ms == 0 { + return Err("Sliding cleanup requires positive window and slide sizes".to_string()); + } + if !data_range_ms.is_multiple_of(window_size_ms) { + return Err(format!( + "Sliding query lookback ({data_range_ms}ms) must be a multiple of window_size_ms ({window_size_ms}ms)" + )); + } + let num_steps = if is_range_query { range_duration_ms / step_ms + 1 } else { 1 }; - return Ok(result); + return match cleanup_policy { + CleanupPolicy::ReadBased => (data_range_ms / window_size_ms) + .checked_mul(num_steps) + .ok_or_else(|| "Sliding read-count cleanup threshold overflowed".to_string()), + CleanupPolicy::CircularBuffer => data_range_ms + .checked_add(range_duration_ms) + .map(|span_ms| span_ms.div_ceil(slide_interval_ms)) + .ok_or_else(|| "Sliding circular-buffer retention span overflowed".to_string()), + CleanupPolicy::NoCleanup => unreachable!(), + }; } // Tumbling @@ -101,6 +124,8 @@ mod tests { 300_000, 300_000, WindowType::Tumbling, + 300_000, + 300_000, 0, 0, ) @@ -117,6 +142,8 @@ mod tests { 300_000, 300_000, WindowType::Tumbling, + 30_000, + 30_000, 3_600_000, 30_000, ) @@ -132,6 +159,8 @@ mod tests { 300_000, 300_000, WindowType::Tumbling, + 300_000, + 300_000, 0, 0, ) @@ -148,6 +177,8 @@ mod tests { 300_000, 300_000, WindowType::Tumbling, + 30_000, + 30_000, 3_600_000, 30_000, ) @@ -164,6 +195,8 @@ mod tests { 300_000, 60_000, WindowType::Tumbling, + 60_000, + 60_000, 0, 0, ) @@ -178,6 +211,8 @@ mod tests { 300_000, 300_000, WindowType::Tumbling, + 300_000, + 300_000, 0, 0, ); @@ -192,9 +227,80 @@ mod tests { 300_000, 300_000, WindowType::Tumbling, + 300_000, + 300_000, 3_600_000, 0, ); assert!(result.is_err()); } + + #[test] + fn cleanup_param_read_based_sliding_counts_each_constituent_window() { + let result = get_cleanup_param( + CleanupPolicy::ReadBased, + 10_000, + 5_000, + WindowType::Sliding, + 5_000, + 1_000, + 0, + 0, + ) + .unwrap(); + + assert_eq!(result, 2); + } + + #[test] + fn cleanup_param_circular_sliding_retains_cover_span() { + let result = get_cleanup_param( + CleanupPolicy::CircularBuffer, + 10_000, + 5_000, + WindowType::Sliding, + 5_000, + 1_000, + 5_000, + 1_000, + ) + .unwrap(); + assert_eq!(result, 15); + } + + #[test] + fn cleanup_param_sliding_rejects_invalid_shape_and_sizes() { + for (window_size_ms, slide_interval_ms, lookback_ms) in [ + (0, 1_000, 10_000), + (5_000, 0, 10_000), + (6_000, 1_000, 10_000), + ] { + assert!(get_cleanup_param( + CleanupPolicy::ReadBased, + lookback_ms, + 5_000, + WindowType::Sliding, + window_size_ms, + slide_interval_ms, + 0, + 0, + ) + .is_err()); + } + } + + #[test] + fn cleanup_param_no_cleanup_sliding_is_rejected() { + assert!(get_cleanup_param( + CleanupPolicy::NoCleanup, + 10_000, + 5_000, + WindowType::Sliding, + 5_000, + 1_000, + 0, + 0, + ) + .is_err()); + } } diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs index d47b9a9..7257e55 100644 --- a/asap-planner-rs/src/planner/promql.rs +++ b/asap-planner-rs/src/planner/promql.rs @@ -289,6 +289,8 @@ impl SingleQueryProcessor { requirements.data_range_ms, self.t_repeat_ms, window_cfg.window_type, + window_cfg.window_size_ms, + window_cfg.slide_interval_ms, self.range_duration_ms, self.step_ms, ) diff --git a/asap-query-engine/src/engines/mod.rs b/asap-query-engine/src/engines/mod.rs index b81b5ac..65da465 100644 --- a/asap-query-engine/src/engines/mod.rs +++ b/asap-query-engine/src/engines/mod.rs @@ -1,6 +1,7 @@ pub(crate) mod merge_utils; pub mod query_result; pub mod simple_engine; +pub(crate) mod sliding_window_composition; pub mod window_merger; pub use query_result::{InstantVector, QueryResult, RangeVector, RangeVectorElement, Sample}; diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 4e583ea..43d2dda 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -7,11 +7,12 @@ use crate::data_model::{ StreamingConfig, }; use crate::engines::query_result::{InstantVectorElement, QueryResult}; +use crate::engines::sliding_window_composition::{plan_exact_cover, SlidingWindowSpec}; // use crate::stores::promsketch_store::{ // self, is_usampling_function, metrics as ps_metrics, PromSketchStore, // }; use crate::stores::{Store, TimestampedBucketsMap}; -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; use std::sync::{Arc, RwLock}; use std::time::Instant; use tracing::{debug, warn}; @@ -19,6 +20,7 @@ use tracing::{debug, warn}; use crate::precompute_operators::AccumulatorError; use crate::AggregateCore; +use asap_types::capability_matching::key_agg_compatible_with_value; use asap_types::enums::WindowType; use promql_utilities::ast_matching::{PromQLPattern, PromQLPatternBuilder}; use promql_utilities::data_model::KeyByLabelNames; @@ -114,23 +116,20 @@ pub struct RangeQueryExecutionContext { /// Tumbling window size in ms pub tumbling_window_ms: u64, /// The value aggregation's `WindowType`. Picks how the per-step loop - /// composes a step's window from `bucket_map`: Sliding buckets are each - /// already a complete `window_size_ms`-wide merged window (see - /// `worker.rs::merge_panes_for_window`), so a step takes exactly the one - /// bucket at `current_time - lookback_ms` (`lookback_ms` == - /// `window_size_ms` here); Tumbling buckets are genuinely disjoint, so a - /// step sums every bucket `sum_window` finds across the lookback span - /// (#608). + /// composes a step's window from `bucket_map`: Sliding selects a + /// non-overlapping `window_size_ms`-spaced exact cover from the denser + /// slide grid (#554); Tumbling sums every disjoint bucket in the + /// lookback span. pub window_type: WindowType, /// The value aggregation's actual `window_size_ms`, independent of /// `tumbling_window_ms` (which is `bucket_step_ms`, not the window - /// size). Used only to assert `lookback_ms == window_size_ms` for - /// Sliding before `single_window` relies on that equality (#608 review). + /// size). For Sliding, this is the stride between constituent windows in + /// an exact cover; `tumbling_window_ms` remains the stored grid stride. pub window_size_ms: u64, /// Same as `window_type`, for the keys aggregation -- `None` when /// there's no separate `keys_query`. Can legitimately differ from - /// `window_type` (e.g. a Sliding SetAggregator keys aggregation paired - /// with a Tumbling value aggregation, or vice versa). + /// `window_type` for the Tumbling DeltaSetAggregator exception; a + /// SetAggregator must share the value aggregation's complete grid. pub keys_window_type: Option, /// Same as `window_size_ms`, for the keys aggregation. `None` under the /// same condition as `keys_window_type`. @@ -386,6 +385,7 @@ impl SimpleEngine { fn create_keys_query_params( &self, metric: &str, + query_start_timestamp: u64, end_timestamp: u64, agg_info: &AggregationIdInfo, ) -> Result { @@ -395,31 +395,19 @@ impl SimpleEngine { (0, end_timestamp) } AggregationType::SetAggregator => { - // Latest window only - let window_size = self - .streaming_config - .read() - .unwrap() - .get_aggregation_config(agg_info.aggregation_id_for_key) - .map(|config| config.window_size_ms) - .ok_or_else(|| { - format!( - "Failed to get window size for aggregation {}", - agg_info.aggregation_id_for_key - ) - })?; - (end_timestamp - window_size, end_timestamp) + // Resolve keys over the query's full requested lookback. A + // Sliding SetAggregator uses the same exact-cover planner as + // the value side; a Tumbling one uses the normal grid walk. + (query_start_timestamp, end_timestamp) } other => { return Err(format!("Unsupported key aggregation type: {other:?}")); } }; - // 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." + // Fetch strategy is selected later from the key config: Sliding + // SetAggregator uses the exact-cover path; Tumbling SetAggregator + // and DeltaSetAggregator use the ordinary grid/range path. Ok(StoreQueryParams { metric: metric.to_string(), aggregation_id: agg_info.aggregation_id_for_key, @@ -457,30 +445,43 @@ impl SimpleEngine { 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. 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) - } else { - // Tumbling window: range query - (timestamps.start_timestamp, timestamps.end_timestamp) - }; - let values_query = StoreQueryParams { metric: metric.to_string(), aggregation_id: agg_info.aggregation_id_for_value, - start_timestamp: values_start, - end_timestamp: values_end, + start_timestamp: timestamps.start_timestamp, + end_timestamp: timestamps.end_timestamp, }; // Determine if we need a separate keys query let keys_query = if agg_info.aggregation_id_for_key != agg_info.aggregation_id_for_value { - Some(self.create_keys_query_params(metric, timestamps.end_timestamp, agg_info)?) + let key_config = sc + .get_aggregation_config(agg_info.aggregation_id_for_key) + .ok_or_else(|| { + format!( + "Aggregation config not found for key aggregation_id: {}", + agg_info.aggregation_id_for_key + ) + })?; + if !key_agg_compatible_with_value(aggregation_config_for_value, key_config) { + return Err(format!( + "Key aggregation {} ({:?}, W={}ms, S={}ms) is not grid-compatible with \ + value aggregation {} ({:?}, W={}ms, S={}ms)", + key_config.aggregation_id, + key_config.window_type, + key_config.window_size_ms, + Self::bucket_step_ms(key_config), + aggregation_config_for_value.aggregation_id, + aggregation_config_for_value.window_type, + aggregation_config_for_value.window_size_ms, + Self::bucket_step_ms(aggregation_config_for_value), + )); + } + Some(self.create_keys_query_params( + metric, + timestamps.start_timestamp, + timestamps.end_timestamp, + agg_info, + )?) } else { None }; @@ -514,6 +515,20 @@ impl SimpleEngine { } } + fn align_down_with_warning(timestamp_ms: u64, step_ms: u64, context: &str) -> u64 { + let aligned = timestamp_ms - (timestamp_ms % step_ms); + if aligned != timestamp_ms { + warn!( + timestamp_ms, + aligned_timestamp_ms = aligned, + step_ms, + context, + "Timestamp was aligned down; the requested timestamp will not be used as-is" + ); + } + aligned + } + /// Widens `query`'s window to `[start_ms - lookback, end_ms]`, where /// `lookback` is the width `query` already had (`end_timestamp - /// start_timestamp`) before this call. Re-anchors whatever window an @@ -587,10 +602,20 @@ impl SimpleEngine { ); let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; - let keys_lookback_ms = extended_store_plan - .keys_query - .as_mut() - .map(|keys_query| Self::widen_query_window(keys_query, query_time, query_time)); + let keys_lookback_ms = extended_store_plan.keys_query.as_mut().map(|keys_query| { + if base_context.agg_info.aggregation_type_for_key == AggregationType::DeltaSetAggregator + { + // DeltaSetAggregator replays from epoch zero. Re-anchoring + // its [0, aligned_end) span at a misaligned evaluation time + // would move the start forward and silently discard early + // deltas. + keys_query.start_timestamp = 0; + keys_query.end_timestamp = query_time; + query_time + } else { + Self::widen_query_window(keys_query, query_time, query_time) + } + }); let (keys_tumbling_window_ms, keys_window_type, keys_window_size_ms) = match keys_lookback_ms { Some(_) => { @@ -634,11 +659,10 @@ impl SimpleEngine { /// 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. + /// with an exact match, merging the sparse per-window results. Sliding + /// execution uses `execute_sliding_cover_query` instead so it requests + /// only the non-overlapping W-spaced subset rather than this entire + /// dense S-grid. fn fetch_window_grid_via_exact_lookups( &self, params: &StoreQueryParams, @@ -688,6 +712,14 @@ impl SimpleEngine { let mut windows: Vec = Vec::new(); let mut window_start = params.start_timestamp.div_ceil(step_ms) * step_ms; + if window_start != params.start_timestamp { + warn!( + requested_start_timestamp = params.start_timestamp, + aligned_start_timestamp = window_start, + step_ms, + "Window-grid query start was aligned up; the requested timestamp will not be used as-is" + ); + } while window_start + window_size_ms <= params.end_timestamp { windows.push((window_start, window_start + window_size_ms)); window_start += step_ms; @@ -733,6 +765,78 @@ impl SimpleEngine { result } + fn execute_sliding_cover_query( + &self, + params: &StoreQueryParams, + output_timestamps: &[u64], + lookback_ms: u64, + window_size_ms: u64, + slide_interval_ms: u64, + ) -> Result { + let spec = SlidingWindowSpec { + window_size_ms, + slide_interval_ms, + }; + // Range-step covers overlap heavily. Keep one sorted set so each + // exact stored window is fetched once, in deterministic order, even + // when multiple output timestamps require it. + let mut windows = BTreeSet::new(); + for &output_timestamp in output_timestamps { + let cover = plan_exact_cover(output_timestamp, lookback_ms, spec).map_err(|error| { + format!( + "Cannot compose Sliding aggregation {} for lookback {}ms (W={}ms, S={}ms): {:?}", + params.aggregation_id, lookback_ms, window_size_ms, slide_interval_ms, error + ) + })?; + windows.extend(cover.windows); + } + let windows: Vec<_> = windows.into_iter().collect(); + debug!( + aggregation_id = params.aggregation_id, + lookback_ms, + window_size_ms, + slide_interval_ms, + window_count = windows.len(), + "Querying exact non-overlapping Sliding-window cover" + ); + let outputs = self + .store + .query_precomputed_output_exact_batch(¶ms.metric, params.aggregation_id, &windows) + .map_err(|error| { + format!( + "Error querying store for metric {}, agg {}, {} exact Sliding windows: {}", + params.metric, + params.aggregation_id, + windows.len(), + error + ) + })?; + + // An instant query has one all-or-nothing cover. Range queries defer + // completeness to each output step below, so a series that appears + // partway through the range does not invalidate unrelated steps. + if output_timestamps.len() == 1 { + let required: BTreeSet<_> = windows.iter().copied().collect(); + for (group_key, buckets) in &outputs { + let found: BTreeSet<_> = buckets.iter().map(|(range, _)| *range).collect(); + if found != required { + let missing: Vec<_> = required.difference(&found).copied().collect(); + return Err(format!( + "Incomplete Sliding-window cover for metric {}, agg {}, group {:?}: \ + requested {} exact windows, missing {:?}", + params.metric, + params.aggregation_id, + group_key, + windows.len(), + missing + )); + } + } + } + + Ok(outputs) + } + /// Executes the full store query plan and returns merged results #[allow(dead_code)] fn execute_and_merge_store_queries( @@ -760,28 +864,12 @@ impl SimpleEngine { let merge_start_time = Instant::now(); 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 - // expected for one exact window; merge whatever came back - // instead of arbitrarily keeping the first and dropping the - // rest (see #567). - const EXPECTED_BUCKETS_PER_KEY: usize = 1; + // Legacy instant helper: merge every Sliding bucket supplied by + // its caller. The live instant path now shares the range + // pipeline and plans the exact non-overlapping cover upstream. debug!("Sliding window mode: merging {} keys", values_map.len()); - for timestamped_buckets in values_map.values() { - if timestamped_buckets.is_empty() { - continue; - } - if timestamped_buckets.len() != EXPECTED_BUCKETS_PER_KEY { - warn!( - "Sliding window expected {} precompute(s) per key, found {}. Merging all.", - EXPECTED_BUCKETS_PER_KEY, - timestamped_buckets.len() - ); - } - } - // Sliding windows always merge (all buckets belong to one - // logical window) — reuse the same merge path as Tumbling. + // Reuse the common accumulator merge path for the caller's + // already-selected Sliding cover. self.merge_precomputed_outputs(&values_map, true, agg_info.aggregation_type_for_value) } else { // Tumbling window: merge needed @@ -1675,25 +1763,6 @@ impl SimpleEngine { window_buckets } - /// Returns whatever bucket(s) `bucket_map` has at exactly - /// `window_start`, or empty if none. Unlike `sum_window`, does not walk - /// or sum multiple grid positions: for a Sliding aggregation, the bucket - /// at `window_start` is already the complete, correctly-merged answer - /// for its window (`worker.rs::merge_panes_for_window` pre-merges before - /// storing), so summing it with neighboring positions would double-count - /// overlapping data (#608). Used identically by - /// `execute_range_query_pipeline` for both the value side and the keys - /// side. - fn single_window( - bucket_map: &HashMap>, - window_start: u64, - ) -> Vec> { - bucket_map - .get(&window_start) - .map(|buckets| buckets.iter().map(|b| b.clone_boxed_core()).collect()) - .unwrap_or_default() - } - /// Collects every bucket in `bucket_map` with a start timestamp strictly /// before `before`, without walking grid positions -- cost proportional /// to however many buckets actually exist in `bucket_map`, never to a @@ -1754,10 +1823,11 @@ impl SimpleEngine { .collect() } - /// Picks how a step's window is composed from `bucket_map`: Sliding -> - /// `single_window` (one lookup); Tumbling -> `sum_window` - /// (scan-and-sum). Used identically by `execute_range_query_pipeline` - /// for both the value side and the keys side (#608). + /// Picks how a step's window is composed from `bucket_map`: Sliding walks + /// the non-overlapping stored-window stride `W`; Tumbling walks the + /// bucket grid stride. Used identically for values and SetAggregator + /// keys. Walking Sliding by its slide `S` would double-count overlapping + /// full-window aggregates (#608/#621). /// /// NOT used for `AggregationType::DeltaSetAggregator` keys -- callers on /// that path must call `collect_bucket_map_entries_before` directly @@ -1768,9 +1838,10 @@ impl SimpleEngine { window_end: u64, step_increment: u64, window_type: WindowType, + stored_window_size_ms: u64, ) -> Vec> { if window_type == WindowType::Sliding { - Self::single_window(bucket_map, window_start) + Self::sum_window(bucket_map, window_start, window_end, stored_window_size_ms) } else { Self::sum_window(bucket_map, window_start, window_end, step_increment) } @@ -1799,8 +1870,22 @@ impl SimpleEngine { use crate::engines::query_result::RangeVectorElement; use crate::engines::window_merger::create_window_merger; - // Step 1: Fetch all data needed for the entire range - let all_data = self.execute_store_query(&context.base.store_plan.values_query)?; + let lookback_ms = (context.lookback_bucket_count as u64) * context.tumbling_window_ms; + + // Step 1: Fetch all data needed for the entire range. Sliding + // aggregates are already full, overlapping windows in the store, so + // request only the W-spaced exact cover for each output timestamp. + let all_data = if context.window_type == WindowType::Sliding { + self.execute_sliding_cover_query( + &context.base.store_plan.values_query, + &context.output_timestamps, + lookback_ms, + context.window_size_ms, + context.tumbling_window_ms, + )? + } else { + self.execute_store_query(&context.base.store_plan.values_query)? + }; if all_data.is_empty() { return Err(format!("No data found for metric: {}", context.base.metric)); @@ -1822,6 +1907,21 @@ impl SimpleEngine { // mirroring the values loop, is the fix. let keys_raw_data: Option = match &context.base.store_plan.keys_query { + Some(keys_query) if context.keys_window_type == Some(WindowType::Sliding) => Some( + self.execute_sliding_cover_query( + keys_query, + &context.output_timestamps, + context + .keys_lookback_ms + .ok_or("Sliding keys query is missing its lookback")?, + context + .keys_window_size_ms + .ok_or("Sliding keys query is missing its window size")?, + context + .keys_tumbling_window_ms + .ok_or("Sliding keys query is missing its slide interval")?, + )?, + ), Some(keys_query) => Some(self.execute_store_query(keys_query)?), None => None, }; @@ -1836,28 +1936,14 @@ impl SimpleEngine { let buckets_per_step = context.buckets_per_step; let lookback_bucket_count = context.lookback_bucket_count; let tumbling_window_ms = context.tumbling_window_ms; - let lookback_ms = (lookback_bucket_count as u64) * tumbling_window_ms; let window_type = context.window_type; - // single_window's correctness for Sliding depends on this equality - // holding -- it looks up exactly one bucket at - // `current_time - lookback_ms` and trusts that position to be the - // step's whole window. Active assert (not debug_assert!): a broken - // equality here means silently wrong data, the same failure mode - // #608 fixed, not just a debug-time nicety (#608 review). - assert!( - window_type != WindowType::Sliding || lookback_ms == context.window_size_ms, - "Sliding range query: lookback_ms ({lookback_ms}) must equal window_size_ms \ - ({}) -- single_window's per-step lookup is only correct under this invariant", - context.window_size_ms - ); let keys_lookback_ms = context.keys_lookback_ms; let keys_tumbling_window_ms = context.keys_tumbling_window_ms; let keys_window_type = context.keys_window_type; let keys_window_size_ms = context.keys_window_size_ms; // Named distinctly from `WindowType` (Sliding/Tumbling, picks how a - // step's window is composed from `bucket_map` below -- one lookup vs. - // a scan-and-sum, see #608) -- this describes step-to-step overlap in + // step's window is composed from `bucket_map` below) -- this describes step-to-step overlap in // the OUTPUT iteration, an unrelated concept that happens to reuse // the words "sliding"/"hopping". See #581. let step_overlap_mode = if buckets_per_step <= lookback_bucket_count { @@ -1919,6 +2005,7 @@ impl SimpleEngine { bucket_map: GroupBucketMap<'a>, lookback_ms: u64, tumbling_window_ms: u64, + stored_window_size_ms: u64, window_type: WindowType, }, } @@ -1946,15 +2033,6 @@ impl SimpleEngine { keys_window_type.expect("keys_raw_data implies keys_window_type is Some"); let keys_window_size_ms = keys_window_size_ms.expect("keys_raw_data implies keys_window_size_ms is Some"); - // Same invariant as the value side's assert above, for the - // keys aggregation (#608 review). - assert!( - keys_window_type != WindowType::Sliding - || keys_lookback_ms == keys_window_size_ms, - "Sliding range query: keys_lookback_ms ({keys_lookback_ms}) must equal \ - keys_window_size_ms ({keys_window_size_ms}) -- single_window's per-step \ - keys lookup is only correct under this invariant" - ); keys_map .iter() .filter_map( @@ -1965,6 +2043,7 @@ impl SimpleEngine { bucket_map: Self::build_bucket_map(raw_keys_buckets), lookback_ms: keys_lookback_ms, tumbling_window_ms: keys_tumbling_window_ms, + stored_window_size_ms: keys_window_size_ms, window_type: keys_window_type, }, )), @@ -2066,6 +2145,7 @@ impl SimpleEngine { bucket_map: keys_bucket_map, lookback_ms: keys_lookback_ms, tumbling_window_ms: keys_tumbling_window_ms, + stored_window_size_ms: keys_stored_window_size_ms, window_type: keys_window_type, } => { // DeltaSetAggregator's keys window is always @@ -2078,38 +2158,69 @@ impl SimpleEngine { // already bounded by real data. Bypass that walk // entirely for this aggregation type (#581 stage // E.4 review). - let keys_window_buckets = if key_accumulator_type - == AggregationType::DeltaSetAggregator - { - // #588/#606 force DeltaSetAggregator's own - // config to Tumbling at planning time -- but - // that's a planner convention, not a runtime - // invariant this code can trust blindly. - // AggregationConfig can be (and in this crate's - // own tests routinely is) constructed directly, - // bypassing the planner. A Sliding DeltaSetAgg - // has no coherent "replay from the beginning" - // semantics to begin with, so this asserts - // rather than silently reinterpreting it (#581 - // stage E.4 review). - assert_eq!( + let keys_window_buckets = + if key_accumulator_type == AggregationType::DeltaSetAggregator { + // #588/#606 force DeltaSetAggregator's own + // config to Tumbling at planning time -- but + // that's a planner convention, not a runtime + // invariant this code can trust blindly. + // AggregationConfig can be (and in this crate's + // own tests routinely is) constructed directly, + // bypassing the planner. A Sliding DeltaSetAgg + // has no coherent "replay from the beginning" + // semantics to begin with, so this asserts + // rather than silently reinterpreting it (#581 + // stage E.4 review). + assert_eq!( *keys_window_type, WindowType::Tumbling, "DeltaSetAggregator keys config must be Tumbling (#588/#606) -- \ the replay-from-the-beginning fast path has no correct meaning \ for Sliding" ); - Self::collect_bucket_map_entries_before(keys_bucket_map, current_time) - } else { - let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); - Self::window_buckets_for_step( - keys_bucket_map, - keys_window_start, - current_time, - *keys_tumbling_window_ms, - *keys_window_type, - ) - }; + let replay_end = if window_type == WindowType::Sliding { + Self::align_down_with_warning( + current_time, + tumbling_window_ms, + "DeltaSetAggregator replay end", + ) + } else { + current_time + }; + Self::collect_bucket_map_entries_before(keys_bucket_map, replay_end) + } else { + let keys_window_end = if *keys_window_type == WindowType::Sliding { + Self::align_down_with_warning( + current_time, + *keys_tumbling_window_ms, + "Sliding key window end", + ) + } else { + current_time + }; + let keys_window_start = + keys_window_end.saturating_sub(*keys_lookback_ms); + Self::window_buckets_for_step( + keys_bucket_map, + keys_window_start, + keys_window_end, + *keys_tumbling_window_ms, + *keys_window_type, + *keys_stored_window_size_ms, + ) + }; + + if *keys_window_type == WindowType::Sliding { + let expected = + (*keys_lookback_ms / *keys_stored_window_size_ms) as usize; + if keys_window_buckets.len() < expected { + debug!( + "Skipping incomplete Sliding key cover at t={}", + current_time + ); + continue; + } + } if keys_window_buckets.is_empty() { debug!( @@ -2134,16 +2245,37 @@ impl SimpleEngine { // Window covers [current_time - lookback_ms, current_time) // This means we look at buckets that START within this range - let window_start = current_time.saturating_sub(lookback_ms); + let window_end = if window_type == WindowType::Sliding { + Self::align_down_with_warning( + current_time, + tumbling_window_ms, + "Sliding value window end", + ) + } else { + current_time + }; + let window_start = window_end.saturating_sub(lookback_ms); let window_buckets = Self::window_buckets_for_step( bucket_map, window_start, - current_time, + window_end, tumbling_window_ms, window_type, + context.window_size_ms, ); + if window_type == WindowType::Sliding { + let expected = (lookback_ms / context.window_size_ms) as usize; + if window_buckets.len() < expected { + debug!( + "Skipping incomplete Sliding value cover at t={}", + current_time + ); + continue; + } + } + if window_buckets.is_empty() { // No data at all for this window - skip sample debug!( @@ -3928,6 +4060,11 @@ mod stage_e4_instant_wrapper_equivalence_tests { KeysConfig::DeltaSetAgg => AggregationType::DeltaSetAggregator, KeysConfig::None => unreachable!(), }; + let (key_window_size_ms, key_slide_interval_ms, key_window_type) = match keys { + KeysConfig::SetAgg => (window_size_ms, slide_interval_ms, window_type), + KeysConfig::DeltaSetAgg => (1000, 1000, WindowType::Tumbling), + KeysConfig::None => unreachable!(), + }; aggregation_configs.insert( 2u64, AggregationConfig { @@ -3939,9 +4076,9 @@ mod stage_e4_instant_wrapper_equivalence_tests { aggregated_labels: KeyByLabelNames::new(vec!["host".to_string()]), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, - window_type: WindowType::Tumbling, + 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: "cpu_load".to_string(), @@ -3986,7 +4123,12 @@ mod stage_e4_instant_wrapper_equivalence_tests { } KeysConfig::None => unreachable!(), }; - let output = PrecomputedOutput::new(2000, 3000, host_a.clone(), 2); + let key_window_size_ms = match keys { + KeysConfig::SetAgg => window_size_ms, + KeysConfig::DeltaSetAgg => 1000, + KeysConfig::None => unreachable!(), + }; + let output = PrecomputedOutput::new(3000 - key_window_size_ms, 3000, host_a.clone(), 2); store.insert_precomputed_output(output, acc).unwrap(); } diff --git a/asap-query-engine/src/engines/sliding_window_composition.rs b/asap-query-engine/src/engines/sliding_window_composition.rs new file mode 100644 index 0000000..214f5ed --- /dev/null +++ b/asap-query-engine/src/engines/sliding_window_composition.rs @@ -0,0 +1,246 @@ +use crate::stores::TimestampRange; +use tracing::warn; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct SlidingWindowSpec { + pub(crate) window_size_ms: u64, + pub(crate) slide_interval_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ExactWindowCover { + pub(crate) aligned_end_ms: u64, + pub(crate) windows: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum CompositionError { + ZeroWindowSize, + ZeroSlideInterval, + ZeroLookback, + WindowNotMultipleOfSlide { + window_size_ms: u64, + slide_interval_ms: u64, + }, + LookbackBeforeEpoch { + aligned_end_ms: u64, + lookback_ms: u64, + }, + WindowAllocationFailed { + window_count: u64, + }, + LookbackNotMultiple { + lookback_ms: u64, + window_size_ms: u64, + }, +} + +pub(crate) fn plan_exact_cover( + query_end_ms: u64, + lookback_ms: u64, + spec: SlidingWindowSpec, +) -> Result { + if spec.window_size_ms == 0 { + return Err(CompositionError::ZeroWindowSize); + } + if spec.slide_interval_ms == 0 { + return Err(CompositionError::ZeroSlideInterval); + } + if lookback_ms == 0 { + return Err(CompositionError::ZeroLookback); + } + if !spec.window_size_ms.is_multiple_of(spec.slide_interval_ms) { + return Err(CompositionError::WindowNotMultipleOfSlide { + window_size_ms: spec.window_size_ms, + slide_interval_ms: spec.slide_interval_ms, + }); + } + if !lookback_ms.is_multiple_of(spec.window_size_ms) { + return Err(CompositionError::LookbackNotMultiple { + lookback_ms, + window_size_ms: spec.window_size_ms, + }); + } + + let aligned_end_ms = query_end_ms - (query_end_ms % spec.slide_interval_ms); + if aligned_end_ms != query_end_ms { + warn!( + query_end_ms, + aligned_end_ms, + slide_interval_ms = spec.slide_interval_ms, + "Sliding query end was aligned down; the requested timestamp will not be used as-is" + ); + } + let start_ms = + aligned_end_ms + .checked_sub(lookback_ms) + .ok_or(CompositionError::LookbackBeforeEpoch { + aligned_end_ms, + lookback_ms, + })?; + let window_count = lookback_ms / spec.window_size_ms; + let capacity = usize::try_from(window_count) + .map_err(|_| CompositionError::WindowAllocationFailed { window_count })?; + let mut windows = Vec::new(); + windows + .try_reserve_exact(capacity) + .map_err(|_| CompositionError::WindowAllocationFailed { window_count })?; + for index in 0..window_count { + let start = start_ms + index * spec.window_size_ms; + windows.push((start, start + spec.window_size_ms)); + } + + Ok(ExactWindowCover { + aligned_end_ms, + windows, + }) +} + +#[cfg(test)] +mod tests { + use super::{plan_exact_cover, CompositionError, SlidingWindowSpec}; + + #[test] + fn wider_lookback_is_covered_by_non_overlapping_stored_windows() { + let cover = plan_exact_cover( + 10_000, + 10_000, + SlidingWindowSpec { + window_size_ms: 5_000, + slide_interval_ms: 1_000, + }, + ) + .expect("the lookback is exactly composable"); + + assert_eq!(cover.aligned_end_ms, 10_000); + assert_eq!(cover.windows, vec![(0, 5_000), (5_000, 10_000)]); + } + + #[test] + fn rejects_lookback_that_is_not_a_multiple_of_the_stored_window() { + let error = plan_exact_cover( + 12_000, + 12_000, + SlidingWindowSpec { + window_size_ms: 5_000, + slide_interval_ms: 1_000, + }, + ) + .expect_err("partial stored windows cannot form an exact cover"); + + assert_eq!( + error, + CompositionError::LookbackNotMultiple { + lookback_ms: 12_000, + window_size_ms: 5_000, + } + ); + } + + #[test] + fn rejects_zero_sized_window_without_panicking() { + let error = plan_exact_cover( + 10_000, + 10_000, + SlidingWindowSpec { + window_size_ms: 0, + slide_interval_ms: 1_000, + }, + ) + .expect_err("a zero-sized stored window is invalid"); + + assert_eq!(error, CompositionError::ZeroWindowSize); + } + + #[test] + fn rejects_zero_slide_without_panicking() { + let error = plan_exact_cover( + 10_000, + 10_000, + SlidingWindowSpec { + window_size_ms: 5_000, + slide_interval_ms: 0, + }, + ) + .expect_err("a zero slide is invalid"); + + assert_eq!(error, CompositionError::ZeroSlideInterval); + } + + #[test] + fn rejects_zero_lookback() { + let error = plan_exact_cover( + 10_000, + 0, + SlidingWindowSpec { + window_size_ms: 5_000, + slide_interval_ms: 1_000, + }, + ) + .expect_err("a query must cover a positive interval"); + + assert_eq!(error, CompositionError::ZeroLookback); + } + + #[test] + fn rejects_window_width_that_does_not_land_on_the_slide_grid() { + let error = plan_exact_cover( + 12_000, + 12_000, + SlidingWindowSpec { + window_size_ms: 6_000, + slide_interval_ms: 4_000, + }, + ) + .expect_err("W-spaced cover boundaries must exist on the S grid"); + + assert_eq!( + error, + CompositionError::WindowNotMultipleOfSlide { + window_size_ms: 6_000, + slide_interval_ms: 4_000, + } + ); + } + + #[test] + fn rejects_lookback_before_the_unix_epoch_without_panicking() { + let error = plan_exact_cover( + 4_500, + 10_000, + SlidingWindowSpec { + window_size_ms: 5_000, + slide_interval_ms: 1_000, + }, + ) + .expect_err("the aligned query interval starts before epoch zero"); + + assert_eq!( + error, + CompositionError::LookbackBeforeEpoch { + aligned_end_ms: 4_000, + lookback_ms: 10_000, + } + ); + } + + #[test] + fn rejects_an_exact_cover_too_large_to_allocate() { + let error = plan_exact_cover( + u64::MAX, + u64::MAX, + SlidingWindowSpec { + window_size_ms: 1, + slide_interval_ms: 1, + }, + ) + .expect_err("allocation failure must be recoverable"); + + assert_eq!( + error, + CompositionError::WindowAllocationFailed { + window_count: u64::MAX, + } + ); + } +} 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 d5bad02..238cd08 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -257,11 +257,8 @@ mod tests { /// but the KEY aggregation is a Sliding window with /// `key_slide_interval_ms < key_window_size_ms` (#600). Real Sliding /// buckets are persisted on the slide_interval_ms grid, not the - /// window_size_ms grid (`precompute_engine/window_manager.rs`), so the - /// keys bucket span here is `key_slide_interval_ms`, not - /// `key_window_size_ms` -- unlike the value side, which stays Tumbling - /// (span == window) exactly as `create_range_engine_dual_input_with_windows` - /// already does. + /// window_size_ms grid (`precompute_engine/window_manager.rs`), and the + /// value side uses the same Sliding grid for these tests. #[allow(clippy::too_many_arguments)] fn create_range_engine_dual_input_sliding_keys( metric: &str, @@ -299,8 +296,8 @@ mod tests { rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), window_size_ms: value_window_ms, - slide_interval_ms: value_window_ms, - window_type: WindowType::Tumbling, + slide_interval_ms: key_slide_interval_ms, + window_type: WindowType::Sliding, spatial_filter: String::new(), spatial_filter_normalized: String::new(), metric: metric.to_string(), @@ -411,8 +408,8 @@ mod tests { // 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 + // - value_data is also placed at timestamp=5000 (Sliding, 2000ms + // wide -> bucket [3000,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 { @@ -435,13 +432,13 @@ mod tests { // 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) + "sum by (host, event) (count_over_time(event_frequency[2s]))", + 2000, // value_window_ms, same Sliding grid as SetAggregator 2000, // key_window_size_ms 1000, // key_slide_interval_ms ); - let query = "count(event_frequency) by (host, event)"; + let query = "sum by (host, event) (count_over_time(event_frequency[2s]))"; 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); @@ -455,6 +452,107 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread")] + async fn range_query_tumbling_dual_population_returns_key_expansion() { + let mut value = CountMinSketchAccumulator::new(2, 3); + value.inner.update("host-a;evt-1", 1.0); + let mut keys = SetAggregatorAccumulator::new(); + keys.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input_with_windows( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![(1_000, None, Box::new(value) as Box)], + vec![(1_000, None, Box::new(keys) as Box)], + "count(event_frequency) by (host, event)", + 1_000, + 1_000, + ); + + let result = engine.handle_range_query_promql( + "count(event_frequency) by (host, event)".to_string(), + 1.0, + 1.5, + 1.0, + ); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + assert!(labels_have_sample_at( + &elements, + &["host-a", "evt-1"], + 1_000 + )); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_tumbling_dual_population_keeps_key_steps_isolated() { + let mut value_1 = CountMinSketchAccumulator::new(2, 3); + value_1.inner.update("host-a;evt-1", 1.0); + let mut value_2 = CountMinSketchAccumulator::new(2, 3); + value_2.inner.update("host-a;evt-2", 1.0); + let mut keys_1 = SetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = SetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-2".to_string()], + }); + + let engine = create_range_engine_dual_input_with_windows( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1_000, None, Box::new(value_1) as Box), + (2_000, None, Box::new(value_2) as Box), + ], + vec![ + (1_000, None, Box::new(keys_1) as Box), + (2_000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + 1_000, + 1_000, + ); + + let result = engine.handle_range_query_promql( + "count(event_frequency) by (host, event)".to_string(), + 1.0, + 2.0, + 1.0, + ); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + assert!(labels_have_sample_at( + &elements, + &["host-a", "evt-1"], + 1_000 + )); + assert!(labels_have_sample_at( + &elements, + &["host-a", "evt-2"], + 2_000 + )); + assert!(!labels_have_sample_at( + &elements, + &["host-a", "evt-2"], + 1_000 + )); + assert!(!labels_have_sample_at( + &elements, + &["host-a", "evt-1"], + 2_000 + )); + } + /// #608's keys-side counterpart: SetAggregator is a real Sliding-capable /// keys aggregation (unlike DeltaSetAggregator, restricted to Tumbling /// by #606), so the keys-side per-step composition has the same @@ -510,13 +608,13 @@ mod tests { (4000, None, Box::new(keys_2) as Box), (5000, None, Box::new(keys_3) as Box), ], - "count(event_frequency) by (host, event)", - 1000, // value_window_ms (Tumbling, unaffected by #608) + "sum by (host, event) (count_over_time(event_frequency[2s]))", + 2000, // value_window_ms, same Sliding grid as SetAggregator 2000, // key_window_size_ms 1000, // key_slide_interval_ms ); - let query = "count(event_frequency) by (host, event)"; + let query = "sum by (host, event) (count_over_time(event_frequency[2s]))"; let result = engine.handle_range_query_promql(query.to_string(), 3.0, 4.0, 1.0); let (_, qr) = result.expect("range query failed"); let elements = matrix_values(qr); @@ -843,7 +941,7 @@ mod tests { Box::new(SumAccumulator::with_sum(5.0)) as Box, ), ]; - let query = "sum_over_time(http_requests[1s])"; + let query = "sum_over_time(http_requests[2s])"; let engine = create_engine_multi_timestamp_with_window( "http_requests", AggregationType::Sum, diff --git a/asap-query-engine/src/tests/query_equivalence_tests.rs b/asap-query-engine/src/tests/query_equivalence_tests.rs index 80890a8..01d9966 100644 --- a/asap-query-engine/src/tests/query_equivalence_tests.rs +++ b/asap-query-engine/src/tests/query_equivalence_tests.rs @@ -89,6 +89,136 @@ impl Store for NoOpStore { mod tests { use super::*; + #[test] + fn sql_executes_wider_sliding_window_exact_cover() { + use crate::data_model::{CleanupPolicy, KeyByLabelValues, PrecomputedOutput}; + use crate::engines::query_result::QueryResult; + use crate::precompute_operators::sum_accumulator::SumAccumulator; + use crate::stores::simple_map_store::SimpleMapStore; + + let promql_query = "sum_over_time(cpu_usage[10s])"; + let sql_query = "SELECT SUM(value) FROM cpu_usage WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY L1, L2, L3, L4"; + let (_, sql_config, streaming_config) = TestConfigBuilder::new("cpu_usage") + .with_grouping_labels(vec!["L1", "L2", "L3", "L4"]) + .with_scrape_interval_ms(1_000) + .add_temporal_query(promql_query, sql_query, 1, 5_000, WindowType::Sliding) + .build_both(); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + let group = Some(KeyByLabelValues { + labels: vec!["a".into(), "b".into(), "c".into(), "d".into()], + }); + for (start, end, value) in [(0, 5_000, 7.0), (5_000, 10_000, 11.0)] { + store + .insert_precomputed_output( + PrecomputedOutput::new(start, end, group.clone(), 1), + Box::new(SumAccumulator::with_sum(value)), + ) + .unwrap(); + } + let engine = SimpleEngine::new( + store, + sql_config, + streaming_config, + 1_000, + QueryLanguage::sql, + ); + + let (_, result) = engine + .handle_query_sql(sql_query.to_string(), 10.0) + .expect("SQL should execute the wider Sliding exact cover"); + let QueryResult::Vector(vector) = result else { + panic!("expected an instant SQL vector"); + }; + + assert_eq!(vector.values.len(), 1); + assert_eq!(vector.values[0].value, 18.0); + } + + #[test] + fn sql_executes_wider_sliding_dual_population_query() { + use crate::data_model::{ + AggregationConfig, AggregationType, CleanupPolicy, KeyByLabelValues, PrecomputedOutput, + }; + use crate::engines::query_result::QueryResult; + use crate::precompute_operators::{CountMinSketchAccumulator, SetAggregatorAccumulator}; + use crate::stores::simple_map_store::SimpleMapStore; + + let sql_query = "SELECT COUNT(value) FROM events WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY host"; + let (_, mut sql_config, mut streaming_config) = TestConfigBuilder::new("events") + .with_grouping_labels(vec!["host"]) + .add_temporal_query( + "count_over_time(events[10s])", + sql_query, + 1, + 5_000, + WindowType::Sliding, + ) + .build_both(); + + let value_config = streaming_config + .get_aggregation_config(1) + .expect("value config") + .clone(); + let mut value_config = AggregationConfig { + aggregation_type: AggregationType::CountMinSketch, + ..value_config + }; + value_config.slide_interval_ms = 1_000; + let mut key_config = value_config.clone(); + key_config.aggregation_id = 2; + key_config.aggregation_type = AggregationType::SetAggregator; + streaming_config = Arc::new(crate::data_model::StreamingConfig { + aggregation_configs: HashMap::from([(1, value_config), (2, key_config)]), + }); + sql_config.query_configs[1] = sql_config.query_configs[1] + .clone() + .add_aggregation(crate::data_model::AggregationReference::new(2, None)); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + for (start, end, count) in [(0, 5_000, 2.0), (5_000, 10_000, 3.0)] { + let mut value = CountMinSketchAccumulator::new(4, 64); + value.inner.update("host-a", count); + let mut keys = SetAggregatorAccumulator::new(); + keys.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + store + .insert_precomputed_output( + PrecomputedOutput::new(start, end, None, 1), + Box::new(value), + ) + .unwrap(); + store + .insert_precomputed_output( + PrecomputedOutput::new(start, end, None, 2), + Box::new(keys), + ) + .unwrap(); + } + + let engine = SimpleEngine::new( + store, + sql_config, + streaming_config, + 1_000, + QueryLanguage::sql, + ); + let (_, result) = engine + .handle_query_sql(sql_query.to_string(), 10.0) + .expect("SQL dual-population query should execute"); + let QueryResult::Vector(vector) = result else { + panic!("expected an instant SQL vector"); + }; + assert_eq!(vector.values.len(), 1); + assert_eq!(vector.values[0].value, 5.0); + } + #[test] fn test_temporal_sum_equivalence() { let scrape_interval_ms = 1000; diff --git a/asap-query-engine/src/tests/stage_e_instant_range_equivalence_tests.rs b/asap-query-engine/src/tests/stage_e_instant_range_equivalence_tests.rs index c9711f3..fa84ef0 100644 --- a/asap-query-engine/src/tests/stage_e_instant_range_equivalence_tests.rs +++ b/asap-query-engine/src/tests/stage_e_instant_range_equivalence_tests.rs @@ -13,15 +13,11 @@ //! these (both from reading `execute_range_query_pipeline` and its callers, //! not from any doc): //! -//! 1. A Sliding-window aggregation's query lookback is REQUIRED to equal its -//! `window_size_ms` -- an active `assert!` in `execute_range_query_pipeline` -//! (mod.rs, #608 review), not just a convention. So "Sliding, -//! window_size == lookback/2, /3, ..." isn't a constructible scenario -- -//! it would panic, not merge incorrectly. The window-shape axis below -//! instead varies Tumbling's lookback-to-bucket-width ratio (1/2/3, via -//! `sum_over_time(metric[Ns])` selector width) and adds one genuinely -//! overlapping Sliding case (`window_size_ms > slide_interval_ms`, -//! lookback == window_size_ms as required). +//! 1. Sliding lookbacks may now be positive integer multiples of +//! `window_size_ms` (#554), composed from non-overlapping W-spaced stored +//! windows. This older matrix retains an exact-width overlapping Sliding +//! case; wider instant/range cases live in +//! `window_semantics_consistency_tests`. //! //! 2. `SetAggregator`/`DeltaSetAggregator` are keys-side (dual-population) //! aggregation types in this codebase -- never a general value-aggregation @@ -159,6 +155,15 @@ mod tests { KeysConfig::DeltaSetAgg => AggregationType::DeltaSetAggregator, KeysConfig::None => unreachable!(), }; + let (key_window_size_ms, key_slide_interval_ms, key_window_type) = match keys { + KeysConfig::SetAgg => ( + shape.window_size_ms, + shape.slide_interval_ms, + shape.window_type, + ), + KeysConfig::DeltaSetAgg => (1000, 1000, WindowType::Tumbling), + KeysConfig::None => unreachable!(), + }; aggregation_configs.insert( 2u64, AggregationConfig { @@ -170,9 +175,9 @@ mod tests { aggregated_labels: KeyByLabelNames::new(vec!["host".to_string()]), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: 1000, - slide_interval_ms: 1000, - window_type: WindowType::Tumbling, + 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: "cpu_load".to_string(), @@ -232,7 +237,12 @@ mod tests { } KeysConfig::None => unreachable!(), }; - let output = PrecomputedOutput::new(2000, 3000, host_a.clone(), 2); + let key_window_size_ms = match keys { + KeysConfig::SetAgg => shape.window_size_ms, + KeysConfig::DeltaSetAgg => 1000, + KeysConfig::None => unreachable!(), + }; + let output = PrecomputedOutput::new(3000 - key_window_size_ms, 3000, host_a.clone(), 2); store.insert_precomputed_output(output, acc).unwrap(); } diff --git a/asap-query-engine/src/tests/window_semantics_consistency_tests.rs b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs index ad9c5bb..6b2144b 100644 --- a/asap-query-engine/src/tests/window_semantics_consistency_tests.rs +++ b/asap-query-engine/src/tests/window_semantics_consistency_tests.rs @@ -37,7 +37,9 @@ mod tests { 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::precompute_operators::{ + CountMinSketchAccumulator, DeltaSetAggregatorAccumulator, 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; @@ -203,6 +205,124 @@ mod tests { } } + #[tokio::test(flavor = "multi_thread")] + async fn wider_sliding_instant_query_merges_only_a_non_overlapping_exact_cover() { + let data = [1.0, 10.0, 100.0, 1_000.0, 10_000.0, 100_000.0] + .into_iter() + .enumerate() + .map(|(index, value)| { + ( + (index as u64 + 1) * 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(value)) as Box, + ) + }) + .collect(); + let query = "sum_over_time(cpu_load[6s])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 3_000, + 1_000, + WindowType::Sliding, + ); + + let result = engine + .handle_query_promql(query.to_string(), 6.0) + .expect("the wider Sliding query should be accelerated"); + + assert_close( + single_host_a_value(result.1), + 111_111.0, + "[0, 6s) must be composed from [0, 3s) and [3s, 6s)", + ); + + let misaligned_result = engine + .handle_query_promql(query.to_string(), 6.5) + .expect("the endpoint should align down to the latest complete slide boundary"); + assert_close( + single_host_a_value(misaligned_result.1), + 111_111.0, + "a 6.5s evaluation must read the complete cover ending at 6s", + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn wider_sliding_query_with_a_missing_constituent_falls_back_as_a_whole() { + // Omitting the pane ending at 2s prevents [0, 3s) from being emitted, + // while all panes for [3s, 6s) remain present. + let data = [ + (1, 1.0), + (3, 100.0), + (4, 1_000.0), + (5, 10_000.0), + (6, 100_000.0), + ] + .into_iter() + .map(|(second, value)| { + ( + second * 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(value)) as Box, + ) + }) + .collect(); + let query = "sum_over_time(cpu_load[6s])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 3_000, + 1_000, + WindowType::Sliding, + ); + + assert!( + engine.handle_query_promql(query.to_string(), 6.0).is_none(), + "a partial exact cover must fall back instead of returning partial data" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn wider_sliding_range_query_composes_each_output_step_without_overlap() { + let data = [1.0, 10.0, 100.0, 1_000.0, 10_000.0, 100_000.0, 1_000_000.0] + .into_iter() + .enumerate() + .map(|(index, value)| { + ( + (index as u64 + 1) * 1_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(value)) as Box, + ) + }) + .collect(); + let query = "sum_over_time(cpu_load[6s])"; + let engine = create_engine_multi_timestamp_with_window( + "cpu_load", + AggregationType::Sum, + vec!["host"], + data, + query, + 3_000, + 1_000, + WindowType::Sliding, + ); + + let result = engine + .handle_range_query_promql(query.to_string(), 6.0, 7.0, 1.0) + .expect("the wider Sliding range query should be accelerated"); + + assert_eq!( + host_a_samples(&matrix_values(result.1)), + vec![(6_000, 111_111.0), (7_000, 1_111_110.0)] + ); + } + // ════════════════════════════════════════════════════════════════════ // 2. Tumbling: instant and range paths must AGREE, and both must // correctly SUM every disjoint bucket in the query's window (proving @@ -401,15 +521,10 @@ mod tests { ) } - /// 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. + /// A pre-bound SetAggregator on a different window grid is rejected, + /// while a matching Tumbling key grid remains valid. #[tokio::test(flavor = "multi_thread")] - async fn instant_keys_query_correct_for_both_tumbling_and_sliding_key_aggregation() { + async fn prebound_set_aggregator_on_a_different_grid_is_rejected() { let tumbling_engine = build_dual_engine_with_key_window(5_000, WindowType::Tumbling, 1_000, 1_000); let sliding_engine = @@ -422,11 +537,6 @@ mod tests { .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() @@ -434,17 +544,228 @@ mod tests { "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:?}" + sliding_engine + .handle_query_promql(query.to_string(), 5.0) + .is_none(), + "a pre-bound SetAggregator must share the value aggregation's window grid" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn wider_sliding_set_aggregator_unions_keys_from_the_same_exact_cover() { + let query = "sum by (host) (count_over_time(event_frequency[6s]))"; + let common = |aggregation_id, aggregation_type| AggregationConfig { + aggregation_id, + aggregation_type, + 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: 3_000, + slide_interval_ms: 1_000, + window_type: WindowType::Sliding, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "event_frequency".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }; + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs: HashMap::from([ + (1, common(1, AggregationType::CountMinSketch)), + (2, common(2, AggregationType::SetAggregator)), + ]), + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + let host_a = KeyByLabelValues { + labels: vec!["host-a".to_string()], + }; + let host_b = KeyByLabelValues { + labels: vec!["host-b".to_string()], + }; + for (start, end, a_count, b_count) in [(0, 3_000, 1.0, 2.0), (3_000, 6_000, 10.0, 20.0)] { + let mut cms = CountMinSketchAccumulator::new(4, 128); + cms.inner.update(&host_a.to_semicolon_str(), a_count); + cms.inner.update(&host_b.to_semicolon_str(), b_count); + store + .insert_precomputed_output( + PrecomputedOutput::new(start, end, None, 1), + Box::new(cms), + ) + .unwrap(); + + let mut keys = SetAggregatorAccumulator::new(); + keys.add_key(if start == 0 { + host_a.clone() + } else { + host_b.clone() + }); + store + .insert_precomputed_output( + PrecomputedOutput::new(start, end, None, 2), + Box::new(keys), + ) + .unwrap(); + } + + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(PromQLSchema::new().add_metric( + "event_frequency".to_string(), + KeyByLabelNames::new(vec!["host".to_string()]), + )), + query_configs: vec![QueryConfig::new(query.to_string()) + .add_aggregation(AggregationReference::new(1, None)) + .add_aggregation(AggregationReference::new(2, None))], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let engine = SimpleEngine::new( + store, + inference_config, + streaming_config, + 1_000, + QueryLanguage::promql, + ); + + let (_, result) = engine + .handle_query_promql(query.to_string(), 6.0) + .expect("the wider dual-population Sliding query should be accelerated"); + let mut values = vector_values(result); + values.sort_by(|left, right| left.0.cmp(&right.0)); + + assert_eq!( + values, + vec![ + (vec!["host-a".to_string()], 11.0), + (vec!["host-b".to_string()], 22.0), + ] + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn wider_sliding_values_use_compatible_tumbling_delta_set_keys() { + let query = "sum by (host) (count_over_time(event_frequency[4s]))"; + let value_config = AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::CountMinSketch, + 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: 2_000, + slide_interval_ms: 1_000, + window_type: WindowType::Sliding, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "event_frequency".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }; + let delta_config = AggregationConfig { + aggregation_id: 2, + aggregation_type: AggregationType::DeltaSetAggregator, + window_size_ms: 1_000, + slide_interval_ms: 1_000, + window_type: WindowType::Tumbling, + ..value_config.clone() + }; + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs: HashMap::from([(1, value_config), (2, delta_config)]), + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + let host_a = KeyByLabelValues { + labels: vec!["host-a".to_string()], + }; + let host_b = KeyByLabelValues { + labels: vec!["host-b".to_string()], + }; + let host_c = KeyByLabelValues { + labels: vec!["host-c".to_string()], + }; + for (start, end, a_count, b_count) in [(0, 2_000, 1.0, 2.0), (2_000, 4_000, 10.0, 20.0)] { + let mut cms = CountMinSketchAccumulator::new(4, 128); + cms.inner.update(&host_a.to_semicolon_str(), a_count); + cms.inner.update(&host_b.to_semicolon_str(), b_count); + store + .insert_precomputed_output( + PrecomputedOutput::new(start, end, None, 1), + Box::new(cms), + ) + .unwrap(); + } + for (start, key) in [ + (0, host_a.clone()), + (2_000, host_b.clone()), + // This delta belongs to the next value-grid interval. A query + // evaluated at 4.5s aligns values down to 4s and must not see it. + (4_000, host_c), + ] { + let mut delta = DeltaSetAggregatorAccumulator::new(); + delta.add_key(key); + store + .insert_precomputed_output( + PrecomputedOutput::new(start, start + 1_000, None, 2), + Box::new(delta), + ) + .unwrap(); + } + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(PromQLSchema::new().add_metric( + "event_frequency".to_string(), + KeyByLabelNames::new(vec!["host".to_string()]), + )), + query_configs: vec![QueryConfig::new(query.to_string()) + .add_aggregation(AggregationReference::new(1, None)) + .add_aggregation(AggregationReference::new(2, None))], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let engine = SimpleEngine::new( + store, + inference_config, + streaming_config, + 1_000, + QueryLanguage::promql, ); + + let (_, result) = engine + .handle_query_promql(query.to_string(), 4.0) + .expect("compatible Tumbling DeltaSet keys should resolve Sliding values"); + let mut values = vector_values(result); + values.sort_by(|left, right| left.0.cmp(&right.0)); + + assert_eq!( + values, + vec![ + (vec!["host-a".to_string()], 11.0), + (vec!["host-b".to_string()], 22.0), + ] + ); + + let (_, misaligned_result) = engine + .handle_query_promql(query.to_string(), 4.5) + .expect("misaligned evaluation should use the latest complete value grid point"); + let mut misaligned_values = vector_values(misaligned_result); + misaligned_values.sort_by(|left, right| left.0.cmp(&right.0)); assert_eq!( - tumbling_values.len(), - sliding_values.len(), - "same logical key set must be returned regardless of the key \ - aggregation's WindowType" + misaligned_values, + vec![ + (vec!["host-a".to_string()], 11.0), + (vec!["host-b".to_string()], 22.0), + ], + "DeltaSet replay must stop at the same aligned endpoint as Sliding values" ); } diff --git a/asap-query-engine/tests/e2e_precompute_equivalence.rs b/asap-query-engine/tests/e2e_precompute_equivalence.rs index c1a7658..ccfacfa 100644 --- a/asap-query-engine/tests/e2e_precompute_equivalence.rs +++ b/asap-query-engine/tests/e2e_precompute_equivalence.rs @@ -9,7 +9,7 @@ use asap_sketchlib::KllSketch; use asap_types::aggregation_config::AggregationConfig; -use asap_types::enums::{AggregationType, WindowType}; +use asap_types::enums::{AggregationType, CleanupPolicy, QueryLanguage, WindowType}; use flate2::{write::GzEncoder, Compression}; use prost::Message; use serde_json::json; @@ -17,7 +17,10 @@ use std::collections::HashMap; use std::io::Write; use std::sync::Arc; -use query_engine_rust::data_model::{PrecomputedOutput, StreamingConfig}; +use query_engine_rust::data_model::{ + AggregationReference, InferenceConfig, PrecomputedOutput, PromQLSchema, QueryConfig, + SchemaConfig, StreamingConfig, +}; use query_engine_rust::drivers::ingest::prometheus_remote_write::{ Label, Sample, TimeSeries, WriteRequest, }; @@ -28,6 +31,7 @@ use query_engine_rust::precompute_engine::{ }; use query_engine_rust::precompute_operators::datasketches_kll_accumulator::DatasketchesKLLAccumulator; use query_engine_rust::precompute_operators::multiple_sum_accumulator::MultipleSumAccumulator; +use query_engine_rust::{QueryResult, SimpleEngine, SimpleMapStore, Store}; // ─── helpers ──────────────────────────────────────────────────────────────── @@ -385,3 +389,92 @@ async fn e2e_multiple_sum_output_matches_arroyo() { "MultipleSum sums map mismatch" ); } + +#[tokio::test] +async fn e2e_sliding_precompute_outputs_compose_a_wider_query() { + let port = 19402u16; + let agg_id = 3u64; + let window_size_ms = 5_000u64; + let slide_interval_ms = 1_000u64; + let metric = "requests"; + let query = "sum_over_time(requests[10s])"; + + let config = make_agg_config( + agg_id, + metric, + AggregationType::Sum, + "", + window_size_ms, + slide_interval_ms, + vec![], + ); + let streaming_config = Arc::new(StreamingConfig::new(HashMap::from([(agg_id, config)]))); + let sink = Arc::new(CapturingOutputSink::new()); + let engine = PrecomputeEngine::new( + engine_config(), + streaming_config.clone(), + sink.clone(), + vec![Box::new(HttpIngestSource::new(HttpIngestConfig { port }))], + ); + tokio::spawn(async move { + let _ = engine.run().await; + }); + tokio::time::sleep(tokio::time::Duration::from_millis(300)).await; + + let client = reqwest::Client::new(); + for second in 1..10i64 { + send_remote_write( + &client, + port, + vec![make_timeseries( + metric, + vec![], + second * 1_000, + second as f64, + )], + ) + .await; + } + send_remote_write( + &client, + port, + vec![make_timeseries(metric, vec![], 15_000, 0.0)], + ) + .await; + tokio::time::sleep(tokio::time::Duration::from_millis(600)).await; + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + for (output, accumulator) in sink.drain() { + store + .insert_precomputed_output(output, accumulator) + .unwrap(); + } + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL( + PromQLSchema::new().add_metric(metric.to_string(), Default::default()), + ), + query_configs: vec![QueryConfig::new(query.to_string()) + .add_aggregation(AggregationReference::new(agg_id, None))], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let query_engine = SimpleEngine::new( + store, + inference_config, + streaming_config, + slide_interval_ms, + QueryLanguage::promql, + ); + + let (_, result) = query_engine + .handle_query_promql(query.to_string(), 10.0) + .expect("worker-emitted Sliding windows should answer the wider query"); + let QueryResult::Vector(vector) = result else { + panic!("expected instant vector result"); + }; + + assert_eq!(vector.values.len(), 1); + assert_eq!(vector.values[0].value, 45.0); +}