From debbcd7d47715c2dc78d5007fca7a5e8e63184f6 Mon Sep 17 00:00:00 2001
From: Sreekanth Sivasankaran <sreekanth.sivasankaran@couchbase.com>
Date: Tue, 13 Mar 2018 17:29:05 +0530
Subject: [PATCH] adding maxsegment size limit checks

---
 index/scorch/merge.go                     |  5 ++
 index/scorch/mergeplan/merge_plan.go      | 17 +++++++
 index/scorch/mergeplan/merge_plan_test.go | 58 +++++++++++++++++++++++
 3 files changed, 80 insertions(+)

diff --git a/index/scorch/merge.go b/index/scorch/merge.go
index ec2c8d4b..41086ad3 100644
--- a/index/scorch/merge.go
+++ b/index/scorch/merge.go
@@ -111,6 +111,11 @@ func (s *Scorch) parseMergePlannerOptions() (*mergeplan.MergePlanOptions,
 		if err != nil {
 			return &mergePlannerOptions, err
 		}
+
+		err = mergeplan.ValidateMergePlannerOptions(&mergePlannerOptions)
+		if err != nil {
+			return nil, err
+		}
 	}
 	return &mergePlannerOptions, nil
 }
diff --git a/index/scorch/mergeplan/merge_plan.go b/index/scorch/mergeplan/merge_plan.go
index 62f643f4..f0d6f162 100644
--- a/index/scorch/mergeplan/merge_plan.go
+++ b/index/scorch/mergeplan/merge_plan.go
@@ -18,6 +18,7 @@
 package mergeplan
 
 import (
+	"errors"
 	"fmt"
 	"math"
 	"sort"
@@ -115,6 +116,14 @@ func (o *MergePlanOptions) RaiseToFloorSegmentSize(s int64) int64 {
 	return o.FloorSegmentSize
 }
 
+// MaxSegmentSizeLimit represents the maximum size of a segment,
+// this limit comes as the roaring lib supports uint32.
+const MaxSegmentSizeLimit = 1<<32 - 1
+
+// ErrMaxSegmentSizeTooLarge is returned when the size of the segment
+// exceeds the MaxSegmentSizeLimit
+var ErrMaxSegmentSizeTooLarge = errors.New("MaxSegmentSize exceeds the size limit")
+
 // Suggested default options.
 var DefaultMergePlanOptions = MergePlanOptions{
 	MaxSegmentsPerTier:   10,
@@ -367,3 +376,11 @@ func ToBarChart(prefix string, barMax int, segments []Segment, plan *MergePlan)
 
 	return strings.Join(rv, "\n")
 }
+
+// ValidateMergePlannerOptions validates the merge planner options
+func ValidateMergePlannerOptions(options *MergePlanOptions) error {
+	if options.MaxSegmentSize > MaxSegmentSizeLimit {
+		return ErrMaxSegmentSizeTooLarge
+	}
+	return nil
+}
diff --git a/index/scorch/mergeplan/merge_plan_test.go b/index/scorch/mergeplan/merge_plan_test.go
index 419ab825..3adc1f4b 100644
--- a/index/scorch/mergeplan/merge_plan_test.go
+++ b/index/scorch/mergeplan/merge_plan_test.go
@@ -17,10 +17,12 @@ package mergeplan
 import (
 	"encoding/json"
 	"fmt"
+	"math/rand"
 	"os"
 	"reflect"
 	"sort"
 	"testing"
+	"time"
 )
 
 // Implements the Segment interface for testing,
@@ -401,6 +403,62 @@ func TestManySameSizedSegmentsWithDeletesBetweenMerges(t *testing.T) {
 	}
 }
 
+func TestValidateMergePlannerOptions(t *testing.T) {
+	o := &MergePlanOptions{
+		MaxSegmentSize:       1 << 32,
+		MaxSegmentsPerTier:   3,
+		TierGrowth:           3.0,
+		SegmentsPerMergeTask: 3,
+	}
+	err := ValidateMergePlannerOptions(o)
+	if err != ErrMaxSegmentSizeTooLarge {
+		t.Error("Validation expected to fail as the MaxSegmentSize exceeds limit")
+	}
+}
+
+func TestPlanMaxSegmentSizeLimit(t *testing.T) {
+	o := &MergePlanOptions{
+		MaxSegmentSize:       20,
+		MaxSegmentsPerTier:   5,
+		TierGrowth:           3.0,
+		SegmentsPerMergeTask: 5,
+		FloorSegmentSize:     5,
+	}
+	segments := makeLinearSegments(20)
+
+	s := rand.NewSource(time.Now().UnixNano())
+	r := rand.New(s)
+
+	max := 20
+	min := 5
+	randomInRange := func() int64 {
+		return int64(r.Intn(max-min) + min)
+	}
+	for i := 1; i < 20; i++ {
+		o.MaxSegmentSize = randomInRange()
+		plans, err := Plan(segments, o)
+		if err != nil {
+			t.Errorf("Plan failed, err: %v", err)
+		}
+		if len(plans.Tasks) == 0 {
+			t.Errorf("expected some plans with tasks")
+		}
+
+		for _, task := range plans.Tasks {
+			var totalLiveSize int64
+			for _, segs := range task.Segments {
+				totalLiveSize += segs.LiveSize()
+
+			}
+			if totalLiveSize >= o.MaxSegmentSize {
+				t.Errorf("merged segments size: %d exceeding the MaxSegmentSize"+
+					"limit: %d", totalLiveSize, o.MaxSegmentSize)
+			}
+		}
+	}
+
+}
+
 // ----------------------------------------
 
 type testCyclesSpec struct {