Thread: pg_logical_slot_peek_changes with slots created with pglogical_output missed DML records,

Hi,

  We test pglogical replication,  with pglogical.conflict_resolution=error, and from source database run

 Insert into pgbench_branches values(109,0,’test’);

Commit;

At target database, since already exist bid=109, so it failed there and from the error log:

2023-01-10 13:06:33.161 UTC::@:[1700321]:[30-1]::2023-01-10 11:22:38 UTCLOG:  background worker "pglogical apply 16389:2210095472" (PID 1716767) exited with exit code 1

2023-01-10 13:06:38.168 UTC::[unknown]@pgddweb:[1723585]:[1-1]:pglogical apply 16389:2210095472:2023-01-10 13:06:38 UTCLOG:  starting apply for subscription topgddweb

2023-01-10 13:06:38.606 UTC::[unknown]@pgddweb:[1723585]:[2-1]:pglogical apply 16389:2210095472:2023-01-10 13:06:38 UTCERROR:  cannot apply conflicting row

2023-01-10 13:06:38.606 UTC::[unknown]@pgddweb:[1723585]:[3-1]:pglogical apply 16389:2210095472:2023-01-10 13:06:38 UTCCONTEXT:  apply INSERT from remote relation pgbench.pgbench_branches in commit before 0/FA002200, xid 1147 committed at 2023-01-10 13:06:33.156494+00 (action #2) from node replorigin 1

Go back to source database, and check details about pg wal records,

SELECT *

FROM pg_logical_slot_peek_changes('pgl_pgddweb_pgtest1_topgddweb', NULL, 100,

         'min_proto_version', '1', 'max_proto_version', '1',

         'startup_params_format', '1', 'proto_format', 'json');

    lsn     | xid  |

0/FA002110 | 1147 | {"action":"S", "params": {"max_proto_version":"1","min_proto_version":"1","coltypes":"f","pg_version_num":"130008","pg_version":"13.8","pg_catversion":"202007201","da

tabase_encoding":"LATIN1","encoding":"SQL_ASCII","forward_changeset_origins":"t","walsender_pid":"1685609","pglogical_version":"2.4.2","pglogical_version_num":"20402","binary.internal_bas

etypes":"f","binary.binary_basetypes":"f","binary.basetypes_major_version":"1300","binary.sizeof_int":"4","binary.sizeof_long":"8","binary.sizeof_datum":"8","binary.maxalign":"8","binary.

bigendian":"f","binary.float4_byval":"f","binary.float8_byval":"t","binary.integer_datetimes":"f","binary.binary_pg_version":"1300","no_txinfo":"f"}}

0/FA002110 | 1147 | {"action":"B", "has_catalog_changes":"f", "xid":"1147", "first_lsn":"0/FA002110", "commit_time":"2023-01-10 13:06:33.156494+00"}

0/FA002230 | 1147 | {"action":"C", "final_lsn":"0/FA002200", "end_lsn":"0/FA002230"}

We only get ‘B’ and ‘C’ but no ‘I’ fo the failed insert statement in transaction 1147.

We create another test slot use test_decoding plugin, it works, it will print all sql statements records together ‘B’, ‘C’.

select * from pg_replication_slots;

           slot_name           |      plugin      | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status

| safe_wal_size

-------------------------------+------------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------

+---------------

pgl_pgddweb_pgtest1_topgddweb | pglogical_output | logical   |  25634 | pgtest1  | f         | f      |            |      |         1147 | 0/FA0020D8  | 0/FA002110          | reserved

|

test                          | test_decoding    | logical   |  25634 | pgtest1  | f         | f      |            |      |         1147 | 0/FA0027C0  | 0/FA0027F8          | reserved

|

Could you help check why pglogical_output does not ?

Thanks,

James