SSE: Warn on dropped items in Union in Math Operation (#72682)

pkg/expr/mathexp/exp.go

@@ -5,6 +5,7 @@ import (
 	"math"
 	"reflect"
 	"runtime"
+	"strings"
 
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 
@@ -25,7 +26,9 @@ type State struct {
 	// Could hold more properties that change behavior around:
 	//  - Unions (How many result A and many Result B in case A + B are joined)
 	//  - NaN/Null behavior
-	RefID string
+	RefID     string
+	Drops     map[string]map[string][]data.Labels // binary node text -> LH/RH -> Drop Labels
+	DropCount int64
 
 	tracer tracing.Tracer
 }
@@ -61,6 +64,7 @@ func (e *Expr) Execute(refID string, vars Vars, tracer tracing.Tracer) (r Result
 func (e *Expr) executeState(s *State) (r Results, err error) {
 	defer errRecover(&err, s)
 	r, err = s.walk(e.Tree.Root)
+	s.addDropNotices(&r)
 	return
 }
 
@@ -198,25 +202,64 @@ type Union struct {
 // within a collection of Series or Numbers. The Unions are used with binary
 // operations. The labels of the Union will the taken from result with a greater
 // number of tags.
-func union(aResults, bResults Results) []*Union {
+func (e *State) union(aResults, bResults Results, biNode *parse.BinaryNode) []*Union {
 	unions := []*Union{}
+	appendUnions := func(u *Union) {
+		unions = append(unions, u)
+	}
+
+	aVar := biNode.Args[0].String()
+	bVar := biNode.Args[1].String()
+
+	aMatched := make([]bool, len(aResults.Values))
+	bMatched := make([]bool, len(bResults.Values))
+	collectDrops := func() {
+		check := func(v string, matchArray []bool, r *Results) {
+			for i, b := range matchArray {
+				if b {
+					continue
+				}
+				if e.Drops == nil {
+					e.Drops = make(map[string]map[string][]data.Labels)
+				}
+				if e.Drops[biNode.String()] == nil {
+					e.Drops[biNode.String()] = make(map[string][]data.Labels)
+				}
+
+				if r.Values[i].Type() == parse.TypeNoData {
+					continue
+				}
+
+				e.DropCount++
+				e.Drops[biNode.String()][v] = append(e.Drops[biNode.String()][v], r.Values[i].GetLabels())
+			}
+		}
+		check(aVar, aMatched, &aResults)
+		check(bVar, bMatched, &bResults)
+	}
+
 	aValueLen := len(aResults.Values)
 	bValueLen := len(bResults.Values)
 	if aValueLen == 0 || bValueLen == 0 {
 		return unions
 	}
+
 	if aValueLen == 1 || bValueLen == 1 {
-		if aResults.Values[0].Type() == parse.TypeNoData || bResults.Values[0].Type() == parse.TypeNoData {
+		aNoData := aResults.Values[0].Type() == parse.TypeNoData
-			unions = append(unions, &Union{
+		bNoData := bResults.Values[0].Type() == parse.TypeNoData
+		if aNoData || bNoData {
+			appendUnions(&Union{
 				Labels: nil,
 				A:      aResults.Values[0],
 				B:      bResults.Values[0],
 			})
+			collectDrops()
 			return unions
 		}
 	}
-	for _, a := range aResults.Values {
+
-		for _, b := range bResults.Values {
+	for iA, a := range aResults.Values {
+		for iB, b := range bResults.Values {
 			var labels data.Labels
 			aLabels := a.GetLabels()
 			bLabels := b.GetLabels()
@@ -241,20 +284,25 @@ func union(aResults, bResults Results) []*Union {
 				A:      a,
 				B:      b,
 			}
-			unions = append(unions, u)
+			appendUnions(u)
+			aMatched[iA] = true
+			bMatched[iB] = true
 		}
 	}
+
 	if len(unions) == 0 && len(aResults.Values) == 1 && len(bResults.Values) == 1 {
 		// In the case of only 1 thing on each side of the operator, we combine them
 		// and strip the tags.
 		// This isn't ideal for understanding behavior, but will make more stuff work when
 		// combining different datasources without munging.
 		// This choice is highly questionable in the long term.
-		unions = append(unions, &Union{
+		appendUnions(&Union{
 			A: aResults.Values[0],
 			B: bResults.Values[0],
 		})
 	}
+
+	collectDrops()
 	return unions
 }
 
@@ -268,7 +316,7 @@ func (e *State) walkBinary(node *parse.BinaryNode) (Results, error) {
 	if err != nil {
 		return res, err
 	}
-	unions := union(ar, br)
+	unions := e.union(ar, br, node)
 	for _, uni := range unions {
 		var value Value
 		switch at := uni.A.(type) {
@@ -548,3 +596,59 @@ func (e *State) walkFunc(node *parse.FuncNode) (Results, error) {
 	}
 	return res, nil
 }
+
+func (e *State) addDropNotices(r *Results) {
+	nT := strings.Builder{}
+
+	if e.DropCount > 0 && len(r.Values) > 0 {
+		itemsPerNodeLimit := 5 // Limit on dropped items shown per each node in the binary node
+
+		nT.WriteString(fmt.Sprintf("%v items dropped from union(s)", e.DropCount))
+		if len(e.Drops) > 0 {
+			nT.WriteString(": ")
+
+			biNodeDropCount := 0
+			for biNodeText, biNodeDrops := range e.Drops {
+				nT.WriteString(fmt.Sprintf(`["%s": `, biNodeText))
+
+				nodeCount := 0
+				for inputNode, droppedItems := range biNodeDrops {
+					nT.WriteString(fmt.Sprintf("(%s: ", inputNode))
+
+					itemCount := 0
+					for _, item := range droppedItems {
+						nT.WriteString(fmt.Sprintf("{%s}", item))
+
+						itemCount++
+						if itemCount == itemsPerNodeLimit {
+							nT.WriteString(fmt.Sprintf("...%v more...", len(droppedItems)-itemsPerNodeLimit))
+							break
+						}
+						if itemCount < len(droppedItems) {
+							nT.WriteString(" ")
+						}
+					}
+
+					nT.WriteString(")")
+
+					nodeCount++
+					if nodeCount < len(biNodeDrops) {
+						nT.WriteString(" ")
+					}
+				}
+
+				nT.WriteString("]")
+
+				biNodeDropCount++
+				if biNodeDropCount < len(biNodeDrops) {
+					nT.WriteString(" ")
+				}
+			}
+		}
+
+		r.Values[0].AddNotice(data.Notice{
+			Severity: data.NoticeSeverityWarning,
+			Text:     nT.String(),
+		})
+	}
+}

pkg/expr/mathexp/exp_nan_null_val_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/grafana/grafana/pkg/expr/mathexp/parse"
 	"github.com/grafana/grafana/pkg/infra/tracing"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -491,21 +492,21 @@ func TestNoData(t *testing.T) {
 					res, err := e.Execute("", makeVars(NewNoData(), NewNoData()), tracing.NewFakeTracer())
 					require.NoError(t, err)
 					require.Len(t, res.Values, 1)
-					require.Equal(t, NewNoData(), res.Values[0])
+					require.Equal(t, parse.TypeNoData, res.Values[0].Type())
 				})
 
 				t.Run("$A=nodata, $B=series", func(t *testing.T) {
 					res, err := e.Execute("", makeVars(NewNoData(), series), tracing.NewFakeTracer())
 					require.NoError(t, err)
 					require.Len(t, res.Values, 1)
-					require.Equal(t, NewNoData(), res.Values[0])
+					require.Equal(t, parse.TypeNoData, res.Values[0].Type())
 				})
 
 				t.Run("$A=series, $B=nodata", func(t *testing.T) {
 					res, err := e.Execute("", makeVars(NewNoData(), series), tracing.NewFakeTracer())
 					require.NoError(t, err)
 					require.Len(t, res.Values, 1)
-					require.Equal(t, NewNoData(), res.Values[0])
+					require.Equal(t, parse.TypeNoData, res.Values[0].Type())
 				})
 			}
 		})

pkg/expr/mathexp/parse/node.go

@@ -77,6 +77,8 @@ func (t NodeType) String() string {
 		return "NodeString"
 	case NodeNumber:
 		return "NodeNumber"
+	case NodeVar:
+		return "NodeVar"
 	default:
 		return "NodeUnknown"
 	}

pkg/expr/mathexp/union_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/grafana/grafana/pkg/expr/mathexp/parse"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -293,7 +294,8 @@ func Test_union(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			unions := union(tt.aResults, tt.bResults)
+			fakeNode := &parse.BinaryNode{Args: [2]parse.Node{&parse.VarNode{}, &parse.VarNode{}}}
+			unions := (&State{}).union(tt.aResults, tt.bResults, fakeNode)
 			tt.unionsAre(t, tt.unions, unions)
 		})
 	}