iterator.gen.go

// Generated by tmpl
// https://github.com/benbjohnson/tmpl
//
// DO NOT EDIT!
// Source: iterator.gen.go.tmpl

package influxql

import (
	"container/heap"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"log"
	"sort"
	"sync"
	"time"

	"github.com/gogo/protobuf/proto"
	"github.com/influxdata/influxdb/influxql/internal"
)

// DefaultStatsInterval is the default value for IteratorEncoder.StatsInterval.
const DefaultStatsInterval = 10 * time.Second

// FloatIterator represents a stream of float points.
type FloatIterator interface {
	Iterator
	Next() *FloatPoint
}

// newFloatIterators converts a slice of Iterator to a slice of FloatIterator.
// Drop and closes any iterator in itrs that is not a FloatIterator and cannot
// be cast to a FloatIterator.
func newFloatIterators(itrs []Iterator) []FloatIterator {
	a := make([]FloatIterator, 0, len(itrs))
	for _, itr := range itrs {
		switch itr := itr.(type) {
		case FloatIterator:
			a = append(a, itr)

		case IntegerIterator:
			a = append(a, &integerFloatCastIterator{input: itr})

		default:
			itr.Close()
		}
	}
	return a
}

// bufFloatIterator represents a buffered FloatIterator.
type bufFloatIterator struct {
	itr FloatIterator
	buf *FloatPoint
}

// newBufFloatIterator returns a buffered FloatIterator.
func newBufFloatIterator(itr FloatIterator) *bufFloatIterator {
	return &bufFloatIterator{itr: itr}
}

// Stats returns statistics from the input iterator.
func (itr *bufFloatIterator) Stats() IteratorStats { return itr.itr.Stats() }

// Close closes the underlying iterator.
func (itr *bufFloatIterator) Close() error { return itr.itr.Close() }

// peek returns the next point without removing it from the iterator.
func (itr *bufFloatIterator) peek() *FloatPoint {
	p := itr.Next()
	itr.unread(p)
	return p
}

// peekTime returns the time of the next point.
// Returns zero time if no more points available.
func (itr *bufFloatIterator) peekTime() int64 {
	p := itr.peek()
	if p == nil {
		return ZeroTime
	}
	return p.Time
}

// Next returns the current buffer, if exists, or calls the underlying iterator.
func (itr *bufFloatIterator) Next() *FloatPoint {
	if itr.buf != nil {
		buf := itr.buf
		itr.buf = nil
		return buf
	}
	return itr.itr.Next()
}

// NextInWindow returns the next value if it is between [startTime, endTime).
// If the next value is outside the range then it is moved to the buffer.
func (itr *bufFloatIterator) NextInWindow(startTime, endTime int64) *FloatPoint {
	v := itr.Next()
	if v == nil {
		return nil
	} else if v.Time < startTime || v.Time >= endTime {
		itr.unread(v)
		return nil
	}
	return v
}

// unread sets v to the buffer. It is read on the next call to Next().
func (itr *bufFloatIterator) unread(v *FloatPoint) { itr.buf = v }

// floatMergeIterator represents an iterator that combines multiple float iterators.
type floatMergeIterator struct {
	inputs []FloatIterator
	heap   *floatMergeHeap
	init   bool

	// Current iterator and window.
	curr   *floatMergeHeapItem
	window struct {
		name      string
		tags      string
		startTime int64
		endTime   int64
	}
}

// newFloatMergeIterator returns a new instance of floatMergeIterator.
func newFloatMergeIterator(inputs []FloatIterator, opt IteratorOptions) *floatMergeIterator {
	itr := &floatMergeIterator{
		inputs: inputs,
		heap: &floatMergeHeap{
			items: make([]*floatMergeHeapItem, 0, len(inputs)),
			opt:   opt,
		},
	}

	// Initialize heap items.
	for _, input := range inputs {
		// Wrap in buffer, ignore any inputs without anymore points.
		bufInput := newBufFloatIterator(input)

		// Append to the heap.
		itr.heap.items = append(itr.heap.items, &floatMergeHeapItem{itr: bufInput})
	}

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *floatMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *floatMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *floatMergeIterator) Next() *FloatPoint {
	// Initialize the heap. This needs to be done lazily on the first call to this iterator
	// so that iterator initialization done through the Select() call returns quickly.
	// Queries can only be interrupted after the Select() call completes so any operations
	// done during iterator creation cannot be interrupted, which is why we do it here
	// instead so an interrupt can happen while initializing the heap.
	if !itr.init {
		items := itr.heap.items
		itr.heap.items = make([]*floatMergeHeapItem, 0, len(items))
		for _, item := range items {
			if item.itr.peek() == nil {
				continue
			}
			itr.heap.items = append(itr.heap.items, item)
		}
		heap.Init(itr.heap)
		itr.init = true
	}

	for {
		// Retrieve the next iterator if we don't have one.
		if itr.curr == nil {
			if len(itr.heap.items) == 0 {
				return nil
			}
			itr.curr = heap.Pop(itr.heap).(*floatMergeHeapItem)

			// Read point and set current window.
			p := itr.curr.itr.Next()
			itr.window.name, itr.window.tags = p.Name, p.Tags.ID()
			itr.window.startTime, itr.window.endTime = itr.heap.opt.Window(p.Time)
			return p
		}

		// Read the next point from the current iterator.
		p := itr.curr.itr.Next()

		// If there are no more points then remove iterator from heap and find next.
		if p == nil {
			itr.curr = nil
			continue
		}

		// Check if the point is inside of our current window.
		inWindow := true
		if itr.window.name != p.Name {
			inWindow = false
		} else if itr.window.tags != p.Tags.ID() {
			inWindow = false
		} else if itr.heap.opt.Ascending && p.Time >= itr.window.endTime {
			inWindow = false
		} else if !itr.heap.opt.Ascending && p.Time < itr.window.startTime {
			inWindow = false
		}

		// If it's outside our window then push iterator back on the heap and find new iterator.
		if !inWindow {
			itr.curr.itr.unread(p)
			heap.Push(itr.heap, itr.curr)
			itr.curr = nil
			continue
		}

		return p
	}
}

// floatMergeHeap represents a heap of floatMergeHeapItems.
// Items are sorted by their next window and then by name/tags.
type floatMergeHeap struct {
	opt   IteratorOptions
	items []*floatMergeHeapItem
}

func (h floatMergeHeap) Len() int      { return len(h.items) }
func (h floatMergeHeap) Swap(i, j int) { h.items[i], h.items[j] = h.items[j], h.items[i] }
func (h floatMergeHeap) Less(i, j int) bool {
	x, y := h.items[i].itr.peek(), h.items[j].itr.peek()

	if h.opt.Ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() < y.Tags.ID()
		}
	} else {
		if x.Name != y.Name {
			return x.Name > y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() > y.Tags.ID()
		}
	}

	xt, _ := h.opt.Window(x.Time)
	yt, _ := h.opt.Window(y.Time)

	if h.opt.Ascending {
		return xt < yt
	}
	return xt > yt
}

func (h *floatMergeHeap) Push(x interface{}) {
	h.items = append(h.items, x.(*floatMergeHeapItem))
}

func (h *floatMergeHeap) Pop() interface{} {
	old := h.items
	n := len(old)
	item := old[n-1]
	h.items = old[0 : n-1]
	return item
}

type floatMergeHeapItem struct {
	itr *bufFloatIterator
}

// floatSortedMergeIterator is an iterator that sorts and merges multiple iterators into one.
type floatSortedMergeIterator struct {
	inputs []FloatIterator
	opt    IteratorOptions
	heap   floatSortedMergeHeap
}

// newFloatSortedMergeIterator returns an instance of floatSortedMergeIterator.
func newFloatSortedMergeIterator(inputs []FloatIterator, opt IteratorOptions) Iterator {
	itr := &floatSortedMergeIterator{
		inputs: inputs,
		heap:   make(floatSortedMergeHeap, 0, len(inputs)),
		opt:    opt,
	}

	// Initialize heap.
	for _, input := range inputs {
		// Read next point.
		p := input.Next()
		if p == nil {
			continue
		}

		// Append to the heap.
		itr.heap = append(itr.heap, &floatSortedMergeHeapItem{point: p, itr: input, ascending: opt.Ascending})
	}
	heap.Init(&itr.heap)

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *floatSortedMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *floatSortedMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next points from the iterator.
func (itr *floatSortedMergeIterator) Next() *FloatPoint { return itr.pop() }

// pop returns the next point from the heap.
// Reads the next point from item's cursor and puts it back on the heap.
func (itr *floatSortedMergeIterator) pop() *FloatPoint {
	if len(itr.heap) == 0 {
		return nil
	}

	// Read the next item from the heap.
	item := heap.Pop(&itr.heap).(*floatSortedMergeHeapItem)

	// Copy the point for return.
	p := item.point.Clone()

	// Read the next item from the cursor. Push back to heap if one exists.
	if item.point = item.itr.Next(); item.point != nil {
		heap.Push(&itr.heap, item)
	}

	return p
}

// floatSortedMergeHeap represents a heap of floatSortedMergeHeapItems.
type floatSortedMergeHeap []*floatSortedMergeHeapItem

func (h floatSortedMergeHeap) Len() int      { return len(h) }
func (h floatSortedMergeHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h floatSortedMergeHeap) Less(i, j int) bool {
	x, y := h[i].point, h[j].point

	if h[i].ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if !x.Tags.Equals(&y.Tags) {
			return x.Tags.ID() < y.Tags.ID()
		}
		return x.Time < y.Time
	}

	if x.Name != y.Name {
		return x.Name > y.Name
	} else if !x.Tags.Equals(&y.Tags) {
		return x.Tags.ID() > y.Tags.ID()
	}
	return x.Time > y.Time
}

func (h *floatSortedMergeHeap) Push(x interface{}) {
	*h = append(*h, x.(*floatSortedMergeHeapItem))
}

func (h *floatSortedMergeHeap) Pop() interface{} {
	old := *h
	n := len(old)
	item := old[n-1]
	*h = old[0 : n-1]
	return item
}

type floatSortedMergeHeapItem struct {
	point     *FloatPoint
	itr       FloatIterator
	ascending bool
}

// floatLimitIterator represents an iterator that limits points per group.
type floatLimitIterator struct {
	input FloatIterator
	opt   IteratorOptions
	n     int

	prev struct {
		name string
		tags Tags
	}
}

// newFloatLimitIterator returns a new instance of floatLimitIterator.
func newFloatLimitIterator(input FloatIterator, opt IteratorOptions) *floatLimitIterator {
	return &floatLimitIterator{
		input: input,
		opt:   opt,
	}
}

// Stats returns stats from the underlying iterator.
func (itr *floatLimitIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the underlying iterators.
func (itr *floatLimitIterator) Close() error { return itr.input.Close() }

// Next returns the next point from the iterator.
func (itr *floatLimitIterator) Next() *FloatPoint {
	for {
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Reset window and counter if a new window is encountered.
		if p.Name != itr.prev.name || !p.Tags.Equals(&itr.prev.tags) {
			itr.prev.name = p.Name
			itr.prev.tags = p.Tags
			itr.n = 0
		}

		// Increment counter.
		itr.n++

		// Read next point if not beyond the offset.
		if itr.n <= itr.opt.Offset {
			continue
		}

		// Read next point if we're beyond the limit.
		if itr.opt.Limit > 0 && (itr.n-itr.opt.Offset) > itr.opt.Limit {
			// If there's no interval, no groups, and a single source then simply exit.
			if itr.opt.Interval.IsZero() && len(itr.opt.Dimensions) == 0 && len(itr.opt.Sources) == 1 {
				return nil
			}
			continue
		}

		return p
	}
}

type floatFillIterator struct {
	input     *bufFloatIterator
	prev      *FloatPoint
	startTime int64
	endTime   int64
	auxFields []interface{}
	done      bool
	opt       IteratorOptions

	window struct {
		name string
		tags Tags
		time int64
	}
}

func newFloatFillIterator(input FloatIterator, expr Expr, opt IteratorOptions) *floatFillIterator {
	if opt.Fill == NullFill {
		if expr, ok := expr.(*Call); ok && expr.Name == "count" {
			opt.Fill = NumberFill
			opt.FillValue = float64(0)
		}
	}

	var startTime, endTime int64
	if opt.Ascending {
		startTime, _ = opt.Window(opt.StartTime)
		_, endTime = opt.Window(opt.EndTime)
	} else {
		_, startTime = opt.Window(opt.EndTime)
		endTime, _ = opt.Window(opt.StartTime)
	}

	var auxFields []interface{}
	if len(opt.Aux) > 0 {
		auxFields = make([]interface{}, len(opt.Aux))
	}

	itr := &floatFillIterator{
		input:     newBufFloatIterator(input),
		startTime: startTime,
		endTime:   endTime,
		auxFields: auxFields,
		opt:       opt,
	}

	p := itr.input.peek()
	if p != nil {
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
	} else {
		itr.window.time = itr.endTime
	}
	return itr
}

func (itr *floatFillIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *floatFillIterator) Close() error         { return itr.input.Close() }

func (itr *floatFillIterator) Next() *FloatPoint {
	p := itr.input.Next()

	// Check if the next point is outside of our window or is nil.
	for p == nil || p.Name != itr.window.name || p.Tags.ID() != itr.window.tags.ID() {
		// If we are inside of an interval, unread the point and continue below to
		// constructing a new point.
		if itr.opt.Ascending {
			if itr.window.time < itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		} else {
			if itr.window.time >= itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		}

		// We are *not* in a current interval. If there is no next point,
		// we are at the end of all intervals.
		if p == nil {
			return nil
		}

		// Set the new interval.
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
		itr.prev = nil
		break
	}

	// Check if the point is our next expected point.
	if p == nil || p.Time > itr.window.time {
		if p != nil {
			itr.input.unread(p)
		}

		p = &FloatPoint{
			Name: itr.window.name,
			Tags: itr.window.tags,
			Time: itr.window.time,
			Aux:  itr.auxFields,
		}

		switch itr.opt.Fill {
		case NullFill:
			p.Nil = true
		case NumberFill:
			p.Value = castToFloat(itr.opt.FillValue)
		case PreviousFill:
			if itr.prev != nil {
				p.Value = itr.prev.Value
				p.Nil = itr.prev.Nil
			} else {
				p.Nil = true
			}
		}
	} else {
		itr.prev = p
	}

	// Advance the expected time. Do not advance to a new window here
	// as there may be lingering points with the same timestamp in the previous
	// window.
	if itr.opt.Ascending {
		itr.window.time = p.Time + int64(itr.opt.Interval.Duration)
	} else {
		itr.window.time = p.Time - int64(itr.opt.Interval.Duration)
	}
	return p
}

// floatIntervalIterator represents a float implementation of IntervalIterator.
type floatIntervalIterator struct {
	input FloatIterator
	opt   IteratorOptions
}

func newFloatIntervalIterator(input FloatIterator, opt IteratorOptions) *floatIntervalIterator {
	return &floatIntervalIterator{input: input, opt: opt}
}

func (itr *floatIntervalIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *floatIntervalIterator) Close() error         { return itr.input.Close() }

func (itr *floatIntervalIterator) Next() *FloatPoint {
	p := itr.input.Next()
	if p == nil {
		return p
	}
	p.Time, _ = itr.opt.Window(p.Time)
	return p
}

// floatInterruptIterator represents a float implementation of InterruptIterator.
type floatInterruptIterator struct {
	input   FloatIterator
	closing <-chan struct{}
	count   int
}

func newFloatInterruptIterator(input FloatIterator, closing <-chan struct{}) *floatInterruptIterator {
	return &floatInterruptIterator{input: input, closing: closing}
}

func (itr *floatInterruptIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *floatInterruptIterator) Close() error         { return itr.input.Close() }

func (itr *floatInterruptIterator) Next() *FloatPoint {
	// Only check if the channel is closed every 256 points. This
	// intentionally checks on both 0 and 256 so that if the iterator
	// has been interrupted before the first point is emitted it will
	// not emit any points.
	if itr.count&0x100 == 0 {
		select {
		case <-itr.closing:
			return nil
		default:
			// Reset iterator count to zero and fall through to emit the next point.
			itr.count = 0
		}
	}

	// Increment the counter for every point read.
	itr.count++
	return itr.input.Next()
}

// floatAuxIterator represents a float implementation of AuxIterator.
type floatAuxIterator struct {
	input      *bufFloatIterator
	output     chan *FloatPoint
	fields     auxIteratorFields
	background bool
}

func newFloatAuxIterator(input FloatIterator, seriesKeys SeriesList, opt IteratorOptions) *floatAuxIterator {
	return &floatAuxIterator{
		input:  newBufFloatIterator(input),
		output: make(chan *FloatPoint, 1),
		fields: newAuxIteratorFields(seriesKeys, opt),
	}
}

func (itr *floatAuxIterator) Background() {
	itr.background = true
	itr.Start()
	go drainIterator(itr)
}

func (itr *floatAuxIterator) Start()                        { go itr.stream() }
func (itr *floatAuxIterator) Stats() IteratorStats          { return itr.input.Stats() }
func (itr *floatAuxIterator) Close() error                  { return itr.input.Close() }
func (itr *floatAuxIterator) Next() *FloatPoint             { return <-itr.output }
func (itr *floatAuxIterator) Iterator(name string) Iterator { return itr.fields.iterator(name) }

func (itr *floatAuxIterator) CreateIterator(opt IteratorOptions) (Iterator, error) {
	expr := opt.Expr
	if expr == nil {
		panic("unable to create an iterator with no expression from an aux iterator")
	}

	switch expr := expr.(type) {
	case *VarRef:
		return itr.Iterator(expr.Val), nil
	default:
		panic(fmt.Sprintf("invalid expression type for an aux iterator: %T", expr))
	}
}

func (itr *floatAuxIterator) FieldDimensions(sources Sources) (fields, dimensions map[string]struct{}, err error) {
	return nil, nil, errors.New("not implemented")
}

func (itr *floatAuxIterator) SeriesKeys(opt IteratorOptions) (SeriesList, error) {
	return nil, errors.New("not implemented")
}

func (itr *floatAuxIterator) ExpandSources(sources Sources) (Sources, error) {
	return nil, errors.New("not implemented")
}

func (itr *floatAuxIterator) stream() {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			break
		}

		// Send point to output and to each field iterator.
		itr.output <- p
		if ok := itr.fields.send(p); !ok && itr.background {
			break
		}
	}

	close(itr.output)
	itr.fields.close()
}

