Merge pull request #6560 from benbjohnson/optimize-tsm1-call-iterator

Move call iterator to series level

influxql/iterator.go

@@ -117,8 +117,10 @@ func (a Iterators) cast() interface{} {
 // interval.
 func NewMergeIterator(inputs []Iterator, opt IteratorOptions) Iterator {
 	inputs = Iterators(inputs).filterNonNil()
-	if len(inputs) == 0 {
+	if n := len(inputs); n == 0 {
 		return nil
+	} else if n == 1 {
+		return inputs[0]
 	}
 
 	// Aggregate functions can use a more relaxed sorting so that points

tsdb/engine/tsm1/engine.go

@@ -746,16 +746,24 @@ func (e *Engine) CreateIterator(opt influxql.IteratorOptions) (influxql.Iterator
 		inputs, err := e.createVarRefIterator(refOpt)
 		if err != nil {
 			return nil, err
+		} else if len(inputs) == 0 {
+			return nil, nil
 		}
 
-		input := influxql.NewMergeIterator(inputs, opt)
+		// Wrap each series in a call iterator.
-		if input != nil {
+		for i, input := range inputs {
 			if opt.InterruptCh != nil {
 				input = influxql.NewInterruptIterator(input, opt.InterruptCh)
 			}
-			return influxql.NewCallIterator(input, opt)
+
+			itr, err := influxql.NewCallIterator(input, opt)
+			if err != nil {
+				return nil, err
+			}
+			inputs[i] = itr
 		}
-		return nil, nil
+
+		return influxql.NewMergeIterator(inputs, opt), nil
 	}
 
 	itrs, err := e.createVarRefIterator(opt)