golang-github-influxdb-influxdb-dev 1.1.1+dfsg1-4 / usr / share / gocode / src / github.com / influxdata / influxdb / tsdb / engine / tsm1 / iterator.gen.go.tmpl

package tsm1

import (
	"sort"
	"fmt"
	"sync"

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

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 }

// 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 {
	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 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

{{range .}}

type {{.name}}Iterator struct {
	cur   {{.name}}Cursor
	aux   []cursorAt
	conds struct {
		names []string
		curs  []cursorAt
	}
	opt   influxql.IteratorOptions

	m map[string]interface{}      // map used for condition evaluation
	point influxql.{{.Name}}Point // reusable buffer

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

func new{{.Name}}Iterator(name string, tags influxql.Tags, opt influxql.IteratorOptions, cur {{.name}}Cursor, aux []cursorAt, conds []cursorAt, condNames []string) *{{.name}}Iterator {
	itr := &{{.name}}Iterator{
		cur:   cur,
		aux:   aux,
		opt:   opt,
		point: influxql.{{.Name}}Point{
			Name: name,
			Tags: tags,
		},
		statsBuf: influxql.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 *{{.name}}Iterator) Next() (*influxql.{{.Name}}Point, 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.next{{.Name}}()
			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.
		if itr.opt.Condition != nil && !influxql.EvalBool(itr.opt.Condition, itr.m) {
			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 *{{.name}}Iterator) copyStats() {
	itr.statsLock.Lock()
	itr.stats = itr.statsBuf
	itr.statsLock.Unlock()
}

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

// Close closes the iterator.
func (itr *{{.name}}Iterator) Close() error {
	for _, c := range itr.aux {
		c.close()
	}
	itr.aux = nil
	for _, c := range itr.conds.curs {
	    c.close()
	}
	itr.conds.curs = nil
	if itr.cur != nil {
		err := itr.cur.close()
		itr.cur = nil
		return err
	}
	return nil
}

// {{.name}}LimitIterator
type {{.name}}LimitIterator struct {
	input influxql.{{.Name}}Iterator
	opt   influxql.IteratorOptions
	n     int
}

func new{{.Name}}LimitIterator(input influxql.{{.Name}}Iterator, opt influxql.IteratorOptions) *{{.name}}LimitIterator {
	return &{{.name}}LimitIterator{
		input: input,
		opt:   opt,
	}
}

func (itr *{{.name}}LimitIterator) Stats() influxql.IteratorStats { return itr.input.Stats() }
func (itr *{{.name}}LimitIterator) Close() error                  { return itr.input.Close() }

func (itr *{{.name}}LimitIterator) Next() (*influxql.{{.Name}}Point, error) {
	for {
		// 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
	}
}

// {{.name}}Cursor represents an object for iterating over a single {{.name}} field.
type {{.name}}Cursor interface {
	cursor
	next{{.Name}}() (t int64, v {{.Type}})
}

func new{{.Name}}Cursor(seek int64, ascending bool, cacheValues Values, tsmKeyCursor *KeyCursor) {{.name}}Cursor {
	if ascending {
		return new{{.Name}}AscendingCursor(seek, cacheValues, tsmKeyCursor)
	}
	return new{{.Name}}DescendingCursor(seek, cacheValues, tsmKeyCursor)
}

type {{.name}}AscendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		buf       []{{.Name}}Value
		values    []{{.Name}}Value
		pos       int
		keyCursor *KeyCursor
	}
}

func new{{.Name}}AscendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *{{.name}}AscendingCursor {
	c := &{{.name}}AscendingCursor{}

	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.buf = make([]{{.Name}}Value, 10)
	c.tsm.values, _ = c.tsm.keyCursor.Read{{.Name}}Block(&c.tsm.buf)
	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 *{{.name}}AscendingCursor) peekCache() (t int64, v {{.Type}}) {
	if c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, {{.Nil}}
	}

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

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

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

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

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

// next{{.Name}} returns the next key/value for the cursor.
func (c *{{.name}}AscendingCursor) next{{.Name}}() (int64, {{.Type}}) {
	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, {{.Nil}}
	}

	// 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 *{{.name}}AscendingCursor) nextCache() {
	if c.cache.pos >= len(c.cache.values) {
		return
	}
	c.cache.pos++
}

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

type {{.name}}DescendingCursor struct {
	cache struct {
		values Values
		pos    int
	}

	tsm struct {
		buf       []{{.Name}}Value
		values    []{{.Name}}Value
		pos       int
		keyCursor *KeyCursor
	}
}

func new{{.Name}}DescendingCursor(seek int64, cacheValues Values, tsmKeyCursor *KeyCursor) *{{.name}}DescendingCursor {
	c := &{{.name}}DescendingCursor{}

	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.buf = make([]{{.Name}}Value, 10)
	c.tsm.values, _ = c.tsm.keyCursor.Read{{.Name}}Block(&c.tsm.buf)
	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 *{{.name}}DescendingCursor) peekCache() (t int64, v {{.Type}}) {
	if c.cache.pos < 0 || c.cache.pos >= len(c.cache.values) {
		return tsdb.EOF, {{.Nil}}
	}

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

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

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

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

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

// next{{.Name}} returns the next key/value for the cursor.
func (c *{{.name}}DescendingCursor) next{{.Name}}() (int64, {{.Type}}) {
	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, {{.Nil}}
	}

	// 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 *{{.name}}DescendingCursor) nextCache() {
	if c.cache.pos < 0 {
		return
	}
	c.cache.pos--
}

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

// {{.name}}LiteralCursor represents a cursor that always returns a single value.
// It doesn't not have a time value so it can only be used with nextAt().
type {{.name}}LiteralCursor struct {
	value {{.Type}}
}

func (c *{{.name}}LiteralCursor) close() error { return nil }
func (c *{{.name}}LiteralCursor) peek() (t int64, v interface{}) { return tsdb.EOF, c.value }
func (c *{{.name}}LiteralCursor) next() (t int64, v interface{}) { return tsdb.EOF, c.value }
func (c *{{.name}}LiteralCursor) nextAt(seek int64) interface{} { return c.value }


// {{.name}}NilLiteralCursor represents a cursor that always returns a typed nil value.
// It doesn't not have a time value so it can only be used with nextAt().
type {{.name}}NilLiteralCursor struct {}

func (c *{{.name}}NilLiteralCursor) close() error { return nil }
func (c *{{.name}}NilLiteralCursor) peek() (t int64, v interface{}) { return tsdb.EOF, (*{{.Type}})(nil) }
func (c *{{.name}}NilLiteralCursor) next() (t int64, v interface{}) { return tsdb.EOF, (*{{.Type}})(nil) }
func (c *{{.name}}NilLiteralCursor) nextAt(seek int64) interface{} { return (*{{.Type}})(nil) }

{{end}}

var _ = fmt.Print