Merge pull request #1635 from molecula/backup-dir

CORE-638: Refactor backup/restore to use directory archive

cmd/backup.go

@@ -36,7 +36,8 @@ Backs up a pilosa server to a local, tar-formatted snapshot file.
 	}
 
 	flags := ccmd.Flags()
-	flags.StringVarP(&cmd.OutputPath, "output", "o", "", "output path to write to; specify '-' to send to stdout")
+	flags.StringVarP(&cmd.OutputDir, "output", "o", "", "output dir to write to")
+	flags.BoolVar(&cmd.NoSync, "no-sync", false, "disable file sync")
 	flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of Pilosa.")
 	ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification)
 	return ccmd

ctl/backup.go

@@ -15,17 +15,14 @@
 package ctl
 
 import (
-	"archive/tar"
-	"bytes"
 	"context"
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
-	"path"
 	"path/filepath"
-	"time"
 
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/http"
@@ -41,7 +38,10 @@ type BackupCommand struct { // nolint: maligned
 	Host string `json:"host"`
 
 	// Path to write the backup to.
-	OutputPath string
+	OutputDir string
+
+	// If true, skips file sync.
+	NoSync bool
 
 	// Reusable client.
 	client pilosa.InternalClient
@@ -59,21 +59,12 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand
 	}
 }
 
-// TempPath returns the path to the temporary file to write the archive to.
-func (cmd *BackupCommand) TempPath() string {
-	dir, base := filepath.Split(cmd.OutputPath)
-	return filepath.Join(dir, "."+base)
-}
-
 // Run executes the main program execution.
 func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
-	logger := cmd.Logger()
-
 	// Validate arguments.
-	if cmd.OutputPath == "" {
+	if cmd.OutputDir == "" {
 		return fmt.Errorf("-o flag required")
 	}
-	useStdout := cmd.OutputPath == "-"
 
 	// Parse TLS configuration for node-specific clients.
 	tls := cmd.TLSConfiguration()
@@ -95,46 +86,23 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
 	}
 	schema := &pilosa.Schema{Indexes: indexes}
 
