From 59c5a2407ef79fdacd749ec258d63c6f70b4e6f7 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Sun, 28 Jun 2026 22:01:55 -0400 Subject: [PATCH] fix(query-engine, planner): change data_range to u64, removed None semantics (was used for OnlySpatial) --- .../rs/asap_types/src/capability_matching.rs | 132 +++++++----------- .../rs/asap_types/src/query_requirements.rs | 9 +- .../src/optimizer/aqe_extractor.rs | 29 ++-- .../src/optimizer/candidate_gen.rs | 20 +-- asap-planner-rs/src/optimizer/cost_model.rs | 8 +- asap-planner-rs/src/optimizer/greedy.rs | 8 +- asap-planner-rs/src/optimizer/pipeline.rs | 2 +- .../src/engines/simple_engine/promql.rs | 1 + .../src/engines/simple_engine/sql.rs | 6 +- 9 files changed, 93 insertions(+), 122 deletions(-) diff --git a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs index a1240b9..5ddfe13 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -62,21 +62,17 @@ fn is_key_agg_type(agg_type: AggregationType) -> bool { /// Window compatibility: can `config` serve a query needing `data_range_ms`? /// -/// - `None` (spatial-only): always compatible. /// - Tumbling: `data_range_ms` must be a positive integer multiple of `window_size_ms`. /// - Sliding: `data_range_ms` must equal `window_size_ms` exactly (a sliding window /// precomputes one fixed range per timestamp; overlapping windows cannot be merged). -pub fn window_compatible(config: &AggregationConfig, data_range_ms: Option) -> bool { - let Some(range) = data_range_ms else { - return true; - }; +pub fn window_compatible(config: &AggregationConfig, data_range_ms: u64) -> bool { let window_ms = config.window_size_ms; - if window_ms == 0 || range == 0 { + if window_ms == 0 || data_range_ms == 0 { return false; } match config.window_type { - WindowType::Sliding => range == window_ms, - WindowType::Tumbling => range % window_ms == 0, + WindowType::Sliding => data_range_ms == window_ms, + WindowType::Tumbling => data_range_ms.is_multiple_of(window_ms), } } @@ -339,7 +335,7 @@ mod tests { fn req( metric: &str, stats: &[Statistic], - data_range_ms: Option, + data_range_ms: u64, grouping: &[&str], spatial_filter: &str, ) -> QueryRequirements { @@ -373,10 +369,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], "")); assert!(result.is_some()); assert_eq!(result.unwrap().aggregation_id_for_value, 1); } @@ -396,11 +390,11 @@ mod tests { // quantile value (0.5 or 0.9) is NOT part of QueryRequirements — both should find the same config let r1 = find_compatible_aggregation( &configs, - &req("lat", &[Statistic::Quantile], Some(300_000), &[], ""), + &req("lat", &[Statistic::Quantile], 300_000, &[], ""), ); let r2 = find_compatible_aggregation( &configs, - &req("lat", &[Statistic::Quantile], Some(300_000), &[], ""), + &req("lat", &[Statistic::Quantile], 300_000, &[], ""), ); assert_eq!(r1.unwrap().aggregation_id_for_value, 2); assert_eq!(r2.unwrap().aggregation_id_for_value, 2); @@ -420,7 +414,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req("lat", &[Statistic::Quantile], Some(300_000), &[], ""), + &req("lat", &[Statistic::Quantile], 300_000, &[], ""), ); assert_eq!(result.unwrap().aggregation_id_for_value, 3); } @@ -437,10 +431,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("mem", &[Statistic::Sum], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("mem", &[Statistic::Sum], 300_000, &[], "")); assert!(result.is_none()); } @@ -456,10 +448,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], "")); assert!(result.is_none()); } @@ -477,10 +467,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], "")); assert!(result.is_some()); } @@ -497,10 +485,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(900_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 900_000, &[], "")); assert!(result.is_some()); } @@ -517,10 +503,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(600_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 600_000, &[], "")); assert!(result.is_none()); } @@ -536,10 +520,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], "")); assert!(result.is_some()); } @@ -556,10 +538,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(600_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 600_000, &[], "")); assert!(result.is_none()); } @@ -575,28 +555,26 @@ mod tests { make_config(2, "cpu", "Sum", "", 900_000, "tumbling", &[], ""), ); // 900_000 ms is divisible by both 300_000 ms and 900_000 ms — prefer 900_000 ms - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(900_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 900_000, &[], "")); assert_eq!(result.unwrap().aggregation_id_for_value, 2); } #[test] fn spatial_only_no_range() { - // data_range_ms = None → any window size is compatible + // spatial-only queries use data_range_ms = scrape interval; window must match let configs = single_config(make_config( 1, "cpu", "Sum", "", - 900_000, + 15_000, "tumbling", &[], "", )); let result = - find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], None, &[], "")); + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 15_000, &[], "")); assert!(result.is_some()); } @@ -616,7 +594,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &["job"], ""), + &req("cpu", &[Statistic::Sum], 300_000, &["job"], ""), ); assert!(result.is_some()); } @@ -636,7 +614,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &["job"], ""), + &req("cpu", &[Statistic::Sum], 300_000, &["job"], ""), ); assert!(result.is_none()); } @@ -655,7 +633,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &["job"], ""), + &req("cpu", &[Statistic::Sum], 300_000, &["job"], ""), ); assert!(result.is_none()); } @@ -674,10 +652,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], "")); assert!(result.is_some()); } @@ -694,10 +670,8 @@ mod tests { &[], "env=prod", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], "")); assert!(result.is_none()); } @@ -715,7 +689,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], "env=prod"), + &req("cpu", &[Statistic::Sum], 300_000, &[], "env=prod"), ); assert!(result.is_some()); } @@ -734,7 +708,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req("cpu", &[Statistic::Sum], Some(300_000), &[], "env=staging"), + &req("cpu", &[Statistic::Sum], 300_000, &[], "env=staging"), ); assert!(result.is_none()); } @@ -753,10 +727,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Min], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Min], 300_000, &[], "")); assert!(result.is_some()); } @@ -773,10 +745,8 @@ mod tests { &[], "", )); - let result = find_compatible_aggregation( - &configs, - &req("cpu", &[Statistic::Max], Some(300_000), &[], ""), - ); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Max], 300_000, &[], "")); assert!(result.is_none()); } @@ -813,7 +783,7 @@ mod tests { ); let result = find_compatible_aggregation( &configs, - &req("req", &[Statistic::Topk], Some(300_000), &[], ""), + &req("req", &[Statistic::Topk], 300_000, &[], ""), ); let info = result.unwrap(); assert_eq!(info.aggregation_id_for_value, 10); @@ -835,7 +805,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req("req", &[Statistic::Topk], Some(300_000), &[], ""), + &req("req", &[Statistic::Topk], 300_000, &[], ""), ); assert!(result.is_none()); } @@ -867,7 +837,7 @@ mod tests { &req( "cpu", &[Statistic::Sum, Statistic::Count], - Some(300_000), + 300_000, &["job"], "", ), @@ -895,13 +865,7 @@ mod tests { )); let result = find_compatible_aggregation( &configs, - &req( - "peers", - &[Statistic::Cardinality], - Some(1_000), - &["srcip"], - "", - ), + &req("peers", &[Statistic::Cardinality], 1_000, &["srcip"], ""), ); let info = result.expect("HLL should serve Cardinality"); assert_eq!(info.aggregation_id_for_value, 42); @@ -953,7 +917,7 @@ mod tests { &req( "cpu", &[Statistic::Sum, Statistic::Count], - Some(300_000), + 300_000, &["job"], "", ), @@ -1003,7 +967,7 @@ mod tests { } fn topk_req(metric: &str, count_events: Option) -> QueryRequirements { - let mut r = req(metric, &[Statistic::Topk], Some(1_000), &[], ""); + let mut r = req(metric, &[Statistic::Topk], 1_000, &[], ""); r.topk_count_events = count_events; r } diff --git a/asap-common/dependencies/rs/asap_types/src/query_requirements.rs b/asap-common/dependencies/rs/asap_types/src/query_requirements.rs index 876ef2c..3dc9e79 100644 --- a/asap-common/dependencies/rs/asap_types/src/query_requirements.rs +++ b/asap-common/dependencies/rs/asap_types/src/query_requirements.rs @@ -20,8 +20,8 @@ pub struct QueryRequirements { /// window_size and grouping_labels. pub statistics: Vec, /// The span of historical data the query reads, in milliseconds. - /// None for spatial-only queries (no time range). - pub data_range_ms: Option, + /// For spatial-only queries (no time range), set to the scrape interval. + pub data_range_ms: u64, /// GROUP BY labels expected in the query result. pub grouping_labels: KeyByLabelNames, /// Normalized label filter (produced by normalize_spatial_filter). @@ -48,6 +48,7 @@ pub fn build_query_requirements_promql( match_result: &PromQLMatchResult, pattern_type: QueryPatternType, metric_schema: &PromQLSchema, + data_ingestion_interval_ms: u64, ) -> Option { let (metric, spatial_filter) = get_metric_and_spatial_filter(match_result); @@ -63,12 +64,12 @@ pub fn build_query_requirements_promql( .ok()?; let data_range_ms = match pattern_type { - QueryPatternType::OnlySpatial => None, + QueryPatternType::OnlySpatial => data_ingestion_interval_ms, // promql-parser supports a literal `ms` duration suffix (e.g. `[500ms]`), // so .num_seconds() would truncate sub-second ranges to 0. _ => match_result .get_range_duration() - .map(|d| d.num_milliseconds() as u64), + .map(|d| d.num_milliseconds() as u64)?, }; let all_labels = metric_schema diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs index 9078ef7..ece963e 100644 --- a/asap-planner-rs/src/optimizer/aqe_extractor.rs +++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs @@ -27,7 +27,7 @@ struct AQEKey { metric: String, /// Statistics are produced in a stable order by get_statistics_to_compute. statistics: Vec, - data_range_ms: Option, + data_range_ms: u64, grouping_labels: KeyByLabelNames, spatial_filter_normalized: String, topk_count_events: Option, @@ -81,13 +81,8 @@ pub fn extract_aqes( let leaves = decompose_to_leaves(&rqe.query_string); for leaf in leaves { - match extract_requirements(&leaf, metric_schema) { - Some(mut req) => { - // Spatial-only queries have no range vector; treat them as a - // single scrape interval so downstream window logic is explicit. - if req.data_range_ms.is_none() { - req.data_range_ms = Some(scrape_interval_ms); - } + match extract_requirements(&leaf, metric_schema, scrape_interval_ms) { + Some(req) => { let key = AQEKey::from_requirements(&req); let entry = acc .entry(key) @@ -167,7 +162,11 @@ fn decompose_to_leaves(query: &str) -> Vec { /// Try to extract `QueryRequirements` from a single leaf PromQL query string. /// Returns `None` if the query cannot be parsed or does not match any pattern. -fn extract_requirements(query: &str, metric_schema: &PromQLSchema) -> Option { +fn extract_requirements( + query: &str, + metric_schema: &PromQLSchema, + data_ingestion_interval_ms: u64, +) -> Option { let ast = promql_parser::parser::parse(query).ok()?; let patterns = build_patterns(); @@ -180,7 +179,13 @@ fn extract_requirements(query: &str, metric_schema: &PromQLSchema) -> Option Vec, min_t: u64) -> AQE { + fn make_aqe(stat: Statistic, range_ms: u64, min_t: u64) -> AQE { use asap_types::query_requirements::QueryRequirements; AQE { requirements: QueryRequirements { @@ -314,7 +314,7 @@ mod tests { #[test] fn always_includes_exact_fallback() { - let aqe = make_aqe(Statistic::Sum, Some(300_000), 60_000); + let aqe = make_aqe(Statistic::Sum, 300_000, 60_000); let candidates = enumerate_candidates(&aqe, 15_000); assert!(candidates .iter() @@ -324,7 +324,7 @@ mod tests { #[test] fn spatial_only_produces_direct_candidates() { // Spatial-only: range = scrape_interval (set by extract_aqes). One Direct window. - let aqe = make_aqe(Statistic::Sum, Some(15_000), 60_000); + let aqe = make_aqe(Statistic::Sum, 15_000, 60_000); let candidates = enumerate_candidates(&aqe, 15_000); for c in candidates.iter().filter(|c| c.config.is_some()) { assert_eq!(c.query_method, QueryMethod::Direct); @@ -334,7 +334,7 @@ mod tests { #[test] fn tumbling_w_equals_range_produces_neither() { // range_a = 60_000ms, scrape = 60_000ms → only W=60_000, n=1 → Direct - let aqe = make_aqe(Statistic::Sum, Some(60_000), 60_000); + let aqe = make_aqe(Statistic::Sum, 60_000, 60_000); let candidates = enumerate_candidates(&aqe, 60_000); for c in candidates.iter().filter(|c| c.config.is_some()) { assert_eq!(c.query_method, QueryMethod::Direct); @@ -345,7 +345,7 @@ mod tests { fn mergeable_sketch_with_multiple_windows_produces_merge() { // Min → MinMax (mergeable, not subtractable): range_a=300_000ms, scrape=60_000ms, min_t=300_000ms // → W=60_000 → n=5, Merge{5}. (Sum would prefer Subtract since it's also subtractable.) - let aqe = make_aqe(Statistic::Min, Some(300_000), 300_000); + let aqe = make_aqe(Statistic::Min, 300_000, 300_000); let candidates = enumerate_candidates(&aqe, 60_000); let merge_candidates: Vec<_> = candidates .iter() @@ -360,7 +360,7 @@ mod tests { #[test] fn cms_with_heap_only_neither_no_merge() { // CMS+Heap is neither mergeable nor subtractable → only n=1 (Direct) valid. - let aqe = make_aqe(Statistic::Topk, Some(300_000), 300_000); + let aqe = make_aqe(Statistic::Topk, 300_000, 300_000); let candidates = enumerate_candidates(&aqe, 60_000); for c in candidates.iter().filter(|c| c.config.is_some()) { assert_eq!( @@ -375,7 +375,7 @@ mod tests { fn partial_width_sliding_candidates_are_generated() { // range_a=600_000ms, min_t=30_000ms, scrape=30_000ms. // W=300_000 (k=2) with S=30_000 should be emitted alongside the full-width W=600_000. - let aqe = make_aqe(Statistic::Min, Some(600_000), 30_000); + let aqe = make_aqe(Statistic::Min, 600_000, 30_000); let candidates = enumerate_candidates(&aqe, 30_000); let partial = candidates.iter().find(|c| { c.config.as_ref().is_some_and(|cfg| { @@ -402,7 +402,7 @@ mod tests { // range_a=600_000ms > min_t_repeat=30_000ms. The old guard (range_a <= min_t_repeat) // incorrectly rejected all Sliding candidates here. New guard: S ≤ min_t_repeat. // W=600_000, S=30_000 ≤ min_t_repeat → full-width Direct Sliding should be present. - let aqe = make_aqe(Statistic::Sum, Some(600_000), 30_000); + let aqe = make_aqe(Statistic::Sum, 600_000, 30_000); let candidates = enumerate_candidates(&aqe, 30_000); assert!( candidates.iter().any(|c| { @@ -418,7 +418,7 @@ mod tests { #[test] fn partial_sliding_subtractable_sketch_gets_subtract() { // Sum → CMS (subtractable): partial Sliding with k=2 should produce Subtract. - let aqe = make_aqe(Statistic::Sum, Some(600_000), 30_000); + let aqe = make_aqe(Statistic::Sum, 600_000, 30_000); let candidates = enumerate_candidates(&aqe, 30_000); assert!( candidates.iter().any(|c| { diff --git a/asap-planner-rs/src/optimizer/cost_model.rs b/asap-planner-rs/src/optimizer/cost_model.rs index 910e060..b43797e 100644 --- a/asap-planner-rs/src/optimizer/cost_model.rs +++ b/asap-planner-rs/src/optimizer/cost_model.rs @@ -167,7 +167,7 @@ mod tests { use promql_utilities::data_model::KeyByLabelNames; use promql_utilities::query_logics::enums::Statistic; - fn make_aqe(stat: Statistic, range_ms: Option, min_t: u64) -> AQE { + fn make_aqe(stat: Statistic, range_ms: u64, min_t: u64) -> AQE { AQE { requirements: QueryRequirements { metric: "test_metric".into(), @@ -193,7 +193,7 @@ mod tests { }; let costs = AtomicCosts::default(); let weights = CostWeights::default(); - let a = make_aqe(Statistic::Sum, Some(300_000), 300_000); + let a = make_aqe(Statistic::Sum, 300_000, 300_000); assert_eq!(ingest_cost(&candidate, 1.0, &costs, &weights), 0.0); assert!(query_cost(&a, &candidate, &costs, &weights) > 0.0); } @@ -202,7 +202,7 @@ mod tests { fn ingest_cost_independent_of_n_windows() { // Retained windows are a transient per-query cost (Mem_query in query_cost), // not a continuous allocation — so ingest_cost must not vary with n_windows. - let a = make_aqe(Statistic::Sum, Some(300_000), 300_000); + let a = make_aqe(Statistic::Sum, 300_000, 300_000); let candidates = enumerate_candidates(&a, 60_000); let template = candidates .iter() @@ -233,7 +233,7 @@ mod tests { // Calibration-independent: Subtract is O(1) (one subtract + one read) while // Merge is O(n) (n-1 merges + one read), so for the same n and same // underlying config, Subtract must cost less regardless of weight tuning. - let a = make_aqe(Statistic::Sum, Some(300_000), 300_000); + let a = make_aqe(Statistic::Sum, 300_000, 300_000); let candidates = enumerate_candidates(&a, 60_000); let template = candidates .iter() diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs index 719c73a..bb98dce 100644 --- a/asap-planner-rs/src/optimizer/greedy.rs +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -94,7 +94,7 @@ mod tests { use promql_utilities::query_logics::enums::Statistic; use std::collections::HashMap as StdHashMap; - fn make_aqe(stat: Statistic, range_ms: Option, min_t: u64, freq_hz: f64) -> AQE { + fn make_aqe(stat: Statistic, range_ms: u64, min_t: u64, freq_hz: f64) -> AQE { AQE { requirements: QueryRequirements { metric: "test_metric".into(), @@ -114,8 +114,8 @@ mod tests { #[test] fn assigns_unique_ids_to_each_deployed_config() { let aqes = vec![ - make_aqe(Statistic::Min, Some(300_000), 300_000, 1.0 / 60.0), - make_aqe(Statistic::Max, Some(300_000), 300_000, 1.0 / 60.0), + make_aqe(Statistic::Min, 300_000, 300_000, 1.0 / 60.0), + make_aqe(Statistic::Max, 300_000, 300_000, 1.0 / 60.0), ]; let solution = greedy_assign( aqes, @@ -141,7 +141,7 @@ mod tests { requirements: QueryRequirements { metric: "test_metric".into(), statistics: vec![Statistic::Sum, Statistic::Count], // avg-style, unsupported - data_range_ms: Some(60_000), + data_range_ms: 60_000, grouping_labels: KeyByLabelNames::empty(), spatial_filter_normalized: String::new(), topk_count_events: None, diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs index 8251812..0954764 100644 --- a/asap-planner-rs/src/optimizer/pipeline.rs +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -156,7 +156,7 @@ mod tests { let rqes = config_to_rqes(&config); let aqes = extract_aqes(&rqes, &PromQLSchema::new(), 15_000); assert_eq!(aqes.len(), 1); - assert_eq!(aqes[0].requirements.data_range_ms, Some(15_000)); + assert_eq!(aqes[0].requirements.data_range_ms, 15_000); } #[test] diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index e8502d8..18c995f 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -1049,6 +1049,7 @@ impl SimpleEngine { &match_result, query_pattern_type, metric_schema, + self.data_ingestion_interval_ms, )?; self.streaming_config .read() diff --git a/asap-query-engine/src/engines/simple_engine/sql.rs b/asap-query-engine/src/engines/simple_engine/sql.rs index fc6b488..fdbff25 100644 --- a/asap-query-engine/src/engines/simple_engine/sql.rs +++ b/asap-query-engine/src/engines/simple_engine/sql.rs @@ -299,10 +299,10 @@ impl SimpleEngine { }; let data_range_ms = match query_pattern_type { - QueryPatternType::OnlySpatial => None, + QueryPatternType::OnlySpatial => self.data_ingestion_interval_ms, QueryPatternType::OnlyTemporal => { let duration_secs = query_data.time_info.clone().get_duration() as u64; - Some(duration_secs * 1000) + duration_secs * 1000 } QueryPatternType::OneTemporalOneSpatial => { let duration_secs = match_result @@ -311,7 +311,7 @@ impl SimpleEngine { .time_info .clone() .get_duration() as u64; - Some(duration_secs * 1000) + duration_secs * 1000 } };