Read Postgres' NUMERIC as big.Rat

Author: hariso

destination.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math/big"
 	"strings"
 
 	sq "github.com/Masterminds/squirrel"
@@ -26,6 +27,7 @@ import (
 	"github.com/conduitio/conduit-connector-postgres/internal"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"github.com/jackc/pgx/v5"
+	"github.com/shopspring/decimal"
 )
 
 type Destination struct {
@@ -35,6 +37,7 @@ type Destination struct {
 	getTableName destination.TableFn
 
 	conn        *pgx.Conn
+	dbInfo      *internal.DbInfo
 	stmtBuilder sq.StatementBuilderType
 }
 
@@ -61,6 +64,7 @@ func (d *Destination) Open(ctx context.Context) error {
 		return fmt.Errorf("invalid table name or table name function: %w", err)
 	}
 
+	d.dbInfo = internal.NewDbInfo(conn)
 	return nil
 }
 
@@ -215,7 +219,11 @@ func (d *Destination) insert(ctx context.Context, r opencdc.Record, b *pgx.Batch
 		return err
 	}
 
-	colArgs, valArgs := d.formatColumnsAndValues(key, payload)
+	colArgs, valArgs, err := d.formatColumnsAndValues(tableName, key, payload)
+	if err != nil {
+		return fmt.Errorf("error formatting columns and values: %w", err)
+	}
+
 	sdk.Logger(ctx).Trace().
 		Str("table_name", tableName).
 		Msg("inserting record")
@@ -294,10 +302,13 @@ func (d *Destination) formatUpsertQuery(
 	// remove the last comma from the list of tuples
 	upsertQuery = strings.TrimSuffix(upsertQuery, ",")
 
-	// we have to manually append a semi colon to the upsert sql;
+	// we have to manually append a semicolon to the upsert sql;
 	upsertQuery += ";"
 
-	colArgs, valArgs := d.formatColumnsAndValues(key, payload)
+	colArgs, valArgs, err := d.formatColumnsAndValues(tableName, key, payload)
+	if err != nil {
+		return "", nil, fmt.Errorf("error formatting columns and values: %w", err)
+	}
 
 	return d.stmtBuilder.
 		Insert(internal.WrapSQLIdent(tableName)).
@@ -309,24 +320,32 @@ func (d *Destination) formatUpsertQuery(
 
 // formatColumnsAndValues turns the key and payload into a slice of ordered
 // columns and values for upserting into Postgres.
-func (d *Destination) formatColumnsAndValues(key, payload opencdc.StructuredData) ([]string, []interface{}) {
+func (d *Destination) formatColumnsAndValues(table string, key, payload opencdc.StructuredData) ([]string, []interface{}, error) {
 	var colArgs []string
 	var valArgs []interface{}
 
 	// range over both the key and payload values in order to format the
 	// query for args and values in proper order
 	for key, val := range key {
 		colArgs = append(colArgs, internal.WrapSQLIdent(key))
-		valArgs = append(valArgs, val)
+		formatted, err := d.formatValue(table, key, val)
+		if err != nil {
+			return nil, nil, fmt.Errorf("error formatting value: %w", err)
+		}
+		valArgs = append(valArgs, formatted)
 		delete(payload, key) // NB: Delete Key from payload arguments
 	}
 
-	for field, value := range payload {
+	for field, val := range payload {
 		colArgs = append(colArgs, internal.WrapSQLIdent(field))
-		valArgs = append(valArgs, value)
+		formatted, err := d.formatValue(table, field, val)
+		if err != nil {
+			return nil, nil, fmt.Errorf("error formatting value: %w", err)
+		}
+		valArgs = append(valArgs, formatted)
 	}
 
-	return colArgs, valArgs
+	return colArgs, valArgs, nil
 }
 
 // getKeyColumnName will return the name of the first item in the key or the
@@ -346,3 +365,28 @@ func (d *Destination) getKeyColumnName(key opencdc.StructuredData, defaultKeyNam
 func (d *Destination) hasKey(e opencdc.Record) bool {
 	return e.Key != nil && len(e.Key.Bytes()) > 0
 }
+
+func (d *Destination) formatValue(table string, column string, val interface{}) (interface{}, error) {
+	switch v := val.(type) {
+	case *big.Rat:
+		return d.formatBigRat(table, column, v)
+	case big.Rat:
+		return d.formatBigRat(table, column, &v)
+	default:
+		return val, nil
+	}
+}
+
+func (d *Destination) formatBigRat(table string, column string, v *big.Rat) (string, error) {
+	scale, err := d.dbInfo.GetNumericColumnScale(table, column)
+	if err != nil {
+		return "", fmt.Errorf("failed getting scale of numeric column: %w", err)
+	}
+
+	if v == nil {
+		return "", nil
+	}
+
+	//nolint:gosec // no risk of overflow, because the scale in Pg is always <= 16383
+	return decimal.NewFromBigRat(v, int32(scale)).String(), nil
+}

destination_integration_test.go

@@ -17,12 +17,14 @@ package postgres
 import (
 	"context"
 	"fmt"
+	"math/big"
 	"strings"
 	"testing"
 
 	"github.com/conduitio/conduit-commons/opencdc"
 	"github.com/conduitio/conduit-connector-postgres/test"
 	sdk "github.com/conduitio/conduit-connector-sdk"
+	"github.com/google/go-cmp/cmp"
 	"github.com/jackc/pgx/v5"
 	"github.com/matryer/is"
 )
@@ -71,11 +73,13 @@ func TestDestination_Write(t *testing.T) {
 						"column1":          "foo",
 						"column2":          123,
 						"column3":          true,
+						"column4":          nil,
 						"UppercaseColumn1": 222,
 					},
 				},
 			},
-		}, {
+		},
+		{
 			name: "create",
 			record: opencdc.Record{
 				Position:  opencdc.Position("foo"),
@@ -87,11 +91,13 @@ func TestDestination_Write(t *testing.T) {
 						"column1":          "foo",
 						"column2":          456,
 						"column3":          false,
+						"column4":          nil,
 						"UppercaseColumn1": 333,
 					},
 				},
 			},
-		}, {
+		},
+		{
 			name: "insert on update (upsert)",
 			record: opencdc.Record{
 				Position:  opencdc.Position("foo"),
@@ -103,11 +109,13 @@ func TestDestination_Write(t *testing.T) {
 						"column1":          "bar",
 						"column2":          567,
 						"column3":          true,
+						"column4":          nil,
 						"UppercaseColumn1": 444,
 					},
 				},
 			},
-		}, {
+		},
+		{
 			name: "update on conflict",
 			record: opencdc.Record{
 				Position:  opencdc.Position("foo"),
@@ -119,11 +127,13 @@ func TestDestination_Write(t *testing.T) {
 						"column1":          "foobar",
 						"column2":          567,
 						"column3":          true,
+						"column4":          nil,
 						"UppercaseColumn1": 555,
 					},
 				},
 			},
