From d6da271d85f846466c31bb1632846dcd9f466fd3 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Fri, 20 Jun 2025 16:16:49 +0800
Subject: [PATCH v39 3/8] Migrate the conflict detection slot to the new node
 during an upgrade.

This patch enables pg_upgrade to create the replication slot that retains
information necessary for conflict detection such as dead tuples, commit
timestamps, and origins, for migrated subscriptions with retain_conflict_info
enabled.
---
 src/backend/replication/logical/launcher.c |  7 ++--
 src/backend/replication/slot.c             |  7 ++--
 src/backend/utils/adt/pg_upgrade_support.c | 19 ++++++++++
 src/bin/pg_upgrade/pg_upgrade.c            | 42 ++++++++++++++++++++--
 src/bin/pg_upgrade/t/004_subscription.pl   | 39 +++++++++-----------
 src/include/catalog/pg_proc.dat            |  4 +++
 src/include/replication/logicallauncher.h  |  2 ++
 7 files changed, 88 insertions(+), 32 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 83935de9141..409cd7f2214 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -105,7 +105,6 @@ static void compute_min_nonremovable_xid(LogicalRepWorker *worker,
 										 TransactionId *xmin,
 										 bool *can_advance_xmin);
 static bool acquire_conflict_slot_if_exists(void);
-static void create_conflict_slot_if_not_exists(void);
 static void advance_conflict_slot_xmin(TransactionId new_xmin);
 
 
