DDSketch

A DDSketch quantile estimator: a sketch that answers "what was the median response time?" or "what's the 99th-percentile latency?" from a stream of measurements, using a small amount of memory instead of keeping every value.

Think of it as a histogram that sizes its own buckets: instead of you guessing bucket boundaries up front, the buckets grow logarithmically so that every bucket is accurate to the same relative precision. A sketch built with relativeAccuracy = 0.01 answers any quantile within 1% of the true value — whether the true value is 2 milliseconds or 2 minutes.

As a CRDT. Each bucket's count is a GCounter, so piece is a pointwise GCounter join — idempotent, commutative, and associative. Many peers can record measurements independently and merge sketches in any order, with any duplication, and converge. The merge is also lossless: merging two replicas' sketches produces exactly the sketch of the combined stream, so a merge adds zero error on top of the α bound. (This is what rules out t-digest, whose merge is order-dependent and lossy.)

Accuracy guarantee. With relative accuracy α, bucket i covers (γ^(i−1), γ^i] where γ = (1+α)/(1−α); a value v is indexed at ⌈log_γ |v|⌉ and estimated by the bucket representative 2γ^i/(γ+1), which is within relative error α of every value in the bucket. Hence every quantile estimate is within α of the exact quantile, for values inside the indexable range (below). Zeros are counted exactly in a dedicated zero bucket; negative values go through a mirrored bucket store with the same guarantee. Reference: Masson, Rim, Lee — DDSketch: A Fast and Fully-Mergeable Quantile Sketch with Relative-Error Guarantees, PVLDB 12(12), 2019.

Bounded memory — insert-time clamping, not state collapse. A lazy "collapse the lowest buckets when the map grows" rule is not a lattice operation (two replicas collapsing at different times would diverge), so the memory bound is enforced where it stays merge-safe — at insert:

• magnitudes below minIndexedValue count as zeros (the OTLP zero_threshold semantics);

• magnitudes above maxIndexedValue clamp into the top bucket and are counted in the mergeable overflowCount, so the cap is observable, never silent — alert on overflowCount > 0 if the range matters.

The bucket maps are sparse (only touched buckets exist) and their key space is bounded by ⌈ln(max/min)/ln γ⌉ per sign — ≈3110 buckets per sign at the defaults (α = 0.01, range 1e−9…1e18). Only values at the clamped extremes trade away the α guarantee; everything in range keeps it.

Configuration is a cluster-wide constant. Two sketches merge only if their (relativeAccuracy, minIndexedValue, maxIndexedValue) match exactly — the same must-match discipline as HyperLogLog's precision. Fix it once per deployment; piece rejects mismatches.

Immutable. add does not mutate the receiver; it returns a Patch whose delta carries a single bucket cell (or a single zero/overflow counter cell) — the minimal sparse fragment idiom shared by the zoo's sketches.

OTel interop. The state is structurally an OTLP ExponentialHistogramDataPoint: zeroCount plus positive/negative log-bucket arrays (positiveBuckets/negativeBuckets). The OTLP mapping itself lives with the metrics exporter, not in this module.

Samples

val replica = ReplicaId("api-server-1")

// α = 0.01 → every quantile estimate is within 1% of the true value.
var latencies = DDSketch.empty(relativeAccuracy = 0.01)

// add() returns a one-bucket delta; absorb it with piece().
for (ms in listOf(12.0, 15.0, 14.0, 250.0, 13.0, 16.0, 900.0, 14.5)) {
    latencies = latencies.piece(latencies.add(replica, ms))
}

// The p50 sits among the fast requests; the p99 reflects the slow tail.
check(latencies.quantile(0.5) in 13.0..17.0)
check(latencies.quantile(1.0) in 890.0..910.0) // within 1% of 900

val serverA = ReplicaId("server-a")
val serverB = ReplicaId("server-b")

// Two servers record their own request latencies.
var a = DDSketch.empty()
var b = DDSketch.empty()
repeat(100) { a = a.piece(a.add(serverA, 10.0 + it)) }   // 10–109 ms
repeat(100) { b = b.piece(b.add(serverB, 500.0 + it)) }  // 500–599 ms

// Merge: pointwise GCounter join of the bucket counts.
val merged = a.piece(b)
check(merged.count == 200L)

// The merged p50 sits at the boundary between the two servers' ranges.
check(merged.quantile(0.5) in 100.0..120.0)

// Idempotent: merging again with either side changes nothing.
check(merged.piece(a) == merged)
check(merged.piece(b) == merged)