// startingGroup returns a tagset that only has tags from `original` that are found in `tags`.
func startingGroup(original api.TagSet, tags []string) api.TagSet {
	result := api.NewTagSet()
	for _, tag := range tags {
		result[tag] = original[tag]
	}
	return result
}

// setTagSeries returns a copy of the timeseries where the given `newTag` has been set to `newValue`, or added if it wasn't present.
func setTagSeries(series api.Timeseries, newTag string, newValue string) api.Timeseries {
	tagSet := api.NewTagSet()
	for tag, val := range series.TagSet {
		tagSet[tag] = val
	}
	tagSet[newTag] = newValue
	series.TagSet = tagSet
	return series
}

// dropTagSeries returns a copy of the timeseries where the given `dropTag` has been removed from its TagSet.
func dropTagSeries(series api.Timeseries, dropTag string) api.Timeseries {
	tagSet := api.NewTagSet()
	for tag, val := range series.TagSet {
		if tag != dropTag {
			tagSet[tag] = val
		}
	}
	series.TagSet = tagSet
	return series
}

func (b movingAverageBackend) FetchSingleTimeseries(r api.FetchTimeseriesRequest) (api.Timeseries, error) {
	t := r.Timerange
	values := []float64{9, 2, 1, 6, 4, 5}
	startIndex := t.Start()/100 - 10
	result := make([]float64, t.Slots())
	for i := range result {
		result[i] = values[i+int(startIndex)]
	}
	return api.Timeseries{Values: values, TagSet: api.NewTagSet()}, nil
}

// startingCollapse returns a tagset copy of `original` but with all tags in `tags` deleted.
func startingCollase(original api.TagSet, tags []string) api.TagSet {
	result := api.NewTagSet()
	for tag, value := range original {
		result[tag] = value
	}
	for _, tagToDelete := range tags {
		delete(result, tagToDelete)
	}
	return result
}

func (value ScalarValue) ToSeriesList(timerange api.Timerange) (api.SeriesList, error) {

	series := make([]float64, timerange.Slots())
	for i := range series {
		series[i] = float64(value)
	}

	return api.SeriesList{
		Series:    []api.Timeseries{api.Timeseries{Values: series, TagSet: api.NewTagSet()}},
		Timerange: timerange,
	}, nil
}

func TestMovingAverage(t *testing.T) {
	fakeAPI := mocks.NewFakeMetricMetadataAPI()
	fakeAPI.AddPairWithoutGraphite(api.TaggedMetric{"series", api.NewTagSet()})

	fakeBackend := movingAverageBackend{}
	timerange, err := api.NewTimerange(1200, 1500, 100)
	if err != nil {
		t.Fatalf(err.Error())
	}

	expression := &functionExpression{
		functionName: "transform.moving_average",
		groupBy:      []string{},
		arguments: []function.Expression{
			&metricFetchExpression{"series", api.TruePredicate},
			durationExpression{"300ms", 300 * time.Millisecond},
		},
	}

	backend := fakeBackend
	result, err := evaluateToSeriesList(expression,
		&function.EvaluationContext{
			MetricMetadataAPI:         fakeAPI,
			TimeseriesStorageAPI:      backend,
			Timerange:                 timerange,
			SampleMethod:              api.SampleMean,
			FetchLimit:                function.NewFetchCounter(1000),
			Registry:                  registry.Default(),
			Cancellable:               api.NewCancellable(),
			OptimizationConfiguration: optimize.NewOptimizationConfiguration(),
		})
	if err != nil {
		t.Errorf(err.Error())
	}

	expected := []float64{4, 3, 11.0 / 3, 5}

	if len(result.Series) != 1 {
		t.Fatalf("expected exactly 1 returned series")
	}
	if len(result.Series[0].Values) != len(expected) {
		t.Fatalf("expected exactly %d values in returned series, but got %d", len(expected), len(result.Series[0].Values))
	}
	const eps = 1e-7
	for i := range expected {
		if math.Abs(result.Series[0].Values[i]-expected[i]) > eps {
			t.Fatalf("expected %+v but got %+v", expected, result.Series[0].Values)
		}
	}
}

