Skip to content

feat(sketch_db): fold sketch_kind/sketch_config into agg_kind #154

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 13, 2026
Merged

feat(sketch_db): fold sketch_kind/sketch_config into agg_kind #154

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
10 changes: 6 additions & 4 deletions data_plane/src/drivers/ingest/otel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
19 changes: 13 additions & 6 deletions data_plane/src/query_engines/asap_query_engine/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -5946,10 +5951,12 @@ mod warm_tier_classify_tests {
.iter()
.map(|s| s.to_string())
.collect::<BTreeSet<_>>(),
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,
Expand Down
28 changes: 21 additions & 7 deletions data_plane/src/query_engines/asap_query_engine/warm_tier/sketch_reducer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,18 @@ impl<'a> SketchReducer<'a> {
family: QueryFamily,
meta: &SketchInstanceMetadata,
) -> Result<Capability, WarmTierError> {
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(_))
Expand All @@ -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(),
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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
Expand All @@ -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 => {
Expand All @@ -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),
});
}
};
Expand Down
52 changes: 31 additions & 21 deletions data_plane/src/query_engines/asap_query_engine/warm_tier/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};

Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down
48 changes: 40 additions & 8 deletions data_plane/src/stores/sketch_db/index/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>,
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<Capability>,
/// 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<AccuracyBound>,
pub first_seen_unix_ms: i64,

/// Wall-clock millis when the sid was retired (removed from the
Expand Down Expand Up @@ -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<SketchKindHandle> {
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
Expand Down Expand Up @@ -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,
Expand Down