batch_cursor.gen.go

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

package storage

import (
	"context"
	"errors"

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

// ********************
// Float BatchCursor

type floatFilterBatchCursor struct {
	tsdb.FloatBatchCursor
	cond expression
	m    *singleValue
	t    []int64
	v    []float64
	tb   []int64
	vb   []float64
}

func newFloatFilterBatchCursor(cond expression) *floatFilterBatchCursor {
	return &floatFilterBatchCursor{
		cond: cond,
		m:    &singleValue{},
		t:    make([]int64, tsdb.DefaultMaxPointsPerBlock),
		v:    make([]float64, tsdb.DefaultMaxPointsPerBlock),
	}
}

func (c *floatFilterBatchCursor) reset(cur tsdb.FloatBatchCursor) {
	c.FloatBatchCursor = cur
	c.tb, c.vb = nil, nil
}

func (c *floatFilterBatchCursor) Next() (key []int64, value []float64) {
	pos := 0
	var ks []int64
	var vs []float64

	if len(c.tb) > 0 {
		ks, vs = c.tb, c.vb
		c.tb, c.vb = nil, nil
	} else {
		ks, vs = c.FloatBatchCursor.Next()
	}

	for len(ks) > 0 {
		for i, v := range vs {
			c.m.v = v
			if c.cond.EvalBool(c.m) {
				c.t[pos], c.v[pos] = ks[i], v
				pos++
				if pos >= tsdb.DefaultMaxPointsPerBlock {
					c.tb, c.vb = ks[i+1:], vs[i+1:]
					return c.t[:pos], c.v[:pos]
				}
			}
		}
		ks, vs = c.FloatBatchCursor.Next()
	}

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

type floatMultiShardBatchCursor struct {
	tsdb.FloatBatchCursor
	ctx    context.Context
	filter *floatFilterBatchCursor
	shards []*tsdb.Shard
	req    *tsdb.CursorRequest
	err    error
	limit  uint64
	count  uint64
}

func newFloatMultiShardBatchCursor(ctx context.Context, cur tsdb.FloatBatchCursor, rr *readRequest, req *tsdb.CursorRequest, shards []*tsdb.Shard, cond expression) *floatMultiShardBatchCursor {
	var filter *floatFilterBatchCursor
	if cond != nil {
		filter = newFloatFilterBatchCursor(cond)
		filter.reset(cur)
		cur = filter
	}

	c := &floatMultiShardBatchCursor{
		FloatBatchCursor: cur,
		ctx:              ctx,
		filter:           filter,
		req:              req,
		shards:           shards,
		limit:            rr.limit,
	}

	return c
}

func (c *floatMultiShardBatchCursor) Err() error        { return c.err }
func (c *floatMultiShardBatchCursor) SeriesKey() string { return c.req.Series }

func (c *floatMultiShardBatchCursor) Next() (key []int64, value []float64) {
	for {
		ks, vs := c.FloatBatchCursor.Next()
		if len(ks) == 0 {
			if c.nextBatchCursor() {
				continue
			}
		}
		c.count += uint64(len(ks))
		if c.count > c.limit {
			diff := c.count - c.limit
			c.count -= diff
			rem := uint64(len(ks)) - diff
			ks = ks[:rem]
			vs = vs[:rem]
		}
		return ks, vs
	}
}

func (c *floatMultiShardBatchCursor) nextBatchCursor() bool {
	if len(c.shards) == 0 {
		return false
	}

	c.FloatBatchCursor.Close()

	var shard *tsdb.Shard
	var cur tsdb.Cursor
	for cur == nil && len(c.shards) > 0 {
		shard, c.shards = c.shards[0], c.shards[1:]
		cur, _ = shard.CreateCursor(c.ctx, c.req)
	}

	var ok bool
	if cur != nil {
		var next tsdb.FloatBatchCursor
		next, ok = cur.(tsdb.FloatBatchCursor)
		if !ok {
			cur.Close()
			next = FloatEmptyBatchCursor
			c.shards = nil
			c.err = errors.New("expected float cursor")
		} else {
			if c.filter != nil {
				c.filter.reset(next)
				next = c.filter
			}
		}
		c.FloatBatchCursor = next
	} else {
		c.FloatBatchCursor = FloatEmptyBatchCursor
	}

	return ok
}

type floatSumBatchCursor struct {
	tsdb.FloatBatchCursor
}

func (c *floatSumBatchCursor) Next() (key []int64, value []float64) {
	ks, vs := c.FloatBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc float64

	for {
		for _, v := range vs {
			acc += v
		}
		ks, vs = c.FloatBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []float64{acc}
		}
	}
}

type integerFloatCountBatchCursor struct {
	tsdb.FloatBatchCursor
}

func (c *integerFloatCountBatchCursor) Next() (key []int64, value []int64) {
	ks, _ := c.FloatBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc int64
	for {
		acc += int64(len(ks))
		ks, _ = c.FloatBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []int64{acc}
		}
	}
}

