Merge pull request #486 from influxdb/fix-486-remove-extra-columns

Output of interpolated time series should not include the columns used to derive the result series name

src/coordinator/coordinator.go

@@ -523,15 +523,39 @@ func (self *CoordinatorImpl) InterpolateValuesAndCommit(query string, db string,
 	sequenceMap := make(map[sequenceKey]int)
 	r, _ := regexp.Compile(`\[.*?\]`)
 
+	// get the fields that are used in the target name
+	fieldsInTargetName := r.FindAllString(targetName, -1)
+	fieldsIndeces := make([]int, 0, len(fieldsInTargetName))
+	for i, f := range fieldsInTargetName {
+		f = f[1 : len(f)-1]
+		fieldsIndeces = append(fieldsIndeces, series.GetFieldIndex(f))
+		fieldsInTargetName[i] = f
+	}
+
+	// remove the fields used in the target name from the series fields
+	for i, fi := range fieldsIndeces {
+		// move the fields to the left
+		copy(series.Fields[fi-i:], series.Fields[fi-i+1:])
+		series.Fields = series.Fields[:len(series.Fields)-1]
+	}
+
 	if r.MatchString(targetName) {
 		serieses := map[string]*protocol.Series{}
 		for _, point := range series.Points {
-			targetNameWithValues := r.ReplaceAllStringFunc(targetName, func(match string) string {
-				fieldName := match[1 : len(match)-1]
-				fieldIndex := series.GetFieldIndex(fieldName)
-				return point.GetFieldValueAsString(fieldIndex)
+			fieldIndex := 0
+			targetNameWithValues := r.ReplaceAllStringFunc(targetName, func(_ string) string {
+				value := point.GetFieldValueAsString(fieldsIndeces[fieldIndex])
+				fieldIndex++
+				return value
 			})
 
+			// remove the fields used in the target name from the series fields
+			for i, fi := range fieldsIndeces {
+				// move the fields to the left
+				copy(point.Values[fi-i:], point.Values[fi-i+1:])
+				point.Values = point.Values[:len(point.Values)-1]
+			}
+
 			if assignSequenceNumbers {
 				key := sequenceKey{targetNameWithValues, *point.Timestamp}
 				sequenceMap[key] += 1

src/integration/multiple_servers_test.go

@@ -745,15 +745,11 @@ func (self *ServerSuite) TestContinuousQueryInterpolation(c *C) {
 
 	collection = self.serverProcesses[0].Query("test_cq", "select * from s4.foo.1.a;", false, c)
 	series = collection.GetSeries("s4.foo.1.a", c)
-	c.Assert(series.GetValueForPointAndColumn(0, "c6", c), Equals, float64(1))
-	c.Assert(series.GetValueForPointAndColumn(0, "c7", c), Equals, "a")
-	c.Assert(series.GetValueForPointAndColumn(0, "c8", c), Equals, float64(10))
+	c.Assert(series.GetValueForPointAndColumn(0, "c8", c), Equals, 10.0)
 
 	collection = self.serverProcesses[0].Query("test_cq", "select * from s4.foo.2.b;", false, c)
 	series = collection.GetSeries("s4.foo.2.b", c)
-	c.Assert(series.GetValueForPointAndColumn(0, "c6", c), Equals, float64(2))
-	c.Assert(series.GetValueForPointAndColumn(0, "c7", c), Equals, "b")
-	c.Assert(series.GetValueForPointAndColumn(0, "c8", c), Equals, float64(11))
+	c.Assert(series.GetValueForPointAndColumn(0, "c8", c), Equals, 11.0)
 
 	self.serverProcesses[0].QueryAsRoot("test_cq", "drop continuous query 1;", false, c)
 	self.serverProcesses[0].QueryAsRoot("test_cq", "drop continuous query 2;", false, c)
@@ -821,8 +817,11 @@ func (self *ServerSuite) TestContinuousQuerySequenceNumberAssignmentWithInterpol
 	subseries := []string{"a", "aa", "b", "bb"}
 	for i := range subseries {
 		series = collection.GetSeries("points.count."+subseries[i], c)
-		c.Assert(series.Points, HasLen, 1)
-		c.Assert(series.Points[0][1], Equals, 1.0)
+		maps := ToMap(series)
+		c.Assert(maps, HasLen, 1)
+		c.Assert(maps[0]["sequence_number"], Equals, 1.0)
+		// c2 shouldn't be included as a column
+		c.Assert(maps[0]["c2"], IsNil)
 	}
 }