Attempt to reduce memory footprint of compaction

block.go

@@ -75,7 +75,9 @@ type IndexReader interface {
 	// The Postings here contain the offsets to the series inside the index.
 	// Found IDs are not strictly required to point to a valid Series, e.g. during
 	// background garbage collections.
-	Postings(name, value string) (index.Postings, error)
+	// 'reusePos' is the Postings object that can be re-used instead of allocating
+	// a new object. Re-use of this object need not be guaranteed.
+	Postings(name, value string, reusePos index.Postings) (index.Postings, error)
 
 	// SortedPostings returns a postings list that is reordered to be sorted
 	// by the label set of the underlying series.
@@ -109,9 +111,10 @@ type StringTuples interface {
 type ChunkWriter interface {
 	// WriteChunks writes several chunks. The Chunk field of the ChunkMetas
 	// must be populated.
+	// b is the byte buffer that can be used by WriteChunks.
 	// After returning successfully, the Ref fields in the ChunkMetas
 	// are set and can be used to retrieve the chunks from the written data.
-	WriteChunks(chunks ...chunks.Meta) error
+	WriteChunks(b []byte, chunks ...chunks.Meta) ([]byte, error)
 
 	// Close writes any required finalization and closes the resources
 	// associated with the underlying writer.
@@ -143,6 +146,9 @@ type BlockReader interface {
 
 	// MaxTime returns the max time of the block.
 	MaxTime() int64
+
+	// NumSeries returns the total number of series in the block.
+	NumSeries() uint64
 }
 
 // Appendable defines an entity to which data can be appended.
@@ -455,21 +461,29 @@ func (r blockIndexReader) LabelValues(names ...string) (index.StringTuples, erro
 	return st, errors.Wrapf(err, "block: %s", r.b.Meta().ULID)
 }
 
-func (r blockIndexReader) Postings(name, value string) (index.Postings, error) {
-	p, err := r.ir.Postings(name, value)
-	return p, errors.Wrapf(err, "block: %s", r.b.Meta().ULID)
+func (r blockIndexReader) Postings(name, value string, reusePos index.Postings) (index.Postings, error) {
+	p, err := r.ir.Postings(name, value, reusePos)
+	if err != nil {
+		// Checking for error before wrapping saves some allocs.
+		// This results in big savings of allocs if this function
+		// is called a lot of time, example during compaction.
+		return p, errors.Wrapf(err, "block: %s", r.b.Meta().ULID)
+	}
+	return p, nil
 }
 
 func (r blockIndexReader) SortedPostings(p index.Postings) index.Postings {
 	return r.ir.SortedPostings(p)
 }
 
 func (r blockIndexReader) Series(ref uint64, lset *labels.Labels, chks *[]chunks.Meta) error {
-	return errors.Wrapf(
-		r.ir.Series(ref, lset, chks),
-		"block: %s",
-		r.b.Meta().ULID,
-	)
+	if err := r.ir.Series(ref, lset, chks); err != nil {
+		// Checking for error before wrapping saves some allocs.
+		// This results in big savings of allocs if this function
+		// is called a lot of time, example during compaction.
+		return errors.Wrapf(err, "block: %s", r.b.Meta().ULID)
+	}
+	return nil
 }
 
 func (r blockIndexReader) LabelIndices() ([][]string, error) {
@@ -643,6 +657,11 @@ func (pb *Block) LabelNames() ([]string, error) {
 	return pb.indexr.LabelNames()
 }
 
+// NumSeries returns number of series in the block.
+func (pb *Block) NumSeries() uint64 {
+	return pb.meta.Stats.NumSeries
+}
+
 func clampInterval(a, b, mint, maxt int64) (int64, int64) {
 	if a < mint {
 		a = mint

chunks/chunks.go

@@ -55,8 +55,9 @@ type Meta struct {
 }
 
 // writeHash writes the chunk encoding and raw data into the provided hash.
-func (cm *Meta) writeHash(h hash.Hash) error {
-	if _, err := h.Write([]byte{byte(cm.Chunk.Encoding())}); err != nil {
+func (cm *Meta) writeHash(h hash.Hash, buf []byte) error {
+	buf = append(buf, byte(cm.Chunk.Encoding()))
+	if _, err := h.Write(buf[:1]); err != nil {
 		return err
 	}
 	if _, err := h.Write(cm.Chunk.Bytes()); err != nil {
@@ -280,7 +281,7 @@ func MergeChunks(a, b chunkenc.Chunk) (*chunkenc.XORChunk, error) {
 	return newChunk, nil
 }
 
-func (w *Writer) WriteChunks(chks ...Meta) error {
+func (w *Writer) WriteChunks(b []byte, chks ...Meta) ([]byte, error) {
 	// Calculate maximum space we need and cut a new segment in case
 	// we don't fit into the current one.
 	maxLen := int64(binary.MaxVarintLen32) // The number of chunks.
@@ -293,14 +294,14 @@ func (w *Writer) WriteChunks(chks ...Meta) error {
 
 	if w.wbuf == nil || w.n > w.segmentSize || newsz > w.segmentSize && maxLen <= w.segmentSize {
 		if err := w.cut(); err != nil {
-			return err
+			return b, err
 		}
 	}
 
-	var (
-		b   = [binary.MaxVarintLen32]byte{}
-		seq = uint64(w.seq()) << 32
-	)
+	var seq = uint64(w.seq()) << 32
+	for len(b) < binary.MaxVarintLen32 {
+		b = append(b, 0)
+	}
 	for i := range chks {
 		chk := &chks[i]
 
@@ -309,26 +310,26 @@ func (w *Writer) WriteChunks(chks ...Meta) error {
 		n := binary.PutUvarint(b[:], uint64(len(chk.Chunk.Bytes())))
 
 		if err := w.write(b[:n]); err != nil {
-			return err
+			return b, err
 		}
 		b[0] = byte(chk.Chunk.Encoding())
 		if err := w.write(b[:1]); err != nil {
-			return err
+			return b, err
 		}
 		if err := w.write(chk.Chunk.Bytes()); err != nil {
-			return err
+			return b, err
 		}
 
 		w.crc32.Reset()
-		if err := chk.writeHash(w.crc32); err != nil {
-			return err
+		if err := chk.writeHash(w.crc32, b[:0]); err != nil {
+			return b, err
 		}
 		if err := w.write(w.crc32.Sum(b[:0])); err != nil {
-			return err
+			return b, err
 		}
 	}
 
-	return nil
+	return b, nil
 }
 
 func (w *Writer) seq() int {

cmd/tsdb/main.go

@@ -34,6 +34,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/tsdb"
 	"github.com/prometheus/tsdb/chunks"
+	"github.com/prometheus/tsdb/index"
 	"github.com/prometheus/tsdb/labels"
 	"gopkg.in/alecthomas/kingpin.v2"
 )
@@ -456,7 +457,8 @@ func analyzeBlock(b *tsdb.Block, limit int) {
 	labelpairsUncovered := map[string]uint64{}
 	labelpairsCount := map[string]uint64{}
 	entries := 0
-	p, err := ir.Postings("", "") // The special all key.
+	apkName, apkValue := index.AllPostingsKey()
+	p, err := ir.Postings(apkName, apkValue, nil) // The special all key.
 	if err != nil {
 		exitWithError(err)
 	}
@@ -528,7 +530,7 @@ func analyzeBlock(b *tsdb.Block, limit int) {
 			exitWithError(err)
 		}
 		for _, n := range names {
-			postings, err := ir.Postings("__name__", n)
+			postings, err := ir.Postings("__name__", n, nil)
 			if err != nil {
 				exitWithError(err)
 			}