type floatEmptyBatchCursor struct{}

var FloatEmptyBatchCursor tsdb.FloatBatchCursor = &floatEmptyBatchCursor{}

func (*floatEmptyBatchCursor) Err() error                           { return nil }
func (*floatEmptyBatchCursor) Close()                               {}
func (*floatEmptyBatchCursor) SeriesKey() string                    { return "" }
func (*floatEmptyBatchCursor) Next() (key []int64, value []float64) { return nil, nil }

// ********************
// Integer BatchCursor

type integerFilterBatchCursor struct {
	tsdb.IntegerBatchCursor
	cond expression
	m    *singleValue
	t    []int64
	v    []int64
	tb   []int64
	vb   []int64
}

func newIntegerFilterBatchCursor(cond expression) *integerFilterBatchCursor {
	return &integerFilterBatchCursor{
		cond: cond,
		m:    &singleValue{},
		t:    make([]int64, tsdb.DefaultMaxPointsPerBlock),
		v:    make([]int64, tsdb.DefaultMaxPointsPerBlock),
	}
}

func (c *integerFilterBatchCursor) reset(cur tsdb.IntegerBatchCursor) {
	c.IntegerBatchCursor = cur
	c.tb, c.vb = nil, nil
}

func (c *integerFilterBatchCursor) Next() (key []int64, value []int64) {
	pos := 0
	var ks []int64
	var vs []int64

	if len(c.tb) > 0 {
		ks, vs = c.tb, c.vb
		c.tb, c.vb = nil, nil
	} else {
		ks, vs = c.IntegerBatchCursor.Next()
	}

	for len(ks) > 0 {
		for i, v := range vs {
			c.m.v = v
			if c.cond.EvalBool(c.m) {
				c.t[pos], c.v[pos] = ks[i], v
				pos++
				if pos >= tsdb.DefaultMaxPointsPerBlock {
					c.tb, c.vb = ks[i+1:], vs[i+1:]
					return c.t[:pos], c.v[:pos]
				}
			}
		}
		ks, vs = c.IntegerBatchCursor.Next()
	}

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

type integerMultiShardBatchCursor struct {
	tsdb.IntegerBatchCursor
	ctx    context.Context
	filter *integerFilterBatchCursor
	shards []*tsdb.Shard
	req    *tsdb.CursorRequest
	err    error
	limit  uint64
	count  uint64
}

func newIntegerMultiShardBatchCursor(ctx context.Context, cur tsdb.IntegerBatchCursor, rr *readRequest, req *tsdb.CursorRequest, shards []*tsdb.Shard, cond expression) *integerMultiShardBatchCursor {
	var filter *integerFilterBatchCursor
	if cond != nil {
		filter = newIntegerFilterBatchCursor(cond)
		filter.reset(cur)
		cur = filter
	}

	c := &integerMultiShardBatchCursor{
		IntegerBatchCursor: cur,
		ctx:                ctx,
		filter:             filter,
		req:                req,
		shards:             shards,
		limit:              rr.limit,
	}

	return c
}

func (c *integerMultiShardBatchCursor) Err() error        { return c.err }
func (c *integerMultiShardBatchCursor) SeriesKey() string { return c.req.Series }

func (c *integerMultiShardBatchCursor) Next() (key []int64, value []int64) {
	for {
		ks, vs := c.IntegerBatchCursor.Next()
		if len(ks) == 0 {
			if c.nextBatchCursor() {
				continue
			}
		}
		c.count += uint64(len(ks))
		if c.count > c.limit {
			diff := c.count - c.limit
			c.count -= diff
			rem := uint64(len(ks)) - diff
			ks = ks[:rem]
			vs = vs[:rem]
		}
		return ks, vs
	}
}

func (c *integerMultiShardBatchCursor) nextBatchCursor() bool {
	if len(c.shards) == 0 {
		return false
	}

	c.IntegerBatchCursor.Close()

	var shard *tsdb.Shard
	var cur tsdb.Cursor
	for cur == nil && len(c.shards) > 0 {
		shard, c.shards = c.shards[0], c.shards[1:]
		cur, _ = shard.CreateCursor(c.ctx, c.req)
	}

	var ok bool
	if cur != nil {
		var next tsdb.IntegerBatchCursor
		next, ok = cur.(tsdb.IntegerBatchCursor)
		if !ok {
			cur.Close()
			next = IntegerEmptyBatchCursor
			c.shards = nil
			c.err = errors.New("expected integer cursor")
		} else {
			if c.filter != nil {
				c.filter.reset(next)
				next = c.filter
			}
		}
		c.IntegerBatchCursor = next
	} else {
		c.IntegerBatchCursor = IntegerEmptyBatchCursor
	}

	return ok
}

type integerSumBatchCursor struct {
	tsdb.IntegerBatchCursor
}

func (c *integerSumBatchCursor) Next() (key []int64, value []int64) {
	ks, vs := c.IntegerBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc int64

	for {
		for _, v := range vs {
			acc += v
		}
		ks, vs = c.IntegerBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []int64{acc}
		}
	}
}

