[zzylol]

Part of #250

Why

A DDSketch's readouts are its quantiles, and they come from the bucket stores. DdSketch also kept sum, min and max, and the ASAPv1 payload carried them. Nothing a DDSketch answers needed them: they only clamped quantile answers to [min, max], plus three accessors. They also broke deltas. A delta names buckets, not values, so merge_delta / apply_delta could only fold in bucket representatives, and a receiver fed by deltas ended with a different sum and different extremes than the producer.

The design decision: sum, count, min and max are a separate exact aggregate, not sketch state. This PR is the DDSketch part of the umbrella issue.

What

• DdSketch's state is the two bucket stores, the zero count and the admission rate (count stays derived). sum(), min() and max() are removed.
• Quantiles are no longer clamped. Every rank, the extremes included, reads its bucket's representative, so q = 0 and q = 1 are within alpha of the true minimum and maximum (for indexable magnitudes: a nonzero magnitude below the indexable range counts as zero and reads 0.0).
• merge_delta and Octo's apply_delta touch buckets and counts only, so a delta-fed receiver holds the producer's buckets exactly.
• ASAPv1, per Q-VER: new kinds 0x05 0x01 (state, five-field payload) and 0x05 0x81 (delta, same payload as before). The retired 0x05 0x00 / 0x05 0x80 still decode, with the scalars dropped, and are never encoded. EhBucketSketch nests 0x05 0x01 and accepts 0x05 0x00, so it has eleven decodable nested ids, ten of them encoded.
• EhBucketSketch's "min" / "max" commands on a DDSketch read quantiles 0 and 1.
• Golden fixtures for the new kinds: submodule bumped to ddsketch: fixtures for the buckets-only kinds 05 01 / 05 81 sketchlib-golden-bytes#4 (08440a1).
• memory_bytes / estimated_memory_bytes need no edit: both start from size_of::<DdSketch>(), which is now 24 bytes smaller.
• The capability table is not touched here: feat(capability)!: Planner vocabulary, exhaustive serde enums, rows without exact aggregates #254 (on main) already left the sum / min / max out of the DDSketch row.
• Docs: wire format §1 registry, §3.8, §3.18, §3.23, the admission-rate table (DDSketch rescales count only), Q-DERIVED; the DDSketch, Octo and EhBucketSketch API pages; CHANGELOG.

How

• ddsketch.rs: fields, add_to_sum and the scalar updates in try_add, merge, apply_delta and apply_zero_delta are removed. get_value_at_quantile clamps the nearest rank to 1..=n and reads value_at_rank, the same bucket walk quantile_interpolated uses, now without the rank-0 / rank-n−1 special cases or the clamp. If count ever exceeded the buckets (no code path allows it), value_at_rank returns the last occupied bucket and does not panic.
• wire.rs: DdPayload is the five fields. DdPayloadWithScalars reads the retired eight-field payload and converts by dropping the scalars. The old scalar decode rule is gone.
• delta.rs: decodes either delta kind, encodes 0x05 0x81.
• octo.rs (octo-runtime test): checks quantiles 0 and 1 within alpha of the stream's extremes instead of the removed min() / max().
• New golden fixtures were generated by decoding each retired fixture and re-encoding it. Tests check both directions.

Key code interfaces added / modified

src/sketches/ddsketch.rs

/// Mergeable, relative-error quantile sketch (stays). Its state is the
/// buckets alone; keep a sum / min / max with a separate exact aggregate.
pub struct DdSketch {
    /// Relative accuracy. // unchanged
    alpha: f64,
    /// Derived from `alpha`. // unchanged
    gamma: f64,
    /// Derived from `alpha`. // unchanged
    log_gamma: f64,
    /// Derived from `alpha`. // unchanged
    inv_log_gamma: f64,
    /// Positive bucket store. // unchanged
    positive_store: Buckets,
    /// Total samples, derived from the stores. // unchanged
    count: u64,
    /// Running sum of the values. // removed
    // sum: f64,
    /// Smallest value seen. // removed
    // min: f64,
    /// Largest value seen. // removed
    // max: f64,
    /// Negative-magnitude bucket store. // unchanged
    negative_store: Buckets,
    /// Values counted as zero. // unchanged
    zero_count: u64,
    /// Probability each sample was admitted. // unchanged
    admission_rate: f64,
}

