[CF 5781] v3 - Make wal_receiver_timeout configurable per subscription

This branch was automatically generated by a robot using patches from an
email thread registered at:

https://commitfest.postgresql.org/patch/5781

The branch will be overwritten each time a new patch version is posted to
the thread, and also periodically to check for bitrot caused by changes
on the master branch.

Patch(es): https://www.postgresql.org/message-id/CAHGQGwG82P4s6tmYK=aEm-T7QfGJBZvXo=WZfckMkffsX6DZjQ@mail.gmail.com
Author(s): Fujii Masao

Author: Commitfest Bot

doc/src/sgml/catalogs.sgml

@@ -8162,6 +8162,16 @@ 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>subwalrcvtimeout</structfield> <type>text</type>
+      </para>
+      <para>
+       The <varname>wal_receiver_timeout</varname>
+       setting for the subscription's workers to use
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>subpublications</structfield> <type>text[]</type>

doc/src/sgml/config.sgml

@@ -5154,9 +5154,6 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
         If this value is specified without units, it is taken as milliseconds.
         The default value is 60 seconds.
         A value of zero disables the timeout mechanism.
-        This parameter can only be set in
-        the <filename>postgresql.conf</filename> file or on the server
-        command line.
        </para>
       </listitem>
      </varlistentry>

doc/src/sgml/ref/alter_subscription.sgml

@@ -265,8 +265,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       <link linkend="sql-createsubscription-params-with-origin"><literal>origin</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>,
-      <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>, and
-      <link linkend="sql-createsubscription-params-with-max-retention-duration"><literal>max_retention_duration</literal></link>.
+      <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>,
+      <link linkend="sql-createsubscription-params-with-max-retention-duration"><literal>max_retention_duration</literal></link>, and
+      <link linkend="sql-createsubscription-params-with-wal-receiver-timeout"><literal>wal_receiver_timeout</literal></link>.
       Only a superuser can set <literal>password_required = false</literal>.
      </para>
 

doc/src/sgml/ref/create_subscription.sgml

@@ -563,8 +563,21 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
          </warning>
         </listitem>
        </varlistentry>
-      </variablelist></para>
 
+      <varlistentry id="sql-createsubscription-params-with-wal-receiver-timeout">
+        <term><literal>wal_receiver_timeout</literal> (<type>text</type>)</term>
+        <listitem>
+         <para>
+          The value of this parameter overrides the
+          <xref linkend="guc-wal-receiver-timeout"/> setting within this
+          subscription's apply worker processes.  The default value is
+          <literal>-1</literal>, which means it does not override the global setting,
+          i.e., the value from the server configuration, command line, role or
+          database settings will be used instead.
+         </para>
+        </listitem>
+       </varlistentry>
+      </variablelist></para>
     </listitem>
    </varlistentry>
   </variablelist>

src/backend/catalog/pg_subscription.c

@@ -129,6 +129,12 @@ GetSubscription(Oid subid, bool missing_ok)
 								   Anum_pg_subscription_subsynccommit);
 	sub->synccommit = TextDatumGetCString(datum);
 
+	/* Get walrcvtimeout */
+	datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
+								   tup,
+								   Anum_pg_subscription_subwalrcvtimeout);
+	sub->walrcvtimeout = TextDatumGetCString(datum);
+
 	/* Get publications */
 	datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
 								   tup,

src/backend/commands/subscriptioncmds.c

@@ -73,8 +73,9 @@
 #define SUBOPT_FAILOVER				0x00002000
 #define SUBOPT_RETAIN_DEAD_TUPLES	0x00004000
 #define SUBOPT_MAX_RETENTION_DURATION	0x00008000
-#define SUBOPT_LSN					0x00010000
-#define SUBOPT_ORIGIN				0x00020000
+#define SUBOPT_WAL_RECEIVER_TIMEOUT			0x00010000
+#define SUBOPT_LSN					0x00020000
+#define SUBOPT_ORIGIN				0x00040000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -104,6 +105,7 @@ typedef struct SubOpts
 	int32		maxretention;
 	char	   *origin;
 	XLogRecPtr	lsn;
