Fix rate combine when a middle split has no samples

Author: fionaliao

pkg/streamingpromql/operators/functions/split_operator_test.go

@@ -0,0 +1,124 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package functions
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/mimir/pkg/mimirpb"
+	"github.com/grafana/mimir/pkg/streamingpromql/types"
+)
+
+func TestResultGetter_OutOfOrderAccess(t *testing.T) {
+	const totalSeries = 3
+	callCount := 0
+
+	nextSeriesFunc := func(ctx context.Context) ([]string, error) {
+		callCount++
+		if callCount > totalSeries {
+			// Simulate the "no more series to pop" panic
+			return nil, types.EOS
+		}
+		return []string{string(rune('A' + callCount - 1))}, nil
+	}
+
+	getter := NewResultGetter(nextSeriesFunc)
+
+	result, err := getter.GetResultsAtIdx(context.Background(), 2)
+	require.NoError(t, err)
+	require.Equal(t, []string{"C"}, result)
+
+	require.Equal(t, 3, callCount)
+
+	result, err = getter.GetResultsAtIdx(context.Background(), 0)
+	require.NoError(t, err)
+	require.Equal(t, []string{"A"}, result)
+
+	result, err = getter.GetResultsAtIdx(context.Background(), 1)
+	require.NoError(t, err)
+	require.Equal(t, []string{"B"}, result)
+}
+
+func TestResultGetter_SequentialAccess(t *testing.T) {
+	callCount := 0
+
+	nextSeriesFunc := func(ctx context.Context) ([]string, error) {
+		callCount++
+		if callCount > 3 {
+			return nil, types.EOS
+		}
+		return []string{string(rune('A' + callCount - 1))}, nil
+	}
+
+	getter := NewResultGetter(nextSeriesFunc)
+
+	result0, err := getter.GetResultsAtIdx(context.Background(), 0)
+	require.NoError(t, err)
+	require.Equal(t, []string{"A"}, result0)
+	require.Equal(t, 1, callCount)
+
+	result1, err := getter.GetResultsAtIdx(context.Background(), 1)
+	require.NoError(t, err)
+	require.Equal(t, []string{"B"}, result1)
+	require.Equal(t, 2, callCount)
+
+	result2, err := getter.GetResultsAtIdx(context.Background(), 2)
+	require.NoError(t, err)
+	require.Equal(t, []string{"C"}, result2)
+	require.Equal(t, 3, callCount)
+}
+
+func TestRateCombineFloat_WithEmptySplits(t *testing.T) {
+	splits := []RateIntermediate{
+		// Split 0: has samples
+		{
+			FirstSample: &mimirpb.Sample{
+				TimestampMs: 1000,
+				Value:       100.0,
+			},
+			LastSample: &mimirpb.Sample{
+				TimestampMs: 2000,
+				Value:       110.0,
+			},
+			Delta:           10.0,
+			SampleCount:     5,
+			IsHistogram:     false,
+			SplitRangeStart: 0,
+			SplitRangeEnd:   3000,
+		},
+		// Split 1: empty (no samples)
+		{
+			FirstSample:     nil,
+			LastSample:      nil,
+			Delta:           0,
+			SampleCount:     0,
+			IsHistogram:     false,
+			SplitRangeStart: 3000,
+			SplitRangeEnd:   6000,
+		},
+		// Split 2: has samples
+		{
+			FirstSample: &mimirpb.Sample{
+				TimestampMs: 6000,
+				Value:       120.0,
+			},
+			LastSample: &mimirpb.Sample{
+				TimestampMs: 7000,
+				Value:       130.0,
+			},
+			Delta:           10.0,
+			SampleCount:     5,
+			IsHistogram:     false,
+			SplitRangeStart: 6000,
+			SplitRangeEnd:   9000,
+		},
+	}
+
+	result, hasResult, err := rateCombineFloat(splits, true)
+	require.NoError(t, err)
+	require.True(t, hasResult)
+	require.NotZero(t, result)
+}

pkg/streamingpromql/operators/functions/split_rate_functions.go

@@ -185,16 +185,17 @@ func rateCombineFloat(splits []RateIntermediate, isRate bool) (float64, bool, er
 	totalDelta := firstPiece.Delta
 	totalCount := int(firstPiece.SampleCount)
 
+	lastSplitIdx := startIdx
 	for i := startIdx + 1; i < len(splits); i++ {
 		split := splits[i]
 		if split.SampleCount == 0 {
 			continue
 		}
 
-		if split.FirstSample.Value < splits[i-1].LastSample.Value {
+		if split.FirstSample.Value < splits[lastSplitIdx].LastSample.Value {
 			totalDelta += split.FirstSample.Value
 		} else {
-			interSplitDelta := split.FirstSample.Value - splits[i-1].LastSample.Value
+			interSplitDelta := split.FirstSample.Value - splits[lastSplitIdx].LastSample.Value
 			totalDelta += interSplitDelta
 		}
 
@@ -205,6 +206,7 @@ func rateCombineFloat(splits []RateIntermediate, isRate bool) (float64, bool, er
 			T: split.LastSample.TimestampMs,
 			F: split.LastSample.Value,
 		}
+		lastSplitIdx = i
 	}
 
 	// Need at least 2 samples total to calculate rate