Repository: blevesearch/bleve
Pull request #1393 (2020-05-07 05:54:23)

Api to trigger a force merge operation on an index
Introducing a new synchronous ForceMerge api to enable
users conveniently invoke a manual merge operation
on an online scorch index. This api would let advanced
bleve users to compact their index all the way down to a
single file segment based index.

The api takes an optional mergePlanOptions argument which
would be applied for the forceful merge operations, while
the original mergePlanOptions property of the index remains
intact after the forceful merge cycle exits.
In the absense of a mergePlanOptions argument, default policy
assumed would be an aggressive one to target a single file
segment index with a maximum segment size of 1B approx.

The api would trigger the merger's work loop only once
during it's invocation and hence it is the caller's
responsibility to actively monitor the expected
file segments count at the root and trigger any
subsequent forceMerge invocations if needed to
meet the expected segment's count at root.

The caller could also pass a ctx for cancelling
an ongoing forceMerge operation in case of any
runtime anomalies during the merge operation.

The caller should also ensure that there is enough disk
space before invoking this compaction/merge operation
as it needs at least double the disk space
whenever there is a full compaction.


Patch:
diff --git a/builder.go b/builder.go
deleted file mode 100644
index de00c97b..00000000
--- a/builder.go
+++ /dev/null
@@ -1,94 +0,0 @@
-//  Copyright (c) 2019 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 bleve
-
-import (
-	"encoding/json"
-	"fmt"
-
-	"github.com/blevesearch/bleve/document"
-	"github.com/blevesearch/bleve/index"
-	"github.com/blevesearch/bleve/index/scorch"
-	"github.com/blevesearch/bleve/mapping"
-)
-
-type builderImpl struct {
-	b index.IndexBuilder
-	m mapping.IndexMapping
-}
-
-func (b *builderImpl) Index(id string, data interface{}) error {
-	if id == "" {
-		return ErrorEmptyID
-	}
-
-	doc := document.NewDocument(id)
-	err := b.m.MapDocument(doc, data)
-	if err != nil {
-		return err
-	}
-	err = b.b.Index(doc)
-	return err
-}
-
-func (b *builderImpl) Close() error {
-	return b.b.Close()
-}
-
-func newBuilder(path string, mapping mapping.IndexMapping, config map[string]interface{}) (Builder, error) {
-	if path == "" {
-		return nil, fmt.Errorf("builder requires path")
-	}
-
-	err := mapping.Validate()
-	if err != nil {
-		return nil, err
-	}
-
-	if config == nil {
-		config = map[string]interface{}{}
-	}
-
-	// the builder does not have an API to interact with internal storage
-	// however we can pass k/v pairs through the config
-	mappingBytes, err := json.Marshal(mapping)
-	if err != nil {
-		return nil, err
-	}
-	config["internal"] = map[string][]byte{
-		string(mappingInternalKey): mappingBytes,
-	}
-
-	// do not use real config, as these are options for the builder,
-	// not the resulting index
-	meta := newIndexMeta(scorch.Name, scorch.Name, map[string]interface{}{})
-	err = meta.Save(path)
-	if err != nil {
-		return nil, err
-	}
-
-	config["path"] = indexStorePath(path)
-
-	b, err := scorch.NewBuilder(config)
-	if err != nil {
-		return nil, err
-	}
-	rv := &builderImpl{
-		b: b,
-		m: mapping,
-	}
-
-	return rv, nil
-}
diff --git a/index.go b/index.go
index 974358b8..ef6ede93 100644
--- a/index.go
+++ b/index.go
@@ -293,17 +293,3 @@ func Open(path string) (Index, error) {
 func OpenUsing(path string, runtimeConfig map[string]interface{}) (Index, error) {
 	return openIndexUsing(path, runtimeConfig)
 }
-
-// Builder is a limited interface, used to build indexes in an offline mode.
-// Items cannot be updated or deleted, and the caller MUST ensure a document is
-// indexed only once.
-type Builder interface {
-	Index(id string, data interface{}) error
-	Close() error
-}
-
-// NewBuilder creates a builder, which will build an index at the specified path,
-// using the specified mapping and options.
-func NewBuilder(path string, mapping mapping.IndexMapping, config map[string]interface{}) (Builder, error) {
-	return newBuilder(path, mapping, config)
-}
diff --git a/index/index.go b/index/index.go
index 551f8de8..3e866f3a 100644
--- a/index/index.go
+++ b/index/index.go
@@ -367,10 +367,3 @@ type OptimizableContext interface {
 type DocValueReader interface {
 	VisitDocValues(id IndexInternalID, visitor DocumentFieldTermVisitor) error
 }
-
-// IndexBuilder is an interface supported by some index schemes
-// to allow direct write-only index building
-type IndexBuilder interface {
-	Index(doc *document.Document) error
-	Close() error
-}
diff --git a/index/scorch/builder.go b/index/scorch/builder.go
deleted file mode 100644
index cd400bd1..00000000
--- a/index/scorch/builder.go
+++ /dev/null
@@ -1,333 +0,0 @@
-//  Copyright (c) 2019 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 (
-	"fmt"
-	"io/ioutil"
-	"os"
-	"sync"
-
-	"github.com/RoaringBitmap/roaring"
-	"github.com/blevesearch/bleve/document"
-	"github.com/blevesearch/bleve/index"
-	"github.com/blevesearch/bleve/index/scorch/segment"
-	bolt "go.etcd.io/bbolt"
-)
-
-const DefaultBuilderBatchSize = 1000
-const DefaultBuilderMergeMax = 10
-
-type Builder struct {
-	m         sync.Mutex
-	segCount  uint64
-	path      string
-	buildPath string
-	segPaths  []string
-	batchSize int
-	mergeMax  int
-	batch     *index.Batch
-	internal  map[string][]byte
-	segPlugin segment.Plugin
-}
-
-func NewBuilder(config map[string]interface{}) (*Builder, error) {
-	path, ok := config["path"].(string)
-	if !ok {
-		return nil, fmt.Errorf("must specify path")
-	}
-
-	buildPathPrefix, _ := config["buildPathPrefix"].(string)
-	buildPath, err := ioutil.TempDir(buildPathPrefix, "scorch-offline-build")
-	if err != nil {
-		return nil, err
-	}
-
-	rv := &Builder{
-		path:      path,
-		buildPath: buildPath,
-		mergeMax:  DefaultBuilderMergeMax,
-		batchSize: DefaultBuilderBatchSize,
-		batch:     index.NewBatch(),
-		segPlugin: defaultSegmentPlugin,
-	}
-
-	err = rv.parseConfig(config)
-	if err != nil {
-		return nil, fmt.Errorf("error parsing builder config: %v", err)
-	}
-
-	return rv, nil
-}
-
-func (o *Builder) parseConfig(config map[string]interface{}) (err error) {
-	if v, ok := config["mergeMax"]; ok {
-		var t int
-		if t, err = parseToInteger(v); err != nil {
-			return fmt.Errorf("mergeMax parse err: %v", err)
-		}
-		if t > 0 {
-			o.mergeMax = t
-		}
-	}
-
-	if v, ok := config["batchSize"]; ok {
-		var t int
-		if t, err = parseToInteger(v); err != nil {
-			return fmt.Errorf("batchSize parse err: %v", err)
-		}
-		if t > 0 {
-			o.batchSize = t
-		}
-	}
-
-	if v, ok := config["internal"]; ok {
-		if vinternal, ok := v.(map[string][]byte); ok {
-			o.internal = vinternal
-		}
-	}
-
-	forcedSegmentType, forcedSegmentVersion, err := configForceSegmentTypeVersion(config)
-	if err != nil {
-		return err
-	}
-	if forcedSegmentType != "" && forcedSegmentVersion != 0 {
-		segPlugin, err := chooseSegmentPlugin(forcedSegmentType,
-			uint32(forcedSegmentVersion))
-		if err != nil {
-			o.segPlugin = segPlugin
-		}
-	}
-
-	return nil
-}
-
-// Index will place the document into the index.
-// It is invalid to index the same document multiple times.
-func (o *Builder) Index(doc *document.Document) error {
-	o.m.Lock()
-	defer o.m.Unlock()
-
-	o.batch.Update(doc)
-
-	return o.maybeFlushBatchLOCKED(o.batchSize)
-}
-
-func (o *Builder) maybeFlushBatchLOCKED(moreThan int) error {
-	if len(o.batch.IndexOps) >= moreThan {
-		defer o.batch.Reset()
-		return o.executeBatchLOCKED(o.batch)
-	}
-	return nil
-}
-
-func (o *Builder) executeBatchLOCKED(batch *index.Batch) (err error) {
-	analysisResults := make([]*index.AnalysisResult, 0, len(batch.IndexOps))
-	for _, doc := range batch.IndexOps {
-		if doc != nil {
-			// insert _id field
-			doc.AddField(document.NewTextFieldCustom("_id", nil, []byte(doc.ID), document.IndexField|document.StoreField, nil))
-			// perform analysis directly
-			analysisResult := analyze(doc)
-			analysisResults = append(analysisResults, analysisResult)
-		}
-	}
-
-	seg, _, err := o.segPlugin.New(analysisResults)
-	if err != nil {
-		return fmt.Errorf("error building segment base: %v", err)
-	}
-
-	filename := zapFileName(o.segCount)
-	o.segCount++
-	path := o.buildPath + string(os.PathSeparator) + filename
-
-	if segUnpersisted, ok := seg.(segment.UnpersistedSegment); ok {
-		err = segUnpersisted.Persist(path)
-		if err != nil {
-			return fmt.Errorf("error persisting segment base to %s: %v", path, err)
-		}
-
-		o.segPaths = append(o.segPaths, path)
-		return nil
-	}
-
-	return fmt.Errorf("new segment does not implement unpersisted: %T", seg)
-}
-
-func (o *Builder) doMerge() error {
-	// as long as we have more than 1 segment, keep merging
-	for len(o.segPaths) > 1 {
-
-		// merge the next <mergeMax> number of segments into one new one
-		// or, if there are fewer than <mergeMax> remaining, merge them all
-		mergeCount := o.mergeMax
-		if mergeCount > len(o.segPaths) {
-			mergeCount = len(o.segPaths)
-		}
-
-		mergePaths := o.segPaths[0:mergeCount]
-		o.segPaths = o.segPaths[mergeCount:]
-
-		// open each of the segments to be merged
-		mergeSegs := make([]segment.Segment, 0, mergeCount)
-
-		// closeOpenedSegs attempts to close all opened
-		// segments even if an error occurs, in which case
-		// the first error is returned
-		closeOpenedSegs := func() error {
-			var err error
-			for _, seg := range mergeSegs {
-				clErr := seg.Close()
-				if clErr != nil && err == nil {
-					err = clErr
-				}
-			}
-			return err
-		}
-
-		for _, mergePath := range mergePaths {
-			seg, err := o.segPlugin.Open(mergePath)
-			if err != nil {
-				_ = closeOpenedSegs()
-				return fmt.Errorf("error opening segment (%s) for merge: %v", mergePath, err)
-			}
-			mergeSegs = append(mergeSegs, seg)
-		}
-
-		// do the merge
-		mergedSegPath := o.buildPath + string(os.PathSeparator) + zapFileName(o.segCount)
-		drops := make([]*roaring.Bitmap, mergeCount)
-		_, _, err := o.segPlugin.Merge(mergeSegs, drops, mergedSegPath, nil, nil)
-		if err != nil {
-			_ = closeOpenedSegs()
-			return fmt.Errorf("error merging segments (%v): %v", mergePaths, err)
-		}
-		o.segCount++
-		o.segPaths = append(o.segPaths, mergedSegPath)
-
-		// close segments opened for merge
-		err = closeOpenedSegs()
-		if err != nil {
-			return fmt.Errorf("error closing opened segments: %v", err)
-		}
-
-		// remove merged segments
-		for _, mergePath := range mergePaths {
-			err = os.RemoveAll(mergePath)
-			if err != nil {
-				return fmt.Errorf("error removing segment %s after merge: %v", mergePath, err)
-			}
-		}
-	}
-
-	return nil
-}
-
-func (o *Builder) Close() error {
-	o.m.Lock()
-	defer o.m.Unlock()
-
-	// see if there is a partial batch
-	err := o.maybeFlushBatchLOCKED(1)
-	if err != nil {
-		return fmt.Errorf("error flushing batch before close: %v", err)
-	}
-
-	// perform all the merging
-	err = o.doMerge()
-	if err != nil {
-		return fmt.Errorf("error while merging: %v", err)
-	}
-
-	// ensure the store path exists
-	err = os.MkdirAll(o.path, 0700)
-	if err != nil {
-		return err
-	}
-
-	// move final segment into place
-	// segment id 2 is chosen to match the behavior of a scorch
-	// index which indexes a single batch of data
-	finalSegPath := o.path + string(os.PathSeparator) + zapFileName(2)
-	err = os.Rename(o.segPaths[0], finalSegPath)
-	if err != nil {
-		return fmt.Errorf("error moving final segment into place: %v", err)
-	}
-
-	// remove the buildPath, as it is no longer needed
-	err = os.RemoveAll(o.buildPath)
-	if err != nil {
-		return fmt.Errorf("error removing build path: %v", err)
-	}
-
-	// prepare wrapping
-	seg, err := o.segPlugin.Open(finalSegPath)
-	if err != nil {
-		return fmt.Errorf("error opening final segment")
-	}
-
-	// create a segment snapshot for this segment
-	ss := &SegmentSnapshot{
-		segment: seg,
-	}
-	is := &IndexSnapshot{
-		epoch:    3, // chosen to match scorch behavior when indexing a single batch
-		segment:  []*SegmentSnapshot{ss},
-		creator:  "scorch-builder",
-		internal: o.internal,
-	}
-
-	// create the root bolt
-	rootBoltPath := o.path + string(os.PathSeparator) + "root.bolt"
-	rootBolt, err := bolt.Open(rootBoltPath, 0600, nil)
-	if err != nil {
-		return err
-	}
-
-	// start a write transaction
-	tx, err := rootBolt.Begin(true)
-	if err != nil {
-		return err
-	}
-
-	// fill the root bolt with this fake index snapshot
-	_, _, err = prepareBoltSnapshot(is, tx, o.path, o.segPlugin)
-	if err != nil {
-		_ = tx.Rollback()
-		_ = rootBolt.Close()
-		return fmt.Errorf("error preparing bolt snapshot in root.bolt: %v", err)
-	}
-
-	// commit bolt data
-	err = tx.Commit()
-	if err != nil {
-		_ = rootBolt.Close()
-		return fmt.Errorf("error committing bolt tx in root.bolt: %v", err)
-	}
-
-	// close bolt
-	err = rootBolt.Close()
-	if err != nil {
-		return fmt.Errorf("error closing root.bolt: %v", err)
-	}
-
-	// close final segment
-	err = seg.Close()
-	if err != nil {
-		return fmt.Errorf("error closing final segment: %v", err)
-	}
-	return nil
-}
diff --git a/index/scorch/merge.go b/index/scorch/merge.go
index 37dca529..3eccee52 100644
--- a/index/scorch/merge.go
+++ b/index/scorch/merge.go
@@ -15,6 +15,7 @@
 package scorch
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -29,12 +30,16 @@ import (
 
 func (s *Scorch) mergerLoop() {
 	var lastEpochMergePlanned uint64
+	var ctrlMsg *mergerCtrl
 	mergePlannerOptions, err := s.parseMergePlannerOptions()
 	if err != nil {
 		s.fireAsyncError(fmt.Errorf("mergePlannerOption json parsing err: %v", err))
 		s.asyncTasks.Done()
 		return
 	}
+	ctrlMsgDflt := &mergerCtrl{ctx: context.Background(),
+		options: mergePlannerOptions,
+		doneCh:  nil}
 
 OUTER:
 	for {
@@ -53,16 +58,30 @@ OUTER:
 			atomic.StoreUint64(&s.iStats.mergeEpoch, ourSnapshot.epoch)
 			s.rootLock.Unlock()
 
-			if ourSnapshot.epoch != lastEpochMergePlanned {
+			if ctrlMsg == nil && ourSnapshot.epoch != lastEpochMergePlanned {
+				ctrlMsg = ctrlMsgDflt
+			}
+			if ctrlMsg != nil {
 				startTime := time.Now()
 
 				// lets get started
-				err := s.planMergeAtSnapshot(ourSnapshot, mergePlannerOptions)
+				err := s.planMergeAtSnapshot(ctrlMsg.ctx, ctrlMsg.options,
+					ourSnapshot)
 				if err != nil {
 					atomic.StoreUint64(&s.iStats.mergeEpoch, 0)
 					if err == segment.ErrClosed {
 						// index has been closed
 						_ = ourSnapshot.DecRef()
+
+						// continue the workloop on a user triggered cancel
+						if ctrlMsg.doneCh != nil {
+							close(ctrlMsg.doneCh)
+							ctrlMsg = nil
+							continue OUTER
+						}
+
+						// exit the workloop on index closure
+						ctrlMsg = nil
 						break OUTER
 					}
 					s.fireAsyncError(fmt.Errorf("merging err: %v", err))
@@ -70,6 +89,12 @@ OUTER:
 					atomic.AddUint64(&s.stats.TotFileMergeLoopErr, 1)
 					continue OUTER
 				}
+
+				if ctrlMsg.doneCh != nil {
+					close(ctrlMsg.doneCh)
+				}
+				ctrlMsg = nil
+
 				lastEpochMergePlanned = ourSnapshot.epoch
 
 				atomic.StoreUint64(&s.stats.LastMergedEpoch, ourSnapshot.epoch)
@@ -90,6 +115,8 @@ OUTER:
 			case <-s.closeCh:
 				break OUTER
 			case s.persisterNotifier <- ew:
+			case ctrlMsg = <-s.forceMergeRequestCh:
+				continue OUTER
 			}
 
 			// now wait for persister (but also detect close)
@@ -97,6 +124,7 @@ OUTER:
 			case <-s.closeCh:
 				break OUTER
 			case <-ew.notifyCh:
+			case ctrlMsg = <-s.forceMergeRequestCh:
 			}
 		}
 
@@ -106,6 +134,58 @@ OUTER:
 	s.asyncTasks.Done()
 }
 
+type mergerCtrl struct {
+	ctx     context.Context
+	options *mergeplan.MergePlanOptions
+	doneCh  chan struct{}
+}
+
+// ForceMerge helps users trigger a merge operation on
+// an online scorch index.
+func (s *Scorch) ForceMerge(ctx context.Context,
+	mo *mergeplan.MergePlanOptions) error {
+	// check whether force merge is already under processing
+	s.rootLock.Lock()
+	if s.stats.TotFileMergeForceOpsStarted >
+		s.stats.TotFileMergeForceOpsCompleted {
+		s.rootLock.Unlock()
+		return fmt.Errorf("force merge already in progress")
+	}
+
+	s.stats.TotFileMergeForceOpsStarted++
+	s.rootLock.Unlock()
+
+	if mo != nil {
+		err := mergeplan.ValidateMergePlannerOptions(mo)
+		if err != nil {
+			return err
+		}
+	} else {
+		// assume the default single segment merge policy
+		mo = &mergeplan.SingleSegmentMergePlanOptions
+	}
+	msg := &mergerCtrl{options: mo,
+		doneCh: make(chan struct{}),
+		ctx:    ctx,
+	}
+
+	// request the merger perform a force merge
+	select {
+	case s.forceMergeRequestCh <- msg:
+	case <-s.closeCh:
+		return nil
+	}
+
+	// wait for the force merge operation completion
+	select {
+	case <-msg.doneCh:
+		atomic.AddUint64(&s.stats.TotFileMergeForceOpsCompleted, 1)
+	case <-s.closeCh:
+	}
+
+	return nil
+}
+
 func (s *Scorch) parseMergePlannerOptions() (*mergeplan.MergePlanOptions,
 	error) {
 	mergePlannerOptions := mergeplan.DefaultMergePlanOptions
@@ -128,8 +208,39 @@ func (s *Scorch) parseMergePlannerOptions() (*mergeplan.MergePlanOptions,
 	return &mergePlannerOptions, nil
 }
 
-func (s *Scorch) planMergeAtSnapshot(ourSnapshot *IndexSnapshot,
-	options *mergeplan.MergePlanOptions) error {
+type closeChWrapper struct {
+	ch1     chan struct{}
+	ctx     context.Context
+	closeCh chan struct{}
+}
+
+func newCloseChWrapper(ch1 chan struct{},
+	ctx context.Context) *closeChWrapper {
+	return &closeChWrapper{ch1: ch1,
+		ctx:     ctx,
+		closeCh: make(chan struct{})}
+}
+
+func (w *closeChWrapper) close() {
+	select {
+	case <-w.closeCh:
+	default:
+		close(w.closeCh)
+	}
+}
+
+func (w *closeChWrapper) listen() {
+	select {
+	case <-w.ch1:
+		w.close()
+	case <-w.ctx.Done():
+		w.close()
+	case <-w.closeCh:
+	}
+}
+
+func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
+	options *mergeplan.MergePlanOptions, ourSnapshot *IndexSnapshot) error {
 	// build list of persisted segments in this snapshot
 	var onlyPersistedSnapshots []mergeplan.Segment
 	for _, segmentSnapshot := range ourSnapshot.segment {
@@ -158,6 +269,11 @@ func (s *Scorch) planMergeAtSnapshot(ourSnapshot *IndexSnapshot,
 	// process tasks in serial for now
 	var filenames []string
 
+	cw := newCloseChWrapper(s.closeCh, ctx)
+	defer cw.close()
+
+	go cw.listen()
+
 	for _, task := range resultMergePlan.Tasks {
 		if len(task.Segments) == 0 {
 			atomic.AddUint64(&s.stats.TotFileMergePlanTasksSegmentsEmpty, 1)
@@ -203,7 +319,7 @@ func (s *Scorch) planMergeAtSnapshot(ourSnapshot *IndexSnapshot,
 
 			atomic.AddUint64(&s.stats.TotFileMergeZapBeg, 1)
 			newDocNums, _, err := s.segPlugin.Merge(segmentsToMerge, docsToDrop, path,
-				s.closeCh, s)
+				cw.closeCh, s)
 			atomic.AddUint64(&s.stats.TotFileMergeZapEnd, 1)
 
 			fileMergeZapTime := uint64(time.Since(fileMergeZapStartTime))
diff --git a/index/scorch/mergeplan/merge_plan.go b/index/scorch/mergeplan/merge_plan.go
index c2a0d3c6..e02923cc 100644
--- a/index/scorch/mergeplan/merge_plan.go
+++ b/index/scorch/mergeplan/merge_plan.go
@@ -134,6 +134,17 @@ var DefaultMergePlanOptions = MergePlanOptions{
 	ReclaimDeletesWeight: 2.0,
 }
 
+// SingleSegmentMergePlanOptions helps in creating a
+// single segment index.
+var SingleSegmentMergePlanOptions = MergePlanOptions{
+	MaxSegmentsPerTier:   1,
+	MaxSegmentSize:       1 << 30,
+	TierGrowth:           1.0,
+	SegmentsPerMergeTask: 10,
+	FloorSegmentSize:     1 << 30,
+	ReclaimDeletesWeight: 2.0,
+}
+
 // -------------------------------------------
 
 func plan(segmentsIn []Segment, o *MergePlanOptions) (*MergePlan, error) {
diff --git a/index/scorch/persister.go b/index/scorch/persister.go
index ea3667bb..ffa65669 100644
--- a/index/scorch/persister.go
+++ b/index/scorch/persister.go
@@ -428,44 +428,55 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot) (
 	return true, nil
 }
 
-func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
-	segPlugin segment.Plugin) ([]string, map[uint64]string, error) {
+func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot) (err error) {
+	// start a write transaction
+	tx, err := s.rootBolt.Begin(true)
+	if err != nil {
+		return err
+	}
+	// defer rollback on error
+	defer func() {
+		if err != nil {
+			_ = tx.Rollback()
+		}
+	}()
+
 	snapshotsBucket, err := tx.CreateBucketIfNotExists(boltSnapshotsBucket)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 	newSnapshotKey := segment.EncodeUvarintAscending(nil, snapshot.epoch)
 	snapshotBucket, err := snapshotsBucket.CreateBucketIfNotExists(newSnapshotKey)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 
 	// persist meta values
 	metaBucket, err := snapshotBucket.CreateBucketIfNotExists(boltMetaDataKey)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
-	err = metaBucket.Put(boltMetaDataSegmentTypeKey, []byte(segPlugin.Type()))
+	err = metaBucket.Put(boltMetaDataSegmentTypeKey, []byte(s.segPlugin.Type()))
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 	buf := make([]byte, binary.MaxVarintLen32)
-	binary.BigEndian.PutUint32(buf, segPlugin.Version())
+	binary.BigEndian.PutUint32(buf, s.segPlugin.Version())
 	err = metaBucket.Put(boltMetaDataSegmentVersionKey, buf)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 
 	// persist internal values
 	internalBucket, err := snapshotBucket.CreateBucketIfNotExists(boltInternalKey)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 	// TODO optimize writing these in order?
 	for k, v := range snapshot.internal {
 		err = internalBucket.Put([]byte(k), v)
 		if err != nil {
-			return nil, nil, err
+			return err
 		}
 	}
 
@@ -477,69 +488,49 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
 		snapshotSegmentKey := segment.EncodeUvarintAscending(nil, segmentSnapshot.id)
 		snapshotSegmentBucket, err := snapshotBucket.CreateBucketIfNotExists(snapshotSegmentKey)
 		if err != nil {
-			return nil, nil, err
+			return err
 		}
 		switch seg := segmentSnapshot.segment.(type) {
 		case segment.PersistedSegment:
-			segPath := seg.Path()
-			filename := strings.TrimPrefix(segPath, path+string(os.PathSeparator))
+			path := seg.Path()
+			filename := strings.TrimPrefix(path, s.path+string(os.PathSeparator))
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
 			if err != nil {
-				return nil, nil, err
+				return err
 			}
 			filenames = append(filenames, filename)
 		case segment.UnpersistedSegment:
 			// need to persist this to disk
 			filename := zapFileName(segmentSnapshot.id)
-			path := path + string(os.PathSeparator) + filename
+			path := s.path + string(os.PathSeparator) + filename
 			err = seg.Persist(path)
 			if err != nil {
-				return nil, nil, fmt.Errorf("error persisting segment: %v", err)
+				return fmt.Errorf("error persisting segment: %v", err)
 			}
 			newSegmentPaths[segmentSnapshot.id] = path
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
 			if err != nil {
-				return nil, nil, err
+				return err
 			}
 			filenames = append(filenames, filename)
+
 		default:
-			return nil, nil, fmt.Errorf("unknown segment type: %T", seg)
+			return fmt.Errorf("unknown segment type: %T", seg)
 		}
 		// store current deleted bits
 		var roaringBuf bytes.Buffer
 		if segmentSnapshot.deleted != nil {
 			_, err = segmentSnapshot.deleted.WriteTo(&roaringBuf)
 			if err != nil {
-				return nil, nil, fmt.Errorf("error persisting roaring bytes: %v", err)
+				return fmt.Errorf("error persisting roaring bytes: %v", err)
 			}
 			err = snapshotSegmentBucket.Put(boltDeletedKey, roaringBuf.Bytes())
 			if err != nil {
-				return nil, nil, err
+				return err
 			}
 		}
 	}
 
-	return filenames, newSegmentPaths, nil
-}
-
-func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot) (err error) {
-	// start a write transaction
-	tx, err := s.rootBolt.Begin(true)
-	if err != nil {
-		return err
-	}
-	// defer rollback on error
-	defer func() {
-		if err != nil {
-			_ = tx.Rollback()
-		}
-	}()
-
-	filenames, newSegmentPaths, err := prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin)
-	if err != nil {
-		return err
-	}
-
 	// we need to swap in a new root only when we've persisted 1 or
 	// more segments -- whereby the new root would have 1-for-1
 	// replacements of in-memory segments with file-based segments
diff --git a/index/scorch/scorch.go b/index/scorch/scorch.go
index d1e1c6af..06a5e6b1 100644
--- a/index/scorch/scorch.go
+++ b/index/scorch/scorch.go
@@ -77,6 +77,8 @@ type Scorch struct {
 
 	pauseCount uint64
 
+	forceMergeRequestCh chan *mergerCtrl
+
 	segPlugin segment.Plugin
 }
 
@@ -101,14 +103,19 @@ func NewScorch(storeName string,
 		nextSnapshotEpoch:    1,
 		closeCh:              make(chan struct{}),
 		ineligibleForRemoval: map[string]bool{},
+		forceMergeRequestCh:  make(chan *mergerCtrl, 1),
 		segPlugin:            defaultSegmentPlugin,
 	}
 
-	forcedSegmentType, forcedSegmentVersion, err := configForceSegmentTypeVersion(config)
-	if err != nil {
-		return nil, err
-	}
-	if forcedSegmentType != "" && forcedSegmentVersion != 0 {
+	// check if the caller has requested a specific segment type/version
+	forcedSegmentVersion, ok := config["forceSegmentVersion"].(int)
+	if ok {
+		forcedSegmentType, ok2 := config["forceSegmentType"].(string)
+		if !ok2 {
+			return nil, fmt.Errorf(
+				"forceSegmentVersion set to %d, must also specify forceSegmentType", forcedSegmentVersion)
+		}
+
 		err := rv.loadSegmentPlugin(forcedSegmentType,
 			uint32(forcedSegmentVersion))
 		if err != nil {
@@ -136,22 +143,6 @@ func NewScorch(storeName string,
 	return rv, nil
 }
 
-// configForceSegmentTypeVersion checks if the caller has requested a
-// specific segment type/version
-func configForceSegmentTypeVersion(config map[string]interface{}) (string, uint32, error) {
-	forcedSegmentVersion, ok := config["forceSegmentVersion"].(int)
-	if ok {
-		forcedSegmentType, ok2 := config["forceSegmentType"].(string)
-		if !ok2 {
-			return "", 0, fmt.Errorf(
-				"forceSegmentVersion set to %d, must also specify forceSegmentType", forcedSegmentVersion)
-		}
-
-		return forcedSegmentType, uint32(forcedSegmentVersion), nil
-	}
-	return "", 0, nil
-}
-
 func (s *Scorch) paused() uint64 {
 	s.pauseLock.Lock()
 	pc := s.pauseCount
@@ -253,6 +244,7 @@ func (s *Scorch) openBolt() error {
 	s.introducerNotifier = make(chan *epochWatcher, 1)
 	s.persisterNotifier = make(chan *epochWatcher, 1)
 	s.closeCh = make(chan struct{})
+	s.forceMergeRequestCh = make(chan *mergerCtrl, 1)
 
 	if !s.readOnly && s.path != "" {
 		err := s.removeOldZapFiles() // Before persister or merger create any new files.
@@ -579,10 +571,6 @@ func (s *Scorch) StatsMap() map[string]interface{} {
 }
 
 func (s *Scorch) Analyze(d *document.Document) *index.AnalysisResult {
-	return analyze(d)
-}
-
-func analyze(d *document.Document) *index.AnalysisResult {
 	rv := &index.AnalysisResult{
 		Document: d,
 		Analyzed: make([]analysis.TokenFrequencies, len(d.Fields)+len(d.CompositeFields)),
diff --git a/index/scorch/segment_plugin.go b/index/scorch/segment_plugin.go
index 6dfa3b28..01eda7fb 100644
--- a/index/scorch/segment_plugin.go
+++ b/index/scorch/segment_plugin.go
@@ -60,28 +60,18 @@ func SupportedSegmentTypeVersions(typ string) (rv []uint32) {
 	return rv
 }
 
-func chooseSegmentPlugin(forcedSegmentType string,
-	forcedSegmentVersion uint32) (segment.Plugin, error) {
+func (s *Scorch) loadSegmentPlugin(forcedSegmentType string,
+	forcedSegmentVersion uint32) error {
 	if versions, ok := supportedSegmentPlugins[forcedSegmentType]; ok {
 		if segPlugin, ok := versions[uint32(forcedSegmentVersion)]; ok {
-			return segPlugin, nil
+			s.segPlugin = segPlugin
+			return nil
 		}
-		return nil, fmt.Errorf(
+		return fmt.Errorf(
 			"unsupported version %d for segment type: %s, supported: %v",
 			forcedSegmentVersion, forcedSegmentType,
 			SupportedSegmentTypeVersions(forcedSegmentType))
 	}
-	return nil, fmt.Errorf("unsupported segment type: %s, supported: %v",
+	return fmt.Errorf("unsupported segment type: %s, supported: %v",
 		forcedSegmentType, SupportedSegmentTypes())
 }
-
-func (s *Scorch) loadSegmentPlugin(forcedSegmentType string,
-	forcedSegmentVersion uint32) error {
-		segPlugin, err := chooseSegmentPlugin(forcedSegmentType,
-			forcedSegmentVersion)
-		if err != nil {
-			return err
-		}
-		s.segPlugin = segPlugin
-		return nil
-}
diff --git a/index/scorch/stats.go b/index/scorch/stats.go
index e638362a..2900ac0d 100644
--- a/index/scorch/stats.go
+++ b/index/scorch/stats.go
@@ -82,6 +82,9 @@ type Stats struct {
 	TotFileMergeLoopErr uint64
 	TotFileMergeLoopEnd uint64
 
+	TotFileMergeForceOpsStarted   uint64
+	TotFileMergeForceOpsCompleted uint64
+
 	TotFileMergePlan     uint64
 	TotFileMergePlanErr  uint64
 	TotFileMergePlanNone uint64
diff --git a/index_alias_impl.go b/index_alias_impl.go
index 4366fc79..bd5fabc8 100644
--- a/index_alias_impl.go
+++ b/index_alias_impl.go
@@ -44,6 +44,16 @@ func NewIndexAlias(indexes ...Index) *indexAliasImpl {
 	}
 }
 
+// VisitIndexes invokes the visit callback on every
+// indexes included in the index alias.
+func (i *indexAliasImpl) VisitIndexes(visit func(Index)) {
+	i.mutex.RLock()
+	for _, idx := range i.indexes {
+		visit(idx)
+	}
+	i.mutex.RUnlock()
+}
+
 func (i *indexAliasImpl) isAliasToSingleIndex() error {
 	if len(i.indexes) < 1 {
 		return ErrorAliasEmpty
diff --git a/mapping/document.go b/mapping/document.go
index 355a602e..15cb6b5f 100644
--- a/mapping/document.go
+++ b/mapping/document.go
@@ -251,6 +251,7 @@ func (dm *DocumentMapping) AddFieldMapping(fm *FieldMapping) {
 
 // UnmarshalJSON offers custom unmarshaling with optional strict validation
 func (dm *DocumentMapping) UnmarshalJSON(data []byte) error {
+
 	var tmp map[string]json.RawMessage
 	err := json.Unmarshal(data, &tmp)
 	if err != nil {
@@ -307,8 +308,8 @@ func (dm *DocumentMapping) UnmarshalJSON(data []byte) error {
 }
 
 func (dm *DocumentMapping) defaultAnalyzerName(path []string) string {
+	rv := ""
 	current := dm
-	rv := current.DefaultAnalyzer
 	for _, pathElement := range path {
 		var ok bool
 		current, ok = current.Properties[pathElement]


Test patch:
diff --git a/builder_test.go b/builder_test.go
deleted file mode 100644
index 119f39f5..00000000
--- a/builder_test.go
+++ /dev/null
@@ -1,89 +0,0 @@
-//  Copyright (c) 2019 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 bleve
-
-import (
-	"fmt"
-	"io/ioutil"
-	"os"
-	"testing"
-)
-
-func TestBuilder(t *testing.T) {
-	tmpDir, err := ioutil.TempDir("", "bleve-scorch-builder-test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(tmpDir)
-		if err != nil {
-			t.Fatalf("error cleaning up test index")
-		}
-	}()
-
-	conf := map[string]interface{}{
-		"batchSize": 2,
-		"mergeMax":  2,
-	}
-	b, err := NewBuilder(tmpDir, NewIndexMapping(), conf)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	for i := 0; i < 10; i++ {
-		doc := map[string]interface{}{
-			"name": "hello",
-		}
-		err = b.Index(fmt.Sprintf("%d", i), doc)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	err = b.Close()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	idx, err := Open(tmpDir)
-	if err != nil {
-		t.Fatalf("error opening index: %v", err)
-	}
-	defer func() {
-		err = idx.Close()
-		if err != nil {
-			t.Errorf("error closing index: %v", err)
-		}
-	}()
-
-	docCount, err := idx.DocCount()
-	if err != nil {
-		t.Errorf("error checking doc count: %v", err)
-	}
-	if docCount != 10 {
-		t.Errorf("expected doc count to be 10, got %d", docCount)
-	}
-
-	q := NewTermQuery("hello")
-	q.SetField("name")
-	req := NewSearchRequest(q)
-	res, err := idx.Search(req)
-	if err != nil {
-		t.Errorf("error searching index: %v", err)
-	}
-	if res.Total != 10 {
-		t.Errorf("expected 10 search hits, got %d", res.Total)
-	}
-}
diff --git a/index/scorch/builder_test.go b/index/scorch/builder_test.go
deleted file mode 100644
index 1d0ce716..00000000
--- a/index/scorch/builder_test.go
+++ /dev/null
@@ -1,160 +0,0 @@
-//  Copyright (c) 2019 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 (
-	"fmt"
-	"io/ioutil"
-	"os"
-	"testing"
-
-	"github.com/blevesearch/bleve/document"
-	"github.com/blevesearch/bleve/index"
-)
-
-func TestBuilder(t *testing.T) {
-	tmpDir, err := ioutil.TempDir("", "scorch-builder-test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(tmpDir)
-		if err != nil {
-			t.Fatalf("error cleaning up test index: %v", err)
-		}
-	}()
-	options := map[string]interface{}{
-		"path":      tmpDir,
-		"batchSize": 2,
-		"mergeMax":  2,
-	}
-	b, err := NewBuilder(options)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	for i := 0; i < 10; i++ {
-		doc := document.NewDocument(fmt.Sprintf("%d", i))
-		doc.AddField(document.NewTextField("name", nil, []byte("hello")))
-		err = b.Index(doc)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	err = b.Close()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	checkIndex(t, tmpDir, []byte("hello"), "name", 10)
-
-}
-
-func checkIndex(t *testing.T, path string, term []byte, field string, expectCount int) {
-	cfg := make(map[string]interface{})
-	cfg["path"] = path
-	analysisQueue := index.NewAnalysisQueue(1)
-	idx, err := NewScorch(Name, cfg, 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.Fatalf("error closing index: %v", err)
-		}
-	}()
-
-	r, err := idx.Reader()
-	if err != nil {
-		t.Fatalf("error accessing index reader: %v", err)
-	}
-	defer func() {
-		err = r.Close()
-		if err != nil {
-			t.Fatalf("error closing reader: %v", err)
-		}
-	}()
-
-	// check the count, expect 10 docs
-	count, err := r.DocCount()
-	if err != nil {
-		t.Errorf("error accessing index doc count: %v", err)
-	} else if count != uint64(expectCount) {
-		t.Errorf("expected %d docs, got %d", expectCount, count)
-	}
-
-	// run a search for hello
-	tfr, err := r.TermFieldReader(term, field, false, false, false)
-	if err != nil {
-		t.Errorf("error accessing term field reader: %v", err)
-	} else {
-		var rows int
-		tfd, err := tfr.Next(nil)
-		for err == nil && tfd != nil {
-			rows++
-			tfd, err = tfr.Next(nil)
-		}
-		if err != nil {
-			t.Errorf("error calling next on term field reader: %v", err)
-		}
-		if rows != expectCount {
-			t.Errorf("expected %d rows for term hello, field name, got %d", expectCount, rows)
-		}
-	}
-}
-
-func TestBuilderFlushFinalBatch(t *testing.T) {
-	tmpDir, err := ioutil.TempDir("", "scorch-builder-test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(tmpDir)
-		if err != nil {
-			t.Fatalf("error cleaning up test index: %v", err)
-		}
-	}()
-	options := map[string]interface{}{
-		"path":      tmpDir,
-		"batchSize": 2,
-		"mergeMax":  2,
-	}
-	b, err := NewBuilder(options)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	for i := 0; i < 9; i++ {
-		doc := document.NewDocument(fmt.Sprintf("%d", i))
-		doc.AddField(document.NewTextField("name", nil, []byte("hello")))
-		err = b.Index(doc)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	err = b.Close()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	checkIndex(t, tmpDir, []byte("hello"), "name", 9)
-}
diff --git a/index/scorch/scorch_test.go b/index/scorch/scorch_test.go
index b168728e..5a9ef40c 100644
--- a/index/scorch/scorch_test.go
+++ b/index/scorch/scorch_test.go
@@ -15,6 +15,7 @@
 package scorch
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"math/rand"
@@ -23,6 +24,7 @@ import (
 	"regexp"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -32,6 +34,7 @@ import (
 	regexpTokenizer "github.com/blevesearch/bleve/analysis/tokenizer/regexp"
 	"github.com/blevesearch/bleve/document"
 	"github.com/blevesearch/bleve/index"
+	"github.com/blevesearch/bleve/index/scorch/mergeplan"
 	"github.com/blevesearch/bleve/mapping"
 )
 
@@ -2149,4 +2152,231 @@ func TestForceVersion(t *testing.T) {
 	if err == nil {
 		t.Fatalf("expected an error opening an unsupported vesion, got nil")
 	}
-}
\ No newline at end of file
+}
+
+func TestIndexForceMerge(t *testing.T) {
+	cfg := CreateConfig("TestIndexForceMerge")
+	err := InitTest(cfg)
+	defer func() {
+		err := DestroyTest(cfg)
+		if err != nil {
+			t.Log(err)
+		}
+	}()
+	tmp := struct {
+		MaxSegmentsPerTier   int   `json:"maxSegmentsPerTier"`
+		SegmentsPerMergeTask int   `json:"segmentsPerMergeTask"`
+		FloorSegmentSize     int64 `json:"floorSegmentSize"`
+	}{
+		int(1),
+		int(1),
+		int64(2),
+	}
+	cfg["scorchMergePlanOptions"] = &tmp
+
+	analysisQueue := index.NewAnalysisQueue(1)
+	idx, err := NewScorch(Name, cfg, analysisQueue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = idx.Open()
+	if err != nil {
+		t.Fatalf("error opening index: %v", err)
+	}
+	var expectedCount uint64
+	batch := index.NewBatch()
+	for i := 0; i < 10; i++ {
+		doc := document.NewDocument(fmt.Sprintf("doc1-%d", i))
+		doc.AddField(document.NewTextField("name", []uint64{}, []byte(fmt.Sprintf("text1-%d", i))))
+		batch.Update(doc)
+		doc = document.NewDocument(fmt.Sprintf("doc2-%d", i))
+		doc.AddField(document.NewTextField("name", []uint64{}, []byte(fmt.Sprintf("text2-%d", i))))
+		batch.Update(doc)
+		err = idx.Batch(batch)
+		if err != nil {
+			t.Error(err)
+		}
+		batch.Reset()
+		expectedCount += 2
+	}
+
+	// verify doc count
+	indexReader, err := idx.Reader()
+	if err != nil {
+		t.Error(err)
+	}
+	docCount, err := indexReader.DocCount()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if docCount != expectedCount {
+		t.Errorf("Expected document count to be %d got %d", expectedCount, docCount)
+	}
+	err = indexReader.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+	var si *Scorch
+	var ok bool
+	if si, ok = idx.(*Scorch); !ok {
+		t.Errorf("expects a scorch index")
+	}
+
+	nfs := atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot)
+	if nfs != 10 {
+		t.Errorf("expected 10 root file segments, got: %d", nfs)
+	}
+
+	ctx := context.Background()
+	for {
+		if atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot) == 1 {
+			break
+		}
+		err := si.ForceMerge(ctx, &mergeplan.MergePlanOptions{
+			MaxSegmentsPerTier:   1,
+			MaxSegmentSize:       10000,
+			SegmentsPerMergeTask: 10,
+			FloorSegmentSize:     10000})
+		if err != nil {
+			t.Errorf("ForceMerge failed, err: %v", err)
+		}
+	}
+
+	// verify the final root segment count
+	if atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot) != 1 {
+		t.Errorf("expected a single root file segments, got: %d",
+			atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot))
+	}
+
+	// verify with an invalid merge plan
+	err = si.ForceMerge(ctx, &mergeplan.MergePlanOptions{
+		MaxSegmentsPerTier:   1,
+		MaxSegmentSize:       1 << 33,
+		SegmentsPerMergeTask: 10,
+		FloorSegmentSize:     10000})
+	if err != mergeplan.ErrMaxSegmentSizeTooLarge {
+		t.Errorf("ForceMerge expected to fail with ErrMaxSegmentSizeTooLarge")
+	}
+
+	err = idx.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCancelIndexForceMerge(t *testing.T) {
+	cfg := CreateConfig("TestCancelIndexForceMerge")
+	err := InitTest(cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		err := DestroyTest(cfg)
+		if err != nil {
+			t.Log(err)
+		}
+	}()
+
+	tmp := struct {
+		MaxSegmentsPerTier   int   `json:"maxSegmentsPerTier"`
+		SegmentsPerMergeTask int   `json:"segmentsPerMergeTask"`
+		FloorSegmentSize     int64 `json:"floorSegmentSize"`
+	}{
+		int(1),
+		int(1),
+		int64(2),
+	}
+	cfg["scorchMergePlanOptions"] = &tmp
+
+	analysisQueue := index.NewAnalysisQueue(1)
+	idx, err := NewScorch(Name, cfg, analysisQueue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = idx.Open()
+	if err != nil {
+		t.Fatalf("error opening index: %v", err)
+	}
+
+	var expectedCount uint64
+	batch := index.NewBatch()
+	for i := 0; i < 20; i++ {
+		doc := document.NewDocument(fmt.Sprintf("doc1-%d", i))
+		doc.AddField(document.NewTextField("name", []uint64{}, []byte(fmt.Sprintf("text1-%d", i))))
+		batch.Update(doc)
+		doc = document.NewDocument(fmt.Sprintf("doc2-%d", i))
+		doc.AddField(document.NewTextField("name", []uint64{}, []byte(fmt.Sprintf("text2-%d", i))))
+		batch.Update(doc)
+		err = idx.Batch(batch)
+		if err != nil {
+			t.Error(err)
+		}
+		batch.Reset()
+		expectedCount += 2
+	}
+
+	// verify doc count
+	indexReader, err := idx.Reader()
+	if err != nil {
+		t.Error(err)
+	}
+	docCount, err := indexReader.DocCount()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if docCount != expectedCount {
+		t.Errorf("Expected document count to be %d got %d", expectedCount, docCount)
+	}
+	err = indexReader.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var si *Scorch
+	var ok bool
+	if si, ok = idx.(*Scorch); !ok {
+		t.Fatal("expects a scorch index")
+	}
+
+	// no merge operations are expected as per the original merge policy.
+	nfsr := atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot)
+	if nfsr != 20 {
+		t.Errorf("expected 20 root file segments, got: %d", nfsr)
+	}
+
+	ctx := context.Background()
+	ctx, cancel := context.WithCancel(ctx)
+
+	// cancel the force merge operation once the root has some new merge
+	// introductions. ie if the root has lesser file segments than earlier.
+	go func() {
+		for {
+			nval := atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot)
+			if nval < nfsr {
+				cancel()
+				return
+			}
+			time.Sleep(time.Millisecond * 5)
+		}
+	}()
+
+	err = si.ForceMerge(ctx, &mergeplan.MergePlanOptions{
+		MaxSegmentsPerTier:   1,
+		MaxSegmentSize:       10000,
+		SegmentsPerMergeTask: 5,
+		FloorSegmentSize:     10000})
+	if err != nil {
+		t.Errorf("ForceMerge failed, err: %v", err)
+	}
+
+	// verify the final root file segment count or forceMerge completion
+	if atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot) == 1 {
+		t.Errorf("expected many files at root, but got: %d segments",
+			atomic.LoadUint64(&si.stats.TotFileSegmentsAtRoot))
+	}
+
+	err = idx.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/mapping/mapping_test.go b/mapping/mapping_test.go
index 2b687096..b57283f3 100644
--- a/mapping/mapping_test.go
+++ b/mapping/mapping_test.go
@@ -1111,6 +1111,7 @@ func TestClosestDocDynamicMapping(t *testing.T) {
 }
 
 func TestMappingPointerToTimeBug1152(t *testing.T) {
+
 	when, err := time.Parse(time.RFC3339, "2019-03-06T15:04:05Z")
 	if err != nil {
 		t.Fatal(err)
@@ -1140,14 +1141,3 @@ func TestMappingPointerToTimeBug1152(t *testing.T) {
 		t.Fatalf("expected field to be type *document.DateTimeField, got %T", doc.Fields[0])
 	}
 }
-
-func TestDefaultAnalyzerInheritance(t *testing.T) {
-	docMapping := NewDocumentMapping()
-	docMapping.DefaultAnalyzer = "xyz"
-	childMapping := NewTextFieldMapping()
-	docMapping.AddFieldMappingsAt("field", childMapping)
-
-	if analyzer := docMapping.defaultAnalyzerName([]string{"field"}); analyzer != "xyz" {
-		t.Fatalf("Expected analyzer: xyz to be inherited by field, but got: '%v'", analyzer)
-	}
-}
diff --git a/search_test.go b/search_test.go
index 98623e1a..755999db 100644
--- a/search_test.go
+++ b/search_test.go
@@ -1693,119 +1693,3 @@ func TestSearchHighlightingWithRegexpReplacement(t *testing.T) {
 		t.Fatalf("Expected 1 hit, got: %v", sres.Total)
 	}
 }
-
-func TestAnalyzerInheritance(t *testing.T) {
-	tests := []struct {
-		name       string
-		mappingStr string
-		doc        map[string]interface{}
-		queryField string
-		queryTerm  string
-	}{
-		{
-			/*
-				index_mapping: keyword
-				default_mapping: ""
-					-> child field (should inherit keyword)
-			*/
-			name: "Child field to inherit index mapping's default analyzer",
-			mappingStr: `{"default_mapping":{"enabled":true,"dynamic":false,"properties":` +
-				`{"city":{"enabled":true,"dynamic":false,"fields":[{"name":"city","type":"text",` +
-				`"store":false,"index":true}]}}},"default_analyzer":"keyword"}`,
-			doc:        map[string]interface{}{"city": "San Francisco"},
-			queryField: "city",
-			queryTerm:  "San Francisco",
-		},
-		{
-			/*
-				index_mapping: standard
-				default_mapping: keyword
-				    -> child field (should inherit keyword)
-			*/
-			name: "Child field to inherit default mapping's default analyzer",
-			mappingStr: `{"default_mapping":{"enabled":true,"dynamic":false,"properties":` +
-				`{"city":{"enabled":true,"dynamic":false,"fields":[{"name":"city","type":"text",` +
-				`"index":true}]}},"default_analyzer":"keyword"},"default_analyzer":"standard"}`,
-			doc:        map[string]interface{}{"city": "San Francisco"},
-			queryField: "city",
-			queryTerm:  "San Francisco",
-		},
-		{
-			/*
-				index_mapping: standard
-				default_mapping: keyword
-				    -> child mapping: ""
-					    -> child field: (should inherit keyword)
-			*/
-			name: "Nested child field to inherit default mapping's default analyzer",
-			mappingStr: `{"default_mapping":{"enabled":true,"dynamic":false,"default_analyzer":` +
-				`"keyword","properties":{"address":{"enabled":true,"dynamic":false,"properties":` +
-				`{"city":{"enabled":true,"dynamic":false,"fields":[{"name":"city","type":"text",` +
-				`"index":true}]}}}}},"default_analyzer":"standard"}`,
-			doc: map[string]interface{}{
-				"address": map[string]interface{}{"city": "San Francisco"},
-			},
-			queryField: "address.city",
-			queryTerm:  "San Francisco",
-		},
-		{
-			/*
-				index_mapping: standard
-				default_mapping: ""
-				    -> child mapping: "keyword"
-					    -> child mapping: ""
-						    -> child field: (should inherit keyword)
-			*/
-			name: "Nested child field to inherit first child mapping's default analyzer",
-			mappingStr: `{"default_mapping":{"enabled":true,"dynamic":false,"properties":` +
-				`{"address":{"enabled":true,"dynamic":false,"default_analyzer":"keyword",` +
-				`"properties":{"state":{"enabled":true,"dynamic":false,"properties":{"city":` +
-				`{"enabled":true,"dynamic":false,"fields":[{"name":"city","type":"text",` +
-				`"store":false,"index":true}]}}}}}}},"default_analyer":"standard"}`,
-			doc: map[string]interface{}{
-				"address": map[string]interface{}{
-					"state": map[string]interface{}{"city": "San Francisco"},
-				},
-			},
-			queryField: "address.state.city",
-			queryTerm:  "San Francisco",
-		},
-	}
-
-	for i := range tests {
-		t.Run(fmt.Sprintf("%s", tests[i].name), func(t *testing.T) {
-			idxMapping := NewIndexMapping()
-			if err := idxMapping.UnmarshalJSON([]byte(tests[i].mappingStr)); err != nil {
-				t.Fatal(err)
-			}
-
-			tmpIndexPath := createTmpIndexPath(t)
-			idx, err := New(tmpIndexPath, idxMapping)
-			if err != nil {
-				t.Fatal(err)
-			}
-
-			defer func() {
-				if err := idx.Close(); err != nil {
-					t.Fatal(err)
-				}
-			}()
-
-			if err = idx.Index("doc", tests[i].doc); err != nil {
-				t.Fatal(err)
-			}
-
-			q := NewTermQuery(tests[i].queryTerm)
-			q.SetField(tests[i].queryField)
-
-			res, err := idx.Search(NewSearchRequest(q))
-			if err != nil {
-				t.Fatal(err)
-			}
-
-			if len(res.Hits) != 1 {
-				t.Errorf("Unexpected number of hits: %v", len(res.Hits))
-			}
-		})
-	}
-}