From fdf34d435632ce7e22b90b73283985ce32056681 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Thu, 5 May 2016 09:59:03 -0600
Subject: [PATCH] move call iterator to series level

This commit moves the `CallIterator` to wrap the individual series
instead of wrapping a shard. This allows individual points to be
aggregated before being merged.

This will cause a small increase in memory usuage per series but
it shows a 20% decrease in query time when there are a moderate
number of points per series.
---
 influxql/iterator.go       |  4 +++-
 tsdb/engine/tsm1/engine.go | 16 ++++++++++++----
 2 files changed, 15 insertions(+), 5 deletions(-)

diff --git a/influxql/iterator.go b/influxql/iterator.go
index c691d8fd91..1d86cfad9f 100644
--- a/influxql/iterator.go
+++ b/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
diff --git a/tsdb/engine/tsm1/engine.go b/tsdb/engine/tsm1/engine.go
index b3a38bbaa0..19b1d81281 100644
--- a/tsdb/engine/tsm1/engine.go
+++ b/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)
-		if input != nil {
+		// Wrap each series in a call iterator.
+		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)