iterator.gen.go

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

package tsm1

import (
	"fmt"
	"runtime"
	"sort"
	"sync"

	"github.com/ivopetiz/influxdb/pkg/metrics"
	"github.com/ivopetiz/influxdb/pkg/tracing"
	"github.com/ivopetiz/influxdb/pkg/tracing/fields"
	"github.com/ivopetiz/influxdb/query"
	"github.com/ivopetiz/influxdb/tsdb"
	"github.com/ivopetiz/influxql"
	"go.uber.org/zap"
)

type cursor interface {
	close() error
	next() (t int64, v interface{})
}

// cursorAt provides a bufferred cursor interface.
// This required for literal value cursors which don't have a time value.
type cursorAt interface {
	close() error
	peek() (k int64, v interface{})
	nextAt(seek int64) interface{}
}

type nilCursor struct{}

func (nilCursor) next() (int64, interface{}) { return tsdb.EOF, nil }
func (nilCursor) close() error               { return nil }

// bufCursor implements a bufferred cursor.
type bufCursor struct {
	cur cursor
	buf struct {
		key    int64
		value  interface{}
		filled bool
	}
	ascending bool
}

// newBufCursor returns a bufferred wrapper for cur.
func newBufCursor(cur cursor, ascending bool) *bufCursor {
	return &bufCursor{cur: cur, ascending: ascending}
}

func (c *bufCursor) close() error {
	if c.cur == nil {
		return nil
	}

	err := c.cur.close()
	c.cur = nil
	return err
}

// next returns the buffer, if filled. Otherwise returns the next key/value from the cursor.
func (c *bufCursor) next() (int64, interface{}) {
	if c.buf.filled {
		k, v := c.buf.key, c.buf.value
		c.buf.filled = false
		return k, v
	}
	return c.cur.next()
}

// unread pushes k and v onto the buffer.
func (c *bufCursor) unread(k int64, v interface{}) {
	c.buf.key, c.buf.value = k, v
	c.buf.filled = true
}

// peek reads next next key/value without removing them from the cursor.
func (c *bufCursor) peek() (k int64, v interface{}) {
	k, v = c.next()
	c.unread(k, v)
	return
}

// nextAt returns the next value where key is equal to seek.
// Skips over any keys that are less than seek.
// If the key doesn't exist then a nil value is returned instead.
func (c *bufCursor) nextAt(seek int64) interface{} {
	for {
		k, v := c.next()
		if k != tsdb.EOF {
			if k == seek {
				return v
			} else if c.ascending && k < seek {
				continue
			} else if !c.ascending && k > seek {
				continue
			}
			c.unread(k, v)
		}

		// Return "nil" value for type.
		switch c.cur.(type) {
		case floatCursor:
			return (*float64)(nil)
		case integerCursor:
			return (*int64)(nil)
		case unsignedCursor:
			return (*uint64)(nil)
		case stringCursor:
			return (*string)(nil)
		case booleanCursor:
			return (*bool)(nil)
		default:
			panic("unreachable")
		}
	}
}

// statsBufferCopyIntervalN is the number of points that are read before
// copying the stats buffer to the iterator's stats field. This is used to
// amortize the cost of using a mutex when updating stats.
const statsBufferCopyIntervalN = 100

type floatFinalizerIterator struct {
	query.FloatIterator
	logger *zap.Logger
}

func newFloatFinalizerIterator(inner query.FloatIterator, logger *zap.Logger) *floatFinalizerIterator {
	itr := &floatFinalizerIterator{FloatIterator: inner, logger: logger}
	runtime.SetFinalizer(itr, (*floatFinalizerIterator).closeGC)
	return itr
}

func (itr *floatFinalizerIterator) closeGC() {
	go func() {
		itr.logger.Error("FloatIterator finalized by GC")
		itr.Close()
	}()
}

func (itr *floatFinalizerIterator) Close() error {
	runtime.SetFinalizer(itr, nil)
	return itr.FloatIterator.Close()
}

type floatInstrumentedIterator struct {
	query.FloatIterator
	span  *tracing.Span
	group *metrics.Group
}

func newFloatInstrumentedIterator(inner query.FloatIterator, span *tracing.Span, group *metrics.Group) *floatInstrumentedIterator {
	return &floatInstrumentedIterator{FloatIterator: inner, span: span, group: group}
}

func (itr *floatInstrumentedIterator) Close() error {
	var f fields.Fields
	itr.group.ForEach(func(v metrics.Metric) {
		switch m := v.(type) {
		case *metrics.Counter:
			f = append(f, fields.Int64(m.Name(), m.Value()))

		case *metrics.Timer:
			f = append(f, fields.Duration(m.Name(), m.Value()))

		default:
			panic("unexpected metrics")
		}
	})
	itr.span.SetFields(f)
	itr.span.Finish()

	return itr.FloatIterator.Close()
}

type floatIterator struct {
	cur   floatCursor
	aux   []cursorAt
	conds struct {
		names []string
		curs  []cursorAt
	}
	opt query.IteratorOptions

	m     map[string]interface{} // map used for condition evaluation
	point query.FloatPoint       // reusable buffer

	statsLock sync.Mutex
	stats     query.IteratorStats
	statsBuf  query.IteratorStats
}

func newFloatIterator(name string, tags query.Tags, opt query.IteratorOptions, cur floatCursor, aux []cursorAt, conds []cursorAt, condNames []string) *floatIterator {
	itr := &floatIterator{
		cur: cur,
		aux: aux,
		opt: opt,
		point: query.FloatPoint{
			Name: name,
			Tags: tags,
		},
		statsBuf: query.IteratorStats{
			SeriesN: 1,
		},
	}
	itr.stats = itr.statsBuf

	if len(aux) > 0 {
		itr.point.Aux = make([]interface{}, len(aux))
	}

	if opt.Condition != nil {
		itr.m = make(map[string]interface{}, len(aux)+len(conds))
	}
	itr.conds.names = condNames
	itr.conds.curs = conds

	return itr
}

