ccl/cdc: change when drop column is detected in schema feed

Previously, before automatic schema_locked toggle was added in the
declarative schema changer, we had a guarantee that schema_locked would
be unset before any column mutations are added. Unfortunately, the
automatic toggle logic will disable schema_locked at the same time as
the column is removed. This doesn't play nice with the existing schema
feed logic. To address this, this patch detects the mutation once the
new primary index is backfilled, so that is separate from the
schema_locked being disabled. This is still valid behavior because the
schema change is still in flight at this point, and doesn't have a
usable new primary index.

Informs: #150003

Release note (bug fix): Addressed a bug on schema_locked tables when a
column is dropped, and schema_locked is toggled for the user.

Author: fqazi

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2180,6 +2180,54 @@ func TestChangefeedColumnDropsOnTheSameTableWithMultipleFamilies(t *testing.T) {
 	})
 }
 
+func TestChangefeedColumnDropsOnTheSameTableWithMultipleFamiliesWithManualSchemaLocked(
+	t *testing.T,
+) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	require.NoError(t, log.SetVModule("kv_feed=2,changefeed_processors=2"))
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+
+		sqlDB.Exec(t, `CREATE TABLE hasfams (id int primary key, a string, b string, c string, FAMILY id_a (id, a), FAMILY b_and_c (b, c))`)
+		sqlDB.Exec(t, `INSERT INTO hasfams values (0, 'a', 'b', 'c')`)
+
+		var args []any
+		if _, ok := f.(*webhookFeedFactory); ok {
+			args = append(args, optOutOfMetamorphicEnrichedEnvelope{reason: "metamorphic enriched envelope does not support column families for webhook sinks"})
+		}
+		// Open up the changefeed.
+		cf := feed(t, f, `CREATE CHANGEFEED FOR TABLE hasfams FAMILY id_a, TABLE hasfams FAMILY b_and_c`, args...)
+		defer closeFeed(t, cf)
+		assertPayloads(t, cf, []string{
+			`hasfams.b_and_c: [0]->{"after": {"b": "b", "c": "c"}}`,
+			`hasfams.id_a: [0]->{"after": {"a": "a", "id": 0}}`,
+		})
+
+		// Check that dropping a watched column will backfill the changefeed.
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=false)`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams DROP COLUMN a`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=true)`)
+		assertPayloads(t, cf, []string{
+			`hasfams.id_a: [0]->{"after": {"id": 0}}`,
+		})
+
+		// Check that dropping a watched column will backfill the changefeed.
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=false)`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams DROP COLUMN b`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=true)`)
+		assertPayloads(t, cf, []string{
+			`hasfams.b_and_c: [0]->{"after": {"c": "c"}}`,
+		})
+	}
+
+	runWithAndWithoutRegression141453(t, testFn, func(t *testing.T, testFn cdcTestFn) {
+		cdcTest(t, testFn)
+	})
+}
+
 func TestNoStopAfterNonTargetAddColumnWithBackfill(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -3203,7 +3251,7 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 				`drop_column: [2]->{"after": {"a": 2, "b": "2"}}`,
 			})
 			sqlDB.Exec(t, `ALTER TABLE drop_column DROP COLUMN b`)
-			ts := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `drop_column`, 2)
+			ts := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `drop_column`, 6)
 
 			// Backfill for DROP COLUMN b.
 			assertPayloads(t, dropColumn, []string{
@@ -3256,7 +3304,7 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 			// the 7th step (version 15). Finally, when adding column d, it goes from 17->25 ith the schema change
 			// being visible at the 7th step (version 23).
 			// TODO(#142936): Investigate if this descriptor version hardcoding is sound.
-			dropTS := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 2)
+			dropTS := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 6)
 			addTS := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 15)
 			addTS2 := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 23)
 

pkg/ccl/changefeedccl/schemafeed/table_event_filter.go

@@ -188,6 +188,26 @@ func shouldFilterAddColumnEvent(e TableEvent, targets changefeedbase.Targets) (b
 	return !watched, nil
 }
 
+// notDeclarativeOrHasMergedIndex returns true if the descriptor has a declarative
+// schema changer with a merged index.
+func notDeclarativeOrHasMergedIndex(desc catalog.TableDescriptor) bool {
+	// If there are not declarative schema changes then this will always be
+	// true.
+	if desc.GetDeclarativeSchemaChangerState() == nil {
+		return true
+	}
+	// For declarative schema changes detect when a new primary index becomes
+	// WRITE_ONLY (i.e. backfill has been completed).
+	for idx, target := range desc.GetDeclarativeSchemaChangerState().Targets {
+		if target.GetPrimaryIndex() != nil &&
+			target.TargetStatus == scpb.Status_PUBLIC &&
+			desc.GetDeclarativeSchemaChangerState().CurrentStatuses[idx] == scpb.Status_WRITE_ONLY {
+			return true
+		}
+	}
+	return false
+}
+
 // Returns true if the changefeed targets a column which has a drop mutation inside the table event.
 func droppedColumnIsWatched(e TableEvent, targets changefeedbase.Targets) (bool, error) {
 	// If no column families are specified, then all columns are targeted.
@@ -212,7 +232,13 @@ func droppedColumnIsWatched(e TableEvent, targets changefeedbase.Targets) (bool,
 		if m.AsColumn() == nil || m.AsColumn().IsHidden() {
 			continue
 		}
-		if m.Dropped() && m.WriteAndDeleteOnly() && watchedColumnIDs.Contains(int(m.AsColumn().GetID())) {
+		// For dropped columns wait for WriteAndDeleteOnly to be hit. When using
+		// the declarative schema changer we will wait a bit later in the plan to
+		// publish the dropped column, since schema_locked and the column being
+		// write and delete only happen at the same stage. Since the schema change
+		// is still in progress, there is a gray area in terms of when the change
+		// should be visible.
+		if m.Dropped() && m.WriteAndDeleteOnly() && notDeclarativeOrHasMergedIndex(e.After) && watchedColumnIDs.Contains(int(m.AsColumn().GetID())) {
 			return true, nil
 		}
 	}
@@ -273,7 +299,13 @@ func dropVisibleColumnMutationExists(desc catalog.TableDescriptor) bool {
 		if m.AsColumn() == nil || m.AsColumn().IsHidden() {
 			continue
 		}
-		if m.Dropped() && m.WriteAndDeleteOnly() {
+		// For dropped columns wait for WriteAndDeleteOnly to be hit. When using
+		// the declarative schema changer we will wait a bit later in the plan to
+		// publish the dropped column, since schema_locked and the column being
+		// write and delete only happen at the same stage. Since the schema change
+		// is still in progress, there is a gray area in terms of when the change
+		// should be visible.
+		if m.Dropped() && m.WriteAndDeleteOnly() && notDeclarativeOrHasMergedIndex(desc) {
 			return true
 		}
 	}

pkg/ccl/changefeedccl/schemafeed/testdata/drop_column

@@ -12,11 +12,11 @@ ALTER TABLE t DROP COLUMN j;
 
 pop f=1
 ----
-t 1->2: DropColumn
+t 1->2: Unknown
 t 2->3: Unknown
 t 3->4: Unknown
 t 4->5: Unknown
-t 5->6: Unknown
+t 5->6: DropColumn
 t 6->7: PrimaryKeyChange (no column changes)
 t 7->8: Unknown
 t 8->9: AddHiddenColumn