-		}, {
+		},
+		{
 			name: "delete",
 			record: opencdc.Record{
 				Position:  opencdc.Position("foo"),
@@ -132,6 +142,24 @@ func TestDestination_Write(t *testing.T) {
 				Key:       opencdc.StructuredData{"id": 4},
 			},
 		},
+		{
+			name: "write a big.Rat",
+			record: opencdc.Record{
+				Position:  opencdc.Position("foo"),
+				Operation: opencdc.OperationSnapshot,
+				Metadata:  map[string]string{opencdc.MetadataCollection: tableName},
+				Key:       opencdc.StructuredData{"id": 123},
+				Payload: opencdc.Change{
+					After: opencdc.StructuredData{
+						"column1":          "abcdef",
+						"column2":          567,
+						"column3":          true,
+						"column4":          big.NewRat(123, 100),
+						"UppercaseColumn1": 555,
+					},
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -146,7 +174,16 @@ func TestDestination_Write(t *testing.T) {
 			switch tt.record.Operation {
 			case opencdc.OperationCreate, opencdc.OperationSnapshot, opencdc.OperationUpdate:
 				is.NoErr(err)
-				is.Equal(tt.record.Payload.After, got)
+				is.Equal(
+					"",
+					cmp.Diff(
+						tt.record.Payload.After,
+						got,
+						cmp.Comparer(func(x, y *big.Rat) bool {
+							return x.Cmp(y) == 0
+						}),
+					),
+				) // -want, +got
 			case opencdc.OperationDelete:
 				is.Equal(err, pgx.ErrNoRows)
 			}
@@ -179,43 +216,50 @@ func TestDestination_Batch(t *testing.T) {
 		is.NoErr(err)
 	}()
 
-	records := []opencdc.Record{{
-		Position:  opencdc.Position("foo1"),
-		Operation: opencdc.OperationCreate,
-		Key:       opencdc.StructuredData{"id": 5},
-		Payload: opencdc.Change{
-			After: opencdc.StructuredData{
-				"column1":          "foo1",
-				"column2":          1,
-				"column3":          false,
-				"UppercaseColumn1": 111,
+	records := []opencdc.Record{
+		{
+			Position:  opencdc.Position("foo1"),
+			Operation: opencdc.OperationCreate,
+			Key:       opencdc.StructuredData{"id": 5},
+			Payload: opencdc.Change{
+				After: opencdc.StructuredData{
+					"column1":          "foo1",
+					"column2":          1,
+					"column3":          false,
+					"column4":          nil,
+					"UppercaseColumn1": 111,
+				},
 			},
 		},
-	}, {
-		Position:  opencdc.Position("foo2"),
-		Operation: opencdc.OperationCreate,
-		Key:       opencdc.StructuredData{"id": 6},
-		Payload: opencdc.Change{
-			After: opencdc.StructuredData{
-				"column1":          "foo2",
-				"column2":          2,
-				"column3":          true,
-				"UppercaseColumn1": 222,
+		{
+			Position:  opencdc.Position("foo2"),
+			Operation: opencdc.OperationCreate,
+			Key:       opencdc.StructuredData{"id": 6},
+			Payload: opencdc.Change{
+				After: opencdc.StructuredData{
+					"column1":          "foo2",
+					"column2":          2,
+					"column3":          true,
+					"column4":          nil,
+					"UppercaseColumn1": 222,
+				},
 			},
 		},
-	}, {
-		Position:  opencdc.Position("foo3"),
-		Operation: opencdc.OperationCreate,
-		Key:       opencdc.StructuredData{"id": 7},
-		Payload: opencdc.Change{
-			After: opencdc.StructuredData{
-				"column1":          "foo3",
-				"column2":          3,
-				"column3":          false,
-				"UppercaseColumn1": 333,
+		{
+			Position:  opencdc.Position("foo3"),
+			Operation: opencdc.OperationCreate,
+			Key:       opencdc.StructuredData{"id": 7},
+			Payload: opencdc.Change{
+				After: opencdc.StructuredData{
+					"column1":          "foo3",
+					"column2":          3,
+					"column3":          false,
+					"column4":          nil,
+					"UppercaseColumn1": 333,
+				},
 			},
 		},
-	}}
+	}
 
 	i, err := d.Write(ctx, records)
 	is.NoErr(err)
@@ -231,25 +275,36 @@ func TestDestination_Batch(t *testing.T) {
 func queryTestTable(ctx context.Context, conn test.Querier, tableName string, id any) (opencdc.StructuredData, error) {
 	row := conn.QueryRow(
 		ctx,
-		fmt.Sprintf(`SELECT column1, column2, column3, "UppercaseColumn1" FROM %q WHERE id = $1`, tableName),
+		fmt.Sprintf(`SELECT column1, column2, column3, column4, "UppercaseColumn1" FROM %q WHERE id = $1`, tableName),
 		id,
 	)
 
 	var (
 		col1          string
 		col2          int
 		col3          bool
+		col4Str       *string
 		uppercaseCol1 int
 	)
-	err := row.Scan(&col1, &col2, &col3, &uppercaseCol1)
+
+	err := row.Scan(&col1, &col2, &col3, &col4Str, &uppercaseCol1)
 	if err != nil {
 		return nil, err
 	}
 
+	// Handle the potential nil case for col4
+	var col4 interface{}
+	if col4Str != nil {
+		r := new(big.Rat)
+		r.SetString(*col4Str)
+		col4 = r
+	}
+
 	return opencdc.StructuredData{
 		"column1":          col1,
 		"column2":          col2,
 		"column3":          col3,
+		"column4":          col4,
 		"UppercaseColumn1": uppercaseCol1,
 	}, nil
 }

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/jackc/pgx/v5 v5.7.5
 	github.com/matryer/is v1.4.1
 	github.com/rs/zerolog v1.34.0
+	github.com/shopspring/decimal v1.4.0
 	gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637
 )
 
@@ -181,7 +182,6 @@ require (
 	github.com/sashamelentyev/interfacebloat v1.1.0 // indirect
 	github.com/sashamelentyev/usestdlibvars v1.28.0 // indirect
 	github.com/securego/gosec/v2 v2.22.2 // indirect
-	github.com/shopspring/decimal v1.4.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/sivchari/containedctx v1.0.3 // indirect
 	github.com/sivchari/tenv v1.12.1 // indirect