mirror of
https://github.com/prometheus/prometheus.git
synced 2025-03-05 20:59:13 -08:00
Add chunk based series iterator
This commit is contained in:
parent
b334c3ade8
commit
9b400b4c58
67
querier.go
67
querier.go
|
@ -1,5 +1,7 @@
|
||||||
package tsdb
|
package tsdb
|
||||||
|
|
||||||
|
import "github.com/fabxc/tsdb/chunks"
|
||||||
|
|
||||||
// Matcher matches a string.
|
// Matcher matches a string.
|
||||||
type Matcher interface {
|
type Matcher interface {
|
||||||
// Match returns true if the matcher applies to the string value.
|
// Match returns true if the matcher applies to the string value.
|
||||||
|
@ -52,3 +54,68 @@ type SeriesIterator interface {
|
||||||
// Err returns the current error.
|
// Err returns the current error.
|
||||||
Err() error
|
Err() error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// chunkSeriesIterator implements a series iterator on top
|
||||||
|
// of a list of time-sorted, non-overlapping chunks.
|
||||||
|
type chunkSeriesIterator struct {
|
||||||
|
// minTimes []int64
|
||||||
|
chunks []chunks.Chunk
|
||||||
|
|
||||||
|
i int
|
||||||
|
cur chunks.Iterator
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func newChunkSeriesIterator(cs []chunks.Chunk) *chunkSeriesIterator {
|
||||||
|
return &chunkSeriesIterator{
|
||||||
|
chunks: cs,
|
||||||
|
i: 0,
|
||||||
|
cur: cs[0].Iterator(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (it *chunkSeriesIterator) Seek(t int64) (ok bool) {
|
||||||
|
// TODO(fabxc): skip to relevant chunk.
|
||||||
|
for it.Next() {
|
||||||
|
if ts, _ := it.Values(); ts >= t {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
func (it *chunkSeriesIterator) Values() (t int64, v float64) {
|
||||||
|
return it.cur.Values()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (it *chunkSeriesIterator) Next() bool {
|
||||||
|
if it.cur.Next() {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
if err := it.cur.Err(); err != nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if it.i == len(it.chunks)-1 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
it.i++
|
||||||
|
it.cur = it.chunks[it.i].Iterator()
|
||||||
|
|
||||||
|
return it.Next()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (it *chunkSeriesIterator) Err() error {
|
||||||
|
return it.cur.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
type bufferedSeriesIterator struct {
|
||||||
|
// TODO(fabxc): time-based look back buffer for time-aggregating
|
||||||
|
// queries such as rate. It should allow us to re-use an iterator
|
||||||
|
// within a range query while calculating time-aggregates at any point.
|
||||||
|
//
|
||||||
|
// It also allows looking up/seeking at-or-before without modifying
|
||||||
|
// the simpler interface.
|
||||||
|
//
|
||||||
|
// Consider making this the main external interface.
|
||||||
|
}
|
||||||
|
|
13
reader.go
13
reader.go
|
@ -70,6 +70,8 @@ type StringTuples interface {
|
||||||
}
|
}
|
||||||
|
|
||||||
type indexReader struct {
|
type indexReader struct {
|
||||||
|
series SeriesReader
|
||||||
|
|
||||||
// The underlying byte slice holding the encoded series data.
|
// The underlying byte slice holding the encoded series data.
|
||||||
b []byte
|
b []byte
|
||||||
|
|
||||||
|
@ -83,11 +85,14 @@ var (
|
||||||
errInvalidFlag = fmt.Errorf("invalid flag")
|
errInvalidFlag = fmt.Errorf("invalid flag")
|
||||||
)
|
)
|
||||||
|
|
||||||
func newIndexReader(b []byte) (*indexReader, error) {
|
func newIndexReader(s SeriesReader, b []byte) (*indexReader, error) {
|
||||||
if len(b) < 16 {
|
if len(b) < 16 {
|
||||||
return nil, errInvalidSize
|
return nil, errInvalidSize
|
||||||
}
|
}
|
||||||
r := &indexReader{b: b}
|
r := &indexReader{
|
||||||
|
series: s,
|
||||||
|
b: b,
|
||||||
|
}
|
||||||
|
|
||||||
// Verify magic number.
|
// Verify magic number.
|
||||||
if m := binary.BigEndian.Uint32(b[:4]); m != MagicIndex {
|
if m := binary.BigEndian.Uint32(b[:4]); m != MagicIndex {
|
||||||
|
@ -276,13 +281,15 @@ func (r *indexReader) Series(ref uint32) (Series, error) {
|
||||||
s := &series{
|
s := &series{
|
||||||
labels: labels,
|
labels: labels,
|
||||||
offsets: coffsets,
|
offsets: coffsets,
|
||||||
|
chunk: r.series.Chunk,
|
||||||
}
|
}
|
||||||
return s, nil
|
return s, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type series struct {
|
type series struct {
|
||||||
labels Labels
|
labels Labels
|
||||||
offsets []ChunkOffset
|
offsets []ChunkOffset // in-order chunk refs
|
||||||
|
chunk func(ref uint32) (chunks.Chunk, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *series) Labels() (Labels, error) {
|
func (s *series) Labels() (Labels, error) {
|
||||||
|
|
Loading…
Reference in a new issue