sql/catalog/dbdesc: repair descriptors corrupted due to a dropped schema

This commit adds a mechanism to delete wrong schema-info entries from
database descriptors as part of post-deserialization changes.

Release note: None

pkg/ccl/backupccl/BUILD.bazel

@@ -177,6 +177,7 @@ go_test(
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catalogkv",
         "//pkg/sql/catalog/descpb",
+        "//pkg/sql/catalog/descs",
         "//pkg/sql/catalog/lease",
         "//pkg/sql/catalog/systemschema",
         "//pkg/sql/catalog/tabledesc",

pkg/ccl/backupccl/restore_old_versions_test.go

@@ -19,7 +19,14 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkv"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
@@ -503,3 +510,57 @@ func TestRestoreOldBackupMissingOfflineIndexes(t *testing.T) {
 		}
 	}
 }
+
+func TestRestoreWithDroppedSchemaCorruption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+
+	const (
+		dbName    = "foo"
+		backupDir = "testdata/restore_with_dropped_schema/exports/v20.2.7"
+		fromDir   = "nodelocal://0/"
+	)
+
+	args := base.TestServerArgs{ExternalIODir: backupDir}
+	s, sqlDB, _ := serverutils.StartServer(t, args)
+	tdb := sqlutils.MakeSQLRunner(sqlDB)
+	defer s.Stopper().Stop(ctx)
+
+	tdb.Exec(t, fmt.Sprintf("RESTORE DATABASE %s FROM '%s'", dbName, fromDir))
+	query := fmt.Sprintf("SELECT database_name FROM [SHOW DATABASES] WHERE database_name = '%s'", dbName)
+	tdb.CheckQueryResults(t, query, [][]string{{dbName}})
+
+	// Read descriptor without validation.
+	execCfg := s.ExecutorConfig().(sql.ExecutorConfig)
+	hasSameNameSchema := func(dbName string) bool {
+		exists := false
+		var desc catalog.DatabaseDescriptor
+		require.NoError(t, sql.DescsTxn(ctx, &execCfg, func(
+			ctx context.Context, txn *kv.Txn, collection *descs.Collection,
+		) error {
+			// Using this method to avoid validation.
+			allDescs, err := catalogkv.GetAllDescriptors(ctx, txn, execCfg.Codec, false)
+			if err != nil {
+				return err
+			}
+			for _, d := range allDescs {
+				if d.GetName() == dbName {
+					desc, err = catalog.AsDatabaseDescriptor(d)
+					require.NoError(t, err, "unable to cast to database descriptor")
+					return nil
+				}
+			}
+			return nil
+		}))
+		require.NoError(t, desc.ForEachSchemaInfo(
+			func(id descpb.ID, name string, isDropped bool) error {
+				if name == dbName {
+					exists = true
+				}
+				return nil
+			}))
+		return exists
+	}
+	require.Falsef(t, hasSameNameSchema(dbName), "corrupted descriptor exists")
+}

pkg/ccl/backupccl/testdata/restore_with_dropped_schema/create.sql

@@ -0,0 +1,14 @@
+-- The below SQL is used to create a backup of a database that
+-- contains a corrupted database descriptor. Data is produced
+-- using version 20.2.7. This backup is used in
+-- TestRestoreWithDroppedSchemaCorruption test.
+
+CREATE DATABASE foo;
+
+SET DATABASE = foo;
+
+CREATE SCHEMA bar;
+
+DROP SCHEMA bar;
+
+BACKUP DATABASE foo to 'nodelocal://0/foo_backup';

pkg/ccl/backupccl/testdata/restore_with_dropped_schema/exports/v20.2.7/BACKUP-CHECKPOINT-683775825115512833-CHECKSUM

@@ -0,0 +1 @@
+���T

pkg/ccl/backupccl/testdata/restore_with_dropped_schema/exports/v20.2.7/BACKUP-CHECKPOINT-CHECKSUM

@@ -0,0 +1 @@
+���T

pkg/ccl/backupccl/testdata/restore_with_dropped_schema/exports/v20.2.7/BACKUP_MANIFEST-CHECKSUM

@@ -0,0 +1 @@
+=�f

pkg/sql/catalog/dbdesc/BUILD.bazel

@@ -38,6 +38,7 @@ go_test(
         "//pkg/sql/parser",
         "//pkg/sql/sem/tree",
         "//pkg/util/leaktest",
+        "//pkg/util/log",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_stretchr_testify//require",
         "@in_gopkg_yaml_v2//:yaml_v2",

pkg/sql/catalog/dbdesc/database_desc.go

@@ -437,3 +437,18 @@ func (desc *Mutable) SetDefaultPrivilegeDescriptor(
 ) {
 	desc.DefaultPrivileges = defaultPrivilegeDescriptor
 }
+
+// maybeRemoveDroppedSelfEntryFromSchemas removes an entry in the Schemas map corresponding to the
+// database itself which was added due to a bug in prior versions when dropping any user-defined schema.
+// The bug inserted an entry for the database rather than the schema being dropped. This function fixes the
+// problem by deleting the erroneous entry.
+func maybeRemoveDroppedSelfEntryFromSchemas(dbDesc *descpb.DatabaseDescriptor) bool {
+	if dbDesc == nil {
+		return false
+	}
+	if sc, ok := dbDesc.Schemas[dbDesc.Name]; ok && sc.ID == dbDesc.ID {
+		delete(dbDesc.Schemas, dbDesc.Name)
+		return true
+	}
+	return false
+}

pkg/sql/catalog/dbdesc/database_desc_builder.go

@@ -58,8 +58,10 @@ func (ddb *databaseDescriptorBuilder) RunPostDeserializationChanges(
 	_ context.Context, _ catalog.DescGetter,
 ) error {
 	ddb.maybeModified = protoutil.Clone(ddb.original).(*descpb.DatabaseDescriptor)
-	ddb.changed = descpb.MaybeFixPrivileges(ddb.maybeModified.ID, ddb.maybeModified.ID,
+	privsChanged := descpb.MaybeFixPrivileges(ddb.maybeModified.ID, ddb.maybeModified.ID,
 		&ddb.maybeModified.Privileges, privilege.Database)
+	removedSelfEntryInSchemas := maybeRemoveDroppedSelfEntryFromSchemas(ddb.maybeModified)
+	ddb.changed = privsChanged || removedSelfEntryInSchemas
 	return nil
 }
 

pkg/sql/catalog/dbdesc/database_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/redact"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v2"
@@ -297,3 +298,30 @@ func TestValidateCrossDatabaseReferences(t *testing.T) {
 		}
 	}
 }
+
+// TestFixDroppedSchemaName tests fixing a corrupted descriptor as part of
+// RunPostDeserializationChanges. It tests for a particular corruption that
+// happened when a schema was dropped that had the same name as its parent
+// database name.
+func TestFixDroppedSchemaName(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	const (
+		dbName = "foo"
+		dbID   = 1
+	)
+	dbDesc := descpb.DatabaseDescriptor{
+		Name: dbName,
+		ID:   dbID,
+		Schemas: map[string]descpb.DatabaseDescriptor_SchemaInfo{
+			dbName: {ID: dbID, Dropped: true},
+		},
+	}
+	b := NewBuilder(&dbDesc)
+	require.NoError(t, b.RunPostDeserializationChanges(ctx, nil))
+	desc := b.BuildCreatedMutableDatabase()
+	require.Truef(t, desc.HasPostDeserializationChanges(), "expected changes in descriptor, found none")
+	_, ok := desc.Schemas[dbName]
+	require.Falsef(t, ok, "erroneous entry exists")
+}