// floatChanIterator represents a new instance of floatChanIterator.
type floatChanIterator struct {
	buf  *FloatPoint
	cond *sync.Cond
	done bool
}

func (itr *floatChanIterator) Stats() IteratorStats { return IteratorStats{} }

func (itr *floatChanIterator) Close() error {
	itr.cond.L.Lock()
	// Mark the channel iterator as done and signal all waiting goroutines to start again.
	itr.done = true
	itr.cond.Broadcast()
	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return nil
}

func (itr *floatChanIterator) setBuf(name string, tags Tags, time int64, value interface{}) bool {
	itr.cond.L.Lock()
	defer itr.cond.L.Unlock()

	// Wait for either the iterator to be done (so we don't have to set the value)
	// or for the buffer to have been read and ready for another write.
	for !itr.done && itr.buf != nil {
		itr.cond.Wait()
	}

	// Do not set the value and return false to signal that the iterator is closed.
	// Do this after the above wait as the above for loop may have exited because
	// the iterator was closed.
	if itr.done {
		return false
	}

	switch v := value.(type) {
	case float64:
		itr.buf = &FloatPoint{Name: name, Tags: tags, Time: time, Value: v}

	case int64:
		itr.buf = &FloatPoint{Name: name, Tags: tags, Time: time, Value: float64(v)}

	default:
		itr.buf = &FloatPoint{Name: name, Tags: tags, Time: time, Nil: true}
	}
	// Signal to all waiting goroutines that a new value is ready to read.
	itr.cond.Signal()
	return true
}

func (itr *floatChanIterator) Next() *FloatPoint {
	itr.cond.L.Lock()

	// Wait until either a value is available in the buffer or
	// the iterator is closed.
	for !itr.done && itr.buf == nil {
		itr.cond.Wait()
	}

	// Always read from the buffer if it exists, even if the iterator
	// is closed. This prevents the last value from being truncated by
	// the parent iterator.
	p := itr.buf
	itr.buf = nil
	itr.cond.Signal()

	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return p
}

// floatReduceFloatIterator executes a reducer for every interval and buffers the result.
type floatReduceFloatIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	points []FloatPoint
}

// Stats returns stats from the input iterator.
func (itr *floatReduceFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatReduceFloatIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *floatReduceFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// floatReduceFloatPoint stores the reduced data for a name/tag combination.
type floatReduceFloatPoint struct {
	Name       string
	Tags       Tags
	Aggregator FloatPointAggregator
	Emitter    FloatPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *floatReduceFloatIterator) reduce() []FloatPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*floatReduceFloatPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]FloatPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// floatStreamFloatIterator
type floatStreamFloatIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	m      map[string]*floatReduceFloatPoint
	points []FloatPoint
}

func newFloatStreamFloatIterator(input FloatIterator, createFn func() (FloatPointAggregator, FloatPointEmitter), opt IteratorOptions) *floatStreamFloatIterator {
	return &floatStreamFloatIterator{
		input:  newBufFloatIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*floatReduceFloatPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *floatStreamFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatStreamFloatIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *floatStreamFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *floatStreamFloatIterator) reduce() []FloatPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// floatExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type floatExprIterator struct {
	left  *bufFloatIterator
	right *bufFloatIterator
	fn    floatExprFunc
}

func (itr *floatExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *floatExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *floatExprIterator) Next() *FloatPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// floatExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type floatExprFunc func(a *FloatPoint, b *FloatPoint) *FloatPoint

// floatReduceIntegerIterator executes a reducer for every interval and buffers the result.
type floatReduceIntegerIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	points []IntegerPoint
}

// Stats returns stats from the input iterator.
func (itr *floatReduceIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatReduceIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *floatReduceIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// floatReduceIntegerPoint stores the reduced data for a name/tag combination.
type floatReduceIntegerPoint struct {
	Name       string
	Tags       Tags
	Aggregator FloatPointAggregator
	Emitter    IntegerPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *floatReduceIntegerIterator) reduce() []IntegerPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*floatReduceIntegerPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]IntegerPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// floatStreamIntegerIterator
type floatStreamIntegerIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	m      map[string]*floatReduceIntegerPoint
	points []IntegerPoint
}

func newFloatStreamIntegerIterator(input FloatIterator, createFn func() (FloatPointAggregator, IntegerPointEmitter), opt IteratorOptions) *floatStreamIntegerIterator {
	return &floatStreamIntegerIterator{
		input:  newBufFloatIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*floatReduceIntegerPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *floatStreamIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatStreamIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *floatStreamIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *floatStreamIntegerIterator) reduce() []IntegerPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// floatIntegerExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type floatIntegerExprIterator struct {
	left  *bufFloatIterator
	right *bufFloatIterator
	fn    floatIntegerExprFunc
}

func (itr *floatIntegerExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *floatIntegerExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *floatIntegerExprIterator) Next() *IntegerPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// floatIntegerExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type floatIntegerExprFunc func(a *FloatPoint, b *FloatPoint) *IntegerPoint

// floatReduceStringIterator executes a reducer for every interval and buffers the result.
type floatReduceStringIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	points []StringPoint
}

// Stats returns stats from the input iterator.
func (itr *floatReduceStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatReduceStringIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *floatReduceStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// floatReduceStringPoint stores the reduced data for a name/tag combination.
type floatReduceStringPoint struct {
	Name       string
	Tags       Tags
	Aggregator FloatPointAggregator
	Emitter    StringPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *floatReduceStringIterator) reduce() []StringPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*floatReduceStringPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]StringPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// floatStreamStringIterator
type floatStreamStringIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	m      map[string]*floatReduceStringPoint
	points []StringPoint
}

func newFloatStreamStringIterator(input FloatIterator, createFn func() (FloatPointAggregator, StringPointEmitter), opt IteratorOptions) *floatStreamStringIterator {
	return &floatStreamStringIterator{
		input:  newBufFloatIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*floatReduceStringPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *floatStreamStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatStreamStringIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *floatStreamStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *floatStreamStringIterator) reduce() []StringPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// floatStringExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type floatStringExprIterator struct {
	left  *bufFloatIterator
	right *bufFloatIterator
	fn    floatStringExprFunc
}

func (itr *floatStringExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *floatStringExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *floatStringExprIterator) Next() *StringPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// floatStringExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type floatStringExprFunc func(a *FloatPoint, b *FloatPoint) *StringPoint

// floatReduceBooleanIterator executes a reducer for every interval and buffers the result.
type floatReduceBooleanIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	points []BooleanPoint
}

// Stats returns stats from the input iterator.
func (itr *floatReduceBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatReduceBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *floatReduceBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// floatReduceBooleanPoint stores the reduced data for a name/tag combination.
type floatReduceBooleanPoint struct {
	Name       string
	Tags       Tags
	Aggregator FloatPointAggregator
	Emitter    BooleanPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *floatReduceBooleanIterator) reduce() []BooleanPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*floatReduceBooleanPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]BooleanPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// floatStreamBooleanIterator
type floatStreamBooleanIterator struct {
	input  *bufFloatIterator
	create func() (FloatPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	m      map[string]*floatReduceBooleanPoint
	points []BooleanPoint
}

func newFloatStreamBooleanIterator(input FloatIterator, createFn func() (FloatPointAggregator, BooleanPointEmitter), opt IteratorOptions) *floatStreamBooleanIterator {
	return &floatStreamBooleanIterator{
		input:  newBufFloatIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*floatReduceBooleanPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *floatStreamBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatStreamBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *floatStreamBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *floatStreamBooleanIterator) reduce() []BooleanPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &floatReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateFloat(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// floatBooleanExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type floatBooleanExprIterator struct {
	left  *bufFloatIterator
	right *bufFloatIterator
	fn    floatBooleanExprFunc
}

func (itr *floatBooleanExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *floatBooleanExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *floatBooleanExprIterator) Next() *BooleanPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// floatBooleanExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type floatBooleanExprFunc func(a *FloatPoint, b *FloatPoint) *BooleanPoint

// floatTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type floatTransformIterator struct {
	input FloatIterator
	fn    floatTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *floatTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *floatTransformIterator) Next() *FloatPoint {
	p := itr.input.Next()
	if p != nil {
		p = itr.fn(p)
	}
	return p
}

// floatTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type floatTransformFunc func(p *FloatPoint) *FloatPoint

// floatBoolTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type floatBoolTransformIterator struct {
	input FloatIterator
	fn    floatBoolTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *floatBoolTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatBoolTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *floatBoolTransformIterator) Next() *BooleanPoint {
	p := itr.input.Next()
	if p != nil {
		return itr.fn(p)
	}
	return nil
}

// floatBoolTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type floatBoolTransformFunc func(p *FloatPoint) *BooleanPoint

// floatDedupeIterator only outputs unique points.
// This differs from the DistinctIterator in that it compares all aux fields too.
// This iterator is relatively inefficient and should only be used on small
// datasets such as meta query results.
type floatDedupeIterator struct {
	input FloatIterator
	m     map[string]struct{} // lookup of points already sent
}

// newFloatDedupeIterator returns a new instance of floatDedupeIterator.
func newFloatDedupeIterator(input FloatIterator) *floatDedupeIterator {
	return &floatDedupeIterator{
		input: input,
		m:     make(map[string]struct{}),
	}
}

// Stats returns stats from the input iterator.
func (itr *floatDedupeIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *floatDedupeIterator) Close() error { return itr.input.Close() }

// Next returns the next unique point from the input iterator.
func (itr *floatDedupeIterator) Next() *FloatPoint {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Serialize to bytes to store in lookup.
		buf, err := proto.Marshal(encodeFloatPoint(p))
		if err != nil {
			log.Println("error marshaling dedupe point:", err)
			continue
		}

		// If the point has already been output then move to the next point.
		if _, ok := itr.m[string(buf)]; ok {
			continue
		}

		// Otherwise mark it as emitted and return point.
		itr.m[string(buf)] = struct{}{}
		return p
	}
}

// floatReaderIterator represents an iterator that streams from a reader.
type floatReaderIterator struct {
	r     io.Reader
	dec   *FloatPointDecoder
	first *FloatPoint
}

// newFloatReaderIterator returns a new instance of floatReaderIterator.
func newFloatReaderIterator(r io.Reader, first *FloatPoint, stats IteratorStats) *floatReaderIterator {
	dec := NewFloatPointDecoder(r)
	dec.stats = stats

	return &floatReaderIterator{
		r:     r,
		dec:   dec,
		first: first,
	}
}

// Stats returns stats about points processed.
func (itr *floatReaderIterator) Stats() IteratorStats { return itr.dec.stats }

// Close closes the underlying reader, if applicable.
func (itr *floatReaderIterator) Close() error {
	if r, ok := itr.r.(io.ReadCloser); ok {
		return r.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *floatReaderIterator) Next() *FloatPoint {
	// Send first point if it hasn't been sent yet.
	if itr.first != nil {
		p := itr.first
		itr.first = nil
		return p
	}

	// OPTIMIZE(benbjohnson): Reuse point on iterator.

	// Unmarshal next point.
	p := &FloatPoint{}
	if err := itr.dec.DecodeFloatPoint(p); err == io.EOF {
		return nil
	} else if err != nil {
		log.Printf("error reading iterator point: %s", err)
		return nil
	}
	return p
}

// IntegerIterator represents a stream of integer points.
type IntegerIterator interface {
	Iterator
	Next() *IntegerPoint
}

// newIntegerIterators converts a slice of Iterator to a slice of IntegerIterator.
// Drop and closes any iterator in itrs that is not a IntegerIterator and cannot
// be cast to a IntegerIterator.
func newIntegerIterators(itrs []Iterator) []IntegerIterator {
	a := make([]IntegerIterator, 0, len(itrs))
	for _, itr := range itrs {
		switch itr := itr.(type) {
		case IntegerIterator:
			a = append(a, itr)

		default:
			itr.Close()
		}
	}
	return a
}

// bufIntegerIterator represents a buffered IntegerIterator.
type bufIntegerIterator struct {
	itr IntegerIterator
	buf *IntegerPoint
}

// newBufIntegerIterator returns a buffered IntegerIterator.
func newBufIntegerIterator(itr IntegerIterator) *bufIntegerIterator {
	return &bufIntegerIterator{itr: itr}
}

// Stats returns statistics from the input iterator.
func (itr *bufIntegerIterator) Stats() IteratorStats { return itr.itr.Stats() }

// Close closes the underlying iterator.
func (itr *bufIntegerIterator) Close() error { return itr.itr.Close() }

// peek returns the next point without removing it from the iterator.
func (itr *bufIntegerIterator) peek() *IntegerPoint {
	p := itr.Next()
	itr.unread(p)
	return p
}

// peekTime returns the time of the next point.
// Returns zero time if no more points available.
func (itr *bufIntegerIterator) peekTime() int64 {
	p := itr.peek()
	if p == nil {
		return ZeroTime
	}
	return p.Time
}

// Next returns the current buffer, if exists, or calls the underlying iterator.
func (itr *bufIntegerIterator) Next() *IntegerPoint {
	if itr.buf != nil {
		buf := itr.buf
		itr.buf = nil
		return buf
	}
	return itr.itr.Next()
}

// NextInWindow returns the next value if it is between [startTime, endTime).
// If the next value is outside the range then it is moved to the buffer.
func (itr *bufIntegerIterator) NextInWindow(startTime, endTime int64) *IntegerPoint {
	v := itr.Next()
	if v == nil {
		return nil
	} else if v.Time < startTime || v.Time >= endTime {
		itr.unread(v)
		return nil
	}
	return v
}

// unread sets v to the buffer. It is read on the next call to Next().
func (itr *bufIntegerIterator) unread(v *IntegerPoint) { itr.buf = v }

// integerMergeIterator represents an iterator that combines multiple integer iterators.
type integerMergeIterator struct {
	inputs []IntegerIterator
	heap   *integerMergeHeap
	init   bool

	// Current iterator and window.
	curr   *integerMergeHeapItem
	window struct {
		name      string
		tags      string
		startTime int64
		endTime   int64
	}
}

// newIntegerMergeIterator returns a new instance of integerMergeIterator.
func newIntegerMergeIterator(inputs []IntegerIterator, opt IteratorOptions) *integerMergeIterator {
	itr := &integerMergeIterator{
		inputs: inputs,
		heap: &integerMergeHeap{
			items: make([]*integerMergeHeapItem, 0, len(inputs)),
			opt:   opt,
		},
	}

	// Initialize heap items.
	for _, input := range inputs {
		// Wrap in buffer, ignore any inputs without anymore points.
		bufInput := newBufIntegerIterator(input)

		// Append to the heap.
		itr.heap.items = append(itr.heap.items, &integerMergeHeapItem{itr: bufInput})
	}

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *integerMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *integerMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *integerMergeIterator) Next() *IntegerPoint {
	// Initialize the heap. This needs to be done lazily on the first call to this iterator
	// so that iterator initialization done through the Select() call returns quickly.
	// Queries can only be interrupted after the Select() call completes so any operations
	// done during iterator creation cannot be interrupted, which is why we do it here
	// instead so an interrupt can happen while initializing the heap.
	if !itr.init {
		items := itr.heap.items
		itr.heap.items = make([]*integerMergeHeapItem, 0, len(items))
		for _, item := range items {
			if item.itr.peek() == nil {
				continue
			}
			itr.heap.items = append(itr.heap.items, item)
		}
		heap.Init(itr.heap)
		itr.init = true
	}

	for {
		// Retrieve the next iterator if we don't have one.
		if itr.curr == nil {
			if len(itr.heap.items) == 0 {
				return nil
			}
			itr.curr = heap.Pop(itr.heap).(*integerMergeHeapItem)

			// Read point and set current window.
			p := itr.curr.itr.Next()
			itr.window.name, itr.window.tags = p.Name, p.Tags.ID()
			itr.window.startTime, itr.window.endTime = itr.heap.opt.Window(p.Time)
			return p
		}

		// Read the next point from the current iterator.
		p := itr.curr.itr.Next()

		// If there are no more points then remove iterator from heap and find next.
		if p == nil {
			itr.curr = nil
			continue
		}

		// Check if the point is inside of our current window.
		inWindow := true
		if itr.window.name != p.Name {
			inWindow = false
		} else if itr.window.tags != p.Tags.ID() {
			inWindow = false
		} else if itr.heap.opt.Ascending && p.Time >= itr.window.endTime {
			inWindow = false
		} else if !itr.heap.opt.Ascending && p.Time < itr.window.startTime {
			inWindow = false
		}

		// If it's outside our window then push iterator back on the heap and find new iterator.
		if !inWindow {
			itr.curr.itr.unread(p)
			heap.Push(itr.heap, itr.curr)
			itr.curr = nil
			continue
		}

		return p
	}
}

// integerMergeHeap represents a heap of integerMergeHeapItems.
// Items are sorted by their next window and then by name/tags.
type integerMergeHeap struct {
	opt   IteratorOptions
	items []*integerMergeHeapItem
}

func (h integerMergeHeap) Len() int      { return len(h.items) }
func (h integerMergeHeap) Swap(i, j int) { h.items[i], h.items[j] = h.items[j], h.items[i] }
func (h integerMergeHeap) Less(i, j int) bool {
	x, y := h.items[i].itr.peek(), h.items[j].itr.peek()

	if h.opt.Ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() < y.Tags.ID()
		}
	} else {
		if x.Name != y.Name {
			return x.Name > y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() > y.Tags.ID()
		}
	}

	xt, _ := h.opt.Window(x.Time)
	yt, _ := h.opt.Window(y.Time)

	if h.opt.Ascending {
		return xt < yt
	}
	return xt > yt
}

func (h *integerMergeHeap) Push(x interface{}) {
	h.items = append(h.items, x.(*integerMergeHeapItem))
}

func (h *integerMergeHeap) Pop() interface{} {
	old := h.items
	n := len(old)
	item := old[n-1]
	h.items = old[0 : n-1]
	return item
}

type integerMergeHeapItem struct {
	itr *bufIntegerIterator
}

// integerSortedMergeIterator is an iterator that sorts and merges multiple iterators into one.
type integerSortedMergeIterator struct {
	inputs []IntegerIterator
	opt    IteratorOptions
	heap   integerSortedMergeHeap
}

// newIntegerSortedMergeIterator returns an instance of integerSortedMergeIterator.
func newIntegerSortedMergeIterator(inputs []IntegerIterator, opt IteratorOptions) Iterator {
	itr := &integerSortedMergeIterator{
		inputs: inputs,
		heap:   make(integerSortedMergeHeap, 0, len(inputs)),
		opt:    opt,
	}

	// Initialize heap.
	for _, input := range inputs {
		// Read next point.
		p := input.Next()
		if p == nil {
			continue
		}

		// Append to the heap.
		itr.heap = append(itr.heap, &integerSortedMergeHeapItem{point: p, itr: input, ascending: opt.Ascending})
	}
	heap.Init(&itr.heap)

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *integerSortedMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *integerSortedMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next points from the iterator.
func (itr *integerSortedMergeIterator) Next() *IntegerPoint { return itr.pop() }

// pop returns the next point from the heap.
// Reads the next point from item's cursor and puts it back on the heap.
func (itr *integerSortedMergeIterator) pop() *IntegerPoint {
	if len(itr.heap) == 0 {
		return nil
	}

	// Read the next item from the heap.
	item := heap.Pop(&itr.heap).(*integerSortedMergeHeapItem)

	// Copy the point for return.
	p := item.point.Clone()

	// Read the next item from the cursor. Push back to heap if one exists.
	if item.point = item.itr.Next(); item.point != nil {
		heap.Push(&itr.heap, item)
	}

	return p
}

// integerSortedMergeHeap represents a heap of integerSortedMergeHeapItems.
type integerSortedMergeHeap []*integerSortedMergeHeapItem

func (h integerSortedMergeHeap) Len() int      { return len(h) }
func (h integerSortedMergeHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h integerSortedMergeHeap) Less(i, j int) bool {
	x, y := h[i].point, h[j].point

	if h[i].ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if !x.Tags.Equals(&y.Tags) {
			return x.Tags.ID() < y.Tags.ID()
		}
		return x.Time < y.Time
	}

	if x.Name != y.Name {
		return x.Name > y.Name
	} else if !x.Tags.Equals(&y.Tags) {
		return x.Tags.ID() > y.Tags.ID()
	}
	return x.Time > y.Time
}

func (h *integerSortedMergeHeap) Push(x interface{}) {
	*h = append(*h, x.(*integerSortedMergeHeapItem))
}

func (h *integerSortedMergeHeap) Pop() interface{} {
	old := *h
	n := len(old)
	item := old[n-1]
	*h = old[0 : n-1]
	return item
}

type integerSortedMergeHeapItem struct {
	point     *IntegerPoint
	itr       IntegerIterator
	ascending bool
}

// integerLimitIterator represents an iterator that limits points per group.
type integerLimitIterator struct {
	input IntegerIterator
	opt   IteratorOptions
	n     int

	prev struct {
		name string
		tags Tags
	}
}

// newIntegerLimitIterator returns a new instance of integerLimitIterator.
func newIntegerLimitIterator(input IntegerIterator, opt IteratorOptions) *integerLimitIterator {
	return &integerLimitIterator{
		input: input,
		opt:   opt,
	}
}

// Stats returns stats from the underlying iterator.
func (itr *integerLimitIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the underlying iterators.
func (itr *integerLimitIterator) Close() error { return itr.input.Close() }

// Next returns the next point from the iterator.
func (itr *integerLimitIterator) Next() *IntegerPoint {
	for {
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Reset window and counter if a new window is encountered.
		if p.Name != itr.prev.name || !p.Tags.Equals(&itr.prev.tags) {
			itr.prev.name = p.Name
			itr.prev.tags = p.Tags
			itr.n = 0
		}

		// Increment counter.
		itr.n++

		// Read next point if not beyond the offset.
		if itr.n <= itr.opt.Offset {
			continue
		}

		// Read next point if we're beyond the limit.
		if itr.opt.Limit > 0 && (itr.n-itr.opt.Offset) > itr.opt.Limit {
			// If there's no interval, no groups, and a single source then simply exit.
			if itr.opt.Interval.IsZero() && len(itr.opt.Dimensions) == 0 && len(itr.opt.Sources) == 1 {
				return nil
			}
			continue
		}

		return p
	}
}

type integerFillIterator struct {
	input     *bufIntegerIterator
	prev      *IntegerPoint
	startTime int64
	endTime   int64
	auxFields []interface{}
	done      bool
	opt       IteratorOptions

	window struct {
		name string
		tags Tags
		time int64
	}
}

func newIntegerFillIterator(input IntegerIterator, expr Expr, opt IteratorOptions) *integerFillIterator {
	if opt.Fill == NullFill {
		if expr, ok := expr.(*Call); ok && expr.Name == "count" {
			opt.Fill = NumberFill
			opt.FillValue = int64(0)
		}
	}

	var startTime, endTime int64
	if opt.Ascending {
		startTime, _ = opt.Window(opt.StartTime)
		_, endTime = opt.Window(opt.EndTime)
	} else {
		_, startTime = opt.Window(opt.EndTime)
		endTime, _ = opt.Window(opt.StartTime)
	}

	var auxFields []interface{}
	if len(opt.Aux) > 0 {
		auxFields = make([]interface{}, len(opt.Aux))
	}

	itr := &integerFillIterator{
		input:     newBufIntegerIterator(input),
		startTime: startTime,
		endTime:   endTime,
		auxFields: auxFields,
		opt:       opt,
	}

	p := itr.input.peek()
	if p != nil {
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
	} else {
		itr.window.time = itr.endTime
	}
	return itr
}

func (itr *integerFillIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *integerFillIterator) Close() error         { return itr.input.Close() }

func (itr *integerFillIterator) Next() *IntegerPoint {
	p := itr.input.Next()

	// Check if the next point is outside of our window or is nil.
	for p == nil || p.Name != itr.window.name || p.Tags.ID() != itr.window.tags.ID() {
		// If we are inside of an interval, unread the point and continue below to
		// constructing a new point.
		if itr.opt.Ascending {
			if itr.window.time < itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		} else {
			if itr.window.time >= itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		}

		// We are *not* in a current interval. If there is no next point,
		// we are at the end of all intervals.
		if p == nil {
			return nil
		}

		// Set the new interval.
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
		itr.prev = nil
		break
	}

	// Check if the point is our next expected point.
	if p == nil || p.Time > itr.window.time {
		if p != nil {
			itr.input.unread(p)
		}

		p = &IntegerPoint{
			Name: itr.window.name,
			Tags: itr.window.tags,
			Time: itr.window.time,
			Aux:  itr.auxFields,
		}

		switch itr.opt.Fill {
		case NullFill:
			p.Nil = true
		case NumberFill:
			p.Value = castToInteger(itr.opt.FillValue)
		case PreviousFill:
			if itr.prev != nil {
				p.Value = itr.prev.Value
				p.Nil = itr.prev.Nil
			} else {
				p.Nil = true
			}
		}
	} else {
		itr.prev = p
	}

	// Advance the expected time. Do not advance to a new window here
	// as there may be lingering points with the same timestamp in the previous
	// window.
	if itr.opt.Ascending {
		itr.window.time = p.Time + int64(itr.opt.Interval.Duration)
	} else {
		itr.window.time = p.Time - int64(itr.opt.Interval.Duration)
	}
	return p
}

// integerIntervalIterator represents a integer implementation of IntervalIterator.
type integerIntervalIterator struct {
	input IntegerIterator
	opt   IteratorOptions
}

func newIntegerIntervalIterator(input IntegerIterator, opt IteratorOptions) *integerIntervalIterator {
	return &integerIntervalIterator{input: input, opt: opt}
}

func (itr *integerIntervalIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *integerIntervalIterator) Close() error         { return itr.input.Close() }

func (itr *integerIntervalIterator) Next() *IntegerPoint {
	p := itr.input.Next()
	if p == nil {
		return p
	}
	p.Time, _ = itr.opt.Window(p.Time)
	return p
}

// integerInterruptIterator represents a integer implementation of InterruptIterator.
type integerInterruptIterator struct {
	input   IntegerIterator
	closing <-chan struct{}
	count   int
}

func newIntegerInterruptIterator(input IntegerIterator, closing <-chan struct{}) *integerInterruptIterator {
	return &integerInterruptIterator{input: input, closing: closing}
}

func (itr *integerInterruptIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *integerInterruptIterator) Close() error         { return itr.input.Close() }

func (itr *integerInterruptIterator) Next() *IntegerPoint {
	// Only check if the channel is closed every 256 points. This
	// intentionally checks on both 0 and 256 so that if the iterator
	// has been interrupted before the first point is emitted it will
	// not emit any points.
	if itr.count&0x100 == 0 {
		select {
		case <-itr.closing:
			return nil
		default:
			// Reset iterator count to zero and fall through to emit the next point.
			itr.count = 0
		}
	}

	// Increment the counter for every point read.
	itr.count++
	return itr.input.Next()
}

// integerAuxIterator represents a integer implementation of AuxIterator.
type integerAuxIterator struct {
	input      *bufIntegerIterator
	output     chan *IntegerPoint
	fields     auxIteratorFields
	background bool
}

func newIntegerAuxIterator(input IntegerIterator, seriesKeys SeriesList, opt IteratorOptions) *integerAuxIterator {
	return &integerAuxIterator{
		input:  newBufIntegerIterator(input),
		output: make(chan *IntegerPoint, 1),
		fields: newAuxIteratorFields(seriesKeys, opt),
	}
}

func (itr *integerAuxIterator) Background() {
	itr.background = true
	itr.Start()
	go drainIterator(itr)
}

func (itr *integerAuxIterator) Start()                        { go itr.stream() }
func (itr *integerAuxIterator) Stats() IteratorStats          { return itr.input.Stats() }
func (itr *integerAuxIterator) Close() error                  { return itr.input.Close() }
func (itr *integerAuxIterator) Next() *IntegerPoint           { return <-itr.output }
func (itr *integerAuxIterator) Iterator(name string) Iterator { return itr.fields.iterator(name) }

func (itr *integerAuxIterator) CreateIterator(opt IteratorOptions) (Iterator, error) {
	expr := opt.Expr
	if expr == nil {
		panic("unable to create an iterator with no expression from an aux iterator")
	}

	switch expr := expr.(type) {
	case *VarRef:
		return itr.Iterator(expr.Val), nil
	default:
		panic(fmt.Sprintf("invalid expression type for an aux iterator: %T", expr))
	}
}

func (itr *integerAuxIterator) FieldDimensions(sources Sources) (fields, dimensions map[string]struct{}, err error) {
	return nil, nil, errors.New("not implemented")
}

func (itr *integerAuxIterator) SeriesKeys(opt IteratorOptions) (SeriesList, error) {
	return nil, errors.New("not implemented")
}

func (itr *integerAuxIterator) ExpandSources(sources Sources) (Sources, error) {
	return nil, errors.New("not implemented")
}

func (itr *integerAuxIterator) stream() {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			break
		}

		// Send point to output and to each field iterator.
		itr.output <- p
		if ok := itr.fields.send(p); !ok && itr.background {
			break
		}
	}

	close(itr.output)
	itr.fields.close()
}

// integerChanIterator represents a new instance of integerChanIterator.
type integerChanIterator struct {
	buf  *IntegerPoint
	cond *sync.Cond
	done bool
}

