diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 20af1251..410b810c 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -114,6 +114,19 @@ pub struct RangeQueryExecutionContext { pub lookback_bucket_count: usize, /// Tumbling window size in ms pub tumbling_window_ms: u64, + /// Per-step lookback for the keys aggregation (#583): `keys_query.end - + /// keys_query.start` from the instant window `create_keys_query_params` + /// computed before widening. `None` when there's no separate + /// `keys_query`. Reused, unmodified, as the per-step + /// `current_time.saturating_sub(keys_lookback_ms)` window start for + /// every output step -- this single value is what makes `SetAggregator` + /// a normal sliding window and `DeltaSetAggregator` always replay from + /// `0`, with no `AggregationType` branching needed here. + pub keys_lookback_ms: Option, + /// Bucket width for the keys aggregation, separate from + /// `tumbling_window_ms` (which is the *value* aggregation's width) -- + /// the two can legitimately differ. + pub keys_tumbling_window_ms: Option, } // /// Parsed components of a sketch query, extracted either via the PromQL AST @@ -604,9 +617,11 @@ impl SimpleEngine { } /// Fetches and merges the keys side of a dual-population query plan, if - /// present. Shared by `execute_and_merge_store_queries` (instant) and - /// `execute_range_query_pipeline` (range) — both build a `StoreQueryPlan` - /// that may carry a separate `keys_query` and need it merged the same way. + /// present. Used by `execute_and_merge_store_queries` (instant) only — + /// `execute_range_query_pipeline` (range) fetches keys raw via + /// `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." fn fetch_and_merge_keys( &self, keys_query: &Option, @@ -1463,6 +1478,56 @@ impl SimpleEngine { // Some((output_labels, QueryResult::matrix(range_elements))) // } + /// Builds a lookup from bucket start-timestamp to every bucket sharing + /// that start. A Sliding aggregation can legitimately return more than + /// one bucket per start timestamp (#567/#570) — every one of them must + /// be merged, not just the last one collected here. Used identically by + /// `execute_range_query_pipeline` for both the value side and (#583) + /// the keys side. + fn build_bucket_map( + buckets: &[crate::stores::TimestampedBucket], + ) -> HashMap> { + let mut bucket_map: HashMap> = HashMap::new(); + for ((start, _), bucket) in buckets { + bucket_map.entry(*start).or_default().push(bucket.as_ref()); + } + bucket_map + } + + /// Collects every bucket in `bucket_map` whose start falls in + /// `[window_start, window_end)`, stepping by `step_increment`. Missing + /// buckets at a given start are skipped (partial data is okay). Used + /// 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( + bucket_map: &HashMap>, + window_start: u64, + window_end: u64, + step_increment: u64, + ) -> Vec> { + // A zero step never advances `t`, so the loop below would never + // terminate. This is the single caller-facing guard for that hazard + // -- callers upstream may also validate their own step sources, but + // this function has two callers (values, keys) and should not rely + // on either of them to have done so. Kept active in release builds + // (assert!, not debug_assert!): a hung query is a production + // incident, not just a debug-time nicety. + assert!( + step_increment > 0, + "scan_window: step_increment must be nonzero, or this loop never terminates" + ); + let mut window_buckets: Vec> = Vec::new(); + let mut t = window_start; + while t < window_end { + if let Some(buckets) = bucket_map.get(&t) { + window_buckets.extend(buckets.iter().map(|b| b.clone_boxed_core())); + } + t += step_increment; + } + window_buckets + } + /// Execute the range query pipeline fn execute_range_query_pipeline( &self, @@ -1484,20 +1549,25 @@ impl SimpleEngine { all_data.values().map(|v| v.len()).sum::() ); - // Dual-population metrics (separate key/value aggregations) need the - // keys side merged once up front, then expanded per value group below - // — mirrors execute_and_merge_store_queries/collect_results_separate_keys - // (#580: this pipeline previously never read keys_query at all). - let merged_keys = self.fetch_and_merge_keys( - &context.base.store_plan.keys_query, - context.base.agg_info.aggregation_type_for_key, - context.base.do_merge, - )?; + // #583: fetch keys raw (no merge). Unlike keys, values have always + // been fetched raw here and merged per-step below (see the loop); + // keys used to go through fetch_and_merge_keys, which collapses + // every fetched bucket into ONE snapshot before this function ever + // sees it. That collapse is the bug: once buckets are merged + // together there's no way to ask what the key set looked like at + // any specific earlier timestamp. Fetching raw and merging per-step, + // mirroring the values loop, is the fix. + let keys_raw_data: Option = match &context.base.store_plan.keys_query + { + Some(keys_query) => Some(self.execute_store_query(keys_query)?), + None => None, + }; let mut results: HashMap = HashMap::new(); // Determine accumulator type for merger selection let accumulator_type = &context.base.agg_info.aggregation_type_for_value; + let key_accumulator_type = context.base.agg_info.aggregation_type_for_key; // Calculate step parameters let step_ms = context.range_params.step; @@ -1507,6 +1577,8 @@ impl SimpleEngine { let lookback_bucket_count = context.lookback_bucket_count; let tumbling_window_ms = context.tumbling_window_ms; let lookback_ms = (lookback_bucket_count as u64) * tumbling_window_ms; + let keys_lookback_ms = context.keys_lookback_ms; + let keys_tumbling_window_ms = context.keys_tumbling_window_ms; let window_mode = if buckets_per_step <= lookback_bucket_count { "sliding (slide <= size)" @@ -1528,82 +1600,190 @@ impl SimpleEngine { // Whether the value accumulator's own get_keys() is even consulted // depends on the query SHAPE (dual- vs single-population), not on a // per-group fallback — mirrors collect_all_results exactly: - // - dual-population (separate keys_query present): always expand - // via the keys aggregation's get_keys(), full stop. The value - // accumulator's own get_keys() is never consulted, even if the - // value accumulator itself happens to be self-keyed (e.g. a - // CountMinSketchWithHeap value paired with a DeltaSetAggregator - // keys aggregation is a real capability-matched config, see - // sql.rs). Otherwise a self-keyed value accumulator's own - // (possibly different, window-to-window-shifting) keys would - // silently override the keys aggregation's expansion. See #587 - // review. - // - single-population (no separate keys_query): the value - // accumulator's own get_keys() takes priority whenever present - // (#584, self-keyed accumulators like top-k), falling back to - // the store-level group key otherwise. - let is_dual_population = merged_keys.is_some(); - - // Resolve, for every value group, a fallback key list — used - // directly for dual-population groups, or as a fallback for - // single-population groups whose value accumulator doesn't self-key. - let groups: Vec<( - &Vec, - Vec, - )> = match &merged_keys { - Some(keys_map) => keys_map - .iter() - .map(|(group_key, keys_precompute)| { - let timestamped_buckets = all_data - .get(group_key) - .ok_or_else(|| format!("No value for key: {:?}", group_key))?; - let fallback_keys = keys_precompute - .get_keys() - .ok_or_else(|| "Keys required for separate aggregation".to_string())?; - Ok((timestamped_buckets, fallback_keys)) - }) - .collect::, String>>()?, + // - dual-population (KeysSource::PerStep below, separate + // keys_query present): always expand via the keys aggregation's + // per-step merge (#583). The value accumulator's own get_keys() + // is never consulted, even if the value accumulator itself + // happens to be self-keyed (e.g. a CountMinSketchWithHeap value + // paired with a DeltaSetAggregator keys aggregation is a real + // capability-matched config, see sql.rs). Otherwise a + // self-keyed value accumulator's own (possibly different, + // window-to-window-shifting) keys would silently override the + // keys aggregation's expansion. See #587 review. + // - single-population (KeysSource::Fixed below, no separate + // keys_query): the value accumulator's own get_keys() takes + // priority whenever present (#584, self-keyed accumulators like + // top-k), evaluated AFTER merging the window's value buckets + // (a top-k heap's keys can depend on that window's data), + // falling back to the store-level group key otherwise. + // PerStep bundles everything a dual-population group's per-step + // resolution needs (bucket_map, lookback_ms, tumbling_window_ms) in + // one place, built once at group-construction time — rather than + // three separate Option fields at function scope that only + // happened to be Some together by convention, each re-unwrapped via + // .expect() on every iteration of the per-step loop. Making the + // invalid state (PerStep present but one companion value missing) + // 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. + enum KeysSource<'a> { + Fixed(Vec), + PerStep { + bucket_map: HashMap>, + lookback_ms: u64, + tumbling_window_ms: u64, + }, + } + + // 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 + // anywhere in the queried range is skipped with a warning instead + // of failing the whole range query (#583; previously + // `.ok_or_else(...)?` here hard-failed everything for one missing + // group). See #582 review for collect_results_separate_keys parity. + let groups: Vec<(&Vec, KeysSource)> = match &keys_raw_data + { + Some(keys_map) => { + // keys_raw_data is Some, so context.keys_lookback_ms / + // context.keys_tumbling_window_ms are guaranteed Some too + // (both derived from the same keys_query.is_some() check in + // finish_range_context) -- resolved once here instead of + // re-unwrapped per group per step. + let keys_lookback_ms = + keys_lookback_ms.expect("keys_raw_data implies keys_lookback_ms is Some"); + let keys_tumbling_window_ms = keys_tumbling_window_ms + .expect("keys_raw_data implies keys_tumbling_window_ms is Some"); + keys_map + .iter() + .filter_map( + |(group_key, raw_keys_buckets)| match all_data.get(group_key) { + Some(timestamped_buckets) => Some(( + timestamped_buckets, + KeysSource::PerStep { + bucket_map: Self::build_bucket_map(raw_keys_buckets), + lookback_ms: keys_lookback_ms, + tumbling_window_ms: keys_tumbling_window_ms, + }, + )), + None => { + warn!( + "Range query: group {:?} has keys data but no value data \ + anywhere in the queried range — skipping this group instead \ + of failing the whole query (#583)", + group_key + ); + None + } + }, + ) + .collect() + } + // #584/#587: keep every group, including group_key=None — that's + // exactly where a self-keyed single-population accumulator + // (e.g. top-k) is typically stored. An empty fallback list here + // is fine; the per-step loop below tries the value + // accumulator's own get_keys() first and only falls back to + // this list. None => all_data .iter() - .map(|(group_key, buckets)| (buckets, group_key.clone().into_iter().collect())) + .map(|(group_key, buckets)| { + ( + buckets, + KeysSource::Fixed(group_key.clone().into_iter().collect()), + ) + }) .collect(), }; // Process each value group independently - for (timestamped_buckets, fallback_keys) in groups { - // Build lookup: bucket_start_timestamp -> all buckets sharing - // that start. A Sliding aggregation can legitimately return more - // than one bucket per start timestamp (#567/#570) — every one of - // them must be merged, not just the last one collected here. - let mut bucket_map: HashMap> = HashMap::new(); - for ((start, _), bucket) in timestamped_buckets { - bucket_map.entry(*start).or_default().push(bucket.as_ref()); - } + for (timestamped_buckets, keys_source) in groups { + let bucket_map = Self::build_bucket_map(timestamped_buckets); debug!( - "Group with {} start-timestamps, fallback keys: {:?}", + "Group with {} start-timestamps ({} keys start-timestamps)", bucket_map.len(), - fallback_keys + match &keys_source { + KeysSource::PerStep { bucket_map, .. } => bucket_map.len(), + KeysSource::Fixed(_) => 0, + } ); // Iterate by OUTPUT timestamp, not by bucket index let mut current_time = start_ms; while current_time <= end_ms { + // #583: dual-population groups resolve their expansion keys + // from the keys aggregation, per step — not a single + // snapshot reused for every step — and (#587) never from + // the value accumulator's own get_keys(). If nothing + // resolves at this step, skip it before ever touching the + // value merge below. Fixed (single-population) groups defer + // key resolution until after the value merge (#584/#587). + let per_step_keys: Option> = match &keys_source { + KeysSource::PerStep { + bucket_map: keys_bucket_map, + lookback_ms: keys_lookback_ms, + tumbling_window_ms: keys_tumbling_window_ms, + } => { + let keys_window_start = current_time.saturating_sub(*keys_lookback_ms); + let keys_window_buckets = Self::scan_window( + keys_bucket_map, + keys_window_start, + current_time, + *keys_tumbling_window_ms, + ); + + let expansion_keys = if keys_window_buckets.is_empty() { + Vec::new() + } else { + let mut key_merger = create_window_merger(key_accumulator_type); + key_merger.initialize(keys_window_buckets); + match key_merger.get_merged() { + Ok(merged) => match merged.get_keys() { + Some(keys) => keys, + None => { + // e.g. a DeltaSetAggregator "remove" with no + // matching "add" resolved in this window -- + // distinct from (and louder than) the routine, + // expected "no buckets in this window at all" + // case below, since it means a merge DID + // happen but couldn't resolve a key set. + warn!( + "Keys merge at t={} produced an unresolved key \ + set (get_keys() returned None) -- skipping \ + this step for this group", + current_time + ); + Vec::new() + } + }, + Err(e) => { + debug!("Failed to merge keys at t={}: {}", current_time, e); + Vec::new() + } + } + }; + + if expansion_keys.is_empty() { + debug!( + "No expansion keys resolved at t={} — skipping this step for this group", + current_time + ); + current_time += step_ms; + continue; + } + Some(expansion_keys) + } + KeysSource::Fixed(_) => None, + }; + // Window covers [current_time - lookback_ms, current_time) // This means we look at buckets that START within this range let window_start = current_time.saturating_sub(lookback_ms); // Collect all AVAILABLE buckets in this window (skip missing ones) - let mut window_buckets: Vec> = Vec::new(); - - let mut t = window_start; - while t < current_time { - if let Some(buckets) = bucket_map.get(&t) { - window_buckets.extend(buckets.iter().map(|b| b.clone_boxed_core())); - } - // If no bucket at timestamp t, just skip it (partial data is okay) - t += tumbling_window_ms; - } + let window_buckets = + Self::scan_window(&bucket_map, window_start, current_time, tumbling_window_ms); if !window_buckets.is_empty() { // Merge available buckets @@ -1612,18 +1792,21 @@ impl SimpleEngine { match merger.get_merged() { Ok(merged) => { - // See is_dual_population above: dual-population - // always trusts the keys aggregation's expansion; - // only single-population lets the value - // accumulator's own get_keys() (read after + // See the note above KeysSource: dual-population + // (per_step_keys already resolved, non-empty) + // always uses that; single-population lets the + // value accumulator's own get_keys() (read after // merging this window, since e.g. a top-k heap's // keys depend on the window's data) take // priority, falling back to the store-level // group key otherwise. - let resolved_keys = if is_dual_population { - fallback_keys.clone() - } else { - merged.get_keys().unwrap_or_else(|| fallback_keys.clone()) + let resolved_keys = match &keys_source { + KeysSource::PerStep { .. } => per_step_keys.expect( + "PerStep always sets per_step_keys above, or continues", + ), + KeysSource::Fixed(fallback_keys) => { + merged.get_keys().unwrap_or_else(|| fallback_keys.clone()) + } }; // Query statistic and emit a sample at current_time // for every expanded key sharing this value group. @@ -1651,16 +1834,16 @@ impl SimpleEngine { } Err(e) => { debug!( - "Failed to get merged result at t={} for keys {:?}: {}", - current_time, fallback_keys, e + "Failed to get merged result at t={} (per_step_keys={:?}): {}", + current_time, per_step_keys, e ); } } } else { // No data at all for this window - skip sample debug!( - "Keys {:?}: skipping sample at {} - no data in window [{}, {})", - fallback_keys, current_time, window_start, current_time + "Skipping sample at {} (per_step_keys={:?}) - no data in window [{}, {})", + current_time, per_step_keys, window_start, current_time ); } diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index bd51a883..10e072ed 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, + RangeQueryParams, StoreQueryParams, }; use crate::data_model::{AggregationIdInfo, KeyByLabelValues, QueryConfig, SchemaConfig}; use crate::engines::query_result::{InstantVectorElement, QueryResult, RangeVectorElement}; @@ -540,6 +540,23 @@ 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 @@ -573,16 +590,47 @@ impl SimpleEngine { }) .ok()?; - let lookback_ms = base_context.store_plan.values_query.end_timestamp - - base_context.store_plan.values_query.start_timestamp; + let mut extended_store_plan = base_context.store_plan.clone(); + let lookback_ms = + Self::widen_query_window(&mut extended_store_plan.values_query, start_ms, end_ms); + extended_store_plan.values_query.is_exact_query = false; let buckets_per_step = (step_ms / tumbling_window_ms) as usize; let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; - let mut extended_store_plan = base_context.store_plan.clone(); - extended_store_plan.values_query.start_timestamp = start_ms.saturating_sub(lookback_ms); - extended_store_plan.values_query.end_timestamp = end_ms; - extended_store_plan.values_query.is_exact_query = false; + // #583: widen keys_query the same way, using the instant window + // create_keys_query_params already computed for it as the source of + // truth for "how far back does this aggregation type look." This + // needs no AggregationType branching: for SetAggregator the instant + // window is [end-window_size, end], so keys_lookback_ms == window_size + // and this widens to a normal sliding window; for DeltaSetAggregator + // the instant window is [0, end], so keys_lookback_ms == end_ms, + // which saturating_sub's to 0 for every current_time <= end_ms in + // the per-step loop -- i.e. "replay from the beginning," for free. + let keys_lookback_ms = extended_store_plan + .keys_query + .as_mut() + .map(|keys_query| Self::widen_query_window(keys_query, start_ms, end_ms)); + let keys_tumbling_window_ms = match keys_lookback_ms { + Some(_) => Some( + self.streaming_config + .read() + .unwrap() + .get_aggregation_config(base_context.agg_info.aggregation_id_for_key) + .map(|c| c.window_size_ms)?, + ), + 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 }`) + // 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 + // multiple of 0); keys has no equivalent check, so guard explicitly. + if keys_tumbling_window_ms == Some(0) { + warn!("Range query validation failed: key aggregation window_size_ms is 0"); + return None; + } Some(RangeQueryExecutionContext { base: QueryExecutionContext { @@ -597,6 +645,8 @@ impl SimpleEngine { buckets_per_step, lookback_bucket_count, tumbling_window_ms, + keys_lookback_ms, + keys_tumbling_window_ms, }) } diff --git a/asap-query-engine/src/engines/window_merger.rs b/asap-query-engine/src/engines/window_merger.rs index 552249e3..bb576322 100644 --- a/asap-query-engine/src/engines/window_merger.rs +++ b/asap-query-engine/src/engines/window_merger.rs @@ -401,4 +401,89 @@ mod tests { .unwrap(); assert_eq!(mock3.value, 120.0); } + + /// Pins a subtle, easy-to-miss requirement that `execute_range_query_pipeline`'s + /// per-step keys merge (#583) depends on: `NaiveMerger` folds buckets + /// *pairwise, left-to-right* (each `get_merged` walks `buckets[0].merge_with(buckets[1])`, + /// then that result `.merge_with(buckets[2])`, and so on) rather than + /// passing the whole slice to a single flat N-way merge. For most + /// accumulators the distinction is invisible (merging is associative and + /// commutative). It is NOT invisible for `DeltaSetAggregatorAccumulator`: + /// its merge treats a key present in both the added set and the removed + /// set as a cancelling conflict (see `merge_accumulators`), so a key that + /// toggles more than once within the merged buckets only nets out to the + /// chronologically correct state if the deltas are folded in order. + /// + /// Grows the window one bucket at a time (add, remove, add, remove, add) + /// via `slide(0, ..)` -- mirroring how the range pipeline's per-step + /// window for `DeltaSetAggregator` only ever grows, never expires -- and + /// checks `get_merged()` after *every* addition, not just the final one, + /// so a fold that's only correct at the boundary (e.g. an implementation + /// that happens to get the last step right by luck) can't hide. + /// + /// If a future change to the range-query pipeline (or to `WindowMerger` + /// itself) ever collects a `DeltaSetAggregator` window's buckets and + /// merges them with one flat call instead of `NaiveMerger`'s sequential + /// fold, this test is the tripwire that catches it. + #[test] + fn naive_merger_sequential_fold_replays_delta_set_toggles_at_every_window() { + use crate::data_model::traits::MergeableAccumulator; + use crate::precompute_operators::DeltaSetAggregatorAccumulator; + + let key = KeyByLabelValues::new_with_labels(vec!["host-a".to_string()]); + + let mut add = DeltaSetAggregatorAccumulator::new(); + add.add_key(key.clone()); + let mut remove = DeltaSetAggregatorAccumulator::new(); + remove.remove_key(key.clone()); + + // add, remove, add, remove, add -> present, absent, present, absent, present + let deltas = [ + add.clone(), + remove.clone(), + add.clone(), + remove.clone(), + add.clone(), + ]; + let expected_present = [true, false, true, false, true]; + + let mut merger = NaiveMerger::new(); + merger.initialize(vec![Box::new(deltas[0].clone())]); + let mismatches: Vec = expected_present + .iter() + .enumerate() + .filter_map(|(i, &expected)| { + if i > 0 { + merger.slide(0, vec![Box::new(deltas[i].clone())]); + } + let merged = merger.get_merged().unwrap(); + let actual = merged + .get_keys() + .expect("no unresolved removals after a sequential fold") + .contains(&key); + (actual != expected) + .then(|| format!("window {}: expected {expected}, got {actual}", i + 1)) + }) + .collect(); + assert!( + mismatches.is_empty(), + "NaiveMerger's sequential left-fold must replay each toggle chronologically \ + at every window, not just the final one -- diverged at: {mismatches:?}" + ); + + // Contrast: the same 5 buckets merged in one flat call (not a + // sequential fold) lose the key entirely, proving these are NOT + // interchangeable for DeltaSetAggregator. + let flat_result = DeltaSetAggregatorAccumulator::merge_accumulators(deltas.to_vec()) + .expect("flat merge should still succeed, just give the wrong answer"); + assert!( + !flat_result + .get_keys() + .expect("no unresolved removals after the flat merge either") + .contains(&key), + "a flat (non-sequential) merge_accumulators call over the same 5 buckets \ + must NOT net the key to present -- this is exactly the mistake the \ + sequential fold above avoids" + ); + } } 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 556d3568..70899fae 100644 --- a/asap-query-engine/src/tests/native_range_query_tests.rs +++ b/asap-query-engine/src/tests/native_range_query_tests.rs @@ -30,7 +30,8 @@ mod tests { use crate::engines::simple_engine::SimpleEngine; use crate::precompute_operators::sum_accumulator::SumAccumulator; use crate::precompute_operators::{ - CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, DeltaSetAggregatorAccumulator, + CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, + DeltaSetAggregatorAccumulator, SetAggregatorAccumulator, }; use crate::stores::simple_map_store::SimpleMapStore; use crate::stores::Store; @@ -49,6 +50,60 @@ mod tests { } } + /// True if the output series for `host_label` (matched by label value) has + /// a sample at `ts`. False if the series is absent entirely, or present + /// but without a sample at that timestamp. + fn key_has_sample_at(elements: &[RangeVectorElement], host_label: &str, ts: u64) -> bool { + elements + .iter() + .find(|e| e.labels.labels.contains(&host_label.to_string())) + .is_some_and(|e| e.samples.iter().any(|s| s.timestamp == ts)) + } + + /// Like `key_has_sample_at`, but matches a series only if ALL given label + /// values are present -- needed once there's more than one grouping + /// label (e.g. region + host), since a bare host value alone can't tell + /// "host-b under region=eu" apart from "host-b under region=us". + fn labels_have_sample_at( + elements: &[RangeVectorElement], + label_values: &[&str], + ts: u64, + ) -> bool { + elements + .iter() + .find(|e| { + label_values + .iter() + .all(|lv| e.labels.labels.contains(&lv.to_string())) + }) + .is_some_and(|e| e.samples.iter().any(|s| s.timestamp == ts)) + } + + /// Checks every `(label_values, ts, expected_present, reason)` case + /// against `elements` and reports ALL mismatches in one panic, instead of + /// stopping at the first failing `assert!` -- each of these tests makes + /// several independent claims (about different keys/timestamps), and a + /// fix attempt that gets some right and some wrong is much easier to + /// debug when every divergence is visible at once. + fn assert_all_at(elements: &[RangeVectorElement], cases: &[(&[&str], u64, bool, &str)]) { + let mismatches: Vec = cases + .iter() + .filter_map(|(labels, ts, expected, reason)| { + let actual = labels_have_sample_at(elements, labels, *ts); + (actual != *expected).then(|| { + format!( + "{labels:?}@{ts}: expected present={expected}, got {actual} -- {reason}" + ) + }) + }) + .collect(); + assert!( + mismatches.is_empty(), + "diverged from expectations at:\n{}", + mismatches.join("\n") + ); + } + /// One tumbling-window bucket: (bucket end timestamp ms, label values, accumulator). type TimeSeriesData = Vec<(u64, Option>, Box)>; @@ -67,6 +122,37 @@ mod tests { value_data: TimeSeriesData, keys_data: TimeSeriesData, promql_query: &str, + ) -> SimpleEngine { + create_range_engine_dual_input_with_windows( + metric, + value_agg_type, + key_agg_type, + grouping_labels, + aggregated_labels, + value_data, + keys_data, + promql_query, + WINDOW_MS, + WINDOW_MS, + ) + } + + /// Same as `create_range_engine_dual_input`, but lets the value and key + /// aggregations use different bucket widths — the value aggregation's + /// `tumbling_window_ms` must not be assumed to also be the key + /// aggregation's bucket width when scanning the keys `bucket_map` (#583). + #[allow(clippy::too_many_arguments)] + fn create_range_engine_dual_input_with_windows( + metric: &str, + value_agg_type: AggregationType, + key_agg_type: AggregationType, + grouping_labels: Vec<&str>, + aggregated_labels: Vec<&str>, + value_data: TimeSeriesData, + keys_data: TimeSeriesData, + promql_query: &str, + value_window_ms: u64, + key_window_ms: u64, ) -> SimpleEngine { let grouping_label_strings: Vec = grouping_labels.iter().map(|s| s.to_string()).collect(); @@ -90,8 +176,8 @@ mod tests { aggregated_labels: KeyByLabelNames::empty(), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: WINDOW_MS, - slide_interval_ms: WINDOW_MS, + window_size_ms: value_window_ms, + slide_interval_ms: value_window_ms, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -113,8 +199,8 @@ mod tests { aggregated_labels: KeyByLabelNames::new(aggregated_label_strings), rollup_labels: KeyByLabelNames::empty(), original_yaml: String::new(), - window_size_ms: WINDOW_MS, - slide_interval_ms: WINDOW_MS, + window_size_ms: key_window_ms, + slide_interval_ms: key_window_ms, window_type: WindowType::Tumbling, spatial_filter: String::new(), spatial_filter_normalized: String::new(), @@ -135,10 +221,13 @@ mod tests { CleanupPolicy::NoCleanup, )); - for (agg_id, data) in [(1u64, value_data), (2u64, keys_data)] { + for (agg_id, window_ms, data) in [ + (1u64, value_window_ms, value_data), + (2u64, key_window_ms, keys_data), + ] { for (timestamp, label_values_opt, acc) in data { let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); - let output = PrecomputedOutput::new(timestamp - WINDOW_MS, timestamp, key, agg_id); + let output = PrecomputedOutput::new(timestamp - window_ms, timestamp, key, agg_id); store.insert_precomputed_output(output, acc).unwrap(); } } @@ -210,17 +299,35 @@ mod tests { // pipeline that calls merged.get_keys() unconditionally on the merged // VALUE accumulator would incorrectly let the heap's own internal // top-k keys override the keys_query's expansion. - let mut heap = CountMinSketchWithHeapAccumulator::new(3, 1024, 32); + // + // Extended (not just the original single-timestamp check) to also + // exercise #583's per-step keys resolution in the SAME test: host-b + // is added to the keys aggregation only at t=2000, so a correct + // implementation must show it absent at t=1000 and present at + // t=2000 -- while the self-keyed heap's own top-k keys (host-x, + // host-y, present in EVERY value bucket) must never appear at + // either step. The two mechanisms are provably orthogonal (dual- + // population never touches the value accumulator's own get_keys() + // at all), but this pins both invariants holding simultaneously in + // one test rather than trusting that orthogonality alone. + let mut heap_1 = CountMinSketchWithHeapAccumulator::new(3, 1024, 32); + let mut heap_2 = CountMinSketchWithHeapAccumulator::new(3, 1024, 32); // Heap's own top-k keys are deliberately disjoint from the keys_query's - // key, so a wrong implementation is caught by key-set mismatch, not - // just a wrong value. - heap.inner.update("host-x;evt-x", 30.0); - heap.inner.update("host-y;evt-y", 20.0); + // keys, at every step, so a wrong implementation is caught by + // key-set mismatch, not just a wrong value. + for heap in [&mut heap_1, &mut heap_2] { + heap.inner.update("host-x;evt-x", 30.0); + heap.inner.update("host-y;evt-y", 20.0); + } - let mut keys = DeltaSetAggregatorAccumulator::new(); - keys.add_key(KeyByLabelValues { + let mut keys_1000 = DeltaSetAggregatorAccumulator::new(); + keys_1000.add_key(KeyByLabelValues { labels: vec!["host-a".to_string(), "evt-1".to_string()], }); + let mut keys_2000 = DeltaSetAggregatorAccumulator::new(); + keys_2000.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); let engine = create_range_engine_dual_input( "event_frequency", @@ -228,24 +335,73 @@ mod tests { AggregationType::DeltaSetAggregator, vec![], vec!["host", "event"], - vec![(1000, None, Box::new(heap) as Box)], - vec![(1000, None, Box::new(keys) as Box)], + vec![ + (1000, None, Box::new(heap_1) as Box), + (2000, None, Box::new(heap_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1000) as Box), + (2000, None, Box::new(keys_2000) as Box), + ], "count(event_frequency) by (host, event)", ); let query = "count(event_frequency) by (host, event)"; - let result = engine.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0); + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); let (_, qr) = result.expect("range query failed"); let elements = matrix_values(qr); - let returned: std::collections::HashSet> = - elements.iter().map(|e| e.labels.labels.clone()).collect(); - assert_eq!( - returned, - std::collections::HashSet::from([vec!["host-a".to_string(), "evt-1".to_string()]]), - "expected the keys_query's key (host-a, evt-1), not the value accumulator's own \ - top-k keys (host-x/host-y), got: {:?}", - returned + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a is the keys_query's key from the start", + ), + ( + &["host-b"], + 1000, + false, + "host-b's key delta only appears at t=2000", + ), + ( + &["host-x"], + 1000, + false, + "the self-keyed heap's own top-k keys must never override the \ + keys_query's expansion for a dual-population group", + ), + ( + &["host-y"], + 1000, + false, + "the self-keyed heap's own top-k keys must never override the \ + keys_query's expansion for a dual-population group", + ), + ( + &["host-a"], + 2000, + true, + "host-a persists (DeltaSetAggregator never removes it)", + ), + (&["host-b"], 2000, true, "host-b was added by t=2000"), + ( + &["host-x"], + 2000, + false, + "the self-keyed heap's own top-k keys must never override the \ + keys_query's expansion for a dual-population group", + ), + ( + &["host-y"], + 2000, + false, + "the self-keyed heap's own top-k keys must never override the \ + keys_query's expansion for a dual-population group", + ), + ], ); } @@ -423,6 +579,953 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_key_appearing_midrange_has_no_phantom_earlier_sample() { + // Issue #583: execute_range_query_pipeline fetches/merges keys_query + // once (anchored at the range's end) and reuses that single snapshot + // for every output step. host-b's DeltaSetAggregator "added" delta + // only appears in the bucket at t=2000, so the correct per-step key + // set at t=1000 must not include host-b yet — but the current + // single-snapshot merge folds host-b's later add into the whole + // range, giving it a phantom sample at t=1000 before it existed. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = DeltaSetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = DeltaSetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + // DeltaSetAggregator accumulates: once added, a key stays in the + // reconstructed set at every later step too (no removal here). + // host-a is added at t=1000, so it should be present at both steps. + // host-b is added only at t=2000, so it must be absent at t=1000 + // (not yet added) and present at t=2000. + assert_all_at( + &elements, + &[ + (&["host-a"], 1000, true, "host-a was added at t=1000"), + ( + &["host-a"], + 2000, + true, + "host-a was added at t=1000 and DeltaSetAggregator never removes it", + ), + ( + &["host-b"], + 1000, + false, + "host-b's key delta only appears at t=2000, so it must not have a \ + phantom sample at t=1000 (before it existed)", + ), + ( + &["host-b"], + 2000, + true, + "host-b should have a sample at t=2000, once its key delta appears", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_set_aggregator_earlier_key_not_silently_dropped() { + // Issue #583 (second half): for SetAggregator ("latest window only"), + // create_keys_query_params scopes keys_query to [end-window_size, end] + // — a single instant-anchored window at the *range's* end, not each + // step's own window. host-a's full-snapshot bucket only exists at + // t=1000; with range end=2000 and window_size=1000, the keys fetch + // window is [1000, 2000], which excludes host-a's bucket (0..1000) + // entirely — its start (0) falls before the query window's start (1000). + // Since the range loop iterates merged_keys (not all_data), host-a + // never gets iterated at all — its whole series silently vanishes + // from the output, even though it had a real sample at t=1000. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = SetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = SetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + // SetAggregator is a full snapshot per window, not a delta: host-a is + // the live set only for the window ending at t=1000, and is replaced + // by host-b in the window ending at t=2000 — each step's own snapshot + // is disjoint from the other's, unlike DeltaSetAggregator's accumulation. + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a existed at t=1000 (its own window) but was dropped entirely \ + from the range output because the final keys snapshot (scoped to \ + the range's end window) no longer contains it", + ), + ( + &["host-a"], + 2000, + false, + "host-a's SetAggregator snapshot at t=2000 no longer contains it \ + (host-b replaced it)", + ), + ( + &["host-b"], + 1000, + false, + "host-b doesn't appear until the window ending at t=2000", + ), + ( + &["host-b"], + 2000, + true, + "host-b is the live set for the window ending at t=2000", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_binary_expr_arm_key_appearing_midrange_has_no_phantom_earlier_sample() { + // Issue #583's "check other candidates" section names + // build_arm_range_context specifically: handle_binary_expr_range_promql + // builds each arm's RangeQueryExecutionContext via the same + // finish_range_context/execute_range_query_pipeline used by the plain + // range path, so the same single-snapshot keys bug should reproduce + // through a binary expression's scalar arm. Same fixture as + // range_query_dual_population_key_appearing_midrange_has_no_phantom_earlier_sample, + // wrapped in `* 1` so it takes the detect_scalar_arm path in + // handle_binary_expr_range_promql instead of the plain-query dispatch. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = DeltaSetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = DeltaSetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event) * 1"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + (&["host-a"], 1000, true, "host-a was added at t=1000"), + ( + &["host-a"], + 2000, + true, + "host-a was added at t=1000 and DeltaSetAggregator never removes it", + ), + ( + &["host-b"], + 1000, + false, + "host-b's key delta only appears at t=2000, so it must not have a \ + phantom sample at t=1000 (before it existed) — through the binary \ + expr arm path (build_arm_range_context) this time, not the plain range path", + ), + ( + &["host-b"], + 2000, + true, + "host-b should have a sample at t=2000, once its key delta appears", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_delta_set_aggregator_oscillating_add_remove_across_five_windows() { + // Generalizes the phantom-sample tests beyond a single add: host-a + // toggles membership every window (add, remove, add, remove, add) + // across 5 tumbling windows. A correct per-step fix must replay + // deltas only up to *each step's own* end, so presence should + // alternate present/absent/present/absent/present across the 5 + // output steps. The current single end-anchored snapshot instead + // merges all 5 deltas into one net state (present, since the last + // delta is an add) and reuses it for every step — so it would wrongly + // show host-a present at every step, including the two "removed" + // windows (t=2000, t=4000). + let value_data: TimeSeriesData = (1..=5) + .map(|i| { + ( + i * 1000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ) + }) + .collect(); + + let mut keys_add = DeltaSetAggregatorAccumulator::new(); + keys_add.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_remove = DeltaSetAggregatorAccumulator::new(); + keys_remove.remove_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let keys_data: TimeSeriesData = vec![ + ( + 1000, + None, + Box::new(keys_add.clone()) as Box, + ), + ( + 2000, + None, + Box::new(keys_remove.clone()) as Box, + ), + ( + 3000, + None, + Box::new(keys_add.clone()) as Box, + ), + ( + 4000, + None, + Box::new(keys_remove.clone()) as Box, + ), + (5000, None, Box::new(keys_add) as Box), + ]; + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + value_data, + keys_data, + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 5.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + let expected_present = [ + (1000, true), + (2000, false), + (3000, true), + (4000, false), + (5000, true), + ]; + let mismatches: Vec = expected_present + .iter() + .filter_map(|&(ts, expected)| { + let actual = key_has_sample_at(&elements, "host-a", ts); + (actual != expected).then(|| format!("t={ts}: expected {expected}, got {actual}")) + }) + .collect(); + assert!( + mismatches.is_empty(), + "host-a's net membership diverged from the per-step expectation (deltas \ + replayed only up to each step's own end) at: {mismatches:?}" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_binary_expr_arm_set_aggregator_earlier_key_not_silently_dropped() { + // SetAggregator counterpart to + // range_query_binary_expr_arm_key_appearing_midrange_has_no_phantom_earlier_sample: + // closes the matrix by driving + // range_query_set_aggregator_earlier_key_not_silently_dropped's fixture + // through the binary-expr arm path (build_arm_range_context) too, + // instead of only the plain range dispatch. + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + let mut keys_1 = SetAggregatorAccumulator::new(); + keys_1.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = SetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + (1000, None, Box::new(keys_1) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event) * 1"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a existed at t=1000 (its own window) but was dropped entirely \ + from the range output because the final keys snapshot (scoped to \ + the range's end window) no longer contains it — through the binary \ + expr arm path (build_arm_range_context) this time", + ), + ( + &["host-a"], + 2000, + false, + "host-a's SetAggregator snapshot at t=2000 no longer contains it \ + (host-b replaced it)", + ), + ( + &["host-b"], + 1000, + false, + "host-b doesn't appear until the window ending at t=2000", + ), + ( + &["host-b"], + 2000, + true, + "host-b is the live set for the window ending at t=2000", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_delta_set_aggregator_key_change_on_middle_step_not_just_boundary() { + // All prior #583 tests query exactly 2 output steps, where the key + // change lands on one of the range's own boundaries (start or end). + // A fix that special-cases the first/last iteration of the per-step + // loop (rather than genuinely scoping every iteration) could pass + // those while still getting an interior step wrong. This test spans + // 3 output steps (1000, 2000, 3000) with the key added only on the + // *middle* one, so t=1000's snapshot must differ from t=2000's and + // t=3000's purely by virtue of being an interior loop iteration. + let value_data: TimeSeriesData = vec![ + ( + 1000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ( + 2000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ( + 3000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ]; + + let mut keys_2000 = DeltaSetAggregatorAccumulator::new(); + keys_2000.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let keys_data: TimeSeriesData = vec![ + ( + 1000, + None, + Box::new(DeltaSetAggregatorAccumulator::new()) as Box, + ), + (2000, None, Box::new(keys_2000) as Box), + ( + 3000, + None, + Box::new(DeltaSetAggregatorAccumulator::new()) as Box, + ), + ]; + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + value_data, + keys_data, + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 3.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + let expected_present = [(1000, false), (2000, true), (3000, true)]; + let mismatches: Vec = expected_present + .iter() + .filter_map(|&(ts, expected)| { + let actual = key_has_sample_at(&elements, "host-a", ts); + (actual != expected).then(|| format!("t={ts}: expected {expected}, got {actual}")) + }) + .collect(); + assert!( + mismatches.is_empty(), + "host-a's key is added only on the middle step (t=2000), so t=1000 must not \ + see it yet while t=2000 and t=3000 must — diverged at: {mismatches:?}" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_delta_set_aggregator_key_bucket_width_differs_from_value_bucket_width() { + // Issue #583's fix scans the keys bucket_map using a per-step + // increment. That increment must come from the KEY aggregation's own + // window_size_ms, not the VALUE aggregation's tumbling_window_ms — an + // implementation that (incorrectly) reuses the value side's bucket + // width to walk the keys bucket_map would silently skip every keys + // bucket whose start isn't a multiple of that width. All other + // #583 tests use the same window size (WINDOW_MS) for both + // aggregations, so none of them would catch that mistake — this one + // deliberately sets them apart: value buckets are 1000ms wide, key + // (DeltaSetAggregator) buckets are 500ms wide, and both key delta + // buckets are placed off the 1000ms grid (starts at 500 and 1500). + let value_window_ms = 1000; + let key_window_ms = 500; + + let value_data: TimeSeriesData = vec![ + ( + 1000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ( + 2000, + None, + Box::new(CountMinSketchAccumulator::new(2, 3)) as Box, + ), + ]; + + // host-a's delta bucket: start=500, end=1000 (off the 1000ms grid). + let mut keys_host_a = DeltaSetAggregatorAccumulator::new(); + keys_host_a.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + // host-b's delta bucket: start=1500, end=2000 (also off the grid). + let mut keys_host_b = DeltaSetAggregatorAccumulator::new(); + keys_host_b.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + let keys_data: TimeSeriesData = vec![ + (1000, None, Box::new(keys_host_a) as Box), + (2000, None, Box::new(keys_host_b) as Box), + ]; + + let engine = create_range_engine_dual_input_with_windows( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec![], + vec!["host", "event"], + value_data, + keys_data, + "count(event_frequency) by (host, event)", + value_window_ms, + key_window_ms, + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + // host-a's bucket (start=500) falls inside [0,1000), so it must be + // visible at t=1000. host-b's bucket (start=1500) doesn't exist yet + // at t=1000, but is inside [0,2000) by t=2000, and DeltaSetAggregator + // accumulates, so host-a stays present at t=2000 too. + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a's delta bucket starts at t=500, which is inside [0,1000) -- it \ + must be found even though 500 isn't a multiple of the value \ + aggregation's 1000ms bucket width", + ), + ( + &["host-a"], + 2000, + true, + "host-a was added by t=1000 and DeltaSetAggregator never removes it", + ), + ( + &["host-b"], + 1000, + false, + "host-b's delta bucket starts at t=1500, which is not yet inside [0,1000)", + ), + ( + &["host-b"], + 2000, + true, + "host-b's delta bucket starts at t=1500, which is inside [0,2000) -- it \ + must be found even though 1500 isn't a multiple of the value \ + aggregation's 1000ms bucket width", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_per_step_key_change_does_not_leak_across_groups() { + // All prior #583 tests use grouping_labels: vec![] -- a single + // implicit group (group_key = None) -- so "the per-step key set" and + // "the per-step key set for THIS group" are the same computation, + // and a bug that pools state across groups would be invisible. + // Here the value aggregation genuinely groups by `region`, giving + // two independent groups (region=us, region=eu), each with its own + // keys aggregation state. region=us gains host-b mid-range (added at + // t=2000); region=eu's host set (host-x) never changes. A fix whose + // per-step keys merge isn't correctly scoped per group_key -- e.g. + // one that merges all groups' DeltaSetAggregator buckets together + // before re-splitting by group -- would leak host-b into region=eu's + // output at t=2000, since nothing about merge_with prevents merging + // two different regions' delta-sets together. + let cms_us_1 = CountMinSketchAccumulator::new(2, 3); + let cms_us_2 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_1 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_us_1000 = DeltaSetAggregatorAccumulator::new(); + keys_us_1000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_us_2000 = DeltaSetAggregatorAccumulator::new(); + keys_us_2000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-b".to_string(), "evt-1".to_string()], + }); + let mut keys_eu_1000 = DeltaSetAggregatorAccumulator::new(); + keys_eu_1000.add_key(KeyByLabelValues { + labels: vec!["eu".to_string(), "host-x".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec!["region"], + vec!["host", "event"], + vec![ + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(cms_us_1) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(cms_us_2) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_1) as Box, + ), + ( + 2000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_2) as Box, + ), + ], + vec![ + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(keys_us_1000) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(keys_us_2000) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(keys_eu_1000) as Box, + ), + ], + "count(event_frequency) by (region, host, event)", + ); + + let query = "count(event_frequency) by (region, host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + ( + &["us", "host-a"], + 1000, + true, + "region=us has host-a from the start", + ), + ( + &["us", "host-b"], + 1000, + false, + "region=us's host-b is only added at t=2000", + ), + ( + &["us", "host-b"], + 2000, + true, + "region=us's host-b was added by t=2000", + ), + ( + &["eu", "host-x"], + 1000, + true, + "region=eu has host-x from the start", + ), + ( + &["eu", "host-x"], + 2000, + true, + "region=eu's host-x is unaffected by region=us's mid-range change", + ), + ( + &["eu", "host-b"], + 2000, + false, + "region=us's host-b addition must not leak into region=eu's output -- \ + each group's per-step key expansion must be scoped to its own \ + group_key, not pooled across groups", + ), + ( + &["us", "host-x"], + 1000, + false, + "region=eu's host-x must not leak into region=us's output either", + ), + ( + &["us", "host-x"], + 2000, + false, + "region=eu's host-x must not leak into region=us's output either", + ), + ], + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_simultaneous_cross_group_adds_stay_isolated() { + // Sharper variant of ...does_not_leak_across_groups: there, region=eu + // never changes, so a failure could plausibly be misread as plain + // Bug 1 (single-snapshot-per-group) rather than cross-group bleed. + // Here BOTH groups add a DIFFERENT new host at the SAME timestamp + // (t=2000), so any implementation that pools keys across group_keys + // before re-splitting would show up as *both* regions gaining *both* + // new hosts -- a decisive, unambiguous signature distinct from + // Bug 1's "one group's own key is phantom-early" symptom. + let cms_us_1 = CountMinSketchAccumulator::new(2, 3); + let cms_us_2 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_1 = CountMinSketchAccumulator::new(2, 3); + let cms_eu_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_us_1000 = DeltaSetAggregatorAccumulator::new(); + keys_us_1000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_us_2000 = DeltaSetAggregatorAccumulator::new(); + keys_us_2000.add_key(KeyByLabelValues { + labels: vec!["us".to_string(), "host-b".to_string(), "evt-1".to_string()], + }); + let mut keys_eu_1000 = DeltaSetAggregatorAccumulator::new(); + keys_eu_1000.add_key(KeyByLabelValues { + labels: vec!["eu".to_string(), "host-p".to_string(), "evt-1".to_string()], + }); + let mut keys_eu_2000 = DeltaSetAggregatorAccumulator::new(); + keys_eu_2000.add_key(KeyByLabelValues { + labels: vec!["eu".to_string(), "host-q".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec!["region"], + vec!["host", "event"], + vec![ + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(cms_us_1) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(cms_us_2) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_1) as Box, + ), + ( + 2000, + Some(vec!["eu".to_string()]), + Box::new(cms_eu_2) as Box, + ), + ], + vec![ + ( + 1000, + Some(vec!["us".to_string()]), + Box::new(keys_us_1000) as Box, + ), + ( + 2000, + Some(vec!["us".to_string()]), + Box::new(keys_us_2000) as Box, + ), + ( + 1000, + Some(vec!["eu".to_string()]), + Box::new(keys_eu_1000) as Box, + ), + ( + 2000, + Some(vec!["eu".to_string()]), + Box::new(keys_eu_2000) as Box, + ), + ], + "count(event_frequency) by (region, host, event)", + ); + + let query = "count(event_frequency) by (region, host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + let expected = [ + (["us", "host-a"], 1000, true), + (["us", "host-b"], 1000, false), + (["us", "host-a"], 2000, true), + (["us", "host-b"], 2000, true), + (["eu", "host-p"], 1000, true), + (["eu", "host-q"], 1000, false), + (["eu", "host-p"], 2000, true), + (["eu", "host-q"], 2000, true), + // Cross-group: neither region's host should ever appear under the other. + (["us", "host-p"], 1000, false), + (["us", "host-q"], 1000, false), + (["us", "host-p"], 2000, false), + (["us", "host-q"], 2000, false), + (["eu", "host-a"], 1000, false), + (["eu", "host-b"], 1000, false), + (["eu", "host-a"], 2000, false), + (["eu", "host-b"], 2000, false), + ]; + let mismatches: Vec = expected + .iter() + .filter_map(|(labels, ts, expected_present)| { + let actual = labels_have_sample_at(&elements, labels, *ts); + (actual != *expected_present) + .then(|| format!("{labels:?}@{ts}: expected {expected_present}, got {actual}")) + }) + .collect(); + assert!( + mismatches.is_empty(), + "simultaneous cross-group adds must stay isolated per group_key -- \ + diverged at: {mismatches:?}" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn range_query_dual_population_group_with_no_value_data_is_skipped_not_fatal() { + // Today, execute_range_query_pipeline hard-fails the ENTIRE range + // query if any group resolved from merged_keys has no matching entry + // in all_data: `all_data.get(group_key).ok_or_else(|| "No value for + // key")?`. region=orphan has keys data (a real DeltaSetAggregator + // delta) but NEVER has any value/CMS data at all -- under today's + // code this poisons the WHOLE query, so even region=normal's + // perfectly good data disappears. + // + // Per #583's design discussion: this should become non-fatal -- + // skip the orphaned group (with a loud warning, not asserted here + // since this is a unit test, not a log-capture test) and still + // return the rest of the query's results. This test pins the + // EXPECTED (fixed) behavior, so it's RED today: today the whole + // `handle_range_query_promql` call returns None and the `.expect()` + // below panics before any of the assertions run. + let cms_normal_1 = CountMinSketchAccumulator::new(2, 3); + let cms_normal_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_normal = DeltaSetAggregatorAccumulator::new(); + keys_normal.add_key(KeyByLabelValues { + labels: vec![ + "normal".to_string(), + "host-a".to_string(), + "evt-1".to_string(), + ], + }); + let mut keys_orphan = DeltaSetAggregatorAccumulator::new(); + keys_orphan.add_key(KeyByLabelValues { + labels: vec![ + "orphan".to_string(), + "host-z".to_string(), + "evt-1".to_string(), + ], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::DeltaSetAggregator, + vec!["region"], + vec!["host", "event"], + vec![ + ( + 1000, + Some(vec!["normal".to_string()]), + Box::new(cms_normal_1) as Box, + ), + ( + 2000, + Some(vec!["normal".to_string()]), + Box::new(cms_normal_2) as Box, + ), + // Deliberately NO value data for region=orphan, at any timestamp. + ], + vec![ + ( + 1000, + Some(vec!["normal".to_string()]), + Box::new(keys_normal) as Box, + ), + ( + 1000, + Some(vec!["orphan".to_string()]), + Box::new(keys_orphan) as Box, + ), + ], + "count(event_frequency) by (region, host, event)", + ); + + let query = "count(event_frequency) by (region, host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect( + "range query should succeed by skipping the value-less region=orphan group, \ + not fail the entire query because of it", + ); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + ( + &["normal", "host-a"], + 1000, + true, + "region=normal has real value data and should be unaffected by \ + region=orphan having none", + ), + ( + &["normal", "host-a"], + 2000, + true, + "region=normal's host-a persists (DeltaSetAggregator never removes it)", + ), + ], + ); + assert!( + !elements + .iter() + .any(|e| e.labels.labels.contains(&"orphan".to_string())), + "region=orphan has keys data but no value data anywhere -- it must be \ + silently skipped, not appear as an (empty or otherwise) series" + ); + } + /// Single-population counterpart to `create_range_engine_dual_input`: one /// `CountMinSketchWithHeap` (self-keyed, top-k) aggregation, no separate /// keys aggregation, values stored with `group_key = None`. @@ -558,6 +1661,115 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread")] + async fn range_query_set_aggregator_merges_multiple_buckets_within_one_window() { + // Every other SetAggregator test uses window_size_ms == bucket + // width, so each window only ever contains exactly ONE bucket -- + // DeltaSetAggregator gets multi-bucket-per-window coverage "for + // free" via its always-widened [0,t] window, but SetAggregator's + // bounded sliding window has never been exercised with more than one + // bucket in play. Mirrors the values-side sliding-collision fixture + // (range_query_sliding_window_merges_both_buckets: two buckets + // sharing the same (start,end)) but for keys: two SetAggregator + // buckets both at t=1000 (start=0,end=1000) must UNION into + // {host-a, host-b}, exactly like the existing bucket_map handling + // for values (`bucket_map.entry(*start).or_default().push(..)`, + // #567/#570) -- a fix that naively does `.insert()` instead of + // accumulating for the keys side would silently drop one of them. + // + // t=2000 adds a third key (host-c) via a single non-colliding + // bucket; since SetAggregator is "latest window only" (not + // cumulative), it replaces the t=1000 pair entirely -- this also + // keeps the test genuinely RED against today's code (Bug 2: the + // single end-anchored keys fetch window [1000,2000] excludes the + // t=1000 collision pair's bucket, whose start=0 falls outside it). + let cms_1 = CountMinSketchAccumulator::new(2, 3); + let cms_2 = CountMinSketchAccumulator::new(2, 3); + + let mut keys_1_a = SetAggregatorAccumulator::new(); + keys_1_a.add_key(KeyByLabelValues { + labels: vec!["host-a".to_string(), "evt-1".to_string()], + }); + let mut keys_1_b = SetAggregatorAccumulator::new(); + keys_1_b.add_key(KeyByLabelValues { + labels: vec!["host-b".to_string(), "evt-1".to_string()], + }); + let mut keys_2 = SetAggregatorAccumulator::new(); + keys_2.add_key(KeyByLabelValues { + labels: vec!["host-c".to_string(), "evt-1".to_string()], + }); + + let engine = create_range_engine_dual_input( + "event_frequency", + AggregationType::CountMinSketch, + AggregationType::SetAggregator, + vec![], + vec!["host", "event"], + vec![ + (1000, None, Box::new(cms_1) as Box), + (2000, None, Box::new(cms_2) as Box), + ], + vec![ + // Both buckets share (start=0, end=1000): a genuine collision. + (1000, None, Box::new(keys_1_a) as Box), + (1000, None, Box::new(keys_1_b) as Box), + (2000, None, Box::new(keys_2) as Box), + ], + "count(event_frequency) by (host, event)", + ); + + let query = "count(event_frequency) by (host, event)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("range query failed"); + let elements = matrix_values(qr); + + assert_all_at( + &elements, + &[ + ( + &["host-a"], + 1000, + true, + "host-a's bucket collides with host-b's at (start=0,end=1000) -- \ + both must be merged (unioned), not one dropped in favor of the other", + ), + ( + &["host-b"], + 1000, + true, + "host-b's bucket collides with host-a's at (start=0,end=1000) -- \ + both must be merged (unioned), not one dropped in favor of the other", + ), + ( + &["host-c"], + 1000, + false, + "host-c's bucket doesn't exist until t=2000", + ), + ( + &["host-a"], + 2000, + false, + "SetAggregator is latest-window-only: host-c's window replaces \ + host-a/host-b's, it doesn't accumulate alongside them", + ), + ( + &["host-b"], + 2000, + false, + "SetAggregator is latest-window-only: host-c's window replaces \ + host-a/host-b's, it doesn't accumulate alongside them", + ), + ( + &["host-c"], + 2000, + true, + "host-c is the live set for the window ending at t=2000", + ), + ], + ); + } + #[tokio::test(flavor = "multi_thread")] async fn range_query_self_keyed_topk_expands_with_non_none_outer_key() { // Same bug as range_query_self_keyed_topk_expands_without_keys_query, diff --git a/docs/583-range-keys-per-step-design.md b/docs/583-range-keys-per-step-design.md new file mode 100644 index 00000000..6412456d --- /dev/null +++ b/docs/583-range-keys-per-step-design.md @@ -0,0 +1,275 @@ +# #583 range-query per-step keys — grilling session recap + +Design discussion for the fix to [#583](https://github.com/ProjectASAP/ASAPQuery/issues/583) +("Range query key expansion uses one snapshot instead of per-step keys"). +No fix code has been written yet — this session was scoped to (1) confirming +understanding of the bug, (2) working out the fix design question-by-question, +and (3) building out a RED test suite that pins every failure mode discussed. +Recorded here so the eventual implementation turn doesn't have to re-derive +any of this. + +## Bug recap + +`execute_range_query_pipeline` (`asap-query-engine/src/engines/simple_engine/mod.rs`) +fetches and merges `keys_query` **once**, anchored at the range's `end`, and +reuses that single snapshot for every output timestamp in the range. Two +distinct problems fall out of this: + +1. **Stale snapshot reused across steps** — a key that starts or stops + existing partway through the range gets phantom samples before it existed, + or is silently dropped even at timestamps where it legitimately existed. + Confirmed to also affect the binary-expr arm path + (`handle_binary_expr_range_promql` → `build_arm_range_context` → + `finish_range_context`), which the issue only flagged as "worth checking." +2. **`keys_query`'s window is instant-anchored, not range-aware** — + `create_keys_query_params` computes the keys window purely from + `end_timestamp`. `finish_range_context` widens `values_query` to + `[start-lookback, end]` but clones `keys_query` unchanged. For + `SetAggregator` ("latest window only") this drops labels that existed + earlier in the range but fell out of the final instant's window. + `DeltaSetAggregator`'s `[0, end]` window happens to already be correct + (it's a full replay-from-start aggregator), so this half of the bug is + `SetAggregator`-specific. + +Both collapse to one fix: fetch/merge keys **per output step**, scoped to +that step's own window, not once at `end`. + +## Q&A + +### Q1 — architecture: per-step store queries, or one fetch + per-step in-memory merge? + +**Answer: one fetch + per-step in-memory merge.** The values side of +`execute_range_query_pipeline` already solves this exact problem: one +`execute_store_query` fetches all raw buckets across the whole range once, +then the `current_time` loop does an in-memory windowed merge per step +(`bucket_map` lookup + `create_window_merger`). The keys side should mirror +this instead of issuing N separate store queries (one per output timestamp). + +### Q2 — how does `finish_range_context` compute each side's widened window? + +**Answer: generalize the existing `values_query` widening formula to +`keys_query`, with zero `AggregationType` branching in the fix.** + +`finish_range_context` already does, for values: +```rust +let lookback_ms = base_context.store_plan.values_query.end_timestamp + - base_context.store_plan.values_query.start_timestamp; +extended_store_plan.values_query.start_timestamp = start_ms.saturating_sub(lookback_ms); +extended_store_plan.values_query.end_timestamp = end_ms; +``` +It doesn't know or care *why* that window is that width. The same trick +works for `keys_query` and happens to handle both aggregation types +correctly for free: + +- `create_keys_query_params` already computed the instant keys window: + `[end-window_size, end]` for `SetAggregator`, `[0, end]` for + `DeltaSetAggregator`. +- `keys_lookback_ms = keys_query.end - keys_query.start` → `window_size` for + `SetAggregator`, `end_ms` (the range's end) for `DeltaSetAggregator`. +- Widen the same way: `start_ms.saturating_sub(keys_lookback_ms)`. For + `SetAggregator` that's a normal sliding window. For `DeltaSetAggregator`, + since `keys_lookback_ms == end_ms` and the per-step loop invariant is + `current_time <= end_ms`, `current_time.saturating_sub(keys_lookback_ms)` + saturates to `0` at **every** step — "replay from the beginning," for free, + provably (not by luck): the `while current_time <= end_ms` loop condition + guarantees `current_time <= end_ms` before every iteration body runs. + +This also means "always merge from t=0 for DeltaSetAggregator" (a specific +follow-up question raised) is already guaranteed by this derivation — no +separate `if key_agg_type == DeltaSetAggregator { start = 0 }` branch is +needed anywhere. + +### Q3 — recompute-from-scratch per step, or incremental carry-forward? + +**Answer: recompute-from-scratch first; optimize later if measured.** + +For `DeltaSetAggregator`, each step's window is `[0, t]`, growing every +step. Recomputing fresh each step (mirroring the existing values loop +exactly) costs `O(N²)` bucket-merges total across `N` output steps (step 1 +processes 1 bucket, step 2 processes 2, ..., step N processes N). An +incremental version — carry a running merged accumulator forward across the +loop, only folding in new buckets since the last step — gives the same +result (verified, see below) in `O(N)` total, but requires restructuring the +loop to persist per-key-group merger state across iterations. + +Decision: ship recompute-from-scratch (least new code, easiest to verify, +matches the existing values-loop pattern exactly). `SetAggregator`'s window +is bounded by `window_size` regardless of range length, so it never has this +blowup — only `DeltaSetAggregator` does. Leave a comment flagging the +`O(N²)` replay cost as a known ceiling; revisit only if it's an actual +measured problem. + +**Correctness check performed before trusting this:** `NaiveMerger.merge_all()` +folds buckets **pairwise, left-to-right** (`buckets[0].merge_with(buckets[1])`, +then that result `.merge_with(buckets[2])`, ...) rather than passing the +whole slice to `DeltaSetAggregatorAccumulator::merge_accumulators` in one +flat N-way call. This distinction matters: `merge_accumulators`'s "key in +both added and removed → cancel both" conflict logic is symmetric and +order-blind when given the whole batch at once, but `NaiveMerger`'s +sequential fold means each binary step only ever resolves conflicts between +"the running state so far" and "the next bucket" — which correctly threads +chronological order through, including for a key that toggles 3+ times +within one merge window. Hand-traced against a 5-step add/remove/add/remove/add +sequence and confirmed correct at every intermediate step, not just the +final one. Pinned as an explicit, readable regression test (see below) since +this is easy to miss if `WindowMerger`'s implementation ever changes. + +### Q4 — does the keys bucket_map walk need its own step increment? + +**Answer: yes.** The per-step window-building loop scans `bucket_map` in +increments of the tumbling bucket width (`t += tumbling_window_ms` in the +existing values loop). That `tumbling_window_ms` is the **value** +aggregation's bucket width. The **key** aggregation (`aggregation_id_for_key`) +can have a different `window_size_ms`. The fix needs a separate +`keys_tumbling_window_ms`, fetched from the key aggregation's own config +(the same way `create_keys_query_params` already does internally), used to +walk the keys `bucket_map` — reusing the value side's width would silently +skip every keys bucket whose start isn't a multiple of that width. + +### Q5 — does `do_merge` still matter for keys once per-step merging is in place? + +**Answer: no, drop it for the range path.** Today `fetch_and_merge_keys` +takes a `do_merge` flag derived from the **value** aggregation's window +(`create_store_query_plan`: `do_merge = range_ms > value_aggregation.window_size_ms`) +and reuses it wholesale for the keys merge. Once keys mirror the values loop +(Q1), `do_merge` becomes moot the same way it already is for values (the +values `current_time` loop never consults it — it always does the +bucket-map-window-merge dance regardless of window count). The range +pipeline's keys fetch should become a raw `execute_store_query(keys_params)` +call (no merge), replacing `fetch_and_merge_keys`, for the range path only — +the instant-query path is unaffected. + +Related: [#581](https://github.com/ProjectASAP/ASAPQuery/issues/581) +"Unify PromQL instant and range query fetch/merge paths" independently names +this exact gap (range path's "fresh `WindowMerger` re-merged from scratch +per step" vs instant's `do_merge` short-circuit) as one of three drift bugs +(#570, #582, #587) between the two paths. This fix is a partial step toward +#581, not a full closer. + +### Q6 — a group with keys but no value data anywhere: fatal, or skip? + +**Answer: non-fatal skip + loud warning, per (step, group).** Today, +`execute_range_query_pipeline` hard-fails the **entire** range query if any +group resolved from `merged_keys` has no matching entry in `all_data`: +`all_data.get(group_key).ok_or_else(|| "No value for key: ...")?`. This is a +one-time check against a single global `groups` list today. Once key +expansion is per-step, a group having zero value data isn't an anomaly — a +key can legitimately exist per the keys aggregation before/after the value +aggregation ever has data for it (e.g. ingestion boundaries). This should +downgrade from hard-error (poisoning every other group's results in the same +query) to skip-this-group + warn loudly. + +### Q7 — multiple independent groups (real `grouping_labels`): does per-step scoping leak across groups? + +Explored via a concrete example (region=us gains host-b mid-range, +region=eu's host-x never changes — does host-b leak into region=eu's +output?). **Finding: not a real distinct risk in today's code.** Store +query results already come back partitioned by `group_key` +(`MergedOutputsMap = HashMap, ...>`), and +`merge_precomputed_outputs` merges *within* each group_key's own buckets, +not across groups. The RED tests written for this (see below) confirmed it: +failures were always "this group's own key is phantom-early" (Bug 1), never +actual cross-group bleed. Kept as regression guards for once Bug 1 is fixed, +not because they proved a second bug. + +A related candidate — "a group that only appears partway through the +range" — was traced through by hand and found to **not** actually be RED +against today's code as originally described (a `DeltaSetAggregator`'s +`keys_query` already spans `[0, end]` today, so a brand-new group is picked +up correctly by coincidence when its only key never changes). It was +replaced with the Q6 scenario instead (a group with keys but zero value +data, which **does** genuinely hard-fail today). + +### Q8 — a toggle within a single output step's window, at the full-pipeline level + +Raised as "no existing test has two key-delta buckets landing inside the +same step's window" — **this claim was checked and found wrong**: since +`DeltaSetAggregator`'s window always starts at `0` (Q2), the oscillating +5-window test's own `t=2000` checkpoint already merges 2 toggle-buckets +(`[0,1000)` add, `[1000,2000)` remove) within one window. No new test needed +for that framing. + +Re-checking turned up a real, different gap: every `SetAggregator` test uses +`window_size_ms == bucket_width_ms`, so each window only ever contains +**exactly one** bucket. `DeltaSetAggregator` gets multi-bucket-per-window +coverage "for free" via its always-widened `[0,t]` window; `SetAggregator`'s +bounded sliding window has never been exercised merging more than one +bucket. Added `range_query_set_aggregator_merges_multiple_buckets_within_one_window`: +two `SetAggregator` buckets colliding at the same `(start=0,end=1000)` (same +construction as `range_query_sliding_window_merges_both_buckets`, applied to +keys instead of values) must union into `{host-a, host-b}`, guarding against +a per-step `bucket_map` implementation that does `.insert()` instead of +`.entry().or_default().push()` for the keys side and silently drops one of +the colliding buckets. + +## RED test inventory + +All in `asap-query-engine/src/tests/native_range_query_tests.rs` unless noted. +16 tests total: 5 green (regression guards / already-correct behavior), 11 RED +(pin the fix's target behavior). + +| Test | Pins | +|---|---| +| `range_query_dual_population_key_appearing_midrange_has_no_phantom_earlier_sample` | Bug 1, `DeltaSetAggregator`, plain range path | +| `range_query_set_aggregator_earlier_key_not_silently_dropped` | Bug 2, `SetAggregator`, plain range path | +| `range_query_binary_expr_arm_key_appearing_midrange_has_no_phantom_earlier_sample` | Bug 1 through `build_arm_range_context` (binary-expr scalar arm) | +| `range_query_binary_expr_arm_set_aggregator_earlier_key_not_silently_dropped` | Bug 2 through `build_arm_range_context` | +| `range_query_delta_set_aggregator_oscillating_add_remove_across_five_windows` | Multi-toggle replay correctness (add/remove ×5), asserted at every step | +| `range_query_delta_set_aggregator_key_change_on_middle_step_not_just_boundary` | 3-step range, change lands on the interior step, not a boundary | +| `range_query_delta_set_aggregator_key_bucket_width_differs_from_value_bucket_width` | Q4 — keys bucket_map must step by the key aggregation's own width | +| `range_query_dual_population_per_step_key_change_does_not_leak_across_groups` | Q7 — one group changes, one doesn't | +| `range_query_dual_population_simultaneous_cross_group_adds_stay_isolated` | Q7 — both groups change at the same timestamp (sharper signature) | +| `range_query_dual_population_group_with_no_value_data_is_skipped_not_fatal` | Q6 — orphaned group must not poison the whole query | +| `range_query_set_aggregator_merges_multiple_buckets_within_one_window` | Q8 — `SetAggregator` must union, not drop, colliding same-timestamp buckets within one window | +| `naive_merger_sequential_fold_replays_delta_set_toggles_at_every_window` (`window_merger.rs`, **GREEN**) | Explicit, readable pin of the NaiveMerger-sequential-fold requirement Q3 depends on; contrasts against a flat `merge_accumulators` call to make the distinction undeniable | + +`assert_all_at` (test-module helper) collects every mismatched +`(labels, timestamp, expected)` case into one panic message instead of +stopping at the first failing `assert!`, per explicit request — applied to +all multi-assertion tests in the file. + +`create_range_engine_dual_input_with_windows` (thin superset of +`create_range_engine_dual_input`) lets tests set different bucket widths for +the value vs. key aggregation configs, needed for the Q4 test; existing call +sites untouched. + +## Status + +The fix landed (PR [#595](https://github.com/ProjectASAP/ASAPQuery/pull/595), +draft), staged as planned: + +- Stage 1 — widened `keys_query`'s window in `finish_range_context` + (`keys_lookback_ms`/`keys_tumbling_window_ms` on + `RangeQueryExecutionContext`), behavior-inert on its own. +- Stage 2 — the actual per-step keys merge in + `execute_range_query_pipeline` (raw fetch, `KeysSource` enum, per-step + windowed merge, Q6's non-fatal group skip). +- Stage 3 — confirmed the binary-expr arm path needed no separate code + change, as predicted. +- Stage 4 — full regression pass, 0 failures. + +All 16 `native_range_query_tests` pass; full workspace suite (`cargo test +--workspace`) passes with 0 failures. + +Follow-up refactor (`widen_query_window`, dedupping the values/keys window +formula) and a broader duplication survey against the instant-query path +were also done post-fix. Two items came out of that survey as genuine +design decisions rather than mechanical refactors, and were filed as +separate issues rather than folded into #583: + +- [#596](https://github.com/ProjectASAP/ASAPQuery/issues/596) — range + queries never got the CMS/KLL batch-merge fast path the instant path has + (`NaiveMerger` only does the sequential fallback); the two fallback + implementations also differ in error-handling policy, so unifying them + isn't free. +- [#597](https://github.com/ProjectASAP/ASAPQuery/issues/597) — #583's Q6 + (skip a group with keys-but-no-value-data instead of hard-failing) was + only applied to the range path; the instant path + (`collect_results_separate_keys`) still hard-fails on the identical case. + +## Explicitly not done here + +- Incremental carry-forward merging for `DeltaSetAggregator` (Q3) — deferred + until proven necessary. +- Broader unification with #581 — out of scope for #583, but this fix moves + in that direction.