Merge #153790 #153792

153790: catalog/replication: delete existing descriptors before updates r=fqazi a=msbutler

Fixes #153788

Release note: none

153792: changefeedccl: disable roachtest on IBM due to missing Kafka support r=rickystewart,log-head a=aerfrei

Marks the 10 minute tpcc-100 test as disabled on IBM platforms, since kafka is not supported on s390x.

Fixes: #153759

Release note: None

Co-authored-by: Michael Butler <[email protected]>
Co-authored-by: Aerin Freilich <[email protected]>

Author: 3 people

pkg/cmd/roachtest/tests/cdc.go

@@ -3031,12 +3031,13 @@ func registerCDC(r registry.Registry) {
 		Run:              runCDCMultipleSchemaChanges,
 	})
 	r.Add(registry.TestSpec{
-		Name:             "cdc/tpcc-100/10min/sink=kafka/envelope=enriched",
-		Owner:            registry.OwnerCDC,
-		Benchmark:        true,
-		Cluster:          r.MakeClusterSpec(4, spec.WorkloadNode(), spec.CPU(16)),
-		Leases:           registry.MetamorphicLeases,
-		CompatibleClouds: registry.AllClouds,
+		Name:      "cdc/tpcc-100/10min/sink=kafka/envelope=enriched",
+		Owner:     registry.OwnerCDC,
+		Benchmark: true,
+		Cluster:   r.MakeClusterSpec(4, spec.WorkloadNode(), spec.CPU(16)),
+		Leases:    registry.MetamorphicLeases,
+		// Disabled on IBM due to lack of Kafka support on s390x.
+		CompatibleClouds: registry.AllClouds.NoIBM(),
 		Suites:           registry.Suites(registry.Nightly),
 		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 			ct := newCDCTester(ctx, t, c)

pkg/sql/catalog/replication/reader_catalog.go

@@ -133,25 +133,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.GetName(), 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.GetName(), desc.GetParentID()) ||
@@ -178,6 +166,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.GetName(), 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")
 			}