On 2025-Nov-21, cca5507 wrote: > Attach the rebased patches.
I rebased this again, for CFbot. I haven't reviewed it. > Discussion: > https://www.postgresql.org/message-id/flat/[email protected] > https://www.postgresql.org/message-id/flat/[email protected] > https://www.postgresql.org/message-id/flat/[email protected] > > Bug Report: > https://www.postgresql.org/message-id/flat/18509-983f064d174ea880%40postgresql.org > https://www.postgresql.org/message-id/flat/19109-4a9ce36715d69632%40postgresql.org > > CF Entry: > https://commitfest.postgresql.org/patch/5029/ -- Álvaro Herrera PostgreSQL Developer — https://www.EnterpriseDB.com/ "Selbst das größte Genie würde nicht weit kommen, wenn es alles seinem eigenen Innern verdanken wollte." (Johann Wolfgang von Goethe) Ni aún el genio más grande llegaría muy lejos si quisiera sacarlo todo de su propio interior.
>From 3c235e5d65daef0d0a82354ee302960e769e62be Mon Sep 17 00:00:00 2001 From: ChangAo Chen <[email protected]> Date: Fri, 21 Nov 2025 15:19:22 +0800 Subject: [PATCH v7 1/2] Track transactions committed in BUILDING_SNAPSHOT. The historic snapshot previously didn't track transactions committed in BUILDING_SNAPSHOT, this might result in a transaction taking an incorrect snapshot and logical decoding being interrupted. So we need to track these transactions. We also need to handle the xlog which means a catalog change in BUILDING_SNAPSHOT because the historic snapshot only tracks catalog modifying transactions. --- src/backend/replication/logical/decode.c | 33 ++++++++++++++++++++---- 1 file changed, 28 insertions(+), 5 deletions(-) diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index f44f94d9f07..8d0c2b9cba5 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -218,12 +218,16 @@ xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) uint8 info = XLogRecGetInfo(r) & XLOG_XACT_OPMASK; /* - * If the snapshot isn't yet fully built, we cannot decode anything, so - * bail out. + * If the snapshot hasn't started building yet, the transaction won't be + * decoded or tracked by the snapshot, so bail out. */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) + if (SnapBuildCurrentState(builder) < SNAPBUILD_BUILDING_SNAPSHOT) return; + /* + * Note that if the snapshot isn't yet fully built, the xlog is only used + * to build the snapshot and won't be decoded. + */ switch (info) { case XLOG_XACT_COMMIT: @@ -294,18 +298,24 @@ xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) { TransactionId xid; xl_xact_invals *invals; + bool has_snapshot; xid = XLogRecGetXid(r); invals = (xl_xact_invals *) XLogRecGetData(r); + has_snapshot = + SnapBuildCurrentState(builder) >= SNAPBUILD_FULL_SNAPSHOT; /* * Execute the invalidations for xid-less transactions, * otherwise, accumulate them so that they can be processed at * the commit time. + * + * Note that we only need to do this when we are not fast-forwarding + * and there is a snapshot. */ if (TransactionIdIsValid(xid)) { - if (!ctx->fast_forward) + if (!ctx->fast_forward && has_snapshot) ReorderBufferAddInvalidations(reorder, xid, buf->origptr, invals->nmsgs, @@ -313,7 +323,7 @@ xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr); } - else if (!ctx->fast_forward) + else if (!ctx->fast_forward && has_snapshot) ReorderBufferImmediateInvalidation(ctx->reorder, invals->nmsgs, invals->msgs); @@ -436,7 +446,19 @@ heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) * SnapBuildProcessRunningXacts(). */ if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) + { + /* + * If we are building snapshot and the xlog means a catalog + * change, we need to mark it in the reorder buffer. + * + * Now only XLOG_HEAP2_NEW_CID means a catalog change. + */ + if (SnapBuildCurrentState(builder) >= SNAPBUILD_BUILDING_SNAPSHOT && + TransactionIdIsValid(xid) && info == XLOG_HEAP2_NEW_CID) + ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr); + return; + } switch (info) { @@ -1343,6 +1365,7 @@ DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid txn_dbid, ReplOriginId origin_id) { if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) || + SnapBuildCurrentState(ctx->snapshot_builder) < SNAPBUILD_CONSISTENT || (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) || FilterByOrigin(ctx, origin_id)) return true; -- 2.47.3
>From a6c49e68ea4864a724f107496ad15914517c816f Mon Sep 17 00:00:00 2001 From: ChangAo Chen <[email protected]> Date: Fri, 21 Nov 2025 15:48:27 +0800 Subject: [PATCH v7 2/2] Add test case snapshot_build for test_decoding. --- contrib/test_decoding/Makefile | 2 +- .../test_decoding/expected/snapshot_build.out | 33 +++++++++++++ contrib/test_decoding/meson.build | 1 + .../test_decoding/specs/snapshot_build.spec | 46 +++++++++++++++++++ 4 files changed, 81 insertions(+), 1 deletion(-) create mode 100644 contrib/test_decoding/expected/snapshot_build.out create mode 100644 contrib/test_decoding/specs/snapshot_build.spec diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile index a33e6295d8c..540d46418ff 100644 --- a/contrib/test_decoding/Makefile +++ b/contrib/test_decoding/Makefile @@ -16,7 +16,7 @@ ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \ oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \ twophase_snapshot $(SLOT_CREATION_TST) catalog_change_snapshot \ skip_snapshot_restore invalidation_distribution parallel_session_origin \ - tuplecid_restart tuplecid_nested + snapshot_build tuplecid_restart tuplecid_nested REGRESS_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf diff --git a/contrib/test_decoding/expected/snapshot_build.out b/contrib/test_decoding/expected/snapshot_build.out new file mode 100644 index 00000000000..0fcf20cce86 --- /dev/null +++ b/contrib/test_decoding/expected/snapshot_build.out @@ -0,0 +1,33 @@ +Parsed test spec with 4 sessions + +starting permutation: s1_begin s1_insert s2_init s3_begin s3_insert s4_create s1_commit s4_begin s4_insert s3_commit s4_commit s2_get_changes +step s1_begin: BEGIN; +step s1_insert: INSERT INTO tbl1 VALUES (1); +step s2_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); <waiting ...> +step s3_begin: BEGIN; +step s3_insert: INSERT INTO tbl1 VALUES (1); +step s4_create: CREATE TABLE tbl2 (val1 integer); +step s1_commit: COMMIT; +step s4_begin: BEGIN; +step s4_insert: INSERT INTO tbl2 VALUES (1); +step s3_commit: COMMIT; +step s2_init: <... completed> +?column? +-------- +init +(1 row) + +step s4_commit: COMMIT; +step s2_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); +data +------------------------------------------ +BEGIN +table public.tbl2: INSERT: val1[integer]:1 +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/meson.build b/contrib/test_decoding/meson.build index 667ea76518f..dc69dab0a35 100644 --- a/contrib/test_decoding/meson.build +++ b/contrib/test_decoding/meson.build @@ -30,6 +30,7 @@ test_decoding_isolation = [ 'skip_snapshot_restore', 'invalidation_distribution', 'parallel_session_origin', + 'snapshot_build', 'tuplecid_nested', 'tuplecid_restart', ] diff --git a/contrib/test_decoding/specs/snapshot_build.spec b/contrib/test_decoding/specs/snapshot_build.spec new file mode 100644 index 00000000000..334531dd219 --- /dev/null +++ b/contrib/test_decoding/specs/snapshot_build.spec @@ -0,0 +1,46 @@ +# Test snapshot build correctly, it must track committed transactions during BUILDING_SNAPSHOT + +setup +{ + DROP TABLE IF EXISTS tbl1; + DROP TABLE IF EXISTS tbl2; + CREATE TABLE tbl1 (val1 integer); +} + +teardown +{ + DROP TABLE tbl1; + DROP TABLE tbl2; + SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot'); +} + +session "s1" +setup { SET synchronous_commit=on; } +step "s1_begin" { BEGIN; } +step "s1_insert" { INSERT INTO tbl1 VALUES (1); } +step "s1_commit" { COMMIT; } + +session "s2" +setup { SET synchronous_commit=on; } +step "s2_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); } +step "s2_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); } + +session "s3" +setup { SET synchronous_commit=on; } +step "s3_begin" { BEGIN; } +step "s3_insert" { INSERT INTO tbl1 VALUES (1); } +step "s3_commit" { COMMIT; } + +session "s4" +setup { SET synchronous_commit=on; } +step "s4_create" { CREATE TABLE tbl2 (val1 integer); } +step "s4_begin" { BEGIN; } +step "s4_insert" { INSERT INTO tbl2 VALUES (1); } +step "s4_commit" { COMMIT; } + +# T1: s1_begin -> s1_insert -> BUILDING_SNAPSHOT -> s1_commit -> FULL_SNAPSHOT +# T2: BUILDING_SNAPSHOT -> s3_begin -> s3_insert -> FULL_SNAPSHOT -> s3_commit -> CONSISTENT +# T3: BUILDING_SNAPSHOT -> s4_create -> FULL_SNAPSHOT +# T4: FULL_SNAPSHOT -> s4_begin -> s4_insert -> CONSISTENT -> s4_commit +# The snapshot must track T3 or the replay of T4 will fail because its snapshot cannot see tbl2 +permutation "s1_begin" "s1_insert" "s2_init" "s3_begin" "s3_insert" "s4_create" "s1_commit" "s4_begin" "s4_insert" "s3_commit" "s4_commit" "s2_get_changes" -- 2.47.3
