Skip to content

Range query scan_window steps by window_size_ms, not slide_interval_ms, for Sliding aggregations #600

Description

@milindsrivastava1997

execute_range_query_pipeline's per-step scan_window (mod.rs) steps the bucket_map by tumbling_window_ms/keys_tumbling_window_ms, both sourced from window_size_ms (finish_range_context, promql.rs). For WindowType::Sliding aggregations, buckets are actually persisted at slide_interval_ms granularity (precompute_engine/window_manager.rs:111, num_panes = window_size_ms / slide_interval_ms), not window_size_ms. When slide_interval_ms < window_size_ms, real buckets land on timestamps the scan never visits and are silently dropped.

Affects both the value side (pre-existing, predates #583/#587) and the keys side (introduced by #583, same root cause). Confirmed via 2 RED tests.

#[tokio::test(flavor = "multi_thread")]
async fn diagnostic_bug2_sliding_keys_bucket_off_window_size_grid() {
    // window_size_ms=2000, slide_interval_ms=1000. Delta bucket at
    // start=1000 (on the slide_interval_ms grid, NOT window_size_ms grid).
    let mut keys_add = DeltaSetAggregatorAccumulator::new();
    keys_add.add_key(KeyByLabelValues {
        labels: vec!["host-a".to_string(), "evt-1".to_string()],
    });

    let engine = create_range_engine_dual_input_sliding_keys(
        "event_frequency",
        AggregationType::CountMinSketch,
        AggregationType::DeltaSetAggregator,
        vec!["host", "event"],
        vec![(3000, None, Box::new(CountMinSketchAccumulator::new(2, 3)) as Box<dyn AggregateCore>)],
        vec![(1000, 2000, None, Box::new(keys_add) as Box<dyn AggregateCore>)],
        2000,
        1000,
        "count(event_frequency) by (host, event)",
    );

    let query = "count(event_frequency) by (host, event)";
    let result = engine.handle_range_query_promql(query.to_string(), 3.0, 3.5, 1.0);
    let (_, qr) = result.expect("range query failed");
    let elements = matrix_values(qr);

    assert!(
        key_has_sample_at(&elements, "host-a", 3000),
        "BUG CONFIRMED: host-a's delta bucket (start=1000, on the \
         slide_interval_ms=1000 grid but not the window_size_ms=2000 grid) \
         was NOT found -- scan_window is stepping the keys bucket_map by \
         window_size_ms instead of slide_interval_ms for this Sliding key \
         aggregation"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn diagnostic_bug2_sliding_values_bucket_off_window_size_grid() {
    // Same hypothesis, VALUE side. Sliding SumAccumulator aggregation,
    // window_size_ms=2000, slide_interval_ms=1000. Two buckets at start=0
    // (value=10.0) and start=1000 (value=5.0) -- both on the
    // slide_interval_ms grid, only one on the window_size_ms grid.
    let query = "sum_over_time(http_requests[1s])";
    let engine = create_engine_multi_timestamp_with_window(
        "http_requests",
        AggregationType::Sum,
        vec!["host"],
        vec![
            (1000, Some(vec!["host-a".to_string()]), Box::new(SumAccumulator::with_sum(10.0)) as Box<dyn AggregateCore>),
            (2000, Some(vec!["host-a".to_string()]), Box::new(SumAccumulator::with_sum(5.0)) as Box<dyn AggregateCore>),
        ],
        query,
        2000, // window_size_ms
        WindowType::Sliding,
    );
    // create_engine_multi_timestamp_with_window hardcodes slide_interval_ms:
    // 1000 internally, matching buckets at start=0 and start=1000.
    // step must be a multiple of window_size_ms (2000) or
    // validate_range_query_params rejects the query outright.
    let result = engine.handle_range_query_promql(query.to_string(), 2.0, 2.5, 2.0);
    let (_, qr) = result.expect("range query failed");
    let elements = matrix_values(qr);
    assert_eq!(elements.len(), 1, "expected one series for host-a");
    let value = elements[0].samples.first().map(|s| s.value);

    assert!(
        (value.unwrap_or(f64::NAN) - 15.0).abs() < 1e-9,
        "BUG CONFIRMED (pre-existing, value side): expected both buckets \
         (10.0 + 5.0 = 15.0) merged, got {:?} -- tumbling_window_ms is \
         stepping by window_size_ms=2000 instead of slide_interval_ms=1000",
        value
    );
}

Both FAIL against current code (keys test: bucket never found, elements empty; values test: merged value 10.0 instead of 15.0, second bucket silently dropped).

Metadata

Metadata

Labels

No labels
No labels

Type

No type

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions