diff --git a/docs/generated/sql/bnf/stmt_block.bnf b/docs/generated/sql/bnf/stmt_block.bnf
index c7bf3339aeee..6f55ef4d3381 100644
--- a/docs/generated/sql/bnf/stmt_block.bnf
+++ b/docs/generated/sql/bnf/stmt_block.bnf
@@ -1262,6 +1262,7 @@ unreserved_keyword ::=
 	| 'RUNNING'
 	| 'SCHEDULE'
 	| 'SCHEDULES'
+	| 'SCHEMA_ONLY'
 	| 'SCROLL'
 	| 'SETTING'
 	| 'SETTINGS'
@@ -2453,6 +2454,7 @@ restore_options ::=
 	| 'NEW_DB_NAME' '=' string_or_placeholder
 	| 'INCREMENTAL_LOCATION' '=' string_or_placeholder_opt_list
 	| 'TENANT' '=' string_or_placeholder
+	| 'SCHEMA_ONLY'
 
 scrub_option_list ::=
 	( scrub_option ) ( ( ',' scrub_option ) )*
diff --git a/pkg/ccl/backupccl/backup_telemetry.go b/pkg/ccl/backupccl/backup_telemetry.go
index 2e80fde6c729..9134f54eb4ec 100644
--- a/pkg/ccl/backupccl/backup_telemetry.go
+++ b/pkg/ccl/backupccl/backup_telemetry.go
@@ -65,6 +65,7 @@ const (
 	telemetryOptionSkipMissingSequenceOwners = "skip_missing_sequence_owners"
 	telemetryOptionSkipMissingViews          = "skip_missing_views"
 	telemetryOptionSkipLocalitiesCheck       = "skip_localities_check"
+	telemetryOptionSchemaOnly                = "schema_only"
 )
 
 // logBackupTelemetry publishes an eventpb.RecoveryEvent about a manually
