code cleanup and refactor

Author: Thejas-bhat

index/scorch/introducer.go

@@ -468,9 +468,7 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 	if skipped {
 		atomic.AddUint64(&s.stats.TotFileMergeIntroductionsObsoleted, 1)
 	} else {
-		// tbd: should this stat correspond to total number of merged segments introduced?
-		// or is it like number of merge introductions done
-		atomic.AddUint64(&s.stats.TotIntroducedSegmentsMerge, 1)
+		atomic.AddUint64(&s.stats.TotIntroducedSegmentsMerge, uint64(len(nextMerge.new)))
 	}
 
 	atomic.StoreUint64(&s.stats.TotItemsToPersist, docsToPersistCount)

index/scorch/merge.go

@@ -334,7 +334,6 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 			}
 		}
 
-		var oldNewDocNums map[uint64][]uint64
 		var seg segment.Segment
 		var filename string
 		if len(segmentsToMerge) > 0 {
@@ -375,10 +374,10 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 			totalBytesRead := seg.BytesRead() + prevBytesReadTotal
 			seg.ResetBytesRead(totalBytesRead)
 
-			oldNewDocNums = make(map[uint64][]uint64, len(newDocNums))
 			for i, segNewDocNums := range newDocNums {
-				oldNewDocNums[task.Segments[i].Id()] = segNewDocNums
-				mergedSegHistory[task.Segments[i].Id()].oldNewDocIDs = segNewDocNums
+				if mergedSegHistory[task.Segments[i].Id()] != nil {
+					mergedSegHistory[task.Segments[i].Id()].oldNewDocIDs = segNewDocNums
+				}
 			}
 
 			atomic.AddUint64(&s.stats.TotFileMergeSegments, uint64(len(segmentsToMerge)))
@@ -489,7 +488,7 @@ func (s *Scorch) mergeSegmentBasesParallel(snapshot *IndexSnapshot, flushableObj
 	// we're tracking the merged segments and their doc number per worker
 	// to be able to introduce them all at once, so the first dimension of the
 	// slices here correspond to workerID
-	newDocNumsSet := make([][][]uint64, len(flushableObjs))
+	newDocIDsSet := make([][][]uint64, len(flushableObjs))
 	newMergedSegments := make([]segment.Segment, len(flushableObjs))
 	newMergedSegmentIDs := make([]uint64, len(flushableObjs))
 	numFlushes := len(flushableObjs)
@@ -507,15 +506,15 @@ func (s *Scorch) mergeSegmentBasesParallel(snapshot *IndexSnapshot, flushableObj
 
 			// the newly merged segment is already flushed out to disk, just needs
 			// to be opened using mmap.
-			newDocNums, _, err :=
+			newDocIDs, _, err :=
 				s.segPlugin.Merge(segsBatch, dropsBatch, path, s.closeCh, s)
 			if err != nil {
 				errs[id] = err
 				atomic.AddUint64(&s.stats.TotMemMergeErr, 1)
 				return
 			}
 			newMergedSegmentIDs[id] = newSegmentID
-			newDocNumsSet[id] = newDocNums
+			newDocIDsSet[id] = newDocIDs
 			newMergedSegments[id], err = s.segPlugin.Open(path)
 			if err != nil {
 				errs[id] = err
@@ -559,7 +558,7 @@ func (s *Scorch) mergeSegmentBasesParallel(snapshot *IndexSnapshot, flushableObj
 			// oldSegmentSnapshot.id -> {workerID, oldSegmentSnapshot, docIDs}
 			sm.mergedSegHistory[ss.id] = &mergedSegmentHistory{
 				workerID:     uint64(i),
-				oldNewDocIDs: newDocNumsSet[i][j],
+				oldNewDocIDs: newDocIDsSet[i][j],
 				oldSegment:   ss,
 			}
 		}

index/scorch/persister.go

@@ -24,6 +24,7 @@ import (
 	"math"
 	"os"
 	"path/filepath"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -381,11 +382,11 @@ type flushable struct {
 
 // number workers which parallely perform an in-memory merge of the segments
 // followed by a flush operation.
-var DefaultNumPersisterWorkers = 8
+var DefaultNumPersisterWorkers = 1
 
 // maximum size of data that a single worker is allowed to perform the in-memory
 // merge operation.
-var DefaultMaxSizeInMemoryMerge = 200 * 1024 * 1024
+var DefaultMaxSizeInMemoryMerge = 0
 
 func legacyFlushBehaviour(maxSizeInMemoryMerge, numPersisterWorkers int) bool {
 	// DefaultMaxSizeInMemoryMerge = 0 is a special value to preserve the leagcy
@@ -432,27 +433,20 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot, po *persiste
 		// constructs a flushSet where each flushable object contains a set of segments
 		// to be merged and flushed out to disk.
 		for i, snapshot := range snapshot.segment {
-			if totSize >= po.MaxSizeInMemoryMerge {
-				if len(sbs) >= DefaultMinSegmentsForInMemoryMerge {
-					numSegsToFlushOut += len(sbs)
-					val := &flushable{
-						segments: make([]segment.Segment, len(sbs)),
-						drops:    make([]*roaring.Bitmap, len(sbsDrops)),
-						sbIdxs:   make([]int, len(sbsIndexes)),
-						totDocs:  totDocs,
-					}
-					copy(val.segments, sbs)
-					copy(val.drops, sbsDrops)
-					copy(val.sbIdxs, sbsIndexes)
-					flushSet = append(flushSet, val)
-
-					oldSegIdxs = append(oldSegIdxs, sbsIndexes...)
-					sbs = sbs[:0]
-					sbsDrops = sbsDrops[:0]
-					sbsIndexes = sbsIndexes[:0]
-					totSize = 0
-					totDocs = 0
+			if totSize >= po.MaxSizeInMemoryMerge &&
+				len(sbs) >= DefaultMinSegmentsForInMemoryMerge {
+				numSegsToFlushOut += len(sbs)
+				val := &flushable{
+					segments: slices.Clone(sbs),
+					drops:    slices.Clone(sbsDrops),
+					sbIdxs:   slices.Clone(sbsIndexes),
+					totDocs:  totDocs,
 				}
+				flushSet = append(flushSet, val)
+				oldSegIdxs = append(oldSegIdxs, sbsIndexes...)
+
+				sbs, sbsDrops, sbsIndexes = sbs[:0], sbsDrops[:0], sbsIndexes[:0]
+				totSize, totDocs = 0, 0
 			}
 
 			if len(flushSet) >= int(po.NumPersisterWorkers) {
@@ -471,22 +465,13 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot, po *persiste
 		if len(flushSet) < po.NumPersisterWorkers {
 			numSegsToFlushOut += len(sbs)
 			val := &flushable{
-				segments: make([]segment.Segment, len(sbs)),
-				drops:    make([]*roaring.Bitmap, len(sbsDrops)),
-				sbIdxs:   make([]int, len(sbsIndexes)),
+				segments: slices.Clone(sbs),
+				drops:    slices.Clone(sbsDrops),
+				sbIdxs:   slices.Clone(sbsIndexes),
 				totDocs:  totDocs,
 			}
-			copy(val.segments, sbs)
-			copy(val.drops, sbsDrops)
-			copy(val.sbIdxs, sbsIndexes)
 			flushSet = append(flushSet, val)
-
 			oldSegIdxs = append(oldSegIdxs, sbsIndexes...)
-			sbs = sbs[:0]
-			sbsDrops = sbsDrops[:0]
-			sbsIndexes = sbsIndexes[:0]
-			totSize = 0
-			totDocs = 0
 		}
 	}
 

search/scorer/scorer_term.go

@@ -272,6 +272,5 @@ func (s *TermQueryScorer) Score(ctx *search.SearchContext, termMatch *index.Term
 				})
 		}
 	}
-
 	return rv
 }