Merge #145972

145972: catalog/replication: fix validation errors when updating catalog r=fqazi a=fqazi

Previously, updating PCR reader catalog descriptors involved processing and flushing one at a time. This worked unless a subsequent descriptor depended on the current one. However, it could fail when, for example, removing a column dependency (i.e., ALTER COLUMN .. SET DEFAULT NULL). To address this, this patch first generates the mutated descriptors and then flushes them into the collections.

Fixes: #145971

Release note (bug fix): addressed a bug where the PCR reader catalog job could hit validation errors, when schema object dependencies between them (for example a sequence default expression was being removed).

Co-authored-by: Faizan Qazi <[email protected]>

Author: craig[bot]

pkg/sql/catalog/descs/collection.go

@@ -334,8 +334,9 @@ func (tc *Collection) WriteDescToBatch(
 	}
 	desc.MaybeIncrementVersion()
 	// Replicated PCR descriptors cannot be modified unless the collection
-	// is setup for updating them.
-	if !tc.readerCatalogSetup && desc.GetReplicatedPCRVersion() != 0 {
+	// is setup for updating them. If write validation is disabled then, allow
+	// PCR reader catalog descriptors to be modified, otherwise repair is impossible.
+	if !tc.readerCatalogSetup && desc.GetReplicatedPCRVersion() != 0 && !tc.skipValidationOnWrite {
 		return pgerror.Newf(pgcode.ReadOnlySQLTransaction,
 			"replicated %s %s (%d) cannot be mutated",
 			desc.GetObjectTypeString(),

pkg/sql/catalog/replication/reader_catalog.go

@@ -60,6 +60,9 @@ func SetupOrAdvanceStandbyReaderCatalog(
 			// Resolve any existing descriptors within the tenant, which
 			// will be use to compute old values for writing.
 			b := txn.KV().NewBatch()
+			// Batch the write updates, since we mutate descriptors one at a time
+			// below.
+			descriptorsToWrite := make([]catalog.MutableDescriptor, 0, len(allExistingDescs.OrderedDescriptorIDs()))
 			if err := extracted.ForEachDescriptor(func(fromDesc catalog.Descriptor) error {
 				if !shouldSetupForReader(fromDesc.GetID(), fromDesc.GetParentID()) {
 					return nil
@@ -71,7 +74,7 @@ func SetupOrAdvanceStandbyReaderCatalog(
 				existingDesc, err := txn.Descriptors().MutableByID(txn.KV()).Desc(ctx, fromDesc.GetID())
 				if err != nil &&
 					!errors.Is(err, catalog.ErrDescriptorNotFound) {
-					return err
+					return errors.Wrapf(err, "failed to get mutable descriptor")
 				} else {
 					err = nil
 				}
@@ -118,11 +121,22 @@ func SetupOrAdvanceStandbyReaderCatalog(
 						descriptorsRenamed.Add(existingDesc.GetID())
 					}
 				}
-				return errors.Wrapf(txn.Descriptors().WriteDescToBatch(ctx, true, mut, b),
-					"unable to create replicated descriptor: %d %T", mut.GetID(), mut)
+				descriptorsToWrite = append(descriptorsToWrite, mut)
+				return nil
 			}); err != nil {
 				return err
 			}
+			// Write all the descriptors into a single batch, we previously would write
+			// them in the loop above. But that interferes with mutable descriptor
+			// validation, which will include uncommitted descriptors. For example,
+			// 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

pkg/sql/catalog/replication/reader_catalog_test.go

@@ -276,6 +276,8 @@ INSERT INTO t3(n) VALUES (3);
 	// Manipulate the schema first.
 	ddlToExec = []string{
 		"ALTER TABLE t1 ADD COLUMN j int default 32",
+		"CREATE SEQUENCE sq2;",
+		"ALTER TABLE t1 ALTER COLUMN n SET default nextval('sq2')",
 		"INSERT INTO t1(val, j) VALUES('open', 1);",
 		"INSERT INTO t1(val, j) VALUES('closed', 2);",
 		"INSERT INTO t1(val, j) VALUES('inactive', 3);",
@@ -301,6 +303,19 @@ INSERT INTO t3(n) VALUES (3);
 	r.destRunner.ExpectErr(t, "schema changes are not allowed on a reader catalog", "ALTER SEQUENCE sq1 RENAME TO sq4")
 	r.destRunner.ExpectErr(t, "schema changes are not allowed on a reader catalog", "ALTER TYPE status ADD VALUE 'newval' ")
 
+	// As a final test intentionally drop dependencies between descriptors. If we
+	// don't batch descriptor updates this will cause a validation error, since
+	// the sequence and table depend on each other.
+	ddlToExec = []string{
+		"ALTER TABLE t1 ALTER COLUMN n SET default NULL",
+	}
+	for _, ddl := range ddlToExec {
+		r.srcRunner.Exec(t, ddl)
+	}
+	// Confirm that everything matches at the old timestamp.
+	require.NoError(t, r.advanceTS(ctx, r.ts.Clock().Now(), true))
+	r.compareEqual(t, "SELECT * FROM v1 ORDER BY 1")
+	r.compareEqual(t, "SELECT * FROM t2 ORDER BY j")
 }
 
 // TestReaderCatalogTSAdvance tests repeated advances of timestamp