From e03d0587b1e3eac2ff7de3684616296961dd3901 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 12 May 2026 20:58:48 -0600 Subject: [PATCH] feat(sketch_db): fold sketch_kind/sketch_config into agg_kind on SketchInstanceMetadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 5 M2.3.2 — replaces the M2-era `sketch_kind: SketchKindHandle` + `sketch_config: SketchConfig` field pair on `SketchInstanceMetadata` with a single `agg_kind: AggKind`. Same struct can now describe either a sketch-backed sid (`AggKind::Sketch { kind, config }`) or a precompute-backed sid (`AggKind::Precompute { agg_type, parameters_canonical }`) — the latter lights up in M2.3.3+. `capability: Option` and `accuracy: Option` because precomputes answer exact statistics and have no warm-tier approximation profile. The warm-tier reducer (sketch-only) handles a `None` capability as a defensive `UnsupportedCapability` rather than a panic. `sketch_kind()` and `sketch_config()` accessor methods return `Option<_>` so existing sketch-only consumers (`sketch_reducer.rs`) keep working with a thin `.expect("only handles sketches")` — upstream classification already filters precompute sids out of the reducer's caller list. 8 ingest + test construction sites updated. 798/800 lib tests pass (the 2 pre-existing `schema_timeline_dispatch_tests` failures predate this PR). Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/drivers/ingest/otel.rs | 10 ++-- .../query_engines/asap_query_engine/engine.rs | 19 ++++--- .../warm_tier/sketch_reducer.rs | 28 +++++++--- .../asap_query_engine/warm_tier/tests.rs | 52 +++++++++++-------- data_plane/src/stores/sketch_db/index/mod.rs | 48 ++++++++++++++--- 5 files changed, 111 insertions(+), 46 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 48d360dd..ee2223c2 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -934,10 +934,12 @@ async fn route_modified_otlp_sketches_to_precompute( sid, metric_name: metric.name.clone(), group_by_keys, - capability: cap, - sketch_kind: kind, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(cap), + agg_kind: crate::stores::sketch_db::index::AggKind::Sketch { + kind, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: ts_ms, retired_at_ms: None, expires_at_ms: None, 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 0b29f2d5..27150a8d 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -3564,8 +3564,13 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu let meta = match idx.instance(*sid) { Some(m) => m, None => continue}; - if required.is_satisfied_by(&meta.capability) { - hit_sids.push(*sid); + // Precompute-backed sids (M2.3) have `capability: None` + // — the analyzer doesn't route them through this path, + // but skip defensively if one slips in. + if let Some(cap) = meta.capability.as_ref() { + if required.is_satisfied_by(cap) { + hit_sids.push(*sid); + } } } if hit_sids.is_empty() { @@ -5946,10 +5951,12 @@ mod warm_tier_classify_tests { .iter() .map(|s| s.to_string()) .collect::>(), - capability: Capability::QuantileApprox(SketchKindHandle::DDSketch), - sketch_kind: SketchKindHandle::DDSketch, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(Capability::QuantileApprox(SketchKindHandle::DDSketch)), + agg_kind: crate::stores::sketch_db::index::AggKind::Sketch { + kind: SketchKindHandle::DDSketch, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None, diff --git a/data_plane/src/query_engines/asap_query_engine/warm_tier/sketch_reducer.rs b/data_plane/src/query_engines/asap_query_engine/warm_tier/sketch_reducer.rs index eb3d04cb..0876d8dc 100644 --- a/data_plane/src/query_engines/asap_query_engine/warm_tier/sketch_reducer.rs +++ b/data_plane/src/query_engines/asap_query_engine/warm_tier/sketch_reducer.rs @@ -252,7 +252,18 @@ impl<'a> SketchReducer<'a> { family: QueryFamily, meta: &SketchInstanceMetadata, ) -> Result { - match (family, &meta.capability) { + // The warm-tier reducer only ever runs on sketch-backed sids + // (the analyzer's `instances_matching` filters on `Capability`, + // which is `None` for precompute-backed sids). A `None` here + // means upstream classification broke — surface as a missing + // capability rather than panicking the request path. + let Some(cap) = meta.capability.as_ref() else { + return Err(WarmTierError::UnsupportedCapability { + function: function_name.to_string(), + capability: Capability::CardinalityApprox, + }); + }; + match (family, cap) { (QueryFamily::Quantile, Capability::QuantileApprox(_)) | (QueryFamily::Cardinality, Capability::CardinalityApprox) | (QueryFamily::FrequencyTopk, Capability::FrequencyTopk(_)) @@ -262,7 +273,7 @@ impl<'a> SketchReducer<'a> { // sketch matrix, so the underlying CMS / CountSketch matrix can // be queried point-wise without consulting it. | (QueryFamily::FrequencyEstimate, Capability::FrequencyTopk(_)) => { - Ok(meta.capability.clone()) + Ok(cap.clone()) } (_, other) => Err(WarmTierError::UnsupportedCapability { function: function_name.to_string(), @@ -355,7 +366,7 @@ impl<'a> SketchReducer<'a> { if w > cov_hi { cov_hi = w; } - let total = decode_frequency_total(sid, meta.sketch_kind, state)?; + let total = decode_frequency_total(sid, meta.sketch_kind().expect("warm-tier reducer only handles sketch-backed sids"), state)?; samples_out.push((*w_end, total)); } out_series.push((ts.series_label_values, samples_out)); @@ -388,7 +399,7 @@ impl<'a> SketchReducer<'a> { if w_end_u64 > cov_hi { cov_hi = w_end_u64; } - let cms_heap = match meta.sketch_kind { + let cms_heap = match meta.sketch_kind().expect("warm-tier reducer only handles sketch-backed sids") { SketchKindHandle::CmsWithHeap | SketchKindHandle::CountSketchWithHeap => { // Both heap-bearing variants serialize the // outer `CountMinSketchWithHeap` envelope via @@ -409,7 +420,7 @@ impl<'a> SketchReducer<'a> { // routes through QueryFamily::FrequencyEstimate). return Err(WarmTierError::MissingHeap { sid, - sketch_kind: meta.sketch_kind, + sketch_kind: meta.sketch_kind().expect("warm-tier reducer only handles sketch-backed sids"), }); } other => { @@ -436,14 +447,17 @@ impl<'a> SketchReducer<'a> { } // Quantile / Cardinality with delta stitching. - let delta_kind = match (family, meta.sketch_kind) { + let delta_kind = match (family, meta.sketch_kind().expect("warm-tier reducer only handles sketch-backed sids")) { (QueryFamily::Quantile, SketchKindHandle::DDSketch) => DeltaSketchKind::DDSketch, (QueryFamily::Quantile, SketchKindHandle::Kll) => DeltaSketchKind::Kll, (QueryFamily::Cardinality, SketchKindHandle::Hll) => DeltaSketchKind::Hll, _ => { return Err(WarmTierError::UnsupportedCapability { function: function_name.to_string(), - capability: meta.capability.clone(), + capability: meta + .capability + .clone() + .unwrap_or(Capability::CardinalityApprox), }); } }; diff --git a/data_plane/src/query_engines/asap_query_engine/warm_tier/tests.rs b/data_plane/src/query_engines/asap_query_engine/warm_tier/tests.rs index e022dade..4d764747 100644 --- a/data_plane/src/query_engines/asap_query_engine/warm_tier/tests.rs +++ b/data_plane/src/query_engines/asap_query_engine/warm_tier/tests.rs @@ -17,7 +17,7 @@ use asap_sketchlib::sketches::hll::{HllSketch, HllVariant}; use crate::query_engines::asap_query_engine::warm_tier::{SketchReducer, WarmTierError}; use crate::stores::sketch_db::index::{ - AccuracyBound, Capability, SketchConfig, SketchEncoding, SketchIndex, SketchInstanceMetadata, + AccuracyBound, AggKind, Capability, SketchConfig, SketchEncoding, SketchIndex, SketchInstanceMetadata, SketchKindHandle, SketchSampleState, }; @@ -104,10 +104,12 @@ fn dd_meta(sid: u64) -> SketchInstanceMetadata { sid, metric_name: "http_latency_ms".to_string(), group_by_keys: BTreeSet::new(), - capability: Capability::QuantileApprox(SketchKindHandle::DDSketch), - sketch_kind: SketchKindHandle::DDSketch, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(Capability::QuantileApprox(SketchKindHandle::DDSketch)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::DDSketch, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None, @@ -120,10 +122,12 @@ fn kll_meta(sid: u64, k: u32) -> SketchInstanceMetadata { sid, metric_name: "http_latency_ms".to_string(), group_by_keys: BTreeSet::new(), - capability: Capability::QuantileApprox(SketchKindHandle::Kll), - sketch_kind: SketchKindHandle::Kll, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(Capability::QuantileApprox(SketchKindHandle::Kll)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::Kll, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None, @@ -136,10 +140,12 @@ fn hll_meta(sid: u64, precision: u32) -> SketchInstanceMetadata { sid, metric_name: "uniq_users".to_string(), group_by_keys: BTreeSet::new(), - capability: Capability::CardinalityApprox, - sketch_kind: SketchKindHandle::Hll, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(Capability::CardinalityApprox), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::Hll, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None, @@ -459,10 +465,12 @@ fn cms_heap_meta(sid: u64) -> SketchInstanceMetadata { sid, metric_name: "endpoint_hits".to_string(), group_by_keys: BTreeSet::new(), - capability: Capability::FrequencyTopk(SketchKindHandle::CmsWithHeap), - sketch_kind: SketchKindHandle::CmsWithHeap, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(Capability::FrequencyTopk(SketchKindHandle::CmsWithHeap)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::CmsWithHeap, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None, @@ -475,10 +483,12 @@ fn cms_only_meta(sid: u64) -> SketchInstanceMetadata { sid, metric_name: "endpoint_hits".to_string(), group_by_keys: BTreeSet::new(), - capability: Capability::FrequencyTopk(SketchKindHandle::CountMin), - sketch_kind: SketchKindHandle::CountMin, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(Capability::FrequencyTopk(SketchKindHandle::CountMin)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::CountMin, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None, diff --git a/data_plane/src/stores/sketch_db/index/mod.rs b/data_plane/src/stores/sketch_db/index/mod.rs index 59c80285..2e4aa0e4 100644 --- a/data_plane/src/stores/sketch_db/index/mod.rs +++ b/data_plane/src/stores/sketch_db/index/mod.rs @@ -327,10 +327,21 @@ pub struct SketchInstanceMetadata { /// The group-by KEY set — `dp.attributes.keys()` after the agent's /// `AggregateBy` rollup folded other labels into the sketch state. pub group_by_keys: BTreeSet, - pub capability: Capability, - pub sketch_kind: SketchKindHandle, - pub sketch_config: SketchConfig, - pub accuracy: AccuracyBound, + /// Warm-tier capability surfaced to the analyzer. For sketch-backed + /// instances this is one of the `*Approx` variants; for precompute- + /// backed instances (M2.3+) it's `None` because precomputes answer + /// exact statistics — the analyzer routes them via `agg_kind` / + /// `agg_type` instead. + pub capability: Option, + /// M2.3 — the canonical "what kind of aggregation lives at this + /// sid" descriptor. Replaces the M2-era `sketch_kind` + + /// `sketch_config` field pair so a single registry can host both + /// sketches and partial-accumulator (Sum/Count/Avg/Rate/MinMax) + /// state. + pub agg_kind: AggKind, + /// Approximate accuracy bound — `Some` for sketch-backed sids, + /// `None` for exact precomputes. + pub accuracy: Option, pub first_seen_unix_ms: i64, /// Wall-clock millis when the sid was retired (removed from the @@ -374,6 +385,25 @@ impl SketchInstanceMetadata { self.retired_at_ms = Some(now); self.expires_at_ms = Some(now + retention.as_millis() as u64); } + + /// Sketch-handle accessor for the legacy sketch path. Returns + /// `Some(handle)` iff this sid is sketch-backed; `None` for + /// precompute-backed sids. Consumers that only meaningfully run on + /// sketches (e.g. the warm-tier reducer) `.expect` it. + pub fn sketch_kind(&self) -> Option { + match &self.agg_kind { + AggKind::Sketch { kind, .. } => Some(*kind), + AggKind::Precompute { .. } => None, + } + } + + /// Sketch-config accessor mirroring [`Self::sketch_kind`]. + pub fn sketch_config(&self) -> Option<&SketchConfig> { + match &self.agg_kind { + AggKind::Sketch { config, .. } => Some(config), + AggKind::Precompute { .. } => None, + } + } } /// Per-sample sketch state. Stored as the payload column inside the @@ -672,10 +702,12 @@ mod tests { sid, metric_name: "m".into(), group_by_keys: BTreeSet::new(), - capability: Capability::QuantileApprox(SketchKindHandle::DDSketch), - sketch_kind: SketchKindHandle::DDSketch, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), + capability: Some(Capability::QuantileApprox(SketchKindHandle::DDSketch)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::DDSketch, + config: cfg.clone(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), first_seen_unix_ms: 0, retired_at_ms: None, expires_at_ms: None,