From a5302a665126d82dea49095cc97a4d7085e45b0a Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@gmail.com>
Date: Tue, 8 Oct 2019 12:28:06 +0300
Subject: [PATCH] app/vmselect/promql: take into account the previous point
 when calculating `max_over_time` and `min_over_time`

This lines up with `first_over_time` function used in `rollup_candlestick`, so `rollup=low` always returns
the minimum value.

Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/204
---
 app/vmselect/promql/exec_test.go   |  4 ++--
 app/vmselect/promql/rollup.go      | 23 ++++++++++++++++-------
 app/vmselect/promql/rollup_test.go |  2 +-
 3 files changed, 19 insertions(+), 10 deletions(-)

diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go
index a8268cc47..71e6b001a 100644
--- a/app/vmselect/promql/exec_test.go
+++ b/app/vmselect/promql/exec_test.go
@@ -3566,7 +3566,7 @@ func TestExecSuccess(t *testing.T) {
 		}}
 		r4 := netstorage.Result{
 			MetricName: metricNameExpected,
-			Values:     []float64{0.85, 0.94, 0.97, 0.93, 0.98, 0.92},
+			Values:     []float64{0.9, 0.94, 0.97, 0.93, 0.98, 0.92},
 			Timestamps: timestampsExpected,
 		}
 		r4.MetricName.Tags = []storage.Tag{{
@@ -3614,7 +3614,7 @@ func TestExecSuccess(t *testing.T) {
 		q := `sort(rollup(time()[:50s]))`
 		r1 := netstorage.Result{
 			MetricName: metricNameExpected,
-			Values:     []float64{850, 1050, 1250, 1450, 1650, 1850},
+			Values:     []float64{800, 1000, 1200, 1400, 1600, 1800},
 			Timestamps: timestampsExpected,
 		}
 		r1.MetricName.Tags = []storage.Tag{{
diff --git a/app/vmselect/promql/rollup.go b/app/vmselect/promql/rollup.go
index d5467c8bc..5e0571668 100644
--- a/app/vmselect/promql/rollup.go
+++ b/app/vmselect/promql/rollup.go
@@ -533,11 +533,14 @@ func rollupAvg(rfa *rollupFuncArg) float64 {
 func rollupMin(rfa *rollupFuncArg) float64 {
 	// There is no need in handling NaNs here, since they must be cleaned up
 	// before calling rollup funcs.
+	minValue := rfa.prevValue
 	values := rfa.values
-	if len(values) == 0 {
-		return rfa.prevValue
+	if math.IsNaN(minValue) {
+		if len(values) == 0 {
+			return nan
+		}
+		minValue = values[0]
 	}
-	minValue := values[0]
 	for _, v := range values {
 		if v < minValue {
 			minValue = v
@@ -549,11 +552,14 @@ func rollupMin(rfa *rollupFuncArg) float64 {
 func rollupMax(rfa *rollupFuncArg) float64 {
 	// There is no need in handling NaNs here, since they must be cleaned up
 	// before calling rollup funcs.
+	maxValue := rfa.prevValue
 	values := rfa.values
-	if len(values) == 0 {
-		return rfa.prevValue
+	if math.IsNaN(maxValue) {
+		if len(values) == 0 {
+			return nan
+		}
+		maxValue = values[0]
 	}
-	maxValue := values[0]
 	for _, v := range values {
 		if v > maxValue {
 			maxValue = v
@@ -567,7 +573,10 @@ func rollupSum(rfa *rollupFuncArg) float64 {
 	// before calling rollup funcs.
 	values := rfa.values
 	if len(values) == 0 {
-		return rfa.prevValue
+		if math.IsNaN(rfa.prevValue) {
+			return nan
+		}
+		return 0
 	}
 	var sum float64
 	for _, v := range values {
diff --git a/app/vmselect/promql/rollup_test.go b/app/vmselect/promql/rollup_test.go
index 5dff302b8..8b99a635d 100644
--- a/app/vmselect/promql/rollup_test.go
+++ b/app/vmselect/promql/rollup_test.go
@@ -527,7 +527,7 @@ func TestRollupFuncsNoWindow(t *testing.T) {
 		}
 		rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step)
 		values := rc.Do(nil, testValues, testTimestamps)
-		valuesExpected := []float64{nan, 21, 12, 32, 34}
+		valuesExpected := []float64{nan, 21, 12, 12, 34}
 		timestampsExpected := []int64{0, 40, 80, 120, 160}
 		testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected)
 	})