Merge #155878

155878: logical: decode into local types r=jeffswenson a=jeffswenson

Previously, the event decoder used by the crud writer copied the behavior of the legacy sql writer and did not decode all UDTs into local types. Once we start using prepared statements, we need datums to have the correct type. This reworks the decoder so the types are mapped into their local representation.

Release note: none
Epic: CRDB-48647

Co-authored-by: Jeff Swenson <jeffswenson@betterthannull.com>

Author: craig[bot]

pkg/crosscluster/logical/event_decoder.go

@@ -26,7 +26,7 @@ import (
 // that are appropriate for the destination table.
 type eventDecoder struct {
 	decoder   cdcevent.Decoder
-	srcToDest map[descpb.ID]destinationTable
+	srcToDest map[descpb.ID]*destinationTable
 
 	// TODO(jeffswenson): clean this interface up. There's a problem with
 	// layering that requires the event decoder to know about the most recent
@@ -70,7 +70,7 @@ func newEventDecoder(
 	settings *cluster.Settings,
 	procConfigByDestID map[descpb.ID]sqlProcessorTableConfig,
 ) (*eventDecoder, error) {
-	srcToDest := make(map[descpb.ID]destinationTable, len(procConfigByDestID))
+	srcToDest := make(map[descpb.ID]*destinationTable, len(procConfigByDestID))
 	err := descriptors.DescsTxn(ctx, func(ctx context.Context, txn descs.Txn) error {
 		for dstID, s := range procConfigByDestID {
 			descriptor, err := txn.Descriptors().GetLeasedImmutableTableByID(ctx, txn.KV(), dstID)
@@ -80,13 +80,16 @@ func newEventDecoder(
 
 			columns := getColumnSchema(descriptor)
 			columnNames := make([]string, 0, len(columns))
+			columnTypes := make([]*types.T, 0, len(columns))
 			for _, column := range columns {
 				columnNames = append(columnNames, column.column.GetName())
+				columnTypes = append(columnTypes, column.column.GetType().Canonical())
 			}
 
-			srcToDest[s.srcDesc.GetID()] = destinationTable{
-				id:      dstID,
-				columns: columnNames,
+			srcToDest[s.srcDesc.GetID()] = &destinationTable{
+				id:          dstID,
+				columns:     columnNames,
+				columnTypes: columnTypes,
 			}
 		}
 		return nil
@@ -185,8 +188,7 @@ func (d *eventDecoder) decodeEvent(
 		return decodedEvent{}, errors.AssertionFailedf("table %d not found", decodedRow.TableID)
 	}
 
-	row := make(tree.Datums, 0, len(dstTable.columns))
-	row, err = appendDatums(row, decodedRow, dstTable.columns)
+	row, err := dstTable.toLocalDatums(decodedRow)
 	if err != nil {
 		return decodedEvent{}, err
 	}
@@ -201,8 +203,7 @@ func (d *eventDecoder) decodeEvent(
 		return decodedEvent{}, err
 	}
 
-	prevRow := make(tree.Datums, 0, len(dstTable.columns))
-	prevRow, err = appendDatums(prevRow, decodedPrevRow, dstTable.columns)
+	prevRow, err := dstTable.toLocalDatums(decodedPrevRow)
 	if err != nil {
 		return decodedEvent{}, err
 	}
@@ -216,37 +217,43 @@ func (d *eventDecoder) decodeEvent(
 	}, nil
 }
 
-// appendDatums appends datums for the specified column names from the cdcevent.Row
-// to the datums slice and returns the updated slice.
-func appendDatums(datums tree.Datums, row cdcevent.Row, columnNames []string) (tree.Datums, error) {
-	it, err := row.DatumsNamed(columnNames)
+// toLocalDatums creates a row with the types of the datums converted to to the
+// types required by the remote descriptor. toLocalDatums takes ownerhsip of
+// the input row and may modify datums from the decoded input row.
+func (d *destinationTable) toLocalDatums(row cdcevent.Row) (tree.Datums, error) {
+	localRow := make(tree.Datums, len(d.columns))
+
+	it, err := row.DatumsNamed(d.columns)
 	if err != nil {
 		return nil, err
 	}
 
-	if err := it.Datum(func(d tree.Datum, col cdcevent.ResultColumn) error {
-		if dEnum, ok := d.(*tree.DEnum); ok {
-			// Override the type to Unknown to avoid a mismatched type OID error
-			// during execution. Note that Unknown is the type used by default
-			// when a SQL statement is executed without type hints.
-			//
-			// TODO(jeffswenson): this feels like the wrong place to do this,
-			// but its inspired by the implementation in queryBuilder.AddRow.
-			//
-			// Really we should be mapping from the source datum type to the
-			// destination datum type.
-			dEnum.EnumTyp = types.Unknown
+	columnIndex := 0
+	if err := it.Datum(func(datum tree.Datum, col cdcevent.ResultColumn) error {
+		typ := d.columnTypes[columnIndex]
+
+		switch d := datum.(type) {
+		case *tree.DEnum:
+			d.EnumTyp = typ
+		case *tree.DArray:
+			d.ParamTyp = typ
+		case *tree.DTuple:
+			datum = tree.NewDTuple(typ, d.D...)
 		}
-		datums = append(datums, d)
+
+		localRow[columnIndex] = datum
+
+		columnIndex += 1
 		return nil
 	}); err != nil {
 		return nil, err
 	}
 
-	return datums, nil
+	return localRow, nil
 }
 
 type destinationTable struct {
-	id      descpb.ID
-	columns []string
+	id          descpb.ID
+	columns     []string
+	columnTypes []*types.T
 }

pkg/crosscluster/logical/event_decoder_test.go

@@ -14,14 +14,17 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
+	"github.com/cockroachdb/cockroach/pkg/sql/randgen"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -197,3 +200,81 @@ func TestEventDecoder_DeduplicationWithDiscardDelete(t *testing.T) {
 	require.Equal(t, tree.NewDString("inserted"), events[1].row[1])
 	require.Equal(t, tree.DNull, events[1].prevRow[1])
 }
+
+func TestEventDecoder_UserDefinedTypes(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// TODO(jeffswenson): it would be nice if we could implement this test using
+	// the randgen package, but randgen appears to be missing a utililty that
+	// lets us create a random table random dependent UDTs.
+
+	ctx := context.Background()
+	rng, _ := randutil.NewTestRand()
+
+	srv, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(ctx)
+	server := srv.ApplicationLayer()
+	runner := sqlutils.MakeSQLRunner(sqlDB)
+
+	for _, db := range []string{"srcdb", "dstdb"} {
+		runner.Exec(t, "CREATE DATABASE "+db)
+		runner.Exec(t, "USE "+db)
+		runner.Exec(t, "CREATE TYPE status_enum AS ENUM ('active', 'inactive')")
+		runner.Exec(t, "CREATE TYPE metadata_type AS (key TEXT, value INT)")
+		runner.Exec(t, `CREATE TABLE user_types (
+			id INT PRIMARY KEY,
+			status status_enum,
+			tags TEXT[],
+			metadata metadata_type
+		)`)
+	}
+
+	srcDesc := cdctest.GetHydratedTableDescriptor(t, server.ExecutorConfig(), "srcdb", "public", "user_types")
+	dstDesc := cdctest.GetHydratedTableDescriptor(t, server.ExecutorConfig(), "dstdb", "public", "user_types")
+
+	decoder, err := newEventDecoder(ctx, server.InternalDB().(descs.DB), server.ClusterSettings(), map[descpb.ID]sqlProcessorTableConfig{
+		dstDesc.GetID(): {srcDesc: srcDesc},
+	})
+	require.NoError(t, err)
+
+	eb := newKvEventBuilder(t, srcDesc.TableDesc())
+
+	// Build test row with user-defined types
+	enumType := catalog.FindColumnByName(srcDesc, "status").GetType()
+	arrayType := catalog.FindColumnByName(srcDesc, "tags").GetType()
+	tupleType := catalog.FindColumnByName(srcDesc, "metadata").GetType()
+	testRow := tree.Datums{
+		tree.NewDInt(1),
+		&tree.DEnum{
+			EnumTyp:     enumType,
+			PhysicalRep: enumType.TypeMeta.EnumData.PhysicalRepresentations[0], // Use correct physical rep
+			LogicalRep:  enumType.TypeMeta.EnumData.LogicalRepresentations[0],  // Use correct logical rep
+		},
+		randgen.RandDatum(rng, arrayType, false).(*tree.DArray),
+		randgen.RandDatum(rng, tupleType, false).(*tree.DTuple),
+	}
+
+	insertEvent := eb.insertEvent(server.Clock().Now(), testRow)
+	events, err := decoder.decodeAndCoalesceEvents(ctx, []streampb.StreamEvent_KV{insertEvent}, jobspb.LogicalReplicationDetails_DiscardNothing)
+	require.NoError(t, err)
+	require.Len(t, events, 1)
+
+	row := events[0].row
+	require.Len(t, row, 4)
+
+	// Verify enum type mapping
+	enum := row[1].(*tree.DEnum)
+	dstEnumType := catalog.FindColumnByName(dstDesc, "status").GetType()
+	require.Equal(t, dstEnumType, enum.EnumTyp)
+
+	// Verify array type mapping
+	array := row[2].(*tree.DArray)
+	dstArrayType := catalog.FindColumnByName(dstDesc, "tags").GetType()
+	require.Equal(t, dstArrayType, array.ParamTyp)
+
+	// Verify tuple type mapping
+	tuple := row[3].(*tree.DTuple)
+	dstTupleType := catalog.FindColumnByName(dstDesc, "metadata").GetType()
+	require.Equal(t, dstTupleType, tuple.ResolvedType())
+}