Skip to content

fix(data_plane): overlap-scan sketch reads for short/instant windows #326

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
zzylol merged 1 commit into from
May 25, 2026
Merged

fix(data_plane): overlap-scan sketch reads for short/instant windows #326

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
138 changes: 138 additions & 0 deletions data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,45 @@ impl<P> MutableEpoch<P> {
}
}

/// Range query with HALF-OPEN OVERLAP semantics: include any window
/// whose `[w.0, w.1)` intersects `[start, end)` (i.e. `w.1 > start &&
/// w.0 < end`). Sister of [`Self::range_query_into`], which uses
/// strict CONTAINMENT (`w.0 >= start && w.1 <= end`).
///
/// The sketch read path ([`SketchStore::query_range`]) needs overlap,
/// not containment: the agent emits ~30s tumbling panes, and a short
/// query window (e.g. `quantile_over_time(...[30s])`, or an instant
/// query whose freshest pane straddles `now`) routinely fails to FULLY
/// CONTAIN any single pane — the pane that covers the window's edge
/// starts before `start` or ends after `end`. Containment then returns
/// ZERO in-window samples, so the delta-stitching carry-in never fires
/// (it keys off the earliest in-window sample) and the reducer yields
/// an empty series → the live "No result for query" bug for `[30s]`
/// and bare-instant selectors. Overlap admits the straddling pane; the
/// reducer's per-window `w_end >= t0` filter and cumulative `latest_end`
/// projection already drop windows that fall outside the requested
/// `[t0, t1]` time domain, so no out-of-range value leaks into the
/// answer. This matches the overlap filter [`Self::range_query_into_grouped`]
/// already uses (and the legacy `SketchStore`'s 30s-pane fix).
pub fn range_query_overlap_into<'a>(
&'a self,
start: u64,
end: u64,
out: &mut Vec<(TimestampRange, LabelValuesId, &'a P)>,
) {
// O(1) skip if the epoch's bounds don't overlap the query range.
if let (Some(min_s), Some(max_e)) = (self.min_start, self.max_end) {
if min_s >= end || max_e <= start {
return;
}
}
for (i, w) in self.windows_col.iter().enumerate() {
if w.1 > start && w.0 < end {
out.push((*w, self.label_ids_col[i], &self.payloads_col[i]));
}
}
}

/// Push every entry whose window-END is at or before `before` into
/// `out`. Used by the sketch read path to fetch a delta-stitching
/// "carry-in base" — the most-recent Full snapshot that landed
Expand Down Expand Up @@ -459,6 +498,43 @@ impl<P> SealedEpoch<P> {
}
}

/// Range query with HALF-OPEN OVERLAP semantics: include any window
/// whose `[w.0, w.1)` intersects `[start, end)` (`w.1 > start && w.0 <
/// end`). Sealed sister of [`MutableEpoch::range_query_overlap_into`] —
/// see its doc for why the sketch read path needs overlap rather than
/// the containment that [`Self::range_query_into`] applies.
///
/// Entries are sorted by window-START, but an overlapping window may
/// START before `start` (it straddles the left edge), so the
/// `partition_point(|e| e.0.0 < start)` seek the containment variant
/// uses would WRONGLY skip it. We instead scan from the front while
/// `w.0 < end`, testing `w.1 > start` per entry. O(k) over the prefix
/// of windows that start before `end` — bounded by the epoch size and
/// fine for the warm read path (epochs are small; nothing is sealed in
/// the live single-epoch deployment anyway).
pub fn range_query_overlap_into<'a>(
&'a self,
start: u64,
end: u64,
out: &mut Vec<(TimestampRange, LabelValuesId, &'a P)>,
) {
if let (Some(min_s), Some(max_e)) = (self.min_start, self.max_end) {
if min_s >= end || max_e <= start {
return;
}
}
for entry in &self.entries {
// Sorted by start; once a window starts at/after `end` it (and
// every later one) cannot overlap `[start, end)`.
if entry.0 .0 >= end {
break;
}
if entry.0 .1 > start {
out.push((entry.0, entry.1, &entry.2));
}
}
}

/// Push every entry whose window-END is at or before `before` into
/// `out`. Sealed sister of
/// [`MutableEpoch::collect_ending_at_or_before`]. Entries sort by
Expand Down Expand Up @@ -693,6 +769,38 @@ mod tests {
assert_eq!(ids2, vec![2, 3]); // (0,100) is below; (200,300) is above
}

