From 256fd9a87e99ef672bccb28619978fa0de6c7c1d Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@gmail.com>
Date: Thu, 9 Jul 2020 13:05:42 +0300
Subject: [PATCH] app/vmselect/promql: refactor implementations for `and` and
 `unless` binary operations, so they are closer to `or` implementation

---
 app/vmselect/promql/binary_op.go | 37 ++++++++++++++------------------
 1 file changed, 16 insertions(+), 21 deletions(-)

diff --git a/app/vmselect/promql/binary_op.go b/app/vmselect/promql/binary_op.go
index d144000ea..d26282330 100644
--- a/app/vmselect/promql/binary_op.go
+++ b/app/vmselect/promql/binary_op.go
@@ -290,12 +290,14 @@ func binaryOpAnd(bfa *binaryOpFuncArg) ([]*timeseries, error) {
 		if tssLeft == nil {
 			continue
 		}
-		for i := range tssLeft[0].Values {
-			if !isAllNaNs(tssRight, i) {
-				continue
-			}
-			for _, tsLeft := range tssLeft {
-				tsLeft.Values[i] = nan
+		// Add gaps to tssLeft if there are gaps at valuesRight.
+		valuesRight := tssRight[0].Values
+		for _, tsLeft := range tssLeft {
+			valuesLeft := tsLeft.Values
+			for i, v := range valuesRight {
+				if math.IsNaN(v) {
+					valuesLeft[i] = nan
+				}
 			}
 		}
 		tssLeft = removeNaNs(tssLeft)
@@ -340,12 +342,14 @@ func binaryOpUnless(bfa *binaryOpFuncArg) ([]*timeseries, error) {
 			rvs = append(rvs, tssLeft...)
 			continue
 		}
-		for i := range tssLeft[0].Values {
-			if isAllNaNs(tssRight, i) {
-				continue
-			}
-			for _, tsLeft := range tssLeft {
-				tsLeft.Values[i] = nan
+		// Add gaps to tssLeft if the are no gaps at valuesRight.
+		valuesRight := tssRight[0].Values
+		for _, tsLeft := range tssLeft {
+			valuesLeft := tsLeft.Values
+			for i, v := range valuesRight {
+				if !math.IsNaN(v) {
+					valuesLeft[i] = nan
+				}
 			}
 		}
 		tssLeft = removeNaNs(tssLeft)
@@ -354,15 +358,6 @@ func binaryOpUnless(bfa *binaryOpFuncArg) ([]*timeseries, error) {
 	return rvs, nil
 }
 
-func isAllNaNs(tss []*timeseries, idx int) bool {
-	for _, ts := range tss {
-		if !math.IsNaN(ts.Values[idx]) {
-			return false
-		}
-	}
-	return true
-}
-
 func createTimeseriesMapByTagSet(be *metricsql.BinaryOpExpr, left, right []*timeseries) (map[string][]*timeseries, map[string][]*timeseries) {
 	groupTags := be.GroupModifier.Args
 	groupOp := strings.ToLower(be.GroupModifier.Op)