fix(query-engine): derived do_merge from window sizes, not QueryPatternType enum

asap-query-engine/src/engines/simple_engine/elastic.rs

@@ -51,16 +51,14 @@ 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
 
         // 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, &timestamps, &agg_info)
             .map_err(|e| {
                 warn!("Failed to create store query plan: {}", e);

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<StoreQueryPlan, String> {
+    ) -> 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

asap-query-engine/src/engines/simple_engine/promql.rs

@@ -303,17 +303,14 @@ impl SimpleEngine {
             query_kwargs,
         };
 
-        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);
                 e
             })
             .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)

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,
             &timestamps,
             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<QueryExecutionContext> {
-        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 {
             &timestamps,
             metadata,
             agg_info,
-            true,
             String::new(),
             query_time,
         )

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,14 +502,17 @@ 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",
             AggregationType::Sum,
             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,
         );
 

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"
+        );
+    }
 }