Merge pull request #6676 from roidelapluie/postings

Make head Postings only return series in time range

tsdb/block.go

@@ -111,8 +111,9 @@ type ChunkReader interface {
 
 // BlockReader provides reading access to a data block.
 type BlockReader interface {
-	// Index returns an IndexReader over the block's data.
+	// Index returns an IndexReader over the block's data within the specified
-	Index() (IndexReader, error)
+	// timeframe.
+	Index(mint, maxt int64) (IndexReader, error)
 
 	// Chunks returns a ChunkReader over the block's data.
 	Chunks() (ChunkReader, error)
@@ -372,7 +373,7 @@ func (pb *Block) startRead() error {
 }
 
 // Index returns a new IndexReader against the block data.
-func (pb *Block) Index() (IndexReader, error) {
+func (pb *Block) Index(mint, maxt int64) (IndexReader, error) {
 	if err := pb.startRead(); err != nil {
 		return nil, err
 	}

tsdb/cmd/tsdb/main.go

@@ -470,7 +470,7 @@ func analyzeBlock(b tsdb.BlockReader, limit int) error {
 	// Presume 1ms resolution that Prometheus uses.
 	fmt.Printf("Duration: %s\n", (time.Duration(meta.MaxTime-meta.MinTime) * 1e6).String())
 	fmt.Printf("Series: %d\n", meta.Stats.NumSeries)
-	ir, err := b.Index()
+	ir, err := b.Index(math.MinInt64, math.MaxInt64)
 	if err != nil {
 		return err
 	}

tsdb/compact.go

@@ -683,7 +683,7 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
 			}
 		}
 
-		indexr, err := b.Index()
+		indexr, err := b.Index(math.MinInt64, globalMaxt)
 		if err != nil {
 			return errors.Wrapf(err, "open index reader for block %s", b)
 		}

tsdb/compact_test.go

@@ -456,10 +456,10 @@ func metaRange(name string, mint, maxt int64, stats *BlockStats) dirMeta {
 
 type erringBReader struct{}
 
-func (erringBReader) Index() (IndexReader, error)            { return nil, errors.New("index") }
+func (erringBReader) Index(int64, int64) (IndexReader, error) { return nil, errors.New("index") }
-func (erringBReader) Chunks() (ChunkReader, error)           { return nil, errors.New("chunks") }
+func (erringBReader) Chunks() (ChunkReader, error)            { return nil, errors.New("chunks") }
-func (erringBReader) Tombstones() (tombstones.Reader, error) { return nil, errors.New("tombstones") }
+func (erringBReader) Tombstones() (tombstones.Reader, error)  { return nil, errors.New("tombstones") }
-func (erringBReader) Meta() BlockMeta                        { return BlockMeta{} }
+func (erringBReader) Meta() BlockMeta                         { return BlockMeta{} }
 
 type nopChunkWriter struct{}
 
@@ -652,7 +652,7 @@ func TestCompaction_populateBlock(t *testing.T) {
 			expErr:         errors.New("found chunk with minTime: 10 maxTime: 30 outside of compacted minTime: 0 maxTime: 20"),
 		},
 		{
-			// Introduced by https://github.com/prometheus/prometheus/tsdb/issues/347.
+			// Introduced by https://github.com/prometheus/tsdb/issues/347.
 			title: "Populate from single block containing extra chunk",
 			inputSeriesSamples: [][]seriesSamples{
 				{
@@ -692,7 +692,7 @@ func TestCompaction_populateBlock(t *testing.T) {
 			},
 		},
 		{
-			// Introduced by https://github.com/prometheus/prometheus/tsdb/pull/539.
+			// Introduced by https://github.com/prometheus/tsdb/pull/539.
 			title: "Populate from three blocks that the last two are overlapping.",
 			inputSeriesSamples: [][]seriesSamples{
 				{

tsdb/db_test.go

@@ -1384,7 +1384,7 @@ func TestOverlappingBlocksDetectsAllOverlaps(t *testing.T) {
 	}, OverlappingBlocks(nc1))
 }
 
-// Regression test for https://github.com/prometheus/prometheus/tsdb/issues/347
+// Regression test for https://github.com/prometheus/tsdb/issues/347
 func TestChunkAtBlockBoundary(t *testing.T) {
 	db, closeFn := openTestDB(t, nil, nil)
 	defer func() {
@@ -1411,7 +1411,7 @@ func TestChunkAtBlockBoundary(t *testing.T) {
 	testutil.Ok(t, err)
 
 	for _, block := range db.Blocks() {
-		r, err := block.Index()
+		r, err := block.Index(math.MinInt64, math.MaxInt64)
 		testutil.Ok(t, err)
 		defer r.Close()
 
@@ -1761,7 +1761,7 @@ func TestDB_LabelNames(t *testing.T) {
 		appendSamples(db, 0, 4, tst.sampleLabels1)
 
 		// Testing head.
-		headIndexr, err := db.head.Index()
+		headIndexr, err := db.head.Index(math.MinInt64, math.MaxInt64)
 		testutil.Ok(t, err)
 		labelNames, err := headIndexr.LabelNames()
 		testutil.Ok(t, err)
@@ -1774,7 +1774,7 @@ func TestDB_LabelNames(t *testing.T) {
 		// All blocks have same label names, hence check them individually.
 		// No need to aggregate and check.
 		for _, b := range db.Blocks() {
-			blockIndexr, err := b.Index()
+			blockIndexr, err := b.Index(math.MinInt64, math.MaxInt64)
 			testutil.Ok(t, err)
 			labelNames, err = blockIndexr.LabelNames()
 			testutil.Ok(t, err)

tsdb/head.go

@@ -754,7 +754,7 @@ type RangeHead struct {
 	mint, maxt int64
 }
 
-// NewRangeHead returns a *rangeHead.
+// NewRangeHead returns a *RangeHead.
 func NewRangeHead(head *Head, mint, maxt int64) *RangeHead {
 	return &RangeHead{
 		head: head,
@@ -763,8 +763,15 @@ func NewRangeHead(head *Head, mint, maxt int64) *RangeHead {
 	}
 }
 
-func (h *RangeHead) Index() (IndexReader, error) {
+func (h *RangeHead) Index(mint, maxt int64) (IndexReader, error) {
-	return h.head.indexRange(h.mint, h.maxt), nil
+	// rangeHead guarantees that the series returned are within its range.
+	if mint < h.mint {
+		mint = h.mint
+	}
+	if maxt > h.maxt {
+		maxt = h.maxt
+	}
+	return h.head.indexRange(mint, maxt), nil
 }
 
 func (h *RangeHead) Chunks() (ChunkReader, error) {
@@ -1162,8 +1169,8 @@ func (h *Head) Tombstones() (tombstones.Reader, error) {
 }
 
 // Index returns an IndexReader against the block.
-func (h *Head) Index() (IndexReader, error) {
+func (h *Head) Index(mint, maxt int64) (IndexReader, error) {
-	return h.indexRange(math.MinInt64, math.MaxInt64), nil
+	return h.indexRange(mint, maxt), nil
 }
 
 func (h *Head) indexRange(mint, maxt int64) *headIndexReader {
@@ -1347,9 +1354,35 @@ func (h *headIndexReader) LabelNames() ([]string, error) {
 
 // Postings returns the postings list iterator for the label pairs.
 func (h *headIndexReader) Postings(name string, values ...string) (index.Postings, error) {
+	fullRange := h.mint <= h.head.MinTime() && h.maxt >= h.head.MaxTime()
 	res := make([]index.Postings, 0, len(values))
 	for _, value := range values {
-		res = append(res, h.head.postings.Get(name, value))
+		p := h.head.postings.Get(name, value)
+		if fullRange {
+			// The head timerange covers the full index reader timerange.
+			// All the series can the be appended without filtering.
+			res = append(res, p)
+			continue
+		}
+
+		// Filter out series not in the time range, to avoid
+		// later on building up all the chunk metadata just to
+		// discard it.
+		filtered := []uint64{}
+		for p.Next() {
+			s := h.head.series.getByID(p.At())
+			if s == nil {
+				level.Debug(h.head.logger).Log("msg", "looked up series not found")
+				continue
+			}
+			if s.minTime() <= h.maxt && s.maxTime() >= h.mint {
+				filtered = append(filtered, p.At())
+			}
+		}
+		if p.Err() != nil {
+			return nil, p.Err()
+		}
+		res = append(res, index.NewListPostings(filtered))
 	}
 	return index.Merge(res...), nil
 }

tsdb/head_test.go

@@ -1322,3 +1322,91 @@ func TestAddDuplicateLabelName(t *testing.T) {
 	add(labels.Labels{{Name: "a", Value: "c"}, {Name: "a", Value: "c"}}, "a")
 	add(labels.Labels{{Name: "__name__", Value: "up"}, {Name: "job", Value: "prometheus"}, {Name: "le", Value: "500"}, {Name: "le", Value: "400"}, {Name: "unit", Value: "s"}}, "le")
 }
+
+func TestHeadSeriesWithTimeBoundaries(t *testing.T) {
+	h, err := NewHead(nil, nil, nil, 15, DefaultStripeSize)
+	testutil.Ok(t, err)
+	defer h.Close()
+	app := h.Appender()
+
+	s1, err := app.Add(labels.FromStrings("foo1", "bar"), 2, 0)
+	testutil.Ok(t, err)
+	for ts := int64(3); ts < 13; ts++ {
+		err = app.AddFast(s1, ts, 0)
+		testutil.Ok(t, err)
+	}
+	s2, err := app.Add(labels.FromStrings("foo2", "bar"), 5, 0)
+	testutil.Ok(t, err)
+	for ts := int64(6); ts < 11; ts++ {
+		err = app.AddFast(s2, ts, 0)
+		testutil.Ok(t, err)
+	}
+	s3, err := app.Add(labels.FromStrings("foo3", "bar"), 5, 0)
+	testutil.Ok(t, err)
+	err = app.AddFast(s3, 6, 0)
+	testutil.Ok(t, err)
+	_, err = app.Add(labels.FromStrings("foo4", "bar"), 9, 0)
+	testutil.Ok(t, err)
+
+	testutil.Ok(t, app.Commit())
+
+	cases := []struct {
+		mint         int64
+		maxt         int64
+		seriesCount  int
+		samplesCount int
+	}{
+		// foo1 ..00000000000..
+		// foo2 .....000000....
+		// foo3 .....00........
+		// foo4 .........0.....
+		{mint: 0, maxt: 0, seriesCount: 0, samplesCount: 0},
+		{mint: 0, maxt: 1, seriesCount: 0, samplesCount: 0},
+		{mint: 0, maxt: 2, seriesCount: 1, samplesCount: 1},
+		{mint: 2, maxt: 2, seriesCount: 1, samplesCount: 1},
+		{mint: 0, maxt: 4, seriesCount: 1, samplesCount: 3},
+		{mint: 0, maxt: 5, seriesCount: 3, samplesCount: 6},
+		{mint: 0, maxt: 6, seriesCount: 3, samplesCount: 9},
+		{mint: 0, maxt: 7, seriesCount: 3, samplesCount: 11},
+		{mint: 0, maxt: 8, seriesCount: 3, samplesCount: 13},
+		{mint: 0, maxt: 9, seriesCount: 4, samplesCount: 16},
+		{mint: 0, maxt: 10, seriesCount: 4, samplesCount: 18},
+		{mint: 0, maxt: 11, seriesCount: 4, samplesCount: 19},
+		{mint: 0, maxt: 12, seriesCount: 4, samplesCount: 20},
+		{mint: 0, maxt: 13, seriesCount: 4, samplesCount: 20},
+		{mint: 0, maxt: 14, seriesCount: 4, samplesCount: 20},
+		{mint: 2, maxt: 14, seriesCount: 4, samplesCount: 20},
+		{mint: 3, maxt: 14, seriesCount: 4, samplesCount: 19},
+		{mint: 4, maxt: 14, seriesCount: 4, samplesCount: 18},
+		{mint: 8, maxt: 9, seriesCount: 3, samplesCount: 5},
+		{mint: 9, maxt: 9, seriesCount: 3, samplesCount: 3},
+		{mint: 6, maxt: 9, seriesCount: 4, samplesCount: 10},
+		{mint: 11, maxt: 11, seriesCount: 1, samplesCount: 1},
+		{mint: 11, maxt: 12, seriesCount: 1, samplesCount: 2},
+		{mint: 11, maxt: 14, seriesCount: 1, samplesCount: 2},
+		{mint: 12, maxt: 14, seriesCount: 1, samplesCount: 1},
+	}
+
+	for i, c := range cases {
+		matcher := labels.MustNewMatcher(labels.MatchEqual, "", "")
+		q, err := NewBlockQuerier(h, c.mint, c.maxt)
+		testutil.Ok(t, err)
+
+		seriesCount := 0
+		samplesCount := 0
+		ss, _, err := q.Select(nil, matcher)
+		testutil.Ok(t, err)
+		for ss.Next() {
+			i := ss.At().Iterator()
+			for i.Next() {
+				samplesCount++
+			}
+			seriesCount++
+		}
+		testutil.Ok(t, ss.Err())
+		testutil.Equals(t, c.seriesCount, seriesCount, "test series %d", i)
+		testutil.Equals(t, c.samplesCount, samplesCount, "test samples %d", i)
+		q.Close()
+	}
+
+}

tsdb/mocks_test.go

@@ -71,8 +71,8 @@ type mockBReader struct {
 	maxt int64
 }
 
-func (r *mockBReader) Index() (IndexReader, error)  { return r.ir, nil }
+func (r *mockBReader) Index(mint, maxt int64) (IndexReader, error) { return r.ir, nil }
-func (r *mockBReader) Chunks() (ChunkReader, error) { return r.cr, nil }
+func (r *mockBReader) Chunks() (ChunkReader, error)                { return r.cr, nil }
 func (r *mockBReader) Tombstones() (tombstones.Reader, error) {
 	return tombstones.NewMemTombstones(), nil
 }

tsdb/querier.go

@@ -160,7 +160,7 @@ func (q *verticalQuerier) sel(p *storage.SelectParams, qs []storage.Querier, ms
 
 // NewBlockQuerier returns a querier against the reader.
 func NewBlockQuerier(b BlockReader, mint, maxt int64) (storage.Querier, error) {
-	indexr, err := b.Index()
+	indexr, err := b.Index(mint, maxt)
 	if err != nil {
 		return nil, errors.Wrapf(err, "open index reader")
 	}

tsdb/querier_bench_test.go

@@ -16,6 +16,7 @@ package tsdb
 import (
 	"fmt"
 	"io/ioutil"
+	"math"
 	"os"
 	"strconv"
 	"testing"
@@ -54,7 +55,7 @@ func BenchmarkPostingsForMatchers(b *testing.B) {
 	}
 	testutil.Ok(b, app.Commit())
 
-	ir, err := h.Index()
+	ir, err := h.Index(math.MinInt64, math.MaxInt64)
 	testutil.Ok(b, err)
 	b.Run("Head", func(b *testing.B) {
 		benchmarkPostingsForMatchers(b, ir)
@@ -72,7 +73,7 @@ func BenchmarkPostingsForMatchers(b *testing.B) {
 	defer func() {
 		testutil.Ok(b, block.Close())
 	}()
-	ir, err = block.Index()
+	ir, err = block.Index(math.MinInt64, math.MaxInt64)
 	testutil.Ok(b, err)
 	defer ir.Close()
 	b.Run("Block", func(b *testing.B) {

tsdb/querier_test.go

@@ -1044,7 +1044,7 @@ func TestSeriesIterator(t *testing.T) {
 	})
 }
 
-// Regression for: https://github.com/prometheus/prometheus/tsdb/pull/97
+// Regression for: https://github.com/prometheus/tsdb/pull/97
 func TestChunkSeriesIterator_DoubleSeek(t *testing.T) {
 	chkMetas := []chunks.Meta{
 		tsdbutil.ChunkFromSamples([]tsdbutil.Sample{}),
@@ -2137,7 +2137,7 @@ func TestPostingsForMatchers(t *testing.T) {
 		},
 	}
 
-	ir, err := h.Index()
+	ir, err := h.Index(math.MinInt64, math.MaxInt64)
 	testutil.Ok(t, err)
 
 	for _, c := range cases {

tsdb/wal/wal_test.go

@@ -49,7 +49,7 @@ func TestWALRepair_ReadingError(t *testing.T) {
 		},
 		// Ensures that the page buffer is big enough to fit
 		// an entire page size without panicking.
-		// https://github.com/prometheus/prometheus/tsdb/pull/414
+		// https://github.com/prometheus/tsdb/pull/414
 		"bad_header": {
 			1,
 			func(f *os.File) {