MQE: improve performance of subqueries when the parent query is a range query with many steps

CHANGELOG.md

@@ -15,7 +15,7 @@
   * `cortex_alertmanager_silences`
 * [CHANGE] Distributor: Drop experimental `-distributor.direct-otlp-translation-enabled` flag, since direct OTLP translation is well tested at this point. #9647
 * [FEATURE] Distributor: Add support for `lz4` OTLP compression. #9763
-* [FEATURE] Querier: add experimental streaming PromQL engine, enabled with `-querier.query-engine=mimir`. #9367 #9368 #9398 #9399 #9403 #9417 #9418 #9419 #9420 #9482 #9504 #9505 #9507 #9518 #9531 #9532 #9533 #9553 #9558 #9588 #9589 #9639 #9641 #9642 #9651 #9664 #9681 #9717
+* [FEATURE] Querier: add experimental streaming PromQL engine, enabled with `-querier.query-engine=mimir`. #9367 #9368 #9398 #9399 #9403 #9417 #9418 #9419 #9420 #9482 #9504 #9505 #9507 #9518 #9531 #9532 #9533 #9553 #9558 #9588 #9589 #9639 #9641 #9642 #9651 #9664 #9681 #9717 #9719
 * [FEATURE] Query-frontend: added experimental configuration options `query-frontend.cache-errors` and `query-frontend.results-cache-ttl-for-errors` to allow non-transient responses to be cached. When set to `true` error responses from hitting limits or bad data are cached for a short TTL. #9028
 * [FEATURE] gRPC: Support S2 compression. #9322
   * `-alertmanager.alertmanager-client.grpc-compression=s2`

pkg/streamingpromql/operators/functions/common.go

