Add a retain_conflict_info option to subscriptions

This patch adds a subscription option allowing users to specify whether
information on the subscriber, which is useful for detecting update_deleted
conflicts, should be retained. The default setting is false. If set to true,
the detection of update_deleted will be enabled, and an additional replication
slot named pg_conflict_detection will be created on the subscriber to prevent
conflict information from being removed. Note that if multiple subscriptions on
one node enable this option, only one replication slot will be created.

The logical launcher will create and maintain a replication slot named
pg_conflict_detection only if any local subscription has the
retain_conflict_info option enabled.

Enabling retain_conflict_info is prohibited if the publisher is currently
in recovery mode (operating as a standby server).

Bump catalog version

Author: Zhijie Hou

doc/src/sgml/catalogs.sgml

@@ -8088,6 +8088,18 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
       </para></entry>
      </row>
 
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>subretainconflictinfo</structfield> <type>bool</type>
+      </para>
+      <para>
+       If true, the detection of <xref linkend="conflict-update-deleted"/> is
+       enabled and the information (e.g., dead tuples, commit timestamps, and
+       origins) on the subscriber that is still useful for conflict detection
+       is retained.
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>subconninfo</structfield> <type>text</type>

doc/src/sgml/logical-replication.sgml

@@ -2408,7 +2408,8 @@ CONTEXT:  processing remote data for replication origin "pg_16395" during "INSER
    <para>
     <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link>
     must be set to at least the number of subscriptions expected to connect,
-    plus some reserve for table synchronization.
+    plus some reserve for table synchronization and one if
+    <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</literal></link> is enabled.
    </para>
 
    <para>

doc/src/sgml/ref/alter_subscription.sgml

@@ -235,8 +235,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       <link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
       <link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
       <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
-      <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
-      <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>.
+      <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>,
+      <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>, and
+      <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</literal></link>.
       Only a superuser can set <literal>password_required = false</literal>.
      </para>
 
@@ -285,6 +286,13 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       option is changed from <literal>true</literal> to <literal>false</literal>,
       the publisher will replicate the transactions again when they are committed.
      </para>
+
+     <para>
+      If the <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</literal></link>
+      option is altered to <literal>false</literal> and no other subscription
+      has this option enabled, the additional replication slot that was created
+      to retain conflict information will be dropped.
+     </para>
     </listitem>
    </varlistentry>
 

doc/src/sgml/ref/create_subscription.sgml

@@ -437,6 +437,35 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
          </para>
         </listitem>
        </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-retain-conflict-info">
+        <term><literal>retain_conflict_info</literal> (<type>boolean</type>)</term>
+        <listitem>
+         <para>
+          Specifies whether the information (e.g., dead tuples, commit
+          timestamps, and origins) on the subscriber that is still useful for
+          conflict detection is retained. The default is
+          <literal>false</literal>. If set to true, the detection of
+          <xref linkend="conflict-update-deleted"/> is enabled, and an
+          additional replication slot named
+          <quote><literal>pg_conflict_detection</literal></quote> will be
+          created on the subscriber to prevent the conflict information from
+          being removed.
+         </para>
+
+         <para>
+          Note that the information useful for conflict detection is retained
+          only after the creation of the additional slot. You can verify the
+          existence of this slot by querying <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflicting</structfield>
+          And even if multiple subscriptions on one node enable this option,
+          only one replication slot will be created.
+         </para>
+
+         <para>
+          This option cannot be enabled if the publisher is also a physical standby.
+         </para>
+        </listitem>
+       </varlistentry>
       </variablelist></para>
 
     </listitem>

src/backend/catalog/pg_subscription.c

@@ -103,6 +103,7 @@ GetSubscription(Oid subid, bool missing_ok)
 	sub->passwordrequired = subform->subpasswordrequired;
 	sub->runasowner = subform->subrunasowner;
 	sub->failover = subform->subfailover;
+	sub->retainconflictinfo = subform->subretainconflictinfo;
 
 	/* Get conninfo */
 	datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,

src/backend/catalog/system_views.sql

@@ -1383,7 +1383,8 @@ REVOKE ALL ON pg_subscription FROM public;
 GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
               subbinary, substream, subtwophasestate, subdisableonerr,
 			  subpasswordrequired, subrunasowner, subfailover,
-              subslotname, subsynccommit, subpublications, suborigin)
+			  subretainconflictinfo, subslotname, subsynccommit,
+			  subpublications, suborigin)
     ON pg_subscription TO public;
 
 CREATE VIEW pg_stat_subscription_stats AS

src/backend/commands/subscriptioncmds.c

@@ -14,6 +14,7 @@
 
 #include "postgres.h"
 
+#include "access/commit_ts.h"
 #include "access/htup_details.h"
 #include "access/table.h"
 #include "access/twophase.h"
@@ -71,8 +72,9 @@
 #define SUBOPT_PASSWORD_REQUIRED	0x00000800
 #define SUBOPT_RUN_AS_OWNER			0x00001000
 #define SUBOPT_FAILOVER				0x00002000
