diff --git a/contrib/test_decoding/expected/catalog_change_snapshot.out b/contrib/test_decoding/expected/catalog_change_snapshot.out index 1d75cf5af0..4c76f98d3b 100644 --- a/contrib/test_decoding/expected/catalog_change_snapshot.out +++ b/contrib/test_decoding/expected/catalog_change_snapshot.out @@ -87,3 +87,46 @@ COMMIT stop (1 row) + +starting permutation: s0_create_temp s0_init s0_begin s0_txid_current s1_checkpoint s0_savepoint s0_insert_temp s0_savepoint_release s0_savepoint s0_insert_temp s0_savepoint_release s0_savepoint s0_insert_temp s0_savepoint_release s0_create_part1 s1_get_changes s0_commit s1_get_changes +step s0_create_temp: CREATE TEMP TABLE temp_t (id int, data TEXT); +step s0_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); +?column? +-------- +init +(1 row) + +step s0_begin: BEGIN; +step s0_txid_current: SELECT 'xid' FROM txid_current(); +?column? +-------- +xid +(1 row) + +step s1_checkpoint: CHECKPOINT; +step s0_savepoint: SAVEPOINT sp1; +step s0_insert_temp: INSERT INTO temp_t VALUES(1); +step s0_savepoint_release: RELEASE SAVEPOINT sp1; +step s0_savepoint: SAVEPOINT sp1; +step s0_insert_temp: INSERT INTO temp_t VALUES(1); +step s0_savepoint_release: RELEASE SAVEPOINT sp1; +step s0_savepoint: SAVEPOINT sp1; +step s0_insert_temp: INSERT INTO temp_t VALUES(1); +step s0_savepoint_release: RELEASE SAVEPOINT sp1; +step s0_create_part1: CREATE TABLE tbl1_part_p1 PARTITION OF tbl1_part FOR VALUES FROM (0) TO (10); +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); +data +---- +(0 rows) + +step s0_commit: COMMIT; +step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); +data +---- +(0 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/specs/catalog_change_snapshot.spec b/contrib/test_decoding/specs/catalog_change_snapshot.spec index 2ad1edeaa8..09602cfc73 100644 --- a/contrib/test_decoding/specs/catalog_change_snapshot.spec +++ b/contrib/test_decoding/specs/catalog_change_snapshot.spec @@ -4,24 +4,32 @@ setup { DROP TABLE IF EXISTS tbl1; CREATE TABLE tbl1 (val1 integer, val2 integer); + CREATE TABLE tbl1_part (val1 integer) PARTITION BY RANGE (val1); CREATE TABLE user_cat (val1 integer) WITH (user_catalog_table = true); } teardown { DROP TABLE tbl1; + DROP TABLE tbl1_part; DROP TABLE user_cat; SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot'); } session "s0" setup { SET synchronous_commit=on; } +step "s0_create_temp" { CREATE TEMP TABLE temp_t (id int, data TEXT); } step "s0_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); } step "s0_begin" { BEGIN; } +step "s0_txid_current" { SELECT 'xid' FROM txid_current(); } step "s0_savepoint" { SAVEPOINT sp1; } +step "s0_savepoint_release" { RELEASE SAVEPOINT sp1; } step "s0_truncate" { TRUNCATE tbl1; } step "s0_insert" { INSERT INTO tbl1 VALUES (1); } step "s0_insert2" { INSERT INTO user_cat VALUES (1); } +step "s0_insert_temp" { INSERT INTO temp_t VALUES(1); } +step "s0_create_part1" { CREATE TABLE tbl1_part_p1 PARTITION OF tbl1_part FOR VALUES FROM (0) TO (10); } + step "s0_commit" { COMMIT; } session "s1" @@ -53,3 +61,11 @@ permutation "s0_init" "s0_begin" "s0_savepoint" "s0_truncate" "s1_checkpoint" "s # transaction to do timetravel since one of its subtransactions has been marked as # containing catalog changes. permutation "s0_init" "s0_begin" "s0_savepoint" "s0_insert" "s1_checkpoint" "s1_get_changes" "s0_insert2" "s0_commit" "s0_begin" "s0_insert" "s1_checkpoint" "s1_get_changes" "s0_commit" "s1_get_changes" + +# DML operations on temporary table temp_t consume XIDs, but do not generate WAL records. +# The last decoding restarts from the first checkpoint and doesn't get any information of +# last two substransactions that performed s0_insert_temp. While processing the commit record +# for the corresponding top-level transaction which will be marked as containing catalog +# change even before commit, we ensure that the corresponding substransaction is added into +# ReorderBuffer as subxact. +permutation "s0_create_temp" "s0_init" "s0_begin" "s0_txid_current" "s1_checkpoint" "s0_savepoint" "s0_insert_temp" "s0_savepoint_release" "s0_savepoint" "s0_insert_temp" "s0_savepoint_release" "s0_savepoint" "s0_insert_temp" "s0_savepoint_release" "s0_create_part1" "s1_get_changes" "s0_commit" "s1_get_changes" diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index d8345a75b6..22dc4787ab 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -829,7 +829,8 @@ ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, * list of top-level txns. Now that we know it's not top-level, * remove it from there. */ - dlist_delete(&subtxn->node); + if (subtxn->node.next != NULL) + dlist_delete(&subtxn->node); } } @@ -1293,7 +1294,8 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) * count too high, but we don't care. Otherwise, we are deleting the TXN * from the LSN-ordered list of toplevel TXNs. */ - dlist_delete(&txn->node); + if (txn->node.next != NULL) + dlist_delete(&txn->node); /* now remove reference from buffer */ hash_search(rb->by_txn, @@ -2211,6 +2213,17 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, txn->has_catalog_changes = true; } +void +ReorderBufferXidSetCatalogChangesEx(ReorderBuffer *rb, TransactionId xid, + XLogRecPtr lsn, bool is_top) +{ + ReorderBufferTXN *txn; + + txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, is_top); + + txn->has_catalog_changes = true; +} + /* * Query whether a transaction is already *known* to contain catalog * changes. This can be wrong until directly before the commit! diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index bdfd1f0228..5140aa8f74 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -2170,6 +2170,7 @@ SnapBuildXidSetCatalogChanges(SnapBuild *builder, TransactionId xid, int subxcnt sizeof(TransactionId), xidComparator) != NULL) { for (int i = 0; i < subxcnt; i++) - ReorderBufferXidSetCatalogChanges(builder->reorder, subxacts[i], lsn); + ReorderBufferXidSetCatalogChangesEx(builder->reorder, subxacts[i], + lsn, false); } } diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index bc97b08a90..69581fcdb3 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -429,6 +429,8 @@ void ReorderBufferImmediateInvalidation(ReorderBuffer *, uint32 ninvalidations, SharedInvalidationMessage *invalidations); void ReorderBufferProcessXid(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn); void ReorderBufferXidSetCatalogChanges(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn); +void ReorderBufferXidSetCatalogChangesEx(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn, bool is_top); + bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *, TransactionId xid); bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *, TransactionId xid);