Skip to content

feat(sketch_db): Phase 5 M2 — derive sketch sid from content hash #150

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): Phase 5 M2 — derive sketch sid from content hash #150

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
101 changes: 55 additions & 46 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 @@ -817,54 +817,59 @@ async fn route_modified_otlp_sketches_to_precompute(
let series_key = format_series_key(&metric.name, &dp.attrs);
let ts_ms = (dp.time_unix_nano / 1_000_000) as i64;

// Phase 4 — sid resolution gate. The four cases mirror
// the design doc §5.4 invariant:
// (sid=0, attrs) → mint a fresh sid and use it
// (sid!=0, attrs) → trust attrs; if cached value
// disagrees, the sender's sid
// is stale → push to
// Phase 5 M2 — sid is content-addressed: a 64-bit
// xxhash of `(metric_name, attrs, sketch_kind,
// sketch_config)`. Same inputs → same sid across
// restarts and across hosts, so the controller no
// longer needs to emit `aggregationId`; the backend
// derives it. Four wire cases:
// (sid=0, attrs) → compute hash, use it
// (sid!=0, attrs) → compute hash; if it
// disagrees with the sender's
// sid the sender's sid is
// stale → push to
// `unknown_sids` so the
// response evicts it
// (sid!=0, no attrs) → reverse-lookup; if unknown,
// push to `unknown_sids` and
// drop this DP (sender will
// re-emit with attrs next pass)
// (sid!=0, no attrs) → can't recompute the hash;
// fall back to "is this sid
// registered?" via
// SketchIndex. Unknown → push
// to `unknown_sids` and drop
// this DP (sender will
// re-emit with attrs next
// pass)
// (sid=0, no attrs) → invalid wire shape, drop
let attrs_pairs: Vec<(&str, &str)> =
dp.attrs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&attrs_pairs);
let kind_for_sid = sketch_kind_handle_for(&dp);
let resolved_sid: Option<u64> = if attrs_pairs.is_empty() {
// No attrs on the wire — sid alone must be
// recognized, otherwise signal stale.
// No attrs on the wire — can't recompute hash.
// Accept the sid iff we've registered it before.
match dp.series_id {
0 => None,
sid => {
if ingest_state.series_resolver.is_known(sid) {
if ingest_state.sketch_index.instance(sid).is_some() {
Some(sid)
} else {
unknown_sids.push(sid);
None
}
}
}
} else if dp.series_id == 0 {
// Attrs present, no sid yet → mint or fetch.
Some(
ingest_state
.series_resolver
.resolve(&metric.name, &fp),
)
} else {
// Both populated: attrs are the source of truth.
// Backend-resolved value wins; mismatched sender
// sid is signalled stale.
let cached = ingest_state
.series_resolver
.resolve(&metric.name, &fp);
if cached != dp.series_id {
let computed = crate::stores::sketch_db::index::compute_sketch_sid(
&metric.name,
&fp,
kind_for_sid,
&dp.container_config,
);
if dp.series_id != 0 && dp.series_id != computed {
// Sender's cached sid disagrees with what the
// backend would now compute — signal stale.
unknown_sids.push(dp.series_id);
}
Some(cached)
Some(computed)
};
let Some(sid) = resolved_sid else {
continue;
Expand Down Expand Up @@ -1975,7 +1980,8 @@ mod sid_resolution_tests {

let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await;
assert!(unknown.is_empty(), "no unknown sids on a fresh-attrs DP");
assert_eq!(state.series_resolver.len(), 1, "resolver minted one sid");
// M2 — sketch sid is hash-derived; resolver is not consulted on
// the sketch ingest path. SketchIndex is the registration set.
assert_eq!(
state.sketch_index.instance_count(),
1,
Expand All @@ -1989,8 +1995,9 @@ mod sid_resolution_tests {
#[tokio::test]
async fn unknown_sid_with_empty_attrs_is_returned_in_response() {
let (state, drain) = make_state().await;
// sid != 0, no attrs — resolver doesn't know it; should land in
// unknown_sids and the DP must be dropped (no instance registered).
// sid != 0, no attrs — SketchIndex doesn't know it; should land
// in unknown_sids and the DP must be dropped (no instance
// registered). Hash recomputation isn't possible without attrs.
let dp = DdSketchDataPoint {
attributes: Vec::new(),
start_time_unix_nano: 0,
Expand All @@ -2005,7 +2012,6 @@ mod sid_resolution_tests {

let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await;
assert_eq!(unknown, vec![7777]);
assert_eq!(state.series_resolver.len(), 0);
assert_eq!(state.sketch_index.instance_count(), 0);

drop(state);
Expand All @@ -2014,8 +2020,11 @@ mod sid_resolution_tests {

#[tokio::test]
async fn sid_attrs_disagreement_signals_stale_sid_but_uses_resolved_value() {
use crate::stores::sketch_db::index::{compute_sketch_sid, SketchConfig, SketchKindHandle};

let (state, drain) = make_state().await;
// First, mint the resolver's view by sending sid=0 with attrs.
// First, register the sid by sending sid=0 with attrs. Sketch
// sid is now hash-derived, not resolver-minted.
let dp_seed = DdSketchDataPoint {
attributes: vec![kv("zone", "z0")],
start_time_unix_nano: 1_000_000,
Expand All @@ -2031,14 +2040,21 @@ mod sid_resolution_tests {
&state,
)
.await;
let resolved_sid = state.series_resolver.lookup(
let expected_sid = compute_sketch_sid(
"http_latency_ms",
&crate::drivers::ingest::canonical_attrs_fingerprint(&[("zone", "z0")]),
SketchKindHandle::DDSketch,
&SketchConfig::DDSketch {
relative_accuracy: 0.01,
},
);
assert!(
state.sketch_index.instance(expected_sid).is_some(),
"seed registers the hash-derived sid"
);
let resolved_sid = resolved_sid.expect("seed mints");

// Now arrive with the same attrs but a STALE sid.
let stale = resolved_sid.wrapping_add(123);
let stale = expected_sid.wrapping_add(123);
let dp_disagree = DdSketchDataPoint {
attributes: vec![kv("zone", "z0")],
start_time_unix_nano: 1_000_000,
Expand All @@ -2055,16 +2071,9 @@ mod sid_resolution_tests {
)
.await;
assert_eq!(unknown, vec![stale], "stale sid should be signalled");
// Cache stays at the originally-resolved value — the second call
// returns the same sid via the canonical fingerprint.
let still_resolved = state
.series_resolver
.lookup(
"http_latency_ms",
&crate::drivers::ingest::canonical_attrs_fingerprint(&[("zone", "z0")]),
)
.expect("still cached");
assert_eq!(still_resolved, resolved_sid);
// The expected sid stays registered — the second DP routed to
// it via hash recomputation.
assert!(state.sketch_index.instance(expected_sid).is_some());

drop(state);
let _ = drain.await;
Expand Down
137 changes: 137 additions & 0 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 @@ -24,6 +24,7 @@ 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::stores::sketch_db::schema::AggStatus;
Expand Down Expand Up @@ -59,6 +60,87 @@ pub enum SketchConfig {
CountMin { rows: i32, cols: i32 },
}

/// 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 {
let mut buf: Vec<u8> =
Vec::with_capacity(metric_name.len() + attrs_fingerprint.len() + 24);
buf.extend_from_slice(metric_name.as_bytes());
buf.push(0);
buf.extend_from_slice(attrs_fingerprint.as_bytes());
buf.push(0);
buf.push(sketch_kind_tag(sketch_kind));
buf.push(0);
encode_sketch_config(sketch_config, &mut buf);
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<u8>) {
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.
Expand Down Expand Up @@ -697,6 +779,61 @@ mod tests {
assert_eq!(series.len(), 1);
assert_eq!(series[0].samples.len(), 4);
}

#[test]
fn compute_sketch_sid_is_deterministic() {
let cfg = SketchConfig::DDSketch {
relative_accuracy: 0.01,
};
let a = compute_sketch_sid("http_requests_total", "zone=z0;", SketchKindHandle::DDSketch, &cfg);
let b = compute_sketch_sid("http_requests_total", "zone=z0;", SketchKindHandle::DDSketch, &cfg);
assert_eq!(a, b);
assert_ne!(a, 0);
}

#[test]
fn compute_sketch_sid_distinguishes_metric() {
let cfg = SketchConfig::DDSketch {
relative_accuracy: 0.01,
};
let a = compute_sketch_sid("metric_a", "zone=z0;", SketchKindHandle::DDSketch, &cfg);
let b = compute_sketch_sid("metric_b", "zone=z0;", SketchKindHandle::DDSketch, &cfg);
assert_ne!(a, b);
}

#[test]
fn compute_sketch_sid_distinguishes_attrs_values() {
let cfg = SketchConfig::DDSketch {
relative_accuracy: 0.01,
};
let a = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg);
let b = compute_sketch_sid("m", "zone=z1;", SketchKindHandle::DDSketch, &cfg);
assert_ne!(a, b);
}

#[test]
fn compute_sketch_sid_distinguishes_sketch_kind() {
let cfg_dd = SketchConfig::DDSketch {
relative_accuracy: 0.01,
};
let cfg_kll = SketchConfig::Kll { k: 200 };
let a = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg_dd);
let b = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::Kll, &cfg_kll);
assert_ne!(a, b);
}

#[test]
fn compute_sketch_sid_distinguishes_container_config() {
let cfg_a = SketchConfig::DDSketch {
relative_accuracy: 0.01,
};
let cfg_b = SketchConfig::DDSketch {
relative_accuracy: 0.005,
};
let a = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg_a);
let b = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg_b);
assert_ne!(a, b);
}
}

// 2026-05 reorg: generic epoch-partitioned columnar storage lives
Expand Down