[query] Implemented the Graphite timeSlice function

src/query/graphite/native/builtin_functions.go

@@ -32,6 +32,7 @@ import (
 
 	"github.com/m3db/m3/src/query/graphite/common"
 	"github.com/m3db/m3/src/query/graphite/errors"
+	"github.com/m3db/m3/src/query/graphite/graphite"
 	"github.com/m3db/m3/src/query/graphite/ts"
 	"github.com/m3db/m3/src/query/util"
 )
@@ -242,6 +243,56 @@ func timeShift(
 	}, nil
 }
 
+
+func timeSlice(ctx *common.Context, input singlePathSpec, start string, end string) (*unaryContextShifter, error) {
+
+	contextShiftingFn := func(c *common.Context) *common.Context {
+		// no need to shift the context here
+		return c;
+	}
+
+	tzOffsetForAbsoluteTime := time.Duration(0)
+	startTime, err := graphite.ParseTime(start, time.Now(), tzOffsetForAbsoluteTime)
+	endTime, err := graphite.ParseTime(end, time.Now(), tzOffsetForAbsoluteTime)
+
+	if (err != nil) {
+		return nil, err
+	}
+
+	transformerFn := func(input ts.SeriesList) (ts.SeriesList, error) {
+		output := make([]*ts.Series, input.Len())
+
+		for i, series := range input.Values {
+			nanoSecondsPerStep := series.MillisPerStep() * 1000000
+			stepDuration := time.Duration(nanoSecondsPerStep)
+			truncatedValues := ts.NewValues(ctx, series.MillisPerStep(), series.Len())
+
+			currentTime := series.StartTime()
+			for i := 0; i < series.Len(); i++ {
+				ctString := currentTime.String()
+				fmt.Println(ctString)
+				if ( currentTime.After(startTime) && currentTime.Before(endTime)) {
+					truncatedValues.SetValueAt(i, series.ValueAtTime(currentTime))
+				}
+				currentTime = currentTime.Add(stepDuration)
+			}
+
+			slicedSeries := ts.NewSeries(ctx, series.Name(), series.StartTime(), truncatedValues)
+			renamedSlicedSeries := slicedSeries.RenamedTo(fmt.Sprintf("timeSlice(%s,%s, %s)", slicedSeries.Name(), start, end))
+			output[i] = renamedSlicedSeries
+		}
+		input.Values = output
+		return input, nil
+	}
+
+	return &unaryContextShifter{
+		ContextShiftFunc: contextShiftingFn,
+		UnaryTransformer: transformerFn,
+	}, nil
+}
+
+
+
 // 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,
