diff --git a/asap-planner-rs/docker-compose.yml.j2 b/asap-planner-rs/docker-compose.yml.j2
index 19d797d7..9978b670 100644
--- a/asap-planner-rs/docker-compose.yml.j2
+++ b/asap-planner-rs/docker-compose.yml.j2
@@ -12,8 +12,8 @@ services:
       "--streaming_engine", "{{ streaming_engine }}",
       "--prometheus-url", "{{ prometheus_url }}",
       "--query-language", "{{ query_language }}"{% if punting %},
-      "--enable-punting"{% endif %}{% if data_ingestion_interval is not none %},
-      "--data-ingestion-interval", "{{ data_ingestion_interval }}"{% endif %}
+      "--enable-punting"{% endif %}{% if data_ingestion_interval_ms is not none %},
+      "--data-ingestion-interval-ms", "{{ data_ingestion_interval_ms }}"{% endif %}
     ]
     network_mode: "host"
     restart: no
diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs
index 6752ca33..413e9f5c 100644
--- a/asap-planner-rs/src/config/input.rs
+++ b/asap-planner-rs/src/config/input.rs
@@ -4,6 +4,7 @@ use asap_types::streaming_config::StreamingConfig;
 use asap_types::PromQLSchema;
 use promql_utilities::data_model::KeyByLabelNames;
 use serde::Deserialize;
+use tracing::warn;
 
 #[derive(Debug, Clone, Deserialize)]
 #[serde(deny_unknown_fields)]
@@ -30,6 +31,22 @@ pub struct ControllerConfig {
 }
 
 impl ControllerConfig {
+    /// Warn if any query group has both SLAs at 0.0 (the serde Default),
+    /// which indicates `controller_options` was omitted from the config.
+    pub fn warn_default_slas(&self) {
+        for qg in &self.query_groups {
+            let opts = &qg.controller_options;
+            if opts.accuracy_sla == 0.0 && opts.latency_sla == 0.0 {
+                warn!(
+                    query_group_id = ?qg.id,
+                    "controller_options not set in query group; \
+                     accuracy_sla=0.0 and latency_sla=0.0 will be used — \
+                     add controller_options to your config"
+                );
+            }
+        }
+    }
+
     /// Build a `PromQLSchema` from the `metrics` hints in this config.
     /// Returns an empty schema if no hints are present.
     pub fn schema_from_hints(&self) -> PromQLSchema {
@@ -133,7 +150,7 @@ pub struct SQLControllerConfig {
 pub struct SQLQueryGroup {
     pub id: Option<u32>,
     pub queries: Vec<String>,
-    pub repetition_delay: u64,
+    pub repetition_delay_ms: u64,
     pub controller_options: ControllerOptions,
 }
 
@@ -157,7 +174,7 @@ pub struct ElasticDSLControllerConfig {
 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,
diff --git a/asap-planner-rs/src/elastic_dsl/generator.rs b/asap-planner-rs/src/elastic_dsl/generator.rs
index 51e0c846..df43b5f1 100644
--- a/asap-planner-rs/src/elastic_dsl/generator.rs
+++ b/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(),
diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs
index 28f796a1..f9c325eb 100644
--- a/asap-planner-rs/src/main.rs
+++ b/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,8 +123,8 @@ 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
@@ -132,7 +132,7 @@ fn main() -> anyhow::Result<()> {
             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)?;
         }
diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs
index cb093af5..b09ed8be 100644
--- a/asap-planner-rs/src/optimizer/pipeline.rs
+++ b/asap-planner-rs/src/optimizer/pipeline.rs
@@ -78,7 +78,7 @@ pub fn run_greedy_pipeline(
 }
 
 /// Convert a `ControllerConfig`'s query groups into a flat list of RQEs.
-/// Each (query, repetition_delay) pair becomes one RQE.
+/// Each (query, repetition_delay_ms) pair becomes one RQE.
 fn config_to_rqes(config: &ControllerConfig) -> Vec<RQE> {
     config
         .query_groups
diff --git a/asap-planner-rs/src/planner/cleanup.rs b/asap-planner-rs/src/planner/cleanup.rs
index ff0182d9..444b977c 100644
--- a/asap-planner-rs/src/planner/cleanup.rs
+++ b/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())
diff --git a/asap-planner-rs/src/planner/elastic_dsl.rs b/asap-planner-rs/src/planner/elastic_dsl.rs
index 6e6b9483..a440874e 100644
--- a/asap-planner-rs/src/planner/elastic_dsl.rs
+++ b/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,8 +33,8 @@ 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>,
@@ -42,8 +42,8 @@ impl ElasticSingleQueryProcessor {
     ) -> 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)?,
             )
         };
diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs
index 168ce39d..950a09da 100644
--- a/asap-planner-rs/src/planner/sql.rs
+++ b/asap-planner-rs/src/planner/sql.rs
@@ -20,8 +20,8 @@ use crate::StreamingEngine;
 
 pub struct SQLSingleQueryProcessor {
     query_string: String,
-    t_repeat: u64,
-    data_ingestion_interval: u64,
+    t_repeat_ms: u64,
+    data_ingestion_interval_ms: u64,
     table_definitions: Vec<TableDefinition>,
     #[allow(dead_code)]
     streaming_engine: StreamingEngine,
@@ -33,8 +33,8 @@ impl SQLSingleQueryProcessor {
     #[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,
         table_definitions: Vec<TableDefinition>,
         streaming_engine: StreamingEngine,
         sketch_parameters: Option<SketchParameterOverrides>,
@@ -42,8 +42,8 @@ impl SQLSingleQueryProcessor {
     ) -> Self {
         Self {
             query_string,
-            t_repeat,
-            data_ingestion_interval,
+            t_repeat_ms,
+            data_ingestion_interval_ms,
             table_definitions,
             streaming_engine,
             sketch_parameters,
@@ -72,8 +72,12 @@ impl SQLSingleQueryProcessor {
             })?;
 
         // Match query to pattern
-        let sql_query = SQLPatternMatcher::new(schema.clone(), self.data_ingestion_interval as f64)
-            .query_info_to_pattern(&qdata);
+        // SQLPatternMatcher.scrape_interval is in seconds (SQL timestamps are seconds-based).
+        let sql_query = SQLPatternMatcher::new(
+            schema.clone(),
+            self.data_ingestion_interval_ms as f64 / 1000.0,
+        )
+        .query_info_to_pattern(&qdata);
 
         if !sql_query.is_valid() {
             return Err(ControllerError::SqlParse(sql_query.msg.unwrap_or_default()));
@@ -97,8 +101,11 @@ impl SQLSingleQueryProcessor {
         let value_column = agg_info.get_value_column_name().to_string();
 
         // Compute window
-        let window_cfg =
-            compute_sql_window(query_type, self.data_ingestion_interval, self.t_repeat);
+        let window_cfg = compute_sql_window(
+            query_type,
+            self.data_ingestion_interval_ms,
+            self.t_repeat_ms,
+        );
 
         // Get all metadata columns for the table
         let all_metadata = get_all_metadata_columns(&self.table_definitions, table_name)?;
@@ -157,16 +164,17 @@ impl SQLSingleQueryProcessor {
             }
         }
 
-        let t_lookback = match query_type {
-            QueryType::Spatial => self.data_ingestion_interval,
-            _ => sql_query.query_data[0].time_info.get_duration() as u64,
+        let t_lookback_ms = match query_type {
+            QueryType::Spatial => self.data_ingestion_interval_ms,
+            // SQLPatternParser always produces second-based durations; convert to ms.
+            _ => (sql_query.query_data[0].time_info.get_duration() * 1000.0).round() as u64,
         };
 
         let cleanup_param = if self.cleanup_policy == CleanupPolicy::NoCleanup {
             None
         } else {
             Some(
-                get_sql_cleanup_param(self.cleanup_policy, t_lookback, self.t_repeat)
+                get_sql_cleanup_param(self.cleanup_policy, t_lookback_ms, self.t_repeat_ms)
                     .map_err(ControllerError::PlannerError)?,
             )
         };
@@ -215,16 +223,12 @@ fn get_sql_statistics(name: &str) -> Result<Vec<Statistic>, ControllerError> {
 
 fn compute_sql_window(
     query_type: &QueryType,
-    data_ingestion_interval: u64,
-    t_repeat: u64,
+    data_ingestion_interval_ms: u64,
+    t_repeat_ms: u64,
 ) -> IntermediateWindowConfig {
-    // data_ingestion_interval and t_repeat are seconds (SQL mode 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 the pre-existing pattern in
-    // asap-query-engine/src/engines/simple_engine/sql.rs.
     let window_size_ms = match query_type {
-        QueryType::Spatial => data_ingestion_interval * 1000,
-        _ => t_repeat * 1000,
+        QueryType::Spatial => data_ingestion_interval_ms,
+        _ => t_repeat_ms,
     };
     IntermediateWindowConfig {
         window_size_ms,
diff --git a/asap-planner-rs/src/promql/controller.rs b/asap-planner-rs/src/promql/controller.rs
index 88c404f8..30dea006 100644
--- a/asap-planner-rs/src/promql/controller.rs
+++ b/asap-planner-rs/src/promql/controller.rs
@@ -37,6 +37,7 @@ impl Controller {
     ) -> Result<Self, ControllerError> {
         let yaml_str = std::fs::read_to_string(path)?;
         let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
+        config.warn_default_slas();
         let all_queries: Vec<String> = config
             .query_groups
             .iter()
@@ -78,6 +79,7 @@ impl Controller {
     ) -> Result<Self, ControllerError> {
         let yaml_str = std::fs::read_to_string(path)?;
         let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
+        config.warn_default_slas();
         Ok(Self {
             config,
             schema,
@@ -92,6 +94,7 @@ impl Controller {
         opts: RuntimeOptions,
     ) -> Result<Self, ControllerError> {
         let config: ControllerConfig = serde_yaml::from_str(yaml)?;
+        config.warn_default_slas();
         Ok(Self {
             config,
             schema,
diff --git a/asap-planner-rs/src/query_log/frequency.rs b/asap-planner-rs/src/query_log/frequency.rs
index 5ec6e3b1..9be65f64 100644
--- a/asap-planner-rs/src/query_log/frequency.rs
+++ b/asap-planner-rs/src/query_log/frequency.rs
@@ -83,7 +83,7 @@ pub fn infer_queries(
         }
 
         let repetition_delay_ms =
-            infer_repetition_delay(&variant_entries, scrape_interval_ms, &query);
+            infer_repetition_delay_ms(&variant_entries, scrape_interval_ms, &query);
 
         if step == 0 {
             instant_results.push(InstantQueryInfo {
@@ -105,7 +105,7 @@ pub fn infer_queries(
 }
 
 /// Compute median inter-arrival time from timestamps and round to nearest scrape interval.
-fn infer_repetition_delay(entries: &[&LogEntry], scrape_interval_ms: u64, query: &str) -> u64 {
+fn infer_repetition_delay_ms(entries: &[&LogEntry], scrape_interval_ms: u64, query: &str) -> u64 {
     let mut timestamps: Vec<DateTime<Utc>> = entries.iter().map(|e| e.ts).collect();
     timestamps.sort();
 
@@ -124,7 +124,7 @@ fn infer_repetition_delay(entries: &[&LogEntry], scrape_interval_ms: u64, query:
             raw_median_ms = raw_median,
             rounded_ms = rounded,
             misalignment_pct = misalignment * 100.0,
-            "inferred repetition_delay is poorly aligned with scrape_interval; result may be inaccurate"
+            "inferred repetition_delay_ms is poorly aligned with scrape_interval; result may be inaccurate"
         );
     }
 
diff --git a/asap-planner-rs/src/sql/generator.rs b/asap-planner-rs/src/sql/generator.rs
index 9982a6b8..560f35f1 100644
--- a/asap-planner-rs/src/sql/generator.rs
+++ b/asap-planner-rs/src/sql/generator.rs
@@ -18,7 +18,7 @@ use crate::StreamingEngine;
 pub struct SQLRuntimeOptions {
     pub streaming_engine: StreamingEngine,
     pub query_evaluation_time: Option<f64>,
-    pub data_ingestion_interval: u64,
+    pub data_ingestion_interval_ms: u64,
 }
 
 pub fn generate_sql_plan(
@@ -38,12 +38,12 @@ pub fn generate_sql_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
             )));
         }
     }
@@ -79,8 +79,8 @@ pub fn generate_sql_plan(
         for query_string in &qg.queries {
             let processor = SQLSingleQueryProcessor::new(
                 query_string.clone(),
-                qg.repetition_delay,
-                opts.data_ingestion_interval,
+                qg.repetition_delay_ms,
+                opts.data_ingestion_interval_ms,
                 config.tables.clone(),
                 opts.streaming_engine,
                 config.sketch_parameters.clone(),
diff --git a/asap-planner-rs/tests/elastic_dsl_integration.rs b/asap-planner-rs/tests/elastic_dsl_integration.rs
index 7d248853..de2138db 100644
--- a/asap-planner-rs/tests/elastic_dsl_integration.rs
+++ b/asap-planner-rs/tests/elastic_dsl_integration.rs
@@ -13,7 +13,7 @@ fn indent_block(text: &str, indent: usize) -> String {
         .join("\n")
 }
 
-fn elastic_yaml(index: &str, time_field: &str, query: &str, t_repeat: u64) -> String {
+fn elastic_yaml(index: &str, time_field: &str, query: &str, t_repeat_ms: u64) -> String {
     format!(
         r#"
 query_groups:
@@ -23,7 +23,7 @@ query_groups:
     queries:
       - |
 {query}
-    repetition_delay: {t_repeat}
+    repetition_delay_ms: {t_repeat_ms}
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 1.0
@@ -33,18 +33,28 @@ aggregate_cleanup:
         index = index,
         time_field = time_field,
         query = indent_block(query, 8),
-        t_repeat = t_repeat,
+        t_repeat_ms = t_repeat_ms,
     )
 }
 
-fn elastic_output(index: &str, time_field: &str, query: &str, t_repeat: u64) -> PlannerOutput {
-    let yaml = elastic_yaml(index, time_field, query, t_repeat);
+fn elastic_output(index: &str, time_field: &str, query: &str, t_repeat_ms: u64) -> PlannerOutput {
+    elastic_output_with_interval(index, time_field, query, t_repeat_ms, 15_000)
+}
+
+fn elastic_output_with_interval(
+    index: &str,
+    time_field: &str,
+    query: &str,
+    t_repeat_ms: u64,
+    data_ingestion_interval_ms: u64,
+) -> PlannerOutput {
+    let yaml = elastic_yaml(index, time_field, query, t_repeat_ms);
     let mut file = NamedTempFile::new().unwrap();
     file.write_all(yaml.as_bytes()).unwrap();
 
     let opts = ElasticRuntimeOptions {
         streaming_engine: StreamingEngine::Arroyo,
-        data_ingestion_interval: 15,
+        data_ingestion_interval_ms,
     };
 
     ElasticController::from_file(Path::new(file.path()), opts)
@@ -86,7 +96,7 @@ fn assert_index_schema(
 fn elastic_querydsl_emits_index_schema() {
     let opts = ElasticRuntimeOptions {
         streaming_engine: StreamingEngine::Arroyo,
-        data_ingestion_interval: 15,
+        data_ingestion_interval_ms: 15_000,
     };
     let c = ElasticController::from_file(Path::new("tests/elastic_example.yaml"), opts).unwrap();
     let out = c.generate().unwrap();
@@ -143,7 +153,7 @@ fn elastic_sum_produces_basic_plan_and_schema() {
     }
 }
 "#;
-    let out = elastic_output("metrics", "\"@timestamp\"", query, 300);
+    let out = elastic_output("metrics", "\"@timestamp\"", query, 300_000);
 
     assert_eq!(out.streaming_aggregation_count(), 2);
     assert_eq!(out.inference_query_count(), 1);
@@ -195,7 +205,7 @@ fn elastic_avg_produces_three_configs() {
     }
 }
 "#;
-    let out = elastic_output("metrics", "\"@timestamp\"", query, 300);
+    let out = elastic_output("metrics", "\"@timestamp\"", query, 300_000);
 
     assert_eq!(out.streaming_aggregation_count(), 2);
     assert_eq!(out.inference_query_count(), 1);
@@ -248,7 +258,7 @@ fn elastic_min_produces_exact_plan() {
     }
 }
 "#;
-    let out = elastic_output("metrics", "\"@timestamp\"", query, 300);
+    let out = elastic_output("metrics", "\"@timestamp\"", query, 300_000);
 
     assert_eq!(out.streaming_aggregation_count(), 1);
     assert_eq!(out.inference_query_count(), 1);
@@ -301,7 +311,7 @@ fn elastic_percentiles_produce_kll_plan() {
     }
 }
 "#;
-    let out = elastic_output("metrics", "\"@timestamp\"", query, 300);
+    let out = elastic_output("metrics", "\"@timestamp\"", query, 300_000);
 
     assert_eq!(out.streaming_aggregation_count(), 1);
     assert_eq!(out.inference_query_count(), 1);
@@ -358,7 +368,7 @@ query_groups:
                         }
                     }
                 }
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 1.0
@@ -397,7 +407,7 @@ query_groups:
                         }
                     }
                 }
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 1.0
@@ -410,7 +420,7 @@ aggregate_cleanup:
 
     let opts = ElasticRuntimeOptions {
         streaming_engine: StreamingEngine::Arroyo,
-        data_ingestion_interval: 15,
+        data_ingestion_interval_ms: 15_000,
     };
 
     let c = ElasticController::from_file(Path::new(file.path()), opts).unwrap();
@@ -449,3 +459,37 @@ aggregate_cleanup:
         other => panic!("expected elastic querydsl schema, got {:?}", other),
     }
 }
+
+// ── sub-second precision ──────────────────────────────────────────────────────
+
+/// repetition_delay_ms = 500 (sub-second): validates ms-precision plumbing end-to-end.
+#[test]
+fn sub_second_repetition_delay_ms() {
+    let query = r#"
+{
+    "aggs": {
+        "avg_cpu": {
+            "avg": {
+                "field": "cpu_usage"
+            }
+        }
+    },
+    "query": {
+        "bool": {
+            "filter": [
+                {
+                    "range": {
+                        "@timestamp": {
+                            "gte": "now-5m",
+                            "lte": "now"
+                        }
+                    }
+                }
+            ]
+        }
+    }
+}"#;
+    let out = elastic_output_with_interval("metrics", "\"@timestamp\"", query, 500, 500);
+    assert_eq!(out.streaming_aggregation_count(), 2);
+    assert!(out.all_tumbling_window_sizes_eq(500));
+}
diff --git a/asap-planner-rs/tests/elastic_example.yaml b/asap-planner-rs/tests/elastic_example.yaml
index ac311de7..66014c45 100644
--- a/asap-planner-rs/tests/elastic_example.yaml
+++ b/asap-planner-rs/tests/elastic_example.yaml
@@ -27,7 +27,7 @@ query_groups:
             }
           }
         }
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 1.0
diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs
index 5c281547..8b7bb1b0 100644
--- a/asap-planner-rs/tests/sql_integration.rs
+++ b/asap-planner-rs/tests/sql_integration.rs
@@ -7,7 +7,7 @@ fn sql_opts() -> SQLRuntimeOptions {
         streaming_engine: StreamingEngine::Arroyo,
         // Fixed evaluation time so NOW()-relative timestamps are deterministic.
         query_evaluation_time: Some(1_000_000.0),
-        data_ingestion_interval: 15,
+        data_ingestion_interval_ms: 15_000,
     }
 }
 
@@ -17,8 +17,8 @@ fn sql_opts() -> SQLRuntimeOptions {
 ///   time_column       : time
 ///   value_columns     : [cpu_usage]
 ///   metadata_columns  : [hostname, datacenter, region]
-/// data_ingestion_interval = 15 s
-fn one_query_config(query: &str, t_repeat: u64) -> String {
+/// data_ingestion_interval_ms = 15_000 ms
+fn one_query_config(query: &str, t_repeat_ms: u64) -> String {
     format!(
         r#"
 tables:
@@ -28,7 +28,7 @@ tables:
     metadata_columns: [hostname, datacenter, region]
 query_groups:
   - id: 1
-    repetition_delay: {t_repeat}
+    repetition_delay_ms: {t_repeat_ms}
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -68,7 +68,7 @@ aggregate_cleanup:
 #[test]
 fn temporal_sum() {
     let q = "SELECT SUM(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -105,7 +105,7 @@ fn temporal_sum() {
 #[test]
 fn temporal_count() {
     let q = "SELECT COUNT(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -142,7 +142,7 @@ fn temporal_count() {
 #[test]
 fn temporal_min() {
     let q = "SELECT MIN(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -179,7 +179,7 @@ fn temporal_min() {
 #[test]
 fn temporal_max() {
     let q = "SELECT MAX(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -216,7 +216,7 @@ fn temporal_max() {
 #[test]
 fn temporal_avg() {
     let q = "SELECT AVG(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -253,7 +253,7 @@ fn temporal_avg() {
 #[test]
 fn temporal_quantile() {
     let q = "SELECT quantile(0.95)(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -299,7 +299,7 @@ fn temporal_quantile() {
 #[test]
 fn temporal_quantile_half_open() {
     let q = "SELECT quantile(0.95)(cpu_usage) FROM metrics_table WHERE time >= DATEADD(s, -300, NOW()) AND time < NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -338,7 +338,7 @@ fn temporal_quantile_half_open() {
 #[test]
 fn half_open_gt_lte_returns_sql_parse_error() {
     let q = "SELECT quantile(0.95)(cpu_usage) FROM metrics_table WHERE time > DATEADD(s, -300, NOW()) AND time <= NOW() GROUP BY datacenter";
-    let result = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let result = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate();
     assert!(matches!(result, Err(ControllerError::SqlParse(_))));
@@ -360,7 +360,7 @@ fn half_open_gt_lte_returns_sql_parse_error() {
 #[test]
 fn temporal_quantile_percentile_syntax() {
     let q = "SELECT PERCENTILE(cpu_usage, 95) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -397,7 +397,7 @@ fn temporal_quantile_percentile_syntax() {
 #[test]
 fn temporal_quantile_quoted_dateadd_unit() {
     let q = "SELECT PERCENTILE(cpu_usage, 95) FROM metrics_table WHERE time BETWEEN DATEADD('s', -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -434,7 +434,7 @@ fn temporal_quantile_quoted_dateadd_unit() {
 #[test]
 fn temporal_quantile_cast_datetime_bounds() {
     let q = "SELECT PERCENTILE(cpu_usage, 95) FROM metrics_table WHERE time BETWEEN DATEADD('s', -300, CAST('2024-01-01T00:05:00Z' AS DATETIME)) AND CAST('2024-01-01T00:05:00Z' AS DATETIME) GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -468,7 +468,7 @@ fn temporal_quantile_cast_datetime_bounds() {
 
 // ── COUNT(DISTINCT) / HLL (spatial 1 s) ───────────────────────────────────────
 
-fn netflow_one_query_config(query: &str, t_repeat: u64) -> String {
+fn netflow_one_query_config(query: &str, t_repeat_ms: u64) -> String {
     format!(
         r#"
 tables:
@@ -478,7 +478,7 @@ tables:
     metadata_columns: [srcip, dstip, proto]
 query_groups:
   - id: 1
-    repetition_delay: {t_repeat}
+    repetition_delay_ms: {t_repeat_ms}
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -495,7 +495,7 @@ fn sql_opts_1s_ingest() -> SQLRuntimeOptions {
     SQLRuntimeOptions {
         streaming_engine: StreamingEngine::Arroyo,
         query_evaluation_time: Some(1_000_000.0),
-        data_ingestion_interval: 1,
+        data_ingestion_interval_ms: 1_000,
     }
 }
 
@@ -503,7 +503,7 @@ fn sql_opts_1s_ingest() -> SQLRuntimeOptions {
 #[test]
 fn spatial_count_distinct_hll() {
     let q = "SELECT srcip, COUNT(DISTINCT dstip) AS unique_peers FROM netflow_table WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) GROUP BY srcip";
-    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1_000), sql_opts_1s_ingest())
         .unwrap()
         .generate()
         .unwrap();
@@ -537,7 +537,7 @@ fn spatial_count_distinct_hll() {
 
 // ── T-value variants for SUM (range = 300 s fixed) ───────────────────────────
 //
-// These three tests use the same query and differ only in repetition_delay (T).
+// These three tests use the same query and differ only in repetition_delay_ms (T).
 // Window size = T (the repetition delay), not the query range.
 // They verify cleanup scales correctly with T and that T > range is valid.
 
@@ -545,7 +545,7 @@ fn spatial_count_distinct_hll() {
 #[test]
 fn temporal_sum_t30() {
     let q = "SELECT SUM(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 30), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 30_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -583,7 +583,7 @@ fn temporal_sum_t30() {
 #[test]
 fn temporal_sum_t600() {
     let q = "SELECT SUM(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let out = SQLController::from_yaml(&one_query_config(q, 600), sql_opts())
+    let out = SQLController::from_yaml(&one_query_config(q, 600_000), sql_opts())
         .unwrap()
         .generate()
         .unwrap();
@@ -628,7 +628,7 @@ tables:
     metadata_columns: [hostname, datacenter, region]
 query_groups:
   - id: 1
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -636,7 +636,7 @@ query_groups:
       - >-
         SELECT MIN(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter
   - id: 2
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -669,7 +669,7 @@ tables:
     metadata_columns: [hostname, datacenter, region]
 query_groups:
   - id: 1
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -703,7 +703,7 @@ tables:
     metadata_columns: [hostname, datacenter, region]
 query_groups:
   - id: 1
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -711,7 +711,7 @@ query_groups:
       - >-
         {q}
   - id: 2
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -740,7 +740,7 @@ tables:
     metadata_columns: [hostname, datacenter, region]
 query_groups:
   - id: 1
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -748,7 +748,7 @@ query_groups:
       - >-
         SELECT SUM(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter
   - id: 2
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -774,7 +774,7 @@ aggregate_cleanup:
 fn generate_to_dir_writes_both_yaml_files() {
     let dir = tempfile::tempdir().unwrap();
     let q = "SELECT SUM(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate_to_dir(dir.path())
         .unwrap();
@@ -793,7 +793,7 @@ fn malformed_yaml_returns_parse_error() {
 #[test]
 fn malformed_sql_returns_sql_parse_error() {
     let q = "NOT VALID SQL AT ALL %%%";
-    let result = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let result = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate();
     assert!(matches!(result, Err(ControllerError::SqlParse(_))));
@@ -802,7 +802,7 @@ fn malformed_sql_returns_sql_parse_error() {
 #[test]
 fn query_referencing_unknown_table_returns_error() {
     let q = "SELECT SUM(cpu_usage) FROM nonexistent_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let result = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let result = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate();
     assert!(matches!(
@@ -814,7 +814,7 @@ fn query_referencing_unknown_table_returns_error() {
 #[test]
 fn query_referencing_unknown_value_column_returns_sql_parse_error() {
     let q = "SELECT SUM(nonexistent_col) FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter";
-    let result = SQLController::from_yaml(&one_query_config(q, 300), sql_opts())
+    let result = SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts())
         .unwrap()
         .generate();
     assert!(matches!(result, Err(ControllerError::SqlParse(_))));
@@ -832,7 +832,7 @@ tables:
     metadata_columns: [hostname, datacenter, region]
 query_groups:
   - id: 1
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -863,7 +863,7 @@ tables:
     metadata_columns: [hostname, datacenter, region]
 query_groups:
   - id: 1
-    repetition_delay: 300
+    repetition_delay_ms: 300000
     controller_options:
       accuracy_sla: 0.95
       latency_sla: 100.0
@@ -882,13 +882,13 @@ aggregate_cleanup:
     ));
 }
 
-/// T that is not a multiple of data_ingestion_interval is invalid: sketch windows
+/// T that is not a multiple of data_ingestion_interval_ms is invalid: sketch windows
 /// must align with the ingestion cadence.
-/// data_ingestion_interval = 15 s, T = 200 s → 200 mod 15 ≠ 0 → PlannerError.
+/// data_ingestion_interval_ms = 15_000, T_ms = 200_000 → 200_000 mod 15_000 ≠ 0 → PlannerError.
 #[test]
 fn t_not_multiple_of_data_ingestion_interval_returns_planner_error() {
     let q = "SELECT SUM(cpu_usage) FROM metrics_table WHERE time BETWEEN DATEADD(s, -200, NOW()) AND NOW() GROUP BY datacenter";
-    let result = SQLController::from_yaml(&one_query_config(q, 200), sql_opts())
+    let result = SQLController::from_yaml(&one_query_config(q, 200_000), sql_opts())
         .unwrap()
         .generate();
     assert!(matches!(result, Err(ControllerError::PlannerError(_))));
@@ -902,7 +902,7 @@ fn spatial_count_topk_heap() {
     let q = "SELECT srcip, COUNT(pkt_len) AS transfer_events FROM netflow_table \
              WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) \
              GROUP BY srcip ORDER BY transfer_events DESC LIMIT 10";
-    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1_000), sql_opts_1s_ingest())
         .unwrap()
         .generate()
         .unwrap();
@@ -943,7 +943,7 @@ fn spatial_sum_topk_heap() {
     let q = "SELECT srcip, SUM(pkt_len) AS total_bytes FROM netflow_table \
              WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) \
              GROUP BY srcip ORDER BY total_bytes DESC LIMIT 10";
-    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1_000), sql_opts_1s_ingest())
         .unwrap()
         .generate()
         .unwrap();
@@ -984,7 +984,7 @@ fn spatial_count_without_order_by_is_not_topk() {
     let q = "SELECT srcip, COUNT(pkt_len) AS transfer_events FROM netflow_table \
              WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) \
              GROUP BY srcip";
-    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1_000), sql_opts_1s_ingest())
         .unwrap()
         .generate()
         .unwrap();
@@ -1001,7 +1001,7 @@ fn spatial_count_order_by_asc_limit_is_not_topk() {
     let q = "SELECT srcip, COUNT(pkt_len) AS transfer_events FROM netflow_table \
              WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) \
              GROUP BY srcip ORDER BY transfer_events ASC LIMIT 10";
-    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1_000), sql_opts_1s_ingest())
         .unwrap()
         .generate()
         .unwrap();
@@ -1012,13 +1012,37 @@ fn spatial_count_order_by_asc_limit_is_not_topk() {
     assert!(!out.has_aggregation_type("CountMinSketchWithHeap"));
 }
 
+// ── sub-second precision ──────────────────────────────────────────────────────
+
+/// data_ingestion_interval_ms = 500 (sub-second): spatial HLL with 500 ms window.
+/// Validates that ms-precision plumbing works end-to-end below 1 s.
+#[test]
+fn sub_second_data_ingestion_interval_ms() {
+    let q = "SELECT srcip, COUNT(DISTINCT dstip) AS unique_peers FROM netflow_table \
+             WHERE time BETWEEN DATEADD(s, -2, NOW()) AND NOW() \
+             GROUP BY srcip";
+    let opts = SQLRuntimeOptions {
+        streaming_engine: StreamingEngine::Arroyo,
+        query_evaluation_time: Some(1_000_000.0),
+        data_ingestion_interval_ms: 500,
+    };
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 500), opts)
+        .unwrap()
+        .generate()
+        .unwrap();
+
+    assert_eq!(out.streaming_aggregation_count(), 1);
+    assert!(out.has_aggregation_type("HLL"));
+    assert!(out.all_tumbling_window_sizes_eq(500));
+}
+
 /// LIMIT 0 is treated as non-top-k and uses the normal CMS + DeltaSet path.
 #[test]
 fn spatial_count_limit_zero_is_not_topk() {
     let q = "SELECT srcip, COUNT(pkt_len) AS transfer_events FROM netflow_table \
              WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) \
              GROUP BY srcip ORDER BY transfer_events DESC LIMIT 0";
-    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1_000), sql_opts_1s_ingest())
         .unwrap()
         .generate()
         .unwrap();
diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py
index e20dff84..046509e6 100644
--- a/asap-tools/experiments/experiment_run_clickhouse.py
+++ b/asap-tools/experiments/experiment_run_clickhouse.py
@@ -296,7 +296,7 @@ def main(cfg: DictConfig) -> None:
     mode_server_urls = {m["mode"]: servers_by_name[m["server"]] for m in experiment_cfg}
     clickhouse_url = servers_by_name["clickhouse"]
     clickhouse_http_port = urlparse(clickhouse_url).port
-    data_ingestion_interval = int(ep.data_ingestion_interval)
+    data_ingestion_interval_ms = int(ep.data_ingestion_interval_ms)
 
     # --- generate prometheus-client config YAMLs for each experiment mode ---
     experiment_modes = config.generate_clickhouse_client_configs(
@@ -420,7 +420,7 @@ def main(cfg: DictConfig) -> None:
                     database=CLICKHOUSE_DATABASE,
                 ),
                 query_language="sql",
-                data_ingestion_interval=data_ingestion_interval,
+                data_ingestion_interval_ms=data_ingestion_interval_ms,
             )
             sync.rsync_controller_config_remote_to_local(
                 provider, remote_controller_dir, local_controller_dir, node_offset
diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py
index 0797a01c..a08a9a50 100644
--- a/asap-tools/experiments/experiment_utils/config.py
+++ b/asap-tools/experiments/experiment_utils/config.py
@@ -212,8 +212,8 @@ def validate_experiment_config(
             raise ValueError(
                 f"Query group {i} missing 'client_options.repetitions' field"
             )
-        if "repetition_delay" not in group:
-            raise ValueError(f"Query group {i} missing 'repetition_delay' field")
+        if "repetition_delay_ms" not in group:
+            raise ValueError(f"Query group {i} missing 'repetition_delay_ms' field")
 
     # Validate exporters structure
     if "exporter_list" not in experiment_params.exporters:
@@ -284,7 +284,7 @@ def get_minimum_experiment_running_time(experiment_params: DictConfig) -> int:
     for query_group in query_groups:
         query_group_starting_delay = query_group.client_options.starting_delay
         query_group_repetitions = query_group.client_options.repetitions
-        query_group_reptition_delay = query_group.repetition_delay
+        query_group_reptition_delay = query_group.repetition_delay_ms / 1000
 
         query_group_running_time = (
             query_group_starting_delay
@@ -369,18 +369,6 @@ def generate_controller_client_configs(
         controller_only_config = {
             k: v for k, v in full_config.items() if k in CONTROLLER_ALLOWED_KEYS
         }
-        # asap-planner-rs renamed QueryGroup.repetition_delay -> repetition_delay_ms
-        # (issue #398). full_config keeps the original seconds value for
-        # prometheus_client (a separate tool, unaffected by that rename) —
-        # only the controller-bound copy needs the renamed/scaled key.
-        if "query_groups" in controller_only_config:
-            controller_only_config["query_groups"] = [
-                {
-                    **{k: v for k, v in qg.items() if k != "repetition_delay"},
-                    "repetition_delay_ms": qg["repetition_delay"] * 1000,
-                }
-                for qg in controller_only_config["query_groups"]
-            ]
         with open(
             os.path.join(
                 output_dir, "{}_controller_input.yaml".format(experiment_mode["mode"])
@@ -692,7 +680,7 @@ def generate_clickhouse_client_configs(
     Args:
         query_groups: Iterable of query-group dicts (or DictConfig/ListConfig).
             Each entry must have ``sql_file`` and may have ``client_options``
-            (``starting_delay``, ``repetitions``) and ``repetition_delay``.
+            (``starting_delay``, ``repetitions``) and ``repetition_delay_ms``.
         local_experiment_dir: Local directory under which
             ``controller_client_configs/`` is created.
         mode_server_urls: Mapping of mode name to ClickHouse server URL, e.g.
@@ -734,7 +722,7 @@ def generate_clickhouse_client_configs(
             {
                 "id": idx,
                 "queries": queries,
-                "repetition_delay": group.get("repetition_delay", 0),
+                "repetition_delay_ms": group.get("repetition_delay_ms", 0),
                 "client_options": client_opts,
                 "time_window_seconds": group.get("time_window_seconds"),
             }
@@ -775,7 +763,7 @@ def generate_sql_planner_input(query_groups: Any, dataset_cfg: Any) -> str:
 
     Args:
         query_groups: ListConfig of query group dicts.
-            Each entry must have ``sql_file``, ``repetition_delay``, and
+            Each entry must have ``sql_file``, ``repetition_delay_ms``, and
             ``controller_options`` (``accuracy_sla``, ``latency_sla``).
         dataset_cfg: DictConfig with ``table``/``name``, and ``precompute``
             sub-config (``timestamp_col``, ``value_col``, ``label_cols``).
@@ -813,7 +801,7 @@ def generate_sql_planner_input(query_groups: Any, dataset_cfg: Any) -> str:
         planner_query_groups.append(
             {
                 "id": idx + 1,
-                "repetition_delay": int(group.get("repetition_delay", 0)),
+                "repetition_delay_ms": int(group.get("repetition_delay_ms", 0)),
                 "queries": queries,
                 "controller_options": {
                     "accuracy_sla": float(ctrl_opts.get("accuracy_sla", 0.95)),
diff --git a/asap-tools/experiments/experiment_utils/services/misc.py b/asap-tools/experiments/experiment_utils/services/misc.py
index 5ca942ec..bae507a8 100644
--- a/asap-tools/experiments/experiment_utils/services/misc.py
+++ b/asap-tools/experiments/experiment_utils/services/misc.py
@@ -207,7 +207,7 @@ def start(
         discovery_backend: DiscoveryBackend,
         prometheus_scrape_interval_ms: Optional[int] = None,
         query_language: str = "promql",
-        data_ingestion_interval: Optional[int] = None,
+        data_ingestion_interval_ms: Optional[int] = None,
         **kwargs,
     ) -> None:
         """
@@ -224,7 +224,7 @@ def start(
             prometheus_scrape_interval_ms: Required for PromQL mode (milliseconds; passed as
                 --prometheus_scrape_interval_ms / --prometheus-scrape-interval-ms)
             query_language: 'promql' (default) or 'sql'
-            data_ingestion_interval: Required for SQL mode (seconds)
+            data_ingestion_interval_ms: Required for SQL mode (milliseconds)
             **kwargs: Additional configuration
         """
         if self.use_container:
@@ -236,7 +236,7 @@ def start(
                 discovery_backend,
                 prometheus_scrape_interval_ms,
                 query_language,
-                data_ingestion_interval,
+                data_ingestion_interval_ms,
             )
         else:
             return self._start_bare_metal(
@@ -247,7 +247,7 @@ def start(
                 discovery_backend,
                 prometheus_scrape_interval_ms,
                 query_language,
-                data_ingestion_interval,
+                data_ingestion_interval_ms,
             )
 
     def _start_bare_metal(
@@ -259,7 +259,7 @@ def _start_bare_metal(
         discovery_backend: DiscoveryBackend,
         prometheus_scrape_interval_ms: Optional[int],
         query_language: str,
-        data_ingestion_interval: Optional[int],
+        data_ingestion_interval_ms: Optional[int],
     ) -> None:
         controller_log = os.path.join(controller_remote_output_dir, "controller.log")
         cmd = (
@@ -271,8 +271,8 @@ def _start_bare_metal(
         )
         if prometheus_scrape_interval_ms is not None:
             cmd += f" --prometheus_scrape_interval_ms {prometheus_scrape_interval_ms}"
-        if data_ingestion_interval is not None:
-            cmd += f" --data-ingestion-interval {data_ingestion_interval}"
+        if data_ingestion_interval_ms is not None:
+            cmd += f" --data-ingestion-interval-ms {data_ingestion_interval_ms}"
         if discovery_backend.type == "prometheus":
             cmd += f" --prometheus-url {discovery_backend.url}"
         elif discovery_backend.type == "clickhouse":
@@ -302,7 +302,7 @@ def _start_containerized(
         discovery_backend: DiscoveryBackend,
         prometheus_scrape_interval_ms: Optional[int],
         query_language: str,
-        data_ingestion_interval: Optional[int],
+        data_ingestion_interval_ms: Optional[int],
     ):
         controller_dir = os.path.join(
             self.provider.get_home_dir(), "code", "asap-planner-rs"
@@ -334,8 +334,10 @@ def _start_containerized(
             generate_cmd += (
                 f" --prometheus-scrape-interval-ms {prometheus_scrape_interval_ms}"
             )
-        if data_ingestion_interval is not None:
-            generate_cmd += f" --data-ingestion-interval {data_ingestion_interval}"
+        if data_ingestion_interval_ms is not None:
+            generate_cmd += (
+                f" --data-ingestion-interval-ms {data_ingestion_interval_ms}"
+            )
         if discovery_backend.type == "prometheus":
             generate_cmd += f" --prometheus-url {discovery_backend.url}"
         elif discovery_backend.type == "clickhouse":
diff --git a/asap-tools/experiments/generate_controller_compose.py b/asap-tools/experiments/generate_controller_compose.py
index 79311d66..15c643dc 100644
--- a/asap-tools/experiments/generate_controller_compose.py
+++ b/asap-tools/experiments/generate_controller_compose.py
@@ -21,7 +21,7 @@ def generate_compose_file(
     punting: bool,
     prometheus_url: str,
     query_language: str,
-    data_ingestion_interval: Optional[int],
+    data_ingestion_interval_ms: Optional[int],
 ):
     """Generate docker-compose.yml from template with provided variables."""
 
@@ -47,7 +47,7 @@ def generate_compose_file(
         "punting": punting,
         "prometheus_url": prometheus_url,
         "query_language": query_language,
-        "data_ingestion_interval": data_ingestion_interval,
+        "data_ingestion_interval_ms": data_ingestion_interval_ms,
     }
 
     # Render the template
@@ -136,17 +136,17 @@ def main():
         help="Query language for the controller",
     )
     parser.add_argument(
-        "--data-ingestion-interval",
+        "--data-ingestion-interval-ms",
         type=int,
         default=None,
-        help="Data ingestion interval in seconds (SQL mode only)",
+        help="Data ingestion interval in milliseconds (SQL mode only)",
     )
 
     args = parser.parse_args()
 
-    if args.data_ingestion_interval is not None and args.query_language != "sql":
+    if args.data_ingestion_interval_ms is not None and args.query_language != "sql":
         parser.error(
-            "--data-ingestion-interval is only valid when --query-language is 'sql'"
+            "--data-ingestion-interval-ms is only valid when --query-language is 'sql'"
         )
 
     generate_compose_file(
@@ -161,7 +161,7 @@ def main():
         punting=args.punting,
         prometheus_url=args.prometheus_url,
         query_language=args.query_language,
-        data_ingestion_interval=args.data_ingestion_interval,
+        data_ingestion_interval_ms=args.data_ingestion_interval_ms,
     )
 
 
diff --git a/asap-tools/experiments/generate_workload.py b/asap-tools/experiments/generate_workload.py
index d1047bb4..1259f7bb 100644
--- a/asap-tools/experiments/generate_workload.py
+++ b/asap-tools/experiments/generate_workload.py
@@ -378,7 +378,7 @@ def get_base_config() -> Dict:
             {
                 "id": 1,
                 "queries": [],  # Will be populated
-                "repetition_delay": 10,
+                "repetition_delay_ms": 10000,
                 "client_options": {
                     "repetitions": 10,
                     "query_time_offset": 10,
diff --git a/asap-tools/experiments/grafana_config.py b/asap-tools/experiments/grafana_config.py
index 4ab14ba4..7d5d6950 100644
--- a/asap-tools/experiments/grafana_config.py
+++ b/asap-tools/experiments/grafana_config.py
@@ -10,6 +10,7 @@
 import os
 import sys
 import typing
+import warnings
 import requests
 import re
 from dataclasses import dataclass
@@ -46,7 +47,7 @@ class QueryConfig:
     """Configuration for a query with timing parameters."""
 
     query: str
-    repetition_delay: int
+    repetition_delay_ms: int
     query_time_offset: int
 
 
@@ -306,7 +307,7 @@ def from_experiment_config(cls, cfg: DictConfig) -> "ExperimentDashboardConfig":
             if not hasattr(query_group, "queries") or not query_group.queries:
                 continue
 
-            repetition_delay = getattr(query_group, "repetition_delay", 30)
+            repetition_delay_ms = getattr(query_group, "repetition_delay_ms", 30000)
             query_time_offset = 0
 
             if hasattr(query_group, "client_options"):
@@ -317,7 +318,7 @@ def from_experiment_config(cls, cfg: DictConfig) -> "ExperimentDashboardConfig":
                 queries.append(
                     QueryConfig(
                         query=query_str,
-                        repetition_delay=repetition_delay,
+                        repetition_delay_ms=repetition_delay_ms,
                         query_time_offset=query_time_offset,
                     )
                 )
@@ -499,8 +500,20 @@ def _calculate_refresh_interval(self, queries: List[QueryConfig]) -> str:
         if not queries:
             return "30s"
 
-        min_delay = min(q.repetition_delay for q in queries)
-        return f"{min_delay}s"
+        min_delay = min(q.repetition_delay_ms for q in queries)
+        if min_delay < 1000:
+            warnings.warn(
+                f"Minimum repetition_delay_ms ({min_delay} ms) is sub-second; "
+                "Grafana refresh intervals must be at least 1 second.",
+                UserWarning,
+                stacklevel=2,
+            )
+            raise ValueError(
+                f"Cannot generate Grafana dashboard: minimum repetition_delay_ms "
+                f"({min_delay} ms) is less than 1000 ms. Grafana requires a refresh "
+                "interval of at least 1 second."
+            )
+        return f"{min_delay // 1000}s"
 
     def _calculate_time_range(self, queries: List[QueryConfig]) -> tuple[str, str]:
         """
diff --git a/asap-tools/queriers/prometheus-client/classes/config.py b/asap-tools/queriers/prometheus-client/classes/config.py
index a7fe215f..01df6534 100644
--- a/asap-tools/queriers/prometheus-client/classes/config.py
+++ b/asap-tools/queriers/prometheus-client/classes/config.py
@@ -37,7 +37,7 @@ def __init__(
         id: int,
         queries: List[str],
         # repetitions: int,
-        repetition_delay: int,
+        repetition_delay_ms: int,
         options: Dict[str, Any],
         time_window_seconds: Optional[int],
         # starting_delay: int,
@@ -50,7 +50,7 @@ def __init__(
         self.id = id
         self.queries = queries
         # self.repetitions = repetitions
-        self.repetition_delay = repetition_delay
+        self.repetition_delay_ms = repetition_delay_ms
         self.time_window_seconds = time_window_seconds
         self.__dict__.update(options)
         # self.starting_delay = starting_delay
@@ -60,18 +60,10 @@ def __init__(
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "QueryGroupConfig":
-        # return cls(
-        #     id=data["id"],
-        #     repetitions=data["repetitions"],
-        #     repetition_delay=data["repetition_delay"],
-        #     starting_delay=data["starting_delay"] if "starting_delay" in data else 0,
-        #     options=data["options"],
-        #     queries=data["queries"],
-        # )
         return cls(
             id=data["id"],
             queries=data["queries"],
-            repetition_delay=data["repetition_delay"],
+            repetition_delay_ms=data["repetition_delay_ms"],
             options=data["client_options"],
             time_window_seconds=data.get("time_window_seconds"),
         )
diff --git a/asap-tools/queriers/prometheus-client/main_prometheus_client.py b/asap-tools/queriers/prometheus-client/main_prometheus_client.py
index 121f834d..104e4d68 100644
--- a/asap-tools/queriers/prometheus-client/main_prometheus_client.py
+++ b/asap-tools/queriers/prometheus-client/main_prometheus_client.py
@@ -142,17 +142,16 @@ def get_query_unix_time(
     query: Query,
     query_unix_time: UnixTimestamp,
     query_start_times: Optional[QueryStartTimes],
-    repetition_delay: int,
+    repetition_delay_ms: int,
 ) -> UnixTimestamp:
     if query_start_times is None or query not in query_start_times:
         return query_unix_time
 
     query_alignment_time = query_start_times[query]
-    # we want the latest timestamp that is query_aligment_time + N * repetition_delay
-    query_unix_time = int(
-        query_unix_time - (query_unix_time - query_alignment_time) % repetition_delay
-    )
-    return query_unix_time
+    # we want the latest timestamp that is query_alignment_time + N * (repetition_delay_ms / 1000)
+    # Keep arithmetic in integer ms to avoid IEEE 754 drift from non-round divisions.
+    diff_ms = round((query_unix_time - query_alignment_time) * 1000)
+    return int(query_unix_time - (diff_ms % repetition_delay_ms) / 1000)
 
 
 def execute_single_query(
@@ -342,7 +341,7 @@ def handle_query_group(
                         query,
                         query_unix_time,
                         query_start_times,
-                        query_group.repetition_delay,
+                        query_group.repetition_delay_ms,
                     )
 
                     for server_name, server_object in servers.items():
@@ -385,7 +384,7 @@ def handle_query_group(
                     query,
                     query_unix_time,
                     query_start_times,
-                    query_group.repetition_delay,
+                    query_group.repetition_delay_ms,
                 )
 
                 logger.debug("Unix time for query: {}", current_query_unix_time)
@@ -430,7 +429,7 @@ def handle_query_group(
             latency_exporter.export_repetition(repetition_idx, result)
 
         if repetition_idx < query_group.repetitions - 1:
-            time.sleep(query_group.repetition_delay)
+            time.sleep(query_group.repetition_delay_ms / 1000)
 
     if latency_exporter is not None:
         latency_exporter.shutdown()
@@ -677,9 +676,9 @@ def main(args: Any) -> None:
     max_duration = 0
     for query_group in config.query_groups:
         assert query_group.repetitions is not None
-        assert query_group.repetition_delay is not None
+        assert query_group.repetition_delay_ms is not None
         duration = (
-            query_group.repetition_delay * query_group.repetitions
+            query_group.repetition_delay_ms / 1000 * query_group.repetitions
             + query_group.starting_delay
             - min_starting_delay
         )