sctest: Augmented BACKUP/RESTORE tests with table-level restore

Previously, the Backup test in declarative schema changer backups the
whole database and restore the whole database with `RESTORE DATABASE`.
This PR augments the test by adding another flavor to restore:
`RESTORE TABLE tbl1,...,tblN` where `tblx` are *all* the tables in the
backup. This will nicely give us coverage for `RESTORE TABLE` when
a declarative schema changer job is running.

Note that ideally we want to randomly restore only a subset of all the
table. Indeed I tried to implement that but realize it was blocked by
one limitation in the declarative shcema changer: We don't yet support
restore schema changer job that skips missing sequences (E.g. if we
have a table `t` and a sequence `s`, and I want to
`ALTER TABLE t ADD COLUMN c DEFAULT nextval('s')`, we backup database
in PostCommit phase. Later when we restore just `t`, the schema changer
job will run into error `error executing 'missing rewrite for id 109 in
<column_default_expression:<table_id:108 column_id:2
embedded_expr:<expr:"nextval(109:::REGCLASS)" uses_sequence_ids:109 >)`)
In other words, we have not implemented the equivalent of
`skip_missing_sequences`, which is an option for RESTORE,
for schema changer job.

Release justification: test-only changes
Release note: None

pkg/sql/schemachanger/sctest/cumulative.go

@@ -549,21 +549,68 @@ func Backup(t *testing.T, path string, newCluster NewClusterFunc) {
 		t.Logf("finished")
 
 		for i, b := range backups {
-			for _, isSchemaOnly := range []bool{true, false} {
-				name := ""
-				if isSchemaOnly {
-					name = "schema-only"
-				}
-				t.Run(name, func(t *testing.T) {
+			// For each backup, we restore it in three flavors.
+			// 1. RESTORE DATABASE
+			// 2. RESTORE DATABASE WITH schema_only
+			// 3. RESTORE TABLE tbl1, tbl2, ..., tblN
+			// We then assert that the restored database should correctly finish
+			// the ongoing schema change job when the backup was taken, and
+			// reaches the expected state as if the back/restore had not happened at all.
+			// Skip a backup randomly.
+			type BackupConsumptionFlavor struct {
+				name         string
+				restoreSetup []string
+				restoreQuery string
+			}
+			flavors := []BackupConsumptionFlavor{
+				{
+					name: "restore database",
+					restoreSetup: []string{
+						fmt.Sprintf("DROP DATABASE IF EXISTS %q CASCADE", dbName),
+						"SET use_declarative_schema_changer = 'off'",
+					},
+					restoreQuery: fmt.Sprintf("RESTORE DATABASE %s FROM LATEST IN '%s'", dbName, b.url),
+				},
+				{
+					name: "restore database with schema-only",
+					restoreSetup: []string{
+						fmt.Sprintf("DROP DATABASE IF EXISTS %q CASCADE", dbName),
+						"SET use_declarative_schema_changer = 'off'",
+					},
+					restoreQuery: fmt.Sprintf("RESTORE DATABASE %s FROM LATEST IN '%s' with schema_only", dbName, b.url),
+				},
+			}
+
+			// For the third flavor, we restore all tables in the backup.
+			// Skip it if there is no tables.
+			rows := tdb.QueryStr(t, `
+			SELECT parent_schema_name, object_name
+			FROM [SHOW BACKUP FROM LATEST IN $1]
+			WHERE database_name = $2 AND object_type = 'table'`, b.url, dbName)
+			var tablesToRestore []string
+			for _, row := range rows {
+				tablesToRestore = append(tablesToRestore, fmt.Sprintf("%s.%s.%s", dbName, row[0], row[1]))
+			}
+
+			if len(tablesToRestore) > 0 {
+				flavors = append(flavors, BackupConsumptionFlavor{
+					name: "restore all tables in database",
+					restoreSetup: []string{
+						fmt.Sprintf("DROP DATABASE IF EXISTS %q CASCADE", dbName),
+						fmt.Sprintf("CREATE DATABASE %q", dbName),
+						"SET use_declarative_schema_changer = 'off'",
+					},
+					restoreQuery: fmt.Sprintf("RESTORE TABLE %s FROM LATEST IN '%s' WITH skip_missing_sequences",
+						strings.Join(tablesToRestore, ","), b.url),
+				})
+			}
+
+			for _, flavor := range flavors {
+				t.Run(flavor.name, func(t *testing.T) {
 					maybeRandomlySkip(t)
-					t.Logf("testing backup %d %v", i, b.isRollback)
-					tdb.Exec(t, fmt.Sprintf("DROP DATABASE IF EXISTS %q CASCADE", dbName))
-					tdb.Exec(t, "SET use_declarative_schema_changer = 'off'")
-					restoreQuery := fmt.Sprintf("RESTORE DATABASE %s FROM LATEST IN '%s'", dbName, b.url)
-					if isSchemaOnly {
-						restoreQuery = restoreQuery + " with schema_only"
-					}
-					tdb.Exec(t, restoreQuery)
+					t.Logf("testing backup %d (rollback=%v)", i, b.isRollback)
+					tdb.ExecMultiple(t, flavor.restoreSetup...)
+					tdb.Exec(t, flavor.restoreQuery)
 					tdb.Exec(t, fmt.Sprintf("USE %q", dbName))
 					waitForSchemaChangesToFinish(t, tdb)
 					afterRestore := tdb.QueryStr(t, fetchDescriptorStateQuery)