@@ -1348,7 +1347,7 @@ compute_min_nonremovable_xid(LogicalRepWorker *worker,
 		 * conflict-related information is available when applying remote
 		 * changes that occurred before the subscription was enabled.
 		 */
-		create_conflict_slot_if_not_exists();
+		ApplyLauncherCreateConflictDetectionSlot();
 
 		/*
 		 * Can't advance xmin of the slot unless all the subscriptions with
@@ -1379,8 +1378,8 @@ acquire_conflict_slot_if_exists(void)
  * Create and acquire the replication slot used to retain information for
  * conflict detection, if not yet.
  */
-static void
-create_conflict_slot_if_not_exists(void)
+void
+ApplyLauncherCreateConflictDetectionSlot(void)
 {
 	TransactionId xmin_horizon;
 
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index d8dd458e93b..aa7ada418f8 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -359,10 +359,11 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 	Assert(MyReplicationSlot == NULL);
 
 	/*
-	 * The logical launcher might be creating an internal slot, so using a
-	 * reserved name is allowed in this case.
+	 * The logical launcher or pg_upgrade may create or migrate an internal
+	 * slot, so using a reserved name is allowed in these cases.
 	 */
-	ReplicationSlotValidateName(name, IsLogicalLauncher(), ERROR);
+	ReplicationSlotValidateName(name, IsBinaryUpgrade || IsLogicalLauncher(),
+								ERROR);
 
 	if (failover)
 	{
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index d44f8c262ba..fdb853bb344 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -21,6 +21,7 @@
 #include "commands/extension.h"
 #include "miscadmin.h"
 #include "replication/logical.h"
+#include "replication/logicallauncher.h"
 #include "replication/origin.h"
 #include "replication/worker_internal.h"
 #include "storage/lmgr.h"
@@ -410,3 +411,21 @@ binary_upgrade_replorigin_advance(PG_FUNCTION_ARGS)
 
 	PG_RETURN_VOID();
 }
+
+/*
+ * binary_upgrade_create_conflict_detection_slot
+ *
+ * Create a replication slot to retain information necessary for conflict
+ * detection such as dead tuples, commit timestamps, and origins.
+ */
+Datum
+binary_upgrade_create_conflict_detection_slot(PG_FUNCTION_ARGS)
+{
+	CHECK_IS_BINARY_UPGRADE;
+
+	ApplyLauncherCreateConflictDetectionSlot();
+
+	ReplicationSlotRelease();
+
+	PG_RETURN_VOID();
+}
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index 536e49d2616..f6b967bc3a9 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -67,6 +67,7 @@ static void set_frozenxids(bool minmxid_only);
 static void make_outputdirs(char *pgdata);
 static void setup(char *argv0);
 static void create_logical_replication_slots(void);
+static void create_conflict_detection_slot(void);
 
 ClusterInfo old_cluster,
 			new_cluster;
@@ -204,12 +205,15 @@ main(int argc, char **argv)
 	 * removed and slots would become unusable.  There is a possibility that
 	 * background processes might generate some WAL before we could create the
 	 * slots in the new cluster but we can ignore that WAL as that won't be
-	 * required downstream.
+	 * required downstream. The conflict detection slots is not affected by
+	 * these concerns, but is created here for consistency.
 	 */
-	if (count_old_cluster_logical_slots())
+	if (count_old_cluster_logical_slots() ||
+		old_cluster.sub_retain_conflict_info)
 	{
 		start_postmaster(&new_cluster, true);
 		create_logical_replication_slots();
+		create_conflict_detection_slot();
 		stop_postmaster(false);
 	}
 
@@ -971,11 +975,15 @@ set_frozenxids(bool minmxid_only)
 /*
  * create_logical_replication_slots()
  *
- * Similar to create_new_objects() but only restores logical replication slots.
+ * Similar to create_new_objects() but only restores logical replication slots
+ * if any.
  */
 static void
 create_logical_replication_slots(void)
 {
+	if (!count_old_cluster_logical_slots())
+		return;
+
 	prep_status_progress("Restoring logical replication slots in the new cluster");
 
 	for (int dbnum = 0; dbnum < old_cluster.dbarr.ndbs; dbnum++)
@@ -1025,3 +1033,31 @@ create_logical_replication_slots(void)
 
 	return;
 }
+
+/*
+ * create_conflict_detection_slot()
+ *
+ * Create a replication slot to retain information necessary for conflict
+ * detection such as dead tuples, commit timestamps, and origins, for migrated
+ * subscriptions with retain_conflict_info enabled.
+ */
+static void
+create_conflict_detection_slot(void)
+{
+	PGconn	   *conn_new_template1;
+
+	/*
+	 * Quickly return if no subscriptions with retain_conflict_info enabled are
+	 * migrated.
+	 */
+	if (!old_cluster.sub_retain_conflict_info)
+		return;
+
+	prep_status("Creating the logical replication conflict detection slot in the new cluster");
+
+	conn_new_template1 = connectToServer(&new_cluster, "template1");
+	PQclear(executeQueryOrDie(conn_new_template1, "SELECT pg_catalog.binary_upgrade_create_conflict_detection_slot()"));
+	PQfinish(conn_new_template1);
+
+	check_ok();
+}
diff --git a/src/bin/pg_upgrade/t/004_subscription.pl b/src/bin/pg_upgrade/t/004_subscription.pl
index 4978c9d5636..d4a9e194b68 100644
--- a/src/bin/pg_upgrade/t/004_subscription.pl
+++ b/src/bin/pg_upgrade/t/004_subscription.pl
@@ -22,13 +22,13 @@ $publisher->start;
 
 # Initialize the old subscriber node
 my $old_sub = PostgreSQL::Test::Cluster->new('old_sub');
-$old_sub->init;
+$old_sub->init(allows_streaming => 'physical');
 $old_sub->start;
 my $oldbindir = $old_sub->config_data('--bindir');
 
 # Initialize the new subscriber
 my $new_sub = PostgreSQL::Test::Cluster->new('new_sub');
-$new_sub->init;
+$new_sub->init(allows_streaming => 'physical');
 my $newbindir = $new_sub->config_data('--bindir');
 
 # In a VPATH build, we'll be started in the source directory, but we want
@@ -82,9 +82,6 @@ command_checks_all(
 # Reset max_active_replication_origins
 $new_sub->append_conf('postgresql.conf', "max_active_replication_origins = 10");
 
-# Increase wal_level to allow enabling retain_conflict_info in next test
-$old_sub->append_conf('postgresql.conf', "wal_level = replica");
-
 # Cleanup
 $publisher->safe_psql('postgres', "DROP PUBLICATION regress_pub1");
 $old_sub->start;
@@ -104,10 +101,7 @@ $old_sub->safe_psql('postgres',
 
 $old_sub->stop;
 
-$new_sub->append_conf(
-	'postgresql.conf',
-	qq{max_replication_slots = 0
-    wal_level = replica});
+$new_sub->append_conf('postgresql.conf', 'max_replication_slots = 0');
 
 # pg_upgrade will fail because the new cluster has insufficient
 # max_replication_slots.
@@ -133,13 +127,8 @@ command_checks_all(
 	'run of pg_upgrade where the new cluster has insufficient max_replication_slots'
 );
 
-# Reset max_replication_slots and wal_level
-$new_sub->append_conf(
-	'postgresql.conf',
-	qq{max_replication_slots = 10
-    wal_level = minimal});
-
-$old_sub->append_conf('postgresql.conf', "wal_level = minimal");
+# Reset max_replication_slots
+$new_sub->append_conf('postgresql.conf', 'max_replication_slots = 10');
 
 # Cleanup
 $publisher->safe_psql('postgres', "DROP PUBLICATION regress_pub1");
@@ -258,7 +247,7 @@ rmtree($new_sub->data_dir . "/pg_upgrade_output.d");
 
 # Verify that the upgrade should be successful with tables in 'ready'/'init'
 # state along with retaining the replication origin's remote lsn, subscription's
-# running status, and failover option.
+# running status, failover, and retain_conflict_info option.
 $publisher->safe_psql(
 	'postgres', qq[
 		CREATE TABLE tab_upgraded1(id int);
@@ -268,7 +257,7 @@ $publisher->safe_psql(
 $old_sub->safe_psql(
 	'postgres', qq[
 		CREATE TABLE tab_upgraded1(id int);
-		CREATE SUBSCRIPTION regress_sub4 CONNECTION '$connstr' PUBLICATION regress_pub4 WITH (failover = true);
+		CREATE SUBSCRIPTION regress_sub4 CONNECTION '$connstr' PUBLICATION regress_pub4 WITH (failover = true, retain_conflict_info = true);
 ]);
 
 # Wait till the table tab_upgraded1 reaches 'ready' state
@@ -350,7 +339,8 @@ ok( !-d $new_sub->data_dir . "/pg_upgrade_output.d",
 # ------------------------------------------------------
 # Check that the data inserted to the publisher when the new subscriber is down
 # will be replicated once it is started. Also check that the old subscription
-# states and relations origins are all preserved.
+# states and relations origins are all preserved, existence of the conflict
+# detection slot.
 # ------------------------------------------------------
 $publisher->safe_psql(
 	'postgres', qq[
@@ -364,10 +354,10 @@ $new_sub->start;
 # in the upgraded instance. So regress_sub4 should still have subenabled and
 # subfailover set to true, while regress_sub5 should have both set to false.
 $result = $new_sub->safe_psql('postgres',
-	"SELECT subname, subenabled, subfailover FROM pg_subscription ORDER BY subname"
+	"SELECT subname, subenabled, subfailover, subretainconflictinfo FROM pg_subscription ORDER BY subname"
 );
-is( $result, qq(regress_sub4|t|t
-regress_sub5|f|f),
+is( $result, qq(regress_sub4|t|t|t
+regress_sub5|f|f|f),
 	"check that the subscription's running status and failover are preserved"
 );
 
@@ -387,6 +377,11 @@ $result = $new_sub->safe_psql('postgres',
 );
 is($result, qq($remote_lsn), "remote_lsn should have been preserved");
 
+# The conflict detection slot should exist
+$result = $new_sub->safe_psql('postgres',
+	"SELECT xmin IS NOT NULL from pg_replication_slots WHERE slot_name = 'pg_conflict_detection'");
+is($result, qq(t), "conflict detection slot exists");
+
 # Resume the initial sync and wait until all tables of subscription
 # 'regress_sub5' are synchronized
 $new_sub->append_conf('postgresql.conf',
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index d3d28a263fa..359f03cd331 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11773,6 +11773,10 @@
   proname => 'binary_upgrade_replorigin_advance', proisstrict => 'f',
   provolatile => 'v', proparallel => 'u', prorettype => 'void',
   proargtypes => 'text pg_lsn', prosrc => 'binary_upgrade_replorigin_advance' },
+{ oid => '9159', descr => 'for use by pg_upgrade (conflict detection slot)',
+  proname => 'binary_upgrade_create_conflict_detection_slot', proisstrict => 'f',
+  provolatile => 'v', proparallel => 'u', prorettype => 'void',
+  proargtypes => '', prosrc => 'binary_upgrade_create_conflict_detection_slot' },
 
 # conversion functions
 { oid => '4302',
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index 7b29f1814db..d8eb7ebed39 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -28,6 +28,8 @@ extern void ApplyLauncherWakeupAtCommit(void);
 extern void ApplyLauncherWakeup(void);
 extern void AtEOXact_ApplyLauncher(bool isCommit);
 
+extern void ApplyLauncherCreateConflictDetectionSlot(void);
+
 extern bool IsLogicalLauncher(void);
 
 extern pid_t GetLeaderApplyWorkerPid(pid_t pid);
-- 
2.30.0.windows.2