@@ -70,7 +70,7 @@ func PassthroughData(seriesData types.InstantVectorSeriesData, _ []types.ScalarD
 //   - h *histogram.FloatHistogram: nil if no histogram is present.
 //   - err error.
 type RangeVectorStepFunction func(
-	step types.RangeVectorStepData,
+	step *types.RangeVectorStepData,
 	rangeSeconds float64,
 	emitAnnotation types.EmitAnnotationFunc,
 ) (f float64, hasFloat bool, h *histogram.FloatHistogram, err error)

pkg/streamingpromql/operators/functions/range_vectors.go

@@ -21,9 +21,9 @@ var CountOverTime = FunctionOverRangeVectorDefinition{
 	StepFunc:               countOverTime,
 }
 
-func countOverTime(step types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-	fPointCount := step.Floats.CountAtOrBefore(step.RangeEnd)
-	hPointCount := step.Histograms.CountAtOrBefore(step.RangeEnd)
+func countOverTime(step *types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+	fPointCount := step.Floats.Count()
+	hPointCount := step.Histograms.Count()
 
 	if fPointCount == 0 && hPointCount == 0 {
 		return 0, false, nil, nil
@@ -37,9 +37,9 @@ var LastOverTime = FunctionOverRangeVectorDefinition{
 	StepFunc: lastOverTime,
 }
 
-func lastOverTime(step types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-	lastFloat, floatAvailable := step.Floats.LastAtOrBefore(step.RangeEnd)
-	lastHistogram, histogramAvailable := step.Histograms.LastAtOrBefore(step.RangeEnd)
+func lastOverTime(step *types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+	lastFloat, floatAvailable := step.Floats.Last()
+	lastHistogram, histogramAvailable := step.Histograms.Last()
 
 	if !floatAvailable && !histogramAvailable {
 		return 0, false, nil, nil
@@ -58,8 +58,8 @@ var PresentOverTime = FunctionOverRangeVectorDefinition{
 	StepFunc:               presentOverTime,
 }
 
-func presentOverTime(step types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-	if step.Floats.AnyAtOrBefore(step.RangeEnd) || step.Histograms.AnyAtOrBefore(step.RangeEnd) {
+func presentOverTime(step *types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+	if step.Floats.Any() || step.Histograms.Any() {
 		return 1, true, nil, nil
 	}
 
@@ -71,22 +71,15 @@ var MaxOverTime = FunctionOverRangeVectorDefinition{
 	StepFunc:               maxOverTime,
 }
 
-func maxOverTime(step types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-	head, tail := step.Floats.UnsafePoints(step.RangeEnd)
+func maxOverTime(step *types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+	head, tail := step.Floats.UnsafePoints()
 
 	if len(head) == 0 && len(tail) == 0 {
 		return 0, false, nil, nil
 	}
 
-	var maxSoFar float64
-
-	if len(head) > 0 {
-		maxSoFar = head[0].F
-		head = head[1:]
-	} else {
-		maxSoFar = tail[0].F
-		tail = tail[1:]
-	}
+	maxSoFar := head[0].F
+	head = head[1:]
 
 	for _, p := range head {
 		if p.F > maxSoFar || math.IsNaN(maxSoFar) {
@@ -108,22 +101,15 @@ var MinOverTime = FunctionOverRangeVectorDefinition{
 	StepFunc:               minOverTime,
 }
 
-func minOverTime(step types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-	head, tail := step.Floats.UnsafePoints(step.RangeEnd)
+func minOverTime(step *types.RangeVectorStepData, _ float64, _ types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+	head, tail := step.Floats.UnsafePoints()
 
 	if len(head) == 0 && len(tail) == 0 {
 		return 0, false, nil, nil
 	}
 
-	var minSoFar float64
-
-	if len(head) > 0 {
-		minSoFar = head[0].F
-		head = head[1:]
-	} else {
-		minSoFar = tail[0].F
-		tail = tail[1:]
-	}
+	minSoFar := head[0].F
+	head = head[1:]
 
 	for _, p := range head {
 		if p.F < minSoFar || math.IsNaN(minSoFar) {
@@ -146,9 +132,9 @@ var SumOverTime = FunctionOverRangeVectorDefinition{
 	NeedsSeriesNamesForAnnotations: true,
 }
 
-func sumOverTime(step types.RangeVectorStepData, _ float64, emitAnnotation types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-	fHead, fTail := step.Floats.UnsafePoints(step.RangeEnd)
-	hHead, hTail := step.Histograms.UnsafePoints(step.RangeEnd)
+func sumOverTime(step *types.RangeVectorStepData, _ float64, emitAnnotation types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+	fHead, fTail := step.Floats.UnsafePoints()
+	hHead, hTail := step.Histograms.UnsafePoints()
 
 	haveFloats := len(fHead) > 0 || len(fTail) > 0
 	haveHistograms := len(hHead) > 0 || len(hTail) > 0
@@ -185,18 +171,8 @@ func sumFloats(head, tail []promql.FPoint) float64 {
 }
 
 func sumHistograms(head, tail []promql.HPoint, emitAnnotation types.EmitAnnotationFunc) (*histogram.FloatHistogram, error) {
-	var sum *histogram.FloatHistogram
-
-	if len(head) > 0 {
-		sum = head[0].H
-		head = head[1:]
-	} else {
-		sum = tail[0].H
-		tail = tail[1:]
-	}
-
-	// We must make a copy of the histogram, as the ring buffer may reuse the FloatHistogram instance on subsequent steps.
-	sum = sum.Copy()
+	sum := head[0].H.Copy() // We must make a copy of the histogram, as the ring buffer may reuse the FloatHistogram instance on subsequent steps.
+	head = head[1:]
 
 	for _, p := range head {
 		if _, err := sum.Add(p.H); err != nil {
@@ -221,9 +197,9 @@ var AvgOverTime = FunctionOverRangeVectorDefinition{
 	NeedsSeriesNamesForAnnotations: true,
 }
 
-func avgOverTime(step types.RangeVectorStepData, _ float64, emitAnnotation types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-	fHead, fTail := step.Floats.UnsafePoints(step.RangeEnd)
-	hHead, hTail := step.Histograms.UnsafePoints(step.RangeEnd)
+func avgOverTime(step *types.RangeVectorStepData, _ float64, emitAnnotation types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+	fHead, fTail := step.Floats.UnsafePoints()
+	hHead, hTail := step.Histograms.UnsafePoints()
 
 	haveFloats := len(fHead) > 0 || len(fTail) > 0
 	haveHistograms := len(hHead) > 0 || len(hTail) > 0
@@ -306,20 +282,10 @@ func avgFloats(head, tail []promql.FPoint) float64 {
 }
 
 func avgHistograms(head, tail []promql.HPoint) (*histogram.FloatHistogram, error) {
-	var avgSoFar *histogram.FloatHistogram
+	avgSoFar := head[0].H.Copy() // We must make a copy of the histogram, as the ring buffer may reuse the FloatHistogram instance on subsequent steps.
+	head = head[1:]
 	count := 1.0
 
-	if len(head) > 0 {
-		avgSoFar = head[0].H
-		head = head[1:]
-	} else {
-		avgSoFar = tail[0].H
-		tail = tail[1:]
-	}
-
-	// We must make a copy of the histogram, as the ring buffer may reuse the FloatHistogram instance on subsequent steps.
-	avgSoFar = avgSoFar.Copy()
-
 	// Reuse these instances if we need them, to avoid allocating two FloatHistograms for every remaining histogram in the range.
 	var contributionByP *histogram.FloatHistogram
 	var contributionByAvgSoFar *histogram.FloatHistogram

pkg/streamingpromql/operators/functions/rate_increase.go

@@ -31,11 +31,11 @@ var Increase = FunctionOverRangeVectorDefinition{
 
 // isRate is true for `rate` function, or false for `instant` function
 func rate(isRate bool) RangeVectorStepFunction {
-	return func(step types.RangeVectorStepData, rangeSeconds float64, emitAnnotation types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
-		fHead, fTail := step.Floats.UnsafePoints(step.RangeEnd)
+	return func(step *types.RangeVectorStepData, rangeSeconds float64, emitAnnotation types.EmitAnnotationFunc) (float64, bool, *histogram.FloatHistogram, error) {
+		fHead, fTail := step.Floats.UnsafePoints()
 		fCount := len(fHead) + len(fTail)
 
-		hHead, hTail := step.Histograms.UnsafePoints(step.RangeEnd)
+		hHead, hTail := step.Histograms.UnsafePoints()
 		hCount := len(hHead) + len(hTail)
 
 		if fCount > 0 && hCount > 0 {
@@ -46,12 +46,13 @@ func rate(isRate bool) RangeVectorStepFunction {
 		}
 
 		if fCount >= 2 {
-			val := floatRate(isRate, fCount, step.Floats, step, fHead, fTail, rangeSeconds)
+			// TODO: just pass step here? (and below)
+			val := floatRate(isRate, fCount, fHead, fTail, step.RangeStart, step.RangeEnd, rangeSeconds)
 			return val, true, nil, nil
 		}
 
 		if hCount >= 2 {
-			val, err := histogramRate(isRate, step, hHead, hTail, rangeSeconds, hCount, emitAnnotation)
+			val, err := histogramRate(isRate, hCount, hHead, hTail, step.RangeStart, step.RangeEnd, rangeSeconds, emitAnnotation)
 			if err != nil {
 				err = NativeHistogramErrorToAnnotation(err, emitAnnotation)
 				return 0, false, nil, err
@@ -63,15 +64,9 @@ func rate(isRate bool) RangeVectorStepFunction {
 	}
 }
 
-func histogramRate(isRate bool, step types.RangeVectorStepData, hHead []promql.HPoint, hTail []promql.HPoint, rangeSeconds float64, hCount int, emitAnnotation types.EmitAnnotationFunc) (*histogram.FloatHistogram, error) {
-	var firstPoint promql.HPoint
-	if len(hHead) > 0 {
-		firstPoint = hHead[0]
-		hHead = hHead[1:]
-	} else {
-		firstPoint = hTail[0]
-		hTail = hTail[1:]
-	}
+func histogramRate(isRate bool, hCount int, hHead []promql.HPoint, hTail []promql.HPoint, rangeStart int64, rangeEnd int64, rangeSeconds float64, emitAnnotation types.EmitAnnotationFunc) (*histogram.FloatHistogram, error) {
+	firstPoint := hHead[0]
+	hHead = hHead[1:]
 
 	var lastPoint promql.HPoint
 	if len(hTail) > 0 {
@@ -142,12 +137,13 @@ func histogramRate(isRate bool, step types.RangeVectorStepData, hHead []promql.H
 		delta = delta.CopyToSchema(desiredSchema)
 	}
 
-	val := calculateHistogramRate(isRate, step.RangeStart, step.RangeEnd, rangeSeconds, firstPoint, lastPoint, delta, hCount)
+	val := calculateHistogramRate(isRate, rangeStart, rangeEnd, rangeSeconds, firstPoint, lastPoint, delta, hCount)
 	return val, err
 }
 
-func floatRate(isRate bool, fCount int, floatBuffer *types.FPointRingBuffer, step types.RangeVectorStepData, fHead []promql.FPoint, fTail []promql.FPoint, rangeSeconds float64) float64 {
-	firstPoint := floatBuffer.First()
+func floatRate(isRate bool, fCount int, fHead []promql.FPoint, fTail []promql.FPoint, rangeStart int64, rangeEnd int64, rangeSeconds float64) float64 {
+	firstPoint := fHead[0]
+	fHead = fHead[1:]
 
 	var lastPoint promql.FPoint
 	if len(fTail) > 0 {
@@ -173,7 +169,7 @@ func floatRate(isRate bool, fCount int, floatBuffer *types.FPointRingBuffer, ste
 	accumulate(fHead)
 	accumulate(fTail)
 
-	val := calculateFloatRate(isRate, step.RangeStart, step.RangeEnd, rangeSeconds, firstPoint, lastPoint, delta, fCount)
+	val := calculateFloatRate(isRate, rangeStart, rangeEnd, rangeSeconds, firstPoint, lastPoint, delta, fCount)
 	return val
 }
 

pkg/streamingpromql/operators/selectors/range_vector_selector.go

@@ -24,9 +24,10 @@ type RangeVectorSelector struct {
 
 	rangeMilliseconds int64
 	chunkIterator     chunkenc.Iterator
-	nextT             int64
+	nextStepT         int64
 	floats            *types.FPointRingBuffer
 	histograms        *types.HPointRingBuffer
+	stepData          *types.RangeVectorStepData // Retain the last step data instance we used to avoid allocating it for every step.
 }
 
 var _ types.RangeVectorOperator = &RangeVectorSelector{}
@@ -36,6 +37,7 @@ func NewRangeVectorSelector(selector *Selector, memoryConsumptionTracker *limiti
 		Selector:   selector,
 		floats:     types.NewFPointRingBuffer(memoryConsumptionTracker),
 		histograms: types.NewHPointRingBuffer(memoryConsumptionTracker),
+		stepData:   &types.RangeVectorStepData{},
 	}
 }
 
@@ -65,19 +67,20 @@ func (m *RangeVectorSelector) NextSeries(ctx context.Context) error {
 		return err
 	}
 
-	m.nextT = m.Selector.TimeRange.StartT
+	m.nextStepT = m.Selector.TimeRange.StartT
 	m.floats.Reset()
 	m.histograms.Reset()
 	return nil
 }
 
-func (m *RangeVectorSelector) NextStepSamples() (types.RangeVectorStepData, error) {
-	if m.nextT > m.Selector.TimeRange.EndT {
-		return types.RangeVectorStepData{}, types.EOS
+func (m *RangeVectorSelector) NextStepSamples() (*types.RangeVectorStepData, error) {
+	if m.nextStepT > m.Selector.TimeRange.EndT {
+		return nil, types.EOS
 	}
 
-	stepT := m.nextT
-	rangeEnd := stepT
+	m.stepData.StepT = m.nextStepT
+	rangeEnd := m.nextStepT
+	m.nextStepT += m.Selector.TimeRange.IntervalMilliseconds
 
 	if m.Selector.Timestamp != nil {
 		// Timestamp from @ modifier takes precedence over query evaluation timestamp.
@@ -91,18 +94,15 @@ func (m *RangeVectorSelector) NextStepSamples() (types.RangeVectorStepData, erro
 	m.histograms.DiscardPointsBefore(rangeStart)
 
 	if err := m.fillBuffer(m.floats, m.histograms, rangeStart, rangeEnd); err != nil {
-		return types.RangeVectorStepData{}, err
+		return nil, err
 	}
 
-	m.nextT += m.Selector.TimeRange.IntervalMilliseconds
+	m.stepData.Floats = m.floats.ViewUntilSearchingBackwards(rangeEnd, m.stepData.Floats)
+	m.stepData.Histograms = m.histograms.ViewUntilSearchingBackwards(rangeEnd, m.stepData.Histograms)
+	m.stepData.RangeStart = rangeStart
+	m.stepData.RangeEnd = rangeEnd
 
-	return types.RangeVectorStepData{
-		Floats:     m.floats,
-		Histograms: m.histograms,
-		StepT:      stepT,
-		RangeStart: rangeStart,
-		RangeEnd:   rangeEnd,
-	}, nil
+	return m.stepData, nil
 }
 
 func (m *RangeVectorSelector) fillBuffer(floats *types.FPointRingBuffer, histograms *types.HPointRingBuffer, rangeStart, rangeEnd int64) error {

pkg/streamingpromql/operators/subquery.go

@@ -26,6 +26,7 @@ type Subquery struct {
 	rangeMilliseconds int64
 	floats            *types.FPointRingBuffer
 	histograms        *types.HPointRingBuffer
+	stepData          *types.RangeVectorStepData // Retain the last step data instance we used to avoid allocating it for every step.
 }
 
 var _ types.RangeVectorOperator = &Subquery{}
@@ -49,6 +50,7 @@ func NewSubquery(
 		rangeMilliseconds:    subqueryRange.Milliseconds(),
 		floats:               types.NewFPointRingBuffer(memoryConsumptionTracker),
 		histograms:           types.NewHPointRingBuffer(memoryConsumptionTracker),
+		stepData:             &types.RangeVectorStepData{},
 	}
 }
 
@@ -72,13 +74,14 @@ func (s *Subquery) NextSeries(ctx context.Context) error {
 	return nil
 }
 
-func (s *Subquery) NextStepSamples() (types.RangeVectorStepData, error) {
+func (s *Subquery) NextStepSamples() (*types.RangeVectorStepData, error) {
 	if s.nextStepT > s.ParentQueryTimeRange.EndT {
-		return types.RangeVectorStepData{}, types.EOS
+		return nil, types.EOS
 	}
 
-	stepT := s.nextStepT
-	rangeEnd := stepT
+	s.stepData.StepT = s.nextStepT
+	rangeEnd := s.nextStepT
+	s.nextStepT += s.ParentQueryTimeRange.IntervalMilliseconds
 
 	if s.SubqueryTimestamp != nil {
 		// Timestamp from @ modifier takes precedence over query evaluation timestamp.
@@ -91,15 +94,12 @@ func (s *Subquery) NextStepSamples() (types.RangeVectorStepData, error) {
 	s.floats.DiscardPointsBefore(rangeStart)
 	s.histograms.DiscardPointsBefore(rangeStart)
 
-	s.nextStepT += s.ParentQueryTimeRange.IntervalMilliseconds
+	s.stepData.Floats = s.floats.ViewUntilSearchingForwards(rangeEnd, s.stepData.Floats)
+	s.stepData.Histograms = s.histograms.ViewUntilSearchingForwards(rangeEnd, s.stepData.Histograms)
+	s.stepData.RangeStart = rangeStart
+	s.stepData.RangeEnd = rangeEnd
 
-	return types.RangeVectorStepData{
-		Floats:     s.floats,
-		Histograms: s.histograms,
-		StepT:      stepT,
-		RangeStart: rangeStart,
-		RangeEnd:   rangeEnd,
-	}, nil
+	return s.stepData, nil
 }
 
 func (s *Subquery) StepCount() int {

pkg/streamingpromql/query.go

@@ -730,12 +730,12 @@ func (q *Query) populateMatrixFromRangeVectorOperator(ctx context.Context, o typ
 			return nil, err
 		}
 
-		floats, err := step.Floats.CopyPoints(step.RangeEnd)
+		floats, err := step.Floats.CopyPoints()
 		if err != nil {
 			return nil, err
 		}
 
-		histograms, err := step.Histograms.CopyPoints(step.RangeEnd)
+		histograms, err := step.Histograms.CopyPoints()
 		if err != nil {
 			return nil, err
 		}