roaring-migrate bug;performance improvements

cmd/roaring-migrate/main.go

@@ -4,6 +4,7 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -17,11 +18,20 @@ import (
 	"github.com/molecula/featurebase/v3/rbf/cfg"
 	"github.com/molecula/featurebase/v3/roaring"
 	txkey "github.com/molecula/featurebase/v3/short_txkey"
+	"github.com/molecula/featurebase/v3/vprint"
 	"github.com/spf13/cobra"
 )
 
+var visited map[string]int64
+
+const (
+	Version = "1.0"
+)
+
 func main() {
+	visited = make(map[string]int64)
 	var dataDir, backupPath string
+	var verbose bool
 	cmdMigrate := &cobra.Command{
 		Use:   "roaring-migrate",
 		Short: "convert roaring pilosa backup to rbf",
@@ -29,7 +39,7 @@ func main() {
 		Run: func(cmd *cobra.Command, args []string) {
 			nodes := strings.Split(dataDir, ",")
 			for _, nodePath := range nodes {
-				err := Migrate(nodePath, backupPath)
+				err := Migrate(nodePath, backupPath, verbose)
 				if err != nil {
 					fmt.Println("Error", err)
 					return
@@ -40,6 +50,7 @@ func main() {
 	}
 	cmdMigrate.Flags().StringVarP(&dataDir, "data-dir", "d", "", "source directories for each node seperated by commas")
 	cmdMigrate.Flags().StringVarP(&backupPath, "backup-dir", "b", "", "location of backup directory")
+	cmdMigrate.Flags().BoolVar(&verbose, "verbose", false, "addition progress information")
 	err := cmdMigrate.MarkFlagRequired("data-dir")
 	if err != nil {
 		fmt.Println("Error setting flag data-dir")
@@ -52,6 +63,9 @@ func main() {
 		os.Exit(1)
 		return
 	}
+	if verbose {
+		vprint.VV("Version: %v", Version)
+	}
 
 	err = cmdMigrate.Execute()
 	if err != nil {
@@ -94,6 +108,14 @@ type local struct {
 	Fields    []*pilosa.FieldInfo `json:"fields,omitempty"`
 }
 
+func fileExists(filename string) (bool, int64) {
+	info, err := os.Stat(filename)
+	if os.IsNotExist(err) {
+		return false, 0
+	}
+	return !info.IsDir(), info.Size()
+}
+
 func BuildSchema(dataDir string) ([]byte, error) {
 	//need to find all the ".meta" files and load as field options
 
@@ -197,20 +219,34 @@ func (d *rbfFile) getDB(path, index string, shard uint64) (*rbf.DB, error) {
 	return d.working, nil
 }
 func (d *rbfFile) Close() error {
+	defer func() error {
+		//cleanup the tempdirectory
+		err := os.RemoveAll(d.temp)
+		if err != nil {
+			return err
+		}
+		return nil
+	}()
+
 	if d.last != "" {
 		d.working.Close()
+
 		//if d.last exists only keep the biggest
-		err := os.MkdirAll(filepath.Dir(d.last), 0777)
-		if err != nil {
-			return err
+		exists, sz := fileExists(d.last)
+		src := filepath.Join(d.temp, "data")
+		if !exists {
+			err := os.MkdirAll(filepath.Dir(d.last), 0777)
+			if err != nil {
+				return err
+			}
+		} else {
+			_, sz2 := fileExists(src)
+			if sz > sz2 {
+				return nil
+			}
 		}
 		// move the datafile backup shard
-		err = os.Rename(filepath.Join(d.temp, "data"), d.last)
-		if err != nil {
-			return err
-		}
-		//cleanup the tempdirectory
-		err = os.RemoveAll(d.temp)
+		err := os.Rename(src, d.last)
 		if err != nil {
 			return err
 		}
@@ -218,19 +254,26 @@ func (d *rbfFile) Close() error {
 	return nil
 }
 func copyFile(src, dest string) error {
-	input, err := ioutil.ReadFile(src)
+	from, err := os.Open(src)
 	if err != nil {
 		return err
 	}
+	defer from.Close()
 
-	err = ioutil.WriteFile(dest, input, 0644)
+	to, err := os.OpenFile(dest, os.O_RDWR|os.O_CREATE, 0644)
+	if err != nil {
+		return err
+	}
+	defer to.Close()
+
+	_, err = io.Copy(to, from)
 	if err != nil {
 		return err
 	}
 	return nil
 }
 
-func Migrate(dataDir, backupPath string) error {
+func Migrate(dataDir, backupPath string, verbose bool) error {
 	dataDir = strings.TrimSuffix(dataDir, "/")
 
 	err := os.MkdirAll(backupPath, 0777)
@@ -279,7 +322,21 @@ func Migrate(dataDir, backupPath string) error {
 	bm := roaring.NewSliceBitmap()
 	for _, filename := range raw {
 		index, field, view, shard := Extract(filename)
-
+		sz, before := visited[filename]
+		fi, _ := os.Stat(dataDir + filename)
+		if field != "_exists" {
+			if !before {
+				visited[filename] = fi.Size()
+			} else {
+				if fi.Size() <= sz {
+					continue //skipp it
+				}
+				visited[filename] = fi.Size()
+			}
+		}
+		if verbose {
+			vprint.VV("processing: %v", dataDir+filename)
+		}
 		content, err := ioutil.ReadFile(dataDir + filename)
 		if err != nil {
 			return err
@@ -293,35 +350,19 @@ func Migrate(dataDir, backupPath string) error {
 		if err != nil {
 			return err
 		}
-		tx, err := db.Begin(true)
-		if err != nil {
-			return err
-		}
 		key := string(txkey.Prefix(index, field, view, shard))
-		itr, ok := bm.Containers.Iterator(0)
-		if ok {
-			for itr.Next() {
-				k, v := itr.Value()
-				tx.PutContainer(key, k, v)
-
-			}
-		}
+		tx, err := db.Begin(true)
+		tx.AddRoaring(key, bm)
 		err = tx.Commit()
-		if err != nil {
-			return err
-		}
 	}
 	cache.Close()
 	keys := FetchIndexKeys(dataDir)
 	for _, filename := range keys {
 		fmt.Println("index keys", filename)
-		content, err := ioutil.ReadFile(filepath.Join(dataDir, filename))
-		if err != nil {
-			return err
-		}
+		srcFile := filepath.Join(dataDir, filename)
 		parts := strings.Split(filename, "/")
 		destFile := filepath.Join(backupPath, "indexes", parts[1], "translate", parts[3])
-		err = writeIfBigger(destFile, content)
+		err = writeIfBigger(destFile, srcFile)
 		if err != nil {
 			return err
 		}
@@ -331,13 +372,10 @@ func Migrate(dataDir, backupPath string) error {
 	keys = FetchRowkeys(dataDir)
 	for _, filename := range keys {
 		fmt.Println("field", filename)
-		content, err := ioutil.ReadFile(dataDir + filename)
-		if err != nil {
-			return err
-		}
+		srcFile := dataDir + filename
 		parts := strings.Split(filename, "/")
 		destFile := filepath.Join(backupPath, "indexes", parts[1], "fields", parts[2], "translate")
-		err = writeIfBigger(destFile, content)
+		err = writeIfBigger(destFile, srcFile)
 		if err != nil {
 			return err
 		}
@@ -345,16 +383,21 @@ func Migrate(dataDir, backupPath string) error {
 	return nil
 }
 
-func writeIfBigger(dst string, content []byte) error {
+func writeIfBigger(dst string, srcFile string) error {
 	if stats, err := os.Stat(dst); os.IsNotExist(err) {
 		err = os.MkdirAll(filepath.Dir(dst), 0777)
 		if err != nil {
 			return err
 		}
-		return ioutil.WriteFile(dst, content, 0644)
+		return copyFile(srcFile, dst)
 	} else {
-		if stats.Size() < int64(len(content)) {
-			return ioutil.WriteFile(dst, content, 0644)
+		stats2, err := os.Stat(srcFile)
+		if err != nil {
+			return err
+		}
+		if stats.Size() < stats2.Size() {
+			vprint.VV("Bigger %v %v", stats.Size(), stats2.Size())
+			return copyFile(srcFile, dst)
 		}
 	}
 	return nil //simply skip it