Refactor tombstone reader types

block.go

@@ -142,10 +142,9 @@ type Block struct {
 	dir  string
 	meta BlockMeta
 
-	chunkr ChunkReader
-	indexr IndexReader
-
-	tombstones tombstoneReader
+	chunkr     ChunkReader
+	indexr     IndexReader
+	tombstones TombstoneReader
 }
 
 // OpenBlock opens the block in the directory. It can be passed a chunk pool, which is used
@@ -293,7 +292,7 @@ func (pb *Block) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 	ir := pb.indexr
 
 	// Choose only valid postings which have chunks in the time-range.
-	stones := map[uint64]Intervals{}
+	stones := memTombstones{}
 
 	var lset labels.Labels
 	var chks []ChunkMeta
@@ -325,16 +324,21 @@ Outer:
 		return p.Err()
 	}
 
-	// Merge the current and new tombstones.
-	for k, v := range stones {
-		pb.tombstones.add(k, v[0])
+	err = pb.tombstones.Iter(func(id uint64, ivs Intervals) error {
+		for _, iv := range ivs {
+			stones.add(id, iv)
+			pb.meta.Stats.NumTombstones++
+		}
+		return nil
+	})
+	if err != nil {
+		return err
 	}
+	pb.tombstones = stones
 
 	if err := writeTombstoneFile(pb.dir, pb.tombstones); err != nil {
 		return err
 	}
-
-	pb.meta.Stats.NumTombstones = uint64(len(pb.tombstones))
 	return writeMetaFile(pb.dir, &pb.meta)
 }
 

compact.go

@@ -418,7 +418,7 @@ func (c *LeveledCompactor) write(dest string, meta *BlockMeta, blocks ...BlockRe
 	}
 
 	// Create an empty tombstones file.
