ringbuffer: simplify interface (#636)

The Len() and PushIntoLast() methods of the ringbuffer interface were
redundant. Len() was only used to check if the buffer is empty, we did
just implement a convenience function that implements this logic using
the already present MaxT method and we adjusted the scanner action to
address the extlookback logic.

Signed-off-by: Michael Hoffmann <mhoffmann@cloudflare.com>

Author: MichaHoffmann

execution/scan/subquery.go

@@ -216,14 +216,14 @@ func (o *subqueryOperator) collect(v model.StepVector, mint int64) {
 	}
 	for i, s := range v.Samples {
 		buffer := o.buffers[v.SampleIDs[i]]
-		if buffer.Len() > 0 && v.T <= buffer.MaxT() {
+		if !ringbuffer.Empty(buffer) && v.T <= buffer.MaxT() {
 			continue
 		}
 		buffer.Push(v.T, ringbuffer.Value{F: s})
 	}
 	for i, s := range v.Histograms {
 		buffer := o.buffers[v.HistogramIDs[i]]
-		if buffer.Len() > 0 && v.T < buffer.MaxT() {
+		if !ringbuffer.Empty(buffer) && v.T < buffer.MaxT() {
 			continue
 		}
 		// Set any "NotCounterReset" and "CounterReset" hints in native

ringbuffer/generic.go

@@ -12,6 +12,16 @@ import (
 	"github.com/prometheus/prometheus/model/histogram"
 )
 
+type Buffer interface {
+	MaxT() int64
+	Push(t int64, v Value)
+	Reset(mint int64, evalt int64)
+	Eval(ctx context.Context, _, _ float64, _ *int64) (float64, *histogram.FloatHistogram, bool, error)
+	SampleCount() int
+}
+
+func Empty(b Buffer) bool { return b.MaxT() != math.MinInt64 }
+
 type Value struct {
 	F float64
 	H *histogram.FloatHistogram
@@ -49,8 +59,6 @@ func NewWithExtLookback(ctx context.Context, size int, selectRange, offset, extL
 	}
 }
 
-func (r *GenericRingBuffer) Len() int { return len(r.items) }
-
 func (r *GenericRingBuffer) SampleCount() int {
 	c := 0
 	for _, s := range r.items {
@@ -72,11 +80,6 @@ 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/rate.go

@@ -14,16 +14,6 @@ import (
 	"github.com/prometheus/prometheus/model/histogram"
 )
 
-type Buffer interface {
-	Len() int
-	MaxT() int64
-	Push(t int64, v Value)
-	Reset(mint int64, evalt int64)
-	Eval(ctx context.Context, _, _ float64, _ *int64) (float64, *histogram.FloatHistogram, bool, error)
-	ReadIntoLast(f func(*Sample))
-	SampleCount() int
-}
-
 // RateBuffer is a Buffer which can calculate rate, increase and delta for a
 // series in a streaming manner, calculating the value incrementally for each
 // step where the sample is used.
@@ -95,8 +85,6 @@ func NewRateBuffer(ctx context.Context, opts query.Options, isCounter, isRate bo
 	}
 }
 
-func (r *RateBuffer) Len() int { return r.stepRanges[0].numSamples }
-
 func (r *RateBuffer) SampleCount() int {
 	return r.stepRanges[0].sampleCount
 }
@@ -198,8 +186,6 @@ 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,8 +183,7 @@ 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.isExtFunction); err != nil {
+			if err := scanner.selectPoints(mint, maxt, seriesTs, o.fhReader, o.extLookbackDelta, o.isExtFunction); err != nil {
 				return nil, err
 			}
 			// TODO(saswatamcode): Handle multi-arg functions for matrixSelectors.
@@ -307,6 +306,7 @@ 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)
@@ -324,7 +324,12 @@ func (m *matrixScanner) selectPoints(
 		mint = maxInt64(mint, m.buffer.MaxT()+1)
 	}
 
-	appendedPointBeforeMint := m.buffer.Len() > 0
+	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}
+	)
+
 	for valType := m.iterator.Next(); valType != chunkenc.ValNone; valType = m.iterator.Next() {
 		switch valType {
 		case chunkenc.ValHistogram, chunkenc.ValFloatHistogram:
@@ -361,19 +366,17 @@ func (m *matrixScanner) selectPoints(
 				m.lastSample.T, m.lastSample.V.F, m.lastSample.V.H = t, v, nil
 				return nil
 			}
-			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})
+			if t > mint {
+				if extSample.T != -1 && isExtFunction {
+					m.buffer.Push(extSample.T, ringbuffer.Value{F: extSample.V.F})
+					extSample.T = -1
 				}
+				m.buffer.Push(t, ringbuffer.Value{F: v})
+				continue
+			}
+			if isExtFunction && t > mint-extLookbackDelta {
+				extSample.T = t
+				extSample.V.F = v
 			}
 		}
 	}