public inbox for [email protected]
help / color / mirror / Atom feedFrom: =?ISO-8859-1?B?Y2NhNTUwNw==?= <[email protected]>
To: =?ISO-8859-1?B?QmVydHJhbmQgRHJvdXZvdA==?= <[email protected]>
Cc: =?ISO-8859-1?B?TWljaGFlbCBQYXF1aWVy?= <[email protected]>
Cc: =?ISO-8859-1?B?cGdzcWwtaGFja2Vycw==?= <[email protected]>
Subject: Re: Historic snapshot doesn't track txns committed in BUILDING_SNAPSHOT state
Date: Tue, 13 Aug 2024 12:23:04 +0800
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<[email protected]>
Hi,
I refactor the code and fix the git apply warning according to [1].
Here are the new version patches.
--
Regards,
ChangAo Chen
[1] https://www.postgresql.org/message-id/Zrmh7X8jYCbFYXjH%40ip-10-97-1-34.eu-west-3.compute.internal
Attachments:
[application/octet-stream] v3-0002-Add-test-case-snapshot_build-for-test_decoding.patch (5.0K, ../[email protected]/3-v3-0002-Add-test-case-snapshot_build-for-test_decoding.patch)
download | inline diff:
From 51838f4cba785231ad6bd6d0f1bc105b9778392f Mon Sep 17 00:00:00 2001
From: ChangAo Chen <[email protected]>
Date: Sat, 10 Aug 2024 17:34:26 +0800
Subject: [PATCH v3 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 a4ba1a509a..8113e2d99c 100644
--- a/contrib/test_decoding/Makefile
+++ b/contrib/test_decoding/Makefile
@@ -9,7 +9,7 @@ REGRESS = ddl xact rewrite toast permissions decoding_in_xact \
ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \
oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \
twophase_snapshot slot_creation_error catalog_change_snapshot \
- skip_snapshot_restore
+ skip_snapshot_restore snapshot_build
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 0000000000..0fcf20cce8
--- /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 f643dc81a2..2b7e80ba71 100644
--- a/contrib/test_decoding/meson.build
+++ b/contrib/test_decoding/meson.build
@@ -63,6 +63,7 @@ tests += {
'twophase_snapshot',
'slot_creation_error',
'skip_snapshot_restore',
+ 'snapshot_build',
],
'regress_args': [
'--temp-config', files('logical.conf'),
diff --git a/contrib/test_decoding/specs/snapshot_build.spec b/contrib/test_decoding/specs/snapshot_build.spec
new file mode 100644
index 0000000000..334531dd21
--- /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.34.1
[application/octet-stream] v4-0001-Track-transactions-committed-in-BUILDING_SNAPSHOT.patch (5.4K, ../[email protected]/4-v4-0001-Track-transactions-committed-in-BUILDING_SNAPSHOT.patch)
download | inline diff:
From b35711f5ef0941f79127acbec55a8ede5e27a5eb Mon Sep 17 00:00:00 2001
From: ChangAo Chen <[email protected]>
Date: Tue, 13 Aug 2024 11:45:07 +0800
Subject: [PATCH v4] Track transactions committed in BUILDING_SNAPSHOT.
The historic snapshot previously didn't track transactions committed
in BUILDING_SNAPSHOT, and this might result in a transaction taking
an incorrect snapshot and logical decoding being interrupted. So we
need track these transactions.
We also need handle the xlog that marks a transaction as containing
catalog changes in BUILDING_SNAPSHOT because the historic snapshot
only tracks catalog modifying transactions.
---
src/backend/replication/logical/decode.c | 45 +++++++++++++++++++++---
1 file changed, 40 insertions(+), 5 deletions(-)
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index d687ceee33..6df558ad18 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -206,12 +206,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:
@@ -282,9 +286,11 @@ 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,
@@ -293,7 +299,7 @@ xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
*/
if (TransactionIdIsValid(xid))
{
- if (!ctx->fast_forward)
+ if (!ctx->fast_forward && has_snapshot)
ReorderBufferAddInvalidations(reorder, xid,
buf->origptr,
invals->nmsgs,
@@ -301,7 +307,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);
@@ -407,6 +413,8 @@ heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
TransactionId xid = XLogRecGetXid(buf->record);
SnapBuild *builder = ctx->snapshot_builder;
+ /* True if the xlog marks the transaction as containing catalog changes */
+ bool set_catalog_changes = (info == XLOG_HEAP2_NEW_CID);
ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
@@ -416,7 +424,19 @@ heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
*/
if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
ctx->fast_forward)
+ {
+ /*
+ * If the transaction contains catalog changes, we need mark it in
+ * reorder buffer before return as the snapshot only tracks catalog
+ * modifying transactions. The transaction before BUILDING_SNAPSHOT
+ * won't be tracked anyway(see SnapBuildCommitTxn), so skip it.
+ */
+ if (set_catalog_changes && TransactionIdIsValid(xid) &&
+ SnapBuildCurrentState(builder) >= SNAPBUILD_BUILDING_SNAPSHOT)
+ ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);
+
return;
+ }
switch (info)
{
@@ -466,6 +486,8 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
uint8 info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
TransactionId xid = XLogRecGetXid(buf->record);
SnapBuild *builder = ctx->snapshot_builder;
+ /* True if the xlog marks the transaction as containing catalog changes */
+ bool set_catalog_changes = (info == XLOG_HEAP_INPLACE);
ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
@@ -475,7 +497,19 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
*/
if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
ctx->fast_forward)
+ {
+ /*
+ * If the transaction contains catalog changes, we need mark it in
+ * reorder buffer before return as the snapshot only tracks catalog
+ * modifying transactions. The transaction before BUILDING_SNAPSHOT
+ * won't be tracked anyway(see SnapBuildCommitTxn), so skip it.
+ */
+ if (set_catalog_changes && TransactionIdIsValid(xid) &&
+ SnapBuildCurrentState(builder) >= SNAPBUILD_BUILDING_SNAPSHOT)
+ ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);
+
return;
+ }
switch (info)
{
@@ -1301,6 +1335,7 @@ DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
Oid txn_dbid, RepOriginId 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.34.1
view thread (10+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected]
Subject: Re: Historic snapshot doesn't track txns committed in BUILDING_SNAPSHOT state
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox