avro schema generator (#174)

* schema generator
* snapshot emits schema
* add logrepl schema
* add uuid type formatting

go.mod

@@ -1,6 +1,6 @@
 module github.com/conduitio/conduit-connector-postgres
 
-go 1.22
+go 1.22.0
 
 require (
  github.com/Masterminds/sprig/v3 v3.2.3
@@ -11,6 +11,7 @@ require (
  github.com/golangci/golangci-lint v1.59.1
  github.com/google/go-cmp v0.6.0
  github.com/google/uuid v1.6.0
+ github.com/hamba/avro/v2 v2.22.1
  github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438
  github.com/jackc/pglogrepl v0.0.0-20240307033717-828fbfe908e9
  github.com/jackc/pgx/v5 v5.6.0
@@ -110,6 +111,7 @@ require (
  github.com/jirfag/go-printf-func-name v0.0.0-20200119135958-7558a9eaa5af // indirect
  github.com/jjti/go-spancheck v0.6.1 // indirect
  github.com/jpillora/backoff v1.0.0 // indirect
+ github.com/json-iterator/go v1.1.12 // indirect
  github.com/julz/importas v0.1.0 // indirect
  github.com/karamaru-alpha/copyloopvar v1.1.0 // indirect
  github.com/kisielk/errcheck v1.7.0 // indirect
@@ -139,6 +141,8 @@ require (
  github.com/mitchellh/go-testing-interface v1.14.1 // indirect
  github.com/mitchellh/mapstructure v1.5.0 // indirect
  github.com/mitchellh/reflectwalk v1.0.0 // indirect
+ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+ github.com/modern-go/reflect2 v1.0.2 // indirect
  github.com/moricho/tparallel v0.3.1 // indirect
  github.com/nakabonne/nestif v0.3.1 // indirect
  github.com/nishanths/exhaustive v0.12.0 // indirect

go.sum

@@ -305,6 +305,8 @@ github.com/gostaticanalysis/nilerr v0.1.1/go.mod h1:wZYb6YI5YAxxq0i1+VJbY0s2YONW
 github.com/gostaticanalysis/testutil v0.3.1-0.20210208050101-bfb5c8eec0e4/go.mod h1:D+FIZ+7OahH3ePw/izIEeH5I06eKs1IKI4Xr64/Am3M=
 github.com/gostaticanalysis/testutil v0.4.0 h1:nhdCmubdmDF6VEatUNjgUZBJKWRqugoISdUv3PPQgHY=
 github.com/gostaticanalysis/testutil v0.4.0/go.mod h1:bLIoPefWXrRi/ssLFWX1dx7Repi5x3CuviD3dgAZaBU=
+github.com/hamba/avro/v2 v2.22.1 h1:q1rAbfJsrbMaZPDLQvwUQMfQzp6H+hGXvckmU/lXemk=
+github.com/hamba/avro/v2 v2.22.1/go.mod h1:HOeTrE3kvWnBAgsufqhAzDDV5gvS0QXs65Z6BHfGgbg=
 github.com/hashicorp/go-hclog v1.5.0 h1:bI2ocEMgcVlz55Oj1xZNBsVi900c7II+fWDyV9o+13c=
 github.com/hashicorp/go-hclog v1.5.0/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
 github.com/hashicorp/go-plugin v1.6.0 h1:wgd4KxHJTVGGqWBq4QPB1i5BZNEx9BR8+OFmHDmTk8A=
@@ -356,6 +358,7 @@ github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
@@ -440,9 +443,11 @@ github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RR
 github.com/mitchellh/reflectwalk v1.0.0 h1:9D+8oIskB4VJBN5SFlmc27fSlIBZaov1Wpk/IfikLNY=
 github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/moricho/tparallel v0.3.1 h1:fQKD4U1wRMAYNngDonW5XupoB/ZGJHdpzrWqgyg9krA=
 github.com/moricho/tparallel v0.3.1/go.mod h1:leENX2cUv7Sv2qDgdi0D0fCftN8fRC67Bcn8pqzeYNI=

source.go

@@ -104,6 +104,7 @@ func (s *Source) Open(ctx context.Context, pos sdk.Position) error {
  TableKeys: s.tableKeys,
  WithSnapshot: s.config.SnapshotMode == source.SnapshotModeInitial,
  SnapshotFetchSize: s.config.SnapshotFetchSize,
+ WithAvroSchema: s.config.WithAvroSchema,
  })
  if err != nil {
  return fmt.Errorf("failed to create logical replication iterator: %w", err)

source/config.go

@@ -49,6 +49,7 @@ const (
 type Config struct {
  // URL is the connection string for the Postgres database.
  URL string `json:"url" validate:"required"`
+
  // Tables is a List of table names to read from, separated by a comma, e.g.:"table1,table2".
  // Use "*" if you'd like to listen to all tables.
  Tables []string `json:"tables"`
@@ -74,6 +75,10 @@ type Config struct {
  // LogreplAutoCleanup determines if the replication slot and publication should be
  // removed when the connector is deleted.
  LogreplAutoCleanup bool `json:"logrepl.autoCleanup" default:"true"`
+
+ // WithAvroSchema determines whether the connector should attach an avro schema on each
+ // record.
+ WithAvroSchema bool `json:"logrepl.withAvroSchema" default:"false"`
 }
 
 // Validate validates the provided config values.

source/logrepl/cdc.go

@@ -33,6 +33,7 @@ type CDCConfig struct {
  PublicationName string
  Tables []string
  TableKeys map[string]string
+ WithAvroSchema bool
 }
 
 // CDCIterator asynchronously listens for events from the logical replication
@@ -63,6 +64,7 @@ func NewCDCIterator(ctx context.Context, pool *pgxpool.Pool, c CDCConfig) (*CDCI
  }
 
  records := make(chan sdk.Record)
+ handler := NewCDCHandler(internal.NewRelationSet(), c.TableKeys, c.WithAvroSchema, records)
 
  sub, err := internal.CreateSubscription(
  ctx,
@@ -71,7 +73,7 @@ func NewCDCIterator(ctx context.Context, pool *pgxpool.Pool, c CDCConfig) (*CDCI
  c.PublicationName,
  c.Tables,
  c.LSN,
- NewCDCHandler(internal.NewRelationSet(), c.TableKeys, records).Handle,
+ handler.Handle,
  )
  if err != nil {
  return nil, fmt.Errorf("failed to initialize subscription: %w", err)

source/logrepl/combined.go

@@ -48,6 +48,7 @@ type Config struct {
  TableKeys map[string]string
  WithSnapshot bool
  SnapshotFetchSize int
+ WithAvroSchema bool
 }
 
 // Validate performs validation tasks on the config.
@@ -177,6 +178,7 @@ func (c *CombinedIterator) initCDCIterator(ctx context.Context, pos position.Pos
  PublicationName: c.conf.PublicationName,
  Tables: c.conf.Tables,
  TableKeys: c.conf.TableKeys,
+ WithAvroSchema: c.conf.WithAvroSchema,
  })
  if err != nil {
  return fmt.Errorf("failed to create CDC iterator: %w", err)
@@ -198,11 +200,12 @@ func (c *CombinedIterator) initSnapshotIterator(ctx context.Context, pos positio
  }
 
  snapshotIterator, err := snapshot.NewIterator(ctx, c.pool, snapshot.Config{
- Position: c.conf.Position,
- Tables: c.conf.Tables,
- TableKeys: c.conf.TableKeys,
- TXSnapshotID: c.cdcIterator.TXSnapshotID(),
- FetchSize: c.conf.SnapshotFetchSize,
+ Position: c.conf.Position,
+ Tables: c.conf.Tables,
+ TableKeys: c.conf.TableKeys,
+ TXSnapshotID: c.cdcIterator.TXSnapshotID(),
+ FetchSize: c.conf.SnapshotFetchSize,
+ WithAvroSchema: c.conf.WithAvroSchema,
  })
  if err != nil {
  return fmt.Errorf("failed to create snapshot iterator: %w", err)

source/logrepl/handler.go

@@ -20,7 +20,9 @@ import (
 
  "github.com/conduitio/conduit-connector-postgres/source/logrepl/internal"
  "github.com/conduitio/conduit-connector-postgres/source/position"
+ "github.com/conduitio/conduit-connector-postgres/source/schema"
  sdk "github.com/conduitio/conduit-connector-sdk"
+ "github.com/hamba/avro/v2"
  "github.com/jackc/pglogrepl"
 )
 
@@ -31,17 +33,23 @@ type CDCHandler struct {
  relationSet *internal.RelationSet
  out chan<- sdk.Record
  lastTXLSN pglogrepl.LSN
+
+ relAvroSchema map[string]avro.Schema
+ withAvroSchema bool
 }
 
 func NewCDCHandler(
  rs *internal.RelationSet,
  tableKeys map[string]string,
+ withAvroSchema bool,
  out chan<- sdk.Record,
 ) *CDCHandler {
  return &CDCHandler{
- tableKeys: tableKeys,
- relationSet: rs,
- out: out,
+ tableKeys: tableKeys,
+ relationSet: rs,
+ out: out,
+ withAvroSchema: withAvroSchema,
+ relAvroSchema: make(map[string]avro.Schema),
  }
 }
 
@@ -100,6 +108,10 @@ func (h *CDCHandler) handleInsert(
  return fmt.Errorf("failed to decode new values: %w", err)
  }
 
+ if err := h.updateAvroSchema(rel, msg.Tuple); err != nil {
+ return fmt.Errorf("failed to update avro schema: %w", err)
+ }
+
  rec := sdk.Util.Source.NewRecordCreate(
  h.buildPosition(lsn),
  h.buildRecordMetadata(rel),
@@ -127,6 +139,10 @@ func (h *CDCHandler) handleUpdate(
  return fmt.Errorf("failed to decode new values: %w", err)
  }
 
+ if err := h.updateAvroSchema(rel, msg.NewTuple); err != nil {
+ return fmt.Errorf("failed to update avro schema: %w", err)
+ }
+
  oldValues, err := h.relationSet.Values(msg.RelationID, msg.OldTuple)
  if err != nil {
  // this is not a critical error, old values are optional, just log it
@@ -180,10 +196,16 @@ func (h *CDCHandler) send(ctx context.Context, rec sdk.Record) error {
  }
 }
 
-func (h *CDCHandler) buildRecordMetadata(relation *pglogrepl.RelationMessage) map[string]string {
- return map[string]string{
- sdk.MetadataCollection: relation.RelationName,
+func (h *CDCHandler) buildRecordMetadata(rel *pglogrepl.RelationMessage) map[string]string {
+ m := map[string]string{
+ sdk.MetadataCollection: rel.RelationName,
  }
+
+ if h.withAvroSchema {
+ m[schema.AvroMetadataKey] = h.relAvroSchema[rel.RelationName].String()
+ }
+
+ return m
 }
 
 // buildRecordKey takes the values from the message and extracts the key that
@@ -209,9 +231,27 @@ func (h *CDCHandler) buildRecordPayload(values map[string]any) sdk.Data {
  return sdk.StructuredData(values)
 }
 
+// buildPosition stores the LSN in position and converts it to bytes.
 func (*CDCHandler) buildPosition(lsn pglogrepl.LSN) sdk.Position {
  return position.Position{
  Type: position.TypeCDC,
  LastLSN: lsn.String(),
  }.ToSDKPosition()
 }
+
+// updateAvroSchema generates and stores avro schema based on the relation's row,
+// when usage of avro schema is requested.
+func (h *CDCHandler) updateAvroSchema(rel *pglogrepl.RelationMessage, row *pglogrepl.TupleData) error {
+ if !h.withAvroSchema {
+ return nil
+ }
+
+ sch, err := schema.Avro.ExtractLogrepl(rel, row)
+ if err != nil {
+ return err
+ }
+
+ h.relAvroSchema[rel.RelationName] = sch
+
+ return nil
+}

source/logrepl/internal/relationset.go

@@ -71,7 +71,7 @@ func (rs *RelationSet) Values(id uint32, row *pglogrepl.TupleData) (map[string]a
  return nil, fmt.Errorf("failed to decode tuple %d: %w", i, err)
  }
 
- v, err := types.Format(val)
+ v, err := types.Format(col.DataType, val)
  if err != nil {
  return nil, fmt.Errorf("failed to format column %q type %T: %w", col.Name, val, err)
  }

source/logrepl/internal/relationset_test.go

@@ -344,7 +344,7 @@ func isValuesAllTypes(is *is.I, got map[string]any) {
  "col_timestamptz": time.Date(2022, 3, 14, 15+8, 16, 17, 0, time.UTC).UTC(),
  "col_tsquery": "'fat' & ( 'rat' | 'cat' )",
  "col_tsvector": "'a' 'and' 'ate' 'cat' 'fat' 'mat' 'on' 'rat' 'sat'",
- "col_uuid": [16]uint8{0xbd, 0x94, 0xee, 0x0b, 0x56, 0x4f, 0x40, 0x88, 0xbf, 0x4e, 0x8d, 0x5e, 0x62, 0x6c, 0xaf, 0x66},
+ "col_uuid": "bd94ee0b-564f-4088-bf4e-8d5e626caf66", // [16]uint8{0xbd, 0x94, 0xee, 0x0b, 0x56, 0x4f, 0x40, 0x88, 0xbf, 0x4e, 0x8d, 0x5e, 0x62, 0x6c, 0xaf, 0x66}
  "col_xml": "<foo>bar</foo>",
  }
  is.Equal("", cmp.Diff(want, got,
@@ -440,7 +440,7 @@ func isValuesAllTypesStandalone(is *is.I, got map[string]any) {
  "col_timestamptz": time.Date(2022, 3, 14, 15+8, 16, 17, 0, time.UTC).UTC().String(),
  "col_tsquery": "'fat' & ( 'rat' | 'cat' )",
  "col_tsvector": "'a' 'and' 'ate' 'cat' 'fat' 'mat' 'on' 'rat' 'sat'",
- "col_uuid": [16]uint8{0xbd, 0x94, 0xee, 0x0b, 0x56, 0x4f, 0x40, 0x88, 0xbf, 0x4e, 0x8d, 0x5e, 0x62, 0x6c, 0xaf, 0x66},
+ "col_uuid": "bd94ee0b-564f-4088-bf4e-8d5e626caf66", // [16]uint8{0xbd, 0x94, 0xee, 0x0b, 0x56, 0x4f, 0x40, 0x88, 0xbf, 0x4e, 0x8d, 0x5e, 0x62, 0x6c, 0xaf, 0x66}
  "col_xml": "<foo>bar</foo>",
  }
  is.Equal("", cmp.Diff(want, got,