type integerIntegerCountBatchCursor struct {
	tsdb.IntegerBatchCursor
}

func (c *integerIntegerCountBatchCursor) Next() (key []int64, value []int64) {
	ks, _ := c.IntegerBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc int64
	for {
		acc += int64(len(ks))
		ks, _ = c.IntegerBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []int64{acc}
		}
	}
}

type integerEmptyBatchCursor struct{}

var IntegerEmptyBatchCursor tsdb.IntegerBatchCursor = &integerEmptyBatchCursor{}

func (*integerEmptyBatchCursor) Err() error                         { return nil }
func (*integerEmptyBatchCursor) Close()                             {}
func (*integerEmptyBatchCursor) SeriesKey() string                  { return "" }
func (*integerEmptyBatchCursor) Next() (key []int64, value []int64) { return nil, nil }

// ********************
// Unsigned BatchCursor

type unsignedFilterBatchCursor struct {
	tsdb.UnsignedBatchCursor
	cond expression
	m    *singleValue
	t    []int64
	v    []uint64
	tb   []int64
	vb   []uint64
}

func newUnsignedFilterBatchCursor(cond expression) *unsignedFilterBatchCursor {
	return &unsignedFilterBatchCursor{
		cond: cond,
		m:    &singleValue{},
		t:    make([]int64, tsdb.DefaultMaxPointsPerBlock),
		v:    make([]uint64, tsdb.DefaultMaxPointsPerBlock),
	}
}

func (c *unsignedFilterBatchCursor) reset(cur tsdb.UnsignedBatchCursor) {
	c.UnsignedBatchCursor = cur
	c.tb, c.vb = nil, nil
}

