Allow merge() to work with one series and to take a list of time series

coordinator/coordinator.go

@@ -278,16 +278,11 @@ func (self *Coordinator) expandRegex(spec *parser.QuerySpec) {
 		return
 	}
 
-	if f := q.FromClause; f.Type == parser.FromClauseMergeFun {
-		series := self.clusterConfiguration.MetaStore.GetSeriesForDatabaseAndRegex(spec.Database(), q.FromClause.Regex)
-		f.Type = parser.FromClauseMerge
-		f.Regex = nil
-		for _, s := range series {
-			f.Names = append(f.Names, &parser.TableName{
-				Name: &parser.Value{Name: s, Type: parser.ValueTableName},
-			})
-		}
+	f := func(r *regexp.Regexp) []string {
+		return self.clusterConfiguration.MetaStore.GetSeriesForDatabaseAndRegex(spec.Database(), r)
 	}
+
+	parser.RewriteMergeQuery(q, f)
 }
 
 // We call this function only if we have a Select query (not continuous) or Delete query

engine/engine.go

@@ -25,7 +25,7 @@ func NewQueryEngine(next Processor, query *parser.SelectQuery, shards []uint32)
 			tables[i] = name.Name.Name
 		}
 		engine = NewMergeEngine(shards, query.Ascending, engine)
-	case parser.FromClauseMergeFun:
+	case parser.FromClauseMergeRegex:
 		// At this point the regex should be expanded to the list of
 		// tables that will be queries
 		panic("QueryEngine cannot be called with merge function")

integration/data_test.go

@@ -267,6 +267,28 @@ func (self *DataTestSuite) TestModeWithNils(c *C) {
 	c.Assert(maps[0]["m2"], Equals, nil)
 }
 