-	// Create output file in temporary location, or send to stdout if a dash is specified.
-	var w io.Writer
-	if useStdout {
-		w = os.Stdout
-	} else {
-		f, err := os.Create(cmd.OutputPath + ".tmp")
-		if err != nil {
-			return err
-		}
-		defer f.Close()
-		w = f
+	// Ensure output directory doesn't exist; then create output directory.
+	if _, err := os.Stat(cmd.OutputDir); !os.IsNotExist(err) {
+		return fmt.Errorf("output directory already exists")
+	} else if err := os.MkdirAll(cmd.OutputDir, 0777); err != nil {
+		return err
 	}
 
-	// Open a tar writer to the temporary file.
-	tw := tar.NewWriter(w)
-	defer tw.Close()
-
 	// Backup schema.
-	if err := cmd.backupSchema(ctx, tw, schema); err != nil {
+	if err := cmd.backupSchema(ctx, schema); err != nil {
 		return fmt.Errorf("cannot back up schema: %w", err)
-	} else if err := cmd.backupIDAllocData(ctx, tw); err != nil {
+	} else if err := cmd.backupIDAllocData(ctx); err != nil {
 		return fmt.Errorf("cannot back up id alloc data: %w", err)
 	}
 
 	// Backup data for each index.
 	for _, ii := range schema.Indexes {
-		if err := cmd.backupIndex(ctx, tw, ii); err != nil {
-			return err
-		}
-	}
-
-	// Close archive.
-	if err := tw.Close(); err != nil {
-		return err
-	}
-
-	// Move data file to final location.
-	if !useStdout {
-		logger.Printf("writing backup: %s", cmd.OutputPath)
-		if err := os.Rename(cmd.OutputPath+".tmp", cmd.OutputPath); err != nil {
+		if err := cmd.backupIndex(ctx, ii); err != nil {
 			return err
 		}
 	}
@@ -143,7 +111,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
 }
 
 // backupSchema writes the schema to the archive.
-func (cmd *BackupCommand) backupSchema(ctx context.Context, tw *tar.Writer, schema *pilosa.Schema) error {
+func (cmd *BackupCommand) backupSchema(ctx context.Context, schema *pilosa.Schema) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up schema")
 
@@ -152,22 +120,14 @@ func (cmd *BackupCommand) backupSchema(ctx context.Context, tw *tar.Writer, sche
 		return fmt.Errorf("marshaling schema: %w", err)
 	}
 
-	// Build header & copy data to archive.
-	if err = tw.WriteHeader(&tar.Header{
-		Name:    "schema",
-		Mode:    0666,
-		Size:    int64(len(buf)),
-		ModTime: time.Now(),
-	}); err != nil {
-		return err
-	} else if _, err := tw.Write(buf); err != nil {
-		return fmt.Errorf("copying schema to archive: %w", err)
+	if err := ioutil.WriteFile(filepath.Join(cmd.OutputDir, "schema"), buf, 0666); err != nil {
+		return fmt.Errorf("writing schema: %w", err)
 	}
 
 	return nil
 }
 
-func (cmd *BackupCommand) backupIDAllocData(ctx context.Context, tw *tar.Writer) error {
+func (cmd *BackupCommand) backupIDAllocData(ctx context.Context) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up id alloc data")
 
@@ -177,29 +137,22 @@ func (cmd *BackupCommand) backupIDAllocData(ctx context.Context, tw *tar.Writer)
 	}
 	defer rc.Close()
 
-	// Read to buffer to determine size.
-	var buf bytes.Buffer
-	if _, err := buf.ReadFrom(rc); err != nil {
-		return fmt.Errorf("copying id alloc data to memory: %w", err)
-	}
-
-	// Build header & copy data to archive.
-	if err = tw.WriteHeader(&tar.Header{
-		Name:    "idalloc",
-		Mode:    0666,
-		Size:    int64(buf.Len()),
-		ModTime: time.Now(),
-	}); err != nil {
+	f, err := os.Create(filepath.Join(cmd.OutputDir, "idalloc"))
+	if err != nil {
 		return err
-	} else if _, err := io.Copy(tw, &buf); err != nil {
-		return fmt.Errorf("copying id alloc data to archive: %w", err)
 	}
+	defer f.Close()
 
-	return nil
+	if _, err := io.Copy(f, rc); err != nil {
+		return err
+	} else if err := cmd.syncFile(f); err != nil {
+		return err
+	}
+	return f.Close()
 }
 
 // backupIndex backs up all shards for a given index.
-func (cmd *BackupCommand) backupIndex(ctx context.Context, tw *tar.Writer, ii *pilosa.IndexInfo) error {
+func (cmd *BackupCommand) backupIndex(ctx context.Context, ii *pilosa.IndexInfo) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up index: %q", ii.Name)
 
@@ -210,19 +163,19 @@ func (cmd *BackupCommand) backupIndex(ctx context.Context, tw *tar.Writer, ii *p
 
 	// Back up all bitmap data for the index.
 	for _, shard := range shards {
-		if err := cmd.backupShard(ctx, tw, ii.Name, shard); err != nil {
+		if err := cmd.backupShard(ctx, ii.Name, shard); err != nil {
 			return fmt.Errorf("cannot backup shard %d on index %q: %w", shard, ii.Name, err)
 		}
 	}
 
 	// Back up translation data after bitmap data so we ensure we can translate all data.
-	if err := cmd.backupIndexTranslateData(ctx, tw, ii.Name); err != nil {
+	if err := cmd.backupIndexTranslateData(ctx, ii.Name); err != nil {
 		return err
 	}
 
 	// Back up field translation data.
 	for _, fi := range ii.Fields {
-		if err := cmd.backupFieldTranslateData(ctx, tw, ii.Name, fi.Name); err != nil {
+		if err := cmd.backupFieldTranslateData(ctx, ii.Name, fi.Name); err != nil {
 			return fmt.Errorf("cannot backup field translation data for field %q on index %q: %w", fi.Name, ii.Name, err)
 		}
 	}
@@ -231,7 +184,7 @@ func (cmd *BackupCommand) backupIndex(ctx context.Context, tw *tar.Writer, ii *p
 }
 
 // backupShard backs up a single shard from a single index.
-func (cmd *BackupCommand) backupShard(ctx context.Context, tw *tar.Writer, indexName string, shard uint64) (err error) {
+func (cmd *BackupCommand) backupShard(ctx context.Context, indexName string, shard uint64) (err error) {
 	nodes, err := cmd.client.FragmentNodes(ctx, indexName, shard)
 	if err != nil {
 		return fmt.Errorf("cannot determine fragment nodes: %w", err)
@@ -240,7 +193,7 @@ func (cmd *BackupCommand) backupShard(ctx context.Context, tw *tar.Writer, index
 	}
 
 	for _, node := range nodes {
-		if e := cmd.backupShardNode(ctx, tw, indexName, shard, node); e == nil {
+		if e := cmd.backupShardNode(ctx, indexName, shard, node); e == nil {
 			return nil // backup ok, exit
 		} else if err == nil {
 			err = e // save first error, try next node
@@ -250,12 +203,10 @@ func (cmd *BackupCommand) backupShard(ctx context.Context, tw *tar.Writer, index
 }
 
 // backupShardNode backs up a single shard from a single index on a specific node.
-func (cmd *BackupCommand) backupShardNode(ctx context.Context, tw *tar.Writer, indexName string, shard uint64, node *topology.Node) error {
+func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, shard uint64, node *topology.Node) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
 
-	filename := path.Join("indexes", indexName, "shards", fmt.Sprintf("%04d", shard))
-
 	client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig))
 	rc, err := client.ShardReader(ctx, indexName, shard)
 	if err != nil {
@@ -263,40 +214,37 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, tw *tar.Writer, i
 	}
 	defer rc.Close()
 
-	// Read to buffer to determine size.
-	// TODO: Provide size via the reader itself.
-	var buf bytes.Buffer
-	if _, err := buf.ReadFrom(rc); err != nil {
-		return fmt.Errorf("copying shard data to memory: %w", err)
-	}
-
-	// Build header & copy data to archive.
-	if err = tw.WriteHeader(&tar.Header{
-		Name:    filename,
-		Mode:    0666,
-		Size:    int64(buf.Len()),
-		ModTime: time.Now(),
-	}); err != nil {
+	filename := filepath.Join(cmd.OutputDir, "indexes", indexName, "shards", fmt.Sprintf("%04d", shard))
+	if err := os.MkdirAll(filepath.Dir(filename), 0777); err != nil {
 		return err
-	} else if _, err := io.Copy(tw, &buf); err != nil {
-		return fmt.Errorf("copying shard data to archive: %w", err)
 	}
 
-	return nil
+	f, err := os.Create(filename)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	if _, err := io.Copy(f, rc); err != nil {
+		return err
+	} else if err := cmd.syncFile(f); err != nil {
+		return err
+	}
+	return f.Close()
 }
 
-func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, tw *tar.Writer, name string) error {
+func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, name string) error {
 	// TODO: Fetch holder partition count.
 	partitionN := topology.DefaultPartitionN
 	for partitionID := 0; partitionID < partitionN; partitionID++ {
-		if err := cmd.backupIndexPartitionTranslateData(ctx, tw, name, partitionID); err != nil {
+		if err := cmd.backupIndexPartitionTranslateData(ctx, name, partitionID); err != nil {
 			return fmt.Errorf("cannot backup index translation data for partition %d on %q: %w", partitionID, name, err)
 		}
 	}
 	return nil
 }
 
-func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context, tw *tar.Writer, name string, partitionID int) error {
+func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context, name string, partitionID int) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up index translation data: %s/%d", name, partitionID)
 
@@ -308,28 +256,26 @@ func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context,
 	}
 	defer rc.Close()
 
-	// Read to buffer to determine size.
-	var buf bytes.Buffer
-	if _, err := buf.ReadFrom(rc); err != nil {
-		return fmt.Errorf("copying translate data to memory: %w", err)
-	}
-
-	// Build header & copy data to archive.
-	if err = tw.WriteHeader(&tar.Header{
-		Name:    path.Join("indexes", name, "translate", fmt.Sprintf("%04d", partitionID)),
-		Mode:    0666,
-		Size:    int64(buf.Len()),
-		ModTime: time.Now(),
-	}); err != nil {
+	filename := filepath.Join(cmd.OutputDir, "indexes", name, "translate", fmt.Sprintf("%04d", partitionID))
+	if err := os.MkdirAll(filepath.Dir(filename), 0777); err != nil {
 		return err
-	} else if _, err := io.Copy(tw, &buf); err != nil {
-		return fmt.Errorf("copying translate data to archive: %w", err)
 	}
 
-	return nil
+	f, err := os.Create(filename)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	if _, err := io.Copy(f, rc); err != nil {
+		return err
+	} else if err := cmd.syncFile(f); err != nil {
+		return err
+	}
+	return f.Close()
 }
 
-func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, tw *tar.Writer, indexName, fieldName string) error {
+func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, indexName, fieldName string) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up field translation data: %s/%s", indexName, fieldName)
 
@@ -341,24 +287,30 @@ func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, tw *tar.
 	}
 	defer rc.Close()
 
-	// Read to buffer to determine size.
-	var buf bytes.Buffer
-	if _, err := buf.ReadFrom(rc); err != nil {
-		return fmt.Errorf("copying translate data to memory: %w", err)
+	filename := filepath.Join(cmd.OutputDir, "indexes", indexName, "fields", fieldName, "translate")
+	if err := os.MkdirAll(filepath.Dir(filename), 0777); err != nil {
+		return err
 	}
 
-	// Build header & copy data to archive.
-	if err = tw.WriteHeader(&tar.Header{
-		Name:    path.Join("indexes", indexName, "fields", fieldName, "translate"),
-		Mode:    0666,
-		Size:    int64(buf.Len()),
-		ModTime: time.Now(),
-	}); err != nil {
+	f, err := os.Create(filename)
+	if err != nil {
 		return err
-	} else if _, err := io.Copy(tw, &buf); err != nil {
-		return fmt.Errorf("copying translate data to archive: %w", err)
 	}
-	return nil
+	defer f.Close()
+
+	if _, err := io.Copy(f, rc); err != nil {
+		return err
+	} else if err := cmd.syncFile(f); err != nil {
+		return err
+	}
+	return f.Close()
+}
+
+func (cmd *BackupCommand) syncFile(f *os.File) error {
+	if cmd.NoSync {
+		return nil
+	}
+	return f.Sync()
 }
 
 func (cmd *BackupCommand) TLSHost() string { return cmd.Host }

ctl/restore.go

@@ -15,17 +15,14 @@
 package ctl
 
 import (
-	"archive/tar"
-	"bytes"
-	"compress/gzip"
 	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
-	gohttp "net/http"
+	"net/http"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 
@@ -65,19 +62,6 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 	if cmd.Path == "" {
 		return fmt.Errorf("-s flag required")
 	}
-	useStdin := cmd.Path == "-"
-
-	var f *os.File
-	// read from Stdin if path specified as -
-	if useStdin {
-		f = os.Stdin
-	} else {
-		f, err = os.Open(cmd.Path)
-		if err != nil {
-			return (err)
-		}
-		defer f.Close()
-	}
 
 	// Parse TLS configuration for node-specific clients.
 	tls := cmd.TLSConfiguration()
@@ -90,16 +74,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 		return fmt.Errorf("creating client: %w", err)
 	}
 	cmd.client = client
-	var tarReader *tar.Reader
-	if strings.HasSuffix(cmd.Path, "gz") {
-		gzf, err := gzip.NewReader(f)
-		if err != nil {
-			return err
-		}
-		tarReader = tar.NewReader(gzf)
-	} else {
-		tarReader = tar.NewReader(f)
-	}
+
 	nodes, err := cmd.client.Nodes(ctx)
 	if err != nil {
 		return err
@@ -116,126 +91,21 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 	if primary == nil {
 		return errors.New("no primary")
 	}
-	c := &gohttp.Client{}
-	for {
-		header, err := tarReader.Next()
-		if err == io.EOF {
-			break
-		}
-		record := strings.Split(header.Name, "/")
-		if len(record) == 1 {
-			switch record[0] {
-			case "schema":
-				logger.Printf("Load Schema")
-				url := primary.URI.Path("/schema")
-				_, err = c.Post(url, "application/json", tarReader)
-				if err != nil {
-					return err
-				}
-			case "idalloc":
-				logger.Printf("Load ids")
-				url := primary.URI.Path("/internal/idalloc/restore")
-				_, err = c.Post(url, "application/octet-stream", tarReader)
-				if err != nil {
-					return err
-				}
-			default:
-				return err
-
-			}
-			continue
-		}
-		indexName := record[1]
-		switch record[2] {
-		case "shards":
-			shard, err := strconv.ParseUint(record[3], 10, 64)
-			if err != nil {
-				return err
-			}
-			nodes, err := cmd.client.FragmentNodes(ctx, indexName, shard)
-			if err != nil {
-				return fmt.Errorf("cannot determine fragment nodes: %w", err)
-			} else if len(nodes) == 0 {
-				return fmt.Errorf("no nodes available")
-			}
-
-			shardBytes, err := ioutil.ReadAll(tarReader) // this feels wrong but works for now
-			if err != nil {
-				return err
-			}
-			g, _ := errgroup.WithContext(ctx)
-			for _, node := range nodes {
-				node := node
-				g.Go(func() error {
-					client := &gohttp.Client{}
-					rd := bytes.NewReader(shardBytes)
-					logger.Printf("shard %v %v", shard, indexName)
-					url := node.URI.Path(fmt.Sprintf("/internal/restore/%v/%v", indexName, shard))
-					_, err = client.Post(url, "application/octet-stream", rd)
-					return err
-				})
-			}
-			if err := g.Wait(); err != nil {
-				return err
-			}
-		case "translate":
-			partitionID, err := strconv.Atoi(record[3])
-			logger.Printf("column keys %v (%v)", indexName, partitionID)
-			if err != nil {
-				return err
-			}
-			partitionNodes, err := cmd.client.PartitionNodes(ctx, partitionID)
-			if err != nil {
-				return err
-			}
-			shardBytes, err := ioutil.ReadAll(tarReader) // this feels wrong but works for now
-			if err != nil {
-				return err
-			}
-			g, _ := errgroup.WithContext(ctx)
-			for _, node := range partitionNodes {
-				node := node
-				g.Go(func() error {
-					rd := bytes.NewReader(shardBytes)
-					return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, rd)
-				})
-			}
-			if err := g.Wait(); err != nil {
-				return err
-			}
-
-		case "attributes":
-			//skip
-		case "fields":
-			fieldName := record[3]
-			switch action := record[4]; action {
-			case "translate":
-				logger.Printf("field keys %v %v", indexName, fieldName)
-				//needs to go to all nodes
-				shardBytes, err := ioutil.ReadAll(tarReader) // this feels wrong but works for now
-				if err != nil {
-					return err
-				}
-				g, _ := errgroup.WithContext(ctx)
-				for _, node := range nodes {
-					node := node
-					g.Go(func() error {
-						rd := bytes.NewReader(shardBytes)
-						return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, rd)
-					})
-				}
-				if err := g.Wait(); err != nil {
-					return err
-				}
-			case "attributes":
-			//skip
-			default:
-				return fmt.Errorf("unknown restore action: %v", action)
-			}
-
-		}
 
+	if err := cmd.restoreSchema(ctx, primary); err != nil {
+		return fmt.Errorf("cannot restore schema: %w", err)
+	} else if err := cmd.restoreIDAlloc(ctx, primary); err != nil {
+		return fmt.Errorf("cannot restore idalloc: %w", err)
 	}
+
+	if err := cmd.restoreShards(ctx); err != nil {
+		return fmt.Errorf("cannot restore shards: %w", err)
+	} else if err := cmd.restoreIndexTranslation(ctx); err != nil {
+		return fmt.Errorf("cannot restore index translation: %w", err)
+	} else if err := cmd.restoreFieldTranslation(ctx, nodes); err != nil {
+		return fmt.Errorf("cannot restore field translation: %w", err)
+	}
+
 	/*	Fetch the cluster nodes from the target host.
 		For each index:
 		Upload the RBF snapshot for each shard to the nodes that own the shard.
@@ -245,6 +115,199 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
 
 	return nil
 }
+
+func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.Node) error {
+	f, err := os.Open(filepath.Join(cmd.Path, "schema"))
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	cmd.Logger().Printf("Load Schema")
+	url := primary.URI.Path("/schema")
+
+	var client http.Client
+	_, err = client.Post(url, "application/json", f)
+	return err
+}
+
+func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error {
+	logger := cmd.Logger()
+
+	f, err := os.Open(filepath.Join(cmd.Path, "idalloc"))
+	if os.IsNotExist(err) {
+		logger.Printf("No idalloc, skipping")
+		return nil
+	} else if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	logger.Printf("Load idalloc")
+	url := primary.URI.Path("/internal/idalloc/restore")
+
+	var client http.Client
+	_, err = client.Post(url, "application/octet-stream", f)
+	return err
+}
+
+func (cmd *RestoreCommand) restoreShards(ctx context.Context) error {
+	logger := cmd.Logger()
+
+	filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "shards", "*"))
+	if err != nil {
+		return err
+	}
+
+	for _, filename := range filenames {
+		rel, err := filepath.Rel(cmd.Path, filename)
+		if err != nil {
+			return err
+		}
+
+		record := strings.Split(rel, string(os.PathSeparator))
+		indexName := record[1]
+
+		shard, err := strconv.ParseUint(record[3], 10, 64)
+		if err != nil {
+			continue
+		}
+
+		nodes, err := cmd.client.FragmentNodes(ctx, indexName, shard)
+		if err != nil {
+			return fmt.Errorf("cannot determine fragment nodes: %w", err)
+		} else if len(nodes) == 0 {
+			return fmt.Errorf("no nodes available")
+		}
+
+		g, ctx := errgroup.WithContext(ctx)
+		for _, node := range nodes {
+			node := node
+			g.Go(func() error {
+				logger.Printf("shard %v %v", shard, indexName)
+
+				f, err := os.Open(filename)
+				if err != nil {
+					return err
+				}
+				defer f.Close()
+
+				url := node.URI.Path(fmt.Sprintf("/internal/restore/%v/%v", indexName, shard))
+				req, err := http.NewRequest("POST", url, f)
+				if err != nil {
+					return err
+				}
+				req = req.WithContext(ctx)
+				req.Header.Set("Content-Type", "application/octet-stream")
+
+				var client http.Client
+				resp, err := client.Do(req)
+				if err != nil {
+					return err
+				} else if err := resp.Body.Close(); err != nil {
+					return err
+				} else if resp.StatusCode != http.StatusOK {
+					return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+				}
+				return nil
+			})
+		}
+		if err := g.Wait(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (cmd *RestoreCommand) restoreIndexTranslation(ctx context.Context) error {
+	logger := cmd.Logger()
+
+	filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "translate", "*"))
+	if err != nil {
+		return err
+	}
+
+	for _, filename := range filenames {
+		rel, err := filepath.Rel(cmd.Path, filename)
+		if err != nil {
+			return err
+		}
+
+		record := strings.Split(rel, string(os.PathSeparator))
+		indexName := record[1]
+
+		partitionID, err := strconv.Atoi(record[3])
+		if err != nil {
+			return err
+		}
+		logger.Printf("column keys %v (%v)", indexName, partitionID)
+
+		nodes, err := cmd.client.PartitionNodes(ctx, partitionID)
+		if err != nil {
+			return err
+		}
+
+		g, ctx := errgroup.WithContext(ctx)
+		for _, node := range nodes {
+			node := node
+			g.Go(func() error {
+				f, err := os.Open(filename)
+				if err != nil {
+					return err
+				}
+				defer f.Close()
+
+				return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, f)
+			})
+		}
+		if err := g.Wait(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []*topology.Node) error {
+	logger := cmd.Logger()
+
+	filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "fields", "*", "translate"))
+	if err != nil {
+		return err
+	}
+
+	for _, filename := range filenames {
+		rel, err := filepath.Rel(cmd.Path, filename)
+		if err != nil {
+			return err
+		}
+
+		record := strings.Split(rel, string(os.PathSeparator))
+		indexName, fieldName := record[1], record[3]
+
+		logger.Printf("field keys %v %v", indexName, fieldName)
+
+		g, ctx := errgroup.WithContext(ctx)
+		for _, node := range nodes {
+			node := node
+			g.Go(func() error {
+				f, err := os.Open(filename)
+				if err != nil {
+					return err
+				}
+				defer f.Close()
+
+				return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, f)
+			})
+		}
+		if err := g.Wait(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (cmd *RestoreCommand) TLSHost() string { return cmd.Host }
 
 func (cmd *RestoreCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS }