Commit 34426766 by Bartlomiej Plotka

Unify Iterator interfaces. All point to storage now.

This is part of https://github.com/prometheus/prometheus/pull/5882 that can be done to simplify things.
All todos I added will be fixed in follow up PRs.

* querier.Querier, querier.Appender, querier.SeriesSet, and querier.Series interfaces merged
with storage interface.go. All imports that.
* querier.SeriesIterator replaced by chunkenc.Iterator
* Added chunkenc.Iterator.Seek method and tests for xor implementation (?)
* Since we properly handle SelectParams for Select methods I adjusted min max
based on that. This should help in terms of performance for queries with functions like offset.
* added Seek to deletedIterator and test.
* storage/tsdb was removed as it was only a unnecessary glue with incompatible structs.

No logic was changed, only different source of abstractions, so no need for benchmarks.
Signed-off-by: 's avatarBartlomiej Plotka <bwplotka@gmail.com>
parent 489a9aa7
......@@ -56,7 +56,7 @@ import (
"github.com/prometheus/prometheus/scrape"
"github.com/prometheus/prometheus/storage"
"github.com/prometheus/prometheus/storage/remote"
"github.com/prometheus/prometheus/storage/tsdb"
"github.com/prometheus/prometheus/tsdb"
"github.com/prometheus/prometheus/util/strutil"
"github.com/prometheus/prometheus/web"
)
......
......@@ -428,10 +428,7 @@ func (t *Test) exec(tc testCommand) error {
t.clear()
case *loadCmd:
app, err := t.storage.Appender()
if err != nil {
return err
}
app := t.storage.Appender()
if err := cmd.append(app); err != nil {
app.Rollback()
return err
......@@ -641,10 +638,7 @@ func (ll *LazyLoader) clear() {
// appendTill appends the defined time series to the storage till the given timestamp (in milliseconds).
func (ll *LazyLoader) appendTill(ts int64) error {
app, err := ll.storage.Appender()
if err != nil {
return err
}
app := ll.storage.Appender()
for h, smpls := range ll.loadCmd.defs {
m := ll.loadCmd.metrics[h]
for i, s := range smpls {
......
......@@ -20,6 +20,7 @@ import (
"strings"
"github.com/pkg/errors"
"github.com/prometheus/prometheus/tsdb/chunkenc"
"github.com/prometheus/prometheus/pkg/labels"
"github.com/prometheus/prometheus/storage"
......@@ -275,7 +276,7 @@ func (ss *StorageSeries) Labels() labels.Labels {
}
// Iterator returns a new iterator of the data of the series.
func (ss *StorageSeries) Iterator() storage.SeriesIterator {
func (ss *StorageSeries) Iterator() chunkenc.Iterator {
return newStorageSeriesIterator(ss.series)
}
......
......@@ -587,12 +587,7 @@ func (g *Group) Eval(ctx context.Context, ts time.Time) {
numDuplicates = 0
)
app, err := g.opts.Appendable.Appender()
if err != nil {
level.Warn(g.logger).Log("msg", "creating appender failed", "err", err)
return
}
app := g.opts.Appendable.Appender()
seriesReturned := make(map[string]labels.Labels, len(g.seriesInPreviousEval[i]))
for _, s := range vector {
if _, err := app.Add(s.Metric, s.T, s.V); err != nil {
......@@ -645,14 +640,10 @@ func (g *Group) cleanupStaleSeries(ts time.Time) {
if len(g.staleSeries) == 0 {
return
}
app, err := g.opts.Appendable.Appender()
if err != nil {
level.Warn(g.logger).Log("msg", "creating appender failed", "err", err)
return
}
app := g.opts.Appendable.Appender()
for _, s := range g.staleSeries {
// Rule that produced series no longer configured, mark it stale.
_, err = app.Add(s, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN))
_, err := app.Add(s, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN))
switch err {
case nil:
case storage.ErrOutOfOrderSample, storage.ErrDuplicateSampleForTimestamp:
......@@ -836,11 +827,6 @@ type Manager struct {
logger log.Logger
}
// Appendable returns an Appender.
type Appendable interface {
Appender() (storage.Appender, error)
}
// NotifyFunc sends notifications about a set of alerts generated by the given expression.
type NotifyFunc func(ctx context.Context, expr string, alerts ...*Alert)
......@@ -850,7 +836,7 @@ type ManagerOptions struct {
QueryFunc QueryFunc
NotifyFunc NotifyFunc
Context context.Context
Appendable Appendable
Appendable storage.Appendable
TSDB storage.Storage
Logger log.Logger
Registerer prometheus.Registerer
......
......@@ -540,7 +540,7 @@ func TestStaleness(t *testing.T) {
})
// A time series that has two samples and then goes stale.
app, _ := storage.Appender()
app := storage.Appender()
app.Add(labels.FromStrings(model.MetricNameLabel, "a"), 0, 1)
app.Add(labels.FromStrings(model.MetricNameLabel, "a"), 1000, 2)
app.Add(labels.FromStrings(model.MetricNameLabel, "a"), 2000, math.Float64frombits(value.StaleNaN))
......@@ -868,7 +868,7 @@ func TestNotify(t *testing.T) {
Opts: opts,
})
app, _ := storage.Appender()
app := storage.Appender()
app.Add(labels.FromStrings(model.MetricNameLabel, "a"), 1000, 2)
app.Add(labels.FromStrings(model.MetricNameLabel, "a"), 2000, 3)
app.Add(labels.FromStrings(model.MetricNameLabel, "a"), 5000, 3)
......
......@@ -20,16 +20,16 @@ import (
type nopAppendable struct{}
func (a nopAppendable) Appender() (storage.Appender, error) {
return nopAppender{}, nil
func (a nopAppendable) Appender() storage.Appender {
return nopAppender{}
}
type nopAppender struct{}
func (a nopAppender) Add(labels.Labels, int64, float64) (uint64, error) { return 0, nil }
func (a nopAppender) AddFast(labels.Labels, uint64, int64, float64) error { return nil }
func (a nopAppender) Commit() error { return nil }
func (a nopAppender) Rollback() error { return nil }
func (a nopAppender) Add(labels.Labels, int64, float64) (uint64, error) { return 0, nil }
func (a nopAppender) AddFast(uint64, int64, float64) error { return nil }
func (a nopAppender) Commit() error { return nil }
func (a nopAppender) Rollback() error { return nil }
type sample struct {
metric labels.Labels
......@@ -42,18 +42,21 @@ type sample struct {
type collectResultAppender struct {
next storage.Appender
result []sample
mapper map[uint64]labels.Labels
}
func (a *collectResultAppender) AddFast(m labels.Labels, ref uint64, t int64, v float64) error {
func (a *collectResultAppender) AddFast(ref uint64, t int64, v float64) error {
if a.next == nil {
return storage.ErrNotFound
}
err := a.next.AddFast(m, ref, t, v)
err := a.next.AddFast(ref, t, v)
if err != nil {
return err
}
a.result = append(a.result, sample{
metric: m,
metric: a.mapper[ref],
t: t,
v: v,
})
......@@ -69,7 +72,17 @@ func (a *collectResultAppender) Add(m labels.Labels, t int64, v float64) (uint64
if a.next == nil {
return 0, nil
}
return a.next.Add(m, t, v)
if a.mapper == nil {
a.mapper = map[uint64]labels.Labels{}
}
ref, err := a.next.Add(m, t, v)
if err != nil {
return 0, err
}
a.mapper[ref] = m
return ref, nil
}
func (a *collectResultAppender) Commit() error { return nil }
......
......@@ -100,13 +100,8 @@ func (mc *MetadataMetricsCollector) Collect(ch chan<- prometheus.Metric) {
}
}
// Appendable returns an Appender.
type Appendable interface {
Appender() (storage.Appender, error)
}
// NewManager is the Manager constructor
func NewManager(logger log.Logger, app Appendable) *Manager {
func NewManager(logger log.Logger, app storage.Appendable) *Manager {
if logger == nil {
logger = log.NewNopLogger()
}
......@@ -127,7 +122,7 @@ func NewManager(logger log.Logger, app Appendable) *Manager {
// when receiving new target groups form the discovery manager.
type Manager struct {
logger log.Logger
append Appendable
append storage.Appendable
graceShut chan struct{}
jitterSeed uint64 // Global jitterSeed seed is used to spread scrape workload across HA setup.
......
......@@ -156,7 +156,7 @@ func init() {
// scrapePool manages scrapes for sets of targets.
type scrapePool struct {
appendable Appendable
appendable storage.Appendable
logger log.Logger
mtx sync.RWMutex
......@@ -187,7 +187,7 @@ const maxAheadTime = 10 * time.Minute
type labelsMutator func(labels.Labels) labels.Labels
func newScrapePool(cfg *config.ScrapeConfig, app Appendable, jitterSeed uint64, logger log.Logger) (*scrapePool, error) {
func newScrapePool(cfg *config.ScrapeConfig, app storage.Appendable, jitterSeed uint64, logger log.Logger) (*scrapePool, error) {
targetScrapePools.Inc()
if logger == nil {
logger = log.NewNopLogger()
......@@ -228,13 +228,7 @@ func newScrapePool(cfg *config.ScrapeConfig, app Appendable, jitterSeed uint64,
return mutateSampleLabels(l, opts.target, opts.honorLabels, opts.mrc)
},
func(l labels.Labels) labels.Labels { return mutateReportSampleLabels(l, opts.target) },
func() storage.Appender {
app, err := app.Appender()
if err != nil {
panic(err)
}
return appender(app, opts.limit)
},
func() storage.Appender { return appender(app.Appender(), opts.limit) },
cache,
jitterSeed,
opts.honorTimestamps,
......@@ -1112,7 +1106,7 @@ loop:
}
ce, ok := sl.cache.get(yoloString(met))
if ok {
switch err = app.AddFast(ce.lset, ce.ref, t, v); err {
switch err = app.AddFast(ce.ref, t, v); err {
case nil:
if tp == nil {
sl.cache.trackStaleness(ce.hash, ce.lset)
......@@ -1323,7 +1317,7 @@ func (sl *scrapeLoop) reportStale(start time.Time) error {
func (sl *scrapeLoop) addReportSample(app storage.Appender, s string, t int64, v float64) error {
ce, ok := sl.cache.get(s)
if ok {
err := app.AddFast(ce.lset, ce.ref, t, v)
err := app.AddFast(ce.ref, t, v)
switch err {
case nil:
return nil
......
......@@ -644,8 +644,7 @@ func TestScrapeLoopSeriesAdded(t *testing.T) {
s := teststorage.New(t)
defer s.Close()
app, err := s.Appender()
testutil.Ok(t, err)
app := s.Appender()
ctx, cancel := context.WithCancel(context.Background())
sl := newScrapeLoop(ctx,
......@@ -788,8 +787,7 @@ func TestScrapeLoopCache(t *testing.T) {
s := teststorage.New(t)
defer s.Close()
sapp, err := s.Appender()
testutil.Ok(t, err)
sapp := s.Appender()
appender := &collectResultAppender{next: sapp}
var (
......@@ -866,8 +864,7 @@ func TestScrapeLoopCacheMemoryExhaustionProtection(t *testing.T) {
s := teststorage.New(t)
defer s.Close()
sapp, err := s.Appender()
testutil.Ok(t, err)
sapp := s.Appender()
appender := &collectResultAppender{next: sapp}
var (
......@@ -1092,8 +1089,7 @@ func TestScrapeLoop_ChangingMetricString(t *testing.T) {
s := teststorage.New(t)
defer s.Close()
app, err := s.Appender()
testutil.Ok(t, err)
app := s.Appender()
capp := &collectResultAppender{next: app}
......@@ -1108,7 +1104,7 @@ func TestScrapeLoop_ChangingMetricString(t *testing.T) {
)
now := time.Now()
_, _, _, err = sl.append([]byte(`metric_a{a="1",b="1"} 1`), "", now)
_, _, _, err := sl.append([]byte(`metric_a{a="1",b="1"} 1`), "", now)
testutil.Ok(t, err)
_, _, _, err = sl.append([]byte(`metric_a{b="1",a="1"} 2`), "", now.Add(time.Minute))
......@@ -1273,8 +1269,8 @@ func (app *errorAppender) Add(lset labels.Labels, t int64, v float64) (uint64, e
}
}
func (app *errorAppender) AddFast(lset labels.Labels, ref uint64, t int64, v float64) error {
return app.collectResultAppender.AddFast(lset, ref, t, v)
func (app *errorAppender) AddFast(ref uint64, t int64, v float64) error {
return app.collectResultAppender.AddFast(ref, t, v)
}
func TestScrapeLoopAppendGracefullyIfAmendOrOutOfOrderOrOutOfBounds(t *testing.T) {
......@@ -1498,8 +1494,7 @@ func TestScrapeLoop_RespectTimestamps(t *testing.T) {
s := teststorage.New(t)
defer s.Close()
app, err := s.Appender()
testutil.Ok(t, err)
app := s.Appender()
capp := &collectResultAppender{next: app}
......@@ -1513,7 +1508,7 @@ func TestScrapeLoop_RespectTimestamps(t *testing.T) {
)
now := time.Now()
_, _, _, err = sl.append([]byte(`metric_a{a="1",b="1"} 1 0`), "", now)
_, _, _, err := sl.append([]byte(`metric_a{a="1",b="1"} 1 0`), "", now)
testutil.Ok(t, err)
want := []sample{
......@@ -1530,8 +1525,7 @@ func TestScrapeLoop_DiscardTimestamps(t *testing.T) {
s := teststorage.New(t)
defer s.Close()
app, err := s.Appender()
testutil.Ok(t, err)
app := s.Appender()
capp := &collectResultAppender{next: app}
......@@ -1545,7 +1539,7 @@ func TestScrapeLoop_DiscardTimestamps(t *testing.T) {
)
now := time.Now()
_, _, _, err = sl.append([]byte(`metric_a{a="1",b="1"} 1 0`), "", now)
_, _, _, err := sl.append([]byte(`metric_a{a="1",b="1"} 1 0`), "", now)
testutil.Ok(t, err)
want := []sample{
......@@ -1562,8 +1556,7 @@ func TestScrapeLoopDiscardDuplicateLabels(t *testing.T) {
s := teststorage.New(t)
defer s.Close()
app, err := s.Appender()
testutil.Ok(t, err)
app := s.Appender()
ctx, cancel := context.WithCancel(context.Background())
sl := newScrapeLoop(ctx,
......@@ -1579,7 +1572,7 @@ func TestScrapeLoopDiscardDuplicateLabels(t *testing.T) {
defer cancel()
// We add a good and a bad metric to check that both are discarded.
_, _, _, err = sl.append([]byte("test_metric{le=\"500\"} 1\ntest_metric{le=\"600\",le=\"700\"} 1\n"), "", time.Time{})
_, _, _, err := sl.append([]byte("test_metric{le=\"500\"} 1\ntest_metric{le=\"600\",le=\"700\"} 1\n"), "", time.Time{})
testutil.NotOk(t, err)
q, err := s.Querier(ctx, time.Time{}.UnixNano(), 0)
......
......@@ -303,14 +303,14 @@ func (app *limitAppender) Add(lset labels.Labels, t int64, v float64) (uint64, e
return ref, nil
}
func (app *limitAppender) AddFast(lset labels.Labels, ref uint64, t int64, v float64) error {
func (app *limitAppender) AddFast(ref uint64, t int64, v float64) error {
if !value.IsStaleNaN(v) {
app.i++
if app.i > app.limit {
return errSampleLimit
}
}
err := app.Appender.AddFast(lset, ref, t, v)
err := app.Appender.AddFast(ref, t, v)
return err
}
......@@ -332,11 +332,11 @@ func (app *timeLimitAppender) Add(lset labels.Labels, t int64, v float64) (uint6
return ref, nil
}
func (app *timeLimitAppender) AddFast(lset labels.Labels, ref uint64, t int64, v float64) error {
func (app *timeLimitAppender) AddFast(ref uint64, t int64, v float64) error {
if t > app.maxTime {
return storage.ErrOutOfBounds
}
err := app.Appender.AddFast(lset, ref, t, v)
err := app.Appender.AddFast(ref, t, v)
return err
}
......
......@@ -15,11 +15,13 @@ package storage
import (
"math"
"github.com/prometheus/prometheus/tsdb/chunkenc"
)
// BufferedSeriesIterator wraps an iterator with a look-back buffer.
type BufferedSeriesIterator struct {
it SeriesIterator
it chunkenc.Iterator
buf *sampleRing
delta int64
......@@ -36,7 +38,7 @@ func NewBuffer(delta int64) *BufferedSeriesIterator {
// NewBufferIterator returns a new iterator that buffers the values within the
// time range of the current element and the duration of delta before.
func NewBufferIterator(it SeriesIterator, delta int64) *BufferedSeriesIterator {
func NewBufferIterator(it chunkenc.Iterator, delta int64) *BufferedSeriesIterator {
bit := &BufferedSeriesIterator{
buf: newSampleRing(delta, 16),
delta: delta,
......@@ -48,7 +50,7 @@ func NewBufferIterator(it SeriesIterator, delta int64) *BufferedSeriesIterator {
// Reset re-uses the buffer with a new iterator, resetting the buffered time
// delta to its original value.
func (b *BufferedSeriesIterator) Reset(it SeriesIterator) {
func (b *BufferedSeriesIterator) Reset(it chunkenc.Iterator) {
b.it = it
b.lastTime = math.MinInt64
b.ok = true
......@@ -70,7 +72,7 @@ func (b *BufferedSeriesIterator) PeekBack(n int) (t int64, v float64, ok bool) {
// Buffer returns an iterator over the buffered data. Invalidates previously
// returned iterators.
func (b *BufferedSeriesIterator) Buffer() SeriesIterator {
func (b *BufferedSeriesIterator) Buffer() chunkenc.Iterator {
return b.buf.iterator()
}
......@@ -159,7 +161,7 @@ func (r *sampleRing) reset() {
}
// Returns the current iterator. Invalidates previously returned iterators.
func (r *sampleRing) iterator() SeriesIterator {
func (r *sampleRing) iterator() chunkenc.Iterator {
r.it.r = r
r.it.i = -1
return &r.it
......
......@@ -19,6 +19,7 @@ import (
"testing"
"github.com/prometheus/prometheus/pkg/labels"
"github.com/prometheus/prometheus/tsdb/chunkenc"
"github.com/prometheus/prometheus/util/testutil"
)
......@@ -190,7 +191,7 @@ func (m *mockSeriesIterator) Err() error { return m.err() }
type mockSeries struct {
labels func() labels.Labels
iterator func() SeriesIterator
iterator func() chunkenc.Iterator
}
func newMockSeries(lset labels.Labels, samples []sample) Series {
......@@ -198,14 +199,14 @@ func newMockSeries(lset labels.Labels, samples []sample) Series {
labels: func() labels.Labels {
return lset
},
iterator: func() SeriesIterator {
iterator: func() chunkenc.Iterator {
return newListSeriesIterator(samples)
},
}
}
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() }
type listSeriesIterator struct {
list []sample
......
......@@ -24,6 +24,7 @@ import (
"github.com/pkg/errors"
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/pkg/labels"
"github.com/prometheus/prometheus/tsdb/chunkenc"
)
type fanout struct {
......@@ -87,25 +88,17 @@ func (f *fanout) Querier(ctx context.Context, mint, maxt int64) (Querier, error)
return NewMergeQuerier(primaryQuerier, queriers), nil
}
func (f *fanout) Appender() (Appender, error) {
primary, err := f.primary.Appender()
if err != nil {
return nil, err
}
func (f *fanout) Appender() Appender {
primary := f.primary.Appender()
secondaries := make([]Appender, 0, len(f.secondaries))
for _, storage := range f.secondaries {
appender, err := storage.Appender()
if err != nil {
return nil, err
}
secondaries = append(secondaries, appender)
secondaries = append(secondaries, storage.Appender())
}
return &fanoutAppender{
logger: f.logger,
primary: primary,
secondaries: secondaries,
}, nil
}
}
// Close closes the storage and all its underlying resources.
......@@ -146,13 +139,13 @@ func (f *fanoutAppender) Add(l labels.Labels, t int64, v float64) (uint64, error
return ref, nil
}
func (f *fanoutAppender) AddFast(l labels.Labels, ref uint64, t int64, v float64) error {
if err := f.primary.AddFast(l, ref, t, v); err != nil {
func (f *fanoutAppender) AddFast(ref uint64, t int64, v float64) error {
if err := f.primary.AddFast(ref, t, v); err != nil {
return err
}
for _, appender := range f.secondaries {
if _, err := appender.Add(l, t, v); err != nil {
if err := appender.AddFast(ref, t, v); err != nil {
return err
}
}
......@@ -521,8 +514,8 @@ func (m *mergeSeries) Labels() labels.Labels {
return m.labels
}
func (m *mergeSeries) Iterator() SeriesIterator {
iterators := make([]SeriesIterator, 0, len(m.series))
func (m *mergeSeries) Iterator() chunkenc.Iterator {
iterators := make([]chunkenc.Iterator, 0, len(m.series))
for _, s := range m.series {
iterators = append(iterators, s.Iterator())
}
......@@ -530,11 +523,11 @@ func (m *mergeSeries) Iterator() SeriesIterator {
}
type mergeIterator struct {
iterators []SeriesIterator
iterators []chunkenc.Iterator
h seriesIteratorHeap
}
func newMergeIterator(iterators []SeriesIterator) SeriesIterator {
func newMergeIterator(iterators []chunkenc.Iterator) chunkenc.Iterator {
return &mergeIterator{
iterators: iterators,
h: nil,
......@@ -581,7 +574,7 @@ func (c *mergeIterator) Next() bool {
break
}
iter := heap.Pop(&c.h).(SeriesIterator)
iter := heap.Pop(&c.h).(chunkenc.Iterator)
if iter.Next() {
heap.Push(&c.h, iter)
}
......@@ -599,7 +592,7 @@ func (c *mergeIterator) Err() error {
return nil
}
type seriesIteratorHeap []SeriesIterator
type seriesIteratorHeap []chunkenc.Iterator
func (h seriesIteratorHeap) Len() int { return len(h) }
func (h seriesIteratorHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
......@@ -611,7 +604,7 @@ func (h seriesIteratorHeap) Less(i, j int) bool {
}
func (h *seriesIteratorHeap) Push(x interface{}) {
*h = append(*h, x.(SeriesIterator))
*h = append(*h, x.(chunkenc.Iterator))
}
func (h *seriesIteratorHeap) Pop() interface{} {
......
......@@ -19,6 +19,7 @@ import (
"testing"
"github.com/prometheus/prometheus/pkg/labels"
"github.com/prometheus/prometheus/tsdb/chunkenc"
"github.com/prometheus/prometheus/util/testutil"
)
......@@ -122,24 +123,24 @@ func TestMergeSeriesSet(t *testing.T) {
func TestMergeIterator(t *testing.T) {
for _, tc := range []struct {
input []SeriesIterator
input []chunkenc.Iterator
expected []sample
}{
{
input: []SeriesIterator{
input: []chunkenc.Iterator{
newListSeriesIterator([]sample{{0, 0}, {1, 1}}),
},
expected: []sample{{0, 0}, {1, 1}},
},
{
input: []SeriesIterator{
input: []chunkenc.Iterator{
newListSeriesIterator([]sample{{0, 0}, {1, 1}}),
newListSeriesIterator([]sample{{2, 2}, {3, 3}}),
},
expected: []sample{{0, 0}, {1, 1}, {2, 2}, {3, 3}},
},
{
input: []SeriesIterator{
input: []chunkenc.Iterator{
newListSeriesIterator([]sample{{0, 0}, {3, 3}}),
newListSeriesIterator([]sample{{1, 1}, {4, 4}}),
newListSeriesIterator([]sample{{2, 2}, {5, 5}}),
......@@ -147,7 +148,7 @@ func TestMergeIterator(t *testing.T) {
expected: []sample{{0, 0}, {1, 1}, {2, 2}, {3, 3}, {4, 4}, {5, 5}},
},
{
input: []SeriesIterator{
input: []chunkenc.Iterator{
newListSeriesIterator([]sample{{0, 0}, {1, 1}}),
newListSeriesIterator([]sample{{0, 0}, {2, 2}}),
newListSeriesIterator([]sample{{2, 2}, {3, 3}}),
......@@ -163,19 +164,19 @@ func TestMergeIterator(t *testing.T) {
func TestMergeIteratorSeek(t *testing.T) {
for _, tc := range []struct {
input []SeriesIterator
input []chunkenc.Iterator
seek int64
expected []sample
}{
{
input: []SeriesIterator{
input: []chunkenc.Iterator{
newListSeriesIterator([]sample{{0, 0}, {1, 1}, {2, 2}}),
},
seek: 1,
expected: []sample{{1, 1}, {2, 2}},
},
{
input: []SeriesIterator{
input: []chunkenc.Iterator{
newListSeriesIterator([]sample{{0, 0}, {1, 1}}),
newListSeriesIterator([]sample{{2, 2}, {3, 3}}),
},
......@@ -183,7 +184,7 @@ func TestMergeIteratorSeek(t *testing.T) {
expected: []sample{{2, 2}, {3, 3}},
},
{
input: []SeriesIterator{
input: []chunkenc.Iterator{
newListSeriesIterator([]sample{{0, 0}, {3, 3}}),
newListSeriesIterator([]sample{{1, 1}, {4, 4}}),
newListSeriesIterator([]sample{{2, 2}, {5, 5}}),
......@@ -203,7 +204,7 @@ func TestMergeIteratorSeek(t *testing.T) {
}
}
func drainSamples(iter SeriesIterator) []sample {
func drainSamples(iter chunkenc.Iterator) []sample {
result := []sample{}
for iter.Next() {