Repository: prometheus/tsdb
Pull request #451 (2018-11-13 18:24:48)

return an error when the  last wal segment record is torn.
this ensures that a repair will be run when the last record in a segment is torn.

to reproduce:
 - kill Prometheus just after a first record is written (before a last record is written).
 - Start Prometheus. It show no signs of the issue and continues to write records after the torn record. This triggers a dead loop at checkpointing.

most likely the cause for:
https://github.com/prometheus/prometheus/issues/4695
~https://github.com/prometheus/prometheus/issues/4842~

cc @fabxc , @SuperQ 	


note to myself :smiling_imp: 
the wal records are grouped and written in pages, each  page is [32kb](https://github.com/prometheus/tsdb/blob/171fc4ab5da16793145d0e1df4403a073b1d2f78/wal/wal.go#L40) and when the last record doesn't fit in the current page it is split in parts(first,last). So the last record of one page would have the first part of a record and the first record in the next page would have the last part of a record.
So in case of a crash just after writing the first part the next WAL read didn't have a  check to trigger a repair in such cases and the PR fixes this.


Signed-off-by: Krasi Georgiev <kgeorgie@redhat.com>


krasi-georgiev: ping @gouthamve 	

Patch:
diff --git a/.travis.yml b/.travis.yml
index c24773b..78fe685 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -4,8 +4,8 @@ dist: trusty
 language: go
 
 go:
-  - 1.9.x
   - 1.10.x
+  - 1.11.x
 
 go_import_path: github.com/prometheus/tsdb
 
diff --git a/Makefile.common b/Makefile.common
index c9d8323..741579e 100644
--- a/Makefile.common
+++ b/Makefile.common
@@ -16,7 +16,7 @@
 # !!! Open PRs only against the prometheus/prometheus/Makefile.common repository!
 
 # Example usage :
-# Create the main Makefile in the root project directory. 
+# Create the main Makefile in the root project directory.
 # include Makefile.common
 # customTarget:
 # 	@echo ">> Running customTarget"
@@ -28,18 +28,53 @@ unexport GOBIN
 GO           ?= go
 GOFMT        ?= $(GO)fmt
 FIRST_GOPATH := $(firstword $(subst :, ,$(shell $(GO) env GOPATH)))
+GOOPTS       ?=
+
+GO_VERSION        ?= $(shell $(GO) version)
+GO_VERSION_NUMBER ?= $(word 3, $(GO_VERSION))
+PRE_GO_111        ?= $(shell echo $(GO_VERSION_NUMBER) | grep -E 'go1\.(10|[0-9])\.')
+
+unexport GOVENDOR
+ifeq (, $(PRE_GO_111))
+	ifneq (,$(wildcard go.mod))
+		# Enforce Go modules support just in case the directory is inside GOPATH (and for Travis CI).
+		GO111MODULE := on
+
+		ifneq (,$(wildcard vendor))
+			# Always use the local vendor/ directory to satisfy the dependencies.
+			GOOPTS := $(GOOPTS) -mod=vendor
+		endif
+	endif
+else
+	ifneq (,$(wildcard go.mod))
+		ifneq (,$(wildcard vendor))
+$(warning This repository requires Go >= 1.11 because of Go modules)
+$(warning Some recipes may not work as expected as the current Go runtime is '$(GO_VERSION_NUMBER)')
+		endif
+	else
+		# This repository isn't using Go modules (yet).
+		GOVENDOR := $(FIRST_GOPATH)/bin/govendor
+	endif
+
+	unexport GO111MODULE
+endif
 PROMU        := $(FIRST_GOPATH)/bin/promu
 STATICCHECK  := $(FIRST_GOPATH)/bin/staticcheck
-GOVENDOR     := $(FIRST_GOPATH)/bin/govendor
 pkgs          = ./...
 
+GO_VERSION        ?= $(shell $(GO) version)
+GO_BUILD_PLATFORM ?= $(subst /,-,$(lastword $(GO_VERSION)))
+
+PROMU_VERSION ?= 0.2.0
+PROMU_URL     := https://github.com/prometheus/promu/releases/download/v$(PROMU_VERSION)/promu-$(PROMU_VERSION).$(GO_BUILD_PLATFORM).tar.gz
+
 PREFIX                  ?= $(shell pwd)
 BIN_DIR                 ?= $(shell pwd)
 DOCKER_IMAGE_TAG        ?= $(subst /,-,$(shell git rev-parse --abbrev-ref HEAD))
 DOCKER_REPO             ?= prom
 
 .PHONY: all
-all: style staticcheck unused build test
+all: precheck style staticcheck unused build test
 
 # This rule is used to forward a target like "build" to "common-build".  This
 # allows a new "build" target to be defined in a Makefile which includes this
@@ -70,37 +105,54 @@ common-check_license:
 .PHONY: common-test-short
 common-test-short:
 	@echo ">> running short tests"
-	$(GO) test -short $(pkgs)
+	GO111MODULE=$(GO111MODULE) $(GO) test -short $(GOOPTS) $(pkgs)
 
 .PHONY: common-test
 common-test:
 	@echo ">> running all tests"
-	$(GO) test -race $(pkgs)
+	GO111MODULE=$(GO111MODULE) $(GO) test -race $(GOOPTS) $(pkgs)
 
 .PHONY: common-format
 common-format:
 	@echo ">> formatting code"
-	$(GO) fmt $(pkgs)
+	GO111MODULE=$(GO111MODULE) $(GO) fmt $(GOOPTS) $(pkgs)
 
 .PHONY: common-vet
 common-vet:
 	@echo ">> vetting code"
-	$(GO) vet $(pkgs)
+	GO111MODULE=$(GO111MODULE) $(GO) vet $(GOOPTS) $(pkgs)
 
 .PHONY: common-staticcheck
 common-staticcheck: $(STATICCHECK)
 	@echo ">> running staticcheck"
+ifdef GO111MODULE
+	GO111MODULE=$(GO111MODULE) $(STATICCHECK) -ignore "$(STATICCHECK_IGNORE)" -checks "SA*" $(pkgs)
+else
 	$(STATICCHECK) -ignore "$(STATICCHECK_IGNORE)" $(pkgs)
+endif
 
 .PHONY: common-unused
 common-unused: $(GOVENDOR)
+ifdef GOVENDOR
 	@echo ">> running check for unused packages"
 	@$(GOVENDOR) list +unused | grep . && exit 1 || echo 'No unused packages'
+else
+ifdef GO111MODULE
+	@echo ">> running check for unused/missing packages in go.mod"
+	GO111MODULE=$(GO111MODULE) $(GO) mod tidy
+	@git diff --exit-code -- go.sum go.mod
+ifneq (,$(wildcard vendor))
+	@echo ">> running check for unused packages in vendor/"
+	GO111MODULE=$(GO111MODULE) $(GO) mod vendor
+	@git diff --exit-code -- go.sum go.mod vendor/
+endif
+endif
+endif
 
 .PHONY: common-build
 common-build: promu
 	@echo ">> building binaries"
-	$(PROMU) build --prefix $(PREFIX)
+	GO111MODULE=$(GO111MODULE) $(PROMU) build --prefix $(PREFIX)
 
 .PHONY: common-tarball
 common-tarball: promu
@@ -120,13 +172,52 @@ common-docker-tag-latest:
 	docker tag "$(DOCKER_REPO)/$(DOCKER_IMAGE_NAME):$(DOCKER_IMAGE_TAG)" "$(DOCKER_REPO)/$(DOCKER_IMAGE_NAME):latest"
 
 .PHONY: promu
-promu:
-	GOOS= GOARCH= $(GO) get -u github.com/prometheus/promu
+promu: $(PROMU)
+
+$(PROMU):
+	curl -s -L $(PROMU_URL) | tar -xvz -C /tmp
+	mkdir -v -p $(FIRST_GOPATH)/bin
+	cp -v /tmp/promu-$(PROMU_VERSION).$(GO_BUILD_PLATFORM)/promu $(PROMU)
+
+.PHONY: proto
+proto:
+	@echo ">> generating code from proto files"
+	@./scripts/genproto.sh
 
 .PHONY: $(STATICCHECK)
 $(STATICCHECK):
-	GOOS= GOARCH= $(GO) get -u honnef.co/go/tools/cmd/staticcheck
-
+ifdef GO111MODULE
+# Get staticcheck from a temporary directory to avoid modifying the local go.{mod,sum}.
+# See https://github.com/golang/go/issues/27643.
+# For now, we are using the next branch of staticcheck because master isn't compatible yet with Go modules.
+	tmpModule=$$(mktemp -d 2>&1) && \
+	mkdir -p $${tmpModule}/staticcheck && \
+	cd "$${tmpModule}"/staticcheck && \
+	GO111MODULE=on $(GO) mod init example.com/staticcheck && \
+	GO111MODULE=on GOOS= GOARCH= $(GO) get -u honnef.co/go/tools/cmd/staticcheck@next && \
+	rm -rf $${tmpModule};
+else
+	GOOS= GOARCH= GO111MODULE=off $(GO) get -u honnef.co/go/tools/cmd/staticcheck
+endif
+
+ifdef GOVENDOR
 .PHONY: $(GOVENDOR)
 $(GOVENDOR):
 	GOOS= GOARCH= $(GO) get -u github.com/kardianos/govendor
+endif
+
+.PHONY: precheck
+precheck::
+
+define PRECHECK_COMMAND_template =
+precheck:: $(1)_precheck
+
+
+PRECHECK_COMMAND_$(1) ?= $(1) $$(strip $$(PRECHECK_OPTIONS_$(1)))
+.PHONY: $(1)_precheck
+$(1)_precheck:
+	@if ! $$(PRECHECK_COMMAND_$(1)) 1>/dev/null 2>&1; then \
+		echo "Execution of '$$(PRECHECK_COMMAND_$(1))' command failed. Is $(1) installed?"; \
+		exit 1; \
+	fi
+endef
diff --git a/block.go b/block.go
index 7081db0..e5a66bd 100644
--- a/block.go
+++ b/block.go
@@ -15,7 +15,6 @@
 package tsdb
 
 import (
-	"encoding/binary"
 	"encoding/json"
 	"io/ioutil"
 	"os"
@@ -278,23 +277,13 @@ func OpenBlock(dir string, pool chunkenc.Pool) (*Block, error) {
 		return nil, err
 	}
 
-	// Calculating symbol table size.
-	tmp := make([]byte, 8)
-	symTblSize := uint64(0)
-	for _, v := range ir.SymbolTable() {
-		// Size of varint length of the symbol.
-		symTblSize += uint64(binary.PutUvarint(tmp, uint64(len(v))))
-		// Size of the symbol.
-		symTblSize += uint64(len(v))
-	}
-
 	pb := &Block{
 		dir:             dir,
 		meta:            *meta,
 		chunkr:          cr,
 		indexr:          ir,
 		tombstones:      tr,
-		symbolTableSize: symTblSize,
+		symbolTableSize: ir.SymbolTableSize(),
 	}
 	return pb, nil
 }
diff --git a/chunkenc/bstream.go b/chunkenc/bstream.go
index a352f07..ef04d44 100644
--- a/chunkenc/bstream.go
+++ b/chunkenc/bstream.go
@@ -49,8 +49,8 @@ type bstream struct {
 	count  uint8  // how many bits are valid in current byte
 }
 
-func newBReader(b []byte) *bstream {
-	return &bstream{stream: b, count: 8}
+func newBReader(b []byte) bstream {
+	return bstream{stream: b, count: 8}
 }
 
 func newBWriter(size int) *bstream {
diff --git a/chunkenc/xor.go b/chunkenc/xor.go
index e5e9c9a..77cc320 100644
--- a/chunkenc/xor.go
+++ b/chunkenc/xor.go
@@ -221,7 +221,7 @@ func (a *xorAppender) writeVDelta(v float64) {
 }
 
 type xorIterator struct {
-	br       *bstream
+	br       bstream
 	numTotal uint16
 	numRead  uint16
 
@@ -249,7 +249,7 @@ func (it *xorIterator) Next() bool {
 	}
 
 	if it.numRead == 0 {
-		t, err := binary.ReadVarint(it.br)
+		t, err := binary.ReadVarint(&it.br)
 		if err != nil {
 			it.err = err
 			return false
@@ -266,7 +266,7 @@ func (it *xorIterator) Next() bool {
 		return true
 	}
 	if it.numRead == 1 {
-		tDelta, err := binary.ReadUvarint(it.br)
+		tDelta, err := binary.ReadUvarint(&it.br)
 		if err != nil {
 			it.err = err
 			return false
diff --git a/compact.go b/compact.go
index ca027ad..f8e6ff5 100644
--- a/compact.go
+++ b/compact.go
@@ -59,7 +59,9 @@ type Compactor interface {
 
 	// Compact runs compaction against the provided directories. Must
 	// only be called concurrently with results of Plan().
-	Compact(dest string, dirs ...string) (ulid.ULID, error)
+	// Can optionally pass a list of already open blocks,
+	// to avoid having to reopen them.
+	Compact(dest string, dirs []string, open []*Block) (ulid.ULID, error)
 }
 
 // LeveledCompactor implements the Compactor interface.
@@ -317,24 +319,39 @@ func compactBlockMetas(uid ulid.ULID, blocks ...*BlockMeta) *BlockMeta {
 
 // Compact creates a new block in the compactor's directory from the blocks in the
 // provided directories.
-func (c *LeveledCompactor) Compact(dest string, dirs ...string) (uid ulid.ULID, err error) {
+func (c *LeveledCompactor) Compact(dest string, dirs []string, open []*Block) (uid ulid.ULID, err error) {
 	var (
 		blocks []BlockReader
 		bs     []*Block
 		metas  []*BlockMeta
 		uids   []string
 	)
+	start := time.Now()
 
 	for _, d := range dirs {
-		b, err := OpenBlock(d, c.chunkPool)
+		meta, err := readMetaFile(d)
 		if err != nil {
 			return uid, err
 		}
-		defer b.Close()
 
-		meta, err := readMetaFile(d)
-		if err != nil {
-			return uid, err
+		var b *Block
+
+		// Use already open blocks if we can, to avoid
+		// having the index data in memory twice.
+		for _, o := range open {
+			if meta.ULID == o.Meta().ULID {
+				b = o
+				break
+			}
+		}
+
+		if b == nil {
+			var err error
+			b, err = OpenBlock(d, c.chunkPool)
+			if err != nil {
+				return uid, err
+			}
+			defer b.Close()
 		}
 
 		metas = append(metas, meta)
@@ -356,6 +373,7 @@ func (c *LeveledCompactor) Compact(dest string, dirs ...string) (uid ulid.ULID,
 			"maxt", meta.MaxTime,
 			"ulid", meta.ULID,
 			"sources", fmt.Sprintf("%v", uids),
+			"duration", time.Since(start),
 		)
 		return uid, nil
 	}
diff --git a/db.go b/db.go
index 8f45213..40025bd 100644
--- a/db.go
+++ b/db.go
@@ -112,9 +112,13 @@ type DB struct {
 	donec    chan struct{}
 	stopc    chan struct{}
 
-	// cmtx is used to control compactions and deletions.
-	cmtx               sync.Mutex
-	compactionsEnabled bool
+	// cmtx ensures that compactions and deletions don't run simultaneously.
+	cmtx sync.Mutex
+
+	// autoCompactMtx ensures that no compaction gets triggered while
+	// changing the autoCompact var.
+	autoCompactMtx sync.Mutex
+	autoCompact    bool
 }
 
 type dbMetrics struct {
@@ -123,6 +127,7 @@ type dbMetrics struct {
 	reloads              prometheus.Counter
 	reloadsFailed        prometheus.Counter
 	compactionsTriggered prometheus.Counter
+	compactionsSkipped   prometheus.Counter
 	cutoffs              prometheus.Counter
 	cutoffsFailed        prometheus.Counter
 	startTime            prometheus.GaugeFunc
@@ -165,6 +170,10 @@ func newDBMetrics(db *DB, r prometheus.Registerer) *dbMetrics {
 		Name: "prometheus_tsdb_compactions_triggered_total",
 		Help: "Total number of triggered compactions for the partition.",
 	})
+	m.compactionsSkipped = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "prometheus_tsdb_compactions_skipped_total",
+		Help: "Total number of skipped compactions due to disabled auto compaction.",
+	})
 	m.cutoffs = prometheus.NewCounter(prometheus.CounterOpts{
 		Name: "prometheus_tsdb_retention_cutoffs_total",
 		Help: "Number of times the database cut off block data from disk.",
@@ -226,14 +235,14 @@ func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db
 	}
 
 	db = &DB{
-		dir:                dir,
-		logger:             l,
-		opts:               opts,
-		compactc:           make(chan struct{}, 1),
-		donec:              make(chan struct{}),
-		stopc:              make(chan struct{}),
-		compactionsEnabled: true,
-		chunkPool:          chunkenc.NewPool(),
+		dir:         dir,
+		logger:      l,
+		opts:        opts,
+		compactc:    make(chan struct{}, 1),
+		donec:       make(chan struct{}),
+		stopc:       make(chan struct{}),
+		autoCompact: true,
+		chunkPool:   chunkenc.NewPool(),
 	}
 	db.metrics = newDBMetrics(db, r)
 
@@ -262,12 +271,12 @@ func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db
 	if err != nil {
 		return nil, err
 	}
-	if err := db.reload(); err != nil {
-		return nil, err
-	}
 	if err := db.head.Init(); err != nil {
 		return nil, errors.Wrap(err, "read WAL")
 	}
+	if err := db.reload(); err != nil {
+		return nil, err
+	}
 
 	go db.run()
 
@@ -300,14 +309,18 @@ func (db *DB) run() {
 		case <-db.compactc:
 			db.metrics.compactionsTriggered.Inc()
 
-			err := db.compact()
-			if err != nil {
-				level.Error(db.logger).Log("msg", "compaction failed", "err", err)
-				backoff = exponential(backoff, 1*time.Second, 1*time.Minute)
+			db.autoCompactMtx.Lock()
+			if db.autoCompact {
+				if err := db.compact(); err != nil {
+					level.Error(db.logger).Log("msg", "compaction failed", "err", err)
+					backoff = exponential(backoff, 1*time.Second, 1*time.Minute)
+				} else {
+					backoff = 0
+				}
 			} else {
-				backoff = 0
+				db.metrics.compactionsSkipped.Inc()
 			}
-
+			db.autoCompactMtx.Unlock()
 		case <-db.stopc:
 			return
 		}
@@ -369,11 +382,6 @@ func (a dbAppender) Commit() error {
 func (db *DB) compact() (err error) {
 	db.cmtx.Lock()
 	defer db.cmtx.Unlock()
-
-	if !db.compactionsEnabled {
-		return nil
-	}
-
 	// Check whether we have pending head blocks that are ready to be persisted.
 	// They have the highest priority.
 	for {
@@ -429,7 +437,7 @@ func (db *DB) compact() (err error) {
 		default:
 		}
 
-		if _, err := db.compactor.Compact(db.dir, plan...); err != nil {
+		if _, err := db.compactor.Compact(db.dir, plan, db.blocks); err != nil {
 			return errors.Wrapf(err, "compact %s", plan)
 		}
 		runtime.GC()
@@ -731,21 +739,21 @@ func (db *DB) Close() error {
 	return merr.Err()
 }
 
-// DisableCompactions disables compactions.
+// DisableCompactions disables auto compactions.
 func (db *DB) DisableCompactions() {
-	db.cmtx.Lock()
-	defer db.cmtx.Unlock()
+	db.autoCompactMtx.Lock()
+	defer db.autoCompactMtx.Unlock()
 
-	db.compactionsEnabled = false
+	db.autoCompact = false
 	level.Info(db.logger).Log("msg", "compactions disabled")
 }
 
-// EnableCompactions enables compactions.
+// EnableCompactions enables auto compactions.
 func (db *DB) EnableCompactions() {
-	db.cmtx.Lock()
-	defer db.cmtx.Unlock()
+	db.autoCompactMtx.Lock()
+	defer db.autoCompactMtx.Unlock()
 
-	db.compactionsEnabled = true
+	db.autoCompact = true
 	level.Info(db.logger).Log("msg", "compactions enabled")
 }
 
@@ -882,49 +890,6 @@ func (db *DB) CleanTombstones() (err error) {
 	return errors.Wrap(db.reload(), "reload blocks")
 }
 
-// labelNames returns all the unique label names from the Block Readers.
-func labelNames(brs ...BlockReader) (map[string]struct{}, error) {
-	labelNamesMap := make(map[string]struct{})
-	for _, br := range brs {
-		ir, err := br.Index()
-		if err != nil {
-			return nil, errors.Wrap(err, "get IndexReader")
-		}
-		names, err := ir.LabelNames()
-		if err != nil {
-			return nil, errors.Wrap(err, "LabelNames() from IndexReader")
-		}
-		for _, name := range names {
-			labelNamesMap[name] = struct{}{}
-		}
-		if err = ir.Close(); err != nil {
-			return nil, errors.Wrap(err, "close IndexReader")
-		}
-	}
-	return labelNamesMap, nil
-}
-
-// LabelNames returns all the unique label names present in the DB in sorted order.
-func (db *DB) LabelNames() ([]string, error) {
-	brs := []BlockReader{db.head}
-	for _, b := range db.Blocks() {
-		brs = append(brs, b)
-	}
-
-	labelNamesMap, err := labelNames(brs...)
-	if err != nil {
-		return nil, err
-	}
-
-	labelNames := make([]string, 0, len(labelNamesMap))
-	for name := range labelNamesMap {
-		labelNames = append(labelNames, name)
-	}
-	sort.Strings(labelNames)
-
-	return labelNames, nil
-}
-
 func isBlockDir(fi os.FileInfo) bool {
 	if !fi.IsDir() {
 		return false
diff --git a/go.mod b/go.mod
new file mode 100644
index 0000000..df0015d
--- /dev/null
+++ b/go.mod
@@ -0,0 +1,28 @@
+module github.com/prometheus/tsdb
+
+require (
+	github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc // indirect
+	github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect
+	github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 // indirect
+	github.com/cespare/xxhash v1.1.0
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954
+	github.com/go-kit/kit v0.8.0
+	github.com/go-logfmt/logfmt v0.3.0 // indirect
+	github.com/go-stack/stack v1.8.0 // indirect
+	github.com/gogo/protobuf v1.1.1 // indirect
+	github.com/golang/protobuf v1.2.0 // indirect
+	github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/oklog/ulid v1.3.1
+	github.com/pkg/errors v0.8.0
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v0.9.1
+	github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910
+	github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce // indirect
+	github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d // indirect
+	github.com/stretchr/testify v1.2.2 // indirect
+	golang.org/x/sync v0.0.0-20181108010431-42b317875d0f
+	golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8
+	gopkg.in/alecthomas/kingpin.v2 v2.2.6
+)
diff --git a/go.sum b/go.sum
new file mode 100644
index 0000000..266fbe9
--- /dev/null
+++ b/go.sum
@@ -0,0 +1,52 @@
+github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE=
+github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
+github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU=
+github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
+github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY=
+github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0=
+github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
+github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
+github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954 h1:RMLoZVzv4GliuWafOuPuQDKSm1SJph7uCRnnS61JAn4=
+github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
+github.com/go-kit/kit v0.8.0 h1:Wz+5lgoB0kkuqLEc6NVmwRknTKP6dTGbSqvhZtBI/j0=
+github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
+github.com/go-logfmt/logfmt v0.3.0 h1:8HUsc87TaSWLKwrnumgC8/YconD2fJQsRJAsWaPg2ic=
+github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
+github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
+github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo=
+github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
+github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515 h1:T+h1c/A9Gawja4Y9mFVWj2vyii2bbUNDw3kt9VxK2EY=
+github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
+github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
+github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
+github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
+github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
+github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
+github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v0.9.1 h1:K47Rk0v/fkEfwfQet2KWhscE0cJzjgCCDBG2KHZoVno=
+github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
+github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 h1:idejC8f05m9MGOsuEi1ATq9shN03HrxNkD/luQvxCv8=
+github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
+github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce h1:X0jFYGnHemYDIW6jlc+fSI8f9Cg+jqCnClYP2WgZT/A=
+github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
+github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d h1:GoAlyOgbOEIFdaDqxJVlbOQ1DtGmZWs/Qau0hIlk+WQ=
+github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
+github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+golang.org/x/sync v0.0.0-20181108010431-42b317875d0f h1:Bl/8QSvNqXvPGPGXa2z5xUTmV7VDcZyvRZ+QQXkXTZQ=
+golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8 h1:YoY1wS6JYVRpIfFngRf2HHo9R9dAne3xbkGOQ5rJXjU=
+golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
+gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
diff --git a/index/index.go b/index/index.go
index 133799a..880fbb1 100644
--- a/index/index.go
+++ b/index/index.go
@@ -544,13 +544,15 @@ type Reader struct {
 
 	// Cached hashmaps of section offsets.
 	labels   map[string]uint64
-	postings map[labels.Label]uint64
+	postings map[string]map[string]uint64
 	// Cache of read symbols. Strings that are returned when reading from the
 	// block are always backed by true strings held in here rather than
 	// strings that are backed by byte slices from the mmap'd index file. This
 	// prevents memory faults when applications work with read symbols after
-	// the block has been unmapped.
-	symbols map[uint32]string
+	// the block has been unmapped. The older format has sparse indexes so a map
+	// must be used, but the new format is not so we can use a slice.
+	symbols     map[uint32]string
+	symbolSlice []string
 
 	dec *Decoder
 
@@ -606,7 +608,7 @@ func newReader(b ByteSlice, c io.Closer) (*Reader, error) {
 		c:        c,
 		symbols:  map[uint32]string{},
 		labels:   map[string]uint64{},
-		postings: map[labels.Label]uint64{},
+		postings: map[string]map[string]uint64{},
 		crc32:    newCRC32(),
 	}
 
@@ -631,28 +633,42 @@ func newReader(b ByteSlice, c io.Closer) (*Reader, error) {
 	}
 	var err error
 
+	// Use the strings already allocated by symbols, rather than
+	// re-allocating them again below.
+	symbols := make(map[string]string, len(r.symbols)+len(r.symbolSlice))
+	for _, s := range r.symbols {
+		symbols[s] = s
+	}
+	for _, s := range r.symbolSlice {
+		symbols[s] = s
+	}
+
 	err = r.readOffsetTable(r.toc.labelIndicesTable, func(key []string, off uint64) error {
 		if len(key) != 1 {
 			return errors.Errorf("unexpected key length %d", len(key))
 		}
-		r.labels[key[0]] = off
+		r.labels[symbols[key[0]]] = off
 		return nil
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "read label index table")
 	}
+	r.postings[""] = map[string]uint64{}
 	err = r.readOffsetTable(r.toc.postingsTable, func(key []string, off uint64) error {
 		if len(key) != 2 {
 			return errors.Errorf("unexpected key length %d", len(key))
 		}
-		r.postings[labels.Label{Name: key[0], Value: key[1]}] = off
+		if _, ok := r.postings[key[0]]; !ok {
+			r.postings[symbols[key[0]]] = map[string]uint64{}
+		}
+		r.postings[key[0]][symbols[key[1]]] = off
 		return nil
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "read postings table")
 	}
 
-	r.dec = &Decoder{symbols: r.symbols}
+	r.dec = &Decoder{lookupSymbol: r.lookupSymbol}
 
 	return r, nil
 }
@@ -672,14 +688,16 @@ type Range struct {
 func (r *Reader) PostingsRanges() (map[labels.Label]Range, error) {
 	m := map[labels.Label]Range{}
 
-	for l, start := range r.postings {
-		d := r.decbufAt(int(start))
-		if d.err() != nil {
-			return nil, d.err()
-		}
-		m[l] = Range{
-			Start: int64(start) + 4,
-			End:   int64(start) + 4 + int64(d.len()),
+	for k, e := range r.postings {
+		for v, start := range e {
+			d := r.decbufAt(int(start))
+			if d.err() != nil {
+				return nil, d.err()
+			}
+			m[labels.Label{Name: k, Value: v}] = Range{
+				Start: int64(start) + 4,
+				End:   int64(start) + 4 + int64(d.len()),
+			}
 		}
 	}
 	return m, nil
@@ -777,18 +795,17 @@ func (r *Reader) readSymbols(off int) error {
 		basePos = uint32(off) + 4
 		nextPos = basePos + uint32(origLen-d.len())
 	)
-
 	if r.version == 2 {
-		nextPos = 0
+		r.symbolSlice = make([]string, 0, cnt)
 	}
 
 	for d.err() == nil && d.len() > 0 && cnt > 0 {
 		s := d.uvarintStr()
-		r.symbols[nextPos] = s
 
 		if r.version == 2 {
-			nextPos++
+			r.symbolSlice = append(r.symbolSlice, s)
 		} else {
+			r.symbols[nextPos] = s
 			nextPos = basePos + uint32(origLen-d.len())
 		}
 		cnt--
@@ -828,6 +845,9 @@ func (r *Reader) Close() error {
 }
 
 func (r *Reader) lookupSymbol(o uint32) (string, error) {
+	if int(o) < len(r.symbolSlice) {
+		return r.symbolSlice[o], nil
+	}
 	s, ok := r.symbols[o]
 	if !ok {
 		return "", errors.Errorf("unknown symbol offset %d", o)
@@ -842,12 +862,22 @@ func (r *Reader) Symbols() (map[string]struct{}, error) {
 	for _, s := range r.symbols {
 		res[s] = struct{}{}
 	}
+	for _, s := range r.symbolSlice {
+		res[s] = struct{}{}
+	}
 	return res, nil
 }
 
 // SymbolTable returns the symbol table that is used to resolve symbol references.
-func (r *Reader) SymbolTable() map[uint32]string {
-	return r.symbols
+func (r *Reader) SymbolTableSize() uint64 {
+	var size int
+	for _, s := range r.symbols {
+		size += len(s) + 8
+	}
+	for _, s := range r.symbolSlice {
+		size += len(s) + 8
+	}
+	return uint64(size)
 }
 
 // LabelValues returns value tuples that exist for the given label name tuples.
@@ -910,10 +940,11 @@ func (r *Reader) Series(id uint64, lbls *labels.Labels, chks *[]chunks.Meta) err
 
 // Postings returns a postings list for the given label pair.
 func (r *Reader) Postings(name, value string) (Postings, error) {
-	off, ok := r.postings[labels.Label{
-		Name:  name,
-		Value: value,
-	}]
+	e, ok := r.postings[name]
+	if !ok {
+		return EmptyPostings(), nil
+	}
+	off, ok := e[value]
 	if !ok {
 		return EmptyPostings(), nil
 	}
@@ -1031,21 +1062,7 @@ func (t *serializedStringTuples) At(i int) ([]string, error) {
 // It currently does not contain decoding methods for all entry types but can be extended
 // by them if there's demand.
 type Decoder struct {
-	symbols map[uint32]string
-}
-
-func (dec *Decoder) lookupSymbol(o uint32) (string, error) {
-	s, ok := dec.symbols[o]
-	if !ok {
-		return "", errors.Errorf("unknown symbol offset %d", o)
-	}
-	return s, nil
-}
-
-// SetSymbolTable set the symbol table to be used for lookups when decoding series
-// and label indices
-func (dec *Decoder) SetSymbolTable(t map[uint32]string) {
-	dec.symbols = t
+	lookupSymbol func(uint32) (string, error)
 }
 
 // Postings returns a postings list for b and its number of elements.
diff --git a/index/postings.go b/index/postings.go
index 53449f3..1f2ee0d 100644
--- a/index/postings.go
+++ b/index/postings.go
@@ -36,14 +36,14 @@ func AllPostingsKey() (name, value string) {
 // unordered batch fills on startup.
 type MemPostings struct {
 	mtx     sync.RWMutex
-	m       map[labels.Label][]uint64
+	m       map[string]map[string][]uint64
 	ordered bool
 }
 
 // NewMemPostings returns a memPostings that's ready for reads and writes.
 func NewMemPostings() *MemPostings {
 	return &MemPostings{
-		m:       make(map[labels.Label][]uint64, 512),
+		m:       make(map[string]map[string][]uint64, 512),
 		ordered: true,
 	}
 }
@@ -52,7 +52,7 @@ func NewMemPostings() *MemPostings {
 // until ensureOrder was called once.
 func NewUnorderedMemPostings() *MemPostings {
 	return &MemPostings{
-		m:       make(map[labels.Label][]uint64, 512),
+		m:       make(map[string]map[string][]uint64, 512),
 		ordered: false,
 	}
 }
@@ -62,8 +62,10 @@ func (p *MemPostings) SortedKeys() []labels.Label {
 	p.mtx.RLock()
 	keys := make([]labels.Label, 0, len(p.m))
 
-	for l := range p.m {
-		keys = append(keys, l)
+	for n, e := range p.m {
+		for v := range e {
+			keys = append(keys, labels.Label{Name: n, Value: v})
+		}
 	}
 	p.mtx.RUnlock()
 
@@ -78,14 +80,18 @@ func (p *MemPostings) SortedKeys() []labels.Label {
 
 // Get returns a postings list for the given label pair.
 func (p *MemPostings) Get(name, value string) Postings {
+	var lp []uint64
 	p.mtx.RLock()
-	l := p.m[labels.Label{Name: name, Value: value}]
+	l := p.m[name]
+	if l != nil {
+		lp = l[value]
+	}
 	p.mtx.RUnlock()
 
-	if l == nil {
+	if lp == nil {
 		return EmptyPostings()
 	}
-	return newListPostings(l)
+	return newListPostings(lp)
 }
 
 // All returns a postings list over all documents ever added.
@@ -118,8 +124,10 @@ func (p *MemPostings) EnsureOrder() {
 		}()
 	}
 
-	for _, l := range p.m {
-		workc <- l
+	for _, e := range p.m {
+		for _, l := range e {
+			workc <- l
+		}
 	}
 	close(workc)
 	wg.Wait()
@@ -129,44 +137,58 @@ func (p *MemPostings) EnsureOrder() {
 
 // Delete removes all ids in the given map from the postings lists.
 func (p *MemPostings) Delete(deleted map[uint64]struct{}) {
-	var keys []labels.Label
+	var keys, vals []string
 
 	// Collect all keys relevant for deletion once. New keys added afterwards
 	// can by definition not be affected by any of the given deletes.
 	p.mtx.RLock()
-	for l := range p.m {
-		keys = append(keys, l)
+	for n := range p.m {
+		keys = append(keys, n)
 	}
 	p.mtx.RUnlock()
 
-	// For each key we first analyse whether the postings list is affected by the deletes.
-	// If yes, we actually reallocate a new postings list.
-	for _, l := range keys {
-		// Only lock for processing one postings list so we don't block reads for too long.
-		p.mtx.Lock()
-
-		found := false
-		for _, id := range p.m[l] {
-			if _, ok := deleted[id]; ok {
-				found = true
-				break
-			}
-		}
-		if !found {
-			p.mtx.Unlock()
-			continue
+	for _, n := range keys {
+		p.mtx.RLock()
+		vals = vals[:0]
+		for v := range p.m[n] {
+			vals = append(vals, v)
 		}
-		repl := make([]uint64, 0, len(p.m[l]))
+		p.mtx.RUnlock()
+
+		// For each posting we first analyse whether the postings list is affected by the deletes.
+		// If yes, we actually reallocate a new postings list.
+		for _, l := range vals {
+			// Only lock for processing one postings list so we don't block reads for too long.
+			p.mtx.Lock()
+
+			found := false
+			for _, id := range p.m[n][l] {
+				if _, ok := deleted[id]; ok {
+					found = true
+					break
+				}
+			}
+			if !found {
+				p.mtx.Unlock()
+				continue
+			}
+			repl := make([]uint64, 0, len(p.m[n][l]))
 
-		for _, id := range p.m[l] {
-			if _, ok := deleted[id]; !ok {
-				repl = append(repl, id)
+			for _, id := range p.m[n][l] {
+				if _, ok := deleted[id]; !ok {
+					repl = append(repl, id)
+				}
 			}
+			if len(repl) > 0 {
+				p.m[n][l] = repl
+			} else {
+				delete(p.m[n], l)
+			}
+			p.mtx.Unlock()
 		}
-		if len(repl) > 0 {
-			p.m[l] = repl
-		} else {
-			delete(p.m, l)
+		p.mtx.Lock()
+		if len(p.m[n]) == 0 {
+			delete(p.m, n)
 		}
 		p.mtx.Unlock()
 	}
@@ -177,9 +199,11 @@ func (p *MemPostings) Iter(f func(labels.Label, Postings) error) error {
 	p.mtx.RLock()
 	defer p.mtx.RUnlock()
 
-	for l, p := range p.m {
-		if err := f(l, newListPostings(p)); err != nil {
-			return err
+	for n, e := range p.m {
+		for v, p := range e {
+			if err := f(labels.Label{Name: n, Value: v}, newListPostings(p)); err != nil {
+				return err
+			}
 		}
 	}
 	return nil
@@ -198,8 +222,13 @@ func (p *MemPostings) Add(id uint64, lset labels.Labels) {
 }
 
 func (p *MemPostings) addFor(id uint64, l labels.Label) {
-	list := append(p.m[l], id)
-	p.m[l] = list
+	nm, ok := p.m[l.Name]
+	if !ok {
+		nm = map[string][]uint64{}
+		p.m[l.Name] = nm
+	}
+	list := append(nm[l.Value], id)
+	nm[l.Value] = list
 
 	if !p.ordered {
 		return
diff --git a/querier.go b/querier.go
index e49ab9c..7459c6b 100644
--- a/querier.go
+++ b/querier.go
@@ -33,10 +33,14 @@ type Querier interface {
 
 	// LabelValues returns all potential values for a label name.
 	LabelValues(string) ([]string, error)
+
 	// LabelValuesFor returns all potential values for a label name.
 	// under the constraint of another label.
 	LabelValuesFor(string, labels.Label) ([]string, error)
 
+	// LabelNames returns all the unique label names present in the block in sorted order.
+	LabelNames() ([]string, error)
+
 	// Close releases the resources of the Querier.
 	Close() error
 }
@@ -60,6 +64,28 @@ func (q *querier) LabelValues(n string) ([]string, error) {
 	return q.lvals(q.blocks, n)
 }
 
+// LabelNames returns all the unique label names present querier blocks.
+func (q *querier) LabelNames() ([]string, error) {
+	labelNamesMap := make(map[string]struct{})
+	for _, b := range q.blocks {
+		names, err := b.LabelNames()
+		if err != nil {
+			return nil, errors.Wrap(err, "LabelNames() from Querier")
+		}
+		for _, name := range names {
+			labelNamesMap[name] = struct{}{}
+		}
+	}
+
+	labelNames := make([]string, 0, len(labelNamesMap))
+	for name := range labelNamesMap {
+		labelNames = append(labelNames, name)
+	}
+	sort.Strings(labelNames)
+
+	return labelNames, nil
+}
+
 func (q *querier) lvals(qs []Querier, n string) ([]string, error) {
 	if len(qs) == 0 {
 		return nil, nil
@@ -187,6 +213,10 @@ func (q *blockQuerier) LabelValues(name string) ([]string, error) {
 	return res, nil
 }
 
+func (q *blockQuerier) LabelNames() ([]string, error) {
+	return q.index.LabelNames()
+}
+
 func (q *blockQuerier) LabelValuesFor(string, labels.Label) ([]string, error) {
 	return nil, fmt.Errorf("not implemented")
 }
diff --git a/wal/wal.go b/wal/wal.go
index 1aae430..f1f13c7 100644
--- a/wal/wal.go
+++ b/wal/wal.go
@@ -231,7 +231,7 @@ func NewSize(logger log.Logger, reg prometheus.Registerer, dir string, segmentSi
 			return nil, err
 		}
 	} else {
-		if w.segment, err = OpenWriteSegment(w.logger, w.dir, j); err != nil {
+		if w.segment, err = OpenWriteSegment(logger, w.dir, j); err != nil {
 			return nil, err
 		}
 		// Correctly initialize donePages.
@@ -713,11 +713,12 @@ func (r *segmentBufReader) Read(b []byte) (n int, err error) {
 
 // Reader reads WAL records from an io.Reader.
 type Reader struct {
-	rdr   io.Reader
-	err   error
-	rec   []byte
-	buf   [pageSize]byte
-	total int64 // total bytes processed.
+	rdr       io.Reader
+	err       error
+	rec       []byte
+	buf       [pageSize]byte
+	total     int64   // Total bytes processed.
+	curRecTyp recType // Used for checking that the last record is not torn.
 }
 
 // NewReader returns a new reader.
@@ -730,6 +731,12 @@ func NewReader(r io.Reader) *Reader {
 func (r *Reader) Next() bool {
 	err := r.next()
 	if errors.Cause(err) == io.EOF {
+		// The last WAL segment record shouldn't be torn(should be full or last).
+		// The last record would be torn after a crash just before
+		// the last record part could be persisted to disk.
+		if recType(r.curRecTyp) == recFirst || recType(r.curRecTyp) == recMiddle {
+			r.err = errors.New("last record is torn")
+		}
 		return false
 	}
 	r.err = err
@@ -750,12 +757,13 @@ func (r *Reader) next() (err error) {
 			return errors.Wrap(err, "read first header byte")
 		}
 		r.total++
-		typ := recType(hdr[0])
+		r.curRecTyp = recType(hdr[0])
 
 		// Gobble up zero bytes.
-		if typ == recPageTerm {
-			// recPageTerm is a single byte that indicates that the rest of the page is padded.
-			// If it's the first byte in a page, buf is too small and we have to resize buf to fit pageSize-1 bytes.
+		if r.curRecTyp == recPageTerm {
+			// recPageTerm is a single byte that indicates the rest of the page is padded.
+			// If it's the first byte in a page, buf is too small and
+			// needs to be resized to fit pageSize-1 bytes.
 			buf = r.buf[1:]
 
 			// We are pedantic and check whether the zeros are actually up
@@ -806,7 +814,7 @@ func (r *Reader) next() (err error) {
 		}
 		r.rec = append(r.rec, buf[:length]...)
 
-		switch typ {
+		switch r.curRecTyp {
 		case recFull:
 			if i != 0 {
 				return errors.New("unexpected full record")
@@ -826,7 +834,7 @@ func (r *Reader) next() (err error) {
 			}
 			return nil
 		default:
-			return errors.Errorf("unexpected record type %d", typ)
+			return errors.Errorf("unexpected record type %d", r.curRecTyp)
 		}
 		// Only increment i for non-zero records since we use it
 		// to determine valid content record sequences.


Test patch:
diff --git a/compact_test.go b/compact_test.go
index acbbb51..4d99ce0 100644
--- a/compact_test.go
+++ b/compact_test.go
@@ -19,10 +19,13 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/go-kit/kit/log"
 	"github.com/pkg/errors"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/prometheus/tsdb/chunks"
+	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/testutil"
 )
 
@@ -682,3 +685,59 @@ func TestCompaction_populateBlock(t *testing.T) {
 		}
 	}
 }
+
+// TestDisableAutoCompactions checks that we can
+// disable and enable the auto compaction.
+// This is needed for unit tests that rely on
+// checking state before and after a compaction.
+func TestDisableAutoCompactions(t *testing.T) {
+	db, close := openTestDB(t, nil)
+	defer close()
+	defer db.Close()
+
+	blockRange := DefaultOptions.BlockRanges[0]
+	label := labels.FromStrings("foo", "bar")
+
+	// Trigger a compaction to check that it was skipped and
+	// no new blocks were created when compaction is disabled.
+	db.DisableCompactions()
+	app := db.Appender()
+	for i := int64(0); i < 3; i++ {
+		_, err := app.Add(label, i*blockRange, 0)
+		testutil.Ok(t, err)
+		_, err = app.Add(label, i*blockRange+1000, 0)
+		testutil.Ok(t, err)
+	}
+	testutil.Ok(t, app.Commit())
+
+	select {
+	case db.compactc <- struct{}{}:
+	default:
+	}
+
+	m := &dto.Metric{}
+	for x := 0; x < 10; x++ {
+		db.metrics.compactionsSkipped.Write(m)
+		if *m.Counter.Value > float64(0) {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	testutil.Assert(t, *m.Counter.Value > float64(0), "No compaction was skipped after the set timeout.")
+	testutil.Equals(t, 0, len(db.blocks))
+
+	// Enable the compaction, trigger it and check that the block is persisted.
+	db.EnableCompactions()
+	select {
+	case db.compactc <- struct{}{}:
+	default:
+	}
+	for x := 0; x < 10; x++ {
+		if len(db.Blocks()) > 0 {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	testutil.Assert(t, len(db.Blocks()) > 0, "No block was persisted after the set timeout.")
+}
diff --git a/db_test.go b/db_test.go
index 014a126..6fd59f3 100644
--- a/db_test.go
+++ b/db_test.go
@@ -27,6 +27,8 @@ import (
 
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	prom_testutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/tsdb/chunks"
 	"github.com/prometheus/tsdb/index"
 	"github.com/prometheus/tsdb/labels"
@@ -876,7 +878,7 @@ func (c *mockCompactorFailing) Write(dest string, b BlockReader, mint, maxt int6
 	return block.Meta().ULID, nil
 }
 
-func (*mockCompactorFailing) Compact(dest string, dirs ...string) (ulid.ULID, error) {
+func (*mockCompactorFailing) Compact(dest string, dirs []string, open []*Block) (ulid.ULID, error) {
 	return ulid.ULID{}, nil
 
 }
@@ -1295,11 +1297,15 @@ func TestInitializeHeadTimestamp(t *testing.T) {
 		testutil.Ok(t, err)
 		testutil.Ok(t, w.Close())
 
-		db, err := Open(dir, nil, nil, nil)
+		r := prometheus.NewRegistry()
+
+		db, err := Open(dir, nil, r, nil)
 		testutil.Ok(t, err)
 
 		testutil.Equals(t, int64(6000), db.head.MinTime())
 		testutil.Equals(t, int64(15000), db.head.MaxTime())
+		// Check that old series has been GCed.
+		testutil.Equals(t, 1.0, prom_testutil.ToFloat64(db.head.metrics.series))
 	})
 }
 
@@ -1388,12 +1394,15 @@ func TestDB_LabelNames(t *testing.T) {
 		}
 
 		// Addings more samples to head with new label names
-		// so that we can test db.LabelNames() (the union).
+		// so that we can test (head+disk).LabelNames() (the union).
 		appendSamples(db, 5, 9, tst.sampleLabels2)
 
 		// Testing DB (union).
-		labelNames, err = db.LabelNames()
+		q, err := db.Querier(math.MinInt64, math.MaxInt64)
+		testutil.Ok(t, err)
+		labelNames, err = q.LabelNames()
 		testutil.Ok(t, err)
+		testutil.Ok(t, q.Close())
 		testutil.Equals(t, tst.exp2, labelNames)
 	}
 }
diff --git a/index/index_test.go b/index/index_test.go
index d512240..f7a8156 100644
--- a/index/index_test.go
+++ b/index/index_test.go
@@ -320,10 +320,12 @@ func TestPersistence_index_e2e(t *testing.T) {
 	testutil.Ok(t, err)
 	mi.WritePostings("", "", newListPostings(all))
 
-	for l := range postings.m {
-		err = iw.WritePostings(l.Name, l.Value, postings.Get(l.Name, l.Value))
-		testutil.Ok(t, err)
-		mi.WritePostings(l.Name, l.Value, postings.Get(l.Name, l.Value))
+	for n, e := range postings.m {
+		for v := range e {
+			err = iw.WritePostings(n, v, postings.Get(n, v))
+			testutil.Ok(t, err)
+			mi.WritePostings(n, v, postings.Get(n, v))
+		}
 	}
 
 	err = iw.Close()
diff --git a/index/postings_test.go b/index/postings_test.go
index e7e2e61..53a9d95 100644
--- a/index/postings_test.go
+++ b/index/postings_test.go
@@ -20,21 +20,22 @@ import (
 	"sort"
 	"testing"
 
-	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/testutil"
 )
 
 func TestMemPostings_addFor(t *testing.T) {
 	p := NewMemPostings()
-	p.m[allPostingsKey] = []uint64{1, 2, 3, 4, 6, 7, 8}
+	p.m[allPostingsKey.Name] = map[string][]uint64{}
+	p.m[allPostingsKey.Name][allPostingsKey.Value] = []uint64{1, 2, 3, 4, 6, 7, 8}
 
 	p.addFor(5, allPostingsKey)
 
-	testutil.Equals(t, []uint64{1, 2, 3, 4, 5, 6, 7, 8}, p.m[allPostingsKey])
+	testutil.Equals(t, []uint64{1, 2, 3, 4, 5, 6, 7, 8}, p.m[allPostingsKey.Name][allPostingsKey.Value])
 }
 
 func TestMemPostings_ensureOrder(t *testing.T) {
 	p := NewUnorderedMemPostings()
+	p.m["a"] = map[string][]uint64{}
 
 	for i := 0; i < 100; i++ {
 		l := make([]uint64, 100)
@@ -43,17 +44,19 @@ func TestMemPostings_ensureOrder(t *testing.T) {
 		}
 		v := fmt.Sprintf("%d", i)
 
-		p.m[labels.Label{"a", v}] = l
+		p.m["a"][v] = l
 	}
 
 	p.EnsureOrder()
 
-	for _, l := range p.m {
-		ok := sort.SliceIsSorted(l, func(i, j int) bool {
-			return l[i] < l[j]
-		})
-		if !ok {
-			t.Fatalf("postings list %v is not sorted", l)
+	for _, e := range p.m {
+		for _, l := range e {
+			ok := sort.SliceIsSorted(l, func(i, j int) bool {
+				return l[i] < l[j]
+			})
+			if !ok {
+				t.Fatalf("postings list %v is not sorted", l)
+			}
 		}
 	}
 }
diff --git a/wal/wal_test.go b/wal/wal_test.go
index 29b7fb0..922126d 100644
--- a/wal/wal_test.go
+++ b/wal/wal_test.go
@@ -216,44 +216,84 @@ func TestWAL_FuzzWriteRead(t *testing.T) {
 }
 
 func TestWAL_Repair(t *testing.T) {
-	for name, cf := range map[string]func(f *os.File){
-		// Ensures that the page buffer is big enough to fit and entyre page size without panicing.
+
+	for name, test := range map[string]struct {
+		corrSgm    int              // Which segment to corrupt.
+		corrFunc   func(f *os.File) // Func that applies the corruption.
+		intactRecs int              // Total expected records left after the repair.
+	}{
+		"torn_last_record": {
+			2,
+			func(f *os.File) {
+				_, err := f.Seek(pageSize*2, 0)
+				testutil.Ok(t, err)
+				_, err = f.Write([]byte{byte(recFirst)})
+				testutil.Ok(t, err)
+			},
+			8,
+		},
+		// Ensures that the page buffer is big enough to fit
+		// an entire page size without panicing.
 		// https://github.com/prometheus/tsdb/pull/414
-		"bad_header": func(f *os.File) {
-			_, err := f.Seek(pageSize, 0)
-			testutil.Ok(t, err)
-			_, err = f.Write([]byte{byte(recPageTerm)})
-			testutil.Ok(t, err)
+		"bad_header": {
+			1,
+			func(f *os.File) {
+				_, err := f.Seek(pageSize, 0)
+				testutil.Ok(t, err)
+				_, err = f.Write([]byte{byte(recPageTerm)})
+				testutil.Ok(t, err)
+			},
+			4,
 		},
-		"bad_fragment_sequence": func(f *os.File) {
-			_, err := f.Seek(pageSize, 0)
-			testutil.Ok(t, err)
-			_, err = f.Write([]byte{byte(recLast)})
-			testutil.Ok(t, err)
+		"bad_fragment_sequence": {
+			1,
+			func(f *os.File) {
+				_, err := f.Seek(pageSize, 0)
+				testutil.Ok(t, err)
+				_, err = f.Write([]byte{byte(recLast)})
+				testutil.Ok(t, err)
+			},
+			4,
 		},
-		"bad_fragment_flag": func(f *os.File) {
-			_, err := f.Seek(pageSize, 0)
-			testutil.Ok(t, err)
-			_, err = f.Write([]byte{123})
-			testutil.Ok(t, err)
+		"bad_fragment_flag": {
+			1,
+			func(f *os.File) {
+				_, err := f.Seek(pageSize, 0)
+				testutil.Ok(t, err)
+				_, err = f.Write([]byte{123})
+				testutil.Ok(t, err)
+			},
+			4,
 		},
-		"bad_checksum": func(f *os.File) {
-			_, err := f.Seek(pageSize+4, 0)
-			testutil.Ok(t, err)
-			_, err = f.Write([]byte{0})
-			testutil.Ok(t, err)
+		"bad_checksum": {
+			1,
+			func(f *os.File) {
+				_, err := f.Seek(pageSize+4, 0)
+				testutil.Ok(t, err)
+				_, err = f.Write([]byte{0})
+				testutil.Ok(t, err)
+			},
+			4,
 		},
-		"bad_length": func(f *os.File) {
-			_, err := f.Seek(pageSize+2, 0)
-			testutil.Ok(t, err)
-			_, err = f.Write([]byte{0})
-			testutil.Ok(t, err)
+		"bad_length": {
+			1,
+			func(f *os.File) {
+				_, err := f.Seek(pageSize+2, 0)
+				testutil.Ok(t, err)
+				_, err = f.Write([]byte{0})
+				testutil.Ok(t, err)
+			},
+			4,
 		},
-		"bad_content": func(f *os.File) {
-			_, err := f.Seek(pageSize+100, 0)
-			testutil.Ok(t, err)
-			_, err = f.Write([]byte("beef"))
-			testutil.Ok(t, err)
+		"bad_content": {
+			1,
+			func(f *os.File) {
+				_, err := f.Seek(pageSize+100, 0)
+				testutil.Ok(t, err)
+				_, err = f.Write([]byte("beef"))
+				testutil.Ok(t, err)
+			},
+			4,
 		},
 	} {
 		t.Run(name, func(t *testing.T) {
@@ -261,10 +301,9 @@ func TestWAL_Repair(t *testing.T) {
 			testutil.Ok(t, err)
 			defer os.RemoveAll(dir)
 
-			// We create 3 segments with 3 records each and then corrupt the 2nd record
-			// of the 2nd segment.
-			// As a result we want a repaired WAL with the first 4 records intact.
-			intactRecords := 4
+			// We create 3 segments with 3 records each and
+			// then corrupt a given record in a given segment.
+			// As a result we want a repaired WAL with given intact records.
 			w, err := NewSize(nil, nil, dir, 3*pageSize)
 			testutil.Ok(t, err)
 
@@ -278,11 +317,11 @@ func TestWAL_Repair(t *testing.T) {
 			}
 			testutil.Ok(t, w.Close())
 
-			f, err := os.OpenFile(SegmentName(dir, 1), os.O_RDWR, 0666)
+			f, err := os.OpenFile(SegmentName(dir, test.corrSgm), os.O_RDWR, 0666)
 			testutil.Ok(t, err)
 
 			// Apply corruption function.
-			cf(f)
+			test.corrFunc(f)
 
 			testutil.Ok(t, f.Close())
 
@@ -315,7 +354,7 @@ func TestWAL_Repair(t *testing.T) {
 				result = append(result, append(b, r.Record()...))
 			}
 			testutil.Ok(t, r.Err())
-			testutil.Equals(t, intactRecords, len(result), "Wrong number of intact records")
+			testutil.Equals(t, test.intactRecs, len(result), "Wrong number of intact records")
 
 			for i, r := range result {
 				if !bytes.Equal(records[i], r) {