Merge pull request #151527 from fqazi/backport25.3-150435-150952

release-25.3: address change feed failure with drop column on a schema_locked  table

Author: fqazi

pkg/ccl/changefeedccl/cdcevent/BUILD.bazel

@@ -30,6 +30,7 @@ go_library(
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/row",
         "//pkg/sql/rowenc",
+        "//pkg/sql/sem/catid",
         "//pkg/sql/sem/eval",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondatapb",

pkg/ccl/changefeedccl/cdcevent/event.go

@@ -8,6 +8,7 @@ package cdcevent
 import (
 	"context"
 	"fmt"
+	"slices"
 	"strings"
 	"time"
 
@@ -21,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
@@ -292,7 +294,7 @@ func NewEventDescriptor(
 	}
 
 	// addColumn is a helper to add a column to this descriptor.
-	addColumn := func(col catalog.Column, ord int) int {
+	addColumn := func(col catalog.Column, nameOverride string, ord int) int {
 		resultColumn := ResultColumn{
 			ResultColumn: colinfo.ResultColumn{
 				Name:           col.GetName(),
@@ -305,10 +307,12 @@ func NewEventDescriptor(
 			ord:       ord,
 			sqlString: col.ColumnDesc().SQLStringNotHumanReadable(),
 		}
-
+		if nameOverride != "" {
+			resultColumn.Name = nameOverride
+		}
 		colIdx := len(sd.cols)
 		sd.cols = append(sd.cols, resultColumn)
-		sd.colsByName[col.GetName()] = colIdx
+		sd.colsByName[resultColumn.Name] = colIdx
 
 		if col.GetType().UserDefined() {
 			sd.udtCols = append(sd.udtCols, colIdx)
@@ -319,7 +323,46 @@ func NewEventDescriptor(
 	// Primary key columns must be added in the same order they
 	// appear in the primary key index.
 	primaryIdx := desc.GetPrimaryIndex()
-	colOrd := catalog.ColumnIDToOrdinalMap(desc.PublicColumns())
+	// The declarative schema changer requires special handling for dropped columns
+	// to ensure their previous values are available to the changefeed.
+	// We select a prior descriptor where the column is in the WRITE_ONLY state
+	// (instead of PUBLIC) and treat it as visible.
+	// This is necessary because the schema change stages for dropped columns were
+	// intentionally shifted to support automatic `schema_locked` management. The
+	// declarative schema changer cannot make a column WRITE_ONLY and flip
+	// `schema_locked` in two separate stages, so we instead treat the WRITE_ONLY
+	//  stage as a non-backfilling change for CDC.
+	columnsToTrack := desc.PublicColumns()
+	colNameOverrides := make(map[catid.ColumnID]string)
+	if desc.GetDeclarativeSchemaChangerState() != nil {
+		hasMergedIndex := catalog.HasDeclarativeMergedPrimaryIndex(desc)
+		allColumns := desc.AllColumns()
+		columnsToTrack = make([]catalog.Column, 0, len(allColumns))
+		for _, column := range allColumns {
+			// Skip all adding columns and non-public column.
+			if column.Adding() {
+				continue
+			}
+			// Pick up any write-only columns.
+			if column.Dropped() {
+				if !column.WriteAndDeleteOnly() || hasMergedIndex || column.IsHidden() {
+					continue
+				}
+				if found, previousName := catalog.FindPreviousColumnNameForDeclarativeSchemaChange(desc, column.GetID()); found {
+					colNameOverrides[column.GetID()] = previousName
+				}
+			}
+			columnsToTrack = append(columnsToTrack, column)
+		}
+		// Recover the order of the original columns.
+		slices.SortStableFunc(columnsToTrack, func(a, b catalog.Column) int {
+			return int(a.GetPGAttributeNum()) - int(b.GetPGAttributeNum())
+		})
+	}
+	var colOrd catalog.TableColMap
+	for ord, col := range columnsToTrack {
+		colOrd.Set(col.GetID(), ord)
+	}
 	writeOnlyAndPublic := catalog.ColumnIDToOrdinalMap(desc.WritableColumns())
 	var primaryKeyOrdinal catalog.TableColMap
 
@@ -335,20 +378,20 @@ func NewEventDescriptor(
 			}
 			return nil, errors.AssertionFailedf("expected to find column %d", ord)
 		}
-		primaryKeyOrdinal.Set(desc.PublicColumns()[ord].GetID(), ordIdx)
+		primaryKeyOrdinal.Set(columnsToTrack[ord].GetID(), ordIdx)
 		ordIdx += 1
 	}
 	sd.keyCols = make([]int, ordIdx)
 
 	switch {
 	case keyOnly:
 		ord := 0
-		for _, col := range desc.PublicColumns() {
+		for _, col := range columnsToTrack {
 			pKeyOrd, isPKey := primaryKeyOrdinal.Get(col.GetID())
 			if !isPKey {
 				continue
 			}
-			colIdx := addColumn(col, ord)
+			colIdx := addColumn(col, colNameOverrides[col.GetID()], ord)
 			ord++
 			sd.keyCols[pKeyOrd] = colIdx
 			sd.valueCols = append(sd.valueCols, colIdx)
@@ -359,9 +402,9 @@ func NewEventDescriptor(
 		// a sentinel ordinal position of virtualColOrd.
 		inFamily := catalog.MakeTableColSet(family.ColumnIDs...)
 		ord := 0
-		for _, col := range desc.PublicColumns() {
+		for _, col := range columnsToTrack {
 			if isVirtual := col.IsVirtual(); isVirtual && includeVirtualColumns {
-				colIdx := addColumn(col, virtualColOrd)
+				colIdx := addColumn(col, colNameOverrides[col.GetID()], virtualColOrd)
 				sd.valueCols = append(sd.valueCols, colIdx)
 				continue
 			}
@@ -370,7 +413,7 @@ func NewEventDescriptor(
 			if !isPKey && !isInFamily {
 				continue
 			}
-			colIdx := addColumn(col, ord)
+			colIdx := addColumn(col, colNameOverrides[col.GetID()], ord)
 			ord++
 			if isPKey {
 				sd.keyCols[pKeyOrd] = colIdx
@@ -871,7 +914,29 @@ func getRelevantColumnsForFamily(
 	// matches the order of columns in the SQL table.
 	idx := 0
 	result := make([]descpb.ColumnID, cols.Len())
-	for _, colID := range tableDesc.PublicColumnIDs() {
+	visibleColumns := tableDesc.PublicColumns()
+	if tableDesc.GetDeclarativeSchemaChangerState() != nil {
+		hasMergedIndex := catalog.HasDeclarativeMergedPrimaryIndex(tableDesc)
+		visibleColumns = make([]catalog.Column, 0, cols.Len())
+		for _, col := range tableDesc.AllColumns() {
+			if col.Adding() {
+				continue
+			}
+			if tableDesc.GetDeclarativeSchemaChangerState() == nil && !col.Public() {
+				continue
+			}
+			if col.Dropped() && (!col.WriteAndDeleteOnly() || hasMergedIndex) {
+				continue
+			}
+			visibleColumns = append(visibleColumns, col)
+		}
+		// Recover the order of the original columns.
+		slices.SortStableFunc(visibleColumns, func(a, b catalog.Column) int {
+			return int(a.GetPGAttributeNum()) - int(b.GetPGAttributeNum())
+		})
+	}
+	for _, col := range visibleColumns {
+		colID := col.GetID()
 		if cols.Contains(colID) {
 			result[idx] = colID
 			idx++

pkg/ccl/changefeedccl/cdcevent/event_test.go

@@ -8,6 +8,7 @@ package cdcevent
 import (
 	"context"
 	"fmt"
+	"slices"
 	"sync/atomic"
 	"testing"
 
@@ -173,7 +174,6 @@ CREATE TABLE foo (
   FAMILY main (a, b, e),
   FAMILY only_c (c)
 )`)
-
 	for _, tc := range []struct {
 		schemaChange string
 		// Each new primary index generated during the test will pause at each stage
@@ -210,16 +210,21 @@ CREATE TABLE foo (
 			expectedUDTCols: [][]string{{"e"}, {"e"}, {"e"}},
 		},
 		{
-			// We are going to execute a mix of add, drop and alter primary key operations,
+			// We are going to execute a mix of add, drop and alter primary key operations;
 			// this will result in 3 primary indexes being swapped.
-			// 1) The first primary index key will be the same as previous test
+			// 1) The first primary index key will be the same as the previous test
 			// 2) The second primary key will use the column "a", without a hash
 			//    sharding column since that needs to be created next.
-			// 3) The final primary key will "a" and have hash sharding on it.
+			// 3) The final primary key will "a" and have hash sharding on it (repeated
+			//    for the column removal).
+			// The values stored will have the following transitions:
+			// 1) Existing columns in the table
+			// 2) New column j added (repeated for the final PK switch)
+			// 3) Old column b removed (final state)
 			schemaChange:    "ALTER TABLE foo ADD COLUMN j INT DEFAULT 32, DROP COLUMN d, DROP COLUMN crdb_internal_a_b_shard_16, DROP COLUMN b, ALTER PRIMARY KEY USING COLUMNS(a) USING HASH",
-			expectedKeyCols: [][]string{{"crdb_internal_c_e_shard_16", "c", "e"}, {"a"}, {"crdb_internal_a_shard_16", "a"}},
-			expectedColumns: [][]string{{"a", "b", "e"}, {"a", "b", "e"}, {"a", "e", "j"}},
-			expectedUDTCols: [][]string{{"e"}, {"e"}, {"e"}},
+			expectedKeyCols: [][]string{{"crdb_internal_c_e_shard_16", "c", "e"}, {"a"}, {"crdb_internal_a_shard_16", "a"}, {"crdb_internal_a_shard_16", "a"}},
+			expectedColumns: [][]string{{"a", "b", "e"}, {"a", "b", "e"}, {"a", "b", "e", "j"}, {"a", "e", "j"}},
+			expectedUDTCols: [][]string{{"e"}, {"e"}, {"e"}, {"e"}},
 		},
 	} {
 		t.Run(tc.schemaChange, func(t *testing.T) {
@@ -917,8 +922,25 @@ func expectResultColumnsWithFamily(
 		colNamesSet[colName] = -1
 	}
 	ord := 0
-	for _, col := range desc.PublicColumns() {
+	nameOverrides := make(map[string]string)
+	// Sort the columns based on the attribute number and if they are PK columns.
+	sortedAllColumns := desc.AllColumns()
+	slices.SortStableFunc(sortedAllColumns, func(a, b catalog.Column) int {
+		return int(a.GetPGAttributeNum()) - int(b.GetPGAttributeNum())
+	})
+	for _, col := range sortedAllColumns {
+		// Skip add mutations.
+		if col.Adding() {
+			continue
+		}
+		// Only include WriteAndDeleteOnly columns.
+		if col.Dropped() && !col.WriteAndDeleteOnly() {
+			continue
+		}
 		colName := string(col.ColName())
+		if found, oldName := catalog.FindPreviousColumnNameForDeclarativeSchemaChange(desc, col.GetID()); found {
+			nameOverrides[oldName] = colName
+		}
 		if _, ok := colNamesSet[colName]; ok {
 			switch {
 			case col.IsVirtual():
@@ -936,18 +958,31 @@ func expectResultColumnsWithFamily(
 	}
 
 	for _, colName := range colNames {
-		col, err := catalog.MustFindColumnByName(desc, colName)
-		require.NoError(t, err)
+		searchName := colName
+		if nameOverrides[colName] != "" {
+			searchName = nameOverrides[colName]
+		}
+		// If a column is missing add a dummy column, which
+		// will force a mismatch / skip this set.
+		col := catalog.FindColumnByName(desc, searchName)
+		if col == nil {
+			res = append(res, ResultColumn{
+				ResultColumn: colinfo.ResultColumn{
+					Name: "Missing column",
+				},
+			})
+			continue
+		}
 		res = append(res, ResultColumn{
 			ResultColumn: colinfo.ResultColumn{
-				Name:           col.GetName(),
+				Name:           colName,
 				Typ:            col.GetType(),
 				TableID:        desc.GetID(),
 				PGAttributeNum: uint32(col.GetPGAttributeNum()),
 			},
 			Computed:  col.IsComputed(),
 			Nullable:  col.IsNullable(),
-			ord:       colNamesSet[colName],
+			ord:       colNamesSet[searchName],
 			sqlString: col.ColumnDesc().SQLStringNotHumanReadable(),
 		})
 	}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2106,6 +2106,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)
@@ -3129,7 +3177,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{
@@ -3182,7 +3230,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,17 @@ 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 no declarative schema changes then this will always be
+	// true.
+	if desc.GetDeclarativeSchemaChangerState() == nil {
+		return true
+	}
+	return catalog.HasDeclarativeMergedPrimaryIndex(desc)
+}
+
 // 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 +223,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 +290,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
 		}
 	}