Repository: blevesearch/bleve
Pull request #1417 (2020-06-18 10:47:08)

MB-40007 -Obsolete segment file leaks during merge introductions
During merge segment introduction, if the segment contents
becomes totally obsolete, then the introducer skips
its introduction to the root. But it wasn't handling the clean
up ceremonies for the segment like Closing and cleaning
the entries from the ineligibleToRemove file list.


Patch:
diff --git a/index/scorch/event.go b/index/scorch/event.go
index dd79d6d0..8f3fc191 100644
--- a/index/scorch/event.go
+++ b/index/scorch/event.go
@@ -54,3 +54,11 @@ var EventKindBatchIntroductionStart = EventKind(5)
 
 // EventKindBatchIntroduction is fired when Batch() completes.
 var EventKindBatchIntroduction = EventKind(6)
+
+// EventKindMergeTaskIntroductionStart is fired when the merger is about to
+// start the introduction of merged segment from a single merge task.
+var EventKindMergeTaskIntroductionStart = EventKind(7)
+
+// EventKindMergeTaskIntroduction is fired when the merger has completed
+// the introduction of merged segment from a single merge task.
+var EventKindMergeTaskIntroduction = EventKind(8)
diff --git a/index/scorch/introducer.go b/index/scorch/introducer.go
index 64ca969b..7770c41c 100644
--- a/index/scorch/introducer.go
+++ b/index/scorch/introducer.go
@@ -383,6 +383,7 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 			}
 		}
 	}
+	var skipped bool
 	// In case where all the docs in the newly merged segment getting
 	// deleted by the time we reach here, can skip the introduction.
 	if nextMerge.new != nil &&
@@ -405,6 +406,9 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 			docsToPersistCount += nextMerge.new.Count() - newSegmentDeleted.GetCardinality()
 			memSegments++
 		}
+	} else {
+		skipped = true
+		atomic.AddUint64(&s.stats.TotFileMergeIntroductionsObsoleted, 1)
 	}
 
 	atomic.StoreUint64(&s.stats.TotItemsToPersist, docsToPersistCount)
@@ -429,8 +433,10 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 	}
 
 	// notify requester that we incorporated this
