ringbuffer: revert interface changes (#644)

Signed-off-by: Michael Hoffmann <[email protected]>

Author: MichaHoffmann

engine/engine_test.go

@@ -2601,6 +2601,128 @@ func TestEdgeCases(t *testing.T) {
 	}
 }
 
+func TestXFunctionsRangeQuery(t *testing.T) {
+	// Negative offset and at modifier are enabled by default
+	// since Prometheus v2.33.0, so we also enable them.
+	opts := promql.EngineOpts{
+		Timeout:              1 * time.Hour,
+		MaxSamples:           1e10,
+		EnableNegativeOffset: true,
+		EnableAtModifier:     true,
+	}
+
+	cases := []struct {
+		name      string
+		load      string
+		query     string
+		startTime time.Time
+		endTime   time.Time
+		step      time.Duration
+
+		expected promql.Matrix
+	}{
+		{
+			name: "gaps between steps",
+			load: `load 10s
+			    http_requests 1 5 10 20 _ 40`,
+			query: "xincrease(http_requests[10s])",
+
+			startTime: time.Unix(0, 0),
+			endTime:   time.Unix(60, 0),
+			step:      20 * time.Second,
+
+			expected: promql.Matrix{
+				promql.Series{
+					Metric: labels.New(),
+					Floats: []promql.FPoint{
+						{T: 00_000, F: 1},
+						{T: 20_000, F: 9}, // TODO: this seems odd, feels like it should be 5
+						{T: 40_000, F: 0},
+						{T: 60_000, F: 0},
+					},
+				},
+			},
+		},
+		{
+			name: "back to back steps",
+			load: `load 10s
+			    http_requests 1 5 10 20 _ 40`,
+			query: "xincrease(http_requests[10s])",
+
+			startTime: time.Unix(0, 0),
+			endTime:   time.Unix(60, 0),
+			step:      10 * time.Second,
+
+			expected: promql.Matrix{
+				promql.Series{
+					Metric: labels.New(),
+					Floats: []promql.FPoint{
+						{T: 00_000, F: 1},
+						{T: 10_000, F: 4},
+						{T: 20_000, F: 5},
+						{T: 30_000, F: 10},
+						{T: 40_000, F: 0},
+						{T: 50_000, F: 20},
+						{T: 60_000, F: 0},
+					},
+				},
+			},
+		},
+		{
+			name: "overlapping steps",
+			load: `load 10s
+			    http_requests 1 5 10 20 _ 40`,
+			query: "xincrease(http_requests[20s])",
+
+			startTime: time.Unix(0, 0),
+			endTime:   time.Unix(60, 0),
+			step:      10 * time.Second,
+
+			expected: promql.Matrix{
+				promql.Series{
+					Metric: labels.New(),
+					Floats: []promql.FPoint{
+						{T: 00_000, F: 1},
+						{T: 10_000, F: 4},
+						{T: 20_000, F: 9},
+						{T: 30_000, F: 15},
+						{T: 40_000, F: 10},
+						{T: 50_000, F: 20},
+						{T: 60_000, F: 20},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			storage := promqltest.LoadedStorage(t, tc.load)
+			defer storage.Close()
+
+			ctx := context.Background()
+			newEngine := engine.New(engine.Opts{
+				EngineOpts:        opts,
+				LogicalOptimizers: logicalplan.AllOptimizers,
+				EnableXFunctions:  true,
+			})
+			query, err := newEngine.NewRangeQuery(ctx, storage, nil, tc.query, tc.startTime, tc.endTime, tc.step)
+			testutil.Ok(t, err)
+			defer query.Close()
+
+			engineResult := query.Exec(ctx)
+			testutil.Ok(t, engineResult.Err)
+
+			gotMatrix, err := engineResult.Matrix()
+			require.NoError(t, err)
+
+			for i := range tc.expected {
+				testutil.WithGoCmp(comparer).Equals(t, tc.expected[i].Floats, gotMatrix[i].Floats, queryExplanation(query))
+			}
+		})
+	}
+}
+
 func TestXFunctionsWithNativeHistograms(t *testing.T) {
 	defaultQueryTime := time.Unix(50, 0)
 

ringbuffer/generic.go

@@ -18,9 +18,12 @@ type Buffer interface {
 	Reset(mint int64, evalt int64)
 	Eval(ctx context.Context, _, _ float64, _ int64) (float64, *histogram.FloatHistogram, bool, error)
 	SampleCount() int
+
+	// to handle extlookback properly, only used by buffers that implement xincrease or xrate
+	ReadIntoLast(f func(*Sample))
 }
 
-func Empty(b Buffer) bool { return b.MaxT() != math.MinInt64 }
+func Empty(b Buffer) bool { return b.MaxT() == math.MinInt64 }
 
 type Value struct {
 	F float64
@@ -80,6 +83,11 @@ func (r *GenericRingBuffer) MaxT() int64 {
 	return r.items[len(r.items)-1].T
 }
 
+// ReadIntoLast reads a sample into the last slot in the buffer, replacing the existing sample.
+func (r *GenericRingBuffer) ReadIntoLast(f func(*Sample)) {
+	f(&r.items[len(r.items)-1])
+}
+
 // Push adds a new sample to the buffer.
 func (r *GenericRingBuffer) Push(t int64, v Value) {
 	n := len(r.items)

ringbuffer/overtime.go

@@ -131,6 +131,8 @@ func (r *CountOverTimeBuffer) Reset(mint int64, evalt int64) {
 	r.firstTimestamps[lastSample] = math.MaxInt64
 }
 
+func (r *CountOverTimeBuffer) ReadIntoLast(func(*Sample)) {}
+
 func (r *CountOverTimeBuffer) Eval(ctx context.Context, _, _ float64, _ int64) (float64, *histogram.FloatHistogram, bool, error) {
 	if r.firstTimestamps[0] == math.MaxInt64 {
 		return 0, nil, false, nil

ringbuffer/rate.go

@@ -192,6 +192,8 @@ func (r *RateBuffer) Eval(ctx context.Context, _, _ float64, _ int64) (float64,
 	return extrapolatedRate(ctx, r.rateBuffer, numSamples, r.isCounter, r.isRate, r.evalTs, r.selectRange, r.offset)
 }
 
+func (r *RateBuffer) ReadIntoLast(func(*Sample)) {}
+
 func querySteps(o query.Options) int64 {
 	// Instant evaluation is executed as a range evaluation with one step.
 	if o.Step.Milliseconds() == 0 {

storage/prometheus/matrix_selector.go

@@ -183,7 +183,8 @@ func (o *matrixSelector) Next(ctx context.Context) ([]model.StepVector, error) {
 		for currStep := 0; currStep < o.numSteps && seriesTs <= o.maxt; currStep++ {
 			maxt := seriesTs - o.offset
 			mint := maxt - o.selectRange
-			if err := scanner.selectPoints(mint, maxt, seriesTs, o.fhReader, o.extLookbackDelta, o.isExtFunction); err != nil {
+
+			if err := scanner.selectPoints(mint, maxt, seriesTs, o.fhReader, o.isExtFunction); err != nil {
 				return nil, err
 			}
 			// TODO(saswatamcode): Handle multi-arg functions for matrixSelectors.
@@ -308,7 +309,6 @@ func (o *matrixSelector) String() string {
 func (m *matrixScanner) selectPoints(
 	mint, maxt, evalt int64,
 	fh *histogram.FloatHistogram,
-	extLookbackDelta int64,
 	isExtFunction bool,
 ) error {
 	m.buffer.Reset(mint, evalt)
@@ -319,19 +319,14 @@ func (m *matrixScanner) selectPoints(
 	if bufMaxt := m.buffer.MaxT() + 1; bufMaxt > mint {
 		mint = bufMaxt
 	}
-	mint = maxInt64(mint, m.buffer.MaxT()+1)
+	mint = max(mint, m.buffer.MaxT()+1)
 	if m.lastSample.T > mint {
 		m.buffer.Push(m.lastSample.T, m.lastSample.V)
 		m.lastSample.T = math.MinInt64
-		mint = maxInt64(mint, m.buffer.MaxT()+1)
+		mint = max(mint, m.buffer.MaxT()+1)
 	}
 
-	var (
-		// The sample that we add for x-functions, -1 is a canary value for the situation
-		// where we have no sample in the extended lookback delta
-		extSample = ringbuffer.Sample{T: -1}
-	)
-
+	appendedPointBeforeMint := !ringbuffer.Empty(m.buffer)
 	for valType := m.iterator.Next(); valType != chunkenc.ValNone; valType = m.iterator.Next() {
 		switch valType {
 		case chunkenc.ValHistogram, chunkenc.ValFloatHistogram:
@@ -367,27 +362,21 @@ func (m *matrixScanner) selectPoints(
 				m.lastSample.T, m.lastSample.V.F, m.lastSample.V.H = t, v, nil
 				return nil
 			}
-			if t > mint {
-				if extSample.T != -1 && isExtFunction {
-					m.buffer.Push(extSample.T, ringbuffer.Value{F: extSample.V.F})
-					extSample.T = -1
+			if isExtFunction {
+				if t > mint || !appendedPointBeforeMint {
+					m.buffer.Push(t, ringbuffer.Value{F: v})
+					appendedPointBeforeMint = true
+				} else {
+					m.buffer.ReadIntoLast(func(s *ringbuffer.Sample) {
+						s.T, s.V.F, s.V.H = t, v, nil
+					})
+				}
+			} else {
+				if t > mint {
+					m.buffer.Push(t, ringbuffer.Value{F: v})
 				}
-				m.buffer.Push(t, ringbuffer.Value{F: v})
-				continue
-			}
-			if isExtFunction && t > mint-extLookbackDelta {
-				extSample.T = t
-				extSample.V.F = v
 			}
 		}
 	}
 	return m.iterator.Err()
 }
-
-func maxInt64(a, b int64) int64 {
-	if a > b {
-		return a
-	}
-	return b
-
-}