// Next returns the next point from the iterator.
func (itr *floatIterator) Next() (*query.FloatPoint, error) {
	for {
		seek := tsdb.EOF

		if itr.cur != nil {
			// Read from the main cursor if we have one.
			itr.point.Time, itr.point.Value = itr.cur.nextFloat()
			seek = itr.point.Time
		} else {
			// Otherwise find lowest aux timestamp.
			for i := range itr.aux {
				if k, _ := itr.aux[i].peek(); k != tsdb.EOF {
					if seek == tsdb.EOF || (itr.opt.Ascending && k < seek) || (!itr.opt.Ascending && k > seek) {
						seek = k
					}
				}
			}
			itr.point.Time = seek
		}

		// Exit if we have no more points or we are outside our time range.
		if itr.point.Time == tsdb.EOF {
			itr.copyStats()
			return nil, nil
		} else if itr.opt.Ascending && itr.point.Time > itr.opt.EndTime {
			itr.copyStats()
			return nil, nil
		} else if !itr.opt.Ascending && itr.point.Time < itr.opt.StartTime {
			itr.copyStats()
			return nil, nil
		}

		// Read from each auxiliary cursor.
		for i := range itr.opt.Aux {
			itr.point.Aux[i] = itr.aux[i].nextAt(seek)
		}

		// Read from condition field cursors.
		for i := range itr.conds.curs {
			itr.m[itr.conds.names[i]] = itr.conds.curs[i].nextAt(seek)
		}

		// Evaluate condition, if one exists. Retry if it fails.
		valuer := influxql.ValuerEval{
			Valuer: influxql.MultiValuer(
				query.MathValuer{},
				influxql.MapValuer(itr.m),
			),
		}
		if itr.opt.Condition != nil && !valuer.EvalBool(itr.opt.Condition) {
			continue
		}

		// Track points returned.
		itr.statsBuf.PointN++

		// Copy buffer to stats periodically.
		if itr.statsBuf.PointN%statsBufferCopyIntervalN == 0 {
			itr.copyStats()
		}

		return &itr.point, nil
	}
}

// copyStats copies from the itr stats buffer to the stats under lock.
func (itr *floatIterator) copyStats() {
	itr.statsLock.Lock()
	itr.stats = itr.statsBuf
	itr.statsLock.Unlock()
}

// Stats returns stats on the points processed.
func (itr *floatIterator) Stats() query.IteratorStats {
	itr.statsLock.Lock()
	stats := itr.stats
	itr.statsLock.Unlock()
	return stats
}

// Close closes the iterator.
func (itr *floatIterator) Close() error {
	cursorsAt(itr.aux).close()
	itr.aux = nil
	cursorsAt(itr.conds.curs).close()
	itr.conds.curs = nil
	if itr.cur != nil {
		err := itr.cur.close()
		itr.cur = nil
		return err
	}
	return nil
}

// floatLimitIterator
type floatLimitIterator struct {
	input query.FloatIterator
	opt   query.IteratorOptions
	n     int
}

func newFloatLimitIterator(input query.FloatIterator, opt query.IteratorOptions) *floatLimitIterator {
	return &floatLimitIterator{
		input: input,
		opt:   opt,
	}
}

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

func (itr *floatLimitIterator) Next() (*query.FloatPoint, error) {
	// Check if we are beyond the limit.
	if (itr.n - itr.opt.Offset) > itr.opt.Limit {
		return nil, nil
	}

	// Read the next point.
	p, err := itr.input.Next()
	if p == nil || err != nil {
		return nil, err
	}

	// Increment counter.
	itr.n++

	// Offsets are handled by a higher level iterator so return all points.
	return p, nil
}

// floatCursor represents an object for iterating over a single float field.
type floatCursor interface {
	cursor
	nextFloat() (t int64, v float64)
}

func newFloatCursor(seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) floatCursor {
	if ascending {
		return newFloatAscendingCursor(seek, cacheValues, tsmKeyCursor)
	}
	return newFloatDescendingCursor(seek, cacheValues, tsmKeyCursor)
}

type floatAscendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []FloatValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newFloatAscendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *floatAscendingCursor {
	c := &floatAscendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadFloatBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})

	return c
}

// peekCache returns the current time/value from the cache.
func (c *floatAscendingCursor) peekCache() (t int64, v float64) {
	if c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, 0
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(FloatValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *floatAscendingCursor) peekTSM() (t int64, v float64) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, 0
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *floatAscendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *floatAscendingCursor) next() (int64, interface{}) { return c.nextFloat() }

// nextFloat returns the next key/value for the cursor.
func (c *floatAscendingCursor) nextFloat() (int64, float64) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, 0
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey < tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *floatAscendingCursor) nextCache() {
	if c.cache.pos >= len(c.cache.values) {
		return
	}
	c.cache.pos++
}

// nextTSM returns the next value from the TSM files.
func (c *floatAscendingCursor) nextTSM() {
	c.tsm.pos++
	if c.tsm.pos >= len(c.tsm.values) {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadFloatBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = 0
	}
}

type floatDescendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []FloatValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newFloatDescendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *floatDescendingCursor {
	c := &floatDescendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekCache(); t != seek {
		c.cache.pos--
	}

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadFloatBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekTSM(); t != seek {
		c.tsm.pos--
	}

	return c
}

// peekCache returns the current time/value from the cache.
func (c *floatDescendingCursor) peekCache() (t int64, v float64) {
	if c.cache.pos < 0 || c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, 0
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(FloatValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *floatDescendingCursor) peekTSM() (t int64, v float64) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, 0
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *floatDescendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *floatDescendingCursor) next() (int64, interface{}) { return c.nextFloat() }

// nextFloat returns the next key/value for the cursor.
func (c *floatDescendingCursor) nextFloat() (int64, float64) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, 0
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey > tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *floatDescendingCursor) nextCache() {
	if c.cache.pos < 0 {
		return
	}
	c.cache.pos--
}

// nextTSM returns the next value from the TSM files.
func (c *floatDescendingCursor) nextTSM() {
	c.tsm.pos--
	if c.tsm.pos < 0 {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadFloatBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = len(c.tsm.values) - 1
	}
}

type integerFinalizerIterator struct {
	query.IntegerIterator
	logger *zap.Logger
}

func newIntegerFinalizerIterator(inner query.IntegerIterator, logger *zap.Logger) *integerFinalizerIterator {
	itr := &integerFinalizerIterator{IntegerIterator: inner, logger: logger}
	runtime.SetFinalizer(itr, (*integerFinalizerIterator).closeGC)
	return itr
}

