Repository: prometheus/tsdb
Pull request #602 (2019-05-14 04:24:54)

Optimize queries using regex matchers for set lookups
fixes https://github.com/prometheus/prometheus/issues/2651

brian-brazil: You need to check for and exclude the wrapper we put around the regex the user specifies.
naivewong: Oh yes, I forgot to exclude the wrapper. But sorry what did you mean by "check for"?
naivewong: Is that we exclude the wrapper only if we find it?
brian-brazil: If the wrapper is missing, treat it as a regex.
brian-brazil: That looks about right. Can you add it to the postings benchmark too?

I'd be interested to know if there's a point where it's better to use the regex rather than the set.
gouthamve: > I'd be interested to know if there's a point where it's better to use the regex rather than the set.

I don't think that'll ever be the case, as with regex, first we'll get all the matching label values (the set) and doing what the set matcher does anyways. With set, we'll be avoiding a step.
brian-brazil: At some point, us doing parsing of the strings might be slower than the regex. Would be good to check one way or the other.
bboreham: What's the motivation for only allowing escaping of special characters?  What if I want to match `\t` or `\n` in a label value?
codesome: Adding to @bboreham's comment, isn't it enough to just include any character that is escaped rather than checking for special character or '\\'?
naivewong: Benchmark
```
BenchmarkSetMatcher/SetMatch,nSeries=15,pattern="^(?:1|2|3)$"-8         	  300000	      4579 ns/op	    3425 B/op	      59 allocs/op
BenchmarkSetMatcher/RegexMatch,nSeries=15,pattern="1|2|3"-8             	  200000	      6349 ns/op	    2836 B/op	      64 allocs/op
BenchmarkSetMatcher/SetMatch,nSeries=15,pattern="^(?:1|2|3|4|5|6|7|8|9|10)$"-8         	  200000	      8249 ns/op	    5443 B/op	      96 allocs/op
BenchmarkSetMatcher/RegexMatch,nSeries=15,pattern="1|2|3|4|5|6|7|8|9|10"-8             	  200000	      8844 ns/op	    4208 B/op	      81 allocs/op
BenchmarkSetMatcher/SetMatch,nSeries=200,pattern="^(?:1|2|3)$"-8                       	  300000	      5011 ns/op	    3425 B/op	      59 allocs/op
BenchmarkSetMatcher/RegexMatch,nSeries=200,pattern="1|2|3"-8                           	   20000	     70441 ns/op	   31928 B/op	     545 allocs/op
BenchmarkSetMatcher/SetMatch,nSeries=200,pattern="^(?:1|2|3|4|5|6|7|8|9|10)$"-8        	  200000	      9755 ns/op	    5546 B/op	      96 allocs/op
BenchmarkSetMatcher/RegexMatch,nSeries=200,pattern="1|2|3|4|5|6|7|8|9|10"-8            	   20000	     83923 ns/op	   35351 B/op	     644 allocs/op
```

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 87b6420..610899d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,8 +1,15 @@
 ## master / unreleased
- - [BUGFIX] Calling `Close` more than once on a querier returns an error instead of a panic.
- - [ENHANCEMENT] Add (again) FromData function to easily create a chunk from bytes.
 
 
+## 0.8.0
+ - [BUGFIX] Calling `Close` more than once on a querier returns an error instead of a panic.
+ - [BUGFIX] Don't panic and recover nicely when running out of disk space.
+ - [BUGFIX] Correctly handle empty labels.
+ - [BUGFIX] Don't crash on an unknown tombstone ref.
+ - [ENHANCEMENT] Re-add FromData function to create a chunk from bytes. It is used by Cortex and Thanos.
+ - [ENHANCEMENT] Simplify mergedPostings.Seek.
+ - [FEATURE]  Added `currentSegment` metric for the current WAL segment it is being written to.
+
 ## 0.7.1
  - [ENHANCEMENT] Reduce memory usage in mergedPostings.Seek
 
diff --git a/Makefile.common b/Makefile.common
index 4f18ea5..c7f9ea6 100644
--- a/Makefile.common
+++ b/Makefile.common
@@ -69,7 +69,7 @@ else
 	GO_BUILD_PLATFORM ?= $(GOHOSTOS)-$(GOHOSTARCH)
 endif
 
-PROMU_VERSION ?= 0.3.0
+PROMU_VERSION ?= 0.4.0
 PROMU_URL     := https://github.com/prometheus/promu/releases/download/v$(PROMU_VERSION)/promu-$(PROMU_VERSION).$(GO_BUILD_PLATFORM).tar.gz
 
 GOLANGCI_LINT :=
