[query] Implemented the Graphite integralByInterval function

src/query/graphite/native/builtin_functions.go

@@ -1086,6 +1086,46 @@ func integral(ctx *common.Context, input singlePathSpec) (ts.SeriesList, error)
 	return r, nil
 }
 
+// integralByInterval will do the same as integral funcion, except it resets the total to 0
+// at the given time in the parameter “from”. Useful for finding totals per hour/day/week.
+func integralByInterval(ctx *common.Context, input singlePathSpec, intervalString string) (ts.SeriesList, error) {
+	intervalUnit, err := common.ParseInterval(intervalString)
+	if err != nil {
+		return ts.NewSeriesList(), err
+	}
+	results := make([]*ts.Series, 0, len(input.Values))
+
+	for _, series := range input.Values {
+		var (
+		  stepsPerInterval = intervalUnit.Milliseconds() /  int64(series.MillisPerStep())
+		  outVals = ts.NewValues(ctx, series.MillisPerStep(), series.Len())
+		  stepCounter int64 = 0
+		  currentSum float64
+		)
+
+		for i := 0; i < series.Len(); i++ {
+			if stepCounter > stepsPerInterval {
+				// startNewInterval
+				stepCounter = 0
+				currentSum = 0.0
+			}
+			n := series.ValueAt(i)
+			if !math.IsNaN(n) {
+				currentSum += n
+				outVals.SetValueAt(i, currentSum)
+			}
+			stepCounter += 1
+		}
+
+		newName := fmt.Sprintf("integralByInterval(%s, %s)", series.Name(), intervalString)
+		results = append(results, ts.NewSeries(ctx, newName, series.StartTime(), outVals))
+	}
+
+	r := ts.SeriesList(input)
+	r.Values = results
+	return r, nil
+}
+
 // This is the opposite of the integral function.  This is useful for taking a
 // running total metric and calculating the delta between subsequent data
 // points.
@@ -2145,6 +2185,7 @@ func init() {
 	MustRegisterFunction(holtWintersForecast)
 	MustRegisterFunction(identity)
 	MustRegisterFunction(integral)
+	MustRegisterFunction(integralByInterval)
 	MustRegisterFunction(isNonNull)
 	MustRegisterFunction(keepLastValue).WithDefaultParams(map[uint8]interface{}{
 		2: -1, // limit

src/query/graphite/native/builtin_functions_test.go

@@ -1823,6 +1823,36 @@ func TestIntegral(t *testing.T) {
 	}
 }
 
+func TestIntegralByInterval(t *testing.T) {
+	ctx := common.NewTestContext()
+	defer ctx.Close()
+
+	invals := []float64{
+		0, 1, 2, 3, 4, 5, 6, math.NaN(), 8, math.NaN(),
+	}
+
+	outvals := []float64{
+		0, 1, 3, 6, 4, 9, 15, math.NaN(), 8, math.NaN(),
+	}
+
+	series := ts.NewSeries(ctx, "hello", time.Now(),
+		common.NewTestSeriesValues(ctx, 10000, invals))
+
+	r, err := integralByInterval(ctx, singlePathSpec{
+		Values: []*ts.Series{series},
+	}, "30s")
+	require.NoError(t, err)
+
+	output := r.Values
+	require.Equal(t, 1, len(output))
+	assert.Equal(t, "integralByInterval(hello, 30s)", output[0].Name())
+	assert.Equal(t, series.StartTime(), output[0].StartTime())
+	require.Equal(t, len(outvals), output[0].Len())
+	for i, expected := range outvals {
+		xtest.Equalish(t, expected, output[0].ValueAt(i), "incorrect value at %d", i)
+	}
+}
+
 func TestDerivative(t *testing.T) {
 	ctx := common.NewTestContext()
 	defer ctx.Close()
@@ -3134,6 +3164,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"holtWintersForecast",
 		"identity",
 		"integral",
+		"integralByInterval",
 		"isNonNull",
 		"keepLastValue",
 		"legendValue",