From a7ca91d45a9f6432d0c4455a493a1366384bb905 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Fri, 1 May 2026 12:48:04 -0400 Subject: [PATCH 1/2] fix(store): range_query_into uses overlap filter instead of fully-contained MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PromQL queries against sketch-aggregated metrics returned empty even though the precompute store demonstrably had data — worker logs `Worker emitting 1 sketch outputs for group (1, )`, runtime-info `earliest_timestamp_per_aggregation_id` populates, yet the engine reports `No precomputed outputs found`. ## Root cause `MutableEpoch::range_query_into` and `SealedEpoch::range_query_into` in `simple_map_store/common.rs` (and the on-disk parts variant in `per_key.rs:query_disk_parts`) used a "fully contained" filter: ```rust if tr.0 < start || tr.0 > end || tr.1 > end { continue; // pane must satisfy: start ≤ tr.0 ≤ tr.1 ≤ end } ``` For tumbling windows of size W with a query range of size R, this matches at most floor(R / W) panes — and only when both query endpoints land exactly on the pane grid. PromQL queries don't align: `quantile_over_time(...[1m])` with `prometheus_scrape_interval=30` and a pane size of 30s, the query range `[query_time - 60s, query_time]` is offset from the 30s grid by the wall-clock fractional portion of `query_time`, so neither of the two panes that should match is fully contained — both have `tr.0 < start` (they straddle the lower boundary) or `tr.1 > end` (they straddle the upper boundary). Result: empty. This is not specific to PromQL — any query whose endpoints don't align to the window grid hits the same wall. ## Fix Replace fully-contained with the standard half-open interval overlap test: a window `[tr.0, tr.1)` overlaps the query range `[start, end)` when `tr.1 > start && tr.0 < end`. Skip iff neither (i.e. `tr.1 <= start || tr.0 >= end`). Same change in three places that all share the same dead semantics: - `common.rs::MutableEpoch::range_query_into` (columnar, used by current/in-flight epoch). - `common.rs::SealedEpoch::range_query_into` (sorted-entries variant, used by older sealed epochs). - `per_key.rs::query_disk_parts` (on-disk parts iteration — matches the in-memory contract). Note the `SealedEpoch` variant retained its `partition_point(tr.0 < start)` upper-bound binary search but was rewritten to bound by `tr.0 < end` instead, since entries where `tr.0 < start` *can* now overlap (when `tr.1 > start`). ## Trade-off acknowledged in the comments A pane that crosses the query boundary contributes its sketch state — including data points slightly outside `[start, end)` — to the merged result. For sketch-based aggregations this is the right trade vs. silently returning empty: the alternative is to either align query timestamps to the grid (changing PromQL semantics: a query at 16:43:54 reports data ending at 16:43:30) or to require callers to pre-align their ranges. Both are worse ergonomics than slightly-imprecise sketch values at the edges. ## Verification End-to-end with b3-delta agent (60s agent window, 30s tumbling panes), querying after 3+ window flushes have populated the store: ``` $ curl 'http://localhost:19091/api/v1/query?query=quantile_over_time(0.5, http_requests_total_latency_ms_quantile[1m])&time=$(now-90s)' {"data":{"result":[{"metric":{"node":""}, "value":[1777653931.0, "9.488447485932145"]}], "resultType":"vector"}, "accuracy":{"epsilon":0.01,"kind":"relative_quantile"}} ``` Pre-fix at the same offset: `result: []` with backend log `No precomputed outputs found for metric: ..., aggregation_id: 1`. A query at `now-30s` or `now-0s` still returns empty when the most recent pane is younger than the query end — the agent hasn't flushed it yet (this is timing, not the filter, and is fine for `[1m]` queries that land 60-90s after a window closes). Co-Authored-By: Claude Opus 4.7 (1M context) --- .../sketch_db/simple_map_store/common.rs | 34 +++++++++++++++---- .../sketch_db/simple_map_store/per_key.rs | 11 ++++-- 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/asap-query-engine/src/stores/sketch_db/simple_map_store/common.rs b/asap-query-engine/src/stores/sketch_db/simple_map_store/common.rs index 2ac3ff1a..68109c65 100644 --- a/asap-query-engine/src/stores/sketch_db/simple_map_store/common.rs +++ b/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, ) { + // 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,6 +316,13 @@ 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, @@ -311,12 +330,13 @@ impl SealedEpoch { out: &mut MetricBucketMap, matched_windows: &mut Vec, ) { - 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) diff --git a/asap-query-engine/src/stores/sketch_db/simple_map_store/per_key.rs b/asap-query-engine/src/stores/sketch_db/simple_map_store/per_key.rs index dde41d7e..f10ccf39 100644 --- a/asap-query-engine/src/stores/sketch_db/simple_map_store/per_key.rs +++ b/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) { From 86aafbc9067f791519ce058c25991e737442daff Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Fri, 1 May 2026 13:10:36 -0400 Subject: [PATCH 2/2] fix(engine): pick closest pane for window queries + annotate response MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #71's overlap-filter change made `range_query_into` return every pane that intersected the query range. The merge path then merged 2-3 panes per series, which works for sum/count but gives a slightly wrong answer for sketch summaries — the merged sketch covers more time than the query asked for, and the user has no way to see which time range was actually consulted. Per user request: for a window query, pick the *closest single window* and annotate the response with which one it was. ## Changes ### Pick the closest pane In `execute_and_merge_store_queries`, after the store returns all overlapping panes, compute the global "closest" pane (max `tr.1`, tie-break on max `tr.0` — i.e. the latest pane that overlaps the request range). Then: - Tumbling case: keep only that pane per series, run through the existing merge path (which is a no-op for a single bucket but preserves accumulator-side cleanup). - Sliding case: untouched — was already exact-match per key. The chosen `(start_ms, end_ms)` is bubbled out as a third element of the result tuple. ### Thread `window_used` through QueryResult `InstantVector` and `RangeVector` get a new `window_used: Option<(u64, u64)>` field, paired with `QueryResult::with_window_used` (chainable like `with_accuracy`). `execute_query_pipeline` returns `(elements, window_used)`. `execute_context` attaches the window onto the `QueryResult`. The schema-timeline dispatch path drops the per-segment window because a combined-result spans multiple agg_ids/windows; only the single-agg path surfaces it. ### Annotate the Prometheus HTTP response `PrometheusResponse::with_precompute_window((start, end))` pushes a human-readable line onto the existing `infos` array: ``` precompute_window: [1777655280000, 1777655310000) ms (width 30000 ms) ``` Mirrors the `with_accuracy` pattern — no new top-level field on the wire, just an item in the existing `infos`. Grafana 11+ already renders these inline. ## Verification End-to-end with b3-delta agent (60s window, 30s tumbling pane): ``` $ for offset in -90 -75 -60 -45 -30; do curl …time=$(now$offset)…; done offset=-90s → value=19.493849507395904 precompute_window: [1777655280000, 1777655310000) ms (width 30000 ms) offset=-75s → value=19.493849507395904 precompute_window: [1777655280000, 1777655310000) ms (width 30000 ms) … offset=-30s → value=19.493849507395904 precompute_window: [1777655280000, 1777655310000) ms (width 30000 ms) ``` Same `value` and same `precompute_window` across the offsets — the engine consistently picked the same closest pane ([17:08:00, 17:08:30)) and reported it. Pre-fix, `infos` only carried the `accuracy` line; the caller had to guess which time range produced the value. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../drivers/query/adapters/prometheus_http.rs | 25 ++++ asap-query-engine/src/engines/query_result.rs | 40 +++++++ .../src/engines/simple_engine.rs | 113 ++++++++++++++---- 3 files changed, 153 insertions(+), 25 deletions(-) diff --git a/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs b/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs index 534e1301..dcef608e 100644 --- a/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs +++ b/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()) } diff --git a/asap-query-engine/src/engines/query_result.rs b/asap-query-engine/src/engines/query_result.rs index 3ca4591a..9c89d152 100644 --- a/asap-query-engine/src/engines/query_result.rs +++ b/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, + /// `[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, + /// See [`InstantVector::window_used`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub window_used: Option<(u64, u64)>, } /// Individual element in a range vector diff --git a/asap-query-engine/src/engines/simple_engine.rs b/asap-query-engine/src/engines/simple_engine.rs index 2f8c7a6d..0f5b48e0 100644 --- a/asap-query-engine/src/engines/simple_engine.rs +++ b/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), String> { + ) -> Result<(MergedOutputsMap, Option, 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, String> { + ) -> Result<(Vec, 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,