diff --git a/cmd/roaring-migrate/main.go b/cmd/roaring-migrate/main.go index 72e6fa94a..f215996ae 100644 --- a/cmd/roaring-migrate/main.go +++ b/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