func (itr *integerChanIterator) Stats() IteratorStats { return IteratorStats{} }

func (itr *integerChanIterator) Close() error {
	itr.cond.L.Lock()
	// Mark the channel iterator as done and signal all waiting goroutines to start again.
	itr.done = true
	itr.cond.Broadcast()
	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return nil
}

func (itr *integerChanIterator) setBuf(name string, tags Tags, time int64, value interface{}) bool {
	itr.cond.L.Lock()
	defer itr.cond.L.Unlock()

	// Wait for either the iterator to be done (so we don't have to set the value)
	// or for the buffer to have been read and ready for another write.
	for !itr.done && itr.buf != nil {
		itr.cond.Wait()
	}

	// Do not set the value and return false to signal that the iterator is closed.
	// Do this after the above wait as the above for loop may have exited because
	// the iterator was closed.
	if itr.done {
		return false
	}

	switch v := value.(type) {
	case int64:
		itr.buf = &IntegerPoint{Name: name, Tags: tags, Time: time, Value: v}

	default:
		itr.buf = &IntegerPoint{Name: name, Tags: tags, Time: time, Nil: true}
	}
	// Signal to all waiting goroutines that a new value is ready to read.
	itr.cond.Signal()
	return true
}

func (itr *integerChanIterator) Next() *IntegerPoint {
	itr.cond.L.Lock()

	// Wait until either a value is available in the buffer or
	// the iterator is closed.
	for !itr.done && itr.buf == nil {
		itr.cond.Wait()
	}

	// Always read from the buffer if it exists, even if the iterator
	// is closed. This prevents the last value from being truncated by
	// the parent iterator.
	p := itr.buf
	itr.buf = nil
	itr.cond.Signal()

	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return p
}

// integerReduceFloatIterator executes a reducer for every interval and buffers the result.
type integerReduceFloatIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	points []FloatPoint
}

// Stats returns stats from the input iterator.
func (itr *integerReduceFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerReduceFloatIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *integerReduceFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// integerReduceFloatPoint stores the reduced data for a name/tag combination.
type integerReduceFloatPoint struct {
	Name       string
	Tags       Tags
	Aggregator IntegerPointAggregator
	Emitter    FloatPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *integerReduceFloatIterator) reduce() []FloatPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*integerReduceFloatPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]FloatPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// integerStreamFloatIterator
type integerStreamFloatIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	m      map[string]*integerReduceFloatPoint
	points []FloatPoint
}

func newIntegerStreamFloatIterator(input IntegerIterator, createFn func() (IntegerPointAggregator, FloatPointEmitter), opt IteratorOptions) *integerStreamFloatIterator {
	return &integerStreamFloatIterator{
		input:  newBufIntegerIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*integerReduceFloatPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *integerStreamFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerStreamFloatIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *integerStreamFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *integerStreamFloatIterator) reduce() []FloatPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// integerFloatExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type integerFloatExprIterator struct {
	left  *bufIntegerIterator
	right *bufIntegerIterator
	fn    integerFloatExprFunc
}

func (itr *integerFloatExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *integerFloatExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *integerFloatExprIterator) Next() *FloatPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// integerFloatExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type integerFloatExprFunc func(a *IntegerPoint, b *IntegerPoint) *FloatPoint

// integerReduceIntegerIterator executes a reducer for every interval and buffers the result.
type integerReduceIntegerIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	points []IntegerPoint
}

// Stats returns stats from the input iterator.
func (itr *integerReduceIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerReduceIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *integerReduceIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// integerReduceIntegerPoint stores the reduced data for a name/tag combination.
type integerReduceIntegerPoint struct {
	Name       string
	Tags       Tags
	Aggregator IntegerPointAggregator
	Emitter    IntegerPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *integerReduceIntegerIterator) reduce() []IntegerPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*integerReduceIntegerPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]IntegerPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// integerStreamIntegerIterator
type integerStreamIntegerIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	m      map[string]*integerReduceIntegerPoint
	points []IntegerPoint
}

func newIntegerStreamIntegerIterator(input IntegerIterator, createFn func() (IntegerPointAggregator, IntegerPointEmitter), opt IteratorOptions) *integerStreamIntegerIterator {
	return &integerStreamIntegerIterator{
		input:  newBufIntegerIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*integerReduceIntegerPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *integerStreamIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerStreamIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *integerStreamIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *integerStreamIntegerIterator) reduce() []IntegerPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// integerExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type integerExprIterator struct {
	left  *bufIntegerIterator
	right *bufIntegerIterator
	fn    integerExprFunc
}

func (itr *integerExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *integerExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *integerExprIterator) Next() *IntegerPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// integerExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type integerExprFunc func(a *IntegerPoint, b *IntegerPoint) *IntegerPoint

// integerReduceStringIterator executes a reducer for every interval and buffers the result.
type integerReduceStringIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	points []StringPoint
}

// Stats returns stats from the input iterator.
func (itr *integerReduceStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerReduceStringIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *integerReduceStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// integerReduceStringPoint stores the reduced data for a name/tag combination.
type integerReduceStringPoint struct {
	Name       string
	Tags       Tags
	Aggregator IntegerPointAggregator
	Emitter    StringPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *integerReduceStringIterator) reduce() []StringPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*integerReduceStringPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]StringPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// integerStreamStringIterator
type integerStreamStringIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	m      map[string]*integerReduceStringPoint
	points []StringPoint
}

func newIntegerStreamStringIterator(input IntegerIterator, createFn func() (IntegerPointAggregator, StringPointEmitter), opt IteratorOptions) *integerStreamStringIterator {
	return &integerStreamStringIterator{
		input:  newBufIntegerIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*integerReduceStringPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *integerStreamStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerStreamStringIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *integerStreamStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *integerStreamStringIterator) reduce() []StringPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// integerStringExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type integerStringExprIterator struct {
	left  *bufIntegerIterator
	right *bufIntegerIterator
	fn    integerStringExprFunc
}

func (itr *integerStringExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *integerStringExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *integerStringExprIterator) Next() *StringPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// integerStringExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type integerStringExprFunc func(a *IntegerPoint, b *IntegerPoint) *StringPoint

// integerReduceBooleanIterator executes a reducer for every interval and buffers the result.
type integerReduceBooleanIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	points []BooleanPoint
}

// Stats returns stats from the input iterator.
func (itr *integerReduceBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerReduceBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *integerReduceBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// integerReduceBooleanPoint stores the reduced data for a name/tag combination.
type integerReduceBooleanPoint struct {
	Name       string
	Tags       Tags
	Aggregator IntegerPointAggregator
	Emitter    BooleanPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *integerReduceBooleanIterator) reduce() []BooleanPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*integerReduceBooleanPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]BooleanPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// integerStreamBooleanIterator
type integerStreamBooleanIterator struct {
	input  *bufIntegerIterator
	create func() (IntegerPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	m      map[string]*integerReduceBooleanPoint
	points []BooleanPoint
}

func newIntegerStreamBooleanIterator(input IntegerIterator, createFn func() (IntegerPointAggregator, BooleanPointEmitter), opt IteratorOptions) *integerStreamBooleanIterator {
	return &integerStreamBooleanIterator{
		input:  newBufIntegerIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*integerReduceBooleanPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *integerStreamBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerStreamBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *integerStreamBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *integerStreamBooleanIterator) reduce() []BooleanPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &integerReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateInteger(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// integerBooleanExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type integerBooleanExprIterator struct {
	left  *bufIntegerIterator
	right *bufIntegerIterator
	fn    integerBooleanExprFunc
}

func (itr *integerBooleanExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *integerBooleanExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *integerBooleanExprIterator) Next() *BooleanPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// integerBooleanExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type integerBooleanExprFunc func(a *IntegerPoint, b *IntegerPoint) *BooleanPoint

// integerTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type integerTransformIterator struct {
	input IntegerIterator
	fn    integerTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *integerTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *integerTransformIterator) Next() *IntegerPoint {
	p := itr.input.Next()
	if p != nil {
		p = itr.fn(p)
	}
	return p
}

// integerTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type integerTransformFunc func(p *IntegerPoint) *IntegerPoint

// integerBoolTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type integerBoolTransformIterator struct {
	input IntegerIterator
	fn    integerBoolTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *integerBoolTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerBoolTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *integerBoolTransformIterator) Next() *BooleanPoint {
	p := itr.input.Next()
	if p != nil {
		return itr.fn(p)
	}
	return nil
}

// integerBoolTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type integerBoolTransformFunc func(p *IntegerPoint) *BooleanPoint

// integerDedupeIterator only outputs unique points.
// This differs from the DistinctIterator in that it compares all aux fields too.
// This iterator is relatively inefficient and should only be used on small
// datasets such as meta query results.
type integerDedupeIterator struct {
	input IntegerIterator
	m     map[string]struct{} // lookup of points already sent
}

// newIntegerDedupeIterator returns a new instance of integerDedupeIterator.
func newIntegerDedupeIterator(input IntegerIterator) *integerDedupeIterator {
	return &integerDedupeIterator{
		input: input,
		m:     make(map[string]struct{}),
	}
}

// Stats returns stats from the input iterator.
func (itr *integerDedupeIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *integerDedupeIterator) Close() error { return itr.input.Close() }

// Next returns the next unique point from the input iterator.
func (itr *integerDedupeIterator) Next() *IntegerPoint {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Serialize to bytes to store in lookup.
		buf, err := proto.Marshal(encodeIntegerPoint(p))
		if err != nil {
			log.Println("error marshaling dedupe point:", err)
			continue
		}

		// If the point has already been output then move to the next point.
		if _, ok := itr.m[string(buf)]; ok {
			continue
		}

		// Otherwise mark it as emitted and return point.
		itr.m[string(buf)] = struct{}{}
		return p
	}
}

// integerReaderIterator represents an iterator that streams from a reader.
type integerReaderIterator struct {
	r     io.Reader
	dec   *IntegerPointDecoder
	first *IntegerPoint
}

// newIntegerReaderIterator returns a new instance of integerReaderIterator.
func newIntegerReaderIterator(r io.Reader, first *IntegerPoint, stats IteratorStats) *integerReaderIterator {
	dec := NewIntegerPointDecoder(r)
	dec.stats = stats

	return &integerReaderIterator{
		r:     r,
		dec:   dec,
		first: first,
	}
}

// Stats returns stats about points processed.
func (itr *integerReaderIterator) Stats() IteratorStats { return itr.dec.stats }

// Close closes the underlying reader, if applicable.
func (itr *integerReaderIterator) Close() error {
	if r, ok := itr.r.(io.ReadCloser); ok {
		return r.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *integerReaderIterator) Next() *IntegerPoint {
	// Send first point if it hasn't been sent yet.
	if itr.first != nil {
		p := itr.first
		itr.first = nil
		return p
	}

	// OPTIMIZE(benbjohnson): Reuse point on iterator.

	// Unmarshal next point.
	p := &IntegerPoint{}
	if err := itr.dec.DecodeIntegerPoint(p); err == io.EOF {
		return nil
	} else if err != nil {
		log.Printf("error reading iterator point: %s", err)
		return nil
	}
	return p
}

// StringIterator represents a stream of string points.
type StringIterator interface {
	Iterator
	Next() *StringPoint
}

// newStringIterators converts a slice of Iterator to a slice of StringIterator.
// Drop and closes any iterator in itrs that is not a StringIterator and cannot
// be cast to a StringIterator.
func newStringIterators(itrs []Iterator) []StringIterator {
	a := make([]StringIterator, 0, len(itrs))
	for _, itr := range itrs {
		switch itr := itr.(type) {
		case StringIterator:
			a = append(a, itr)

		default:
			itr.Close()
		}
	}
	return a
}

// bufStringIterator represents a buffered StringIterator.
type bufStringIterator struct {
	itr StringIterator
	buf *StringPoint
}

// newBufStringIterator returns a buffered StringIterator.
func newBufStringIterator(itr StringIterator) *bufStringIterator {
	return &bufStringIterator{itr: itr}
}

// Stats returns statistics from the input iterator.
func (itr *bufStringIterator) Stats() IteratorStats { return itr.itr.Stats() }

// Close closes the underlying iterator.
func (itr *bufStringIterator) Close() error { return itr.itr.Close() }

// peek returns the next point without removing it from the iterator.
func (itr *bufStringIterator) peek() *StringPoint {
	p := itr.Next()
	itr.unread(p)
	return p
}

// peekTime returns the time of the next point.
// Returns zero time if no more points available.
func (itr *bufStringIterator) peekTime() int64 {
	p := itr.peek()
	if p == nil {
		return ZeroTime
	}
	return p.Time
}

// Next returns the current buffer, if exists, or calls the underlying iterator.
func (itr *bufStringIterator) Next() *StringPoint {
	if itr.buf != nil {
		buf := itr.buf
		itr.buf = nil
		return buf
	}
	return itr.itr.Next()
}

// NextInWindow returns the next value if it is between [startTime, endTime).
// If the next value is outside the range then it is moved to the buffer.
func (itr *bufStringIterator) NextInWindow(startTime, endTime int64) *StringPoint {
	v := itr.Next()
	if v == nil {
		return nil
	} else if v.Time < startTime || v.Time >= endTime {
		itr.unread(v)
		return nil
	}
	return v
}

// unread sets v to the buffer. It is read on the next call to Next().
func (itr *bufStringIterator) unread(v *StringPoint) { itr.buf = v }

// stringMergeIterator represents an iterator that combines multiple string iterators.
type stringMergeIterator struct {
	inputs []StringIterator
	heap   *stringMergeHeap
	init   bool

	// Current iterator and window.
	curr   *stringMergeHeapItem
	window struct {
		name      string
		tags      string
		startTime int64
		endTime   int64
	}
}

// newStringMergeIterator returns a new instance of stringMergeIterator.
func newStringMergeIterator(inputs []StringIterator, opt IteratorOptions) *stringMergeIterator {
	itr := &stringMergeIterator{
		inputs: inputs,
		heap: &stringMergeHeap{
			items: make([]*stringMergeHeapItem, 0, len(inputs)),
			opt:   opt,
		},
	}

	// Initialize heap items.
	for _, input := range inputs {
		// Wrap in buffer, ignore any inputs without anymore points.
		bufInput := newBufStringIterator(input)

		// Append to the heap.
		itr.heap.items = append(itr.heap.items, &stringMergeHeapItem{itr: bufInput})
	}

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *stringMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *stringMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *stringMergeIterator) Next() *StringPoint {
	// Initialize the heap. This needs to be done lazily on the first call to this iterator
	// so that iterator initialization done through the Select() call returns quickly.
	// Queries can only be interrupted after the Select() call completes so any operations
	// done during iterator creation cannot be interrupted, which is why we do it here
	// instead so an interrupt can happen while initializing the heap.
	if !itr.init {
		items := itr.heap.items
		itr.heap.items = make([]*stringMergeHeapItem, 0, len(items))
		for _, item := range items {
			if item.itr.peek() == nil {
				continue
			}
			itr.heap.items = append(itr.heap.items, item)
		}
		heap.Init(itr.heap)
		itr.init = true
	}

	for {
		// Retrieve the next iterator if we don't have one.
		if itr.curr == nil {
			if len(itr.heap.items) == 0 {
				return nil
			}
			itr.curr = heap.Pop(itr.heap).(*stringMergeHeapItem)

			// Read point and set current window.
			p := itr.curr.itr.Next()
			itr.window.name, itr.window.tags = p.Name, p.Tags.ID()
			itr.window.startTime, itr.window.endTime = itr.heap.opt.Window(p.Time)
			return p
		}

		// Read the next point from the current iterator.
		p := itr.curr.itr.Next()

		// If there are no more points then remove iterator from heap and find next.
		if p == nil {
			itr.curr = nil
			continue
		}

		// Check if the point is inside of our current window.
		inWindow := true
		if itr.window.name != p.Name {
			inWindow = false
		} else if itr.window.tags != p.Tags.ID() {
			inWindow = false
		} else if itr.heap.opt.Ascending && p.Time >= itr.window.endTime {
			inWindow = false
		} else if !itr.heap.opt.Ascending && p.Time < itr.window.startTime {
			inWindow = false
		}

		// If it's outside our window then push iterator back on the heap and find new iterator.
		if !inWindow {
			itr.curr.itr.unread(p)
			heap.Push(itr.heap, itr.curr)
			itr.curr = nil
			continue
		}

		return p
	}
}

// stringMergeHeap represents a heap of stringMergeHeapItems.
// Items are sorted by their next window and then by name/tags.
type stringMergeHeap struct {
	opt   IteratorOptions
	items []*stringMergeHeapItem
}

func (h stringMergeHeap) Len() int      { return len(h.items) }
func (h stringMergeHeap) Swap(i, j int) { h.items[i], h.items[j] = h.items[j], h.items[i] }
func (h stringMergeHeap) Less(i, j int) bool {
	x, y := h.items[i].itr.peek(), h.items[j].itr.peek()

	if h.opt.Ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() < y.Tags.ID()
		}
	} else {
		if x.Name != y.Name {
			return x.Name > y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() > y.Tags.ID()
		}
	}

	xt, _ := h.opt.Window(x.Time)
	yt, _ := h.opt.Window(y.Time)

	if h.opt.Ascending {
		return xt < yt
	}
	return xt > yt
}

