From 2d757447fac52c18947ce8897d98bff12cca2ed3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 28 Oct 2026 22:14:04 +0000 Subject: [PATCH 1/2] feat(ddsketch)!: buckets-only state, ASAPv1 kind 05 01; 05 00 decodes with scalars dropped asap_sketchlib drops sum / min / max from the DDSketch state: they are a separate exact aggregate, and a delta (bucket indices and counts) cannot rebuild them. Go follows. - MarshalASAPv1 writes kind 05 01, metadata_version 1 and alpha, payload [counts, offset, negative_counts, negative_offset, zero_count]. - UnmarshalASAPv1 accepts 05 01, and the retired 05 00 (eight fields, sum / min / max read and dropped). - DDSketch loses its sum / min / max fields, Min() and Max(), the scalar decode rules, and the representative cache that only fed the sum. - Quantile reads the nearest-rank bucket representative for every q, q = 0 and q = 1 included (within alpha of the extremes), with no clamp. - asapv1_golden points at 08440a1 (ddsketch_buckets_* fixtures). Co-Authored-By: Claude Opus 5.5 --- asapv1_golden | 2 +- sketches/DDSketch/DDSketch.go | 182 ++----------- sketches/DDSketch/DDSketch_test.go | 8 +- sketches/DDSketch/asapv1.go | 104 +++----- sketches/DDSketch/asapv1_test.go | 404 +++++++++++------------------ wire/asapv1/kind.go | 21 +- wire/asapv1/kind_test.go | 22 +- 7 files changed, 238 insertions(+), 505 deletions(-) diff --git a/asapv1_golden b/asapv1_golden index a70fa1a..08440a1 160000 --- a/asapv1_golden +++ b/asapv1_golden @@ -1 +1 @@ -Subproject commit a70fa1ab1e9d98cb6cc722d558e1efa5a04885ae +Subproject commit 08440a1196027e5b75fe779462f0262fa1e34544 diff --git a/sketches/DDSketch/DDSketch.go b/sketches/DDSketch/DDSketch.go index 4d917e4..43ee473 100644 --- a/sketches/DDSketch/DDSketch.go +++ b/sketches/DDSketch/DDSketch.go @@ -270,7 +270,9 @@ func (b Buckets) Clone() Buckets { // ---------------- DDSketch ---------------- // DDSketch is a relative-accuracy quantile sketch. Update records positive -// values; negative magnitudes and zeros are held when merged or decoded. +// values; negative magnitudes and zeros are held when merged or decoded. Its +// state is the buckets alone: a caller that needs the sum, minimum or maximum +// of the values keeps a separate exact aggregate. type DDSketch struct { mapping IndexMapping store Buckets @@ -278,9 +280,6 @@ type DDSketch struct { zeroCount uint64 count uint64 - sum float64 - min float64 - max float64 // gosPopulated is B — the number of currently-populated (nonzero-count) // buckets — tracked incrementally by UpdateGOS for the insert-time GOS @@ -293,10 +292,6 @@ type DDSketch struct { // and never mixes it with the plain insert path. See PopulatedBuckets. gosPopulated uint32 - // reps caches bucket representatives for addOneFast, direct-mapped by - // absolute bucket index; allocated on first use. - reps *[repCacheSize]repEntry - // sampler implements optional NitroSketch geometric skip-sampling. When nil // (the default) every value is recorded and the sketch is byte-identical to // an unsampled one. When set with p<1, a Geometric(p)-distributed run of @@ -333,8 +328,6 @@ func (d *DDSketch) admit() bool { func NewDDSketch(alpha float64) *DDSketch { return &DDSketch{ mapping: NewIndexMapping(alpha), - min: math.Inf(1), - max: math.Inf(-1), } } @@ -360,8 +353,6 @@ func NewDDSketchWithMaxBins(alpha float64, maxBins int32) *DDSketch { return &DDSketch{ mapping: NewIndexMapping(alpha), store: Buckets{maxBins: maxBins}, - min: math.Inf(1), - max: math.Inf(-1), } } @@ -379,9 +370,6 @@ func (d *DDSketch) Clear() { d.negative = Buckets{} d.zeroCount = 0 d.count = 0 - d.sum = 0 - d.min = math.Inf(1) - d.max = math.Inf(-1) d.gosPopulated = 0 } @@ -415,37 +403,10 @@ func (d *DDSketch) Update(v float64) { } d.count++ - d.addToSum(v) - - if v < d.min { - d.min = v - } - if v > d.max { - d.max = v - } - k := d.mapping.Index(v) d.store.addOne(k) } -// addToSum advances sum, saturating at ±math.MaxFloat64 instead of overflowing. -func (d *DDSketch) addToSum(delta float64) { - next := d.sum + delta - if math.IsInf(next, 0) { - next = math.Copysign(math.MaxFloat64, next) - } - d.sum = next -} - -// resetScalarsIfEmpty restores the empty sum/min/max once the count reaches 0. -func (d *DDSketch) resetScalarsIfEmpty() { - if d.count == 0 { - d.sum = 0 - d.min = math.Inf(1) - d.max = math.Inf(-1) - } -} - // InsertWithHash implements common.Sketch. // It interprets the hash as a numerical value (casting uint64 to float64). // This allows tracking distributions of integer-like values (e.g., latencies in ns). @@ -461,20 +422,6 @@ func (d *DDSketch) Count() uint64 { return d.count } -func (d *DDSketch) Min() (float64, bool) { - if d.count == 0 { - return 0, false - } - return d.min, true -} - -func (d *DDSketch) Max() (float64, bool) { - if d.count == 0 { - return 0, false - } - return d.max, true -} - // ---------------- Safe Merge ---------------- // Merge implements common.Sketch. @@ -502,9 +449,6 @@ func (d *DDSketch) Merge(other common.Sketch) error { if d.count == 0 { // Clone fundamental fields d.count = o.count - d.sum = o.sum - d.min = o.min - d.max = o.max d.store = o.store.Clone() d.negative = o.negative.Clone() d.zeroCount = o.zeroCount @@ -514,15 +458,6 @@ func (d *DDSketch) Merge(other common.Sketch) error { } d.count += o.count - d.addToSum(o.sum) - - if o.min < d.min { - d.min = o.min - } - if o.max > d.max { - d.max = o.max - } - d.mergeBuckets(&d.store, &o.store) d.mergeBuckets(&d.negative, &o.negative) d.zeroCount += o.zeroCount @@ -587,22 +522,17 @@ func (d *DDSketch) mergeBuckets(a *Buckets, b *Buckets) { // ---------------- Quantile ---------------- -// Quantile returns the value at quantile q in [0, 1], walking negative buckets -// in descending index order, then zeros, then positive buckets ascending. Bucket -// values are capped to [min, max]; q == 0 returns min and q == 1 returns max. +// Quantile returns the value at quantile q in [0, 1]: the representative of +// the bucket holding the nearest-rank order statistic ceil(q*n), clamped to +// 1..n, walking negative buckets in descending index order, then zeros, then +// positive buckets ascending. q == 0 and q == 1 read the lowest and highest +// occupied buckets, within alpha of the true extremes. func (d *DDSketch) Quantile(q float64) (float64, bool) { - if d.count == 0 || q < 0 || q > 1 { + if d.count == 0 || !(q >= 0 && q <= 1) { return 0, false } - if q == 0 { - return d.min, true - } - if q == 1 { - return d.max, true - } - - rank := uint64(math.Ceil(q * float64(d.count))) - clamp := func(v float64) float64 { return math.Min(math.Max(v, d.min), d.max) } + rank := max(uint64(math.Ceil(q*float64(d.count))), 1) + rank = min(rank, d.count) var seen uint64 if !d.negative.IsEmpty() { @@ -613,7 +543,7 @@ func (d *DDSketch) Quantile(q float64) (float64, bool) { } seen += neg[i] if seen >= rank { - return clamp(-d.mapping.Value(d.negative.offset + int32(i))), true + return -d.mapping.Value(d.negative.offset + int32(i)), true } } } @@ -622,18 +552,20 @@ func (d *DDSketch) Quantile(q float64) (float64, bool) { return 0, true } + var last float64 if !d.store.IsEmpty() { for i, c := range d.store.counts.AsSlice() { if c == 0 { continue } seen += c + last = d.mapping.Value(d.store.offset + int32(i)) if seen >= rank { - return clamp(d.mapping.Value(d.store.offset + int32(i))), true + return last, true } } } - return d.max, true + return last, true } // QueryWithHash implements common.Sketch. @@ -665,9 +597,6 @@ func (d *DDSketch) Clone() *DDSketch { negative: d.negative.Clone(), zeroCount: d.zeroCount, count: d.count, - sum: d.sum, - min: d.min, - max: d.max, gosPopulated: d.gosPopulated, } } @@ -699,11 +628,7 @@ func (d *DDSketch) PopulatedBuckets() uint32 { return d.gosPopulated } // back to the general ensure()+increment for a new/out-of-range bucket) // that also reports whether this was the bucket's first-ever touch // (prev==0 -> nonzero) in the SAME lookup, so the caller can maintain -// d.gosPopulated in O(1) without a second bucket read. Unlike addOneFast, -// this does NOT touch d.min/d.max — UpdateGOS already updates those from the -// raw inserted value (mirroring Update's semantics) before calling this, so -// doing it again here from the bucket's representative value would just be -// redundant work on every first-touch bucket. +// d.gosPopulated in O(1) without a second bucket read. func (d *DDSketch) addOneGOS(k int32) (newCount uint64, firstTouch bool) { if !d.store.IsEmpty() { idx := int(k - d.store.offset) @@ -729,8 +654,7 @@ func (d *DDSketch) addOneGOS(k int32) (newCount uint64, firstTouch bool) { // adjusted to match — the local edge-side copy now reflects "accumulated // since this bucket was last sent", exactly like Update's running d.count // already does across resets) and the crossing is reported via -// DDSketchGOSUpdate; sum/min/max return to their empty values when the count -// reaches 0. threshold==0 disables the check entirely and behaves +// DDSketchGOSUpdate. threshold==0 disables the check entirely and behaves // exactly like Update (no gosPopulated bookkeeping either — mirrors // CountSketch.UpdateStringGOS's threshold<=0 escape hatch), so callers can // toggle GOS mode without a second insert path. @@ -749,14 +673,6 @@ func (d *DDSketch) UpdateGOS(v float64, threshold uint64) (crossed bool, update return false, DDSketchGOSUpdate{} } - d.addToSum(v) - if v < d.min { - d.min = v - } - if v > d.max { - d.max = v - } - if threshold == 0 { // GOS disabled: plain insert, no gosPopulated tracking. k := d.mapping.Index(v) @@ -782,7 +698,6 @@ func (d *DDSketch) UpdateGOS(v float64, threshold uint64) (crossed bool, update } else { d.count = 0 } - d.resetScalarsIfEmpty() d.gosPopulated-- return true, DDSketchGOSUpdate{Index: k, Count: newCount} } @@ -819,8 +734,7 @@ func (d *DDSketch) BucketCount(k int32) uint64 { return counts[idx] } -// AddToBucket increments bucket k's count by delta, adjusts d.count, advances -// d.sum by the bucket representative times delta, updates d.min/d.max, and +// AddToBucket increments bucket k's count by delta, adjusts d.count, and // returns the new bucket count. Grows the bucket store if k // is outside the current range (or, when the store is maxBins-capped, // collapses the lowest bins — in which case k's delta lands in the @@ -833,47 +747,9 @@ func (d *DDSketch) AddToBucket(k int32, delta uint64) uint64 { counts := d.store.counts.AsMutSlice() counts[idx] += delta d.count += delta - // Use the representative of the bucket ACTUALLY written to (which may be - // the collapsed floor bucket, not k's own), so min/max stay consistent - // with what Quantile() can resolve post-collapse. - rep := d.mapping.Value(d.store.offset + int32(idx)) - d.addToSum(rep * float64(delta)) - if rep < d.min { - d.min = rep - } - if rep > d.max { - d.max = rep - } return counts[idx] } -const repCacheSize = 256 - -// repEntry is one representative cache slot; v == 0 means empty. -type repEntry struct { - k int32 - v float64 -} - -// cachedValue returns d.mapping.Value(k) through the reps cache. -func (d *DDSketch) cachedValue(k int32) float64 { - if d.reps != nil { - if e := &d.reps[uint32(k)%repCacheSize]; e.k == k && e.v != 0 { - return e.v - } - } - return d.fillValue(k) -} - -func (d *DDSketch) fillValue(k int32) float64 { - if d.reps == nil { - d.reps = new([repCacheSize]repEntry) - } - v := d.mapping.Value(k) - d.reps[uint32(k)%repCacheSize] = repEntry{k: k, v: v} - return v -} - // addOneFast is the hot-path variant of AddToBucket(k, 1). // It avoids the ensure() call when bucket k is already within the store's range, // saving a function call and two bounds checks on every insert of a hot bucket. @@ -883,28 +759,16 @@ func (d *DDSketch) addOneFast(k int32) uint64 { idx := int(k - d.store.offset) counts := d.store.counts.AsMutSlice() if uint(idx) < uint(len(counts)) { - prev := counts[idx] counts[idx]++ d.count++ - rep := d.cachedValue(k) - d.addToSum(rep) - if prev == 0 { - // First write to this bucket: initialise min/max from representative. - if rep < d.min { - d.min = rep - } - if rep > d.max { - d.max = rep - } - } return counts[idx] } } return d.AddToBucket(k, 1) } -// ResetBucket zeroes bucket k and decrements d.count by the bucket's current count; -// sum/min/max return to their empty values when the count reaches 0. Used by +// ResetBucket zeroes bucket k and decrements d.count by the bucket's current +// count. Used by // DDSketchOcto.ResetCell to drain a worker-local bucket after emitting a delta. func (d *DDSketch) ResetBucket(k int32) { if d.store.IsEmpty() { @@ -925,11 +789,10 @@ func (d *DDSketch) ResetBucket(k int32) { } else { d.count = 0 } - d.resetScalarsIfEmpty() } // DrainBuckets calls f(k, count) for every non-zero positive bucket, zeroes it and -// subtracts it from d.count, resetting sum/min/max once the count reaches 0. +// subtracts it from d.count. // Used by DDSketchOcto.Flush to ship worker-local sub-τ residuals at end-of-window. func (d *DDSketch) DrainBuckets(f func(k int32, count uint64)) { if d.store.IsEmpty() { @@ -949,7 +812,6 @@ func (d *DDSketch) DrainBuckets(f func(k int32, count uint64)) { } else { d.count = 0 } - d.resetScalarsIfEmpty() } // EachBucket calls f(k, count) for every non-zero bucket without modifying diff --git a/sketches/DDSketch/DDSketch_test.go b/sketches/DDSketch/DDSketch_test.go index 9102af2..a4a4250 100644 --- a/sketches/DDSketch/DDSketch_test.go +++ b/sketches/DDSketch/DDSketch_test.go @@ -268,11 +268,11 @@ func TestDDSketchRustStyleAPI(t *testing.T) { if got := s.Count(); got != 3 { t.Fatalf("unexpected count: got %d", got) } - if min, ok := s.Min(); !ok || min != 1 { - t.Fatalf("unexpected min: %v %v", min, ok) + if lo, ok := s.Quantile(0); !ok || math.Abs(lo-1) > 0.01+1e-12 { + t.Fatalf("unexpected Quantile(0): %v %v", lo, ok) } - if max, ok := s.Max(); !ok || max != 100 { - t.Fatalf("unexpected max: %v %v", max, ok) + if hi, ok := s.Quantile(1); !ok || math.Abs(hi-100) > 0.01*100+1e-9 { + t.Fatalf("unexpected Quantile(1): %v %v", hi, ok) } } diff --git a/sketches/DDSketch/asapv1.go b/sketches/DDSketch/asapv1.go index cbe14e0..7eef82e 100644 --- a/sketches/DDSketch/asapv1.go +++ b/sketches/DDSketch/asapv1.go @@ -9,9 +9,9 @@ import ( "github.com/ProjectASAP/sketchlib-go/wire/asapv1" ) -// MarshalASAPv1 encodes the sketch as an ASAPv1 DDSketch envelope. It writes -// metadata version 1 for a positive-only state and version 2 when the sketch -// holds a negative store or a nonzero zero count. +// MarshalASAPv1 encodes the sketch as an ASAPv1 DDSketch envelope, kind +// 05 01: metadata_version 1 and alpha, then the payload +// [counts, offset, negative_counts, negative_offset, zero_count]. func (d *DDSketch) MarshalASAPv1() ([]byte, error) { alpha := d.mapping.alpha if err := checkAlpha(alpha); err != nil { @@ -19,40 +19,40 @@ func (d *DDSketch) MarshalASAPv1() ([]byte, error) { } counts, offset := wireStore(&d.store) negCounts, negOffset := wireStore(&d.negative) - signed := len(negCounts) > 0 || d.zeroCount != 0 - if err := checkState(signed, counts, offset, negCounts, negOffset, d.zeroCount, d.sum, d.min, d.max); err != nil { + if err := checkState(counts, offset, negCounts, negOffset, d.zeroCount); err != nil { return nil, err } - version, fields := uint8(1), 5 - if signed { - version, fields = 2, 8 - } - md := asapv1.NewMetadataWriter(version) + md := asapv1.NewMetadataWriter(1) md.Float64("alpha", alpha) p := asapv1.NewEncoder() - p.Array(fields) + p.Array(5) asapv1.EncodeUints(p, counts) p.Int(int64(offset)) - p.Float64(d.sum) - p.Float64(d.min) - p.Float64(d.max) - if signed { - asapv1.EncodeUints(p, negCounts) - p.Int(int64(negOffset)) - p.Uint(d.zeroCount) - } + asapv1.EncodeUints(p, negCounts) + p.Int(int64(negOffset)) + p.Uint(d.zeroCount) return asapv1.Marshal(asapv1.KindDDSketch, md, p) } -// UnmarshalASAPv1 replaces the sketch with the one encoded in b, accepting -// metadata versions 1 and 2. The result has no bin cap and no sampler. +// UnmarshalASAPv1 replaces the sketch with the one encoded in b: kind 05 01, +// or the retired 05 00, whose eight-field payload also carries sum / min / +// max between offset and negative_counts; those are read as float64 and +// dropped. The result has no bin cap and no sampler. func (d *DDSketch) UnmarshalASAPv1(b []byte) error { - md, p, err := asapv1.Open(b, asapv1.KindDDSketch) + kind, metadata, payload, err := asapv1.Split(b) + if err != nil { + return err + } + if kind != asapv1.KindDDSketch && kind != asapv1.KindDDSketchWithScalars { + return fmt.Errorf("asapv1: kind_id %v, want %v (or the retired %v)", + kind, asapv1.KindDDSketch, asapv1.KindDDSketchWithScalars) + } + md, err := asapv1.ReadMetadata(metadata) if err != nil { return err } - md.ExpectVersion(1, 2) + md.ExpectVersion(1) alpha := md.Float64("alpha") if err := md.Finish(); err != nil { return err @@ -60,28 +60,26 @@ func (d *DDSketch) UnmarshalASAPv1(b []byte) error { if err := checkAlpha(alpha); err != nil { return err } - signed := md.Version() == 2 - if signed { + withScalars := kind == asapv1.KindDDSketchWithScalars + p := asapv1.NewDecoder(payload) + if withScalars { p.ExpectArray(8) } else { p.ExpectArray(5) } counts := asapv1.DecodeUints[uint64](p) offset := p.Int32() - sum, minV, maxV := p.Float64(), p.Float64(), p.Float64() - var negCounts []uint64 - var negOffset int32 - var zeroCount uint64 - if signed { - negCounts = asapv1.DecodeUints[uint64](p) - negOffset = p.Int32() - zeroCount = p.Uint() + if withScalars { + _, _, _ = p.Float64(), p.Float64(), p.Float64() // sum, min, max: dropped } + negCounts := asapv1.DecodeUints[uint64](p) + negOffset := p.Int32() + zeroCount := p.Uint() if err := p.Finish(); err != nil { return err } - if err := checkState(signed, counts, offset, negCounts, negOffset, zeroCount, sum, minV, maxV); err != nil { + if err := checkState(counts, offset, negCounts, negOffset, zeroCount); err != nil { return err } total, _ := totalCount(counts, negCounts, zeroCount) @@ -93,9 +91,6 @@ func (d *DDSketch) UnmarshalASAPv1(b []byte) error { negative: storeFromWire(negCounts, negOffset), zeroCount: zeroCount, count: total, - sum: sum, - min: minV, - max: maxV, gosPopulated: populatedBucketCount(&store), } return nil @@ -122,40 +117,17 @@ func checkAlpha(alpha float64) error { return nil } -// checkState applies the ASAPv1 decode rules to a store pair, zero count and -// running scalars. A state with no negative or zero mass gets the positive-only -// scalar rule even when it is written as version 2. -func checkState(signed bool, counts []uint64, offset int32, negCounts []uint64, negOffset int32, - zeroCount uint64, sum, minV, maxV float64) error { +// checkState applies the ASAPv1 decode rules to a store pair and zero count. +func checkState(counts []uint64, offset int32, negCounts []uint64, negOffset int32, zeroCount uint64) error { if err := checkSpan(counts, offset); err != nil { return err } - if signed { - if err := checkSpan(negCounts, negOffset); err != nil { - return err - } + if err := checkSpan(negCounts, negOffset); err != nil { + return err } - total, ok := totalCount(counts, negCounts, zeroCount) - if !ok { + if _, ok := totalCount(counts, negCounts, zeroCount); !ok { return errors.New("ddsketch: bucket counts overflow the total sample count") } - if total == 0 { - if sum != 0 || !math.IsInf(minV, 1) || !math.IsInf(maxV, -1) { - return fmt.Errorf("ddsketch: empty sketch must carry sum=0, min=+Inf, max=-Inf, got %v, %v, %v", sum, minV, maxV) - } - return nil - } - if !isFinite(sum) || !isFinite(minV) || !isFinite(maxV) { - return fmt.Errorf("ddsketch: scalars must be finite for a populated sketch: %v, %v, %v", sum, minV, maxV) - } - negTotal, _ := totalCount(negCounts, nil, zeroCount) - if negTotal == 0 { - if !(minV > 0 && minV <= maxV && sum >= minV) { - return fmt.Errorf("ddsketch: scalars out of order: sum=%v, min=%v, max=%v", sum, minV, maxV) - } - } else if !(minV <= maxV) { - return fmt.Errorf("ddsketch: min %v above max %v", minV, maxV) - } return nil } @@ -187,5 +159,3 @@ func totalCount(counts, negCounts []uint64, zeroCount uint64) (uint64, bool) { } return total, true } - -func isFinite(v float64) bool { return !math.IsNaN(v) && !math.IsInf(v, 0) } diff --git a/sketches/DDSketch/asapv1_test.go b/sketches/DDSketch/asapv1_test.go index 746e8c3..e1953d1 100644 --- a/sketches/DDSketch/asapv1_test.go +++ b/sketches/DDSketch/asapv1_test.go @@ -15,8 +15,7 @@ const testAlpha = 0.01 // literal builds a sketch from raw wire state, with count summed here rather // than by the codec. -func literal(alpha float64, pos []uint64, posOffset int32, neg []uint64, negOffset int32, - zero uint64, sum, minV, maxV float64) *DDSketch { +func literal(alpha float64, pos []uint64, posOffset int32, neg []uint64, negOffset int32, zero uint64) *DDSketch { d := NewDDSketch(alpha) if len(pos) > 0 { d.store = Buckets{counts: storage.Vector1DFromVec(pos), offset: posOffset} @@ -29,7 +28,6 @@ func literal(alpha float64, pos []uint64, posOffset int32, neg []uint64, negOffs for _, c := range append(append([]uint64(nil), pos...), neg...) { d.count += c } - d.sum, d.min, d.max = sum, minV, maxV d.gosPopulated = populatedBucketCount(&d.store) return d } @@ -59,39 +57,51 @@ func sameState(got, want *DDSketch) bool { sameBuckets(&got.store, &want.store) && sameBuckets(&got.negative, &want.negative) && got.zeroCount == want.zeroCount && - got.count == want.count && - sameFloat(got.sum, want.sum) && - sameFloat(got.min, want.min) && - sameFloat(got.max, want.max) + got.count == want.count } func signedFixture() *DDSketch { - return literal(testAlpha, []uint64{3, 0, 2}, 310, []uint64{5, 1}, -208, 7, 2523.90625, -0.016, 515.0) + return literal(testAlpha, []uint64{3, 0, 2}, 310, []uint64{5, 1}, -208, 7) +} + +func positiveFixture() *DDSketch { + return literal(testAlpha, []uint64{1, 0, 127, 128, 300, 65536, 4294967296}, -40, nil, 0, 0) } func TestASAPv1Golden(t *testing.T) { cases := map[string]*DDSketch{ - "ddsketch_empty_a001": NewDDSketch(testAlpha), - "ddsketch_positive_a001": literal(testAlpha, []uint64{1, 0, 127, 128, 300, 65536, 4294967296}, -40, - nil, 0, 0, 2181071000.0, 0.453125, 0.5078125), - "ddsketch_signed_a001": signedFixture(), + "ddsketch_buckets_empty_a001": NewDDSketch(testAlpha), + "ddsketch_buckets_positive_a001": positiveFixture(), + "ddsketch_buckets_signed_a001": signedFixture(), } for name, known := range cases { t.Run(name, func(t *testing.T) { asapv1test.CheckGolden(t, name, known, sameState) }) } } -func metadataVersion(t *testing.T, b []byte) uint8 { - t.Helper() - _, metadata, _, err := asapv1.Split(b) - if err != nil { - t.Fatal(err) - } - md, err := asapv1.ReadMetadata(metadata) - if err != nil { - t.Fatal(err) +// The retired 05 00 fixtures decode, with sum / min / max dropped, into the +// state of their 05 01 twins and re-encode as those twins' bytes. +func TestASAPv1RetiredFixturesDecodeToBucketTwins(t *testing.T) { + for retired, want := range map[string]*DDSketch{ + "ddsketch_empty_a001": NewDDSketch(testAlpha), + "ddsketch_positive_a001": positiveFixture(), + "ddsketch_signed_a001": signedFixture(), + } { + t.Run(retired, func(t *testing.T) { + var got DDSketch + if err := got.UnmarshalASAPv1(asapv1test.Golden(t, retired)); err != nil { + t.Fatal(err) + } + if !sameState(&got, want) { + t.Fatalf("decoded state differs\n got: %+v\nwant: %+v", got, *want) + } + b, err := got.MarshalASAPv1() + if err != nil { + t.Fatal(err) + } + asapv1test.Equal(t, b, asapv1test.Golden(t, strings.Replace(retired, "ddsketch_", "ddsketch_buckets_", 1))) + }) } - return md.Version() } func roundTrip(t *testing.T, d *DDSketch) *DDSketch { @@ -120,13 +130,6 @@ func TestASAPv1RoundTripAfterUpdates(t *testing.T) { for i := 1; i <= 2000; i++ { d.Update(float64(i) * 0.37) } - b, err := d.MarshalASAPv1() - if err != nil { - t.Fatal(err) - } - if v := metadataVersion(t, b); v != 1 { - t.Fatalf("metadata_version %d, want 1", v) - } got := roundTrip(t, d) for _, q := range []float64{0, 0.01, 0.25, 0.5, 0.9, 0.99, 1} { gv, _ := got.Quantile(q) @@ -143,15 +146,15 @@ func TestASAPv1RoundTripAfterUpdates(t *testing.T) { func TestASAPv1EmptyRoundTrip(t *testing.T) { got := roundTrip(t, NewDDSketch(testAlpha)) - if _, ok := got.Min(); ok || got.Count() != 0 { - t.Fatalf("decoded empty sketch: count %d, min ok %v", got.Count(), ok) + if _, ok := got.Quantile(0); ok || got.Count() != 0 { + t.Fatalf("decoded empty sketch: count %d, quantile ok %v", got.Count(), ok) } got.Update(42) if got.Count() != 1 { t.Fatalf("Update after decode: count %d, want 1", got.Count()) } - if q, _ := got.Quantile(0); q != 42 { - t.Fatalf("min after decode and Update: %v, want 42", q) + if q, _ := got.Quantile(0); math.Abs(q-42) > testAlpha*42 { + t.Fatalf("Quantile(0) after decode and Update: %v, want 42 within alpha", q) } } @@ -166,20 +169,8 @@ func TestASAPv1ClearedSketchEncodesAsEmpty(t *testing.T) { if err != nil { t.Fatal(err) } - asapv1test.Equal(t, got, asapv1test.Golden(t, "ddsketch_empty_a001")) - } -} - -func TestASAPv1ZeroCountAloneSelectsVersion2(t *testing.T) { - d := literal(testAlpha, []uint64{2}, 40, nil, 0, 3, 3, 0, 1.5) - b, err := d.MarshalASAPv1() - if err != nil { - t.Fatal(err) - } - if v := metadataVersion(t, b); v != 2 { - t.Fatalf("metadata_version %d, want 2", v) + asapv1test.Equal(t, got, asapv1test.Golden(t, "ddsketch_buckets_empty_a001")) } - roundTrip(t, d) } func TestASAPv1MergedRoundTrip(t *testing.T) { @@ -193,23 +184,15 @@ func TestASAPv1MergedRoundTrip(t *testing.T) { } roundTrip(t, left) - signed := signedFixture() - if err := left.Merge(signed); err != nil { + if err := left.Merge(signedFixture()); err != nil { t.Fatal(err) } - b, err := left.MarshalASAPv1() - if err != nil { - t.Fatal(err) - } - if v := metadataVersion(t, b); v != 2 { - t.Fatalf("metadata_version %d after merging a signed sketch, want 2", v) - } got := roundTrip(t, left) if got.Count() != 400+18 || got.zeroCount != 7 { t.Fatalf("merged count %d zero %d, want 418 and 7", got.Count(), got.zeroCount) } - if lo, _ := got.Min(); lo != -0.016 { - t.Fatalf("merged min %v, want -0.016", lo) + if lo, _ := got.Quantile(0); lo != -got.mapping.Value(-207) { + t.Fatalf("merged Quantile(0) %v, want -Value(-207) %v", lo, -got.mapping.Value(-207)) } } @@ -233,24 +216,25 @@ func TestASAPv1CollapsedStoreRoundTrip(t *testing.T) { func TestASAPv1SignedQuantileWalk(t *testing.T) { var d DDSketch - if err := d.UnmarshalASAPv1(asapv1test.Golden(t, "ddsketch_signed_a001")); err != nil { + if err := d.UnmarshalASAPv1(asapv1test.Golden(t, "ddsketch_buckets_signed_a001")); err != nil { t.Fatal(err) } if d.Count() != 18 { t.Fatalf("count %d, want 18", d.Count()) } m := d.mapping - // Ranks 1, 2-6, 7-13, 14-16, 17-18 fall in buckets -207, -208, zero, 310, 312; - // values are capped to [min, max] = [-0.016, 515]. + // Ranks 1, 2-6, 7-13, 14-16, 17-18 fall in buckets -207, -208, zero, 310, + // 312; every rank reads its bucket's representative. want := map[int]float64{ - 1: -0.016, + 1: -m.Value(-207), 2: -m.Value(-208), 6: -m.Value(-208), 7: 0, 13: 0, 14: m.Value(310), 16: m.Value(310), - 17: 515.0, + 17: m.Value(312), + 18: m.Value(312), } for rank, w := range want { q := (float64(rank) - 0.5) / 18 @@ -260,42 +244,32 @@ func TestASAPv1SignedQuantileWalk(t *testing.T) { } } -func TestUpdateSumSaturates(t *testing.T) { - d := NewDDSketch(testAlpha) - v := d.mapping.MaxIndexableValue() - d.Update(v) - d.Update(v) - d.Update(v) - if d.sum != math.MaxFloat64 { - t.Fatalf("sum %v, want MaxFloat64", d.sum) - } - roundTrip(t, d) -} - -func craft(version uint8, alpha float64, payload func(p *asapv1.Encoder)) []byte { +func craft(kind asapv1.KindID, version uint8, alpha float64, payload func(p *asapv1.Encoder)) []byte { md := asapv1.NewMetadataWriter(version) md.Float64("alpha", alpha) p := asapv1.NewEncoder() payload(p) - b, err := asapv1.Marshal(asapv1.KindDDSketch, md, p) + b, err := asapv1.Marshal(kind, md, p) if err != nil { panic(err) } return b } -func v1Payload(counts []uint64, offset int64, sum, minV, maxV float64) func(p *asapv1.Encoder) { +// bucketsPayload is the 05 01 payload. +func bucketsPayload(counts []uint64, offset int64, neg []uint64, negOffset int64, zero uint64) func(p *asapv1.Encoder) { return func(p *asapv1.Encoder) { p.Array(5) asapv1.EncodeUints(p, counts) p.Int(offset) - p.Float64(sum) - p.Float64(minV) - p.Float64(maxV) + asapv1.EncodeUints(p, neg) + p.Int(negOffset) + p.Uint(zero) } } -func v2Payload(counts []uint64, offset int64, sum, minV, maxV float64, neg []uint64, negOffset int64, zero uint64) func(p *asapv1.Encoder) { +// scalarsPayload is the retired 05 00 payload. +func scalarsPayload(counts []uint64, offset int64, sum, minV, maxV float64, neg []uint64, negOffset int64, zero uint64) func(p *asapv1.Encoder) { return func(p *asapv1.Encoder) { p.Array(8) asapv1.EncodeUints(p, counts) @@ -311,21 +285,14 @@ func v2Payload(counts []uint64, offset int64, sum, minV, maxV float64, neg []uin func TestASAPv1RejectsCraftedBytes(t *testing.T) { inf := math.Inf(1) - valid := v1Payload([]uint64{3}, 7, 6, 1.5, 2.5) - foreign := func() []byte { - md := asapv1.NewMetadataWriter(1) - md.Float64("alpha", testAlpha) - p := asapv1.NewEncoder() - valid(p) - b, _ := asapv1.Marshal(asapv1.KindKLL, md, p) - return b - }() + const dd, retired = asapv1.KindDDSketch, asapv1.KindDDSketchWithScalars + valid := bucketsPayload([]uint64{3}, 7, nil, 0, 0) withMetadata := func(write func(md *asapv1.MetadataWriter)) []byte { md := asapv1.NewMetadataWriter(1) write(md) p := asapv1.NewEncoder() valid(p) - b, _ := asapv1.Marshal(asapv1.KindDDSketch, md, p) + b, _ := asapv1.Marshal(dd, md, p) return b } @@ -334,40 +301,33 @@ func TestASAPv1RejectsCraftedBytes(t *testing.T) { b []byte want string }{ - {"foreign kind_id", foreign, ""}, - {"metadata version 3", craft(3, testAlpha, valid), ""}, + {"foreign kind_id", craft(asapv1.KindKLL, 1, testAlpha, valid), "kind_id"}, + {"metadata version 2", craft(dd, 2, testAlpha, valid), ""}, {"unknown metadata key", withMetadata(func(md *asapv1.MetadataWriter) { md.Float64("alpha", testAlpha) md.Uint("bogus", 7) }), ""}, {"missing alpha", withMetadata(func(md *asapv1.MetadataWriter) {}), ""}, - {"alpha 0", craft(1, 0, valid), "alpha"}, - {"alpha 1", craft(1, 1, valid), "alpha"}, - {"alpha negative", craft(1, -0.5, valid), "alpha"}, - {"alpha 2", craft(1, 2, valid), "alpha"}, - {"alpha NaN", craft(1, math.NaN(), valid), "alpha"}, - {"alpha +Inf", craft(1, inf, valid), "alpha"}, - {"v1 with eight fields", craft(1, testAlpha, v2Payload([]uint64{3}, 7, 6, 1.5, 2.5, nil, 0, 1)), ""}, - {"v2 with five fields", craft(2, testAlpha, valid), ""}, - {"offset past int32", craft(1, testAlpha, v1Payload([]uint64{3}, 1<<40, 6, 1.5, 2.5)), ""}, - {"span past int32", craft(1, testAlpha, v1Payload([]uint64{1, 1, 1, 1}, math.MaxInt32-2, 5, 1, 2)), "span past int32"}, - {"negative span past int32", craft(2, testAlpha, - v2Payload([]uint64{1}, 0, 5, -2, 2, []uint64{1, 1, 1, 1}, math.MaxInt32-2, 0)), "span past int32"}, - {"empty store at nonzero offset", craft(1, testAlpha, v1Payload(nil, 42, 0, inf, -inf)), "offset 0"}, - {"empty negative store at nonzero offset", craft(2, testAlpha, - v2Payload([]uint64{1}, 0, 1, 0, 1, nil, 42, 1)), "offset 0"}, - {"total overflow", craft(1, testAlpha, v1Payload([]uint64{math.MaxUint64, 1}, 0, 5, 1, 2)), "overflow"}, - {"signed total overflow", craft(2, testAlpha, - v2Payload([]uint64{1}, 0, 5, -1, 2, []uint64{1}, 0, math.MaxUint64-1)), "overflow"}, - {"populated with empty sentinels", craft(1, testAlpha, v1Payload([]uint64{3}, 7, 6, inf, -inf)), "finite"}, - {"min above max", craft(1, testAlpha, v1Payload([]uint64{3}, 7, 6, 9, 2)), "out of order"}, - {"nonpositive min", craft(1, testAlpha, v1Payload([]uint64{3}, 7, 6, 0, 2)), "out of order"}, - {"sum below min", craft(1, testAlpha, v1Payload([]uint64{3}, 7, 0.5, 1.5, 2.5)), "out of order"}, - {"empty with nonzero sum", craft(1, testAlpha, v1Payload(nil, 0, 5, inf, -inf)), "empty sketch"}, - {"zero buckets with populated scalars", craft(1, testAlpha, v1Payload([]uint64{0, 0}, 7, 6, 1.5, 2.5)), "empty sketch"}, - {"signed min above max", craft(2, testAlpha, v2Payload([]uint64{1}, 0, 5, 3, -1, []uint64{1}, 0, 0)), "above max"}, - {"signed infinite sum", craft(2, testAlpha, v2Payload(nil, 0, inf, 0, 0, nil, 0, 1)), "finite"}, - {"v2 positive-only sum below min", craft(2, testAlpha, v2Payload([]uint64{3}, 7, 0.5, 1.5, 2.5, nil, 0, 0)), "out of order"}, + {"alpha 0", craft(dd, 1, 0, valid), "alpha"}, + {"alpha 1", craft(dd, 1, 1, valid), "alpha"}, + {"alpha negative", craft(dd, 1, -0.5, valid), "alpha"}, + {"alpha 2", craft(dd, 1, 2, valid), "alpha"}, + {"alpha NaN", craft(dd, 1, math.NaN(), valid), "alpha"}, + {"alpha +Inf", craft(dd, 1, inf, valid), "alpha"}, + {"05 01 with eight fields", craft(dd, 1, testAlpha, scalarsPayload([]uint64{3}, 7, 6, 1.5, 2.5, nil, 0, 0)), ""}, + {"05 00 with five fields", craft(retired, 1, testAlpha, valid), ""}, + {"offset past int32", craft(dd, 1, testAlpha, bucketsPayload([]uint64{3}, 1<<40, nil, 0, 0)), ""}, + {"span past int32", craft(dd, 1, testAlpha, bucketsPayload([]uint64{1, 1, 1, 1}, math.MaxInt32-2, nil, 0, 0)), "span past int32"}, + {"negative span past int32", craft(dd, 1, testAlpha, + bucketsPayload([]uint64{1}, 0, []uint64{1, 1, 1, 1}, math.MaxInt32-2, 0)), "span past int32"}, + {"retired span past int32", craft(retired, 1, testAlpha, + scalarsPayload([]uint64{1, 1, 1, 1}, math.MaxInt32-2, 4, 1, 2, nil, 0, 0)), "span past int32"}, + {"empty store at nonzero offset", craft(dd, 1, testAlpha, bucketsPayload(nil, 42, nil, 0, 0)), "offset 0"}, + {"empty negative store at nonzero offset", craft(dd, 1, testAlpha, + bucketsPayload([]uint64{1}, 0, nil, 42, 1)), "offset 0"}, + {"total overflow", craft(dd, 1, testAlpha, bucketsPayload([]uint64{math.MaxUint64, 1}, 0, nil, 0, 0)), "overflow"}, + {"signed total overflow", craft(dd, 1, testAlpha, + bucketsPayload([]uint64{1}, 0, []uint64{1}, 0, math.MaxUint64-1)), "overflow"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -387,37 +347,50 @@ func TestASAPv1RejectsCraftedBytes(t *testing.T) { }) } - if err := new(DDSketch).UnmarshalASAPv1(craft(1, testAlpha, valid)); err != nil { + if err := new(DDSketch).UnmarshalASAPv1(craft(dd, 1, testAlpha, valid)); err != nil { t.Fatalf("control payload rejected: %v", err) } - boundary := craft(1, testAlpha, v1Payload([]uint64{1, 1, 1}, math.MaxInt32-2, 5, 1, 2)) + boundary := craft(dd, 1, testAlpha, bucketsPayload([]uint64{1, 1, 1}, math.MaxInt32-2, nil, 0, 0)) if err := new(DDSketch).UnmarshalASAPv1(boundary); err != nil { t.Fatalf("store ending at MaxInt32 rejected: %v", err) } } -func TestASAPv1PositiveOnlyVersion2DecodesAndEncodesAsVersion1(t *testing.T) { - var d DDSketch - if err := d.UnmarshalASAPv1(craft(2, testAlpha, v2Payload([]uint64{3}, 7, 6, 1.5, 2.5, nil, 0, 0))); err != nil { - t.Fatal(err) - } - got, err := d.MarshalASAPv1() - if err != nil { - t.Fatal(err) +// The retired kind's scalars are read and dropped without any check, so +// values the old decoder refused still decode, into the buckets alone. +func TestASAPv1RetiredScalarsAreDropped(t *testing.T) { + want := literal(testAlpha, []uint64{3}, 7, []uint64{1}, -2, 4) + for name, scalars := range map[string][3]float64{ + "consistent": {6, -1.5, 2.5}, + "NaN": {math.NaN(), math.NaN(), math.NaN()}, + "min > max": {0, 9, -9}, + "empty-style": {0, math.Inf(1), math.Inf(-1)}, + } { + t.Run(name, func(t *testing.T) { + b := craft(asapv1.KindDDSketchWithScalars, 1, testAlpha, + scalarsPayload([]uint64{3}, 7, scalars[0], scalars[1], scalars[2], []uint64{1}, -2, 4)) + var got DDSketch + if err := got.UnmarshalASAPv1(b); err != nil { + t.Fatal(err) + } + if !sameState(&got, want) { + t.Fatalf("decoded state differs\n got: %+v\nwant: %+v", got, *want) + } + again, err := got.MarshalASAPv1() + if err != nil { + t.Fatal(err) + } + asapv1test.Equal(t, again, craft(asapv1.KindDDSketch, 1, testAlpha, + bucketsPayload([]uint64{3}, 7, []uint64{1}, -2, 4))) + }) } - asapv1test.Equal(t, got, craft(1, testAlpha, v1Payload([]uint64{3}, 7, 6, 1.5, 2.5))) } func TestASAPv1MarshalRejectsUndecodableState(t *testing.T) { cases := map[string]*DDSketch{ - "span past int32": literal(testAlpha, []uint64{1, 1, 1, 1}, math.MaxInt32-2, nil, 0, 0, 5, 1, 2), - "total overflow": literal(testAlpha, []uint64{math.MaxUint64, 1}, 0, nil, 0, 0, 5, 1, 2), - "infinite sum": literal(testAlpha, []uint64{1}, 0, nil, 0, 0, math.Inf(1), 1, 2), - "sum below min": literal(testAlpha, []uint64{3}, 7, nil, 0, 0, 0.5, 1.5, 2.5), - "drained to empty": literal(testAlpha, []uint64{0, 0}, 7, nil, 0, 0, 6, 1.5, 2.5), - "signed min > max": literal(testAlpha, []uint64{1}, 0, []uint64{1}, 0, 0, 5, 3, -1), - "signed span": literal(testAlpha, []uint64{1}, 0, []uint64{1, 1, 1, 1}, math.MaxInt32-2, 0, 5, -2, 2), - "zero count with NaN": literal(testAlpha, nil, 0, nil, 0, 1, math.NaN(), 0, 0), + "span past int32": literal(testAlpha, []uint64{1, 1, 1, 1}, math.MaxInt32-2, nil, 0, 0), + "total overflow": literal(testAlpha, []uint64{math.MaxUint64, 1}, 0, nil, 0, 0), + "signed span": literal(testAlpha, []uint64{1}, 0, []uint64{1, 1, 1, 1}, math.MaxInt32-2, 0), } for name, d := range cases { t.Run(name, func(t *testing.T) { @@ -428,21 +401,44 @@ func TestASAPv1MarshalRejectsUndecodableState(t *testing.T) { } } -func TestQuantileCapsAtCarriedMax(t *testing.T) { +// q = 0 and q = 1 read the lowest and highest occupied buckets, within alpha +// of the true extremes; there is no exact min / max to clamp to. +func TestQuantileEndpointsAreWithinAlpha(t *testing.T) { d := NewDDSketch(testAlpha) - d.Update(1.0) - for _, q := range []float64{0.5, 1} { - if got, _ := d.Quantile(q); got != 1.0 { - t.Fatalf("Quantile(%v) = %v, want 1.0", q, got) + values := []float64{0.37, 12, 15.5, 230, 1200} + for _, v := range values { + d.Update(v) + } + lo, _ := d.Quantile(0) + hi, _ := d.Quantile(1) + if lo != d.mapping.Value(d.mapping.Index(0.37)) || math.Abs(lo-0.37) > testAlpha*0.37 { + t.Fatalf("Quantile(0) = %v, want the bucket of 0.37 within alpha", lo) + } + if hi != d.mapping.Value(d.mapping.Index(1200)) || math.Abs(hi-1200) > testAlpha*1200 { + t.Fatalf("Quantile(1) = %v, want the bucket of 1200 within alpha", hi) + } + for _, q := range []float64{-0.1, 1.1, math.NaN()} { + if _, ok := d.Quantile(q); ok { + t.Fatalf("Quantile(%v) ok, want false", q) } } +} - top := NewDDSketch(testAlpha) - v := top.mapping.LowerBound(11) * 0.9999999 - top.Update(top.mapping.LowerBound(10)) - top.Update(v) - if got, _ := top.Quantile(1); got != v { - t.Fatalf("Quantile(1) = %v, want max %v", got, v) +// A sketch rebuilt from its buckets alone (the Octo aggregator path) answers +// every quantile exactly as the sketch the values were added to. +func TestBucketFedSketchMatchesTheProducer(t *testing.T) { + src := NewDDSketch(testAlpha) + for _, v := range []float64{12, 15.5, 9.8, 230, 14.1, 18.9, 1200} { + src.Update(v) + } + agg := NewDDSketch(testAlpha) + src.EachBucket(func(k int32, c uint64) { agg.AddToBucket(k, c) }) + for _, q := range []float64{0, 0.25, 0.5, 0.99, 1} { + a, _ := agg.Quantile(q) + s, _ := src.Quantile(q) + if a != s { + t.Fatalf("Quantile(%v): aggregator %v, producer %v", q, a, s) + } } } @@ -451,11 +447,12 @@ func TestQuantileCapsAtCarriedMax(t *testing.T) { // Rust's powf can differ in the last bits, so bucket values get a tolerance. func TestSignedFixtureQuantilesMatchRust(t *testing.T) { var d DDSketch - if err := d.UnmarshalASAPv1(asapv1test.Golden(t, "ddsketch_signed_a001")); err != nil { + if err := d.UnmarshalASAPv1(asapv1test.Golden(t, "ddsketch_buckets_signed_a001")); err != nil { t.Fatal(err) } for q, want := range map[float64]float64{ - 0.05: -0.016, 0.1: -0.01576144762907453, 0.4: 0, 0.75: 497.77940145581556, 0.9: 515.0, 1: 515.0, + 0: -0.016079860712490172, 0.05: -0.016079860712490172, 0.1: -0.01576144762907453, 0.4: 0, + 0.75: 497.77940145581556, 0.9: 518.0948550403809, 1: 518.0948550403809, } { got, _ := d.Quantile(q) if math.Abs(got-want) > 1e-13*math.Abs(want) { @@ -473,109 +470,6 @@ func TestMappingUsesConstructorAlpha(t *testing.T) { } } -func TestASAPv1EmptiedWorkerEncodesAsEmpty(t *testing.T) { - fill := func() *DDSketch { - d := NewDDSketch(testAlpha) - d.Update(5) - d.Update(5) - return d - } - drained := fill() - drained.DrainBuckets(func(int32, uint64) {}) - reset := fill() - reset.ResetBucket(reset.BucketIndex(5)) - crossed := NewDDSketch(testAlpha) - crossed.UpdateGOS(5, 1) - - empty := NewDDSketch(testAlpha) - for name, d := range map[string]*DDSketch{"DrainBuckets": drained, "ResetBucket": reset, "UpdateGOS": crossed} { - t.Run(name, func(t *testing.T) { - b, err := d.MarshalASAPv1() - if err != nil { - t.Fatalf("MarshalASAPv1: %v", err) - } - var got DDSketch - if err := got.UnmarshalASAPv1(b); err != nil { - t.Fatalf("UnmarshalASAPv1: %v", err) - } - if got.count != 0 || !sameFloat(got.sum, empty.sum) || - !sameFloat(got.min, empty.min) || !sameFloat(got.max, empty.max) { - t.Fatalf("decoded count %d sum %v min %v max %v, want the empty state", - got.count, got.sum, got.min, got.max) - } - }) - } -} - -func TestASAPv1SumAfterBucketInserts(t *testing.T) { - d := NewDDSketch(testAlpha) - d.AddToBucket(10, 3) - d.addOneFast(10) - d.addOneFast(20) - m := d.mapping - want := 0.0 - want += m.Value(10) * 3 - want += m.Value(10) - want += m.Value(20) - got := roundTrip(t, d) - if !sameFloat(got.sum, want) { - t.Fatalf("encoded sum %v, want %v", got.sum, want) - } -} - -func TestAddOneFastSumAcrossGrowthCollapseAndDecode(t *testing.T) { - check := func(t *testing.T, d *DDSketch, ks []int32) { - t.Helper() - want := d.sum - for _, k := range ks { - d.addOneFast(k) - want += d.mapping.Value(k) - } - if !sameFloat(d.sum, want) { - t.Fatalf("sum %v, want %v", d.sum, want) - } - } - grown := NewDDSketch(testAlpha) - grown.AddToBucket(0, 1) - check(t, grown, []int32{0, 0, 5, 5, 0}) - grown.AddToBucket(-500, 1) - check(t, grown, []int32{0, -500, 0, 5, 5}) - grown.AddToBucket(repCacheSize, 1) - check(t, grown, []int32{0, repCacheSize, 0, repCacheSize, -500}) - - collapsed := NewDDSketchWithMaxBins(testAlpha, 8) - collapsed.AddToBucket(0, 1) - check(t, collapsed, []int32{0, 3, 3}) - collapsed.AddToBucket(100, 1) - check(t, collapsed, []int32{100, 95, 95, 100}) - - b, err := NewDDSketch(0.05).MarshalASAPv1() - if err != nil { - t.Fatal(err) - } - if err := grown.UnmarshalASAPv1(b); err != nil { - t.Fatal(err) - } - grown.AddToBucket(5, 1) - check(t, grown, []int32{5, 5}) -} - -func TestMergeSumSaturates(t *testing.T) { - a, b := NewDDSketch(testAlpha), NewDDSketch(testAlpha) - v := a.mapping.MaxIndexableValue() - for _, d := range []*DDSketch{a, b} { - d.Update(v) - d.Update(v) - } - if err := a.Merge(b); err != nil { - t.Fatal(err) - } - if a.sum != math.MaxFloat64 { - t.Fatalf("merged sum %v, want MaxFloat64", a.sum) - } - roundTrip(t, a) -} - func TestASAPv1EncodesAggregatorFedByAddToBucket(t *testing.T) { src := NewDDSketch(testAlpha) for i := 1; i <= 500; i++ { diff --git a/wire/asapv1/kind.go b/wire/asapv1/kind.go index 90137f5..ab97660 100644 --- a/wire/asapv1/kind.go +++ b/wire/asapv1/kind.go @@ -12,14 +12,18 @@ type KindID string // The kind_id registry. An id is allocated once and never reassigned; ids // without a designed payload are reserved. const ( - KindHLLUnspecified KindID = "\x01\x00" // reserved - KindHLLClassic KindID = "\x01\x01" - KindHLLErtlMLE KindID = "\x01\x02" - KindHLLHIP KindID = "\x01\x03" - KindCountMin KindID = "\x02\x00" - KindCMSHeap KindID = "\x03\x00" - KindCountSketch KindID = "\x04\x00" - KindDDSketch KindID = "\x05\x00" + KindHLLUnspecified KindID = "\x01\x00" // reserved + KindHLLClassic KindID = "\x01\x01" + KindHLLErtlMLE KindID = "\x01\x02" + KindHLLHIP KindID = "\x01\x03" + KindCountMin KindID = "\x02\x00" + KindCMSHeap KindID = "\x03\x00" + KindCountSketch KindID = "\x04\x00" + KindDDSketch KindID = "\x05\x01" + // KindDDSketchWithScalars is the retired DDSketch kind whose payload also + // carried sum / min / max: decoded with those scalars dropped, never + // encoded. + KindDDSketchWithScalars KindID = "\x05\x00" KindKLL KindID = "\x06\x00" KindKLLDynamic KindID = "\x06\x01" KindHydraKLL KindID = "\x07\x00" @@ -60,6 +64,7 @@ var kindNames = map[KindID]string{ KindCMSHeap: "CMSHeap", KindCountSketch: "Count Sketch", KindDDSketch: "DDSketch", + KindDDSketchWithScalars: "DDSketch with scalars", KindKLL: "KLL", KindKLLDynamic: "KLL Dynamic", KindHydraKLL: "Hydra KLL", diff --git a/wire/asapv1/kind_test.go b/wire/asapv1/kind_test.go index 95d0ac6..012ab2c 100644 --- a/wire/asapv1/kind_test.go +++ b/wire/asapv1/kind_test.go @@ -3,8 +3,8 @@ package asapv1 import "testing" func TestRegistry(t *testing.T) { - if len(kindNames) != 37 { - t.Fatalf("registry has %d ids, want 37", len(kindNames)) + if len(kindNames) != 38 { + t.Fatalf("registry has %d ids, want 38", len(kindNames)) } for kind, name := range kindNames { if len(kind) != 2 || name == "" { @@ -12,14 +12,16 @@ func TestRegistry(t *testing.T) { } } cases := map[KindID]string{ - KindHLLClassic: "01 01 (HLL Classic)", - KindCountMin: "02 00 (Count-Min)", - KindCSHeap: "0a 00 (CSHeap)", - KindUnivMonQ: "1a 00 (UnivMon-Q)", - KindRetired1D: "1d 00 (retired)", - "\x1e\x00": "1e 00 (unregistered)", - "": " (unregistered)", - KindHydraUnivMon: "07 04 (Hydra UnivMon)", + KindHLLClassic: "01 01 (HLL Classic)", + KindCountMin: "02 00 (Count-Min)", + KindCSHeap: "0a 00 (CSHeap)", + KindDDSketch: "05 01 (DDSketch)", + KindDDSketchWithScalars: "05 00 (DDSketch with scalars)", + KindUnivMonQ: "1a 00 (UnivMon-Q)", + KindRetired1D: "1d 00 (retired)", + "\x1e\x00": "1e 00 (unregistered)", + "": " (unregistered)", + KindHydraUnivMon: "07 04 (Hydra UnivMon)", } for kind, want := range cases { if got := kind.String(); got != want { From 1d5ca3d27d9a74e19fc1f710eb4536a02b171c36 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 28 Oct 2026 22:43:45 +0000 Subject: [PATCH 2/2] fix(ddsketch): Quantile's fallback reads the last occupied bucket of any sign If count ever exceeded the buckets, Quantile returned the last positive bucket, or 0 when the positive store was empty. It now returns the last occupied bucket walked (negative, zero or positive), as Rust's value_at_rank does. Co-Authored-By: Claude Opus 5.5 --- sketches/DDSketch/DDSketch.go | 16 +++++++++++----- sketches/DDSketch/asapv1_test.go | 10 ++++++++++ 2 files changed, 21 insertions(+), 5 deletions(-) diff --git a/sketches/DDSketch/DDSketch.go b/sketches/DDSketch/DDSketch.go index 43ee473..a11030e 100644 --- a/sketches/DDSketch/DDSketch.go +++ b/sketches/DDSketch/DDSketch.go @@ -534,7 +534,10 @@ func (d *DDSketch) Quantile(q float64) (float64, bool) { rank := max(uint64(math.Ceil(q*float64(d.count))), 1) rank = min(rank, d.count) + // last is the last occupied bucket walked, returned if the buckets ever + // held fewer samples than count says (no code path allows it). var seen uint64 + var last float64 if !d.negative.IsEmpty() { neg := d.negative.counts.AsSlice() for i := len(neg) - 1; i >= 0; i-- { @@ -542,17 +545,20 @@ func (d *DDSketch) Quantile(q float64) (float64, bool) { continue } seen += neg[i] + last = -d.mapping.Value(d.negative.offset + int32(i)) if seen >= rank { - return -d.mapping.Value(d.negative.offset + int32(i)), true + return last, true } } } - seen += d.zeroCount - if seen >= rank { - return 0, true + if d.zeroCount > 0 { + seen += d.zeroCount + last = 0 + if seen >= rank { + return last, true + } } - var last float64 if !d.store.IsEmpty() { for i, c := range d.store.counts.AsSlice() { if c == 0 { diff --git a/sketches/DDSketch/asapv1_test.go b/sketches/DDSketch/asapv1_test.go index e1953d1..3c989ae 100644 --- a/sketches/DDSketch/asapv1_test.go +++ b/sketches/DDSketch/asapv1_test.go @@ -442,6 +442,16 @@ func TestBucketFedSketchMatchesTheProducer(t *testing.T) { } } +// If count ever exceeded the buckets, Quantile reads the last occupied +// bucket, a negative one included. +func TestQuantilePastTheBucketsReadsTheLastOccupiedBucket(t *testing.T) { + d := literal(testAlpha, nil, 0, []uint64{2}, 30, 0) + d.count += 5 + if got, _ := d.Quantile(1); got != -d.mapping.Value(30) { + t.Fatalf("Quantile(1) = %v, want %v", got, -d.mapping.Value(30)) + } +} + // TestSignedFixtureQuantilesMatchRust compares against the values Rust's // get_value_at_quantile returns for the decoded signed fixture; math.Pow and // Rust's powf can differ in the last bits, so bucket values get a tolerance.