fix(store + engine): closest-pane window queries + precompute_window annotation

asap-query-engine/src/drivers/query/adapters/prometheus_http.rs

@@ -93,6 +93,23 @@ impl PrometheusResponse {
         self
     }
 
+    /// Attach the actual `[start_ms, end_ms)` precompute window
+    /// the engine consulted to answer this query. Surfaced as a
+    /// `precompute_window: ...` line in the response's `infos`
+    /// array so the caller can see which pane produced the
+    /// answer — important for window queries where the request
+    /// range and the answered range may differ (the engine picks
+    /// the latest closest pane that overlaps the request).
+    pub fn with_precompute_window(mut self, window: (u64, u64)) -> Self {
+        self.infos.push(format!(
+            "precompute_window: [{}, {}) ms (width {} ms)",
+            window.0,
+            window.1,
+            window.1.saturating_sub(window.0),
+        ));
+        self
+    }
+
     pub fn error(error_type: &str, error: &str) -> Self {
         Self {
             status: "error".to_string(),
@@ -265,6 +282,7 @@ impl QueryResponseAdapter for PrometheusHttpAdapter {
         // Prometheus's native API.
         let warnings = result.query_result.warnings().to_vec();
         let accuracy = result.query_result.accuracy().cloned();
+        let window_used = result.query_result.window_used();
         let mut response = if warnings.is_empty() {
             PrometheusResponse::success(prometheus_data)
         } else {
@@ -273,6 +291,9 @@ impl QueryResponseAdapter for PrometheusHttpAdapter {
         if let Some(envelope) = accuracy {
             response = response.with_accuracy(envelope);
         }
+        if let Some(window) = window_used {
+            response = response.with_precompute_window(window);
+        }
         Ok(Json(serde_json::to_value(response).unwrap()).into_response())
     }
 
@@ -289,6 +310,7 @@ impl QueryResponseAdapter for PrometheusHttpAdapter {
         })?;
         let warnings = result.warnings().to_vec();
         let accuracy = result.accuracy().cloned();
+        let window_used = result.window_used();
         let mut response = if warnings.is_empty() {
             PrometheusResponse::success(prometheus_data)
         } else {
@@ -297,6 +319,9 @@ impl QueryResponseAdapter for PrometheusHttpAdapter {
         if let Some(envelope) = accuracy {
             response = response.with_accuracy(envelope);
         }
+        if let Some(window) = window_used {
+            response = response.with_precompute_window(window);
+        }
         Ok(Json(serde_json::to_value(response).unwrap()).into_response())
     }
 

asap-query-engine/src/engines/query_result.rs

@@ -25,6 +25,7 @@ impl QueryResult {
             timestamp,
             warnings: Vec::new(),
             accuracy: None,
+            window_used: None,
         })
     }
 
@@ -45,6 +46,7 @@ impl QueryResult {
             timestamp,
             warnings,
             accuracy: None,
+            window_used: None,
         })
     }
 
@@ -53,6 +55,7 @@ impl QueryResult {
             values,
             warnings: Vec::new(),
             accuracy: None,
+            window_used: None,
         })
     }
 
@@ -88,6 +91,30 @@ impl QueryResult {
         }
         self
     }
+
+    /// Actual `[start_ms, end_ms)` precompute window used to answer
+    /// the query. Set by the engine when a window-style query (e.g.
+    /// `quantile_over_time(...[1m])`) resolved to a single closest
+    /// pane rather than a merge across the request range — the
+    /// caller's request range and the answered range are not the
+    /// same in that case, and the user needs to know which window
+    /// was actually consulted.
+    pub fn window_used(&self) -> Option<(u64, u64)> {
+        match self {
+            QueryResult::Vector(iv) => iv.window_used,
+            QueryResult::Matrix(m) => m.window_used,
+        }
+    }
+
+    /// Attach the actual window range that produced this answer.
+    /// Chainable, mirroring `with_accuracy`.
+    pub fn with_window_used(mut self, window: (u64, u64)) -> Self {
+        match &mut self {
+            QueryResult::Vector(iv) => iv.window_used = Some(window),
+            QueryResult::Matrix(m) => m.window_used = Some(window),
+        }
+        self
+    }
 }
 
 /// Instant vector - a set of time series containing a single sample for each time series, all sharing the same timestamp
