sql: fix namespace table name collision in backup

As part of the namespace table migration to support schemas, we changed
the name in the old namespace table's descriptor from "namespace" to
"namespace_deprecated". However, we only made this change for new
clusters, not clusters that were migrated from 19.2. This resulted in
breaking backup on migrated clusters, since it expects that table names
are unique.

I supplemented our namespace table migration to retrieve the deprecated
namespace table's descriptor and update its name as desired. I also
updated the existing migration to use Puts rather than CPuts since it
should be fine to overwrite the existing KVs.

Fixes #43979

Release note: None

pkg/sqlmigrations/migrations.go

@@ -663,8 +663,22 @@ func createProtectedTimestampsRecordsTable(ctx context.Context, r runner) error
 }
 
 func createNewSystemNamespaceDescriptor(ctx context.Context, r runner) error {
-	err := r.db.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {
+
+	return r.db.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {
 		b := txn.NewBatch()
+
+		// Retrieve the existing namespace table's descriptor and change its name to
+		// "namespace_deprecated". This prevents name collisions with the new
+		// namespace table, for instance during backup.
+		deprecatedKey := sqlbase.MakeDescMetadataKey(keys.DeprecatedNamespaceTableID)
+		deprecatedDesc := &sqlbase.Descriptor{}
+		ts, err := txn.GetProtoTs(ctx, deprecatedKey, deprecatedDesc)
+		if err != nil {
+			return err
+		}
+		deprecatedDesc.Table(ts).Name = sqlbase.DeprecatedNamespaceTable.Name
+		b.Put(deprecatedKey, deprecatedDesc)
+
 		// The 19.2 namespace table contains an entry for "namespace" which maps to
 		// the deprecated namespace tables ID. Even though the cluster version at
 		// this point is 19.2, we construct a metadata name key in the 20.1 format.
@@ -676,17 +690,13 @@ func createNewSystemNamespaceDescriptor(ctx context.Context, r runner) error {
 		//    idempotent semantics of the migration ensure that "namespace" maps to
 		//    the correct ID in the new system.namespace table after all tables are
 		//    copied over.
-		nameKey := sqlbase.NewPublicTableKey(sqlbase.NamespaceTable.GetParentID(), sqlbase.NamespaceTable.GetName())
-		b.CPut(nameKey.Key(), sqlbase.NamespaceTable.GetID(), nil)
-		b.CPut(sqlbase.MakeDescMetadataKey(sqlbase.NamespaceTable.GetID()), sqlbase.WrapDescriptor(&sqlbase.NamespaceTable), nil)
+		nameKey := sqlbase.NewPublicTableKey(
+			sqlbase.NamespaceTable.GetParentID(), sqlbase.NamespaceTable.GetName())
+		b.Put(nameKey.Key(), sqlbase.NamespaceTable.GetID())
+		b.Put(sqlbase.MakeDescMetadataKey(
+			sqlbase.NamespaceTable.GetID()), sqlbase.WrapDescriptor(&sqlbase.NamespaceTable))
 		return txn.Run(ctx, b)
 	})
-	// CPuts only provide idempotent inserts if we ignore the errors that arise
-	// when the condition isn't met.
-	if _, ok := err.(*roachpb.ConditionFailedError); ok {
-		return nil
-	}
-	return err
 }
 
 // migrateSystemNamespace migrates entries from the deprecated system.namespace

pkg/sqlmigrations/migrations_test.go

@@ -705,3 +705,53 @@ func TestUpdateSystemLocationData(t *testing.T) {
 		t.Fatalf("Exected to find 0 rows in system.locations. Found  %d instead", count)
 	}
 }
+
+func TestMigrateNamespaceTableDescriptors(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	ctx := context.Background()
+
+	mt := makeMigrationTest(ctx, t)
+	defer mt.close(ctx)
+
+	migration := mt.pop(t, "create new system.namespace table")
+	mt.start(t, base.TestServerArgs{})
+
+	// Since we're already on 20.1, mimic the beginning state by deleting the
+	// new namespace descriptor and changing the old one's name to "namespace".
+	key := sqlbase.MakeDescMetadataKey(keys.NamespaceTableID)
+	require.NoError(t, mt.kvDB.Del(ctx, key))
+
+	deprecatedKey := sqlbase.MakeDescMetadataKey(keys.DeprecatedNamespaceTableID)
+	desc := &sqlbase.Descriptor{}
+	require.NoError(t, mt.kvDB.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {
+		ts, err := txn.GetProtoTs(ctx, deprecatedKey, desc)
+		require.NoError(t, err)
+		desc.Table(ts).Name = sqlbase.NamespaceTable.Name
+		return txn.Put(ctx, deprecatedKey, desc)
+	}))
+
+	// Run the migration.
+	require.NoError(t, mt.runMigration(ctx, migration))
+
+	require.NoError(t, mt.kvDB.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {
+		// Check that the persisted descriptors now match our in-memory versions,
+		// ignoring create and modification times.
+		{
+			ts, err := txn.GetProtoTs(ctx, key, desc)
+			require.NoError(t, err)
+			table := desc.Table(ts)
+			table.CreateAsOfTime = sqlbase.NamespaceTable.CreateAsOfTime
+			table.ModificationTime = sqlbase.NamespaceTable.ModificationTime
+			require.True(t, table.Equal(sqlbase.NamespaceTable))
+		}
+		{
+			ts, err := txn.GetProtoTs(ctx, deprecatedKey, desc)
+			require.NoError(t, err)
+			table := desc.Table(ts)
+			table.CreateAsOfTime = sqlbase.DeprecatedNamespaceTable.CreateAsOfTime
+			table.ModificationTime = sqlbase.DeprecatedNamespaceTable.ModificationTime
+			require.True(t, table.Equal(sqlbase.DeprecatedNamespaceTable))
+		}
+		return nil
+	}))
+}