func (c *unsignedFilterBatchCursor) Next() (key []int64, value []uint64) {
	pos := 0
	var ks []int64
	var vs []uint64

	if len(c.tb) > 0 {
		ks, vs = c.tb, c.vb
		c.tb, c.vb = nil, nil
	} else {
		ks, vs = c.UnsignedBatchCursor.Next()
	}

	for len(ks) > 0 {
		for i, v := range vs {
			c.m.v = v
			if c.cond.EvalBool(c.m) {
				c.t[pos], c.v[pos] = ks[i], v
				pos++
				if pos >= tsdb.DefaultMaxPointsPerBlock {
					c.tb, c.vb = ks[i+1:], vs[i+1:]
					return c.t[:pos], c.v[:pos]
				}
			}
		}
		ks, vs = c.UnsignedBatchCursor.Next()
	}

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

type unsignedMultiShardBatchCursor struct {
	tsdb.UnsignedBatchCursor
	ctx    context.Context
	filter *unsignedFilterBatchCursor
	shards []*tsdb.Shard
	req    *tsdb.CursorRequest
	err    error
	limit  uint64
	count  uint64
}

func newUnsignedMultiShardBatchCursor(ctx context.Context, cur tsdb.UnsignedBatchCursor, rr *readRequest, req *tsdb.CursorRequest, shards []*tsdb.Shard, cond expression) *unsignedMultiShardBatchCursor {
	var filter *unsignedFilterBatchCursor
	if cond != nil {
		filter = newUnsignedFilterBatchCursor(cond)
		filter.reset(cur)
		cur = filter
	}

	c := &unsignedMultiShardBatchCursor{
		UnsignedBatchCursor: cur,
		ctx:                 ctx,
		filter:              filter,
		req:                 req,
		shards:              shards,
		limit:               rr.limit,
	}

	return c
}

func (c *unsignedMultiShardBatchCursor) Err() error        { return c.err }
func (c *unsignedMultiShardBatchCursor) SeriesKey() string { return c.req.Series }

func (c *unsignedMultiShardBatchCursor) Next() (key []int64, value []uint64) {
	for {
		ks, vs := c.UnsignedBatchCursor.Next()
		if len(ks) == 0 {
			if c.nextBatchCursor() {
				continue
			}
		}
		c.count += uint64(len(ks))
		if c.count > c.limit {
			diff := c.count - c.limit
			c.count -= diff
			rem := uint64(len(ks)) - diff
			ks = ks[:rem]
			vs = vs[:rem]
		}
		return ks, vs
	}
}

func (c *unsignedMultiShardBatchCursor) nextBatchCursor() bool {
	if len(c.shards) == 0 {
		return false
	}

	c.UnsignedBatchCursor.Close()

	var shard *tsdb.Shard
	var cur tsdb.Cursor
	for cur == nil && len(c.shards) > 0 {
		shard, c.shards = c.shards[0], c.shards[1:]
		cur, _ = shard.CreateCursor(c.ctx, c.req)
	}

	var ok bool
	if cur != nil {
		var next tsdb.UnsignedBatchCursor
		next, ok = cur.(tsdb.UnsignedBatchCursor)
		if !ok {
			cur.Close()
			next = UnsignedEmptyBatchCursor
			c.shards = nil
			c.err = errors.New("expected unsigned cursor")
		} else {
			if c.filter != nil {
				c.filter.reset(next)
				next = c.filter
			}
		}
		c.UnsignedBatchCursor = next
	} else {
		c.UnsignedBatchCursor = UnsignedEmptyBatchCursor
	}

	return ok
}

type unsignedSumBatchCursor struct {
	tsdb.UnsignedBatchCursor
}

func (c *unsignedSumBatchCursor) Next() (key []int64, value []uint64) {
	ks, vs := c.UnsignedBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc uint64

	for {
		for _, v := range vs {
			acc += v
		}
		ks, vs = c.UnsignedBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []uint64{acc}
		}
	}
}

type integerUnsignedCountBatchCursor struct {
	tsdb.UnsignedBatchCursor
}

func (c *integerUnsignedCountBatchCursor) Next() (key []int64, value []int64) {
	ks, _ := c.UnsignedBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc int64
	for {
		acc += int64(len(ks))
		ks, _ = c.UnsignedBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []int64{acc}
		}
	}
}

type unsignedEmptyBatchCursor struct{}

var UnsignedEmptyBatchCursor tsdb.UnsignedBatchCursor = &unsignedEmptyBatchCursor{}

func (*unsignedEmptyBatchCursor) Err() error                          { return nil }
func (*unsignedEmptyBatchCursor) Close()                              {}
func (*unsignedEmptyBatchCursor) SeriesKey() string                   { return "" }
func (*unsignedEmptyBatchCursor) Next() (key []int64, value []uint64) { return nil, nil }

// ********************
// String BatchCursor

type stringFilterBatchCursor struct {
	tsdb.StringBatchCursor
	cond expression
	m    *singleValue
	t    []int64
	v    []string
	tb   []int64
	vb   []string
}

func newStringFilterBatchCursor(cond expression) *stringFilterBatchCursor {
	return &stringFilterBatchCursor{
		cond: cond,
		m:    &singleValue{},
		t:    make([]int64, tsdb.DefaultMaxPointsPerBlock),
		v:    make([]string, tsdb.DefaultMaxPointsPerBlock),
	}
}

func (c *stringFilterBatchCursor) reset(cur tsdb.StringBatchCursor) {
	c.StringBatchCursor = cur
	c.tb, c.vb = nil, nil
}