impl DdSketch {
    /// Nearest-rank quantile `ceil(q·n)`; `q <= 0` / `q >= 1` read the lowest /
    /// highest occupied bucket. None when empty or q is NaN.
    pub fn get_value_at_quantile(&self, q: f64) -> Option<f64>; // was: q<=0 / q>=1 returned the exact min / max, other ranks clamped to [min, max]
    /// PromQL-style interpolated quantile; every order statistic is its
    /// bucket's representative.
    pub fn quantile_interpolated(&self, q: f64) -> Result<Option<f64>, String>; // was: ranks 0 and n-1 exact min/max, others clamped
    /// Running sum of ingested values. // removed
    // pub fn sum(&self) -> f64;
    /// Smallest ingested value. // removed
    // pub fn min(&self) -> Option<f64>;
    /// Largest ingested value. // removed
    // pub fn max(&self) -> Option<f64>;
    /// Adds an Octo bucket delta: bucket and count only. // was: also folded the representative into sum/min/max
    pub fn apply_delta(&mut self, delta: DdSketchBucketDelta);
    /// Adds an Octo zero-count delta: zero count and count only. // was: also moved min/max to 0.0
    pub fn apply_zero_delta(&mut self, count: u64);
    /// Sets the admission rate; `count` covers admitted samples. // unchanged signature, doc no longer mentions sum
    pub fn with_admission_rate(self, p: f64) -> Self;
}

src/sketches/ddsketch/wire.rs

impl DdSketch {
    /// Encodes kind 0x05 0x01: alpha in the metadata; payload
    /// [counts, offset, negative_counts, negative_offset, zero_count].
    pub fn to_asapv1(&self) -> Result<Vec<u8>, RmpEncodeError>; // was kind 0x05 0x00 with sum/min/max in the payload
    /// Decodes kind 0x05 0x01, or the retired 0x05 0x00 with its
    /// sum/min/max read and dropped. Fails closed otherwise.
    pub fn from_asapv1(bytes: &[u8]) -> Result<Self, RmpDecodeError>; // was 0x05 0x00 only
}

src/sketches/ddsketch/delta.rs

impl DdSketchDelta {
    /// Encodes kind 0x05 0x81 (payload unchanged).
    pub fn to_asapv1(&self) -> Result<Vec<u8>, RmpEncodeError>; // was 0x05 0x80
    /// Decodes 0x05 0x81, or the retired 0x05 0x80 (same payload).
    pub fn from_asapv1(bytes: &[u8]) -> Result<Self, RmpDecodeError>; // was 0x05 0x80 only
}
impl DdSketch {
    /// Adds each count to its bucket and the zero count; the result is the
    /// state the same samples' adds would build. Fails, changing nothing, on an
    /// alpha / rate mismatch, a count overflow or a store past i32.
    pub fn merge_delta(&mut self, delta: &DdSketchDelta) -> Result<(), String>; // was: also folded representatives into sum/min/max
}

src/asapv1/kind.rs (crate-private; the wire ids)

/// DDSketch: the two bucket stores and the zero count (spec §3.8).
pub(crate) const DDSKETCH: &[u8] = &[0x05, 0x01]; // was [0x05, 0x00]
/// The DDSketch delta (spec §3.23).
pub(crate) const DDSKETCH_DELTA: &[u8] = &[0x05, 0x81]; // was [0x05, 0x80]
/// Retired kind that also carried sum/min/max: decoded, never encoded.
pub(crate) const DDSKETCH_WITH_SCALARS: &[u8] = &[0x05, 0x00]; // new
/// Retired delta of DDSKETCH_WITH_SCALARS: decoded, never encoded.
pub(crate) const DDSKETCH_WITH_SCALARS_DELTA: &[u8] = &[0x05, 0x80]; // new

src/sketch_framework/eh_bucket_sketch.rs

impl EhBucketSketch {
    /// Sketch-specific query. On a DDSketch, "min" / "max" are quantiles 0 / 1
    /// (within alpha of the extremes); "count" and numeric quantiles unchanged.
    pub fn query(&self, key: &DataInput) -> Result<f64, &'static str>; // was: "min" / "max" returned the exact tracked extremes
}

Wire format (docs/asapv1_wire_format.md)

