From 2716ae72f8db8d226c3a01c784befc72acad40c2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 24 May 2026 22:51:00 -0600 Subject: [PATCH] fix(data_plane): read delta-only sketch windows + count(HLL) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two warm-tier sketch-query bugs that made quantile_over_time / count(HLL) return empty `asap_query` results (silent "No result", never failing over to the archive): 1. Delta-stitching carry-in. The agent emits a periodic Full snapshot then many cheap Delta frames, so a short query window (e.g. `[30s]`) routinely contains ONLY deltas — the Full landed earlier, outside the window. `query_range`'s strict containment filter dropped that Full, the delta-apply reducer couldn't establish a rolling base, and the engine returned `Ok(empty)` instead of a value. `query_range` now splices in the most-recent Full ending before `start` as a carry-in base; the reducer drops the out-of-range base from per-window output. 2. count(HLL) distinct-count idiom. `count(metric)` lifts the outer count into both `outer_agg=Count` and the `CardinalityApprox` capability while the bare-selector inner leaves the trace function empty. The engine passed the empty function to the reducer (UnsupportedFunction) AND re-applied the Count fold (collapsing the estimate to row-count 1). The engine now derives the reducer family from the capability when the function string is empty, and suppresses the already-consumed Count fold so the HLL distinct-count is returned directly. Fixes the previously-failing controller_plan_to_query_full_roundtrip_hll e2e and adds index-, reducer-, analyzer-, and engine-level regression tests for both shapes. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/asap_tier_analysis.rs | 23 ++ .../query_engines/asap_query_engine/engine.rs | 380 +++++++++++++++++- .../sketch_db/index/epoch_columnar.rs | 50 +++ .../storage_engines/sketch_db/index/mod.rs | 149 +++++++ .../sketch_db/query/sketch_reducer.rs | 12 + 5 files changed, 610 insertions(+), 4 deletions(-) diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs index 787cda39c..16e951f56 100644 --- a/control_plane/src/asap_tier_analysis.rs +++ b/control_plane/src/asap_tier_analysis.rs @@ -825,6 +825,29 @@ mod tests { ); } + /// `count(metric)` is the distinct-count idiom: the analyzer lifts + /// the outer `count` into BOTH `outer_agg = Count` and the + /// `CardinalityApprox` capability, and the bare-selector inner leaves + /// `function` empty. The data-plane engine must (a) derive the + /// reducer function from the capability when `function` is empty, and + /// (b) NOT re-apply the outer `Count` fold (the cardinality estimate + /// IS the count). This test pins the analyzer-side shape those + /// engine fixes rely on. + #[test] + fn analyze_count_bare_metric_trace_shape() { + let a = analyze_promql_for_asap_tier("count(unique_users_per_min)"); + let c = &a.candidates[0]; + assert_eq!( + c.function, "", + "bare-selector inner leaves the trace function empty: {a:?}" + ); + assert!( + matches!(c.outer_agg, OuterAgg::Count(_)), + "outer count is lifted into outer_agg: {a:?}" + ); + assert_eq!(c.range_seconds, 0, "instant query: {a:?}"); + } + #[test] fn analyze_quantile_over_time() { let a = analyze_promql_for_asap_tier("quantile_over_time(0.99, http_latency_ms[5m])"); diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index b05536854..ce6beb9db 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -789,7 +789,7 @@ impl ASAPQueryEngine { _ => reducer .evaluate( &hit_sids, - &candidate.function, + effective_sketch_function(candidate), &candidate.function_args, start_ms, end_ms, @@ -808,7 +808,9 @@ impl ASAPQueryEngine { // the range-query path too (issue #296) — same identity // case + fold semantics as the instant-query trait // adapter above. - let result = if candidate.outer_agg.is_some() { + let result = if candidate.outer_agg.is_some() + && !outer_fold_already_consumed(candidate) + { apply_outer_agg_fold(result, &candidate.outer_agg) } else { result @@ -870,6 +872,67 @@ impl ASAPQueryEngine { /// /// Coverage is preserved from the inner result — the fold doesn't /// change which time-range the underlying sids covered. +/// Effective sketch reducer function-name for a candidate. +/// +/// `SketchReducer::evaluate` keys its query-family dispatch off a +/// function-NAME string. The analyzer's `trace.function` is usually that +/// name (`quantile_over_time`, `cardinality_estimate`, …), BUT for an +/// outer-aggregation idiom whose inner is a BARE selector — e.g. +/// `count(metric)` (the HLL distinct-count idiom) — `trace_from_promql` +/// unwraps the outer `count` into `outer_agg` and then walks the inner +/// bare selector, which carries no function name. The trace's `function` +/// is then EMPTY, and the reducer maps `""` → `UnsupportedFunction` → +/// the engine returns an empty/capability-miss result for a query the +/// warm tier can actually answer. +/// +/// When `function` is empty we fall back to a canonical name derived +/// from the analyzer's typed `required_capability` (the load-bearing +/// signal), so `count(hll_metric)` dispatches to the Cardinality family +/// and `quantile(...)` to the Quantile family even when the AST walk +/// couldn't recover a string. Non-empty function names pass through +/// unchanged so existing aliases keep their exact semantics. +fn effective_sketch_function( + candidate: &control_plane::asap_tier_analysis::ASAPTierCandidate, +) -> &str { + if !candidate.function.is_empty() { + return &candidate.function; + } + use crate::storage_engines::sketch_db::index::Capability; + match &candidate.required_capability { + Capability::QuantileApprox(_) => "quantile", + Capability::CardinalityApprox => "cardinality_estimate", + Capability::FrequencyTopk(_) => "topk", + Capability::FrequencyEstimate(_) => "frequency", + // ExactAgg never reaches the sketch `evaluate` path (handled by + // the ExactAgg dispatch branch), but return a benign default so + // a stray ExactAgg still surfaces as UnsupportedFunction rather + // than silently mis-dispatching. + Capability::ExactAgg(_) => "", + } +} + +/// Whether the analyzer's `outer_agg` should still be folded over the +/// reducer's result, or has already been CONSUMED by the +/// capability dispatch. +/// +/// `count(hll_metric)` is the distinct-count idiom: the analyzer lifts +/// the outer `count` into both `outer_agg = Count` AND +/// `required_capability = CardinalityApprox`. The HLL reducer answers +/// the distinct count directly (one cardinality scalar per window), so +/// re-applying the `Count` fold would collapse that estimate to the +/// row-count (`values.len()` → 1) — the wrong answer. Suppress the fold +/// in that case; the cardinality estimate IS the count. +fn outer_fold_already_consumed( + candidate: &control_plane::asap_tier_analysis::ASAPTierCandidate, +) -> bool { + use crate::storage_engines::sketch_db::index::Capability; + use control_plane::asap_tier_analysis::OuterAgg; + matches!( + (&candidate.required_capability, &candidate.outer_agg), + (Capability::CardinalityApprox, OuterAgg::Count(_)) + ) +} + fn apply_outer_agg_fold( inner: crate::storage_engines::sketch_db::query::ASAPTierResult, outer: &control_plane::asap_tier_analysis::OuterAgg, @@ -1415,7 +1478,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu ), _ => reducer.evaluate( &hit_sids, - &candidate.function, + effective_sketch_function(candidate), &candidate.function_args, t0_ms, now_ms, @@ -1498,7 +1561,9 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // a single-value group, returning the same value // unchanged. No special case needed; the general fold // handles it. - let result = if candidate.outer_agg.is_some() { + let result = if candidate.outer_agg.is_some() + && !outer_fold_already_consumed(candidate) + { apply_outer_agg_fold(result, &candidate.outer_agg) } else { result @@ -2447,6 +2512,313 @@ mod asap_tier_classify_tests { assert_eq!(by_zone.get("z3").copied(), Some(400.0)); } + // Build a now-anchored KLL `SketchInstanceMetadata` + sample so the + // engine's instant/range default lookbacks reach it. Mirrors the + // live MVP workload: the agent emits a bare-named KLL sketch + // (`http_requests_total_latency_ms`) into the SketchStore. + fn kll_meta(sid: u64, metric: &str) -> SketchInstanceMetadata { + let cfg = SketchConfig::Kll { k: 200 }; + SketchInstanceMetadata { + sid, + metric_name: metric.to_string(), + group_by_keys: BTreeSet::new(), + capability: Some(Capability::QuantileApprox(SketchKindHandle::Kll)), + agg_kind: crate::storage_engines::sketch_db::index::AggKind::Sketch { + kind: SketchKindHandle::Kll, + config: cfg.clone(), + spatial_filter_canonical: String::new()}, + accuracy: Some(AccuracyBound::from_config(&cfg)), + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET} + } + + fn encode_kll_items_proto(k: u16, items: &[f64]) -> Vec { + use asap_sketchlib::proto::sketchlib::{sketch_envelope, KllState, SketchEnvelope}; + use prost::Message; + let state = KllState { + k: k as u32, + items: items.to_vec(), + levels: vec![], + num_levels: 0, + ..Default::default() + }; + let env = SketchEnvelope { + sketch_state: Some(sketch_envelope::SketchState::Kll(state)), + ..Default::default() + }; + env.encode_to_vec() + } + + fn hll_meta(sid: u64, metric: &str) -> SketchInstanceMetadata { + let cfg = SketchConfig::Hll { precision: 10 }; + SketchInstanceMetadata { + sid, + metric_name: metric.to_string(), + group_by_keys: BTreeSet::new(), + capability: Some(Capability::CardinalityApprox), + agg_kind: crate::storage_engines::sketch_db::index::AggKind::Sketch { + kind: SketchKindHandle::Hll, + config: cfg.clone(), + spatial_filter_canonical: String::new()}, + accuracy: Some(AccuracyBound::from_config(&cfg)), + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET} + } + + fn encode_hll_with_cardinality(precision: u32, distinct: usize) -> Vec { + use asap_sketchlib::proto::sketchlib::{ + sketch_envelope, HllVariant as ProtoVariant, HyperLogLogState, SketchEnvelope, + }; + use asap_sketchlib::sketches::hll::{HllSketch, HllVariant}; + use prost::Message; + let mut sk = HllSketch::new(HllVariant::Regular, precision); + for i in 0..distinct { + sk.update(format!("user-{i}").as_bytes()); + } + let state = HyperLogLogState { + variant: ProtoVariant::Regular as i32, + precision: sk.precision, + registers: sk.registers.clone(), + hip_kxq0: sk.hip_kxq0, + hip_kxq1: sk.hip_kxq1, + hip_est: sk.hip_est}; + let env = SketchEnvelope { + sketch_state: Some(sketch_envelope::SketchState::Hll(state)), + ..Default::default() + }; + env.encode_to_vec() + } + + /// REGRESSION of the HLL `count(metric)` "No result" e2e failure + /// (`controller_plan_to_query_full_roundtrip_hll`) isolated to the + /// engine layer. `count(unique_users_per_min)` is the distinct-count + /// idiom: the analyzer lifts the outer `count` into `outer_agg=Count` + /// AND `required_capability=CardinalityApprox`, and the bare-selector + /// inner leaves the trace `function` EMPTY. Before the fix the engine + /// passed the empty function to the reducer (→ `UnsupportedFunction`) + /// AND re-applied the `Count` fold (→ row-count 1.0). The fix derives + /// the reducer family from the capability and suppresses the + /// already-consumed `Count` fold, so the HLL distinct-count is + /// returned directly. A single FULL HLL frame (~500 users) is used so + /// the instant projection reads the real estimate. + #[tokio::test] + async fn execute_count_hll_returns_cardinality_not_rowcount() { + let idx = Arc::new(SketchStore::new()); + let sid = 7500u64; + idx.register(hll_meta(sid, "unique_users_per_min")); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + idx.append_sample( + sid, + BTreeMap::new(), + (now_ms.saturating_sub(3_000), now_ms.saturating_sub(2_000)), + SketchSampleState { + bytes: encode_hll_with_cardinality(10, 500), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull}, + ); + + let engine = build_engine_with_index(idx); + let result = engine + .execute("count(unique_users_per_min)") + .await + .expect( + "count(hll_metric) must dispatch to the Cardinality family \ + via the candidate capability (empty trace function) and \ + return the HLL distinct-count, NOT capability-miss", + ); + assert!( + result_nonempty(&result), + "count(unique_users_per_min) over an HLL sid must return a \ + non-empty cardinality estimate (regression: empty `asap_query` \ + No-result)" + ); + // The value must be the HLL distinct-count estimate (~500), NOT + // the outer-Count fold collapsing it to the row-count (1.0). + let est = match &result { + crate::query_engines::query_result::QueryResult::Vector(v) => v.values[0].value, + crate::query_engines::query_result::QueryResult::Matrix(m) => { + m.values[0].samples.last().map(|s| s.value).unwrap_or(0.0) + } + }; + assert!( + est > 100.0, + "expected the HLL distinct-count estimate (~500), not the \ + row-count fold (1.0); got {est}" + ); + } + + /// REPRODUCTION (root-cause hunt): `quantile_over_time(0.99, + /// http_requests_total_latency_ms[30s])` end-to-end via + /// `execute(&str)` against a now-anchored KLL sid carrying real + /// sketch state. The window is inside the engine's `[now-30s, now]` + /// range. This pins the exact end-to-end behaviour the live deploy + /// shows ("No result" tagged `asap_query`) so we can see whether the + /// engine produces `Ok(populated)`, `Ok(empty)`, or `CapabilityMiss`. + #[tokio::test] + async fn execute_quantile_over_time_kll_now_anchored() { + let idx = Arc::new(SketchStore::new()); + let sid = 7100u64; + idx.register(kll_meta(sid, "http_requests_total_latency_ms")); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + // A 10s window ending 5s ago — comfortably inside the 30s range. + let window_start = now_ms.saturating_sub(15_000); + let window_end = now_ms.saturating_sub(5_000); + + let items: Vec = (1..=50).map(|i| i as f64).collect(); + let bytes = encode_kll_items_proto(200, &items); + idx.append_sample( + sid, + BTreeMap::new(), + (window_start, window_end), + SketchSampleState { + bytes, + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull}, + ); + + // The sid must classify as Hit (in-memory unsealed state counts). + assert_eq!( + idx.classify(sid), + crate::storage_engines::sketch_db::index::SidLookup::Hit, + "KLL sid with appended in-memory state must classify Hit" + ); + + let result = engine_quantile_result(idx, now_ms).await; + let nonempty = match result { + crate::query_engines::query_result::QueryResult::Vector(v) => !v.values.is_empty(), + crate::query_engines::query_result::QueryResult::Matrix(m) => { + m.values.iter().any(|s| !s.samples.is_empty()) + } + }; + assert!( + nonempty, + "quantile_over_time over a now-anchored KLL sid must return a \ + non-empty result (got empty → reproduces the live `asap_query` \ + + No-result bug)" + ); + } + + async fn engine_quantile_result( + idx: Arc, + _now_ms: u64, + ) -> crate::query_engines::query_result::QueryResult { + let engine = build_engine_with_index(idx); + engine + .execute("quantile_over_time(0.99, http_requests_total_latency_ms[30s])") + .await + .expect( + "quantile_over_time over a Hit KLL sid must NOT capability-miss \ + (if it does, the bug is upstream of the reducer)", + ) + } + + fn result_nonempty(r: &crate::query_engines::query_result::QueryResult) -> bool { + match r { + crate::query_engines::query_result::QueryResult::Vector(v) => !v.values.is_empty(), + crate::query_engines::query_result::QueryResult::Matrix(m) => { + m.values.iter().any(|s| !s.samples.is_empty()) + } + } + } + + /// REGRESSION (delta-stitching carry-in): the live agent emits a + /// periodic Full snapshot followed by many cheap Delta frames to + /// save bandwidth, so a short query window (`[30s]`) routinely + /// contains ONLY deltas — the Full landed earlier, outside the + /// window. Before the fix, `SketchStore::query_range`'s strict + /// containment filter (`w.0 >= start`) dropped the out-of-window + /// Full, the delta-apply reducer couldn't establish a rolling base, + /// and the engine returned `Ok(empty)` (NOT a capability-miss) — so + /// the router never failed over and the client saw "No result" + /// tagged `asap_query`. The fix splices in the most-recent Full + /// ending before `start` as a carry-in base. This test pins that: + /// a Full at now-60s + a Delta at now-10s with a `[30s]` window must + /// produce a NON-EMPTY answer. + #[tokio::test] + async fn quantile_over_time_kll_full_before_window_carries_in_base() { + let idx = Arc::new(SketchStore::new()); + let sid = 7400u64; + idx.register(kll_meta(sid, "http_requests_total_latency_ms")); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + let items: Vec = (1..=50).map(|i| i as f64).collect(); + // Full at now-60s..now-55s — OUTSIDE the 30s window. + idx.append_sample( + sid, + BTreeMap::new(), + (now_ms.saturating_sub(60_000), now_ms.saturating_sub(55_000)), + SketchSampleState { + bytes: encode_kll_items_proto(200, &items), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoFull}, + ); + // Delta at now-15s..now-5s — INSIDE the window. + idx.append_sample( + sid, + BTreeMap::new(), + (now_ms.saturating_sub(15_000), now_ms.saturating_sub(5_000)), + SketchSampleState { + bytes: encode_kll_items_proto(200, &items), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoDelta}, + ); + + let result = engine_quantile_result(idx, now_ms).await; + assert!( + result_nonempty(&result), + "quantile_over_time with a Full BEFORE the window + a Delta \ + inside it must carry in the Full as a base and return a \ + non-empty result (regression: returned empty `asap_query` \ + No-result)" + ); + } + + /// A delta-ONLY window with NO Full anywhere is a genuine data gap — + /// there is no base to stitch from. The carry-in fix does not (and + /// cannot) fabricate one, so the result is empty. Documents the + /// boundary so the carry-in change isn't mistaken for "always + /// non-empty"; a follow-up may convert this to a NoData → archive + /// failover. + #[tokio::test] + async fn quantile_over_time_kll_delta_only_no_base_is_empty() { + let idx = Arc::new(SketchStore::new()); + let sid = 7300u64; + idx.register(kll_meta(sid, "http_requests_total_latency_ms")); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let items: Vec = (1..=50).map(|i| i as f64).collect(); + idx.append_sample( + sid, + BTreeMap::new(), + (now_ms.saturating_sub(15_000), now_ms.saturating_sub(5_000)), + SketchSampleState { + bytes: encode_kll_items_proto(200, &items), + encoding: crate::storage_engines::sketch_db::index::SketchEncoding::ProtoDelta}, + ); + + let result = engine_quantile_result(idx, now_ms).await; + assert!( + !result_nonempty(&result), + "delta-only window with no Full base anywhere has nothing to \ + stitch from → empty result expected" + ); + } + /// `rate(http_requests_total[5m])` end-to-end via `execute(&str)`. /// The analyzer hands the engine `Capability::ExactAgg(Sum)` with /// `function="rate"` and `range_seconds=300`; the engine must diff --git a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs index a29634cc8..70fc7dfb6 100644 --- a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs +++ b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs @@ -256,6 +256,30 @@ impl