diff --git a/head.go b/head.go
index 30cbe61..740eff0 100644
--- a/head.go
+++ b/head.go
@@ -14,6 +14,7 @@
 package tsdb
 
 import (
+	"fmt"
 	"math"
 	"runtime"
 	"sort"
@@ -421,6 +422,10 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 					if itv.Maxt < h.minValidTime {
 						continue
 					}
+					if m := h.series.getByID(s.ref); m == nil {
+						unknownRefs++
+						continue
+					}
 					allStones.addInterval(s.ref, itv)
 				}
 			}
@@ -488,22 +493,32 @@ func (h *Head) Init(minValidTime int64) error {
 		startFrom++
 	}
 
-	// Backfill segments from the last checkpoint onwards
-	sr, err := wal.NewSegmentsRangeReader(wal.SegmentRange{Dir: h.wal.Dir(), First: startFrom, Last: -1})
+	// Find the last segment.
+	_, last, err := h.wal.Segments()
 	if err != nil {
-		return errors.Wrap(err, "open WAL segments")
+		return errors.Wrap(err, "finding WAL segments")
 	}
 
-	err = h.loadWAL(wal.NewReader(sr))
-	sr.Close() // Close the reader so that if there was an error the repair can remove the corrupted file under Windows.
-	if err == nil {
-		return nil
-	}
-	level.Warn(h.logger).Log("msg", "encountered WAL error, attempting repair", "err", err)
-	h.metrics.walCorruptionsTotal.Inc()
-	if err := h.wal.Repair(err); err != nil {
-		return errors.Wrap(err, "repair corrupted WAL")
+	// Backfill segments from the most recent checkpoint onwards.
+	for i := startFrom; i <= last; i++ {
+		s, err := wal.OpenReadSegment(wal.SegmentName(h.wal.Dir(), i))
+		if err != nil {
+			return errors.Wrap(err, fmt.Sprintf("open WAL segment: %d", i))
+		}
+
+		sr := wal.NewSegmentBufReader(s)
+		err = h.loadWAL(wal.NewReader(sr))
+		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
+		}
+		level.Warn(h.logger).Log("msg", "encountered WAL error, attempting repair", "err", err)
+		h.metrics.walCorruptionsTotal.Inc()
+		if err := h.wal.Repair(err); err != nil {
+			return errors.Wrap(err, "repair corrupted WAL")
+		}
 	}
+
 	return nil
 }
 
diff --git a/labels/selector.go b/labels/selector.go
index a0565f5..c94ebb3 100644
--- a/labels/selector.go
+++ b/labels/selector.go
@@ -63,14 +63,15 @@ func NewEqualMatcher(name, value string) Matcher {
 	return &EqualMatcher{name: name, value: value}
 }
 
