Encode key/payload with schema

cmd/connector/main.go

@@ -16,13 +16,9 @@ package main
 
 import (
 	postgres "github.com/conduitio/conduit-connector-postgres"
-	"github.com/conduitio/conduit-connector-postgres/source/types"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 )
 
 func main() {
-	// Running as standalone plugin
-	types.WithBuiltinPlugin = false
-
 	sdk.Serve(postgres.Connector)
 }

source/logrepl/internal/relationset_test.go

@@ -96,11 +96,9 @@ func TestRelationSetAllTypes(t *testing.T) {
 	t.Run("with standalone plugin", func(t *testing.T) {
 		is := is.New(t)
 
-		types.WithBuiltinPlugin = false
 		values, err := rs.Values(ins.RelationID, ins.Tuple)
 		is.NoErr(err)
 		isValuesAllTypesStandalone(is, values)
-		types.WithBuiltinPlugin = true
 	})
 }
 

source/snapshot/iterator.go

@@ -95,7 +95,7 @@ func (i *Iterator) Next(ctx context.Context) (opencdc.Record, error) {
 		}
 
 		i.acks.Add(1)
-		return i.buildRecord(d), nil
+		return i.buildRecord(d)
 	}
 }
 
@@ -112,7 +112,7 @@ func (i *Iterator) Teardown(_ context.Context) error {
 	return nil
 }
 
-func (i *Iterator) buildRecord(d FetchData) opencdc.Record {
+func (i *Iterator) buildRecord(d FetchData) (opencdc.Record, error) {
 	// merge this position with latest position
 	i.lastPosition.Type = position.TypeSnapshot
 	i.lastPosition.Snapshots[d.Table] = d.Position
@@ -121,11 +121,41 @@ func (i *Iterator) buildRecord(d FetchData) opencdc.Record {
 	metadata := make(opencdc.Metadata)
 	metadata["postgres.table"] = d.Table
 
-	rec := sdk.Util.Source.NewRecordSnapshot(pos, metadata, d.Key, d.Payload)
+	encodedPayload, err := i.encodeWithSchema(d.PayloadSchema, d.Payload)
+	if err != nil {
+		return opencdc.Record{}, fmt.Errorf("failed to encode payload with schema: %w", err)
+	}
+
+	encodedKey, err := i.encodeWithSchema(d.KeySchema, d.Key)
+	if err != nil {
+		return opencdc.Record{}, fmt.Errorf("failed to encode key with schema: %w", err)
+	}
+
+	rec := sdk.Util.Source.NewRecordSnapshot(
+		pos,
+		metadata,
+		opencdc.RawData(encodedKey),
+		opencdc.RawData(encodedPayload),
+	)
+
 	cschema.AttachKeySchemaToRecord(rec, d.KeySchema)
 	cschema.AttachPayloadSchemaToRecord(rec, d.PayloadSchema)
 
-	return rec
+	return rec, nil
+}
+
+func (i *Iterator) encodeWithSchema(sch cschema.Schema, data any) ([]byte, error) {
+	srd, err := sch.Serde()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get serde for schema: %w", err)
+	}
+
+	encoded, err := srd.Marshal(data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal data with schema: %w", err)
+	}
+
+	return encoded, nil
 }
 
 func (i *Iterator) initFetchers(ctx context.Context) error {

source/types/types.go

@@ -15,8 +15,6 @@
 package types
 
 import (
-	"time"
-
 	"github.com/jackc/pgx/v5/pgtype"
 )
 
@@ -26,8 +24,6 @@ var (
 	UUID    = UUIDFormatter{}
 )
 
-var WithBuiltinPlugin = true
-
 func Format(oid uint32, v any) (any, error) {
 	if oid == pgtype.UUIDOID {
 		return UUID.Format(v)
@@ -38,10 +34,6 @@ func Format(oid uint32, v any) (any, error) {
 		return Numeric.Format(t)
 	case *pgtype.Numeric:
 		return Numeric.Format(*t)
-	case time.Time:
-		return Time.Format(t)
-	case *time.Time:
-		return Time.Format(*t)
 	case []uint8:
 		if oid == pgtype.XMLOID {
 			return string(t), nil

source/types/types_test.go

@@ -30,7 +30,6 @@ func Test_Format(t *testing.T) {
 		input       []any
 		inputOID    []uint32
 		expect      []any
-		withBuiltin bool
 	}{
 		{
 			name: "int float string bool",
@@ -58,25 +57,6 @@ func Test_Format(t *testing.T) {
 		},
 		{
 			name: "time.Time",
-			input: []any{
-				func() time.Time {
-					is := is.New(t)
-					is.Helper()
-					t, err := time.Parse(time.DateTime, "2009-11-10 23:00:00")
-					is.NoErr(err)
-					return t
-				}(),
-				nil,
-			},
-			inputOID: []uint32{
-				0, 0,
-			},
-			expect: []any{
-				"2009-11-10 23:00:00 +0000 UTC", nil,
-			},
-		},
-		{
-			name: "builtin time.Time",
 			input: []any{
 				now,
 			},
@@ -86,7 +66,6 @@ func Test_Format(t *testing.T) {
 			expect: []any{
 				now,
 			},
-			withBuiltin: true,
 		},
 		{
 			name: "uuid",
@@ -107,13 +86,6 @@ func Test_Format(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			is := is.New(t)
 
-			prevWithBuiltinPlugin := WithBuiltinPlugin
-			WithBuiltinPlugin = tc.withBuiltin
-
-			t.Cleanup(func() {
-				WithBuiltinPlugin = prevWithBuiltinPlugin
-			})
-
 			for i, in := range tc.input {
 				v, err := Format(tc.inputOID[i], in)
 				is.NoErr(err)