batch_cursor.gen.go

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

package tsm1

import (
	"context"
	"sort"

	"github.com/influxdata/influxdb/query"
	"github.com/influxdata/influxdb/tsdb"
)

// buildFloatCursor creates a cursor for a float field.
func (e *Engine) buildFloatCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) floatCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newFloatCursor(opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildFloatBatchCursor creates a batch cursor for a float field.
func (e *Engine) buildFloatBatchCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) tsdb.FloatBatchCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newFloatBatchCursor(seriesKey, opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildIntegerCursor creates a cursor for a integer field.
func (e *Engine) buildIntegerCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) integerCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newIntegerCursor(opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildIntegerBatchCursor creates a batch cursor for a integer field.
func (e *Engine) buildIntegerBatchCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) tsdb.IntegerBatchCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newIntegerBatchCursor(seriesKey, opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildUnsignedCursor creates a cursor for a unsigned field.
func (e *Engine) buildUnsignedCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) unsignedCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newUnsignedCursor(opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildUnsignedBatchCursor creates a batch cursor for a unsigned field.
func (e *Engine) buildUnsignedBatchCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) tsdb.UnsignedBatchCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newUnsignedBatchCursor(seriesKey, opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildStringCursor creates a cursor for a string field.
func (e *Engine) buildStringCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) stringCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newStringCursor(opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildStringBatchCursor creates a batch cursor for a string field.
func (e *Engine) buildStringBatchCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) tsdb.StringBatchCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newStringBatchCursor(seriesKey, opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildBooleanCursor creates a cursor for a boolean field.
func (e *Engine) buildBooleanCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) booleanCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newBooleanCursor(opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// buildBooleanBatchCursor creates a batch cursor for a boolean field.
func (e *Engine) buildBooleanBatchCursor(ctx context.Context, measurement, seriesKey, field string, opt query.IteratorOptions) tsdb.BooleanBatchCursor {
	key := SeriesFieldKeyBytes(seriesKey, field)
	cacheValues := e.Cache.Values(key)
	keyCursor := e.KeyCursor(ctx, key, opt.SeekTime(), opt.Ascending)
	return newBooleanBatchCursor(seriesKey, opt.SeekTime(), opt.Ascending, cacheValues, keyCursor)
}

// Cursors

func newFloatBatchCursor(key string, seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) tsdb.FloatBatchCursor {
	if ascending {
		return newFloatAscendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
	}
	return newFloatDescendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
}

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

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

	key string
	t   []int64
	v   []float64
}

func newFloatAscendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *floatAscendingBatchCursor {
	c := &floatAscendingBatchCursor{key: key}

	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
	})

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]float64, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *floatAscendingBatchCursor) Err() error        { return nil }
func (c *floatAscendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *floatAscendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// Next returns the next key/value for the cursor.
func (c *floatAscendingBatchCursor) Next() ([]int64, []float64) {
	var ckey, tkey int64
	var cvalue, tvalue float64

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos < len(c.cache.values) {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey < tkey || tkey == tsdb.EOF {
				// Buffered cache key precedes that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key precedes that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *floatAscendingBatchCursor) peekCache() (t int64, v float64) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(FloatValue).value
}

// nextCache returns the next value from the cache.
func (c *floatAscendingBatchCursor) nextCache() {
	if c.cache.pos < len(c.cache.values) {
		c.cache.pos++
	}
}

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

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

// nextTSM returns the next value from the TSM files.
func (c *floatAscendingBatchCursor) 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)
		c.tsm.pos = 0
	}
}

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

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

	key string
	t   []int64
	v   []float64
}

func newFloatDescendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *floatDescendingBatchCursor {
	c := &floatDescendingBatchCursor{key: key}

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

	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 len(c.tsm.values) > 0 {
		if c.tsm.pos == len(c.tsm.values) {
			c.tsm.pos--
		} else if t, _ := c.peekTSM(); t != seek {
			c.tsm.pos--
		}
	} else {
		c.tsm.pos = -1
	}

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]float64, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *floatDescendingBatchCursor) Err() error        { return nil }
func (c *floatDescendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *floatDescendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// nextFloat returns the next key/value for the cursor.
func (c *floatDescendingBatchCursor) Next() ([]int64, []float64) {
	var ckey, tkey int64
	var cvalue, tvalue float64

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos >= 0 {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey > tkey || tkey == tsdb.EOF {
				// Buffered cache key succeeds that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key succeeds that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *floatDescendingBatchCursor) peekCache() (t int64, v float64) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(FloatValue).value
}

// nextCache returns the next value from the cache.
func (c *floatDescendingBatchCursor) nextCache() {
	if c.cache.pos >= 0 {
		c.cache.pos--
	}
}

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

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

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

func newFloatRangeBatchCursor(time int64, asc bool, cur tsdb.FloatBatchCursor) tsdb.FloatBatchCursor {
	if asc {
		return &floatAscendingRangeBatchCursor{FloatBatchCursor: cur, t: time}
	}
	return &floatDescendingRangeBatchCursor{FloatBatchCursor: cur, t: time}
}

type floatAscendingRangeBatchCursor struct {
	tsdb.FloatBatchCursor
	t int64
}

func (l *floatAscendingRangeBatchCursor) Next() ([]int64, []float64) {
	k, v := l.FloatBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] > l.t {
		i := len(k) - 2
		for i >= 0 && k[i] > l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

type floatDescendingRangeBatchCursor struct {
	tsdb.FloatBatchCursor
	t int64
}

func (l *floatDescendingRangeBatchCursor) Next() ([]int64, []float64) {
	k, v := l.FloatBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] < l.t {
		i := len(k) - 2
		for i >= 0 && k[i] < l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

func newIntegerBatchCursor(key string, seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) tsdb.IntegerBatchCursor {
	if ascending {
		return newIntegerAscendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
	}
	return newIntegerDescendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
}

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

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

	key string
	t   []int64
	v   []int64
}

func newIntegerAscendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *integerAscendingBatchCursor {
	c := &integerAscendingBatchCursor{key: key}

	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
	})

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]int64, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *integerAscendingBatchCursor) Err() error        { return nil }
func (c *integerAscendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *integerAscendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// Next returns the next key/value for the cursor.
func (c *integerAscendingBatchCursor) Next() ([]int64, []int64) {
	var ckey, tkey int64
	var cvalue, tvalue int64

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos < len(c.cache.values) {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey < tkey || tkey == tsdb.EOF {
				// Buffered cache key precedes that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key precedes that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *integerAscendingBatchCursor) peekCache() (t int64, v int64) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(IntegerValue).value
}

// nextCache returns the next value from the cache.
func (c *integerAscendingBatchCursor) nextCache() {
	if c.cache.pos < len(c.cache.values) {
		c.cache.pos++
	}
}

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

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

// nextTSM returns the next value from the TSM files.
func (c *integerAscendingBatchCursor) 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)
		c.tsm.pos = 0
	}
}

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

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

	key string
	t   []int64
	v   []int64
}

func newIntegerDescendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *integerDescendingBatchCursor {
	c := &integerDescendingBatchCursor{key: key}

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

	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 len(c.tsm.values) > 0 {
		if c.tsm.pos == len(c.tsm.values) {
			c.tsm.pos--
		} else if t, _ := c.peekTSM(); t != seek {
			c.tsm.pos--
		}
	} else {
		c.tsm.pos = -1
	}

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]int64, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *integerDescendingBatchCursor) Err() error        { return nil }
func (c *integerDescendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *integerDescendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// nextInteger returns the next key/value for the cursor.
func (c *integerDescendingBatchCursor) Next() ([]int64, []int64) {
	var ckey, tkey int64
	var cvalue, tvalue int64

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos >= 0 {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey > tkey || tkey == tsdb.EOF {
				// Buffered cache key succeeds that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key succeeds that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *integerDescendingBatchCursor) peekCache() (t int64, v int64) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(IntegerValue).value
}

// nextCache returns the next value from the cache.
func (c *integerDescendingBatchCursor) nextCache() {
	if c.cache.pos >= 0 {
		c.cache.pos--
	}
}

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

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

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

func newIntegerRangeBatchCursor(time int64, asc bool, cur tsdb.IntegerBatchCursor) tsdb.IntegerBatchCursor {
	if asc {
		return &integerAscendingRangeBatchCursor{IntegerBatchCursor: cur, t: time}
	}
	return &integerDescendingRangeBatchCursor{IntegerBatchCursor: cur, t: time}
}

type integerAscendingRangeBatchCursor struct {
	tsdb.IntegerBatchCursor
	t int64
}

func (l *integerAscendingRangeBatchCursor) Next() ([]int64, []int64) {
	k, v := l.IntegerBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] > l.t {
		i := len(k) - 2
		for i >= 0 && k[i] > l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

type integerDescendingRangeBatchCursor struct {
	tsdb.IntegerBatchCursor
	t int64
}

func (l *integerDescendingRangeBatchCursor) Next() ([]int64, []int64) {
	k, v := l.IntegerBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] < l.t {
		i := len(k) - 2
		for i >= 0 && k[i] < l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

func newUnsignedBatchCursor(key string, seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) tsdb.UnsignedBatchCursor {
	if ascending {
		return newUnsignedAscendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
	}
	return newUnsignedDescendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
}

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

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

	key string
	t   []int64
	v   []uint64
}

func newUnsignedAscendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *unsignedAscendingBatchCursor {
	c := &unsignedAscendingBatchCursor{key: key}

	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
	})

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]uint64, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *unsignedAscendingBatchCursor) Err() error        { return nil }
func (c *unsignedAscendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *unsignedAscendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// Next returns the next key/value for the cursor.
func (c *unsignedAscendingBatchCursor) Next() ([]int64, []uint64) {
	var ckey, tkey int64
	var cvalue, tvalue uint64

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos < len(c.cache.values) {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey < tkey || tkey == tsdb.EOF {
				// Buffered cache key precedes that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key precedes that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *unsignedAscendingBatchCursor) peekCache() (t int64, v uint64) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(UnsignedValue).value
}

// nextCache returns the next value from the cache.
func (c *unsignedAscendingBatchCursor) nextCache() {
	if c.cache.pos < len(c.cache.values) {
		c.cache.pos++
	}
}

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

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

// nextTSM returns the next value from the TSM files.
func (c *unsignedAscendingBatchCursor) 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)
		c.tsm.pos = 0
	}
}

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

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

	key string
	t   []int64
	v   []uint64
}

func newUnsignedDescendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *unsignedDescendingBatchCursor {
	c := &unsignedDescendingBatchCursor{key: key}

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

	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 len(c.tsm.values) > 0 {
		if c.tsm.pos == len(c.tsm.values) {
			c.tsm.pos--
		} else if t, _ := c.peekTSM(); t != seek {
			c.tsm.pos--
		}
	} else {
		c.tsm.pos = -1
	}

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]uint64, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *unsignedDescendingBatchCursor) Err() error        { return nil }
func (c *unsignedDescendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *unsignedDescendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// nextUnsigned returns the next key/value for the cursor.
func (c *unsignedDescendingBatchCursor) Next() ([]int64, []uint64) {
	var ckey, tkey int64
	var cvalue, tvalue uint64

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos >= 0 {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey > tkey || tkey == tsdb.EOF {
				// Buffered cache key succeeds that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key succeeds that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *unsignedDescendingBatchCursor) peekCache() (t int64, v uint64) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(UnsignedValue).value
}

// nextCache returns the next value from the cache.
func (c *unsignedDescendingBatchCursor) nextCache() {
	if c.cache.pos >= 0 {
		c.cache.pos--
	}
}

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

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

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

func newUnsignedRangeBatchCursor(time int64, asc bool, cur tsdb.UnsignedBatchCursor) tsdb.UnsignedBatchCursor {
	if asc {
		return &unsignedAscendingRangeBatchCursor{UnsignedBatchCursor: cur, t: time}
	}
	return &unsignedDescendingRangeBatchCursor{UnsignedBatchCursor: cur, t: time}
}

type unsignedAscendingRangeBatchCursor struct {
	tsdb.UnsignedBatchCursor
	t int64
}

func (l *unsignedAscendingRangeBatchCursor) Next() ([]int64, []uint64) {
	k, v := l.UnsignedBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] > l.t {
		i := len(k) - 2
		for i >= 0 && k[i] > l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

type unsignedDescendingRangeBatchCursor struct {
	tsdb.UnsignedBatchCursor
	t int64
}

func (l *unsignedDescendingRangeBatchCursor) Next() ([]int64, []uint64) {
	k, v := l.UnsignedBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] < l.t {
		i := len(k) - 2
		for i >= 0 && k[i] < l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

func newStringBatchCursor(key string, seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) tsdb.StringBatchCursor {
	if ascending {
		return newStringAscendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
	}
	return newStringDescendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
}

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

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

	key string
	t   []int64
	v   []string
}

func newStringAscendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *stringAscendingBatchCursor {
	c := &stringAscendingBatchCursor{key: key}

	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
	})

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]string, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *stringAscendingBatchCursor) Err() error        { return nil }
func (c *stringAscendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *stringAscendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// Next returns the next key/value for the cursor.
func (c *stringAscendingBatchCursor) Next() ([]int64, []string) {
	var ckey, tkey int64
	var cvalue, tvalue string

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos < len(c.cache.values) {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey < tkey || tkey == tsdb.EOF {
				// Buffered cache key precedes that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key precedes that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *stringAscendingBatchCursor) peekCache() (t int64, v string) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(StringValue).value
}

// nextCache returns the next value from the cache.
func (c *stringAscendingBatchCursor) nextCache() {
	if c.cache.pos < len(c.cache.values) {
		c.cache.pos++
	}
}

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

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

// nextTSM returns the next value from the TSM files.
func (c *stringAscendingBatchCursor) 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)
		c.tsm.pos = 0
	}
}

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

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

	key string
	t   []int64
	v   []string
}

func newStringDescendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *stringDescendingBatchCursor {
	c := &stringDescendingBatchCursor{key: key}

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

	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 len(c.tsm.values) > 0 {
		if c.tsm.pos == len(c.tsm.values) {
			c.tsm.pos--
		} else if t, _ := c.peekTSM(); t != seek {
			c.tsm.pos--
		}
	} else {
		c.tsm.pos = -1
	}

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]string, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *stringDescendingBatchCursor) Err() error        { return nil }
func (c *stringDescendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *stringDescendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// nextString returns the next key/value for the cursor.
func (c *stringDescendingBatchCursor) Next() ([]int64, []string) {
	var ckey, tkey int64
	var cvalue, tvalue string

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos >= 0 {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey > tkey || tkey == tsdb.EOF {
				// Buffered cache key succeeds that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key succeeds that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *stringDescendingBatchCursor) peekCache() (t int64, v string) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(StringValue).value
}

// nextCache returns the next value from the cache.
func (c *stringDescendingBatchCursor) nextCache() {
	if c.cache.pos >= 0 {
		c.cache.pos--
	}
}

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

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

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

func newStringRangeBatchCursor(time int64, asc bool, cur tsdb.StringBatchCursor) tsdb.StringBatchCursor {
	if asc {
		return &stringAscendingRangeBatchCursor{StringBatchCursor: cur, t: time}
	}
	return &stringDescendingRangeBatchCursor{StringBatchCursor: cur, t: time}
}

type stringAscendingRangeBatchCursor struct {
	tsdb.StringBatchCursor
	t int64
}

func (l *stringAscendingRangeBatchCursor) Next() ([]int64, []string) {
	k, v := l.StringBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] > l.t {
		i := len(k) - 2
		for i >= 0 && k[i] > l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

type stringDescendingRangeBatchCursor struct {
	tsdb.StringBatchCursor
	t int64
}

func (l *stringDescendingRangeBatchCursor) Next() ([]int64, []string) {
	k, v := l.StringBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] < l.t {
		i := len(k) - 2
		for i >= 0 && k[i] < l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

func newBooleanBatchCursor(key string, seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) tsdb.BooleanBatchCursor {
	if ascending {
		return newBooleanAscendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
	}
	return newBooleanDescendingBatchCursor(key, seek, cacheValues, tsmKeyCursor)
}

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

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

	key string
	t   []int64
	v   []bool
}

func newBooleanAscendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *booleanAscendingBatchCursor {
	c := &booleanAscendingBatchCursor{key: key}

	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
	})

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]bool, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *booleanAscendingBatchCursor) Err() error        { return nil }
func (c *booleanAscendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *booleanAscendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// Next returns the next key/value for the cursor.
func (c *booleanAscendingBatchCursor) Next() ([]int64, []bool) {
	var ckey, tkey int64
	var cvalue, tvalue bool

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos < len(c.cache.values) {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey < tkey || tkey == tsdb.EOF {
				// Buffered cache key precedes that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key precedes that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *booleanAscendingBatchCursor) peekCache() (t int64, v bool) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(BooleanValue).value
}

// nextCache returns the next value from the cache.
func (c *booleanAscendingBatchCursor) nextCache() {
	if c.cache.pos < len(c.cache.values) {
		c.cache.pos++
	}
}

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

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

// nextTSM returns the next value from the TSM files.
func (c *booleanAscendingBatchCursor) 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)
		c.tsm.pos = 0
	}
}

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

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

	key string
	t   []int64
	v   []bool
}