func (h *stringMergeHeap) Push(x interface{}) {
	h.items = append(h.items, x.(*stringMergeHeapItem))
}

func (h *stringMergeHeap) Pop() interface{} {
	old := h.items
	n := len(old)
	item := old[n-1]
	h.items = old[0 : n-1]
	return item
}

type stringMergeHeapItem struct {
	itr *bufStringIterator
}

// stringSortedMergeIterator is an iterator that sorts and merges multiple iterators into one.
type stringSortedMergeIterator struct {
	inputs []StringIterator
	opt    IteratorOptions
	heap   stringSortedMergeHeap
}

// newStringSortedMergeIterator returns an instance of stringSortedMergeIterator.
func newStringSortedMergeIterator(inputs []StringIterator, opt IteratorOptions) Iterator {
	itr := &stringSortedMergeIterator{
		inputs: inputs,
		heap:   make(stringSortedMergeHeap, 0, len(inputs)),
		opt:    opt,
	}

	// Initialize heap.
	for _, input := range inputs {
		// Read next point.
		p := input.Next()
		if p == nil {
			continue
		}

		// Append to the heap.
		itr.heap = append(itr.heap, &stringSortedMergeHeapItem{point: p, itr: input, ascending: opt.Ascending})
	}
	heap.Init(&itr.heap)

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *stringSortedMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *stringSortedMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next points from the iterator.
func (itr *stringSortedMergeIterator) Next() *StringPoint { return itr.pop() }

// pop returns the next point from the heap.
// Reads the next point from item's cursor and puts it back on the heap.
func (itr *stringSortedMergeIterator) pop() *StringPoint {
	if len(itr.heap) == 0 {
		return nil
	}

	// Read the next item from the heap.
	item := heap.Pop(&itr.heap).(*stringSortedMergeHeapItem)

	// Copy the point for return.
	p := item.point.Clone()

	// Read the next item from the cursor. Push back to heap if one exists.
	if item.point = item.itr.Next(); item.point != nil {
		heap.Push(&itr.heap, item)
	}

	return p
}

// stringSortedMergeHeap represents a heap of stringSortedMergeHeapItems.
type stringSortedMergeHeap []*stringSortedMergeHeapItem

func (h stringSortedMergeHeap) Len() int      { return len(h) }
func (h stringSortedMergeHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h stringSortedMergeHeap) Less(i, j int) bool {
	x, y := h[i].point, h[j].point

	if h[i].ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if !x.Tags.Equals(&y.Tags) {
			return x.Tags.ID() < y.Tags.ID()
		}
		return x.Time < y.Time
	}

	if x.Name != y.Name {
		return x.Name > y.Name
	} else if !x.Tags.Equals(&y.Tags) {
		return x.Tags.ID() > y.Tags.ID()
	}
	return x.Time > y.Time
}

func (h *stringSortedMergeHeap) Push(x interface{}) {
	*h = append(*h, x.(*stringSortedMergeHeapItem))
}

func (h *stringSortedMergeHeap) Pop() interface{} {
	old := *h
	n := len(old)
	item := old[n-1]
	*h = old[0 : n-1]
	return item
}

type stringSortedMergeHeapItem struct {
	point     *StringPoint
	itr       StringIterator
	ascending bool
}

// stringLimitIterator represents an iterator that limits points per group.
type stringLimitIterator struct {
	input StringIterator
	opt   IteratorOptions
	n     int

	prev struct {
		name string
		tags Tags
	}
}

// newStringLimitIterator returns a new instance of stringLimitIterator.
func newStringLimitIterator(input StringIterator, opt IteratorOptions) *stringLimitIterator {
	return &stringLimitIterator{
		input: input,
		opt:   opt,
	}
}

// Stats returns stats from the underlying iterator.
func (itr *stringLimitIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the underlying iterators.
func (itr *stringLimitIterator) Close() error { return itr.input.Close() }

// Next returns the next point from the iterator.
func (itr *stringLimitIterator) Next() *StringPoint {
	for {
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Reset window and counter if a new window is encountered.
		if p.Name != itr.prev.name || !p.Tags.Equals(&itr.prev.tags) {
			itr.prev.name = p.Name
			itr.prev.tags = p.Tags
			itr.n = 0
		}

		// Increment counter.
		itr.n++

		// Read next point if not beyond the offset.
		if itr.n <= itr.opt.Offset {
			continue
		}

		// Read next point if we're beyond the limit.
		if itr.opt.Limit > 0 && (itr.n-itr.opt.Offset) > itr.opt.Limit {
			// If there's no interval, no groups, and a single source then simply exit.
			if itr.opt.Interval.IsZero() && len(itr.opt.Dimensions) == 0 && len(itr.opt.Sources) == 1 {
				return nil
			}
			continue
		}

		return p
	}
}

type stringFillIterator struct {
	input     *bufStringIterator
	prev      *StringPoint
	startTime int64
	endTime   int64
	auxFields []interface{}
	done      bool
	opt       IteratorOptions

	window struct {
		name string
		tags Tags
		time int64
	}
}

func newStringFillIterator(input StringIterator, expr Expr, opt IteratorOptions) *stringFillIterator {
	if opt.Fill == NullFill {
		if expr, ok := expr.(*Call); ok && expr.Name == "count" {
			opt.Fill = NumberFill
			opt.FillValue = ""
		}
	}

	var startTime, endTime int64
	if opt.Ascending {
		startTime, _ = opt.Window(opt.StartTime)
		_, endTime = opt.Window(opt.EndTime)
	} else {
		_, startTime = opt.Window(opt.EndTime)
		endTime, _ = opt.Window(opt.StartTime)
	}

	var auxFields []interface{}
	if len(opt.Aux) > 0 {
		auxFields = make([]interface{}, len(opt.Aux))
	}

	itr := &stringFillIterator{
		input:     newBufStringIterator(input),
		startTime: startTime,
		endTime:   endTime,
		auxFields: auxFields,
		opt:       opt,
	}

	p := itr.input.peek()
	if p != nil {
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
	} else {
		itr.window.time = itr.endTime
	}
	return itr
}

func (itr *stringFillIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *stringFillIterator) Close() error         { return itr.input.Close() }

func (itr *stringFillIterator) Next() *StringPoint {
	p := itr.input.Next()

	// Check if the next point is outside of our window or is nil.
	for p == nil || p.Name != itr.window.name || p.Tags.ID() != itr.window.tags.ID() {
		// If we are inside of an interval, unread the point and continue below to
		// constructing a new point.
		if itr.opt.Ascending {
			if itr.window.time < itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		} else {
			if itr.window.time >= itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		}

		// We are *not* in a current interval. If there is no next point,
		// we are at the end of all intervals.
		if p == nil {
			return nil
		}

		// Set the new interval.
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
		itr.prev = nil
		break
	}

	// Check if the point is our next expected point.
	if p == nil || p.Time > itr.window.time {
		if p != nil {
			itr.input.unread(p)
		}

		p = &StringPoint{
			Name: itr.window.name,
			Tags: itr.window.tags,
			Time: itr.window.time,
			Aux:  itr.auxFields,
		}

		switch itr.opt.Fill {
		case NullFill:
			p.Nil = true
		case NumberFill:
			p.Value = castToString(itr.opt.FillValue)
		case PreviousFill:
			if itr.prev != nil {
				p.Value = itr.prev.Value
				p.Nil = itr.prev.Nil
			} else {
				p.Nil = true
			}
		}
	} else {
		itr.prev = p
	}

	// Advance the expected time. Do not advance to a new window here
	// as there may be lingering points with the same timestamp in the previous
	// window.
	if itr.opt.Ascending {
		itr.window.time = p.Time + int64(itr.opt.Interval.Duration)
	} else {
		itr.window.time = p.Time - int64(itr.opt.Interval.Duration)
	}
	return p
}

// stringIntervalIterator represents a string implementation of IntervalIterator.
type stringIntervalIterator struct {
	input StringIterator
	opt   IteratorOptions
}

func newStringIntervalIterator(input StringIterator, opt IteratorOptions) *stringIntervalIterator {
	return &stringIntervalIterator{input: input, opt: opt}
}

func (itr *stringIntervalIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *stringIntervalIterator) Close() error         { return itr.input.Close() }

func (itr *stringIntervalIterator) Next() *StringPoint {
	p := itr.input.Next()
	if p == nil {
		return p
	}
	p.Time, _ = itr.opt.Window(p.Time)
	return p
}

// stringInterruptIterator represents a string implementation of InterruptIterator.
type stringInterruptIterator struct {
	input   StringIterator
	closing <-chan struct{}
	count   int
}

func newStringInterruptIterator(input StringIterator, closing <-chan struct{}) *stringInterruptIterator {
	return &stringInterruptIterator{input: input, closing: closing}
}

func (itr *stringInterruptIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *stringInterruptIterator) Close() error         { return itr.input.Close() }

func (itr *stringInterruptIterator) Next() *StringPoint {
	// Only check if the channel is closed every 256 points. This
	// intentionally checks on both 0 and 256 so that if the iterator
	// has been interrupted before the first point is emitted it will
	// not emit any points.
	if itr.count&0x100 == 0 {
		select {
		case <-itr.closing:
			return nil
		default:
			// Reset iterator count to zero and fall through to emit the next point.
			itr.count = 0
		}
	}

	// Increment the counter for every point read.
	itr.count++
	return itr.input.Next()
}

// stringAuxIterator represents a string implementation of AuxIterator.
type stringAuxIterator struct {
	input      *bufStringIterator
	output     chan *StringPoint
	fields     auxIteratorFields
	background bool
}

func newStringAuxIterator(input StringIterator, seriesKeys SeriesList, opt IteratorOptions) *stringAuxIterator {
	return &stringAuxIterator{
		input:  newBufStringIterator(input),
		output: make(chan *StringPoint, 1),
		fields: newAuxIteratorFields(seriesKeys, opt),
	}
}

func (itr *stringAuxIterator) Background() {
	itr.background = true
	itr.Start()
	go drainIterator(itr)
}

func (itr *stringAuxIterator) Start()                        { go itr.stream() }
func (itr *stringAuxIterator) Stats() IteratorStats          { return itr.input.Stats() }
func (itr *stringAuxIterator) Close() error                  { return itr.input.Close() }
func (itr *stringAuxIterator) Next() *StringPoint            { return <-itr.output }
func (itr *stringAuxIterator) Iterator(name string) Iterator { return itr.fields.iterator(name) }

func (itr *stringAuxIterator) CreateIterator(opt IteratorOptions) (Iterator, error) {
	expr := opt.Expr
	if expr == nil {
		panic("unable to create an iterator with no expression from an aux iterator")
	}

	switch expr := expr.(type) {
	case *VarRef:
		return itr.Iterator(expr.Val), nil
	default:
		panic(fmt.Sprintf("invalid expression type for an aux iterator: %T", expr))
	}
}

func (itr *stringAuxIterator) FieldDimensions(sources Sources) (fields, dimensions map[string]struct{}, err error) {
	return nil, nil, errors.New("not implemented")
}

func (itr *stringAuxIterator) SeriesKeys(opt IteratorOptions) (SeriesList, error) {
	return nil, errors.New("not implemented")
}

func (itr *stringAuxIterator) ExpandSources(sources Sources) (Sources, error) {
	return nil, errors.New("not implemented")
}

func (itr *stringAuxIterator) stream() {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			break
		}

		// Send point to output and to each field iterator.
		itr.output <- p
		if ok := itr.fields.send(p); !ok && itr.background {
			break
		}
	}

	close(itr.output)
	itr.fields.close()
}

// stringChanIterator represents a new instance of stringChanIterator.
type stringChanIterator struct {
	buf  *StringPoint
	cond *sync.Cond
	done bool
}

func (itr *stringChanIterator) Stats() IteratorStats { return IteratorStats{} }

func (itr *stringChanIterator) Close() error {
	itr.cond.L.Lock()
	// Mark the channel iterator as done and signal all waiting goroutines to start again.
	itr.done = true
	itr.cond.Broadcast()
	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return nil
}

func (itr *stringChanIterator) setBuf(name string, tags Tags, time int64, value interface{}) bool {
	itr.cond.L.Lock()
	defer itr.cond.L.Unlock()

	// Wait for either the iterator to be done (so we don't have to set the value)
	// or for the buffer to have been read and ready for another write.
	for !itr.done && itr.buf != nil {
		itr.cond.Wait()
	}

	// Do not set the value and return false to signal that the iterator is closed.
	// Do this after the above wait as the above for loop may have exited because
	// the iterator was closed.
	if itr.done {
		return false
	}

	switch v := value.(type) {
	case string:
		itr.buf = &StringPoint{Name: name, Tags: tags, Time: time, Value: v}

	default:
		itr.buf = &StringPoint{Name: name, Tags: tags, Time: time, Nil: true}
	}
	// Signal to all waiting goroutines that a new value is ready to read.
	itr.cond.Signal()
	return true
}

func (itr *stringChanIterator) Next() *StringPoint {
	itr.cond.L.Lock()

	// Wait until either a value is available in the buffer or
	// the iterator is closed.
	for !itr.done && itr.buf == nil {
		itr.cond.Wait()
	}

	// Always read from the buffer if it exists, even if the iterator
	// is closed. This prevents the last value from being truncated by
	// the parent iterator.
	p := itr.buf
	itr.buf = nil
	itr.cond.Signal()

	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return p
}

// stringReduceFloatIterator executes a reducer for every interval and buffers the result.
type stringReduceFloatIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	points []FloatPoint
}

// Stats returns stats from the input iterator.
func (itr *stringReduceFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringReduceFloatIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *stringReduceFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// stringReduceFloatPoint stores the reduced data for a name/tag combination.
type stringReduceFloatPoint struct {
	Name       string
	Tags       Tags
	Aggregator StringPointAggregator
	Emitter    FloatPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *stringReduceFloatIterator) reduce() []FloatPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*stringReduceFloatPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]FloatPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// stringStreamFloatIterator
type stringStreamFloatIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	m      map[string]*stringReduceFloatPoint
	points []FloatPoint
}

func newStringStreamFloatIterator(input StringIterator, createFn func() (StringPointAggregator, FloatPointEmitter), opt IteratorOptions) *stringStreamFloatIterator {
	return &stringStreamFloatIterator{
		input:  newBufStringIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*stringReduceFloatPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *stringStreamFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringStreamFloatIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *stringStreamFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *stringStreamFloatIterator) reduce() []FloatPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// stringFloatExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type stringFloatExprIterator struct {
	left  *bufStringIterator
	right *bufStringIterator
	fn    stringFloatExprFunc
}

