v11-0005-Add-a-tap-test-to-verify-the-new-slot-xmin-mecha.patch

From 12505e5dcd047decde15739edb2b5c875a1214c6 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Sun, 1 Sep 2024 17:50:31 +0800
Subject: [PATCH v11 5/5] Add a tap test to verify the new slot xmin mechanism

---
 src/test/subscription/meson.build             |   1 +
 .../t/034_confl_update_deleted.pl             | 172 ++++++++++++++++++
 2 files changed, 173 insertions(+)
 create mode 100644 src/test/subscription/t/034_confl_update_deleted.pl

diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d61..6303322d41 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
       't/031_column_list.pl',
       't/032_subscribe_use_index.pl',
       't/033_run_as_table_owner.pl',
+      't/034_confl_update_deleted.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/034_confl_update_deleted.pl b/src/test/subscription/t/034_confl_update_deleted.pl
new file mode 100644
index 0000000000..8873f758d5
--- /dev/null
+++ b/src/test/subscription/t/034_confl_update_deleted.pl
@@ -0,0 +1,172 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test the CREATE SUBSCRIPTION 'detect_update_deleted' parameter and its
+# interaction with the xmin value of replication slots.
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $subname_AB = 'tap_sub_a_b';
+my $subname_BA = 'tap_sub_b_a';
+
+###############################################################################
+# Setup a bidirectional logical replication between node_A & node_B
+###############################################################################
+
+# Initialize nodes. Enable the track_commit_timestamp on both nodes to detect
+# the conflict when attempting to update a row that was previously modified by
+# a different origin.
+
+# node_A
+my $node_A = PostgreSQL::Test::Cluster->new('node_A');
+$node_A->init(allows_streaming => 'logical');
+$node_A->append_conf(
+	'postgresql.conf',
+	qq{track_commit_timestamp = on
+    autovacuum = off});
+$node_A->start;
+
+# node_B
+my $node_B = PostgreSQL::Test::Cluster->new('node_B');
+$node_B->init(allows_streaming => 'logical');
+$node_B->append_conf(
+	'postgresql.conf',
+	qq{track_commit_timestamp = on
+    autovacuum = off});
+$node_B->start;
+
+# Create table on node_A
+$node_A->safe_psql('postgres', "CREATE TABLE tab (a int PRIMARY KEY, b int)");
+
+# Create the same table on node_B
+$node_B->safe_psql('postgres', "CREATE TABLE tab (a int PRIMARY KEY, b int)");
+
+# Setup logical replication
+# node_A (pub) -> node_B (sub)
+my $node_A_connstr = $node_A->connstr . ' dbname=postgres';
+$node_A->safe_psql('postgres', "CREATE PUBLICATION tap_pub_A FOR TABLE tab");
+$node_B->safe_psql(
+	'postgres', "
+	CREATE SUBSCRIPTION $subname_BA
+	CONNECTION '$node_A_connstr application_name=$subname_BA'
+	PUBLICATION tap_pub_A
+	WITH (origin = none, detect_update_deleted = true)");
+
+# node_B (pub) -> node_A (sub)
+my $node_B_connstr = $node_B->connstr . ' dbname=postgres';
+$node_B->safe_psql('postgres', "CREATE PUBLICATION tap_pub_B FOR TABLE tab");
+$node_A->safe_psql(
+	'postgres', "
+	CREATE SUBSCRIPTION $subname_AB
+	CONNECTION '$node_B_connstr application_name=$subname_AB'
+	PUBLICATION tap_pub_B
+	WITH (origin = none, copy_data = off, detect_update_deleted = true)");
+
+# Wait for initial table sync to finish
+$node_A->wait_for_subscription_sync($node_B, $subname_AB);
+$node_B->wait_for_subscription_sync($node_A, $subname_BA);
+
+is(1, 1, 'Bidirectional replication setup is complete');
+
+# Confirm that the additional replication slot is created on both nodes and the
+# xmin value is valid.
+
+ok( $node_A->poll_query_until(
+		'postgres',
+		"SELECT xmin IS NOT NULL from pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"the xmin value of slot 'pg_conflict_detection' is valid on Node A");
+
+ok( $node_B->poll_query_until(
+		'postgres',
+		"SELECT xmin IS NOT NULL from pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"the xmin value of slot 'pg_conflict_detection' is valid on Node B");
+
+###############################################################################
+# Check that dead tuples on node A cannot be cleaned by VACUUM until the
+# concurrent transactions on Node B have been applied and flushed on Node A.
+###############################################################################
+
+# Insert a record
+$node_A->safe_psql('postgres', "INSERT INTO tab VALUES (1, 1), (2, 2);");
+$node_A->wait_for_catchup($subname_BA);
+
+my $result = $node_B->safe_psql('postgres', "SELECT * FROM tab;");
+is($result, qq(1|1
+2|2), 'check replicated insert on node B');
+
+# Disable the logical replication from node B to node A
+$node_A->safe_psql('postgres', "ALTER SUBSCRIPTION $subname_AB DISABLE");
+
+my $log_location = -s $node_B->logfile;
+
+$node_B->safe_psql('postgres', "UPDATE tab SET b = 3 WHERE a = 1;");
+$node_A->safe_psql('postgres', "DELETE FROM tab WHERE a = 1;");
+
+$node_A->wait_for_catchup($subname_BA);
+
+my ($cmdret, $stdout, $stderr) = $node_A->psql(
+	'postgres', qq(VACUUM (verbose) public.tab;)
+);
+
+ok( $stderr =~
+	  qr/1 are dead but not yet removable/,
+	'the deleted column is non-removable');
+
+my $logfile = slurp_file($node_B->logfile(), $log_location);
+ok( $logfile =~
+	  qr/conflict detected on relation "public.tab": conflict=delete_origin_differs.*\n.*DETAIL:.* Deleting the row that was modified locally in transaction [0-9]+ at .*\n.*Existing local tuple \(1, 3\); replica identity \(a\)=\(1\)/,
+	'delete target row was modified in tab');
+
+$log_location = -s $node_A->logfile;
+
+# Remember the next transaction ID to be assigned
+my $next_xid = $node_A->safe_psql('postgres', "SELECT txid_current() + 1;");
+
+$node_A->safe_psql(
+	'postgres', "ALTER SUBSCRIPTION $subname_AB ENABLE;");
+$node_B->wait_for_catchup($subname_AB);
+
+# Account for the transaction ID increment caused by enabling the subscription
+$next_xid++;
+
+$logfile = slurp_file($node_A->logfile(), $log_location);
+ok( $logfile =~
+	  qr/conflict detected on relation "public.tab": conflict=update_deleted.*\n.*DETAIL:.* The row to be updated was deleted locally in transaction [0-9]+ at .*\n.*Remote tuple \(1, 3\); replica identity \(a\)=\(1\)/,
+	'update target row was deleted in tab');
+
+# Confirm that the xmin value is updated
+ok( $node_A->poll_query_until(
+		'postgres',
+		"SELECT xmin = $next_xid from pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"the xmin value of slot 'pg_conflict_detection' is updated on Node A");
+
+###############################################################################
+# Check that the replication slot pg_conflict_detection is dropped after
+# removing the subscriptions that have the detect_update_deleted option enabled.
+###############################################################################
+
+$node_B->safe_psql(
+	'postgres', "DROP SUBSCRIPTION $subname_BA");
+
+ok( $node_B->poll_query_until(
+		'postgres',
+		"SELECT count(*) = 0 FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"the slot 'pg_conflict_detection' has been dropped on Node B");
+
+$node_A->safe_psql(
+	'postgres', "DROP SUBSCRIPTION $subname_AB");
+
+ok( $node_A->poll_query_until(
+		'postgres',
+		"SELECT count(*) = 0 FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"the slot 'pg_conflict_detection' has been dropped on Node A");
+
+done_testing();
-- 
2.30.0.windows.2