func (itr *integerFinalizerIterator) closeGC() {
	go func() {
		itr.logger.Error("IntegerIterator finalized by GC")
		itr.Close()
	}()
}

func (itr *integerFinalizerIterator) Close() error {
	runtime.SetFinalizer(itr, nil)
	return itr.IntegerIterator.Close()
}

type integerInstrumentedIterator struct {
	query.IntegerIterator
	span  *tracing.Span
	group *metrics.Group
}

func newIntegerInstrumentedIterator(inner query.IntegerIterator, span *tracing.Span, group *metrics.Group) *integerInstrumentedIterator {
	return &integerInstrumentedIterator{IntegerIterator: inner, span: span, group: group}
}

func (itr *integerInstrumentedIterator) Close() error {
	var f fields.Fields
	itr.group.ForEach(func(v metrics.Metric) {
		switch m := v.(type) {
		case *metrics.Counter:
			f = append(f, fields.Int64(m.Name(), m.Value()))

		case *metrics.Timer:
			f = append(f, fields.Duration(m.Name(), m.Value()))

		default:
			panic("unexpected metrics")
		}
	})
	itr.span.SetFields(f)
	itr.span.Finish()

	return itr.IntegerIterator.Close()
}

type integerIterator struct {
	cur   integerCursor
	aux   []cursorAt
	conds struct {
		names []string
		curs  []cursorAt
	}
	opt query.IteratorOptions

	m     map[string]interface{} // map used for condition evaluation
	point query.IntegerPoint     // reusable buffer

	statsLock sync.Mutex
	stats     query.IteratorStats
	statsBuf  query.IteratorStats
}

func newIntegerIterator(name string, tags query.Tags, opt query.IteratorOptions, cur integerCursor, aux []cursorAt, conds []cursorAt, condNames []string) *integerIterator {
	itr := &integerIterator{
		cur: cur,
		aux: aux,
		opt: opt,
		point: query.IntegerPoint{
			Name: name,
			Tags: tags,
		},
		statsBuf: query.IteratorStats{
			SeriesN: 1,
		},
	}
	itr.stats = itr.statsBuf

	if len(aux) > 0 {
		itr.point.Aux = make([]interface{}, len(aux))
	}

	if opt.Condition != nil {
		itr.m = make(map[string]interface{}, len(aux)+len(conds))
	}
	itr.conds.names = condNames
	itr.conds.curs = conds

	return itr
}

// Next returns the next point from the iterator.
func (itr *integerIterator) Next() (*query.IntegerPoint, error) {
	for {
		seek := tsdb.EOF

		if itr.cur != nil {
			// Read from the main cursor if we have one.
			itr.point.Time, itr.point.Value = itr.cur.nextInteger()
			seek = itr.point.Time
		} else {
			// Otherwise find lowest aux timestamp.
			for i := range itr.aux {
				if k, _ := itr.aux[i].peek(); k != tsdb.EOF {
					if seek == tsdb.EOF || (itr.opt.Ascending && k < seek) || (!itr.opt.Ascending && k > seek) {
						seek = k
					}
				}
			}
			itr.point.Time = seek
		}

		// Exit if we have no more points or we are outside our time range.
		if itr.point.Time == tsdb.EOF {
			itr.copyStats()
			return nil, nil
		} else if itr.opt.Ascending && itr.point.Time > itr.opt.EndTime {
			itr.copyStats()
			return nil, nil
		} else if !itr.opt.Ascending && itr.point.Time < itr.opt.StartTime {
			itr.copyStats()
			return nil, nil
		}

		// Read from each auxiliary cursor.
		for i := range itr.opt.Aux {
			itr.point.Aux[i] = itr.aux[i].nextAt(seek)
		}

		// Read from condition field cursors.
		for i := range itr.conds.curs {
			itr.m[itr.conds.names[i]] = itr.conds.curs[i].nextAt(seek)
		}

		// Evaluate condition, if one exists. Retry if it fails.
		valuer := influxql.ValuerEval{
			Valuer: influxql.MultiValuer(
				query.MathValuer{},
				influxql.MapValuer(itr.m),
			),
		}
		if itr.opt.Condition != nil && !valuer.EvalBool(itr.opt.Condition) {
			continue
		}

		// Track points returned.
		itr.statsBuf.PointN++

		// Copy buffer to stats periodically.
		if itr.statsBuf.PointN%statsBufferCopyIntervalN == 0 {
			itr.copyStats()
		}

		return &itr.point, nil
	}
}

// copyStats copies from the itr stats buffer to the stats under lock.
func (itr *integerIterator) copyStats() {
	itr.statsLock.Lock()
	itr.stats = itr.statsBuf
	itr.statsLock.Unlock()
}

// Stats returns stats on the points processed.
func (itr *integerIterator) Stats() query.IteratorStats {
	itr.statsLock.Lock()
	stats := itr.stats
	itr.statsLock.Unlock()
	return stats
}

// Close closes the iterator.
func (itr *integerIterator) Close() error {
	cursorsAt(itr.aux).close()
	itr.aux = nil
	cursorsAt(itr.conds.curs).close()
	itr.conds.curs = nil
	if itr.cur != nil {
		err := itr.cur.close()
		itr.cur = nil
		return err
	}
	return nil
}

// integerLimitIterator
type integerLimitIterator struct {
	input query.IntegerIterator
	opt   query.IteratorOptions
	n     int
}

func newIntegerLimitIterator(input query.IntegerIterator, opt query.IteratorOptions) *integerLimitIterator {
	return &integerLimitIterator{
		input: input,
		opt:   opt,
	}
}

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

func (itr *integerLimitIterator) Next() (*query.IntegerPoint, error) {
	// Check if we are beyond the limit.
	if (itr.n - itr.opt.Offset) > itr.opt.Limit {
		return nil, nil
	}

	// Read the next point.
	p, err := itr.input.Next()
	if p == nil || err != nil {
		return nil, err
	}

	// Increment counter.
	itr.n++

	// Offsets are handled by a higher level iterator so return all points.
	return p, nil
}

// integerCursor represents an object for iterating over a single integer field.
type integerCursor interface {
	cursor
	nextInteger() (t int64, v int64)
}

func newIntegerCursor(seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) integerCursor {
	if ascending {
		return newIntegerAscendingCursor(seek, cacheValues, tsmKeyCursor)
	}
	return newIntegerDescendingCursor(seek, cacheValues, tsmKeyCursor)
}

type integerAscendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []IntegerValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newIntegerAscendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *integerAscendingCursor {
	c := &integerAscendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadIntegerBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})

	return c
}

// peekCache returns the current time/value from the cache.
func (c *integerAscendingCursor) peekCache() (t int64, v int64) {
	if c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, 0
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(IntegerValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *integerAscendingCursor) peekTSM() (t int64, v int64) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, 0
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *integerAscendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *integerAscendingCursor) next() (int64, interface{}) { return c.nextInteger() }

// nextInteger returns the next key/value for the cursor.
func (c *integerAscendingCursor) nextInteger() (int64, int64) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, 0
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey < tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *integerAscendingCursor) nextCache() {
	if c.cache.pos >= len(c.cache.values) {
		return
	}
	c.cache.pos++
}

// nextTSM returns the next value from the TSM files.
func (c *integerAscendingCursor) nextTSM() {
	c.tsm.pos++
	if c.tsm.pos >= len(c.tsm.values) {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadIntegerBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = 0
	}
}

type integerDescendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []IntegerValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newIntegerDescendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *integerDescendingCursor {
	c := &integerDescendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekCache(); t != seek {
		c.cache.pos--
	}

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadIntegerBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekTSM(); t != seek {
		c.tsm.pos--
	}

	return c
}

// peekCache returns the current time/value from the cache.
func (c *integerDescendingCursor) peekCache() (t int64, v int64) {
	if c.cache.pos < 0 || c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, 0
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(IntegerValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *integerDescendingCursor) peekTSM() (t int64, v int64) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, 0
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *integerDescendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *integerDescendingCursor) next() (int64, interface{}) { return c.nextInteger() }

// nextInteger returns the next key/value for the cursor.
func (c *integerDescendingCursor) nextInteger() (int64, int64) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, 0
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey > tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *integerDescendingCursor) nextCache() {
	if c.cache.pos < 0 {
		return
	}
	c.cache.pos--
}

// nextTSM returns the next value from the TSM files.
func (c *integerDescendingCursor) nextTSM() {
	c.tsm.pos--
	if c.tsm.pos < 0 {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadIntegerBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = len(c.tsm.values) - 1
	}
}

type unsignedFinalizerIterator struct {
	query.UnsignedIterator
	logger *zap.Logger
}

func newUnsignedFinalizerIterator(inner query.UnsignedIterator, logger *zap.Logger) *unsignedFinalizerIterator {
	itr := &unsignedFinalizerIterator{UnsignedIterator: inner, logger: logger}
	runtime.SetFinalizer(itr, (*unsignedFinalizerIterator).closeGC)
	return itr
}

func (itr *unsignedFinalizerIterator) closeGC() {
	go func() {
		itr.logger.Error("UnsignedIterator finalized by GC")
		itr.Close()
	}()
}

func (itr *unsignedFinalizerIterator) Close() error {
	runtime.SetFinalizer(itr, nil)
	return itr.UnsignedIterator.Close()
}

type unsignedInstrumentedIterator struct {
	query.UnsignedIterator
	span  *tracing.Span
	group *metrics.Group
}

func newUnsignedInstrumentedIterator(inner query.UnsignedIterator, span *tracing.Span, group *metrics.Group) *unsignedInstrumentedIterator {
	return &unsignedInstrumentedIterator{UnsignedIterator: inner, span: span, group: group}
}

func (itr *unsignedInstrumentedIterator) Close() error {
	var f fields.Fields
	itr.group.ForEach(func(v metrics.Metric) {
		switch m := v.(type) {
		case *metrics.Counter:
			f = append(f, fields.Int64(m.Name(), m.Value()))

		case *metrics.Timer:
			f = append(f, fields.Duration(m.Name(), m.Value()))

		default:
			panic("unexpected metrics")
		}
	})
	itr.span.SetFields(f)
	itr.span.Finish()

	return itr.UnsignedIterator.Close()
}

type unsignedIterator struct {
	cur   unsignedCursor
	aux   []cursorAt
	conds struct {
		names []string
		curs  []cursorAt
	}
	opt query.IteratorOptions

	m     map[string]interface{} // map used for condition evaluation
	point query.UnsignedPoint    // reusable buffer

	statsLock sync.Mutex
	stats     query.IteratorStats
	statsBuf  query.IteratorStats
}

func newUnsignedIterator(name string, tags query.Tags, opt query.IteratorOptions, cur unsignedCursor, aux []cursorAt, conds []cursorAt, condNames []string) *unsignedIterator {
	itr := &unsignedIterator{
		cur: cur,
		aux: aux,
		opt: opt,
		point: query.UnsignedPoint{
			Name: name,
			Tags: tags,
		},
		statsBuf: query.IteratorStats{
			SeriesN: 1,
		},
	}
	itr.stats = itr.statsBuf

	if len(aux) > 0 {
		itr.point.Aux = make([]interface{}, len(aux))
	}

	if opt.Condition != nil {
		itr.m = make(map[string]interface{}, len(aux)+len(conds))
	}
	itr.conds.names = condNames
	itr.conds.curs = conds

	return itr
}

// Next returns the next point from the iterator.
func (itr *unsignedIterator) Next() (*query.UnsignedPoint, error) {
	for {
		seek := tsdb.EOF

		if itr.cur != nil {
			// Read from the main cursor if we have one.
			itr.point.Time, itr.point.Value = itr.cur.nextUnsigned()
			seek = itr.point.Time
		} else {
			// Otherwise find lowest aux timestamp.
			for i := range itr.aux {
				if k, _ := itr.aux[i].peek(); k != tsdb.EOF {
					if seek == tsdb.EOF || (itr.opt.Ascending && k < seek) || (!itr.opt.Ascending && k > seek) {
						seek = k
					}
				}
			}
			itr.point.Time = seek
		}

		// Exit if we have no more points or we are outside our time range.
		if itr.point.Time == tsdb.EOF {
			itr.copyStats()
			return nil, nil
		} else if itr.opt.Ascending && itr.point.Time > itr.opt.EndTime {
			itr.copyStats()
			return nil, nil
		} else if !itr.opt.Ascending && itr.point.Time < itr.opt.StartTime {
			itr.copyStats()
			return nil, nil
		}

		// Read from each auxiliary cursor.
		for i := range itr.opt.Aux {
			itr.point.Aux[i] = itr.aux[i].nextAt(seek)
		}

		// Read from condition field cursors.
		for i := range itr.conds.curs {
			itr.m[itr.conds.names[i]] = itr.conds.curs[i].nextAt(seek)
		}

		// Evaluate condition, if one exists. Retry if it fails.
		valuer := influxql.ValuerEval{
			Valuer: influxql.MultiValuer(
				query.MathValuer{},
				influxql.MapValuer(itr.m),
			),
		}
		if itr.opt.Condition != nil && !valuer.EvalBool(itr.opt.Condition) {
			continue
		}

		// Track points returned.
		itr.statsBuf.PointN++

		// Copy buffer to stats periodically.
		if itr.statsBuf.PointN%statsBufferCopyIntervalN == 0 {
			itr.copyStats()
		}

		return &itr.point, nil
	}
}

