ringbuffers: cleanup subquery logic (#631)

The ringbuffer should not be aware if its used in a subquery or not,
thats an improper hierarchy. Its better if the subquery owns that logic.

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

Author: MichaHoffmann

execution/scan/subquery.go

@@ -16,6 +16,7 @@ import (
 	"github.com/thanos-io/promql-engine/query"
 	"github.com/thanos-io/promql-engine/ringbuffer"
 
+	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
 )
 
@@ -225,6 +226,24 @@ func (o *subqueryOperator) collect(v model.StepVector, mint int64) {
 		if buffer.Len() > 0 && v.T < buffer.MaxT() {
 			continue
 		}
+		// Set any "NotCounterReset" and "CounterReset" hints in native
+		// histograms to "UnknownCounterReset" because we might
+		// otherwise miss a counter reset happening in samples not
+		// returned by the subquery, or we might over-detect counter
+		// resets if the sample with a counter reset is returned
+		// multiple times by a high-res subquery. This intentionally
+		// does not attempt to be clever (like detecting if we are
+		// really missing underlying samples or returning underlying
+		// samples multiple times) because subqueries on counters are
+		// inherently problematic WRT counter reset handling, so we
+		// cannot really solve the problem for good. We only want to
+		// avoid problems that happen due to the explicitly set counter
+		// reset hints and go back to the behavior we already know from
+		// float samples.
+		switch s.CounterResetHint {
+		case histogram.NotCounterReset, histogram.CounterReset:
+			s.CounterResetHint = histogram.UnknownCounterReset
+		}
 		buffer.Push(v.T, ringbuffer.Value{H: s})
 	}
 	o.next.GetPool().PutStepVector(v)
@@ -249,7 +268,7 @@ func (o *subqueryOperator) initSeries(ctx context.Context) error {
 		o.series = make([]labels.Labels, len(series))
 		o.buffers = make([]*ringbuffer.GenericRingBuffer, len(series))
 		for i := range o.buffers {
-			o.buffers[i] = ringbuffer.New(ctx, 8, o.subQuery.Range.Milliseconds(), o.subQuery.Offset.Milliseconds(), o.call, true)
+			o.buffers[i] = ringbuffer.New(ctx, 8, o.subQuery.Range.Milliseconds(), o.subQuery.Offset.Milliseconds(), o.call)
 		}
 		var b labels.ScratchBuilder
 		for i, s := range series {

ringbuffer/generic.go

@@ -23,10 +23,9 @@ type Sample struct {
 }
 
 type GenericRingBuffer struct {
-	ctx      context.Context
-	items    []Sample
-	tail     []Sample
-	subquery bool
+	ctx   context.Context
+	items []Sample
+	tail  []Sample
 
 	currentStep int64
 	offset      int64
@@ -35,19 +34,18 @@ type GenericRingBuffer struct {
 	call        FunctionCall
 }
 
-func New(ctx context.Context, size int, selectRange, offset int64, call FunctionCall, subquery bool) *GenericRingBuffer {
-	return NewWithExtLookback(ctx, size, selectRange, offset, 0, call, subquery)
+func New(ctx context.Context, size int, selectRange, offset int64, call FunctionCall) *GenericRingBuffer {
+	return NewWithExtLookback(ctx, size, selectRange, offset, 0, call)
 }
 
-func NewWithExtLookback(ctx context.Context, size int, selectRange, offset, extLookback int64, call FunctionCall, subquery bool) *GenericRingBuffer {
+func NewWithExtLookback(ctx context.Context, size int, selectRange, offset, extLookback int64, call FunctionCall) *GenericRingBuffer {
 	return &GenericRingBuffer{
 		ctx:         ctx,
 		items:       make([]Sample, 0, size),
 		selectRange: selectRange,
 		offset:      offset,
 		extLookback: extLookback,
 		call:        call,
-		subquery:    subquery,
 	}
 }
 
@@ -93,28 +91,7 @@ func (r *GenericRingBuffer) Push(t int64, v Value) {
 	if v.H != nil {
 		if r.items[n].V.H == nil {
 			h := v.H.Copy()
-			if r.subquery {
-				// Set any "NotCounterReset" and "CounterReset" hints in native
-				// histograms to "UnknownCounterReset" because we might
-				// otherwise miss a counter reset happening in samples not
-				// returned by the subquery, or we might over-detect counter
-				// resets if the sample with a counter reset is returned
-				// multiple times by a high-res subquery. This intentionally
-				// does not attempt to be clever (like detecting if we are
-				// really missing underlying samples or returning underlying
-				// samples multiple times) because subqueries on counters are
-				// inherently problematic WRT counter reset handling, so we
-				// cannot really solve the problem for good. We only want to
-				// avoid problems that happen due to the explicitly set counter
-				// reset hints and go back to the behavior we already know from
-				// float samples.
-				switch h.CounterResetHint {
-				case histogram.NotCounterReset, histogram.CounterReset:
-					h.CounterResetHint = histogram.UnknownCounterReset
-				}
-			}
 			r.items[n].V.H = h
-
 		} else {
 			v.H.CopyTo(r.items[n].V.H)
 		}

storage/prometheus/matrix_selector.go

@@ -281,9 +281,9 @@ func (o *matrixSelector) newBuffer(ctx context.Context) ringbuffer.Buffer {
 	}
 
 	if o.isExtFunction {
-		return ringbuffer.NewWithExtLookback(ctx, 8, o.selectRange, o.offset, o.opts.ExtLookbackDelta.Milliseconds()-1, o.call, false)
+		return ringbuffer.NewWithExtLookback(ctx, 8, o.selectRange, o.offset, o.opts.ExtLookbackDelta.Milliseconds()-1, o.call)
 	}
-	return ringbuffer.New(ctx, 8, o.selectRange, o.offset, o.call, false)
+	return ringbuffer.New(ctx, 8, o.selectRange, o.offset, o.call)
 
 }