-type regexpMatcher struct {
+type RegexpMatcher struct {
 	name string
 	re   *regexp.Regexp
 }
 
-func (m regexpMatcher) Name() string          { return m.name }
-func (m regexpMatcher) Matches(v string) bool { return m.re.MatchString(v) }
-func (m regexpMatcher) String() string        { return fmt.Sprintf("%s=~%q", m.name, m.re.String()) }
+func (m RegexpMatcher) Name() string          { return m.name }
+func (m RegexpMatcher) Matches(v string) bool { return m.re.MatchString(v) }
+func (m RegexpMatcher) String() string        { return fmt.Sprintf("%s=~%q", m.name, m.re.String()) }
+func (m RegexpMatcher) Value() string         { return m.re.String() }
 
 // NewRegexpMatcher returns a new matcher verifying that a value matches
 // the regular expression pattern.
@@ -79,7 +80,7 @@ func NewRegexpMatcher(name, pattern string) (Matcher, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &regexpMatcher{name: name, re: re}, nil
+	return &RegexpMatcher{name: name, re: re}, nil
 }
 
 // NewMustRegexpMatcher returns a new matcher verifying that a value matches
@@ -90,7 +91,7 @@ func NewMustRegexpMatcher(name, pattern string) Matcher {
 	if err != nil {
 		panic(err)
 	}
-	return &regexpMatcher{name: name, re: re}
+	return &RegexpMatcher{name: name, re: re}
 
 }
 
diff --git a/querier.go b/querier.go
index 9d99de0..253102b 100644
--- a/querier.go
+++ b/querier.go
@@ -17,6 +17,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"unicode/utf8"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/tsdb/chunkenc"
@@ -266,6 +267,62 @@ func (q *blockQuerier) Close() error {
 	return merr.Err()
 }
 
+// Bitmap used by func isRegexMetaCharacter to check whether a character needs to be escaped.
+var regexMetaCharacterBytes [16]byte
+
+// isRegexMetaCharacter reports whether byte b needs to be escaped.
+func isRegexMetaCharacter(b byte) bool {
+	return b < utf8.RuneSelf && regexMetaCharacterBytes[b%16]&(1<<(b/16)) != 0
+}
+
+func init() {
+	for _, b := range []byte(`.+*?()|[]{}^$`) {
+		regexMetaCharacterBytes[b%16] |= 1 << (b / 16)
+	}
+}
+
+func findSetMatches(pattern string) []string {
+	// Return empty matches if the wrapper from Prometheus is missing.
+	if len(pattern) < 6 || pattern[:4] != "^(?:" || pattern[len(pattern)-2:] != ")$" {
+		return nil
+	}
+	escaped := false
+	sets := []*strings.Builder{&strings.Builder{}}
+	for i := 4; i < len(pattern)-2; i++ {
+		if escaped {
+			switch {
+			case isRegexMetaCharacter(pattern[i]):
+				sets[len(sets)-1].WriteByte(pattern[i])
+			case pattern[i] == '\\':
+				sets[len(sets)-1].WriteByte('\\')
+			default:
+				return nil
+			}
+			escaped = false
+		} else {
+			switch {
+			case isRegexMetaCharacter(pattern[i]):
+				if pattern[i] == '|' {
+					sets = append(sets, &strings.Builder{})
+				} else {
+					return nil
+				}
+			case pattern[i] == '\\':
+				escaped = true
+			default:
+				sets[len(sets)-1].WriteByte(pattern[i])
+			}
+		}
+	}
+	matches := make([]string, 0, len(sets))
+	for _, s := range sets {
+		if s.Len() > 0 {
+			matches = append(matches, s.String())
+		}
+	}
+	return matches
+}
+
 // PostingsForMatchers assembles a single postings iterator against the index reader
 // based on the given matchers.
 func PostingsForMatchers(ix IndexReader, ms ...labels.Matcher) (index.Postings, error) {
@@ -346,6 +403,14 @@ func postingsForMatcher(ix IndexReader, m labels.Matcher) (index.Postings, error
 		return ix.Postings(em.Name(), em.Value())
 	}
 
+	// Fast-path for set matching.
+	if em, ok := m.(*labels.RegexpMatcher); ok {
+		setMatches := findSetMatches(em.Value())
+		if len(setMatches) > 0 {
+			return postingsForSetMatcher(ix, em.Name(), setMatches)
+		}
+	}
+
 	tpls, err := ix.LabelValues(m.Name())
 	if err != nil {
 		return nil, err
@@ -411,6 +476,18 @@ func inversePostingsForMatcher(ix IndexReader, m labels.Matcher) (index.Postings
 	return index.Merge(rit...), nil
 }
 
+func postingsForSetMatcher(ix IndexReader, name string, matches []string) (index.Postings, error) {
+	var its []index.Postings
+	for _, match := range matches {
+		if it, err := ix.Postings(name, match); err == nil {
+			its = append(its, it)
+		} else {
+			return nil, err
+		}
+	}
+	return index.Merge(its...), nil
+}
+
 func mergeStrings(a, b []string) []string {
 	maxl := len(a)
 	if len(b) > len(a) {
diff --git a/wal/wal.go b/wal/wal.go
index a51d436..cb2e11f 100644
--- a/wal/wal.go
+++ b/wal/wal.go
@@ -171,6 +171,7 @@ type WAL struct {
 	pageCompletions prometheus.Counter
 	truncateFail    prometheus.Counter
 	truncateTotal   prometheus.Counter
+	currentSegment  prometheus.Gauge
 }
 
 // New returns a new WAL over the given directory.
@@ -218,34 +219,34 @@ func NewSize(logger log.Logger, reg prometheus.Registerer, dir string, segmentSi
 		Name: "prometheus_tsdb_wal_truncations_total",
 		Help: "Total number of WAL truncations attempted.",
 	})
+	w.currentSegment = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "prometheus_tsdb_wal_segment_current",
+		Help: "WAL segment index that TSDB is currently writing to.",
+	})
 	if reg != nil {
-		reg.MustRegister(w.fsyncDuration, w.pageFlushes, w.pageCompletions, w.truncateFail, w.truncateTotal)
+		reg.MustRegister(w.fsyncDuration, w.pageFlushes, w.pageCompletions, w.truncateFail, w.truncateTotal, w.currentSegment)
 	}
 
 	_, j, err := w.Segments()
+	// Index of the Segment we want to open and write to.
+	writeSegmentIndex := 0
 	if err != nil {
 		return nil, errors.Wrap(err, "get segment range")
 	}
-	// Fresh dir, no segments yet.
-	if j == -1 {
-		segment, err := CreateSegment(w.dir, 0)
-		if err != nil {
-			return nil, err
-		}
+	// If some segments already exist create one with a higher index than the last segment.
+	if j != -1 {
+		writeSegmentIndex = j + 1
+	}
 
-		if err := w.setSegment(segment); err != nil {
-			return nil, err
-		}
-	} else {
-		segment, err := OpenWriteSegment(logger, w.dir, j)
-		if err != nil {
-			return nil, err
-		}
+	segment, err := CreateSegment(w.dir, writeSegmentIndex)
+	if err != nil {
+		return nil, err
+	}
 
-		if err := w.setSegment(segment); err != nil {
-			return nil, err
-		}
+	if err := w.setSegment(segment); err != nil {
+		return nil, err
 	}
+
 	go w.run()
 
 	return w, nil
@@ -358,6 +359,9 @@ func (w *WAL) Repair(origErr error) error {
 	}
 	// We expect an error here from r.Err(), so nothing to handle.
 
+	// We need to pad to the end of the last page in the repaired segment
+	w.flushPage(true)
+
 	// We explicitly close even when there is a defer for Windows to be
 	// able to delete it. The defer is in place to close it in-case there
 	// are errors above.
@@ -367,6 +371,20 @@ func (w *WAL) Repair(origErr error) error {
 	if err := os.Remove(tmpfn); err != nil {
 		return errors.Wrap(err, "delete corrupted segment")
 	}
+
+	// Explicitly close the the segment we just repaired to avoid issues with Windows.
+	s.Close()
+
+	// We always want to start writing to a new Segment rather than an existing
+	// Segment, which is handled by NewSize, but earlier in Repair we're deleting
+	// all segments that come after the corrupted Segment. Recreate a new Segment here.
+	s, err = CreateSegment(w.dir, cerr.Segment+1)
+	if err != nil {
+		return err
+	}
+	if err := w.setSegment(s); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -413,7 +431,7 @@ func (w *WAL) setSegment(segment *Segment) error {
 		return err
 	}
 	w.donePages = int(stat.Size() / pageSize)
-
+	w.currentSegment.Set(float64(segment.Index()))
 	return nil
 }
 
@@ -426,10 +444,10 @@ func (w *WAL) flushPage(clear bool) error {
 	p := w.page
 	clear = clear || p.full()
 
-	// No more data will fit into the page. Enqueue and clear it.
+	// No more data will fit into the page or an implicit clear.
+	// Enqueue and clear it.
 	if clear {
 		p.alloc = pageSize // Write till end of page.
-		w.pageCompletions.Inc()
 	}
 	n, err := w.segment.Write(p.buf[p.flushed:p.alloc])
 	if err != nil {
@@ -445,6 +463,7 @@ func (w *WAL) flushPage(clear bool) error {
 		p.alloc = 0
 		p.flushed = 0
 		w.donePages++
+		w.pageCompletions.Inc()
 	}
 	return nil
 }
@@ -495,10 +514,18 @@ func (w *WAL) Log(recs ...[]byte) error {
 	return nil
 }
 
-// log writes rec to the log and forces a flush of the current page if its
-// the final record of a batch, the record is bigger than the page size or
-// the current page is full.
+// log writes rec to the log and forces a flush of the current page if:
+// - the final record of a batch
+// - the record is bigger than the page size
+// - the current page is full.
 func (w *WAL) log(rec []byte, final bool) error {
+	// When the last page flush failed the page will remain full.
+	// When the page is full, need to flush it before trying to add more records to it.
+	if w.page.full() {
+		if err := w.flushPage(true); err != nil {
+			return err
+		}
+	}
 	// If the record is too big to fit within the active page in the current
 	// segment, terminate the active segment and advance to the next one.
 	// This ensures that records do not cross segment boundaries.
@@ -696,7 +723,7 @@ func NewSegmentsRangeReader(sr ...SegmentRange) (io.ReadCloser, error) {
 			segs = append(segs, s)
 		}
 	}
-	return newSegmentBufReader(segs...), nil
+	return NewSegmentBufReader(segs...), nil
 }
 
 // segmentBufReader is a buffered reader that reads in multiples of pages.
@@ -711,7 +738,7 @@ type segmentBufReader struct {
 	off  int // Offset of read data into current segment.
 }
 
-func newSegmentBufReader(segs ...*Segment) *segmentBufReader {
+func NewSegmentBufReader(segs ...*Segment) *segmentBufReader {
 	return &segmentBufReader{
 		buf:  bufio.NewReaderSize(segs[0], 16*pageSize),
 		segs: segs,


Test patch:
diff --git a/block_test.go b/block_test.go
index bdfd58f..3ae2ec2 100644
--- a/block_test.go
+++ b/block_test.go
@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	"os"
 	"path/filepath"
+	"strconv"
 	"testing"
 
 	"github.com/go-kit/kit/log"
@@ -184,6 +185,11 @@ func createBlock(tb testing.TB, dir string, series []Series) string {
 	return filepath.Join(dir, ulid.String())
 }
 
+const (
+	defaultLabelName  = "labelName"
+	defaultLabelValue = "labelValue"
+)
+
 // genSeries generates series with a given number of labels and values.
 func genSeries(totalSeries, labelCount int, mint, maxt int64) []Series {
 	if totalSeries == 0 || labelCount == 0 {
@@ -193,8 +199,9 @@ 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)
-		for len(lbls) < labelCount {
-			lbls[randString()] = randString()
+		lbls[defaultLabelName] = strconv.Itoa(i)
+		for j := 1; len(lbls) < labelCount; j++ {
+			lbls[defaultLabelName+strconv.Itoa(j)] = defaultLabelValue + strconv.Itoa(j)
 		}
 		samples := make([]tsdbutil.Sample, 0, maxt-mint+1)
 		for t := mint; t <= maxt; t++ {
@@ -224,31 +231,3 @@ func populateSeries(lbls []map[string]string, mint, maxt int64) []Series {
 	}
 	return series
 }
-
-const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
-const (
-	letterIdxBits = 6                    // 6 bits to represent a letter index
-	letterIdxMask = 1<<letterIdxBits - 1 // All 1-bits, as many as letterIdxBits
-	letterIdxMax  = 63 / letterIdxBits   // # of letter indices fitting in 63 bits
-)
-
-// randString generates random string.
-func randString() string {
-	maxLength := int32(50)
-	length := rand.Int31n(maxLength)
-	b := make([]byte, length+1)
-	// A rand.Int63() generates 63 random bits, enough for letterIdxMax characters!
-	for i, cache, remain := length, rand.Int63(), letterIdxMax; i >= 0; {
-		if remain == 0 {
-			cache, remain = rand.Int63(), letterIdxMax
-		}
-		if idx := int(cache & letterIdxMask); idx < len(letterBytes) {
-			b[i] = letterBytes[idx]
-			i--
-		}
-		cache >>= letterIdxBits
-		remain--
-	}
-
-	return string(b)
-}
diff --git a/head_test.go b/head_test.go
index 9e78b6b..79092dd 100644
--- a/head_test.go
+++ b/head_test.go
@@ -114,6 +114,9 @@ func TestHead_ReadWAL(t *testing.T) {
 			{Ref: 10, T: 101, V: 5},
 			{Ref: 50, T: 101, V: 6},
 		},
+		[]Stone{
+			{ref: 0, intervals: []Interval{{Mint: 99, Maxt: 101}}},
+		},
 	}
 	dir, err := ioutil.TempDir("", "test_read_wal")
 	testutil.Ok(t, err)
diff --git a/querier_test.go b/querier_test.go
index cb53462..dfbc6a7 100644
--- a/querier_test.go
+++ b/querier_test.go
@@ -1691,6 +1691,192 @@ func BenchmarkQuerySeek(b *testing.B) {
 	}
 }
 
+// Refer to https://github.com/prometheus/prometheus/issues/2651.
+func BenchmarkSetMatcher(b *testing.B) {
+	cases := []struct {
+		numBlocks                   int
+		numSeries                   int
+		numSamplesPerSeriesPerBlock int
+		cardinality                 int
+		pattern                     string
+	}{
+		// The first three cases are to find out whether the set
+		// matcher is always faster than regex matcher.
+		{
+			numBlocks:                   1,
+			numSeries:                   1,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 100,
+			pattern:                     "^(?:1|2|3|4|5|6|7|8|9|10)$",
+		},
+		{
+			numBlocks:                   1,
+			numSeries:                   15,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 100,
+			pattern:                     "^(?:1|2|3|4|5|6|7|8|9|10)$",
+		},
+		{
+			numBlocks:                   1,
+			numSeries:                   15,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 100,
+			pattern:                     "^(?:1|2|3)$",
+		},
+		// Big data sizes benchmarks.
+		{
+			numBlocks:                   20,
+			numSeries:                   1000,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 100,
+			pattern:                     "^(?:1|2|3)$",
+		},
+		{
+			numBlocks:                   20,
+			numSeries:                   1000,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 100,
+			pattern:                     "^(?:1|2|3|4|5|6|7|8|9|10)$",
+		},
+		// Increase cardinality.
+		{
+			numBlocks:                   1,
+			numSeries:                   100000,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 100000,
+			pattern:                     "^(?:1|2|3|4|5|6|7|8|9|10)$",
+		},
+		{
+			numBlocks:                   1,
+			numSeries:                   500000,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 500000,
+			pattern:                     "^(?:1|2|3|4|5|6|7|8|9|10)$",
+		},
+		{
+			numBlocks:                   10,
+			numSeries:                   500000,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 500000,
+			pattern:                     "^(?:1|2|3|4|5|6|7|8|9|10)$",
+		},
+		{
+			numBlocks:                   1,
+			numSeries:                   1000000,
+			numSamplesPerSeriesPerBlock: 10,
+			cardinality:                 1000000,
+			pattern:                     "^(?:1|2|3|4|5|6|7|8|9|10)$",
+		},
+	}
+
+	for _, c := range cases {
+		dir, err := ioutil.TempDir("", "bench_postings_for_matchers")
+		testutil.Ok(b, err)
+		defer func() {
+			testutil.Ok(b, os.RemoveAll(dir))
+		}()
+
+		var (
+			blocks          []*Block
+			prefilledLabels []map[string]string
+			generatedSeries []Series
+		)
+		for i := int64(0); i < int64(c.numBlocks); i++ {
+			mint := i * int64(c.numSamplesPerSeriesPerBlock)
+			maxt := mint + int64(c.numSamplesPerSeriesPerBlock) - 1
+			if len(prefilledLabels) == 0 {
+				generatedSeries = genSeries(c.numSeries, 10, mint, maxt)
+				for _, s := range generatedSeries {
+					prefilledLabels = append(prefilledLabels, s.Labels().Map())
+				}
+			} else {
+				generatedSeries = populateSeries(prefilledLabels, mint, maxt)
+			}
+			block, err := OpenBlock(nil, createBlock(b, dir, generatedSeries), nil)
+			testutil.Ok(b, err)
+			blocks = append(blocks, block)
+			defer block.Close()
+		}
+
+		que := &querier{
+			blocks: make([]Querier, 0, len(blocks)),
+		}
+		for _, blk := range blocks {
+			q, err := NewBlockQuerier(blk, math.MinInt64, math.MaxInt64)
+			testutil.Ok(b, err)
+			que.blocks = append(que.blocks, q)
+		}
+		defer que.Close()
+
+		benchMsg := fmt.Sprintf("nSeries=%d,nBlocks=%d,cardinality=%d,pattern=\"%s\"", c.numSeries, c.numBlocks, c.cardinality, c.pattern)
+		b.Run(benchMsg, func(b *testing.B) {
+			b.ResetTimer()
+			b.ReportAllocs()
+			for n := 0; n < b.N; n++ {
+				_, err := que.Select(labels.NewMustRegexpMatcher("test", c.pattern))
+				testutil.Ok(b, err)
+
+			}
+		})
+	}
+}
+
+// Refer to https://github.com/prometheus/prometheus/issues/2651.
+func TestFindSetMatches(t *testing.T) {
+	cases := []struct {
+		pattern string
+		exp     []string
+	}{
+		// Simple sets.
+		{
+			pattern: "^(?:foo|bar|baz)$",
+			exp: []string{
+				"foo",
+				"bar",
+				"baz",
+			},
+		},
+		// Simple sets containing escaped characters.
+		{
+			pattern: "^(?:fo\\.o|bar\\?|\\^baz)$",
+			exp: []string{
+				"fo.o",
+				"bar?",
+				"^baz",
+			},
+		},
+		// Simple sets containing special characters without escaping.
+		{
+			pattern: "^(?:fo.o|bar?|^baz)$",
+			exp:     nil,
+		},
+		// Missing wrapper.
+		{
+			pattern: "foo|bar|baz",
+			exp:     nil,
+		},
+	}
+
+	for _, c := range cases {
+		matches := findSetMatches(c.pattern)
+		if len(c.exp) == 0 {
+			if len(matches) != 0 {
+				t.Errorf("Evaluating %s, unexpected result %v", c.pattern, matches)
+			}
+		} else {
+			if len(matches) != len(c.exp) {
+				t.Errorf("Evaluating %s, length of result not equal to exp", c.pattern)
+			} else {
+				for i := 0; i < len(c.exp); i++ {
+					if c.exp[i] != matches[i] {
+						t.Errorf("Evaluating %s, unexpected result %s", c.pattern, matches[i])
+					}
+				}
+			}
+		}
+	}
+}
+
 func TestPostingsForMatchers(t *testing.T) {
 	h, err := NewHead(nil, nil, nil, 1000)
 	testutil.Ok(t, err)
@@ -1703,6 +1889,7 @@ func TestPostingsForMatchers(t *testing.T) {
 	app.Add(labels.FromStrings("n", "1", "i", "a"), 0, 0)
 	app.Add(labels.FromStrings("n", "1", "i", "b"), 0, 0)
 	app.Add(labels.FromStrings("n", "2"), 0, 0)
+	app.Add(labels.FromStrings("n", "2.5"), 0, 0)
 	testutil.Ok(t, app.Commit())
 
 	cases := []struct {
@@ -1735,6 +1922,7 @@ func TestPostingsForMatchers(t *testing.T) {
 				labels.FromStrings("n", "1", "i", "a"),
 				labels.FromStrings("n", "1", "i", "b"),
 				labels.FromStrings("n", "2"),
+				labels.FromStrings("n", "2.5"),
 			},
 		},
 		// Not equals.
@@ -1742,6 +1930,7 @@ func TestPostingsForMatchers(t *testing.T) {
 			matchers: []labels.Matcher{labels.Not(labels.NewEqualMatcher("n", "1"))},
 			exp: []labels.Labels{
 				labels.FromStrings("n", "2"),
+				labels.FromStrings("n", "2.5"),
 			},
 		},
 		{
@@ -1796,6 +1985,7 @@ func TestPostingsForMatchers(t *testing.T) {
 			exp: []labels.Labels{
 				labels.FromStrings("n", "1"),
 				labels.FromStrings("n", "2"),
+				labels.FromStrings("n", "2.5"),
 			},
 		},
 		{
@@ -1824,6 +2014,7 @@ func TestPostingsForMatchers(t *testing.T) {
 			matchers: []labels.Matcher{labels.Not(labels.NewMustRegexpMatcher("n", "^1$"))},
 			exp: []labels.Labels{
 				labels.FromStrings("n", "2"),
+				labels.FromStrings("n", "2.5"),
 			},
 		},
 		{
@@ -1869,6 +2060,46 @@ func TestPostingsForMatchers(t *testing.T) {
 				labels.FromStrings("n", "1", "i", "a"),
 			},
 		},
+		// Set optimization for Regex.
+		// Refer to https://github.com/prometheus/prometheus/issues/2651.
+		{
+			matchers: []labels.Matcher{labels.NewMustRegexpMatcher("n", "^(?:1|2)$")},
+			exp: []labels.Labels{
+				labels.FromStrings("n", "1"),
+				labels.FromStrings("n", "1", "i", "a"),
+				labels.FromStrings("n", "1", "i", "b"),
+				labels.FromStrings("n", "2"),
+			},
+		},
+		{
+			matchers: []labels.Matcher{labels.NewMustRegexpMatcher("i", "^(?:a|b)$")},
+			exp: []labels.Labels{
+				labels.FromStrings("n", "1", "i", "a"),
+				labels.FromStrings("n", "1", "i", "b"),
+			},
+		},
+		{
+			matchers: []labels.Matcher{labels.NewMustRegexpMatcher("n", "^(?:x1|2)$")},
+			exp: []labels.Labels{
+				labels.FromStrings("n", "2"),
+			},
+		},
+		{
+			matchers: []labels.Matcher{labels.NewMustRegexpMatcher("n", "^(?:2|2\\.5)$")},
+			exp: []labels.Labels{
+				labels.FromStrings("n", "2"),
+				labels.FromStrings("n", "2.5"),
+			},
+		},
+		// Empty value.
+		{
+			matchers: []labels.Matcher{labels.NewMustRegexpMatcher("i", "^(?:c||d)$")},
+			exp: []labels.Labels{
+				labels.FromStrings("n", "1"),
+				labels.FromStrings("n", "2"),
+				labels.FromStrings("n", "2.5"),
+			},
+		},
 	}
 
 	ir, err := h.Index()
diff --git a/wal/wal_test.go b/wal/wal_test.go
index 391c4b6..577ae5f 100644
--- a/wal/wal_test.go
+++ b/wal/wal_test.go
@@ -23,6 +23,7 @@ import (
 	"path/filepath"
 	"testing"
 
+	client_testutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/tsdb/testutil"
 )
 
@@ -142,20 +143,35 @@ func TestWAL_Repair(t *testing.T) {
 			testutil.Ok(t, err)
 			defer w.Close()
 
-			sr, err := NewSegmentsReader(dir)
+			first, last, err := w.Segments()
 			testutil.Ok(t, err)
-			r := NewReader(sr)
 
-			for r.Next() {
+			// Backfill segments from the most recent checkpoint onwards.
+			for i := first; i <= last; i++ {
+				s, err := OpenReadSegment(SegmentName(w.Dir(), i))
+				testutil.Ok(t, err)
+
+				sr := NewSegmentBufReader(s)
+				testutil.Ok(t, err)
+				r := NewReader(sr)
+				for r.Next() {
+				}
+
+				//Close the segment so we don't break things on Windows.
+				s.Close()
+
+				// No corruption in this segment.
+				if r.Err() == nil {
+					continue
+				}
+				testutil.Ok(t, w.Repair(r.Err()))
+				break
 			}
-			testutil.NotOk(t, r.Err())
-			testutil.Ok(t, sr.Close())
 
-			testutil.Ok(t, w.Repair(r.Err()))
-			sr, err = NewSegmentsReader(dir)
+			sr, err := NewSegmentsReader(dir)
 			testutil.Ok(t, err)
 			defer sr.Close()
-			r = NewReader(sr)
+			r := NewReader(sr)
 
 			var result [][]byte
 			for r.Next() {
@@ -171,10 +187,13 @@ func TestWAL_Repair(t *testing.T) {
 				}
 			}
 
-			// Make sure the last segment is the corrupt segment.
-			_, last, err := w.Segments()
+			// Make sure there is a new 0 size Segment after the corrupted Segment.
+			_, last, err = w.Segments()
 			testutil.Ok(t, err)
-			testutil.Equals(t, test.corrSgm, last)
+			testutil.Equals(t, test.corrSgm+1, last)
+			fi, err := os.Stat(SegmentName(dir, last))
+			testutil.Ok(t, err)
+			testutil.Equals(t, int64(0), fi.Size())
 		})
 	}
 }
@@ -275,6 +294,10 @@ func TestCorruptAndCarryOn(t *testing.T) {
 		err = w.Repair(corruptionErr)
 		testutil.Ok(t, err)
 
+		// Ensure that we have a completely clean slate after reapiring.
+		testutil.Equals(t, w.segment.Index(), 1) // We corrupted segment 0.
+		testutil.Equals(t, w.donePages, 0)
+
 		for i := 0; i < 5; i++ {
 			buf := make([]byte, recordSize)
 			_, err := rand.Read(buf)
@@ -318,6 +341,35 @@ func TestClose(t *testing.T) {
 	testutil.NotOk(t, w.Close())
 }
 
+func TestSegmentMetric(t *testing.T) {
+	var (
+		segmentSize = pageSize
+		recordSize  = (pageSize / 2) - recordHeaderSize
+	)
+
+	dir, err := ioutil.TempDir("", "segment_metric")
+	testutil.Ok(t, err)
+	defer func() {
+		testutil.Ok(t, os.RemoveAll(dir))
+	}()
+	w, err := NewSize(nil, nil, dir, segmentSize)
+	testutil.Ok(t, err)
+
+	initialSegment := client_testutil.ToFloat64(w.currentSegment)
+
+	// Write 3 records, each of which is half the segment size, meaning we should rotate to the next segment.
+	for i := 0; i < 3; i++ {
+		buf := make([]byte, recordSize)
+		_, err := rand.Read(buf)
+		testutil.Ok(t, err)
+
+		err = w.Log(buf)
+		testutil.Ok(t, err)
+	}
+	testutil.Assert(t, client_testutil.ToFloat64(w.currentSegment) == initialSegment+1, "segment metric did not increment after segment rotation")
+	testutil.Ok(t, w.Close())
+}
+
 func BenchmarkWAL_LogBatched(b *testing.B) {
 	dir, err := ioutil.TempDir("", "bench_logbatch")
 	testutil.Ok(b, err)