fix file transfer logic

Author: Thejas-bhat

index/scorch/merge.go

@@ -363,7 +363,6 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context,
 
 		var seg segment.Segment
 		var filename string
-		var trainingSample []float32
 		if len(segmentsToMerge) > 0 {
 			filename = zapFileName(newSegmentID)
 			s.markIneligibleForRemoval(filename)

index/scorch/scorch.go

@@ -1084,13 +1084,6 @@ func (s *Scorch) CopyFile(file string, d index.IndexDirectory) error {
 	s.rootLock.Lock()
 	defer s.rootLock.Unlock()
 
-	// this code is currently specific to copying trained data but is future proofed for other files
-	// to be updated in the dest's bolt
-	err := s.trainer.copyFileLOCKED(file, d)
-	if err != nil {
-		return err
-	}
-
 	dest, err := d.GetWriter(filepath.Join("store", file))
 	if err != nil {
 		return err
@@ -1104,6 +1097,13 @@ func (s *Scorch) CopyFile(file string, d index.IndexDirectory) error {
 	defer source.Close()
 	defer dest.Close()
 	_, err = io.Copy(dest, source)
+	if err != nil {
+		return err
+	}
+
+	// this code is currently specific to copying trained data but is future proofed for other files
+	// to be updated in the dest's bolt
+	err = s.trainer.copyFileLOCKED(file, d)
 	return err
 }
 

index/scorch/train_vector.go

@@ -315,6 +315,12 @@ func (t *vectorTrainer) updateBolt(snapshotsBucket *bolt.Bucket, key []byte, val
 		if err != nil {
 			return err
 		}
+
+		// update the centroid index pointer
+		t.centroidIndex, err = t.parent.loadSegment(trainerBucket)
+		if err != nil {
+			return err
+		}
 	}
 
 	return nil

index_alias_impl.go

@@ -106,9 +106,6 @@ func (i *indexAliasImpl) IndexSynonym(id string, collection string, definition *
 func (i *indexAliasImpl) Train(batch *Batch) error {
 	i.mutex.RLock()
 	defer i.mutex.RUnlock()
-	if !i.open {
-		return ErrorIndexClosed
-	}
 
 	if !i.open {
 		return ErrorIndexClosed

index_impl.go

@@ -1432,7 +1432,6 @@ func (m *searchHitSorter) Less(i, j int) bool {
 	return c < 0
 }
 
-// CopyTo (index.Directory, filter)
 func (i *indexImpl) CopyTo(d index.Directory) (err error) {
 	i.mutex.RLock()
 	defer i.mutex.RUnlock()
@@ -1594,3 +1593,35 @@ func (i *indexImpl) buildTopNCollector(ctx context.Context, req *SearchRequest,
 	}
 	return newCollector(), nil
 }
+
+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)
+}