Skip to content

pkg/serializer/internal/metrics: replace Cols() with Range() in V3 sketch path to eliminate allocations #52498

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

Open
AlexanderYastrebov wants to merge 2 commits into
base: main
Choose a base branch
from
Open

pkg/serializer/internal/metrics: replace Cols() with Range() in V3 sketch path to eliminate allocations #52498

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a1e8df3
pkg/serializer/internal/metrics: add writeSketch sub-benchmark to Ben…
AlexanderYastrebov Jun 19, 2026
53911db
pkg/serializer/internal/metrics: replace Cols() with Range() in V3 sk…
AlexanderYastrebov Jun 19, 2026
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
8 changes: 8 additions & 0 deletions comp/dogstatsd/http/impl/sketch_iterator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,14 @@ func (s *sketchData) Cols() ([]int32, []uint32) {
return s.k, s.n
}

func (s *sketchData) Range(f func(k int32, n uint32) bool) {
for i, k := range s.k {
if !f(k, s.n[i]) {

@RiantZ RiantZ Jun 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are iterating here s.k and accessing к s.n[i]
If len(s.n) < len(s.k) will it lead to panic? or such case is impossible?

@AlexanderYastrebov AlexanderYastrebov Jun 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

k and n are the same size by construction, e.g. clients iterate them in lockstep when they are obtained via existing Cols method.

@RiantZ RiantZ Jun 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My comment was driven by the fact that the relationship between the sizes of the two arrays is not immediately obvious. In the future, if this code is reused or modified, it could be confusing.

return
}
}
}

func (s *sketchData) BasicStats() (int64, float64, float64, float64, float64) {
return s.cnt, s.min, s.max, s.sum, s.avg
}
Expand Down
3 changes: 3 additions & 0 deletions pkg/metrics/sketch_series.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,9 @@ func (sl SketchSeries) String() string {
type SketchData interface {
// Cols returns bin keys and per-bin counts in ascending key order.
Cols() (k []int32, n []uint32)
// Range calls f for each (k, n) bin pair in ascending key order.
// If f returns false, range stops the iteration.
Range(f func(k int32, n uint32) bool)
// BasicStats returns the five summary fields used in the wire format.
BasicStats() (cnt int64, min, max, sum, avg float64)
}
Expand Down
14 changes: 8 additions & 6 deletions pkg/serializer/internal/metrics/iterable_series_v3.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -577,13 +577,15 @@ func (pb *payloadsBuilderV3) writeSketchToTxn(sketch *metrics.SketchSeries) {
pb.txn.Sint64(columnValueSint64, bCnt)
pb.stats.valuesSint64++

k, n := pnt.Sketch.Cols()
kDelta := deltaEncoder{}
for i := range k {
pb.txn.Sint64(columnSketchBinKeys, kDelta.encode(int64(k[i])))
pb.txn.Uint64(columnSketchBinCnts, uint64(n[i]))
}
pb.txn.Uint64(columnSketchNumBins, uint64(len(k)))
numBins := uint64(0)
pnt.Sketch.Range(func(k int32, n uint32) bool {
pb.txn.Sint64(columnSketchBinKeys, kDelta.encode(int64(k)))
pb.txn.Uint64(columnSketchBinCnts, uint64(n))
numBins++
Comment thread
RiantZ marked this conversation as resolved.
return true
})
pb.txn.Uint64(columnSketchNumBins, numBins)
}
}

