Vertical query merging and compaction

block.go

@@ -100,9 +100,11 @@ type StringTuples interface {
 type ChunkWriter interface {
 	// WriteChunks writes several chunks. The Chunk field of the ChunkMetas
 	// must be populated.
+	// After successful write, it returns the actual chunks that it wrote.
+	// It might differ from original if there was overlapping between chunks.
 	// 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(chunks ...chunks.Meta) ([]chunks.Meta, error)
 
 	// Close writes any required finalization and closes the resources
 	// associated with the underlying writer.

chunkenc/xor.go

@@ -384,3 +384,42 @@ func (it *xorIterator) readValue() bool {
 	it.numRead++
 	return true
 }
+
+// MergeChunks vertically merges a and b, i.e., if there is any sample
+// with same timestamp in both a and b, the sample in a is discarded.
+func MergeChunks(a, b Chunk) (*XORChunk, error) {
+	newChunk := NewXORChunk()
+	app, err := newChunk.Appender()
+	if err != nil {
+		return nil, err
+	}
+	ait := a.Iterator()
+	bit := b.Iterator()
+	aok, bok := ait.Next(), bit.Next()
+	for aok && bok {
+		at, av := ait.At()
+		bt, bv := bit.At()
+		if at < bt {
+			app.Append(at, av)
+			aok = ait.Next()
+		} else if bt < at {
+			app.Append(bt, bv)
+			bok = bit.Next()
+		} else {
+			app.Append(bt, bv)
+			aok = ait.Next()
+			bok = bit.Next()
+		}
+	}
+	for aok {
+		at, av := ait.At()
+		app.Append(at, av)
+		aok = ait.Next()
+	}
+	for bok {
+		bt, bv := bit.At()
+		app.Append(bt, bv)
+		bok = bit.Next()
+	}
+	return newChunk, nil
+}

chunks/chunks.go

@@ -198,7 +198,44 @@ func (w *Writer) write(b []byte) error {
 	return err
 }
 
-func (w *Writer) WriteChunks(chks ...Meta) error {
+// mergeOverlappingChunks removes the samples whose timestamp is overlapping.
+// The first appearing sample is retained in case there is overlapping.
+func (w *Writer) mergeOverlappingChunks(chks []Meta) ([]Meta, error) {
+	if len(chks) < 2 {
+		return chks, nil
+	}
+	var newChks []Meta // Will contain the merged chunks.
+	newChks = append(newChks, chks[0])
+	last := 0
+	for _, c := range chks[1:] {
+		// We need to check only the last chunk in newChks.
+		// Reason: (1) newChks[last-1].MaxTime < newChks[last].MinTime (non overlapping)
+		//         (2) As chks are sorted w.r.t. MinTime, newChks[last].MinTime < c.MinTime.
+		// So never overlaps with newChks[last-1] or anything before that.
+		if c.MinTime > newChks[last].MaxTime {
+			newChks = append(newChks, c)
+			continue
+		}
+		nc := &newChks[last]
+		if c.MaxTime > nc.MaxTime {
+			nc.MaxTime = c.MaxTime
+		}
+		chk, err := chunkenc.MergeChunks(nc.Chunk, c.Chunk)
+		if err != nil {
+			return nil, err
+		}
+		nc.Chunk = chk
+	}
+
+	return newChks, nil
+}
+
+func (w *Writer) WriteChunks(chks ...Meta) ([]Meta, error) {
+	chks, err := w.mergeOverlappingChunks(chks)
+	if err != nil {
+		return nil, err
+	}
+
 	// 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.
@@ -210,7 +247,7 @@ 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 nil, err
 		}
 	}
 
@@ -226,26 +263,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 nil, err
 		}
 		b[0] = byte(chk.Chunk.Encoding())
 		if err := w.write(b[:1]); err != nil {
-			return err
+			return nil, err
 		}
 		if err := w.write(chk.Chunk.Bytes()); err != nil {
-			return err
+			return nil, err
 		}
 
 		w.crc32.Reset()
 		if err := chk.writeHash(w.crc32); err != nil {
-			return err
+			return nil, err
 		}
 		if err := w.write(w.crc32.Sum(b[:0])); err != nil {
-			return err
+			return nil, err
 		}
 	}
 
-	return nil
+	return chks, nil
 }
 
 func (w *Writer) seq() int {

compact.go

@@ -172,11 +172,20 @@ func (c *LeveledCompactor) plan(dms []dirMeta) ([]string, error) {
 		return dms[i].meta.MinTime < dms[j].meta.MinTime
 	})
 
+	var res []string
+	// Checking for overlapping blocks.
+	for _, dm := range c.selectOverlappingDirs(dms) {
+		res = append(res, dm.dir)
+	}
+	if len(res) > 0 {
+		return res, nil
+	}
+	// No overlapping blocks, do compaction the usual way.
+
 	// We do not include a recently created block with max(minTime), so the block which was just created from WAL.
 	// This gives users a window of a full block size to piece-wise backup new data without having to care about data overlap.
 	dms = dms[:len(dms)-1]
 
-	var res []string
 	for _, dm := range c.selectDirs(dms) {
 		res = append(res, dm.dir)
 	}
