changefeedccl: support resolved option with protobuf format

This change introduces support for the resolved option
with protobuf format for changefeeds

Fixes #148934
Epic: CRDB-44007

Release note (general change): the changefeeds with protobuf
format now supports the resolved option for emitting
resolved timestamps.

Author: elizaMkraule

pkg/ccl/changefeedccl/encoder_protobuf.go

@@ -15,7 +15,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedpb"
 	"github.com/cockroachdb/cockroach/pkg/geo"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
-	"github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
@@ -77,9 +76,29 @@ func (e *protobufEncoder) EncodeValue(
 
 // EncodeResolvedTimestamp encodes a resolved timestamp message for the specified topic.
 func (e *protobufEncoder) EncodeResolvedTimestamp(
-	context.Context, string, hlc.Timestamp,
+	ctx context.Context, topic string, ts hlc.Timestamp,
 ) ([]byte, error) {
-	return nil, unimplemented.NewWithIssuef(148934, "protobuf encoder does not support resolved timestamps yet")
+	var msg *changefeedpb.Message
+	if e.envelopeType == changefeedbase.OptEnvelopeBare {
+		msg = &changefeedpb.Message{
+			Data: &changefeedpb.Message_BareResolved{
+				BareResolved: &changefeedpb.BareResolved{
+					XCrdb__: &changefeedpb.Resolved{
+						Resolved: ts.AsOfSystemTime(),
+					},
+				},
+			},
+		}
+	} else {
+		msg = &changefeedpb.Message{
+			Data: &changefeedpb.Message_Resolved{
+				Resolved: &changefeedpb.Resolved{
+					Resolved: ts.AsOfSystemTime(),
+				},
+			},
+		}
+	}
+	return protoutil.Marshal(msg)
 }
 
 // buildBare constructs a BareEnvelope with optional metadata and serializes it.

pkg/ccl/changefeedccl/encoder_protobuf_test.go

@@ -70,3 +70,60 @@ func TestProtoEncoder_BareEnvelope_WithMetadata(t *testing.T) {
 	require.Equal(t, int64(1), bare.Values["id"].GetInt64Value())
 	require.Equal(t, "Alice", bare.Values["name"].GetStringValue())
 }
+
+func TestProtoEncoder_ResolvedEnvelope(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	tableDesc, err := parseTableDesc(`CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
+	require.NoError(t, err)
+	targets := mkTargets(tableDesc)
+
+	ts := hlc.Timestamp{WallTime: 123, Logical: 456}
+
+	tests := []struct {
+		name          string
+		envelopeType  changefeedbase.EnvelopeType
+		expectWrapped bool
+	}{
+		{
+			name:         "wrapped envelope",
+			envelopeType: changefeedbase.OptEnvelopeWrapped,
+		},
+		{
+			name:         "bare envelope",
+			envelopeType: changefeedbase.OptEnvelopeBare,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opts := changefeedbase.EncodingOptions{
+				Envelope: tc.envelopeType,
+				Format:   changefeedbase.OptFormatProtobuf,
+			}
+
+			enc, err := getEncoder(context.Background(), opts, targets, false, nil, nil, nil)
+			require.NoError(t, err)
+
+			b, err := enc.EncodeResolvedTimestamp(context.Background(), "test-topic", ts)
+			require.NoError(t, err)
+
+			var msg changefeedpb.Message
+			require.NoError(t, protoutil.Unmarshal(b, &msg))
+
+			switch tc.envelopeType {
+			case changefeedbase.OptEnvelopeWrapped:
+				res := msg.GetResolved()
+				require.NotNil(t, res, "wrapped envelope should populate Resolved field")
+				require.Equal(t, ts.AsOfSystemTime(), res.Resolved)
+			case changefeedbase.OptEnvelopeBare:
+				res := msg.GetBareResolved()
+				require.NotNil(t, res, "bare envelope should populate BareResolved field")
+				require.Equal(t, ts.AsOfSystemTime(), res.XCrdb__.Resolved)
+			default:
+				t.Fatalf("unexpected envelope type: %v", tc.envelopeType)
+			}
+		})
+	}
+}