Expand Down
62 changes: 44 additions & 18 deletions pkg/serializer/internal/metrics/iterable_series_v3_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -238,33 +238,59 @@ func TestPayloadsBuilderV3_Unit(t *testing.T) {

func BenchmarkPayloadsBuilderV3(b *testing.B) {
const ts = 1756737057.1
serie := &metrics.Serie{
Name: "serie1",
Tags: tagset.NewCompositeTags([]string{"foo", "bar"}, []string{"ook", "eek"}),
Points: []metrics.Point{{Ts: ts, Value: 3.14}}}

pipelineConfig := PipelineConfig{
Filter: AllowAllFilter{},
V3: true,
}
pipelineContext := &PipelineContext{}

pb, err := newPayloadsBuilderV3(500_000, 2_000_000, 10_000, noopimpl.New(), pipelineConfig, pipelineContext)
if err != nil {
b.Fatalf("new: %v", err)
}
b.Run("writeSerie", func(b *testing.B) {
serie := &metrics.Serie{
Name: "serie1",
Tags: tagset.NewCompositeTags([]string{"foo", "bar"}, []string{"ook", "eek"}),
Points: []metrics.Point{{Ts: ts, Value: 3.14}}}

b.ReportAllocs()
b.ResetTimer()
pipelineContext := &PipelineContext{}
pb, err := newPayloadsBuilderV3(500_000, 2_000_000, 10_000, noopimpl.New(), pipelineConfig, pipelineContext)
if err != nil {
b.Fatalf("new: %v", err)
}

for i := 0; i < b.N; i++ {
err = pb.writeSerie(serie)
pipelineContext.payloads = pipelineContext.payloads[:]
}
b.ReportAllocs()
b.ResetTimer()

if err != nil {
b.Fatalf("writeSerie: %v", err)
}
for i := 0; i < b.N; i++ {
err = pb.writeSerie(serie)
pipelineContext.payloads = pipelineContext.payloads[:]
}

if err != nil {
b.Fatalf("writeSerie: %v", err)
}
})

b.Run("writeSketch", func(b *testing.B) {
// Makeseries(200) produces a sketch with 200 bins and 205 points — realistic distribution shape.
sketch := Makeseries(200)

pipelineContext := &PipelineContext{}
pb, err := newPayloadsBuilderV3(500_000, 2_000_000, 10_000, noopimpl.New(), pipelineConfig, pipelineContext)
if err != nil {
b.Fatalf("new: %v", err)
}

b.ReportAllocs()
b.ResetTimer()

for i := 0; i < b.N; i++ {
err = pb.writeSketch(sketch)
pipelineContext.payloads = pipelineContext.payloads[:]
}

if err != nil {
b.Fatalf("writeSketch: %v", err)
}
})
}

func TestPayloadBuildersV3_Split(t *testing.T) {
Expand Down
10 changes: 10 additions & 0 deletions pkg/util/quantile/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,16 @@ func (s *sparseStore) Cols() (k []int32, n []uint32) {
return
}

// Range calls f for each (k, n) bin pair in ascending key order.
// If f returns false, range stops the iteration.
func (s *sparseStore) Range(f func(k int32, n uint32) bool) {
for _, b := range s.bins {
if !f(int32(b.k), uint32(b.n)) {
return
}
}
}

// MemSize returns memory use in bytes:
//
// used: uses len(bins)
Expand Down
43 changes: 43 additions & 0 deletions pkg/util/quantile/store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,3 +220,46 @@ func TestCols(t *testing.T) {
assert.Equal(t, n, tt.n, "values don't match")
}
}

func TestRange(t *testing.T) {
for _, tt := range []struct {
store string
k []int32
n []uint32
}{
{
store: "",
},
{
store: "0:1 1:1 2:2 3:1 4:1 5:1 8:1 9:1 10:max",
k: []int32{0, 1, 2, 3, 4, 5, 8, 9, 10},
n: []uint32{1, 1, 2, 1, 1, 1, 1, 1, math.MaxUint16},
},
{
store: "0:1 0:max",
k: []int32{0, 0},
n: []uint32{1, math.MaxUint16},
},
} {
st := buildStore(t, tt.store)
var k []int32
var n []uint32
st.Range(func(bk int32, bn uint32) bool {
k = append(k, bk)
n = append(n, bn)
return true
})
assert.Equal(t, tt.k, k, "keys don't match")
assert.Equal(t, tt.n, n, "values don't match")
}

t.Run("early stop", func(t *testing.T) {
st := buildStore(t, "0:1 1:1 2:1")
var k []int32
st.Range(func(bk int32, _ uint32) bool {
k = append(k, bk)
return bk < 1
})
assert.Equal(t, []int32{0, 1}, k)
})
}
Loading