-#define SUBOPT_LSN					0x00004000
-#define SUBOPT_ORIGIN				0x00008000
+#define SUBOPT_RETAIN_CONFLICT_INFO		0x00004000
+#define SUBOPT_LSN					0x00008000
+#define SUBOPT_ORIGIN				0x00010000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -98,6 +100,7 @@ typedef struct SubOpts
 	bool		passwordrequired;
 	bool		runasowner;
 	bool		failover;
+	bool		retainconflictinfo;
 	char	   *origin;
 	XLogRecPtr	lsn;
 } SubOpts;
@@ -107,6 +110,8 @@ static void check_publications_origin(WalReceiverConn *wrconn,
 									  List *publications, bool copydata,
 									  char *origin, Oid *subrel_local_oids,
 									  int subrel_count, char *subname);
+static void check_conflict_info_retaintion(WalReceiverConn *wrconn,
+										   bool retain_conflict_info);
 static void check_duplicates_in_publist(List *publist, Datum *datums);
 static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname);
 static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
@@ -162,6 +167,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 		opts->runasowner = false;
 	if (IsSet(supported_opts, SUBOPT_FAILOVER))
 		opts->failover = false;
+	if (IsSet(supported_opts, SUBOPT_RETAIN_CONFLICT_INFO))
+		opts->retainconflictinfo = false;
 	if (IsSet(supported_opts, SUBOPT_ORIGIN))
 		opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
 
@@ -307,6 +314,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			opts->specified_opts |= SUBOPT_FAILOVER;
 			opts->failover = defGetBoolean(defel);
 		}
+		else if (IsSet(supported_opts, SUBOPT_RETAIN_CONFLICT_INFO) &&
+				 strcmp(defel->defname, "retain_conflict_info") == 0)
+		{
+			if (IsSet(opts->specified_opts, SUBOPT_RETAIN_CONFLICT_INFO))
+				errorConflictingDefElem(defel, pstate);
+
+			opts->specified_opts |= SUBOPT_RETAIN_CONFLICT_INFO;
+			opts->retainconflictinfo = defGetBoolean(defel);
+		}
 		else if (IsSet(supported_opts, SUBOPT_ORIGIN) &&
 				 strcmp(defel->defname, "origin") == 0)
 		{
@@ -563,7 +579,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
 					  SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
 					  SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
-					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
+					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
+					  SUBOPT_RETAIN_CONFLICT_INFO | SUBOPT_ORIGIN);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/*
@@ -608,6 +625,12 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 				 errmsg("password_required=false is superuser-only"),
 				 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
 
+	if (opts.retainconflictinfo && !track_commit_timestamp)
+		ereport(WARNING,
+				errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("information for detecting conflicts cannot be fully retained when \"%s\" is disabled",
+					   "track_commit_timestamp"));
+
 	/*
 	 * If built with appropriate switch, whine when regression-testing
 	 * conventions for subscription names are violated.
@@ -670,6 +693,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
 	values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner);
 	values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
+	values[Anum_pg_subscription_subretainconflictinfo - 1] =
+		BoolGetDatum(opts.retainconflictinfo);
 	values[Anum_pg_subscription_subconninfo - 1] =
 		CStringGetTextDatum(conninfo);
 	if (opts.slot_name)
@@ -724,6 +749,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 			check_publications_origin(wrconn, publications, opts.copy_data,
 									  opts.origin, NULL, 0, stmt->subname);
 
+			check_conflict_info_retaintion(wrconn, opts.retainconflictinfo);
+
 			/*
 			 * Set sync state based on if we were asked to do data copy or
 			 * not.
@@ -1110,6 +1137,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 	bool		update_tuple = false;
 	bool		update_failover = false;
 	bool		update_two_phase = false;
+	bool		retain_conflict_info = false;
 	Subscription *sub;
 	Form_pg_subscription form;
 	bits32		supported_opts;
@@ -1165,7 +1193,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 								  SUBOPT_DISABLE_ON_ERR |
 								  SUBOPT_PASSWORD_REQUIRED |
 								  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
-								  SUBOPT_ORIGIN);
+								  SUBOPT_RETAIN_CONFLICT_INFO | SUBOPT_ORIGIN);
 
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
@@ -1325,6 +1353,29 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					replaces[Anum_pg_subscription_subfailover - 1] = true;
 				}
 
+				if (IsSet(opts.specified_opts, SUBOPT_RETAIN_CONFLICT_INFO))
+				{
+					values[Anum_pg_subscription_subretainconflictinfo - 1] =
+						BoolGetDatum(opts.retainconflictinfo);
+					replaces[Anum_pg_subscription_subretainconflictinfo - 1] = true;
+
+					if (opts.retainconflictinfo && !track_commit_timestamp)
+						ereport(WARNING,
+								errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+								errmsg("information for detecting conflicts cannot be fully retained when \"%s\" is disabled",
+									   "track_commit_timestamp"));
+
+					/*
+					 * Notify the launcher to manage the replication slot for
+					 * conflict detection. This ensures that replication slot
+					 * is efficiently handled (created, updated, or dropped)
+					 * in response to any configuration changes.
+					 */
+					ApplyLauncherWakeupAtCommit();
+
+					retain_conflict_info = opts.retainconflictinfo;
+				}
+
 				if (IsSet(opts.specified_opts, SUBOPT_ORIGIN))
 				{
 					values[Anum_pg_subscription_suborigin - 1] =
@@ -1355,6 +1406,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					ApplyLauncherWakeupAtCommit();
 
 				update_tuple = true;
+
+				/*
+				 * The subscription might be initially created with
+				 * connect=false and retain_conflict_info=true, meaning the
+				 * remote server's status may not be checked. Ensure this
+				 * check is conducted now.
+				 */
+				retain_conflict_info = sub->retainconflictinfo;
 				break;
 			}
 
