diff --git a/asap-query-engine/src/engines/simple_engine/elastic.rs b/asap-query-engine/src/engines/simple_engine/elastic.rs index fd7f903..c21d1eb 100644 --- a/asap-query-engine/src/engines/simple_engine/elastic.rs +++ b/asap-query-engine/src/engines/simple_engine/elastic.rs @@ -51,8 +51,6 @@ impl SimpleEngine { }) .ok()?; - let do_merge = true; // No "instant" queries in ElasticSearch supported for now, so we always need to merge. - let (metric, query_metadata) = self.build_query_metadata_elastic(&query_info)?; let spatial_filter = String::new(); // Placeholder - extract from query if applicable @@ -60,7 +58,7 @@ impl SimpleEngine { // Parse time range information from first query predicate if available, otherwise default to entire history up to query_time. let timestamps = self.resolve_query_time_range_elastic(query_time, query_info); - let query_plan = self + let (query_plan, do_merge) = self .create_store_query_plan(&metric, ×tamps, &agg_info) .map_err(|e| { warn!("Failed to create store query plan: {}", e); diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 4fe429a..b59e213 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -399,13 +399,15 @@ impl SimpleEngine { }) } - /// Creates a plan for querying the store based on aggregation configuration + /// Creates a plan for querying the store based on aggregation configuration. + /// Also derives `do_merge`: true when the requested time range spans more + /// than one stored window, i.e. `range_ms > window_size_ms`. fn create_store_query_plan( &self, metric: &str, timestamps: &QueryTimestamps, agg_info: &AggregationIdInfo, - ) -> Result { + ) -> Result<(StoreQueryPlan, bool), String> { let sc = self.streaming_config.read().unwrap().clone(); // Get aggregation config for value to determine window type let aggregation_config_for_value = sc @@ -419,6 +421,8 @@ impl SimpleEngine { let window_type = aggregation_config_for_value.window_type; let is_exact_query = window_type == WindowType::Sliding; + let range_ms = timestamps.end_timestamp - timestamps.start_timestamp; + let do_merge = range_ms > aggregation_config_for_value.window_size_ms; // Determine start/end for values query based on window type let (values_start, values_end) = if is_exact_query { @@ -446,10 +450,13 @@ impl SimpleEngine { None }; - Ok(StoreQueryPlan { - values_query, - keys_query, - }) + Ok(( + StoreQueryPlan { + values_query, + keys_query, + }, + do_merge, + )) } /// Executes a single store query based on parameters diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 18c995f..ddf8137 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -303,7 +303,7 @@ impl SimpleEngine { query_kwargs, }; - let query_plan = self + let (query_plan, do_merge) = self .create_store_query_plan(&metric, ×tamps, &agg_info) .map_err(|e| { warn!("Failed to create store query plan: {}", e); @@ -311,9 +311,6 @@ impl SimpleEngine { }) .ok()?; - let do_merge = query_pattern_type == QueryPatternType::OnlyTemporal - || query_pattern_type == QueryPatternType::OneTemporalOneSpatial; - let sc = self.streaming_config.read().unwrap().clone(); let grouping_labels = sc .get_aggregation_config(agg_info.aggregation_id_for_value) diff --git a/asap-query-engine/src/engines/simple_engine/sql.rs b/asap-query-engine/src/engines/simple_engine/sql.rs index fdbff25..ba2c70b 100644 --- a/asap-query-engine/src/engines/simple_engine/sql.rs +++ b/asap-query-engine/src/engines/simple_engine/sql.rs @@ -615,15 +615,11 @@ impl SimpleEngine { String::new() }; - let do_merge = query_pattern_type == QueryPatternType::OnlyTemporal - || query_pattern_type == QueryPatternType::OneTemporalOneSpatial; - let ctx = self.build_sql_execution_context_tail( metric, ×tamps, metadata, agg_info, - do_merge, spatial_filter, query_time, )?; @@ -643,11 +639,10 @@ impl SimpleEngine { timestamps: &QueryTimestamps, metadata: QueryMetadata, agg_info: AggregationIdInfo, - do_merge: bool, spatial_filter: String, query_time: u64, ) -> Option { - let query_plan = self + let (query_plan, do_merge) = self .create_store_query_plan(metric, timestamps, &agg_info) .map_err(|e| { warn!("Failed to create store query plan: {}", e); @@ -763,7 +758,6 @@ impl SimpleEngine { ×tamps, metadata, agg_info, - true, String::new(), query_time, ) diff --git a/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs b/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs index 3908c30..ffad408 100644 --- a/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs +++ b/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs @@ -240,6 +240,9 @@ mod tests { #[tokio::test] async fn test_temporal_context_has_do_merge_true() { + // window_size_ms (1_000) must be smaller than the query's [5s] range so the + // derived do_merge (range_ms > window_size_ms) reflects a realistic config + // where the aggregation's bucket is finer than the requested range. let query = "sum_over_time(http_requests[5s])"; let engine = create_engine_multi_timestamp_with_window( "http_requests", @@ -251,7 +254,7 @@ mod tests { Box::new(SumAccumulator::with_sum(1.0)), )], query, - 5_000, + 1_000, WindowType::Tumbling, ); @@ -404,6 +407,9 @@ mod tests { #[tokio::test] async fn test_collapsable_context_has_do_merge_true() { + // window_size_ms (1_000) must be smaller than the query's [5s] range so the + // derived do_merge (range_ms > window_size_ms) reflects a realistic config + // where the aggregation's bucket is finer than the requested range. let query = "sum by (host) (sum_over_time(http_requests[5s]))"; let engine = create_engine_multi_timestamp_with_window( "http_requests", @@ -415,7 +421,7 @@ mod tests { Box::new(SumAccumulator::with_sum(1.0)), )], query, - 5_000, + 1_000, WindowType::Tumbling, ); @@ -496,6 +502,9 @@ mod tests { }) .collect(); + // window_size_ms (1_000) must be smaller than the query's [5s] range so the + // derived do_merge (range_ms > window_size_ms) reflects a realistic config + // where the aggregation's bucket is finer than the requested range. let query = "sum_over_time(http_requests[5s])"; let engine = create_engine_multi_timestamp_with_window( "http_requests", @@ -503,7 +512,7 @@ mod tests { vec!["host"], data, query, - 5_000, + 1_000, WindowType::Tumbling, ); @@ -533,7 +542,7 @@ mod tests { vec!["host"], data, query, - 5_000, + 1_000, WindowType::Tumbling, ); diff --git a/asap-query-engine/src/tests/query_equivalence_tests.rs b/asap-query-engine/src/tests/query_equivalence_tests.rs index da1141f..bd9cbec 100644 --- a/asap-query-engine/src/tests/query_equivalence_tests.rs +++ b/asap-query-engine/src/tests/query_equivalence_tests.rs @@ -418,4 +418,115 @@ mod tests { "[500ms] range vector must produce a 500ms window, got [{start_ms}, {end_ms})" ); } + + // --- do_merge derivation (issue #486) --- + // + // do_merge is derived in `create_store_query_plan` as `range_ms > window_size_ms`, + // not from QueryPatternType. These three tests are exhaustive over the only + // three patterns that exist, so together they pin down the iff: + // do_merge == true iff pattern is OnlyTemporal or OneTemporalOneSpatial + // do_merge == false iff pattern is OnlySpatial + // Each uses a window_size_ms smaller than the query's range so the derived + // check reflects a realistic config (bucket finer than the requested range), + // matching how OnlySpatial queries are always exactly one bucket wide + // (range_ms == data_ingestion_interval_ms == window_size_ms, so do_merge is + // always false there) while temporal/collapsable queries normally span many. + + #[test] + fn test_do_merge_true_for_temporal_context() { + let scrape_interval_ms = 1000; + let promql_query = "sum_over_time(cpu_usage[5m])"; + let window_size_ms = 1000; // << 5m range + + let (promql_config, _, streaming_config) = TestConfigBuilder::new("cpu_usage") + .with_grouping_labels(vec!["L1"]) + .with_scrape_interval_ms(scrape_interval_ms) + .add_temporal_query( + promql_query, + promql_query, + 1, + window_size_ms, + WindowType::Tumbling, + ) + .build_both(); + + let promql_engine = SimpleEngine::new( + Arc::new(NoOpStore), + promql_config, + streaming_config, + scrape_interval_ms, + QueryLanguage::promql, + ); + + let context = promql_engine + .build_query_execution_context_promql(promql_query.to_string(), 1_000.0) + .expect("Failed to build PromQL context"); + + assert!( + context.do_merge, + "OnlyTemporal queries must have do_merge=true" + ); + } + + #[test] + fn test_do_merge_true_for_collapsable_context() { + let scrape_interval_ms = 1000; + let promql_query = "sum(sum_over_time(cpu_usage[5m])) by (L1)"; + let window_size_ms = 1000; // << 5m range + + let (promql_config, _, streaming_config) = TestConfigBuilder::new("cpu_usage") + .with_grouping_labels(vec!["L1"]) + .with_rollup_labels(vec!["L2", "L3", "L4"]) + .with_scrape_interval_ms(scrape_interval_ms) + .add_spatial_of_temporal_query(promql_query, promql_query, 1, window_size_ms) + .build_both(); + + let promql_engine = SimpleEngine::new( + Arc::new(NoOpStore), + promql_config, + streaming_config, + scrape_interval_ms, + QueryLanguage::promql, + ); + + let context = promql_engine + .build_query_execution_context_promql(promql_query.to_string(), 1_000.0) + .expect("Failed to build PromQL context"); + + assert!( + context.do_merge, + "OneTemporalOneSpatial (collapsable) queries must have do_merge=true" + ); + } + + #[test] + fn test_do_merge_false_for_spatial_context() { + let scrape_interval_ms = 1000; + let promql_query = "sum(cpu_usage) by (L1)"; + + // add_spatial_query sets window_size_ms = scrape_interval_ms, and a spatial + // query's range is exactly one scrape interval, so range_ms == window_size_ms. + let (promql_config, _, streaming_config) = TestConfigBuilder::new("cpu_usage") + .with_grouping_labels(vec!["L1"]) + .with_scrape_interval_ms(scrape_interval_ms) + .add_spatial_query(promql_query, promql_query, 1) + .build_both(); + + let promql_engine = SimpleEngine::new( + Arc::new(NoOpStore), + promql_config, + streaming_config, + scrape_interval_ms, + QueryLanguage::promql, + ); + + let context = promql_engine + .build_query_execution_context_promql(promql_query.to_string(), 1_000.0) + .expect("Failed to build PromQL context"); + + assert!( + !context.do_merge, + "OnlySpatial queries must have do_merge=false" + ); + } }