// copyStats copies from the itr stats buffer to the stats under lock.
func (itr *unsignedIterator) copyStats() {
	itr.statsLock.Lock()
	itr.stats = itr.statsBuf
	itr.statsLock.Unlock()
}

// Stats returns stats on the points processed.
func (itr *unsignedIterator) Stats() query.IteratorStats {
	itr.statsLock.Lock()
	stats := itr.stats
	itr.statsLock.Unlock()
	return stats
}

// Close closes the iterator.
func (itr *unsignedIterator) Close() error {
	cursorsAt(itr.aux).close()
	itr.aux = nil
	cursorsAt(itr.conds.curs).close()
	itr.conds.curs = nil
	if itr.cur != nil {
		err := itr.cur.close()
		itr.cur = nil
		return err
	}
	return nil
}

// unsignedLimitIterator
type unsignedLimitIterator struct {
	input query.UnsignedIterator
	opt   query.IteratorOptions
	n     int
}

func newUnsignedLimitIterator(input query.UnsignedIterator, opt query.IteratorOptions) *unsignedLimitIterator {
	return &unsignedLimitIterator{
		input: input,
		opt:   opt,
	}
}

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

func (itr *unsignedLimitIterator) Next() (*query.UnsignedPoint, error) {
	// Check if we are beyond the limit.
	if (itr.n - itr.opt.Offset) > itr.opt.Limit {
		return nil, nil
	}

	// Read the next point.
	p, err := itr.input.Next()
	if p == nil || err != nil {
		return nil, err
	}

	// Increment counter.
	itr.n++

	// Offsets are handled by a higher level iterator so return all points.
	return p, nil
}

// unsignedCursor represents an object for iterating over a single unsigned field.
type unsignedCursor interface {
	cursor
	nextUnsigned() (t int64, v uint64)
}

func newUnsignedCursor(seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) unsignedCursor {
	if ascending {
		return newUnsignedAscendingCursor(seek, cacheValues, tsmKeyCursor)
	}
	return newUnsignedDescendingCursor(seek, cacheValues, tsmKeyCursor)
}

type unsignedAscendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []UnsignedValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newUnsignedAscendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *unsignedAscendingCursor {
	c := &unsignedAscendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadUnsignedBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})

	return c
}

// peekCache returns the current time/value from the cache.
func (c *unsignedAscendingCursor) peekCache() (t int64, v uint64) {
	if c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, 0
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(UnsignedValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *unsignedAscendingCursor) peekTSM() (t int64, v uint64) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, 0
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *unsignedAscendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *unsignedAscendingCursor) next() (int64, interface{}) { return c.nextUnsigned() }

// nextUnsigned returns the next key/value for the cursor.
func (c *unsignedAscendingCursor) nextUnsigned() (int64, uint64) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, 0
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey < tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *unsignedAscendingCursor) nextCache() {
	if c.cache.pos >= len(c.cache.values) {
		return
	}
	c.cache.pos++
}

// nextTSM returns the next value from the TSM files.
func (c *unsignedAscendingCursor) nextTSM() {
	c.tsm.pos++
	if c.tsm.pos >= len(c.tsm.values) {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadUnsignedBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = 0
	}
}

type unsignedDescendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []UnsignedValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newUnsignedDescendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *unsignedDescendingCursor {
	c := &unsignedDescendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekCache(); t != seek {
		c.cache.pos--
	}

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadUnsignedBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekTSM(); t != seek {
		c.tsm.pos--
	}

	return c
}

// peekCache returns the current time/value from the cache.
func (c *unsignedDescendingCursor) peekCache() (t int64, v uint64) {
	if c.cache.pos < 0 || c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, 0
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(UnsignedValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *unsignedDescendingCursor) peekTSM() (t int64, v uint64) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, 0
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *unsignedDescendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *unsignedDescendingCursor) next() (int64, interface{}) { return c.nextUnsigned() }

// nextUnsigned returns the next key/value for the cursor.
func (c *unsignedDescendingCursor) nextUnsigned() (int64, uint64) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, 0
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey > tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *unsignedDescendingCursor) nextCache() {
	if c.cache.pos < 0 {
		return
	}
	c.cache.pos--
}

// nextTSM returns the next value from the TSM files.
func (c *unsignedDescendingCursor) nextTSM() {
	c.tsm.pos--
	if c.tsm.pos < 0 {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadUnsignedBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = len(c.tsm.values) - 1
	}
}

type stringFinalizerIterator struct {
	query.StringIterator
	logger *zap.Logger
}

func newStringFinalizerIterator(inner query.StringIterator, logger *zap.Logger) *stringFinalizerIterator {
	itr := &stringFinalizerIterator{StringIterator: inner, logger: logger}
	runtime.SetFinalizer(itr, (*stringFinalizerIterator).closeGC)
	return itr
}

func (itr *stringFinalizerIterator) closeGC() {
	go func() {
		itr.logger.Error("StringIterator finalized by GC")
		itr.Close()
	}()
}

func (itr *stringFinalizerIterator) Close() error {
	runtime.SetFinalizer(itr, nil)
	return itr.StringIterator.Close()
}

type stringInstrumentedIterator struct {
	query.StringIterator
	span  *tracing.Span
	group *metrics.Group
}

func newStringInstrumentedIterator(inner query.StringIterator, span *tracing.Span, group *metrics.Group) *stringInstrumentedIterator {
	return &stringInstrumentedIterator{StringIterator: inner, span: span, group: group}
}