func newBooleanDescendingBatchCursor(key string, seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *booleanDescendingBatchCursor {
	c := &booleanDescendingBatchCursor{key: key}

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

	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 len(c.tsm.values) > 0 {
		if c.tsm.pos == len(c.tsm.values) {
			c.tsm.pos--
		} else if t, _ := c.peekTSM(); t != seek {
			c.tsm.pos--
		}
	} else {
		c.tsm.pos = -1
	}

	c.t = make([]int64, tsdb.DefaultMaxPointsPerBlock)
	c.v = make([]bool, tsdb.DefaultMaxPointsPerBlock)

	return c
}

func (c *booleanDescendingBatchCursor) Err() error        { return nil }
func (c *booleanDescendingBatchCursor) SeriesKey() string { return c.key }

// close closes the cursor and any dependent cursors.
func (c *booleanDescendingBatchCursor) Close() {
	c.tsm.keyCursor.Close()
	c.tsm.keyCursor = nil
	c.cache.values = nil
	c.tsm.values = nil
}

// nextBoolean returns the next key/value for the cursor.
func (c *booleanDescendingBatchCursor) Next() ([]int64, []bool) {
	var ckey, tkey int64
	var cvalue, tvalue bool

	pos := 0
	for ; pos < cap(c.t); pos++ {
		tkey, tvalue = c.peekTSM()

		if c.cache.pos >= 0 {
			ckey, cvalue = c.peekCache()

			var cache, tsm bool

			// Both cache and tsm files have the same key, cache takes precedence.
			if ckey == tkey {
				cache, tsm = true, true
				tkey = ckey
				tvalue = cvalue
			} else if ckey > tkey || tkey == tsdb.EOF {
				// Buffered cache key succeeds that in TSM file.
				cache = true
				tkey = ckey
				tvalue = cvalue
			} else {
				// Buffered TSM key succeeds that in cache.
				tsm = true
			}

			if cache {
				c.nextCache()
			}

			if tsm {
				c.nextTSM()
			}
		} else {
			if tkey == tsdb.EOF {
				break
			}
			c.nextTSM()
		}

		c.t[pos] = tkey
		c.v[pos] = tvalue
	}

	return c.t[:pos], c.v[:pos]
}

// peekCache returns the current time/value from the cache.
func (c *booleanDescendingBatchCursor) peekCache() (t int64, v bool) {
	item := c.cache.values[c.cache.pos]
	return item.UnixNano(), item.(BooleanValue).value
}

// nextCache returns the next value from the cache.
func (c *booleanDescendingBatchCursor) nextCache() {
	if c.cache.pos >= 0 {
		c.cache.pos--
	}
}

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

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

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

func newBooleanRangeBatchCursor(time int64, asc bool, cur tsdb.BooleanBatchCursor) tsdb.BooleanBatchCursor {
	if asc {
		return &booleanAscendingRangeBatchCursor{BooleanBatchCursor: cur, t: time}
	}
	return &booleanDescendingRangeBatchCursor{BooleanBatchCursor: cur, t: time}
}

type booleanAscendingRangeBatchCursor struct {
	tsdb.BooleanBatchCursor
	t int64
}

func (l *booleanAscendingRangeBatchCursor) Next() ([]int64, []bool) {
	k, v := l.BooleanBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] > l.t {
		i := len(k) - 2
		for i >= 0 && k[i] > l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}

type booleanDescendingRangeBatchCursor struct {
	tsdb.BooleanBatchCursor
	t int64
}

func (l *booleanDescendingRangeBatchCursor) Next() ([]int64, []bool) {
	k, v := l.BooleanBatchCursor.Next()

	// strip out remaining time that is outside the range
	if len(k) > 0 && k[len(k)-1] < l.t {
		i := len(k) - 2
		for i >= 0 && k[i] < l.t {
			i--
		}
		k = k[:i+1]
		v = v[:i+1]
	}

	return k, v
}