Repository: blevesearch/bleve
Pull request #1389 (2020-05-01 13:17:25)

update datetime table test to use subtests


Patch:
diff --git a/builder.go b/builder.go
new file mode 100644
index 00000000..de00c97b
--- /dev/null
+++ b/builder.go
@@ -0,0 +1,94 @@
+//  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 ef6ede93..974358b8 100644
--- a/index.go
+++ b/index.go
@@ -293,3 +293,17 @@ 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 3e866f3a..551f8de8 100644
--- a/index/index.go
+++ b/index/index.go
@@ -367,3 +367,10 @@ 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
new file mode 100644
index 00000000..cd400bd1
--- /dev/null
+++ b/index/scorch/builder.go
@@ -0,0 +1,333 @@
+//  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 ffa65669..ea3667bb 100644
--- a/index/scorch/persister.go
+++ b/index/scorch/persister.go
@@ -428,55 +428,44 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot) (
 	return true, 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()
-		}
-	}()
-
+func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
+	segPlugin segment.Plugin) ([]string, map[uint64]string, error) {
 	snapshotsBucket, err := tx.CreateBucketIfNotExists(boltSnapshotsBucket)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	newSnapshotKey := segment.EncodeUvarintAscending(nil, snapshot.epoch)
 	snapshotBucket, err := snapshotsBucket.CreateBucketIfNotExists(newSnapshotKey)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 
 	// persist meta values
 	metaBucket, err := snapshotBucket.CreateBucketIfNotExists(boltMetaDataKey)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
-	err = metaBucket.Put(boltMetaDataSegmentTypeKey, []byte(s.segPlugin.Type()))
+	err = metaBucket.Put(boltMetaDataSegmentTypeKey, []byte(segPlugin.Type()))
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	buf := make([]byte, binary.MaxVarintLen32)
-	binary.BigEndian.PutUint32(buf, s.segPlugin.Version())
+	binary.BigEndian.PutUint32(buf, segPlugin.Version())
 	err = metaBucket.Put(boltMetaDataSegmentVersionKey, buf)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 
 	// persist internal values
 	internalBucket, err := snapshotBucket.CreateBucketIfNotExists(boltInternalKey)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	// TODO optimize writing these in order?
 	for k, v := range snapshot.internal {
 		err = internalBucket.Put([]byte(k), v)
 		if err != nil {
-			return err
+			return nil, nil, err
 		}
 	}
 
@@ -488,49 +477,69 @@ func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot) (err error) {
 		snapshotSegmentKey := segment.EncodeUvarintAscending(nil, segmentSnapshot.id)
 		snapshotSegmentBucket, err := snapshotBucket.CreateBucketIfNotExists(snapshotSegmentKey)
 		if err != nil {
-			return err
+			return nil, nil, err
 		}
 		switch seg := segmentSnapshot.segment.(type) {
 		case segment.PersistedSegment:
-			path := seg.Path()
-			filename := strings.TrimPrefix(path, s.path+string(os.PathSeparator))
+			segPath := seg.Path()
+			filename := strings.TrimPrefix(segPath, path+string(os.PathSeparator))
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
 			if err != nil {
-				return err
+				return nil, nil, err
 			}
 			filenames = append(filenames, filename)
 		case segment.UnpersistedSegment:
 			// need to persist this to disk
 			filename := zapFileName(segmentSnapshot.id)
-			path := s.path + string(os.PathSeparator) + filename
+			path := path + string(os.PathSeparator) + filename
 			err = seg.Persist(path)
 			if err != nil {
-				return fmt.Errorf("error persisting segment: %v", err)
+				return nil, nil, fmt.Errorf("error persisting segment: %v", err)
 			}
 			newSegmentPaths[segmentSnapshot.id] = path
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
 			if err != nil {
-				return err
+				return nil, nil, err
 			}
 			filenames = append(filenames, filename)
-
 		default:
-			return fmt.Errorf("unknown segment type: %T", seg)
+			return nil, nil, 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 fmt.Errorf("error persisting roaring bytes: %v", err)
+				return nil, nil, fmt.Errorf("error persisting roaring bytes: %v", err)
 			}
 			err = snapshotSegmentBucket.Put(boltDeletedKey, roaringBuf.Bytes())
 			if err != nil {
-				return err
+				return nil, nil, 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 698aaf16..c33f760e 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,18 +103,15 @@ func NewScorch(storeName string,
 		nextSnapshotEpoch:    1,
 		closeCh:              make(chan struct{}),
 		ineligibleForRemoval: map[string]bool{},
+		forceMergeRequestCh:  make(chan *mergerCtrl, 1),
 		segPlugin:            defaultSegmentPlugin,
 	}
 
-	// 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)
-		}
-
+	forcedSegmentType, forcedSegmentVersion, err := configForceSegmentTypeVersion(config)
+	if err != nil {
+		return nil, err
+	}
+	if forcedSegmentType != "" && forcedSegmentVersion != 0 {
 		err := rv.loadSegmentPlugin(forcedSegmentType,
 			uint32(forcedSegmentVersion))
 		if err != nil {
@@ -140,6 +139,22 @@ 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
@@ -241,6 +256,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.
@@ -567,6 +583,10 @@ 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 01eda7fb..6dfa3b28 100644
--- a/index/scorch/segment_plugin.go
+++ b/index/scorch/segment_plugin.go
@@ -60,18 +60,28 @@ func SupportedSegmentTypeVersions(typ string) (rv []uint32) {
 	return rv
 }
 
-func (s *Scorch) loadSegmentPlugin(forcedSegmentType string,
-	forcedSegmentVersion uint32) error {
+func chooseSegmentPlugin(forcedSegmentType string,
+	forcedSegmentVersion uint32) (segment.Plugin, error) {
 	if versions, ok := supportedSegmentPlugins[forcedSegmentType]; ok {
 		if segPlugin, ok := versions[uint32(forcedSegmentVersion)]; ok {
-			s.segPlugin = segPlugin
-			return nil
+			return segPlugin, nil
 		}
-		return fmt.Errorf(
+		return nil, fmt.Errorf(
 			"unsupported version %d for segment type: %s, supported: %v",
 			forcedSegmentVersion, forcedSegmentType,
 			SupportedSegmentTypeVersions(forcedSegmentType))
 	}
-	return fmt.Errorf("unsupported segment type: %s, supported: %v",
+	return nil, 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 15cb6b5f..355a602e 100644
--- a/mapping/document.go
+++ b/mapping/document.go
@@ -251,7 +251,6 @@ 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 {
@@ -308,8 +307,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]
diff --git a/search/sort.go b/search/sort.go
index 6e4ed80f..3dc11851 100644
--- a/search/sort.go
+++ b/search/sort.go
@@ -233,7 +233,11 @@ func (so SortOrder) Compare(cachedScoring, cachedDesc []bool, i, j *DocumentMatc
 		} else {
 			iVal := i.Sort[x]
 			jVal := j.Sort[x]
-			c = strings.Compare(iVal, jVal)
+			if iVal < jVal {
+				c = -1
+			} else if iVal > jVal {
+				c = 1
+			}
 		}
 
 		if c == 0 {


Test patch:
diff --git a/analysis/char/asciifolding/asciifolding_test.go b/analysis/char/asciifolding/asciifolding_test.go
index 216583d1..d79542a8 100644
--- a/analysis/char/asciifolding/asciifolding_test.go
+++ b/analysis/char/asciifolding/asciifolding_test.go
@@ -15,6 +15,7 @@
 package asciifolding
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 )
@@ -52,9 +53,11 @@ func TestAsciiFoldingFilter(t *testing.T) {
 
 	for _, test := range tests {
 		filter := New()
-		output := filter.Filter(test.input)
-		if !reflect.DeepEqual(output, test.output) {
-			t.Errorf("Expected:\n`%s`\ngot:\n`%s`\nfor:\n`%s`\n", string(test.output), string(output), string(test.input))
-		}
+		t.Run(fmt.Sprintf("on %s", test.input), func(t *testing.T) {
+			output := filter.Filter(test.input)
+			if !reflect.DeepEqual(output, test.output) {
+				t.Errorf("\nExpected:\n`%s`\ngot:\n`%s`\n", string(test.output), string(output))
+			}
+		})
 	}
 }
diff --git a/analysis/char/regexp/regexp_test.go b/analysis/char/regexp/regexp_test.go
index a3430555..ff1b04b3 100644
--- a/analysis/char/regexp/regexp_test.go
+++ b/analysis/char/regexp/regexp_test.go
@@ -15,64 +15,32 @@
 package regexp
 
 import (
+	"fmt"
 	"reflect"
 	"regexp"
 	"testing"
 )
 
 func TestRegexpCharFilter(t *testing.T) {
-	htmlTagPattern := `</?[!\w]+((\s+\w+(\s*=\s*(?:".*?"|'.*?'|[^'">\s]+))?)+\s*|\s*)/?>`
-	htmlRegex := regexp.MustCompile(htmlTagPattern)
 
-	tests := []struct {
-		input  []byte
-		output []byte
-	}{
-		{
-			input:  []byte(`<html>test</html>`),
-			output: []byte(` test `),
-		},
-	}
-
-	for _, test := range tests {
-		filter := New(htmlRegex, []byte{' '})
-		output := filter.Filter(test.input)
-		if !reflect.DeepEqual(output, test.output) {
-			t.Errorf("Expected:\n`%s`\ngot:\n`%s`\nfor:\n`%s`\n", string(test.output), string(output), string(test.input))
-		}
-	}
-}
-
-func TestZeroWidthNonJoinerCharFilter(t *testing.T) {
-	zeroWidthNonJoinerPattern := `\x{200C}`
-	zeroWidthNonJoinerRegex := regexp.MustCompile(zeroWidthNonJoinerPattern)
-
-	tests := []struct {
-		input  []byte
-		output []byte
-	}{
-		{
-			input:  []byte("water\u200Cunder\u200Cthe\u200Cbridge"),
-			output: []byte("water under the bridge"),
-		},
-	}
-
-	for _, test := range tests {
-		filter := New(zeroWidthNonJoinerRegex, []byte{' '})
-		output := filter.Filter(test.input)
-		if !reflect.DeepEqual(output, test.output) {
-			t.Errorf("Expected:\n`%s`\ngot:\n`%s`\nfor:\n`%s`\n", string(test.output), string(output), string(test.input))
-		}
-	}
-}
-
-func TestRegexpCustomReplace(t *testing.T) {
 	tests := []struct {
 		regexStr string
 		replace  []byte
 		input    []byte
 		output   []byte
 	}{
+		{
+			regexStr: `</?[!\w]+((\s+\w+(\s*=\s*(?:".*?"|'.*?'|[^'">\s]+))?)+\s*|\s*)/?>`,
+			replace:  []byte{' '},
+			input:    []byte(`<html>test</html>`),
+			output:   []byte(` test `),
+		},
+		{
+			regexStr: `\x{200C}`,
+			replace:  []byte{' '},
+			input:    []byte("water\u200Cunder\u200Cthe\u200Cbridge"),
+			output:   []byte("water under the bridge"),
+		},
 		{
 			regexStr: `([a-z])\s+(\d)`,
 			replace:  []byte(`$1-$2`),
@@ -105,13 +73,16 @@ func TestRegexpCustomReplace(t *testing.T) {
 		},
 	}
 
-	for i := range tests {
-		regex := regexp.MustCompile(tests[i].regexStr)
-		filter := New(regex, tests[i].replace)
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("match %s replace %s", test.regexStr, string(test.replace)), func(t *testing.T) {
+			regex := regexp.MustCompile(test.regexStr)
+			filter := New(regex, test.replace)
+
+			output := filter.Filter(test.input)
+			if !reflect.DeepEqual(test.output, output) {
+				t.Errorf("Expected: `%s`, Got: `%s`\n", string(test.output), string(output))
+			}
+		})
 
-		output := filter.Filter(tests[i].input)
-		if !reflect.DeepEqual(tests[i].output, output) {
-			t.Errorf("[%d] Expected: `%s`, Got: `%s`\n", i, string(tests[i].output), string(output))
-		}
 	}
 }
diff --git a/analysis/datetime/flexible/flexible_test.go b/analysis/datetime/flexible/flexible_test.go
index e97423b7..644d91eb 100644
--- a/analysis/datetime/flexible/flexible_test.go
+++ b/analysis/datetime/flexible/flexible_test.go
@@ -76,14 +76,14 @@ func TestFlexibleDateTimeParser(t *testing.T) {
 		})
 
 	for _, test := range tests {
-		actualTime, actualErr := dateOptionalTimeParser.ParseDateTime(test.input)
-		if actualErr != test.expectedError {
-			t.Errorf("expected error %#v, got %#v", test.expectedError, actualErr)
-			continue
-		}
-		if !reflect.DeepEqual(actualTime, test.expectedTime) {
-			t.Errorf("expected time %#v, got %#v", test.expectedTime, actualTime)
-			t.Errorf("expected location %#v,\n got %#v", test.expectedTime.Location(), actualTime.Location())
-		}
+		t.Run(test.input, func(t *testing.T) {
+			actualTime, actualErr := dateOptionalTimeParser.ParseDateTime(test.input)
+			if actualErr != test.expectedError {
+				t.Fatalf("expected error %#v, got %#v", test.expectedError, actualErr)
+			}
+			if !reflect.DeepEqual(actualTime, test.expectedTime) {
+				t.Errorf("expected time %v, got %v", test.expectedTime, actualTime)
+			}
+		})
 	}
 }
diff --git a/builder_test.go b/builder_test.go
new file mode 100644
index 00000000..119f39f5
--- /dev/null
+++ b/builder_test.go
@@ -0,0 +1,89 @@
+//  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
new file mode 100644
index 00000000..1d0ce716
--- /dev/null
+++ b/index/scorch/builder_test.go
@@ -0,0 +1,160 @@
+//  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 b57283f3..2b687096 100644
--- a/mapping/mapping_test.go
+++ b/mapping/mapping_test.go
@@ -1111,7 +1111,6 @@ 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)
@@ -1141,3 +1140,14 @@ 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 755999db..98623e1a 100644
--- a/search_test.go
+++ b/search_test.go
@@ -1693,3 +1693,119 @@ 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))
+			}
+		})
+	}
+}
diff --git a/test/versus_test.go b/test/versus_test.go
index 20a85d16..4257a984 100644
--- a/test/versus_test.go
+++ b/test/versus_test.go
@@ -53,7 +53,7 @@ func TestScorchVersusUpsideDownBoltAll(t *testing.T) {
 		NumDocs:              1000,
 		MaxWordsPerDoc:       20,
 		NumWords:             10,
-		BatchSize:            10,
+		BatchSize:            1000,
 		NumAttemptsPerSearch: 100,
 	}).run(scorch.Name, boltdb.Name, upsidedown.Name, boltdb.Name, nil, nil)
 }
@@ -275,7 +275,6 @@ func testVersusSearches(vt *VersusTest, searchTemplates []string, idxA, idxB ble
 		if err != nil {
 			t.Fatalf("could not parse search template: %s, err: %v", searchTemplate, err)
 		}
-
 		for j := 0; j < vt.NumAttemptsPerSearch; j++ {
 			vt.CurAttempt = j