func (itr *stringFloatExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *stringFloatExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *stringFloatExprIterator) Next() *FloatPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// stringFloatExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type stringFloatExprFunc func(a *StringPoint, b *StringPoint) *FloatPoint

// stringReduceIntegerIterator executes a reducer for every interval and buffers the result.
type stringReduceIntegerIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	points []IntegerPoint
}

// Stats returns stats from the input iterator.
func (itr *stringReduceIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringReduceIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *stringReduceIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// stringReduceIntegerPoint stores the reduced data for a name/tag combination.
type stringReduceIntegerPoint struct {
	Name       string
	Tags       Tags
	Aggregator StringPointAggregator
	Emitter    IntegerPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *stringReduceIntegerIterator) reduce() []IntegerPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*stringReduceIntegerPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]IntegerPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// stringStreamIntegerIterator
type stringStreamIntegerIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	m      map[string]*stringReduceIntegerPoint
	points []IntegerPoint
}

func newStringStreamIntegerIterator(input StringIterator, createFn func() (StringPointAggregator, IntegerPointEmitter), opt IteratorOptions) *stringStreamIntegerIterator {
	return &stringStreamIntegerIterator{
		input:  newBufStringIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*stringReduceIntegerPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *stringStreamIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringStreamIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *stringStreamIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *stringStreamIntegerIterator) reduce() []IntegerPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// stringIntegerExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type stringIntegerExprIterator struct {
	left  *bufStringIterator
	right *bufStringIterator
	fn    stringIntegerExprFunc
}

func (itr *stringIntegerExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *stringIntegerExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *stringIntegerExprIterator) Next() *IntegerPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// stringIntegerExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type stringIntegerExprFunc func(a *StringPoint, b *StringPoint) *IntegerPoint

// stringReduceStringIterator executes a reducer for every interval and buffers the result.
type stringReduceStringIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	points []StringPoint
}

// Stats returns stats from the input iterator.
func (itr *stringReduceStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringReduceStringIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *stringReduceStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// stringReduceStringPoint stores the reduced data for a name/tag combination.
type stringReduceStringPoint struct {
	Name       string
	Tags       Tags
	Aggregator StringPointAggregator
	Emitter    StringPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *stringReduceStringIterator) reduce() []StringPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*stringReduceStringPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]StringPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// stringStreamStringIterator
type stringStreamStringIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	m      map[string]*stringReduceStringPoint
	points []StringPoint
}

func newStringStreamStringIterator(input StringIterator, createFn func() (StringPointAggregator, StringPointEmitter), opt IteratorOptions) *stringStreamStringIterator {
	return &stringStreamStringIterator{
		input:  newBufStringIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*stringReduceStringPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *stringStreamStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringStreamStringIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *stringStreamStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *stringStreamStringIterator) reduce() []StringPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// stringExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type stringExprIterator struct {
	left  *bufStringIterator
	right *bufStringIterator
	fn    stringExprFunc
}

func (itr *stringExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *stringExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *stringExprIterator) Next() *StringPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// stringExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type stringExprFunc func(a *StringPoint, b *StringPoint) *StringPoint

// stringReduceBooleanIterator executes a reducer for every interval and buffers the result.
type stringReduceBooleanIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	points []BooleanPoint
}

// Stats returns stats from the input iterator.
func (itr *stringReduceBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringReduceBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *stringReduceBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// stringReduceBooleanPoint stores the reduced data for a name/tag combination.
type stringReduceBooleanPoint struct {
	Name       string
	Tags       Tags
	Aggregator StringPointAggregator
	Emitter    BooleanPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *stringReduceBooleanIterator) reduce() []BooleanPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*stringReduceBooleanPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]BooleanPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// stringStreamBooleanIterator
type stringStreamBooleanIterator struct {
	input  *bufStringIterator
	create func() (StringPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	m      map[string]*stringReduceBooleanPoint
	points []BooleanPoint
}

func newStringStreamBooleanIterator(input StringIterator, createFn func() (StringPointAggregator, BooleanPointEmitter), opt IteratorOptions) *stringStreamBooleanIterator {
	return &stringStreamBooleanIterator{
		input:  newBufStringIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*stringReduceBooleanPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *stringStreamBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringStreamBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *stringStreamBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *stringStreamBooleanIterator) reduce() []BooleanPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &stringReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateString(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// stringBooleanExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type stringBooleanExprIterator struct {
	left  *bufStringIterator
	right *bufStringIterator
	fn    stringBooleanExprFunc
}

func (itr *stringBooleanExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *stringBooleanExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *stringBooleanExprIterator) Next() *BooleanPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// stringBooleanExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type stringBooleanExprFunc func(a *StringPoint, b *StringPoint) *BooleanPoint

// stringTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type stringTransformIterator struct {
	input StringIterator
	fn    stringTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *stringTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *stringTransformIterator) Next() *StringPoint {
	p := itr.input.Next()
	if p != nil {
		p = itr.fn(p)
	}
	return p
}

// stringTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type stringTransformFunc func(p *StringPoint) *StringPoint

// stringBoolTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type stringBoolTransformIterator struct {
	input StringIterator
	fn    stringBoolTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *stringBoolTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringBoolTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *stringBoolTransformIterator) Next() *BooleanPoint {
	p := itr.input.Next()
	if p != nil {
		return itr.fn(p)
	}
	return nil
}

// stringBoolTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type stringBoolTransformFunc func(p *StringPoint) *BooleanPoint

// stringDedupeIterator only outputs unique points.
// This differs from the DistinctIterator in that it compares all aux fields too.
// This iterator is relatively inefficient and should only be used on small
// datasets such as meta query results.
type stringDedupeIterator struct {
	input StringIterator
	m     map[string]struct{} // lookup of points already sent
}

// newStringDedupeIterator returns a new instance of stringDedupeIterator.
func newStringDedupeIterator(input StringIterator) *stringDedupeIterator {
	return &stringDedupeIterator{
		input: input,
		m:     make(map[string]struct{}),
	}
}

// Stats returns stats from the input iterator.
func (itr *stringDedupeIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *stringDedupeIterator) Close() error { return itr.input.Close() }

// Next returns the next unique point from the input iterator.
func (itr *stringDedupeIterator) Next() *StringPoint {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Serialize to bytes to store in lookup.
		buf, err := proto.Marshal(encodeStringPoint(p))
		if err != nil {
			log.Println("error marshaling dedupe point:", err)
			continue
		}

		// If the point has already been output then move to the next point.
		if _, ok := itr.m[string(buf)]; ok {
			continue
		}

		// Otherwise mark it as emitted and return point.
		itr.m[string(buf)] = struct{}{}
		return p
	}
}

// stringReaderIterator represents an iterator that streams from a reader.
type stringReaderIterator struct {
	r     io.Reader
	dec   *StringPointDecoder
	first *StringPoint
}

// newStringReaderIterator returns a new instance of stringReaderIterator.
func newStringReaderIterator(r io.Reader, first *StringPoint, stats IteratorStats) *stringReaderIterator {
	dec := NewStringPointDecoder(r)
	dec.stats = stats

	return &stringReaderIterator{
		r:     r,
		dec:   dec,
		first: first,
	}
}

// Stats returns stats about points processed.
func (itr *stringReaderIterator) Stats() IteratorStats { return itr.dec.stats }

// Close closes the underlying reader, if applicable.
func (itr *stringReaderIterator) Close() error {
	if r, ok := itr.r.(io.ReadCloser); ok {
		return r.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *stringReaderIterator) Next() *StringPoint {
	// Send first point if it hasn't been sent yet.
	if itr.first != nil {
		p := itr.first
		itr.first = nil
		return p
	}

	// OPTIMIZE(benbjohnson): Reuse point on iterator.

	// Unmarshal next point.
	p := &StringPoint{}
	if err := itr.dec.DecodeStringPoint(p); err == io.EOF {
		return nil
	} else if err != nil {
		log.Printf("error reading iterator point: %s", err)
		return nil
	}
	return p
}

// BooleanIterator represents a stream of boolean points.
type BooleanIterator interface {
	Iterator
	Next() *BooleanPoint
}

// newBooleanIterators converts a slice of Iterator to a slice of BooleanIterator.
// Drop and closes any iterator in itrs that is not a BooleanIterator and cannot
// be cast to a BooleanIterator.
func newBooleanIterators(itrs []Iterator) []BooleanIterator {
	a := make([]BooleanIterator, 0, len(itrs))
	for _, itr := range itrs {
		switch itr := itr.(type) {
		case BooleanIterator:
			a = append(a, itr)

		default:
			itr.Close()
		}
	}
	return a
}

// bufBooleanIterator represents a buffered BooleanIterator.
type bufBooleanIterator struct {
	itr BooleanIterator
	buf *BooleanPoint
}

// newBufBooleanIterator returns a buffered BooleanIterator.
func newBufBooleanIterator(itr BooleanIterator) *bufBooleanIterator {
	return &bufBooleanIterator{itr: itr}
}

// Stats returns statistics from the input iterator.
func (itr *bufBooleanIterator) Stats() IteratorStats { return itr.itr.Stats() }

// Close closes the underlying iterator.
func (itr *bufBooleanIterator) Close() error { return itr.itr.Close() }

// peek returns the next point without removing it from the iterator.
func (itr *bufBooleanIterator) peek() *BooleanPoint {
	p := itr.Next()
	itr.unread(p)
	return p
}

// peekTime returns the time of the next point.
// Returns zero time if no more points available.
func (itr *bufBooleanIterator) peekTime() int64 {
	p := itr.peek()
	if p == nil {
		return ZeroTime
	}
	return p.Time
}

// Next returns the current buffer, if exists, or calls the underlying iterator.
func (itr *bufBooleanIterator) Next() *BooleanPoint {
	if itr.buf != nil {
		buf := itr.buf
		itr.buf = nil
		return buf
	}
	return itr.itr.Next()
}

// NextInWindow returns the next value if it is between [startTime, endTime).
// If the next value is outside the range then it is moved to the buffer.
func (itr *bufBooleanIterator) NextInWindow(startTime, endTime int64) *BooleanPoint {
	v := itr.Next()
	if v == nil {
		return nil
	} else if v.Time < startTime || v.Time >= endTime {
		itr.unread(v)
		return nil
	}
	return v
}

// unread sets v to the buffer. It is read on the next call to Next().
func (itr *bufBooleanIterator) unread(v *BooleanPoint) { itr.buf = v }

// booleanMergeIterator represents an iterator that combines multiple boolean iterators.
type booleanMergeIterator struct {
	inputs []BooleanIterator
	heap   *booleanMergeHeap
	init   bool

	// Current iterator and window.
	curr   *booleanMergeHeapItem
	window struct {
		name      string
		tags      string
		startTime int64
		endTime   int64
	}
}

// newBooleanMergeIterator returns a new instance of booleanMergeIterator.
func newBooleanMergeIterator(inputs []BooleanIterator, opt IteratorOptions) *booleanMergeIterator {
	itr := &booleanMergeIterator{
		inputs: inputs,
		heap: &booleanMergeHeap{
			items: make([]*booleanMergeHeapItem, 0, len(inputs)),
			opt:   opt,
		},
	}

	// Initialize heap items.
	for _, input := range inputs {
		// Wrap in buffer, ignore any inputs without anymore points.
		bufInput := newBufBooleanIterator(input)

		// Append to the heap.
		itr.heap.items = append(itr.heap.items, &booleanMergeHeapItem{itr: bufInput})
	}

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *booleanMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *booleanMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *booleanMergeIterator) Next() *BooleanPoint {
	// Initialize the heap. This needs to be done lazily on the first call to this iterator
	// so that iterator initialization done through the Select() call returns quickly.
	// Queries can only be interrupted after the Select() call completes so any operations
	// done during iterator creation cannot be interrupted, which is why we do it here
	// instead so an interrupt can happen while initializing the heap.
	if !itr.init {
		items := itr.heap.items
		itr.heap.items = make([]*booleanMergeHeapItem, 0, len(items))
		for _, item := range items {
			if item.itr.peek() == nil {
				continue
			}
			itr.heap.items = append(itr.heap.items, item)
		}
		heap.Init(itr.heap)
		itr.init = true
	}

	for {
		// Retrieve the next iterator if we don't have one.
		if itr.curr == nil {
			if len(itr.heap.items) == 0 {
				return nil
			}
			itr.curr = heap.Pop(itr.heap).(*booleanMergeHeapItem)

			// Read point and set current window.
			p := itr.curr.itr.Next()
			itr.window.name, itr.window.tags = p.Name, p.Tags.ID()
			itr.window.startTime, itr.window.endTime = itr.heap.opt.Window(p.Time)
			return p
		}

		// Read the next point from the current iterator.
		p := itr.curr.itr.Next()

		// If there are no more points then remove iterator from heap and find next.
		if p == nil {
			itr.curr = nil
			continue
		}

		// Check if the point is inside of our current window.
		inWindow := true
		if itr.window.name != p.Name {
			inWindow = false
		} else if itr.window.tags != p.Tags.ID() {
			inWindow = false
		} else if itr.heap.opt.Ascending && p.Time >= itr.window.endTime {
			inWindow = false
		} else if !itr.heap.opt.Ascending && p.Time < itr.window.startTime {
			inWindow = false
		}

		// If it's outside our window then push iterator back on the heap and find new iterator.
		if !inWindow {
			itr.curr.itr.unread(p)
			heap.Push(itr.heap, itr.curr)
			itr.curr = nil
			continue
		}

		return p
	}
}

// booleanMergeHeap represents a heap of booleanMergeHeapItems.
// Items are sorted by their next window and then by name/tags.
type booleanMergeHeap struct {
	opt   IteratorOptions
	items []*booleanMergeHeapItem
}

func (h booleanMergeHeap) Len() int      { return len(h.items) }
func (h booleanMergeHeap) Swap(i, j int) { h.items[i], h.items[j] = h.items[j], h.items[i] }
func (h booleanMergeHeap) Less(i, j int) bool {
	x, y := h.items[i].itr.peek(), h.items[j].itr.peek()

	if h.opt.Ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() < y.Tags.ID()
		}
	} else {
		if x.Name != y.Name {
			return x.Name > y.Name
		} else if x.Tags.ID() != y.Tags.ID() {
			return x.Tags.ID() > y.Tags.ID()
		}
	}

	xt, _ := h.opt.Window(x.Time)
	yt, _ := h.opt.Window(y.Time)

	if h.opt.Ascending {
		return xt < yt
	}
	return xt > yt
}

func (h *booleanMergeHeap) Push(x interface{}) {
	h.items = append(h.items, x.(*booleanMergeHeapItem))
}

func (h *booleanMergeHeap) Pop() interface{} {
	old := h.items
	n := len(old)
	item := old[n-1]
	h.items = old[0 : n-1]
	return item
}

type booleanMergeHeapItem struct {
	itr *bufBooleanIterator
}

// booleanSortedMergeIterator is an iterator that sorts and merges multiple iterators into one.
type booleanSortedMergeIterator struct {
	inputs []BooleanIterator
	opt    IteratorOptions
	heap   booleanSortedMergeHeap
}

// newBooleanSortedMergeIterator returns an instance of booleanSortedMergeIterator.
func newBooleanSortedMergeIterator(inputs []BooleanIterator, opt IteratorOptions) Iterator {
	itr := &booleanSortedMergeIterator{
		inputs: inputs,
		heap:   make(booleanSortedMergeHeap, 0, len(inputs)),
		opt:    opt,
	}

	// Initialize heap.
	for _, input := range inputs {
		// Read next point.
		p := input.Next()
		if p == nil {
			continue
		}

		// Append to the heap.
		itr.heap = append(itr.heap, &booleanSortedMergeHeapItem{point: p, itr: input, ascending: opt.Ascending})
	}
	heap.Init(&itr.heap)

	return itr
}

// Stats returns an aggregation of stats from the underlying iterators.
func (itr *booleanSortedMergeIterator) Stats() IteratorStats {
	var stats IteratorStats
	for _, input := range itr.inputs {
		stats.Add(input.Stats())
	}
	return stats
}

