0001-fix-alter-replication-slot.patch.txt

text/plain

Filename: 0001-fix-alter-replication-slot.patch.txt
Type: text/plain
Part: 0
Message: RE: Slow catchup of 2PC (twophase) transactions on replica in LR
From 3bbaaba53a0cb3db43cc893acbd3ffbedd61bff1 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Sat, 13 Jul 2024 18:31:28 +0800
Subject: [PATCH] fix alter replication slot

---
 doc/src/sgml/protocol.sgml              | 16 ++++++++++++++
 src/backend/commands/subscriptioncmds.c |  2 +-
 src/backend/replication/slot.c          | 16 ++++++++------
 src/backend/replication/walsender.c     | 29 +++++++++++++++----------
 src/include/replication/slot.h          |  4 ++--
 5 files changed, 46 insertions(+), 21 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 1b27d0a547..3ac4a4be28 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2206,6 +2206,22 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
        </varlistentry>
       </variablelist>
 
+      <variablelist>
+       <varlistentry>
+        <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
+        <listitem>
+         <para>
+          If true, this logical replication slot supports decoding of two-phase
+          commit. With this option, commands related to two-phase commit such as
+          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
+          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
+          The transaction will be decoded and transmitted at
+          <literal>PREPARE TRANSACTION</literal> time.
+         </para>
+        </listitem>
+       </varlistentry>
+      </variablelist>
+
      </listitem>
     </varlistentry>
 
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 7604e228e8..c48b6d0549 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1308,7 +1308,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 						ereport(ERROR,
 								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 								 errmsg("cannot disable two_phase when uncommitted prepared transactions present"),
-								 errhint("Resolve these transactions and try again")));
+								 errhint("Resolve these transactions and try again.")));
 
 					/* Change system catalog acoordingly */
 					values[Anum_pg_subscription_subtwophasestate - 1] =
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 2ad6dca993..2f167a2adc 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -804,11 +804,12 @@ ReplicationSlotDrop(const char *name, bool nowait)
  * Change the definition of the slot identified by the specified name.
  */
 void
-ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
+ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase)
 {
 	bool		update_slot = false;
 
 	Assert(MyReplicationSlot == NULL);
+	Assert(failover || two_phase);
 
 	ReplicationSlotAcquire(name, false);
 
@@ -834,7 +835,7 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
 		 * Do not allow users to enable failover on the standby as we do not
 		 * support sync to the cascading standby.
 		 */
-		if (failover)
+		if (failover && *failover)
 			ereport(ERROR,
 					errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 					errmsg("cannot enable failover for a replication slot"
@@ -845,24 +846,25 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
 	 * Do not allow users to enable failover for temporary slots as we do not
 	 * support syncing temporary slots to the standby.
 	 */
-	if (failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
+	if (failover && *failover &&
+		MyReplicationSlot->data.persistency == RS_TEMPORARY)
 		ereport(ERROR,
 				errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				errmsg("cannot enable failover for a temporary replication slot"));
 
-	if (MyReplicationSlot->data.failover != failover)
+	if (failover && MyReplicationSlot->data.failover != *failover)
 	{
 		SpinLockAcquire(&MyReplicationSlot->mutex);
-		MyReplicationSlot->data.failover = failover;
+		MyReplicationSlot->data.failover = *failover;
 		SpinLockRelease(&MyReplicationSlot->mutex);
 
 		update_slot = true;
 	}
 
-	if (MyReplicationSlot->data.two_phase != two_phase)
+	if (two_phase && MyReplicationSlot->data.two_phase != *two_phase)
 	{
 		SpinLockAcquire(&MyReplicationSlot->mutex);
-		MyReplicationSlot->data.two_phase = two_phase;
+		MyReplicationSlot->data.two_phase = *two_phase;
 		SpinLockRelease(&MyReplicationSlot->mutex);
 
 		update_slot = true;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 5224ea6c2c..f3b5068d95 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,30 +1411,31 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
  */
 static void
 ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
-						  bool *failover, bool *two_phase)
+						  bool *failover_given, bool *failover,
+						  bool *two_phase_given, bool *two_phase)
 {
-	bool		failover_given = false;
-	bool		two_phase_given = false;
+	*failover_given = false;
+	*two_phase_given = false;
 
 	/* Parse options */
 	foreach_ptr(DefElem, defel, cmd->options)
 	{
 		if (strcmp(defel->defname, "failover") == 0)
 		{
-			if (failover_given)
+			if (*failover_given)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			failover_given = true;
+			*failover_given = true;
 			*failover = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
-			if (two_phase_given)
+			if (*two_phase_given)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			two_phase_given = true;
+			*two_phase_given = true;
 			*two_phase = defGetBoolean(defel);
 		}
 		else
@@ -1448,11 +1449,17 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
 static void
 AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
 {
-	bool		failover = false;
-	bool		two_phase = false;
+	bool		failover_given;
+	bool		two_phase_given;
+	bool		failover;
+	bool		two_phase;
+
+	ParseAlterReplSlotOptions(cmd, &failover_given, &failover,
+							  &two_phase_given, &two_phase);
 
-	ParseAlterReplSlotOptions(cmd, &failover, &two_phase);
-	ReplicationSlotAlter(cmd->slotname, failover, two_phase);
+	ReplicationSlotAlter(cmd->slotname,
+						 failover_given ? &failover : NULL,
+						 two_phase_given ? &two_phase : NULL);
 }
 
 /*
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 163a4a911a..cde164472a 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -243,8 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
 extern void ReplicationSlotPersist(void);
 extern void ReplicationSlotDrop(const char *name, bool nowait);
 extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover,
-								 bool two_phase);
+extern void ReplicationSlotAlter(const char *name, bool *failover,
+								 bool *two_phase);
 
 extern void ReplicationSlotAcquire(const char *name, bool nowait);
 extern void ReplicationSlotRelease(void);
-- 
2.30.0.windows.2