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