#[test]
fn mutable_epoch_overlap_vs_containment() {
let mut e = MutableEpoch::<u32>::new();
e.insert((0, 100), 1, 100);
e.insert((100, 200), 2, 200);
e.insert((200, 300), 3, 300);

// Containment `[50, 250]`: only (100,200) is fully inside.
let mut buf = Vec::new();
e.range_query_into(50, 250, &mut buf);
let mut ids: Vec<_> = buf.iter().map(|(_, id, _)| *id).collect();
ids.sort();
assert_eq!(ids, vec![2]);

// Overlap `[50, 250)`: (0,100) straddles left, (100,200) inside,
// (200,300) straddles right → all three intersect.
buf.clear();
e.range_query_overlap_into(50, 250, &mut buf);
let mut ids: Vec<_> = buf.iter().map(|(_, id, _)| *id).collect();
ids.sort();
assert_eq!(ids, vec![1, 2, 3]);

// Half-open boundary: a window ending exactly at `start` does NOT
// overlap; one starting exactly at `end` does NOT either.
buf.clear();
e.range_query_overlap_into(100, 200, &mut buf); // start=100, end=200
let mut ids: Vec<_> = buf.iter().map(|(_, id, _)| *id).collect();
ids.sort();
// (0,100) ends at start → out; (100,200) overlaps; (200,300) starts at end → out.
assert_eq!(ids, vec![2]);
}

