backupccl: support restoring views with into_db

The `RESTORE ... into_db = 'foo'` option allows overriding the
destination database for all restored tables, views and sequences.

Previously this option could not be used when restoring views, as the
view query string could contain references to tables qualified with
their original database name(s), and we did not support rewriting that
query to reflect the overridden destination for those tables.

This change adds that rewriting (except for tables/sequences named in
strings, see cockroachdb#24556).

Since a destination override is applied to all tables being restored,
and since restoring a view requires restoring all tables it references,
any table referenced by the view query -- regardless of its prior
qualification -- can be re-qualified with the override DB.

Release note (enterprise change): support restoring views when using the 'into_db' option.

pkg/ccl/backupccl/backup_test.go

@@ -1192,7 +1192,7 @@ func TestBackupRestoreCrossTableReferences(t *testing.T) {
 		)`)
 
 		// and a few views for good measure.
-		origDB.Exec(t, `CREATE VIEW store.early_customers AS SELECT id from store.customers WHERE id < 5`)
+		origDB.Exec(t, `CREATE VIEW store.early_customers AS SELECT id, email from store.customers WHERE id < 5`)
 		origDB.Exec(t, `CREATE VIEW storestats.ordercounts AS
 			SELECT c.id, c.email, COUNT(o.id)
 			FROM store.customers AS c
@@ -1401,6 +1401,13 @@ func TestBackupRestoreCrossTableReferences(t *testing.T) {
 		tc := testcluster.StartTestCluster(t, singleNode, base.TestClusterArgs{ServerArgs: args})
 		defer tc.Stopper().Stop(context.TODO())
 		db := sqlutils.MakeSQLRunner(tc.Conns[0])
+
+		if _, err := db.DB.Exec(
+			`RESTORE DATABASE storestats FROM $1`, localFoo,
+		); !testutils.IsError(err, `cannot restore "ordercounts" without restoring referenced table`) {
+			t.Fatal(err)
+		}
+
 		db.Exec(t, createStore)
 		db.Exec(t, createStoreStats)
 
@@ -1434,6 +1441,25 @@ func TestBackupRestoreCrossTableReferences(t *testing.T) {
 		}
 
 		db.CheckQueryResults(t, `SELECT * FROM storestats.ordercounts ORDER BY id`, origOrderCounts)
+
+		db.Exec(t, `CREATE DATABASE otherstore`)
+		db.Exec(t, `RESTORE store.* FROM $1 WITH into_db = 'otherstore'`, localFoo)
+		// we want to observe just the view-related errors, not fk errors below.
+		db.Exec(t, `ALTER TABLE otherstore.orders DROP CONSTRAINT fk_customerid_ref_customers`)
+		db.Exec(t, `DROP TABLE otherstore.receipts`)
+
+		if _, err := db.DB.Exec(`DROP TABLE otherstore.customers`); !testutils.IsError(err,
+			`cannot drop relation "customers" because view "early_customers" depends on it`,
+		) {
+			t.Fatal(err)
+		}
+		if _, err := db.DB.Exec(`ALTER TABLE otherstore.customers DROP COLUMN email`); !testutils.IsError(
+			err, `cannot drop column "email" because view "early_customers" depends on it`) {
+			t.Fatal(err)
+		}
+		db.Exec(t, `DROP DATABASE store CASCADE`)
+		db.CheckQueryResults(t, `SELECT * FROM otherstore.early_customers ORDER BY id`, origEarlyCustomers)
+
 	})
 }
 

pkg/ccl/backupccl/restore.go

@@ -29,6 +29,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/jobs"
+	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/types"
@@ -152,6 +153,34 @@ func selectTargets(
 	return matched.descs, matched.requestedDBs, nil
 }
 
+// rewriteViewQueryDBNames rewrites the passed table's ViewQuery replacing all
+// non-empty db qualifiers with `newDB`.
+//
+// TODO: this AST traversal misses tables named in strings (#24556).
+func rewriteViewQueryDBNames(table *sqlbase.TableDescriptor, newDB string) error {
+	stmt, err := parser.ParseOne(table.ViewQuery)
+	if err != nil {
+		return errors.Wrapf(err, "failed to parse underlying query from view %q", table.Name)
+	}
+	// Re-format to change all DB names to `newDB`.
+	f := tree.NewFmtCtxWithBuf(tree.FmtParsable)
+	f.WithReformatTableNames(
+		func(ctx *tree.FmtCtx, t *tree.NormalizableTableName) {
+			tn, err := t.Normalize()
+			if err != nil {
+				return
+			}
+			// empty catalog e.g. ``"".information_schema.tables` should stay empty.
+			if tn.CatalogName != "" {
+				tn.CatalogName = tree.Name(newDB)
+			}
+			ctx.FormatNode(tn)
+		})
+	f.FormatNode(stmt)
+	table.ViewQuery = f.CloseAndGetString()
+	return nil
+}
+
 // allocateTableRewrites determines the new ID and parentID (a "TableRewrite")
 // for each table in sqlDescs and returns a mapping from old ID to said
 // TableRewrite. It first validates that the provided sqlDescs can be restored