func (itr *stringInstrumentedIterator) Close() error {
	var f fields.Fields
	itr.group.ForEach(func(v metrics.Metric) {
		switch m := v.(type) {
		case *metrics.Counter:
			f = append(f, fields.Int64(m.Name(), m.Value()))

		case *metrics.Timer:
			f = append(f, fields.Duration(m.Name(), m.Value()))

		default:
			panic("unexpected metrics")
		}
	})
	itr.span.SetFields(f)
	itr.span.Finish()

	return itr.StringIterator.Close()
}

type stringIterator struct {
	cur   stringCursor
	aux   []cursorAt
	conds struct {
		names []string
		curs  []cursorAt
	}
	opt query.IteratorOptions

	m     map[string]interface{} // map used for condition evaluation
	point query.StringPoint      // reusable buffer

	statsLock sync.Mutex
	stats     query.IteratorStats
	statsBuf  query.IteratorStats
}

func newStringIterator(name string, tags query.Tags, opt query.IteratorOptions, cur stringCursor, aux []cursorAt, conds []cursorAt, condNames []string) *stringIterator {
	itr := &stringIterator{
		cur: cur,
		aux: aux,
		opt: opt,
		point: query.StringPoint{
			Name: name,
			Tags: tags,
		},
		statsBuf: query.IteratorStats{
			SeriesN: 1,
		},
	}
	itr.stats = itr.statsBuf

	if len(aux) > 0 {
		itr.point.Aux = make([]interface{}, len(aux))
	}

	if opt.Condition != nil {
		itr.m = make(map[string]interface{}, len(aux)+len(conds))
	}
	itr.conds.names = condNames
	itr.conds.curs = conds

	return itr
}

// Next returns the next point from the iterator.
func (itr *stringIterator) Next() (*query.StringPoint, error) {
	for {
		seek := tsdb.EOF

		if itr.cur != nil {
			// Read from the main cursor if we have one.
			itr.point.Time, itr.point.Value = itr.cur.nextString()
			seek = itr.point.Time
		} else {
			// Otherwise find lowest aux timestamp.
			for i := range itr.aux {
				if k, _ := itr.aux[i].peek(); k != tsdb.EOF {
					if seek == tsdb.EOF || (itr.opt.Ascending && k < seek) || (!itr.opt.Ascending && k > seek) {
						seek = k
					}
				}
			}
			itr.point.Time = seek
		}

		// Exit if we have no more points or we are outside our time range.
		if itr.point.Time == tsdb.EOF {
			itr.copyStats()
			return nil, nil
		} else if itr.opt.Ascending && itr.point.Time > itr.opt.EndTime {
			itr.copyStats()
			return nil, nil
		} else if !itr.opt.Ascending && itr.point.Time < itr.opt.StartTime {
			itr.copyStats()
			return nil, nil
		}

		// Read from each auxiliary cursor.
		for i := range itr.opt.Aux {
			itr.point.Aux[i] = itr.aux[i].nextAt(seek)
		}

		// Read from condition field cursors.
		for i := range itr.conds.curs {
			itr.m[itr.conds.names[i]] = itr.conds.curs[i].nextAt(seek)
		}

		// Evaluate condition, if one exists. Retry if it fails.
		valuer := influxql.ValuerEval{
			Valuer: influxql.MultiValuer(
				query.MathValuer{},
				influxql.MapValuer(itr.m),
			),
		}
		if itr.opt.Condition != nil && !valuer.EvalBool(itr.opt.Condition) {
			continue
		}

		// Track points returned.
		itr.statsBuf.PointN++

		// Copy buffer to stats periodically.
		if itr.statsBuf.PointN%statsBufferCopyIntervalN == 0 {
			itr.copyStats()
		}

		return &itr.point, nil
	}
}

// copyStats copies from the itr stats buffer to the stats under lock.
func (itr *stringIterator) copyStats() {
	itr.statsLock.Lock()
	itr.stats = itr.statsBuf
	itr.statsLock.Unlock()
}

// Stats returns stats on the points processed.
func (itr *stringIterator) Stats() query.IteratorStats {
	itr.statsLock.Lock()
	stats := itr.stats
	itr.statsLock.Unlock()
	return stats
}

// Close closes the iterator.
func (itr *stringIterator) Close() error {
	cursorsAt(itr.aux).close()
	itr.aux = nil
	cursorsAt(itr.conds.curs).close()
	itr.conds.curs = nil
	if itr.cur != nil {
		err := itr.cur.close()
		itr.cur = nil
		return err
	}
	return nil
}

// stringLimitIterator
type stringLimitIterator struct {
	input query.StringIterator
	opt   query.IteratorOptions
	n     int
}

func newStringLimitIterator(input query.StringIterator, opt query.IteratorOptions) *stringLimitIterator {
	return &stringLimitIterator{
		input: input,
		opt:   opt,
	}
}

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

func (itr *stringLimitIterator) Next() (*query.StringPoint, error) {
	// Check if we are beyond the limit.
	if (itr.n - itr.opt.Offset) > itr.opt.Limit {
		return nil, nil
	}

	// Read the next point.
	p, err := itr.input.Next()
	if p == nil || err != nil {
		return nil, err
	}

	// Increment counter.
	itr.n++

	// Offsets are handled by a higher level iterator so return all points.
	return p, nil
}

// stringCursor represents an object for iterating over a single string field.
type stringCursor interface {
	cursor
	nextString() (t int64, v string)
}

func newStringCursor(seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) stringCursor {
	if ascending {
		return newStringAscendingCursor(seek, cacheValues, tsmKeyCursor)
	}
	return newStringDescendingCursor(seek, cacheValues, tsmKeyCursor)
}

type stringAscendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []StringValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newStringAscendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *stringAscendingCursor {
	c := &stringAscendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadStringBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})

	return c
}

// peekCache returns the current time/value from the cache.
func (c *stringAscendingCursor) peekCache() (t int64, v string) {
	if c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, ""
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(StringValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *stringAscendingCursor) peekTSM() (t int64, v string) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, ""
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *stringAscendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *stringAscendingCursor) next() (int64, interface{}) { return c.nextString() }

// nextString returns the next key/value for the cursor.
func (c *stringAscendingCursor) nextString() (int64, string) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, ""
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey < tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *stringAscendingCursor) nextCache() {
	if c.cache.pos >= len(c.cache.values) {
		return
	}
	c.cache.pos++
}