// Close closes the underlying iterators.
func (itr *booleanSortedMergeIterator) Close() error {
	for _, input := range itr.inputs {
		input.Close()
	}
	return nil
}

// Next returns the next points from the iterator.
func (itr *booleanSortedMergeIterator) Next() *BooleanPoint { return itr.pop() }

// pop returns the next point from the heap.
// Reads the next point from item's cursor and puts it back on the heap.
func (itr *booleanSortedMergeIterator) pop() *BooleanPoint {
	if len(itr.heap) == 0 {
		return nil
	}

	// Read the next item from the heap.
	item := heap.Pop(&itr.heap).(*booleanSortedMergeHeapItem)

	// Copy the point for return.
	p := item.point.Clone()

	// Read the next item from the cursor. Push back to heap if one exists.
	if item.point = item.itr.Next(); item.point != nil {
		heap.Push(&itr.heap, item)
	}

	return p
}

// booleanSortedMergeHeap represents a heap of booleanSortedMergeHeapItems.
type booleanSortedMergeHeap []*booleanSortedMergeHeapItem

func (h booleanSortedMergeHeap) Len() int      { return len(h) }
func (h booleanSortedMergeHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h booleanSortedMergeHeap) Less(i, j int) bool {
	x, y := h[i].point, h[j].point

	if h[i].ascending {
		if x.Name != y.Name {
			return x.Name < y.Name
		} else if !x.Tags.Equals(&y.Tags) {
			return x.Tags.ID() < y.Tags.ID()
		}
		return x.Time < y.Time
	}

	if x.Name != y.Name {
		return x.Name > y.Name
	} else if !x.Tags.Equals(&y.Tags) {
		return x.Tags.ID() > y.Tags.ID()
	}
	return x.Time > y.Time
}

func (h *booleanSortedMergeHeap) Push(x interface{}) {
	*h = append(*h, x.(*booleanSortedMergeHeapItem))
}

func (h *booleanSortedMergeHeap) Pop() interface{} {
	old := *h
	n := len(old)
	item := old[n-1]
	*h = old[0 : n-1]
	return item
}

type booleanSortedMergeHeapItem struct {
	point     *BooleanPoint
	itr       BooleanIterator
	ascending bool
}

// booleanLimitIterator represents an iterator that limits points per group.
type booleanLimitIterator struct {
	input BooleanIterator
	opt   IteratorOptions
	n     int

	prev struct {
		name string
		tags Tags
	}
}

// newBooleanLimitIterator returns a new instance of booleanLimitIterator.
func newBooleanLimitIterator(input BooleanIterator, opt IteratorOptions) *booleanLimitIterator {
	return &booleanLimitIterator{
		input: input,
		opt:   opt,
	}
}

// Stats returns stats from the underlying iterator.
func (itr *booleanLimitIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the underlying iterators.
func (itr *booleanLimitIterator) Close() error { return itr.input.Close() }

// Next returns the next point from the iterator.
func (itr *booleanLimitIterator) Next() *BooleanPoint {
	for {
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Reset window and counter if a new window is encountered.
		if p.Name != itr.prev.name || !p.Tags.Equals(&itr.prev.tags) {
			itr.prev.name = p.Name
			itr.prev.tags = p.Tags
			itr.n = 0
		}

		// Increment counter.
		itr.n++

		// Read next point if not beyond the offset.
		if itr.n <= itr.opt.Offset {
			continue
		}

		// Read next point if we're beyond the limit.
		if itr.opt.Limit > 0 && (itr.n-itr.opt.Offset) > itr.opt.Limit {
			// If there's no interval, no groups, and a single source then simply exit.
			if itr.opt.Interval.IsZero() && len(itr.opt.Dimensions) == 0 && len(itr.opt.Sources) == 1 {
				return nil
			}
			continue
		}

		return p
	}
}

type booleanFillIterator struct {
	input     *bufBooleanIterator
	prev      *BooleanPoint
	startTime int64
	endTime   int64
	auxFields []interface{}
	done      bool
	opt       IteratorOptions

	window struct {
		name string
		tags Tags
		time int64
	}
}

func newBooleanFillIterator(input BooleanIterator, expr Expr, opt IteratorOptions) *booleanFillIterator {
	if opt.Fill == NullFill {
		if expr, ok := expr.(*Call); ok && expr.Name == "count" {
			opt.Fill = NumberFill
			opt.FillValue = false
		}
	}

	var startTime, endTime int64
	if opt.Ascending {
		startTime, _ = opt.Window(opt.StartTime)
		_, endTime = opt.Window(opt.EndTime)
	} else {
		_, startTime = opt.Window(opt.EndTime)
		endTime, _ = opt.Window(opt.StartTime)
	}

	var auxFields []interface{}
	if len(opt.Aux) > 0 {
		auxFields = make([]interface{}, len(opt.Aux))
	}

	itr := &booleanFillIterator{
		input:     newBufBooleanIterator(input),
		startTime: startTime,
		endTime:   endTime,
		auxFields: auxFields,
		opt:       opt,
	}

	p := itr.input.peek()
	if p != nil {
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
	} else {
		itr.window.time = itr.endTime
	}
	return itr
}

func (itr *booleanFillIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *booleanFillIterator) Close() error         { return itr.input.Close() }

func (itr *booleanFillIterator) Next() *BooleanPoint {
	p := itr.input.Next()

	// Check if the next point is outside of our window or is nil.
	for p == nil || p.Name != itr.window.name || p.Tags.ID() != itr.window.tags.ID() {
		// If we are inside of an interval, unread the point and continue below to
		// constructing a new point.
		if itr.opt.Ascending {
			if itr.window.time < itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		} else {
			if itr.window.time >= itr.endTime {
				itr.input.unread(p)
				p = nil
				break
			}
		}

		// We are *not* in a current interval. If there is no next point,
		// we are at the end of all intervals.
		if p == nil {
			return nil
		}

		// Set the new interval.
		itr.window.name, itr.window.tags = p.Name, p.Tags
		itr.window.time = itr.startTime
		itr.prev = nil
		break
	}

	// Check if the point is our next expected point.
	if p == nil || p.Time > itr.window.time {
		if p != nil {
			itr.input.unread(p)
		}

		p = &BooleanPoint{
			Name: itr.window.name,
			Tags: itr.window.tags,
			Time: itr.window.time,
			Aux:  itr.auxFields,
		}

		switch itr.opt.Fill {
		case NullFill:
			p.Nil = true
		case NumberFill:
			p.Value = castToBoolean(itr.opt.FillValue)
		case PreviousFill:
			if itr.prev != nil {
				p.Value = itr.prev.Value
				p.Nil = itr.prev.Nil
			} else {
				p.Nil = true
			}
		}
	} else {
		itr.prev = p
	}

	// Advance the expected time. Do not advance to a new window here
	// as there may be lingering points with the same timestamp in the previous
	// window.
	if itr.opt.Ascending {
		itr.window.time = p.Time + int64(itr.opt.Interval.Duration)
	} else {
		itr.window.time = p.Time - int64(itr.opt.Interval.Duration)
	}
	return p
}

// booleanIntervalIterator represents a boolean implementation of IntervalIterator.
type booleanIntervalIterator struct {
	input BooleanIterator
	opt   IteratorOptions
}

func newBooleanIntervalIterator(input BooleanIterator, opt IteratorOptions) *booleanIntervalIterator {
	return &booleanIntervalIterator{input: input, opt: opt}
}

func (itr *booleanIntervalIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *booleanIntervalIterator) Close() error         { return itr.input.Close() }

func (itr *booleanIntervalIterator) Next() *BooleanPoint {
	p := itr.input.Next()
	if p == nil {
		return p
	}
	p.Time, _ = itr.opt.Window(p.Time)
	return p
}

// booleanInterruptIterator represents a boolean implementation of InterruptIterator.
type booleanInterruptIterator struct {
	input   BooleanIterator
	closing <-chan struct{}
	count   int
}

func newBooleanInterruptIterator(input BooleanIterator, closing <-chan struct{}) *booleanInterruptIterator {
	return &booleanInterruptIterator{input: input, closing: closing}
}

func (itr *booleanInterruptIterator) Stats() IteratorStats { return itr.input.Stats() }
func (itr *booleanInterruptIterator) Close() error         { return itr.input.Close() }

func (itr *booleanInterruptIterator) Next() *BooleanPoint {
	// Only check if the channel is closed every 256 points. This
	// intentionally checks on both 0 and 256 so that if the iterator
	// has been interrupted before the first point is emitted it will
	// not emit any points.
	if itr.count&0x100 == 0 {
		select {
		case <-itr.closing:
			return nil
		default:
			// Reset iterator count to zero and fall through to emit the next point.
			itr.count = 0
		}
	}

	// Increment the counter for every point read.
	itr.count++
	return itr.input.Next()
}

// booleanAuxIterator represents a boolean implementation of AuxIterator.
type booleanAuxIterator struct {
	input      *bufBooleanIterator
	output     chan *BooleanPoint
	fields     auxIteratorFields
	background bool
}

func newBooleanAuxIterator(input BooleanIterator, seriesKeys SeriesList, opt IteratorOptions) *booleanAuxIterator {
	return &booleanAuxIterator{
		input:  newBufBooleanIterator(input),
		output: make(chan *BooleanPoint, 1),
		fields: newAuxIteratorFields(seriesKeys, opt),
	}
}

func (itr *booleanAuxIterator) Background() {
	itr.background = true
	itr.Start()
	go drainIterator(itr)
}

func (itr *booleanAuxIterator) Start()                        { go itr.stream() }
func (itr *booleanAuxIterator) Stats() IteratorStats          { return itr.input.Stats() }
func (itr *booleanAuxIterator) Close() error                  { return itr.input.Close() }
func (itr *booleanAuxIterator) Next() *BooleanPoint           { return <-itr.output }
func (itr *booleanAuxIterator) Iterator(name string) Iterator { return itr.fields.iterator(name) }

func (itr *booleanAuxIterator) CreateIterator(opt IteratorOptions) (Iterator, error) {
	expr := opt.Expr
	if expr == nil {
		panic("unable to create an iterator with no expression from an aux iterator")
	}

	switch expr := expr.(type) {
	case *VarRef:
		return itr.Iterator(expr.Val), nil
	default:
		panic(fmt.Sprintf("invalid expression type for an aux iterator: %T", expr))
	}
}

func (itr *booleanAuxIterator) FieldDimensions(sources Sources) (fields, dimensions map[string]struct{}, err error) {
	return nil, nil, errors.New("not implemented")
}

func (itr *booleanAuxIterator) SeriesKeys(opt IteratorOptions) (SeriesList, error) {
	return nil, errors.New("not implemented")
}

func (itr *booleanAuxIterator) ExpandSources(sources Sources) (Sources, error) {
	return nil, errors.New("not implemented")
}

func (itr *booleanAuxIterator) stream() {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			break
		}

		// Send point to output and to each field iterator.
		itr.output <- p
		if ok := itr.fields.send(p); !ok && itr.background {
			break
		}
	}

	close(itr.output)
	itr.fields.close()
}

// booleanChanIterator represents a new instance of booleanChanIterator.
type booleanChanIterator struct {
	buf  *BooleanPoint
	cond *sync.Cond
	done bool
}

func (itr *booleanChanIterator) Stats() IteratorStats { return IteratorStats{} }

func (itr *booleanChanIterator) Close() error {
	itr.cond.L.Lock()
	// Mark the channel iterator as done and signal all waiting goroutines to start again.
	itr.done = true
	itr.cond.Broadcast()
	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return nil
}

func (itr *booleanChanIterator) setBuf(name string, tags Tags, time int64, value interface{}) bool {
	itr.cond.L.Lock()
	defer itr.cond.L.Unlock()

	// Wait for either the iterator to be done (so we don't have to set the value)
	// or for the buffer to have been read and ready for another write.
	for !itr.done && itr.buf != nil {
		itr.cond.Wait()
	}

	// Do not set the value and return false to signal that the iterator is closed.
	// Do this after the above wait as the above for loop may have exited because
	// the iterator was closed.
	if itr.done {
		return false
	}

	switch v := value.(type) {
	case bool:
		itr.buf = &BooleanPoint{Name: name, Tags: tags, Time: time, Value: v}

	default:
		itr.buf = &BooleanPoint{Name: name, Tags: tags, Time: time, Nil: true}
	}
	// Signal to all waiting goroutines that a new value is ready to read.
	itr.cond.Signal()
	return true
}

func (itr *booleanChanIterator) Next() *BooleanPoint {
	itr.cond.L.Lock()

	// Wait until either a value is available in the buffer or
	// the iterator is closed.
	for !itr.done && itr.buf == nil {
		itr.cond.Wait()
	}

	// Always read from the buffer if it exists, even if the iterator
	// is closed. This prevents the last value from being truncated by
	// the parent iterator.
	p := itr.buf
	itr.buf = nil
	itr.cond.Signal()

	// Do not defer the unlock so we don't create an unnecessary allocation.
	itr.cond.L.Unlock()
	return p
}

// booleanReduceFloatIterator executes a reducer for every interval and buffers the result.
type booleanReduceFloatIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	points []FloatPoint
}

// Stats returns stats from the input iterator.
func (itr *booleanReduceFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanReduceFloatIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *booleanReduceFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// booleanReduceFloatPoint stores the reduced data for a name/tag combination.
type booleanReduceFloatPoint struct {
	Name       string
	Tags       Tags
	Aggregator BooleanPointAggregator
	Emitter    FloatPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *booleanReduceFloatIterator) reduce() []FloatPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*booleanReduceFloatPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]FloatPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// booleanStreamFloatIterator
type booleanStreamFloatIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, FloatPointEmitter)
	opt    IteratorOptions
	m      map[string]*booleanReduceFloatPoint
	points []FloatPoint
}

func newBooleanStreamFloatIterator(input BooleanIterator, createFn func() (BooleanPointAggregator, FloatPointEmitter), opt IteratorOptions) *booleanStreamFloatIterator {
	return &booleanStreamFloatIterator{
		input:  newBufBooleanIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*booleanReduceFloatPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *booleanStreamFloatIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanStreamFloatIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *booleanStreamFloatIterator) Next() *FloatPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *booleanStreamFloatIterator) reduce() []FloatPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceFloatPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// booleanFloatExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type booleanFloatExprIterator struct {
	left  *bufBooleanIterator
	right *bufBooleanIterator
	fn    booleanFloatExprFunc
}

func (itr *booleanFloatExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *booleanFloatExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *booleanFloatExprIterator) Next() *FloatPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// booleanFloatExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type booleanFloatExprFunc func(a *BooleanPoint, b *BooleanPoint) *FloatPoint

// booleanReduceIntegerIterator executes a reducer for every interval and buffers the result.
type booleanReduceIntegerIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	points []IntegerPoint
}

// Stats returns stats from the input iterator.
func (itr *booleanReduceIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanReduceIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *booleanReduceIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// booleanReduceIntegerPoint stores the reduced data for a name/tag combination.
type booleanReduceIntegerPoint struct {
	Name       string
	Tags       Tags
	Aggregator BooleanPointAggregator
	Emitter    IntegerPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *booleanReduceIntegerIterator) reduce() []IntegerPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*booleanReduceIntegerPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]IntegerPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// booleanStreamIntegerIterator
type booleanStreamIntegerIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, IntegerPointEmitter)
	opt    IteratorOptions
	m      map[string]*booleanReduceIntegerPoint
	points []IntegerPoint
}

func newBooleanStreamIntegerIterator(input BooleanIterator, createFn func() (BooleanPointAggregator, IntegerPointEmitter), opt IteratorOptions) *booleanStreamIntegerIterator {
	return &booleanStreamIntegerIterator{
		input:  newBufBooleanIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*booleanReduceIntegerPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *booleanStreamIntegerIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanStreamIntegerIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *booleanStreamIntegerIterator) Next() *IntegerPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *booleanStreamIntegerIterator) reduce() []IntegerPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceIntegerPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// booleanIntegerExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type booleanIntegerExprIterator struct {
	left  *bufBooleanIterator
	right *bufBooleanIterator
	fn    booleanIntegerExprFunc
}