@@ -397,6 +398,9 @@ func logRestoreTelemetry(
 	if opts.Detached {
 		options = append(options, telemetryOptionDetached)
 	}
+	if opts.SchemaOnly {
+		options = append(options, telemetryOptionSchemaOnly)
+	}
 	sort.Strings(options)
 
 	event := &eventpb.RecoveryEvent{
diff --git a/pkg/ccl/backupccl/backup_test.go b/pkg/ccl/backupccl/backup_test.go
index 44f34c938df8..f4991dbec63f 100644
--- a/pkg/ccl/backupccl/backup_test.go
+++ b/pkg/ccl/backupccl/backup_test.go
@@ -2675,99 +2675,109 @@ func TestBackupRestoreDuringUserDefinedTypeChange(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			// Protects numTypeChangesStarted and numTypeChangesFinished.
-			var mu syncutil.Mutex
-			numTypeChangesStarted := 0
-			numTypeChangesFinished := 0
-			typeChangesStarted := make(chan struct{})
-			waitForBackup := make(chan struct{})
-			typeChangesFinished := make(chan struct{})
-			_, sqlDB, _, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, 0, InitManualReplication, base.TestClusterArgs{
-				ServerArgs: base.TestServerArgs{
-					Knobs: base.TestingKnobs{
-						SQLTypeSchemaChanger: &sql.TypeSchemaChangerTestingKnobs{
-							RunBeforeEnumMemberPromotion: func(context.Context) error {
-								mu.Lock()
-								if numTypeChangesStarted < len(tc.queries) {
-									numTypeChangesStarted++
-									if numTypeChangesStarted == len(tc.queries) {
-										close(typeChangesStarted)
+		for _, isSchemaOnly := range []bool{true, false} {
+			suffix := ""
+			if isSchemaOnly {
+				suffix = "-schema-only"
+			}
+			t.Run(tc.name+suffix, func(t *testing.T) {
+				// Protects numTypeChangesStarted and numTypeChangesFinished.
+				var mu syncutil.Mutex
+				numTypeChangesStarted := 0
+				numTypeChangesFinished := 0
+				typeChangesStarted := make(chan struct{})
+				waitForBackup := make(chan struct{})
+				typeChangesFinished := make(chan struct{})
+				_, sqlDB, _, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, 0, InitManualReplication, base.TestClusterArgs{
+					ServerArgs: base.TestServerArgs{
+						Knobs: base.TestingKnobs{
+							SQLTypeSchemaChanger: &sql.TypeSchemaChangerTestingKnobs{
+								RunBeforeEnumMemberPromotion: func(context.Context) error {
+									mu.Lock()
+									if numTypeChangesStarted < len(tc.queries) {
+										numTypeChangesStarted++
+										if numTypeChangesStarted == len(tc.queries) {
+											close(typeChangesStarted)
+										}
+										mu.Unlock()
+										<-waitForBackup
+									} else {
+										mu.Unlock()
 									}
-									mu.Unlock()
-									<-waitForBackup
-								} else {
-									mu.Unlock()
-								}
-								return nil
+									return nil
+								},
 							},
 						},
 					},
-				},
-			})
-			defer cleanupFn()
+				})
+				defer cleanupFn()
 
-			// Create a database with a type.
-			sqlDB.Exec(t, `
+				// Create a database with a type.
+				sqlDB.Exec(t, `
 CREATE DATABASE d;
 CREATE TYPE d.greeting AS ENUM ('hello', 'howdy', 'hi');
 `)
 
-			// Start ALTER TYPE statement(s) that will block.
-			for _, query := range tc.queries {
-				go func(query string, totalQueries int) {
-					// Note we don't use sqlDB.Exec here because we can't Fatal from within a goroutine.
-					if _, err := sqlDB.DB.ExecContext(context.Background(), query); err != nil {
-						t.Error(err)
-					}
-					mu.Lock()
-					numTypeChangesFinished++
-					if numTypeChangesFinished == totalQueries {
-						close(typeChangesFinished)
-					}
-					mu.Unlock()
-				}(query, len(tc.queries))
-			}
-
-			// Wait on the type changes to start.
-			<-typeChangesStarted
+				// Start ALTER TYPE statement(s) that will block.
+				for _, query := range tc.queries {
+					go func(query string, totalQueries int) {
+						// Note we don't use sqlDB.Exec here because we can't Fatal from within a goroutine.
+						if _, err := sqlDB.DB.ExecContext(context.Background(), query); err != nil {
+							t.Error(err)
+						}
+						mu.Lock()
+						numTypeChangesFinished++
+						if numTypeChangesFinished == totalQueries {
+							close(typeChangesFinished)
+						}
+						mu.Unlock()
+					}(query, len(tc.queries))
+				}
 
-			// Now create a backup while the type change job is blocked so that
-			// greeting is backed up with some enum members in READ_ONLY state.
-			sqlDB.Exec(t, `BACKUP DATABASE d TO 'nodelocal://0/test/'`)
+				// Wait on the type changes to start.
+				<-typeChangesStarted
 
-			// Let the type change finish.
-			close(waitForBackup)
-			<-typeChangesFinished
+				// Now create a backup while the type change job is blocked so that
+				// greeting is backed up with some enum members in READ_ONLY state.
+				sqlDB.Exec(t, `BACKUP DATABASE d TO 'nodelocal://0/test/'`)
 
-			// Now drop the database and restore.
-			sqlDB.Exec(t, `DROP DATABASE d`)
-			sqlDB.Exec(t, `RESTORE DATABASE d FROM 'nodelocal://0/test/'`)
+				// Let the type change finish.
+				close(waitForBackup)
+				<-typeChangesFinished
 
-			// The type change job should be scheduled and finish. Note that we can't use
-			// sqlDB.CheckQueryResultsRetry as it Fatal's upon an error. The case below
-			// will error until the job completes.
-			for i, query := range tc.succeedAfter {
-				testutils.SucceedsSoon(t, func() error {
-					_, err := sqlDB.DB.ExecContext(context.Background(), query)
-					return err
-				})
-				sqlDB.CheckQueryResults(t, query, [][]string{{tc.expectedSuccess[i]}})
-			}
+				// Now drop the database and restore.
+				sqlDB.Exec(t, `DROP DATABASE d`)
+				restoreQuery := `RESTORE DATABASE d FROM 'nodelocal://0/test/'`
+				if isSchemaOnly {
+					restoreQuery = restoreQuery + " with schema_only"
+				}
+				sqlDB.Exec(t, restoreQuery)
+
+				// The type change job should be scheduled and finish. Note that we can't use
+				// sqlDB.CheckQueryResultsRetry as it Fatal's upon an error. The case below
+				// will error until the job completes.
+				for i, query := range tc.succeedAfter {
+					testutils.SucceedsSoon(t, func() error {
+						_, err := sqlDB.DB.ExecContext(context.Background(), query)
+						return err
+					})
+					sqlDB.CheckQueryResults(t, query, [][]string{{tc.expectedSuccess[i]}})
+				}
 
-			for i, query := range tc.errorAfter {
-				testutils.SucceedsSoon(t, func() error {
-					_, err := sqlDB.DB.ExecContext(context.Background(), query)
-					if err == nil {
-						return errors.New("expected error, found none")
-					}
-					if !testutils.IsError(err, tc.expectedError[i]) {
-						return errors.Newf("expected error %q, found %v", tc.expectedError[i], pgerror.FullError(err))
-					}
-					return nil
-				})
-			}
-		})
+				for i, query := range tc.errorAfter {
+					testutils.SucceedsSoon(t, func() error {
+						_, err := sqlDB.DB.ExecContext(context.Background(), query)
+						if err == nil {
+							return errors.New("expected error, found none")
+						}
+						if !testutils.IsError(err, tc.expectedError[i]) {
+							return errors.Newf("expected error %q, found %v", tc.expectedError[i], pgerror.FullError(err))
+						}
+						return nil
+					})
+				}
+			})
+		}
 	}
 }
 
diff --git a/pkg/ccl/backupccl/backuprand/backup_rand_test.go b/pkg/ccl/backupccl/backuprand/backup_rand_test.go
index 08b5fce1fbdf..7af84d8a56f8 100644
--- a/pkg/ccl/backupccl/backuprand/backup_rand_test.go
+++ b/pkg/ccl/backupccl/backuprand/backup_rand_test.go
@@ -32,7 +32,8 @@ import (
 // TestBackupRestoreRandomDataRoundtrips conducts backup/restore roundtrips on
 // randomly generated tables and verifies their data and schema are preserved.
 // It tests that full database backup as well as all subsets of per-table backup
-// roundtrip properly.
+// roundtrip properly. 50% of the time, the test runs the restore with the
+// schema_only parameter, which does not restore any rows from user tables.
 func TestBackupRestoreRandomDataRoundtrips(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -66,6 +67,11 @@ func TestBackupRestoreRandomDataRoundtrips(t *testing.T) {
 	}
 	numInserts := 20
 
+	runSchemaOnlyExtension := ""
+	if rng.Intn(10)%2 == 0 {
+		runSchemaOnlyExtension = ", schema_only"
+	}
+
 	tables := sqlDB.Query(t, `SELECT name FROM crdb_internal.tables WHERE 
 database_name = 'rand' AND schema_name = 'public'`)
 	var tableNames []string
@@ -87,7 +93,9 @@ database_name = 'rand' AND schema_name = 'public'`)
 	expectedData := make(map[string][][]string)
 	for _, tableName := range tableNames {
 		expectedCreateTableStmt[tableName] = sqlDB.QueryStr(t, fmt.Sprintf(`SELECT create_statement FROM [SHOW CREATE TABLE %s]`, tableName))[0][0]
-		expectedData[tableName] = sqlDB.QueryStr(t, fmt.Sprintf(`SELECT * FROM %s`, tableName))
+		if runSchemaOnlyExtension == "" {
+			expectedData[tableName] = sqlDB.QueryStr(t, fmt.Sprintf(`SELECT * FROM %s`, tableName))
+		}
 	}
 
 	// Now that we've created our random tables, backup and restore the whole DB
@@ -97,12 +105,12 @@ database_name = 'rand' AND schema_name = 'public'`)
 	tablesBackup := localFoo + "alltables"
 	dbBackups := []string{dbBackup, tablesBackup}
 	if err := backuputils.VerifyBackupRestoreStatementResult(
-		t, sqlDB, "BACKUP DATABASE rand TO $1", dbBackup,
+		t, sqlDB, "BACKUP DATABASE rand INTO $1", dbBackup,
 	); err != nil {
 		t.Fatal(err)
 	}
 	if err := backuputils.VerifyBackupRestoreStatementResult(
-		t, sqlDB, "BACKUP TABLE rand.* TO $1", tablesBackup,
+		t, sqlDB, "BACKUP TABLE rand.* INTO $1", tablesBackup,
 	); err != nil {
 		t.Fatal(err)
 	}
@@ -118,7 +126,12 @@ database_name = 'rand' AND schema_name = 'public'`)
 				fmt.Sprintf(`SELECT create_statement FROM [SHOW CREATE TABLE %s]`, restoreTable))[0][0]
 			assert.Equal(t, expectedCreateTableStmt[tableName], createStmt,
 				"SHOW CREATE %s not equal after RESTORE", tableName)
-			sqlDB.CheckQueryResults(t, fmt.Sprintf(`SELECT * FROM %s`, tableName), expectedData[tableName])
+			if runSchemaOnlyExtension == "" {
+				sqlDB.CheckQueryResults(t, fmt.Sprintf(`SELECT * FROM %s`, restoreTable), expectedData[tableName])
+			} else {
+				sqlDB.CheckQueryResults(t, fmt.Sprintf(`SELECT count(*) FROM %s`, restoreTable),
+					[][]string{{"0"}})
+			}
 		}
 	}
 
@@ -128,17 +141,17 @@ database_name = 'rand' AND schema_name = 'public'`)
 	for _, backup := range dbBackups {
 		sqlDB.Exec(t, "DROP DATABASE IF EXISTS restoredb")
 		sqlDB.Exec(t, "CREATE DATABASE restoredb")
+		tableQuery := fmt.Sprintf("RESTORE rand.* FROM LATEST IN $1 WITH OPTIONS (into_db='restoredb'%s)", runSchemaOnlyExtension)
 		if err := backuputils.VerifyBackupRestoreStatementResult(
-			t, sqlDB, "RESTORE rand.* FROM $1 WITH OPTIONS (into_db='restoredb')", backup,
+			t, sqlDB, tableQuery, backup,
 		); err != nil {
 			t.Fatal(err)
 		}
 		verifyTables(t, tableNames)
 		sqlDB.Exec(t, "DROP DATABASE IF EXISTS restoredb")
 
-		if err := backuputils.VerifyBackupRestoreStatementResult(
-			t, sqlDB, "RESTORE DATABASE rand FROM $1 WITH OPTIONS (new_db_name='restoredb')", backup,
-		); err != nil {
+		dbQuery := fmt.Sprintf("RESTORE DATABASE rand FROM LATEST IN $1 WITH OPTIONS (new_db_name='restoredb'%s)", runSchemaOnlyExtension)
+		if err := backuputils.VerifyBackupRestoreStatementResult(t, sqlDB, dbQuery, backup); err != nil {
 			t.Fatal(err)
 		}
 		verifyTables(t, tableNames)
@@ -155,8 +168,9 @@ database_name = 'rand' AND schema_name = 'public'`)
 		}
 		tables := strings.Join(combo, ", ")
 		t.Logf("Testing subset backup/restore %s", tables)
-		sqlDB.Exec(t, fmt.Sprintf(`BACKUP TABLE %s TO $1`, tables), backupTarget)
-		_, err := tc.Conns[0].Exec(fmt.Sprintf("RESTORE TABLE %s FROM $1 WITH OPTIONS (into_db='restoredb')", tables),
+		sqlDB.Exec(t, fmt.Sprintf(`BACKUP TABLE %s INTO $1`, tables), backupTarget)
+		_, err := tc.Conns[0].Exec(
+			fmt.Sprintf("RESTORE TABLE %s FROM LATEST IN $1 WITH OPTIONS (into_db='restoredb' %s)", tables, runSchemaOnlyExtension),
 			backupTarget)
 		if err != nil {
 			if strings.Contains(err.Error(), "skip_missing_foreign_keys") {
diff --git a/pkg/ccl/backupccl/restore_data_processor.go b/pkg/ccl/backupccl/restore_data_processor.go
index c7d902e86b5c..0cc824448778 100644
--- a/pkg/ccl/backupccl/restore_data_processor.go
+++ b/pkg/ccl/backupccl/restore_data_processor.go
@@ -17,7 +17,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl/backuppb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/storageccl"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
-	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/bulk"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -138,10 +137,6 @@ func newRestoreDataProcessor(
 ) (execinfra.Processor, error) {
 	sv := &flowCtx.Cfg.Settings.SV
 
-	if spec.Validation != jobspb.RestoreValidation_DefaultRestore {
-		return nil, errors.New("Restore Data Processor does not support validation yet")
-	}
-
 	rd := &restoreDataProcessor{
 		flowCtx:    flowCtx,
 		input:      input,
diff --git a/pkg/ccl/backupccl/restore_job.go b/pkg/ccl/backupccl/restore_job.go
index ba4deb83fabe..6a12cd848dd9 100644
--- a/pkg/ccl/backupccl/restore_job.go
+++ b/pkg/ccl/backupccl/restore_job.go
@@ -133,6 +133,7 @@ func restoreWithRetry(
 	encryption *jobspb.BackupEncryptionOptions,
 	kmsEnv cloud.KMSEnv,
 ) (roachpb.RowCount, error) {
+
 	// We retry on pretty generic failures -- any rpc error. If a worker node were
 	// to restart, it would produce this kind of error, but there may be other
 	// errors that are also rpc errors. Don't retry to aggressively.
@@ -578,13 +579,26 @@ func spansForAllRestoreTableIndexes(
 	codec keys.SQLCodec,
 	tables []catalog.TableDescriptor,
 	revs []backuppb.BackupManifest_DescriptorRevision,
+	schemaOnly bool,
 ) []roachpb.Span {
 
+	skipTableData := func(table catalog.TableDescriptor) bool {
+		// We only import spans for physical tables.
+		if !table.IsPhysicalTable() {
+			return true
+		}
+		// The only table data restored during a schemaOnly restore are from system tables,
+		// which only get restored during a cluster restore.
+		if table.GetParentID() != keys.SystemDatabaseID && schemaOnly {
+			return true
+		}
+		return false
+	}
+
 	added := make(map[tableAndIndex]bool, len(tables))
 	sstIntervalTree := interval.NewTree(interval.ExclusiveOverlapper)
 	for _, table := range tables {
-		// We only import spans for physical tables.
-		if !table.IsPhysicalTable() {
+		if skipTableData(table) {
 			continue
 		}
 		for _, index := range table.ActiveIndexes() {
@@ -607,8 +621,7 @@ func spansForAllRestoreTableIndexes(
 		rawTbl, _, _, _, _ := descpb.FromDescriptor(rev.Desc)
 		if rawTbl != nil && !rawTbl.Dropped() {
 			tbl := tabledesc.NewBuilder(rawTbl).BuildImmutableTable()
-			// We only import spans for physical tables.
-			if !tbl.IsPhysicalTable() {
+			if skipTableData(tbl) {
 				continue
 			}
 			for _, idx := range tbl.ActiveIndexes() {
@@ -713,8 +726,8 @@ func createImportingDescriptors(
 
 	// We get the spans of the restoring tables _as they appear in the backup_,
 	// that is, in the 'old' keyspace, before we reassign the table IDs.
-	preRestoreSpans := spansForAllRestoreTableIndexes(backupCodec, preRestoreTables, nil)
-	postRestoreSpans := spansForAllRestoreTableIndexes(backupCodec, postRestoreTables, nil)
+	preRestoreSpans := spansForAllRestoreTableIndexes(backupCodec, preRestoreTables, nil, details.SchemaOnly)
+	postRestoreSpans := spansForAllRestoreTableIndexes(backupCodec, postRestoreTables, nil, details.SchemaOnly)
 
 	log.Eventf(ctx, "starting restore for %d tables", len(mutableTables))
 
@@ -1231,10 +1244,6 @@ func (r *restoreResumer) doResume(ctx context.Context, execCtx interface{}) erro
 	p := execCtx.(sql.JobExecContext)
 	r.execCfg = p.ExecCfg()
 
-	if details.Validation != jobspb.RestoreValidation_DefaultRestore {
-		return errors.Errorf("No restore validation tools are supported")
-	}
-
 	mem := p.ExecCfg().RootMemoryMonitor.MakeBoundAccount()
 	defer mem.Close(ctx)
 
diff --git a/pkg/ccl/backupccl/restore_mid_schema_change_test.go b/pkg/ccl/backupccl/restore_mid_schema_change_test.go
index 84d0edd6e4c5..63463abfe0e0 100644
--- a/pkg/ccl/backupccl/restore_mid_schema_change_test.go
+++ b/pkg/ccl/backupccl/restore_mid_schema_change_test.go
@@ -62,46 +62,53 @@ func TestRestoreMidSchemaChange(t *testing.T) {
 		testdataBase = testutils.TestDataPath(t, "restore_mid_schema_change")
 		exportDirs   = testdataBase + "/exports"
 	)
-	for _, isClusterRestore := range []bool{true, false} {
-		name := "table"
-		if isClusterRestore {
-			name = "cluster"
+	for _, isSchemaOnly := range []bool{true, false} {
+		name := "regular-"
+		if isSchemaOnly {
+			name = "schema-only-"
 		}
-		t.Run(name, func(t *testing.T) {
-			// blockLocations indicates whether the backup taken was blocked before or
-			// after the backfill portion of the schema change.
-			for _, blockLocation := range []string{"before", "after"} {
-				t.Run(blockLocation, func(t *testing.T) {
-					versionDirs, err := ioutil.ReadDir(filepath.Join(exportDirs, blockLocation))
-					require.NoError(t, err)
-
-					for _, clusterVersionDir := range versionDirs {
-						if clusterVersionDir.Name() == "19.2" && isClusterRestore {
-							// 19.2 does not support cluster backups.
-							continue
-						}
-
-						t.Run(clusterVersionDir.Name(), func(t *testing.T) {
-							require.True(t, clusterVersionDir.IsDir())
-							fullClusterVersionDir, err := filepath.Abs(
-								filepath.Join(exportDirs, blockLocation, clusterVersionDir.Name()))
-							require.NoError(t, err)
+		for _, isClusterRestore := range []bool{true, false} {
+			name = name + "table"
+			if isClusterRestore {
+				name = name + "cluster"
+			}
+			t.Run(name, func(t *testing.T) {
+				// blockLocations indicates whether the backup taken was blocked before or
+				// after the backfill portion of the schema change.
+				for _, blockLocation := range []string{"before", "after"} {
+					t.Run(blockLocation, func(t *testing.T) {
+						versionDirs, err := ioutil.ReadDir(filepath.Join(exportDirs, blockLocation))
+						require.NoError(t, err)
+
+						for _, clusterVersionDir := range versionDirs {
+							if clusterVersionDir.Name() == "19.2" && isClusterRestore {
+								// 19.2 does not support cluster backups.
+								continue
+							}
 
-							// In each version folder (e.g. "19.2", "20.1"), there is a backup for
-							// each schema change.
-							backupDirs, err := ioutil.ReadDir(fullClusterVersionDir)
-							require.NoError(t, err)
+							t.Run(clusterVersionDir.Name(), func(t *testing.T) {
+								require.True(t, clusterVersionDir.IsDir())
+								fullClusterVersionDir, err := filepath.Abs(
+									filepath.Join(exportDirs, blockLocation, clusterVersionDir.Name()))
+								require.NoError(t, err)
 
-							for _, backupDir := range backupDirs {
-								fullBackupDir, err := filepath.Abs(filepath.Join(fullClusterVersionDir, backupDir.Name()))
+								// In each version folder (e.g. "19.2", "20.1"), there is a backup for
+								// each schema change.
+								backupDirs, err := ioutil.ReadDir(fullClusterVersionDir)
 								require.NoError(t, err)
-								t.Run(backupDir.Name(), restoreMidSchemaChange(fullBackupDir, backupDir.Name(), isClusterRestore))
-							}
-						})
-					}
-				})
-			}
-		})
+
+								for _, backupDir := range backupDirs {
+									fullBackupDir, err := filepath.Abs(filepath.Join(fullClusterVersionDir, backupDir.Name()))
+									require.NoError(t, err)
+									t.Run(backupDir.Name(), restoreMidSchemaChange(fullBackupDir, backupDir.Name(),
+										isClusterRestore, isSchemaOnly))
+								}
+							})
+						}
+					})
+				}
+			})
+		}
 	}
 }
 
@@ -132,7 +139,12 @@ func expectedSCJobCount(scName string) int {
 }
 
 func validateTable(
-	t *testing.T, kvDB *kv.DB, sqlDB *sqlutils.SQLRunner, dbName string, tableName string,
+	t *testing.T,
+	kvDB *kv.DB,
+	sqlDB *sqlutils.SQLRunner,
+	dbName string,
+	tableName string,
+	isSchemaOnly bool,
 ) {
 	desc := desctestutils.TestingGetPublicTableDescriptor(kvDB, keys.SystemSQLCodec, dbName, tableName)
 	// There should be no mutations on these table descriptors at this point.
@@ -140,7 +152,11 @@ func validateTable(
 
 	var rowCount int
 	sqlDB.QueryRow(t, fmt.Sprintf(`SELECT count(*) FROM %s.%s`, dbName, tableName)).Scan(&rowCount)
-	require.Greater(t, rowCount, 0, "expected table to have some rows")
+	if isSchemaOnly {
+		require.Equal(t, rowCount, 0, "expected table to have no rows")
+	} else {
+		require.Greater(t, rowCount, 0, "expected table to have some rows")
+	}
 	// The number of entries in all indexes should be the same.
 	for _, index := range desc.AllIndexes() {
 		var indexCount int
@@ -162,7 +178,9 @@ func getTablesInTest(scName string) (tableNames []string) {
 	return
 }
 
-func verifyMidSchemaChange(t *testing.T, scName string, kvDB *kv.DB, sqlDB *sqlutils.SQLRunner) {
+func verifyMidSchemaChange(
+	t *testing.T, scName string, kvDB *kv.DB, sqlDB *sqlutils.SQLRunner, isSchemaOnly bool,
+) {
 	tables := getTablesInTest(scName)
 
 	// Check that we are left with the expected number of schema change jobs.
@@ -174,7 +192,7 @@ func verifyMidSchemaChange(t *testing.T, scName string, kvDB *kv.DB, sqlDB *sqlu
 		"Expected %d schema change jobs but found %v", expNumSchemaChangeJobs, synthesizedSchemaChangeJobs)
 
 	for _, tableName := range tables {
-		validateTable(t, kvDB, sqlDB, "defaultdb", tableName)
+		validateTable(t, kvDB, sqlDB, "defaultdb", tableName, isSchemaOnly)
 		// Ensure that a schema change can complete on the restored table.
 		schemaChangeQuery := fmt.Sprintf("ALTER TABLE defaultdb.%s ADD CONSTRAINT post_restore_const CHECK (a > 0)", tableName)
 		sqlDB.Exec(t, schemaChangeQuery)
@@ -183,7 +201,7 @@ func verifyMidSchemaChange(t *testing.T, scName string, kvDB *kv.DB, sqlDB *sqlu
 }
 
 func restoreMidSchemaChange(
-	backupDir, schemaChangeName string, isClusterRestore bool,
+	backupDir, schemaChangeName string, isClusterRestore bool, isSchemaOnly bool,
 ) func(t *testing.T) {
 	return func(t *testing.T) {
 		ctx := context.Background()
@@ -213,16 +231,19 @@ func restoreMidSchemaChange(
 		require.NoError(t, err)
 
 		sqlDB.Exec(t, "USE defaultdb")
-		restoreQuery := "RESTORE defaultdb.* from $1"
+		restoreQuery := "RESTORE defaultdb.* FROM $1"
 		if isClusterRestore {
-			restoreQuery = "RESTORE from $1"
+			restoreQuery = "RESTORE FROM $1"
+		}
+		if isSchemaOnly {
+			restoreQuery = restoreQuery + "with schema_only"
 		}
 		log.Infof(context.Background(), "%+v", sqlDB.QueryStr(t, "SHOW BACKUP $1", localFoo))
 		sqlDB.Exec(t, restoreQuery, localFoo)
 		// Wait for all jobs to terminate. Some may fail since we don't restore
 		// adding spans.
 		sqlDB.CheckQueryResultsRetry(t, "SELECT * FROM crdb_internal.jobs WHERE job_type = 'SCHEMA CHANGE' AND NOT (status = 'succeeded' OR status = 'failed')", [][]string{})
-		verifyMidSchemaChange(t, schemaChangeName, kvDB, sqlDB)
+		verifyMidSchemaChange(t, schemaChangeName, kvDB, sqlDB, isSchemaOnly)
 
 		// Because crdb_internal.invalid_objects is a virtual table, by default, the
 		// query will take a lease on the database sqlDB is connected to and only run
diff --git a/pkg/ccl/backupccl/restore_old_sequences_test.go b/pkg/ccl/backupccl/restore_old_sequences_test.go
index 783f97767702..3e2fb96122b8 100644
--- a/pkg/ccl/backupccl/restore_old_sequences_test.go
+++ b/pkg/ccl/backupccl/restore_old_sequences_test.go
@@ -31,7 +31,7 @@ import (
 //
 //  VERSION=...
 //  roachprod create local
-//  roachprod wipe local
+//  roachprod wipe localÅ
 //  roachprod stage local release ${VERSION}
 //  roachprod start local
 //  roachprod sql local:1 -- -e "$(cat pkg/ccl/backupccl/testdata/restore_old_sequences/create.sql)"
@@ -50,16 +50,22 @@ func TestRestoreOldSequences(t *testing.T) {
 	t.Run("sequences-restore", func(t *testing.T) {
 		dirs, err := ioutil.ReadDir(exportDirs)
 		require.NoError(t, err)
-		for _, dir := range dirs {
-			require.True(t, dir.IsDir())
-			exportDir, err := filepath.Abs(filepath.Join(exportDirs, dir.Name()))
-			require.NoError(t, err)
-			t.Run(dir.Name(), restoreOldSequencesTest(exportDir))
+		for _, isSchemaOnly := range []bool{true, false} {
+			suffix := ""
+			if isSchemaOnly {
+				suffix = "-schema-only"
+			}
+			for _, dir := range dirs {
+				require.True(t, dir.IsDir())
+				exportDir, err := filepath.Abs(filepath.Join(exportDirs, dir.Name()))
+				require.NoError(t, err)
+				t.Run(dir.Name()+suffix, restoreOldSequencesTest(exportDir, isSchemaOnly))
+			}
 		}
 	})
 }
 
-func restoreOldSequencesTest(exportDir string) func(t *testing.T) {
+func restoreOldSequencesTest(exportDir string, isSchemaOnly bool) func(t *testing.T) {
 	return func(t *testing.T) {
 		params := base.TestServerArgs{}
 		const numAccounts = 1000
@@ -71,10 +77,17 @@ func restoreOldSequencesTest(exportDir string) func(t *testing.T) {
 		sqlDB.Exec(t, `CREATE DATABASE test`)
 		var unused string
 		var importedRows int
-		sqlDB.QueryRow(t, `RESTORE test.* FROM $1`, localFoo).Scan(
+		restoreQuery := `RESTORE test.* FROM $1`
+		if isSchemaOnly {
+			restoreQuery = restoreQuery + " with schema_only"
+		}
+		sqlDB.QueryRow(t, restoreQuery, localFoo).Scan(
 			&unused, &unused, &unused, &importedRows, &unused, &unused,
 		)
-		const totalRows = 4
+		totalRows := 4
+		if isSchemaOnly {
+			totalRows = 0
+		}
 		if importedRows != totalRows {
 			t.Fatalf("expected %d rows, got %d", totalRows, importedRows)
 		}
@@ -100,6 +113,12 @@ func restoreOldSequencesTest(exportDir string) func(t *testing.T) {
 			{"1", "1"},
 			{"2", "2"},
 		}
+		if isSchemaOnly {
+			// In a schema_only RESTORE, the restored sequence will be empty
+			expectedRows = [][]string{
+				{"1", "1"},
+			}
+		}
 		sqlDB.CheckQueryResults(t, `SELECT * FROM test.t1 ORDER BY i`, expectedRows)
 		sqlDB.CheckQueryResults(t, `SELECT * FROM test.v`, [][]string{{"1"}})
 		sqlDB.CheckQueryResults(t, `SELECT * FROM test.v2`, [][]string{{"2"}})
diff --git a/pkg/ccl/backupccl/restore_planning.go b/pkg/ccl/backupccl/restore_planning.go
index 9d7dc2057ab3..d0b550810e8a 100644
--- a/pkg/ccl/backupccl/restore_planning.go
+++ b/pkg/ccl/backupccl/restore_planning.go
@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/multiregionccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/storageccl"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/featureflag"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
@@ -1135,6 +1136,7 @@ func resolveOptionsForRestoreJobDescription(
 		SkipMissingSequenceOwners: opts.SkipMissingSequenceOwners,
 		SkipMissingViews:          opts.SkipMissingViews,
 		Detached:                  opts.Detached,
+		SchemaOnly:                opts.SchemaOnly,
 	}
 
 	if opts.EncryptionPassphrase != nil {
@@ -1223,6 +1225,12 @@ func restorePlanHook(
 		return nil, nil, nil, false, err
 	}
 
+	if restoreStmt.Options.SchemaOnly &&
+		!p.ExecCfg().Settings.Version.IsActive(ctx, clusterversion.Start22_2) {
+		return nil, nil, nil, false,
+			errors.New("cannot run RESTORE with schema_only until cluster has fully upgraded to 22.2")
+	}
+
 	fromFns := make([]func() ([]string, error), len(restoreStmt.From))
 	for i := range restoreStmt.From {
 		fromFn, err := p.TypeAsStringArray(ctx, tree.Exprs(restoreStmt.From[i]), "RESTORE")
@@ -2026,7 +2034,7 @@ func doRestorePlan(
 		// TODO(msbutler): Delete in 23.1
 		RestoreSystemUsers: restoreStmt.DescriptorCoverage == tree.SystemUsers,
 		PreRewriteTenantId: oldTenantID,
-		Validation:         jobspb.RestoreValidation_DefaultRestore,
+		SchemaOnly:         restoreStmt.Options.SchemaOnly,
 	}
 
 	jr := jobs.Record{
diff --git a/pkg/ccl/backupccl/restore_processor_planning.go b/pkg/ccl/backupccl/restore_processor_planning.go
index e9388cc05ca3..860b4bab7400 100644
--- a/pkg/ccl/backupccl/restore_processor_planning.go
+++ b/pkg/ccl/backupccl/restore_processor_planning.go
@@ -118,7 +118,6 @@ func distRestore(
 			TableRekeys:  tableRekeys,
 			TenantRekeys: tenantRekeys,
 			PKIDs:        pkIDs,
-			Validation:   jobspb.RestoreValidation_DefaultRestore,
 		}
 
 		if len(splitAndScatterSpecs) == 0 {
@@ -308,7 +307,6 @@ func makeSplitAndScatterSpecs(
 				}},
 				TableRekeys:  tableRekeys,
 				TenantRekeys: tenantRekeys,
-				Validation:   jobspb.RestoreValidation_DefaultRestore,
 			}
 		}
 	}
diff --git a/pkg/ccl/backupccl/split_and_scatter_processor.go b/pkg/ccl/backupccl/split_and_scatter_processor.go
index 9ccbd6ed835b..676231f6b3a8 100644
--- a/pkg/ccl/backupccl/split_and_scatter_processor.go
+++ b/pkg/ccl/backupccl/split_and_scatter_processor.go
@@ -15,7 +15,6 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -213,10 +212,6 @@ func newSplitAndScatterProcessor(
 	output execinfra.RowReceiver,
 ) (execinfra.Processor, error) {
 
-	if spec.Validation != jobspb.RestoreValidation_DefaultRestore {
-		return nil, errors.New("Split and Scatter Processor does not support validation yet")
-	}
-
 	numEntries := 0
 	for _, chunk := range spec.Chunks {
 		numEntries += len(chunk.Entries)
diff --git a/pkg/ccl/backupccl/split_and_scatter_processor_test.go b/pkg/ccl/backupccl/split_and_scatter_processor_test.go
index 8b586e095590..3c13b491a202 100644
--- a/pkg/ccl/backupccl/split_and_scatter_processor_test.go
+++ b/pkg/ccl/backupccl/split_and_scatter_processor_test.go
@@ -15,7 +15,6 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
@@ -247,7 +246,6 @@ func TestSplitAndScatterProcessor(t *testing.T) {
 			require.NoError(t, err)
 
 			post := execinfrapb.PostProcessSpec{}
-			c.procSpec.Validation = jobspb.RestoreValidation_DefaultRestore
 			proc, err := newSplitAndScatterProcessor(&flowCtx, 0 /* processorID */, c.procSpec, &post, out)
 			require.NoError(t, err)
 			ssp, ok := proc.(*splitAndScatterProcessor)
diff --git a/pkg/ccl/backupccl/testdata/backup-restore/restore-schema-only b/pkg/ccl/backupccl/testdata/backup-restore/restore-schema-only
new file mode 100644
index 000000000000..53542234456f
--- /dev/null
+++ b/pkg/ccl/backupccl/testdata/backup-restore/restore-schema-only
@@ -0,0 +1,144 @@
+# Test schema_only restore
+
+new-server name=s1 allow-implicit-access
+----
+
+exec-sql
+CREATE DATABASE d;
+CREATE TYPE d.greeting AS ENUM ('hello', 'howdy', 'hi');
+CREATE TABLE d.t1 (x INT);
+INSERT INTO d.t1 VALUES (1), (2), (3);
+CREATE TABLE d.t2 (x d.greeting);
+INSERT INTO d.t2 VALUES ('hello'), ('howdy');
+COMMENT ON TABLE d.t1 IS 'This comment better get restored from the backed up system table!';
+----
+
+query-sql
+SHOW CREATE TABLE d.t1;
+----
+d.public.t1 CREATE TABLE public.t1 (
+	x INT8 NULL,
+	rowid INT8 NOT VISIBLE NOT NULL DEFAULT unique_rowid(),
+	CONSTRAINT t1_pkey PRIMARY KEY (rowid ASC)
+);
+COMMENT ON TABLE public.t1 IS 'This comment better get restored from the backed up system table!'
+
+# drop and create defaultDB to ensure it has a higher ID than by default. We will check that when
+# this cluster is restored, the default db with the higher id was also restored
+# by default, default db has an id of 100.
+query-sql
+SELECT id FROM system.namespace WHERE name = 'defaultdb'
+----
+100
+
+exec-sql
+DROP DATABASE defaultdb;
+CREATE DATABASE defaultdb;
+----
+
+query-sql
+SELECT count(*) FROM system.namespace WHERE name = 'defaultdb' AND id > 100
+----
+1
+
+exec-sql
+BACKUP INTO 'nodelocal://1/full_cluster_backup/';
+----
+
+exec-sql
+BACKUP Database d INTO 'nodelocal://1/full_database_backup/';
+----
+
+
+# A new cluster in prep for a cluster level schema_only restore.
+new-server name=s2 share-io-dir=s1 allow-implicit-access
+----
+
+# First, ensure cluster level schema_only restore fails fast in same ways as a cluster level restore.
+#
+# Fail fast if the user passes new_db_name.
+exec-sql
+RESTORE FROM LATEST IN 'nodelocal://0/full_cluster_backup/' with schema_only, new_db_name='d2';
+----
+pq: new_db_name can only be used for RESTORE DATABASE with a single target database
+
+
+exec-sql server=s2
+CREATE USER testuser
+----
+
+# Non admins cannot run schema_only cluster restore
+exec-sql user=testuser
+RESTORE FROM LATEST IN 'nodelocal://0/full_cluster_backup/' with schema_only
+----
+pq: only users with the admin role are allowed to restore full cluster backups
+
+# Fail fast using a database backup
+exec-sql
+RESTORE FROM LATEST IN 'nodelocal://0/full_database_backup/' with schema_only;
+----
+pq: full cluster RESTORE can only be used on full cluster BACKUP files
+
+exec-sql
+RESTORE FROM LATEST IN 'nodelocal://0/full_cluster_backup/' with schema_only;
+----
+
+# there should be no data in the restored tables
+query-sql
+SELECT * FROM d.t1;
+----
+
+query-sql
+SELECT * FROM d.t2;
+----
+
+# The backed up cluster was initiated with bank. Ensure it's now empty.
+query-sql
+SELECT * FROM data.bank;
+----
+
+# The backed table d.t1 had a comment stored in a system table. This should have been restored.
+query-sql
+SHOW CREATE TABLE d.t1;
+----
+d.public.t1 CREATE TABLE public.t1 (
+	x INT8 NULL,
+	rowid INT8 NOT VISIBLE NOT NULL DEFAULT unique_rowid(),
+	CONSTRAINT t1_pkey PRIMARY KEY (rowid ASC)
+);
+COMMENT ON TABLE public.t1 IS 'This comment better get restored from the backed up system table!'
+
+# Ensure the defaultdb from the backed up cluster was restored.
+query-sql
+SELECT count(*) FROM system.namespace WHERE name = 'defaultdb' AND id > 100
+----
+1
+
+############################################################
+# Ensure Database Level schema_only restore logic is sound
+############################################################
+
+exec-sql
+RESTORE DATABASE d FROM LATEST IN 'nodelocal://0/full_database_backup/' with schema_only, new_db_name='d2';
+----
+
+# There should be no data in the user tables.
+query-sql
+SELECT * FROM d2.t1;
+----
+
+query-sql
+SELECT * FROM d2.t2;
+----
+
+# Each of the restored types should have namespace entries. Test this by
+# trying to create types that would cause namespace conflicts.
+exec-sql
+CREATE TYPE d2.greeting AS ENUM ('hello', 'hiya')
+----
+pq: type "d2.public.greeting" already exists
+
+# We should be able to resolve each restored type. Test this by inserting
+# into each of the restored tables.
+exec-sql
+INSERT INTO d2.t2 VALUES ('hi');
diff --git a/pkg/ccl/backupccl/testdata/backup-restore/restore-schema-only-multiregion b/pkg/ccl/backupccl/testdata/backup-restore/restore-schema-only-multiregion
new file mode 100644
index 000000000000..a8852688402e
--- /dev/null
+++ b/pkg/ccl/backupccl/testdata/backup-restore/restore-schema-only-multiregion
@@ -0,0 +1,193 @@
+# Test schema only multi region restore this test is exactly the same as the 'multiregion' datadriven
+# test, except schema_only RESTORE is used
+
+new-server name=s1 allow-implicit-access localities=us-east-1,us-west-1,eu-central-1
+----
+
+exec-sql
+CREATE DATABASE d PRIMARY REGION "us-east-1" REGIONS "us-west-1", "eu-central-1";
+CREATE TABLE d.t (x INT);
+INSERT INTO d.t VALUES (1), (2), (3);
+----
+
+query-sql
+SELECT region FROM [SHOW REGIONS FROM DATABASE d] ORDER BY 1;
+----
+eu-central-1
+us-east-1
+us-west-1
+
+exec-sql
+BACKUP DATABASE d INTO 'nodelocal://1/database_backup/';
+----
+
+exec-sql
+BACKUP INTO 'nodelocal://1/full_cluster_backup/';
+----
+
+# A new cluster with the same locality settings.
+new-server name=s2 share-io-dir=s1 allow-implicit-access localities=us-east-1,us-west-1,eu-central-1
+----
+
+exec-sql
+RESTORE FROM LATEST IN 'nodelocal://0/full_cluster_backup/' with schema_only;
+----
+
+exec-sql
+DROP DATABASE d;
+----
+
+exec-sql
+RESTORE DATABASE d FROM LATEST IN 'nodelocal://0/database_backup/' with schema_only;
+----
+
+query-sql
+SHOW DATABASES;
+----
+d root us-east-1 {eu-central-1,us-east-1,us-west-1} zone
+data root <nil> {} <nil>
+defaultdb root <nil> {} <nil>
+postgres root <nil> {} <nil>
+system node <nil> {} <nil>
+
+# A new cluster with different localities settings.
+new-server name=s3 share-io-dir=s1 allow-implicit-access localities=eu-central-1,eu-north-1
+----
+
+exec-sql
+RESTORE DATABASE d FROM LATEST IN 'nodelocal://0/database_backup/' with schema_only;
+----
+pq: detected a mismatch in regions between the restore cluster and the backup cluster, missing regions detected: us-east-1, us-west-1.
+HINT: there are two ways you can resolve this issue: 1) update the cluster to which you're restoring to ensure that the regions present on the nodes' --locality flags match those present in the backup image, or 2) restore with the "skip_localities_check" option
+
+exec-sql
+RESTORE FROM LATEST IN 'nodelocal://0/full_cluster_backup/' with schema_only;
+----
+pq: detected a mismatch in regions between the restore cluster and the backup cluster, missing regions detected: us-east-1, us-west-1.
+HINT: there are two ways you can resolve this issue: 1) update the cluster to which you're restoring to ensure that the regions present on the nodes' --locality flags match those present in the backup image, or 2) restore with the "skip_localities_check" option
+
+# Create a database with no regions to check default primary regions.
+exec-sql
+CREATE DATABASE no_region_db;
+----
+
+exec-sql
+CREATE TABLE no_region_db.t (x INT);
+INSERT INTO no_region_db.t VALUES (1), (2), (3);
+CREATE DATABASE no_region_db_2;
+CREATE TABLE no_region_db_2.t (x INT);
+INSERT INTO no_region_db_2.t VALUES (1), (2), (3);
+----
+
+exec-sql
+BACKUP DATABASE no_region_db INTO 'nodelocal://1/no_region_database_backup/';
+----
+
+exec-sql
+BACKUP INTO 'nodelocal://1/no_region_cluster_backup/';
+----
+
+exec-sql
+DROP DATABASE no_region_db;
+----
+
+exec-sql
+DROP DATABASE no_region_db_2;
+----
+
+exec-sql ignore-notice
+SET CLUSTER SETTING sql.defaults.primary_region = 'non-existent-region';
+----
+
+exec-sql
+RESTORE DATABASE no_region_db FROM LATEST IN 'nodelocal://1/no_region_database_backup/' with schema_only;
+----
+pq: region "non-existent-region" does not exist
+HINT: valid regions: eu-central-1, eu-north-1
+--
+set the default PRIMARY REGION to a region that exists (see SHOW REGIONS FROM CLUSTER) then using SET CLUSTER SETTING sql.defaults.primary_region = 'region'
+
+exec-sql ignore-notice
+SET CLUSTER SETTING sql.defaults.primary_region = 'eu-central-1';
+----
+
+exec-sql
+RESTORE DATABASE no_region_db FROM LATEST IN 'nodelocal://1/no_region_database_backup/' with schema_only;
+----
+NOTICE: setting the PRIMARY REGION as eu-central-1 on database no_region_db
+HINT: to change the default primary region, use SET CLUSTER SETTING sql.defaults.primary_region = 'region' or use RESET CLUSTER SETTING sql.defaults.primary_region to disable this behavior
+
+query-sql
+SHOW DATABASES;
+----
+defaultdb root <nil> {} <nil>
+no_region_db root eu-central-1 {eu-central-1} zone
+postgres root <nil> {} <nil>
+system node <nil> {} <nil>
+
+query-sql
+USE no_region_db;
+SELECT schema_name, table_name, type, owner, locality FROM [SHOW TABLES];
+----
+public t table root REGIONAL BY TABLE IN PRIMARY REGION
+
+exec-sql
+CREATE DATABASE eu_central_db;
+CREATE TABLE eu_central_db.t (x INT);
+INSERT INTO eu_central_db.t VALUES (1), (2), (3);
+----
+NOTICE: setting eu-central-1 as the PRIMARY REGION as no PRIMARY REGION was specified
+
+exec-sql
+BACKUP DATABASE eu_central_db INTO 'nodelocal://1/eu_central_database_backup/';
+----
+
+# New cluster for a cluster backup.
+new-server name=s4 share-io-dir=s1 allow-implicit-access localities=eu-central-1,eu-north-1
+----
+
+exec-sql ignore-notice
+SET CLUSTER SETTING sql.defaults.primary_region = 'eu-north-1';
+----
+
+exec-sql
+RESTORE FROM LATEST IN 'nodelocal://1/no_region_cluster_backup/' with schema_only;
+----
+NOTICE: setting the PRIMARY REGION as eu-north-1 on database defaultdb
+HINT: to change the default primary region, use SET CLUSTER SETTING sql.defaults.primary_region = 'region' or use RESET CLUSTER SETTING sql.defaults.primary_region to disable this behavior
+NOTICE: setting the PRIMARY REGION as eu-north-1 on database postgres
+HINT: to change the default primary region, use SET CLUSTER SETTING sql.defaults.primary_region = 'region' or use RESET CLUSTER SETTING sql.defaults.primary_region to disable this behavior
+NOTICE: setting the PRIMARY REGION as eu-north-1 on database no_region_db
+HINT: to change the default primary region, use SET CLUSTER SETTING sql.defaults.primary_region = 'region' or use RESET CLUSTER SETTING sql.defaults.primary_region to disable this behavior
+NOTICE: setting the PRIMARY REGION as eu-north-1 on database no_region_db_2
+HINT: to change the default primary region, use SET CLUSTER SETTING sql.defaults.primary_region = 'region' or use RESET CLUSTER SETTING sql.defaults.primary_region to disable this behavior
+
+query-sql
+SHOW DATABASES;
+----
+defaultdb root eu-north-1 {eu-north-1} zone
+no_region_db root eu-north-1 {eu-north-1} zone
+no_region_db_2 root eu-north-1 {eu-north-1} zone
+postgres root eu-north-1 {eu-north-1} zone
+system node <nil> {} <nil>
+
+query-sql
+USE no_region_db;
+SELECT schema_name, table_name, type, owner, locality FROM [SHOW TABLES];
+----
+public t table root REGIONAL BY TABLE IN PRIMARY REGION
+
+# Check we can restore without triggering the default primary region.
+exec-sql
+RESTORE DATABASE eu_central_db FROM LATEST IN 'nodelocal://1/eu_central_database_backup/' with schema_only;
+----
+
+query-sql
+SHOW DATABASES;
+----
+defaultdb root eu-north-1 {eu-north-1} zone
+eu_central_db root eu-central-1 {eu-central-1} zone
+no_region_db root eu-north-1 {eu-north-1} zone
+no_region_db_2 root eu-north-1 {eu-north-1} zone
+postgres root eu-north-1 {eu-north-1} zone
+system node <nil> {} <nil>
diff --git a/pkg/jobs/jobspb/jobs.proto b/pkg/jobs/jobspb/jobs.proto
index 1dbb6e204cb5..4cd8d0efdbd1 100644
--- a/pkg/jobs/jobspb/jobs.proto
+++ b/pkg/jobs/jobspb/jobs.proto
@@ -381,17 +381,12 @@ message RestoreDetails {
   // it is only valid to set this if len(tenants) == 1.
   roachpb.TenantID pre_rewrite_tenant_id = 23;
 
-  // RestoreValidation determines whether to skip certain parts of the restore
-  // job if its only purpose is to validate the user's restore command.
-  RestoreValidation validation = 24;
+  // SchemaOnly determines whether to only restore the schema in the backup.
+  bool SchemaOnly = 24;
 
   // NEXT ID: 25.
 }
 
-enum RestoreValidation {
-  DefaultRestore = 0;
-}
-
 
 
 
diff --git a/pkg/sql/execinfrapb/processors_bulk_io.proto b/pkg/sql/execinfrapb/processors_bulk_io.proto
index 6b18e8216a5d..bdbf943fe2de 100644
--- a/pkg/sql/execinfrapb/processors_bulk_io.proto
+++ b/pkg/sql/execinfrapb/processors_bulk_io.proto
@@ -302,7 +302,7 @@ message RestoreDataSpec {
   // PKIDs is used to convert result from an ExportRequest into row count
   // information passed back to track progress in the backup job.
   map<uint64, bool> pk_ids = 4 [(gogoproto.customname) = "PKIDs"];
-  optional jobs.jobspb.RestoreValidation validation = 7 [(gogoproto.nullable) = false];
+  optional bool schema_only = 7 [(gogoproto.nullable) = false];
 
   // NEXT ID: 8.
 }
@@ -316,7 +316,7 @@ message SplitAndScatterSpec {
   repeated RestoreEntryChunk chunks = 1 [(gogoproto.nullable) = false];
   repeated TableRekey table_rekeys = 2 [(gogoproto.nullable) = false];
   repeated TenantRekey tenant_rekeys = 3 [(gogoproto.nullable) = false];
-  optional jobs.jobspb.RestoreValidation validation = 5 [(gogoproto.nullable) = false];
+  optional bool schema_only = 5 [(gogoproto.nullable) = false];
 
   // NEXTID: 6.
 }
diff --git a/pkg/sql/parser/sql.y b/pkg/sql/parser/sql.y
index 03351e83b417..122f4056b286 100644
--- a/pkg/sql/parser/sql.y
+++ b/pkg/sql/parser/sql.y
@@ -908,7 +908,8 @@ func (u *sqlSymUnion) routineBody() *tree.RoutineBody {
 %token <str> RELEASE RESET RESTART RESTORE RESTRICT RESTRICTED RESUME RETURNING RETURN RETURNS RETRY REVISION_HISTORY
 %token <str> REVOKE RIGHT ROLE ROLES ROLLBACK ROLLUP ROUTINES ROW ROWS RSHIFT RULE RUNNING
 
-%token <str> SAVEPOINT SCANS SCATTER SCHEDULE SCHEDULES SCROLL SCHEMA SCHEMAS SCRUB SEARCH SECOND SECONDARY SECURITY SELECT SEQUENCE SEQUENCES
+%token <str> SAVEPOINT SCANS SCATTER SCHEDULE SCHEDULES SCROLL SCHEMA SCHEMA_ONLY SCHEMAS SCRUB
+%token <str> SEARCH SECOND SECONDARY SECURITY SELECT SEQUENCE SEQUENCES
 %token <str> SERIALIZABLE SERVER SESSION SESSIONS SESSION_USER SET SETOF SETS SETTING SETTINGS
 %token <str> SHARE SHOW SIMILAR SIMPLE SKIP SKIP_LOCALITIES_CHECK SKIP_MISSING_FOREIGN_KEYS
 %token <str> SKIP_MISSING_SEQUENCES SKIP_MISSING_SEQUENCE_OWNERS SKIP_MISSING_VIEWS SMALLINT SMALLSERIAL SNAPSHOT SOME SPLIT SQL
@@ -3446,7 +3447,10 @@ restore_options:
   {
     $$.val = &tree.RestoreOptions{AsTenant: $3.expr()}
   }
-
+| SCHEMA_ONLY
+	{
+		$$.val = &tree.RestoreOptions{SchemaOnly: true}
+	}
 import_format:
   name
   {
@@ -14887,6 +14891,7 @@ unreserved_keyword:
 | RUNNING
 | SCHEDULE
 | SCHEDULES
+| SCHEMA_ONLY
 | SCROLL
 | SETTING
 | SETTINGS
diff --git a/pkg/sql/parser/testdata/backup_restore b/pkg/sql/parser/testdata/backup_restore
index a9409e1a45ef..c56976f2fd6c 100644
--- a/pkg/sql/parser/testdata/backup_restore
+++ b/pkg/sql/parser/testdata/backup_restore
@@ -518,6 +518,14 @@ RESTORE DATABASE foo FROM ('bar') WITH new_db_name = ('baz') -- fully parenthesi
 RESTORE DATABASE foo FROM '_' WITH new_db_name = '_' -- literals removed
 RESTORE DATABASE _ FROM 'bar' WITH new_db_name = 'baz' -- identifiers removed
 
+parse
+RESTORE DATABASE foo FROM 'bar' WITH schema_only
+----
+RESTORE DATABASE foo FROM 'bar' WITH schema_only
+RESTORE DATABASE foo FROM ('bar') WITH schema_only -- fully parenthesized
+RESTORE DATABASE foo FROM '_' WITH schema_only -- literals removed
+RESTORE DATABASE _ FROM 'bar' WITH schema_only -- identifiers removed
+
 parse
 RESTORE DATABASE foo FROM 'bar' IN LATEST WITH incremental_location = 'baz'
 ----
diff --git a/pkg/sql/schemachanger/sctest/cumulative.go b/pkg/sql/schemachanger/sctest/cumulative.go
index 83dea9431cba..a706b6200602 100644
--- a/pkg/sql/schemachanger/sctest/cumulative.go
+++ b/pkg/sql/schemachanger/sctest/cumulative.go
@@ -468,28 +468,38 @@ func Backup(t *testing.T, dir string, newCluster NewClusterFunc) {
 		t.Logf("finished")
 
 		for i, b := range backups {
-			t.Run("", func(t *testing.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'")
-				tdb.Exec(t, fmt.Sprintf("RESTORE DATABASE %s FROM LATEST IN '%s'", dbName, b.url))
-				tdb.Exec(t, fmt.Sprintf("USE %q", dbName))
-				waitForSchemaChangesToFinish(t, tdb)
-				afterRestore := tdb.QueryStr(t, fetchDescriptorStateQuery)
-				if b.isRollback {
-					require.Equal(t, before, afterRestore)
-				} else {
-					require.Equal(t, after, afterRestore)
+			for _, isSchemaOnly := range []bool{true, false} {
+				name := ""
+				if isSchemaOnly {
+					name = "schema-only"
 				}
-				// Hack to deal with corrupt userfiles tables due to #76764.
-				const validateQuery = `
+				t.Run(name, func(t *testing.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)
+					tdb.Exec(t, fmt.Sprintf("USE %q", dbName))
+					waitForSchemaChangesToFinish(t, tdb)
+					afterRestore := tdb.QueryStr(t, fetchDescriptorStateQuery)
+					if b.isRollback {
+						require.Equal(t, before, afterRestore)
+					} else {
+						require.Equal(t, after, afterRestore)
+					}
+					// Hack to deal with corrupt userfiles tables due to #76764.
+					const validateQuery = `
 SELECT * FROM crdb_internal.invalid_objects WHERE database_name != 'backups'
 `
-				tdb.CheckQueryResults(t, validateQuery, [][]string{})
-				tdb.Exec(t, fmt.Sprintf("DROP DATABASE %q CASCADE", dbName))
-				tdb.Exec(t, "USE backups")
-				tdb.CheckQueryResults(t, validateQuery, [][]string{})
-			})
+					tdb.CheckQueryResults(t, validateQuery, [][]string{})
+					tdb.Exec(t, fmt.Sprintf("DROP DATABASE %q CASCADE", dbName))
+					tdb.Exec(t, "USE backups")
+					tdb.CheckQueryResults(t, validateQuery, [][]string{})
+				})
+			}
 		}
 	}
 	cumulativeTest(t, dir, testFunc)
diff --git a/pkg/sql/sem/tree/backup.go b/pkg/sql/sem/tree/backup.go
index 6a68a755148a..c82787c7532c 100644
--- a/pkg/sql/sem/tree/backup.go
+++ b/pkg/sql/sem/tree/backup.go
@@ -137,6 +137,7 @@ type RestoreOptions struct {
 	NewDBName                 Expr
 	IncrementalStorage        StringOrPlaceholderOptList
 	AsTenant                  Expr
+	SchemaOnly                bool
 }
 
 var _ NodeFormatter = &RestoreOptions{}
@@ -407,6 +408,10 @@ func (o *RestoreOptions) Format(ctx *FmtCtx) {
 		ctx.WriteString("tenant = ")
 		ctx.FormatNode(o.AsTenant)
 	}
+	if o.SchemaOnly {
+		maybeAddSep()
+		ctx.WriteString("schema_only")
+	}
 }
 
 // CombineWith merges other backup options into this backup options struct.
@@ -502,6 +507,13 @@ func (o *RestoreOptions) CombineWith(other *RestoreOptions) error {
 		return errors.New("tenant option specified multiple times")
 	}
 
+	if o.SchemaOnly {
+		if other.SchemaOnly {
+			return errors.New("schema_only option specified multiple times")
+		}
+	} else {
+		o.SchemaOnly = other.SchemaOnly
+	}
 	return nil
 }
 
@@ -520,7 +532,8 @@ func (o RestoreOptions) IsDefault() bool {
 		o.DebugPauseOn == options.DebugPauseOn &&
 		o.NewDBName == options.NewDBName &&
 		cmp.Equal(o.IncrementalStorage, options.IncrementalStorage) &&
-		o.AsTenant == options.AsTenant
+		o.AsTenant == options.AsTenant &&
+		o.SchemaOnly == options.SchemaOnly
 }
 
 // BackupTargetList represents a list of targets.