Skip to content

Commit 26bde72

Browse files
PG to PG: Add ORDER BY in MERGE to handle non-PK unique columns (#4169)
## Problem statement Consider a Postgres table such with `id` PRIMARY KEY and `email` TEXT with a UNIQUE constraint on it. Let's say there is a mirror of this table to another PG database, and the following changes are made on source: ```sql DELETE FROM table WHERE id = 1 (where email = 'alice@example.com') INSERT INTO table (id, email) VALUES (2, 'alice@example.com') ``` This can result in a unique constraint violation error in normalize: ``` error executing normalize statement for table public.accounts: ERROR: duplicate key value violates unique constraint ``` This is because MERGE in Postgres processes rows in non-deterministic fashion, therefore across PKeys we could attempt replicating the INSERT first before DELETE (note: for rows pertaining to the same PKey this is not possible as we rank there), resulting in us hitting a non-PK unique constraint. ## Fix According to [Postgres docs for MERGE](https://www.postgresql.org/docs/current/sql-merge.html), it says: ``` The order in which rows are generated from the data source is indeterminate by default. A source_query can be used to specify a consistent ordering, if required, which might be needed to avoid deadlocks between concurrent transactions. ``` We are already using the `source_query` syntax in our MERGE. This PR adds an ORDER BY to our subquery which sorts by `_peerdb_timestamp` which is representative of WAL order, thereby not running into this issue. - [x] Functionally tested
1 parent b0735da commit 26bde72

1 file changed

Lines changed: 2 additions & 2 deletions

File tree

flow/connectors/postgres/client.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,12 +55,12 @@ const (
5555
ARRAY_AGG(DISTINCT _peerdb_unchanged_toast_columns) FROM %s.%s WHERE
5656
_peerdb_batch_id=$1 AND _peerdb_record_type!=2 GROUP BY _peerdb_destination_table_name`
5757
mergeStatementSQL = `WITH src_rank AS (
58-
SELECT _peerdb_data,_peerdb_record_type,_peerdb_unchanged_toast_columns,
58+
SELECT _peerdb_data,_peerdb_record_type,_peerdb_unchanged_toast_columns,_peerdb_timestamp,
5959
RANK() OVER (PARTITION BY %s ORDER BY _peerdb_timestamp DESC) AS _peerdb_rank
6060
FROM %s.%s WHERE _peerdb_batch_id = $1 AND _peerdb_destination_table_name=$2
6161
)
6262
MERGE INTO %s dst
63-
USING (SELECT %s,_peerdb_record_type,_peerdb_unchanged_toast_columns FROM src_rank WHERE _peerdb_rank=1) src
63+
USING (SELECT %s,_peerdb_record_type,_peerdb_unchanged_toast_columns FROM src_rank WHERE _peerdb_rank=1 ORDER BY _peerdb_timestamp) src
6464
ON %s
6565
WHEN NOT MATCHED AND src._peerdb_record_type!=2 THEN
6666
INSERT (%s) VALUES (%s) %s

0 commit comments

Comments
 (0)