perf: reduce sparse render decode and merge overhead

pkg/types/encoding/carbonapi_v2/packed_stream_test.go

@@ -55,6 +55,45 @@ func TestPackedStreamDensityAndSplitFields(t *testing.T) {
 	}
 }
 
+func TestPackedStreamMixedMetricOwnership(t *testing.T) {
+	rng := rand.New(rand.NewSource(20261022))
+	msg := &pb.MultiFetchResponse{}
+	// Alternate packed and dense storage, and exercise both buffer growth
+	// and shorter successors. Retained dense arrays must remain independent.
+	for i := 0; i < 60; i++ {
+		n := []int{6000, 6000, 256, 128, 127, 0}[i%6]
+		m := &pb.FetchResponse{Name: string(rune('a' + i)), Values: make([]float64, n), IsAbsent: make([]bool, n)}
+		for j := range m.Values {
+			m.IsAbsent[j] = (i+j)%3 == 0
+			if i%6 == 1 || j%997 == 0 {
+				m.Values[j] = math.Float64frombits(rng.Uint64())
+			}
+		}
+		msg.Metrics = append(msg.Metrics, m)
+	}
+	body, err := msg.MarshalVT()
+	if err != nil {
+		t.Fatal(err)
+	}
+	want, err := RenderDecoder(body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, chunk := range []int{1, 31, 32768} {
+		got, err := RenderStreamDecoderPacked(&fragmentedReader{data: body, chunk: chunk})
+		if err != nil {
+			t.Fatal(err)
+		}
+		// Empty repeated fields retain the reference decoder's nil slices.
+		for i := range want {
+			if want[i].IsAbsent == nil && got[i].IsAbsent != nil {
+				t.Fatalf("metric %d: empty absence field became non-nil", i)
+			}
+		}
+		equalStreamMetrics(t, unpackStream(got), want)
+	}
+}
+
 func BenchmarkSparseStream(b *testing.B) {
 	msg := &pb.MultiFetchResponse{}
 	for i := 0; i < 100; i++ {

pkg/types/encoding/carbonapi_v2/stream.go

@@ -36,6 +36,7 @@ func renderStreamDecoder(src io.Reader, packed bool) ([]types.Metric, error) {
 	defer func() { r.Reset(nil); streamReaders.Put(r) }()
 	s := protoStream{r: r, left: math.MaxInt64}
 	metrics := make([]types.Metric, 0)
+	var absentScratch []bool
 	for {
 		tag, err := s.varint()
 		if err == io.EOF {
@@ -60,22 +61,29 @@ func renderStreamDecoder(src io.Reader, packed bool) ([]types.Metric, error) {
 		if err != nil {
 			return nil, err
 		}
-		child := protoStream{r: r, left: n}
+		child := protoStream{r: r, left: n, absentScratch: absentScratch}
 		m, err := child.metric(packed)
 		if err != nil {
 			return nil, err
 		}
 		s.left -= n
 		if packed {
 			m.PackPoints()
+			// Packed metrics own a bitmap, so their temporary bool storage can
+			// serve the next metric. Dense metrics retain it and must detach.
+			absentScratch = child.absentScratch
+			if m.IsAbsent != nil || cap(absentScratch) > maxStreamPointChunk {
+				absentScratch = nil
+			}
 		}
 		metrics = append(metrics, m)
 	}
 }
 
 type protoStream struct {
-	r    *bufio.Reader
-	left int64
+	r             *bufio.Reader
+	left          int64
+	absentScratch []bool
 }
 
 func validTag(tag uint64) error {
@@ -281,7 +289,12 @@ func (s *protoStream) bools(dst []bool, n int64) ([]bool, error) {
 		return dst, nil
 	}
 	start := len(dst)
-	dst = append(dst, make([]bool, int(n))...)
+	if int(n) <= cap(dst)-start {
+		// read overwrites every byte before any bool is inspected.
+		dst = dst[:start+int(n)]
+	} else {
+		dst = append(dst, make([]bool, int(n))...)
+	}
 	if n == 0 {
 		return dst, nil
 	}
@@ -307,6 +320,9 @@ func (s *protoStream) bools(dst []bool, n int64) ([]bool, error) {
 }
 func (s *protoStream) metric(packed bool) (types.Metric, error) {
 	var m types.Metric
+	if packed {
+		m.IsAbsent = s.absentScratch[:0]
+	}
 	var points types.PointPacker
 	packing := packed
 	for s.left > 0 {
@@ -406,6 +422,10 @@ func (s *protoStream) metric(packed bool) (types.Metric, error) {
 			}
 		}
 	}
+	if len(m.IsAbsent) == 0 {
+		m.IsAbsent = nil
+	}
+	s.absentScratch = m.IsAbsent
 	if packing {
 		points.Finish(&m)
 	}

pkg/types/packed_merge_bench_test.go

@@ -0,0 +1,37 @@
+package types
+
+import (
+	"testing"
+
+	"github.com/bookingcom/carbonapi/pkg/cfg"
+)
+
+func BenchmarkPackedReplicaMerge(b *testing.B) {
+	for _, density := range []string{"zero", "sparse", "dense"} {
+		for _, mode := range []cfg.ReplicaMatchMode{cfg.ReplicaMatchModeNormal, cfg.ReplicaMatchModeCheck, cfg.ReplicaMatchModeMajority} {
+			b.Run(density+"/"+string(mode), func(b *testing.B) {
+				metrics := make([]Metric, 3)
+				for replica := range metrics {
+					m := Metric{StepTime: 60, Values: make([]float64, 10080), IsAbsent: make([]bool, 10080)}
+					for i := range m.Values {
+						m.IsAbsent[i] = (i+replica)%3 == 0
+						if density == "dense" || density == "sparse" && i%997 == 0 {
+							m.Values[i] = 17
+						}
+					}
+					m.PackPoints()
+					metrics[replica] = m
+				}
+				c := cfg.RenderReplicaMismatchConfig{RenderReplicaMatchMode: mode}
+				b.ReportAllocs()
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					m, stats := mergeMetrics(metrics, c)
+					if len(m.Values) != 10080 || stats.MismatchCount != 0 {
+						b.Fatal("invalid merge")
+					}
+				}
+			})
+		}
+	}
+}

pkg/types/packed_points_test.go

@@ -11,6 +11,49 @@ import (
 	"go.uber.org/zap"
 )
 
+func TestPackedZeroBlockMergeOracle(t *testing.T) {
+	rng := rand.New(rand.NewSource(20261022))
+	for seed := 0; seed < 80; seed++ {
+		input := make([][]Metric, 2+seed%3)
+		for replica := range input {
+			n := []int{128, 129, 255, 256, 10080}[seed%5]
+			m := Metric{Name: "zero-blocks", StepTime: 60, Values: make([]float64, n), IsAbsent: make([]bool, n)}
+			if seed%7 == 0 && replica == 1 {
+				m.StepTime = 120
+			}
+			for i := range m.Values {
+				m.IsAbsent[i] = rng.Intn(3) == 0
+				if seed%4 != 0 && i%997 == 0 {
+					m.Values[i] = float64(replica + 1)
+				}
+			}
+			if seed%4 == 2 {
+				m.Values[63] = math.Float64frombits(1 << 63)
+				m.Values[64] = math.Float64frombits(0x7ff8000000000123)
+				m.Values[127] = math.Inf(replica%2*2 - 1)
+			}
+			input[replica] = []Metric{m}
+		}
+		for _, mode := range []cfg.ReplicaMatchMode{cfg.ReplicaMatchModeNormal, cfg.ReplicaMatchModeCheck, cfg.ReplicaMatchModeMajority} {
+			for _, approx := range []bool{false, true} {
+				c := cfg.RenderReplicaMismatchConfig{RenderReplicaMatchMode: mode, RenderReplicaMismatchApproximateCheck: approx}
+				want, ws := MergeMetrics(copyPackedOracleInput(input, false), c, zap.NewNop())
+				for _, mixed := range []bool{false, true} {
+					pending := copyPackedOracleInput(input, true)
+					if mixed {
+						pending[1][0].materializePoints()
+					}
+					got, gs := MergeMetrics(pending, c, zap.NewNop())
+					if gs != ws {
+						t.Fatalf("seed %d: stats differ: %+v / %+v", seed, gs, ws)
+					}
+					assertPackedBits(t, got, want)
+				}
+			}
+		}
+	}
+}
+
 func copyPackedOracleInput(input [][]Metric, packed bool) [][]Metric {
 	out := make([][]Metric, len(input))
 	for i, ms := range input {

pkg/types/types.go

@@ -11,6 +11,7 @@ package types
 import (
 	"errors"
 	"math"
+	"math/bits"
 	"sort"
 
 	"github.com/bookingcom/carbonapi/pkg/cfg"
@@ -319,7 +320,7 @@ func mergeMetrics(metrics []Metric, replicaMismatchConfig cfg.RenderReplicaMisma
 	metric = metrics[0]
 	metric.materializePoints()
 	var cursors []packedPointCursor
-	for _, replica := range metrics[1:] {
+	for _, replica := range metrics {
 		if replica.packed != nil {
 			cursors = make([]packedPointCursor, len(metrics))
 			for i := range cursors {
@@ -337,13 +338,26 @@ func mergeMetrics(metrics []Metric, replicaMismatchConfig cfg.RenderReplicaMisma
 	}
 	valuesForPoint := make([]float64, 0, len(metrics))
 	isMismatchFindConfig := replicaMatchMode != cfg.ReplicaMatchModeNormal
-	for i := range metric.Values {
+	for i := 0; i < len(metric.Values); i++ {
 		if cursors != nil && i&63 == 0 {
-			for j := 1; j < replicaLimit; j++ {
+			for j := 0; j < replicaLimit; j++ {
 				if p := metrics[j].packed; p != nil {
 					cursors[j].advanceToBlock(p, i>>6)
 				}
 			}
+			if metrics[0].packed != nil {
+				if absent, ok := packedZeroBlockAbsent(metrics[:replicaLimit], cursors, i>>6); ok {
+					// Every value in this block is +0. Only absence healing can
+					// change the result; no mismatch comparison is necessary.
+					healedBits := metrics[0].packed.absent[i>>6] &^ absent
+					for healedBits != 0 {
+						metric.IsAbsent[i+bits.TrailingZeros64(healedBits)] = false
+						healedBits &= healedBits - 1
+					}
+					i += min(64, len(metric.Values)-i) - 1
+					continue
+				}
+			}
 		}
 		pointExists := !metric.IsAbsent[i]
 		shouldLookForMismatch := isMismatchFindConfig
@@ -411,6 +425,20 @@ func mergeMetrics(metrics []Metric, replicaMismatchConfig cfg.RenderReplicaMisma
 	}
 }
 
+// packedZeroBlockAbsent combines absence flags only when every compatible
+// replica stores exclusively positive-zero values in this block. Nonzero bits,
+// including negative zero and NaNs, stay on the point-wise merge path.
+func packedZeroBlockAbsent(metrics []Metric, cursors []packedPointCursor, block int) (uint64, bool) {
+	absent := ^uint64(0)
+	for i, m := range metrics {
+		if m.packed == nil || m.packed.values[cursors[i].offset] != 0 {
+			return 0, false
+		}
+		absent &= m.packed.absent[block]
+	}
+	return absent, true
+}
+
 // Info contains metadata about a metric in Graphite.
 type Info struct {
 	Host              string