changefeedccl: fix incorrect previous values for dropped columns

Previously, when we adjusted when the backfill was emitted for the drop
column under the declarative schema changer when schema_locked was
active, we wound up emitting the previous values incorrectly. What this
meant was that in the Avro format the old and new values would be the
same when a column was being dropped. This was observed by the
cdc/schemareg test. To address this, this patch correctly resolves the
previous values in the drop column case from the last descriptor
version.

Fixes: #150540

Release note: None

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/schemafeed/table_event_filter.go

@@ -191,21 +191,12 @@ func shouldFilterAddColumnEvent(e TableEvent, targets changefeedbase.Targets) (b
 // 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
+	// If there are no 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
+	return catalog.HasDeclarativeMergedPrimaryIndex(desc)
 }
 
 // Returns true if the changefeed targets a column which has a drop mutation inside the table event.

pkg/sql/catalog/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "//pkg/sql/privilege",
         "//pkg/sql/schemachanger/scpb",
         "//pkg/sql/sem/catconstants",
+        "//pkg/sql/sem/catid",
         "//pkg/sql/sem/idxtype",
         "//pkg/sql/sem/semenumpb",
         "//pkg/sql/sem/tree",

pkg/sql/catalog/descriptor.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -1207,3 +1208,46 @@ func IsSystemDescriptor(desc Descriptor) bool {
 func HasConcurrentDeclarativeSchemaChange(desc Descriptor) bool {
 	return desc.GetDeclarativeSchemaChangerState() != nil
 }
+
+// HasDeclarativeMergedPrimaryIndex returns if a primary key swap is occurring
+// via the declarative schema changer, and the new index is in a write-only state.
+func HasDeclarativeMergedPrimaryIndex(desc TableDescriptor) bool {
+	// For declarative schema changes detect when a new primary index becomes
+	// WRITE_ONLY (i.e. backfill has been completed).
+	state := desc.GetDeclarativeSchemaChangerState()
+	if state == nil {
+		return false
+	}
+	for idx, target := range state.Targets {
+		if target.GetPrimaryIndex() != nil &&
+			target.GetPrimaryIndex().TableID == desc.GetID() &&
+			(target.TargetStatus == scpb.Status_PUBLIC || target.TargetStatus == scpb.Status_TRANSIENT_ABSENT) &&
+			state.CurrentStatuses[idx] == scpb.Status_WRITE_ONLY {
+			return true
+		}
+	}
+	return false
+}
+
+// FindPreviousColumnNameForDeclarativeSchemaChange attempts to resolve the previous
+// column name for a declarative schema change. This is handy for cases where the
+// original name needs to be known.
+func FindPreviousColumnNameForDeclarativeSchemaChange(
+	desc TableDescriptor, columnID catid.ColumnID,
+) (bool, string) {
+	state := desc.GetDeclarativeSchemaChangerState()
+	if state == nil {
+		// No declarative schema change is active.
+		return false, ""
+	}
+	// Find the dropped column name element name that maps to the old name.
+	for _, elem := range state.Targets {
+		if nameElem, ok := elem.Element().(*scpb.ColumnName); ok &&
+			nameElem.TableID == desc.GetID() &&
+			nameElem.ColumnID == columnID &&
+			elem.TargetStatus == scpb.Status_ABSENT {
+			return true, nameElem.Name
+		}
+	}
+	return false, ""
+}