refactor(query-engine): remove is_exact_query as a threaded bool

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

@@ -58,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, do_merge) = self
+        let (query_plan, do_merge, value_window_type) = self
             .create_store_query_plan(&metric, &timestamps, &agg_info)
             .map_err(|e| {
                 warn!("Failed to create store query plan: {}", e);
@@ -82,6 +82,7 @@ impl SimpleEngine {
             metadata: query_metadata,
             store_plan: query_plan.clone(),
             agg_info: agg_info.clone(),
+            value_window_type,
             do_merge,
             spatial_filter,
             query_time,

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

@@ -51,8 +51,6 @@ pub struct StoreQueryParams {
     pub start_timestamp: u64,
     /// Milliseconds since epoch.
     pub end_timestamp: u64,
-    /// true for sliding windows (exact match), false for tumbling (range)
-    pub is_exact_query: bool,
 }
 
 /// Complete plan for querying store (values + optional separate keys)
@@ -79,6 +77,9 @@ pub struct QueryExecutionContext {
     pub metadata: QueryMetadata,
     pub store_plan: StoreQueryPlan,
     pub agg_info: AggregationIdInfo,
+    /// The value aggregation's WindowType -- Sliding fetches/merges a single
+    /// already-complete window; Tumbling sums the disjoint buckets in range.
+    pub value_window_type: WindowType,
     /// Whether to merge multiple precomputes (true for temporal queries)
     pub do_merge: bool,
     #[allow(dead_code)]
@@ -416,24 +417,33 @@ impl SimpleEngine {
             }
         };
 
+        // Keys always fetch via the window-grid walk (execute_store_query),
+        // never a single exact-window lookup -- this is an explicit,
+        // permanent choice, not a WindowType derivation: a keys query
+        // conceptually always needs to see the key's own bucket(s), not "the
+        // one window ending now."
         Ok(StoreQueryParams {
             metric: metric.to_string(),
             aggregation_id: agg_info.aggregation_id_for_key,
             start_timestamp,
             end_timestamp,
-            is_exact_query: false, // Keys always use range queries
         })
     }
 
     /// 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`.
+    ///
+    /// Returns the value aggregation's `WindowType` alongside the plan --
+    /// callers need it again later (e.g. to pick merge semantics) and it's
+    /// cheaper to hand back what was already looked up here than to
+    /// re-fetch the aggregation config.
     fn create_store_query_plan(
         &self,
         metric: &str,
         timestamps: &QueryTimestamps,
         agg_info: &AggregationIdInfo,
-    ) -> Result<(StoreQueryPlan, bool), String> {
+    ) -> Result<(StoreQueryPlan, bool, WindowType), String> {
         let sc = self.streaming_config.read().unwrap().clone();
         // Get aggregation config for value to determine window type
         let aggregation_config_for_value = sc
@@ -446,13 +456,15 @@ 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 {
-            // Sliding window: exact window match
+        // Determine start/end for values query based on window type. For
+        // Sliding, narrow to exactly the one window ending "now" --
+        // execute_store_query's window-grid walk degenerates to a single
+        // exact lookup when given a range exactly one window wide, so this
+        // narrowing (not a separate flag) is what makes it an "exact" fetch.
+        let (values_start, values_end) = if window_type == WindowType::Sliding {
             let exact_start =
                 timestamps.end_timestamp - aggregation_config_for_value.window_size_ms;
             (exact_start, timestamps.end_timestamp)
@@ -466,7 +478,6 @@ impl SimpleEngine {
             aggregation_id: agg_info.aggregation_id_for_value,
             start_timestamp: values_start,
             end_timestamp: values_end,
-            is_exact_query,
         };
 
         // Determine if we need a separate keys query
@@ -482,6 +493,7 @@ impl SimpleEngine {
                 keys_query,
             },
             do_merge,
+            window_type,
         ))
     }
 
@@ -504,13 +516,14 @@ impl SimpleEngine {
         }
     }
 
-    /// Non-exact store query: walks the aggregation's window grid
-    /// (`bucket_step_ms` apart, each window `window_size_ms` wide, per
-    /// `WindowManager::window_start_for`) and looks up every grid position
-    /// in `[start_timestamp, end_timestamp)` with an exact match, merging
-    /// the sparse per-window results. Used for range queries, key queries,
-    /// and instant queries over tumbling windows — everywhere
-    /// `is_exact_query` is false.
+    /// Walks the aggregation's window grid (`bucket_step_ms` apart, each
+    /// window `window_size_ms` wide, per `WindowManager::window_start_for`)
+    /// and looks up every grid position in `[start_timestamp, end_timestamp)`
+    /// with an exact match, merging the sparse per-window results. A range
+    /// exactly one window wide degenerates to a single exact lookup -- an
+    /// instant Sliding-window fetch gets "the one window ending now" this
+    /// way, by being narrowed to one window's width before calling
+    /// (`create_store_query_plan`), not via a separate exact/scan flag.
     fn scan_windows_via_exact(
         &self,
         params: &StoreQueryParams,
@@ -590,65 +603,20 @@ impl SimpleEngine {
         params: &StoreQueryParams,
     ) -> Result<TimestampedBucketsMap, String> {
         debug!(
-            "Querying store: metric={}, agg_id={}, range=[{}, {}], exact={}",
-            params.metric,
-            params.aggregation_id,
-            params.start_timestamp,
-            params.end_timestamp,
-            params.is_exact_query
+            "Querying store: metric={}, agg_id={}, range=[{}, {}]",
+            params.metric, params.aggregation_id, params.start_timestamp, params.end_timestamp,
         );
 
         let store_query_start_time = Instant::now();
-
-        let result = if params.is_exact_query {
+        let result = self.scan_windows_via_exact(params);
+        if let Ok(ref outputs) = result {
+            let store_query_duration = store_query_start_time.elapsed();
             debug!(
-                "Sliding window query: Looking for exact window [{}, {}]",
-                params.start_timestamp, params.end_timestamp
+                "Window-grid query took: {:.2}ms, found {} unique keys",
+                store_query_duration.as_secs_f64() * 1000.0,
+                outputs.len()
             );
-            let res = self
-                .store
-                .query_precomputed_output_exact(
-                    &params.metric,
-                    params.aggregation_id,
-                    params.start_timestamp,
-                    params.end_timestamp,
-                )
-                .map_err(|e| {
-                    format!(
-                        "Error querying store for metric {}, agg {}, range [{}, {}]: {}",
-                        params.metric,
-                        params.aggregation_id,
-                        params.start_timestamp,
-                        params.end_timestamp,
-                        e
-                    )
-                });
-            if let Ok(ref outputs) = res {
-                let store_query_duration = store_query_start_time.elapsed();
-                debug!(
-                    "Sliding window exact query took: {:.2}ms, found {} unique keys",
-                    store_query_duration.as_secs_f64() * 1000.0,
-                    outputs.len()
-                );
-            }
-            res
-        } else {
-            debug!(
-                "Window-grid query: range [{}, {}]",
-                params.start_timestamp, params.end_timestamp
-            );
-            let res = self.scan_windows_via_exact(params);
-            if let Ok(ref outputs) = res {
-                let store_query_duration = store_query_start_time.elapsed();
-                debug!(
-                    "Window-grid query took: {:.2}ms, found {} unique keys",
-                    store_query_duration.as_secs_f64() * 1000.0,
-                    outputs.len()
-                );
-            }
-            res
-        };
-
+        }
         result
     }
 
@@ -658,6 +626,7 @@ impl SimpleEngine {
         plan: &StoreQueryPlan,
         do_merge: bool,
         agg_info: &AggregationIdInfo,
+        value_window_type: WindowType,
     ) -> Result<(MergedOutputsMap, Option<MergedOutputsMap>), String> {
         // Query and merge values
         let values_map = self.execute_store_query(&plan.values_query).map_err(|e| {
@@ -675,13 +644,8 @@ impl SimpleEngine {
         debug!("Store query returned {} unique keys", values_map.len());
 
         let merge_start_time = Instant::now();
-        let window_type = if plan.values_query.is_exact_query {
-            WindowType::Sliding
-        } else {
-            WindowType::Tumbling
-        };
 
-        let merged_values = if plan.values_query.is_exact_query {
+        let merged_values = if value_window_type == WindowType::Sliding {
             // Sliding window: expected exactly 1 precompute per key today
             // (ponytail: hardcoded, #554 will make >1 legitimate — don't
             // block on it). The store can legitimately return more than
@@ -716,7 +680,7 @@ impl SimpleEngine {
         };
 
         let merge_duration = merge_start_time.elapsed();
-        let did_merge = window_type == WindowType::Sliding
+        let did_merge = value_window_type == WindowType::Sliding
             || do_merge
             || agg_info.aggregation_type_for_value == AggregationType::DeltaSetAggregator;
         debug!(
@@ -897,6 +861,7 @@ impl SimpleEngine {
             &context.store_plan,
             context.do_merge,
             &context.agg_info,
+            context.value_window_type,
         )?;
 
         // Step 2: Collect results

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

@@ -360,7 +360,7 @@ impl SimpleEngine {
             query_kwargs,
         };
 
-        let (query_plan, do_merge) = self
+        let (query_plan, do_merge, value_window_type) = self
             .create_store_query_plan(&metric, &timestamps, &agg_info)
             .map_err(|e| {
                 warn!("Failed to create store query plan: {}", e);
@@ -384,6 +384,7 @@ impl SimpleEngine {
             metadata,
             store_plan: query_plan,
             agg_info,
+            value_window_type,
             do_merge,
             spatial_filter,
             query_time,
@@ -594,10 +595,13 @@ impl SimpleEngine {
             })
             .ok()?;
 
+        // Widening the fetch range to cover the whole step span (rather than
+        // one window's width) is what makes this a window-grid walk instead
+        // of a single exact lookup -- there's no separate flag to set for
+        // that; it falls out of execute_store_query's range-driven behavior.
         let mut extended_store_plan = base_context.store_plan.clone();
         let lookback_ms =
             Self::widen_query_window(&mut extended_store_plan.values_query, start_ms, end_ms);
-        extended_store_plan.values_query.is_exact_query = false;
 
         let buckets_per_step = (step_ms / tumbling_window_ms) as usize;
         let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize;

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

@@ -382,7 +382,7 @@ impl SimpleEngine {
         spatial_filter: String,
         query_time: u64,
     ) -> Option<QueryExecutionContext> {
-        let (query_plan, do_merge) = self
+        let (query_plan, do_merge, value_window_type) = self
             .create_store_query_plan(metric, timestamps, &agg_info)
             .map_err(|e| {
                 warn!("Failed to create store query plan: {}", e);
@@ -406,6 +406,7 @@ impl SimpleEngine {
             metadata,
             store_plan: query_plan,
             agg_info,
+            value_window_type,
             do_merge,
             spatial_filter,
             query_time,

asap-query-engine/src/tests/mod.rs

@@ -15,6 +15,7 @@ pub mod sql_pattern_matching_tests;
 pub mod store_correctness_tests;
 pub mod structural_matching_tests;
 pub mod trait_design_tests;
+pub mod window_semantics_consistency_tests;
 
 #[cfg(test)]
 pub mod test_utilities;

asap-query-engine/src/tests/native_range_query_tests.rs

@@ -8,10 +8,9 @@
 //!    value/key aggregations, values keyed `None`, grouping coming entirely
 //!    from the keys aggregation's `get_keys()`) silently return an empty
 //!    result over a range instead of the expanded key set.
-//! 2. `finish_range_context` (`promql.rs`) unconditionally forces
-//!    `is_exact_query = false`, ignoring the aggregation's real `WindowType`,
-//!    so Sliding-window range queries don't fetch/merge the way the instant
-//!    path does.
+//! 2. The range per-step merge logic didn't distinguish Sliding from
+//!    Tumbling, so Sliding-window range queries didn't fetch/merge the way
+//!    the instant path does (fixed by #608/#621).
 //!
 //! These tests are RED against current code: they mirror instant-query
 //! precedents that already pass (`native_binary_instant_tests.rs`'s
@@ -790,8 +789,8 @@ mod tests {
     async fn range_query_sliding_window_single_bucket_regression() {
         // No-collision counterpart to the merge tests above: a single
         // Sliding bucket per output step must still return its value
-        // unchanged once is_exact_query correctly honors WindowType::Sliding
-        // for range queries. Mirrors
+        // unchanged now that the per-step merge logic correctly honors
+        // WindowType::Sliding for range queries. Mirrors
         // native_pipeline_merge_tests::sliding_single_bucket_returns_its_value.
         let data = vec![(
             1_000_000,

asap-query-engine/src/tests/test_utilities/comparison.rs

@@ -2,7 +2,7 @@
 //!
 //! Provides assertion helpers for deep equality checking of query execution contexts.
 
-use crate::data_model::{AggregationIdInfo, AggregationType};
+use crate::data_model::{AggregationIdInfo, AggregationType, WindowType};
 use crate::engines::simple_engine::{
     QueryExecutionContext, QueryMetadata, StoreQueryParams, StoreQueryPlan,
 };
@@ -30,6 +30,13 @@ pub fn assert_execution_context_equivalent(
         test_name
     );
 
+    // Compare value_window_type
+    assert_eq!(
+        context1.value_window_type, context2.value_window_type,
+        "{}: value_window_type mismatch",
+        test_name
+    );
+
     // Compare metadata
     assert_metadata_equivalent(&context1.metadata, &context2.metadata, test_name);
 
@@ -120,12 +127,6 @@ pub fn assert_store_params_equivalent(
         "{}: End timestamp mismatch - PromQL={}, SQL={}",
         test_name, params1.end_timestamp, params2.end_timestamp
     );
-
-    assert_eq!(
-        params1.is_exact_query, params2.is_exact_query,
-        "{}: Query type mismatch - PromQL={}, SQL={}",
-        test_name, params1.is_exact_query, params2.is_exact_query
-    );
 }
 
 /// Assert that two KeyByLabelNames objects are equivalent
@@ -193,7 +194,6 @@ mod tests {
                     aggregation_id: 1,
                     start_timestamp: 1000,
                     end_timestamp: 2000,
-                    is_exact_query: false,
                 },
                 keys_query: None,
             },
@@ -203,6 +203,7 @@ mod tests {
                 aggregation_type_for_key: AggregationType::Sum,
                 aggregation_type_for_value: AggregationType::Sum,
             },
+            value_window_type: WindowType::Tumbling,
             do_merge: true, // OnlyTemporal queries merge
             spatial_filter: String::new(),
             query_time: 2_000_000, // query timestamp in milliseconds