-	if err := writeTombstoneFile(tmp, newEmptyTombstoneReader()); err != nil {
+	if err := writeTombstoneFile(tmp, EmptyTombstoneReader()); err != nil {
 		return errors.Wrap(err, "write new tombstones file")
 	}
 
@@ -631,7 +631,11 @@ func (c *compactionSeriesSet) Next() bool {
 	}
 	var err error
 
-	c.intervals = c.tombstones.Get(c.p.At())
+	c.intervals, err = c.tombstones.Get(c.p.At())
+	if err != nil {
+		c.err = errors.Wrap(err, "get tombstones")
+		return false
+	}
 
 	if err = c.index.Series(c.p.At(), &c.l, &c.c); err != nil {
 		c.err = errors.Wrapf(err, "get series %d", c.p.At())

head.go

@@ -66,7 +66,7 @@ type Head struct {
 
 	postings *memPostings // postings lists for terms
 
-	tombstones tombstoneReader
+	tombstones memTombstones
 }
 
 type headMetrics struct {
@@ -186,7 +186,7 @@ func NewHead(r prometheus.Registerer, l log.Logger, wal WAL, chunkRange int64) (
 		values:     map[string]stringset{},
 		symbols:    map[string]struct{}{},
 		postings:   newUnorderedMemPostings(),
-		tombstones: newEmptyTombstoneReader(),
+		tombstones: memTombstones{},
 	}
 	h.metrics = newHeadMetrics(h, r)
 

head_test.go

@@ -318,7 +318,7 @@ func TestHeadDeleteSimple(t *testing.T) {
 Outer:
 	for _, c := range cases {
 		// Reset the tombstones.
-		head.tombstones = newEmptyTombstoneReader()
+		head.tombstones = memTombstones{}
 
 		// Delete the ranges.
 		for _, r := range c.intervals {

querier.go

@@ -465,6 +465,7 @@ func (s *baseChunkSeries) Next() bool {
 	var (
 		lset   labels.Labels
 		chunks []ChunkMeta
+		err    error
 	)
 Outer:
 	for s.p.Next() {
@@ -487,7 +488,11 @@ Outer:
 
 		s.lset = lset
 		s.chks = chunks
-		s.intervals = s.tombstones.Get(s.p.At())
+		s.intervals, err = s.tombstones.Get(s.p.At())
+		if err != nil {
+			s.err = errors.Wrap(err, "get tombstones")
+			return false
+		}
 
 		if len(s.intervals) > 0 {
 			// Only those chunks that are not entirely deleted.

querier_test.go

@@ -454,7 +454,7 @@ Outer:
 		querier := &blockQuerier{
 			index:      ir,
 			chunks:     cr,
-			tombstones: newEmptyTombstoneReader(),
+			tombstones: EmptyTombstoneReader(),
 
 			mint: c.mint,
 			maxt: c.maxt,
@@ -506,7 +506,7 @@ func TestBlockQuerierDelete(t *testing.T) {
 			chunks [][]sample
 		}
 
-		tombstones tombstoneReader
+		tombstones TombstoneReader
 		queries    []query
 	}{
 		data: []struct {
@@ -554,13 +554,11 @@ func TestBlockQuerierDelete(t *testing.T) {
 				},
 			},
 		},
-		tombstones: newTombstoneReader(
-			map[uint64]Intervals{
-				1: Intervals{{1, 3}},
-				2: Intervals{{1, 3}, {6, 10}},
-				3: Intervals{{6, 10}},
-			},
-		),
+		tombstones: memTombstones{
+			1: Intervals{{1, 3}},
+			2: Intervals{{1, 3}, {6, 10}},
+			3: Intervals{{6, 10}},
+		},
 
 		queries: []query{
 			{
@@ -736,7 +734,7 @@ func TestBaseChunkSeries(t *testing.T) {
 		bcs := &baseChunkSeries{
 			p:          newListPostings(tc.postings),
 			index:      mi,
-			tombstones: newEmptyTombstoneReader(),
+			tombstones: EmptyTombstoneReader(),
 		}
 
 		i := 0

tombstones.go

@@ -35,12 +35,17 @@ const (
 
 // TombstoneReader gives access to tombstone intervals by series reference.
 type TombstoneReader interface {
-	Get(ref uint64) Intervals
+	// Get returns deletion intervals for the series with the given reference.
+	Get(ref uint64) (Intervals, error)
 
+	// Iter calls the given function for each encountered interval.
+	Iter(func(uint64, Intervals) error) error
+
+	// Close any underlying resources
 	Close() error
 }
 
-func writeTombstoneFile(dir string, tr tombstoneReader) error {
+func writeTombstoneFile(dir string, tr TombstoneReader) error {
 	path := filepath.Join(dir, tombstoneFilename)
 	tmp := path + ".tmp"
 	hash := newCRC32()
@@ -67,19 +72,21 @@ func writeTombstoneFile(dir string, tr tombstoneReader) error {
 
 	mw := io.MultiWriter(f, hash)
 
-	for k, v := range tr {
-		for _, itv := range v {
+	tr.Iter(func(ref uint64, ivs Intervals) error {
+		for _, iv := range ivs {
 			buf.reset()
-			buf.putUvarint64(k)
-			buf.putVarint64(itv.Mint)
-			buf.putVarint64(itv.Maxt)
+
+			buf.putUvarint64(ref)
+			buf.putVarint64(iv.Mint)
+			buf.putVarint64(iv.Maxt)
 
 			_, err = mw.Write(buf.get())
 			if err != nil {
 				return err
 			}
 		}
-	}
+		return nil
+	})
 
 	_, err = f.Write(hash.Sum(nil))
 	if err != nil {
@@ -100,7 +107,7 @@ type Stone struct {
 	intervals Intervals
 }
 
-func readTombstones(dir string) (tombstoneReader, error) {
+func readTombstones(dir string) (memTombstones, error) {
 	b, err := ioutil.ReadFile(filepath.Join(dir, tombstoneFilename))
 	if err != nil {
 		return nil, err
@@ -131,7 +138,8 @@ func readTombstones(dir string) (tombstoneReader, error) {
 		return nil, errors.New("checksum did not match")
 	}
 
-	stonesMap := newEmptyTombstoneReader()
+	stonesMap := memTombstones{}
+
 	for d.len() > 0 {
 		k := d.uvarint64()
 		mint := d.varint64()
@@ -143,28 +151,36 @@ func readTombstones(dir string) (tombstoneReader, error) {
 		stonesMap.add(k, Interval{mint, maxt})
 	}
 
-	return newTombstoneReader(stonesMap), nil
+	return stonesMap, nil
 }
 
-type tombstoneReader map[uint64]Intervals
+type memTombstones map[uint64]Intervals
 
-func newTombstoneReader(ts map[uint64]Intervals) tombstoneReader {
-	return tombstoneReader(ts)
+var emptyTombstoneReader = memTombstones{}
+
+// EmptyTombstoneReader returns a TombstoneReader that is always empty.
+func EmptyTombstoneReader() TombstoneReader {
+	return emptyTombstoneReader
 }
 
-func newEmptyTombstoneReader() tombstoneReader {
-	return tombstoneReader(make(map[uint64]Intervals))
+func (t memTombstones) Get(ref uint64) (Intervals, error) {
+	return t[ref], nil
 }
 
-func (t tombstoneReader) Get(ref uint64) Intervals {
-	return t[ref]
+func (t memTombstones) Iter(f func(uint64, Intervals) error) error {
+	for ref, ivs := range t {
+		if err := f(ref, ivs); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
-func (t tombstoneReader) add(ref uint64, itv Interval) {
+func (t memTombstones) add(ref uint64, itv Interval) {
 	t[ref] = t[ref].add(itv)
 }
 
-func (tombstoneReader) Close() error {
+func (memTombstones) Close() error {
 	return nil
 }
 

tombstones_test.go

@@ -29,7 +29,7 @@ func TestWriteAndReadbackTombStones(t *testing.T) {
 
 	ref := uint64(0)
 
-	stones := make(map[uint64]Intervals)
+	stones := memTombstones{}
 	// Generate the tombstones.
 	for i := 0; i < 100; i++ {
 		ref += uint64(rand.Int31n(10)) + 1
@@ -43,13 +43,13 @@ func TestWriteAndReadbackTombStones(t *testing.T) {
 		stones[ref] = dranges
 	}
 
-	require.NoError(t, writeTombstoneFile(tmpdir, newTombstoneReader(stones)))
+	require.NoError(t, writeTombstoneFile(tmpdir, stones))
 
 	restr, err := readTombstones(tmpdir)
 	require.NoError(t, err)
-	exptr := newTombstoneReader(stones)
+
 	// Compare the two readers.
-	require.Equal(t, exptr, restr)
+	require.Equal(t, stones, restr)
 }
 
 func TestAddingNewIntervals(t *testing.T) {