Hi,
Masahiko Sawada <[email protected]> wrote (in August 2024):
> My question is;
> in order to track just catalog-change transactions, whether it's
> sufficient to check if XLOG_XACT_COMMIT[_PREPARED] has the
> XACT_XINFO_HAS_INVALS flag. If yes, we probably should change only
> xact_decode() to check the commit records even in BUILDING_SNAPSHOT.
> Otherwise, we would need to change mostly all paths where we mark the
> transaction as catalog-change as the patch does.
I tried to answer that with data, and it looks like the answer is no.
I added two LOG lines to master (the attached instrumentation diff, not
meant to be applied): SnapBuildCommitTxn() logs, for the xid and each
subxid, whether the reorder buffer has it as catalog-changing and
whether the commit has XACT_XINFO_HAS_INVALS; SnapBuildProcessNewCid()
logs which relation the NEW_CID is for. Then I ran installcheck-parallel
with a logical slot created beforehand, and decoded all of it. The
attached catchk_run.sh does all of this on a build with that diff,
including the user catalog table check below; three runs gave the same
counts.
catalog changes invals in commit top-level xacts
no no 5087
yes no 37
yes yes 15540
The 37 are two kinds. 33 changed only pg_largeobject and
pg_largeobject_metadata. In the other 4 the catalog changes were made
only in subtransactions that were rolled back (TRUNCATE and DROP TABLE
in stats.sql, and one pg_statistic write), so the top transaction was
marked through its child but had nothing left to invalidate. In that
run, every committed change to a catalog with a syscache or relcache
had invalidations.
But the regression tests don't write to user catalog tables, and those
behave like large objects:
CREATE TABLE uc (k int PRIMARY KEY, v text)
WITH (user_catalog_table = true);
INSERT INTO uc ... catalog changes: yes invals: no
UPDATE uc ... catalog changes: yes invals: no
DELETE FROM uc ... catalog changes: yes invals: no
ALTER TABLE plain ... catalog changes: yes invals: yes
Output plugins may read user catalog tables with the historic snapshot,
so a transaction that writes one during BUILDING_SNAPSHOT has to be
tracked, and XACT_XINFO_HAS_INVALS would miss it. So I think the
patch's approach, marking the transaction from the NEW_CID records as
well, is the one needed.
The same data says something about master that I can't turn into a
failure: SnapBuildXidHasCatalogChanges() relies on "The transactions
that have changed catalogs must have invalidation info" to skip the
catchange array lookup, which doesn't hold for user catalog tables or
large objects. Is that a problem when a snapshot is restored, for a
plugin that reads user catalog tables?
On the patch itself, v6 applied on master e8a3ee5b197, against that
master, with --enable-cassert:
- snapshot_build fails on master with "could not map filenumber
"base/16384/16772" to relation OID" and passes with v6.
- The same test with the catalog change in a subtransaction, and with
it committed by COMMIT PREPARED, fails the same way on master and
passes with v6.
- With ALTER TABLE ... ADD COLUMN on an existing table instead of
CREATE TABLE, master raises no error: the insert that follows is
decoded with the old tuple descriptor, and the new columns are
silently dropped:
master: table public.tbl3: INSERT: val1[integer]:1
v6: table public.tbl3: INSERT: val1[integer]:1 val2[text]:'two'
val3[bigint]:3
That seems worse than the error, since nothing tells the user.
The attached diff, on top of v6-0002, adds these three as permutations
of snapshot_build. They fail on master and pass with v6.
About Ajin's question in March 2025 on the DecodeTXNNeedSkip() change
(the "SnapBuildCurrentState(...) < SNAPBUILD_CONSISTENT" test), which
ChangAo said might be redundant: on v6 I logged every call made before
CONSISTENT, and whether SnapBuildXactNeedsSkip() already skipped it.
Over test_decoding, src/test/subscription and the recovery tests below
plus 040, it was called 41 times before CONSISTENT (28 in
BUILDING_SNAPSHOT, 13 in FULL_SNAPSHOT), and SnapBuildXactNeedsSkip()
had already skipped every one of them. So it never changes the result
there. Maybe an Assert would say the same thing and catch a case where
it does.
With v6: test_decoding (20 and 15, plus the new permutations),
src/test/subscription (605 tests), the logical decoding tests in
src/test/recovery (006, 035 and 038, 119 tests) and make check (239)
pass.
v6-0001 applies cleanly to master and REL_18_STABLE; REL_19_STABLE
conflicts in DecodeTXNNeedSkip(), and REL_17_STABLE and older in the
XLOG_XACT_INVALIDATIONS case of xact_decode(). As Haiyang reported in
bug #19109, it reproduces back to 11.
I haven't looked at Ajin's alternative beyond your question about it.
Regards,
Manu
diff --git a/contrib/test_decoding/expected/snapshot_build.out
b/contrib/test_decoding/expected/snapshot_build.out
index 0fcf20cce86..32e5cd1dd1d 100644
--- a/contrib/test_decoding/expected/snapshot_build.out
+++ b/contrib/test_decoding/expected/snapshot_build.out
@@ -31,3 +31,100 @@ COMMIT
stop
(1 row)
+
+starting permutation: s1_begin s1_insert s2_init s3_begin s3_insert
s4_create_subxact 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_subxact: BEGIN; SAVEPOINT a; CREATE TABLE tbl2 (val1 integer);
RELEASE a; COMMIT;
+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)
+
+
+starting permutation: s1_begin s1_insert s2_init s3_begin s3_insert s4_prepare
s4_commit_prepared 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_prepare: BEGIN; CREATE TABLE tbl2 (val1 integer); PREPARE TRANSACTION
'snapshot_build';
+step s4_commit_prepared: COMMIT PREPARED 'snapshot_build';
+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)
+
+
+starting permutation: s1_begin s1_insert s2_init s3_begin s3_insert s4_alter
s1_commit s4_begin s4_insert3 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_alter: ALTER TABLE tbl3 ADD COLUMN val2 text, ADD COLUMN val3 bigint;
+step s1_commit: COMMIT;
+step s4_begin: BEGIN;
+step s4_insert3: INSERT INTO tbl3 VALUES (1, 'two', 3);
+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.tbl3: INSERT: val1[integer]:1 val2[text]:'two' val3[bigint]:3
+COMMIT
+(3 rows)
+
+?column?
+--------
+stop
+(1 row)
+
diff --git a/contrib/test_decoding/specs/snapshot_build.spec
b/contrib/test_decoding/specs/snapshot_build.spec
index 334531dd219..65625e548ed 100644
--- a/contrib/test_decoding/specs/snapshot_build.spec
+++ b/contrib/test_decoding/specs/snapshot_build.spec
@@ -4,13 +4,16 @@ setup
{
DROP TABLE IF EXISTS tbl1;
DROP TABLE IF EXISTS tbl2;
+ DROP TABLE IF EXISTS tbl3;
CREATE TABLE tbl1 (val1 integer);
+ CREATE TABLE tbl3 (val1 integer);
}
teardown
{
DROP TABLE tbl1;
- DROP TABLE tbl2;
+ DROP TABLE IF EXISTS tbl2;
+ DROP TABLE tbl3;
SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot');
}
@@ -34,8 +37,13 @@ step "s3_commit" { COMMIT; }
session "s4"
setup { SET synchronous_commit=on; }
step "s4_create" { CREATE TABLE tbl2 (val1 integer); }
+step "s4_create_subxact" { BEGIN; SAVEPOINT a; CREATE TABLE tbl2 (val1
integer); RELEASE a; COMMIT; }
+step "s4_prepare" { BEGIN; CREATE TABLE tbl2 (val1 integer); PREPARE
TRANSACTION 'snapshot_build'; }
+step "s4_commit_prepared" { COMMIT PREPARED 'snapshot_build'; }
+step "s4_alter" { ALTER TABLE tbl3 ADD COLUMN val2 text, ADD COLUMN val3
bigint; }
step "s4_begin" { BEGIN; }
step "s4_insert" { INSERT INTO tbl2 VALUES (1); }
+step "s4_insert3" { INSERT INTO tbl3 VALUES (1, 'two', 3); }
step "s4_commit" { COMMIT; }
# T1: s1_begin -> s1_insert -> BUILDING_SNAPSHOT -> s1_commit -> FULL_SNAPSHOT
@@ -44,3 +52,13 @@ step "s4_commit" { COMMIT; }
# 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"
+
+# The same, with the catalog change made in a subtransaction.
+permutation "s1_begin" "s1_insert" "s2_init" "s3_begin" "s3_insert"
"s4_create_subxact" "s1_commit" "s4_begin" "s4_insert" "s3_commit" "s4_commit"
"s2_get_changes"
+
+# The same, with the catalog change committed by COMMIT PREPARED.
+permutation "s1_begin" "s1_insert" "s2_init" "s3_begin" "s3_insert"
"s4_prepare" "s4_commit_prepared" "s1_commit" "s4_begin" "s4_insert"
"s3_commit" "s4_commit" "s2_get_changes"
+
+# An ALTER TABLE instead: without tracking T3, the insert is decoded with the
+# old tuple descriptor and the new columns are silently dropped.
+permutation "s1_begin" "s1_insert" "s2_init" "s3_begin" "s3_insert" "s4_alter"
"s1_commit" "s4_begin" "s4_insert3" "s3_commit" "s4_commit" "s2_get_changes"
diff --git a/src/backend/replication/logical/snapbuild.c
b/src/backend/replication/logical/snapbuild.c
index de491ea0c4b..27ba27f4162 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -701,6 +701,10 @@ SnapBuildProcessNewCid(SnapBuild *builder, TransactionId
xid,
*/
ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
+ /* EXPERIMENT: which catalog does this change touch? */
+ elog(LOG, "NEWCID xid=%u top=%u rel=%u",
+ xid, xlrec->top_xid, xlrec->target_locator.relNumber);
+
ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
xlrec->target_locator, xlrec->target_tid,
xlrec->cmin,
xlrec->cmax,
@@ -968,6 +972,17 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn,
TransactionId xid,
return;
}
+ /* EXPERIMENT: does every catalog-changing xact carry invalidations? */
+ elog(LOG, "CATCHK state=%d xid=%u sub=0 rb=%d invals=%d",
+ builder->state, xid,
+ ReorderBufferXidHasCatalogChanges(builder->reorder, xid),
+ (xinfo & XACT_XINFO_HAS_INVALS) != 0);
+ for (int i = 0; i < nsubxacts; i++)
+ elog(LOG, "CATCHK state=%d xid=%u sub=1 rb=%d invals=%d",
+ builder->state, subxacts[i],
+ ReorderBufferXidHasCatalogChanges(builder->reorder,
subxacts[i]),
+ (xinfo & XACT_XINFO_HAS_INVALS) != 0);
+
if (builder->state < SNAPBUILD_CONSISTENT)
{
/* ensure that only commits after this are getting replayed */
#!/bin/bash
# usage: catchk_run.sh <install prefix> <build dir> <port>
#
# Needs a build with nocfbot-catchk-instrumentation.diff applied.
#
# 1. Creates a logical slot, runs "make installcheck-parallel" from the build
# dir against a throwaway cluster, decodes everything, and counts, per
# committed xid, whether the reorder buffer had it as catalog-changing and
# whether its commit had XACT_XINFO_HAS_INVALS.
# 2. For the xacts with catalog changes but no invals, lists the catalogs
# their NEW_CID records were for, and whether the xids that touched them
# committed (they appear as committed subxids) or were rolled back.
# 3. The same check for INSERT/UPDATE/DELETE on a user catalog table.
P=$1; B=$2; PORT=$3
D=/tmp/catchk_$PORT
rm -rf $D
$P/bin/initdb -D $D -U postgres --no-sync >/dev/null 2>&1
cat >> $D/postgresql.conf <<EOF
wal_level = logical
max_replication_slots = 4
max_wal_size = 8GB
EOF
$P/bin/pg_ctl -D $D -o "-p $PORT -k /tmp" -l $D/log -w start >/dev/null
trap '$P/bin/pg_ctl -D $D -m immediate -w stop >/dev/null 2>&1; rm -rf $D' EXIT
PSQL="$P/bin/psql -h /tmp -p $PORT -U postgres -X -q -At postgres"
$PSQL -c "SELECT 'slot' FROM pg_create_logical_replication_slot('catchk',
'test_decoding')" >/dev/null
make -C $B/src/test/regress installcheck-parallel PGHOST=/tmp PGPORT=$PORT
PGUSER=postgres \
> $D/installcheck.log 2>&1
grep -E '# All|failed' $D/installcheck.log
$PSQL -c "SELECT count(*) FROM pg_logical_slot_get_changes('catchk', NULL,
NULL)" >/dev/null
echo "== committed top-level xacts: catalog changes (rb) vs invals in commit"
grep -oE 'CATCHK state=2 xid=[0-9]+ sub=0 rb=[01] invals=[01]' $D/log |
awk '{print $5, $6}' | sort | uniq -c
echo "== the ones with rb=1 invals=0, by catalog touched"
$PSQL -c "SELECT pg_relation_filenode(oid) || '|' || relname FROM pg_class
WHERE relnamespace = 'pg_catalog'::regnamespace" > $D/filenodes
perl -e '
open F, "$ARGV[0]"; while (<F>) { chomp; my ($n, $r) = split /\|/; $fn{$n} =
$r }
open L, "$ARGV[1]";
while (<L>) {
$bad{$1} = 1 if /CATCHK state=2 xid=(\d+) sub=0 rb=1 invals=0/;
$committed{$1} = 1 if /CATCHK state=2 xid=(\d+) sub=1 /;
push @{ $nc{$2} }, [$1, $3] if /NEWCID xid=(\d+) top=(\d+) rel=(\d+)/;
}
for my $x (keys %bad) {
my (%cat, $aborted);
for my $e (@{ $nc{$x} || [] }) {
$cat{ $fn{$e->[1]} // "relfilenode $e->[1]" } = 1;
$aborted = 1 if $e->[0] != $x && !$committed{ $e->[0] };
}
my $k = join(", ", sort keys %cat) . ($aborted ? " (in rolled back
subxacts)" : "");
$g{$k}++;
}
printf "%5d %s\n", $g{$_}, $_ for sort { $g{$b} <=> $g{$a} } keys %g;
' $D/filenodes $D/log
echo "== user catalog table"
$PSQL -c "CREATE TABLE uc (k int PRIMARY KEY, v text) WITH (user_catalog_table
= true)" \
-c "SELECT 'slot' FROM pg_create_logical_replication_slot('uc',
'test_decoding')" >/dev/null
for q in "INSERT INTO uc VALUES (1, 'a')" "UPDATE uc SET v = 'b' WHERE k = 1" \
"DELETE FROM uc WHERE k = 1" "ALTER TABLE uc ADD COLUMN w int"; do
x=$($PSQL -c "BEGIN" -c "$q" -c "SELECT txid_current()" -c "COMMIT" | tail -1)
echo "$x|$q"
done > $D/uc
$PSQL -c "SELECT count(*) FROM pg_logical_slot_get_changes('uc', NULL, NULL)"
>/dev/null
while IFS='|' read -r x q; do
printf " %-36s %s\n" "$q" \
"$(grep -oE "CATCHK state=2 xid=$x sub=0 rb=[01] invals=[01]" $D/log | tail
-1 | grep -oE 'rb=[01] invals=[01]')"
done < $D/uc
