fix(planner): change more variables to ms precision

asap-planner-rs/src/config/input.rs

@@ -45,6 +45,7 @@ impl ControllerConfig {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct QueryGroup {
     pub id: Option<u32>,
     pub queries: Vec<String>,
@@ -60,23 +61,27 @@ pub struct QueryGroup {
 }
 
 #[derive(Debug, Clone, Deserialize, Default)]
+#[serde(deny_unknown_fields)]
 pub struct ControllerOptions {
     pub accuracy_sla: f64,
     pub latency_sla: f64,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct MetricDefinition {
     pub metric: String,
     pub labels: Vec<String>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct AggregateCleanupConfig {
     pub policy: Option<CleanupPolicy>,
 }
 
 #[derive(Debug, Clone, Deserialize, Default)]
+#[serde(deny_unknown_fields)]
 pub struct SketchParameterOverrides {
     #[serde(rename = "CountMinSketch")]
     pub count_min_sketch: Option<CmsParams>,
@@ -91,37 +96,43 @@ pub struct SketchParameterOverrides {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct CmsParams {
     pub depth: u64,
     pub width: u64,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct CmsHeapParams {
     pub depth: u64,
     pub width: u64,
     pub heap_multiplier: Option<u64>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct KllParams {
     #[serde(rename = "K")]
     pub k: u64,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct HydraParams {
     pub row_num: u64,
     pub col_num: u64,
     pub k: u64,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct HllParams {
     pub precision: u64,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct SQLControllerConfig {
     pub query_groups: Vec<SQLQueryGroup>,
     pub tables: Vec<TableDefinition>,
@@ -130,14 +141,16 @@ pub struct SQLControllerConfig {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct SQLQueryGroup {
     pub id: Option<u32>,
     pub queries: Vec<String>,
-    pub repetition_delay: u64,
+    pub repetition_delay_ms: u64,
     pub controller_options: ControllerOptions,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct TableDefinition {
     pub name: String,
     pub time_column: String,
@@ -147,17 +160,19 @@ pub struct TableDefinition {
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct ElasticDSLControllerConfig {
     pub query_groups: Vec<ElasticDSLQueryGroup>,
     pub sketch_parameters: Option<SketchParameterOverrides>,
     pub aggregate_cleanup: Option<AggregateCleanupConfig>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct ElasticDSLQueryGroup {
     pub id: Option<u32>,
     pub queries: Vec<String>,
-    pub repetition_delay: u64,
+    pub repetition_delay_ms: u64,
     pub index: String,
     pub time_field: String,
     pub controller_options: ControllerOptions,

asap-planner-rs/src/elastic_dsl/generator.rs

@@ -47,7 +47,7 @@ impl ElasticIndexSchemaBuilder {
 
 pub struct ElasticRuntimeOptions {
     pub streaming_engine: StreamingEngine,
-    pub data_ingestion_interval: u64,
+    pub data_ingestion_interval_ms: u64,
 }
 
 pub fn generate_elastic_plan(
@@ -60,12 +60,12 @@ pub fn generate_elastic_plan(
         .and_then(|c| c.policy)
         .unwrap_or(CleanupPolicy::ReadBased);
 
-    // Validate T % data_ingestion_interval == 0
+    // Validate T % data_ingestion_interval_ms == 0
     for qg in &config.query_groups {
-        if qg.repetition_delay % opts.data_ingestion_interval != 0 {
+        if qg.repetition_delay_ms % opts.data_ingestion_interval_ms != 0 {
             return Err(ControllerError::PlannerError(format!(
-                "repetition_delay {} is not a multiple of data_ingestion_interval {}",
-                qg.repetition_delay, opts.data_ingestion_interval
+                "repetition_delay_ms {} is not a multiple of data_ingestion_interval_ms {}",
+                qg.repetition_delay_ms, opts.data_ingestion_interval_ms
             )));
         }
     }
@@ -111,8 +111,8 @@ pub fn generate_elastic_plan(
         for query_string in &qg.queries {
             let processor = ElasticSingleQueryProcessor::new(
                 query_string.clone(),
-                qg.repetition_delay,
-                opts.data_ingestion_interval,
+                qg.repetition_delay_ms,
+                opts.data_ingestion_interval_ms,
                 index_schema_builders[&qg.index].clone(),
                 opts.streaming_engine,
                 config.sketch_parameters.clone(),

asap-planner-rs/src/main.rs

@@ -45,8 +45,8 @@ struct Args {
     #[arg(long = "query-language", value_enum, default_value = "promql")]
     query_language: QueryLanguage,
 
-    #[arg(long = "data-ingestion-interval", required = false)]
-    data_ingestion_interval: Option<u64>,
+    #[arg(long = "data-ingestion-interval-ms", required = false)]
+    data_ingestion_interval_ms: Option<u64>,
 
     /// ClickHouse base URL for auto-inferring metadata_columns when not listed
     /// in the config file. Example: http://localhost:8123
@@ -123,16 +123,16 @@ fn main() -> anyhow::Result<()> {
             controller.generate_to_dir(&args.output_dir)?;
         }
         QueryLanguage::sql | QueryLanguage::elastic_sql => {
-            let interval = args.data_ingestion_interval.ok_or_else(|| {
-                anyhow::anyhow!("--data-ingestion-interval is required for SQL mode")
+            let interval = args.data_ingestion_interval_ms.ok_or_else(|| {
+                anyhow::anyhow!("--data-ingestion-interval-ms is required for SQL mode")
             })?;
             let config_path = args
                 .input_config
                 .ok_or_else(|| anyhow::anyhow!("--input_config is required for SQL mode"))?;
             let opts = SQLRuntimeOptions {
                 streaming_engine: engine,
                 query_evaluation_time: None,
-                data_ingestion_interval: interval,
+                data_ingestion_interval_ms: interval,
             };
             let controller = match args.clickhouse_url {
                 Some(ref url) => SQLController::from_file_with_discovery(
@@ -146,15 +146,17 @@ fn main() -> anyhow::Result<()> {
             controller.generate_to_dir(&args.output_dir)?;
         }
         QueryLanguage::elastic_querydsl => {
-            let interval = args.data_ingestion_interval.ok_or_else(|| {
-                anyhow::anyhow!("--data-ingestion-interval is required for Elasticsearch DSL mode")
+            let interval = args.data_ingestion_interval_ms.ok_or_else(|| {
+                anyhow::anyhow!(
+                    "--data-ingestion-interval-ms is required for Elasticsearch DSL mode"
+                )
             })?;
             let config_path = args.input_config.ok_or_else(|| {
                 anyhow::anyhow!("--input_config is required for Elasticsearch DSL mode")
             })?;
             let opts = ElasticRuntimeOptions {
                 streaming_engine: engine,
-                data_ingestion_interval: interval,
+                data_ingestion_interval_ms: interval,
             };
             ElasticController::from_file(&config_path, opts)?.generate_to_dir(&args.output_dir)?;
         }

asap-planner-rs/src/planner/cleanup.rs

@@ -75,19 +75,19 @@ pub fn get_cleanup_param(
 /// SQL cleanup param — SQL queries are always instant (no range_duration/step).
 pub fn get_sql_cleanup_param(
     cleanup_policy: CleanupPolicy,
-    t_lookback: u64,
-    t_repeat: u64,
+    t_lookback_ms: u64,
+    t_repeat_ms: u64,
 ) -> Result<u64, String> {
     match cleanup_policy {
         CleanupPolicy::CircularBuffer | CleanupPolicy::ReadBased => {
-            if t_repeat == 0 {
+            if t_repeat_ms == 0 {
                 return Err(
-                    "repetition_delay must be > 0 for cleanup param calculation; \
-                     set a non-zero repetition_delay in your query group config"
+                    "repetition_delay_ms must be > 0 for cleanup param calculation; \
+                     set a non-zero repetition_delay_ms in your query group config"
                         .to_string(),
                 );
             }
-            Ok(t_lookback.div_ceil(t_repeat))
+            Ok(t_lookback_ms.div_ceil(t_repeat_ms))
         }
         CleanupPolicy::NoCleanup => {
             Err("NoCleanup policy should not call get_sql_cleanup_param".to_string())

asap-planner-rs/src/planner/elastic_dsl.rs

@@ -19,9 +19,9 @@ use indexmap::IndexSet;
 
 pub struct ElasticSingleQueryProcessor {
     query_string: String,
-    t_repeat: u64,
+    t_repeat_ms: u64,
     #[allow(dead_code)]
-    data_ingestion_interval: u64,
+    data_ingestion_interval_ms: u64,
     index_schema: ElasticIndexSchemaBuilder,
     #[allow(dead_code)]
     streaming_engine: StreamingEngine,
@@ -33,17 +33,17 @@ impl ElasticSingleQueryProcessor {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         query_string: String,
-        t_repeat: u64,
-        data_ingestion_interval: u64,
+        t_repeat_ms: u64,
+        data_ingestion_interval_ms: u64,
         index_schema: ElasticIndexSchemaBuilder,
         streaming_engine: StreamingEngine,
         sketch_parameters: Option<SketchParameterOverrides>,
         cleanup_policy: CleanupPolicy,
     ) -> Self {
         Self {
             query_string,
-            t_repeat,
-            data_ingestion_interval,
+            t_repeat_ms,
+            data_ingestion_interval_ms,
             index_schema,
             streaming_engine,
             sketch_parameters,
@@ -65,11 +65,7 @@ impl ElasticSingleQueryProcessor {
         // Get aggregation type and statistics
         let (treatment_type, statistics) = get_elastic_statistics(&query_info.aggregation)?;
 
-        // Build window config (always tumbling for Elasticsearch queries).
-        // self.t_repeat is seconds (Elastic DSL is out of scope for the ms-precision
-        // rename, see issue #427) — IntermediateWindowConfig is now ms-typed throughout,
-        // so convert at this boundary, same as planner::sql::compute_sql_window.
-        let t_repeat_ms = self.t_repeat * 1000;
+        let t_repeat_ms = self.t_repeat_ms;
         let window_cfg = IntermediateWindowConfig {
             window_size_ms: t_repeat_ms,
             slide_interval_ms: t_repeat_ms,
@@ -134,7 +130,7 @@ impl ElasticSingleQueryProcessor {
                 _ => false,
             })
             .and_then(|p| range_query_to_time_range(p, 0));
-        let t_lookback = match time_range {
+        let t_lookback_ms = match time_range {
             Some(tr) => tr.duration_ms().unwrap_or(t_repeat_ms),
             None => t_repeat_ms,
         };
@@ -144,7 +140,7 @@ impl ElasticSingleQueryProcessor {
             None
         } else {
             Some(
-                get_sql_cleanup_param(self.cleanup_policy, t_lookback, t_repeat_ms)
+                get_sql_cleanup_param(self.cleanup_policy, t_lookback_ms, t_repeat_ms)
                     .map_err(ControllerError::PlannerError)?,
             )
         };