Upgrade pgx to 5.7.1 (#212)

Author: hariso

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/hamba/avro/v2 v2.26.0
 	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
+	github.com/jackc/pgx/v5 v5.7.1
 	github.com/matryer/is v1.4.1
 	golang.org/x/tools v0.26.0
 	gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637

go.sum

@@ -224,8 +224,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
-github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
+github.com/jackc/pgx/v5 v5.7.1 h1:x7SYsPBYDkHDksogeSmZZ5xzThcTgRz++I5E+ePFUcs=
+github.com/jackc/pgx/v5 v5.7.1/go.mod h1:e7O26IywZZ+naJtWWos6i6fvWK+29etgITqrqHLfoZA=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jgautheron/goconst v1.7.1 h1:VpdAG7Ca7yvvJk5n8dMwQhfEZJh95kl/Hl9S1OI5Jkk=

source/logrepl/internal/relationset.go

@@ -65,15 +65,9 @@ func (rs *RelationSet) Values(id uint32, row *pglogrepl.TupleData) (map[string]a
 	// assert same number of row and rel columns
 	for i, tuple := range row.Columns {
 		col := rel.Columns[i]
-		decoder := rs.oidToCodec(col.DataType)
-		val, err := decoder.DecodeValue(rs.connInfo, col.DataType, pgtype.TextFormatCode, tuple.Data)
-		if err != nil {
-			return nil, fmt.Errorf("failed to decode tuple %d: %w", i, err)
-		}
-
-		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)
+		v, decodeErr := rs.decodeValue(col, tuple.Data)
+		if decodeErr != nil {
+			return nil, fmt.Errorf("failed to decode value for column %q: %w", col.Name, err)
 		}
 
 		values[col.Name] = v
@@ -89,3 +83,30 @@ func (rs *RelationSet) oidToCodec(id uint32) pgtype.Codec {
 	}
 	return dt.Codec
 }
+
+func (rs *RelationSet) decodeValue(col *pglogrepl.RelationMessageColumn, data []byte) (any, error) {
+	decoder := rs.oidToCodec(col.DataType)
+	// This workaround is due to an issue in pgx v5.7.1.
+	// Namely, that version introduces an XML codec
+	// (see: https://github.com/jackc/pgx/pull/2083/files#diff-8288d41e69f73d01a874b40de086684e5894da83a627e845e484b06d5e053a44).
+	// The XML codec, however, always return nil when deserializing input bytes
+	// (see: https://github.com/jackc/pgx/pull/2083#discussion_r1755768269).
+	var val any
+	var err error
+	if col.DataType == pgtype.XMLOID || col.DataType == pgtype.XMLArrayOID {
+		val, err = decoder.DecodeDatabaseSQLValue(rs.connInfo, col.DataType, pgtype.TextFormatCode, data)
+	} else {
+		val, err = decoder.DecodeValue(rs.connInfo, col.DataType, pgtype.TextFormatCode, data)
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to decode value of pgtype %v: %w", col.DataType, err)
+	}
+
+	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)
+	}
+
+	return v, nil
+}

source/types/types.go

@@ -42,6 +42,11 @@ func Format(oid uint32, v any) (any, error) {
 		return Time.Format(t)
 	case *time.Time:
 		return Time.Format(*t)
+	case []uint8:
+		if oid == pgtype.XMLOID {
+			return string(t), nil
+		}
+		return t, nil
 	default:
 		return t, nil
 	}