app/vmselect/promql: skip the first value in time series passed to increase() if it exceeds by more than 10x the delta between the next value and the first value

This should prvent from inflated `increase()` results for time series that start from big initial values.
Such cases may occur when a label value changes in a metric without counter reset.

app/vmselect/promql/rollup.go

@@ -1187,11 +1187,19 @@ func rollupDeltaInternal(rfa *rollupFuncArg, canUseRealPrevValue bool) float64 {
 			return nan
 		}
 		// Assume that the previous non-existing value was 0
-		// only if the first value is quite small.
+		// only if the first value doesn't exceed too much the delta with the next value.
+		//
 		// This should prevent from improper increase() results for os-level counters
 		// such as cpu time or bytes sent over the network interface.
 		// These counters may start long ago before the first value appears in the db.
-		if values[0] < 1e6 {
+		//
+		// This also should prevent from improper increase() results when a part of label values are changed
+		// without counter reset.
+		d := float64(10)
+		if len(values) > 1 {
+			d = values[1] - values[0]
+		}
+		if math.Abs(values[0]) < 10*(math.Abs(d)+1) {
 			prevValue = 0
 			if canUseRealPrevValue && !math.IsNaN(rfa.realPrevValue) {
 				prevValue = rfa.realPrevValue

app/vmselect/promql/rollup_test.go

@@ -1024,3 +1024,42 @@ func testRowsEqual(t *testing.T, values []float64, timestamps []int64, valuesExp
 		}
 	}
 }
+
+func TestRollupIncrease(t *testing.T) {
+	f := func(prevValue float64, values []float64, resultExpected float64) {
+		t.Helper()
+		rfa := &rollupFuncArg{
+			prevValue:     prevValue,
+			realPrevValue: prevValue,
+			values:        values,
+		}
+		result := rollupIncrease(rfa)
+		if math.IsNaN(result) {
+			if !math.IsNaN(resultExpected) {
+				t.Fatalf("unexpected result; got %v; want %v", result, resultExpected)
+			}
+			return
+		}
+		if result != resultExpected {
+			t.Fatalf("unexpected result; got %v; want %v", result, resultExpected)
+		}
+	}
+	f(nan, nil, nan)
+
+	// Small initial value
+	f(nan, []float64{1}, 1)
+	f(nan, []float64{10}, 10)
+	f(nan, []float64{100}, 100)
+	f(nan, []float64{1, 2, 3}, 3)
+	f(1, []float64{1, 2, 3}, 2)
+	f(nan, []float64{5, 6, 8}, 8)
+	f(2, []float64{5, 6, 8}, 6)
+
+	// Too big initial value must be skipped.
+	f(nan, []float64{1000}, 0)
+	f(nan, []float64{1000, 1001, 1002}, 2)
+
+	// Empty values
+	f(1, nil, 0)
+	f(100, nil, 0)
+}