diff --git a/pkg/wal/listener/postgres/bytea.go b/pkg/wal/listener/postgres/bytea.go new file mode 100644 index 00000000..5385e2dc --- /dev/null +++ b/pkg/wal/listener/postgres/bytea.go @@ -0,0 +1,59 @@ +// SPDX-License-Identifier: Apache-2.0 + +package postgres + +import ( + "encoding/hex" + "strings" + + "github.com/xataio/pgstream/pkg/wal" +) + +const byteaType = "bytea" + +// decodeByteaColumns converts the hex text wal2json emits for bytea columns +// into the raw bytes the rest of the pipeline expects. +// +// wal2json renders a bytea column as bare hex digits ("deadbeef"); the postgres +// hex format ("\xdeadbeef") is accepted too, since that is the form +// pkg/transformers already documents for the replication path. Snapshots are +// unaffected: pgx hands those values over as []byte to begin with, which is +// what makes this the point where the two paths converge. +// +// Without it the hex text travels all the way to the writer and is handed to +// pgx as the parameter value for a bytea column, storing its ASCII characters +// as the column contents — '\xdeadbeef' on the source becomes +// '\x6465616462656566' on the target. Nothing else about the row changes, so +// row counts and keys still match and the corruption only surfaces when +// something parses the column. +// +// Both Columns and Identity are covered, so insert values, the SET clause, the +// WHERE clause and the bulk-delete builders all see decoded bytes. +// +// A value that is not valid hex is left untouched rather than dropped, so an +// unexpected producer format degrades to the previous behaviour instead of +// failing the batch. +func decodeByteaColumns(d *wal.Data) { + if d == nil { + return + } + decodeByteaValues(d.Columns) + decodeByteaValues(d.Identity) +} + +func decodeByteaValues(cols []wal.Column) { + for i := range cols { + if cols[i].Type != byteaType { + continue + } + strVal, ok := cols[i].Value.(string) + if !ok { + continue + } + decoded, err := hex.DecodeString(strings.TrimPrefix(strVal, `\x`)) + if err != nil { + continue + } + cols[i].Value = decoded + } +} diff --git a/pkg/wal/listener/postgres/bytea_test.go b/pkg/wal/listener/postgres/bytea_test.go new file mode 100644 index 00000000..7e0c9b1a --- /dev/null +++ b/pkg/wal/listener/postgres/bytea_test.go @@ -0,0 +1,89 @@ +// SPDX-License-Identifier: Apache-2.0 + +package postgres + +import ( + "testing" + + "github.com/stretchr/testify/require" + "github.com/xataio/pgstream/pkg/wal" +) + +func Test_decodeByteaColumns(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + data *wal.Data + + want *wal.Data + }{ + { + name: "nil data", + data: nil, + want: nil, + }, + { + name: "wal2json bare hex, in columns and identity", + data: &wal.Data{ + Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "deadbeef"}}, + Identity: []wal.Column{{Name: "key", Type: "bytea", Value: "0102"}}, + }, + want: &wal.Data{ + Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0xde, 0xad, 0xbe, 0xef}}}, + Identity: []wal.Column{{Name: "key", Type: "bytea", Value: []byte{0x01, 0x02}}}, + }, + }, + { + name: "postgres hex format", + data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: `\xdeadbeef`}}}, + want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0xde, 0xad, 0xbe, 0xef}}}}, + }, + { + name: "empty value", + data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: ""}}}, + want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{}}}}, + }, + { + // snapshots hand bytea over as []byte already + name: "already decoded", + data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0x01}}}}, + want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0x01}}}}, + }, + { + // degrade to the previous behaviour rather than failing the batch + name: "not valid hex", + data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "not-hex"}}}, + want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "not-hex"}}}, + }, + { + name: "odd length hex", + data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "abc"}}}, + want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "abc"}}}, + }, + { + name: "nil value", + data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: nil}}}, + want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: nil}}}, + }, + { + name: "other column types are untouched", + data: &wal.Data{Columns: []wal.Column{ + {Name: "name", Type: "text", Value: "deadbeef"}, + {Name: "id", Type: "integer", Value: 1}, + }}, + want: &wal.Data{Columns: []wal.Column{ + {Name: "name", Type: "text", Value: "deadbeef"}, + {Name: "id", Type: "integer", Value: 1}, + }}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + decodeByteaColumns(tc.data) + require.Equal(t, tc.want, tc.data) + }) + } +} diff --git a/pkg/wal/listener/postgres/wal_pg_listener.go b/pkg/wal/listener/postgres/wal_pg_listener.go index 950a6c03..461afeeb 100644 --- a/pkg/wal/listener/postgres/wal_pg_listener.go +++ b/pkg/wal/listener/postgres/wal_pg_listener.go @@ -178,6 +178,11 @@ func (l *Listener) processWALEvent(ctx context.Context, msg *replication.Message if err := l.walDataDeserialiser(msg.Data, event.Data); err != nil { return fmt.Errorf("error unmarshaling wal data: %w", err) } + // Here rather than in the writer: this is the only place that knows the + // value came from wal2json, so one pass covers every downstream call + // site, and transformers then see the same []byte the snapshot path + // gives them instead of hex text. + decodeByteaColumns(event.Data) } event.CommitPosition = wal.CommitPosition(l.lsnParser.ToString(msg.LSN)) if isInternalPgstreamDML(event.Data) {