diff --git a/CHANGELOG.md b/CHANGELOG.md index 12364b09..eed7c04b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ ## master / unreleased +- [CHANGE] `chunks.MergeOverlappingChunks` moved to `tsdb.MergeOverlappingChunks` +- [CHANGE] `Series` interface allows return chunk iterator that allows iterating over encoded chunks. + ## 0.10.0 - [FEATURE] Added `DBReadOnly` to allow opening a database in read only mode. diff --git a/chunkenc/chunk.go b/chunkenc/chunk.go index 5f9349f0..f2e3bb85 100644 --- a/chunkenc/chunk.go +++ b/chunkenc/chunk.go @@ -56,11 +56,19 @@ type Appender interface { Append(int64, float64) } -// Iterator is a simple iterator that can only get the next value. +// Iterator iterates over the data of a time series. type Iterator interface { + // Seek advances the iterator forward to the sample with the timestamp t or first value after t. + // If the current iterator points to the sample with timestamp after t already, Seek should not advance the iterator. + // Seek returns false if there is no such sample with the timestamp equal or larger than t. + // Iterator can be exhausted when the Seek returns false. + Seek(t int64) bool + // At returns the current timestamp/value pair. At() (int64, float64) - Err() error + // Next advances the iterator by one. Next() bool + // Err returns the current error. + Err() error } // NewNopIterator returns a new chunk iterator that does not hold any data. @@ -70,6 +78,7 @@ func NewNopIterator() Iterator { type nopIterator struct{} +func (nopIterator) Seek(t int64) bool { return false } func (nopIterator) At() (int64, float64) { return 0, 0 } func (nopIterator) Next() bool { return false } func (nopIterator) Err() error { return nil } diff --git a/chunkenc/chunk_test.go b/chunkenc/chunk_test.go index f6577b42..84297943 100644 --- a/chunkenc/chunk_test.go +++ b/chunkenc/chunk_test.go @@ -17,7 +17,6 @@ import ( "fmt" "io" "math/rand" - "reflect" "testing" "github.com/prometheus/tsdb/testutil" @@ -35,28 +34,23 @@ func TestChunk(t *testing.T) { t.Run(fmt.Sprintf("%v", enc), func(t *testing.T) { for range make([]struct{}, 1) { c := nc() - if err := testChunk(c); err != nil { - t.Fatal(err) - } + testChunk(t, c) } }) } } -func testChunk(c Chunk) error { +func testChunk(t *testing.T, c Chunk) { app, err := c.Appender() - if err != nil { - return err - } + testutil.Ok(t, err) - var exp []pair + var all []pair var ( ts = int64(1234123324) v = 1243535.123 ) for i := 0; i < 300; i++ { ts += int64(rand.Intn(10000) + 1) - // v = rand.Float64() if i%2 == 0 { v += float64(rand.Intn(1000000)) } else { @@ -67,29 +61,53 @@ func testChunk(c Chunk) error { // appending to a partially filled chunk. if i%10 == 0 { app, err = c.Appender() - if err != nil { - return err - } + testutil.Ok(t, err) } app.Append(ts, v) - exp = append(exp, pair{t: ts, v: v}) - // fmt.Println("appended", len(c.Bytes()), c.Bytes()) + all = append(all, pair{t: ts, v: v}) } - it := c.Iterator(nil) - var res []pair - for it.Next() { - ts, v := it.At() - res = append(res, pair{t: ts, v: v}) + // 1. Expand iterator in simple case. + it1 := c.Iterator(nil) + var res1 []pair + for it1.Next() { + ts, v := it1.At() + res1 = append(res1, pair{t: ts, v: v}) } - if it.Err() != nil { - return it.Err() + testutil.Ok(t, it1.Err()) + testutil.Equals(t, all, res1) + + // 2. Expand second iterator while reusing first one. + it2 := c.Iterator(it1) + var res2 []pair + for it2.Next() { + ts, v := it2.At() + res2 = append(res2, pair{t: ts, v: v}) } - if !reflect.DeepEqual(exp, res) { - return fmt.Errorf("unexpected result\n\ngot: %v\n\nexp: %v", res, exp) + testutil.Ok(t, it2.Err()) + testutil.Equals(t, all, res2) + + // 3. Test Iterator Seek. + mid := len(all) / 2 + + it3 := c.Iterator(nil) + var res3 []pair + testutil.Equals(t, true, it3.Seek(all[mid].t)) + // Below ones should not matter. + testutil.Equals(t, true, it3.Seek(all[mid].t)) + testutil.Equals(t, true, it3.Seek(all[mid].t)) + ts, v = it3.At() + res3 = append(res3, pair{t: ts, v: v}) + + for it3.Next() { + ts, v := it3.At() + res3 = append(res3, pair{t: ts, v: v}) } - return nil + testutil.Ok(t, it3.Err()) + testutil.Equals(t, all[mid:], res3) + + testutil.Equals(t, false, it3.Seek(all[len(all)-1].t+1)) } func benchmarkIterator(b *testing.B, newChunk func() Chunk) { diff --git a/chunkenc/xor.go b/chunkenc/xor.go index ca20309f..86c05f0f 100644 --- a/chunkenc/xor.go +++ b/chunkenc/xor.go @@ -241,6 +241,19 @@ type xorIterator struct { err error } +func (it *xorIterator) Seek(t int64) bool { + if it.err != nil { + return false + } + + for t > it.t || it.numRead == 0 { + if !it.Next() { + return false + } + } + return true +} + func (it *xorIterator) At() (int64, float64) { return it.t, it.val } diff --git a/chunks/chunks.go b/chunks/chunks.go index 8c6945ca..a288ac48 100644 --- a/chunks/chunks.go +++ b/chunks/chunks.go @@ -205,85 +205,6 @@ func (w *Writer) write(b []byte) error { return err } -// MergeOverlappingChunks removes the samples whose timestamp is overlapping. -// The last appearing sample is retained in case there is overlapping. -// This assumes that `chks []Meta` is sorted w.r.t. MinTime. -func MergeOverlappingChunks(chks []Meta) ([]Meta, error) { - if len(chks) < 2 { - return chks, nil - } - newChks := make([]Meta, 0, len(chks)) // Will contain the merged chunks. - newChks = append(newChks, chks[0]) - last := 0 - for _, c := range chks[1:] { - // We need to check only the last chunk in newChks. - // Reason: (1) newChks[last-1].MaxTime < newChks[last].MinTime (non overlapping) - // (2) As chks are sorted w.r.t. MinTime, newChks[last].MinTime < c.MinTime. - // So never overlaps with newChks[last-1] or anything before that. - if c.MinTime > newChks[last].MaxTime { - newChks = append(newChks, c) - last++ - continue - } - nc := &newChks[last] - if c.MaxTime > nc.MaxTime { - nc.MaxTime = c.MaxTime - } - chk, err := MergeChunks(nc.Chunk, c.Chunk) - if err != nil { - return nil, err - } - nc.Chunk = chk - } - - return newChks, nil -} - -// MergeChunks vertically merges a and b, i.e., if there is any sample -// with same timestamp in both a and b, the sample in a is discarded. -func MergeChunks(a, b chunkenc.Chunk) (*chunkenc.XORChunk, error) { - newChunk := chunkenc.NewXORChunk() - app, err := newChunk.Appender() - if err != nil { - return nil, err - } - ait := a.Iterator(nil) - bit := b.Iterator(nil) - aok, bok := ait.Next(), bit.Next() - for aok && bok { - at, av := ait.At() - bt, bv := bit.At() - if at < bt { - app.Append(at, av) - aok = ait.Next() - } else if bt < at { - app.Append(bt, bv) - bok = bit.Next() - } else { - app.Append(bt, bv) - aok = ait.Next() - bok = bit.Next() - } - } - for aok { - at, av := ait.At() - app.Append(at, av) - aok = ait.Next() - } - for bok { - bt, bv := bit.At() - app.Append(bt, bv) - bok = bit.Next() - } - if ait.Err() != nil { - return nil, ait.Err() - } - if bit.Err() != nil { - return nil, bit.Err() - } - return newChunk, nil -} - func (w *Writer) WriteChunks(chks ...Meta) error { // Calculate maximum space we need and cut a new segment in case // we don't fit into the current one. diff --git a/compact.go b/compact.go index 9443c99e..ec7b2ae9 100644 --- a/compact.go +++ b/compact.go @@ -811,7 +811,7 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta, mergedChks := chks if overlapping { - mergedChks, err = chunks.MergeOverlappingChunks(chks) + mergedChks, err = MergeOverlappingChunks(chks) if err != nil { return errors.Wrap(err, "merge overlapping chunks") } @@ -1032,3 +1032,34 @@ func (c *compactionMerger) Err() error { func (c *compactionMerger) At() (labels.Labels, []chunks.Meta, Intervals) { return c.l, c.c, c.intervals } + +// MergeOverlappingChunks removes the samples whose timestamp is overlapping. +// The last appearing sample is retained in case there is overlapping. +// This assumes that `chks []Meta` is sorted w.r.t. MinTime. +func MergeOverlappingChunks(chks []chunks.Meta) ([]chunks.Meta, error) { + if len(chks) < 2 { + return chks, nil + } + newChks := make([]chunks.Meta, 0, len(chks)) // Will contain the merged chunks. + newChks = append(newChks, chks[0]) + last := 0 + var aReuseIter, bReuseIter chunkenc.Iterator + for _, c := range chks[1:] { + // We need to check only the last chunk in newChks. + // Reason: (1) newChks[last-1].MaxTime < newChks[last].MinTime (non overlapping) + // (2) As chks are sorted w.r.t. MinTime, newChks[last].MinTime < c.MinTime. + // So never overlaps with newChks[last-1] or anything before that. + if c.MinTime > newChks[last].MaxTime { + newChks = append(newChks, c) + last++ + continue + } + chk, err := mergeOverlappingChunks(newChks[last], c, aReuseIter, bReuseIter) + if err != nil { + return nil, err + } + newChks[last] = chk + } + + return newChks, nil +} diff --git a/head_test.go b/head_test.go index 040ae828..34bdd57b 100644 --- a/head_test.go +++ b/head_test.go @@ -740,7 +740,7 @@ func TestDelete_e2e(t *testing.T) { smpls = deletedSamples(smpls, del.drange) // Only append those series for which samples exist as mockSeriesSet // doesn't skip series with no samples. - // TODO: But sometimes SeriesSet returns an empty SeriesIterator + // TODO: But sometimes SeriesSet returns an empty chunkenc.Iterator if len(smpls) > 0 { matchedSeries = append(matchedSeries, newSeries( m.Map(), diff --git a/querier.go b/querier.go index fbd9493f..664d2605 100644 --- a/querier.go +++ b/querier.go @@ -15,6 +15,7 @@ package tsdb import ( "fmt" + "math" "sort" "strings" "unicode/utf8" @@ -53,7 +54,23 @@ type Series interface { Labels() labels.Labels // Iterator returns a new iterator of the data of the series. - Iterator() SeriesIterator + Iterator() chunkenc.Iterator + + // ChunkIterator returns a new iterator that iterates over non-overlapping chunks of the series. + ChunkIterator() ChunkIterator +} + +// ChunkIterator iterates over the chunk of a time series. +type ChunkIterator interface { + // Seek advances the iterator forward to the given timestamp. + // It advances to the chunk with min time at t or first chunk with min time after t. + Seek(t int64) bool + // At returns the meta. + At() chunks.Meta + // Next advances the iterator by one. + Next() bool + // Err returns optional error if Next is false. + Err() error } // querier aggregates querying results from time blocks within @@ -872,22 +889,12 @@ func (s *chunkSeries) Labels() labels.Labels { return s.labels } -func (s *chunkSeries) Iterator() SeriesIterator { +func (s *chunkSeries) Iterator() chunkenc.Iterator { return newChunkSeriesIterator(s.chunks, s.intervals, s.mint, s.maxt) } -// SeriesIterator iterates over the data of a time series. -type SeriesIterator interface { - // Seek advances the iterator forward to the given timestamp. - // If there's no value exactly at t, it advances to the first value - // after t. - Seek(t int64) bool - // At returns the current timestamp/value pair. - At() (t int64, v float64) - // Next advances the iterator by one. - Next() bool - // Err returns the current error. - Err() error +func (s *chunkSeries) ChunkIterator() ChunkIterator { + return &chunkIterator{chunks: s.chunks} } // chainedSeries implements a series for a list of time-sorted series. @@ -900,17 +907,25 @@ func (s *chainedSeries) Labels() labels.Labels { return s.series[0].Labels() } -func (s *chainedSeries) Iterator() SeriesIterator { +func (s *chainedSeries) Iterator() chunkenc.Iterator { return newChainedSeriesIterator(s.series...) } -// chainedSeriesIterator implements a series iterater over a list +func (s *chainedSeries) ChunkIterator() ChunkIterator { + ch := &chainedChunkIterator{chain: make([]ChunkIterator, 0, len(s.series))} + for _, s := range s.series { + ch.chain = append(ch.chain, s.ChunkIterator()) + } + return ch +} + +// chainedSeriesIterator implements a series iterated over a list // of time-sorted, non-overlapping iterators. type chainedSeriesIterator struct { series []Series // series in time order i int - cur SeriesIterator + cur chunkenc.Iterator } func newChainedSeriesIterator(s ...Series) *chainedSeriesIterator { @@ -940,7 +955,7 @@ func (it *chainedSeriesIterator) Next() bool { if it.cur.Next() { return true } - if err := it.cur.Err(); err != nil { + if it.cur.Err() != nil { return false } if it.i == len(it.series)-1 { @@ -971,21 +986,25 @@ func (s *verticalChainedSeries) Labels() labels.Labels { return s.series[0].Labels() } -func (s *verticalChainedSeries) Iterator() SeriesIterator { +func (s *verticalChainedSeries) Iterator() chunkenc.Iterator { return newVerticalMergeSeriesIterator(s.series...) } -// verticalMergeSeriesIterator implements a series iterater over a list +func (s *verticalChainedSeries) ChunkIterator() ChunkIterator { + return newVerticalMergeChunkIterator(s.series...) +} + +// verticalMergeSeriesIterator implements a series iterator over a list // of time-sorted, time-overlapping iterators. type verticalMergeSeriesIterator struct { - a, b SeriesIterator + a, b chunkenc.Iterator aok, bok, initialized bool curT int64 curV float64 } -func newVerticalMergeSeriesIterator(s ...Series) SeriesIterator { +func newVerticalMergeSeriesIterator(s ...Series) chunkenc.Iterator { if len(s) == 1 { return s[0].Iterator() } else if len(s) == 2 { @@ -1001,12 +1020,20 @@ func newVerticalMergeSeriesIterator(s ...Series) SeriesIterator { } func (it *verticalMergeSeriesIterator) Seek(t int64) bool { + if it.initialized && it.curT >= t { + return true + } + it.aok, it.bok = it.a.Seek(t), it.b.Seek(t) it.initialized = true return it.Next() } func (it *verticalMergeSeriesIterator) Next() bool { + if it.Err() != nil { + return false + } + if !it.initialized { it.aok = it.a.Next() it.bok = it.b.Next() @@ -1055,6 +1082,141 @@ func (it *verticalMergeSeriesIterator) Err() error { return it.b.Err() } +// verticalMergeChunkIterator implements a ChunkIterator over a list +// of time-sorted, time-overlapping chunk iterators for the same labels (same series). +// Any overlap in chunks will be merged using verticalMergeSeriesIterator. +type verticalMergeChunkIterator struct { + a, b ChunkIterator + aok, bok, initialized bool + + curMeta chunks.Meta + err error + + aReuseIter, bReuseIter chunkenc.Iterator +} + +func newVerticalMergeChunkIterator(s ...Series) ChunkIterator { + if len(s) == 1 { + return s[0].ChunkIterator() + } else if len(s) == 2 { + return &verticalMergeChunkIterator{ + a: s[0].ChunkIterator(), + b: s[1].ChunkIterator(), + } + } + return &verticalMergeChunkIterator{ + a: s[0].ChunkIterator(), + b: newVerticalMergeChunkIterator(s[1:]...), + } +} + +func (it *verticalMergeChunkIterator) Next() bool { + if it.Err() != nil { + return false + } + + if !it.initialized { + it.aok = it.a.Next() + it.bok = it.b.Next() + it.initialized = true + } + + if !it.aok && !it.bok { + return false + } + + if !it.aok { + it.curMeta = it.b.At() + it.bok = it.b.Next() + return true + } + if !it.bok { + it.curMeta = it.a.At() + it.aok = it.a.Next() + return true + } + + aCurMeta := it.a.At() + bCurMeta := it.b.At() + + if aCurMeta.MaxTime < bCurMeta.MinTime { + it.curMeta = aCurMeta + it.aok = it.a.Next() + return true + } + + if bCurMeta.MaxTime < aCurMeta.MinTime { + it.curMeta = bCurMeta + it.bok = it.b.Next() + return true + } + + it.curMeta, it.err = mergeOverlappingChunks(aCurMeta, bCurMeta, it.aReuseIter, it.bReuseIter) + if it.err != nil { + return false + } + + it.aok = it.a.Next() + it.bok = it.b.Next() + return true +} + +// TODO: https://github.com/prometheus/tsdb/issues/670 +func mergeOverlappingChunks(a, b chunks.Meta, aReuseIter, bReuseIter chunkenc.Iterator) (chunks.Meta, error) { + chk := chunkenc.NewXORChunk() + app, err := chk.Appender() + if err != nil { + return chunks.Meta{}, err + } + seriesIter := &verticalMergeSeriesIterator{ + a: a.Chunk.Iterator(aReuseIter), + b: b.Chunk.Iterator(bReuseIter), + } + + mint := int64(math.MaxInt64) + maxt := int64(math.MinInt64) + + // TODO: This can end up being up to 240 samples per chunk, so we need to have a case to split to two. + for seriesIter.Next() { + t, v := seriesIter.At() + app.Append(t, v) + + maxt = t + if mint == math.MaxInt64 { + mint = t + } + } + if err := seriesIter.Err(); err != nil { + return chunks.Meta{}, err + } + + return chunks.Meta{ + MinTime: mint, + MaxTime: maxt, + Chunk: chk, + }, nil +} + +func (it *verticalMergeChunkIterator) Seek(t int64) bool { + it.aok, it.bok = it.a.Seek(t), it.b.Seek(t) + it.initialized = true + return it.Next() +} + +func (it *verticalMergeChunkIterator) At() chunks.Meta { + return it.curMeta +} + +func (it *verticalMergeChunkIterator) Err() error { + if it.err != nil { + return it.err + } + if it.a.Err() != nil { + return it.a.Err() + } + return it.b.Err() +} + // chunkSeriesIterator implements a series iterator on top // of a list of time-sorted, non-overlapping chunks. type chunkSeriesIterator struct { @@ -1098,31 +1260,40 @@ func (it *chunkSeriesIterator) resetCurIterator() { it.cur = it.bufDelIter } -func (it *chunkSeriesIterator) Seek(t int64) (ok bool) { - if t > it.maxt { +func (it *chunkSeriesIterator) Seek(t int64) bool { + if it.Err() != nil || t > it.maxt || it.i > len(it.chunks)-1 { + // Exhaust iterator. + it.i = len(it.chunks) return false } - // Seek to the first valid value after t. if t < it.mint { t = it.mint } + currI := it.i for ; it.chunks[it.i].MaxTime < t; it.i++ { if it.i == len(it.chunks)-1 { + // Exhaust iterator. + it.i = len(it.chunks) return false } } - it.resetCurIterator() + if currI != it.i { + it.resetCurIterator() + } - for it.cur.Next() { - t0, _ := it.cur.At() - if t0 >= t { - return true + tc, _ := it.cur.At() + for t > tc { + if !it.cur.Next() { + // Exhaust iterator. + it.i = len(it.chunks) + return false } + tc, _ = it.cur.At() } - return false + return true } func (it *chunkSeriesIterator) At() (t int64, v float64) { @@ -1130,6 +1301,10 @@ func (it *chunkSeriesIterator) At() (t int64, v float64) { } func (it *chunkSeriesIterator) Next() bool { + if it.Err() != nil || it.i > len(it.chunks)-1 { + return false + } + if it.cur.Next() { t, _ := it.cur.At() @@ -1175,6 +1350,19 @@ func (it *deletedIterator) At() (int64, float64) { return it.it.At() } +func (it *deletedIterator) Seek(t int64) bool { + if atT, _ := it.At(); t >= atT { + return false + } + + for it.Next() { + if atT, _ := it.At(); t >= atT { + return true + } + } + return false +} + func (it *deletedIterator) Next() bool { Outer: for it.it.Next() { @@ -1210,3 +1398,85 @@ type errSeriesSet struct { func (s errSeriesSet) Next() bool { return false } func (s errSeriesSet) At() Series { return nil } func (s errSeriesSet) Err() error { return s.err } + +type chunkIterator struct { + chunks []chunks.Meta // series in time order + i int +} + +func (c *chunkIterator) Seek(t int64) bool { + if c.i >= len(c.chunks) { + return false + } + + for c.Next() { + if t >= c.At().MinTime { + return true + } + } + return false +} + +func (c *chunkIterator) Next() bool { + if c.i >= len(c.chunks) { + return false + } + c.i++ + return true +} + +func (c *chunkIterator) At() chunks.Meta { + return c.chunks[c.i-1] +} + +func (c *chunkIterator) Err() error { return nil } + +// chainedChunkIterator implements flat iteration for chunks iterated over a list +// of time-sorted, non-overlapping iterators for each series. +type chainedChunkIterator struct { + chain []ChunkIterator // chunk iterators for each series in time order + i int + err error +} + +func (c *chainedChunkIterator) Seek(t int64) bool { + if c.Err() != nil { + return false + } + + for c.i < len(c.chain) { + if c.chain[c.i].Seek(t) { + return true + } + if err := c.chain[c.i].Err(); err != nil { + c.err = err + return false + } + c.i++ + } + return false +} + +func (c *chainedChunkIterator) Next() bool { + if c.Err() != nil { + return false + } + + for c.i < len(c.chain) { + if c.chain[c.i].Next() { + return true + } + if err := c.chain[c.i].Err(); err != nil { + c.err = err + return false + } + c.i++ + } + return false +} + +func (c *chainedChunkIterator) At() chunks.Meta { + return c.chain[c.i].At() +} + +func (c *chainedChunkIterator) Err() error { return c.err } diff --git a/querier_test.go b/querier_test.go index 2be48fcd..38925a90 100644 --- a/querier_test.go +++ b/querier_test.go @@ -179,7 +179,7 @@ Outer: } } -func expandSeriesIterator(it SeriesIterator) (r []tsdbutil.Sample, err error) { +func expandSeriesIterator(it chunkenc.Iterator) (r []tsdbutil.Sample, err error) { for it.Next() { t, v := it.At() r = append(r, sample{t: t, v: v}) @@ -188,6 +188,13 @@ func expandSeriesIterator(it SeriesIterator) (r []tsdbutil.Sample, err error) { return r, it.Err() } +func expandChunkIterator(it ChunkIterator) (chks []chunks.Meta) { + for it.Next() { + chks = append(chks, it.At()) + } + return chks +} + type seriesSamples struct { lset map[string]string chunks [][]sample @@ -264,7 +271,7 @@ func TestBlockQuerier(t *testing.T) { newSeries := func(l map[string]string, s []tsdbutil.Sample) Series { return &mockSeries{ labels: func() labels.Labels { return labels.FromMap(l) }, - iterator: func() SeriesIterator { return newListSeriesIterator(s) }, + iterator: func() chunkenc.Iterator { return newListSeriesIterator(s) }, } } @@ -402,7 +409,7 @@ func TestBlockQuerierDelete(t *testing.T) { newSeries := func(l map[string]string, s []tsdbutil.Sample) Series { return &mockSeries{ labels: func() labels.Labels { return labels.FromMap(l) }, - iterator: func() SeriesIterator { return newListSeriesIterator(s) }, + iterator: func() chunkenc.Iterator { return newListSeriesIterator(s) }, } } @@ -658,28 +665,52 @@ func TestBaseChunkSeries(t *testing.T) { // TODO: Remove after simpleSeries is merged type itSeries struct { - si SeriesIterator + si chunkenc.Iterator } -func (s itSeries) Iterator() SeriesIterator { return s.si } -func (s itSeries) Labels() labels.Labels { return labels.Labels{} } +func (s itSeries) Iterator() chunkenc.Iterator { return s.si } +func (s itSeries) Labels() labels.Labels { return labels.Labels{} } +func (s itSeries) ChunkIterator() ChunkIterator { return nil } -func TestSeriesIterator(t *testing.T) { - itcases := []struct { - a, b, c []tsdbutil.Sample - exp []tsdbutil.Sample +type iteratorCase struct { + a, b, c, expected []tsdbutil.Sample - mint, maxt int64 - }{ - { - a: []tsdbutil.Sample{}, - b: []tsdbutil.Sample{}, - c: []tsdbutil.Sample{}, + mint, maxt int64 + + // seek is zero means do not test seek. + seek int64 + seekSuccess bool +} + +func (tc iteratorCase) test(t *testing.T, it chunkenc.Iterator) { + var r []tsdbutil.Sample + if tc.seek != 0 { + testutil.Equals(t, tc.seekSuccess, it.Seek(tc.seek)) + testutil.Equals(t, tc.seekSuccess, it.Seek(tc.seek)) // Next one should be noop. - exp: []tsdbutil.Sample{}, + if tc.seekSuccess { + // After successful seek iterator is ready. Grab the value. + t, v := it.At() + r = append(r, sample{t: t, v: v}) + } + } + expandedResult, err := expandSeriesIterator(it) + testutil.Ok(t, err) + r = append(r, expandedResult...) + testutil.Equals(t, tc.expected, r) +} + +func TestSeriesIterator(t *testing.T) { + cases := []iteratorCase{ + { + a: []tsdbutil.Sample{}, + b: []tsdbutil.Sample{}, + c: []tsdbutil.Sample{}, mint: math.MinInt64, maxt: math.MaxInt64, + + expected: nil, }, { a: []tsdbutil.Sample{ @@ -692,27 +723,12 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{7, 89}, sample{9, 8}, }, - - exp: []tsdbutil.Sample{ - sample{1, 2}, sample{2, 3}, sample{3, 5}, sample{6, 1}, sample{7, 89}, sample{9, 8}, - }, mint: math.MinInt64, maxt: math.MaxInt64, - }, - { - a: []tsdbutil.Sample{}, - b: []tsdbutil.Sample{ - sample{1, 2}, sample{2, 3}, sample{3, 5}, sample{6, 1}, - }, - c: []tsdbutil.Sample{ - sample{7, 89}, sample{9, 8}, - }, - exp: []tsdbutil.Sample{ + expected: []tsdbutil.Sample{ sample{1, 2}, sample{2, 3}, sample{3, 5}, sample{6, 1}, sample{7, 89}, sample{9, 8}, }, - mint: 2, - maxt: 8, }, { a: []tsdbutil.Sample{ @@ -724,32 +740,22 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{10, 22}, sample{203, 3493}, }, + mint: math.MinInt64, + maxt: math.MaxInt64, - exp: []tsdbutil.Sample{ + expected: []tsdbutil.Sample{ sample{1, 2}, sample{2, 3}, sample{3, 5}, sample{6, 1}, sample{7, 89}, sample{9, 8}, sample{10, 22}, sample{203, 3493}, }, - mint: 6, - maxt: 10, }, - } - - seekcases := []struct { - a, b, c []tsdbutil.Sample - - seek int64 - success bool - exp []tsdbutil.Sample - - mint, maxt int64 - }{ + // Seek cases. { - a: []tsdbutil.Sample{}, - b: []tsdbutil.Sample{}, - c: []tsdbutil.Sample{}, + a: []tsdbutil.Sample{}, + b: []tsdbutil.Sample{}, + c: []tsdbutil.Sample{}, + seek: 1, - seek: 0, - success: false, - exp: nil, + seekSuccess: false, + expected: nil, }, { a: []tsdbutil.Sample{ @@ -759,12 +765,12 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{7, 89}, sample{9, 8}, }, + seek: 10, + mint: math.MinInt64, + maxt: math.MaxInt64, - seek: 10, - success: false, - exp: nil, - mint: math.MinInt64, - maxt: math.MaxInt64, + seekSuccess: false, + expected: nil, }, { a: []tsdbutil.Sample{}, @@ -774,14 +780,14 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{7, 89}, sample{9, 8}, }, + seek: 2, + mint: math.MinInt64, + maxt: math.MaxInt64, - seek: 2, - success: true, - exp: []tsdbutil.Sample{ + seekSuccess: true, + expected: []tsdbutil.Sample{ sample{3, 5}, sample{6, 1}, sample{7, 89}, sample{9, 8}, }, - mint: 5, - maxt: 8, }, { a: []tsdbutil.Sample{ @@ -793,14 +799,14 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{10, 22}, sample{203, 3493}, }, + seek: 10, + mint: math.MinInt64, + maxt: math.MaxInt64, - seek: 10, - success: true, - exp: []tsdbutil.Sample{ + seekSuccess: true, + expected: []tsdbutil.Sample{ sample{10, 22}, sample{203, 3493}, }, - mint: 10, - maxt: 203, }, { a: []tsdbutil.Sample{ @@ -812,133 +818,143 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{10, 22}, sample{203, 3493}, }, + seek: 203, + mint: math.MinInt64, + maxt: math.MaxInt64, - seek: 203, - success: true, - exp: []tsdbutil.Sample{ + seekSuccess: true, + expected: []tsdbutil.Sample{ sample{203, 3493}, }, - mint: 7, - maxt: 203, }, } t.Run("Chunk", func(t *testing.T) { - for _, tc := range itcases { - chkMetas := []chunks.Meta{ - tsdbutil.ChunkFromSamples(tc.a), - tsdbutil.ChunkFromSamples(tc.b), - tsdbutil.ChunkFromSamples(tc.c), - } - res := newChunkSeriesIterator(chkMetas, nil, tc.mint, tc.maxt) - - smplValid := make([]tsdbutil.Sample, 0) - for _, s := range tc.exp { - if s.T() >= tc.mint && s.T() <= tc.maxt { - smplValid = append(smplValid, tsdbutil.Sample(s)) - } - } - exp := newListSeriesIterator(smplValid) - - smplExp, errExp := expandSeriesIterator(exp) - smplRes, errRes := expandSeriesIterator(res) - - testutil.Equals(t, errExp, errRes) - testutil.Equals(t, smplExp, smplRes) - } - - t.Run("Seek", func(t *testing.T) { - extra := []struct { - a, b, c []tsdbutil.Sample + // Only chunk implements time filtering, so add those cases here only. + limitedMinMaxtCases := []iteratorCase{ + { + a: []tsdbutil.Sample{}, + b: []tsdbutil.Sample{}, + c: []tsdbutil.Sample{}, + mint: 20, + maxt: 21, - seek int64 - success bool - exp []tsdbutil.Sample + expected: nil, + }, + { + a: []tsdbutil.Sample{ + sample{1, 2}, + sample{2, 3}, + sample{3, 5}, + sample{6, 1}, + }, + b: []tsdbutil.Sample{}, + c: []tsdbutil.Sample{ + sample{7, 89}, sample{9, 8}, + }, + mint: 2, + maxt: 8, - mint, maxt int64 - }{ - { - a: []tsdbutil.Sample{ - sample{6, 1}, - }, - b: []tsdbutil.Sample{ - sample{9, 8}, - }, - c: []tsdbutil.Sample{ - sample{10, 22}, sample{203, 3493}, - }, + expected: []tsdbutil.Sample{ + sample{2, 3}, sample{3, 5}, sample{6, 1}, sample{7, 89}, + }, + }, + { + a: []tsdbutil.Sample{ + sample{1, 2}, sample{2, 3}, sample{3, 5}, sample{6, 1}, + }, + b: []tsdbutil.Sample{ + sample{7, 89}, sample{9, 8}, + }, + c: []tsdbutil.Sample{ + sample{10, 22}, sample{203, 3493}, + }, + mint: 3, + maxt: math.MaxInt64, - seek: 203, - success: false, - exp: nil, - mint: 2, - maxt: 202, + expected: []tsdbutil.Sample{ + sample{3, 5}, sample{6, 1}, sample{7, 89}, sample{9, 8}, sample{10, 22}, sample{203, 3493}, }, - { - a: []tsdbutil.Sample{ - sample{6, 1}, - }, - b: []tsdbutil.Sample{ - sample{9, 8}, - }, - c: []tsdbutil.Sample{ - sample{10, 22}, sample{203, 3493}, - }, + }, + { + a: []tsdbutil.Sample{ + sample{1, 2}, sample{2, 3}, sample{3, 5}, sample{6, 1}, + }, + b: []tsdbutil.Sample{ + sample{7, 89}, sample{9, 8}, + }, + c: []tsdbutil.Sample{ + sample{10, 22}, sample{203, 3493}, + }, + mint: 6, + maxt: 10, - seek: 5, - success: true, - exp: []tsdbutil.Sample{sample{10, 22}}, - mint: 10, - maxt: 202, + expected: []tsdbutil.Sample{ + sample{6, 1}, sample{7, 89}, sample{9, 8}, sample{10, 22}, }, - } + }, + // Same with seek. + { + a: []tsdbutil.Sample{ + sample{6, 1}, + }, + b: []tsdbutil.Sample{ + sample{9, 8}, + }, + c: []tsdbutil.Sample{ + sample{10, 22}, sample{203, 3493}, + }, + seek: 203, + mint: 2, + maxt: 202, - seekcases2 := append(seekcases, extra...) + seekSuccess: false, + expected: nil, + }, + { + a: []tsdbutil.Sample{ + sample{6, 1}, + }, + b: []tsdbutil.Sample{ + sample{9, 8}, + }, + c: []tsdbutil.Sample{ + sample{10, 22}, sample{203, 3493}, + }, + seek: 5, + mint: 10, + maxt: 202, - for _, tc := range seekcases2 { + seekSuccess: true, + expected: []tsdbutil.Sample{sample{10, 22}}, + }, + } + for i, tc := range append(cases, limitedMinMaxtCases...) { + t.Run(fmt.Sprintf("%v", i), func(t *testing.T) { chkMetas := []chunks.Meta{ tsdbutil.ChunkFromSamples(tc.a), tsdbutil.ChunkFromSamples(tc.b), tsdbutil.ChunkFromSamples(tc.c), } - res := newChunkSeriesIterator(chkMetas, nil, tc.mint, tc.maxt) - - smplValid := make([]tsdbutil.Sample, 0) - for _, s := range tc.exp { - if s.T() >= tc.mint && s.T() <= tc.maxt { - smplValid = append(smplValid, tsdbutil.Sample(s)) - } - } - exp := newListSeriesIterator(smplValid) - - testutil.Equals(t, tc.success, res.Seek(tc.seek)) - - if tc.success { - // Init the list and then proceed to check. - remaining := exp.Next() - testutil.Assert(t, remaining == true, "") - - for remaining { - sExp, eExp := exp.At() - sRes, eRes := res.At() - testutil.Equals(t, eExp, eRes) - testutil.Equals(t, sExp, sRes) - - remaining = exp.Next() - testutil.Equals(t, remaining, res.Next()) - } - } - } - }) + tc.test(t, newChunkSeriesIterator(chkMetas, nil, tc.mint, tc.maxt)) + }) + } }) t.Run("Chain", func(t *testing.T) { + for i, tc := range cases { + t.Run(fmt.Sprintf("%v", i), func(t *testing.T) { + a, b, c := itSeries{newListSeriesIterator(tc.a)}, + itSeries{newListSeriesIterator(tc.b)}, + itSeries{newListSeriesIterator(tc.c)} + tc.test(t, newChainedSeriesIterator(a, b, c)) + }) + } + }) + + t.Run("Vertical", func(t *testing.T) { // Extra cases for overlapping series. - itcasesExtra := []struct { - a, b, c []tsdbutil.Sample - exp []tsdbutil.Sample - mint, maxt int64 - }{ + overlappingCases := []iteratorCase{ { a: []tsdbutil.Sample{ sample{1, 2}, sample{2, 3}, sample{3, 5}, sample{6, 1}, @@ -949,12 +965,12 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{2, 33}, sample{4, 44}, sample{10, 3}, }, + mint: math.MinInt64, + maxt: math.MaxInt64, - exp: []tsdbutil.Sample{ + expected: []tsdbutil.Sample{ sample{1, 2}, sample{2, 33}, sample{3, 5}, sample{4, 44}, sample{5, 49}, sample{6, 1}, sample{7, 89}, sample{9, 8}, sample{10, 3}, }, - mint: math.MinInt64, - maxt: math.MaxInt64, }, { a: []tsdbutil.Sample{ @@ -964,84 +980,64 @@ func TestSeriesIterator(t *testing.T) { c: []tsdbutil.Sample{ sample{1, 23}, sample{2, 342}, sample{3, 25}, sample{6, 11}, }, + mint: math.MinInt64, + maxt: math.MaxInt64, - exp: []tsdbutil.Sample{ + expected: []tsdbutil.Sample{ sample{1, 23}, sample{2, 342}, sample{3, 25}, sample{6, 11}, sample{9, 5}, sample{13, 1}, }, - mint: math.MinInt64, - maxt: math.MaxInt64, }, } - - for _, tc := range itcases { - a, b, c := itSeries{newListSeriesIterator(tc.a)}, - itSeries{newListSeriesIterator(tc.b)}, - itSeries{newListSeriesIterator(tc.c)} - - res := newChainedSeriesIterator(a, b, c) - exp := newListSeriesIterator([]tsdbutil.Sample(tc.exp)) - - smplExp, errExp := expandSeriesIterator(exp) - smplRes, errRes := expandSeriesIterator(res) - - testutil.Equals(t, errExp, errRes) - testutil.Equals(t, smplExp, smplRes) - } - - for _, tc := range append(itcases, itcasesExtra...) { - a, b, c := itSeries{newListSeriesIterator(tc.a)}, - itSeries{newListSeriesIterator(tc.b)}, - itSeries{newListSeriesIterator(tc.c)} - - res := newVerticalMergeSeriesIterator(a, b, c) - exp := newListSeriesIterator([]tsdbutil.Sample(tc.exp)) - - smplExp, errExp := expandSeriesIterator(exp) - smplRes, errRes := expandSeriesIterator(res) - - testutil.Equals(t, errExp, errRes) - testutil.Equals(t, smplExp, smplRes) + for i, tc := range append(cases, overlappingCases...) { + t.Run(fmt.Sprintf("%v", i), func(t *testing.T) { + a, b, c := + itSeries{newListSeriesIterator(tc.a)}, + itSeries{newListSeriesIterator(tc.b)}, + itSeries{newListSeriesIterator(tc.c)} + tc.test(t, newVerticalMergeSeriesIterator(a, b, c)) + }) } + }) +} - t.Run("Seek", func(t *testing.T) { - for _, tc := range seekcases { - ress := []SeriesIterator{ - newChainedSeriesIterator( - itSeries{newListSeriesIterator(tc.a)}, - itSeries{newListSeriesIterator(tc.b)}, - itSeries{newListSeriesIterator(tc.c)}, - ), - newVerticalMergeSeriesIterator( - itSeries{newListSeriesIterator(tc.a)}, - itSeries{newListSeriesIterator(tc.b)}, - itSeries{newListSeriesIterator(tc.c)}, - ), - } - - for _, res := range ress { - exp := newListSeriesIterator(tc.exp) +func TestChunkIterator(t *testing.T) { + it := &chunkIterator{} + testutil.Equals(t, []chunks.Meta(nil), expandChunkIterator(it)) + testutil.Equals(t, false, it.Next()) - testutil.Equals(t, tc.success, res.Seek(tc.seek)) + chks := []chunks.Meta{ + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{1, 1}, sample{1, 2}}), + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{2, 1}, sample{2, 2}}), + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{3, 1}, sample{3, 2}}), + } + it = &chunkIterator{chunks: chks} + testutil.Equals(t, chks, expandChunkIterator(it)) + testutil.Equals(t, false, it.Next()) +} - if tc.success { - // Init the list and then proceed to check. - remaining := exp.Next() - testutil.Assert(t, remaining == true, "") +func TestChainedChunkIterator(t *testing.T) { + it := &chainedChunkIterator{} + testutil.Equals(t, []chunks.Meta(nil), expandChunkIterator(it)) + testutil.Equals(t, false, it.Next()) - for remaining { - sExp, eExp := exp.At() - sRes, eRes := res.At() - testutil.Equals(t, eExp, eRes) - testutil.Equals(t, sExp, sRes) + chks1 := []chunks.Meta{ + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{1, 1}, sample{1, 2}}), + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{2, 1}, sample{2, 2}}), + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{3, 1}, sample{3, 2}}), + } + chks2 := []chunks.Meta(nil) + chks3 := []chunks.Meta{ + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{4, 1}, sample{4, 2}}), + tsdbutil.ChunkFromSamples([]tsdbutil.Sample{sample{5, 1}, sample{5, 2}}), + } - remaining = exp.Next() - testutil.Equals(t, remaining, res.Next()) - } - } - } - } - }) - }) + it = &chainedChunkIterator{chain: []ChunkIterator{ + &chunkIterator{chunks: chks1}, + &chunkIterator{chunks: chks2}, + &chunkIterator{chunks: chks3}, + }} + testutil.Equals(t, append(chks1, chks3...), expandChunkIterator(it)) + testutil.Equals(t, false, it.Next()) } // Regression for: https://github.com/prometheus/tsdb/pull/97 @@ -1053,8 +1049,9 @@ func TestChunkSeriesIterator_DoubleSeek(t *testing.T) { } res := newChunkSeriesIterator(chkMetas, nil, 2, 8) - testutil.Assert(t, res.Seek(1) == true, "") - testutil.Assert(t, res.Seek(2) == true, "") + testutil.Assert(t, res.Seek(1), "") + testutil.Assert(t, res.Seek(2), "") + testutil.Assert(t, res.Seek(2), "") ts, v := res.At() testutil.Equals(t, int64(2), ts) testutil.Equals(t, float64(2), v) @@ -1071,12 +1068,12 @@ func TestChunkSeriesIterator_SeekInCurrentChunk(t *testing.T) { it := newChunkSeriesIterator(metas, nil, 1, 7) - testutil.Assert(t, it.Next() == true, "") + testutil.Assert(t, it.Next(), "") ts, v := it.At() testutil.Equals(t, int64(1), ts) testutil.Equals(t, float64(2), v) - testutil.Assert(t, it.Seek(4) == true, "") + testutil.Assert(t, it.Seek(4), "") ts, v = it.At() testutil.Equals(t, int64(5), ts) testutil.Equals(t, float64(6), v) @@ -1430,17 +1427,18 @@ func (m mockIndex) LabelNames() ([]string, error) { type mockSeries struct { labels func() labels.Labels - iterator func() SeriesIterator + iterator func() chunkenc.Iterator } func newSeries(l map[string]string, s []tsdbutil.Sample) Series { return &mockSeries{ labels: func() labels.Labels { return labels.FromMap(l) }, - iterator: func() SeriesIterator { return newListSeriesIterator(s) }, + iterator: func() chunkenc.Iterator { return newListSeriesIterator(s) }, } } -func (m *mockSeries) Labels() labels.Labels { return m.labels() } -func (m *mockSeries) Iterator() SeriesIterator { return m.iterator() } +func (m *mockSeries) Labels() labels.Labels { return m.labels() } +func (m *mockSeries) Iterator() chunkenc.Iterator { return m.iterator() } +func (m *mockSeries) ChunkIterator() ChunkIterator { return nil } type listSeriesIterator struct { list []tsdbutil.Sample @@ -1465,11 +1463,13 @@ func (it *listSeriesIterator) Seek(t int64) bool { if it.idx == -1 { it.idx = 0 } + // Do binary search between current position and end. - it.idx = sort.Search(len(it.list)-it.idx, func(i int) bool { + pos := sort.Search(len(it.list)-it.idx, func(i int) bool { s := it.list[i+it.idx] return s.T() >= t }) + it.idx += pos return it.idx < len(it.list) } diff --git a/tsdbutil/buffer.go b/tsdbutil/buffer.go index dc2d960d..8f615d8e 100644 --- a/tsdbutil/buffer.go +++ b/tsdbutil/buffer.go @@ -15,25 +15,13 @@ package tsdbutil import ( "math" -) -// SeriesIterator iterates over the data of a time series. -type SeriesIterator interface { - // Seek advances the iterator forward to the given timestamp. - // If there's no value exactly at t, it advances to the first value - // after t. - Seek(t int64) bool - // At returns the current timestamp/value pair. - At() (t int64, v float64) - // Next advances the iterator by one. - Next() bool - // Err returns the current error. - Err() error -} + "github.com/prometheus/tsdb/chunkenc" +) // BufferedSeriesIterator wraps an iterator with a look-back buffer. type BufferedSeriesIterator struct { - it SeriesIterator + it chunkenc.Iterator buf *sampleRing lastTime int64 @@ -41,7 +29,7 @@ type BufferedSeriesIterator struct { // NewBuffer returns a new iterator that buffers the values within the time range // of the current element and the duration of delta before. -func NewBuffer(it SeriesIterator, delta int64) *BufferedSeriesIterator { +func NewBuffer(it chunkenc.Iterator, delta int64) *BufferedSeriesIterator { return &BufferedSeriesIterator{ it: it, buf: newSampleRing(delta, 16), @@ -56,7 +44,7 @@ func (b *BufferedSeriesIterator) PeekBack() (t int64, v float64, ok bool) { } // Buffer returns an iterator over the buffered data. -func (b *BufferedSeriesIterator) Buffer() SeriesIterator { +func (b *BufferedSeriesIterator) Buffer() chunkenc.Iterator { return b.buf.iterator() } @@ -145,7 +133,7 @@ func (r *sampleRing) reset() { r.f = 0 } -func (r *sampleRing) iterator() SeriesIterator { +func (r *sampleRing) iterator() chunkenc.Iterator { return &sampleRingIterator{r: r, i: -1} }