Vertical query merging and compaction

block_test.go

@@ -78,7 +78,7 @@ func createEmptyBlock(t *testing.T, dir string, meta *BlockMeta) *Block {
 }
 
 // createPopulatedBlock creates a block with nSeries series, and nSamples samples.
-func createPopulatedBlock(tb testing.TB, dir string, nSeries, nSamples int) *Block {
+func createPopulatedBlock(tb testing.TB, dir string, nSeries, nSamples int, startTime int64) *Block {
 	head, err := NewHead(nil, nil, nil, 2*60*60*1000)
 	testutil.Ok(tb, err)
 	defer head.Close()
@@ -89,15 +89,15 @@ func createPopulatedBlock(tb testing.TB, dir string, nSeries, nSamples int) *Blo
 
 	for n := 0; n < nSamples; n++ {
 		app := head.Appender()
-		ts := n * 1000
+		ts := startTime + int64(n*1000)
 		for i, lbl := range lbls {
 			if refs[i] != 0 {
-				err := app.AddFast(refs[i], int64(ts), rand.Float64())
+				err := app.AddFast(refs[i], ts, rand.Float64())
 				if err == nil {
 					continue
 				}
 			}
-			ref, err := app.Add(lbl, int64(ts), rand.Float64())
+			ref, err := app.Add(lbl, ts, rand.Float64())
 			testutil.Ok(tb, err)
 			refs[i] = ref
 		}

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,6 +198,38 @@ func (w *Writer) write(b []byte) error {
 	return err
 }
 
+// MergeOverlappingChunks removes the samples whose timestamp is overlapping.
+// The first appearing sample is retained in case there is overlapping.
+func MergeOverlappingChunks(chks []Meta) ([]Meta, error) {
+	if len(chks) < 2 {
+		return chks, nil
+	}
+	newChks := make([]Meta, 0, len(chks)) // 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) error {
 	// Calculate maximum space we need and cut a new segment in case
 	// we don't fit into the current one.