From 787aeacbd280920ca8c80c46460f699260a77090 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 13 May 2026 07:58:28 -0600 Subject: [PATCH] refactor(sketch_db): extract data/ module for payload taxonomy Post-M2.3 reorg #1 of 8. Moves stateless payload types out of `store/mod.rs` into a new `sketch_db::data` module. The split clarifies that: - `data/` = stateless data taxonomy (payload kinds, sid hashing, accuracy derivation, capability re-exports) - `store/` = stateful sid registry + per-sid columnar substrate Moved to `sketch_db::data`: - `AggKind` enum (sketch-vs-precompute discriminator) - `AggPayload` enum (sketch bytes vs accumulator) - `SketchConfig` (per-variant tuning params) - `SketchSampleState`, `SketchEncoding` - `SketchTimeSeries` (read-side row shape) - `AccuracyBound` + `from_config` derivation - `compute_sid` / `compute_sketch_sid` (canonical sid hash) - `canonical_parameters` helper - Re-exports: `Capability`, `SketchKindHandle`, `AggregationType` `store/mod.rs` re-exports each type at the legacy path (`sketch_db::store::AggKind` etc.) so no external caller needs to change spelling. The canonical home is now `sketch_db::data::*`. 783/783 lib tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/storage_engines/sketch_db/data/mod.rs | 397 ++++++++++++++++++ .../src/storage_engines/sketch_db/mod.rs | 1 + .../storage_engines/sketch_db/store/mod.rs | 380 +---------------- 3 files changed, 411 insertions(+), 367 deletions(-) create mode 100644 data_plane/src/storage_engines/sketch_db/data/mod.rs diff --git a/data_plane/src/storage_engines/sketch_db/data/mod.rs b/data_plane/src/storage_engines/sketch_db/data/mod.rs new file mode 100644 index 00000000..d279c301 --- /dev/null +++ b/data_plane/src/storage_engines/sketch_db/data/mod.rs @@ -0,0 +1,397 @@ +//! Sketch-DB data taxonomy — payload types, the sketch-vs-precompute +//! kind discriminator, the canonical sid hash, and the accuracy-bound +//! derivation. +//! +//! This module is intentionally **stateless**: every type here is a +//! plain data carrier or a pure function. The `index/` module (sid +//! registry + per-sid columnar substrate) and `lifecycle/` module +//! (Active/Retired/Expired) layer state on top of these. +//! +//! ## What's here +//! +//! - [`AggKind`] — `Sketch { kind, config } | Precompute { agg_type, +//! parameters_canonical }`. The discriminator that decides what +//! shape a sid's payload takes. +//! - [`AggPayload`] — `Sketch(SketchSampleState) | Precompute(Box)`. The actual byte/accumulator stored at each +//! (sid, window, label_values) cell. +//! - [`SketchConfig`] — variant-specific tuning parameters +//! (DDSketch's α, KLL's k, HLL's precision, etc.). Part of +//! `AggKind::Sketch`. +//! - [`SketchSampleState`] — `bytes + encoding` carried by sketch +//! payloads. +//! - [`SketchEncoding`] — wire-format hint (PROTO / PROTO_DELTA / +//! MSGPACK / MSGPACK_DELTA). +//! - [`SketchTimeSeries`] — one materialized series row returned by +//! the read path (sid + label values + per-window samples). +//! - [`AccuracyBound`] — `(epsilon, confidence)` derived from a +//! `SketchConfig`. Surfaces in HTTP response headers. +//! - [`compute_sid`] / [`compute_sketch_sid`] — the canonical sid +//! hash. Deterministic across hosts; defines the sid identity. +//! - [`canonical_parameters`] — helper that renders a parameters +//! `HashMap` into the canonical string form +//! `AggKind::Precompute::parameters_canonical` expects. +//! +//! ## Re-exports for callers +//! +//! - [`Capability`] / [`SketchKindHandle`] — the canonical +//! controller-side capability vocabulary. +//! - [`AggregationType`] — the agg-type enum that +//! `AggKind::Precompute` carries. + +use xxhash_rust::xxh64::xxh64; + +// ── Capability re-exports ──────────────────────────────────────────────────── +// +// Step 2a consolidated all capability state into +// `controller::sketch_algebra::capability`. The backend re-exports the +// canonical types so there's exactly one definition in the codebase. +// `is_satisfied_by` (used by the engine warm-tier hook) lives on the +// controller-side `Capability` impl. + +pub use controller::sketch_algebra::{Capability, SketchKindHandle}; + +/// Re-export so callers don't need to depend on promql_utilities +/// directly for the agg_type tag. +pub use promql_utilities::query_logics::enums::AggregationType; + +// ── Payload taxonomy ──────────────────────────────────────────────────────── + +/// Sketch-instance configuration carried per-Metric on the OTLP wire +/// (Phase 2 lifted these from per-DP up to the parent sketch container). +/// Backend reads the relevant variant at ingest time and stores it in +/// `SketchInstanceMetadata.agg_kind`. +#[derive(Debug, Clone)] +pub enum SketchConfig { + DDSketch { relative_accuracy: f64 }, + Kll { k: u32 }, + Hll { precision: u32 }, + CountSketch { rows: i32, cols: i32 }, + CountMin { rows: i32, cols: i32 }, +} + +/// What kind of aggregation a `sid` identifies. M2.3 generalization +/// — sids cover BOTH opaque-sketch state and partial-accumulator +/// (Sum / Count / Avg / Rate / MinMax) state, so a single store can +/// host both. +/// +/// The sid hash distinguishes the two branches structurally: a +/// `Sketch` sid is content-addressed over `(metric, attrs, +/// sketch_kind, sketch_config)`; a `Precompute` sid is content- +/// addressed over `(metric, attrs, agg_type, parameters)`. +#[derive(Debug, Clone)] +pub enum AggKind { + /// Opaque sketch state — DDSketch, KLL, HLL, CountMin, CountSketch. + /// Payload at storage layer is an encoded byte string + /// (`SketchSampleState`). + Sketch { + kind: SketchKindHandle, + config: SketchConfig, + }, + /// Partial-accumulator state — Sum, Count, Avg, Rate, MinMax, + /// SetAggregator, etc. Payload at storage layer is whatever the + /// per-accumulator serializer produces. + Precompute { + agg_type: AggregationType, + /// Stable canonical encoding of the agg's + /// `parameters: HashMap` — sorted keys, each + /// value rendered via serde_json. Held as a single owned + /// string so equality / hashing stay cheap and independent of + /// the original HashMap's iteration order. + parameters_canonical: String, + }, +} + +/// Render a `HashMap` of parameters into the canonical +/// string form `AggKind::Precompute::parameters_canonical` expects. +/// Keys sorted lexicographically; each value via `serde_json`. +pub fn canonical_parameters( + parameters: &std::collections::HashMap, +) -> String { + let sorted: std::collections::BTreeMap<&String, &serde_json::Value> = + parameters.iter().collect(); + let mut buf = String::new(); + for (k, v) in sorted { + buf.push_str(k); + buf.push('='); + buf.push_str(&serde_json::to_string(v).unwrap_or_default()); + buf.push(';'); + } + buf +} + +// ── sid hash ──────────────────────────────────────────────────────────────── + +/// Compute a deterministic `series_id` (sid) for one sketch instance. +/// +/// Folds the four DataPoint inputs the backend has at ingest time — +/// `metric_name`, `attrs` (keys + values, canonicalized), `sketch_kind`, +/// and `sketch_config` — into a 64-bit xxhash. Same inputs always +/// produce the same sid across restarts and across hosts. +/// +/// `attrs_fingerprint` MUST be the canonical fingerprint string +/// (`canonical_attrs_fingerprint` — keys sorted, joined `k=v;`); the +/// hash is sensitive to whitespace, ordering, and trailing separator. +/// +/// sid=0 is reserved on the wire (means "unresolved"); if a real input +/// hashes to 0 (vanishingly unlikely with 64-bit xxhash), we perturb +/// to 1. +pub fn compute_sketch_sid( + metric_name: &str, + attrs_fingerprint: &str, + sketch_kind: SketchKindHandle, + sketch_config: &SketchConfig, +) -> u64 { + compute_sid( + metric_name, + attrs_fingerprint, + &AggKind::Sketch { + kind: sketch_kind, + config: sketch_config.clone(), + }, + ) +} + +/// Generalized version of [`compute_sketch_sid`] covering both sketch +/// and precompute aggregations. Same `(metric, attrs, agg_kind)` tuple +/// always yields the same sid. +/// +/// The two branches encode disjointly: a `Sketch` payload starts with +/// `sketch_kind_tag` (1..=7), while a `Precompute` payload starts with +/// the byte `b'P'` (ASCII 80), which no sketch tag will ever produce. +/// So an attacker (or a colliding hash input) can't force a sketch sid +/// to overlap a precompute sid at the encoding level. +/// +/// Critically, the `Sketch` branch is bit-identical to the historical +/// `compute_sketch_sid` byte layout — existing sketch sids the M2 +/// wire format introduced stay stable across this generalization. +pub fn compute_sid( + metric_name: &str, + attrs_fingerprint: &str, + agg_kind: &AggKind, +) -> u64 { + let mut buf: Vec = + Vec::with_capacity(metric_name.len() + attrs_fingerprint.len() + 32); + buf.extend_from_slice(metric_name.as_bytes()); + buf.push(0); + buf.extend_from_slice(attrs_fingerprint.as_bytes()); + buf.push(0); + match agg_kind { + AggKind::Sketch { kind, config } => { + buf.push(sketch_kind_tag(*kind)); + buf.push(0); + encode_sketch_config(config, &mut buf); + } + AggKind::Precompute { + agg_type, + parameters_canonical, + } => { + buf.push(b'P'); + buf.push(0); + buf.extend_from_slice(agg_type.as_str().as_bytes()); + buf.push(0); + buf.extend_from_slice(parameters_canonical.as_bytes()); + } + } + let h = xxh64(&buf, 0); + if h == 0 { + 1 + } else { + h + } +} + +fn sketch_kind_tag(k: SketchKindHandle) -> u8 { + match k { + SketchKindHandle::DDSketch => 1, + SketchKindHandle::Kll => 2, + SketchKindHandle::Hll => 3, + SketchKindHandle::CountSketch => 4, + SketchKindHandle::CountMin => 5, + SketchKindHandle::CmsWithHeap => 6, + SketchKindHandle::CountSketchWithHeap => 7, + SketchKindHandle::Any => 0, + } +} + +fn encode_sketch_config(cfg: &SketchConfig, buf: &mut Vec) { + match cfg { + SketchConfig::DDSketch { relative_accuracy } => { + buf.push(b'D'); + buf.extend_from_slice(&relative_accuracy.to_le_bytes()); + } + SketchConfig::Kll { k } => { + buf.push(b'K'); + buf.extend_from_slice(&k.to_le_bytes()); + } + SketchConfig::Hll { precision } => { + buf.push(b'H'); + buf.extend_from_slice(&precision.to_le_bytes()); + } + SketchConfig::CountSketch { rows, cols } => { + buf.push(b'S'); + buf.extend_from_slice(&rows.to_le_bytes()); + buf.extend_from_slice(&cols.to_le_bytes()); + } + SketchConfig::CountMin { rows, cols } => { + buf.push(b'M'); + buf.extend_from_slice(&rows.to_le_bytes()); + buf.extend_from_slice(&cols.to_le_bytes()); + } + } +} + +// ── Accuracy ──────────────────────────────────────────────────────────────── + +/// Accuracy bound derived from `SketchConfig`. Surfaced to the user +/// via query response metadata so they know the precision / confidence +/// of each result. +#[derive(Debug, Clone, Copy)] +pub struct AccuracyBound { + /// Approximate error bound (e.g. DDSketch's α, HLL's std-error). + pub epsilon: f64, + /// Probability of staying within `epsilon` (1.0 - δ). + pub confidence: f64, +} + +impl AccuracyBound { + /// Compute accuracy bound from sketch config. Variant-specific + /// formulas; the HTTP layer can present this in + /// `X-ASAP-Accuracy: 0.01` so callers know the warm-tier answer's + /// error envelope. + pub fn from_config(cfg: &SketchConfig) -> Self { + match cfg { + SketchConfig::DDSketch { relative_accuracy } => Self { + epsilon: *relative_accuracy, + confidence: 1.0, + }, + SketchConfig::Kll { k } => Self { + epsilon: if *k > 0 { 1.0 / (*k as f64) } else { 1.0 }, + confidence: 0.99, + }, + SketchConfig::Hll { precision } => Self { + epsilon: 1.04 / ((1u64 << *precision) as f64).sqrt(), + confidence: 0.68, + }, + SketchConfig::CountSketch { rows, cols } => { + let e = std::f64::consts::E; + let eps = if *cols > 0 { + (e / (*cols as f64)).sqrt() + } else { + 1.0 + }; + let half_rows = (*rows as f64) / 2.0; + let delta = 2f64.powf(-half_rows); + Self { + epsilon: eps, + confidence: 1.0 - delta, + } + } + SketchConfig::CountMin { rows, cols } => { + let e = std::f64::consts::E; + let eps = if *cols > 0 { e / (*cols as f64) } else { 1.0 }; + let delta = (-(*rows as f64)).exp(); + Self { + epsilon: eps, + confidence: 1.0 - delta, + } + } + } + } +} + +// ── Per-sample payload ────────────────────────────────────────────────────── + +/// Per-sample sketch state. Stored as the payload column inside the +/// per-sid `SidStoreData` columnar storage. +#[derive(Debug, Clone)] +pub struct SketchSampleState { + pub bytes: Vec, + /// Wire-encoding hint from the OTLP DataPoint's `encoding` field. + pub encoding: SketchEncoding, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SketchEncoding { + ProtoFull, + ProtoDelta, + MsgpackFull, + MsgpackDelta, +} + +/// One materialized series row returned by the query path. Resolved +/// from the per-sid intern table at read time. +#[derive(Debug, Clone)] +pub struct SketchTimeSeries { + pub sid: u64, + pub series_label_values: std::collections::BTreeMap, + /// `window_end_unix_ms → sketch payload`. BTreeMap so the query + /// path can iterate in time order without an extra sort. + pub samples: std::collections::BTreeMap, +} + +/// Unified payload variant — what one window of one (sid, label-values +/// vector) physically holds. Phase 5 M2.3 generalizes storage so the +/// same store can host both sketch state and partial-accumulator state +/// under content-addressed sids. +/// +/// Each sid stores exactly one variant for its entire lifetime — the +/// variant is fixed by the sid's `agg_kind` at registration. Mixing +/// variants under one sid is a logic error the storage layer doesn't +/// guard against (a mismatch crashes the reducer at runtime). The +/// ingest path is responsible for not doing that. +#[derive(Clone)] +pub enum AggPayload { + /// Opaque sketch state — see [`SketchSampleState`]. + Sketch(SketchSampleState), + /// Partial-accumulator state — Sum / Count / Avg / Rate / MinMax. + /// Cloned via the `Clone` impl on `Box` (which + /// dispatches through `clone_boxed_core`). + Precompute(Box), +} + +impl std::fmt::Debug for AggPayload { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + AggPayload::Sketch(s) => f.debug_tuple("Sketch").field(s).finish(), + AggPayload::Precompute(p) => f + .debug_struct("Precompute") + .field("type_name", &p.type_name()) + .finish(), + } + } +} + +impl AggPayload { + /// Return the sketch payload if this is a sketch variant; `None` + /// otherwise. The warm-tier sketch reducer uses this to filter + /// non-sketch payloads out of its `query_range` results. + pub fn as_sketch(&self) -> Option<&SketchSampleState> { + match self { + AggPayload::Sketch(s) => Some(s), + AggPayload::Precompute(_) => None, + } + } + + /// Return the precompute payload if this is a precompute variant; + /// `None` otherwise. The precompute query path uses this. + pub fn as_precompute(&self) -> Option<&dyn crate::storage_engines::types::AggregateCore> { + match self { + AggPayload::Precompute(p) => Some(p.as_ref()), + AggPayload::Sketch(_) => None, + } + } + + /// Approximate in-memory byte footprint of the payload — used by + /// the persistence layer's memory-pressure trigger. Sketch payloads + /// report their byte buffer length (the dominant cost); precompute + /// payloads forward to the accumulator's own `approx_memory_bytes()`. + pub fn approx_bytes(&self) -> usize { + match self { + AggPayload::Sketch(s) => { + s.bytes.len() + std::mem::size_of::() + } + AggPayload::Precompute(p) => p.approx_memory_bytes(), + } + } +} diff --git a/data_plane/src/storage_engines/sketch_db/mod.rs b/data_plane/src/storage_engines/sketch_db/mod.rs index 393e2c7f..ad8ff587 100644 --- a/data_plane/src/storage_engines/sketch_db/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/mod.rs @@ -30,6 +30,7 @@ pub mod accuracy; pub mod backfill; +pub mod data; pub mod metrics; pub mod schema; pub mod store; diff --git a/data_plane/src/storage_engines/sketch_db/store/mod.rs b/data_plane/src/storage_engines/sketch_db/store/mod.rs index 1d35e062..608a2c1d 100644 --- a/data_plane/src/storage_engines/sketch_db/store/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/store/mod.rs @@ -24,11 +24,20 @@ use std::sync::{Arc, RwLock}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use dashmap::DashMap; -use xxhash_rust::xxh64::xxh64; use self::epoch_columnar::{LabelValuesId, SidStoreData, TimestampRange}; use crate::storage_engines::sketch_db::schema::AggStatus; +// Phase-5 reorg: payload taxonomy + sid hashing + accuracy moved to +// `sketch_db::data`. Re-exported here so existing call sites +// (`crate::storage_engines::sketch_db::store::*`) keep compiling +// during the reorg. +pub use crate::storage_engines::sketch_db::data::{ + canonical_parameters, compute_sid, compute_sketch_sid, AccuracyBound, AggKind, AggPayload, + AggregationType, Capability, SketchConfig, SketchEncoding, SketchKindHandle, + SketchSampleState, SketchTimeSeries, +}; + fn now_ms() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -36,276 +45,6 @@ fn now_ms() -> u64 { .unwrap_or(0) } -// ── Capability re-exports ──────────────────────────────────────────────────── -// -// Step 2a consolidated all capability state into -// `controller::sketch_algebra::capability`. The backend no longer -// defines its own `Capability` / `SketchKindHandle`; it re-exports the -// canonical types so there's exactly one definition in the codebase. -// `is_satisfied_by` (used by the engine warm-tier hook) now lives on -// the controller-side `Capability` impl. - -pub use controller::sketch_algebra::{Capability, SketchKindHandle}; - -/// Sketch-instance configuration carried per-Metric on the OTLP wire -/// (Phase 2 lifted these from per-DP up to the parent sketch container). -/// Backend reads the relevant variant at ingest time and stores it in -/// `SketchInstanceMetadata.sketch_config`. -#[derive(Debug, Clone)] -pub enum SketchConfig { - DDSketch { relative_accuracy: f64 }, - Kll { k: u32 }, - Hll { precision: u32 }, - CountSketch { rows: i32, cols: i32 }, - CountMin { rows: i32, cols: i32 }, -} - -/// What kind of aggregation a `sid` identifies. M2.3 generalization -/// — sids now cover BOTH opaque-sketch state and partial-accumulator -/// (Sum / Count / Avg / Rate / MinMax) state, so the future -/// unified store can host both. -/// -/// The sid hash distinguishes the two branches structurally: a -/// `Sketch` sid is content-addressed over `(metric, attrs, -/// sketch_kind, sketch_config)`; a `Precompute` sid is content- -/// addressed over `(metric, attrs, agg_type, parameters)`. -#[derive(Debug, Clone)] -pub enum AggKind { - /// Opaque sketch state — DDSketch, KLL, HLL, CountMin, CountSketch. - /// Payload at storage layer is an encoded byte string - /// (`SketchSampleState`). - Sketch { - kind: SketchKindHandle, - config: SketchConfig, - }, - /// Partial-accumulator state — Sum, Count, Avg, Rate, MinMax, - /// SetAggregator, etc. Payload at storage layer is whatever the - /// per-accumulator serializer produces. - Precompute { - agg_type: AggregationType, - /// Stable canonical encoding of the agg's `parameters: HashMap` - /// — sorted keys, each value rendered via serde_json. Held as - /// a single owned string so equality / hashing stay cheap and - /// independent of the original HashMap's iteration order. - parameters_canonical: String, - }, -} - -/// Re-export so callers don't need to depend on promql_utilities -/// for the agg_type tag. -pub use promql_utilities::query_logics::enums::AggregationType; - -/// Render a `HashMap` of parameters into the canonical -/// string form `AggKind::Precompute::parameters_canonical` expects. -/// Keys sorted lexicographically; each value via `serde_json`. -pub fn canonical_parameters( - parameters: &std::collections::HashMap, -) -> String { - let sorted: std::collections::BTreeMap<&String, &serde_json::Value> = parameters.iter().collect(); - let mut buf = String::new(); - for (k, v) in sorted { - buf.push_str(k); - buf.push('='); - buf.push_str(&serde_json::to_string(v).unwrap_or_default()); - buf.push(';'); - } - buf -} - -/// Compute a deterministic `series_id` (sid) for one sketch instance. -/// -/// Folds the four DataPoint inputs the backend has at ingest time — -/// `metric_name`, `attrs` (keys + values, canonicalized), `sketch_kind`, -/// and `sketch_config` — into a 64-bit xxhash. Same inputs always -/// produce the same sid across restarts and across hosts, so the -/// controller no longer needs to mint and emit an `aggregation_id` for -/// each (metric, agg-type, params) tuple. Phase 5 M2 directive. -/// -/// `attrs_fingerprint` MUST be the canonical fingerprint string -/// (`canonical_attrs_fingerprint` — keys sorted, joined `k=v;`); the -/// hash is sensitive to whitespace, ordering, and trailing separator, -/// so callers must round-trip through that one function for the -/// pre-flight ResolveSeriesIDs RPC and the ingest path to agree. -/// -/// sid=0 is reserved on the wire (means "unresolved"); if a real input -/// hashes to 0 (vanishingly unlikely with 64-bit xxhash), we perturb to -/// 1. -pub fn compute_sketch_sid( - metric_name: &str, - attrs_fingerprint: &str, - sketch_kind: SketchKindHandle, - sketch_config: &SketchConfig, -) -> u64 { - compute_sid( - metric_name, - attrs_fingerprint, - &AggKind::Sketch { - kind: sketch_kind, - config: sketch_config.clone(), - }, - ) -} - -/// Generalized version of [`compute_sketch_sid`] covering both sketch -/// and precompute aggregations. Same `(metric, attrs, agg_kind)` -/// tuple always yields the same sid. -/// -/// The two branches encode disjointly: a `Sketch` payload starts with -/// `sketch_kind_tag` (1..=7, see [`sketch_kind_tag`]), while a -/// `Precompute` payload starts with the byte `b'P'` (ASCII 80), which -/// no sketch tag will ever produce. So an attacker (or a colliding -/// hash input) can't force a sketch sid to overlap a precompute sid -/// at the encoding level. -/// -/// Critically, the `Sketch` branch is bit-identical to what the -/// retired `compute_sketch_sid` function produced — same encoding, -/// same hash. Existing sketch sids the M2 wire format introduced -/// stay stable across this generalization. -pub fn compute_sid( - metric_name: &str, - attrs_fingerprint: &str, - agg_kind: &AggKind, -) -> u64 { - let mut buf: Vec = - Vec::with_capacity(metric_name.len() + attrs_fingerprint.len() + 32); - buf.extend_from_slice(metric_name.as_bytes()); - buf.push(0); - buf.extend_from_slice(attrs_fingerprint.as_bytes()); - buf.push(0); - match agg_kind { - AggKind::Sketch { kind, config } => { - // Sketch branch — match `compute_sketch_sid`'s historical - // byte layout exactly so M2-issued sketch sids survive. - buf.push(sketch_kind_tag(*kind)); - buf.push(0); - encode_sketch_config(config, &mut buf); - } - AggKind::Precompute { - agg_type, - parameters_canonical, - } => { - // Precompute branch starts with 'P' which can never be a - // `sketch_kind_tag` (those live in 1..=7) — no collision. - buf.push(b'P'); - buf.push(0); - buf.extend_from_slice(agg_type.as_str().as_bytes()); - buf.push(0); - buf.extend_from_slice(parameters_canonical.as_bytes()); - } - } - let h = xxh64(&buf, 0); - if h == 0 { - 1 - } else { - h - } -} - -fn sketch_kind_tag(k: SketchKindHandle) -> u8 { - match k { - SketchKindHandle::DDSketch => 1, - SketchKindHandle::Kll => 2, - SketchKindHandle::Hll => 3, - SketchKindHandle::CountSketch => 4, - SketchKindHandle::CountMin => 5, - SketchKindHandle::CmsWithHeap => 6, - SketchKindHandle::CountSketchWithHeap => 7, - SketchKindHandle::Any => 0, - } -} - -fn encode_sketch_config(cfg: &SketchConfig, buf: &mut Vec) { - match cfg { - SketchConfig::DDSketch { relative_accuracy } => { - buf.push(b'D'); - buf.extend_from_slice(&relative_accuracy.to_le_bytes()); - } - SketchConfig::Kll { k } => { - buf.push(b'K'); - buf.extend_from_slice(&k.to_le_bytes()); - } - SketchConfig::Hll { precision } => { - buf.push(b'H'); - buf.extend_from_slice(&precision.to_le_bytes()); - } - SketchConfig::CountSketch { rows, cols } => { - buf.push(b'S'); - buf.extend_from_slice(&rows.to_le_bytes()); - buf.extend_from_slice(&cols.to_le_bytes()); - } - SketchConfig::CountMin { rows, cols } => { - buf.push(b'M'); - buf.extend_from_slice(&rows.to_le_bytes()); - buf.extend_from_slice(&cols.to_le_bytes()); - } - } -} - -/// Accuracy bound derived from `SketchConfig`. Surfaced to the user via -/// query response metadata so they know the precision / confidence of -/// each result. -#[derive(Debug, Clone, Copy)] -pub struct AccuracyBound { - /// Approximate error bound (e.g. DDSketch's α, HLL's std-error). - pub epsilon: f64, - /// Probability of staying within `epsilon` (1.0 - δ). - pub confidence: f64, -} - -impl AccuracyBound { - /// Compute accuracy bound from sketch config. Variant-specific - /// formulas; backends can present this to PromQL response headers - /// (e.g. `X-ASAP-Accuracy: 0.01`) so callers know the warm-tier - /// answer's error envelope. - pub fn from_config(cfg: &SketchConfig) -> Self { - match cfg { - // DDSketch's relative-accuracy α IS the epsilon; confidence - // is 1.0 (deterministic bucket placement). - SketchConfig::DDSketch { relative_accuracy } => Self { - epsilon: *relative_accuracy, - confidence: 1.0, - }, - // KLL's rank error: ε ≈ 1 / k, confidence 99% by default. - SketchConfig::Kll { k } => Self { - epsilon: if *k > 0 { 1.0 / (*k as f64) } else { 1.0 }, - confidence: 0.99, - }, - // HLL std error: σ ≈ 1.04 / sqrt(2^precision); 1σ ~ 68% - // confidence. Surface the std-error as epsilon. - SketchConfig::Hll { precision } => Self { - epsilon: 1.04 / ((1u64 << *precision) as f64).sqrt(), - confidence: 0.68, - }, - // Count-Sketch: ε ≈ √(e/cols) (L2 estimation), δ ≈ - // 1/(2^(rows/2)). - SketchConfig::CountSketch { rows, cols } => { - let e = std::f64::consts::E; - let eps = if *cols > 0 { - (e / (*cols as f64)).sqrt() - } else { - 1.0 - }; - let half_rows = (*rows as f64) / 2.0; - let delta = 2f64.powf(-half_rows); - Self { - epsilon: eps, - confidence: 1.0 - delta, - } - } - // CMS: ε ≈ e/cols, δ ≈ exp(-rows). - SketchConfig::CountMin { rows, cols } => { - let e = std::f64::consts::E; - let eps = if *cols > 0 { e / (*cols as f64) } else { 1.0 }; - let delta = (-(*rows as f64)).exp(); - Self { - epsilon: eps, - confidence: 1.0 - delta, - } - } - } - } -} - /// Metadata for one logical sketch instance, keyed by `series_id`. /// Populated at ingest time when a sketch DataPoint with a fresh sid /// arrives (or `(metric, attrs)` produces a fresh sid via the @@ -406,102 +145,9 @@ impl SketchInstanceMetadata { } } -/// Per-sample sketch state. Stored as the payload column inside the -/// per-sid `SidStoreData` columnar storage. -#[derive(Debug, Clone)] -pub struct SketchSampleState { - pub bytes: Vec, - /// Wire-encoding hint from the OTLP DataPoint's `encoding` field - /// (PROTO / PROTO_DELTA / MSGPACK / MSGPACK_DELTA). - pub encoding: SketchEncoding, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SketchEncoding { - ProtoFull, - ProtoDelta, - MsgpackFull, - MsgpackDelta, -} - -/// One materialized series row returned by the query path. Resolved -/// from the per-sid intern table at read time. -#[derive(Debug, Clone)] -pub struct SketchTimeSeries { - pub sid: u64, - pub series_label_values: BTreeMap, - /// `window_end_unix_ms → sketch payload`. BTreeMap so the query - /// path can iterate in time order without an extra sort. - pub samples: BTreeMap, -} - -/// Unified payload variant — what one window of one (sid, label-values -/// vector) physically holds. Phase 5 M2.3 generalizes the storage so -/// the same `SketchStore` can host both sketch state and partial- -/// accumulator (Sum / Count / Avg / Rate / MinMax) state under -/// content-addressed sids. -/// -/// Each sid stores exactly one variant for its entire lifetime — the -/// variant is fixed by the sid's `agg_kind` at registration. Mixing -/// variants under one sid is a logic error the storage layer doesn't -/// guard against (a mismatch crashes the reducer at runtime). The -/// ingest path is responsible for not doing that. -#[derive(Clone)] -pub enum AggPayload { - /// Opaque sketch state — see [`SketchSampleState`]. - Sketch(SketchSampleState), - /// Partial-accumulator state — Sum / Count / Avg / Rate / MinMax. - /// Cloned via the `Clone` impl on `Box` (which - /// dispatches through `clone_boxed_core`). - Precompute(Box), -} - -impl std::fmt::Debug for AggPayload { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - AggPayload::Sketch(s) => f.debug_tuple("Sketch").field(s).finish(), - AggPayload::Precompute(p) => f - .debug_struct("Precompute") - .field("type_name", &p.type_name()) - .finish(), - } - } -} - -impl AggPayload { - /// Return the sketch payload if this is a sketch variant; `None` - /// otherwise. The warm-tier sketch reducer uses this to filter - /// non-sketch payloads out of its `query_range` results. - pub fn as_sketch(&self) -> Option<&SketchSampleState> { - match self { - AggPayload::Sketch(s) => Some(s), - AggPayload::Precompute(_) => None, - } - } - - /// Return the precompute payload if this is a precompute variant; - /// `None` otherwise. The precompute query path (M2.3.5) uses this. - pub fn as_precompute(&self) -> Option<&dyn crate::storage_engines::types::AggregateCore> { - match self { - AggPayload::Precompute(p) => Some(p.as_ref()), - AggPayload::Sketch(_) => None, - } - } - - /// Approximate in-memory byte footprint of the payload — used by - /// the persistence layer's memory-pressure trigger. Sketch - /// payloads report their byte buffer length (the dominant cost); - /// precompute payloads forward to the accumulator's own - /// `approx_memory_bytes()`. - pub fn approx_bytes(&self) -> usize { - match self { - AggPayload::Sketch(s) => { - s.bytes.len() + std::mem::size_of::() - } - AggPayload::Precompute(p) => p.approx_memory_bytes(), - } - } -} +// `SketchSampleState`, `SketchEncoding`, `SketchTimeSeries`, and +// `AggPayload` moved to `sketch_db::data` in the Phase-5 reorg; they're +// re-exported at the top of this file for backwards compatibility. /// Per-sid storage value — wraps `SidStoreData` in an `RwLock` so the /// outer DashMap stays read-mostly and per-sid writes don't block one