func Test_ParallelMultiBackend_Success(t *testing.T) {
	a := assert.New(t)
	suite := newSuite()
	defer suite.cleanup()
	go func() {
		_, err := suite.multiBackend.FetchMultipleSeries(api.FetchMultipleRequest{
			Metrics:     []api.TaggedMetric{api.TaggedMetric{"a", api.NewTagSet()}},
			Cancellable: suite.cancellable,
		})
		a.CheckError(err)
		suite.waitGroup.Done()
	}()
	suite.backend.tickets <- struct{}{}
	suite.waitGroup.Wait()
}

// extractTagValues extracts the tagset using the given regex and the list of tags.
func extractTagValues(regex *regexp.Regexp, tagList []string, input string) api.TagSet {
	matches := regex.FindStringSubmatch(input)
	if matches == nil {
		return nil
	}
	tagSet := api.NewTagSet()
	for index, tagValue := range matches {
		if index == 0 {
			continue
		}
		tagKey := tagList[index-1]
		tagSet[tagKey] = tagValue
	}
	return tagSet
}

func TestMovingAverage(t *testing.T) {
	fakeAPI := mocks.NewFakeApi()
	fakeAPI.AddPair(api.TaggedMetric{"series", api.NewTagSet()}, "series")

	fakeBackend := movingAverageBackend{}
	timerange, err := api.NewTimerange(1200, 1500, 100)
	if err != nil {
		t.Fatalf(err.Error())
	}

	expression := &functionExpression{
		functionName: "transform.moving_average",
		groupBy:      []string{},
		arguments: []function.Expression{
			&metricFetchExpression{"series", api.TruePredicate},
			stringExpression{"300ms"},
		},
	}

	result, err := evaluateToSeriesList(expression,
		function.EvaluationContext{
			API:          fakeAPI,
			MultiBackend: backend.NewSequentialMultiBackend(fakeBackend),
			Timerange:    timerange,
			SampleMethod: api.SampleMean,
			FetchLimit:   function.NewFetchCounter(1000),
			Registry:     registry.Default(),
		})
	if err != nil {
		t.Errorf(err.Error())
	}

	expected := []float64{4, 3, 11.0 / 3, 5}
	if len(result.Series) != 1 {
		t.Fatalf("expected exactly 1 returned series")
	}
	if len(result.Series[0].Values) != len(expected) {
		t.Fatalf("expected exactly %d values in returned series", len(expected))
	}
	const eps = 1e-7
	for i := range expected {
		if math.Abs(result.Series[0].Values[i]-expected[i]) > eps {
			t.Fatalf("expected %+v but got %+v", expected, result.Series[0].Values)
		}
	}
}

func TestCaching(t *testing.T) {
	optimizer := NewOptimizationConfiguration()
	optimizer.EnableMetricMetadataCaching = true

	updateFunc := func() ([]api.TagSet, error) {
		// map[string]string
		result := []api.TagSet{api.NewTagSet()}
		return result, nil
	}
	someMetric := api.MetricKey("blah")
	optimizer.AllTagsCacheHitOrExecute(someMetric, updateFunc)

	updateFunc = func() ([]api.TagSet, error) {
		t.Errorf("Should not be called")
		return nil, nil
	}

	optimizer.AllTagsCacheHitOrExecute(someMetric, updateFunc)
}

func Test_ScalarExpression(t *testing.T) {
	timerangeA, err := api.NewTimerange(0, 10, 2)
	if err != nil {
		t.Fatalf("invalid timerange used for testcase")
		return
	}
	for _, test := range []struct {
		expr           scalarExpression
		timerange      api.Timerange
		expectedSeries []api.Timeseries
	}{
		{
			scalarExpression{5},
			timerangeA,
			[]api.Timeseries{
				api.Timeseries{
					Values: []float64{5.0, 5.0, 5.0, 5.0, 5.0, 5.0},
					TagSet: api.NewTagSet(),
				},
			},
		},
	} {
		a := assert.New(t).Contextf("%+v", test)
		result, err := evaluateToSeriesList(test.expr, &function.EvaluationContext{
			TimeseriesStorageAPI: FakeBackend{},
			Timerange:            test.timerange,
			SampleMethod:         api.SampleMean,
			FetchLimit:           function.NewFetchCounter(1000),
			Registry:             registry.Default(),
		})

		if err != nil {
			t.Fatalf("failed to convert number into serieslist")
		}

		a.EqInt(len(result.Series), len(test.expectedSeries))

		for i := 0; i < len(result.Series); i++ {
			a.Eq(result.Series[i].Values, test.expectedSeries[i].Values)
		}
	}
}