MutableEpoch

{ } } + /// 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 + /// before the query window — so a short query window that contains + /// only delta frames can still establish its rolling state. The + /// caller is responsible for picking the latest Full per label (the + /// columnar layer is payload-agnostic). O(M) linear scan. + pub fn collect_ending_at_or_before<'a>( + &'a self, + before: u64, + out: &mut Vec<(TimestampRange, LabelValuesId, &'a P)>, + ) { + if let Some(min_s) = self.min_start { + if min_s > before { + return; + } + } + for (i, w) in self.windows_col.iter().enumerate() { + if w.1 <= before { + out.push((*w, self.label_ids_col[i], &self.payloads_col[i])); + } + } + } + /// Total accumulated entries — caller compares against /// `epoch_capacity` to decide whether to seal + rotate. pub fn distinct_windows(&self) -> usize { @@ -435,6 +459,32 @@ impl

SealedEpoch

{ } } + /// 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 + /// window-START; since `w.1 >= w.0`, any entry with `w.1 <= before` + /// also has `w.0 <= before`, so we can stop the scan once + /// `w.0 > before`. O(log N + k). + pub fn collect_ending_at_or_before<'a>( + &'a self, + before: u64, + out: &mut Vec<(TimestampRange, LabelValuesId, &'a P)>, + ) { + if let Some(min_s) = self.min_start { + if min_s > before { + return; + } + } + for entry in &self.entries { + if entry.0 .0 > before { + break; + } + if entry.0 .1 <= before { + out.push((entry.0, entry.1, &entry.2)); + } + } + } + /// Exact-window query — O(log N + m). Binary search for the /// matching range; linear scan while the range matches. pub fn exact_query(&self, target: TimestampRange) -> Vec<(LabelValuesId, &P)> { diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index da439930d..891e93640 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -421,6 +421,90 @@ impl SketchStore { buf.clear(); } + // Delta-stitching carry-in (issue: quantile/HLL "No result" + // bug). The agent emits a periodic Full snapshot followed by + // many cheap Delta frames. A short query window (e.g. `[30s]`) + // routinely contains ONLY deltas — the Full landed earlier, + // outside `[start, end]`. The downstream delta-apply reducer + // can't establish a rolling base from a leading delta, so it + // silently produces an empty result that the engine returns as + // `Ok(empty)` (NOT a capability-miss), so the router never + // fails over and the caller sees "No result". To fix, for each + // label series whose earliest in-window sample is a Delta, + // splice in the most-recent Full snapshot ending at or before + // `start` as a carry-in base. Its window-end is `< start`, so + // it sorts first in the per-label `BTreeMap` and the reducer's + // cumulative/per-window walk uses it as the base; the reducer + // drops out-of-range output windows so the carry-in never leaks + // into the answer's time domain. + if start_unix_ms > 0 { + // Which labels need a base? Those present in-window whose + // earliest sample is a Delta (a leading Full needs nothing). + let need_base: Vec = by_label_id + .iter() + .filter(|(_, samples)| { + samples + .values() + .next() + .map(|s| { + matches!( + s.encoding, + SketchEncoding::ProtoDelta | SketchEncoding::MsgpackDelta + ) + }) + .unwrap_or(false) + }) + .map(|(label_id, _)| *label_id) + .collect(); + + if !need_base.is_empty() { + let before = start_unix_ms.saturating_sub(1); + // Track the latest Full per label (by window-end). + let mut latest_full: HashMap = + HashMap::new(); + let mut consider = |buf: &Vec<(TimestampRange, LabelValuesId, &AggPayload)>| { + for (win, label_id, payload) in buf { + if !need_base.contains(label_id) { + continue; + } + let Some(s) = payload.as_sketch() else { + continue; + }; + if !matches!( + s.encoding, + SketchEncoding::ProtoFull | SketchEncoding::MsgpackFull + ) { + continue; + } + let w_end = win.1 as i64; + match latest_full.get(label_id) { + Some((prev_end, _)) if *prev_end >= w_end => {} + _ => { + latest_full.insert(*label_id, (w_end, s.clone())); + } + } + } + }; + guard + .current_epoch + .collect_ending_at_or_before(before, &mut buf); + consider(&buf); + buf.clear(); + for sealed in guard.sealed_epochs.values() { + sealed.collect_ending_at_or_before(before, &mut buf); + consider(&buf); + buf.clear(); + } + for (label_id, (w_end, state)) in latest_full { + by_label_id + .entry(label_id) + .or_default() + .entry(w_end) + .or_insert(state); + } + } + } + by_label_id .into_iter() .map(|(label_id, samples)| { @@ -1377,6 +1461,71 @@ mod tests { assert!(s.samples.contains_key(&20)); } + fn delta_sample(b: u8) -> SketchSampleState { + SketchSampleState { + bytes: vec![b], + encoding: SketchEncoding::ProtoDelta, + } + } + + #[test] + fn range_query_carries_in_latest_full_before_window() { + // The delta-stitching carry-in: a Full lands BEFORE the query + // window and only deltas land inside it. `query_range` must + // splice in the most-recent pre-window Full so the downstream + // delta-apply reducer can establish a rolling base. Without it, + // a short window that contains only deltas yields an + // unanswerable series (the live quantile/HLL "No result" bug). + let idx = SketchStore::new(); + idx.register(meta(21)); + let lv = BTreeMap::new(); + // Two Fulls before the window; the LATER one (end=200) is the + // base that must be carried in. + idx.append_sample(21, lv.clone(), (90, 100), sample(1)); + idx.append_sample(21, lv.clone(), (190, 200), sample(2)); + // Delta-only inside the window [300, 400]. + idx.append_sample(21, lv.clone(), (310, 320), delta_sample(3)); + + let series = idx.query_range(21, 300, 400); + assert_eq!(series.len(), 1); + let s = &series[0]; + // In-window delta (end=320) + carried-in latest Full (end=200). + assert!(s.samples.contains_key(&320), "in-window delta present"); + assert!( + s.samples.contains_key(&200), + "latest pre-window Full (end=200) carried in as base" + ); + assert!( + !s.samples.contains_key(&100), + "only the LATEST pre-window Full is carried in, not older ones" + ); + // The carried-in entry must be a Full (the reducer needs a base). + assert_eq!(s.samples.get(&200).unwrap().encoding, SketchEncoding::ProtoFull); + } + + #[test] + fn range_query_no_carry_in_when_window_leads_with_full() { + // If the in-window samples already lead with a Full, no carry-in + // is needed (and none should be spliced — it would be redundant + // and could skew coverage). + let idx = SketchStore::new(); + idx.register(meta(22)); + let lv = BTreeMap::new(); + idx.append_sample(22, lv.clone(), (90, 100), sample(1)); + idx.append_sample(22, lv.clone(), (310, 320), sample(2)); // Full in-window + idx.append_sample(22, lv.clone(), (330, 340), delta_sample(3)); + + let series = idx.query_range(22, 300, 400); + assert_eq!(series.len(), 1); + let s = &series[0]; + assert!(s.samples.contains_key(&320)); + assert!(s.samples.contains_key(&340)); + assert!( + !s.samples.contains_key(&100), + "no carry-in when the window already leads with a Full" + ); + } + #[test] fn ddsketch_accuracy_bound() { let bound = AccuracyBound::from_config(&SketchConfig::DDSketch { diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs index c75f1e455..f68b31415 100644 --- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs +++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs @@ -536,7 +536,19 @@ impl<'a> SketchReducer<'a> { reason: e, } })?; + // Drop carry-in base windows: `SketchStore::query_range` + // may splice in a Full snapshot ending BEFORE `t0_ms` + // so the delta-apply walk can establish a rolling base + // for a delta-only window. That base must not surface + // as an output sample in the requested `[t0, t1]` + // range. Cumulative mode emits a single scalar at the + // latest in-window end so it's unaffected; per-window + // mode emits one sample per window, so filter here. + let lo = t0_ms as i64; per_win + .into_iter() + .filter(|(w_end, _)| *w_end >= lo) + .collect() }; out_series.push((ts.series_label_values, samples_out)); }