From 88bed76db9b446cab6b7d6a429577e1155e5e95a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 10:45:03 -0400 Subject: [PATCH 01/18] test(planner): cover PromQL sliding window config --- asap-planner-rs/tests/integration.rs | 41 +++++++++++++++++++ .../test_data/windowing/promql_sliding.yaml | 20 +++++++++ 2 files changed, 61 insertions(+) create mode 100644 asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index 18b0f098..2073a143 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -27,6 +27,47 @@ fn http_requests_schema() -> PromQLSchema { ) } +#[test] +fn config_file_sliding_window_override_generates_per_query_candidates() { + let controller = Controller::from_file_with_schema( + Path::new("tests/test_data/windowing/promql_sliding.yaml"), + http_requests_schema(), + arroyo_opts(), + ) + .unwrap(); + + let output = controller.generate().unwrap(); + let streaming: serde_yaml::Value = + serde_yaml::from_str(&output.to_streaming_yaml_string().unwrap()).unwrap(); + let aggregations = streaming["aggregations"].as_sequence().unwrap(); + + assert_eq!(output.inference_query_count(), 2); + assert_eq!(aggregations.len(), 2); + + let mut candidates: Vec<(u64, u64, String)> = aggregations + .iter() + .map(|aggregation| { + ( + aggregation["windowSizeMs"].as_u64().unwrap(), + aggregation["slideIntervalMs"].as_u64().unwrap(), + aggregation["windowType"].as_str().unwrap().to_string(), + ) + }) + .collect(); + candidates.sort(); + + assert_eq!( + candidates, + vec![ + (60_000, 15_000, "sliding".to_string()), + (120_000, 30_000, "sliding".to_string()), + ] + ); + assert!(candidates + .iter() + .all(|(_, _, window_type)| window_type != "tumbling")); +} + /// Schema for binary arithmetic tests: errors_total and requests_total. fn binary_arithmetic_schema() -> PromQLSchema { PromQLSchema::new() diff --git a/asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml b/asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml new file mode 100644 index 00000000..d4be1819 --- /dev/null +++ b/asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml @@ -0,0 +1,20 @@ +windowing: + type: "sliding" + slide_divisor: 4 +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[1m])" + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 + - id: 2 + queries: + - "rate(http_requests_total[2m])" + repetition_delay_ms: 120000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +aggregate_cleanup: + policy: "read_based" From abd492f801a2376b0870b817d3494c712c3afdfc Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:05:27 -0400 Subject: [PATCH 02/18] feat(planner): thread windowing override through PromQL --- asap-planner-rs/src/config/input.rs | 17 ++++++++++ asap-planner-rs/src/optimizer/pipeline.rs | 1 + asap-planner-rs/src/planner/promql.rs | 8 ++++- asap-planner-rs/src/planner/window.rs | 37 ++++++++++++++++++++++ asap-planner-rs/src/promql/generator.rs | 1 + asap-planner-rs/src/query_log/converter.rs | 1 + asap-query-engine/src/planner_client.rs | 1 + 7 files changed, 65 insertions(+), 1 deletion(-) diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs index 413e9f5c..187b70d2 100644 --- a/asap-planner-rs/src/config/input.rs +++ b/asap-planner-rs/src/config/input.rs @@ -10,6 +10,7 @@ use tracing::warn; #[serde(deny_unknown_fields)] pub struct ControllerConfig { pub query_groups: Vec, + pub windowing: Option, pub sketch_parameters: Option, pub aggregate_cleanup: Option, /// Optional hint: per-metric label sets used as a fallback when Prometheus @@ -93,6 +94,21 @@ pub struct AggregateCleanupConfig { pub policy: Option, } +#[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct WindowingConfig { + #[serde(rename = "type")] + pub window_type: WindowingType, + pub slide_divisor: Option, +} + +#[derive(Debug, Clone, Copy, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum WindowingType { + Tumbling, + Sliding, +} + #[derive(Debug, Clone, Deserialize, Default)] pub struct SketchParameterOverrides { #[serde(rename = "CountMinSketch")] @@ -142,6 +158,7 @@ pub struct HllParams { pub struct SQLControllerConfig { pub query_groups: Vec, pub tables: Vec, + pub windowing: Option, pub sketch_parameters: Option, pub aggregate_cleanup: Option, } diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs index 627ea9aa..55f4d1b0 100644 --- a/asap-planner-rs/src/optimizer/pipeline.rs +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -123,6 +123,7 @@ mod tests { ControllerConfig { query_groups, + windowing: None, sketch_parameters: None, aggregate_cleanup: None, metrics: None, diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs index 7257e558..1654b196 100644 --- a/asap-planner-rs/src/planner/promql.rs +++ b/asap-planner-rs/src/planner/promql.rs @@ -7,7 +7,7 @@ use promql_utilities::query_logics::enums::{ }; use promql_utilities::query_logics::parsing::get_metric_and_spatial_filter; -use crate::config::input::SketchParameterOverrides; +use crate::config::input::{SketchParameterOverrides, WindowingConfig}; use crate::error::ControllerError; use crate::planner::agg_config::{build_agg_configs_for_statistics, IntermediateAggConfig}; use crate::planner::cleanup::get_cleanup_param; @@ -77,6 +77,7 @@ pub struct SingleQueryProcessor { range_duration_ms: u64, step_ms: u64, cleanup_policy: CleanupPolicy, + windowing: Option, } impl SingleQueryProcessor { @@ -91,6 +92,7 @@ impl SingleQueryProcessor { range_duration_ms: u64, step_ms: u64, cleanup_policy: CleanupPolicy, + windowing: Option, ) -> Self { Self { query, @@ -102,6 +104,7 @@ impl SingleQueryProcessor { range_duration_ms, step_ms, cleanup_policy, + windowing, } } @@ -154,6 +157,7 @@ impl SingleQueryProcessor { self.range_duration_ms, self.step_ms, self.cleanup_policy, + self.windowing.clone(), ) } @@ -254,6 +258,8 @@ impl SingleQueryProcessor { &mut window_cfg, ) .map_err(ControllerError::PlannerError)?; + crate::planner::window::apply_windowing_override(&mut window_cfg, self.windowing.as_ref()) + .map_err(ControllerError::PlannerError)?; let subpopulation_labels = requirements.grouping_labels; let rollup = all_labels.difference(&subpopulation_labels); diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs index 7cbc9be9..9699d512 100644 --- a/asap-planner-rs/src/planner/window.rs +++ b/asap-planner-rs/src/planner/window.rs @@ -1,5 +1,7 @@ use asap_types::enums::WindowType; +use crate::config::input::{WindowingConfig, WindowingType}; + pub fn get_effective_repeat(t_repeat_ms: u64, step_ms: u64) -> u64 { if step_ms > 0 { t_repeat_ms.min(step_ms) @@ -88,6 +90,41 @@ pub fn set_window_parameters( Ok(()) } +pub fn apply_windowing_override( + config: &mut IntermediateWindowConfig, + windowing: Option<&WindowingConfig>, +) -> Result<(), String> { + let Some(windowing) = windowing else { + return Ok(()); + }; + + match windowing.window_type { + WindowingType::Tumbling => { + config.window_type = WindowType::Tumbling; + config.slide_interval_ms = config.window_size_ms; + } + WindowingType::Sliding => { + let divisor = windowing.slide_divisor.ok_or_else(|| { + "windowing.slide_divisor is required for sliding windows".to_string() + })?; + if divisor < 2 { + return Err(format!( + "windowing.slide_divisor must be at least 2, got {divisor}" + )); + } + if !config.window_size_ms.is_multiple_of(divisor) { + return Err(format!( + "window_size_ms ({}) must be evenly divisible by slide_divisor ({divisor})", + config.window_size_ms + )); + } + config.window_type = WindowType::Sliding; + config.slide_interval_ms = config.window_size_ms / divisor; + } + } + Ok(()) +} + /// A mutable window config holder used during planning #[derive(Debug, Clone, Default)] pub struct IntermediateWindowConfig { diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index b211f665..2359c3eb 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -67,6 +67,7 @@ pub fn generate_plan( qg.range_duration_ms.unwrap_or(opts.range_duration_ms), qg.step_ms.unwrap_or(opts.step_ms), cleanup_policy, + controller_config.windowing.clone(), ); let mut should_process = processor.is_supported(); diff --git a/asap-planner-rs/src/query_log/converter.rs b/asap-planner-rs/src/query_log/converter.rs index 9a9d70e3..756dcd6b 100644 --- a/asap-planner-rs/src/query_log/converter.rs +++ b/asap-planner-rs/src/query_log/converter.rs @@ -37,6 +37,7 @@ pub fn to_controller_config( ControllerConfig { query_groups, + windowing: None, sketch_parameters: None, aggregate_cleanup: Some(AggregateCleanupConfig { policy: Some(CleanupPolicy::ReadBased), diff --git a/asap-query-engine/src/planner_client.rs b/asap-query-engine/src/planner_client.rs index b2b794d6..5d3f3e7c 100644 --- a/asap-query-engine/src/planner_client.rs +++ b/asap-query-engine/src/planner_client.rs @@ -113,6 +113,7 @@ mod tests { step_ms: None, range_duration_ms: None, }], + windowing: None, metrics: Some(vec![MetricDefinition { metric: "http_requests_total".to_string(), labels: vec!["method".to_string(), "status".to_string()], From 9007d8454ce2f428e1dd633372af0bc089a1bd6f Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:07:02 -0400 Subject: [PATCH 03/18] test(planner): cover SQL sliding window config --- asap-planner-rs/tests/sql_integration.rs | 40 +++++++++++++++++++ .../test_data/windowing/sql_sliding.yaml | 31 ++++++++++++++ 2 files changed, 71 insertions(+) create mode 100644 asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index 28a61072..4c7587bd 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -11,6 +11,46 @@ fn sql_opts() -> SQLRuntimeOptions { } } +#[test] +fn config_file_sliding_window_override_generates_per_query_candidates() { + let controller = SQLController::from_file( + std::path::Path::new("tests/test_data/windowing/sql_sliding.yaml"), + sql_opts(), + ) + .unwrap(); + + let output = controller.generate().unwrap(); + let streaming: serde_yaml::Value = + serde_yaml::from_str(&output.to_streaming_yaml_string().unwrap()).unwrap(); + let aggregations = streaming["aggregations"].as_sequence().unwrap(); + + assert_eq!(output.inference_query_count(), 2); + assert_eq!(aggregations.len(), 2); + + let mut candidates: Vec<(u64, u64, String)> = aggregations + .iter() + .map(|aggregation| { + ( + aggregation["windowSizeMs"].as_u64().unwrap(), + aggregation["slideIntervalMs"].as_u64().unwrap(), + aggregation["windowType"].as_str().unwrap().to_string(), + ) + }) + .collect(); + candidates.sort(); + + assert_eq!( + candidates, + vec![ + (60_000, 15_000, "sliding".to_string()), + (120_000, 30_000, "sliding".to_string()), + ] + ); + assert!(candidates + .iter() + .all(|(_, _, window_type)| window_type != "tumbling")); +} + /// Single-query config with a 3-column metadata schema. /// /// Schema: metrics_table diff --git a/asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml b/asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml new file mode 100644 index 00000000..54c8489a --- /dev/null +++ b/asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml @@ -0,0 +1,31 @@ +windowing: + type: "sliding" + slide_divisor: 4 +tables: + - name: metrics_table + time_column: time + value_columns: [cpu_usage] + metadata_columns: [hostname, datacenter, region] +query_groups: + - id: 1 + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.95 + latency_sla: 100.0 + queries: + - >- + SELECT MIN(cpu_usage) FROM metrics_table + WHERE time BETWEEN DATEADD(s, -60, NOW()) AND NOW() + GROUP BY datacenter + - id: 2 + repetition_delay_ms: 120000 + controller_options: + accuracy_sla: 0.95 + latency_sla: 100.0 + queries: + - >- + SELECT MIN(cpu_usage) FROM metrics_table + WHERE time BETWEEN DATEADD(s, -120, NOW()) AND NOW() + GROUP BY datacenter +aggregate_cleanup: + policy: "read_based" From 2d9d759e58c6b078f940b99fc27550f5c3e37a82 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:09:57 -0400 Subject: [PATCH 04/18] feat(planner): thread windowing override through SQL --- asap-planner-rs/src/planner/sql.rs | 9 +++++++-- asap-planner-rs/src/sql/generator.rs | 1 + 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs index f7d13f2a..8d5fa2c2 100644 --- a/asap-planner-rs/src/planner/sql.rs +++ b/asap-planner-rs/src/planner/sql.rs @@ -10,7 +10,7 @@ use sql_utilities::ast_matching::SQLSchema; use sqlparser::dialect::ClickHouseDialect; use sqlparser::parser::Parser as SqlParser; -use crate::config::input::{SketchParameterOverrides, TableDefinition}; +use crate::config::input::{SketchParameterOverrides, TableDefinition, WindowingConfig}; use crate::error::ControllerError; use crate::planner::agg_config::{build_agg_configs_for_statistics, IntermediateAggConfig}; use crate::planner::cleanup::get_sql_cleanup_param; @@ -27,6 +27,7 @@ pub struct SQLSingleQueryProcessor { streaming_engine: StreamingEngine, sketch_parameters: Option, cleanup_policy: CleanupPolicy, + windowing: Option, } impl SQLSingleQueryProcessor { @@ -39,6 +40,7 @@ impl SQLSingleQueryProcessor { streaming_engine: StreamingEngine, sketch_parameters: Option, cleanup_policy: CleanupPolicy, + windowing: Option, ) -> Self { Self { query_string, @@ -48,6 +50,7 @@ impl SQLSingleQueryProcessor { streaming_engine, sketch_parameters, cleanup_policy, + windowing, } } @@ -100,11 +103,13 @@ impl SQLSingleQueryProcessor { let value_column = agg_info.get_value_column_name().to_string(); // Compute window - let window_cfg = compute_sql_window( + let mut window_cfg = compute_sql_window( &sql_query.query_data[0].time_info, self.data_ingestion_interval_ms, self.t_repeat_ms, )?; + crate::planner::window::apply_windowing_override(&mut window_cfg, self.windowing.as_ref()) + .map_err(ControllerError::PlannerError)?; // Get all metadata columns for the table let all_metadata = get_all_metadata_columns(&self.table_definitions, table_name)?; diff --git a/asap-planner-rs/src/sql/generator.rs b/asap-planner-rs/src/sql/generator.rs index 560f35f1..bcde4941 100644 --- a/asap-planner-rs/src/sql/generator.rs +++ b/asap-planner-rs/src/sql/generator.rs @@ -85,6 +85,7 @@ pub fn generate_sql_plan( opts.streaming_engine, config.sketch_parameters.clone(), cleanup_policy, + config.windowing.clone(), ); let (configs, cleanup_param) = From 9eb1f893ce55bdeac878366f2392f75ee47d4385 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:14:48 -0400 Subject: [PATCH 05/18] test(planner): reject sliding config without divisor --- asap-planner-rs/tests/integration.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index 2073a143..7194896c 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -68,6 +68,27 @@ fn config_file_sliding_window_override_generates_per_query_candidates() { .all(|(_, _, window_type)| window_type != "tumbling")); } +#[test] +fn sliding_window_config_requires_a_slide_divisor() { + let result = Controller::from_yaml_with_schema( + r#" +windowing: + type: "sliding" +query_groups: [] +"#, + http_requests_schema(), + arroyo_opts(), + ); + let error = match result { + Ok(_) => panic!("sliding config without a divisor should be rejected"), + Err(error) => error, + }; + + assert!(error + .to_string() + .contains("windowing.slide_divisor is required for sliding windows")); +} + /// Schema for binary arithmetic tests: errors_total and requests_total. fn binary_arithmetic_schema() -> PromQLSchema { PromQLSchema::new() From cb8ca1473464c4900ac9bd4fa67d82ad847ffbac Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:16:02 -0400 Subject: [PATCH 06/18] fix(planner): validate sliding window config --- asap-planner-rs/src/config/input.rs | 18 ++++++++++++++++++ asap-planner-rs/src/planner/window.rs | 10 ++-------- asap-planner-rs/src/promql/controller.rs | 15 +++++++++++++++ asap-planner-rs/src/sql/controller.rs | 5 +++++ 4 files changed, 40 insertions(+), 8 deletions(-) diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs index 187b70d2..20014856 100644 --- a/asap-planner-rs/src/config/input.rs +++ b/asap-planner-rs/src/config/input.rs @@ -102,6 +102,24 @@ pub struct WindowingConfig { pub slide_divisor: Option, } +impl WindowingConfig { + pub fn validate(&self) -> Result<(), String> { + match self.window_type { + WindowingType::Tumbling if self.slide_divisor.is_some() => { + Err("windowing.slide_divisor is only valid for sliding windows".to_string()) + } + WindowingType::Sliding => match self.slide_divisor { + None => Err("windowing.slide_divisor is required for sliding windows".to_string()), + Some(divisor) if divisor < 2 => Err(format!( + "windowing.slide_divisor must be at least 2, got {divisor}" + )), + Some(_) => Ok(()), + }, + WindowingType::Tumbling => Ok(()), + } + } +} + #[derive(Debug, Clone, Copy, Deserialize)] #[serde(rename_all = "lowercase")] pub enum WindowingType { diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs index 9699d512..b51def10 100644 --- a/asap-planner-rs/src/planner/window.rs +++ b/asap-planner-rs/src/planner/window.rs @@ -97,6 +97,7 @@ pub fn apply_windowing_override( let Some(windowing) = windowing else { return Ok(()); }; + windowing.validate()?; match windowing.window_type { WindowingType::Tumbling => { @@ -104,14 +105,7 @@ pub fn apply_windowing_override( config.slide_interval_ms = config.window_size_ms; } WindowingType::Sliding => { - let divisor = windowing.slide_divisor.ok_or_else(|| { - "windowing.slide_divisor is required for sliding windows".to_string() - })?; - if divisor < 2 { - return Err(format!( - "windowing.slide_divisor must be at least 2, got {divisor}" - )); - } + let divisor = windowing.slide_divisor.expect("validated sliding divisor"); if !config.window_size_ms.is_multiple_of(divisor) { return Err(format!( "window_size_ms ({}) must be evenly divisible by slide_divisor ({divisor})", diff --git a/asap-planner-rs/src/promql/controller.rs b/asap-planner-rs/src/promql/controller.rs index 6e85f7f6..be08fd36 100644 --- a/asap-planner-rs/src/promql/controller.rs +++ b/asap-planner-rs/src/promql/controller.rs @@ -37,6 +37,11 @@ impl Controller { ) -> Result { let yaml_str = std::fs::read_to_string(path)?; let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; + if let Some(windowing) = &config.windowing { + windowing + .validate() + .map_err(ControllerError::PlannerError)?; + } config.warn_default_slas(); let all_queries: Vec = config .query_groups @@ -79,6 +84,11 @@ impl Controller { ) -> Result { let yaml_str = std::fs::read_to_string(path)?; let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; + if let Some(windowing) = &config.windowing { + windowing + .validate() + .map_err(ControllerError::PlannerError)?; + } config.warn_default_slas(); Ok(Self { config, @@ -94,6 +104,11 @@ impl Controller { opts: RuntimeOptions, ) -> Result { let config: ControllerConfig = serde_yaml::from_str(yaml)?; + if let Some(windowing) = &config.windowing { + windowing + .validate() + .map_err(ControllerError::PlannerError)?; + } config.warn_default_slas(); Ok(Self { config, diff --git a/asap-planner-rs/src/sql/controller.rs b/asap-planner-rs/src/sql/controller.rs index a1ebeb47..faf96124 100644 --- a/asap-planner-rs/src/sql/controller.rs +++ b/asap-planner-rs/src/sql/controller.rs @@ -67,6 +67,11 @@ impl SQLController { pub fn from_yaml(yaml: &str, opts: SQLRuntimeOptions) -> Result { let config: SQLControllerConfig = serde_yaml::from_str(yaml)?; + if let Some(windowing) = &config.windowing { + windowing + .validate() + .map_err(ControllerError::PlannerError)?; + } Ok(Self { config, options: opts, From fd070a0a88944e6d212250a148dba8d110548af6 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:17:52 -0400 Subject: [PATCH 07/18] test(planner): aggregate invalid sliding windows --- asap-planner-rs/tests/integration.rs | 39 ++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index 7194896c..20b86d81 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -89,6 +89,45 @@ query_groups: [] .contains("windowing.slide_divisor is required for sliding windows")); } +#[test] +fn sliding_window_validation_reports_all_invalid_queries() { + let controller = Controller::from_yaml_with_schema( + r#" +windowing: + type: "sliding" + slide_divisor: 3 +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[65s])" + repetition_delay_ms: 65000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 + - id: 2 + queries: + - "rate(http_requests_total[55s])" + repetition_delay_ms: 55000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +"#, + http_requests_schema(), + arroyo_opts(), + ) + .unwrap(); + + let result = controller.generate(); + let error = match result { + Ok(_) => panic!("invalid sliding windows should abort plan generation"), + Err(error) => error, + }; + let message = error.to_string(); + + assert!(message.contains("rate(http_requests_total[65s])")); + assert!(message.contains("rate(http_requests_total[55s])")); +} + /// Schema for binary arithmetic tests: errors_total and requests_total. fn binary_arithmetic_schema() -> PromQLSchema { PromQLSchema::new() From b483301192bf0fdeb369d024a89129f7a176a192 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:19:08 -0400 Subject: [PATCH 08/18] fix(planner): aggregate PromQL window validation errors --- asap-planner-rs/src/promql/generator.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index 2359c3eb..ede5c30f 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -54,6 +54,7 @@ pub fn generate_plan( let mut query_keys_map: IndexMap)>> = IndexMap::new(); let mut punted_queries: Vec = Vec::new(); + let mut windowing_errors: Vec = Vec::new(); for qg in &controller_config.query_groups { for query_string in &qg.queries { @@ -98,6 +99,11 @@ pub fn generate_plan( "skipping query referencing unknown metric" ); } + Err(ControllerError::PlannerError(message)) + if message.starts_with("window_size_ms (") => + { + windowing_errors.push(format!("query '{query_string}': {message}")); + } Err(e) => return Err(e), } } else if let Some(arm_entries) = @@ -112,6 +118,13 @@ pub fn generate_plan( } } + if !windowing_errors.is_empty() { + return Err(ControllerError::PlannerError(format!( + "sliding window validation failed:\n{}", + windowing_errors.join("\n") + ))); + } + // Assign sequential IDs (1-indexed, insertion order) let mut id_map: HashMap = HashMap::new(); for (idx, key) in dedup_map.keys().enumerate() { From 2e5299f50ca227309c765f7800a6a8beb753497e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:21:14 -0400 Subject: [PATCH 09/18] test(planner): aggregate SQL window validation errors --- asap-planner-rs/tests/sql_integration.rs | 19 ++++++++++++ .../windowing/sql_sliding_invalid.yaml | 31 +++++++++++++++++++ 2 files changed, 50 insertions(+) create mode 100644 asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index 4c7587bd..41b9ddf0 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -51,6 +51,25 @@ fn config_file_sliding_window_override_generates_per_query_candidates() { .all(|(_, _, window_type)| window_type != "tumbling")); } +#[test] +fn sliding_window_validation_reports_all_invalid_sql_queries() { + let controller = SQLController::from_file( + std::path::Path::new("tests/test_data/windowing/sql_sliding_invalid.yaml"), + sql_opts(), + ) + .unwrap(); + + let result = controller.generate(); + let error = match result { + Ok(_) => panic!("invalid sliding windows should abort plan generation"), + Err(error) => error, + }; + let message = error.to_string(); + + assert!(message.contains("DATEADD(s, -65, NOW())")); + assert!(message.contains("DATEADD(s, -55, NOW())")); +} + /// Single-query config with a 3-column metadata schema. /// /// Schema: metrics_table diff --git a/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml b/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml new file mode 100644 index 00000000..0541ef9f --- /dev/null +++ b/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml @@ -0,0 +1,31 @@ +windowing: + type: "sliding" + slide_divisor: 3 +tables: + - name: metrics_table + time_column: time + value_columns: [cpu_usage] + metadata_columns: [hostname, datacenter, region] +query_groups: + - id: 1 + repetition_delay_ms: 65000 + controller_options: + accuracy_sla: 0.95 + latency_sla: 100.0 + queries: + - >- + SELECT MIN(cpu_usage) FROM metrics_table + WHERE time BETWEEN DATEADD(s, -65, NOW()) AND NOW() + GROUP BY datacenter + - id: 2 + repetition_delay_ms: 55000 + controller_options: + accuracy_sla: 0.95 + latency_sla: 100.0 + queries: + - >- + SELECT MIN(cpu_usage) FROM metrics_table + WHERE time BETWEEN DATEADD(s, -55, NOW()) AND NOW() + GROUP BY datacenter +aggregate_cleanup: + policy: "read_based" From ea3e97b3200b01c00e5d73958feb53adc6abf909 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:24:09 -0400 Subject: [PATCH 10/18] fix(planner): aggregate SQL window validation errors --- asap-planner-rs/src/sql/generator.rs | 19 ++++++++++++++++++- asap-planner-rs/tests/sql_integration.rs | 4 ++-- .../windowing/sql_sliding_invalid.yaml | 10 +++++----- 3 files changed, 25 insertions(+), 8 deletions(-) diff --git a/asap-planner-rs/src/sql/generator.rs b/asap-planner-rs/src/sql/generator.rs index bcde4941..b1231ff7 100644 --- a/asap-planner-rs/src/sql/generator.rs +++ b/asap-planner-rs/src/sql/generator.rs @@ -74,6 +74,7 @@ pub fn generate_sql_plan( let mut dedup_map: IndexMap = IndexMap::new(); // query_string -> Vec<(key, cleanup_param)> let mut query_keys_map: IndexMap)>> = IndexMap::new(); + let mut windowing_errors: Vec = Vec::new(); for qg in &config.query_groups { for query_string in &qg.queries { @@ -89,7 +90,16 @@ pub fn generate_sql_plan( ); let (configs, cleanup_param) = - processor.get_streaming_aggregation_configs(eval_time)?; + match processor.get_streaming_aggregation_configs(eval_time) { + Ok(result) => result, + Err(ControllerError::PlannerError(message)) + if message.starts_with("window_size_ms (") => + { + windowing_errors.push(format!("query '{query_string}': {message}")); + continue; + } + Err(error) => return Err(error), + }; let mut keys_for_query = Vec::new(); for config_item in configs { @@ -101,6 +111,13 @@ pub fn generate_sql_plan( } } + if !windowing_errors.is_empty() { + return Err(ControllerError::PlannerError(format!( + "sliding window validation failed:\n{}", + windowing_errors.join("\n") + ))); + } + // Assign sequential IDs let mut id_map: HashMap = HashMap::new(); for (idx, key) in dedup_map.keys().enumerate() { diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index 41b9ddf0..aae0e018 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -66,8 +66,8 @@ fn sliding_window_validation_reports_all_invalid_sql_queries() { }; let message = error.to_string(); - assert!(message.contains("DATEADD(s, -65, NOW())")); - assert!(message.contains("DATEADD(s, -55, NOW())")); + assert!(message.contains("DATEADD(s, -75, NOW())")); + assert!(message.contains("DATEADD(s, -90, NOW())")); } /// Single-query config with a 3-column metadata schema. diff --git a/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml b/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml index 0541ef9f..a207d6ce 100644 --- a/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml +++ b/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml @@ -1,6 +1,6 @@ windowing: type: "sliding" - slide_divisor: 3 + slide_divisor: 7 tables: - name: metrics_table time_column: time @@ -8,24 +8,24 @@ tables: metadata_columns: [hostname, datacenter, region] query_groups: - id: 1 - repetition_delay_ms: 65000 + repetition_delay_ms: 75000 controller_options: accuracy_sla: 0.95 latency_sla: 100.0 queries: - >- SELECT MIN(cpu_usage) FROM metrics_table - WHERE time BETWEEN DATEADD(s, -65, NOW()) AND NOW() + WHERE time BETWEEN DATEADD(s, -75, NOW()) AND NOW() GROUP BY datacenter - id: 2 - repetition_delay_ms: 55000 + repetition_delay_ms: 90000 controller_options: accuracy_sla: 0.95 latency_sla: 100.0 queries: - >- SELECT MIN(cpu_usage) FROM metrics_table - WHERE time BETWEEN DATEADD(s, -55, NOW()) AND NOW() + WHERE time BETWEEN DATEADD(s, -90, NOW()) AND NOW() GROUP BY datacenter aggregate_cleanup: policy: "read_based" From dd68340a46dc6f84344a7fb5b434d5c4441fd39c Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:25:33 -0400 Subject: [PATCH 11/18] test(planner): reject redundant sliding divisor --- asap-planner-rs/tests/integration.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index 20b86d81..c7341295 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -128,6 +128,28 @@ query_groups: assert!(message.contains("rate(http_requests_total[55s])")); } +#[test] +fn sliding_window_config_rejects_redundant_divisors() { + let result = Controller::from_yaml_with_schema( + r#" +windowing: + type: "sliding" + slide_divisor: 1 +query_groups: [] +"#, + http_requests_schema(), + arroyo_opts(), + ); + let error = match result { + Ok(_) => panic!("sliding config with divisor 1 should be rejected"), + Err(error) => error, + }; + + assert!(error + .to_string() + .contains("windowing.slide_divisor must be at least 2, got 1")); +} + /// Schema for binary arithmetic tests: errors_total and requests_total. fn binary_arithmetic_schema() -> PromQLSchema { PromQLSchema::new() From 23547e0bc27dd71df70965e437f24b757bb72aee Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:26:50 -0400 Subject: [PATCH 12/18] test(planner): reject divisor on tumbling config --- asap-planner-rs/tests/integration.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index c7341295..d16aaa9b 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -150,6 +150,28 @@ query_groups: [] .contains("windowing.slide_divisor must be at least 2, got 1")); } +#[test] +fn tumbling_window_config_rejects_a_slide_divisor() { + let result = Controller::from_yaml_with_schema( + r#" +windowing: + type: "tumbling" + slide_divisor: 4 +query_groups: [] +"#, + http_requests_schema(), + arroyo_opts(), + ); + let error = match result { + Ok(_) => panic!("tumbling config with a divisor should be rejected"), + Err(error) => error, + }; + + assert!(error + .to_string() + .contains("windowing.slide_divisor is only valid for sliding windows")); +} + /// Schema for binary arithmetic tests: errors_total and requests_total. fn binary_arithmetic_schema() -> PromQLSchema { PromQLSchema::new() From 68481191cf361e4279ec7e39f20cc9d0045a4ca5 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 11:32:00 -0400 Subject: [PATCH 13/18] docs(planner): document windowing override --- asap-planner-rs/src/lib.rs | 1 + asap-planner-rs/src/planner/window.rs | 9 ++++- asap-planner-rs/src/promql/generator.rs | 6 ++++ asap-planner-rs/src/sql/generator.rs | 6 ++++ asap-planner-rs/tests/integration.rs | 33 +++++++++++++++++++ .../CONFIG_PARAMETERS_REFERENCE.md | 19 +++++++++++ asap-tools/experiments/config/config.yaml | 7 ++++ .../experiment_only_ingest_path.py | 1 + .../experiments/experiment_run_clickhouse.py | 5 ++- asap-tools/experiments/experiment_run_e2e.py | 1 + .../experiment_run_grafana_demo.py | 1 + .../experiments/experiment_utils/config.py | 19 ++++++++++- 12 files changed, 105 insertions(+), 3 deletions(-) diff --git a/asap-planner-rs/src/lib.rs b/asap-planner-rs/src/lib.rs index 9f83c515..a59bda27 100644 --- a/asap-planner-rs/src/lib.rs +++ b/asap-planner-rs/src/lib.rs @@ -15,6 +15,7 @@ pub use asap_types::PromQLSchema; pub use config::input::ControllerConfig; pub use config::input::ElasticDSLControllerConfig; pub use config::input::SQLControllerConfig; +pub use config::input::{WindowingConfig, WindowingType}; pub use elastic_dsl::ElasticController; pub use elastic_dsl::ElasticIndexSchemaBuilder; pub use elastic_dsl::ElasticRuntimeOptions; diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs index b51def10..c0a8dbe4 100644 --- a/asap-planner-rs/src/planner/window.rs +++ b/asap-planner-rs/src/planner/window.rs @@ -105,7 +105,14 @@ pub fn apply_windowing_override( config.slide_interval_ms = config.window_size_ms; } WindowingType::Sliding => { - let divisor = windowing.slide_divisor.expect("validated sliding divisor"); + let divisor = match windowing.slide_divisor { + Some(divisor) => divisor, + None => { + return Err( + "windowing.slide_divisor is required for sliding windows".to_string() + ) + } + }; if !config.window_size_ms.is_multiple_of(divisor) { return Err(format!( "window_size_ms ({}) must be evenly divisible by slide_divisor ({divisor})", diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index ede5c30f..0dd77e34 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -31,6 +31,12 @@ pub fn generate_plan( ) -> Result { let metric_schema = schema.clone(); + if let Some(windowing) = &controller_config.windowing { + windowing + .validate() + .map_err(ControllerError::PlannerError)?; + } + // Determine cleanup policy let cleanup_policy = controller_config .aggregate_cleanup diff --git a/asap-planner-rs/src/sql/generator.rs b/asap-planner-rs/src/sql/generator.rs index b1231ff7..5b70c0fc 100644 --- a/asap-planner-rs/src/sql/generator.rs +++ b/asap-planner-rs/src/sql/generator.rs @@ -25,6 +25,12 @@ pub fn generate_sql_plan( config: &SQLControllerConfig, opts: &SQLRuntimeOptions, ) -> Result { + if let Some(windowing) = &config.windowing { + windowing + .validate() + .map_err(ControllerError::PlannerError)?; + } + let eval_time: f64 = opts.query_evaluation_time.unwrap_or_else(|| { SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index d16aaa9b..e2eb6ac2 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -172,6 +172,39 @@ query_groups: [] .contains("windowing.slide_divisor is only valid for sliding windows")); } +#[test] +fn explicit_tumbling_window_override_keeps_tumbling_candidates() { + let controller = Controller::from_yaml_with_schema( + r#" +windowing: + type: "tumbling" +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[1m])" + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +"#, + http_requests_schema(), + arroyo_opts(), + ) + .unwrap(); + + let output = controller.generate().unwrap(); + let streaming: serde_yaml::Value = + serde_yaml::from_str(&output.to_streaming_yaml_string().unwrap()).unwrap(); + let aggregation = &streaming["aggregations"][0]; + + assert_eq!(aggregation["windowType"].as_str(), Some("tumbling")); + assert_eq!( + aggregation["slideIntervalMs"].as_u64(), + aggregation["windowSizeMs"].as_u64() + ); + assert_ne!(aggregation["windowType"].as_str(), Some("sliding")); +} + /// Schema for binary arithmetic tests: errors_total and requests_total. fn binary_arithmetic_schema() -> PromQLSchema { PromQLSchema::new() diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index fba9c6ac..25b7a3e6 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -204,6 +204,25 @@ These parameters must be provided for all experiment scripts: - **Example**: `"10s"` - **Usage**: Controls pre-computed metric updates +## Planner Windowing Override + +#### `windowing` (object, optional) +- **Description**: Global manual override for the planner's window type +- **Default**: Omitted, which preserves tumbling-window planning +- **Choices**: `tumbling`, `sliding` +- **Scope**: Applies to PromQL and SQL planner inputs; ElasticDSL is unchanged + +#### `windowing.type` (string, required when `windowing` is present) +- **Description**: Selects the planner window type +- **Choices**: `tumbling`, `sliding` + +#### `windowing.slide_divisor` (int, required for sliding) +- **Description**: Computes each query's slide interval as + `window_size_ms / slide_divisor` +- **Validation**: Must be at least 2 and must evenly divide every supported + query's computed window size; invalid workloads fail plan generation +- **Constraint**: Do not provide this field for `tumbling` + --- ## Monitoring Configuration diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index 0c6d5ef6..928bc783 100644 --- a/asap-tools/experiments/config/config.yaml +++ b/asap-tools/experiments/config/config.yaml @@ -99,6 +99,13 @@ query_engine: controller: punting: true # Enable query punting based on performance heuristics (should_be_performant check) +# Optional manual planner windowing override. Omit this block to preserve the +# default tumbling-window behavior. For sliding windows, the divisor must be +# at least 2 and must evenly divide every query's computed window size. +# windowing: +# type: "sliding" +# slide_divisor: 4 + # Aggregate cleanup configuration # Policy options: # - "circular_buffer": Keep N most recent aggregates (fixed-count cleanup) diff --git a/asap-tools/experiments/experiment_only_ingest_path.py b/asap-tools/experiments/experiment_only_ingest_path.py index a7c16dc7..d85765d8 100644 --- a/asap-tools/experiments/experiment_only_ingest_path.py +++ b/asap-tools/experiments/experiment_only_ingest_path.py @@ -190,6 +190,7 @@ def main(cfg: DictConfig): local_experiment_root_dir, cfg.aggregate_cleanup, cfg.get("sketch_parameters", None), + cfg.get("windowing", None), ) ) sync.rsync_controller_client_configs( diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index ac218244..d52c7e0c 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -377,7 +377,10 @@ def main(cfg: DictConfig) -> None: # Generate and rsync the planner input config to the node planner_input_yaml = config.generate_sql_planner_input( - ep.query_groups, dataset_cfg, cfg.get("sketch_parameters", None) + ep.query_groups, + dataset_cfg, + cfg.get("sketch_parameters", None), + cfg.get("windowing", None), ) local_planner_input = os.path.join( local_controller_dir, "planner_input.yaml" diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index ce9ccba3..f52c5f08 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -200,6 +200,7 @@ def main(cfg: DictConfig): local_experiment_root_dir, cfg.aggregate_cleanup, cfg.get("sketch_parameters", None), + cfg.get("windowing", None), ) ) sync.rsync_controller_client_configs( diff --git a/asap-tools/experiments/experiment_run_grafana_demo.py b/asap-tools/experiments/experiment_run_grafana_demo.py index d7e98b48..0ecd3ddc 100644 --- a/asap-tools/experiments/experiment_run_grafana_demo.py +++ b/asap-tools/experiments/experiment_run_grafana_demo.py @@ -161,6 +161,7 @@ def main(cfg: DictConfig): local_experiment_root_dir, cfg.aggregate_cleanup, cfg.get("sketch_parameters", None), + cfg.get("windowing", None), ) ) sync.rsync_controller_client_configs( diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index 348c3953..495b663c 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -342,6 +342,7 @@ def generate_controller_client_configs( local_experiment_dir: str, aggregate_cleanup: DictConfig = None, sketch_parameters: DictConfig = None, + windowing: DictConfig = None, ) -> Tuple[List[str], List[str]]: """Generate controller client configurations from experiment parameters.""" # experiment_params is already loaded by Hydra @@ -358,6 +359,10 @@ def generate_controller_client_configs( sketch_params_config = OmegaConf.to_container(sketch_parameters, resolve=True) experiment_config["sketch_parameters"] = sketch_params_config + # Add the optional planner windowing override if provided. + if windowing is not None: + experiment_config["windowing"] = OmegaConf.to_container(windowing, resolve=True) + output_dir = os.path.join(local_experiment_dir, "controller_client_configs") os.makedirs(output_dir, exist_ok=True) @@ -376,6 +381,7 @@ def generate_controller_client_configs( "query_groups", "sketch_parameters", "aggregate_cleanup", + "windowing", "metrics", "existing_streaming_config", } @@ -828,7 +834,10 @@ def generate_clickhouse_client_configs( def generate_sql_planner_input( - query_groups: Any, dataset_cfg: Any, sketch_parameters: Any = None + query_groups: Any, + dataset_cfg: Any, + sketch_parameters: Any = None, + windowing: Any = None, ) -> str: """Generate the YAML input file for asap-planner in SQL mode. @@ -836,6 +845,7 @@ def generate_sql_planner_input( ``SQLControllerConfig`` YAML that contains: - ``tables``: schema of the tables being queried - ``query_groups``: SQL queries with controller options + - ``windowing``: optional global tumbling/sliding window override - ``sketch_parameters``: optional per-sketch-type overrides (e.g. ``DatasketchesKLL.K``), matching ``ControllerConfig``'s PromQL-mode field of the same name (``SketchParameterOverrides`` in @@ -854,6 +864,9 @@ def generate_sql_planner_input( top-level ``sketch_parameters`` section (``CountMinSketch``, ``DatasketchesKLL``, etc.). When ``None``, the planner falls back to its own defaults. + windowing: Optional DictConfig/dict mirroring ``config.yaml``'s + top-level ``windowing`` section. When ``None``, the planner uses + its default tumbling-window behavior. Returns: YAML string ready to write to disk and pass to asap-planner. @@ -913,6 +926,10 @@ def generate_sql_planner_input( if isinstance(sketch_parameters, (DictConfig, ListConfig)): sketch_parameters = OmegaConf.to_container(sketch_parameters, resolve=True) planner_input["sketch_parameters"] = sketch_parameters + if windowing is not None: + if isinstance(windowing, (DictConfig, ListConfig)): + windowing = OmegaConf.to_container(windowing, resolve=True) + planner_input["windowing"] = windowing return yaml.dump(planner_input, default_flow_style=False, allow_unicode=True) From 0e54a38ac740cfa00b47b5ea3d0dc344cd5daf18 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 13:12:54 -0400 Subject: [PATCH 14/18] fix(planner): harden windowing override validation --- asap-planner-rs/src/error.rs | 4 ++ asap-planner-rs/src/planner/promql.rs | 7 +- asap-planner-rs/src/planner/sql.rs | 9 ++- asap-planner-rs/src/planner/window.rs | 88 +++++++++++++++++++++--- asap-planner-rs/src/promql/generator.rs | 41 +++++++---- asap-planner-rs/src/sql/controller.rs | 5 ++ asap-planner-rs/src/sql/generator.rs | 12 +--- asap-planner-rs/tests/integration.rs | 69 +++++++++++++++++++ asap-planner-rs/tests/sql_integration.rs | 70 +++++++++++++++++++ 9 files changed, 267 insertions(+), 38 deletions(-) diff --git a/asap-planner-rs/src/error.rs b/asap-planner-rs/src/error.rs index f257f41d..3fc5a4cf 100644 --- a/asap-planner-rs/src/error.rs +++ b/asap-planner-rs/src/error.rs @@ -1,5 +1,7 @@ use thiserror::Error; +use crate::planner::window::WindowingError; + #[derive(Debug, Error)] pub enum ControllerError { #[error("IO error: {0}")] @@ -12,6 +14,8 @@ pub enum ControllerError { DuplicateQuery(String), #[error("Planner error: {0}")] PlannerError(String), + #[error("Windowing error: {0}")] + Windowing(#[from] WindowingError), #[error("Unknown metric: {0}")] UnknownMetric(String), #[error("SQL parse error: {0}")] diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs index 1654b196..6136537a 100644 --- a/asap-planner-rs/src/planner/promql.rs +++ b/asap-planner-rs/src/planner/promql.rs @@ -258,8 +258,11 @@ impl SingleQueryProcessor { &mut window_cfg, ) .map_err(ControllerError::PlannerError)?; - crate::planner::window::apply_windowing_override(&mut window_cfg, self.windowing.as_ref()) - .map_err(ControllerError::PlannerError)?; + crate::planner::window::apply_windowing_override( + &mut window_cfg, + requirements.data_range_ms, + self.windowing.as_ref(), + )?; let subpopulation_labels = requirements.grouping_labels; let rollup = all_labels.difference(&subpopulation_labels); diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs index 8d5fa2c2..4b054f6a 100644 --- a/asap-planner-rs/src/planner/sql.rs +++ b/asap-planner-rs/src/planner/sql.rs @@ -108,8 +108,13 @@ impl SQLSingleQueryProcessor { self.data_ingestion_interval_ms, self.t_repeat_ms, )?; - crate::planner::window::apply_windowing_override(&mut window_cfg, self.windowing.as_ref()) - .map_err(ControllerError::PlannerError)?; + let data_range_ms = + (sql_query.query_data[0].time_info.get_duration() * 1000.0).round() as u64; + crate::planner::window::apply_windowing_override( + &mut window_cfg, + data_range_ms, + self.windowing.as_ref(), + )?; // Get all metadata columns for the table let all_metadata = get_all_metadata_columns(&self.table_definitions, table_name)?; diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs index c0a8dbe4..0515e91f 100644 --- a/asap-planner-rs/src/planner/window.rs +++ b/asap-planner-rs/src/planner/window.rs @@ -1,4 +1,5 @@ use asap_types::enums::WindowType; +use std::fmt; use crate::config::input::{WindowingConfig, WindowingType}; @@ -92,12 +93,15 @@ pub fn set_window_parameters( pub fn apply_windowing_override( config: &mut IntermediateWindowConfig, + data_range_ms: u64, windowing: Option<&WindowingConfig>, -) -> Result<(), String> { +) -> Result<(), WindowingError> { let Some(windowing) = windowing else { return Ok(()); }; - windowing.validate()?; + windowing + .validate() + .map_err(WindowingError::InvalidConfig)?; match windowing.window_type { WindowingType::Tumbling => { @@ -108,16 +112,22 @@ pub fn apply_windowing_override( let divisor = match windowing.slide_divisor { Some(divisor) => divisor, None => { - return Err( - "windowing.slide_divisor is required for sliding windows".to_string() - ) + return Err(WindowingError::InvalidConfig( + "windowing.slide_divisor is required for sliding windows".to_string(), + )); } }; if !config.window_size_ms.is_multiple_of(divisor) { - return Err(format!( - "window_size_ms ({}) must be evenly divisible by slide_divisor ({divisor})", - config.window_size_ms - )); + return Err(WindowingError::WindowSizeNotDivisible { + window_size_ms: config.window_size_ms, + divisor, + }); + } + if !data_range_ms.is_multiple_of(config.window_size_ms) { + return Err(WindowingError::DataRangeNotDivisible { + data_range_ms, + window_size_ms: config.window_size_ms, + }); } config.window_type = WindowType::Sliding; config.slide_interval_ms = config.window_size_ms / divisor; @@ -126,6 +136,43 @@ pub fn apply_windowing_override( Ok(()) } +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WindowingError { + InvalidConfig(String), + WindowSizeNotDivisible { + window_size_ms: u64, + divisor: u64, + }, + DataRangeNotDivisible { + data_range_ms: u64, + window_size_ms: u64, + }, +} + +impl fmt::Display for WindowingError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidConfig(message) => f.write_str(message), + Self::WindowSizeNotDivisible { + window_size_ms, + divisor, + } => write!( + f, + "window_size_ms ({window_size_ms}) must be evenly divisible by slide_divisor ({divisor})" + ), + Self::DataRangeNotDivisible { + data_range_ms, + window_size_ms, + } => write!( + f, + "data_range_ms ({data_range_ms}) must be evenly divisible by window_size_ms ({window_size_ms})" + ), + } + } +} + +impl std::error::Error for WindowingError {} + /// A mutable window config holder used during planning #[derive(Debug, Clone, Default)] pub struct IntermediateWindowConfig { @@ -213,4 +260,27 @@ mod tests { assert!(result.is_err()); assert!(result.unwrap_err().contains("must be a multiple of")); } + + #[test] + fn sliding_override_rejects_data_range_not_multiple_of_window_size() { + let mut config = IntermediateWindowConfig { + window_size_ms: 60_000, + slide_interval_ms: 60_000, + window_type: WindowType::Tumbling, + }; + let windowing = WindowingConfig { + window_type: WindowingType::Sliding, + slide_divisor: Some(4), + }; + + let error = apply_windowing_override(&mut config, 90_000, Some(&windowing)).unwrap_err(); + + assert_eq!( + error, + WindowingError::DataRangeNotDivisible { + data_range_ms: 90_000, + window_size_ms: 60_000, + } + ); + } } diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index 0dd77e34..e4da99e4 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -31,12 +31,6 @@ pub fn generate_plan( ) -> Result { let metric_schema = schema.clone(); - if let Some(windowing) = &controller_config.windowing { - windowing - .validate() - .map_err(ControllerError::PlannerError)?; - } - // Determine cleanup policy let cleanup_policy = controller_config .aggregate_cleanup @@ -105,16 +99,17 @@ pub fn generate_plan( "skipping query referencing unknown metric" ); } - Err(ControllerError::PlannerError(message)) - if message.starts_with("window_size_ms (") => - { - windowing_errors.push(format!("query '{query_string}': {message}")); + Err(ControllerError::Windowing(error)) => { + windowing_errors.push(format!("query '{query_string}': {error}")); } Err(e) => return Err(e), } - } else if let Some(arm_entries) = - collect_binary_leaf_entries(&processor, &mut dedup_map)? - { + } else if let Some(arm_entries) = collect_binary_leaf_entries( + &processor, + &mut dedup_map, + &mut windowing_errors, + query_string, + )? { // Binary arithmetic: register each leaf arm in dedup_map and query_keys_map for (arm_query, keys_for_arm) in arm_entries { // Use `entry` so a standalone query that duplicates an arm wins @@ -162,6 +157,8 @@ pub fn generate_plan( fn collect_binary_leaf_entries( processor: &SingleQueryProcessor, dedup_map: &mut IndexMap, + windowing_errors: &mut Vec, + query_context: &str, ) -> Result, ControllerError> { let arms = match processor.get_binary_arm_queries() { Some(arms) => arms, @@ -181,7 +178,16 @@ fn collect_binary_leaf_entries( if arm_processor.is_supported() { // Leaf arm: gather its streaming aggregation configs. let (configs, cleanup_param) = - arm_processor.get_streaming_aggregation_configs()?; + match arm_processor.get_streaming_aggregation_configs() { + Ok(result) => result, + Err(ControllerError::Windowing(error)) => { + windowing_errors.push(format!( + "query '{query_context}' (leaf '{arm_query}'): {error}" + )); + return Ok(None); + } + Err(error) => return Err(error), + }; let mut keys_for_arm = Vec::new(); for config in configs { let key = config.identifying_key(); @@ -191,7 +197,12 @@ fn collect_binary_leaf_entries( all_entries.push((arm_query, keys_for_arm)); } else { // The arm might itself be a binary expression — recurse. - match collect_binary_leaf_entries(&arm_processor, dedup_map)? { + match collect_binary_leaf_entries( + &arm_processor, + dedup_map, + windowing_errors, + query_context, + )? { Some(sub_entries) => { all_entries.extend(sub_entries); } diff --git a/asap-planner-rs/src/sql/controller.rs b/asap-planner-rs/src/sql/controller.rs index faf96124..d9da13b9 100644 --- a/asap-planner-rs/src/sql/controller.rs +++ b/asap-planner-rs/src/sql/controller.rs @@ -38,6 +38,11 @@ impl SQLController { ) -> Result { let yaml_str = std::fs::read_to_string(path)?; let mut config: SQLControllerConfig = serde_yaml::from_str(&yaml_str)?; + if let Some(windowing) = &config.windowing { + windowing + .validate() + .map_err(ControllerError::PlannerError)?; + } for table in &mut config.tables { if table.metadata_columns.is_empty() { debug!( diff --git a/asap-planner-rs/src/sql/generator.rs b/asap-planner-rs/src/sql/generator.rs index 5b70c0fc..f889259d 100644 --- a/asap-planner-rs/src/sql/generator.rs +++ b/asap-planner-rs/src/sql/generator.rs @@ -25,12 +25,6 @@ pub fn generate_sql_plan( config: &SQLControllerConfig, opts: &SQLRuntimeOptions, ) -> Result { - if let Some(windowing) = &config.windowing { - windowing - .validate() - .map_err(ControllerError::PlannerError)?; - } - let eval_time: f64 = opts.query_evaluation_time.unwrap_or_else(|| { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -98,10 +92,8 @@ pub fn generate_sql_plan( let (configs, cleanup_param) = match processor.get_streaming_aggregation_configs(eval_time) { Ok(result) => result, - Err(ControllerError::PlannerError(message)) - if message.starts_with("window_size_ms (") => - { - windowing_errors.push(format!("query '{query_string}': {message}")); + Err(ControllerError::Windowing(error)) => { + windowing_errors.push(format!("query '{query_string}': {error}")); continue; } Err(error) => return Err(error), diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index e2eb6ac2..d9fd4357 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -128,6 +128,36 @@ query_groups: assert!(message.contains("rate(http_requests_total[55s])")); } +#[test] +fn sliding_window_validation_rejects_data_range_not_multiple_of_window() { + let controller = Controller::from_yaml_with_schema( + r#" +windowing: + type: "sliding" + slide_divisor: 4 +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[90s])" + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +"#, + http_requests_schema(), + arroyo_opts(), + ) + .unwrap(); + + let error = match controller.generate() { + Ok(_) => panic!("invalid sliding window should abort plan generation"), + Err(error) => error.to_string(), + }; + + assert!(error.contains("data_range_ms (90000)")); + assert!(error.contains("window_size_ms (60000)")); +} + #[test] fn sliding_window_config_rejects_redundant_divisors() { let result = Controller::from_yaml_with_schema( @@ -751,6 +781,45 @@ fn binary_arithmetic_with_non_acceleratable_arm_produces_no_configs() { assert_eq!(out.inference_query_count(), 0); } +#[test] +fn binary_arithmetic_aggregates_windowing_errors_from_all_leaves() { + let controller = Controller::from_yaml_with_schema( + r#" +windowing: + type: "sliding" + slide_divisor: 3 +query_groups: + - id: 1 + queries: + - "rate(errors_total[65s]) / rate(requests_total[65s])" + repetition_delay_ms: 65000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 + - id: 2 + queries: + - "rate(errors_total[55s]) / rate(requests_total[55s])" + repetition_delay_ms: 55000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +"#, + binary_arithmetic_schema(), + arroyo_opts(), + ) + .unwrap(); + + let error = match controller.generate() { + Ok(_) => panic!("invalid binary sliding windows should abort plan generation"), + Err(error) => error.to_string(), + }; + + assert!(error.contains("rate(errors_total[65s])")); + assert!(error.contains("rate(requests_total[65s])")); + assert!(error.contains("rate(errors_total[55s])")); + assert!(error.contains("rate(requests_total[55s])")); +} + #[test] fn temporal_overlapping_rate_increase_deduped() { // rate and increase produce identical MultipleIncrease configs → 1 streaming entry shared, diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index aae0e018..8ff0b4ee 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -1,4 +1,6 @@ use asap_planner::{ControllerError, SQLController, SQLRuntimeOptions, StreamingEngine}; +use std::io::Write; +use tempfile::NamedTempFile; // ── helpers ────────────────────────────────────────────────────────────────── @@ -70,6 +72,74 @@ fn sliding_window_validation_reports_all_invalid_sql_queries() { assert!(message.contains("DATEADD(s, -90, NOW())")); } +#[test] +fn sliding_window_validation_rejects_sql_data_range_not_multiple_of_window() { + let query = "SELECT MIN(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -90, NOW()) AND NOW() GROUP BY hostname"; + let config = format!( + r#" +windowing: + type: sliding + slide_divisor: 4 +tables: + - name: metrics_table + time_column: time + value_columns: [cpu_usage] + metadata_columns: [hostname] +query_groups: + - id: 1 + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.95 + latency_sla: 100.0 + queries: + - "{query}" +aggregate_cleanup: + policy: no_cleanup +"# + ); + + let error = match SQLController::from_yaml(&config, sql_opts()) + .unwrap() + .generate() + { + Ok(_) => panic!("invalid sliding window should abort plan generation"), + Err(error) => error.to_string(), + }; + + assert!(error.contains("data_range_ms (90000)")); + assert!(error.contains("window_size_ms (60000)")); +} + +#[test] +fn discovery_validates_windowing_before_contacting_clickhouse() { + let mut file = NamedTempFile::new().unwrap(); + file.write_all( + br#" +windowing: + type: sliding +tables: + - name: metrics_table + time_column: time + value_columns: [cpu_usage] + metadata_columns: [] +query_groups: [] +"#, + ) + .unwrap(); + + let error = match SQLController::from_file_with_discovery( + file.path(), + "http://127.0.0.1:1", + "default", + sql_opts(), + ) { + Ok(_) => panic!("invalid windowing config should be rejected"), + Err(error) => error.to_string(), + }; + + assert!(error.contains("windowing.slide_divisor is required for sliding windows")); +} + /// Single-query config with a 3-column metadata schema. /// /// Schema: metrics_table From d426a44dacd6ab644bf1f3860fc0c76a0cb404b1 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 13:36:11 -0400 Subject: [PATCH 15/18] fix(planner): validate every windowing leaf --- asap-planner-rs/src/promql/generator.rs | 22 ++++++++++++++++++++-- asap-planner-rs/src/sql/generator.rs | 7 +++++++ asap-planner-rs/tests/integration.rs | 1 + 3 files changed, 28 insertions(+), 2 deletions(-) diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index e4da99e4..a14533e5 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -13,6 +13,7 @@ use crate::generator::{ }; use crate::planner::agg_config::IntermediateAggConfig; use crate::planner::promql::{BinaryArm, SingleQueryProcessor}; +use crate::planner::window::WindowingError; use crate::RuntimeOptions; /// `(query_string, Vec<(identifying_key, cleanup_param)>)` pairs produced by binary leaf decomposition. @@ -31,6 +32,12 @@ pub fn generate_plan( ) -> Result { let metric_schema = schema.clone(); + if let Some(windowing) = &controller_config.windowing { + windowing + .validate() + .map_err(|error| ControllerError::Windowing(WindowingError::InvalidConfig(error)))?; + } + // Determine cleanup policy let cleanup_policy = controller_config .aggregate_cleanup @@ -166,6 +173,7 @@ fn collect_binary_leaf_entries( }; let mut all_entries: LeafEntries = Vec::new(); + let mut found_windowing_error = false; for arm in [arms.0, arms.1] { match arm { @@ -184,7 +192,8 @@ fn collect_binary_leaf_entries( windowing_errors.push(format!( "query '{query_context}' (leaf '{arm_query}'): {error}" )); - return Ok(None); + found_windowing_error = true; + continue; } Err(error) => return Err(error), }; @@ -197,6 +206,7 @@ fn collect_binary_leaf_entries( all_entries.push((arm_query, keys_for_arm)); } else { // The arm might itself be a binary expression — recurse. + let error_count = windowing_errors.len(); match collect_binary_leaf_entries( &arm_processor, dedup_map, @@ -207,6 +217,10 @@ fn collect_binary_leaf_entries( all_entries.extend(sub_entries); } None => { + if windowing_errors.len() > error_count { + found_windowing_error = true; + continue; + } // Arm is neither a supported leaf nor a binary expression. // This entire query cannot be accelerated. return Ok(None); @@ -217,7 +231,11 @@ fn collect_binary_leaf_entries( } } - Ok(Some(all_entries)) + if found_windowing_error { + Ok(None) + } else { + Ok(Some(all_entries)) + } } fn build_streaming_yaml( diff --git a/asap-planner-rs/src/sql/generator.rs b/asap-planner-rs/src/sql/generator.rs index f889259d..17c33aab 100644 --- a/asap-planner-rs/src/sql/generator.rs +++ b/asap-planner-rs/src/sql/generator.rs @@ -13,6 +13,7 @@ use crate::generator::{ }; use crate::planner::agg_config::IntermediateAggConfig; use crate::planner::sql::SQLSingleQueryProcessor; +use crate::planner::window::WindowingError; use crate::StreamingEngine; pub struct SQLRuntimeOptions { @@ -25,6 +26,12 @@ pub fn generate_sql_plan( config: &SQLControllerConfig, opts: &SQLRuntimeOptions, ) -> Result { + if let Some(windowing) = &config.windowing { + windowing + .validate() + .map_err(|error| ControllerError::Windowing(WindowingError::InvalidConfig(error)))?; + } + let eval_time: f64 = opts.query_evaluation_time.unwrap_or_else(|| { SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index d9fd4357..6b9cd7e6 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -818,6 +818,7 @@ query_groups: assert!(error.contains("rate(requests_total[65s])")); assert!(error.contains("rate(errors_total[55s])")); assert!(error.contains("rate(requests_total[55s])")); + assert_eq!(error.matches("(leaf '").count(), 4); } #[test] From 6c238080a84a6595011fc29de4e46ebf5f78f861 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 18:17:30 -0400 Subject: [PATCH 16/18] feat(planner): accept explicit window sizes --- asap-planner-rs/src/config/input.rs | 25 ++++++++---- asap-planner-rs/src/planner/window.rs | 23 ++++++----- asap-planner-rs/tests/integration.rs | 39 ++++++++++--------- asap-planner-rs/tests/sql_integration.rs | 16 +++----- .../test_data/windowing/promql_sliding.yaml | 3 +- .../test_data/windowing/sql_sliding.yaml | 3 +- .../windowing/sql_sliding_invalid.yaml | 3 +- .../CONFIG_PARAMETERS_REFERENCE.md | 16 +++++--- asap-tools/experiments/config/config.yaml | 7 ++-- 9 files changed, 77 insertions(+), 58 deletions(-) diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs index 20014856..66626c18 100644 --- a/asap-planner-rs/src/config/input.rs +++ b/asap-planner-rs/src/config/input.rs @@ -99,19 +99,30 @@ pub struct AggregateCleanupConfig { pub struct WindowingConfig { #[serde(rename = "type")] pub window_type: WindowingType, - pub slide_divisor: Option, + pub window_size_ms: u64, + pub slide_interval_ms: Option, } impl WindowingConfig { pub fn validate(&self) -> Result<(), String> { + if self.window_size_ms == 0 { + return Err("windowing.window_size_ms must be greater than 0".to_string()); + } match self.window_type { - WindowingType::Tumbling if self.slide_divisor.is_some() => { - Err("windowing.slide_divisor is only valid for sliding windows".to_string()) + WindowingType::Tumbling if self.slide_interval_ms.is_some() => { + Err("windowing.slide_interval_ms is only valid for sliding windows".to_string()) } - WindowingType::Sliding => match self.slide_divisor { - None => Err("windowing.slide_divisor is required for sliding windows".to_string()), - Some(divisor) if divisor < 2 => Err(format!( - "windowing.slide_divisor must be at least 2, got {divisor}" + WindowingType::Sliding => match self.slide_interval_ms { + None => Err("windowing.slide_interval_ms is required for sliding windows".to_string()), + Some(0) => { + Err("windowing.slide_interval_ms must be greater than 0".to_string()) + } + Some(slide) if slide > self.window_size_ms => Err( + "windowing.slide_interval_ms must be <= windowing.window_size_ms".to_string(), + ), + Some(slide) if !self.window_size_ms.is_multiple_of(slide) => Err(format!( + "windowing.window_size_ms ({}) must be evenly divisible by windowing.slide_interval_ms ({slide})", + self.window_size_ms )), Some(_) => Ok(()), }, diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs index 0515e91f..5f45b888 100644 --- a/asap-planner-rs/src/planner/window.rs +++ b/asap-planner-rs/src/planner/window.rs @@ -106,21 +106,23 @@ pub fn apply_windowing_override( match windowing.window_type { WindowingType::Tumbling => { config.window_type = WindowType::Tumbling; + config.window_size_ms = windowing.window_size_ms; config.slide_interval_ms = config.window_size_ms; } WindowingType::Sliding => { - let divisor = match windowing.slide_divisor { - Some(divisor) => divisor, + let slide_interval_ms = match windowing.slide_interval_ms { + Some(slide_interval_ms) => slide_interval_ms, None => { return Err(WindowingError::InvalidConfig( - "windowing.slide_divisor is required for sliding windows".to_string(), + "windowing.slide_interval_ms is required for sliding windows".to_string(), )); } }; - if !config.window_size_ms.is_multiple_of(divisor) { + config.window_size_ms = windowing.window_size_ms; + if !config.window_size_ms.is_multiple_of(slide_interval_ms) { return Err(WindowingError::WindowSizeNotDivisible { window_size_ms: config.window_size_ms, - divisor, + slide_interval_ms, }); } if !data_range_ms.is_multiple_of(config.window_size_ms) { @@ -130,7 +132,7 @@ pub fn apply_windowing_override( }); } config.window_type = WindowType::Sliding; - config.slide_interval_ms = config.window_size_ms / divisor; + config.slide_interval_ms = slide_interval_ms; } } Ok(()) @@ -141,7 +143,7 @@ pub enum WindowingError { InvalidConfig(String), WindowSizeNotDivisible { window_size_ms: u64, - divisor: u64, + slide_interval_ms: u64, }, DataRangeNotDivisible { data_range_ms: u64, @@ -155,10 +157,10 @@ impl fmt::Display for WindowingError { Self::InvalidConfig(message) => f.write_str(message), Self::WindowSizeNotDivisible { window_size_ms, - divisor, + slide_interval_ms, } => write!( f, - "window_size_ms ({window_size_ms}) must be evenly divisible by slide_divisor ({divisor})" + "windowing.window_size_ms ({window_size_ms}) must be evenly divisible by windowing.slide_interval_ms ({slide_interval_ms})" ), Self::DataRangeNotDivisible { data_range_ms, @@ -270,7 +272,8 @@ mod tests { }; let windowing = WindowingConfig { window_type: WindowingType::Sliding, - slide_divisor: Some(4), + window_size_ms: 60_000, + slide_interval_ms: Some(15_000), }; let error = apply_windowing_override(&mut config, 90_000, Some(&windowing)).unwrap_err(); diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index 6b9cd7e6..a8da7093 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -42,7 +42,7 @@ fn config_file_sliding_window_override_generates_per_query_candidates() { let aggregations = streaming["aggregations"].as_sequence().unwrap(); assert_eq!(output.inference_query_count(), 2); - assert_eq!(aggregations.len(), 2); + assert_eq!(aggregations.len(), 1); let mut candidates: Vec<(u64, u64, String)> = aggregations .iter() @@ -56,24 +56,19 @@ fn config_file_sliding_window_override_generates_per_query_candidates() { .collect(); candidates.sort(); - assert_eq!( - candidates, - vec![ - (60_000, 15_000, "sliding".to_string()), - (120_000, 30_000, "sliding".to_string()), - ] - ); + assert_eq!(candidates, vec![(60_000, 15_000, "sliding".to_string())]); assert!(candidates .iter() .all(|(_, _, window_type)| window_type != "tumbling")); } #[test] -fn sliding_window_config_requires_a_slide_divisor() { +fn sliding_window_config_requires_a_slide_interval() { let result = Controller::from_yaml_with_schema( r#" windowing: type: "sliding" + window_size_ms: 60000 query_groups: [] "#, http_requests_schema(), @@ -86,7 +81,7 @@ query_groups: [] assert!(error .to_string() - .contains("windowing.slide_divisor is required for sliding windows")); + .contains("windowing.slide_interval_ms is required for sliding windows")); } #[test] @@ -95,7 +90,8 @@ fn sliding_window_validation_reports_all_invalid_queries() { r#" windowing: type: "sliding" - slide_divisor: 3 + window_size_ms: 60000 + slide_interval_ms: 15000 query_groups: - id: 1 queries: @@ -134,7 +130,8 @@ fn sliding_window_validation_rejects_data_range_not_multiple_of_window() { r#" windowing: type: "sliding" - slide_divisor: 4 + window_size_ms: 60000 + slide_interval_ms: 15000 query_groups: - id: 1 queries: @@ -159,12 +156,13 @@ query_groups: } #[test] -fn sliding_window_config_rejects_redundant_divisors() { +fn sliding_window_config_rejects_zero_slide_interval() { let result = Controller::from_yaml_with_schema( r#" windowing: type: "sliding" - slide_divisor: 1 + window_size_ms: 60000 + slide_interval_ms: 0 query_groups: [] "#, http_requests_schema(), @@ -177,16 +175,17 @@ query_groups: [] assert!(error .to_string() - .contains("windowing.slide_divisor must be at least 2, got 1")); + .contains("windowing.slide_interval_ms must be greater than 0")); } #[test] -fn tumbling_window_config_rejects_a_slide_divisor() { +fn tumbling_window_config_rejects_a_slide_interval() { let result = Controller::from_yaml_with_schema( r#" windowing: type: "tumbling" - slide_divisor: 4 + window_size_ms: 60000 + slide_interval_ms: 15000 query_groups: [] "#, http_requests_schema(), @@ -199,7 +198,7 @@ query_groups: [] assert!(error .to_string() - .contains("windowing.slide_divisor is only valid for sliding windows")); + .contains("windowing.slide_interval_ms is only valid for sliding windows")); } #[test] @@ -208,6 +207,7 @@ fn explicit_tumbling_window_override_keeps_tumbling_candidates() { r#" windowing: type: "tumbling" + window_size_ms: 60000 query_groups: - id: 1 queries: @@ -787,7 +787,8 @@ fn binary_arithmetic_aggregates_windowing_errors_from_all_leaves() { r#" windowing: type: "sliding" - slide_divisor: 3 + window_size_ms: 60000 + slide_interval_ms: 15000 query_groups: - id: 1 queries: diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index 8ff0b4ee..f672414d 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -27,7 +27,7 @@ fn config_file_sliding_window_override_generates_per_query_candidates() { let aggregations = streaming["aggregations"].as_sequence().unwrap(); assert_eq!(output.inference_query_count(), 2); - assert_eq!(aggregations.len(), 2); + assert_eq!(aggregations.len(), 1); let mut candidates: Vec<(u64, u64, String)> = aggregations .iter() @@ -41,13 +41,7 @@ fn config_file_sliding_window_override_generates_per_query_candidates() { .collect(); candidates.sort(); - assert_eq!( - candidates, - vec![ - (60_000, 15_000, "sliding".to_string()), - (120_000, 30_000, "sliding".to_string()), - ] - ); + assert_eq!(candidates, vec![(60_000, 15_000, "sliding".to_string())]); assert!(candidates .iter() .all(|(_, _, window_type)| window_type != "tumbling")); @@ -79,7 +73,8 @@ fn sliding_window_validation_rejects_sql_data_range_not_multiple_of_window() { r#" windowing: type: sliding - slide_divisor: 4 + window_size_ms: 60000 + slide_interval_ms: 15000 tables: - name: metrics_table time_column: time @@ -117,6 +112,7 @@ fn discovery_validates_windowing_before_contacting_clickhouse() { br#" windowing: type: sliding + window_size_ms: 60000 tables: - name: metrics_table time_column: time @@ -137,7 +133,7 @@ query_groups: [] Err(error) => error.to_string(), }; - assert!(error.contains("windowing.slide_divisor is required for sliding windows")); + assert!(error.contains("windowing.slide_interval_ms is required for sliding windows")); } /// Single-query config with a 3-column metadata schema. diff --git a/asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml b/asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml index d4be1819..ad1093cf 100644 --- a/asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml +++ b/asap-planner-rs/tests/test_data/windowing/promql_sliding.yaml @@ -1,6 +1,7 @@ windowing: type: "sliding" - slide_divisor: 4 + window_size_ms: 60000 + slide_interval_ms: 15000 query_groups: - id: 1 queries: diff --git a/asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml b/asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml index 54c8489a..acf8a142 100644 --- a/asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml +++ b/asap-planner-rs/tests/test_data/windowing/sql_sliding.yaml @@ -1,6 +1,7 @@ windowing: type: "sliding" - slide_divisor: 4 + window_size_ms: 60000 + slide_interval_ms: 15000 tables: - name: metrics_table time_column: time diff --git a/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml b/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml index a207d6ce..91af6af8 100644 --- a/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml +++ b/asap-planner-rs/tests/test_data/windowing/sql_sliding_invalid.yaml @@ -1,6 +1,7 @@ windowing: type: "sliding" - slide_divisor: 7 + window_size_ms: 60000 + slide_interval_ms: 15000 tables: - name: metrics_table time_column: time diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 25b7a3e6..469321b6 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -216,12 +216,16 @@ These parameters must be provided for all experiment scripts: - **Description**: Selects the planner window type - **Choices**: `tumbling`, `sliding` -#### `windowing.slide_divisor` (int, required for sliding) -- **Description**: Computes each query's slide interval as - `window_size_ms / slide_divisor` -- **Validation**: Must be at least 2 and must evenly divide every supported - query's computed window size; invalid workloads fail plan generation -- **Constraint**: Do not provide this field for `tumbling` +#### `windowing.window_size_ms` (int, required) +- **Description**: Explicit precompute window size in milliseconds +- **Validation**: Must be greater than zero; every supported query's lookback + must be an exact multiple + +#### `windowing.slide_interval_ms` (int, required for sliding) +- **Description**: Explicit sliding interval in milliseconds +- **Validation**: Must be positive, no greater than `window_size_ms`, and evenly + divide it +- **Constraint**: Omit this field for `tumbling` --- diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index 928bc783..6288ef73 100644 --- a/asap-tools/experiments/config/config.yaml +++ b/asap-tools/experiments/config/config.yaml @@ -100,11 +100,12 @@ controller: punting: true # Enable query punting based on performance heuristics (should_be_performant check) # Optional manual planner windowing override. Omit this block to preserve the -# default tumbling-window behavior. For sliding windows, the divisor must be -# at least 2 and must evenly divide every query's computed window size. +# default tumbling-window behavior. For tumbling windows, omit +# slide_interval_ms. For sliding windows, both sizes are explicit. # windowing: # type: "sliding" -# slide_divisor: 4 +# window_size_ms: 60000 +# slide_interval_ms: 15000 # Aggregate cleanup configuration # Policy options: From d268f3fa0361f542eab2eef2bf45945db67ca0d7 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 19:58:09 -0400 Subject: [PATCH 17/18] fix(planner): validate explicit window grids --- asap-planner-rs/src/planner/promql.rs | 1 + asap-planner-rs/src/planner/sql.rs | 1 + asap-planner-rs/src/planner/window.rs | 38 ++++++++-- asap-planner-rs/src/promql/generator.rs | 6 +- asap-planner-rs/tests/integration.rs | 92 +++++++++++++++++++++++- asap-planner-rs/tests/sql_integration.rs | 36 ++++++++++ 6 files changed, 164 insertions(+), 10 deletions(-) diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs index 6136537a..15248735 100644 --- a/asap-planner-rs/src/planner/promql.rs +++ b/asap-planner-rs/src/planner/promql.rs @@ -261,6 +261,7 @@ impl SingleQueryProcessor { crate::planner::window::apply_windowing_override( &mut window_cfg, requirements.data_range_ms, + self.step_ms, self.windowing.as_ref(), )?; diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs index 4b054f6a..64e1efa1 100644 --- a/asap-planner-rs/src/planner/sql.rs +++ b/asap-planner-rs/src/planner/sql.rs @@ -113,6 +113,7 @@ impl SQLSingleQueryProcessor { crate::planner::window::apply_windowing_override( &mut window_cfg, data_range_ms, + 0, self.windowing.as_ref(), )?; diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs index 5f45b888..6d28e974 100644 --- a/asap-planner-rs/src/planner/window.rs +++ b/asap-planner-rs/src/planner/window.rs @@ -94,6 +94,7 @@ pub fn set_window_parameters( pub fn apply_windowing_override( config: &mut IntermediateWindowConfig, data_range_ms: u64, + step_ms: u64, windowing: Option<&WindowingConfig>, ) -> Result<(), WindowingError> { let Some(windowing) = windowing else { @@ -103,6 +104,13 @@ pub fn apply_windowing_override( .validate() .map_err(WindowingError::InvalidConfig)?; + if !data_range_ms.is_multiple_of(windowing.window_size_ms) { + return Err(WindowingError::DataRangeNotDivisible { + data_range_ms, + window_size_ms: windowing.window_size_ms, + }); + } + match windowing.window_type { WindowingType::Tumbling => { config.window_type = WindowType::Tumbling; @@ -125,16 +133,21 @@ pub fn apply_windowing_override( slide_interval_ms, }); } - if !data_range_ms.is_multiple_of(config.window_size_ms) { - return Err(WindowingError::DataRangeNotDivisible { - data_range_ms, - window_size_ms: config.window_size_ms, - }); - } config.window_type = WindowType::Sliding; config.slide_interval_ms = slide_interval_ms; } } + + let grid_interval_ms = match config.window_type { + WindowType::Tumbling => config.window_size_ms, + WindowType::Sliding => config.slide_interval_ms, + }; + if step_ms > 0 && !step_ms.is_multiple_of(grid_interval_ms) { + return Err(WindowingError::StepNotDivisible { + step_ms, + grid_interval_ms, + }); + } Ok(()) } @@ -149,6 +162,10 @@ pub enum WindowingError { data_range_ms: u64, window_size_ms: u64, }, + StepNotDivisible { + step_ms: u64, + grid_interval_ms: u64, + }, } impl fmt::Display for WindowingError { @@ -169,6 +186,13 @@ impl fmt::Display for WindowingError { f, "data_range_ms ({data_range_ms}) must be evenly divisible by window_size_ms ({window_size_ms})" ), + Self::StepNotDivisible { + step_ms, + grid_interval_ms, + } => write!( + f, + "step_ms ({step_ms}) must be evenly divisible by final window grid interval ({grid_interval_ms})" + ), } } } @@ -276,7 +300,7 @@ mod tests { slide_interval_ms: Some(15_000), }; - let error = apply_windowing_override(&mut config, 90_000, Some(&windowing)).unwrap_err(); + let error = apply_windowing_override(&mut config, 90_000, 0, Some(&windowing)).unwrap_err(); assert_eq!( error, diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index a14533e5..7076d5c3 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -174,6 +174,7 @@ fn collect_binary_leaf_entries( let mut all_entries: LeafEntries = Vec::new(); let mut found_windowing_error = false; + let mut found_unsupported_arm = false; for arm in [arms.0, arms.1] { match arm { @@ -223,7 +224,8 @@ fn collect_binary_leaf_entries( } // Arm is neither a supported leaf nor a binary expression. // This entire query cannot be accelerated. - return Ok(None); + found_unsupported_arm = true; + continue; } } } @@ -231,7 +233,7 @@ fn collect_binary_leaf_entries( } } - if found_windowing_error { + if found_windowing_error || found_unsupported_arm { Ok(None) } else { Ok(Some(all_entries)) diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index a8da7093..cd22af7f 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -211,7 +211,7 @@ windowing: query_groups: - id: 1 queries: - - "rate(http_requests_total[1m])" + - "rate(http_requests_total[2m])" repetition_delay_ms: 60000 controller_options: accuracy_sla: 0.99 @@ -228,6 +228,7 @@ query_groups: let aggregation = &streaming["aggregations"][0]; assert_eq!(aggregation["windowType"].as_str(), Some("tumbling")); + assert_eq!(aggregation["windowSizeMs"].as_u64(), Some(60_000)); assert_eq!( aggregation["slideIntervalMs"].as_u64(), aggregation["windowSizeMs"].as_u64() @@ -235,6 +236,64 @@ query_groups: assert_ne!(aggregation["windowType"].as_str(), Some("sliding")); } +#[test] +fn explicit_sliding_window_revalidates_final_step_grid() { + let controller = Controller::from_yaml_with_schema( + r#" +windowing: + type: "sliding" + window_size_ms: 60000 + slide_interval_ms: 30000 +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[2m])" + repetition_delay_ms: 40000 + step_ms: 80000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +"#, + http_requests_schema(), + arroyo_opts(), + ) + .unwrap(); + + let error = match controller.generate() { + Ok(_) => panic!("step misaligned with the explicit sliding grid should fail"), + Err(error) => error.to_string(), + }; + assert!(error.contains("final window grid interval (30000)")); +} + +#[test] +fn explicit_tumbling_window_rejects_non_multiple_lookback() { + let controller = Controller::from_yaml_with_schema( + r#" +windowing: + type: "tumbling" + window_size_ms: 60000 +query_groups: + - id: 1 + queries: + - "rate(http_requests_total[90s])" + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +"#, + http_requests_schema(), + arroyo_opts(), + ) + .unwrap(); + + let error = match controller.generate() { + Ok(_) => panic!("tumbling lookback must be an exact window multiple"), + Err(error) => error.to_string(), + }; + assert!(error.contains("data_range_ms (90000)")); +} + /// Schema for binary arithmetic tests: errors_total and requests_total. fn binary_arithmetic_schema() -> PromQLSchema { PromQLSchema::new() @@ -822,6 +881,37 @@ query_groups: assert_eq!(error.matches("(leaf '").count(), 4); } +#[test] +fn binary_arithmetic_scans_past_unsupported_arms_for_windowing_errors() { + let controller = Controller::from_yaml_with_schema( + r#" +windowing: + type: "sliding" + window_size_ms: 60000 + slide_interval_ms: 15000 +query_groups: + - id: 1 + queries: + - "abs(errors_total) / rate(requests_total[65s])" + - "rate(requests_total[65s]) / abs(errors_total)" + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.99 + latency_sla: 1.0 +"#, + binary_arithmetic_schema(), + arroyo_opts(), + ) + .unwrap(); + + let error = match controller.generate() { + Ok(_) => panic!("invalid binary sliding windows should abort plan generation"), + Err(error) => error.to_string(), + }; + assert!(error.contains("abs(errors_total) / rate(requests_total[65s])")); + assert!(error.contains("rate(requests_total[65s]) / abs(errors_total)")); +} + #[test] fn temporal_overlapping_rate_increase_deduped() { // rate and increase produce identical MultipleIncrease configs → 1 streaming entry shared, diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index f672414d..ae7e109f 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -105,6 +105,42 @@ aggregate_cleanup: assert!(error.contains("window_size_ms (60000)")); } +#[test] +fn explicit_tumbling_window_rejects_sql_non_multiple_lookback() { + let query = "SELECT MIN(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -90, NOW()) AND NOW() GROUP BY hostname"; + let config = format!( + r#" +windowing: + type: tumbling + window_size_ms: 60000 +tables: + - name: metrics_table + time_column: time + value_columns: [cpu_usage] + metadata_columns: [hostname] +query_groups: + - id: 1 + repetition_delay_ms: 60000 + controller_options: + accuracy_sla: 0.95 + latency_sla: 100.0 + queries: + - "{query}" +aggregate_cleanup: + policy: read_based +"# + ); + + let error = match SQLController::from_yaml(&config, sql_opts()) + .unwrap() + .generate() + { + Ok(_) => panic!("tumbling lookback must be an exact window multiple"), + Err(error) => error.to_string(), + }; + assert!(error.contains("data_range_ms (90000)")); +} + #[test] fn discovery_validates_windowing_before_contacting_clickhouse() { let mut file = NamedTempFile::new().unwrap(); From 7b59f0635ee3b98948b73d98a0e8a0eb719aa811 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 27 Aug 2026 20:15:11 -0400 Subject: [PATCH 18/18] fix(planner): complete windowing review fixes --- asap-planner-rs/src/planner/promql.rs | 1 + asap-planner-rs/src/planner/window.rs | 17 +++++++------ asap-planner-rs/src/promql/generator.rs | 33 ++++++++++++++++--------- 3 files changed, 31 insertions(+), 20 deletions(-) diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs index 15248735..a4b2a7e6 100644 --- a/asap-planner-rs/src/planner/promql.rs +++ b/asap-planner-rs/src/planner/promql.rs @@ -256,6 +256,7 @@ impl SingleQueryProcessor { self.data_ingestion_interval_ms, self.step_ms, &mut window_cfg, + self.windowing.is_none(), ) .map_err(ControllerError::PlannerError)?; crate::planner::window::apply_windowing_override( diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs index 6d28e974..5c4c6ce6 100644 --- a/asap-planner-rs/src/planner/window.rs +++ b/asap-planner-rs/src/planner/window.rs @@ -47,6 +47,7 @@ pub fn set_window_parameters( data_ingestion_interval_ms: u64, step_ms: u64, config: &mut IntermediateWindowConfig, + validate_step_alignment: bool, ) -> Result<(), String> { if t_repeat_ms < data_ingestion_interval_ms { return Err(format!( @@ -79,7 +80,7 @@ pub fn set_window_parameters( // Catch a window size incompatible with its own planning-time step_ms // here, at planning time, instead of provisioning a window that would // reject every range query using exactly the step_ms it was planned for. - if step_ms > 0 && !step_ms.is_multiple_of(window_size_ms) { + if validate_step_alignment && step_ms > 0 && !step_ms.is_multiple_of(window_size_ms) { return Err(format!( "step_ms ({step_ms}ms) must be a multiple of the computed window size ({window_size_ms}ms)" )); @@ -229,7 +230,7 @@ mod tests { #[test] fn set_window_parameters_temporal_shape() { let mut config = IntermediateWindowConfig::default(); - set_window_parameters(300_000, 60_000, 15_000, 0, &mut config).unwrap(); + set_window_parameters(300_000, 60_000, 15_000, 0, &mut config, true).unwrap(); assert_eq!(config.window_size_ms, 60_000); assert_eq!(config.slide_interval_ms, 60_000); assert_eq!(config.window_type, WindowType::Tumbling); @@ -240,7 +241,7 @@ mod tests { // data_range_ms == data_ingestion_interval_ms (spatial-only query), // t_repeat_ms also equal to the interval: unaffected by the relaxation. let mut config = IntermediateWindowConfig::default(); - set_window_parameters(15_000, 15_000, 15_000, 0, &mut config).unwrap(); + set_window_parameters(15_000, 15_000, 15_000, 0, &mut config, true).unwrap(); assert_eq!(config.window_size_ms, 15_000); } @@ -252,26 +253,26 @@ mod tests { // any cadence gives the latest available answer. window_size stays // exactly one interval regardless of t_repeat_ms. let mut config = IntermediateWindowConfig::default(); - set_window_parameters(15_000, 60_000, 15_000, 0, &mut config).unwrap(); + set_window_parameters(15_000, 60_000, 15_000, 0, &mut config, true).unwrap(); assert_eq!(config.window_size_ms, 15_000); } #[test] fn set_window_parameters_rejects_t_repeat_below_interval() { let mut config = IntermediateWindowConfig::default(); - assert!(set_window_parameters(300_000, 10_000, 15_000, 0, &mut config).is_err()); + assert!(set_window_parameters(300_000, 10_000, 15_000, 0, &mut config, true).is_err()); } #[test] fn set_window_parameters_rejects_data_range_below_t_repeat() { let mut config = IntermediateWindowConfig::default(); - assert!(set_window_parameters(30_000, 60_000, 15_000, 0, &mut config).is_err()); + assert!(set_window_parameters(30_000, 60_000, 15_000, 0, &mut config, true).is_err()); } #[test] fn set_window_parameters_rejects_step_below_interval() { let mut config = IntermediateWindowConfig::default(); - assert!(set_window_parameters(300_000, 60_000, 15_000, 10_000, &mut config).is_err()); + assert!(set_window_parameters(300_000, 60_000, 15_000, 10_000, &mut config, true).is_err()); } #[test] @@ -282,7 +283,7 @@ mod tests { // otherwise provision a window that query-engine's // validate_range_query_params rejects for exactly this step_ms. let mut config = IntermediateWindowConfig::default(); - let result = set_window_parameters(300_000, 40_000, 10_000, 100_000, &mut config); + let result = set_window_parameters(300_000, 40_000, 10_000, 100_000, &mut config, true); assert!(result.is_err()); assert!(result.unwrap_err().contains("must be a multiple of")); } diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index 7076d5c3..297ce683 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -111,16 +111,21 @@ pub fn generate_plan( } Err(e) => return Err(e), } - } else if let Some(arm_entries) = collect_binary_leaf_entries( - &processor, - &mut dedup_map, - &mut windowing_errors, - query_string, - )? { - // Binary arithmetic: register each leaf arm in dedup_map and query_keys_map - for (arm_query, keys_for_arm) in arm_entries { - // Use `entry` so a standalone query that duplicates an arm wins - query_keys_map.entry(arm_query).or_insert(keys_for_arm); + } else { + let mut pending_dedup_map = IndexMap::new(); + if let Some(arm_entries) = collect_binary_leaf_entries( + &processor, + &dedup_map, + &mut pending_dedup_map, + &mut windowing_errors, + query_string, + )? { + dedup_map.extend(pending_dedup_map); + // Binary arithmetic: register each leaf arm in dedup_map and query_keys_map + for (arm_query, keys_for_arm) in arm_entries { + // Use `entry` so a standalone query that duplicates an arm wins + query_keys_map.entry(arm_query).or_insert(keys_for_arm); + } } } } @@ -163,7 +168,8 @@ pub fn generate_plan( /// Returns `Err` only on internal planner errors. fn collect_binary_leaf_entries( processor: &SingleQueryProcessor, - dedup_map: &mut IndexMap, + dedup_map: &IndexMap, + pending_dedup_map: &mut IndexMap, windowing_errors: &mut Vec, query_context: &str, ) -> Result, ControllerError> { @@ -202,7 +208,9 @@ fn collect_binary_leaf_entries( for config in configs { let key = config.identifying_key(); keys_for_arm.push((key.clone(), cleanup_param)); - dedup_map.entry(key).or_insert(config); + if !dedup_map.contains_key(&key) { + pending_dedup_map.entry(key).or_insert(config); + } } all_entries.push((arm_query, keys_for_arm)); } else { @@ -211,6 +219,7 @@ fn collect_binary_leaf_entries( match collect_binary_leaf_entries( &arm_processor, dedup_map, + pending_dedup_map, windowing_errors, query_context, )? {