Skip to content

fix(query-engine): judge a Sliding range step's cover by distinct windows #847

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
milindsrivastava1997 merged 2 commits into from
Oct 13, 2026
Merged

fix(query-engine): judge a Sliding range step's cover by distinct windows #847

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ae2e496
fix(query-engine): judge a Sliding range step's cover by distinct win…
milindsrivastava1997 Oct 13, 2026
e2e0477
fix(query-engine): check Sliding covers before cloning
milindsrivastava1997 Oct 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 64 additions & 65 deletions asap-query-engine/src/engines/simple_engine/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2342,11 +2342,9 @@ impl SimpleEngine {
}

/// 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.
/// `[window_start, window_end)`. Sliding walks by its stored-window
/// stride; Tumbling walks by `step_increment`. An incomplete Sliding
/// cover is rejected before accumulator sketches are cloned.
///
/// Cost is proportional to `(window_end - window_start) / step_increment`
/// -- the nominal window width -- not to how many buckets actually exist
Expand All @@ -2356,32 +2354,59 @@ impl SimpleEngine {
/// `[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(
fn window_buckets_for_step(
bucket_map: &BucketMap,
window_start: u64,
window_end: u64,
step_increment: u64,
) -> Vec<Box<dyn AggregateCore>> {
window_type: WindowType,
stored_window_size_ms: u64,
expected_sliding_window_count: Option<usize>,
) -> Option<Vec<Box<dyn AggregateCore>>> {
// 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.
let grid_step_ms = if window_type == WindowType::Sliding {
stored_window_size_ms
} else {
step_increment
};
assert!(
step_increment > 0,
"sum_window: step_increment must be nonzero, or this loop never terminates"
grid_step_ms > 0,
"window_buckets_for_step: grid step must be nonzero"
);
let mut window_buckets: Vec<Box<dyn AggregateCore>> = Vec::new();
assert_eq!(
expected_sliding_window_count.is_some(),
window_type == WindowType::Sliding,
"only Sliding windows have a completeness requirement"
);

let mut selected_buckets = Vec::new();
let mut covered_window_count = 0;
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()));
if let Some(window_buckets) = bucket_map.get(&t) {
if !window_buckets.is_empty() {
covered_window_count += 1;
selected_buckets.extend(window_buckets.iter());
}
}
t += step_increment;
t += grid_step_ms;
}
if expected_sliding_window_count.is_some_and(|expected| covered_window_count < expected) {
return None;
}
window_buckets

Some(
selected_buckets
.into_iter()
.map(|bucket| bucket.clone_boxed_core())
.collect(),
)
}

/// Collects every bucket in `bucket_map` with a start timestamp strictly
Expand Down Expand Up @@ -2444,30 +2469,6 @@ impl SimpleEngine {
.collect()
}

/// Picks how a step's window is composed from `bucket_map`: Sliding walks
/// the non-overlapping stored-window stride `W`; Tumbling walks the
/// bucket grid stride. Used identically for values and SetAggregator
/// keys. Walking Sliding by its slide `S` would double-count overlapping
/// full-window aggregates (#608/#621).
///
/// 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: &BucketMap,
window_start: u64,
window_end: u64,
step_increment: u64,
window_type: WindowType,
stored_window_size_ms: u64,
) -> Vec<Box<dyn AggregateCore>> {
if window_type == WindowType::Sliding {
Self::sum_window(bucket_map, window_start, window_end, stored_window_size_ms)
} else {
Self::sum_window(bucket_map, window_start, window_end, step_increment)
}
}

/// Execute the range query pipeline.
///
/// `enable_topk_limiting`/`enable_topk_formatting` mirror
Expand Down Expand Up @@ -2987,28 +2988,28 @@ impl SimpleEngine {
};
let keys_window_start =
keys_window_end.saturating_sub(*keys_lookback_ms);
Self::window_buckets_for_step(
let expected_sliding_window_count =
(*keys_window_type == WindowType::Sliding).then_some(
(*keys_lookback_ms / *keys_stored_window_size_ms) as usize,
);
let Some(keys_window_buckets) = Self::window_buckets_for_step(
keys_bucket_map,
keys_window_start,
keys_window_end,
*keys_tumbling_window_ms,
*keys_window_type,
*keys_stored_window_size_ms,
)
expected_sliding_window_count,
) else {
debug!(
"Skipping incomplete Sliding key cover at t={}",
current_time
);
continue;
};
keys_window_buckets
};

