diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs index 6752ca33..64a177a4 100644 --- a/asap-planner-rs/src/config/input.rs +++ b/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, pub queries: Vec, @@ -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, } #[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] pub struct AggregateCleanupConfig { pub policy: Option, } #[derive(Debug, Clone, Deserialize, Default)] +#[serde(deny_unknown_fields)] pub struct SketchParameterOverrides { #[serde(rename = "CountMinSketch")] pub count_min_sketch: Option, @@ -91,12 +96,14 @@ 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, @@ -104,12 +111,14 @@ pub struct CmsHeapParams { } #[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, @@ -117,11 +126,13 @@ pub struct HydraParams { } #[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, pub tables: Vec, @@ -130,14 +141,16 @@ pub struct SQLControllerConfig { } #[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] pub struct SQLQueryGroup { pub id: Option, pub queries: Vec, - 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,6 +160,7 @@ pub struct TableDefinition { } #[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] pub struct ElasticDSLControllerConfig { pub query_groups: Vec, pub sketch_parameters: Option, @@ -154,10 +168,11 @@ pub struct ElasticDSLControllerConfig { } #[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] pub struct ElasticDSLQueryGroup { pub id: Option, pub queries: Vec, - 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, + #[arg(long = "data-ingestion-interval-ms", required = false)] + data_ingestion_interval_ms: Option, /// 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/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 { 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, @@ -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..8215e052 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, #[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, streaming_engine: StreamingEngine, sketch_parameters: Option, @@ -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) 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, 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/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, - 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..a39bb4e1 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(); @@ -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();