@@ -238,6 +247,33 @@ func (c *LeveledCompactor) selectDirs(ds []dirMeta) []dirMeta {
 	return nil
 }
 
+// selectOverlappingDirs returns one set of dir metas of the blocks whose ranges are overlapping,
+// that should be compacted into a single new block.
+func (c *LeveledCompactor) selectOverlappingDirs(ds []dirMeta) []dirMeta {
+	if len(ds) < 2 {
+		return nil
+	}
+	startBlock, endBlock := -1, -1
+	prevMaxTime := ds[0].meta.MaxTime
+	for i, d := range ds[1:] {
+		if d.meta.MinTime < prevMaxTime {
+			if startBlock < 0 {
+				startBlock = i
+			}
+			endBlock = i + 1
+		} else if startBlock >= 0 {
+			break
+		}
+		if d.meta.MaxTime > prevMaxTime {
+			prevMaxTime = d.meta.MaxTime
+		}
+	}
+	if startBlock < 0 {
+		return nil
+	}
+	return ds[startBlock : endBlock+1]
+}
+
 // splitByRange splits the directories by the time range. The range sequence starts at 0.
 //
 // For example, if we have blocks [0-10, 10-20, 50-60, 90-100] and the split range tr is 30
@@ -409,7 +445,7 @@ type instrumentedChunkWriter struct {
 	trange  prometheus.Histogram
 }
 
-func (w *instrumentedChunkWriter) WriteChunks(chunks ...chunks.Meta) error {
+func (w *instrumentedChunkWriter) WriteChunks(chunks ...chunks.Meta) ([]chunks.Meta, error) {
 	for _, c := range chunks {
 		w.size.Observe(float64(len(c.Chunk.Bytes())))
 		w.samples.Observe(float64(c.Chunk.NumSamples()))
@@ -583,6 +619,10 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
 
 	for set.Next() {
 		lset, chks, dranges := set.At() // The chunks here are not fully deleted.
+		// If blocks are overlapped, it is possible to have unsorted chunks.
+		sort.Slice(chks, func(i, j int) bool {
+			return chks[i].MinTime < chks[j].MinTime
+		})
 
 		// Skip the series with all deleted chunks.
 		if len(chks) == 0 {
@@ -611,7 +651,8 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
 				chks[i].Chunk = newChunk
 			}
 		}
-		if err := chunkw.WriteChunks(chks...); err != nil {
+		var err error
+		if chks, err = chunkw.WriteChunks(chks...); err != nil {
 			return errors.Wrap(err, "write chunks")
 		}
 

compact_test.go

@@ -302,6 +302,40 @@ func TestLeveledCompactor_plan(t *testing.T) {
 			},
 			expected: []string{"7", "8"},
 		},
+		// For overlapping blocks.
+		{
+			metas: []dirMeta{
+				metaRange("1", 0, 20, nil),
+				metaRange("2", 19, 40, nil),
+				metaRange("3", 40, 60, nil),
+			},
+			expected: []string{"1", "2"},
+		},
+		{
+			metas: []dirMeta{
+				metaRange("1", 0, 20, nil),
+				metaRange("2", 20, 40, nil),
+				metaRange("3", 30, 50, nil),
+			},
+			expected: []string{"2", "3"},
+		},
+		{
+			metas: []dirMeta{
+				metaRange("1", 0, 20, nil),
+				metaRange("2", 10, 40, nil),
+				metaRange("3", 30, 50, nil),
+			},
+			expected: []string{"1", "2", "3"},
+		},
+		{
+			metas: []dirMeta{
+				metaRange("5", 0, 360, nil),
+				metaRange("6", 340, 560, nil),
+				metaRange("7", 360, 420, nil),
+				metaRange("8", 420, 540, nil),
+			},
+			expected: []string{"5", "6", "7", "8"},
+		},
 	}
 
 	for _, c := range cases {

db.go

@@ -506,9 +506,6 @@ func (db *DB) reload() (err error) {
 	sort.Slice(blocks, func(i, j int) bool {
 		return blocks[i].Meta().MinTime < blocks[j].Meta().MinTime
 	})
-	if err := validateBlockSequence(blocks); err != nil {
-		return errors.Wrap(err, "invalid block sequence")
-	}
 
 	// Swap in new blocks first for subsequently created readers to be seen.
 	// Then close previous blocks, which may block for pending readers to complete.
@@ -543,25 +540,6 @@ func (db *DB) reload() (err error) {
 	return errors.Wrap(db.head.Truncate(maxt), "head truncate failed")
 }
 
-// validateBlockSequence returns error if given block meta files indicate that some blocks overlaps within sequence.
-func validateBlockSequence(bs []*Block) error {
-	if len(bs) <= 1 {
-		return nil
-	}
-
-	var metas []BlockMeta
-	for _, b := range bs {
-		metas = append(metas, b.meta)
-	}
-
-	overlaps := OverlappingBlocks(metas)
-	if len(overlaps) > 0 {
-		return errors.Errorf("block time ranges overlap: %s", overlaps)
-	}
-
-	return nil
-}
-
 // TimeRange specifies minTime and maxTime range.
 type TimeRange struct {
 	Min, Max int64