+func (self *DataTestSuite) TestMergeRegexOneSeries(c *C) {
+	data := `
+[
+  {
+    "name": "test_merge_1",
+    "columns": ["time", "value"],
+    "points": [
+      [1401321700000, "m11"],
+      [1401321600000, "m12"]
+    ]
+  }
+]`
+
+	self.client.WriteJsonData(data, c, influxdb.Millisecond)
+	serieses := self.client.RunQuery("select * from merge(/.*/)", c, "m")
+	c.Assert(serieses, HasLen, 1)
+	maps := ToMap(serieses[0])
+	c.Assert(maps, HasLen, 2)
+	c.Assert(maps[0]["value"], Equals, "m11")
+	c.Assert(maps[1]["value"], Equals, "m12")
+}
+
 func (self *DataTestSuite) TestMergeRegex(c *C) {
 	data := `
 [

parser/from_clause.go

@@ -13,10 +13,10 @@ import "fmt"
 type FromClauseType int
 
 const (
-	FromClauseArray     FromClauseType = C.FROM_ARRAY
-	FromClauseMerge     FromClauseType = C.FROM_MERGE
-	FromClauseInnerJoin FromClauseType = C.FROM_INNER_JOIN
-	FromClauseMergeFun  FromClauseType = C.FROM_MERGE_FUNCTION
+	FromClauseArray      FromClauseType = C.FROM_ARRAY
+	FromClauseMerge      FromClauseType = C.FROM_MERGE
+	FromClauseInnerJoin  FromClauseType = C.FROM_INNER_JOIN
+	FromClauseMergeRegex FromClauseType = C.FROM_MERGE_REGEX
 )
 
 func (self *TableName) GetAlias() string {
@@ -48,8 +48,14 @@ func (self *FromClause) GetString() string {
 	buffer := bytes.NewBufferString("")
 	switch self.Type {
 	case FromClauseMerge:
-		fmt.Fprintf(buffer, "%s%s merge %s %s", self.Names[0].Name.GetString(), self.Names[1].GetAliasString(),
-			self.Names[1].Name.GetString(), self.Names[1].GetAliasString())
+		fmt.Fprintf(buffer, "merge(")
+		for i, n := range self.Names {
+			if i > 0 {
+				buffer.WriteRune(',')
+			}
+			buffer.WriteString(n.Name.GetString())
+		}
+		buffer.WriteString(")")
 	case FromClauseInnerJoin:
 		fmt.Fprintf(buffer, "%s%s inner join %s%s", self.Names[0].Name.GetString(), self.Names[0].GetAliasString(),
 			self.Names[1].Name.GetString(), self.Names[1].GetAliasString())

parser/parser.go

@@ -473,7 +473,7 @@ func GetFromClause(fromClause *C.from_clause) (*FromClause, error) {
 	var regex *regexp.Regexp
 
 	switch t {
-	case FromClauseMergeFun:
+	case FromClauseMergeRegex:
 		val, err := GetValue(fromClause.regex_value)
 		if err != nil {
 			return nil, err

parser/parser_test.go

@@ -32,6 +32,32 @@ func (self *QueryParserSuite) TestInvalidFromClause(c *C) {
 	c.Assert(err, ErrorMatches, ".*\\$undefined.*")
 }
 
+// Make sure that GetQueryStringWithTimeCondition() works for regex
+// merge queries.
+func (self *QueryParserSuite) TestMergeMultipleRegex(c *C) {
+	query := "select * from merge(/.*foo.*/, /.*bar.*/)"
+	_, err := ParseQuery(query)
+	c.Assert(err, NotNil)
+}
+
+func (self *QueryParserSuite) TestParseMergeGetString(c *C) {
+	f := func(r *regexp.Regexp) []string {
+		return []string{"foobar"}
+	}
+
+	query := "select * from merge(/.*foo.*/)"
+	qs, err := ParseQuery(query)
+	c.Assert(err, IsNil)
+	c.Assert(qs, HasLen, 1)
+	RewriteMergeQuery(qs[0].SelectQuery, f)
+	actualQs, err := ParseQuery(qs[0].GetQueryStringWithTimeCondition())
+	c.Assert(err, IsNil)
+	c.Assert(actualQs, HasLen, 1)
+	RewriteMergeQuery(actualQs[0].SelectQuery, f)
+	actualQs[0].SelectQuery.startTimeSpecified = false
+	c.Assert(actualQs[0].SelectQuery, DeepEquals, qs[0].SelectQuery)
+}
+
 func (self *QueryParserSuite) TestInvalidExplainQueries(c *C) {
 	query := "explain select foo, baz group by time(1d)"
 	_, err := ParseQuery(query)
@@ -266,7 +292,7 @@ func (self *QueryParserSuite) TestParseFromWithMergeRegex(c *C) {
 	q, err := ParseSelectQuery("select count(*) from merge(/.*/) where time>now()-1d;")
 	c.Assert(err, IsNil)
 	fromClause := q.GetFromClause()
-	c.Assert(fromClause.Type, Equals, FromClauseMergeFun)
+	c.Assert(fromClause.Type, Equals, FromClauseMergeRegex)
 	c.Assert(fromClause.Regex, NotNil)
 }
 

parser/query.yacc

@@ -423,10 +423,17 @@ FROM_CLAUSE:
           $$->from_clause_type = FROM_MERGE;
         }
         |
+        FROM MERGE '(' SIMPLE_TABLE_VALUES ')'
+        {
+          $$ = calloc(1, sizeof(from_clause));
+          $$->names = $4;
+          $$->from_clause_type = FROM_MERGE;
+        }
+        |
         FROM MERGE '(' REGEX_VALUE ')'
         {
           $$ = calloc(1, sizeof(from_clause));
-          $$->from_clause_type = FROM_MERGE_FUNCTION;
+          $$->from_clause_type = FROM_MERGE_REGEX;
           $$->regex_value = $4;
         }
         |

parser/query_types.h

@@ -73,7 +73,7 @@ typedef struct {
     FROM_ARRAY,
     FROM_MERGE,
     FROM_INNER_JOIN,
-    FROM_MERGE_FUNCTION
+    FROM_MERGE_REGEX
   } from_clause_type;
   // in case of merge or join, it's guaranteed that the names array
   // will have two table names only and they aren't regex.

parser/rewrite.go

@@ -0,0 +1,28 @@
+package parser
+
+import "regexp"
+
+type RegexMatcher func(r *regexp.Regexp) []string
+
+// Given a function that returns the series names that match the given
+// regex, this function will rewrite the query such that the matching
+// serires names are included in the query. E.g. if we have series
+// names foobar, foobaz and barbaz and a query
+//     select * from merge(/foo.*/)
+// the query will be rewritten to
+//     select * from merge(foobar, foobaz)
+func RewriteMergeQuery(query *SelectQuery, rm RegexMatcher) {
+	if query.FromClause.Type != FromClauseMergeRegex {
+		return
+	}
+
+	series := rm(query.FromClause.Regex)
+	f := query.FromClause
+	f.Type = FromClauseMerge
+	f.Regex = nil
+	for _, s := range series {
+		f.Names = append(f.Names, &TableName{
+			Name: &Value{Name: s, Type: ValueSimpleName},
+		})
+	}
+}