Skip to content

build_bucket_map has no chronological sort; DeltaSetAggregator merge order-sensitive for 3+ same-timestamp collisions #601

Description

@milindsrivastava1997

build_bucket_map (mod.rs) pushes buckets sharing a start-timestamp into a Vec in store-return order, no sort:

for ((start, _), bucket) in buckets {
    bucket_map.entry(*start).or_default().push(bucket.as_ref());
}

NaiveMerger's sequential fold is order-sensitive for DeltaSetAggregatorAccumulator (proven in window_merger.rs's naive_merger_sequential_fold_replays_delta_set_toggles_at_every_window). A 2-bucket add+remove collision is order-independent (conflict-cancellation is symmetric for 2), so it takes 3+ colliding deltas on the same key to actually expose this. Confirmed: same 3 logical deltas (add, remove, add), 3 different insertion orders, 2 orders agree and the 3rd gives a different final answer — proves insertion order alone changes the result.

/// Builds a DeltaSetAggregatorAccumulator "add host-a" delta.
fn diag_add_host_a() -> DeltaSetAggregatorAccumulator {
    let mut acc = DeltaSetAggregatorAccumulator::new();
    acc.add_key(KeyByLabelValues {
        labels: vec!["host-a".to_string(), "evt-1".to_string()],
    });
    acc
}

/// Builds a DeltaSetAggregatorAccumulator "remove host-a" delta.
fn diag_remove_host_a() -> DeltaSetAggregatorAccumulator {
    let mut acc = DeltaSetAggregatorAccumulator::new();
    acc.remove_key(KeyByLabelValues {
        labels: vec!["host-a".to_string(), "evt-1".to_string()],
    });
    acc
}

/// Inserts 3 DeltaSetAggregatorAccumulator deltas -- add, remove, add --
/// all colliding at the EXACT SAME (start=0,end=1000), in the given
/// insertion order, and reports whether host-a is present at t=1000.
async fn diag_bug6_run(insertion_order: [DeltaSetAggregatorAccumulator; 3]) -> bool {
    let engine = create_range_engine_dual_input(
        "event_frequency",
        AggregationType::CountMinSketch,
        AggregationType::DeltaSetAggregator,
        vec![],
        vec!["host", "event"],
        vec![(
            1000,
            None,
            Box::new(CountMinSketchAccumulator::new(2, 3)) as Box<dyn AggregateCore>,
        )],
        insertion_order
            .into_iter()
            .map(|acc| (1000, None, Box::new(acc) as Box<dyn AggregateCore>))
            .collect(),
        "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 (_, qr) = result.expect("range query failed");
    let elements = matrix_values(qr);
    key_has_sample_at(&elements, "host-a", 1000)
}

#[tokio::test(flavor = "multi_thread")]
async fn diagnostic_bug6_delta_set_same_timestamp_three_way_collision() {
    // Same 3 logical deltas (add, remove, add) for host-a, all at
    // (start=0,end=1000), inserted into the store in several different
    // orders. If the result varies by insertion order alone, that's
    // decisive proof build_bucket_map's missing sort matters.
    let order_a =
        diag_bug6_run([diag_add_host_a(), diag_remove_host_a(), diag_add_host_a()]).await;
    let order_b =
        diag_bug6_run([diag_remove_host_a(), diag_add_host_a(), diag_add_host_a()]).await;
    let order_c =
        diag_bug6_run([diag_add_host_a(), diag_add_host_a(), diag_remove_host_a()]).await;

    // Not asserting a specific expected value since chronological order is
    // ambiguous for same-timestamp buckets to begin with -- asserting only
    // that the result is DETERMINISTIC regardless of insertion order.
    assert_eq!(
        order_a, order_b,
        "result differs between insertion orders [add,remove,add] and \
         [remove,add,add] for the SAME 3 logical deltas -- proves \
         build_bucket_map's missing chronological sort affects the \
         final result"
    );
    assert_eq!(
        order_a, order_c,
        "result differs between insertion orders [add,remove,add] and \
         [add,add,remove] for the SAME 3 logical deltas -- proves \
         build_bucket_map's missing chronological sort affects the \
         final result"
    );
}

FAILS against current code: [add,remove,add]=true, [remove,add,add]=true, [add,add,remove]=false -- same 3 deltas, different insertion order, different answer.

Likely fix: sort same-start-timestamp buckets by some deterministic tiebreaker (a sequence number, precise sub-timestamp, or similar) before folding, rather than relying on store-return order.

Metadata

Metadata

Assignees

No one assigned

    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