// This method takes a partial joinrow, and evaluates the validity of appending `series` to it.
// If this is possible, return the new series and true; otherwise return false for "ok"
func extendRow(row JoinRow, series api.Timeseries) (JoinRow, bool) {
	for key, newValue := range series.TagSet {
		oldValue, ok := map[string]string(row.TagSet)[key]
		if ok && newValue != oldValue {
			// If this occurs, then the candidate member (series) and the rest of the row are in
			// conflict about `key`, since they assign it different values. If this occurs, then
			// it is not possible to assign any key here.
			return JoinRow{}, false
		}
	}
	// if this point has been reached, then it is possible to extend the row without conflict
	newTagSet := api.NewTagSet()
	result := JoinRow{newTagSet, append(row.Row, series)}
	for key, newValue := range series.TagSet {
		newTagSet[key] = newValue
	}
	for key, oldValue := range row.TagSet {
		newTagSet[key] = oldValue
	}
	return result, true
}

func Test_ParallelMultiBackend_Timeout(t *testing.T) {
	a := assert.New(t)
	suite := newSuite()
	defer suite.cleanup()
	go func() {
		_, err := suite.multiBackend.FetchMultipleSeries(api.FetchMultipleRequest{
			Metrics:     []api.TaggedMetric{api.TaggedMetric{"a", api.NewTagSet()}},
			Cancellable: suite.cancellable,
		})
		if err == nil {
			t.Errorf("Error expected, but got nil")
		} else {
			casted, ok := err.(api.BackendError)
			if !ok {
				t.Errorf("Invalid error type")
			} else {
				a.Eq(casted.Code, api.FetchTimeoutError)
			}
		}
		suite.waitGroup.Done()
	}()
	close(suite.cancellable.Done())
	suite.waitGroup.Wait()
}

// Join generates a cartesian product of the given series lists, and then returns rows where the tags are matching.
func Join(lists []api.SeriesList) JoinResult {
	// place an empty row inside the results list first
	// this row will be used to build up all others
	emptyRow := JoinRow{api.NewTagSet(), []api.Timeseries{}}
	results := []JoinRow{emptyRow}

	// The `results` list is given an inductive definition:
	// at the end of the `i`th iteration of the outer loop,
	// `results` corresponds to the join of the first `i` seriesLists given as input

	for _, list := range lists {
		next := []JoinRow{}
		// `next` is gradually accumulated into the final join of the first `i` (iteration) seriesLists
		// results already contains the join of the the first (i-1)th series
		for _, series := range list.Series {
			// here we have our series
			// iterator over the results of the previous iteration:
			for _, previous := range results {
				// consider adding this series to each row from the joins of all previous series lists
				// if this is successful, the newly extended list is added to the `next` slice
				extension, ok := extendRow(previous, series)
				if ok {
					next = append(next, extension)
				}
			}
		}
		// `next` now contains the join of the first `i` iterations,
		// while `results` contains the join of the first `i-1` iterations.
		results = next
		// thus we update `results`
	}
	// at this stage, iteration has continued over the entire set of lists,
	// so `results` contains the join of all of the lists.

	return JoinResult{Rows: results}
}

func TestCacheExpiration(t *testing.T) {
	optimizer := NewOptimizationConfiguration()
	optimizer.EnableMetricMetadataCaching = true

	latch := false
	updateFunc := func() ([]api.TagSet, error) {
		// map[string]string
		latch = true
		result := []api.TagSet{api.NewTagSet()}
		return result, nil
	}
	someMetric := api.MetricKey("blah")
	optimizer.AllTagsCacheHitOrExecute(someMetric, updateFunc)
	if !latch {
		t.Errorf("We expected the update function to be called, but it wasn't")
	}
	optimizer.TimeSourceForNow = func() time.Time { return time.Now().Add(5 * time.Hour) }
	latch = false // Reset the latch

	optimizer.AllTagsCacheHitOrExecute(someMetric, updateFunc)
	if !latch {
		t.Errorf("We expected the update function to be called, but it wasn't")
	}
}

