[query] Implemented the Graphite delay function

src/query/graphite/native/builtin_functions.go

@@ -242,6 +242,40 @@ func timeShift(
 	}, nil
 }
 
+
+// delay shifts all samples later by an integer number of steps. This can be used
+// for custom derivative calculations, among other things. Note: this will pad
+// the early end of the data with NaN for every step shifted. delay complements
+// other time-displacement functions such as timeShift and timeSlice, in that
+// delay is indifferent about the step intervals being shifted.
+func delay(
+	ctx *common.Context,
+	singlePath singlePathSpec,
+	steps int,
+) (ts.SeriesList, error) {
+		input := ts.SeriesList(singlePath)
+		output := make([]*ts.Series, input.Len())
+
+		for i, series := range input.Values {
+			delayedVals := delayValuesHelper(ctx, *series, steps)
+			delayedSeries := ts.NewSeries(ctx, series.Name(), series.StartTime(), delayedVals)
+			renamedSeries := delayedSeries.RenamedTo(fmt.Sprintf("delay(%s,%d)", delayedSeries.Name(), steps))
+			output[i] = renamedSeries
+		}
+		input.Values = output
+		return input, nil
+}
+
+// delayValuesHelper takes a series and returns a copy of the values after
+// delaying the values by `steps` number of steps
+func delayValuesHelper(ctx *common.Context, series ts.Series, steps int ) (ts.Values) {
+	output := ts.NewValues(ctx, series.MillisPerStep(), series.Len())
+	for i := steps; i < series.Len(); i++ {
+		output.SetValueAt(i, series.ValueAt(i - steps))
+	}
+	return output
+}
+
 // absolute returns the absolute value of each element in the series.
 func absolute(ctx *common.Context, input singlePathSpec) (ts.SeriesList, error) {
 	return transform(ctx, input,
@@ -1885,6 +1919,7 @@ func init() {
 	MustRegisterFunction(dashed).WithDefaultParams(map[uint8]interface{}{
 		2: 5.0, // dashLength
 	})
+	MustRegisterFunction(delay)
 	MustRegisterFunction(derivative)
 	MustRegisterFunction(diffSeries)
 	MustRegisterFunction(divideSeries)

src/query/graphite/native/builtin_functions_test.go

@@ -2840,6 +2840,61 @@ func TestTimeShift(t *testing.T) {
 		[]common.TestSeries{expected}, res.Values)
 }
 
+func TestDelay(t *testing.T) {
+	var values = [3][]float64{
+		{54.0, 48.0, 92.0, 54.0, 14.0, 1.2},
+		{4.0, 5.0, math.NaN(), 6.4, 7.2, math.NaN()},
+		{math.NaN(), 8.0, 9.0, 10.6, 11.2, 12.2},
+	}
+	expected := [3][]float64{
+		{math.NaN(), math.NaN(), math.NaN(), 54.0, 48.0, 92.0},
+		{math.NaN(), math.NaN(), math.NaN(), 4.0, 5.0, math.NaN()},
+		{math.NaN(), math.NaN(), math.NaN(), math.NaN(), 8.0, 9.0},
+	}
+
+	for index, value := range values {
+		e := expected[index]
+		testDelay(t, "delay(foo.bar.baz, 3)", "delay(foo.bar.baz,3)", value, e)
+	}
+}
+
+var (
+	testDelayStart     = time.Now().Truncate(time.Minute)
+	testDelayEnd       = testMovingFunctionEnd.Add(time.Minute)
+)
+
+func testDelay(t *testing.T, target, expectedName string, values, output []float64) {
+	ctx := common.NewTestContext()
+	defer ctx.Close()
+
+	engine := NewEngine(
+		&common.MovingFunctionStorage{
+			StepMillis:     10000,
+			Values:         values,
+		},
+	)
+	phonyContext := common.NewContext(common.ContextOptions{
+		Start:  testDelayStart,
+		End:    testDelayEnd,
+		Engine: engine,
+	})
+
+	expr, err := phonyContext.Engine.(*Engine).Compile(target)
+	require.NoError(t, err)
+	res, err := expr.Execute(phonyContext)
+	require.NoError(t, err)
+	var expected []common.TestSeries
+
+	if output != nil {
+		expectedSeries := common.TestSeries{
+			Name: expectedName,
+			Data: output,
+		}
+		expected = append(expected, expectedSeries)
+	}
+	common.CompareOutputsAndExpected(t, 10000, testDelayStart, expected, res.Values)
+}
+
 func TestDashed(t *testing.T) {
 	ctx := common.NewTestContext()
 	defer ctx.Close()
@@ -2927,6 +2982,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"currentAbove",
 		"currentBelow",
 		"dashed",
+		"delay",
 		"derivative",
 		"diffSeries",
 		"divideSeries",