Repository: blevesearch/bleve
Pull request #1605 (2021-06-28 13:42:14)

Introduce online Copy for scorch indexes.
Approach used is to acquire an index snapshot
and reuse existing code for persisting
snapshot segments and recording snapshot
information into the root bolt.

sreekanth-cb: @mschoch , changed the signature to accept a path here.
sreekanth-cb: @mschoch, moved the FileSystemDirectory here.
abhinavdangeti: Once the bleve_index_api and zapx changes are merged - we'll need to set up new versions and update the go.mod here absorbing the right version of the two before committing this.
abhinavdangeti: @sreekanth-cb Rebasing your change on master branch (which now includes https://github.com/blevesearch/bleve/pull/1612) should allow for the checks to pass.

Patch:
diff --git a/index.go b/index.go
index e08271e5..e297122f 100644
--- a/index.go
+++ b/index.go
@@ -16,6 +16,7 @@ package bleve
 
 import (
 	"context"
+
 	"github.com/blevesearch/bleve/v2/index/upsidedown"
 
 	"github.com/blevesearch/bleve/v2/document"
@@ -306,3 +307,15 @@ type Builder interface {
 func NewBuilder(path string, mapping mapping.IndexMapping, config map[string]interface{}) (Builder, error) {
 	return newBuilder(path, mapping, config)
 }
+
+// IndexCopyable is an index which supports an online copy operation
+// of the index.
+type IndexCopyable interface {
+	// CopyTo creates a fully functional copy of the index at the
+	// specified destination directory implementation.
+	CopyTo(d index.Directory) error
+}
+
+// FileSystemDirectory is the default implementation for the
+// index.Directory interface.
+type FileSystemDirectory string
diff --git a/index/scorch/builder.go b/index/scorch/builder.go
index 3311bd02..5e64230c 100644
--- a/index/scorch/builder.go
+++ b/index/scorch/builder.go
@@ -304,7 +304,7 @@ func (o *Builder) Close() error {
 	}
 
 	// fill the root bolt with this fake index snapshot
-	_, _, err = prepareBoltSnapshot(is, tx, o.path, o.segPlugin)
+	_, _, err = prepareBoltSnapshot(is, tx, o.path, o.segPlugin, nil)
 	if err != nil {
 		_ = tx.Rollback()
 		_ = rootBolt.Close()
diff --git a/index/scorch/persister.go b/index/scorch/persister.go
index 00c92429..1688edae 100644
--- a/index/scorch/persister.go
+++ b/index/scorch/persister.go
@@ -19,6 +19,7 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"math"
@@ -427,8 +428,59 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot) (
 	return true, nil
 }
 
+func copyToDirectory(srcPath string, d index.Directory) (int64, error) {
+	if d == nil {
+		return 0, nil
+	}
+
+	dest, err := d.GetWriter(filepath.Join("store", filepath.Base(srcPath)))
+	if err != nil {
+		return 0, fmt.Errorf("GetWriter err: %v", err)
+	}
+
+	sourceFileStat, err := os.Stat(srcPath)
+	if err != nil {
+		return 0, err
+	}
+
+	if !sourceFileStat.Mode().IsRegular() {
+		return 0, fmt.Errorf("%s is not a regular file", srcPath)
+	}
+
+	source, err := os.Open(srcPath)
+	if err != nil {
+		return 0, err
+	}
+	defer source.Close()
+	defer dest.Close()
+	return io.Copy(dest, source)
+}
+
+func persistToDirectory(seg segment.UnpersistedSegment, d index.Directory,
+	path string) error {
+	if d == nil {
+		return seg.Persist(path)
+	}
+
+	sg, ok := seg.(io.WriterTo)
+	if !ok {
+		return fmt.Errorf("no io.WriterTo segment implementation found")
+	}
+
+	w, err := d.GetWriter(filepath.Join("store", filepath.Base(path)))
+	if err != nil {
+		return err
+	}
+
+	_, err = sg.WriteTo(w)
+	w.Close()
+
+	return err
+}
+
 func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
-	segPlugin SegmentPlugin) ([]string, map[uint64]string, error) {
+	segPlugin SegmentPlugin, d index.Directory) (
+	[]string, map[uint64]string, error) {
 	snapshotsBucket, err := tx.CreateBucketIfNotExists(boltSnapshotsBucket)
 	if err != nil {
 		return nil, nil, err
@@ -481,7 +533,11 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
 		switch seg := segmentSnapshot.segment.(type) {
 		case segment.PersistedSegment:
 			segPath := seg.Path()
-			filename := strings.TrimPrefix(segPath, path+string(os.PathSeparator))
+			_, err = copyToDirectory(segPath, d)
+			if err != nil {
+				return nil, nil, fmt.Errorf("segment: %s copy err: %v", segPath, err)
+			}
+			filename := filepath.Base(segPath)
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
 			if err != nil {
 				return nil, nil, err
@@ -490,10 +546,10 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string,
 		case segment.UnpersistedSegment:
 			// need to persist this to disk
 			filename := zapFileName(segmentSnapshot.id)
-			path := path + string(os.PathSeparator) + filename
-			err = seg.Persist(path)
+			path := filepath.Join(path, filename)
+			err := persistToDirectory(seg, d, path)
 			if err != nil {
-				return nil, nil, fmt.Errorf("error persisting segment: %v", err)
+				return nil, nil, fmt.Errorf("segment: %s persist err: %v", path, err)
 			}
 			newSegmentPaths[segmentSnapshot.id] = path
 			err = snapshotSegmentBucket.Put(boltPathKey, []byte(filename))
@@ -534,7 +590,7 @@ func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot) (err error) {
 		}
 	}()
 
-	filenames, newSegmentPaths, err := prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin)
+	filenames, newSegmentPaths, err := prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin, nil)
 	if err != nil {
 		return err
 	}
diff --git a/index/scorch/snapshot_index.go b/index/scorch/snapshot_index.go
index ac2f3441..98282eae 100644
--- a/index/scorch/snapshot_index.go
+++ b/index/scorch/snapshot_index.go
@@ -18,6 +18,8 @@ import (
 	"container/heap"
 	"encoding/binary"
 	"fmt"
+	"os"
+	"path/filepath"
 	"reflect"
 	"sort"
 	"sync"
@@ -29,6 +31,7 @@ import (
 	segment "github.com/blevesearch/scorch_segment_api/v2"
 	"github.com/blevesearch/vellum"
 	lev "github.com/blevesearch/vellum/levenshtein"
+	bolt "go.etcd.io/bbolt"
 )
 
 // re usable, threadsafe levenshtein builders
@@ -762,3 +765,46 @@ OUTER:
 	}
 	return rv
 }
+
+func (i *IndexSnapshot) CopyTo(d index.Directory) error {
+	// get the root bolt file.
+	w, err := d.GetWriter(filepath.Join("store", "root.bolt"))
+	if err != nil || w == nil {
+		return fmt.Errorf("failed to create the root.bolt file, err: %v", err)
+	}
+	rootFile, ok := w.(*os.File)
+	if !ok {
+		return fmt.Errorf("invalid root.bolt file found")
+	}
+
+	copyBolt, err := bolt.Open(rootFile.Name(), 0600, nil)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		w.Close()
+		if cerr := copyBolt.Close(); cerr != nil && err == nil {
+			err = cerr
+		}
+	}()
+
+	// start a write transaction
+	tx, err := copyBolt.Begin(true)
+	if err != nil {
+		return err
+	}
+
+	_, _, err = prepareBoltSnapshot(i, tx, "", i.parent.segPlugin, d)
+	if err != nil {
+		_ = tx.Rollback()
+		return fmt.Errorf("error backing up index snapshot: %v", err)
+	}
+
+	// commit bolt data
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("error commit tx to backup root bolt: %v", err)
+	}
+
+	return copyBolt.Sync()
+}
diff --git a/index_impl.go b/index_impl.go
index 879a3663..8a9cfd3b 100644
--- a/index_impl.go
+++ b/index_impl.go
@@ -18,7 +18,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -910,3 +912,49 @@ func (m *searchHitSorter) Less(i, j int) bool {
 	c := m.sort.Compare(m.cachedScoring, m.cachedDesc, m.hits[i], m.hits[j])
 	return c < 0
 }
