Merge pull request #13289 from fpetkovski/fix-histogram-reuse

Fix reusing float histograms

promql/bench_test.go

@@ -296,8 +296,12 @@ func BenchmarkNativeHistograms(b *testing.B) {
 			query: "sum(native_histogram_series)",
 		},
 		{
-			name:  "sum rate",
+			name:  "sum rate with short rate interval",
-			query: "sum(rate(native_histogram_series[1m]))",
+			query: "sum(rate(native_histogram_series[2m]))",
+		},
+		{
+			name:  "sum rate with long rate interval",
+			query: "sum(rate(native_histogram_series[20m]))",
 		},
 	}
 

promql/engine.go

@@ -2052,7 +2052,12 @@ func (ev *evaluator) matrixIterSlice(
 		var drop int
 		for drop = 0; histograms[drop].T < mint; drop++ {
 		}
+		// Rotate the buffer around the drop index so that points before mint can be
+		// reused to store new histograms.
+		tail := make([]HPoint, drop)
+		copy(tail, histograms[:drop])
 		copy(histograms, histograms[drop:])
+		copy(histograms[len(histograms)-drop:], tail)
 		histograms = histograms[:len(histograms)-drop]
 		ev.currentSamples -= totalHPointSize(histograms)
 		// Only append points with timestamps after the last timestamp we have.
@@ -2121,18 +2126,23 @@ loop:
 	// The sought sample might also be in the range.
 	switch soughtValueType {
 	case chunkenc.ValFloatHistogram, chunkenc.ValHistogram:
-		t, h := it.AtFloatHistogram()
+		t := it.AtT()
-		if t == maxt && !value.IsStaleNaN(h.Sum) {
+		if t == maxt {
+			_, h := it.AtFloatHistogram()
+			if !value.IsStaleNaN(h.Sum) {
 				if ev.currentSamples >= ev.maxSamples {
 					ev.error(ErrTooManySamples(env))
 				}
 				if histograms == nil {
 					histograms = getHPointSlice(16)
 				}
-			point := HPoint{T: t, H: h}
+				// The last sample comes directly from the iterator, so we need to copy it to
+				// avoid having the same reference twice in the buffer.
+				point := HPoint{T: t, H: h.Copy()}
 				histograms = append(histograms, point)
 				ev.currentSamples += point.size()
 			}
+		}
 	case chunkenc.ValFloat:
 		t, f := it.At()
 		if t == maxt && !value.IsStaleNaN(f) {

promql/engine_test.go

@@ -3169,7 +3169,8 @@ func TestNativeHistogramRate(t *testing.T) {
 	}
 	require.NoError(t, app.Commit())
 
-	queryString := fmt.Sprintf("rate(%s[1m])", seriesName)
+	queryString := fmt.Sprintf("rate(%s[45s])", seriesName)
+	t.Run("instant_query", func(t *testing.T) {
 		qry, err := engine.NewInstantQuery(context.Background(), storage, nil, queryString, timestamp.Time(int64(5*time.Minute/time.Millisecond)))
 		require.NoError(t, err)
 		res := qry.Exec(context.Background())
@@ -3184,13 +3185,59 @@ func TestNativeHistogramRate(t *testing.T) {
 			ZeroThreshold:    0.001,
 			ZeroCount:        1. / 15.,
 			Count:            9. / 15.,
-		Sum:              1.226666666666667,
+			Sum:              1.2266666666666663,
 			PositiveSpans:    []histogram.Span{{Offset: 0, Length: 2}, {Offset: 1, Length: 2}},
 			PositiveBuckets:  []float64{1. / 15., 1. / 15., 1. / 15., 1. / 15.},
 			NegativeSpans:    []histogram.Span{{Offset: 0, Length: 2}, {Offset: 1, Length: 2}},
 			NegativeBuckets:  []float64{1. / 15., 1. / 15., 1. / 15., 1. / 15.},
 		}
 		require.Equal(t, expectedHistogram, actualHistogram)
+	})
+
+	t.Run("range_query", func(t *testing.T) {
+		step := 30 * time.Second
+		start := timestamp.Time(int64(5 * time.Minute / time.Millisecond))
+		end := start.Add(step)
+		qry, err := engine.NewRangeQuery(context.Background(), storage, nil, queryString, start, end, step)
+		require.NoError(t, err)
+		res := qry.Exec(context.Background())
+		require.NoError(t, res.Err)
+		matrix, err := res.Matrix()
+		require.NoError(t, err)
+		require.Len(t, matrix, 1)
+		require.Len(t, matrix[0].Histograms, 2)
+		actualHistograms := matrix[0].Histograms
+		expectedHistograms := []HPoint{{
+			T: 300000,
+			H: &histogram.FloatHistogram{
+				CounterResetHint: histogram.GaugeType,
+				Schema:           1,
+				ZeroThreshold:    0.001,
+				ZeroCount:        1. / 15.,
+				Count:            9. / 15.,
+				Sum:              1.2266666666666663,
+				PositiveSpans:    []histogram.Span{{Offset: 0, Length: 2}, {Offset: 1, Length: 2}},
+				PositiveBuckets:  []float64{1. / 15., 1. / 15., 1. / 15., 1. / 15.},
+				NegativeSpans:    []histogram.Span{{Offset: 0, Length: 2}, {Offset: 1, Length: 2}},
+				NegativeBuckets:  []float64{1. / 15., 1. / 15., 1. / 15., 1. / 15.},
+			},
+		}, {
+			T: 330000,
+			H: &histogram.FloatHistogram{
+				CounterResetHint: histogram.GaugeType,
+				Schema:           1,
+				ZeroThreshold:    0.001,
+				ZeroCount:        1. / 15.,
+				Count:            9. / 15.,
+				Sum:              1.2266666666666663,
+				PositiveSpans:    []histogram.Span{{Offset: 0, Length: 2}, {Offset: 1, Length: 2}},
+				PositiveBuckets:  []float64{1. / 15., 1. / 15., 1. / 15., 1. / 15.},
+				NegativeSpans:    []histogram.Span{{Offset: 0, Length: 2}, {Offset: 1, Length: 2}},
+				NegativeBuckets:  []float64{1. / 15., 1. / 15., 1. / 15., 1. / 15.},
+			},
+		}}
+		require.Equal(t, expectedHistograms, actualHistograms)
+	})
 }
 
 func TestNativeFloatHistogramRate(t *testing.T) {

tsdb/tsdbutil/histogram.go

@@ -30,6 +30,14 @@ func GenerateTestHistograms(n int) (r []*histogram.Histogram) {
 	return r
 }
 
+func GenerateTestHistogramsWithUnknownResetHint(n int) []*histogram.Histogram {
+	hs := GenerateTestHistograms(n)
+	for i := range hs {
+		hs[i].CounterResetHint = histogram.UnknownCounterReset
+	}
+	return hs
+}
+
 // GenerateTestHistogram but it is up to the user to set any known counter reset hint.
 func GenerateTestHistogram(i int) *histogram.Histogram {
 	return &histogram.Histogram{