feat(bigtable): allow restore backup to different instance (#3489)

* feat: restore table to a different instance

* add check that table restored

* review fixes

* change instance config zone

bigtable/admin.go

@@ -115,6 +115,10 @@ func (ac *AdminClient) instancePrefix() string {
 	return fmt.Sprintf("projects/%s/instances/%s", ac.project, ac.instance)
 }
 
+func (ac *AdminClient) backupPath(cluster, backup string) string {
+	return fmt.Sprintf("projects/%s/instances/%s/clusters/%s/backups/%s", ac.project, ac.instance, cluster, backup)
+}
+
 // Tables returns a list of the tables in the instance.
 func (ac *AdminClient) Tables(ctx context.Context) ([]string, error) {
 	ctx = mergeOutgoingMetadata(ctx, ac.md)
@@ -1441,10 +1445,17 @@ func UpdateInstanceAndSyncClusters(ctx context.Context, iac *InstanceAdminClient
 
 // RestoreTable creates a table from a backup. The table will be created in the same cluster as the backup.
 func (ac *AdminClient) RestoreTable(ctx context.Context, table, cluster, backup string) error {
-	ctx = mergeOutgoingMetadata(ctx, ac.md)
-	prefix := ac.instancePrefix()
-	backupPath := prefix + "/clusters/" + cluster + "/backups/" + backup
+	return ac.RestoreTableTo(ctx, ac.instance, table, cluster, backup)
+}
 
+// RestoreTableTo creates a new table by restoring from this completed backup.
+//
+// diffInstance is an instance in which the new table will be restored to.
+// Instance must be in the same project as the project containing backup.
+func (ac *AdminClient) RestoreTableTo(ctx context.Context, diffInstance, table, cluster, backup string) error {
+	ctx = mergeOutgoingMetadata(ctx, ac.md)
+	prefix := "projects/" + ac.project + "/instances/" + diffInstance
+	backupPath := ac.backupPath(cluster, backup)
 	req := &btapb.RestoreTableRequest{
 		Parent:  prefix,
 		TableId: table,

bigtable/integration_test.go

@@ -2033,6 +2033,26 @@ func TestIntegration_AdminBackup(t *testing.T) {
 	table := testEnv.Config().Table
 	cluster := testEnv.Config().Cluster
 
+	iAdminClient, err := testEnv.NewInstanceAdminClient()
+	if err != nil {
+		t.Fatalf("NewInstanceAdminClient: %v", err)
+	}
+	defer iAdminClient.Close()
+	diffInstance := testEnv.Config().Instance + "-diff"
+	diffCluster := cluster + "-diff"
+	conf := &InstanceConf{
+		InstanceId:   diffInstance,
+		ClusterId:    diffCluster,
+		DisplayName:  "different test instance",
+		Zone:         instanceToCreateZone2,
+		InstanceType: DEVELOPMENT,
+		Labels:       map[string]string{"test-label-key": "test-label-value"},
+	}
+	defer iAdminClient.DeleteInstance(ctx, diffInstance)
+	// Create different instance to restore table.
+	if err := iAdminClient.CreateInstance(ctx, conf); err != nil {
+		t.Fatalf("CreateInstance: %v", err)
+	}
 	// Delete the table at the end of the test. Schedule ahead of time
 	// in case the client fails
 	defer deleteTable(ctx, t, adminClient, table)
@@ -2128,6 +2148,31 @@ func TestIntegration_AdminBackup(t *testing.T) {
 	if _, err := adminClient.TableInfo(ctx, restoredTable); err != nil {
 		t.Fatalf("Restored TableInfo: %v", err)
 	}
+	// Restore backup to different instance
+	diffTable := table + "-diff-restored"
+	diffConf := IntegrationTestConfig{
+		Project:  testEnv.Config().Project,
+		Instance: diffInstance,
+		Cluster:  diffCluster,
+		Table:    diffTable,
+	}
+	env := &ProdEnv{
+		config: diffConf,
+	}
+	dAdminClient, err := env.NewAdminClient()
+	if err != nil {
+		t.Fatalf("NewAdminClient: %v", err)
+	}
+	defer dAdminClient.Close()
+
+	defer deleteTable(ctx, t, dAdminClient, diffTable)
+	if err = adminClient.RestoreTableTo(ctx, diffInstance, diffTable, cluster, "mybackup"); err != nil {
+		t.Fatalf("RestoreTableTo: %v", err)
+	}
+	_, err = dAdminClient.TableInfo(ctx, diffTable)
+	if err != nil {
+		t.Fatalf("Restored to different instance TableInfo: %v", err)
+	}
 
 	// Delete backup
 	if err = adminClient.DeleteBackup(ctx, cluster, "mybackup"); err != nil {