@@ -110,6 +137,16 @@ pub struct InstantVector {
     /// Grafana 11+ inline display.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub accuracy: Option<AccuracyEnvelope>,
+    /// `[start_ms, end_ms)` of the precompute window the engine
+    /// actually used to answer this query. Set when a window query
+    /// resolved to a single closest pane (latest pane that overlaps
+    /// the request range) rather than a merge across the full
+    /// request range — the caller's request and the answered range
+    /// differ in that case, and they need to know which window was
+    /// consulted. Surfaced as a `precompute_window` info line in
+    /// `PrometheusResponse.infos` for the Prometheus HTTP adapter.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub window_used: Option<(u64, u64)>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -134,6 +171,9 @@ pub struct RangeVector {
     /// See [`InstantVector::accuracy`].
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub accuracy: Option<AccuracyEnvelope>,
+    /// See [`InstantVector::window_used`].
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub window_used: Option<(u64, u64)>,
 }
 
 /// Individual element in a range vector

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

@@ -876,7 +876,7 @@ impl SimpleEngine {
         plan: &StoreQueryPlan,
         do_merge: bool,
         agg_info: &AggregationIdInfo,
-    ) -> Result<(MergedOutputsMap, Option<MergedOutputsMap>), String> {
+    ) -> Result<(MergedOutputsMap, Option<MergedOutputsMap>, Option<(u64, u64)>), String> {
         // Query and merge values
         let values_map = self.execute_store_query(&plan.values_query).map_err(|e| {
             warn!("Error querying store for values: {}", e);
@@ -899,7 +899,23 @@ impl SimpleEngine {
             WindowType::Tumbling
         };
 
-        let merged_values = if plan.values_query.is_exact_query {
+        // Pick the single CLOSEST precompute window across all keys —
+        // the latest pane (max tr.1, tie-break on max tr.0) that
+        // overlaps the request range. The store's overlap filter may
+        // have returned multiple tumbling panes that straddle the
+        // request, but a window query
+        // (e.g. `quantile_over_time(...[1m])`) should answer with
+        // *one* concrete window so the caller can see exactly which
+        // pane produced the value (annotated downstream as
+        // `precompute_window`). Keys whose data didn't land in that
+        // chosen window are dropped from the result rather than
+        // contributing a stale answer from an older pane.
+        let chosen_window: Option<(u64, u64)> = values_map
+            .values()
+            .flat_map(|buckets| buckets.iter().map(|(tr, _)| *tr))
+            .max_by_key(|tr| (tr.1, tr.0));
+
+        let merged_values: MergedOutputsMap = if plan.values_query.is_exact_query {
             // Sliding window: no merge needed, extract buckets from timestamped data
             debug!("Sliding window mode: Skipping merge (expecting 1 precompute per key)");
             values_map
@@ -917,10 +933,36 @@ impl SimpleEngine {
                 })
                 .collect()
         } else {
-            // Tumbling window: merge needed
-            debug!("Tumbling window mode: Merging {} outputs", values_map.len());
+            // Tumbling window: keep only the chosen-window bucket per
+            // key, then run through the existing merge code (which is
+            // a no-op for a single bucket but preserves whatever
+            // accumulator-side cleanup the merge path does).
+            let target = chosen_window.expect(
+                "values_map non-empty (checked above) but chosen_window was None — \
+                 invariant: if buckets exist, max_by_key returns Some",
+            );
+            let filtered: TimestampedBucketsMap = values_map
+                .into_iter()
+                .filter_map(|(key, buckets)| {
+                    let kept: Vec<_> = buckets
+                        .into_iter()
+                        .filter(|(tr, _)| *tr == target)
+                        .collect();
+                    if kept.is_empty() {
+                        None
+                    } else {
+                        Some((key, kept))
+                    }
+                })
+                .collect();
+            debug!(
+                "Tumbling window mode: closest pane [{}, {}); {} keys present in that pane",
+                target.0,
+                target.1,
+                filtered.len()
+            );
             self.merge_precomputed_outputs(
-                &values_map,
+                &filtered,
                 do_merge,
                 agg_info.aggregation_type_for_value,
             )
@@ -969,7 +1011,7 @@ impl SimpleEngine {
             None
         };
 
-        Ok((merged_values, merged_keys))
+        Ok((merged_values, merged_keys, chosen_window))
     }
 
     /// Collects all results based on whether keys are separate or not
@@ -995,14 +1037,23 @@ impl SimpleEngine {
         }
     }
 
-    /// Executes the complete query pipeline: plan, execute, collect, and format
+    /// Executes the complete query pipeline: plan, execute, collect, and format.
+    ///
+    /// Returns the formatted instant-vector elements alongside the
+    /// `[start_ms, end_ms)` precompute window the engine actually
+    /// consulted (for tumbling-window queries this is the latest
+    /// pane that overlapped the request range; for sliding-window
+    /// queries it's the exact window). Callers attach this onto the
+    /// outgoing `QueryResult` via `with_window_used` so the
+    /// HTTP-adapter response can annotate it as
+    /// `precompute_window`.
     pub fn execute_query_pipeline(
         &self,
         context: &QueryExecutionContext,
         enable_topk: bool,
-    ) -> Result<Vec<InstantVectorElement>, String> {
+    ) -> Result<(Vec<InstantVectorElement>, Option<(u64, u64)>), String> {
         // Step 1: Execute the query plan (already created in context.store_plan)
-        let (merged_values, merged_keys) = self.execute_and_merge_store_queries(
+        let (merged_values, merged_keys, chosen_window) = self.execute_and_merge_store_queries(
             &context.store_plan,
             context.do_merge,
             &context.agg_info,
@@ -1035,7 +1086,7 @@ impl SimpleEngine {
             results_start_time.elapsed().as_millis()
         );
 
-        Ok(results)
+        Ok((results, chosen_window))
     }
 
     /// Execute a query using the plan-based approach (for testing)
@@ -1922,7 +1973,7 @@ impl SimpleEngine {
         enable_topk: bool,
     ) -> Option<(KeyByLabelNames, QueryResult)> {
         let agg_id = context.agg_info.aggregation_id_for_value;
-        let results = self
+        let (results, window_used) = self
             .execute_query_pipeline(&context, enable_topk)
             .map_err(|e| {
                 warn!("Query execution failed: {}", e);
@@ -1934,6 +1985,10 @@ impl SimpleEngine {
             Some(env) => qr.with_accuracy(env),
             None => qr,
         };
+        let qr = match window_used {
+            Some(w) => qr.with_window_used(w),
+            None => qr,
+        };
         Some((context.metadata.query_output_labels, qr))
     }
 
@@ -3145,20 +3200,28 @@ impl SimpleEngine {
                 keys_q.end_timestamp = segment.end_ms;
             }
 
-            let per_segment_results = match self.execute_query_pipeline(&ctx, true) {
-                Ok(v) => v,
-                Err(e) => {
-                    warn!(
-                        agg_id = segment.agg_id,
-                        start_ms = segment.start_ms,
-                        end_ms = segment.end_ms,
-                        "Timeline segment execution failed: {}",
-                        e
-                    );
-                    unresolved.push(segment.clone());
-                    continue;
-                }
-            };
+            let (per_segment_results, _segment_window) =
+                match self.execute_query_pipeline(&ctx, true) {
+                    Ok(v) => v,
+                    Err(e) => {
+                        warn!(
+                            agg_id = segment.agg_id,
+                            start_ms = segment.start_ms,
+                            end_ms = segment.end_ms,
+                            "Timeline segment execution failed: {}",
+                            e
+                        );
+                        unresolved.push(segment.clone());
+                        continue;
+                    }
+                };
+
+            // The per-segment window isn't surfaced in the combined
+            // result for the schema-timeline dispatch path — the
+            // combined answer spans multiple agg_ids/windows by
+            // design, so a single `precompute_window` annotation
+            // would be misleading. The single-agg path
+            // (execute_context above) carries it through normally.
 
             debug!(
                 agg_id = segment.agg_id,

asap-query-engine/src/stores/sketch_db/simple_map_store/common.rs

@@ -194,8 +194,20 @@ impl MutableEpoch {
         out: &mut MetricBucketMap,
         matched_windows: &mut Vec<TimestampRange>,
     ) {
+        // Overlap filter, not fully-contained: include any window whose
+        // [tr.0, tr.1) interval intersects [start, end). The previous
+        // form (`tr.0 < start || tr.0 > end || tr.1 > end → skip`)
+        // required `start ≤ tr.0 ≤ tr.1 ≤ end`, which excluded windows
+        // that crossed the query boundaries — typical for tumbling
+        // windows with a query range that doesn't align to the window
+        // grid (e.g. 60s query range over 30s panes with an unaligned
+        // query end timestamp returns 0 panes instead of the 2 it
+        // should). `quantile_over_time(...[1m])` against a sketch
+        // emitted into a 30s pane otherwise reports
+        // "No precomputed outputs found" even when the data is
+        // demonstrably in the store.
         for (i, &tr) in self.windows_col.iter().enumerate() {
-            if tr.0 < start || tr.0 > end || tr.1 > end {
+            if tr.1 <= start || tr.0 >= end {
                 continue;
             }
             let metric_id = self.metric_ids_col[i];
@@ -304,19 +316,27 @@ impl SealedEpoch {
     }
 
     /// Binary-search start + linear scan — O(log N + actual_matches), cache-friendly.
+    ///
+    /// Overlap filter (not fully-contained): include any window whose
+    /// `[tr.0, tr.1)` interval intersects `[start, end)`. See the
+    /// matching change on the columnar variant above for the longer
+    /// rationale — short version: tumbling windows that cross the
+    /// query boundary should still match, otherwise unaligned query
+    /// ranges silently return no data.
     pub fn range_query_into(
         &self,
         start: u64,
         end: u64,
         out: &mut MetricBucketMap,
         matched_windows: &mut Vec<TimestampRange>,
     ) {
-        let start_pos = self.entries.partition_point(|(tr, _, _)| tr.0 < start);
-        for (tr, metric_id, agg) in &self.entries[start_pos..] {
-            if tr.0 > end {
-                break;
-            }
-            if tr.1 > end {
+        // Entries are sorted by `tr.0`. Bound the upper end with
+        // `tr.0 < end`; entries past that point can't overlap.
+        let end_pos = self.entries.partition_point(|(tr, _, _)| tr.0 < end);
+        for (tr, metric_id, agg) in &self.entries[..end_pos] {
+            // Lower-end overlap check: skip entries that ended at or
+            // before the query start.
+            if tr.1 <= start {
                 continue;
             }
             out.entry(*metric_id)

asap-query-engine/src/stores/sketch_db/simple_map_store/per_key.rs

@@ -693,9 +693,14 @@ impl SimpleMapStorePerKey {
                 if rec.agg_id != aggregation_id {
                     continue;
                 }
-                // Same overlap semantics as MutableEpoch::range_query_into:
-                // window must be fully inside [start, end].
-                if rec.start_ts < start || rec.start_ts > end || rec.end_ts > end {
+                // Overlap semantics matching MutableEpoch::range_query_into:
+                // include any window whose [start_ts, end_ts) interval
+                // intersects [start, end). The earlier "fully inside"
+                // form silently dropped windows that crossed the query
+                // boundaries, which is what tumbling windows do
+                // virtually always when the query timestamp doesn't
+                // align to the window grid.
+                if rec.end_ts <= start || rec.start_ts >= end {
                     continue;
                 }
                 let disk_entry = match reader.load_entry(&rec) {