ccl: add an option to perform a regionless RESTORE

This patch adds the ability to restore a multi-region database onto a
single region cluster by adding a new option for the RESTORE syntax,
`strip_localities`. This option will strip RegionConfigs from
databases and LocalityConfigs from tables.

Epic: none
Fixes: cockroachdb#105623
Release note (sql change): A new option for the RESTORE
syntax, `strip_localities`, has been added. This can be used to
strip the locality information from a backup when there
are mismatched cluster regions between the backup's
cluster and the target cluster.

The following are behaviors that will most likely not be encountered
with the specific use case that this patch provides, but are
documented nonetheless:

Note: Adding a primary region to a regionless restore (with or
without regional by row table(s)) will not work out-of-the-box, but
does produce an accurate message detailing a user to essentially
`DROP TYPE [database].public.crdb_internal_region;` (and for cluster
restores, `ALTER DATABASE [database] CONFIGURE ZONE DISCARD;`).

Note: Restoring a cluster/database/table with a regional by row table
will not work out-of-the box. In particular, if we are performing
writes, the `crdb_region` column needs to specify the region of the
new row(s) being written to the table. The user will need to alter
said column and set a default that makes sense, along with discarding
the zone config (this latter is due to the fact that the zone config
holds all outdated info related to the partitions, constraints,
etc.).