@@ -165,7 +194,7 @@ func allocateTableRewrites(
 	opts map[string]string,
 ) (tableRewriteMap, error) {
 	tableRewrites := make(tableRewriteMap)
-	_, renaming := opts[restoreOptIntoDB]
+	overrideDB, renaming := opts[restoreOptIntoDB]
 
 	restoreDBNames := make(map[string]*sqlbase.DatabaseDescriptor, len(restoreDBs))
 	for _, db := range restoreDBs {
@@ -193,10 +222,6 @@ func allocateTableRewrites(
 	// options.
 	// Check that foreign key targets exist.
 	for _, table := range tablesByID {
-		if renaming && table.IsView() {
-			return nil, errors.Errorf("cannot restore view when using %q option", restoreOptIntoDB)
-		}
-
 		if err := table.ForeachNonDropIndex(func(index *sqlbase.IndexDescriptor) error {
 			if index.ForeignKey.IsSet() {
 				to := index.ForeignKey.Table
@@ -247,8 +272,8 @@ func allocateTableRewrites(
 
 		for _, table := range tablesByID {
 			var targetDB string
-			if override, ok := opts[restoreOptIntoDB]; ok {
-				targetDB = override
+			if renaming {
+				targetDB = overrideDB
 			} else {
 				database, ok := databasesByID[table.ParentID]
 				if !ok {
@@ -366,12 +391,24 @@ func CheckTableExists(
 // rewriteTableDescs mutates tables to match the ID and privilege specified in
 // tableRewrites, as well as adjusting cross-table references to use the new
 // IDs.
-func rewriteTableDescs(tables []*sqlbase.TableDescriptor, tableRewrites tableRewriteMap) error {
+func rewriteTableDescs(
+	tables []*sqlbase.TableDescriptor, tableRewrites tableRewriteMap, overrideDB string,
+) error {
 	for _, table := range tables {
 		tableRewrite, ok := tableRewrites[table.ID]
 		if !ok {
 			return errors.Errorf("missing table rewrite for table %d", table.ID)
 		}
+		if table.IsView() && overrideDB != "" {
+			// restore checks that all dependencies are also being restored, but if
+			// the restore is overriding the destination database, qualifiers in the
+			// view query string may be wrong. Since the destination override is
+			// applied to everything being restored, anything the view query
+			// references will be in the override DB post-restore, so all database
+			// qualifiers in the view query should be replaced with overrideDB.
+			rewriteViewQueryDBNames(table, overrideDB)
+		}
+
 		table.ID = tableRewrite.TableID
 		table.ParentID = tableRewrite.ParentID
 
@@ -914,6 +951,7 @@ func restore(
 	endTime hlc.Timestamp,
 	sqlDescs []sqlbase.Descriptor,
 	tableRewrites tableRewriteMap,
+	overrideDB string,
 	job *jobs.Job,
 	resultsCh chan<- tree.Datums,
 ) (roachpb.BulkOpSummary, []*sqlbase.DatabaseDescriptor, []*sqlbase.TableDescriptor, error) {
@@ -954,7 +992,7 @@ func restore(
 
 	// Assign new IDs and privileges to the tables, and update all references to
 	// use the new IDs.
-	if err := rewriteTableDescs(tables, tableRewrites); err != nil {
+	if err := rewriteTableDescs(tables, tableRewrites, overrideDB); err != nil {
 		return mu.res, nil, nil, err
 	}
 
@@ -1273,7 +1311,7 @@ func doRestorePlan(
 			tables = append(tables, tableDesc)
 		}
 	}
-	if err := rewriteTableDescs(tables, tableRewrites); err != nil {
+	if err := rewriteTableDescs(tables, tableRewrites, opts[restoreOptIntoDB]); err != nil {
 		return err
 	}
 
@@ -1291,6 +1329,7 @@ func doRestorePlan(
 			TableRewrites: tableRewrites,
 			URIs:          from,
 			TableDescs:    tables,
+			OverrideDB:    opts[restoreOptIntoDB],
 		},
 	})
 	if err != nil {
@@ -1344,6 +1383,7 @@ func (r *restoreResumer) Resume(
 		details.EndTime,
 		sqlDescs,
 		details.TableRewrites,
+		details.OverrideDB,
 		job,
 		resultsCh,
 	)