refactor file transfer

Author: Thejas-bhat

index.go

@@ -51,12 +51,10 @@ func (b *Batch) Index(id string, data interface{}) error {
 		eventIndex.FireIndexEvent()
 	}
 	doc := document.NewDocument(id)
-	// fmt.Printf("data is before mapping %#v\n", data)
 	err := b.index.Mapping().MapDocument(doc, data)
 	if err != nil {
 		return err
 	}
-	// fmt.Printf("data is after mapping %#v\n", doc)
 	b.internal.Update(doc)
 
 	b.lastDocSize = uint64(doc.Size() +
@@ -355,11 +353,6 @@ type IndexCopyable interface {
 	CopyTo(d index.Directory) error
 }
 
-type IndexFileCopyable interface {
-	UpdateFileInBolt(key []byte, value []byte) error
-	CopyFile(file string, d index.IndexDirectory) error
-}
-
 // FileSystemDirectory is the default implementation for the
 // index.Directory interface.
 type FileSystemDirectory string

index/scorch/merge.go

@@ -375,7 +375,6 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 			atomic.AddUint64(&s.stats.TotFileMergeZapBeg, 1)
 			prevBytesReadTotal := cumulateBytesRead(segmentsToMerge)
 
-			fmt.Println("files", files)
 			newDocNums, _, err := s.segPlugin.MergeEx(segmentsToMerge, docsToDrop, path,
 				cw.cancelCh, s, s.segmentConfig)
 			atomic.AddUint64(&s.stats.TotFileMergeZapEnd, 1)
@@ -541,7 +540,6 @@ func (s *Scorch) mergeAndPersistInMemorySegments(snapshot *IndexSnapshot,
 			filename := zapFileName(newSegmentID)
 			path := s.path + string(os.PathSeparator) + filename
 
-			fmt.Println("version while merging", s.segPlugin.Version())
 			// the newly merged segment is already flushed out to disk, just needs
 			// to be opened using mmap.
 			newDocIDs, _, err :=

index/scorch/persister.go

@@ -425,7 +425,6 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot, po *persiste
 	var totSize int
 	var numSegsToFlushOut int
 	var totDocs uint64
-	fmt.Println("persister doing its thing")
 	// legacy behaviour of merge + flush of all in-memory segments in one-shot
 	if legacyFlushBehaviour(po.MaxSizeInMemoryMergePerWorker, po.NumPersisterWorkers) {
 		val := &flushable{
@@ -887,12 +886,7 @@ func (s *Scorch) loadFromBolt() error {
 				s.AddEligibleForRemoval(snapshotEpoch)
 				continue
 			}
-			// fmt.Println("loadFromBolt key %s", k)
-			// if k[0] == util.BoltCentroidIndexKey[0] {
-			// 	fmt.Println("loadFromBolt centroid index key", string(k))
 
-			// 	continue
-			// }
 			snapshot := snapshots.Bucket(k)
 			if snapshot == nil {
 				log.Printf("snapshot key, but bucket missing %x, continuing", k)

index/scorch/scorch.go

@@ -15,10 +15,8 @@
 package scorch
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
-	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -83,7 +81,7 @@ type Scorch struct {
 	persisterNotifier        chan *epochWatcher
 	rootBolt                 *bolt.DB
 	asyncTasks               sync.WaitGroup
-	// not a real searchable segment, singleton
+	// not a real searchable segment
 	centroidIndex *SegmentSnapshot
 	train         chan *trainRequest
 
@@ -184,7 +182,6 @@ func NewScorch(storeName string,
 		}
 	}
 
-	// "pretraining": true
 	segConfig, ok := config["segmentConfig"].(map[string]interface{})
 	if ok {
 		rv.segmentConfig = segConfig
@@ -592,7 +589,7 @@ func (s *Scorch) trainerLoop() {
 	// some init stuff
 	s.segmentConfig["getCentroidIndexCallback"] = s.getCentroidIndex
 	var totalSamplesProcessed int
-	filename := "centroid_index"
+	filename := index.CentroidIndexFileName
 	path := filepath.Join(s.path, filename)
 	for {
 		select {
@@ -620,7 +617,7 @@ func (s *Scorch) trainerLoop() {
 				// persist in a tmp file and then rename - is that a fair strategy?
 				s.segmentConfig["training"] = true
 				_, _, err := s.segPlugin.MergeEx([]segment.Segment{s.centroidIndex.segment, sampleSeg},
-					[]*roaring.Bitmap{nil, nil}, filepath.Join(s.path, "centroid_index.tmp"), s.closeCh, nil, s.segmentConfig)
+					[]*roaring.Bitmap{nil, nil}, filepath.Join(s.path, index.CentroidIndexFileName+".tmp"), s.closeCh, nil, s.segmentConfig)
 				if err != nil {
 					trainReq.ackCh <- fmt.Errorf("error merging centroid index: %v", err)
 					close(trainReq.ackCh)
@@ -630,7 +627,7 @@ func (s *Scorch) trainerLoop() {
 
 				// close the existing centroid segment - it's supposed to be gc'd at this point
 				s.centroidIndex.segment.Close()
-				err = moveFile(filepath.Join(s.path, "centroid_index.tmp"), filepath.Join(s.path, "centroid_index"))
+				err = moveFile(filepath.Join(s.path, index.CentroidIndexFileName+".tmp"), filepath.Join(s.path, index.CentroidIndexFileName))
 				if err != nil {
 					trainReq.ackCh <- fmt.Errorf("error renaming centroid index: %v", err)
 					close(trainReq.ackCh)
@@ -691,7 +688,7 @@ func (s *Scorch) trainerLoop() {
 			}
 
 			// update the centroid index pointer
-			centroidIndex, err := s.segPlugin.OpenEx(filepath.Join(s.path, "centroid_index"), s.segmentConfig)
+			centroidIndex, err := s.segPlugin.OpenEx(filepath.Join(s.path, index.centroidIndexFileName), s.segmentConfig)
 			if err != nil {
 				trainReq.ackCh <- fmt.Errorf("error opening centroid index: %v", err)
 				close(trainReq.ackCh)
@@ -709,8 +706,6 @@ func (s *Scorch) Train(batch *index.Batch) error {
 	// regulate the Train function
 	s.FireIndexEvent()
 
-	// batch.InternalOps
-
 	var trainData []index.Document
 	for key, doc := range batch.IndexOps {
 		if doc != nil {
@@ -732,7 +727,7 @@ func (s *Scorch) Train(batch *index.Batch) error {
 	//
 	// note: this might index text data too, how to handle this? s.segmentConfig?
 	// todo: updates/deletes -> data drift detection
-	seg, n, err := s.segPlugin.NewEx(trainData, s.segmentConfig)
+	seg, _, err := s.segPlugin.NewEx(trainData, s.segmentConfig)
 	if err != nil {
 		return err
 	}
@@ -749,7 +744,6 @@ func (s *Scorch) Train(batch *index.Batch) error {
 		return err
 	}
 
-	fmt.Println("number of bytes written to centroid index", n)
 	return err
 }
 
@@ -1212,88 +1206,6 @@ func (s *Scorch) CopyReader() index.CopyReader {
 	return rv
 }
 
-func (s *Scorch) UpdateFileInBolt(key []byte, value []byte) error {
-	tx, err := s.rootBolt.Begin(true)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err != nil {
-			_ = tx.Rollback()
-		}
-	}()
-
-	snapshotsBucket, err := tx.CreateBucketIfNotExists(util.BoltSnapshotsBucket)
-	if err != nil {
-		return err
-	}
-
-	// currently this is specific to centroid index file update
-	if bytes.Equal(key, util.BoltCentroidIndexKey) {
-		// todo: guard against duplicate updates
-		centroidBucket, err := snapshotsBucket.CreateBucketIfNotExists(util.BoltCentroidIndexKey)
-		if err != nil {
-			return err
-		}
-		if centroidBucket == nil {
-			return fmt.Errorf("centroid bucket not found")
-		}
-		existingValue := centroidBucket.Get(util.BoltPathKey)
-		if existingValue != nil {
-			return fmt.Errorf("key already exists %v %v", s.path, string(existingValue))
-		}
-
-		err = centroidBucket.Put(util.BoltPathKey, value)
-		if err != nil {
-			return err
-		}
-	}
-
-	err = tx.Commit()
-	if err != nil {
-		return err
-	}
-
-	err = s.rootBolt.Sync()
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-// CopyFile copies a specific file to a destination directory which has an access to a bleve index
-// doing a io.Copy() isn't enough because the file needs to be tracked in bolt file as well
-func (s *Scorch) CopyFile(file string, d index.IndexDirectory) error {
-	s.rootLock.Lock()
-	defer s.rootLock.Unlock()
-
-	// this code is currently specific to centroid index file but is future proofed for other files
-	// to be updated in the dest's bolt
-	if strings.HasSuffix(file, "centroid_index") {
-		// centroid index file - this is outside the snapshots domain so the bolt update is different
-		err := d.UpdateFileInBolt(util.BoltCentroidIndexKey, []byte(file))
-		if err != nil {
-			return fmt.Errorf("error updating dest index bolt: %w", err)
-		}
-	}
-
-	dest, err := d.GetWriter(filepath.Join("store", file))
-	if err != nil {
-		return err
-	}
-
-	source, err := os.Open(filepath.Join(s.path, file))
-	if err != nil {
-		return err
-	}
-
-	defer source.Close()
-	defer dest.Close()
-	_, err = io.Copy(dest, source)
-	return err
-}
-
 // external API to fire a scorch event (EventKindIndexStart) externally from bleve
 func (s *Scorch) FireIndexEvent() {
 	s.fireEvent(EventKindIndexStart, 0)

index/scorch/segment_plugin.go

@@ -101,7 +101,6 @@ func RegisterSegmentPlugin(plugin SegmentPlugin, makeDefault bool) {
 	}
 	supportedSegmentPlugins[plugin.Type()][plugin.Version()] = plugin
 	if makeDefault {
-		fmt.Println("registering default segment plugin", plugin.Type(), plugin.Version())
 		defaultSegmentPlugin = plugin
 	}
 }

index/scorch/snapshot_index.go

@@ -66,8 +66,7 @@ func init() {
 }
 
 type IndexSnapshot struct {
-	parent *Scorch
-
+	parent   *Scorch
 	segment  []*SegmentSnapshot
 	offsets  []uint64
 	internal map[string][]byte

index_impl.go

@@ -326,13 +326,11 @@ func (i *indexImpl) Index(id string, data interface{}) (err error) {
 
 	i.FireIndexEvent()
 
-	// fmt.Printf("data is %#v\n", data)
 	doc := document.NewDocument(id)
 	err = i.m.MapDocument(doc, data)
 	if err != nil {
 		return
 	}
-	// fmt.Printf("data is after mapping %#v\n", doc)
 	err = i.i.Update(doc)
 	return
 }
@@ -1432,39 +1430,6 @@ func (m *searchHitSorter) Less(i, j int) bool {
 	return c < 0
 }
 
-func (i *indexImpl) CopyFile(file string, d index.IndexDirectory) (err error) {
-	i.mutex.RLock()
-	defer i.mutex.RUnlock()
-
-	if !i.open {
-		return ErrorIndexClosed
-	}
-
-	copyIndex, ok := i.i.(index.IndexFileCopyable)
-	if !ok {
-		return fmt.Errorf("index implementation does not support copy reader")
-	}
-
-	return copyIndex.CopyFile(file, d)
-}
-
-func (i *indexImpl) UpdateFileInBolt(key []byte, value []byte) error {
-	i.mutex.RLock()
-	defer i.mutex.RUnlock()
-
-	if !i.open {
-		return ErrorIndexClosed
-	}
-
-	copyIndex, ok := i.i.(index.IndexFileCopyable)
-	if !ok {
-		return fmt.Errorf("index implementation does not support file copy")
-	}
-
-	return copyIndex.UpdateFileInBolt(key, value)
-}
-
-// CopyTo (index.Directory, filter)
 func (i *indexImpl) CopyTo(d index.Directory) (err error) {
 	i.mutex.RLock()
 	defer i.mutex.RUnlock()
@@ -1478,8 +1443,6 @@ func (i *indexImpl) CopyTo(d index.Directory) (err error) {
 		return fmt.Errorf("index implementation does not support copy reader")
 	}
 
-	// copyIndex.Copy() -> copies the centroid index
-
 	copyReader := copyIndex.CopyReader()
 	if copyReader == nil {
 		return fmt.Errorf("index's copyReader is nil")