if *keys_window_type == WindowType::Sliding {
let expected =
(*keys_lookback_ms / *keys_stored_window_size_ms) as usize;
if keys_window_buckets.len() < expected {
debug!(
"Skipping incomplete Sliding key cover at t={}",
current_time
);
continue;
}
}

if keys_window_buckets.is_empty() {
debug!(
"No keys data in window at t={} — skipping this step for this group",
Expand Down Expand Up @@ -3043,14 +3044,23 @@ impl SimpleEngine {
};
let window_start = window_end.saturating_sub(lookback_ms);

let window_buckets = Self::window_buckets_for_step(
let expected_sliding_window_count = (window_type == WindowType::Sliding)
.then_some((lookback_ms / context.window_size_ms) as usize);
let Some(window_buckets) = Self::window_buckets_for_step(
bucket_map,
window_start,
window_end,
tumbling_window_ms,
window_type,
context.window_size_ms,
);
expected_sliding_window_count,
) else {
debug!(
"Skipping incomplete Sliding value cover at t={}",
current_time
);
continue;
};

trace!(
current_time,
Expand All @@ -3063,17 +3073,6 @@ impl SimpleEngine {
"Composed query output window from stored buckets"
);

if window_type == WindowType::Sliding {
let expected = (lookback_ms / context.window_size_ms) as usize;
if window_buckets.len() < expected {
debug!(
"Skipping incomplete Sliding value cover at t={}",
current_time
);
continue;
}
}

if window_buckets.is_empty() {
// No data at all for this window - skip sample
debug!(
Expand Down
178 changes: 178 additions & 0 deletions asap-query-engine/src/tests/window_semantics_consistency_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -331,6 +331,184 @@ mod tests {
);
}

/// Two workers can each emit a bucket for the same (key, window), so a
/// range step's Sliding cover must be judged by distinct window starts,
/// not bucket count: two buckets for [3s, 6s) must not stand in for the
/// missing [0s, 3s).
#[tokio::test(flavor = "multi_thread")]
async fn sliding_range_step_with_duplicate_bucket_and_missing_window_is_skipped() {
let query = "sum_over_time(cpu_load[6s])";
let streaming_config = Arc::new(StreamingConfig {
aggregation_configs: HashMap::from([(
1,
AggregationConfig {
aggregation_id: 1,
aggregation_type: AggregationType::Sum,
aggregation_sub_type: String::new(),
parameters: HashMap::new(),
grouping_labels: KeyByLabelNames::new(vec!["host".to_string()]),
aggregated_labels: KeyByLabelNames::empty(),
rollup_labels: KeyByLabelNames::empty(),
original_yaml: String::new(),
window_size_ms: 3_000,
slide_interval_ms: 1_000,
window_type: WindowType::Sliding,
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 store = Arc::new(SimpleMapStore::new(
streaming_config.clone(),
CleanupPolicy::NoCleanup,
));
let host_a = Some(KeyByLabelValues {
labels: vec!["host-a".to_string()],
});
for (start, value) in [
(3_000, 10.0),
(3_000, 10.0),

@milindsrivastava1997 milindsrivastava1997 Oct 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are these partials or duplicates? The test inserts two identical (3_000, 10.0) buckets for the same (key, window) and calls one of them a duplicate, but the engine merges every bucket at a start as a per-worker partial. If a step with a complete cover included [3s,6s) (e.g. a step at 9s with [6s,9s) present), it would sum to 20 for that window. That is right if these are two workers' partials and double-counting if this is a real duplicate (re-emission or retry). The PR fixes completeness only. Please state the invariant (one bucket per worker per window, never a re-emitted copy), or rename the test and use different partial values so it doesn't model a true duplicate.

(1_000, 100.0),
(4_000, 1_000.0),
] {
store
.insert_precomputed_output(
PrecomputedOutput::new(start, start + 3_000, host_a.clone(), 1),
Box::new(SumAccumulator::with_sum(value)),
)
.unwrap();
}
let inference_config = InferenceConfig {
schema: SchemaConfig::PromQL(PromQLSchema::new().add_metric(
"cpu_load".to_string(),
KeyByLabelNames::new(vec!["host".to_string()]),
)),
query_configs: vec![QueryConfig::new(query.to_string())
.add_aggregation(AggregationReference::new(1, None))],
cleanup_policy: CleanupPolicy::NoCleanup,
};
let engine = SimpleEngine::new(
store,
inference_config,
streaming_config,
1_000,
QueryLanguage::promql,
);

let result = engine
.handle_range_query_promql(query.to_string(), 6.0, 7.0, 1.0)
.expect("native query execution should not fail")
.expect("the step at 7s has a complete cover");

assert_eq!(
host_a_samples(&matrix_values(result.1)),
vec![(7_000, 1_100.0)]
);
}

/// The same cover rule applies to per-step keys: worker partials at one
/// window start cannot make a missing key window look complete.
#[tokio::test(flavor = "multi_thread")]
async fn sliding_key_cover_with_duplicate_bucket_and_missing_window_is_skipped() {
let query = "sum by (host) (count_over_time(event_frequency[6s]))";
let config = |aggregation_id, aggregation_type, aggregated_labels| AggregationConfig {
aggregation_id,
aggregation_type,
aggregation_sub_type: String::new(),
parameters: HashMap::new(),
grouping_labels: KeyByLabelNames::empty(),
aggregated_labels,
rollup_labels: KeyByLabelNames::empty(),
original_yaml: String::new(),
window_size_ms: 3_000,
slide_interval_ms: 1_000,
window_type: WindowType::Sliding,
spatial_filter: String::new(),
spatial_filter_normalized: String::new(),
metric: "event_frequency".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: HashMap::from([
(
1,
config(1, AggregationType::CountMinSketch, KeyByLabelNames::empty()),
),
(
2,
config(
2,
AggregationType::SetAggregator,
KeyByLabelNames::new(vec!["host".to_string()]),
),
),
]),
});
let store = Arc::new(SimpleMapStore::new(
streaming_config.clone(),
CleanupPolicy::NoCleanup,
));
let host_a = KeyByLabelValues {
labels: vec!["host-a".to_string()],
};

// Values fully cover both steps. Keys have two worker partials at
// [3s, 6s), no [0s, 3s) bucket, then complete coverage at 7s.
for start in [0, 1_000, 3_000, 4_000] {
let mut values = CountMinSketchAccumulator::new(4, 128);
values.inner.update(&host_a.to_semicolon_str(), 1.0);
store
.insert_precomputed_output(
PrecomputedOutput::new(start, start + 3_000, None, 1),
Box::new(values),
)
.unwrap();
}
for start in [3_000, 3_000, 1_000, 4_000] {
let mut keys = SetAggregatorAccumulator::new();
keys.add_key(host_a.clone());
store
.insert_precomputed_output(
PrecomputedOutput::new(start, start + 3_000, None, 2),
Box::new(keys),
)
.unwrap();
}

let inference_config = InferenceConfig {
schema: SchemaConfig::PromQL(PromQLSchema::new().add_metric(
"event_frequency".to_string(),
KeyByLabelNames::new(vec!["host".to_string()]),
)),
query_configs: vec![QueryConfig::new(query.to_string())
.add_aggregation(AggregationReference::new(1, None))
.add_aggregation(AggregationReference::new(2, None))],
cleanup_policy: CleanupPolicy::NoCleanup,
};
let engine = SimpleEngine::new(
store,
inference_config,
streaming_config,
1_000,
QueryLanguage::promql,
);

let result = engine
.handle_range_query_promql(query.to_string(), 6.0, 7.0, 1.0)
.expect("native query execution should not fail")
.expect("the step at 7s has a complete key cover");

assert_eq!(host_a_samples(&matrix_values(result.1)), vec![(7_000, 2.0)]);
}

// ════════════════════════════════════════════════════════════════════
// 2. Tumbling: instant and range paths must AGREE, and both must
// correctly SUM every disjoint bucket in the query's window (proving
Expand Down
Loading