From f07d82994633f562f96e78d79c7ce5411a7de553 Mon Sep 17 00:00:00 2001
From: Bartek Plotka <bwplotka@gmail.com>
Date: Wed, 28 Mar 2018 23:50:42 +0100
Subject: [PATCH] db: Tiny tuning of algo + added proper print.

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

diff --git a/db.go b/db.go
index ea5b19c83..65ec1e275 100644
--- a/db.go
+++ b/db.go
@@ -29,6 +29,8 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
+	"strings"
+
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	"github.com/nightlyone/lockfile"
@@ -567,11 +569,11 @@ func validateBlockSequence(bs []*Block) error {
 	}
 
 	overlaps := OverlappingBlocks(metas)
-	if len(overlaps) == 0 {
-		return nil
+	if len(overlaps) > 0 {
+		return errors.Errorf("block time ranges overlap: %s", PrintOverlappedBlocks(overlaps))
 	}
 
-	return errors.Errorf("block time ranges overlap (%v)", overlaps)
+	return nil
 }
 
 // OverlappingBlocks returns all overlapping blocks from given meta files.
@@ -587,14 +589,17 @@ func OverlappingBlocks(bm []BlockMeta) (overlaps [][]BlockMeta) {
 		return bm[i].MinTime < bm[j].MinTime
 	})
 
-	pending := []BlockMeta{bm[0]}
+	var (
+		// pending contains not ended blocks in regards to "current" timestamp.
+		pending = []BlockMeta{bm[0]}
+		// Same pending helps to aggregate same overlaps to single group.
+		samePendings = true
+	)
 	for _, b := range bm[1:] {
-		var (
-			newPending   []BlockMeta
-			samePendings = true
-		)
+		var newPending []BlockMeta
 
 		for _, p := range pending {
+			// "b.MinTime" is our current time.
 			if b.MinTime >= p.MaxTime {
 				samePendings = false
 				continue
@@ -603,10 +608,11 @@ func OverlappingBlocks(bm []BlockMeta) (overlaps [][]BlockMeta) {
 			// "p" overlaps with "b" and "p" is still pending.
 			newPending = append(newPending, p)
 		}
+
 		// Our block "b" is now pending.
 		pending = append(newPending, b)
-
 		if len(newPending) == 0 {
+			// No overlaps.
 			continue
 		}
 
@@ -615,11 +621,30 @@ func OverlappingBlocks(bm []BlockMeta) (overlaps [][]BlockMeta) {
 			continue
 		}
 		overlaps = append(overlaps, append(newPending, b))
-
+		samePendings = true
 	}
 	return overlaps
 }
 
+// PrintOverlappedBlocks returns human readable string form of overlapped blocks.
+func PrintOverlappedBlocks(overlaps [][]BlockMeta) string {
+	var res []string
+	for _, o := range overlaps {
+		var groups []string
+		for _, m := range o {
+			groups = append(groups, fmt.Sprintf(
+				"[id: %s mint: %d maxt: %d range: %s]",
+				m.ULID.String(),
+				m.MinTime,
+				m.MaxTime,
+				(time.Duration((m.MaxTime-m.MinTime)/1000)*time.Second).String(),
+			))
+		}
+		res = append(res, fmt.Sprintf("<%s>", strings.Join(groups, "")))
+	}
+	return strings.Join(res, "")
+}
+
 func (db *DB) String() string {
 	return "HEAD"
 }
diff --git a/db_test.go b/db_test.go
index be0fba462..7f4c7bb5f 100644
--- a/db_test.go
+++ b/db_test.go
@@ -930,4 +930,14 @@ func TestOverlappingBlocksDetectsAllOverlaps(t *testing.T) {
 	o6a := BlockMeta{MinTime: 92, MaxTime: 105}
 	o6b := BlockMeta{MinTime: 94, MaxTime: 99}
 	testutil.Equals(t, [][]BlockMeta{{metas[9], o6a, o6b}, {o6a, metas[10]}}, OverlappingBlocks(append(metas, o6a, o6b)))
+
+	// All together.
+	testutil.Equals(t, [][]BlockMeta{
+		{metas[1], o1},
+		{metas[2], o2}, {o2, metas[3]},
+		{metas[3], o3a, o3b},
+		{metas[5], o4},
+		{metas[6], o5}, {o5, metas[7]}, {o5, metas[8]},
+		{metas[9], o6a, o6b}, {o6a, metas[10]},
+	}, OverlappingBlocks(append(metas, o1, o2, o3a, o3b, o4, o5, o6a, o6b)))
 }