func (itr *booleanIntegerExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *booleanIntegerExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *booleanIntegerExprIterator) Next() *IntegerPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// booleanIntegerExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type booleanIntegerExprFunc func(a *BooleanPoint, b *BooleanPoint) *IntegerPoint

// booleanReduceStringIterator executes a reducer for every interval and buffers the result.
type booleanReduceStringIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	points []StringPoint
}

// Stats returns stats from the input iterator.
func (itr *booleanReduceStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanReduceStringIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *booleanReduceStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// booleanReduceStringPoint stores the reduced data for a name/tag combination.
type booleanReduceStringPoint struct {
	Name       string
	Tags       Tags
	Aggregator BooleanPointAggregator
	Emitter    StringPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *booleanReduceStringIterator) reduce() []StringPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*booleanReduceStringPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]StringPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// booleanStreamStringIterator
type booleanStreamStringIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, StringPointEmitter)
	opt    IteratorOptions
	m      map[string]*booleanReduceStringPoint
	points []StringPoint
}

func newBooleanStreamStringIterator(input BooleanIterator, createFn func() (BooleanPointAggregator, StringPointEmitter), opt IteratorOptions) *booleanStreamStringIterator {
	return &booleanStreamStringIterator{
		input:  newBufBooleanIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*booleanReduceStringPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *booleanStreamStringIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanStreamStringIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *booleanStreamStringIterator) Next() *StringPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *booleanStreamStringIterator) reduce() []StringPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceStringPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// booleanStringExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type booleanStringExprIterator struct {
	left  *bufBooleanIterator
	right *bufBooleanIterator
	fn    booleanStringExprFunc
}

func (itr *booleanStringExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *booleanStringExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *booleanStringExprIterator) Next() *StringPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// booleanStringExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type booleanStringExprFunc func(a *BooleanPoint, b *BooleanPoint) *StringPoint

// booleanReduceBooleanIterator executes a reducer for every interval and buffers the result.
type booleanReduceBooleanIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	points []BooleanPoint
}

// Stats returns stats from the input iterator.
func (itr *booleanReduceBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanReduceBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *booleanReduceBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// booleanReduceBooleanPoint stores the reduced data for a name/tag combination.
type booleanReduceBooleanPoint struct {
	Name       string
	Tags       Tags
	Aggregator BooleanPointAggregator
	Emitter    BooleanPointEmitter
}

// reduce executes fn once for every point in the next window.
// The previous value for the dimension is passed to fn.
func (itr *booleanReduceBooleanIterator) reduce() []BooleanPoint {
	// Calculate next window.
	startTime, endTime := itr.opt.Window(itr.input.peekTime())

	// Create points by tags.
	m := make(map[string]*booleanReduceBooleanPoint)
	for {
		// Read next point.
		curr := itr.input.NextInWindow(startTime, endTime)
		if curr == nil {
			break
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)
	}

	// Reverse sort points by name & tag.
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Sort(sort.Reverse(sort.StringSlice(keys)))

	a := make([]BooleanPoint, 0, len(m))
	for _, k := range keys {
		rp := m[k]
		points := rp.Emitter.Emit()
		for i := len(points) - 1; i >= 0; i-- {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
			// Set the points time to the interval time if the reducer didn't provide one.
			if points[i].Time == ZeroTime {
				points[i].Time = startTime
			}
			a = append(a, points[i])
		}
	}

	return a
}

// booleanStreamBooleanIterator
type booleanStreamBooleanIterator struct {
	input  *bufBooleanIterator
	create func() (BooleanPointAggregator, BooleanPointEmitter)
	opt    IteratorOptions
	m      map[string]*booleanReduceBooleanPoint
	points []BooleanPoint
}

func newBooleanStreamBooleanIterator(input BooleanIterator, createFn func() (BooleanPointAggregator, BooleanPointEmitter), opt IteratorOptions) *booleanStreamBooleanIterator {
	return &booleanStreamBooleanIterator{
		input:  newBufBooleanIterator(input),
		create: createFn,
		opt:    opt,
		m:      make(map[string]*booleanReduceBooleanPoint),
	}
}

// Stats returns stats from the input iterator.
func (itr *booleanStreamBooleanIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanStreamBooleanIterator) Close() error { return itr.input.Close() }

// Next returns the next value for the stream iterator.
func (itr *booleanStreamBooleanIterator) Next() *BooleanPoint {
	// Calculate next window if we have no more points.
	if len(itr.points) == 0 {
		itr.points = itr.reduce()
		if len(itr.points) == 0 {
			return nil
		}
	}

	// Pop next point off the stack.
	p := &itr.points[len(itr.points)-1]
	itr.points = itr.points[:len(itr.points)-1]
	return p
}

// reduce creates and manages aggregators for every point from the input.
// After aggregating a point, it always tries to emit a value using the emitter.
func (itr *booleanStreamBooleanIterator) reduce() []BooleanPoint {
	for {
		// Read next point.
		curr := itr.input.Next()
		if curr == nil {
			return nil
		} else if curr.Nil {
			continue
		}
		tags := curr.Tags.Subset(itr.opt.Dimensions)

		id := curr.Name
		if len(tags.m) > 0 {
			id += "\x00" + tags.ID()
		}

		// Retrieve the aggregator for this name/tag combination or create one.
		rp := itr.m[id]
		if rp == nil {
			aggregator, emitter := itr.create()
			rp = &booleanReduceBooleanPoint{
				Name:       curr.Name,
				Tags:       tags,
				Aggregator: aggregator,
				Emitter:    emitter,
			}
			itr.m[id] = rp
		}
		rp.Aggregator.AggregateBoolean(curr)

		// Attempt to emit points from the aggregator.
		points := rp.Emitter.Emit()
		if len(points) == 0 {
			continue
		}

		for i := range points {
			points[i].Name = rp.Name
			points[i].Tags = rp.Tags
		}
		return points
	}
}

// booleanExprIterator executes a function to modify an existing point
// for every output of the input iterator.
type booleanExprIterator struct {
	left  *bufBooleanIterator
	right *bufBooleanIterator
	fn    booleanExprFunc
}

func (itr *booleanExprIterator) Stats() IteratorStats {
	stats := itr.left.Stats()
	stats.Add(itr.right.Stats())
	return stats
}

func (itr *booleanExprIterator) Close() error {
	itr.left.Close()
	itr.right.Close()
	return nil
}

func (itr *booleanExprIterator) Next() *BooleanPoint {
	a := itr.left.Next()
	b := itr.right.Next()
	if a == nil && b == nil {
		return nil
	}
	return itr.fn(a, b)
}

// booleanExprFunc creates or modifies a point by combining two
// points. The point passed in may be modified and returned rather than
// allocating a new point if possible. One of the points may be nil, but at
// least one of the points will be non-nil.
type booleanExprFunc func(a *BooleanPoint, b *BooleanPoint) *BooleanPoint

// booleanTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type booleanTransformIterator struct {
	input BooleanIterator
	fn    booleanTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *booleanTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *booleanTransformIterator) Next() *BooleanPoint {
	p := itr.input.Next()
	if p != nil {
		p = itr.fn(p)
	}
	return p
}

// booleanTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type booleanTransformFunc func(p *BooleanPoint) *BooleanPoint

// booleanBoolTransformIterator executes a function to modify an existing point for every
// output of the input iterator.
type booleanBoolTransformIterator struct {
	input BooleanIterator
	fn    booleanBoolTransformFunc
}

// Stats returns stats from the input iterator.
func (itr *booleanBoolTransformIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanBoolTransformIterator) Close() error { return itr.input.Close() }

// Next returns the minimum value for the next available interval.
func (itr *booleanBoolTransformIterator) Next() *BooleanPoint {
	p := itr.input.Next()
	if p != nil {
		return itr.fn(p)
	}
	return nil
}

// booleanBoolTransformFunc creates or modifies a point.
// The point passed in may be modified and returned rather than allocating a
// new point if possible.
type booleanBoolTransformFunc func(p *BooleanPoint) *BooleanPoint

// booleanDedupeIterator only outputs unique points.
// This differs from the DistinctIterator in that it compares all aux fields too.
// This iterator is relatively inefficient and should only be used on small
// datasets such as meta query results.
type booleanDedupeIterator struct {
	input BooleanIterator
	m     map[string]struct{} // lookup of points already sent
}

// newBooleanDedupeIterator returns a new instance of booleanDedupeIterator.
func newBooleanDedupeIterator(input BooleanIterator) *booleanDedupeIterator {
	return &booleanDedupeIterator{
		input: input,
		m:     make(map[string]struct{}),
	}
}

// Stats returns stats from the input iterator.
func (itr *booleanDedupeIterator) Stats() IteratorStats { return itr.input.Stats() }

// Close closes the iterator and all child iterators.
func (itr *booleanDedupeIterator) Close() error { return itr.input.Close() }

// Next returns the next unique point from the input iterator.
func (itr *booleanDedupeIterator) Next() *BooleanPoint {
	for {
		// Read next point.
		p := itr.input.Next()
		if p == nil {
			return nil
		}

		// Serialize to bytes to store in lookup.
		buf, err := proto.Marshal(encodeBooleanPoint(p))
		if err != nil {
			log.Println("error marshaling dedupe point:", err)
			continue
		}

		// If the point has already been output then move to the next point.
		if _, ok := itr.m[string(buf)]; ok {
			continue
		}

		// Otherwise mark it as emitted and return point.
		itr.m[string(buf)] = struct{}{}
		return p
	}
}

// booleanReaderIterator represents an iterator that streams from a reader.
type booleanReaderIterator struct {
	r     io.Reader
	dec   *BooleanPointDecoder
	first *BooleanPoint
}

// newBooleanReaderIterator returns a new instance of booleanReaderIterator.
func newBooleanReaderIterator(r io.Reader, first *BooleanPoint, stats IteratorStats) *booleanReaderIterator {
	dec := NewBooleanPointDecoder(r)
	dec.stats = stats

	return &booleanReaderIterator{
		r:     r,
		dec:   dec,
		first: first,
	}
}

// Stats returns stats about points processed.
func (itr *booleanReaderIterator) Stats() IteratorStats { return itr.dec.stats }

// Close closes the underlying reader, if applicable.
func (itr *booleanReaderIterator) Close() error {
	if r, ok := itr.r.(io.ReadCloser); ok {
		return r.Close()
	}
	return nil
}

// Next returns the next point from the iterator.
func (itr *booleanReaderIterator) Next() *BooleanPoint {
	// Send first point if it hasn't been sent yet.
	if itr.first != nil {
		p := itr.first
		itr.first = nil
		return p
	}

	// OPTIMIZE(benbjohnson): Reuse point on iterator.

	// Unmarshal next point.
	p := &BooleanPoint{}
	if err := itr.dec.DecodeBooleanPoint(p); err == io.EOF {
		return nil
	} else if err != nil {
		log.Printf("error reading iterator point: %s", err)
		return nil
	}
	return p
}

// IteratorEncoder is an encoder for encoding an iterator's points to w.
type IteratorEncoder struct {
	w io.Writer

	// Frequency with which stats are emitted.
	StatsInterval time.Duration
}

// NewIteratorEncoder encodes an iterator's points to w.
func NewIteratorEncoder(w io.Writer) *IteratorEncoder {
	return &IteratorEncoder{
		w: w,

		StatsInterval: DefaultStatsInterval,
	}
}

// Encode encodes and writes all of itr's points to the underlying writer.
func (enc *IteratorEncoder) EncodeIterator(itr Iterator) error {
	switch itr := itr.(type) {
	case FloatIterator:
		return enc.encodeFloatIterator(itr)
	case IntegerIterator:
		return enc.encodeIntegerIterator(itr)
	case StringIterator:
		return enc.encodeStringIterator(itr)
	case BooleanIterator:
		return enc.encodeBooleanIterator(itr)
	default:
		panic(fmt.Sprintf("unsupported iterator for encoder: %T", itr))
	}
}

// encodeFloatIterator encodes all points from itr to the underlying writer.
func (enc *IteratorEncoder) encodeFloatIterator(itr FloatIterator) error {
	ticker := time.NewTicker(enc.StatsInterval)
	defer ticker.Stop()

	// Emit initial stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}

	// Continually stream points from the iterator into the encoder.
	penc := NewFloatPointEncoder(enc.w)
	for {
		// Emit stats periodically.
		select {
		case <-ticker.C:
			if err := enc.encodeStats(itr.Stats()); err != nil {
				return err
			}
		default:
		}

		// Retrieve the next point from the iterator.
		p := itr.Next()
		if p == nil {
			break
		}

		// Write the point to the point encoder.
		if err := penc.EncodeFloatPoint(p); err != nil {
			return err
		}
	}

	// Emit final stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}
	return nil
}

// encodeIntegerIterator encodes all points from itr to the underlying writer.
func (enc *IteratorEncoder) encodeIntegerIterator(itr IntegerIterator) error {
	ticker := time.NewTicker(enc.StatsInterval)
	defer ticker.Stop()

	// Emit initial stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}

	// Continually stream points from the iterator into the encoder.
	penc := NewIntegerPointEncoder(enc.w)
	for {
		// Emit stats periodically.
		select {
		case <-ticker.C:
			if err := enc.encodeStats(itr.Stats()); err != nil {
				return err
			}
		default:
		}

		// Retrieve the next point from the iterator.
		p := itr.Next()
		if p == nil {
			break
		}

		// Write the point to the point encoder.
		if err := penc.EncodeIntegerPoint(p); err != nil {
			return err
		}
	}

	// Emit final stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}
	return nil
}

// encodeStringIterator encodes all points from itr to the underlying writer.
func (enc *IteratorEncoder) encodeStringIterator(itr StringIterator) error {
	ticker := time.NewTicker(enc.StatsInterval)
	defer ticker.Stop()

	// Emit initial stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}

	// Continually stream points from the iterator into the encoder.
	penc := NewStringPointEncoder(enc.w)
	for {
		// Emit stats periodically.
		select {
		case <-ticker.C:
			if err := enc.encodeStats(itr.Stats()); err != nil {
				return err
			}
		default:
		}

		// Retrieve the next point from the iterator.
		p := itr.Next()
		if p == nil {
			break
		}

		// Write the point to the point encoder.
		if err := penc.EncodeStringPoint(p); err != nil {
			return err
		}
	}

	// Emit final stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}
	return nil
}

// encodeBooleanIterator encodes all points from itr to the underlying writer.
func (enc *IteratorEncoder) encodeBooleanIterator(itr BooleanIterator) error {
	ticker := time.NewTicker(enc.StatsInterval)
	defer ticker.Stop()

	// Emit initial stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}

	// Continually stream points from the iterator into the encoder.
	penc := NewBooleanPointEncoder(enc.w)
	for {
		// Emit stats periodically.
		select {
		case <-ticker.C:
			if err := enc.encodeStats(itr.Stats()); err != nil {
				return err
			}
		default:
		}

		// Retrieve the next point from the iterator.
		p := itr.Next()
		if p == nil {
			break
		}

		// Write the point to the point encoder.
		if err := penc.EncodeBooleanPoint(p); err != nil {
			return err
		}
	}

	// Emit final stats.
	if err := enc.encodeStats(itr.Stats()); err != nil {
		return err
	}
	return nil
}

// encode a stats object in the point stream.
func (enc *IteratorEncoder) encodeStats(stats IteratorStats) error {
	buf, err := proto.Marshal(&internal.Point{
		Name: proto.String(""),
		Tags: proto.String(""),
		Time: proto.Int64(0),
		Nil:  proto.Bool(false),

		Stats: encodeIteratorStats(&stats),
	})
	if err != nil {
		return err
	}

	if err := binary.Write(enc.w, binary.BigEndian, uint32(len(buf))); err != nil {
		return err
	}
	if _, err := enc.w.Write(buf); err != nil {
		return err
	}
	return nil
}