func (expr *LiteralExpression) Evaluate(context *function.EvaluationContext) (function.Value, error) {
	return api.SeriesList{
		Series:    []api.Timeseries{api.Timeseries{Values: expr.Values, TagSet: api.NewTagSet()}},
		Timerange: api.Timerange{},
	}, nil
}

func (a fakeAPI) ToTaggedName(metric api.GraphiteMetric) (api.TaggedMetric, error) {
	return api.TaggedMetric{
		MetricKey: api.MetricKey(metric),
		TagSet:    api.NewTagSet(),
	}, nil
}

func TestCommand_Select(t *testing.T) {
	epsilon := 1e-10
	fakeApi := mocks.NewFakeApi()
	fakeApi.AddPair(api.TaggedMetric{"series_1", api.ParseTagSet("dc=west")}, emptyGraphiteName)
	fakeApi.AddPair(api.TaggedMetric{"series_2", api.ParseTagSet("dc=east")}, emptyGraphiteName)
	fakeApi.AddPair(api.TaggedMetric{"series_2", api.ParseTagSet("dc=west")}, emptyGraphiteName)
	fakeApi.AddPair(api.TaggedMetric{"series_3", api.ParseTagSet("dc=west")}, emptyGraphiteName)
	fakeApi.AddPair(api.TaggedMetric{"series_3", api.ParseTagSet("dc=east")}, emptyGraphiteName)
	fakeApi.AddPair(api.TaggedMetric{"series_3", api.ParseTagSet("dc=north")}, emptyGraphiteName)
	fakeApi.AddPair(api.TaggedMetric{"series_timeout", api.ParseTagSet("dc=west")}, emptyGraphiteName)
	var fakeBackend fakeApiBackend
	testTimerange, err := api.NewTimerange(0, 120, 30)
	if err != nil {
		t.Errorf("Invalid test timerange")
		return
	}
	earlyTimerange, err := api.NewTimerange(0, 60, 30)
	if err != nil {
		t.Errorf("Invalid test timerange")
	}
	lateTimerange, err := api.NewTimerange(60, 120, 30)
	if err != nil {
		t.Errorf("Invalid test timerange")
	}
	for _, test := range []struct {
		query       string
		expectError bool
		expected    api.SeriesList
	}{
		{"select does_not_exist from 0 to 120 resolution 30ms", true, api.SeriesList{}},
		{"select series_1 from 0 to 120 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{1, 2, 3, 4, 5},
				api.ParseTagSet("dc=west"),
			}},
			Timerange: testTimerange,
			Name:      "series_1",
		}},
		{"select series_timeout from 0 to 120 resolution 30ms", true, api.SeriesList{}},
		{"select series_1 + 1 from 0 to 120 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{2, 3, 4, 5, 6},
				api.ParseTagSet("dc=west"),
			}},
			Timerange: testTimerange,
			Name:      "",
		}},
		{"select series_1 * 2 from 0 to 120 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{2, 4, 6, 8, 10},
				api.ParseTagSet("dc=west"),
			}},
			Timerange: testTimerange,
			Name:      "",
		}},
		{"select aggregate.max(series_2) from 0 to 120 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{3, 2, 3, 6, 5},
				api.NewTagSet(),
			}},
			Timerange: testTimerange,
			Name:      "series_2",
		}},
		{"select (1 + series_2) | aggregate.max from 0 to 120 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{4, 3, 4, 7, 6},
				api.NewTagSet(),
			}},
			Timerange: testTimerange,
			Name:      "series_2",
		}},
		{"select series_1 from 0 to 60 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{1, 2, 3},
				api.ParseTagSet("dc=west"),
			}},
			Timerange: earlyTimerange,
			Name:      "series_1",
		}},
		{"select transform.timeshift(series_1,31ms) from 0 to 60 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{2, 3, 4},
				api.ParseTagSet("dc=west"),
			}},
			Timerange: earlyTimerange,
			Name:      "series_1",
		}},
		{"select transform.timeshift(series_1,62ms) from 0 to 60 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{3, 4, 5},
				api.ParseTagSet("dc=west"),
			}},
			Timerange: earlyTimerange,
			Name:      "series_1",
		}},
		{"select transform.timeshift(series_1,29ms) from 0 to 60 resolution 30ms", false, api.SeriesList{
			Series: []api.Timeseries{{
				[]float64{2, 3, 4},
				api.ParseTagSet("dc=west"),
			}},
//.........这里部分代码省略.........