changefeedccl: protobuf encoder randomized test for all types

We recently added support for the Protobuf encoding format in changefeeds.
This test ensures that all types supported in CockroachDB are correctly
encoded and decoded using Protobuf.

Release note: None
Fixes #149797

Author: elizaMkraule

pkg/ccl/changefeedccl/BUILD.bazel

@@ -71,7 +71,6 @@ go_library(
         "//pkg/clusterversion",
         "//pkg/docs",
         "//pkg/featureflag",
-        "//pkg/geo",
         "//pkg/jobs",
         "//pkg/jobs/jobsauth",
         "//pkg/jobs/jobspb",
@@ -254,6 +253,8 @@ go_test(
         "//pkg/cloud",
         "//pkg/cloud/cloudpb",
         "//pkg/cloud/impl:cloudimpl",
+        "//pkg/geo",
+        "//pkg/geo/geopb",
         "//pkg/internal/sqlsmith",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
@@ -322,7 +323,9 @@ go_test(
         "//pkg/testutils/testcluster",
         "//pkg/util",
         "//pkg/util/cidr",
+        "//pkg/util/collatedstring",
         "//pkg/util/ctxgroup",
+        "//pkg/util/duration",
         "//pkg/util/encoding",
         "//pkg/util/hlc",
         "//pkg/util/intsets",
@@ -341,6 +344,7 @@ go_test(
         "//pkg/util/span",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
+        "//pkg/util/timeutil/pgdate",
         "//pkg/util/tracing",
         "//pkg/util/tracing/tracingpb",
         "//pkg/util/uuid",
@@ -359,6 +363,7 @@ go_test(
         "@com_github_jackc_pgx_v5//:pgx",
         "@com_github_klauspost_compress//gzip",
         "@com_github_lib_pq//:pq",
+        "@com_github_lib_pq//oid",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_twmb_franz_go//pkg/kerr",

pkg/ccl/changefeedccl/encoder_protobuf.go

@@ -8,13 +8,13 @@ package changefeedccl
 import (
 	"context"
 	"fmt"
-	"strings"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"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/sql/sessiondatapb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
@@ -227,7 +227,7 @@ func encodeRowToRecord(row cdcevent.Row) (*changefeedpb.Record, error) {
 	}
 	record := &changefeedpb.Record{Values: make(map[string]*changefeedpb.Value, row.NumValueColumns())}
 	if err := row.ForEachColumn().Datum(func(d tree.Datum, col cdcevent.ResultColumn) error {
-		val, err := datumToProtoValue(d)
+		val, err := datumToProtoValue(d, sessiondatapb.DataConversionConfig{}, time.UTC)
 		if err != nil {
 			return err
 		}
@@ -244,7 +244,7 @@ func buildKeyMessage(row cdcevent.Row) (*changefeedpb.Key, error) {
 	keyMap := make(map[string]*changefeedpb.Value, row.NumKeyColumns())
 
 	if err := row.ForEachKeyColumn().Datum(func(d tree.Datum, col cdcevent.ResultColumn) error {
-		val, err := datumToProtoValue(d)
+		val, err := datumToProtoValue(d, sessiondatapb.DataConversionConfig{}, time.UTC)
 		if err != nil {
 			return err
 		}
@@ -259,12 +259,16 @@ func buildKeyMessage(row cdcevent.Row) (*changefeedpb.Key, error) {
 // datumToProtoValue converts a tree.Datum into a changefeedpb.Value.
 // It handles all common CockroachDB datum types and maps them to their
 // corresponding protobuf representation.
-func datumToProtoValue(d tree.Datum) (*changefeedpb.Value, error) {
+func datumToProtoValue(
+	d tree.Datum, dcc sessiondatapb.DataConversionConfig, loc *time.Location,
+) (*changefeedpb.Value, error) {
 	d = tree.UnwrapDOidWrapper(d)
+
 	if d == tree.DNull {
 		return nil, nil
 	}
 	switch v := d.(type) {
+
 	case *tree.DBool:
 		return &changefeedpb.Value{Value: &changefeedpb.Value_BoolValue{BoolValue: bool(*v)}}, nil
 	case *tree.DInt:
@@ -285,7 +289,7 @@ func datumToProtoValue(d tree.Datum) (*changefeedpb.Value, error) {
 	case *tree.DArray:
 		elems := make([]*changefeedpb.Value, 0, v.Len())
 		for _, elt := range v.Array {
-			pv, err := datumToProtoValue(elt)
+			pv, err := datumToProtoValue(elt, sessiondatapb.DataConversionConfig{}, time.UTC)
 			if err != nil {
 				return nil, err
 			}
@@ -297,7 +301,7 @@ func datumToProtoValue(d tree.Datum) (*changefeedpb.Value, error) {
 		labels := v.ResolvedType().TupleLabels()
 		records := make(map[string]*changefeedpb.Value, len(v.D))
 		for i, elem := range v.D {
-			pv, err := datumToProtoValue(elem)
+			pv, err := datumToProtoValue(elem, sessiondatapb.DataConversionConfig{}, time.UTC)
 			if err != nil {
 				return nil, err
 			}
@@ -330,36 +334,25 @@ func datumToProtoValue(d tree.Datum) (*changefeedpb.Value, error) {
 		return &changefeedpb.Value{Value: &changefeedpb.Value_TimestampValue{TimestampValue: ts}}, nil
 	case *tree.DBytes:
 		return &changefeedpb.Value{Value: &changefeedpb.Value_BytesValue{BytesValue: []byte(*v)}}, nil
-
 	case *tree.DGeography:
-		geostr, err := geo.SpatialObjectToWKT(v.Geography.SpatialObject(), -1)
-		if err != nil {
-			return nil, err
-		}
-		return &changefeedpb.Value{Value: &changefeedpb.Value_StringValue{StringValue: string(geostr)}}, nil
+		ewkb := v.EWKB()
+		return &changefeedpb.Value{Value: &changefeedpb.Value_BytesValue{BytesValue: ewkb}}, nil
 	case *tree.DGeometry:
-		geostr, err := geo.SpatialObjectToWKT(v.Geometry.SpatialObject(), -1)
-		if err != nil {
-			return nil, err
-		}
-		return &changefeedpb.Value{Value: &changefeedpb.Value_StringValue{StringValue: string(geostr)}}, nil
+		ewkb := v.EWKB()
+		return &changefeedpb.Value{Value: &changefeedpb.Value_BytesValue{BytesValue: ewkb}}, nil
 	case *tree.DVoid:
 		return nil, nil
+	case *tree.DOid, *tree.DIPAddr, *tree.DBitArray, *tree.DBox2D,
+		*tree.DTSVector, *tree.DTSQuery, *tree.DPGLSN, *tree.DPGVector:
+		return &changefeedpb.Value{Value: &changefeedpb.Value_StringValue{StringValue: tree.AsStringWithFlags(v, tree.FmtBareStrings, tree.FmtDataConversionConfig(dcc), tree.FmtLocation(loc))}}, nil
 	case *tree.DDate:
-		date, err := v.ToTime()
-		if err != nil {
-			return nil, err
-		}
-		return &changefeedpb.Value{Value: &changefeedpb.Value_DateValue{DateValue: date.Format("2006-01-02")}}, nil
+		return &changefeedpb.Value{Value: &changefeedpb.Value_DateValue{DateValue: tree.AsStringWithFlags(v, tree.FmtBareStrings, tree.FmtDataConversionConfig(dcc), tree.FmtLocation(loc))}}, nil
 	case *tree.DInterval:
-		return &changefeedpb.Value{Value: &changefeedpb.Value_IntervalValue{IntervalValue: v.Duration.String()}}, nil
+		return &changefeedpb.Value{Value: &changefeedpb.Value_IntervalValue{IntervalValue: tree.AsStringWithFlags(v, tree.FmtBareStrings, tree.FmtDataConversionConfig(dcc), tree.FmtLocation(loc))}}, nil
 	case *tree.DUuid:
-		return &changefeedpb.Value{Value: &changefeedpb.Value_UuidValue{UuidValue: strings.Trim(v.UUID.String(), "'")}}, nil
+		return &changefeedpb.Value{Value: &changefeedpb.Value_UuidValue{UuidValue: tree.AsStringWithFlags(v, tree.FmtBareStrings, tree.FmtDataConversionConfig(dcc), tree.FmtLocation(loc))}}, nil
 	case *tree.DTime, *tree.DTimeTZ:
-		return &changefeedpb.Value{Value: &changefeedpb.Value_TimeValue{TimeValue: tree.AsStringWithFlags(v, tree.FmtBareStrings)}}, nil
-	case *tree.DOid, *tree.DIPAddr, *tree.DBitArray, *tree.DBox2D,
-		*tree.DTSVector, *tree.DTSQuery, *tree.DPGLSN, *tree.DPGVector:
-		return &changefeedpb.Value{Value: &changefeedpb.Value_StringValue{StringValue: d.String()}}, nil
+		return &changefeedpb.Value{Value: &changefeedpb.Value_TimeValue{TimeValue: tree.AsStringWithFlags(v, tree.FmtBareStrings, tree.FmtDataConversionConfig(dcc), tree.FmtLocation(loc))}}, nil
 	default:
 		return nil, errors.AssertionFailedf("unexpected type %T for datumToProtoValue", d)
 	}