+
+func (i *indexImpl) CopyTo(d index.Directory) (err error) {
+	i.mutex.RLock()
+	defer i.mutex.RUnlock()
+
+	if !i.open {
+		return ErrorIndexClosed
+	}
+
+	indexReader, err := i.i.Reader()
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if cerr := indexReader.Close(); err == nil && cerr != nil {
+			err = cerr
+		}
+	}()
+
+	irc, ok := indexReader.(IndexCopyable)
+	if !ok {
+		return fmt.Errorf("index implementation does not support copy")
+	}
+
+	err = irc.CopyTo(d)
+	if err != nil {
+		return fmt.Errorf("error copying index metadata: %v", err)
+	}
+
+	// copy the metadata
+	return i.meta.CopyTo(d)
+}
+
+func (f FileSystemDirectory) GetWriter(filePath string) (io.WriteCloser,
+	error) {
+	dir, file := filepath.Split(filePath)
+	if dir != "" {
+		err := os.MkdirAll(filepath.Join(string(f), dir), os.ModePerm)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return os.OpenFile(filepath.Join(string(f), dir, file),
+		os.O_RDWR|os.O_CREATE, 0600)
+}
diff --git a/index_meta.go b/index_meta.go
index fe0ddeba..711e0c1b 100644
--- a/index_meta.go
+++ b/index_meta.go
@@ -16,11 +16,13 @@ package bleve
 
 import (
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 
 	"github.com/blevesearch/bleve/v2/index/upsidedown"
+	index "github.com/blevesearch/bleve_index_api"
 )
 
 const metaFilename = "index_meta.json"
@@ -92,6 +94,23 @@ func (i *indexMeta) Save(path string) (err error) {
 	return nil
 }
 
+func (i *indexMeta) CopyTo(d index.Directory) (err error) {
+	metaBytes, err := json.Marshal(i)
+	if err != nil {
+		return err
+	}
+
+	w, err := d.GetWriter(metaFilename)
+	if w == nil || err != nil {
+		return fmt.Errorf("invalid writer for file: %s, err: %v",
+			metaFilename, err)
+	}
+	defer w.Close()
+
+	_, err = w.Write(metaBytes)
+	return err
+}
+
 func indexMetaPath(path string) string {
 	return filepath.Join(path, metaFilename)
 }


Test patch:
diff --git a/index_test.go b/index_test.go
index a3d0d1dc..c0af3689 100644
--- a/index_test.go
+++ b/index_test.go
@@ -2261,3 +2261,227 @@ func TestIndexMappingDocValuesDynamic(t *testing.T) {
 		t.Fatalf("Expected DocValuesDynamic to remain false after the index mapping edit")
 	}
 }
