From 009276269b2c78d8f28f1658f76aea1a525656b9 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 26 Aug 2026 00:43:52 -0400 Subject: [PATCH 1/6] refactor(query-engine): thread output_timestamps through range pipeline (#581 stage E.1) Replaces RangeQueryExecutionContext's start/end/step RangeQueryParams with an explicit output_timestamps: Vec, computed once upstream in finish_range_context instead of re-expanded via a manual while loop inside execute_range_query_pipeline. Zero behavior change -- same timestamp sequence, just threaded as a list instead of three fields re-walked with a mutable loop variable. Stage E prep: this is the shape a future unified instant/range engine takes directly (instant becomes the one-element case). 520/520 tests passing, no new tests needed (pure refactor, existing suite is the regression guard). Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 37 ++++++------------- .../src/engines/simple_engine/promql.rs | 12 +++--- 2 files changed, 18 insertions(+), 31 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 9c63649..4579fc3 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -94,21 +94,18 @@ pub struct QueryExecutionContext { pub aggregated_labels: KeyByLabelNames, } -/// Parameters for a range query -#[derive(Debug, Clone)] -pub struct RangeQueryParams { - pub start: u64, // start timestamp in ms - pub end: u64, // end timestamp in ms - pub step: u64, // step in ms -} - /// Extended execution context for range queries #[derive(Debug, Clone)] pub struct RangeQueryExecutionContext { /// Base context (metric, metadata, store_plan, etc.) pub base: QueryExecutionContext, - /// Range-specific parameters - pub range_params: RangeQueryParams, + /// Every timestamp the per-step loop below produces one output sample + /// for, computed upstream (start..=end stepped by step_ms). Stage E + /// (#581): this is the shape a future unified instant/range engine + /// takes directly -- instant becomes the one-element case of the same + /// list, rather than a start/end/step triple that only ever meant + /// something for range. + pub output_timestamps: Vec, /// Number of buckets per step (step / tumbling_window) pub buckets_per_step: usize, /// Number of buckets in lookback window @@ -1552,9 +1549,6 @@ impl SimpleEngine { let key_accumulator_type = context.base.agg_info.aggregation_type_for_key; // Calculate step parameters - let step_ms = context.range_params.step; - let start_ms = context.range_params.start; - let end_ms = context.range_params.end; let buckets_per_step = context.buckets_per_step; let lookback_bucket_count = context.lookback_bucket_count; let tumbling_window_ms = context.tumbling_window_ms; @@ -1588,11 +1582,11 @@ impl SimpleEngine { "hopping (slide > size)" }; debug!( - "Range query params: start={}, end={}, step_ms={}, tumbling_window_ms={}, \ + "Range query params: {} output timestamp(s) [{}..{}], tumbling_window_ms={}, \ buckets_per_step (slide)={}, lookback_bucket_count (size)={}, mode={}", - start_ms, - end_ms, - step_ms, + context.output_timestamps.len(), + context.output_timestamps.first().copied().unwrap_or(0), + context.output_timestamps.last().copied().unwrap_or(0), tumbling_window_ms, buckets_per_step, lookback_bucket_count, @@ -1722,8 +1716,7 @@ impl SimpleEngine { ); // Iterate by OUTPUT timestamp, not by bucket index - let mut current_time = start_ms; - while current_time <= end_ms { + for ¤t_time in &context.output_timestamps { // #583: dual-population groups resolve their expansion keys // from the keys aggregation, per step — not a single // snapshot reused for every step. If nothing resolves at @@ -1752,7 +1745,6 @@ impl SimpleEngine { "No keys data in window at t={} — skipping this step for this group", current_time ); - current_time += step_ms; continue; } @@ -1762,7 +1754,6 @@ impl SimpleEngine { Ok(merged_keys) => Some(merged_keys), Err(e) => { warn!("Failed to merge keys at t={}: {}", current_time, e); - current_time += step_ms; continue; } } @@ -1788,7 +1779,6 @@ impl SimpleEngine { "Skipping sample at {} - no data in window [{}, {})", current_time, window_start, current_time ); - current_time += step_ms; continue; } @@ -1799,7 +1789,6 @@ impl SimpleEngine { Ok(merged) => merged, Err(e) => { debug!("Failed to get merged result at t={}: {}", current_time, e); - current_time += step_ms; continue; } }; @@ -1834,8 +1823,6 @@ impl SimpleEngine { .or_insert_with(|| RangeVectorElement::new(key)) .add_sample(current_time, value); } - - current_time += step_ms; } } diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index c0bd531..631360c 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -6,7 +6,7 @@ use super::SimpleEngine; use super::{ QueryExecutionContext, QueryMetadata, QueryTimestamps, RangeQueryExecutionContext, - RangeQueryParams, StoreQueryParams, + StoreQueryParams, }; use crate::data_model::{AggregationIdInfo, KeyByLabelValues, QueryConfig, SchemaConfig}; use crate::engines::query_result::{InstantVectorElement, QueryResult, RangeVectorElement}; @@ -649,11 +649,11 @@ impl SimpleEngine { store_plan: extended_store_plan, ..base_context }, - range_params: RangeQueryParams { - start: start_ms, - end: end_ms, - step: step_ms, - }, + // validate_range_query_params (above) already guarantees + // step_ms > 0 and start_ms < end_ms, so this matches the old + // per-step loop's `current_time` sequence exactly: start_ms, + // start_ms+step_ms, ..., the last value <= end_ms. + output_timestamps: (start_ms..=end_ms).step_by(step_ms as usize).collect(), buckets_per_step, lookback_bucket_count, tumbling_window_ms, From 7b195f22d79f996f4c8593d7c35c8d1bb116d34b Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 26 Aug 2026 08:32:00 -0400 Subject: [PATCH 2/6] refactor(query-engine): restructure range pipeline's per-step loop to step-major (#581 stage E.2) execute_range_query_pipeline's fetch/merge loop was group-major (all timestamps for group A, then all timestamps for group B, ...). Restructures it to step-major (all groups for t1, then all groups for t2, ...) -- required for topk correctness (ranking a timestamp's candidates needs every group's value at that timestamp, which a group-major loop can't provide), and the one loop shape topk and non-topk queries now share. Per-group setup (bucket_map, keys_source) still happens exactly once per group, precomputed into a Vec before the step-major loop, not repeated per timestamp. No per-group state carries across timestamps in the existing loop body (every step re-derives its window from bucket_map fresh), so this is a pure reordering: same (group, timestamp, value) triples produced, same per-group sample ordering (chronological, since each group is still visited in ascending timestamp order) -- verified via the full existing suite, including the #629 instant/range equivalence matrix and topk step-major tests, which are exactly what would catch a reordering regression here. 520/520 tests passing, no new tests needed (behavior-preserving refactor). Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 52 +++++++++++++------ 1 file changed, 35 insertions(+), 17 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 4579fc3..8291ac9 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1632,6 +1632,11 @@ impl SimpleEngine { }, } + // Named alias purely to keep the step-major `groups` binding below + // under clippy::type_complexity -- same shape as KeysSource::PerStep's + // own bucket_map field above. + type GroupBucketMap<'a> = HashMap>; + // Resolve, for every value group, which groups exist at all (a // one-time operation — see design doc) and where their expansion // keys come from. A group with keys data but no value data @@ -1702,21 +1707,34 @@ impl SimpleEngine { .collect(), }; - // Process each value group independently - for (timestamped_buckets, keys_source) in groups { - let bucket_map = Self::build_bucket_map(timestamped_buckets); - - debug!( - "Group with {} start-timestamps ({} keys start-timestamps)", - bucket_map.len(), - match &keys_source { - KeysSource::PerStep { bucket_map, .. } => bucket_map.len(), - KeysSource::Fixed(_) => 0, - } - ); + // Precompute per-group setup (bucket_map, keys_source) once, before + // the step-major loop below revisits every group at every output + // timestamp -- doing this per-step instead would repeat identical + // work once per timestamp instead of once per group. + let groups: Vec<(GroupBucketMap, KeysSource)> = groups + .into_iter() + .map(|(timestamped_buckets, keys_source)| { + let bucket_map = Self::build_bucket_map(timestamped_buckets); + debug!( + "Group with {} start-timestamps ({} keys start-timestamps)", + bucket_map.len(), + match &keys_source { + KeysSource::PerStep { bucket_map, .. } => bucket_map.len(), + KeysSource::Fixed(_) => 0, + } + ); + (bucket_map, keys_source) + }) + .collect(); - // Iterate by OUTPUT timestamp, not by bucket index - for ¤t_time in &context.output_timestamps { + // Step-major: for each output timestamp, visit every group, not the + // other way around. Required for topk correctness -- ranking a + // timestamp's candidates means seeing every group's value at that + // timestamp before truncating, which a group-major loop can't do + // (#581). One loop shape for topk and non-topk alike, rather than + // maintaining two. + for ¤t_time in &context.output_timestamps { + for (bucket_map, keys_source) in &groups { // #583: dual-population groups resolve their expansion keys // from the keys aggregation, per step — not a single // snapshot reused for every step. If nothing resolves at @@ -1724,7 +1742,7 @@ impl SimpleEngine { // below (avoids wasted merge work on steps outside the // key's lifetime). Fixed (single-population) groups have no // separate keys accumulator to merge here at all. - let keys_precompute: Option> = match &keys_source { + let keys_precompute: Option> = match keys_source { KeysSource::PerStep { bucket_map: keys_bucket_map, lookback_ms: keys_lookback_ms, @@ -1766,7 +1784,7 @@ impl SimpleEngine { let window_start = current_time.saturating_sub(lookback_ms); let window_buckets = Self::window_buckets_for_step( - &bucket_map, + bucket_map, window_start, current_time, tumbling_window_ms, @@ -1793,7 +1811,7 @@ impl SimpleEngine { } }; - let fallback_key = match &keys_source { + let fallback_key = match keys_source { KeysSource::Fixed(fallback_key) => fallback_key.clone(), KeysSource::PerStep { .. } => None, }; From 9e01ca39451fdb67edde57630dadb64d6fcfbfb8 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 26 Aug 2026 08:39:44 -0400 Subject: [PATCH 3/6] docs(query-engine): fix stale apply_range_topk comment, note step-major memory tradeoff (roborev #32) apply_range_topk's doc comment still said the fetch/merge loop was group-major and that step-major restructuring was "not done here, deliberately" -- both false as of the previous commit (stage E.2). Updates it to describe the loop as it is now, and to explain apply_range_topk still runs as a separate post-pass rather than being folded into that loop (that's stage E.3, not done yet). Also documents the Low finding: building every group's bucket_map eagerly into the `groups` Vec (instead of one group at a time, dropped between groups) raises peak memory for high-cardinality range queries. Inherent to enabling step-major ranking -- no code change, just made the tradeoff explicit where it wasn't before. 520/520 tests passing. Co-Authored-By: Claude Sonnet 5 --- asap-query-engine/src/engines/simple_engine/mod.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 8291ac9..7120f35 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1711,6 +1711,19 @@ impl SimpleEngine { // the step-major loop below revisits every group at every output // timestamp -- doing this per-step instead would repeat identical // work once per timestamp instead of once per group. + // + // Memory tradeoff vs. the old group-major shape (#581 stage E.2 + // review): every group's value bucket_map is now held simultaneously + // for the whole step-major loop's duration, instead of one group's + // bucket_map at a time (built, used, dropped, next group). Keys-side + // PerStep bucket maps were already built eagerly for every group + // beforehand (see `groups` above), so this brings the value side in + // line with that, not a new pattern -- but for a range query over a + // very high-cardinality label set this is a real (if likely modest) + // increase in peak memory. Inherent to step-major: ranking a + // timestamp's candidates needs every group's bucket_map available at + // that timestamp, so they can't be built lazily one group at a time + // anymore. let groups: Vec<(GroupBucketMap, KeysSource)> = groups .into_iter() .map(|(timestamped_buckets, keys_source)| { From 26d07dd789645b3405c48ebd8f2ca6dae2b7660b Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 26 Aug 2026 09:01:54 -0400 Subject: [PATCH 4/6] refactor(query-engine): fold apply_range_topk's ranking into the step-major loop (#581 stage E.3) apply_range_topk ran as a separate pass after the step-major loop finished, re-deriving each timestamp's candidate set (by_timestamp) from the fully assembled results just to rank/truncate for topk. The step-major loop (stage E.2) already visits every group at every output timestamp, so it already has what that re-derivation was reconstructing. Folds the ranking/truncation directly into the loop: each timestamp's (key, value) pairs are collected into step_results, sorted + truncated to k right there when it's a topk query, then inserted into the final result map. apply_range_topk is deleted; formatting (metric-name label prefix) becomes a small tail pass over the final results, since it's a once-per-group operation, not once-per-timestep, and doesn't fit naturally inside the step-major loop the way ranking does. Also fixes tie-break nondeterminism while rewriting this logic: the sort comparator now breaks ties on label value, not just descending value. step_results' order traces back to a HashMap iteration (groups, built from all_data/keys_raw_data) and would otherwise keep a different group at the k-th boundary on every process run. (Same fix separately queued for PR for it there once #629 rebases past this.) 520/520 tests passing (28/28 topk-specific), clippy clean. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 199 ++++++------------ 1 file changed, 62 insertions(+), 137 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 7120f35..591c329 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1502,10 +1502,15 @@ impl SimpleEngine { /// `enable_topk_limiting`/`enable_topk_formatting` mirror /// `execute_query_pipeline`'s flags of the same name (see that method's /// doc comment) -- both no-ops unless - /// `context.base.metadata.statistic_to_compute == Statistic::Topk`. The - /// actual ranking/truncation is delegated to `apply_range_topk` below; - /// see its doc comment for why range's version can't just reuse - /// instant's `format_final_results` truncate-once shape. + /// `context.base.metadata.statistic_to_compute == Statistic::Topk`. + /// Unlike instant's `format_final_results` (sort once, truncate once -- + /// only possible because instant has exactly one value per group), + /// range ranks/truncates per-timestamp ("step-major"): the step-major + /// loop below already visits every group at every output timestamp, so + /// each timestamp's candidates are ranked/truncated inline, right there + /// (#581 stage E.3), before insertion into the final result map. + /// Formatting (metric-name label prefix) is a separate, smaller pass + /// afterward, once per group rather than once per timestep. fn execute_range_query_pipeline( &self, context: &RangeQueryExecutionContext, @@ -1740,6 +1745,23 @@ impl SimpleEngine { }) .collect(); + // Top-k's k, parsed once rather than per timestamp. Some only when + // this is actually a topk query with limiting requested -- gates + // both the per-step sort/truncate below and nothing else, so a + // non-topk query pays zero cost for this. + let topk_k: Option = if enable_topk_limiting + && context.base.metadata.statistic_to_compute == Statistic::Topk + { + context + .base + .metadata + .query_kwargs + .get("k") + .and_then(|s| s.parse::().ok()) + } else { + None + }; + // Step-major: for each output timestamp, visit every group, not the // other way around. Required for topk correctness -- ranking a // timestamp's candidates means seeing every group's value at that @@ -1747,6 +1769,14 @@ impl SimpleEngine { // (#581). One loop shape for topk and non-topk alike, rather than // maintaining two. for ¤t_time in &context.output_timestamps { + // This timestamp's (key, value) pairs across every group, + // collected before insertion into `results` so a topk query can + // rank/truncate them as one step-local set (#581 stage E.3 -- + // folds what used to be a separate apply_range_topk pass, + // re-deriving this same per-timestamp grouping from the + // finished `results` afterward, directly into this loop). + let mut step_results: Vec<(KeyByLabelValues, f64)> = Vec::new(); + for (bucket_map, keys_source) in &groups { // #583: dual-population groups resolve their expansion keys // from the keys aggregation, per step — not a single @@ -1849,152 +1879,47 @@ impl SimpleEngine { // KeyByLabelValues, not Option) -- matches today's // behavior of producing no sample for this combination. let Some(key) = key else { continue }; - results - .entry(key.clone()) - .or_insert_with(|| RangeVectorElement::new(key)) - .add_sample(current_time, value); + step_results.push((key, value)); } } - } - - Ok(self.apply_range_topk( - results, - &context.base.metadata.statistic_to_compute, - &context.base.metadata.query_kwargs, - &context.base.metric, - enable_topk_formatting, - enable_topk_limiting, - )) - } - - /// Applies PromQL top-k semantics to a range query's raw per-group - /// results. No-op unless `statistic == Statistic::Topk` (mirrors - /// `format_final_results`). - /// - /// This is deliberately NOT a straight port of `format_final_results` - /// (sort all groups once by value, then truncate to k): that shape only - /// works because instant queries have exactly one value per group. A - /// range query's `RangeVectorElement` carries many per-timestamp - /// samples, and real PromQL `topk(k, range_vector)` semantics rank - /// independently AT EACH timestamp -- the surviving key set can differ - /// from step to step. So this ranks/truncates per-timestamp - /// ("step-major"), across all groups, as its own pass over the - /// already-assembled results -- rather than restructuring the group-major - /// fetch/merge loop above into a step-major shape. Issue #581's own - /// scoping decided the fetch/merge loop itself becomes step-major only - /// as part of stage E, the full instant/range pipeline collapse (not - /// done here, deliberately -- this is stage-E prep). Doing the ranking - /// as a separate pass gets the same correctness (each timestamp's kept - /// set is decided across all groups, never one group at a time) without - /// front-running that larger, separately-staged restructure. - /// `enable_topk_limiting` and `enable_topk_formatting` are independent - /// flags, mirroring instant's `execute_query_pipeline` contract -- but - /// unlike instant, range has no `(false, true)`-observable case. Instant - /// always sorts Topk results when formatting regardless of limiting, - /// because it returns a flat `Vec` where sort order is part of the - /// output. Range returns a `HashMap` (this function's `results`) whose - /// iteration order was never meaningful, and each surviving group carries - /// many per-timestamp samples rather than one value to sort the outer - /// collection by -- so skipping the ranking block when - /// `enable_topk_limiting` is false has no observable effect here beyond - /// formatting, even though every current call site passes both flags - /// together and never actually exercises `(false, true)`. - fn apply_range_topk( - &self, - mut results: HashMap, - statistic: &Statistic, - query_kwargs: &HashMap, - metric: &str, - enable_topk_formatting: bool, - enable_topk_limiting: bool, - ) -> Vec { - if *statistic != Statistic::Topk { - return results.into_values().collect(); - } - - // Limiting MUST run before formatting: it matches - // `kept_timestamps_by_key`'s keys (read from each element's - // `labels` field) against `results`' own HashMap keys via - // `retain`. Formatting rewrites `elem.labels` (the field) without - // touching the HashMap's outer key, so if formatting ran first the - // two would no longer agree and `retain` would drop every group. - if enable_topk_limiting { - if let Some(k) = query_kwargs.get("k").and_then(|s| s.parse::().ok()) { - use std::collections::HashSet; - - // Index each group once (G clones total) instead of cloning - // its label vector per (group, timestamp) sample -- G*T - // clones otherwise, for G groups over T steps. - let index_keys: Vec = results.keys().cloned().collect(); - let key_to_idx: HashMap = index_keys - .iter() - .cloned() - .enumerate() - .map(|(i, key)| (key, i)) - .collect(); - // Step-major ranking: group every group's samples by - // timestamp first, so each timestamp's top-k decision sees - // every group's value at that timestamp. - let mut by_timestamp: HashMap> = HashMap::new(); - for elem in results.values() { - let idx = key_to_idx[&elem.labels]; - for sample in &elem.samples { - by_timestamp - .entry(sample.timestamp) - .or_default() - .push((idx, sample.value)); - } - } - - let mut kept_timestamps_by_idx: HashMap> = HashMap::new(); - for (timestamp, mut candidates) in by_timestamp { - // Tiebreak on label values: `candidates`'s order comes - // from iterating `results`, a HashMap, whose iteration - // order is randomized per-process -- without this, - // groups tied at the k-th value boundary would keep - // different survivors run to run. - candidates.sort_by(|a, b| { - b.1.partial_cmp(&a.1) - .unwrap_or(std::cmp::Ordering::Equal) - .then_with(|| index_keys[a.0].labels.cmp(&index_keys[b.0].labels)) - }); - candidates.truncate(k); - for (idx, _) in candidates { - kept_timestamps_by_idx - .entry(idx) - .or_default() - .insert(timestamp); - } - } + // Rank this timestamp's candidates across every group and + // truncate to k -- tie-broken by label for determinism, since + // `step_results`' order ultimately traces back to a HashMap + // iteration (`groups`, built from `all_data`/`keys_raw_data`) + // and would otherwise keep a different group on every process + // run when two groups tie at the k-th value. + if let Some(k) = topk_k { + step_results.sort_by(|a, b| { + b.1.partial_cmp(&a.1) + .unwrap_or(std::cmp::Ordering::Equal) + .then_with(|| a.0.labels.cmp(&b.0.labels)) + }); + step_results.truncate(k); + } - results.retain(|key, _| kept_timestamps_by_idx.contains_key(&key_to_idx[key])); - for elem in results.values_mut() { - let idx = key_to_idx[&elem.labels]; - // `keep` is built only from timestamps that already - // appear in this same element's `samples` (see the - // `by_timestamp` loop above), and is non-empty for every - // key that survives the `retain` just above -- so this - // filter can never leave `elem.samples` empty. - let keep = &kept_timestamps_by_idx[&idx]; - elem.samples.retain(|s| keep.contains(&s.timestamp)); - } + for (key, value) in step_results { + results + .entry(key.clone()) + .or_insert_with(|| RangeVectorElement::new(key)) + .add_sample(current_time, value); } } - if enable_topk_formatting { - // Prepend metric name to each key's label values (PromQL shape), - // same rewrite as format_final_results does for instant. Safe to - // mutate `elem.labels` now -- nothing below matches it back - // against the HashMap's outer key. + // Formatting (PromQL topk(...) output shape: prepend the metric + // name to each surviving group's labels) applies once per group, + // not once per timestep -- a separate pass over the final results, + // after every timestamp's ranking above has already decided which + // groups/samples survive. + if enable_topk_formatting && context.base.metadata.statistic_to_compute == Statistic::Topk { for elem in results.values_mut() { - let mut new_labels = vec![metric.to_string()]; + let mut new_labels = vec![context.base.metric.clone()]; new_labels.extend(elem.labels.labels.clone()); elem.labels.labels = new_labels; } } - results.into_values().collect() + Ok(results.into_values().collect()) } } From ce290ac17a33049a325f435ec27355d0ff72d9d4 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 26 Aug 2026 15:37:04 -0400 Subject: [PATCH 5/6] refactor(query-engine): collapse execute_query_pipeline into a thin wrapper around execute_range_query_pipeline (#581 stage E.4) The actual pipeline collapse: execute_query_pipeline now builds a single-timestamp RangeQueryExecutionContext (build_instant_range_context, mirroring finish_range_context but skipping its start --- .../src/engines/simple_engine/mod.rs | 1140 ++++++++++++++++- .../src/engines/simple_engine/promql.rs | 24 +- .../exact_window_grid_adversarial_tests.rs | 65 +- .../src/tests/native_range_query_tests.rs | 6 +- .../src/tests/store_correctness_tests.rs | 2 +- 5 files changed, 1166 insertions(+), 71 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 591c329..dfda39d 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -29,6 +29,7 @@ use promql_utilities::query_logics::enums::{ use serde_json::Value; // Type alias for merged outputs (single aggregate per key after merging) +#[allow(dead_code)] type MergedOutputsMap = HashMap, Box>; /// Metadata extracted from a query, independent of query language @@ -118,7 +119,7 @@ pub struct RangeQueryExecutionContext { /// `worker.rs::merge_panes_for_window`), so a step takes exactly the one /// bucket at `current_time - lookback_ms` (`lookback_ms` == /// `window_size_ms` here); Tumbling buckets are genuinely disjoint, so a - /// step sums every bucket `scan_window` finds across the lookback span + /// step sums every bucket `sum_window` finds across the lookback span /// (#608). pub window_type: WindowType, /// The value aggregation's actual `window_size_ms`, independent of @@ -513,6 +514,123 @@ impl SimpleEngine { } } + /// Widens `query`'s window to `[start_ms - lookback, end_ms]`, where + /// `lookback` is the width `query` already had (`end_timestamp - + /// start_timestamp`) before this call. Re-anchors whatever window an + /// aggregation's instant fetch already computed so it slides across the + /// whole range, without needing to know *why* that window is that width + /// — e.g. it's what lets the same call widen a Tumbling `values_query`, + /// a Sliding-window instant fetch, or (per #583) a `SetAggregator`'s + /// `[end-window_size, end]` keys window or a `DeltaSetAggregator`'s + /// `[0, end]` one, identically. Returns the lookback so callers can + /// derive bucket-count/step metadata from it. + /// + /// Generic (not PromQL-specific) despite `finish_range_context` + /// (promql.rs) being its original/main caller -- lives here so + /// `build_instant_range_context` below can call it too (#581 stage E.4). + fn widen_query_window(query: &mut StoreQueryParams, start_ms: u64, end_ms: u64) -> u64 { + let lookback_ms = query.end_timestamp - query.start_timestamp; + query.start_timestamp = start_ms.saturating_sub(lookback_ms); + query.end_timestamp = end_ms; + lookback_ms + } + + /// Extends an instant `QueryExecutionContext` into a + /// `RangeQueryExecutionContext` with a single output timestamp -- #581 + /// stage E.4's context-builder, so `execute_query_pipeline` can become a + /// thin wrapper around `execute_range_query_pipeline`. Mirrors + /// `finish_range_context` (promql.rs, the range-query equivalent) but + /// for exactly one point instead of a `[start, end]` step range. + /// + /// Deliberately does NOT call `validate_range_query_params`: its + /// start0 / step%window==0 checks are range-query concerns + /// (a real step-through-time query) that don't apply to a single instant + /// point -- there's no "step" here at all, and start==end==query_time is + /// exactly the case that function would reject. + /// + /// `widen_query_window(query, query_time, query_time)` is a + /// mathematical no-op in this single-point case: it re-derives + /// `lookback_ms` from the width `query` already has (computed by + /// `create_store_query_plan`, called upstream to build `base_context`), + /// then resets `start_timestamp = query_time.saturating_sub(lookback_ms)`, + /// `end_timestamp = query_time` -- exactly reproducing the window + /// `create_store_query_plan` already narrowed to, for both Tumbling and + /// Sliding. Verified empirically via the old-vs-new comparison tests + /// (stage_e4_instant_wrapper_equivalence_tests.rs), not just asserted + /// here. + fn build_instant_range_context( + &self, + base_context: QueryExecutionContext, + query_time: u64, + ) -> Option { + let (tumbling_window_ms, window_type, window_size_ms) = { + let sc = self.streaming_config.read().unwrap(); + let config = + sc.get_aggregation_config(base_context.agg_info.aggregation_id_for_value)?; + ( + Self::bucket_step_ms(config), + config.window_type, + config.window_size_ms, + ) + }; + + if tumbling_window_ms == 0 { + warn!("Instant-as-range context: value aggregation window_size_ms is 0"); + return None; + } + + let mut extended_store_plan = base_context.store_plan.clone(); + let lookback_ms = Self::widen_query_window( + &mut extended_store_plan.values_query, + query_time, + query_time, + ); + let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; + + let keys_lookback_ms = extended_store_plan + .keys_query + .as_mut() + .map(|keys_query| Self::widen_query_window(keys_query, query_time, query_time)); + let (keys_tumbling_window_ms, keys_window_type, keys_window_size_ms) = + match keys_lookback_ms { + Some(_) => { + let sc = self.streaming_config.read().unwrap(); + let config = + sc.get_aggregation_config(base_context.agg_info.aggregation_id_for_key)?; + ( + Some(Self::bucket_step_ms(config)), + Some(config.window_type), + Some(config.window_size_ms), + ) + } + None => (None, None, None), + }; + if keys_tumbling_window_ms == Some(0) { + warn!("Instant-as-range context: key aggregation window_size_ms is 0"); + return None; + } + + Some(RangeQueryExecutionContext { + base: QueryExecutionContext { + store_plan: extended_store_plan, + ..base_context + }, + output_timestamps: vec![query_time], + // No real "step" for a single instant point -- 1 is a + // placeholder. Only used for step_overlap_mode's debug-log + // string in execute_range_query_pipeline, never functionally. + buckets_per_step: 1, + lookback_bucket_count, + tumbling_window_ms, + window_type, + window_size_ms, + keys_window_type, + keys_window_size_ms, + keys_lookback_ms, + keys_tumbling_window_ms, + }) + } + /// 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)` @@ -521,7 +639,7 @@ impl SimpleEngine { /// 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( + fn fetch_window_grid_via_exact_lookups( &self, params: &StoreQueryParams, ) -> Result { @@ -603,7 +721,7 @@ impl SimpleEngine { ); let store_query_start_time = Instant::now(); - let result = self.scan_windows_via_exact(params); + let result = self.fetch_window_grid_via_exact_lookups(params); if let Ok(ref outputs) = result { let store_query_duration = store_query_start_time.elapsed(); debug!( @@ -616,6 +734,7 @@ impl SimpleEngine { } /// Executes the full store query plan and returns merged results + #[allow(dead_code)] fn execute_and_merge_store_queries( &self, plan: &StoreQueryPlan, @@ -701,6 +820,7 @@ impl SimpleEngine { /// `execute_store_query` and merges them per output step instead (#583), /// since a single merged snapshot can't answer "what did the key set /// look like at an earlier timestamp." + #[allow(dead_code)] fn fetch_and_merge_keys( &self, keys_query: &Option, @@ -818,6 +938,7 @@ impl SimpleEngine { .collect() } + #[allow(dead_code)] fn collect_all_results( &self, merged_values: &HashMap, Box>, @@ -834,24 +955,21 @@ impl SimpleEngine { } } - /// Executes the complete query pipeline: plan, execute, collect, and format. - /// - /// The two top-k flags are deliberately separate because the two engines - /// need different halves of the behaviour: - /// * `enable_topk_limiting` — enumerate candidate keys from the sketch - /// heap and truncate to `k` during collection. Used by both PromQL and - /// SQL top-k (it's what makes the heap actually drive the result set). - /// * `enable_topk_formatting` — sort by value descending AND prepend the - /// metric name to each key's labels. This is PromQL `topk(...)` output - /// shape only; SQL returns bare `(group-by columns, value)` rows and - /// applies its own ORDER BY / LIMIT, so SQL leaves this `false`. - pub fn execute_query_pipeline( + /// Pre-#581-stage-E.4 instant pipeline, preserved verbatim under + /// `#[cfg(test)]` ONLY as a reference implementation for the old-vs-new + /// comparison tests (stage_e4_instant_wrapper_equivalence_tests.rs) -- + /// not compiled into production, not called by anything else. Delete + /// this (and the comparison tests, and whichever of + /// execute_and_merge_store_queries/fetch_and_merge_keys/collect_all_results/ + /// format_final_results become unreferenced once it's gone) once the + /// wrapper below is confirmed equivalent and the tests are green. + #[cfg(test)] + fn execute_query_pipeline_pre_e4( &self, context: &QueryExecutionContext, enable_topk_limiting: bool, enable_topk_formatting: bool, ) -> Result, String> { - // Step 1: Execute the query plan (already created in context.store_plan) let (merged_values, merged_keys) = self.execute_and_merge_store_queries( &context.store_plan, context.do_merge, @@ -859,29 +977,19 @@ impl SimpleEngine { context.value_window_type, )?; - // Step 2: Collect results - let unformatted_results_start_time = Instant::now(); let unformatted_results = self.collect_all_results( &merged_values, merged_keys.as_ref(), &context.metadata.statistic_to_compute, &context.metadata.query_kwargs, )?; - debug!( - "[LATENCY] Unformatted results collection: {:.2}ms", - unformatted_results_start_time.elapsed().as_secs_f64() * 1000.0 - ); - // Step 3: Format results - let results_start_time = Instant::now(); let mut results = self.format_final_results( unformatted_results, &context.metadata.statistic_to_compute, &context.metric, enable_topk_formatting, ); - // Truncate to k when limiting is active (heap may carry heap_size > k - // candidates; the query only asked for the top k). if enable_topk_limiting { if let Some(k) = context .metadata @@ -892,10 +1000,79 @@ impl SimpleEngine { results.truncate(k); } } - debug!( - "[LATENCY] Results collection: {}ms", - results_start_time.elapsed().as_millis() - ); + + Ok(results) + } + + /// Executes the complete query pipeline: plan, execute, collect, and format. + /// + /// The two top-k flags are deliberately separate because the two engines + /// need different halves of the behaviour: + /// * `enable_topk_limiting` — enumerate candidate keys from the sketch + /// heap and truncate to `k` during collection. Used by both PromQL and + /// SQL top-k (it's what makes the heap actually drive the result set). + /// * `enable_topk_formatting` — sort by value descending AND prepend the + /// metric name to each key's labels. This is PromQL `topk(...)` output + /// shape only; SQL returns bare `(group-by columns, value)` rows and + /// applies its own ORDER BY / LIMIT, so SQL leaves this `false`. + /// + /// #581 stage E.4: a thin wrapper around `execute_range_query_pipeline` + /// -- builds a single-timestamp `RangeQueryExecutionContext` + /// (`build_instant_range_context`) and unwraps the one resulting + /// `RangeVectorElement` per group back into an `InstantVectorElement`. + /// Signature and return type unchanged, per #581's own D2 decision -- + /// SQL/Elastic callers need no changes. + pub fn execute_query_pipeline( + &self, + context: &QueryExecutionContext, + enable_topk_limiting: bool, + enable_topk_formatting: bool, + ) -> Result, String> { + let query_time = context.query_time; + let range_context = self + .build_instant_range_context(context.clone(), query_time) + .ok_or_else(|| { + format!( + "Failed to build instant-as-range context for metric: {}", + context.metric + ) + })?; + + let range_results = self.execute_range_query_pipeline( + &range_context, + enable_topk_limiting, + enable_topk_formatting, + )?; + + let mut results: Vec = range_results + .into_iter() + .map(|elem| { + debug_assert_eq!( + elem.samples.len(), + 1, + "a single-output-timestamp range query must produce exactly one sample per group" + ); + InstantVectorElement::new(elem.labels, elem.samples[0].value) + }) + .collect(); + + // execute_range_query_pipeline's step-major loop ranks/truncates + // correctly, but its final `results.into_values().collect()` + // (mod.rs) comes from a HashMap, whose iteration order does NOT + // preserve that ranking. Instant's own contract (mirrored from the + // old format_final_results, unconditional whenever the statistic is + // Topk, independent of enable_topk_limiting/formatting) is that + // results come back sorted by value descending -- re-sort here to + // restore it. Tie-broken by label for determinism, matching the + // range engine's own topk sort (#581 stage E.3). + if context.metadata.statistic_to_compute == Statistic::Topk { + results.sort_by(|a, b| { + b.value + .partial_cmp(&a.value) + .unwrap_or(std::cmp::Ordering::Equal) + .then_with(|| a.labels.labels.cmp(&b.labels.labels)) + }); + } Ok(results) } @@ -908,6 +1085,7 @@ impl SimpleEngine { /// additionally prepends the metric name to each key's labels — this is the /// PromQL `topk(...)` output shape only; SQL leaves it `false` so rows stay /// as bare `(group-by columns, value)`. + #[allow(dead_code)] fn format_final_results( &self, unformatted_results: HashMap, f64>, @@ -1087,6 +1265,7 @@ impl SimpleEngine { } /// Merge precomputed outputs (extracts buckets from timestamped data) + #[allow(dead_code)] fn merge_precomputed_outputs( &self, precomputed_outputs_map: &TimestampedBucketsMap, @@ -1182,6 +1361,7 @@ impl SimpleEngine { /// Merge multiple accumulators using the merge_with method from AggregateCore trait /// This follows the Python merge_accumulators approach + #[allow(dead_code)] fn merge_accumulators( &self, accumulators: Vec>, @@ -1202,6 +1382,7 @@ impl SimpleEngine { } /// Collects results when key and value use different aggregations + #[allow(dead_code)] fn collect_results_separate_keys( &self, merged_values: &HashMap, Box>, @@ -1235,6 +1416,7 @@ impl SimpleEngine { /// `execute_query_pipeline`) so we must NOT pre-truncate here — the sketch /// heap can hold more than `k` candidates and is not value-sorted, so /// dropping keys now could discard a true top-k member. + #[allow(dead_code)] fn collect_results_same_aggregation( &self, merged_outputs: &HashMap, Box>, @@ -1432,7 +1614,16 @@ impl SimpleEngine { /// identically by `execute_range_query_pipeline` for both the value /// side and (#583) the keys side — the only difference between the two /// call sites is which map/lookback/bucket-width they pass in. - fn scan_window( + /// + /// Cost is proportional to `(window_end - window_start) / step_increment` + /// -- the nominal window width -- not to how many buckets actually exist + /// in `bucket_map`. Fine for Tumbling/Sliding, where that width is + /// bounded by the query itself. Catastrophic for + /// `AggregationType::DeltaSetAggregator`'s keys window, which is always + /// `[0, current_time)` ("replay from the beginning") -- callers on that + /// path MUST use `collect_bucket_map_entries_before` instead, never this + /// (#581 stage E.4 review; see that function's doc for why). + fn sum_window( bucket_map: &HashMap>, window_start: u64, window_end: u64, @@ -1447,7 +1638,7 @@ impl SimpleEngine { // incident, not just a debug-time nicety. assert!( step_increment > 0, - "scan_window: step_increment must be nonzero, or this loop never terminates" + "sum_window: step_increment must be nonzero, or this loop never terminates" ); let mut window_buckets: Vec> = Vec::new(); let mut t = window_start; @@ -1461,7 +1652,7 @@ impl SimpleEngine { } /// Returns whatever bucket(s) `bucket_map` has at exactly - /// `window_start`, or empty if none. Unlike `scan_window`, does not walk + /// `window_start`, or empty if none. Unlike `sum_window`, does not walk /// or sum multiple grid positions: for a Sliding aggregation, the bucket /// at `window_start` is already the complete, correctly-merged answer /// for its window (`worker.rs::merge_panes_for_window` pre-merges before @@ -1479,10 +1670,62 @@ impl SimpleEngine { .unwrap_or_default() } + /// Collects every bucket in `bucket_map` with a start timestamp strictly + /// before `before`, without walking grid positions -- cost proportional + /// to however many buckets actually exist in `bucket_map`, never to a + /// nominal range width. This is `AggregationType::DeltaSetAggregator`'s + /// keys-window composition: its window is always `[0, current_time)` + /// ("replay from the beginning," per `create_keys_query_params` / + /// `widen_query_window`), so `window_start` is always 0 there and + /// filtering `bucket_map`'s own entries by `< before` is exactly that + /// semantics -- while `sum_window`'s position-by-position walk from 0 to + /// `current_time` (which can be ~1e11ms for a real timestamp) is not + /// merely slower, it doesn't complete in any reasonable time. Same fast + /// path `fetch_window_grid_via_exact_lookups` already applies at the + /// fetch layer (tolerant scan instead of a grid walk) -- this is the + /// merge layer's equivalent, needed separately because this function + /// never talks to the store; it only walks whatever + /// `fetch_window_grid_via_exact_lookups` already fetched into + /// `bucket_map`, and that walk was the actual bottleneck (#581 stage E.4 + /// review -- surfaced by instant queries newly routing through this + /// code path, but pre-existing for range queries too, just never + /// exercised by a test wide enough to notice). + /// + /// MUST return buckets in ascending-timestamp order, not `bucket_map`'s + /// own (arbitrary, per-process-randomized) HashMap iteration order: + /// `DeltaSetAggregatorAccumulator::merge_with` is order-sensitive (an + /// oscillating add/remove/add sequence only replays to the correct final + /// membership if applied chronologically, #586) -- `sum_window`'s + /// position-by-position walk gave this for free by construction; this + /// function has to sort for it explicitly instead. Still O(k log k) for + /// k = buckets actually present, not the nominal range width, so the + /// fix stays intact (caught by + /// `range_query_delta_set_aggregator_oscillating_add_remove_across_five_windows` + /// when this was first written without the sort). + fn collect_bucket_map_entries_before( + bucket_map: &HashMap>, + before: u64, + ) -> Vec> { + let mut entries: Vec<(u64, &&dyn AggregateCore)> = bucket_map + .iter() + .filter(|(&t, _)| t < before) + .flat_map(|(&t, buckets)| buckets.iter().map(move |b| (t, b))) + .collect(); + entries.sort_by_key(|(t, _)| *t); + entries + .into_iter() + .map(|(_, b)| b.clone_boxed_core()) + .collect() + } + /// Picks how a step's window is composed from `bucket_map`: Sliding -> - /// `single_window` (one lookup); Tumbling -> `scan_window` + /// `single_window` (one lookup); Tumbling -> `sum_window` /// (scan-and-sum). Used identically by `execute_range_query_pipeline` /// for both the value side and the keys side (#608). + /// + /// NOT used for `AggregationType::DeltaSetAggregator` keys -- callers on + /// that path must call `collect_bucket_map_entries_before` directly + /// instead (see its doc comment). fn window_buckets_for_step( bucket_map: &HashMap>, window_start: u64, @@ -1493,7 +1736,7 @@ impl SimpleEngine { if window_type == WindowType::Sliding { Self::single_window(bucket_map, window_start) } else { - Self::scan_window(bucket_map, window_start, window_end, step_increment) + Self::sum_window(bucket_map, window_start, window_end, step_increment) } } @@ -1792,14 +2035,30 @@ impl SimpleEngine { tumbling_window_ms: keys_tumbling_window_ms, window_type: keys_window_type, } => { - let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); - let keys_window_buckets = Self::window_buckets_for_step( - keys_bucket_map, - keys_window_start, - current_time, - *keys_tumbling_window_ms, - *keys_window_type, - ); + // DeltaSetAggregator's keys window is always + // [0, current_time) ("replay from the beginning"), + // which saturating_sub's keys_window_start to 0 -- + // window_buckets_for_step's sum_window would then + // walk every grid position from 0 to current_time + // (up to ~1e8 positions for a real timestamp) purely + // to see what's in keys_bucket_map, an in-memory map + // already bounded by real data. Bypass that walk + // entirely for this aggregation type (#581 stage + // E.4 review). + let keys_window_buckets = if key_accumulator_type + == AggregationType::DeltaSetAggregator + { + Self::collect_bucket_map_entries_before(keys_bucket_map, current_time) + } else { + let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); + Self::window_buckets_for_step( + keys_bucket_map, + keys_window_start, + current_time, + *keys_tumbling_window_ms, + *keys_window_type, + ) + }; if keys_window_buckets.is_empty() { debug!( @@ -3510,3 +3769,798 @@ mod sketch_query_tests { // assert!(result.is_none()); // } } + +/// Old-vs-new comparison for #581 stage E.4: `execute_query_pipeline_pre_e4` +/// (the pre-E.4 instant implementation, preserved verbatim under +/// `#[cfg(test)]`) vs `execute_query_pipeline` (the new thin wrapper around +/// `execute_range_query_pipeline`). Both run against the exact same +/// `QueryExecutionContext` and must produce identical results -- this is the +/// real safety net for the wrapper swap itself, distinct from +/// `stage_e_instant_range_equivalence_tests.rs` (which compares instant vs +/// range as two *independent* implementations, and stops being a meaningful +/// check for THIS specific change once `execute_query_pipeline` calls into +/// `execute_range_query_pipeline` internally -- at that point both sides of +/// that comparison are the same code). +/// +/// Lives in mod.rs (not `src/tests/`) because it needs direct access to the +/// private `execute_query_pipeline_pre_e4` and `execute_query_pipeline`, +/// matching this file's existing `merge_accumulators_regression_tests_596` +/// convention for the same reason. +#[cfg(test)] +mod stage_e4_instant_wrapper_equivalence_tests { + use crate::data_model::{ + AggregationConfig, AggregationReference, AggregationType, CleanupPolicy, InferenceConfig, + KeyByLabelValues, PrecomputedOutput, PromQLSchema, QueryConfig, QueryLanguage, + SchemaConfig, StreamingConfig, WindowType, + }; + use crate::engines::query_result::{InstantVectorElement, QueryResult}; + use crate::engines::simple_engine::SimpleEngine; + use crate::precompute_operators::sum_accumulator::SumAccumulator; + use crate::precompute_operators::{ + CountMinSketchWithHeapAccumulator, DeltaSetAggregatorAccumulator, SetAggregatorAccumulator, + }; + use crate::stores::simple_map_store::SimpleMapStore; + use crate::stores::Store; + use crate::AggregateCore; + use promql_utilities::data_model::KeyByLabelNames; + use std::collections::HashMap; + use std::sync::Arc; + + enum KeysConfig { + None, + SetAgg, + DeltaSetAgg, + } + + /// Builds an engine with one value aggregation (id=1, Sum, at the given + /// window shape) covering `value_buckets` for group "host-a", and + /// optionally a second keys aggregation (id=2, fixed Tumbling + /// window_size=slide=1000ms with one bucket at [2000,3000)). + #[allow(clippy::too_many_arguments)] + fn build_engine( + window_type: WindowType, + window_size_ms: u64, + slide_interval_ms: u64, + value_buckets: &[(u64, u64, f64)], + keys: KeysConfig, + ) -> SimpleEngine { + let grouping_labels = vec!["host".to_string()]; + let host_a = Some(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + + let mut aggregation_configs = HashMap::new(); + aggregation_configs.insert( + 1u64, + AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_labels.clone()), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms, + slide_interval_ms, + window_type, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "cpu_load".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + + if !matches!(keys, KeysConfig::None) { + let key_agg_type = match keys { + KeysConfig::SetAgg => AggregationType::SetAggregator, + KeysConfig::DeltaSetAgg => AggregationType::DeltaSetAggregator, + KeysConfig::None => unreachable!(), + }; + aggregation_configs.insert( + 2u64, + AggregationConfig { + aggregation_id: 2, + aggregation_type: key_agg_type, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_labels.clone()), + aggregated_labels: KeyByLabelNames::new(vec!["host".to_string()]), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: 1000, + slide_interval_ms: 1000, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "cpu_load".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + } + + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + for (start, end, value) in value_buckets { + let output = PrecomputedOutput::new(*start, *end, host_a.clone(), 1); + store + .insert_precomputed_output(output, Box::new(SumAccumulator::with_sum(*value))) + .unwrap(); + } + + if !matches!(keys, KeysConfig::None) { + let acc: Box = match keys { + KeysConfig::SetAgg => { + let mut a = SetAggregatorAccumulator::new(); + a.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + Box::new(a) + } + KeysConfig::DeltaSetAgg => { + let mut a = DeltaSetAggregatorAccumulator::new(); + a.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + Box::new(a) + } + KeysConfig::None => unreachable!(), + }; + let output = PrecomputedOutput::new(2000, 3000, host_a.clone(), 2); + store.insert_precomputed_output(output, acc).unwrap(); + } + + let promql_schema = PromQLSchema::new().add_metric( + "cpu_load".to_string(), + KeyByLabelNames::new(grouping_labels), + ); + let mut query_config = QueryConfig::new("sum(cpu_load) by (host)".to_string()) + .add_aggregation(AggregationReference::new(1, None)); + if !matches!(keys, KeysConfig::None) { + query_config = query_config.add_aggregation(AggregationReference::new(2, None)); + } + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(promql_schema), + query_configs: vec![query_config], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + + SimpleEngine::new( + store, + inference_config, + streaming_config, + 1000, + QueryLanguage::promql, + ) + } + + fn instant_pairs(elements: Vec) -> Vec<(Vec, f64)> { + let mut pairs: Vec<(Vec, f64)> = elements + .into_iter() + .map(|e| (e.labels.labels, e.value)) + .collect(); + pairs.sort_by(|a, b| a.0.cmp(&b.0)); + pairs + } + + /// Runs `query` at `query_time_s` through both the pre-E.4 and the new + /// wrapper-based `execute_query_pipeline`, against the SAME + /// `QueryExecutionContext`, and asserts they produce the exact same + /// (label, value) set. + fn assert_old_new_match( + engine: &SimpleEngine, + query: &str, + query_time_s: f64, + limiting: bool, + formatting: bool, + case_name: &str, + ) { + let context = engine + .build_query_execution_context_promql(query.to_string(), query_time_s) + .unwrap_or_else(|| panic!("{case_name}: failed to build query execution context")); + + let old_result = engine.execute_query_pipeline_pre_e4(&context, limiting, formatting); + let new_result = engine.execute_query_pipeline(&context, limiting, formatting); + + match (old_result, new_result) { + (Ok(old), Ok(new)) => { + assert_eq!( + instant_pairs(old), + instant_pairs(new), + "{case_name}: old and new instant pipelines diverge" + ); + } + (Err(old_err), Err(new_err)) => { + // Error TEXT is allowed to differ (old/new fetch different + // code paths internally and may phrase the failure + // differently) -- what matters is both sides agree the + // query fails, not fail identically-worded. + eprintln!( + "{case_name}: both sides errored as expected (old: {old_err}, new: {new_err})" + ); + } + (old, new) => panic!( + "{case_name}: old and new disagree on success/failure -- old={old:?}, new={new:?}" + ), + } + } + + // ── Core grid: window type x population x statistic ──────────────── + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_tumbling_multibucket_fallback_key_matches() { + // 3 buckets in range (do_merge=true): [0,1000)=1.0, [1000,2000)=10.0, [2000,3000)=100.0 + let engine = build_engine( + WindowType::Tumbling, + 1000, + 1000, + &[(0, 1000, 1.0), (1000, 2000, 10.0), (2000, 3000, 100.0)], + KeysConfig::None, + ); + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "tumbling_multibucket_fallback_key", + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_tumbling_singlebucket_fallback_key_matches() { + // Exactly 1 bucket in range (do_merge=false): [2000,3000)=100.0 + let engine = build_engine( + WindowType::Tumbling, + 1000, + 1000, + &[(2000, 3000, 100.0)], + KeysConfig::None, + ); + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "tumbling_singlebucket_fallback_key", + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_sliding_fallback_key_matches() { + // Sliding: one already-merged 2000ms-wide window at [1000,3000). + let engine = build_engine( + WindowType::Sliding, + 2000, + 1000, + &[(1000, 3000, 110.0)], + KeysConfig::None, + ); + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "sliding_fallback_key", + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_tumbling_dual_setagg_matches() { + let engine = build_engine( + WindowType::Tumbling, + 1000, + 1000, + &[(2000, 3000, 100.0)], + KeysConfig::SetAgg, + ); + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "tumbling_dual_setagg", + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_sliding_dual_setagg_matches() { + let engine = build_engine( + WindowType::Sliding, + 2000, + 1000, + &[(1000, 3000, 110.0)], + KeysConfig::SetAgg, + ); + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "sliding_dual_setagg", + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_tumbling_dual_deltasetagg_matches() { + let engine = build_engine( + WindowType::Tumbling, + 1000, + 1000, + &[(2000, 3000, 100.0)], + KeysConfig::DeltaSetAgg, + ); + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "tumbling_dual_deltasetagg", + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_sliding_dual_deltasetagg_matches() { + // DeltaSetAgg's own window stays Tumbling (#588/#606) independent of + // the value side's Sliding shape. + let engine = build_engine( + WindowType::Sliding, + 2000, + 1000, + &[(1000, 3000, 110.0)], + KeysConfig::DeltaSetAgg, + ); + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "sliding_dual_deltasetagg", + ); + } + + /// Self-keyed topk: value accumulator's own `get_keys()` (a + /// `CountMinSketchWithHeap`) drives key resolution, not a separate keys + /// aggregation or a fallback group key. + #[tokio::test(flavor = "multi_thread")] + async fn old_new_self_keyed_topk_matches() { + let mut aggregation_configs = HashMap::new(); + aggregation_configs.insert( + 1u64, + AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::CountMinSketchWithHeap, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::empty(), + aggregated_labels: KeyByLabelNames::new(vec!["host".to_string()]), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: 1000, + slide_interval_ms: 1000, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "cpu_load".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32); + for (host, value) in [("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)] { + sketch.inner.update(host, value); + } + let output = PrecomputedOutput::new(2000, 3000, None, 1); + store + .insert_precomputed_output(output, Box::new(sketch)) + .unwrap(); + + let promql_schema = PromQLSchema::new().add_metric( + "cpu_load".to_string(), + KeyByLabelNames::new(vec!["host".to_string()]), + ); + let query = "topk(2, cpu_load)"; + let query_config = + QueryConfig::new(query.to_string()).add_aggregation(AggregationReference::new(1, None)); + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(promql_schema), + query_configs: vec![query_config], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let engine = SimpleEngine::new( + store, + inference_config, + streaming_config, + 1000, + QueryLanguage::promql, + ); + + assert_old_new_match(&engine, query, 3.0, true, true, "self_keyed_topk"); + } + + // ── Edge cases ─────────────────────────────────────────────────────── + // + // No test here for "topk requested over a non-self-keyed accumulator": + // confirmed empirically (probed "topk(2, cpu_load)", + // "topk(2, sum(cpu_load) by (host))", and + // "topk(2, sum_over_time(cpu_load[3s]))" against a Sum-only schema) that + // PromQL capability/pattern matching rejects all three phrasings at + // context-building time, before execute_query_pipeline is ever reached. + // No real query shape reaches execute_query_pipeline with Topk requested + // over non-topk-capable data, so there's nothing for old vs new to + // diverge on -- not a coverage gap, a scenario that doesn't exist. The + // real topk case (self-keyed CountMinSketchWithHeap) is covered above by + // old_new_self_keyed_topk_matches. + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_no_data_returns_err_identically() { + // No precomputed outputs inserted at all. + let engine = build_engine(WindowType::Tumbling, 1000, 1000, &[], KeysConfig::None); + let context = engine + .build_query_execution_context_promql("sum(cpu_load) by (host)".to_string(), 3.0) + .expect("failed to build context"); + let old = engine.execute_query_pipeline_pre_e4(&context, true, false); + let new = engine.execute_query_pipeline(&context, true, false); + assert!(old.is_err(), "old path should fail with no data"); + assert!(new.is_err(), "new path should fail with no data"); + } + + /// host-a has both value and keys data; host-b has keys data but no + /// value data -- #597's warn-and-skip path, not a hard failure. Both + /// old and new must return host-a only. + #[tokio::test(flavor = "multi_thread")] + async fn old_new_keys_data_no_value_data_skips_group() { + let grouping_labels = vec!["host".to_string()]; + let host_a = Some(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + let host_b = Some(KeyByLabelValues { + labels: vec!["host-b".to_string()], + }); + + let mut aggregation_configs = HashMap::new(); + aggregation_configs.insert( + 1u64, + AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_labels.clone()), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: 1000, + slide_interval_ms: 1000, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "cpu_load".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + aggregation_configs.insert( + 2u64, + AggregationConfig { + aggregation_id: 2, + aggregation_type: AggregationType::SetAggregator, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_labels.clone()), + aggregated_labels: KeyByLabelNames::new(vec!["host".to_string()]), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: 1000, + slide_interval_ms: 1000, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "cpu_load".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + // host-a: both value and keys. + store + .insert_precomputed_output( + PrecomputedOutput::new(2000, 3000, host_a.clone(), 1), + Box::new(SumAccumulator::with_sum(100.0)), + ) + .unwrap(); + let mut keys_a = SetAggregatorAccumulator::new(); + keys_a.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + store + .insert_precomputed_output( + PrecomputedOutput::new(2000, 3000, host_a.clone(), 2), + Box::new(keys_a), + ) + .unwrap(); + + // host-b: keys only, no value data. + let mut keys_b = SetAggregatorAccumulator::new(); + keys_b.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string()], + }); + store + .insert_precomputed_output( + PrecomputedOutput::new(2000, 3000, host_b.clone(), 2), + Box::new(keys_b), + ) + .unwrap(); + + let promql_schema = PromQLSchema::new().add_metric( + "cpu_load".to_string(), + KeyByLabelNames::new(grouping_labels), + ); + let query_config = QueryConfig::new("sum(cpu_load) by (host)".to_string()) + .add_aggregation(AggregationReference::new(1, None)) + .add_aggregation(AggregationReference::new(2, None)); + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(promql_schema), + query_configs: vec![query_config], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let engine = SimpleEngine::new( + store, + inference_config, + streaming_config, + 1000, + QueryLanguage::promql, + ); + + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "keys_data_no_value_data_skips_group", + ); + } + + /// Several distinct groups in one query -- sanity-checks the step-major + /// loop (folded in at #581 stage E.2/E.3) holds up under real + /// cardinality when reached via the single-timestamp wrapper. + #[tokio::test(flavor = "multi_thread")] + async fn old_new_multi_group_matches() { + let grouping_labels = vec!["host".to_string()]; + let mut aggregation_configs = HashMap::new(); + aggregation_configs.insert( + 1u64, + AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_labels.clone()), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: 1000, + slide_interval_ms: 1000, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "cpu_load".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + for (host, value) in [ + ("host-a", 100.0), + ("host-b", 50.0), + ("host-c", 10.0), + ("host-d", 5.0), + ] { + let key = Some(KeyByLabelValues { + labels: vec![host.to_string()], + }); + store + .insert_precomputed_output( + PrecomputedOutput::new(2000, 3000, key, 1), + Box::new(SumAccumulator::with_sum(value)), + ) + .unwrap(); + } + let promql_schema = PromQLSchema::new().add_metric( + "cpu_load".to_string(), + KeyByLabelNames::new(grouping_labels), + ); + let query_config = QueryConfig::new("sum(cpu_load) by (host)".to_string()) + .add_aggregation(AggregationReference::new(1, None)); + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(promql_schema), + query_configs: vec![query_config], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let engine = SimpleEngine::new( + store, + inference_config, + streaming_config, + 1000, + QueryLanguage::promql, + ); + + assert_old_new_match( + &engine, + "sum(cpu_load) by (host)", + 3.0, + true, + false, + "multi_group", + ); + } + + // ── Binary-expr composition (expected-value, not old-vs-new) ──────── + // + // handle_binary_expr_promql itself isn't touched by E.4 -- only what + // execute_query_pipeline does internally for each arm's leaf context + // changes, and that's already covered by the old-vs-new tests above (a + // binary-expr arm's leaf context has the same shape as a plain query's). + // So these are expected-value regression tests confirming composition + // still works end-to-end through the new wrapper, not a second + // old-vs-new comparison. + + fn matrix_metric(qr: QueryResult) -> f64 { + match qr { + QueryResult::Vector(v) => { + assert_eq!(v.values.len(), 1, "expected exactly one series"); + v.values[0].value + } + QueryResult::Matrix(_) => panic!("expected an instant Vector, got a Matrix"), + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_binary_expr_scalar_arm_matches() { + let engine = build_engine( + WindowType::Tumbling, + 1000, + 1000, + &[(2000, 3000, 100.0)], + KeysConfig::None, + ); + let (_, qr) = engine + .handle_query_promql("sum(cpu_load) by (host) * 5".to_string(), 3.0) + .expect("scalar binary-expr query should resolve"); + assert_eq!(matrix_metric(qr), 500.0); + } + + #[tokio::test(flavor = "multi_thread")] + async fn old_new_binary_expr_vector_vector_matches() { + let grouping_labels = vec!["host".to_string()]; + let host_a = Some(KeyByLabelValues { + labels: vec!["host-a".to_string()], + }); + let mut aggregation_configs = HashMap::new(); + for (id, metric) in [(1u64, "metric_a"), (2u64, "metric_b")] { + aggregation_configs.insert( + id, + AggregationConfig { + aggregation_id: id, + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(grouping_labels.clone()), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: 1000, + slide_interval_ms: 1000, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: metric.to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + } + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + store + .insert_precomputed_output( + PrecomputedOutput::new(2000, 3000, host_a.clone(), 1), + Box::new(SumAccumulator::with_sum(10.0)), + ) + .unwrap(); + store + .insert_precomputed_output( + PrecomputedOutput::new(2000, 3000, host_a.clone(), 2), + Box::new(SumAccumulator::with_sum(20.0)), + ) + .unwrap(); + let promql_schema = PromQLSchema::new() + .add_metric( + "metric_a".to_string(), + KeyByLabelNames::new(grouping_labels.clone()), + ) + .add_metric( + "metric_b".to_string(), + KeyByLabelNames::new(grouping_labels), + ); + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(promql_schema), + query_configs: vec![ + QueryConfig::new("sum(metric_a) by (host)".to_string()) + .add_aggregation(AggregationReference::new(1, None)), + QueryConfig::new("sum(metric_b) by (host)".to_string()) + .add_aggregation(AggregationReference::new(2, None)), + ], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + let engine = SimpleEngine::new( + store, + inference_config, + streaming_config, + 1000, + QueryLanguage::promql, + ); + + let (_, qr) = engine + .handle_query_promql( + "sum(metric_a) by (host) + sum(metric_b) by (host)".to_string(), + 3.0, + ) + .expect("vector-vector binary-expr query should resolve"); + assert_eq!(matrix_metric(qr), 30.0); + } +} diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index 631360c..73f0094 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -4,10 +4,7 @@ //! dispatch, range-query handling, and query dispatch. use super::SimpleEngine; -use super::{ - QueryExecutionContext, QueryMetadata, QueryTimestamps, RangeQueryExecutionContext, - StoreQueryParams, -}; +use super::{QueryExecutionContext, QueryMetadata, QueryTimestamps, RangeQueryExecutionContext}; use crate::data_model::{AggregationIdInfo, KeyByLabelValues, QueryConfig, SchemaConfig}; use crate::engines::query_result::{InstantVectorElement, QueryResult, RangeVectorElement}; use asap_types::query_requirements::build_query_requirements_promql; @@ -541,23 +538,6 @@ impl SimpleEngine { } } - /// Widens `query`'s window to `[start_ms - lookback, end_ms]`, where - /// `lookback` is the width `query` already had (`end_timestamp - - /// start_timestamp`) before this call. Re-anchors whatever window an - /// aggregation's instant fetch already computed so it slides across the - /// whole range, without needing to know *why* that window is that width - /// — e.g. it's what lets the same call widen a Tumbling `values_query`, - /// a Sliding-window instant fetch, or (per #583) a `SetAggregator`'s - /// `[end-window_size, end]` keys window or a `DeltaSetAggregator`'s - /// `[0, end]` one, identically. Returns the lookback so callers can - /// derive bucket-count/step metadata from it. - fn widen_query_window(query: &mut StoreQueryParams, start_ms: u64, end_ms: u64) -> u64 { - let lookback_ms = query.end_timestamp - query.start_timestamp; - query.start_timestamp = start_ms.saturating_sub(lookback_ms); - query.end_timestamp = end_ms; - lookback_ms - } - /// Extends an instant `QueryExecutionContext` into a `RangeQueryExecutionContext`: /// computes the lookback window from the aggregation's tumbling window size, /// validates the range params, and widens the store plan to cover @@ -634,7 +614,7 @@ impl SimpleEngine { None => (None, None, None), }; // A zero window_size_ms would make execute_range_query_pipeline's - // per-step scan_window (`while t < window_end { ...; t += step_increment }`) + // per-step sum_window (`while t < window_end { ...; t += step_increment }`) // loop forever, since t would never advance. The value side is // accidentally protected from this by validate_range_query_params's // `step.is_multiple_of(tumbling_window_ms)` check (only 0 is a diff --git a/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs b/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs index a2bc49c..c16539d 100644 --- a/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs +++ b/asap-query-engine/src/tests/exact_window_grid_adversarial_tests.rs @@ -509,7 +509,7 @@ mod tests { /// `create_engine_multi_timestamp_with_window`, mirroring /// `worker.rs::merge_panes_for_window`'s real output shape) -- each /// bucket is already a complete answer for its own window. The range - /// pipeline's per-step `scan_window` doesn't know that: it walks every + /// pipeline's per-step `sum_window` doesn't know that: it walks every /// grid position in `[current_time - window_size_ms, current_time)` and /// sums whatever it finds there, which is correct for genuinely disjoint /// Tumbling buckets but over-counts for Sliding, where every position in @@ -632,7 +632,7 @@ mod tests { /// Step 4000: window [2000,4000) -> panes {2000,3000} -> 100+1000=1100 /// /// Same root cause as the test above (#608): each grid position holds a - /// complete, already-merged window, and scan_window sums every position + /// complete, already-merged window, and sum_window sums every position /// in the lookback span instead of taking the single one at /// `current_time - window_size_ms`. #[tokio::test(flavor = "multi_thread")] @@ -908,4 +908,65 @@ mod tests { scan would be expected to blow well past this on a range this wide" ); } + + /// Range-query counterpart to + /// `instant_query_delta_set_keys_wide_range_from_zero_completes_quickly_and_correctly`. + /// `finish_range_context` widens the keys window using the same + /// `create_keys_query_params` `[0, end_timestamp]` span + /// `build_instant_range_context` does for instant, so this bug is + /// pre-existing in `execute_range_query_pipeline` -- not something #581 + /// stage E.4 introduces, just never caught because no test exercised a + /// genuine range query against a DeltaSetAggregator keys span this wide + /// before. Confirmed empirically before writing this: an unfixed run + /// doesn't even complete within 30s (vs. the instant path's ~13.7s) -- + /// bounded here at 5s, same bound as the instant counterpart. + #[tokio::test(flavor = "multi_thread")] + async fn range_query_delta_set_keys_wide_range_from_zero_completes_quickly_and_correctly() { + let base_ts: u64 = 100_000_000_000; // 1e11 ms + let cms = CountMinSketchAccumulator::new(2, 3); + let mut keys = DeltaSetAggregatorAccumulator::new(); + keys.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + + let engine = make_dual_engine_at_timestamp( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + base_ts, + vec![(None, Box::new(cms) as Box)], + vec![(None, Box::new(keys) as Box)], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let t = base_ts as f64 / 1000.0; + + let call_start = Instant::now(); + // Single-step range query (start == end - step, step = 1s) -- + // exercises execute_range_query_pipeline's per-step keys merge + // exactly once, same shape as the instant case, through the range + // entry point instead. + let result = engine.handle_range_query_promql(query.to_string(), t, t + 1.0, 1.0); + let elapsed = call_start.elapsed(); + + let (_, qr) = + result.expect("range query failed to resolve real data near a huge timestamp"); + let elements = matrix_values(qr); + assert!( + elements + .iter() + .any(|e| e.labels.labels.contains(&"host-a".to_string())), + "expected host-a's key to be resolved via the DeltaSetAggregator keys query \ + spanning [0, {base_ts}], got {elements:?}" + ); + assert!( + elapsed < Duration::from_secs(5), + "range query with a DeltaSetAggregator keys span of [0, {base_ts}] (~1e11ms) \ + took {elapsed:?} -- a per-grid-position enumeration substitute for the tolerant \ + scan would be expected to blow well past this on a range this wide" + ); + } } diff --git a/asap-query-engine/src/tests/native_range_query_tests.rs b/asap-query-engine/src/tests/native_range_query_tests.rs index 88bfdcf..d5bad02 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -383,7 +383,7 @@ mod tests { #[tokio::test(flavor = "multi_thread")] async fn range_query_sliding_keys_bucket_found_on_slide_interval_grid() { - // #600: the keys-side scan_window must step by the KEY aggregation's + // #600: the keys-side sum_window must step by the KEY aggregation's // own slide_interval_ms, not its window_size_ms. Key aggregation: // window_size_ms=2000, slide_interval_ms=1000 (Sliding) -- real // buckets land on the 1000ms grid (start=3000), which isn't on the @@ -450,7 +450,7 @@ mod tests { key_has_sample_at(&elements, "host-a", 5000), "BUG #600: host-a's keys delta bucket (start=3000, on the \ slide_interval_ms=1000 grid but not the window_size_ms=2000 \ - grid) was not found -- the keys-side scan_window is stepping \ + grid) was not found -- the keys-side sum_window is stepping \ by window_size_ms instead of slide_interval_ms" ); } @@ -828,7 +828,7 @@ mod tests { // window_size_ms=2000, but create_engine_multi_timestamp_with_window // fixes the bucket span at slide_interval_ms=1000, so the two // buckets below land at start=0 and start=1000 -- only one of which - // is on the window_size_ms=2000 grid ({0, 2000, ...}). A scan_window + // is on the window_size_ms=2000 grid ({0, 2000, ...}). A sum_window // that steps by window_size_ms instead of slide_interval_ms never // visits start=1000 and silently drops that bucket from the merge. let data = vec![ diff --git a/asap-query-engine/src/tests/store_correctness_tests.rs b/asap-query-engine/src/tests/store_correctness_tests.rs index 1636f2f..124fd73 100644 --- a/asap-query-engine/src/tests/store_correctness_tests.rs +++ b/asap-query-engine/src/tests/store_correctness_tests.rs @@ -630,7 +630,7 @@ fn test_batch_exact_query_equivalent_to_sequential_calls(strategy: LockStrategy) .unwrap(); // Windows include a miss (4_000, 5_000) between two hits, mirroring a real - // scan_windows_via_exact grid walk over a range with a gap. + // fetch_window_grid_via_exact_lookups grid walk over a range with a gap. let windows = [ (1_000, 2_000), (2_000, 3_000), From 0689a7c30972c1092c07f138d3e37a5824167a1f Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 26 Aug 2026 16:41:48 -0400 Subject: [PATCH 6/6] fix(query-engine): address stage-E.4 review findings DeltaSetAgg fast-path assert, shared topk comparator, alias reuse, comment fixes. 536/536 passing. Co-Authored-By: Claude Sonnet 5 --- .../src/engines/simple_engine/mod.rs | 80 ++++++++++++++----- 1 file changed, 62 insertions(+), 18 deletions(-) diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index dfda39d..99a1856 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -616,9 +616,9 @@ impl SimpleEngine { ..base_context }, output_timestamps: vec![query_time], - // No real "step" for a single instant point -- 1 is a - // placeholder. Only used for step_overlap_mode's debug-log - // string in execute_range_query_pipeline, never functionally. + // Placeholder: no real "step" for a single instant point. Only + // feeds a debug-log string today -- not type-enforced, recheck + // before using it for anything functional. buckets_per_step: 1, lookback_bucket_count, tumbling_window_ms, @@ -1004,6 +1004,24 @@ impl SimpleEngine { Ok(results) } + /// Topk ranking comparator shared by `execute_query_pipeline`'s + /// post-wrapper re-sort and `execute_range_query_pipeline`'s per-step + /// sort: descending by value, ties broken by ascending label for + /// determinism (both call sites' inputs ultimately trace back to a + /// HashMap iteration order, which is randomized per-process -- #581 + /// stage E.3/E.4 review). + fn cmp_topk_value_desc( + a_value: f64, + a_labels: &[String], + b_value: f64, + b_labels: &[String], + ) -> std::cmp::Ordering { + b_value + .partial_cmp(&a_value) + .unwrap_or(std::cmp::Ordering::Equal) + .then_with(|| a_labels.cmp(b_labels)) + } + /// Executes the complete query pipeline: plan, execute, collect, and format. /// /// The two top-k flags are deliberately separate because the two engines @@ -1067,10 +1085,7 @@ impl SimpleEngine { // range engine's own topk sort (#581 stage E.3). if context.metadata.statistic_to_compute == Statistic::Topk { results.sort_by(|a, b| { - b.value - .partial_cmp(&a.value) - .unwrap_or(std::cmp::Ordering::Equal) - .then_with(|| a.labels.labels.cmp(&b.labels.labels)) + Self::cmp_topk_value_desc(a.value, &a.labels.labels, b.value, &b.labels.labels) }); } @@ -1702,6 +1717,18 @@ impl SimpleEngine { /// fix stays intact (caught by /// `range_query_delta_set_aggregator_oscillating_add_remove_across_five_windows` /// when this was first written without the sort). + /// + /// `sort_by_key` only orders by timestamp, so two buckets sharing an + /// exact timestamp keep whatever relative order they arrive in -- + /// that's NOT `bucket_map`'s HashMap order, though: `build_bucket_map` + /// preserves each `Vec` group's original order when + /// grouping by start, and that Vec was already chronologically sorted + /// once by the store itself (`sort_buckets_chronologically`, called in + /// `query_precomputed_output`/`query_precomputed_output_exact_batch` + /// before this function ever sees the data) -- so same-timestamp order + /// here is deterministic, just not decided by this function; it's + /// inherited from the store's own sort, same as it always was for + /// `sum_window` (#581 stage E.4 review). fn collect_bucket_map_entries_before( bucket_map: &HashMap>, before: u64, @@ -1870,21 +1897,23 @@ impl SimpleEngine { // unrepresentable is the same reasoning that motivated this enum // over two raw Option fields in the first place — just applied all // the way through instead of partway. + // Named alias purely to keep declarations under + // clippy::type_complexity -- used by both KeysSource::PerStep's own + // bucket_map field below and the step-major `groups` binding + // further down (#581 stage E.4 review: previously duplicated as the + // raw type at the PerStep site instead of using this alias). + type GroupBucketMap<'a> = HashMap>; + enum KeysSource<'a> { Fixed(Option), PerStep { - bucket_map: HashMap>, + bucket_map: GroupBucketMap<'a>, lookback_ms: u64, tumbling_window_ms: u64, window_type: WindowType, }, } - // Named alias purely to keep the step-major `groups` binding below - // under clippy::type_complexity -- same shape as KeysSource::PerStep's - // own bucket_map field above. - type GroupBucketMap<'a> = HashMap>; - // Resolve, for every value group, which groups exist at all (a // one-time operation — see design doc) and where their expansion // keys come from. A group with keys data but no value data @@ -2048,6 +2077,24 @@ impl SimpleEngine { let keys_window_buckets = if key_accumulator_type == AggregationType::DeltaSetAggregator { + // #588/#606 force DeltaSetAggregator's own + // config to Tumbling at planning time -- but + // that's a planner convention, not a runtime + // invariant this code can trust blindly. + // AggregationConfig can be (and in this crate's + // own tests routinely is) constructed directly, + // bypassing the planner. A Sliding DeltaSetAgg + // has no coherent "replay from the beginning" + // semantics to begin with, so this asserts + // rather than silently reinterpreting it (#581 + // stage E.4 review). + assert_eq!( + *keys_window_type, + WindowType::Tumbling, + "DeltaSetAggregator keys config must be Tumbling (#588/#606) -- \ + the replay-from-the-beginning fast path has no correct meaning \ + for Sliding" + ); Self::collect_bucket_map_entries_before(keys_bucket_map, current_time) } else { let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); @@ -2149,11 +2196,8 @@ impl SimpleEngine { // and would otherwise keep a different group on every process // run when two groups tie at the k-th value. if let Some(k) = topk_k { - step_results.sort_by(|a, b| { - b.1.partial_cmp(&a.1) - .unwrap_or(std::cmp::Ordering::Equal) - .then_with(|| a.0.labels.cmp(&b.0.labels)) - }); + step_results + .sort_by(|a, b| Self::cmp_topk_value_desc(a.1, &a.0.labels, b.1, &b.0.labels)); step_results.truncate(k); }