These are due to a conflict with the `crdb_region` column already
being present in the regionless restore.  This column specifies each
row's home region and is a prefix to the table's primary key.
Stripping localities does not touch this column as it would be an
expensive operation that includes rewriting the entire table
([internal discussion here](https://cockroachlabs.slack.com/archives/C04N0AS14CT/p1694618761011619)).

docs/generated/sql/bnf/restore_options.bnf

@@ -20,3 +20,4 @@ restore_options ::=
 	| 'VERIFY_BACKUP_TABLE_DATA'
 	| 'UNSAFE_RESTORE_INCOMPATIBLE_VERSION'
 	| 'EXECUTION' 'LOCALITY' '=' string_or_placeholder
+	| 'REMOVE_REGIONS'

docs/generated/sql/bnf/stmt_block.bnf

@@ -1366,6 +1366,7 @@ unreserved_keyword ::=
 	| 'SKIP_MISSING_SEQUENCE_OWNERS'
 	| 'SKIP_MISSING_VIEWS'
 	| 'SKIP_MISSING_UDFS'
+	| 'STRIP_LOCALITIES'
 	| 'SNAPSHOT'
 	| 'SPLIT'
 	| 'SQL'
@@ -2623,6 +2624,7 @@ restore_options ::=
 	| 'VERIFY_BACKUP_TABLE_DATA'
 	| 'UNSAFE_RESTORE_INCOMPATIBLE_VERSION'
 	| 'EXECUTION' 'LOCALITY' '=' string_or_placeholder
+	| 'REMOVE_REGIONS'
 
 scrub_option_list ::=
 	( scrub_option ) ( ( ',' scrub_option ) )*
@@ -3917,6 +3919,7 @@ bare_label_keywords ::=
 	| 'SKIP_MISSING_SEQUENCE_OWNERS'
 	| 'SKIP_MISSING_UDFS'
 	| 'SKIP_MISSING_VIEWS'
+	| 'STRIP_LOCALITIES'
 	| 'SMALLINT'
 	| 'SNAPSHOT'
 	| 'SOME'

pkg/ccl/backupccl/BUILD.bazel

@@ -182,6 +182,7 @@ go_test(
         "partitioned_backup_test.go",
         "restore_data_processor_test.go",
         "restore_mid_schema_change_test.go",
+        "restore_multiregion_rbr_test.go",
         "restore_old_sequences_test.go",
         "restore_old_versions_test.go",
         "restore_planning_test.go",

pkg/ccl/backupccl/restore_job.go

@@ -988,6 +988,17 @@ func createImportingDescriptors(
 		typesByID[types[i].GetID()] = types[i]
 	}
 
+	if details.StripLocalities {
+		// Can't restore multi-region tables into non-multi-region database
+		for _, t := range tables {
+			t.TableDesc().LocalityConfig = nil
+		}
+
+		for _, d := range databases {
+			d.DatabaseDesc().RegionConfig = nil
+		}
+	}
+
 	// Collect all databases, for doing lookups of whether a database is new when
 	// updating schema references later on.
 	dbsByID := make(map[descpb.ID]catalog.DatabaseDescriptor)
@@ -1012,6 +1023,17 @@ func createImportingDescriptors(
 				if regionTypeDesc == nil {
 					continue
 				}
+
+				// When stripping localities, there is no longer a need for a region config. In addition,
+				// we need to make sure that multi-region databases no longer get tagged as such - meaning
+				// that we want to change the TypeDescriptor_MULTIREGION_ENUM to a normal enum. We `continue`
+				// to skip the multi-region work below.
+				if details.StripLocalities {
+					t.TypeDesc().Kind = descpb.TypeDescriptor_ENUM
+					t.TypeDesc().RegionConfig = nil
+					continue
+				}
+
 				// Check to see if we've found more than one multi-region enum on any
 				// given database.
 				if id, ok := mrEnumsFound[regionTypeDesc.GetParentID()]; ok {
@@ -1038,6 +1060,7 @@ func createImportingDescriptors(
 						var regionNames []catpb.RegionName
 						_ = regionTypeDesc.ForEachPublicRegion(func(name catpb.RegionName) error {
 							regionNames = append(regionNames, name)
+
 							return nil
 						})
 						regionConfig := multiregion.MakeRegionConfig(

pkg/ccl/backupccl/restore_multiregion_rbr_test.go

@@ -0,0 +1,100 @@
+// Copyright 2023 The Cockroach Authors.
+//
+// Licensed as a CockroachDB Enterprise file under the Cockroach Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//     https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt
+
+package backupccl
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl/backuptestutils"
+	"github.com/cockroachdb/cockroach/pkg/ccl/multiregionccl/multiregionccltestutils"
+	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+// The goal of this test is to ensure that if a user ever performed a
+// regionless restore where the backed-up target has a regional by row table,
+// they would be able to get themselves out of a stuck state without needing
+// an enterprise license (in addition to testing the ability to use strip_localities
+// without said license).
+func TestMultiRegionRegionlessRestoreNoLicense(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	skip.UnderStressRace(t, "test is too heavy to run under stress")
+
+	ctx := context.Background()
+
+	dir, dirCleanupfn := testutils.TempDir(t)
+	defer dirCleanupfn()
+
+	_, mrSqlDB, cleanup := multiregionccltestutils.TestingCreateMultiRegionCluster(
+		t, 3 /* numServers */, base.TestingKnobs{}, multiregionccltestutils.WithBaseDirectory(dir),
+	)
+	defer cleanup()
+	mrSql := sqlutils.MakeSQLRunner(mrSqlDB)
+
+	// Create the database & table, insert some values.
+	mrSql.Exec(t,
+		`CREATE DATABASE d PRIMARY REGION "us-east1" REGIONS "us-east2";
+            CREATE TABLE d.t (x INT);
+            INSERT INTO d.t VALUES (1), (2), (3);`,
+	)
+
+	// Make table regional by row.
+	mrSql.Exec(t, `ALTER TABLE d.t SET LOCALITY REGIONAL BY ROW;`)
+
+	if err := backuptestutils.VerifyBackupRestoreStatementResult(t, mrSql, `BACKUP DATABASE d INTO $1`, localFoo); err != nil {
+		t.Fatal(err)
+	}
+
+	defer utilccl.TestingDisableEnterprise()()
+
+	sqlTC := testcluster.StartTestCluster(
+		t, singleNode, base.TestClusterArgs{ServerArgs: base.TestServerArgs{
+			ExternalIODir:     dir,
+			DefaultTestTenant: base.TestControlsTenantsExplicitly,
+			Knobs: base.TestingKnobs{
+				JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
+				TenantTestingKnobs: &sql.TenantTestingKnobs{
+					// This test expects a specific tenant ID to be selected after DROP TENANT.
+					EnableTenantIDReuse: true,
+				},
+			},
+		}},
+	)
+	defer sqlTC.Stopper().Stop(ctx)
+	sqlDB := sqlutils.MakeSQLRunner(sqlTC.Conns[0])
+
+	if err := backuptestutils.VerifyBackupRestoreStatementResult(t, sqlDB, `RESTORE DATABASE d FROM LATEST IN $1 WITH strip_localities`, localFoo); err != nil {
+		t.Fatal(err)
+	}
+
+	// Get us in the state that allows us to perform writes.
+	// This is the main purpose of this test - we want to ensure that this process is available
+	// to those without enterprise licenses.
+	sqlDB.Exec(t, `ALTER TABLE d.t ALTER COLUMN crdb_region SET DEFAULT 'us-east1';
+                        ALTER TABLE d.t CONFIGURE ZONE DISCARD;`)
+
+	// Perform some writes to d's table.
+	sqlDB.Exec(t, `INSERT INTO d.t VALUES (4), (5), (6)`)
+
+	var rowCount int
+	sqlDB.QueryRow(t, `SELECT count(x) FROM d.t`).Scan(&rowCount)
+	require.Equal(t, 6, rowCount)
+}

pkg/ccl/backupccl/restore_planning.go

@@ -920,6 +920,7 @@ func resolveOptionsForRestoreJobDescription(
 		SchemaOnly:                       opts.SchemaOnly,
 		VerifyData:                       opts.VerifyData,
 		UnsafeRestoreIncompatibleVersion: opts.UnsafeRestoreIncompatibleVersion,
+		RemoveRegions:                    opts.RemoveRegions,
 	}
 
 	if opts.EncryptionPassphrase != nil {
@@ -1921,6 +1922,15 @@ func doRestorePlan(
 		}
 	}
 
+	// If we are stripping localities, wipe tables of their LocalityConfig before we allocate
+	// descriptor rewrites - as validation in remapTables compares these tables with the non-mr
+	// database and fails otherwise
+	if restoreStmt.Options.StripLocalities {
+		for _, t := range filteredTablesByID {
+			t.TableDesc().LocalityConfig = nil
+		}
+	}
+
 	descriptorRewrites, err := allocateDescriptorRewrites(
 		ctx,
 		p,
@@ -2036,6 +2046,7 @@ func doRestorePlan(
 		VerifyData:          restoreStmt.Options.VerifyData,
 		SkipLocalitiesCheck: restoreStmt.Options.SkipLocalitiesCheck,
 		ExecutionLocality:   execLocality,
+		RemoveRegions:       restoreStmt.Options.RemoveRegions,
 	}
 
 	jr := jobs.Record{