// nextTSM returns the next value from the TSM files.
func (c *stringAscendingCursor) nextTSM() {
	c.tsm.pos++
	if c.tsm.pos >= len(c.tsm.values) {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadStringBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = 0
	}
}

type stringDescendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []StringValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newStringDescendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *stringDescendingCursor {
	c := &stringDescendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekCache(); t != seek {
		c.cache.pos--
	}

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadStringBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekTSM(); t != seek {
		c.tsm.pos--
	}

	return c
}

// peekCache returns the current time/value from the cache.
func (c *stringDescendingCursor) peekCache() (t int64, v string) {
	if c.cache.pos < 0 || c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, ""
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(StringValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *stringDescendingCursor) peekTSM() (t int64, v string) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, ""
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *stringDescendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *stringDescendingCursor) next() (int64, interface{}) { return c.nextString() }

// nextString returns the next key/value for the cursor.
func (c *stringDescendingCursor) nextString() (int64, string) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, ""
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey > tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *stringDescendingCursor) nextCache() {
	if c.cache.pos < 0 {
		return
	}
	c.cache.pos--
}

// nextTSM returns the next value from the TSM files.
func (c *stringDescendingCursor) nextTSM() {
	c.tsm.pos--
	if c.tsm.pos < 0 {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadStringBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = len(c.tsm.values) - 1
	}
}

type booleanFinalizerIterator struct {
	query.BooleanIterator
	logger *zap.Logger
}

func newBooleanFinalizerIterator(inner query.BooleanIterator, logger *zap.Logger) *booleanFinalizerIterator {
	itr := &booleanFinalizerIterator{BooleanIterator: inner, logger: logger}
	runtime.SetFinalizer(itr, (*booleanFinalizerIterator).closeGC)
	return itr
}

func (itr *booleanFinalizerIterator) closeGC() {
	go func() {
		itr.logger.Error("BooleanIterator finalized by GC")
		itr.Close()
	}()
}

func (itr *booleanFinalizerIterator) Close() error {
	runtime.SetFinalizer(itr, nil)
	return itr.BooleanIterator.Close()
}

type booleanInstrumentedIterator struct {
	query.BooleanIterator
	span  *tracing.Span
	group *metrics.Group
}

func newBooleanInstrumentedIterator(inner query.BooleanIterator, span *tracing.Span, group *metrics.Group) *booleanInstrumentedIterator {
	return &booleanInstrumentedIterator{BooleanIterator: inner, span: span, group: group}
}

func (itr *booleanInstrumentedIterator) Close() error {
	var f fields.Fields
	itr.group.ForEach(func(v metrics.Metric) {
		switch m := v.(type) {
		case *metrics.Counter:
			f = append(f, fields.Int64(m.Name(), m.Value()))

		case *metrics.Timer:
			f = append(f, fields.Duration(m.Name(), m.Value()))

		default:
			panic("unexpected metrics")
		}
	})
	itr.span.SetFields(f)
	itr.span.Finish()

	return itr.BooleanIterator.Close()
}

type booleanIterator struct {
	cur   booleanCursor
	aux   []cursorAt
	conds struct {
		names []string
		curs  []cursorAt
	}
	opt query.IteratorOptions

	m     map[string]interface{} // map used for condition evaluation
	point query.BooleanPoint     // reusable buffer

	statsLock sync.Mutex
	stats     query.IteratorStats
	statsBuf  query.IteratorStats
}

func newBooleanIterator(name string, tags query.Tags, opt query.IteratorOptions, cur booleanCursor, aux []cursorAt, conds []cursorAt, condNames []string) *booleanIterator {
	itr := &booleanIterator{
		cur: cur,
		aux: aux,
		opt: opt,
		point: query.BooleanPoint{
			Name: name,
			Tags: tags,
		},
		statsBuf: query.IteratorStats{
			SeriesN: 1,
		},
	}
	itr.stats = itr.statsBuf

	if len(aux) > 0 {
		itr.point.Aux = make([]interface{}, len(aux))
	}

	if opt.Condition != nil {
		itr.m = make(map[string]interface{}, len(aux)+len(conds))
	}
	itr.conds.names = condNames
	itr.conds.curs = conds

	return itr
}

// Next returns the next point from the iterator.
func (itr *booleanIterator) Next() (*query.BooleanPoint, error) {
	for {
		seek := tsdb.EOF

		if itr.cur != nil {
			// Read from the main cursor if we have one.
			itr.point.Time, itr.point.Value = itr.cur.nextBoolean()
			seek = itr.point.Time
		} else {
			// Otherwise find lowest aux timestamp.
			for i := range itr.aux {
				if k, _ := itr.aux[i].peek(); k != tsdb.EOF {
					if seek == tsdb.EOF || (itr.opt.Ascending && k < seek) || (!itr.opt.Ascending && k > seek) {
						seek = k
					}
				}
			}
			itr.point.Time = seek
		}

		// Exit if we have no more points or we are outside our time range.
		if itr.point.Time == tsdb.EOF {
			itr.copyStats()
			return nil, nil
		} else if itr.opt.Ascending && itr.point.Time > itr.opt.EndTime {
			itr.copyStats()
			return nil, nil
		} else if !itr.opt.Ascending && itr.point.Time < itr.opt.StartTime {
			itr.copyStats()
			return nil, nil
		}

		// Read from each auxiliary cursor.
		for i := range itr.opt.Aux {
			itr.point.Aux[i] = itr.aux[i].nextAt(seek)
		}

		// Read from condition field cursors.
		for i := range itr.conds.curs {
			itr.m[itr.conds.names[i]] = itr.conds.curs[i].nextAt(seek)
		}

		// Evaluate condition, if one exists. Retry if it fails.
		valuer := influxql.ValuerEval{
			Valuer: influxql.MultiValuer(
				query.MathValuer{},
				influxql.MapValuer(itr.m),
			),
		}
		if itr.opt.Condition != nil && !valuer.EvalBool(itr.opt.Condition) {
			continue
		}

		// Track points returned.
		itr.statsBuf.PointN++

		// Copy buffer to stats periodically.
		if itr.statsBuf.PointN%statsBufferCopyIntervalN == 0 {
			itr.copyStats()
		}

		return &itr.point, nil
	}
}

// copyStats copies from the itr stats buffer to the stats under lock.
func (itr *booleanIterator) copyStats() {
	itr.statsLock.Lock()
	itr.stats = itr.statsBuf
	itr.statsLock.Unlock()
}

// Stats returns stats on the points processed.
func (itr *booleanIterator) Stats() query.IteratorStats {
	itr.statsLock.Lock()
	stats := itr.stats
	itr.statsLock.Unlock()
	return stats
}

// Close closes the iterator.
func (itr *booleanIterator) Close() error {
	cursorsAt(itr.aux).close()
	itr.aux = nil
	cursorsAt(itr.conds.curs).close()
	itr.conds.curs = nil
	if itr.cur != nil {
		err := itr.cur.close()
		itr.cur = nil
		return err
	}
	return nil
}

// booleanLimitIterator
type booleanLimitIterator struct {
	input query.BooleanIterator
	opt   query.IteratorOptions
	n     int
}

func newBooleanLimitIterator(input query.BooleanIterator, opt query.IteratorOptions) *booleanLimitIterator {
	return &booleanLimitIterator{
		input: input,
		opt:   opt,
	}
}

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

func (itr *booleanLimitIterator) Next() (*query.BooleanPoint, error) {
	// Check if we are beyond the limit.
	if (itr.n - itr.opt.Offset) > itr.opt.Limit {
		return nil, nil
	}

	// Read the next point.
	p, err := itr.input.Next()
	if p == nil || err != nil {
		return nil, err
	}

	// Increment counter.
	itr.n++

	// Offsets are handled by a higher level iterator so return all points.
	return p, nil
}

// booleanCursor represents an object for iterating over a single boolean field.
type booleanCursor interface {
	cursor
	nextBoolean() (t int64, v bool)
}

func newBooleanCursor(seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) booleanCursor {
	if ascending {
		return newBooleanAscendingCursor(seek, cacheValues, tsmKeyCursor)
	}
	return newBooleanDescendingCursor(seek, cacheValues, tsmKeyCursor)
}

type booleanAscendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []BooleanValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newBooleanAscendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *booleanAscendingCursor {
	c := &booleanAscendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadBooleanBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})

	return c
}

// peekCache returns the current time/value from the cache.
func (c *booleanAscendingCursor) peekCache() (t int64, v bool) {
	if c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, false
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(BooleanValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *booleanAscendingCursor) peekTSM() (t int64, v bool) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, false
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *booleanAscendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *booleanAscendingCursor) next() (int64, interface{}) { return c.nextBoolean() }

// nextBoolean returns the next key/value for the cursor.
func (c *booleanAscendingCursor) nextBoolean() (int64, bool) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, false
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey < tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *booleanAscendingCursor) nextCache() {
	if c.cache.pos >= len(c.cache.values) {
		return
	}
	c.cache.pos++
}

// nextTSM returns the next value from the TSM files.
func (c *booleanAscendingCursor) nextTSM() {
	c.tsm.pos++
	if c.tsm.pos >= len(c.tsm.values) {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadBooleanBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = 0
	}
}

type booleanDescendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		values    []BooleanValue
		pos       int
		keyCursor *KeyCursor
	}
}

func newBooleanDescendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *booleanDescendingCursor {
	c := &booleanDescendingCursor{}

	c.cache.values = cacheValues
	c.cache.pos = sort.Search(len(c.cache.values), func(i int) bool {
		return c.cache.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekCache(); t != seek {
		c.cache.pos--
	}

	c.tsm.keyCursor = tsmKeyCursor
	c.tsm.values, _ = c.tsm.keyCursor.ReadBooleanBlock(&c.tsm.values)
	c.tsm.pos = sort.Search(len(c.tsm.values), func(i int) bool {
		return c.tsm.values[i].UnixNano() >= seek
	})
	if t, _ := c.peekTSM(); t != seek {
		c.tsm.pos--
	}

	return c
}

// peekCache returns the current time/value from the cache.
func (c *booleanDescendingCursor) peekCache() (t int64, v bool) {
	if c.cache.pos < 0 || c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, false
	}

	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(BooleanValue).value
}

// peekTSM returns the current time/value from tsm.
func (c *booleanDescendingCursor) peekTSM() (t int64, v bool) {
	if c.tsm.pos < 0 || c.tsm.pos >= len(c.tsm.values) {
		return tsdb.EOF, false
	}

	item := c.tsm.values[c.tsm.pos]
	return item.UnixNano(), item.value
}

// close closes the cursor and any dependent cursors.
func (c *booleanDescendingCursor) close() error {
	if c.tsm.keyCursor == nil {
		return nil
	}

	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
	return nil
}

// next returns the next key/value for the cursor.
func (c *booleanDescendingCursor) next() (int64, interface{}) { return c.nextBoolean() }

// nextBoolean returns the next key/value for the cursor.
func (c *booleanDescendingCursor) nextBoolean() (int64, bool) {
	ckey, cvalue := c.peekCache()
	tkey, tvalue := c.peekTSM()

	// No more data in cache or in TSM files.
	if ckey == tsdb.EOF && tkey == tsdb.EOF {
		return tsdb.EOF, false
	}

	// Both cache and tsm files have the same key, cache takes precedence.
	if ckey == tkey {
		c.nextCache()
		c.nextTSM()
		return ckey, cvalue
	}

	// Buffered cache key precedes that in TSM file.
	if ckey != tsdb.EOF && (ckey > tkey || tkey == tsdb.EOF) {
		c.nextCache()
		return ckey, cvalue
	}

	// Buffered TSM key precedes that in cache.
	c.nextTSM()
	return tkey, tvalue
}

// nextCache returns the next value from the cache.
func (c *booleanDescendingCursor) nextCache() {
	if c.cache.pos < 0 {
		return
	}
	c.cache.pos--
}

// nextTSM returns the next value from the TSM files.
func (c *booleanDescendingCursor) nextTSM() {
	c.tsm.pos--
	if c.tsm.pos < 0 {
		c.tsm.keyCursor.Next()
		c.tsm.values, _ = c.tsm.keyCursor.ReadBooleanBlock(&c.tsm.values)
		if len(c.tsm.values) == 0 {
			return
		}
		c.tsm.pos = len(c.tsm.values) - 1
	}
}

var _ = fmt.Print