+
+func TestCopyIndex(t *testing.T) {
+	tmpIndexPath := createTmpIndexPath(t)
+	defer cleanupTmpIndexPath(t, tmpIndexPath)
+
+	idx, err := New(tmpIndexPath, NewIndexMapping())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		err := idx.Close()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	doca := map[string]interface{}{
+		"name": "tester",
+		"desc": "gophercon india testing",
+	}
+	err = idx.Index("a", doca)
+	if err != nil {
+		t.Error(err)
+	}
+
+	docy := map[string]interface{}{
+		"name": "jasper",
+		"desc": "clojure",
+	}
+	err = idx.Index("y", docy)
+	if err != nil {
+		t.Error(err)
+	}
+
+	err = idx.Delete("y")
+	if err != nil {
+		t.Error(err)
+	}
+
+	docx := map[string]interface{}{
+		"name": "rose",
+		"desc": "xoogler",
+	}
+	err = idx.Index("x", docx)
+	if err != nil {
+		t.Error(err)
+	}
+
+	err = idx.SetInternal([]byte("status"), []byte("pending"))
+	if err != nil {
+		t.Error(err)
+	}
+
+	docb := map[string]interface{}{
+		"name": "sree",
+		"desc": "cbft janitor",
+	}
+	batch := idx.NewBatch()
+	err = batch.Index("b", docb)
+	if err != nil {
+		t.Error(err)
+	}
+	batch.Delete("x")
+	batch.SetInternal([]byte("batchi"), []byte("batchv"))
+	batch.DeleteInternal([]byte("status"))
+	err = idx.Batch(batch)
+	if err != nil {
+		t.Error(err)
+	}
+	val, err := idx.GetInternal([]byte("batchi"))
+	if err != nil {
+		t.Error(err)
+	}
+	if string(val) != "batchv" {
+		t.Errorf("expected 'batchv', got '%s'", val)
+	}
+	val, err = idx.GetInternal([]byte("status"))
+	if err != nil {
+		t.Error(err)
+	}
+	if val != nil {
+		t.Errorf("expected nil, got '%s'", val)
+	}
+
+	err = idx.SetInternal([]byte("seqno"), []byte("7"))
+	if err != nil {
+		t.Error(err)
+	}
+	err = idx.SetInternal([]byte("status"), []byte("ready"))
+	if err != nil {
+		t.Error(err)
+	}
+	err = idx.DeleteInternal([]byte("status"))
+	if err != nil {
+		t.Error(err)
+	}
+	val, err = idx.GetInternal([]byte("status"))
+	if err != nil {
+		t.Error(err)
+	}
+	if val != nil {
+		t.Errorf("expected nil, got '%s'", val)
+	}
+
+	val, err = idx.GetInternal([]byte("seqno"))
+	if err != nil {
+		t.Error(err)
+	}
+	if string(val) != "7" {
+		t.Errorf("expected '7', got '%s'", val)
+	}
+
+	count, err := idx.DocCount()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if count != 2 {
+		t.Errorf("expected doc count 2, got %d", count)
+	}
+
+	doc, err := idx.Document("a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	foundNameField := false
+	doc.VisitFields(func(field index.Field) {
+		if field.Name() == "name" && string(field.Value()) == "tester" {
+			foundNameField = true
+		}
+	})
+	if !foundNameField {
+		t.Errorf("expected to find field named 'name' with value 'tester'")
+	}
+
+	fields, err := idx.Fields()
+	if err != nil {
+		t.Fatal(err)
+	}
+	expectedFields := map[string]bool{
+		"_all": false,
+		"name": false,
+		"desc": false,
+	}
+	if len(fields) < len(expectedFields) {
+		t.Fatalf("expected %d fields got %d", len(expectedFields), len(fields))
+	}
+	for _, f := range fields {
+		expectedFields[f] = true
+	}
+	for ef, efp := range expectedFields {
+		if !efp {
+			t.Errorf("field %s is missing", ef)
+		}
+	}
+
+	// now create a copy of the index, and repeat assertions on it
+	copyableIndex, ok := idx.(IndexCopyable)
+	if !ok {
+		t.Fatal("index doesn't support copy")
+	}
+	backupIndexPath := createTmpIndexPath(t)
+	defer cleanupTmpIndexPath(t, backupIndexPath)
+
+	err = copyableIndex.CopyTo(FileSystemDirectory(backupIndexPath))
+	if err != nil {
+		t.Fatalf("error copying the index: %v", err)
+	}
+
+	// open the copied  index
+	idxCopied, err := Open(backupIndexPath)
+	if err != nil {
+		t.Fatalf("unable to open copy index")
+	}
+	defer func() {
+		err := idxCopied.Close()
+		if err != nil {
+			t.Fatalf("error closing copy index: %v", err)
+		}
+	}()
+
+	// assertions on copied index
+	copyCount, err := idxCopied.DocCount()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if copyCount != 2 {
+		t.Errorf("expected doc count 2, got %d", copyCount)
+	}
+
+	copyDoc, err := idxCopied.Document("a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	copyFoundNameField := false
+	copyDoc.VisitFields(func(field index.Field) {
+		if field.Name() == "name" && string(field.Value()) == "tester" {
+			copyFoundNameField = true
+		}
+	})
+	if !copyFoundNameField {
+		t.Errorf("expected copy index to find field named 'name' with value 'tester'")
+	}
+
+	copyFields, err := idx.Fields()
+	if err != nil {
+		t.Fatal(err)
+	}
+	copyExpectedFields := map[string]bool{
+		"_all": false,
+		"name": false,
+		"desc": false,
+	}
+	if len(copyFields) < len(copyExpectedFields) {
+		t.Fatalf("expected %d fields got %d", len(copyExpectedFields), len(copyFields))
+	}
+	for _, f := range copyFields {
+		copyExpectedFields[f] = true
+	}
+	for ef, efp := range copyExpectedFields {
+		if !efp {
+			t.Errorf("copy field %s is missing", ef)
+		}
+	}
+}