#[test]
fn sealed_epoch_binary_search_range() {
let mut m = MutableEpoch::<u32>::new();
Expand All @@ -709,6 +817,36 @@ mod tests {
assert!(payloads.contains(&300));
}

#[test]
fn sealed_epoch_overlap_admits_straddling_windows() {
let mut m = MutableEpoch::<u32>::new();
m.insert((0, 10), 1, 100);
m.insert((10, 20), 2, 200);
m.insert((20, 30), 3, 300);
m.insert((30, 40), 4, 400);
let s = SealedEpoch::from_mutable(m);

// Overlap `[5, 25)`: (0,10) straddles left edge — a window the
// start-keyed binary search of the containment scan would skip.
let mut buf = Vec::new();
s.range_query_overlap_into(5, 25, &mut buf);
let mut payloads: Vec<u32> = buf.iter().map(|(_, _, p)| **p).collect();
payloads.sort();
// (0,10) overlaps (10>5), (10,20) inside, (20,30) overlaps (20<25);
// (30,40) is entirely after.
assert_eq!(payloads, vec![100, 200, 300]);

// Half-open: window ending at `start` excluded; starting at `end`
// excluded.
buf.clear();
s.range_query_overlap_into(10, 30, &mut buf); // [10,30)
let mut payloads: Vec<u32> = buf.iter().map(|(_, _, p)| **p).collect();
payloads.sort();
// (0,10) ends at 10=start → out; (10,20) & (20,30) overlap;
// (30,40) starts at 30=end → out.
assert_eq!(payloads, vec![200, 300]);
}

#[test]
fn sid_store_rotation() {
let mut s = SidStoreData::<String, String>::new();
Expand Down
78 changes: 73 additions & 5 deletions data_plane/src/storage_engines/sketch_db/index/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -392,9 +392,19 @@ impl SketchStore {
HashMap::new();

let mut buf: Vec<(TimestampRange, LabelValuesId, &AggPayload)> = Vec::new();
// Sketch read path uses HALF-OPEN OVERLAP, not containment: the
// agent emits ~30s tumbling panes, so a short query window (a
// `[30s]` range, or an instant query whose freshest pane straddles
// `now`) can't FULLY CONTAIN any pane. Containment then returns
// zero in-window samples → the carry-in never fires and the
// reducer yields an empty series (the live "No result" bug for
// `[30s]` + bare-instant selectors). The reducer's own
// `w_end >= t0` / `latest_end` filters keep out-of-range values
// from leaking into the answer. See
// `MutableEpoch::range_query_overlap_into`.
guard
.current_epoch
.range_query_into(start_unix_ms, end_unix_ms, &mut buf);
.range_query_overlap_into(start_unix_ms, end_unix_ms, &mut buf);
for (win, label_id, payload) in &buf {
// Filter to sketch-variant payloads only; precompute sids
// (M2.3) are served via the precompute query path
Expand All @@ -409,7 +419,7 @@ impl SketchStore {
buf.clear();

for sealed in guard.sealed_epochs.values() {
sealed.range_query_into(start_unix_ms, end_unix_ms, &mut buf);
sealed.range_query_overlap_into(start_unix_ms, end_unix_ms, &mut buf);
for (win, label_id, payload) in &buf {
if let Some(s) = payload.as_sketch() {
by_label_id
Expand Down Expand Up @@ -1445,22 +1455,80 @@ mod tests {
}

#[test]
fn range_query_clips_to_window_bounds() {
fn range_query_uses_overlap_not_containment() {
// The sketch read path uses HALF-OPEN OVERLAP, not containment:
// any pane intersecting `[start, end)` is returned so the reducer
// can establish a rolling base even when no pane is fully
// contained (the ~30s-pane case behind the live `[30s]` "No
// result" bug). Of `(0,10)`, `(10,20)`, `(20,30)` against `[5,25)`:
// - `(0,10)` overlaps (10 > 5) → included (straddles left edge)
// - `(10,20)` overlaps → included
// - `(20,30)` overlaps (20 < 25) → included (straddles right edge)
let idx = SketchStore::new();
idx.register(meta(13));
let lv = BTreeMap::new();
idx.append_sample(13, lv.clone(), (0, 10), sample(1));
idx.append_sample(13, lv.clone(), (10, 20), sample(2));
idx.append_sample(13, lv.clone(), (20, 30), sample(3));

// Only the middle window is fully within [5, 25].
let series = idx.query_range(13, 5, 25);
assert_eq!(series.len(), 1);
let s = &series[0];
assert_eq!(s.samples.len(), 1);
assert_eq!(s.samples.len(), 3, "all three panes overlap [5,25)");
assert!(s.samples.contains_key(&10));
assert!(s.samples.contains_key(&20));
assert!(s.samples.contains_key(&30));
}

#[test]
fn range_query_excludes_non_overlapping_panes() {
// Overlap must still EXCLUDE panes that don't intersect the
// window — a pane ending exactly at `start` (half-open: `w.1 >
// start` is false) and one starting at/after `end`.
let idx = SketchStore::new();
idx.register(meta(14));
let lv = BTreeMap::new();
idx.append_sample(14, lv.clone(), (0, 10), sample(1)); // ends at start=10 → excluded
idx.append_sample(14, lv.clone(), (10, 20), sample(2)); // overlaps → included
idx.append_sample(14, lv.clone(), (30, 40), sample(3)); // starts at end=30 → excluded

let series = idx.query_range(14, 10, 30);
assert_eq!(series.len(), 1);
let s = &series[0];
assert_eq!(s.samples.len(), 1, "only the (10,20) pane overlaps [10,30)");
assert!(s.samples.contains_key(&20));
}

#[test]
fn range_query_short_window_straddling_pane_with_delta_carry_in() {
// Live gap 1: a `[30s]`-style window narrower than the agent's
// ~30s pane cadence. The freshest pane STRADDLES the window's left
// edge (starts before `start`, ends inside), so strict containment
// returned nothing → "No result". Overlap admits the straddling
// delta pane, and the carry-in splices the prior Full as its base.
let idx = SketchStore::new();
idx.register(meta(15));
let lv = BTreeMap::new();
// Full pane fully before the window.
idx.append_sample(15, lv.clone(), (260, 290), sample(1));
// Delta pane STRADDLING the window's left edge [300,330): starts
// at 295 (< 300), ends at 325 (inside). Containment excludes it
// (295 < 300); overlap includes it.
idx.append_sample(15, lv.clone(), (295, 325), delta_sample(2));

let series = idx.query_range(15, 300, 330);
assert_eq!(series.len(), 1, "straddling delta pane is now visible");
let s = &series[0];
assert!(
s.samples.contains_key(&325),
"straddling in-window delta (end=325) admitted by overlap"
);
assert!(
s.samples.contains_key(&290),
"prior Full (end=290) carried in as the delta's base"
);
}

fn delta_sample(b: u8) -> SketchSampleState {
SketchSampleState {
bytes: vec![b],
Expand Down
Loading