@@ -1972,6 +2023,9 @@ func init() {
 	MustRegisterFunction(timeShift).WithDefaultParams(map[uint8]interface{}{
 		3: true, // resetEnd
 	})
+	MustRegisterFunction(timeSlice).WithDefaultParams(map[uint8]interface{}{
+		3: "now", // endTime
+	})
 	MustRegisterFunction(transformNull).WithDefaultParams(map[uint8]interface{}{
 		2: 0.0, // defaultValue
 	})

src/query/graphite/native/builtin_functions_test.go

@@ -605,7 +605,7 @@ func TestTransformNull(t *testing.T) {
 var (
 	testMovingFunctionBootstrap = testMovingFunctionStart.Add(-30 * time.Second)
 	testMovingFunctionStart     = time.Now().Truncate(time.Minute)
-	testMovingFunctionEnd       = testMovingFunctionStart.Add(time.Minute)
+	testMovingFunctionEnd       = testMovingFunctionStart.Add(time.Minute * 120)
 )
 
 func testMovingFunction(t *testing.T, target, expectedName string, values, bootstrap, output []float64) {
@@ -642,6 +642,44 @@ func testMovingFunction(t *testing.T, target, expectedName string, values, boots
 		expected, res.Values)
 }
 
+var (
+	testGeneralFunctionStart     = time.Now().Add(time.Minute * -11).Truncate(time.Minute)
+	testGeneralFunctionEnd       = time.Now().Add(time.Minute * -3).Truncate(time.Minute)
+)
+
+// testGeneralFunction is a copy of testMovingFunction but without any logic for bootstrapping values
+func testGeneralFunction(t *testing.T, target, expectedName string, values, output []float64) {
+	ctx := common.NewTestContext()
+	defer ctx.Close()
+
+	engine := NewEngine(
+		&common.MovingFunctionStorage{
+			StepMillis:     60000,
+			Values:         values,
+		},
+	)
+	phonyContext := common.NewContext(common.ContextOptions{
+		Start:  testGeneralFunctionStart,
+		End:    testGeneralFunctionEnd,
+		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, 60000, testMovingFunctionStart,
+		expected, res.Values)
+}
+
 func TestMovingAverageSuccess(t *testing.T) {
 	values := []float64{12.0, 19.0, -10.0, math.NaN(), 10.0}
 	bootstrap := []float64{3.0, 4.0, 5.0}
@@ -2549,12 +2587,12 @@ func TestMovingMismatchedLimits(t *testing.T) {
 	// points. When limits do not snap exactly, the first point should be omitted.
 	for _, fn := range []string{"movingAverage", "movingMedian"} {
 		for i := time.Duration(0); i < time.Minute; i += time.Second {
-			testMovingAverageInvalidLimits(t, fn, i)
+			testMovingFunctionInvalidLimits(t, fn, i)
 		}
 	}
 }
 
-func testMovingAverageInvalidLimits(t *testing.T, fn string, offset time.Duration) {
+func testMovingFunctionInvalidLimits(t *testing.T, fn string, offset time.Duration) {
 	ctrl := xgomock.NewController(t)
 	defer ctrl.Finish()
 
@@ -2840,6 +2878,42 @@ func TestTimeShift(t *testing.T) {
 		[]common.TestSeries{expected}, res.Values)
 }
 
+/*
+
+   def test_timeSlice(self):
+       # series starts at 60 seconds past the epoch and continues for 600 seconds (ten minutes)
+       # steps are every 60 seconds
+       seriesList = self._gen_series_list_with_data(
+           key='test.value',
+           start=0,
+           end=600,
+           step=60,
+           data=[None,1,2,3,None,5,6,None,7,8,9]
+       )
+
+       # we're going to slice such that we only include minutes 3 to 8 (of 0 to 9)
+       expectedResult = [
+           TimeSeries('timeSlice(test.value, 180, 480)',0,600,60,[None,None,None,3,None,5,6,None,7,None,None])
+       ]
+
+       results = functions.timeSlice(
+           self._build_requestContext(
+               startTime=datetime(1970, 1, 1, 0, 0, 0, 0, pytz.timezone(settings.TIME_ZONE)),
+               endTime=datetime(1970, 1, 1, 0, 9, 0, 0, pytz.timezone(settings.TIME_ZONE))
+           ),
+           seriesList,
+           '00:03 19700101',
+           '00:08 19700101'
+       )
+       self.assertEqual(results, expectedResult)
+*/
+func TestTimeSlice(t *testing.T) {
+	values := []float64{math.NaN(),1.0,2.0,3.0,math.NaN(),5.0,6.0,math.NaN(),7.0,8.0,9.0}
+	expected := []float64{math.NaN(),math.NaN(),math.NaN(),3.0,math.NaN(),5.0,6.0,math.NaN(),7.0,math.NaN(),math.NaN()}
+
+	testGeneralFunction(t, "timeSlice(foo.bar.baz, '-8min','-2min')", "timeSlice(foo.bar.baz,-8min, -2min)", values, expected)
+}
+
 func TestDashed(t *testing.T) {
 	ctx := common.NewTestContext()
 	defer ctx.Close()
@@ -2989,6 +3063,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"time",
 		"timeFunction",
 		"timeShift",
+		"timeSlice",
 		"transformNull",
 		"weightedAverage",
 	}