feat(planner): add manual windowing override

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<QueryGroup>,
+    pub windowing: Option<WindowingConfig>,
     pub sketch_parameters: Option<SketchParameterOverrides>,
     pub aggregate_cleanup: Option<AggregateCleanupConfig>,
     /// Optional hint: per-metric label sets used as a fallback when Prometheus
@@ -93,6 +94,50 @@ pub struct AggregateCleanupConfig {
     pub policy: Option<CleanupPolicy>,
 }
 
+#[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct WindowingConfig {
+    #[serde(rename = "type")]
+    pub window_type: WindowingType,
+    pub window_size_ms: u64,
+    pub slide_interval_ms: Option<u64>,
+}
+
+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_interval_ms.is_some() => {
+                Err("windowing.slide_interval_ms is only valid for sliding windows".to_string())
+            }
+            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(()),
+            },
+            WindowingType::Tumbling => Ok(()),
+        }
+    }
+}
+
+#[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 +187,7 @@ pub struct HllParams {
 pub struct SQLControllerConfig {
     pub query_groups: Vec<SQLQueryGroup>,
     pub tables: Vec<TableDefinition>,
+    pub windowing: Option<WindowingConfig>,
     pub sketch_parameters: Option<SketchParameterOverrides>,
     pub aggregate_cleanup: Option<AggregateCleanupConfig>,
 }

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}")]

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;

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,

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<WindowingConfig>,
 }
 
 impl SingleQueryProcessor {
@@ -91,6 +92,7 @@ impl SingleQueryProcessor {
         range_duration_ms: u64,
         step_ms: u64,
         cleanup_policy: CleanupPolicy,
+        windowing: Option<WindowingConfig>,
     ) -> 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(),
         )
     }
 
@@ -252,8 +256,15 @@ 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(
+            &mut window_cfg,
+            requirements.data_range_ms,
+            self.step_ms,
+            self.windowing.as_ref(),
+        )?;
 
         let subpopulation_labels = requirements.grouping_labels;
         let rollup = all_labels.difference(&subpopulation_labels);

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<SketchParameterOverrides>,
     cleanup_policy: CleanupPolicy,
+    windowing: Option<WindowingConfig>,
 }
 
 impl SQLSingleQueryProcessor {
@@ -39,6 +40,7 @@ impl SQLSingleQueryProcessor {
         streaming_engine: StreamingEngine,
         sketch_parameters: Option<SketchParameterOverrides>,
         cleanup_policy: CleanupPolicy,
+        windowing: Option<WindowingConfig>,
     ) -> Self {
         Self {
             query_string,
@@ -48,6 +50,7 @@ impl SQLSingleQueryProcessor {
             streaming_engine,
             sketch_parameters,
             cleanup_policy,
+            windowing,
         }
     }
 
@@ -100,11 +103,19 @@ 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,
         )?;
+        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,
+            0,
+            self.windowing.as_ref(),
+        )?;
 
         // Get all metadata columns for the table
         let all_metadata = get_all_metadata_columns(&self.table_definitions, table_name)?;