+	char	   *wal_receiver_timeout;
 } SubOpts;
 
 /*
@@ -402,6 +404,30 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			opts->specified_opts |= SUBOPT_LSN;
 			opts->lsn = lsn;
 		}
+		else if (IsSet(supported_opts, SUBOPT_WAL_RECEIVER_TIMEOUT) &&
+				 strcmp(defel->defname, "wal_receiver_timeout") == 0)
+		{
+			bool		parsed;
+			int			val;
+
+			if (IsSet(opts->specified_opts, SUBOPT_WAL_RECEIVER_TIMEOUT))
+				errorConflictingDefElem(defel, pstate);
+
+			opts->specified_opts |= SUBOPT_WAL_RECEIVER_TIMEOUT;
+			opts->wal_receiver_timeout = defGetString(defel);
+
+			/*
+			 * Test if the given value is valid for wal_receiver_timeeout GUC.
+			 * Skip this test if the value is -1, since -1 is allowed for the
+			 * wal_receiver_timeout subscription option, but not for the GUC
+			 * itself.
+			 */
+			parsed = parse_int(opts->wal_receiver_timeout, &val, 0, NULL);
+			if (!parsed || val != -1)
+				(void) set_config_option("wal_receiver_timeout", opts->wal_receiver_timeout,
+										 PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
+										 false, 0, false);
+		}
 		else
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR),
@@ -610,7 +636,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
 					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
 					  SUBOPT_RETAIN_DEAD_TUPLES |
-					  SUBOPT_MAX_RETENTION_DURATION | SUBOPT_ORIGIN);
+					  SUBOPT_MAX_RETENTION_DURATION |
+					  SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/*
@@ -693,6 +720,14 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	if (opts.synchronous_commit == NULL)
 		opts.synchronous_commit = "off";
 
+	/*
+	 * The default for wal_receiver_timeout of subscriptions is -1, which
+	 * means the value is inherited from the server configuration, command
+	 * line, or role/database settings.
+	 */
+	if (opts.wal_receiver_timeout == NULL)
+		opts.wal_receiver_timeout = "-1";
+
 	conninfo = stmt->conninfo;
 	publications = stmt->publication;
 
@@ -740,6 +775,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 		nulls[Anum_pg_subscription_subslotname - 1] = true;
 	values[Anum_pg_subscription_subsynccommit - 1] =
 		CStringGetTextDatum(opts.synchronous_commit);
+	values[Anum_pg_subscription_subwalrcvtimeout - 1] =
+		CStringGetTextDatum(opts.wal_receiver_timeout);
 	values[Anum_pg_subscription_subpublications - 1] =
 		publicationListToArray(publications);
 	values[Anum_pg_subscription_suborigin - 1] =
@@ -1408,6 +1445,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 								  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
 								  SUBOPT_RETAIN_DEAD_TUPLES |
 								  SUBOPT_MAX_RETENTION_DURATION |
+								  SUBOPT_WAL_RECEIVER_TIMEOUT |
 								  SUBOPT_ORIGIN);
 
 				parse_subscription_options(pstate, stmt->options,
@@ -1663,6 +1701,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					origin = opts.origin;
 				}
 
+				if (IsSet(opts.specified_opts, SUBOPT_WAL_RECEIVER_TIMEOUT))
+				{
+					values[Anum_pg_subscription_subwalrcvtimeout - 1] =
+						CStringGetTextDatum(opts.wal_receiver_timeout);
+					replaces[Anum_pg_subscription_subwalrcvtimeout - 1] = true;
+				}
+
 				update_tuple = true;
 				break;
 			}

src/backend/replication/logical/worker.c

@@ -627,6 +627,8 @@ static inline void reset_apply_error_context_info(void);
 static TransApplyAction get_transaction_apply_action(TransactionId xid,
 													 ParallelApplyWorkerInfo **winfo);
 
+static void set_wal_receiver_timeout(void);
+
 static void replorigin_reset(int code, Datum arg);
 
 /*
@@ -5128,12 +5130,46 @@ maybe_reread_subscription(void)
 	SetConfigOption("synchronous_commit", MySubscription->synccommit,
 					PGC_BACKEND, PGC_S_OVERRIDE);
 
+	/* Change wal_receiver_timeout according to the user's wishes */
+	set_wal_receiver_timeout();
+
 	if (started_tx)
 		CommitTransactionCommand();
 
 	MySubscriptionValid = true;
 }
 
+/*
+ * Change wal_receiver_timeout to MySubscription->walrcvtimeout.
+ */
+static void
+set_wal_receiver_timeout(void)
+{
+	bool		parsed;
+	int			val;
+
+	/*
+	 * Set the wal_receiver_timeout GUC to MySubscription->walrcvtimeout,
+	 * which comes from the subscription's wal_receiver_timeout option. If the
+	 * value is -1, reset the GUC to its default, meaning it will inherit from
+	 * the server config, command line, or role/database settings.
+	 */
+	parsed = parse_int(MySubscription->walrcvtimeout, &val, 0, NULL);
+	if (parsed && val == -1)
+		SetConfigOption("wal_receiver_timeout", NULL,
+						PGC_BACKEND, PGC_S_SESSION);
+	else
+		SetConfigOption("wal_receiver_timeout", MySubscription->walrcvtimeout,
+						PGC_BACKEND, PGC_S_SESSION);
+
+	/*
+	 * Log the current wal_receiver_timeout GUC value (in milliseconds) as a
+	 * debug message to verify it was set correctly.
+	 */
+	elog(DEBUG1, "logical replication worker for subscription \"%s\" wal_receiver_timeout: %d ms",
+		 MySubscription->name, wal_receiver_timeout);
+}
+
 /*
  * Callback from subscription syscache invalidation.
  */
