Merge #143311

143311: crosscluster/logical: fix crash when using UDTs in LDR on multiple nodes r=dt a=dt

Previously running LDR into a table that included user-defined types could crash the nodes due to an error citing missing type hydration. 

This occurs during decoding of an incoming KV into a logical row, and is due to the fact the decode used in the LDR writer processor is configured with table descriptors received from the source cluster during LDR initialization, but the LDR writer processors setup was _not_ hydrating said descriptors before using them.

This hydration step is modeled as an in-place mutation of said metadata that does not change its type signature, so even though downstream code requires its inputs be hydrated (and crashes if they're not); it is up to the specific execution path to ensure this step occurs as this is not checked during compilation. 

Furthermore, when this metadata is serialized and transmitted to another node as part of a distSQL flow spec, the table descriptor seen by the other node does _not_ contain hydrated types, while if the same distsql processor is passed the same spec on a local node, it sees the types in that descriptor already hydrated.

Such cases are not unheard of and just mean we rely on tests rather than the compiler to ensure proper usage. And indeed, when UDT functionality was added, the addition included tests to exercise it. However this is where a subtly complication emerges, which meant that these tests, which ran LDR on a table including UDTs from one node to one node, did not catch this bug.

Apparently when a TableDescriptor that has been hydrated is placed in a DistSQL spec, the TableDescriptor read back out of that spec by a DistSQL processor to which that spec is supplied may also be hydrated or may not be be, depending on whether the processor is on the same node as the coordinator where the spec was created or not. In the tests of UDT behavior, since these tests were configured to use a single source and destination node, this was always the case, so despite missing an explicit hydration in the processor, the table descriptor just so happened to already be hydrated when it was used in that processor. However in a multi-node cluster, the processors on remote nodes would encounter a non-hydrated descriptor when reading the same spec.

Our bias towards using single-node test configurations for exercising features that do not appear to have anything to do with how work is distributed across many nodes in a larger cluster is generally sensible: it speeds up tests and thus allows us to add more test cases without making the test suite unwieldy. However in this case there was a very non-obvious interaction between the routines for decoding of some bytes and executing those routines on different nodes.

This diff changes the UDT tests to now explicitly run on multiple nodes --and scatters many ranges to ensure those nodes are used. This caused these tests to reliably fail until  the missing hydration steps were added to the processor to fix the underlying bug.

Release note (bug fix): Fix a crash due to 'use of enum metadata before hydration' when using LDR with user-defined types.
Epic: none.

Co-authored-by: David Taylor <[email protected]>

Author: craig[bot]

pkg/crosscluster/logical/logical_replication_dist.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -29,6 +30,7 @@ func constructLogicalReplicationWriterSpecs(
 	previousReplicatedTimestamp hlc.Timestamp,
 	checkpoint jobspb.StreamIngestionCheckpoint,
 	tableMetadataByDestID map[int32]execinfrapb.TableReplicationMetadata,
+	srcTypes []*descpb.TypeDescriptor,
 	jobID jobspb.JobID,
 	streamID streampb.StreamID,
 	discard jobspb.LogicalReplicationDetails_Discard,
@@ -47,6 +49,7 @@ func constructLogicalReplicationWriterSpecs(
 		Discard:                     discard,
 		Mode:                        mode,
 		MetricsLabel:                metricsLabel,
+		TypeDescriptors:             srcTypes,
 	}
 
 	writerSpecs := make(map[base.SQLInstanceID][]execinfrapb.LogicalReplicationWriterSpec, len(destSQLInstances))

pkg/crosscluster/logical/logical_replication_job.go

@@ -583,6 +583,7 @@ func (p *logicalReplicationPlanner) generatePlanImpl(
 		progress.ReplicatedTime,
 		progress.Checkpoint,
 		tableMetadataByDestID,
+		plan.SourceTypes,
 		p.job.ID(),
 		streampb.StreamID(payload.StreamID),
 		payload.Discard,

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -2176,6 +2176,7 @@ func TestLogicalReplicationSchemaChanges(t *testing.T) {
 func TestUserDefinedTypes(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
+	skip.UnderDuress(t, "this needs to be multi-node but that tends to be too slow for duressed builds")
 
 	ctx := context.Background()
 	clusterArgs := base.TestClusterArgs{
@@ -2187,7 +2188,7 @@ func TestUserDefinedTypes(t *testing.T) {
 		},
 	}
 
-	server, s, dbA, dbB := setupLogicalTestServer(t, ctx, clusterArgs, 1)
+	server, s, dbA, dbB := setupLogicalTestServer(t, ctx, clusterArgs, 3)
 	defer server.Stopper().Stop(ctx)
 
 	dbBURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("b"))
@@ -2206,6 +2207,8 @@ func TestUserDefinedTypes(t *testing.T) {
 			dbB.Exec(t, "CREATE TABLE data2 (pk INT PRIMARY KEY, val1 my_enum DEFAULT 'two', val2 my_composite)")
 
 			dbB.Exec(t, "INSERT INTO data VALUES (1, 'one', (3, 'cat'))")
+			dbB.Exec(t, "ALTER TABLE data SPLIT AT VALUES (1), (2), (3)")
+			dbB.Exec(t, "ALTER TABLE data SCATTER")
 			// Force default expression evaluation.
 			dbB.Exec(t, "INSERT INTO data (pk, val2) VALUES (2, (4, 'dog'))")
 
@@ -2216,8 +2219,8 @@ func TestUserDefinedTypes(t *testing.T) {
 			).Scan(&jobAID)
 			WaitUntilReplicatedTime(t, s.Clock().Now(), dbA, jobAID)
 			require.NoError(t, replicationtestutils.CheckEmptyDLQs(ctx, dbA.DB, "A"))
-			dbB.CheckQueryResults(t, "SELECT * FROM data", [][]string{{"1", "one", "(3,cat)"}, {"2", "two", "(4,dog)"}})
-			dbA.CheckQueryResults(t, "SELECT * FROM data", [][]string{{"1", "one", "(3,cat)"}, {"2", "two", "(4,dog)"}})
+			dbB.CheckQueryResults(t, "SELECT * FROM data ORDER BY pk", [][]string{{"1", "one", "(3,cat)"}, {"2", "two", "(4,dog)"}})
+			dbA.CheckQueryResults(t, "SELECT * FROM data ORDER BY pk", [][]string{{"1", "one", "(3,cat)"}, {"2", "two", "(4,dog)"}})
 
 			var jobBID jobspb.JobID
 			dbB.QueryRow(t,

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/typedesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
@@ -180,11 +181,17 @@ func newLogicalReplicationWriterProcessor(
 
 	procConfigByDestTableID := make(map[descpb.ID]sqlProcessorTableConfig)
 	destTableBySrcID := make(map[descpb.ID]dstTableMetadata)
+	crossClusterResolver := crosscluster.MakeCrossClusterTypeResolver(spec.TypeDescriptors)
 	for dstTableID, md := range spec.TableMetadataByDestID {
+		srcDesc := tabledesc.NewBuilder(&md.SourceDescriptor).BuildImmutableTable()
+		if err := typedesc.HydrateTypesInDescriptor(ctx, srcDesc, crossClusterResolver); err != nil {
+			return nil, err
+		}
 		procConfigByDestTableID[descpb.ID(dstTableID)] = sqlProcessorTableConfig{
-			srcDesc: tabledesc.NewBuilder(&md.SourceDescriptor).BuildImmutableTable(),
+			srcDesc: srcDesc,
 			dstOID:  md.DestinationFunctionOID,
 		}
+
 		destTableBySrcID[md.SourceDescriptor.GetID()] = dstTableMetadata{
 			database: md.DestinationParentDatabaseName,
 			schema:   md.DestinationParentSchemaName,

pkg/sql/execinfrapb/processors_bulk_io.proto

@@ -535,7 +535,8 @@ message LogicalReplicationWriterSpec {
 
     optional string metrics_label = 11 [(gogoproto.nullable) = false];
 
-    // Next ID: 13.
+    repeated cockroach.sql.sqlbase.TypeDescriptor type_descriptors = 13;
+    // Next ID: 14.
 }
 
 message LogicalReplicationOfflineScanSpec {