changefeedccl: add protobuf encoder support for Kafka with envelope=bare

This adds initial support for the format=protobuf option in changefeeds,
targeting the Kafka sink with the envelope=bare config as well as
the key_in_value, mvcc_timestamp, updated and topic_in_value flags.
At this stage, wrapped envelope and resolved support are partially
implemented but not the focus of this PR.

The encoder serializes key value pairs into Protobuf messages using
protobufEncoder that conforms to the existing Encoder interface.
The schema is defined under changefeedpb.

Part of: #148377 – Support protobuf in Kafka with envelope=bare

Update pkg/ccl/changefeedccl/encoder_protobuf_test.go

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

Author: elizaMkraule

pkg/ccl/changefeedccl/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "encoder_avro.go",
         "encoder_csv.go",
         "encoder_json.go",
+        "encoder_protobuf.go",
         "enriched_source_provider.go",
         "event_processing.go",
         "fetch_table_bytes.go",
@@ -70,6 +71,7 @@ go_library(
         "//pkg/clusterversion",
         "//pkg/docs",
         "//pkg/featureflag",
+        "//pkg/geo",
         "//pkg/jobs",
         "//pkg/jobs/jobsauth",
         "//pkg/jobs/jobspb",
@@ -200,6 +202,7 @@ go_test(
         "changefeed_test.go",
         "csv_test.go",
         "encoder_json_test.go",
+        "encoder_protobuf_test.go",
         "encoder_test.go",
         "event_processing_test.go",
         "fetch_table_bytes_test.go",

pkg/ccl/changefeedccl/cdcevent/event.go

@@ -111,6 +111,16 @@ func (r Row) ForEachUDTColumn() Iterator {
 	return iter{r: r, cols: r.udtCols}
 }
 
+// NumKeyColumns returns the number of primary key columns in the row.
+func (r Row) NumKeyColumns() int {
+	return len(r.keyCols)
+}
+
+// NumValueColumns returns the number of value columns in the row.
+func (r Row) NumValueColumns() int {
+	return len(r.valueCols)
+}
+
 // DatumNamed returns the datum with the specified column name, in the form of an Iterator.
 func (r Row) DatumNamed(n string) (Iterator, error) {
 	idx, ok := r.EventDescriptor.colsByName[n]

pkg/ccl/changefeedccl/changefeed_test.go

@@ -12051,3 +12051,38 @@ func TestDatabaseLevelChangefeed(t *testing.T) {
 	}
 	cdcTest(t, testFn)
 }
+
+func TestChangefeedBareFullProtobuf(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+
+		sqlDB.Exec(t, `
+			CREATE TABLE pricing (
+				id INT PRIMARY KEY,
+				name STRING,
+				discount FLOAT,
+				tax DECIMAL,
+				options STRING[]
+			)`)
+		sqlDB.Exec(t, `
+			INSERT INTO pricing VALUES
+				(1, 'Chair', 15.75, 2.500, ARRAY['Brown', 'Black']), 
+				(2, 'Table', 20.00, 1.23456789, ARRAY['Brown', 'Black'])
+		`)
+		pricingFeed := feed(t, f,
+			`CREATE CHANGEFEED FOR pricing WITH envelope='bare', format='protobuf', key_in_value, topic_in_value`)
+		defer closeFeed(t, pricingFeed)
+
+		expected := []string{
+			`pricing: {"id":1}->{"values":{"discount":15.75,"id":1,"name":"Chair","options":["Brown","Black"],"tax":"2.500"},"__crdb__":{"key":{"id":1},"topic":"pricing"}}`,
+			`pricing: {"id":2}->{"values":{"discount":20,"id":2,"name":"Table","options":["Brown","Black"],"tax":"1.23456789"},"__crdb__":{"key":{"id":2},"topic":"pricing"}}`,
+		}
+
+		assertPayloads(t, pricingFeed, expected)
+	}
+
+	cdcTest(t, testFn, feedTestForceSink("kafka"))
+}

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -169,10 +169,11 @@ const (
 	OptEnvelopeBare          EnvelopeType = `bare`
 	OptEnvelopeEnriched      EnvelopeType = `enriched`
 
-	OptFormatJSON    FormatType = `json`
-	OptFormatAvro    FormatType = `avro`
-	OptFormatCSV     FormatType = `csv`
-	OptFormatParquet FormatType = `parquet`
+	OptFormatJSON     FormatType = `json`
+	OptFormatAvro     FormatType = `avro`
+	OptFormatCSV      FormatType = `csv`
+	OptFormatParquet  FormatType = `parquet`
+	OptFormatProtobuf FormatType = `protobuf`
 
 	OptOnErrorFail  OnErrorType = `fail`
 	OptOnErrorPause OnErrorType = `pause`
@@ -376,7 +377,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptCustomKeyColumn:                    stringOption,
 	OptEndTime:                            timestampOption,
 	OptEnvelope:                           enum("row", "key_only", "wrapped", "deprecated_row", "bare", "enriched"),
-	OptFormat:                             enum("json", "avro", "csv", "experimental_avro", "parquet"),
+	OptFormat:                             enum("json", "avro", "csv", "experimental_avro", "parquet", "protobuf"),
 	OptFullTableName:                      flagOption,
 	OptKeyInValue:                         flagOption,
 	OptTopicInValue:                       flagOption,
@@ -948,7 +949,7 @@ func (e EncodingOptions) Validate() error {
 	}
 
 	// TODO(#140110): refactor this logic.
-	if (e.Envelope != OptEnvelopeWrapped && e.Envelope != OptEnvelopeEnriched) && e.Format != OptFormatJSON && e.Format != OptFormatParquet {
+	if (e.Envelope != OptEnvelopeWrapped && e.Envelope != OptEnvelopeEnriched) && e.Format != OptFormatProtobuf && e.Format != OptFormatJSON && e.Format != OptFormatParquet {
 		requiresWrap := []struct {
 			k string
 			b bool

pkg/ccl/changefeedccl/changefeedpb/BUILD.bazel

@@ -4,9 +4,13 @@ load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
 
 proto_library(
     name = "changefeedpb_proto",
-    srcs = ["scheduled_changefeed.proto"],
+    srcs = [
+        "changefeed.proto",
+        "scheduled_changefeed.proto",
+    ],
     strip_import_prefix = "/pkg",
     visibility = ["//visibility:public"],
+    deps = ["@com_google_protobuf//:timestamp_proto"],
 )
 
 go_proto_library(

pkg/ccl/changefeedccl/changefeedpb/changefeed.proto

@@ -0,0 +1,153 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+// TODO(#149703): Move this .proto file to the new changefeedpb repo,
+// import it into CockroachDB, and update usages accordingly.
+
+syntax = "proto3";
+package cockroach.ccl.changefeedccl;
+option go_package = "github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb";
+
+import "google/protobuf/timestamp.proto";
+
+// MessageBatch is a batch of messages for use in webhook sinks.
+message MessageBatch {
+    repeated Message payload = 1;
+}
+
+// Message is an enum of the different envelope types. This is what will be emitted to the client.
+message Message {
+    oneof data {
+        // wrapped is a Message in WrappedEnvelope format.
+        WrappedEnvelope wrapped = 1;
+
+        // bare is a Message in BareEnvelope format.
+        BareEnvelope bare = 2;
+
+        // enriched is a Message in EnrichedEnvelope format.
+        EnrichedEnvelope enriched = 3;
+
+         // resolved is a Message in Resolved format for Resolved Timestamps.
+        Resolved resolved = 4;
+
+        // bareResolved wraps a resolved envelope inside a BareResolved format for Resolved Timestamps in Bare envelopes.
+        BareResolved bareResolved = 5;
+    }
+}
+
+// WrappedEnvelope includes both the changed data and contextual metadata about the change.
+message WrappedEnvelope {
+    Record after = 1;
+    Record before = 2;
+    string updated = 4;
+    string mvcc_timestamp = 5;
+    Key key = 6;
+    string topic = 7;
+}
+
+// BareEnvelope contains change data as a flat map along with
+// CockroachDB-specific metadata under the '__crdb__' field.
+message BareEnvelope {
+    map<string, Value> values = 1;
+    Metadata __crdb__ = 2;
+}
+
+// BareResolved is a minimal envelope that wraps a resolved timestamp in a '__crdb__' field.
+message BareResolved {
+    Resolved __crdb__ = 1;
+}
+
+// EnrichedEnvelope includes detailed context about the change event and source.
+message EnrichedEnvelope {
+    Record after = 1;
+    Record before = 2;
+    Op op = 3;
+    int64 ts_ns = 4;
+    EnrichedSource source = 5;
+}
+
+// Resolved carries resolved timestamp information for a changefeed span.
+message Resolved {
+    string resolved = 1;
+}
+
+// EnrichedSource records information about the origin and context
+// of a change event, for operational traceability.
+message EnrichedSource {
+    string job_id = 1;
+    string changefeed_sink = 2;
+    string db_version = 3;
+    string cluster_name = 4;
+    string cluster_id = 5;
+    string source_node_locality = 6;
+    string node_name = 7;
+    string node_id = 8;
+    string mvcc_timestamp = 9;
+    int64 ts_ns = 10;
+    string ts_hlc = 11;
+    string origin = 12;
+    string database_name = 13;
+    string schema_name = 14;
+    string table_name = 15;
+    repeated string primary_keys = 16;
+}
+
+// Op enumerates the types of operations represented in a change event.
+enum Op {
+    OP_UNSPECIFIED = 0;
+    OP_CREATE = 1;
+    OP_UPDATE = 2;
+    OP_DELETE = 3;
+}
+
+// Metadata contains CockroachDB-specific metadata about a change event.
+// This message is also referred to as '__crdb__'.
+message Metadata {
+    string updated = 1;
+    string mvcc_timestamp = 2;
+    Key key = 3;
+    string topic = 4;
+}
+
+// Value represents a value of arbitrary type carried in a change event.
+message Value {
+    oneof value {
+        string string_value = 1;
+        bytes bytes_value = 2;
+        int32 int32_value = 3;
+        int64 int64_value = 4;
+        float float_value = 5;
+        double double_value = 6;
+        bool bool_value = 7;
+        google.protobuf.Timestamp timestamp_value = 8;
+        Array array_value = 9;
+        Record tuple_value = 10;
+        Decimal decimal_value = 11;
+        string date_value = 12;
+        string interval_value = 13;
+        string time_value = 14;
+        string uuid_value = 15;
+    }
+}
+
+// Key contains the primary key values for a row.
+message Key {
+	map<string, Value> key = 1;
+}
+
+// Array represents an ordered list of values.
+message Array {
+  repeated Value values = 1;
+}
+
+// Decimal contains a fixed-point decimal value represented as a string.
+message Decimal {
+  string value = 1;
+}
+
+// Record represents a flat mapping of column names to values for a row.
+message Record {
+    map<string, Value> values = 1;
+}

pkg/ccl/changefeedccl/changefeedpb/marshal.go

@@ -40,3 +40,58 @@ func (m ScheduledChangefeedExecutionArgs) MarshalJSONPB(x *jsonpb.Marshaler) ([]
 	m.ChangefeedStatement = tree.AsStringWithFlags(export, tree.FmtShowFullURIs)
 	return json.Marshal(m)
 }
+
+// MarshalJSON provides clean, unwrapped JSON output for changefeedpb.Value.
+func (v *Value) MarshalJSON() ([]byte, error) {
+	if v == nil || v.Value == nil {
+		return []byte("null"), nil
+	}
+
+	var out any
+	switch val := v.Value.(type) {
+	case *Value_BoolValue:
+		out = val.BoolValue
+	case *Value_Int64Value:
+		out = val.Int64Value
+	case *Value_DoubleValue:
+		out = val.DoubleValue
+	case *Value_StringValue:
+		out = val.StringValue
+	case *Value_DecimalValue:
+		out = val.DecimalValue.Value
+	case *Value_TimestampValue:
+		out = val.TimestampValue
+	case *Value_DateValue:
+		out = val.DateValue
+	case *Value_IntervalValue:
+		out = val.IntervalValue
+	case *Value_UuidValue:
+		out = val.UuidValue
+	case *Value_BytesValue:
+		out = string(val.BytesValue)
+	case *Value_TimeValue:
+		out = val.TimeValue
+	case *Value_ArrayValue:
+		arr := make([]any, len(val.ArrayValue.Values))
+		for i, elem := range val.ArrayValue.Values {
+			arr[i] = elem
+		}
+		out = arr
+	case *Value_TupleValue:
+		m := make(map[string]any, len(val.TupleValue.Values))
+		for k, v := range val.TupleValue.Values {
+			m[k] = v
+		}
+		out = m
+	default:
+		return nil, errors.AssertionFailedf("unexpected protobuf value type: %T", v.Value)
+	}
+	return json.Marshal(out)
+}
+
+func (k *Key) MarshalJSON() ([]byte, error) {
+	if k == nil || k.Key == nil {
+		return []byte("null"), nil
+	}
+	return json.Marshal(k.Key)
+}

pkg/ccl/changefeedccl/encoder.go

@@ -62,6 +62,8 @@ func getEncoder(
 		//of both encoder and sink. See parquet_sink_cloudstorage.go file for more
 		//information on why this was needed.
 		return nil, nil
+	case changefeedbase.OptFormatProtobuf:
+		return newProtobufEncoder(ctx, protobufEncoderOptions{EncodingOptions: opts}, targets), nil
 	default:
 		return nil, errors.AssertionFailedf(`unknown format: %s`, opts.Format)
 	}