func (c *stringFilterBatchCursor) Next() (key []int64, value []string) {
	pos := 0
	var ks []int64
	var vs []string

	if len(c.tb) > 0 {
		ks, vs = c.tb, c.vb
		c.tb, c.vb = nil, nil
	} else {
		ks, vs = c.StringBatchCursor.Next()
	}

	for len(ks) > 0 {
		for i, v := range vs {
			c.m.v = v
			if c.cond.EvalBool(c.m) {
				c.t[pos], c.v[pos] = ks[i], v
				pos++
				if pos >= tsdb.DefaultMaxPointsPerBlock {
					c.tb, c.vb = ks[i+1:], vs[i+1:]
					return c.t[:pos], c.v[:pos]
				}
			}
		}
		ks, vs = c.StringBatchCursor.Next()
	}

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

type stringMultiShardBatchCursor struct {
	tsdb.StringBatchCursor
	ctx    context.Context
	filter *stringFilterBatchCursor
	shards []*tsdb.Shard
	req    *tsdb.CursorRequest
	err    error
	limit  uint64
	count  uint64
}

func newStringMultiShardBatchCursor(ctx context.Context, cur tsdb.StringBatchCursor, rr *readRequest, req *tsdb.CursorRequest, shards []*tsdb.Shard, cond expression) *stringMultiShardBatchCursor {
	var filter *stringFilterBatchCursor
	if cond != nil {
		filter = newStringFilterBatchCursor(cond)
		filter.reset(cur)
		cur = filter
	}

	c := &stringMultiShardBatchCursor{
		StringBatchCursor: cur,
		ctx:               ctx,
		filter:            filter,
		req:               req,
		shards:            shards,
		limit:             rr.limit,
	}

	return c
}

func (c *stringMultiShardBatchCursor) Err() error        { return c.err }
func (c *stringMultiShardBatchCursor) SeriesKey() string { return c.req.Series }

func (c *stringMultiShardBatchCursor) Next() (key []int64, value []string) {
	for {
		ks, vs := c.StringBatchCursor.Next()
		if len(ks) == 0 {
			if c.nextBatchCursor() {
				continue
			}
		}
		c.count += uint64(len(ks))
		if c.count > c.limit {
			diff := c.count - c.limit
			c.count -= diff
			rem := uint64(len(ks)) - diff
			ks = ks[:rem]
			vs = vs[:rem]
		}
		return ks, vs
	}
}

func (c *stringMultiShardBatchCursor) nextBatchCursor() bool {
	if len(c.shards) == 0 {
		return false
	}

	c.StringBatchCursor.Close()

	var shard *tsdb.Shard
	var cur tsdb.Cursor
	for cur == nil && len(c.shards) > 0 {
		shard, c.shards = c.shards[0], c.shards[1:]
		cur, _ = shard.CreateCursor(c.ctx, c.req)
	}

	var ok bool
	if cur != nil {
		var next tsdb.StringBatchCursor
		next, ok = cur.(tsdb.StringBatchCursor)
		if !ok {
			cur.Close()
			next = StringEmptyBatchCursor
			c.shards = nil
			c.err = errors.New("expected string cursor")
		} else {
			if c.filter != nil {
				c.filter.reset(next)
				next = c.filter
			}
		}
		c.StringBatchCursor = next
	} else {
		c.StringBatchCursor = StringEmptyBatchCursor
	}

	return ok
}

type integerStringCountBatchCursor struct {
	tsdb.StringBatchCursor
}

func (c *integerStringCountBatchCursor) Next() (key []int64, value []int64) {
	ks, _ := c.StringBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc int64
	for {
		acc += int64(len(ks))
		ks, _ = c.StringBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []int64{acc}
		}
	}
}

type stringEmptyBatchCursor struct{}

var StringEmptyBatchCursor tsdb.StringBatchCursor = &stringEmptyBatchCursor{}

func (*stringEmptyBatchCursor) Err() error                          { return nil }
func (*stringEmptyBatchCursor) Close()                              {}
func (*stringEmptyBatchCursor) SeriesKey() string                   { return "" }
func (*stringEmptyBatchCursor) Next() (key []int64, value []string) { return nil, nil }

// ********************
// Boolean BatchCursor

type booleanFilterBatchCursor struct {
	tsdb.BooleanBatchCursor
	cond expression
	m    *singleValue
	t    []int64
	v    []bool
	tb   []int64
	vb   []bool
}

func newBooleanFilterBatchCursor(cond expression) *booleanFilterBatchCursor {
	return &booleanFilterBatchCursor{
		cond: cond,
		m:    &singleValue{},
		t:    make([]int64, tsdb.DefaultMaxPointsPerBlock),
		v:    make([]bool, tsdb.DefaultMaxPointsPerBlock),
	}
}

func (c *booleanFilterBatchCursor) reset(cur tsdb.BooleanBatchCursor) {
	c.BooleanBatchCursor = cur
	c.tb, c.vb = nil, nil
}

