SSE: Reduce to apply Mode to instant vector (mathexp.Number) (#74859)

pkg/expr/commands.go

@@ -175,12 +175,21 @@ func (gr *ReduceCommand) Execute(ctx context.Context, _ time.Time, vars mathexp.
 			}
 			newRes.Values = append(newRes.Values, num)
 		case mathexp.Number: // if incoming vars is just a number, any reduce op is just a noop, add it as it is
+			value := v.GetFloat64Value()
+			if gr.seriesMapper != nil {
+				value = gr.seriesMapper.MapInput(value)
+				if value == nil { // same logic as in mapSeries
+					continue
+				}
+			}
 			copyV := mathexp.NewNumber(gr.refID, v.GetLabels())
-			copyV.SetValue(v.GetFloat64Value())
+			copyV.SetValue(value)
-			copyV.AddNotice(data.Notice{
+			if gr.seriesMapper == nil {
-				Severity: data.NoticeSeverityWarning,
+				copyV.AddNotice(data.Notice{
-				Text:     fmt.Sprintf("Reduce operation is not needed. Input query or expression %s is already reduced data.", gr.VarToReduce),
+					Severity: data.NoticeSeverityWarning,
-			})
+					Text:     fmt.Sprintf("Reduce operation is not needed. Input query or expression %s is already reduced data.", gr.VarToReduce),
+				})
+			}
 			newRes.Values = append(newRes.Values, copyV)
 		case mathexp.NoData:
 			newRes.Values = append(newRes.Values, v.New())

pkg/expr/commands_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
 	"math/rand"
 	"testing"
 	"time"
@@ -100,42 +101,83 @@ func Test_UnmarshalReduceCommand_Settings(t *testing.T) {
 
 func TestReduceExecute(t *testing.T) {
 	varToReduce := util.GenerateShortUID()
-	cmd, err := NewReduceCommand(util.GenerateShortUID(), randomReduceFunc(), varToReduce, nil)
-	require.NoError(t, err)
 
-	t.Run("should noop if Number", func(t *testing.T) {
+	t.Run("when mapper is nil", func(t *testing.T) {
+		cmd, err := NewReduceCommand(util.GenerateShortUID(), randomReduceFunc(), varToReduce, nil)
+		require.NoError(t, err)
+
+		t.Run("should noop if Number", func(t *testing.T) {
+			var numbers mathexp.Values = []mathexp.Value{
+				mathexp.GenerateNumber(util.Pointer(rand.Float64())),
+				mathexp.GenerateNumber(util.Pointer(rand.Float64())),
+				mathexp.GenerateNumber(util.Pointer(rand.Float64())),
+			}
+
+			vars := map[string]mathexp.Results{
+				varToReduce: {
+					Values: numbers,
+				},
+			}
+
+			execute, err := cmd.Execute(context.Background(), time.Now(), vars, tracing.NewFakeTracer())
+			require.NoError(t, err)
+
+			require.Len(t, execute.Values, len(numbers))
+			for i, value := range execute.Values {
+				expected := numbers[i]
+				require.Equal(t, expected.Type(), value.Type())
+				require.Equal(t, expected.GetLabels(), value.GetLabels())
+
+				expectedValue := expected.Value().(*mathexp.Number).GetFloat64Value()
+				actualValue := value.Value().(*mathexp.Number).GetFloat64Value()
+				require.Equal(t, expectedValue, actualValue)
+			}
+
+			t.Run("should add warn notices to every frame", func(t *testing.T) {
+				frames := execute.Values.AsDataFrames("test")
+				for _, frame := range frames {
+					require.Len(t, frame.Meta.Notices, 1)
+					notice := frame.Meta.Notices[0]
+					require.Equal(t, data.NoticeSeverityWarning, notice.Severity)
+				}
+			})
+		})
+	})
+
+	t.Run("when mapper is not nil", func(t *testing.T) {
 		var numbers mathexp.Values = []mathexp.Value{
 			mathexp.GenerateNumber(util.Pointer(rand.Float64())),
-			mathexp.GenerateNumber(util.Pointer(rand.Float64())),
+			mathexp.GenerateNumber(nil),
+			mathexp.GenerateNumber(util.Pointer(math.NaN())),
+			mathexp.GenerateNumber(util.Pointer(math.Inf(-1))),
+			mathexp.GenerateNumber(util.Pointer(math.Inf(1))),
 			mathexp.GenerateNumber(util.Pointer(rand.Float64())),
 		}
-
+		varToReduce := util.GenerateShortUID()
 		vars := map[string]mathexp.Results{
 			varToReduce: {
 				Values: numbers,
 			},
 		}
 
-		execute, err := cmd.Execute(context.Background(), time.Now(), vars, tracing.NewFakeTracer())
+		t.Run("drop all non numbers if mapper is DropNonNumber", func(t *testing.T) {
-		require.NoError(t, err)
+			cmd, err := NewReduceCommand(util.GenerateShortUID(), randomReduceFunc(), varToReduce, &mathexp.DropNonNumber{})
+			require.NoError(t, err)
+			execute, err := cmd.Execute(context.Background(), time.Now(), vars, tracing.NewFakeTracer())
+			require.NoError(t, err)
+			require.Len(t, execute.Values, 2)
+		})
 
-		require.Len(t, execute.Values, len(numbers))
+		t.Run("replace all non numbers if mapper is ReplaceNonNumberWithValue", func(t *testing.T) {
-		for i, value := range execute.Values {
+			cmd, err := NewReduceCommand(util.GenerateShortUID(), randomReduceFunc(), varToReduce, &mathexp.ReplaceNonNumberWithValue{Value: 1})
-			expected := numbers[i]
+			require.NoError(t, err)
-			require.Equal(t, expected.Type(), value.Type())
+			execute, err := cmd.Execute(context.Background(), time.Now(), vars, tracing.NewFakeTracer())
-			require.Equal(t, expected.GetLabels(), value.GetLabels())
+			require.NoError(t, err)
-
+			require.Len(t, execute.Values, len(numbers))
-			expectedValue := expected.Value().(*mathexp.Number).GetFloat64Value()
+			for _, value := range execute.Values[1 : len(numbers)-1] {
-			actualValue := value.Value().(*mathexp.Number).GetFloat64Value()
+				require.IsType(t, &mathexp.Number{}, value.Value())
-			require.Equal(t, expectedValue, actualValue)
+				f := value.Value().(*mathexp.Number)
-		}
+				require.Equal(t, float64(1), *f.GetFloat64Value())
-
-		t.Run("should add warn notices to every frame", func(t *testing.T) {
-			frames := execute.Values.AsDataFrames("test")
-			for _, frame := range frames {
-				require.Len(t, frame.Meta.Notices, 1)
-				notice := frame.Meta.Notices[0]
-				require.Equal(t, data.NoticeSeverityWarning, notice.Severity)
 			}
 		})
 	})
@@ -150,7 +192,8 @@ func TestReduceExecute(t *testing.T) {
 				Values: noData,
 			},
 		}
-
+		cmd, err := NewReduceCommand(util.GenerateShortUID(), randomReduceFunc(), varToReduce, nil)
+		require.NoError(t, err)
 		results, err := cmd.Execute(context.Background(), time.Now(), vars, tracing.NewFakeTracer())
 		require.NoError(t, err)