From 67807e132947cdb12a41e56df47ad98457bed394 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 12:28:02 -0400 Subject: [PATCH 1/4] fix(planner): force DeltaSetAggregator to Tumbling regardless of query window_type (#588) DeltaSetAggregator only tracks added/removed keys since the last window, so it's only correct for non-overlapping (tumbling) windows. build_agg_configs_for_statistics was copying the surrounding query's window_type straight through, which would silently plan a Sliding DeltaSetAggregator once sliding windows are enabled. Always plan the companion config as Tumbling, sized to the sibling value aggregation's slide_interval_ms so its buckets align with the pane-close cadence the value aggregation already emits at. No-op under Tumbling (the only window_type the live planner produces today). Co-Authored-By: Claude Sonnet 5 --- asap-planner-rs/src/planner/agg_config.rs | 135 +++++++++++++++++++++- 1 file changed, 133 insertions(+), 2 deletions(-) diff --git a/asap-planner-rs/src/planner/agg_config.rs b/asap-planner-rs/src/planner/agg_config.rs index 9e0b4a8c..8e11241a 100644 --- a/asap-planner-rs/src/planner/agg_config.rs +++ b/asap-planner-rs/src/planner/agg_config.rs @@ -111,8 +111,17 @@ pub fn build_agg_configs_for_statistics( configs.push(IntermediateAggConfig { aggregation_type: AggregationType::DeltaSetAggregator, aggregation_sub_type: String::new(), - window_type: window_cfg.window_type, - window_size_ms: window_cfg.window_size_ms, + // DeltaSetAggregator only tracks added/removed keys since the + // last window, so it's only correct for non-overlapping + // (tumbling) windows (#588). Always plan it as Tumbling, + // decoupled from the sibling value aggregation's window_type, + // sized to the sibling's slide_interval_ms -- the pane + // granularity it already emits at (window_manager.rs) -- so + // its buckets align with existing pane-close events. This is + // a no-op when window_cfg is already Tumbling, since + // window_size_ms == slide_interval_ms there. + window_type: WindowType::Tumbling, + window_size_ms: window_cfg.slide_interval_ms, slide_interval_ms: window_cfg.slide_interval_ms, spatial_filter: spatial_filter.to_string(), metric: metric.to_string(), @@ -222,4 +231,126 @@ mod tests { .insert("depth".to_string(), Value::Number(3.into())); assert_eq!(cfg1.identifying_key(), cfg2.identifying_key()); } + + /// Builds configs for a single `CountMinSketch`-triggering statistic + /// (`Statistic::Count` under `Approximate` treatment maps to + /// `AggregationType::CountMinSketch`, see `map_statistic_to_precompute_operator`), + /// which is what makes `build_agg_configs_for_statistics` emit a paired + /// `DeltaSetAggregator` companion config. + fn configs_for_count_min_sketch( + window_cfg: &IntermediateWindowConfig, + ) -> Vec { + build_agg_configs_for_statistics( + &[Statistic::Count], + QueryTreatmentType::Approximate, + &KeyByLabelNames::empty(), + &KeyByLabelNames::empty(), + window_cfg, + "http_requests_total", + None, + None, + "", + |_, _| Ok(HashMap::new()), + ) + .expect("building configs for Statistic::Count should succeed") + } + + /// Issue #588: DeltaSetAggregator only tracks added/removed keys since + /// the last window, so it's only correct for non-overlapping (tumbling) + /// windows. When the surrounding query is planned with a Sliding + /// window_type, the companion DeltaSetAggregator config must still be + /// planned as Tumbling -- never silently copy Sliding through. + #[test] + fn sliding_query_window_forces_delta_set_aggregator_to_tumbling() { + let window_cfg = IntermediateWindowConfig { + window_type: WindowType::Sliding, + window_size_ms: 300_000, + slide_interval_ms: 60_000, + }; + + let configs = configs_for_count_min_sketch(&window_cfg); + + let delta = configs + .iter() + .find(|c| c.aggregation_type == AggregationType::DeltaSetAggregator) + .expect("CountMinSketch statistic must emit a paired DeltaSetAggregator config"); + + assert_eq!( + delta.window_type, + WindowType::Tumbling, + "DeltaSetAggregator must never be planned as Sliding" + ); + } + + /// The decoupled DeltaSetAggregator bucket must be sized to the sibling's + /// slide_interval_ms (the pane granularity the value aggregator already + /// emits at -- see window_manager.rs::panes_for_window), not its full + /// window_size_ms, so its tumbling buckets align with existing pane-close + /// events instead of requiring a second, independent emission schedule. + #[test] + fn sliding_query_window_sizes_delta_set_aggregator_to_slide_interval() { + let window_cfg = IntermediateWindowConfig { + window_type: WindowType::Sliding, + window_size_ms: 300_000, + slide_interval_ms: 60_000, + }; + + let configs = configs_for_count_min_sketch(&window_cfg); + + let delta = configs + .iter() + .find(|c| c.aggregation_type == AggregationType::DeltaSetAggregator) + .unwrap(); + + assert_eq!(delta.window_size_ms, window_cfg.slide_interval_ms); + assert_eq!(delta.slide_interval_ms, window_cfg.slide_interval_ms); + } + + /// The sibling value aggregation (CountMinSketch) keeps the query's + /// original Sliding window untouched -- only the DeltaSetAggregator + /// companion is forced to Tumbling. + #[test] + fn sliding_query_window_leaves_value_aggregation_sliding() { + let window_cfg = IntermediateWindowConfig { + window_type: WindowType::Sliding, + window_size_ms: 300_000, + slide_interval_ms: 60_000, + }; + + let configs = configs_for_count_min_sketch(&window_cfg); + + let value_cfg = configs + .iter() + .find(|c| c.aggregation_type == AggregationType::CountMinSketch) + .expect("value aggregation config must be present"); + + assert_eq!(value_cfg.window_type, WindowType::Sliding); + assert_eq!(value_cfg.window_size_ms, window_cfg.window_size_ms); + assert_eq!(value_cfg.slide_interval_ms, window_cfg.slide_interval_ms); + } + + /// Regression guard: under a Tumbling query window (window_size_ms == + /// slide_interval_ms, the only case the live planner produces today -- + /// see window.rs::should_use_sliding_window), the DeltaSetAggregator + /// companion's window fields must stay exactly what they were before + /// this fix -- copied straight from window_cfg. + #[test] + fn tumbling_query_window_delta_set_aggregator_unchanged() { + let window_cfg = IntermediateWindowConfig { + window_type: WindowType::Tumbling, + window_size_ms: 60_000, + slide_interval_ms: 60_000, + }; + + let configs = configs_for_count_min_sketch(&window_cfg); + + let delta = configs + .iter() + .find(|c| c.aggregation_type == AggregationType::DeltaSetAggregator) + .unwrap(); + + assert_eq!(delta.window_type, WindowType::Tumbling); + assert_eq!(delta.window_size_ms, 60_000); + assert_eq!(delta.slide_interval_ms, 60_000); + } } From bc84444e794cba6cb9fdb10aca6dbd1ebf7ce00c Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 12:32:41 -0400 Subject: [PATCH 2/4] fix(planner): exclude DeltaSetAggregator from Sliding candidate enumeration (#588) candidate_gen.rs's enumerate_candidates was enumerating Sliding window candidates for every agg type compatible with Statistic::Cardinality, including DeltaSetAggregator -- which is only correct under non-overlapping (tumbling) windows. This optimizer module isn't wired into the live planner yet, but should already encode the invariant before it is. Skip Sliding entries from window_candidates() when agg_type is DeltaSetAggregator; Tumbling candidates for it, and Sliding candidates for its siblings (SetAggregator, HLL), are unaffected. Co-Authored-By: Claude Sonnet 5 --- .../src/optimizer/candidate_gen.rs | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs index b3b023cf..495a0ec5 100644 --- a/asap-planner-rs/src/optimizer/candidate_gen.rs +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -51,6 +51,15 @@ pub fn enumerate_candidates(aqe: &AQE, scrape_interval_ms: u64) -> Vec Date: Tue, 25 Aug 2026 12:36:34 -0400 Subject: [PATCH 3/4] fix(asap_types): reject Sliding DeltaSetAggregator in capability matching (#588) DeltaSetAggregator only tracks added/removed keys since the last window, so it's only correct for non-overlapping (tumbling) windows. The two planner-side producers were fixed to never emit one, but find_compatible_aggregation's paired-key-agg lookup matched candidates by (metric, is_key_agg_type) alone, bypassing window_compatible entirely -- a Sliding DeltaSetAggregator built by any other path could still be selected to serve a query and silently produce incorrect merged add/remove sets. Add key_agg_window_valid() as the single source of truth for this invariant, wired into both window_compatible() (the value-slot matching path) and the key-agg find() filter (the actual pairing path CountMinSketch/HydraKLL rely on). When a key agg exists on the metric but gets filtered out for this reason, log the specific cause instead of the generic "none found" warning. Co-Authored-By: Claude Sonnet 5 --- .../rs/asap_types/src/capability_matching.rs | 190 +++++++++++++++++- 1 file changed, 182 insertions(+), 8 deletions(-) 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 5ddfe135..cc748492 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -60,12 +60,29 @@ fn is_key_agg_type(agg_type: AggregationType) -> bool { agg_type.is_key_agg_type() } +/// Whether `agg_type` may legitimately be planned/served with `window_type`. +/// +/// DeltaSetAggregator only tracks added/removed keys since the last window, +/// so it's only correct for non-overlapping (tumbling) windows (#588) -- +/// unlike its sibling key aggregations (SetAggregator, HLL), which are fine +/// under Sliding. This is the single source of truth for that invariant; +/// both producer sites (the live planner's agg_config.rs and the optimizer's +/// candidate_gen.rs) are expected to never emit a Sliding DeltaSetAggregator +/// config, but this predicate is the defensive check that stops one from +/// ever being *selected* to serve a query, regardless of how it was produced. +pub fn key_agg_window_valid(agg_type: AggregationType, window_type: WindowType) -> bool { + !(agg_type == AggregationType::DeltaSetAggregator && window_type == WindowType::Sliding) +} + /// 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). pub fn window_compatible(config: &AggregationConfig, data_range_ms: u64) -> bool { + if !key_agg_window_valid(config.aggregation_type, config.window_type) { + return false; + } let window_ms = config.window_size_ms; if window_ms == 0 || data_range_ms == 0 { return false; @@ -253,15 +270,33 @@ pub fn find_compatible_aggregation( // separate key aggregation just like any other multi-population value type. let key_agg: &AggregationConfig = if is_multi_population_value_type(value_agg.aggregation_type) { - let ka = configs - .values() - .find(|c| c.metric == requirements.metric && is_key_agg_type(c.aggregation_type)); + let is_key_agg_on_metric = |c: &&AggregationConfig| { + c.metric == requirements.metric && is_key_agg_type(c.aggregation_type) + }; + let ka = configs.values().find(|c| { + is_key_agg_on_metric(c) && key_agg_window_valid(c.aggregation_type, c.window_type) + }); if ka.is_none() { - warn!( - metric = %requirements.metric, - value_agg_type = %value_agg.aggregation_type, - "capability matching: multi-population value agg requires a key agg (SetAggregator/DeltaSetAggregator) but none found", - ); + // Distinguish "a key agg exists but its window_type is invalid for + // its aggregation_type" (e.g. a Sliding DeltaSetAggregator) from + // "genuinely no key agg on this metric" -- both mean "treat as + // absent" for matching purposes, but the former is a silent + // planning bug worth calling out specifically (#588). + let invalid_window = configs.values().find(is_key_agg_on_metric); + match invalid_window { + Some(bad) => warn!( + metric = %requirements.metric, + value_agg_type = %value_agg.aggregation_type, + key_agg_type = %bad.aggregation_type, + key_agg_window_type = ?bad.window_type, + "capability matching: found a key agg on this metric but its window_type is invalid for its aggregation_type (e.g. DeltaSetAggregator must be Tumbling) -- treating as absent", + ), + None => warn!( + metric = %requirements.metric, + value_agg_type = %value_agg.aggregation_type, + "capability matching: multi-population value agg requires a key agg (SetAggregator/DeltaSetAggregator) but none found", + ), + } } ka? } else { @@ -810,6 +845,145 @@ mod tests { assert!(result.is_none()); } + // --- issue #588: DeltaSetAggregator must be restricted to tumbling windows --- + + #[test] + fn key_agg_window_valid_rejects_delta_set_aggregator_sliding() { + assert!(!key_agg_window_valid( + AggregationType::DeltaSetAggregator, + WindowType::Sliding + )); + } + + #[test] + fn key_agg_window_valid_accepts_delta_set_aggregator_tumbling() { + assert!(key_agg_window_valid( + AggregationType::DeltaSetAggregator, + WindowType::Tumbling + )); + } + + #[test] + fn key_agg_window_valid_accepts_set_aggregator_sliding() { + // SetAggregator (unlike DeltaSetAggregator) legitimately supports sliding. + assert!(key_agg_window_valid( + AggregationType::SetAggregator, + WindowType::Sliding + )); + } + + #[test] + fn window_compatible_rejects_sliding_delta_set_aggregator_even_on_exact_range_match() { + // data_range_ms == window_size_ms would normally satisfy the Sliding + // rule -- the rejection must come from the aggregation_type check, + // not the range/window_size arithmetic. + let config = make_config( + 1, + "req", + "DeltaSetAggregator", + "", + 300_000, + "sliding", + &[], + "", + ); + assert!(!window_compatible(&config, 300_000)); + } + + #[test] + fn window_compatible_still_accepts_sliding_set_aggregator() { + let config = make_config(1, "req", "SetAggregator", "", 300_000, "sliding", &[], ""); + assert!(window_compatible(&config, 300_000)); + } + + #[test] + fn window_compatible_still_accepts_tumbling_delta_set_aggregator() { + let config = make_config( + 1, + "req", + "DeltaSetAggregator", + "", + 300_000, + "tumbling", + &[], + "", + ); + assert!(window_compatible(&config, 900_000)); + } + + #[test] + fn multi_pop_rejects_sliding_delta_set_aggregator_key_agg() { + // The key-agg pairing lookup used to match by (metric, is_key_agg_type) + // alone, bypassing window_compatible entirely -- a Sliding + // DeltaSetAggregator could be paired even though it can only ever + // give incorrect merged add/remove sets under sliding windows. + let mut configs = HashMap::new(); + configs.insert( + 10, + make_config( + 10, + "req", + "CountMinSketchWithHeap", + "", + 300_000, + "tumbling", + &[], + "", + ), + ); + configs.insert( + 11, + make_config( + 11, + "req", + "DeltaSetAggregator", + "", + 300_000, + "sliding", + &[], + "", + ), + ); + let result = find_compatible_aggregation( + &configs, + &req("req", &[Statistic::Topk], 300_000, &[], ""), + ); + assert!( + result.is_none(), + "a Sliding DeltaSetAggregator must never be selected as the paired key agg" + ); + } + + #[test] + fn multi_pop_accepts_sliding_set_aggregator_key_agg() { + // Regression guard: the DeltaSetAggregator-specific rejection must not + // block SetAggregator, which legitimately supports sliding windows. + let mut configs = HashMap::new(); + configs.insert( + 10, + make_config( + 10, + "req", + "CountMinSketchWithHeap", + "", + 300_000, + "tumbling", + &[], + "", + ), + ); + configs.insert( + 11, + make_config(11, "req", "SetAggregator", "", 300_000, "sliding", &[], ""), + ); + let result = find_compatible_aggregation( + &configs, + &req("req", &[Statistic::Topk], 300_000, &[], ""), + ); + let info = result.expect("Sliding SetAggregator must still be accepted as a key agg"); + assert_eq!(info.aggregation_id_for_key, 11); + } + // --- avg (Vec) --- #[test] From e611a06840451ae9e8ceb60e9b94944404b1296b Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 25 Aug 2026 13:00:33 -0400 Subject: [PATCH 4/4] refactor: address PR #606 review nits - candidate_gen.rs: call the shared key_agg_window_valid predicate instead of duplicating the DeltaSetAggregator+Sliding check inline, so the invariant has one source of truth. - capability_matching.rs: fold the key-agg miss-path diagnostic into the same single pass over configs.values() instead of re-scanning to find the rejected candidate separately. Co-Authored-By: Claude Sonnet 5 --- .../rs/asap_types/src/capability_matching.rs | 24 +++++++++++-------- .../src/optimizer/candidate_gen.rs | 9 ++++--- 2 files changed, 18 insertions(+), 15 deletions(-) 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 cc748492..1c55e6cf 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -270,19 +270,23 @@ pub fn find_compatible_aggregation( // separate key aggregation just like any other multi-population value type. let key_agg: &AggregationConfig = if is_multi_population_value_type(value_agg.aggregation_type) { - let is_key_agg_on_metric = |c: &&AggregationConfig| { - c.metric == requirements.metric && is_key_agg_type(c.aggregation_type) - }; + // Single pass: take the first window-valid key agg on this metric, + // but also remember the first window-*invalid* one seen (e.g. a + // Sliding DeltaSetAggregator) so the miss path below can report the + // specific reason instead of a generic "none found" (#588). + let mut invalid_window: Option<&AggregationConfig> = None; let ka = configs.values().find(|c| { - is_key_agg_on_metric(c) && key_agg_window_valid(c.aggregation_type, c.window_type) + 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) { + true + } else { + invalid_window.get_or_insert(c); + false + } }); if ka.is_none() { - // Distinguish "a key agg exists but its window_type is invalid for - // its aggregation_type" (e.g. a Sliding DeltaSetAggregator) from - // "genuinely no key agg on this metric" -- both mean "treat as - // absent" for matching purposes, but the former is a silent - // planning bug worth calling out specifically (#588). - let invalid_window = configs.values().find(is_key_agg_on_metric); match invalid_window { Some(bad) => warn!( metric = %requirements.metric, diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs index 495a0ec5..c8665e9f 100644 --- a/asap-planner-rs/src/optimizer/candidate_gen.rs +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use asap_types::aggregation_config::AggregationConfig; -use asap_types::capability_matching::compatible_agg_types; +use asap_types::capability_matching::{compatible_agg_types, key_agg_window_valid}; use asap_types::enums::WindowType; use promql_utilities::data_model::KeyByLabelNames; use promql_utilities::query_logics::enums::{AggregationType, Statistic}; @@ -53,10 +53,9 @@ pub fn enumerate_candidates(aqe: &AQE, scrape_interval_ms: u64) -> Vec