@@ -1369,6 +1428,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				CStringGetTextDatum(stmt->conninfo);
 			replaces[Anum_pg_subscription_subconninfo - 1] = true;
 			update_tuple = true;
+
+			/*
+			 * Since the remote server configuration might have changed,
+			 * perform a check to ensure it permits enabling
+			 * retain_conflict_info.
+			 */
+			retain_conflict_info = sub->retainconflictinfo;
 			break;
 
 		case ALTER_SUBSCRIPTION_SET_PUBLICATION:
@@ -1568,14 +1634,15 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 	}
 
 	/*
-	 * Try to acquire the connection necessary for altering the slot, if
-	 * needed.
+	 * Try to acquire the connection necessary either for modifying the slot
+	 * or for checking if the remote server permits enabling
+	 * retain_conflict_info.
 	 *
 	 * This has to be at the end because otherwise if there is an error while
 	 * doing the database operations we won't be able to rollback altered
 	 * slot.
 	 */
-	if (update_failover || update_two_phase)
+	if (update_failover || update_two_phase || retain_conflict_info)
 	{
 		bool		must_use_password;
 		char	   *err;
@@ -1584,10 +1651,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 		/* Load the library providing us libpq calls. */
 		load_file("libpqwalreceiver", false);
 
-		/* Try to connect to the publisher. */
+		/*
+		 * Try to connect to the publisher, using the new connection string if
+		 * available.
+		 */
 		must_use_password = sub->passwordrequired && !sub->ownersuperuser;
-		wrconn = walrcv_connect(sub->conninfo, true, true, must_use_password,
-								sub->name, &err);
+		wrconn = walrcv_connect(stmt->conninfo ? stmt->conninfo : sub->conninfo,
+								true, true, must_use_password, sub->name,
+								&err);
 		if (!wrconn)
 			ereport(ERROR,
 					(errcode(ERRCODE_CONNECTION_FAILURE),
@@ -1596,9 +1667,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 		PG_TRY();
 		{
-			walrcv_alter_slot(wrconn, sub->slotname,
-							  update_failover ? &opts.failover : NULL,
-							  update_two_phase ? &opts.twophase : NULL);
+			check_conflict_info_retaintion(wrconn, retain_conflict_info);
+
+			if (update_failover || update_two_phase)
+				walrcv_alter_slot(wrconn, sub->slotname,
+								  update_failover ? &opts.failover : NULL,
+								  update_two_phase ? &opts.twophase : NULL);
 		}
 		PG_FINALLY();
 		{
@@ -2196,6 +2270,57 @@ check_publications_origin(WalReceiverConn *wrconn, List *publications,
 	walrcv_clear_result(res);
 }
 
+/*
+ * Check if the publisher's status permits enabling retain_conflict_info.
+ *
+ * Enabling retain_conflict_info is not allowed if the publisher's version is
+ * prior to PG18 or if the publisher is in recovery (operating as a standby
+ * server).
+ *
+ * Refer to the comments atop maybe_advance_nonremovable_xid() for detailed
+ * reasons.
+ */
+static void
+check_conflict_info_retaintion(WalReceiverConn *wrconn, bool retain_conflict_info)
+{
+	WalRcvExecResult *res;
+	Oid			RecoveryRow[1] = {BOOLOID};
+	TupleTableSlot *slot;
+	bool		isnull;
+	bool		remote_in_recovery;
+
+	if (!retain_conflict_info)
+		return;
+
+	if (walrcv_server_version(wrconn) < 18000)
+		ereport(ERROR,
+				errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("cannot enable retain_conflict_info if the publisher is running a version earlier than PostgreSQL 18."));
+
+	res = walrcv_exec(wrconn, "SELECT pg_is_in_recovery()", 1, RecoveryRow);
+
+	if (res->status != WALRCV_OK_TUPLES)
+		ereport(ERROR,
+				(errcode(ERRCODE_CONNECTION_FAILURE),
+				 errmsg("could not obtain recovery progress from the publisher: %s",
+						res->err)));
+
+	slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+	if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+		elog(ERROR, "failed to fetch tuple for the recovery progress");
+
+	remote_in_recovery = DatumGetBool(slot_getattr(slot, 1, &isnull));
+
+	if (remote_in_recovery)
+		ereport(ERROR,
+				errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				errmsg("cannot enable retain_conflict_info if the publisher is in recovery."));
+
+	ExecDropSingleTupleTableSlot(slot);
+
+	walrcv_clear_result(res);
+}
+
 /*
  * Get the list of tables which belong to specified publications on the
  * publisher connection.