Repository: prometheus/tsdb
Pull request #467 (2018-12-10 16:29:00)

Add e2e  benchmarks for head and block Queries
closes #240 
~closes #278~ already closed in https://github.com/prometheus/tsdb/pull/572

combined and refactored the above PRs.

will post results from the benchmarks shortly.

Signed-off-by: Krasi Georgiev <kgeorgie@redhat.com>

krasi-georgiev: ```
benchcmp no-expand.txt expand.txt
benchmark                                                                 old ns/op     new ns/op     delta
BenchmarkPersistedQueries_Series_1M_RESelector_1_Expansion-8              157304440     147421099     -6.28%
BenchmarkPersistedQueries_Series_1M_RESelector_2_Expansion-8              19163203      17982639      -6.16%
BenchmarkPersistedQueries_Series_1M_RESelector_3_Expansion-8              8640811       7914559       -8.40%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_1_Expansion-8      515304845     471832010     -8.44%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_2_Expansion-8      64759526      69743052      +7.70%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_3_Expansion-8      29525021      28743209      -2.65%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_1_Expansion-8     572205794     580150933     +1.39%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_2_Expansion-8     72159910      82287972      +14.04%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_3_Expansion-8     33388668      26196775      -21.54%

benchmark                                                                 old allocs     new allocs     delta
BenchmarkPersistedQueries_Series_1M_RESelector_1_Expansion-8              1000182        1000223        +0.00%
BenchmarkPersistedQueries_Series_1M_RESelector_2_Expansion-8              110304         110325         +0.02%
BenchmarkPersistedQueries_Series_1M_RESelector_3_Expansion-8              40215          40232          +0.04%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_1_Expansion-8      3100695        3100832        +0.00%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_2_Expansion-8      341929         341997         +0.02%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_3_Expansion-8      124624         124676         +0.04%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_1_Expansion-8     3100696        3100830        +0.00%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_2_Expansion-8     341929         341996         +0.02%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_3_Expansion-8     124624         124676         +0.04%

benchmark                                                                 old bytes     new bytes     delta
BenchmarkPersistedQueries_Series_1M_RESelector_1_Expansion-8              55219483      59879260      +8.44%
BenchmarkPersistedQueries_Series_1M_RESelector_2_Expansion-8              6081791       6468332       +6.36%
BenchmarkPersistedQueries_Series_1M_RESelector_3_Expansion-8              2208200       2324264       +5.26%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_1_Expansion-8      144936442     158922606     +9.65%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_2_Expansion-8      15965682      17126937      +7.27%
BenchmarkPersistedQueries_3Blocks_Series_1M_RESelector_3_Expansion-8      5801820       6150190       +6.00%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_1_Expansion-8     144937071     158921465     +9.65%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_2_Expansion-8     15965733      17126544      +7.27%
BenchmarkPersistedQueries_10Blocks_Series_1M_RESelector_3_Expansion-8     5801675       6150330       +6.01%
```
krasi-georgiev: ping @codesome 	@gouthamve 	@brian-brazil 	
ready for review (thanks to @VladMasarik for fixing a small bug)
brian-brazil: Hmm, I've some benchmarks in #572 also. We should de-duplicate a bit I think.

Patch:
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
new file mode 100644
index 0000000..bbde613
--- /dev/null
+++ b/.github/PULL_REQUEST_TEMPLATE.md
@@ -0,0 +1,17 @@
+<!--
+    Don't forget!
+    
+    - Most PRs would require a CHANGELOG entry.
+    
+    - If the PR adds or changes a behaviour or fixes a bug of an exported API it would need a unit/e2e test.
+    
+    - Where possible use only exported APIs for tests to simplify the review and make it as close as possible to an actual library usage.
+    
+    - No tests are needed for internal implementation changes.
+    
+    - Performance improvements would need a benchmark test to prove it.
+    
+    - All exposed objects should have a comment.
+    
+    - All comments should start with a capital letter and end with a full stop.
+ -->
\ No newline at end of file
diff --git a/Makefile.common b/Makefile.common
index c7f9ea6..6b94f83 100644
--- a/Makefile.common
+++ b/Makefile.common
@@ -86,6 +86,7 @@ endif
 PREFIX                  ?= $(shell pwd)
 BIN_DIR                 ?= $(shell pwd)
 DOCKER_IMAGE_TAG        ?= $(subst /,-,$(shell git rev-parse --abbrev-ref HEAD))