kind_id	Payload	Status
0x05 0x01	[counts, offset, negative_counts, negative_offset, zero_count]	new, encoded
0x05 0x81	[indices, deltas, negative_indices, negative_deltas, zero_count]	new, encoded
0x05 0x00	[counts, offset, sum, min, max, negative_counts, negative_offset, zero_count]	retired: decoded, scalars dropped
0x05 0x80	same as 0x05 0x81	retired: decoded

Usage

A collector pane sends DDSketch deltas to a receiver. It keeps sum / min / max in its own exact aggregate. This was compiled and run as an integration test at 4234c5c (the same commit before the rebase onto #254, which touches only the capability table):

use asap_sketchlib::{DdSketch, DdSketchWorker};

// Non-zero buckets of a store as (bucket index, count).
fn cells(counts: &[u64], offset: i32) -> Vec<(i32, u64)> {
    (offset..).zip(counts.iter().copied()).filter(|&(_, c)| c != 0).collect()
}

// Latencies one collector sees in one pane, in milliseconds.
let latencies_ms = [12.0, 15.5, 9.8, 230.0, 14.1, 0.0, 18.9, 1200.0];
// Producer: a DDSketch (alpha = 1%) for quantiles ...
let mut producer = DdSketch::new(0.01);
// ... and a separate exact aggregate for sum / min / max, no longer DDSketch state.
let (mut sum_ms, mut min_ms, mut max_ms) = (0.0f64, f64::INFINITY, f64::NEG_INFINITY);
// Octo worker: emits a bucket delta once a held count reaches 2; flush sends the rest.
let mut worker = DdSketchWorker::new(0.01);
// Receiver: built from the deltas alone.
let mut receiver = DdSketch::new(0.01);
for v in latencies_ms {
    producer.add(&v);
    (sum_ms, min_ms, max_ms) = (sum_ms + v, min_ms.min(v), max_ms.max(v));
    worker.add_emit_delta(v, 2, &mut |delta| receiver.apply_delta(delta));
}
worker.flush(&mut |delta| receiver.apply_delta(delta));

// The receiver holds the producer's buckets exactly, so every quantile matches.
assert_eq!(
    cells(receiver.positive_store_counts(), receiver.positive_store_offset()),
    cells(producer.positive_store_counts(), producer.positive_store_offset()),
);
for q in [0.0, 0.5, 0.99, 1.0] {
    assert_eq!(receiver.get_value_at_quantile(q), producer.get_value_at_quantile(q));
}
// On the wire the state is kind 0x05 0x01: buckets and zero count only.
let wire = receiver.to_asapv1().unwrap();
assert_eq!(&wire[8..10], &[0x05, 0x01]);
let restored = DdSketch::from_asapv1(&wire).unwrap();
let top = restored.get_value_at_quantile(1.0).unwrap(); // highest bucket, within 1% of 1200
assert!((top - 1200.0).abs() <= 0.01 * 1200.0 + 1e-9);
// Printed: count=8 p99=1200.131 q1=1200.131
// Printed: exact aggregate: sum=1500.3 min=0 max=1200

Before this PR, the same receiver's sum(), min() and max() were built from bucket representatives and differed from the producer's.

Major algorithm changes

• Problem. At α = 1%, 100.3 and 101.4 fall in bucket 230, whose representative is 100.495 ((1 + α)·γ^230). The producer had sum = 201.7, min = 100.3, max = 101.4. A receiver fed the bucket delta had sum = 200.989 and min = max = 100.495, so the two states differed.
• Before. Quantile = bucket representative clamped to [min, max]; q <= 0 / q >= 1 returned the exact min / max; deltas folded representatives into the scalars.
• After. Quantile = bucket representative for every rank; deltas add counts only.
• Correctness. Every answer is a representative, which is within α of every value in its bucket (DDSketch's guarantee), so the extremes stay within α. The clamp only tightened answers already inside that bound. Delta apply is cell-wise addition, so it is exact and commutative.
• Complexity. Unchanged walks. Each add does one add and two compares less. The state is 24 bytes smaller, and each payload three f64 fields smaller.
• Limits. q = 0 / q = 1 are no longer exact observed values: a consumer that needs the exact extremes keeps a min / max aggregate.

Before this PR

DdSketch tracked sum / min / max, carried them on the wire (0x05 0x00), returned exact extremes at q = 0 / 1, and could not rebuild them from deltas.

After this PR

DdSketch is buckets only (0x05 0x01 / 0x05 0x81). A delta stream rebuilds it exactly, quantiles are within α everywhere (for indexable magnitudes; smaller nonzero values count as zero), and older 0x05 0x00 / 0x05 0x80 bytes still decode.

Verification

Rebased onto main at 21b6357 (after #249, #251 and #254). Commits: d3a5ddc (the change) and f961b90 (review fixes). Only the tests covering the changed code, plus format and lint:

• cargo test --lib: 1010 passed, 0 failed, 4 ignored.
• cargo test --features octo-runtime --lib octo: 65 passed. cargo test --release --features octo-runtime --test e2e octo: 80 passed.
• cargo test --release --features experimental --lib -- ddsketch eh_bucket octo asapv1 admission: 111 passed (on 950bb5c).
• cargo test --test asapv1_golden: 71 passed (default features), 72 passed (experimental).
• cargo test --test ddsketch_signed: 5 passed. cargo test --test conformance_kit: 13 passed.
• cargo test --test capability: 3 passed (default and experimental). cargo test --test heap_memory_bytes: 1 passed on 950bb5c (includes the DDSketch estimated_memory_bytes bound).
• cargo test --test pbt ddsketch: 9 passed; --test pbt octo_delta: 25 passed (on f961b90). cargo test --test pbt -- ddsketch dd_ octo eh wire quantile: 47 passed. cargo test --release --test e2e -- ddsketch dd_ octo eh_ wire: 110 passed. Those two filtered runs were on the pre-feat(capability)!: Planner vocabulary, exhaustive serde enums, rows without exact aggregates #254 base 950bb5c; feat(capability)!: Planner vocabulary, exhaustive serde enums, rows without exact aggregates #254 changes only the capability table, docs and lib.rs re-exports.
• cargo fmt --check: clean. cargo clippy --all-targets, --all-targets --features experimental and --workspace --all-targets --all-features --locked -- -D warnings: no warnings.
• New tests:
  • deltas_rebuild_the_producer_bit_for_bit: the receiver's non-zero cells and quantiles equal the producer's after each delta batch.
  • ddsketch_decodes_the_retired_kind_with_scalars, ddsketch_retired_fixtures_decode_to_their_bucket_twins: old bytes decode.
  • ddsketch_endpoints_are_within_alpha, ddsketch_returns_the_signed_extremes_within_alpha: q = 0 / 1 are within α of the true min / max.
  • ddsketch_min_and_max_commands_are_within_alpha.
  • ddsketch_retired_scalars_must_be_float64: the retired decoder refuses an int sum or a float32 min, and drops NaN scalars.
  • a_rank_past_the_buckets_reads_the_last_occupied_bucket: the value_at_rank fallback.
• Not run: the full pbt / e2e suites, and benchmarks.

Limitations and follow-ups

• sketchlib-go has no merged ASAPv1 DDSketch codec yet. Two companion sketchlib-go PRs mirror it. On main, the registry gains 0x05 0x01 / 0x05 0x81 and keeps 0x05 0x00 / 0x05 0x80 as retired kinds with a payload. On top of the Go DDSketch codec PR (feat(ddsketch)!: ASAPv1 codec with signed values; drop the protobuf, msgpack, delta and gob formats sketchlib-go#90), the codec encodes 0x05 0x01 and decodes both kinds. A Go delta codec for 0x05 0x81 does not exist yet. Go merge order: the Go main registry PR first (after golden RefCell to reuse buffer during insertion when only have Shared Reference (&T) #4), then KLL compactor merge: safe or unsafe #90 rebased onto Go main (taking main's kind.go), then the codec branch restacked on it with its kind.go hunk dropped.
• sketchlib-golden-bytes#4 must merge first, then this submodule pointer moves to the merged commit.
• Downstream pin bumps: ASAPPlanner's executor (its DDSketch kernel only calls quantile_interpolated) and ASAPCollector (get_value_at_quantile). Tests there that compare q = 0 / 1 or ranks 0 / n−1 with exact values need an α tolerance.
• Other sketches with the same pattern are tracked in the umbrella issue.

🤖 Generated with Claude Code