@@ -5795,6 +5831,9 @@ InitializeLogRepWorker(void)
 	SetConfigOption("synchronous_commit", MySubscription->synccommit,
 					PGC_BACKEND, PGC_S_OVERRIDE);
 
+	/* Change wal_receiver_timeout according to the user's wishes */
+	set_wal_receiver_timeout();
+
 	/*
 	 * Keep us informed about subscription or role changes. Note that the
 	 * role's superuser privilege can be revoked.

src/backend/utils/misc/guc_parameters.dat

@@ -1059,7 +1059,7 @@
   max => 'INT_MAX / 1000',
 },
 
-{ name => 'wal_receiver_timeout', type => 'int', context => 'PGC_SIGHUP', group => 'REPLICATION_STANDBY',
+{ name => 'wal_receiver_timeout', type => 'int', context => 'PGC_USERSET', group => 'REPLICATION_STANDBY',
   short_desc => 'Sets the maximum wait time to receive data from the sending server.',
   long_desc => '0 disables the timeout.',
   flags => 'GUC_UNIT_MS',

src/bin/pg_dump/pg_dump.c

@@ -5109,6 +5109,7 @@ getSubscriptions(Archive *fout)
 	int			i_subconninfo;
 	int			i_subslotname;
 	int			i_subsynccommit;
+	int			i_subwalrcvtimeout;
 	int			i_subpublications;
 	int			i_suborigin;
 	int			i_suboriginremotelsn;
@@ -5202,10 +5203,17 @@ getSubscriptions(Archive *fout)
 
 	if (fout->remoteVersion >= 190000)
 		appendPQExpBufferStr(query,
-							 " s.submaxretention\n");
+							 " s.submaxretention,\n");
 	else
 		appendPQExpBuffer(query,
-						  " 0 AS submaxretention\n");
+						  " 0 AS submaxretention,\n");
+
+	if (fout->remoteVersion >= 190000)
+		appendPQExpBufferStr(query,
+							 " s.subwalrcvtimeout\n");
+	else
+		appendPQExpBufferStr(query,
+							 " '-1' AS subwalrcvtimeout\n");
 
 	appendPQExpBufferStr(query,
 						 "FROM pg_subscription s\n");
@@ -5244,6 +5252,7 @@ getSubscriptions(Archive *fout)
 	i_subconninfo = PQfnumber(res, "subconninfo");
 	i_subslotname = PQfnumber(res, "subslotname");
 	i_subsynccommit = PQfnumber(res, "subsynccommit");
+	i_subwalrcvtimeout = PQfnumber(res, "subwalrcvtimeout");
 	i_subpublications = PQfnumber(res, "subpublications");
 	i_suborigin = PQfnumber(res, "suborigin");
 	i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
@@ -5287,6 +5296,8 @@ getSubscriptions(Archive *fout)
 				pg_strdup(PQgetvalue(res, i, i_subslotname));
 		subinfo[i].subsynccommit =
 			pg_strdup(PQgetvalue(res, i, i_subsynccommit));
+		subinfo[i].subwalrcvtimeout =
+			pg_strdup(PQgetvalue(res, i, i_subwalrcvtimeout));
 		subinfo[i].subpublications =
 			pg_strdup(PQgetvalue(res, i, i_subpublications));
 		subinfo[i].suborigin = pg_strdup(PQgetvalue(res, i, i_suborigin));
@@ -5545,6 +5556,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 	if (strcmp(subinfo->subsynccommit, "off") != 0)
 		appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
 
+	if (strcmp(subinfo->subwalrcvtimeout, "-1") != 0)
+		appendPQExpBuffer(query, ", wal_receiver_timeout = %s", fmtId(subinfo->subwalrcvtimeout));
+
 	if (pg_strcasecmp(subinfo->suborigin, LOGICALREP_ORIGIN_ANY) != 0)
 		appendPQExpBuffer(query, ", origin = %s", subinfo->suborigin);
 

src/bin/pg_dump/pg_dump.h

@@ -722,6 +722,7 @@ typedef struct _SubscriptionInfo
 	char	   *subconninfo;
 	char	   *subslotname;
 	char	   *subsynccommit;
+	char	   *subwalrcvtimeout;
 	char	   *subpublications;
 	char	   *suborigin;
 	char	   *suboriginremotelsn;