Merge #143383

143383: changefeedccl: add origin source field to enriched envelope r=asg0451 a=asg0451

Add origin="cockroachdb" to the enriched envelope source object.

Fixes #139693

Release note: None


Co-authored-by: Miles Frankel <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/changefeed_test.go

@@ -4431,6 +4431,7 @@ func TestChangefeedEnrichedSourceWithData(t *testing.T) {
 										"node_name":            map[string]any{"string": nodeName},
 										"changefeed_sink":      map[string]any{"string": sink},
 										"source_node_locality": map[string]any{"string": sourceNodeLocality},
+										"origin":               map[string]any{"string": "cockroachdb"},
 									},
 								}
 								if withMVCCTS {
@@ -4467,6 +4468,7 @@ func TestChangefeedEnrichedSourceWithData(t *testing.T) {
 									"node_name":            nodeName,
 									"changefeed_sink":      sink,
 									"source_node_locality": sourceNodeLocality,
+									"origin":               "cockroachdb",
 								}
 								if withMVCCTS {
 									assertReasonableMVCCTimestamp(t, actualSource["mvcc_timestamp"].(string))

pkg/ccl/changefeedccl/enriched_source_provider.go

@@ -100,6 +100,7 @@ func newEnrichedSourceProvider(
 		fieldNameSourceNodeLocality: json.FromString(sourceData.sourceNodeLocality),
 		fieldNameNodeName:           json.FromString(sourceData.nodeName),
 		fieldNameNodeID:             json.FromString(sourceData.nodeID),
+		fieldNameOrigin:             json.FromString(originCockroachDB),
 	}
 
 	var nonFixedJSONFields []string
@@ -167,6 +168,7 @@ func (p *enrichedSourceProvider) GetAvro(
 			dest[fieldNameSourceNodeLocality] = goavro.Union(avro.SchemaTypeString, p.sourceData.sourceNodeLocality)
 			dest[fieldNameNodeName] = goavro.Union(avro.SchemaTypeString, p.sourceData.nodeName)
 			dest[fieldNameNodeID] = goavro.Union(avro.SchemaTypeString, p.sourceData.nodeID)
+			dest[fieldNameOrigin] = goavro.Union(avro.SchemaTypeString, originCockroachDB)
 		}
 
 		if p.opts.mvccTimestamp {
@@ -196,6 +198,7 @@ const (
 	fieldNameMVCCTimestamp      = "mvcc_timestamp"
 	fieldNameUpdatedTSNS        = "ts_ns"
 	fieldNameUpdatedTSHLC       = "ts_hlc"
+	fieldNameOrigin             = "origin"
 )
 
 type fieldInfo struct {
@@ -208,6 +211,17 @@ type fieldInfo struct {
 // everything is nullable in avro for better backwards compatibility, whereas we
 // use the optional flag in kafka connect more meaningfully.
 var allFieldInfo = map[string]fieldInfo{
+	fieldNameOrigin: {
+		avroSchemaField: avro.SchemaField{
+			Name:       fieldNameOrigin,
+			SchemaType: []avro.SchemaType{avro.SchemaTypeNull, avro.SchemaTypeString},
+		},
+		kafkaConnectSchema: kcjsonschema.Schema{
+			Field:    fieldNameOrigin,
+			TypeName: kcjsonschema.SchemaTypeString,
+			Optional: false,
+		},
+	},
 	fieldNameChangefeedSink: {
 		avroSchemaField: avro.SchemaField{
 			Name:       fieldNameChangefeedSink,
@@ -355,3 +369,5 @@ func init() {
 		Optional: true,
 	}
 }
+
+const originCockroachDB = "cockroachdb"