func (c *booleanFilterBatchCursor) Next() (key []int64, value []bool) {
	pos := 0
	var ks []int64
	var vs []bool

	if len(c.tb) > 0 {
		ks, vs = c.tb, c.vb
		c.tb, c.vb = nil, nil
	} else {
		ks, vs = c.BooleanBatchCursor.Next()
	}

	for len(ks) > 0 {
		for i, v := range vs {
			c.m.v = v
			if c.cond.EvalBool(c.m) {
				c.t[pos], c.v[pos] = ks[i], v
				pos++
				if pos >= tsdb.DefaultMaxPointsPerBlock {
					c.tb, c.vb = ks[i+1:], vs[i+1:]
					return c.t[:pos], c.v[:pos]
				}
			}
		}
		ks, vs = c.BooleanBatchCursor.Next()
	}

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

type booleanMultiShardBatchCursor struct {
	tsdb.BooleanBatchCursor
	ctx    context.Context
	filter *booleanFilterBatchCursor
	shards []*tsdb.Shard
	req    *tsdb.CursorRequest
	err    error
	limit  uint64
	count  uint64
}

func newBooleanMultiShardBatchCursor(ctx context.Context, cur tsdb.BooleanBatchCursor, rr *readRequest, req *tsdb.CursorRequest, shards []*tsdb.Shard, cond expression) *booleanMultiShardBatchCursor {
	var filter *booleanFilterBatchCursor
	if cond != nil {
		filter = newBooleanFilterBatchCursor(cond)
		filter.reset(cur)
		cur = filter
	}

	c := &booleanMultiShardBatchCursor{
		BooleanBatchCursor: cur,
		ctx:                ctx,
		filter:             filter,
		req:                req,
		shards:             shards,
		limit:              rr.limit,
	}

	return c
}

func (c *booleanMultiShardBatchCursor) Err() error        { return c.err }
func (c *booleanMultiShardBatchCursor) SeriesKey() string { return c.req.Series }

func (c *booleanMultiShardBatchCursor) Next() (key []int64, value []bool) {
	for {
		ks, vs := c.BooleanBatchCursor.Next()
		if len(ks) == 0 {
			if c.nextBatchCursor() {
				continue
			}
		}
		c.count += uint64(len(ks))
		if c.count > c.limit {
			diff := c.count - c.limit
			c.count -= diff
			rem := uint64(len(ks)) - diff
			ks = ks[:rem]
			vs = vs[:rem]
		}
		return ks, vs
	}
}

func (c *booleanMultiShardBatchCursor) nextBatchCursor() bool {
	if len(c.shards) == 0 {
		return false
	}

	c.BooleanBatchCursor.Close()

	var shard *tsdb.Shard
	var cur tsdb.Cursor
	for cur == nil && len(c.shards) > 0 {
		shard, c.shards = c.shards[0], c.shards[1:]
		cur, _ = shard.CreateCursor(c.ctx, c.req)
	}

	var ok bool
	if cur != nil {
		var next tsdb.BooleanBatchCursor
		next, ok = cur.(tsdb.BooleanBatchCursor)
		if !ok {
			cur.Close()
			next = BooleanEmptyBatchCursor
			c.shards = nil
			c.err = errors.New("expected boolean cursor")
		} else {
			if c.filter != nil {
				c.filter.reset(next)
				next = c.filter
			}
		}
		c.BooleanBatchCursor = next
	} else {
		c.BooleanBatchCursor = BooleanEmptyBatchCursor
	}

	return ok
}

type integerBooleanCountBatchCursor struct {
	tsdb.BooleanBatchCursor
}

func (c *integerBooleanCountBatchCursor) Next() (key []int64, value []int64) {
	ks, _ := c.BooleanBatchCursor.Next()
	if len(ks) == 0 {
		return nil, nil
	}

	ts := ks[0]
	var acc int64
	for {
		acc += int64(len(ks))
		ks, _ = c.BooleanBatchCursor.Next()
		if len(ks) == 0 {
			return []int64{ts}, []int64{acc}
		}
	}
}

type booleanEmptyBatchCursor struct{}

var BooleanEmptyBatchCursor tsdb.BooleanBatchCursor = &booleanEmptyBatchCursor{}

func (*booleanEmptyBatchCursor) Err() error                        { return nil }
func (*booleanEmptyBatchCursor) Close()                            {}
func (*booleanEmptyBatchCursor) SeriesKey() string                 { return "" }
func (*booleanEmptyBatchCursor) Next() (key []int64, value []bool) { return nil, nil }