catalog/replication: delete existing descriptors before updates

Fixes #153788

Release note: none

Author: msbutler

pkg/sql/catalog/replication/reader_catalog.go

@@ -132,25 +132,13 @@ func SetupOrAdvanceStandbyReaderCatalog(
 			// if a table and sequence depend on each other, then updating one and
 			// fetching the other in a mutable way to remove a dependency will hit
 			// a validation error.
-			for _, mut := range descriptorsToWrite {
-				if err := txn.Descriptors().WriteDescToBatch(ctx, true, mut, b); err != nil {
-					return errors.Wrapf(err, "unable to create replicated descriptor: %d %T", mut.GetID(), mut)
-				}
-			}
-			if err := extracted.ForEachNamespaceEntry(func(e nstree.NamespaceEntry) error {
-				if !shouldSetupForReader(e.GetID(), e.GetParentID()) {
-					return nil
-				}
-				// Do not upsert entries if one already exists.
-				entry := allExistingDescs.LookupNamespaceEntry(catalog.MakeNameInfo(e))
-				if entry != nil && e.GetID() == entry.GetID() {
-					return nil
-				}
-				return errors.Wrapf(txn.Descriptors().UpsertNamespaceEntryToBatch(ctx, true, e, b), "namespace entry %v", e)
-			}); err != nil {
-				return err
-			}
+
 			// Figure out which descriptors should be deleted.
+			//
+			// NB: we issue deletes of existing descriptors/namespace entries before
+			// we upsert new ones in the batch to ensure that if we need to delete and
+			// upsert the same namespace entry but for a different table id, after the
+			// txn, the reader will see the upsert.
 			if err := allExistingDescs.ForEachDescriptor(func(desc catalog.Descriptor) error {
 				// Skip descriptors that were updated above
 				if !shouldSetupForReader(desc.GetID(), desc.GetParentID()) ||
@@ -177,6 +165,25 @@ func SetupOrAdvanceStandbyReaderCatalog(
 			}); err != nil {
 				return err
 			}
+
+			for _, mut := range descriptorsToWrite {
+				if err := txn.Descriptors().WriteDescToBatch(ctx, true, mut, b); err != nil {
+					return errors.Wrapf(err, "unable to create replicated descriptor: %d %T", mut.GetID(), mut)
+				}
+			}
+			if err := extracted.ForEachNamespaceEntry(func(e nstree.NamespaceEntry) error {
+				if !shouldSetupForReader(e.GetID(), e.GetParentID()) {
+					return nil
+				}
+				// Do not upsert entries if one already exists with the same ID.
+				entry := allExistingDescs.LookupNamespaceEntry(catalog.MakeNameInfo(e))
+				if entry != nil && e.GetID() == entry.GetID() {
+					return nil
+				}
+				return errors.Wrapf(txn.Descriptors().UpsertNamespaceEntryToBatch(ctx, true, e, b), "namespace entry %v", e)
+			}); err != nil {
+				return err
+			}
 			if err := maybeBlockSchemaChangesOnSystemDatabase(ctx, txn, b); err != nil {
 				return errors.Wrapf(err, "blocking schema changes")
 			}