+DOCKERFILE_PATH         ?= ./
 DOCKER_REPO             ?= prom
 
 DOCKER_ARCHS            ?= amd64
@@ -212,7 +213,7 @@ $(BUILD_DOCKER_ARCHS): common-docker-%:
 	docker build -t "$(DOCKER_REPO)/$(DOCKER_IMAGE_NAME)-linux-$*:$(DOCKER_IMAGE_TAG)" \
 		--build-arg ARCH="$*" \
 		--build-arg OS="linux" \
-		.
+		$(DOCKERFILE_PATH)
 
 .PHONY: common-docker-publish $(PUBLISH_DOCKER_ARCHS)
 common-docker-publish: $(PUBLISH_DOCKER_ARCHS)
diff --git a/cmd/tsdb/main.go b/cmd/tsdb/main.go
index d4fdc62..e7b33c8 100644
--- a/cmd/tsdb/main.go
+++ b/cmd/tsdb/main.go
@@ -15,7 +15,6 @@ package main
 
 import (
 	"bufio"
-	"flag"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -107,9 +106,6 @@ func main() {
 		}
 		dumpSamples(db, *dumpMinTime, *dumpMaxTime)
 	}
-	if err := flag.CommandLine.Set("log.level", "debug"); err != nil {
-		exitWithError(err)
-	}
 }
 
 type writeBenchmark struct {
diff --git a/head.go b/head.go
index 740eff0..b14c056 100644
--- a/head.go
+++ b/head.go
@@ -313,7 +313,7 @@ func (h *Head) updateMinMaxTime(mint, maxt int64) {
 	}
 }
 
