wired in restore command

api.go

@@ -2235,10 +2235,10 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64
 	if err != nil {
 		return err
 	}
-	dbs.Close()
 	//need to find the path to the db
 	//will not work on blue green
-	finalPath := dbs.W[0].Path()
+	db := dbs.W[0]
+	finalPath := db.Path() + "/data"
 	tempPath := finalPath + ".tmp"
 	vprint.VV("restore to %v", tempPath)
 	o, err := os.OpenFile(tempPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)
@@ -2248,15 +2248,28 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64
 	w := bufio.NewWriter(o)
 	//close db if open
 	vprint.VV("restore index:%v shard:%v", indexName, shard)
-	_, err = io.Copy(w, rd)
+	n, err := io.Copy(w, rd)
+	vprint.VV("Written:%v", n)
 	w.Flush()
 	o.Close()
 	if err != nil {
 		_ = os.Remove(tempPath)
 		return err
 	}
+	err = db.CloseDB()
+	if err != nil {
+		return err
+	}
 	vprint.VV("Rename %v to %v", tempPath, finalPath)
-	return os.Rename(tempPath, finalPath)
+	err = os.Rename(tempPath, finalPath)
+	if err != nil {
+		_ = os.Remove(tempPath)
+		return err
+	}
+	api.holder.recalculateCaches()
+
+	return db.OpenDB()
+
 }
 
 type serverInfo struct {

bolt.go

@@ -217,6 +217,23 @@ func (w *BoltWrapper) HasData() (has bool, err error) {
 func (w *BoltWrapper) CleanupTx(tx Tx) {
 	// inlined into Rollback and Commit, so this is a no-op, just here to satisfy the interface.
 }
+func (w *BoltWrapper) CloseDB() error {
+	w.muDb.Lock()
+	defer w.muDb.Unlock()
+	w.closed = true
+	return w.db.Close()
+}
+func (w *BoltWrapper) OpenDB() error {
+	w.muDb.Lock()
+	defer w.muDb.Unlock()
+	db, err := bolt.Open(w.path, 0666, &bolt.Options{Timeout: 5 * time.Second, InitialMmapSize: TxInitialMmapSize})
+	if err != nil {
+		return err
+	}
+	w.db = db
+	w.closed = false
+	return nil
+}
 
 func (tx *BoltTx) IsDone() (done bool) {
 	return atomic.LoadInt64(&tx.unlocked) == 1

cmd/root.go

@@ -64,6 +64,7 @@ at https://www.pilosa.com/docs/.
 	rc.PersistentFlags().StringP("config", "c", "", "Configuration file to read from.")
 
 	rc.AddCommand(newBackupCommand(stdin, stdout, stderr))
+	rc.AddCommand(newRestoreCommand(stdin, stdout, stderr))
 	rc.AddCommand(newCheckCommand(stdin, stdout, stderr))
 	rc.AddCommand(newConfigCommand(stdin, stdout, stderr))
 	rc.AddCommand(newExportCommand(stdin, stdout, stderr))

ctl/restore.go

@@ -16,17 +16,15 @@ package ctl
 
 import (
 	"archive/tar"
-	"bytes"
 	"compress/gzip"
 	"context"
 	"io"
 	"os"
 	"strings"
 
-	gohttp "net/http"
-
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/http"
+	"github.com/pilosa/pilosa/v2/vprint"
 )
 
 // RestoreCommand represents a command for restoring a backup to
@@ -41,16 +39,18 @@ type RestoreCommand struct {
 
 // NewRestoreCommand returns a new instance of RestoreCommand.
 func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand {
-	h := &gohttp.Client{}
-	host := "SOMETHING"
-	c, err := http.NewInternalClient(host, h)
-	if err != nil {
-		panic(err)
-	}
+	/*
+		h := &gohttp.Client{}
+		host := "SOMETHING"
+		c, err := http.NewInternalClient(host, h)
+		if err != nil {
+			panic(err)
+		}
+	*/
 
 	return &RestoreCommand{
-		CmdIO:  pilosa.NewCmdIO(stdin, stdout, stderr),
-		client: c,
+		CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
+		//		client: c,
 	}
 }
 
@@ -95,14 +95,16 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error {
 		tarReader = tar.NewReader(f)
 	}
 	//maybe begin transaction?
-	schemaJson := readSchema(cmd.Path)
-	//Push the schema from the archive into the cluster belonging to the host.
-	client := &gohttp.Client{}
-	url := "FIXME"
-	_, err = client.Post(url+"/schema", "application/json", bytes.NewBufferString(schemaJson))
-	if err != nil {
-		return err
-	}
+	/*
+		schemaJson := readSchema(cmd.Path)
+		//Push the schema from the archive into the cluster belonging to the host.
+		client := &gohttp.Client{}
+		url := "FIXME"
+		_, err = client.Post(url+"/schema", "application/json", bytes.NewBufferString(schemaJson))
+		if err != nil {
+			return err
+		}
+	*/
 	//TODO (twg) load schema
 	//TODO (twg) load rbf shard
 	//TODO (twg) load row keys
@@ -113,11 +115,44 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error {
 
 	for {
 		header, err := tarReader.Next()
-		//fmt.Println("What %v", header.Name)
-		_ = header
 		if err == io.EOF {
 			break
 		}
+		record := strings.Split(header.Name, "/")
+		if len(record) == 1 {
+			switch record[0] {
+			case "schema":
+				vprint.VV("Load Schema")
+			case "idalloc":
+				vprint.VV("Load ids")
+			default:
+				panic("UNKNOWN " + record[0])
+
+			}
+			continue
+		}
+		indexName := record[1]
+		switch record[2] {
+		case "shards":
+			shard := record[3]
+			vprint.VV("shard %v %v", shard, indexName)
+		case "translate":
+			vprint.VV("column keys %v", indexName)
+		case "attributes":
+			vprint.VV("column attributes %v", indexName)
+		case "fields":
+			fieldName := record[3]
+			switch action := record[4]; action {
+			case "translate":
+				vprint.VV("field keys %v %v", indexName, fieldName)
+			case "attributes":
+				vprint.VV("field attributes %v %v", indexName, fieldName)
+			default:
+				panic("unknown:" + action)
+			}
+
+		}
+
 	}
 	/*	Fetch the cluster nodes from the target host.
 		For each index:

dbshard.go

@@ -66,6 +66,9 @@ type DBWrapper interface {
 	Path() string
 	HasData() (has bool, err error)
 	SetHolder(h *Holder)
+	//needed for restore
+	CloseDB() error
+	OpenDB() error
 }
 
 type DBRegistry interface {

fragment.go

@@ -2909,6 +2909,9 @@ func (f *fragment) flushCache() error {
 		return errors.Wrap(err, "marshalling")
 	}
 
+	if err := os.MkdirAll(filepath.Dir(f.cachePath()), 0777); err != nil {
+		return errors.Wrap(err, "mkdir")
+	}
 	// Write to disk.
 	if err := ioutil.WriteFile(f.cachePath(), buf, 0666); err != nil {
 		return errors.Wrap(err, "writing")

lattice

@@ -1 +1 @@
-Subproject commit 7ea3d77f89771a06cbe59867f9135436fc8ea3b6
+Subproject commit d4bada428e45823a70432321843c38fcf2a384a0

rbf.go

@@ -571,6 +571,26 @@ func (w *RbfDBWrapper) Close() error {
 	return w.db.Close()
 }
 
+// needed to handle the special case on reload, the close method unregisters the wrapper and all that is
+// required is the backing file get reloaded
+
+func (w *RbfDBWrapper) CloseDB() error {
+	w.muDb.Lock()
+	defer w.muDb.Unlock()
+	w.closed = true
+	return w.db.Close()
+}
+func (w *RbfDBWrapper) OpenDB() error {
+	w.muDb.Lock()
+	defer w.muDb.Unlock()
+	err := w.db.Open()
+	if err != nil {
+		return err
+	}
+	w.closed = false
+	return nil
+}
+
 var globalNextTxSnRBFTx int64
 
 func (w *RbfDBWrapper) NewTx(write bool, initialIndex string, o Txo) (_ Tx, err error) {

rrtx.go

@@ -663,6 +663,13 @@ func (w *RoaringWrapper) OpenSnList() (slc []int64) {
 	return nil
 }
 
+func (w *RoaringWrapper) CloseDB() error {
+	return errors.New("CloseDB not supported in roaring")
+}
+func (w *RoaringWrapper) OpenDB() error {
+	return errors.New("OpenDB not supported in roaring")
+}
+
 // statically confirm that RoaringTx satisfies the Tx interface.
 var _ Tx = (*RoaringTx)(nil)