-	nextMerge.notify <- newSnapshot
-	close(nextMerge.notify)
+	nextMerge.notifyCh <- &mergeTaskIntroStatus{
+		indexSnapshot: newSnapshot,
+		skipped:       skipped}
+	close(nextMerge.notifyCh)
 }
 
 func isMemorySegment(s *SegmentSnapshot) bool {
diff --git a/index/scorch/merge.go b/index/scorch/merge.go
index 3eccee52..56c0953f 100644
--- a/index/scorch/merge.go
+++ b/index/scorch/merge.go
@@ -310,8 +310,9 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 
 		var oldNewDocNums map[uint64][]uint64
 		var seg segment.Segment
+		var filename string
 		if len(segmentsToMerge) > 0 {
-			filename := zapFileName(newSegmentID)
+			filename = zapFileName(newSegmentID)
 			s.markIneligibleForRemoval(filename)
 			path := s.path + string(os.PathSeparator) + filename
 
@@ -356,9 +357,11 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 			old:           oldMap,
 			oldNewDocNums: oldNewDocNums,
 			new:           seg,
-			notify:        make(chan *IndexSnapshot),
+			notifyCh:      make(chan *mergeTaskIntroStatus),
 		}
 
+		s.fireEvent(EventKindMergeTaskIntroductionStart, 0)
+
 		// give it to the introducer
 		select {
 		case <-s.closeCh:
@@ -371,18 +374,25 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 		introStartTime := time.Now()
 		// it is safe to blockingly wait for the merge introduction
 		// here as the introducer is bound to handle the notify channel.
-		newSnapshot := <-sm.notify
+		introStatus := <-sm.notifyCh
 		introTime := uint64(time.Since(introStartTime))
 		atomic.AddUint64(&s.stats.TotFileMergeZapIntroductionTime, introTime)
 		if atomic.LoadUint64(&s.stats.MaxFileMergeZapIntroductionTime) < introTime {
 			atomic.StoreUint64(&s.stats.MaxFileMergeZapIntroductionTime, introTime)
 		}
 		atomic.AddUint64(&s.stats.TotFileMergeIntroductionsDone, 1)
-		if newSnapshot != nil {
-			_ = newSnapshot.DecRef()
+		if introStatus != nil && introStatus.indexSnapshot != nil {
+			_ = introStatus.indexSnapshot.DecRef()
+			if introStatus.skipped {
+				// close the segment on skipping introduction.
+				s.unmarkIneligibleForRemoval(filename)
+				_ = seg.Close()
+			}
 		}
 
 		atomic.AddUint64(&s.stats.TotFileMergePlanTasksDone, 1)
+
+		s.fireEvent(EventKindMergeTaskIntroduction, 0)
 	}
 
 	// once all the newly merged segment introductions are done,
@@ -395,12 +405,17 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 	return nil
 }
 
+type mergeTaskIntroStatus struct {
+	indexSnapshot *IndexSnapshot
+	skipped       bool
+}
+
 type segmentMerge struct {
 	id            uint64
 	old           map[uint64]*SegmentSnapshot
 	oldNewDocNums map[uint64][]uint64
 	new           segment.Segment
-	notify        chan *IndexSnapshot
+	notifyCh      chan *mergeTaskIntroStatus
 }
 
 // perform a merging of the given SegmentBase instances into a new,
@@ -450,7 +465,7 @@ func (s *Scorch) mergeSegmentBases(snapshot *IndexSnapshot,
 		old:           make(map[uint64]*SegmentSnapshot),
 		oldNewDocNums: make(map[uint64][]uint64),
 		new:           seg,
-		notify:        make(chan *IndexSnapshot),
+		notifyCh:      make(chan *mergeTaskIntroStatus),
 	}
 
 	for i, idx := range sbsIndexes {
@@ -467,11 +482,20 @@ func (s *Scorch) mergeSegmentBases(snapshot *IndexSnapshot,
 	}
 
 	// blockingly wait for the introduction to complete
-	newSnapshot := <-sm.notify
-	if newSnapshot != nil {
+	var newSnapshot *IndexSnapshot
+	introStatus := <-sm.notifyCh
+	if introStatus != nil && introStatus.indexSnapshot != nil {
+		newSnapshot = introStatus.indexSnapshot
 		atomic.AddUint64(&s.stats.TotMemMergeSegments, uint64(len(sbs)))
 		atomic.AddUint64(&s.stats.TotMemMergeDone, 1)
+		if introStatus.skipped {
+			// close the segment on skipping introduction.
+			_ = newSnapshot.DecRef()
+			_ = seg.Close()
+			newSnapshot = nil
+		}
 	}
+
 	return newSnapshot, newSegmentID, nil
 }
 
diff --git a/index/scorch/snapshot_index_tfr.go b/index/scorch/snapshot_index_tfr.go
index 61537fc4..5d56f194 100644
--- a/index/scorch/snapshot_index_tfr.go
+++ b/index/scorch/snapshot_index_tfr.go
@@ -133,8 +133,6 @@ func (i *IndexSnapshotTermFieldReader) Advance(ID index.IndexInternalID, preAllo
 		if err != nil {
 			return nil, err
 		}
-		// close the current term field reader before replacing it with a new one
-		_ = i.Close()
 		*i = *(i2.(*IndexSnapshotTermFieldReader))
 	}
 	num, err := docInternalToNumber(ID)
diff --git a/index/scorch/stats.go b/index/scorch/stats.go
index 2900ac0d..9d7fbc0e 100644
--- a/index/scorch/stats.go
+++ b/index/scorch/stats.go
@@ -108,9 +108,10 @@ type Stats struct {
 	TotFileMergeZapIntroductionTime uint64
 	MaxFileMergeZapIntroductionTime uint64
 
-	TotFileMergeIntroductions        uint64
-	TotFileMergeIntroductionsDone    uint64
-	TotFileMergeIntroductionsSkipped uint64
+	TotFileMergeIntroductions          uint64
+	TotFileMergeIntroductionsDone      uint64
+	TotFileMergeIntroductionsSkipped   uint64
+	TotFileMergeIntroductionsObsoleted uint64
 
 	CurFilesIneligibleForRemoval     uint64
 	TotSnapshotsRemovedFromMetaStore uint64


Test patch:
diff --git a/index/scorch/merge_test.go b/index/scorch/merge_test.go
new file mode 100644
index 00000000..41f6892e
--- /dev/null
+++ b/index/scorch/merge_test.go
@@ -0,0 +1,154 @@
+//  Copyright (c) 2020 Couchbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// 		http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package scorch
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/blevesearch/bleve/document"
+	"github.com/blevesearch/bleve/index"
+)
+
+func TestObsoleteSegmentMergeIntroduction(t *testing.T) {
+	testConfig := CreateConfig("TestObsoleteSegmentMergeIntroduction")
+	err := InitTest(testConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		err := DestroyTest(testConfig)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	var introComplete, mergeIntroStart, mergeIntroComplete sync.WaitGroup
+	introComplete.Add(1)
+	mergeIntroStart.Add(1)
+	mergeIntroComplete.Add(1)
+	var segIntroCompleted int
+	RegistryEventCallbacks["test"] = func(e Event) {
+		if e.Kind == EventKindBatchIntroduction {
+			segIntroCompleted++
+			if segIntroCompleted == 3 {
+				// all 3 segments introduced
+				introComplete.Done()
+			}
+		} else if e.Kind == EventKindMergeTaskIntroductionStart {
+			// signal the start of merge task introduction so that
+			// we can introduce a new batch which obsoletes the
+			// merged segment's contents.
+			mergeIntroStart.Done()
+			// hold the merge task introduction until the merged segment contents
+			// are obsoleted with the next batch/segment introduction.
+			introComplete.Wait()
+		} else if e.Kind == EventKindMergeTaskIntroduction {
+			// signal the completion of the merge task introduction.
+			mergeIntroComplete.Done()
+
+		}
+	}
+
+	ourConfig := make(map[string]interface{}, len(testConfig))
+	for k, v := range testConfig {
+		ourConfig[k] = v
+	}
+	ourConfig["eventCallbackName"] = "test"
+
+	analysisQueue := index.NewAnalysisQueue(1)
+	idx, err := NewScorch(Name, ourConfig, analysisQueue)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = idx.Open()
+	if err != nil {
+		t.Fatalf("error opening index: %v", err)
+	}
+	defer func() {
+		err := idx.Close()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	// first introduce two documents over two batches.
+	batch := index.NewBatch()
+	doc := document.NewDocument("1")
+	doc.AddField(document.NewTextField("name", []uint64{}, []byte("test3")))
+	batch.Update(doc)
+	err = idx.Batch(batch)
+	if err != nil {
+		t.Error(err)
+	}
+
+	batch.Reset()
+	doc = document.NewDocument("2")
+	doc.AddField(document.NewTextField("name", []uint64{}, []byte("test2updated")))
+	batch.Update(doc)
+	err = idx.Batch(batch)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// wait until the merger trying to introduce the new merged segment.
+	mergeIntroStart.Wait()
+
+	// execute another batch which obsoletes the contents of the new merged
+	// segment awaiting introduction.
+	batch.Reset()
+	batch.Delete("1")
+	batch.Delete("2")
+	doc = document.NewDocument("3")
+	doc.AddField(document.NewTextField("name", []uint64{}, []byte("test3updated")))
+	batch.Update(doc)
+	err = idx.Batch(batch)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// wait until the merge task introduction complete.
+	mergeIntroComplete.Wait()
+
+	idxr, err := idx.Reader()
+	if err != nil {
+		t.Error(err)
+	}
+
+	numSegments := len(idxr.(*IndexSnapshot).segment)
+	if numSegments != 1 {
+		t.Errorf("expected one segment at the root, got: %d", numSegments)
+	}
+
+	skipIntroCount := atomic.LoadUint64(&idxr.(*IndexSnapshot).parent.stats.TotFileMergeIntroductionsObsoleted)
+	if skipIntroCount != 1 {
+		t.Errorf("expected one obsolete merge segment skipping the introduction, got: %d", skipIntroCount)
+	}
+
+	docCount, err := idxr.DocCount()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if docCount != 1 {
+		t.Errorf("Expected document count to be %d got %d", 1, docCount)
+	}
+
+	err = idxr.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/index/scorch/scorch_test.go b/index/scorch/scorch_test.go
index e22fafa2..5a9ef40c 100644
--- a/index/scorch/scorch_test.go
+++ b/index/scorch/scorch_test.go
@@ -2380,88 +2380,3 @@ func TestCancelIndexForceMerge(t *testing.T) {
 		t.Fatal(err)
 	}
 }
-
-func TestIndexSeekBackwardsStats(t *testing.T) {
-	cfg := CreateConfig("TestIndexOpenReopen")
-	err := InitTest(cfg)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err := DestroyTest(cfg)
-		if err != nil {
-			t.Log(err)
-		}
-	}()
-
-	analysisQueue := index.NewAnalysisQueue(1)
-	idx, err := NewScorch(Name, cfg, analysisQueue)
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = idx.Open()
-	if err != nil {
-		t.Errorf("error opening index: %v", err)
-	}
-	defer func() {
-		err := idx.Close()
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-
-	// insert a doc
-	doc := document.NewDocument("1")
-	doc.AddField(document.NewTextField("name", []uint64{}, []byte("cat")))
-	err = idx.Update(doc)
-	if err != nil {
-		t.Fatalf("error updating index: %v", err)
-	}
-
-	// insert another doc
-	doc = document.NewDocument("2")
-	doc.AddField(document.NewTextField("name", []uint64{}, []byte("cat")))
-	err = idx.Update(doc)
-	if err != nil {
-		t.Fatalf("error updating index: %v", err)
-	}
-
-	reader, err := idx.Reader()
-	if err != nil {
-		t.Fatalf("error getting index reader: %v", err)
-	}
-	defer reader.Close()
-
-	tfr, err := reader.TermFieldReader([]byte("cat"), "name", false, false, false)
-	if err != nil {
-		t.Fatalf("error getting term field readyer for name/cat: %v", err)
-	}
-
-	tfdFirst, err := tfr.Next(nil)
-	if err != nil {
-		t.Fatalf("error getting first tfd: %v", err)
-	}
-
-	_, err = tfr.Next(nil)
-	if err != nil {
-		t.Fatalf("error getting second tfd: %v", err)
-	}
-
-	// seek backwards to the first
-	_, err = tfr.Advance(tfdFirst.ID, nil)
-	if err != nil {
-		t.Fatalf("error adancing backwards: %v", err)
-	}
-
-	err = tfr.Close()
-	if err != nil {
-		t.Fatalf("error closing term field reader: %v", err)
-	}
-
-
-	if idx.(*Scorch).stats.TotTermSearchersStarted != idx.(*Scorch).stats.TotTermSearchersFinished {
-		t.Errorf("expected term searchers started %d to equal term searchers finished %d",
-			idx.(*Scorch).stats.TotTermSearchersStarted,
-			idx.(*Scorch).stats.TotTermSearchersFinished)
-	}
-}