-func (h *Head) loadWAL(r *wal.Reader) error {
+func (h *Head) loadWAL(r *wal.Reader, multiRef map[uint64]uint64) error {
 	// Track number of samples that referenced a series we don't know about
 	// for error reporting.
 	var unknownRefs uint64
@@ -322,10 +322,11 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 	// They are connected through a ring of channels which ensures that all sample batches
 	// read from the WAL are processed in order.
 	var (
-		wg      sync.WaitGroup
-		n       = runtime.GOMAXPROCS(0)
-		inputs  = make([]chan []RefSample, n)
-		outputs = make([]chan []RefSample, n)
+		wg           sync.WaitGroup
+		multiRefLock sync.Mutex
+		n            = runtime.GOMAXPROCS(0)
+		inputs       = make([]chan []RefSample, n)
+		outputs      = make([]chan []RefSample, n)
 	)
 	wg.Add(n)
 
@@ -364,7 +365,14 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 				}
 			}
 			for _, s := range series {
-				h.getOrCreateWithID(s.Ref, s.Labels.Hash(), s.Labels)
+				series, created := h.getOrCreateWithID(s.Ref, s.Labels.Hash(), s.Labels)
+
+				if !created {
+					// There's already a different ref for this series.
+					multiRefLock.Lock()
+					multiRef[s.Ref] = series.ref
+					multiRefLock.Unlock()
+				}
 
 				if h.lastSeriesID < s.Ref {
 					h.lastSeriesID = s.Ref
@@ -399,6 +407,9 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 					shards[i] = buf[:0]
 				}
 				for _, sam := range samples[:m] {
+					if r, ok := multiRef[sam.Ref]; ok {
+						sam.Ref = r
+					}
 					mod := sam.Ref % uint64(n)
 					shards[mod] = append(shards[mod], sam)
 				}
@@ -478,6 +489,7 @@ func (h *Head) Init(minValidTime int64) error {
 	if err != nil && err != ErrNotFound {
 		return errors.Wrap(err, "find last checkpoint")
 	}
+	multiRef := map[uint64]uint64{}
 	if err == nil {
 		sr, err := wal.NewSegmentsReader(dir)
 		if err != nil {
@@ -487,7 +499,7 @@ func (h *Head) Init(minValidTime int64) error {
 
 		// A corrupted checkpoint is a hard error for now and requires user
 		// intervention. There's likely little data that can be recovered anyway.
-		if err := h.loadWAL(wal.NewReader(sr)); err != nil {
+		if err := h.loadWAL(wal.NewReader(sr), multiRef); err != nil {
 			return errors.Wrap(err, "backfill checkpoint")
 		}
 		startFrom++
@@ -507,7 +519,7 @@ func (h *Head) Init(minValidTime int64) error {
 		}
 
 		sr := wal.NewSegmentBufReader(s)
-		err = h.loadWAL(wal.NewReader(sr))
+		err = h.loadWAL(wal.NewReader(sr), multiRef)
 		sr.Close() // Close the reader so that if there was an error the repair can remove the corrupted file under Windows.
 		if err == nil {
 			continue
@@ -564,6 +576,12 @@ func (h *Head) Truncate(mint int64) (err error) {
 	if err != nil {
 		return errors.Wrap(err, "get segment range")
 	}
+	// Start a new segment, so low ingestion volume TSDB don't have more WAL than
+	// needed.
+	err = h.wal.NextSegment()
+	if err != nil {
+		return errors.Wrap(err, "next segment")
+	}
 	last-- // Never consider last segment for checkpoint.
 	if last < 0 {
 		return nil // no segments yet.
diff --git a/wal/live_reader.go b/wal/live_reader.go
index 8394bfd..b62845f 100644
--- a/wal/live_reader.go
+++ b/wal/live_reader.go
@@ -24,19 +24,11 @@ import (
 	"github.com/go-kit/kit/log/level"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-)
-
-var (
-	readerCorruptionErrors = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name: "prometheus_tsdb_wal_reader_corruption_errors",
-		Help: "Errors encountered when reading the WAL.",
-	}, []string{"error"})
 )
 
 // NewLiveReader returns a new live reader.
-func NewLiveReader(logger log.Logger, r io.Reader) *LiveReader {
-	return &LiveReader{
+func NewLiveReader(logger log.Logger, reg prometheus.Registerer, r io.Reader) *LiveReader {
+	lr := &LiveReader{
 		logger: logger,
 		rdr:    r,
 
@@ -44,6 +36,17 @@ func NewLiveReader(logger log.Logger, r io.Reader) *LiveReader {
 		// to records spanning pages.
 		permissive: true,
 	}
+
+	lr.readerCorruptionErrors = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "prometheus_tsdb_wal_reader_corruption_errors_total",
+		Help: "Errors encountered when reading the WAL.",
+	}, []string{"error"})
+
+	if reg != nil {
+		reg.MustRegister(lr.readerCorruptionErrors)
+	}
+
+	return lr
 }
 
 // LiveReader reads WAL records from an io.Reader. It allows reading of WALs
@@ -68,6 +71,8 @@ type LiveReader struct {
 	// does.  Until we track down why, set permissive to true to tolerate it.
 	// NB the non-ive Reader implementation allows for this.
 	permissive bool
+
+	readerCorruptionErrors *prometheus.CounterVec
 }
 
 // Err returns any errors encountered reading the WAL.  io.EOFs are not terminal
@@ -258,7 +263,7 @@ func (r *LiveReader) readRecord() ([]byte, int, error) {
 		if !r.permissive {
 			return nil, 0, fmt.Errorf("record would overflow current page: %d > %d", r.readIndex+recordHeaderSize+length, pageSize)
 		}
-		readerCorruptionErrors.WithLabelValues("record_span_page").Inc()
+		r.readerCorruptionErrors.WithLabelValues("record_span_page").Inc()
 		level.Warn(r.logger).Log("msg", "record spans page boundaries", "start", r.readIndex, "end", recordHeaderSize+length, "pageSize", pageSize)
 	}
 	if recordHeaderSize+length > pageSize {
diff --git a/wal/wal.go b/wal/wal.go
index cb2e11f..f0e413f 100644
--- a/wal/wal.go
+++ b/wal/wal.go
@@ -393,6 +393,13 @@ func SegmentName(dir string, i int) string {
 	return filepath.Join(dir, fmt.Sprintf("%08d", i))
 }
 
+// NextSegment creates the next segment and closes the previous one.
+func (w *WAL) NextSegment() error {
+	w.mtx.Lock()
+	defer w.mtx.Unlock()
+	return w.nextSegment()
+}
+
 // nextSegment creates the next segment and closes the previous one.
 func (w *WAL) nextSegment() error {
 	// Only flush the current page if it actually holds data.


Test patch:
diff --git a/block_test.go b/block_test.go
index 3ae2ec2..7cd02dd 100644
--- a/block_test.go
+++ b/block_test.go
@@ -151,6 +151,18 @@ func TestCorruptedChunk(t *testing.T) {
 
 // createBlock creates a block with given set of series and returns its dir.
 func createBlock(tb testing.TB, dir string, series []Series) string {
+	head := createHead(tb, series)
+	compactor, err := NewLeveledCompactor(context.Background(), nil, log.NewNopLogger(), []int64{1000000}, nil)
+	testutil.Ok(tb, err)
+
+	testutil.Ok(tb, os.MkdirAll(dir, 0777))
+
+	ulid, err := compactor.Write(dir, head, head.MinTime(), head.MaxTime(), nil)
+	testutil.Ok(tb, err)
+	return filepath.Join(dir, ulid.String())
+}
+
+func createHead(tb testing.TB, series []Series) *Head {
 	head, err := NewHead(nil, nil, nil, 2*60*60*1000)
 	testutil.Ok(tb, err)
 	defer head.Close()
@@ -174,15 +186,7 @@ func createBlock(tb testing.TB, dir string, series []Series) string {
 	}
 	err = app.Commit()
 	testutil.Ok(tb, err)
-
-	compactor, err := NewLeveledCompactor(context.Background(), nil, log.NewNopLogger(), []int64{1000000}, nil)
-	testutil.Ok(tb, err)
-
-	testutil.Ok(tb, os.MkdirAll(dir, 0777))
-
-	ulid, err := compactor.Write(dir, head, head.MinTime(), head.MaxTime(), nil)
-	testutil.Ok(tb, err)
-	return filepath.Join(dir, ulid.String())
+	return head
 }
 
 const (
@@ -197,6 +201,7 @@ func genSeries(totalSeries, labelCount int, mint, maxt int64) []Series {
 	}
 
 	series := make([]Series, totalSeries)
+
 	for i := 0; i < totalSeries; i++ {
 		lbls := make(map[string]string, labelCount)
 		lbls[defaultLabelName] = strconv.Itoa(i)
diff --git a/head_test.go b/head_test.go
index 79092dd..dbe6fec 100644
--- a/head_test.go
+++ b/head_test.go
@@ -109,10 +109,13 @@ func TestHead_ReadWAL(t *testing.T) {
 		},
 		[]RefSeries{
 			{Ref: 50, Labels: labels.FromStrings("a", "4")},
+			// This series has two refs pointing to it.
+			{Ref: 101, Labels: labels.FromStrings("a", "3")},
 		},
 		[]RefSample{
 			{Ref: 10, T: 101, V: 5},
 			{Ref: 50, T: 101, V: 6},
+			{Ref: 101, T: 101, V: 7},
 		},
 		[]Stone{
 			{ref: 0, intervals: []Interval{{Mint: 99, Maxt: 101}}},
@@ -133,7 +136,7 @@ func TestHead_ReadWAL(t *testing.T) {
 	testutil.Ok(t, err)
 
 	testutil.Ok(t, head.Init(math.MinInt64))
-	testutil.Equals(t, uint64(100), head.lastSeriesID)
+	testutil.Equals(t, uint64(101), head.lastSeriesID)
 
 	s10 := head.series.getByID(10)
 	s11 := head.series.getByID(11)
@@ -156,7 +159,52 @@ func TestHead_ReadWAL(t *testing.T) {
 
 	testutil.Equals(t, []sample{{100, 2}, {101, 5}}, expandChunk(s10.iterator(0)))
 	testutil.Equals(t, []sample{{101, 6}}, expandChunk(s50.iterator(0)))
-	testutil.Equals(t, []sample{{100, 3}}, expandChunk(s100.iterator(0)))
+	testutil.Equals(t, []sample{{100, 3}, {101, 7}}, expandChunk(s100.iterator(0)))
+}
+
+func TestHead_WALMultiRef(t *testing.T) {
+	dir, err := ioutil.TempDir("", "test_wal_multi_ref")
+	testutil.Ok(t, err)
+	defer func() {
+		testutil.Ok(t, os.RemoveAll(dir))
+	}()
+
+	w, err := wal.New(nil, nil, dir)
+	testutil.Ok(t, err)
+
+	head, err := NewHead(nil, nil, w, 1000)
+	testutil.Ok(t, err)
+
+	testutil.Ok(t, head.Init(0))
+	app := head.Appender()
+	ref1, err := app.Add(labels.FromStrings("foo", "bar"), 100, 1)
+	testutil.Ok(t, err)
+	testutil.Ok(t, app.Commit())
+
+	testutil.Ok(t, head.Truncate(200))
+
+	app = head.Appender()
+	ref2, err := app.Add(labels.FromStrings("foo", "bar"), 300, 2)
+	testutil.Ok(t, err)
+	testutil.Ok(t, app.Commit())
+
+	if ref1 == ref2 {
+		t.Fatal("Refs are the same")
+	}
+	testutil.Ok(t, head.Close())
+
+	w, err = wal.New(nil, nil, dir)
+	testutil.Ok(t, err)
+
+	head, err = NewHead(nil, nil, w, 1000)
+	testutil.Ok(t, err)
+	testutil.Ok(t, head.Init(0))
+	defer head.Close()
+
+	q, err := NewBlockQuerier(head, 0, 300)
+	testutil.Ok(t, err)
+	series := query(t, q, labels.NewEqualMatcher("foo", "bar"))
+	testutil.Equals(t, map[string][]tsdbutil.Sample{`{foo="bar"}`: {sample{100, 1}, sample{300, 2}}}, series)
 }
 
 func TestHead_Truncate(t *testing.T) {
@@ -1080,3 +1128,40 @@ func TestWalRepair(t *testing.T) {
 	}
 
 }
+
+func TestNewWalSegmentOnTruncate(t *testing.T) {
+	dir, err := ioutil.TempDir("", "test_wal_segemnts")
+	testutil.Ok(t, err)
+	defer func() {
+		testutil.Ok(t, os.RemoveAll(dir))
+	}()
+	wlog, err := wal.NewSize(nil, nil, dir, 32768)
+	testutil.Ok(t, err)
+
+	h, err := NewHead(nil, nil, wlog, 1000)
+	testutil.Ok(t, err)
+	defer h.Close()
+	add := func(ts int64) {
+		app := h.Appender()
+		_, err := app.Add(labels.Labels{{"a", "b"}}, ts, 0)
+		testutil.Ok(t, err)
+		testutil.Ok(t, app.Commit())
+	}
+
+	add(0)
+	_, last, err := wlog.Segments()
+	testutil.Ok(t, err)
+	testutil.Equals(t, 0, last)
+
+	add(1)
+	testutil.Ok(t, h.Truncate(1))
+	_, last, err = wlog.Segments()
+	testutil.Ok(t, err)
+	testutil.Equals(t, 1, last)
+
+	add(2)
+	testutil.Ok(t, h.Truncate(2))
+	_, last, err = wlog.Segments()
+	testutil.Ok(t, err)
+	testutil.Equals(t, 2, last)
+}
diff --git a/querier_test.go b/querier_test.go
index dfbc6a7..252170d 100644
--- a/querier_test.go
+++ b/querier_test.go
@@ -21,6 +21,7 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strconv"
 	"testing"
 
 	"github.com/pkg/errors"
@@ -1225,45 +1226,6 @@ func BenchmarkMergedSeriesSet(b *testing.B) {
 	}
 }
 
-func BenchmarkPersistedQueries(b *testing.B) {
-	for _, nSeries := range []int{10, 100} {
-		for _, nSamples := range []int64{1000, 10000, 100000} {
-			b.Run(fmt.Sprintf("series=%d,samplesPerSeries=%d", nSeries, nSamples), func(b *testing.B) {
-				dir, err := ioutil.TempDir("", "bench_persisted")
-				testutil.Ok(b, err)
-				defer func() {
-					testutil.Ok(b, os.RemoveAll(dir))
-				}()
-
-				block, err := OpenBlock(nil, createBlock(b, dir, genSeries(nSeries, 10, 1, int64(nSamples))), nil)
-				testutil.Ok(b, err)
-				defer block.Close()
-
-				q, err := NewBlockQuerier(block, block.Meta().MinTime, block.Meta().MaxTime)
-				testutil.Ok(b, err)
-				defer q.Close()
-
-				b.ResetTimer()
-				b.ReportAllocs()
-
-				for i := 0; i < b.N; i++ {
-					ss, err := q.Select(labels.NewMustRegexpMatcher("__name__", ".+"))
-					for ss.Next() {
-						s := ss.At()
-						s.Labels()
-						it := s.Iterator()
-						for it.Next() {
-						}
-						testutil.Ok(b, it.Err())
-					}
-					testutil.Ok(b, ss.Err())
-					testutil.Ok(b, err)
-				}
-			})
-		}
-	}
-}
-
 type mockChunkReader map[uint64]chunkenc.Chunk
 
 func (cr mockChunkReader) Chunk(id uint64) (chunkenc.Chunk, error) {
@@ -1584,19 +1546,7 @@ func BenchmarkQueryIterator(b *testing.B) {
 				}
 				defer sq.Close()
 
-				b.ResetTimer()
-				b.ReportAllocs()
-
-				ss, err := sq.Select(labels.NewMustRegexpMatcher("__name__", ".*"))
-				testutil.Ok(b, err)
-				for ss.Next() {
-					it := ss.At().Iterator()
-					for it.Next() {
-					}
-					testutil.Ok(b, it.Err())
-				}
-				testutil.Ok(b, ss.Err())
-				testutil.Ok(b, err)
+				benchQuery(b, c.numSeries, sq, labels.Selector{labels.NewMustRegexpMatcher("__name__", ".*")})
 			})
 		}
 	}
@@ -2156,3 +2106,124 @@ func TestClose(t *testing.T) {
 	testutil.Ok(t, q.Close())
 	testutil.NotOk(t, q.Close())
 }
+
+func BenchmarkQueries(b *testing.B) {
+	cases := map[string]labels.Selector{
+		"Eq Matcher: Expansion - 1": labels.Selector{
+			labels.NewEqualMatcher("la", "va"),
+		},
+		"Eq Matcher: Expansion - 2": labels.Selector{
+			labels.NewEqualMatcher("la", "va"),
+			labels.NewEqualMatcher("lb", "vb"),
+		},
+
+		"Eq Matcher: Expansion - 3": labels.Selector{
+			labels.NewEqualMatcher("la", "va"),
+			labels.NewEqualMatcher("lb", "vb"),
+			labels.NewEqualMatcher("lc", "vc"),
+		},
+		"Regex Matcher: Expansion - 1": labels.Selector{
+			labels.NewMustRegexpMatcher("la", ".*va"),
+		},
+		"Regex Matcher: Expansion - 2": labels.Selector{
+			labels.NewMustRegexpMatcher("la", ".*va"),
+			labels.NewMustRegexpMatcher("lb", ".*vb"),
+		},
+		"Regex Matcher: Expansion - 3": labels.Selector{
+			labels.NewMustRegexpMatcher("la", ".*va"),
+			labels.NewMustRegexpMatcher("lb", ".*vb"),
+			labels.NewMustRegexpMatcher("lc", ".*vc"),
+		},
+	}
+
+	queryTypes := make(map[string]Querier)
+	defer func() {
+		for _, q := range queryTypes {
+			// Can't run a check for error here as some of these will fail as
+			// queryTypes is using the same slice for the different block queriers
+			// and would have been closed in the previous iterration.
+			q.Close()
+		}
+	}()
+
+	for title, selectors := range cases {
+		for _, nSeries := range []int{10} {
+			for _, nSamples := range []int64{1000, 10000, 100000} {
+				dir, err := ioutil.TempDir("", "test_persisted_query")
+				testutil.Ok(b, err)
+				defer func() {
+					testutil.Ok(b, os.RemoveAll(dir))
+				}()
+
+				series := genSeries(nSeries, 5, 1, int64(nSamples))
+
+				// Add some common labels to make the matchers select these series.
+				{
+					var commonLbls labels.Labels
+					for _, selector := range selectors {
+						switch sel := selector.(type) {
+						case *labels.EqualMatcher:
+							commonLbls = append(commonLbls, labels.Label{Name: sel.Name(), Value: sel.Value()})
+						case *labels.RegexpMatcher:
+							commonLbls = append(commonLbls, labels.Label{Name: sel.Name(), Value: sel.Value()})
+						}
+					}
+					for i := range commonLbls {
+						s := series[i].(*mockSeries)
+						allLabels := append(commonLbls, s.Labels()...)
+						s = &mockSeries{
+							labels:   func() labels.Labels { return allLabels },
+							iterator: s.iterator,
+						}
+						series[i] = s
+					}
+				}
+
+				qs := []Querier{}
+				for x := 0; x <= 10; x++ {
+					block, err := OpenBlock(nil, createBlock(b, dir, series), nil)
+					testutil.Ok(b, err)
+					q, err := NewBlockQuerier(block, 1, int64(nSamples))
+					testutil.Ok(b, err)
+					qs = append(qs, q)
+				}
+				queryTypes["_1-Block"] = &querier{blocks: qs[:1]}
+				queryTypes["_3-Blocks"] = &querier{blocks: qs[0:3]}
+				queryTypes["_10-Blocks"] = &querier{blocks: qs}
+
+				head := createHead(b, series)
+				qHead, err := NewBlockQuerier(head, 1, int64(nSamples))
+				testutil.Ok(b, err)
+				queryTypes["_Head"] = qHead
+
+				for qtype, querier := range queryTypes {
+					b.Run(title+qtype+"_nSeries:"+strconv.Itoa(nSeries)+"_nSamples:"+strconv.Itoa(int(nSamples)), func(b *testing.B) {
+						expExpansions, err := strconv.Atoi(string(title[len(title)-1]))
+						testutil.Ok(b, err)
+						benchQuery(b, expExpansions, querier, selectors)
+					})
+				}
+			}
+		}
+	}
+}
+
+func benchQuery(b *testing.B, expExpansions int, q Querier, selectors labels.Selector) {
+	b.ResetTimer()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		ss, err := q.Select(selectors...)
+		testutil.Ok(b, err)
+		var actualExpansions int
+		for ss.Next() {
+			s := ss.At()
+			s.Labels()
+			it := s.Iterator()
+			for it.Next() {
+			}
+			actualExpansions++
+		}
+		testutil.Equals(b, expExpansions, actualExpansions)
+		testutil.Ok(b, ss.Err())
+	}
+}
diff --git a/wal/reader_test.go b/wal/reader_test.go
index 1178aa5..1a21d7b 100644
--- a/wal/reader_test.go
+++ b/wal/reader_test.go
@@ -51,7 +51,7 @@ var readerConstructors = map[string]func(io.Reader) reader{
 		return NewReader(r)
 	},
 	"LiveReader": func(r io.Reader) reader {
-		lr := NewLiveReader(log.NewNopLogger(), r)
+		lr := NewLiveReader(log.NewNopLogger(), nil, r)
 		lr.eofNonErr = true
 		return lr
 	},
@@ -216,7 +216,7 @@ func TestReader_Live(t *testing.T) {
 			// Read from a second FD on the same file.
 			readFd, err := os.Open(writeFd.Name())
 			testutil.Ok(t, err)
-			reader := NewLiveReader(logger, readFd)
+			reader := NewLiveReader(logger, nil, readFd)
 			for _, exp := range testReaderCases[i].exp {
 				for !reader.Next() {
 					testutil.Assert(t, reader.Err() == io.EOF, "expect EOF, got: %v", reader.Err())
@@ -374,7 +374,7 @@ func TestReaderFuzz_Live(t *testing.T) {
 	testutil.Ok(t, err)
 	defer seg.Close()
 
-	r := NewLiveReader(logger, seg)
+	r := NewLiveReader(logger, nil, seg)
 	segmentTicker := time.NewTicker(100 * time.Millisecond)
 	readTicker := time.NewTicker(10 * time.Millisecond)
 
@@ -410,7 +410,7 @@ outer:
 			seg, err = OpenReadSegment(SegmentName(dir, seg.i+1))
 			testutil.Ok(t, err)
 			defer seg.Close()
-			r = NewLiveReader(logger, seg)
+			r = NewLiveReader(logger, nil, seg)
 
 		case <-readTicker.C:
 			readSegment(r)
@@ -464,7 +464,7 @@ func TestLiveReaderCorrupt_ShortFile(t *testing.T) {
 	testutil.Ok(t, err)
 	defer seg.Close()
 
-	r := NewLiveReader(logger, seg)
+	r := NewLiveReader(logger, nil, seg)
 	testutil.Assert(t, r.Next() == false, "expected no records")
 	testutil.Assert(t, r.Err() == io.EOF, "expected error, got: %v", r.Err())
 }
@@ -512,7 +512,7 @@ func TestLiveReaderCorrupt_RecordTooLongAndShort(t *testing.T) {
 	testutil.Ok(t, err)
 	defer seg.Close()
 
-	r := NewLiveReader(logger, seg)
+	r := NewLiveReader(logger, nil, seg)
 	testutil.Assert(t, r.Next() == false, "expected no records")
 	testutil.Assert(t, r.Err().Error() == "record length greater than a single page: 65542 > 32768", "expected error, got: %v", r.Err())
 }