Hi,
We have written a decoder plugin to use streaming replication in an aggregation framework and we write the changes in chunks using OutputPluginWriter from pg_decode_change. In order to ensure we don't process a message twice we add the lsn of each message to our aggregated value and check as new messages come in that they are not less than the present lsn saved. The problem we have found is that interleaved inserts in 2 separate session have interleaved lsn values, an example of this can be reproduced using the test_decoder:
# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
lsn | xid | data
------------+--------+----------------------------------------------------------
0/3AEB0DC0 | 182312 | BEGIN 182312
0/3AEB0DC0 | 182312 | table public.data: INSERT: id[integer]:11 data[text]:'1'
0/3AEB0F20 | 182312 | COMMIT 182312
0/3AEA4770 | 182265 | BEGIN 182265
0/3AEA47D8 | 182265 | table public.data: INSERT: id[integer]:10 data[text]:'1'
0/3AEB1790 | 182265 | COMMIT 182265
(6 rows)
The lsn of the commit is correctly ordered but the inserts/update/BEGIN seems to be in the order that they happened, this means that if we save value '0/3AEB0DC0’, when we reach '0/3AEA47D8’ we think we have already processed it as it has a value less then the last one processed. I have found that if I change to a batched approach, writing to OutputPluginWriter in the pg_decode_commit_txn, then this will work as the java client used for the aggregation returns the value of the commit lsn as the getLastReceiveLSN, or I can just put the commit_lsn value into our streamed results.
If we would like to use a none-batched approach, what are the suggestions for how to ensure we only process a given record once, whist only saving one value against our aggregation value ie something like an lsn value.
Thanks.
