fix(precompute): bulk single-timestamp ingest silently drops rows when wall-clock fallback closes the window mid-ingest

[akanksha-akkihal]

Summary

When ingesting a large one-shot batch where many rows share the same event-time (e.g. netflow data at 5M rows/s, all packed into one timestamp-second), the precompute engine force-closes the tumbling window ~6s into ingest via the wall-clock fallback. Rows that arrive after that point hit the "evicted pane" path and, because late_data_policy is hardcoded to Drop, are silently discarded.

The result is a uniform undercount across all group keys. Queries read a closed but incomplete window from the store.

Observed behavior

5M rows/s netflow run (5M rows, all in window [13:10:00, 13:10:01)), sketchdb vs ClickHouse baseline:

Query	baseline	sketchdb	error
COUNT(pkt_len) per srcip	500,000	300,800	39.84% low (all 10 srcips)
SUM(pkt_len) per srcip	~453M	~272M	~39.8% low
COUNT(DISTINCT dstip) per srcip	~35,400	~25,700	~27% low

• 300,800 / 500,000 = 0.6016 → exactly ~60% of rows retained, ~40% dropped.
• Error is uniform across every srcip (rows are interleaved, so each key loses the same fraction).

Root cause

All rows share one event-time, so the event-time watermark never advances. The window only closes via the wall-clock fallback in worker.rs, which fires after window_size_ms + wall_clock_grace_period_ms (default 1000 + 5000 = 6000 ms) of wall-clock time:

// asap-query-engine/src/precompute_engine/worker.rs:603
let window_size_ms = state.window_manager.window_size_ms();
for (&pane_start, &pane_birth_ms) in &state.pane_wall_clock_starts_ms {
    if now_ms.saturating_sub(pane_birth_ms) >= window_size_ms + grace_ms {
        let force_to = pane_start.saturating_add(window_size_ms);
        if force_to > effective_wm {
            effective_wm = force_to;
        }
    }
}

Ingesting 5M rows takes longer than 6s, so the window closes mid-ingest. Subsequent rows find their pane already evicted and are dropped:

// asap-query-engine/src/precompute_engine/worker.rs:430
match late_data_policy {
    LateDataPolicy::Drop => {
        debug!(
            "Dropping late sample for evicted pane [{}, {})",
            pane_start, pane_end
        );
        continue;
    }
    ...
}

late_data_policy is hardcoded to Drop and cannot be configured from the engine config:

// asap-query-engine/src/main.rs:210
pass_raw_samples: false,
raw_mode_aggregation_id: 0,
late_data_policy: LateDataPolicy::Drop,

The ForwardToStore variant already exists and would merge late rows into the store at query time, but there's no way to select it:

// asap-query-engine/src/precompute_engine/config.rs:5
pub enum LateDataPolicy {
    /// Drop late samples that arrive after their window has closed.
    Drop,
    /// Forward late samples to the store to be merged with existing window data.
    ForwardToStore,
}