From a9b28a6aa0ca5a420a8d8fb3f82d44aef7760144 Mon Sep 17 00:00:00 2001
From: Bartek Plotka <bwplotka@gmail.com>
Date: Wed, 28 Mar 2018 15:50:52 +0100
Subject: [PATCH] db: Added tests for validateBlockSequence to confirm a bug.

(That's why test fails)

Signed-off-by: Bartek Plotka <bwplotka@gmail.com>
---
 db.go      | 41 +++++++++++++++++++++++++++--------------
 db_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 71 insertions(+), 14 deletions(-)

diff --git a/db.go b/db.go
index 489f3a12e5..6326014795 100644
--- a/db.go
+++ b/db.go
@@ -26,7 +26,6 @@ import (
 	"strconv"
 	"sync"
 	"time"
-	"unsafe"
 
 	"golang.org/x/sync/errgroup"
 
@@ -560,14 +559,35 @@ func validateBlockSequence(bs []*Block) error {
 	if len(bs) == 0 {
 		return nil
 	}
-	sort.Slice(bs, func(i, j int) bool {
-		return bs[i].Meta().MinTime < bs[j].Meta().MinTime
+
+	var metas []BlockMeta
+	for _, b := range bs {
+		metas = append(metas, b.meta)
+	}
+
+	overlappedBlocks := ValidateBlockSequence(metas)
+	if len(overlappedBlocks) == 0 {
+		return nil
+	}
+
+	return errors.Errorf("block time ranges overlap (%v)", overlappedBlocks)
+}
+
+// ValidateBlockSequence returns error if given block meta files indicate that some blocks overlaps within sequence.
+func ValidateBlockSequence(bm []BlockMeta) [][]BlockMeta {
+	if len(bm) == 0 {
+		return nil
+	}
+	sort.Slice(bm, func(i, j int) bool {
+		return bm[i].MinTime < bm[j].MinTime
 	})
-	prev := bs[0]
-	for _, b := range bs[1:] {
-		if b.Meta().MinTime < prev.Meta().MaxTime {
-			return errors.Errorf("block time ranges overlap (%d, %d)", b.Meta().MinTime, prev.Meta().MaxTime)
+
+	prev := bm[0]
+	for _, b := range bm[1:] {
+		if b.MinTime < prev.MaxTime {
+			return [][]BlockMeta{{b, prev}}
 		}
+		//prev = b
 	}
 	return nil
 }
@@ -767,10 +787,6 @@ func intervalOverlap(amin, amax, bmin, bmax int64) bool {
 	return amin <= bmax && bmin <= amax
 }
 
-func intervalContains(min, max, t int64) bool {
-	return t >= min && t <= max
-}
-
 func isBlockDir(fi os.FileInfo) bool {
 	if !fi.IsDir() {
 		return false
@@ -869,9 +885,6 @@ func (es MultiError) Err() error {
 	return es
 }
 
-func yoloString(b []byte) string { return *((*string)(unsafe.Pointer(&b))) }
-func yoloBytes(s string) []byte  { return *((*[]byte)(unsafe.Pointer(&s))) }
-
 func closeAll(cs ...io.Closer) error {
 	var merr MultiError
 
diff --git a/db_test.go b/db_test.go
index 3fbbef74fa..3eb783c616 100644
--- a/db_test.go
+++ b/db_test.go
@@ -892,3 +892,47 @@ func expandSeriesSet(ss SeriesSet) ([]labels.Labels, error) {
 
 	return result, ss.Err()
 }
+
+func TestValidateBlockSequenceDetectsAllOverlaps(t *testing.T) {
+	var metas []BlockMeta
+
+	// Create 10 blocks that does not overlap (0-10, 10-20, ..., 90-100)
+	for i := 0; i < 10; i++ {
+		metas = append(metas, BlockMeta{MinTime: int64(i * 10), MaxTime: int64((i + 1) * 10)})
+	}
+
+	overlappedBlocks := ValidateBlockSequence(metas)
+	testutil.Assert(t, len(overlappedBlocks) == 0, "we found unexpected overlaps")
+
+	// Add overlaping blocks.
+
+	// o1 overlaps with 10-20.
+	o1 := BlockMeta{MinTime: 15, MaxTime: 17}
+
+	overlappedBlocks = ValidateBlockSequence(append(metas, o1))
+	expectedOverlaps := [][]BlockMeta{
+		{metas[1], o1},
+	}
+	testutil.Equals(t, expectedOverlaps, overlappedBlocks)
+
+	//// o2 overlaps with 20-30 and 30-40.
+	//o2 := BlockMeta{MinTime: 21, MaxTime: 31}
+	//
+	//// o3a and o3b overlaps with 30-40 and each other.
+	//o3a := BlockMeta{MinTime: 33, MaxTime: 39}
+	//o3b := BlockMeta{MinTime: 34, MaxTime: 36}
+	//
+	//// o4 is 1:1 overlap with 50-60
+	//o4 := BlockMeta{MinTime: 50, MaxTime: 60}
+	//
+	//// o5 overlaps with 50-60, 60-70 and 70,80
+	//o5 := BlockMeta{MinTime: 60, MaxTime: 80}
+	//
+
+	//expectedOverlaps := [][]block.Meta{
+	//	{metas[1], o1},
+	//	{metas[2], o2},
+	//	{metas[3], o2},
+	//	{metas[3], o3, o3b},
+	//}
+}
\ No newline at end of file