0007-protocol-changes-20230402.patch

From 84640161f11b80b9104d893ab9cdac0c987bcc1f Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas@2ndquadrant.com>
Date: Sun, 2 Apr 2023 18:57:53 +0200
Subject: [PATCH 7/7] protocol changes

---
 src/backend/catalog/pg_subscription.c         |  1 +
 src/backend/commands/subscriptioncmds.c       | 23 +++++++++++++++++-
 .../libpqwalreceiver/libpqwalreceiver.c       |  3 +++
 src/backend/replication/logical/decode.c      | 15 ++++++++++++
 src/backend/replication/logical/logical.c     | 24 +++++++++++++++++++
 src/backend/replication/logical/worker.c      | 11 ++++++++-
 src/backend/replication/pgoutput/pgoutput.c   | 23 +++++++++++++++++-
 src/include/catalog/pg_subscription.h         |  4 ++++
 src/include/replication/logical.h             | 15 ++++++++++++
 src/include/replication/logicalproto.h        |  3 ++-
 src/include/replication/walreceiver.h         |  1 +
 11 files changed, 119 insertions(+), 4 deletions(-)

diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 87e8ea7efa..bc296a6eaf 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -68,6 +68,7 @@ GetSubscription(Oid subid, bool missing_ok)
 	sub->owner = subform->subowner;
 	sub->enabled = subform->subenabled;
 	sub->binary = subform->subbinary;
+	sub->sequences = subform->subsequences;
 	sub->stream = subform->substream;
 	sub->twophasestate = subform->subtwophasestate;
 	sub->disableonerr = subform->subdisableonerr;
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 8f52bc9c37..c1b7de0470 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -70,6 +70,7 @@
 #define SUBOPT_PASSWORD_REQUIRED	0x00000800
 #define SUBOPT_LSN					0x00001000
 #define SUBOPT_ORIGIN				0x00002000
+#define SUBOPT_SEQUENCES			0x00004000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -91,6 +92,7 @@ typedef struct SubOpts
 	bool		binary;
 	char		streaming;
 	bool		twophase;
+	bool		sequences;
 	bool		disableonerr;
 	bool		passwordrequired;
 	char	   *origin;
@@ -144,6 +146,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 		opts->refresh = true;
 	if (IsSet(supported_opts, SUBOPT_BINARY))
 		opts->binary = false;
+	if (IsSet(supported_opts, SUBOPT_SEQUENCES))
+		opts->sequences = false;
 	if (IsSet(supported_opts, SUBOPT_STREAMING))
 		opts->streaming = LOGICALREP_STREAM_OFF;
 	if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
@@ -252,6 +256,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			opts->specified_opts |= SUBOPT_STREAMING;
 			opts->streaming = defGetStreamingMode(defel);
 		}
+		else if (IsSet(supported_opts, SUBOPT_SEQUENCES) &&
+				 strcmp(defel->defname, "sequences") == 0)
+		{
+			if (IsSet(opts->specified_opts, SUBOPT_SEQUENCES))
+				errorConflictingDefElem(defel, pstate);
+
+			opts->specified_opts |= SUBOPT_SEQUENCES;
+			opts->sequences = defGetBoolean(defel);
+		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
 			/*
@@ -579,7 +592,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
 					  SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
 					  SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
-					  SUBOPT_ORIGIN);
+					  SUBOPT_ORIGIN | SUBOPT_SEQUENCES);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/*
@@ -676,6 +689,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(opts.enabled);
 	values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(opts.binary);
 	values[Anum_pg_subscription_substream - 1] = CharGetDatum(opts.streaming);
+	values[Anum_pg_subscription_subsequences - 1] = CharGetDatum(opts.sequences);
 	values[Anum_pg_subscription_subtwophasestate - 1] =
 		CharGetDatum(opts.twophase ?
 					 LOGICALREP_TWOPHASE_STATE_PENDING :
@@ -1163,6 +1177,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					replaces[Anum_pg_subscription_subbinary - 1] = true;
 				}
 
+				if (IsSet(opts.specified_opts, SUBOPT_SEQUENCES))
+				{
+					values[Anum_pg_subscription_subsequences - 1] =
+						BoolGetDatum(opts.sequences);
+					replaces[Anum_pg_subscription_subsequences - 1] = true;
+				}
+
 				if (IsSet(opts.specified_opts, SUBOPT_STREAMING))
 				{
 					values[Anum_pg_subscription_substream - 1] =
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 052505e46f..6f636647ca 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -461,6 +461,9 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
 			appendStringInfo(&cmd, ", streaming '%s'",
 							 options->proto.logical.streaming_str);
 
+		if (options->proto.logical.sequences)
+			appendStringInfoString(&cmd, ", sequences 'on'");
+
 		if (options->proto.logical.twophase &&
 			PQserverVersion(conn->streamConn) >= 150000)
 			appendStringInfoString(&cmd, ", two_phase 'on'");
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index cd2f88e0bd..5dd94779dc 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -1329,6 +1329,21 @@ sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	RepOriginId origin_id = XLogRecGetOrigin(r);
 	bool		transactional;
 
+	/*
+	 * ignore sequences when not requested
+	 *
+	 * XXX Maybe we should differentiate between "callbacks not defined" or
+	 * "subscriber disabled sequence replication" and "subscriber does not
+	 * know about sequence replication" (e.g. old subscriber version).
+	 *
+	 * For the first two it'd be fine to bail out here, but for the last it
+	 * might be better to continue and error out only when the sequence
+	 * would be replicated (e.g. as part of the publication). We don't know
+	 * that here, unfortunately.
+	 */
+	if (!ctx->sequences)
+		return;
+
 	/* only decode changes flagged with XLOG_SEQ_LOG */
 	if (info != XLOG_SEQ_LOG)
 		elog(ERROR, "unexpected RM_SEQ_ID record type: %u", info);
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 675cf12366..0d1497f66b 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -300,6 +300,20 @@ StartupDecodingContext(List *output_plugin_options,
 	 */
 	ctx->reorder->update_progress_txn = update_progress_txn_cb_wrapper;
 
+	/*
+	 * To support logical decoding of sequences, we require the sequence
+	 * callback. We decide it here, but only check it later in the wrappers.
+	 *
+	 * XXX Isn't it wrong to define only one of those callbacks? Say we
+	 * only define the stream_sequence_cb() - that may get strange results
+	 * depending on what gets streamed. Either none or both?
+	 *
+	 * XXX Shouldn't sequence be defined at slot creation time, similar
+	 * to two_phase? Probably not.
+	 */
+	ctx->sequences = ((ctx->callbacks.sequence_cb != NULL) ||
+					  (ctx->callbacks.stream_sequence_cb != NULL));
+
 	ctx->out = makeStringInfo();
 	ctx->prepare_write = prepare_write;
 	ctx->write = do_write;
@@ -566,6 +580,16 @@ CreateDecodingContext(XLogRecPtr start_lsn,
 		startup_cb_wrapper(ctx, &ctx->options, false);
 	MemoryContextSwitchTo(old_context);
 
+	/*
+	 * We allow decoding of sequences when the option is given at the streaming
+	 * start, provided the plugin supports all the callbacks for two-phase.
+	 *
+	 * XXX Similar behavior to the two-phase block below.
+	 *
+	 * XXX Shouldn't this error out if the callbacks are not defined?
+	 */
+	ctx->sequences &= ctx->sequences_opt_given;
+
 	/*
 	 * We allow decoding of prepared transactions when the two_phase is
 	 * enabled at the time of slot creation, or when the two_phase option is
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 28059f9e97..c129123bc4 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -4664,7 +4664,7 @@ ApplyWorkerMain(Datum main_arg)
 
 	server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
 	options.proto.logical.proto_version =
-		server_version >= 160000 ? LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM :
+		server_version >= 160000 ? LOGICALREP_PROTO_SEQUENCES_VERSION_NUM :
 		server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
 		server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
 		LOGICALREP_PROTO_VERSION_NUM;
@@ -4694,9 +4694,18 @@ ApplyWorkerMain(Datum main_arg)
 		MyLogicalRepWorker->parallel_apply = false;
 	}
 
+	options.proto.logical.sequences = false;
 	options.proto.logical.twophase = false;
 	options.proto.logical.origin = pstrdup(MySubscription->origin);
 
+	/*
+	 * Assign the appropriate option value for sequence decoding option according
+	 * to the 'sequences' mode and the publisher's ability to support that mode.
+	 */
+	if (server_version >= 160000 &&
+		MySubscription->sequences)
+		options.proto.logical.sequences = true;
+
 	if (!am_tablesync_worker())
 	{
 		/*
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index af037ec546..50dbd66068 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -299,7 +299,7 @@ parse_output_parameters(List *options, PGOutputData *data)
 	data->streaming = LOGICALREP_STREAM_OFF;
 	data->messages = false;
 	data->two_phase = false;
-	data->sequences = true;
+	data->sequences = false;
 
 	foreach(lc, options)
 	{
@@ -522,6 +522,27 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 		else
 			ctx->twophase_opt_given = true;
 
+		/*
+		 * Here, we just check whether the sequences decoding option is passed
+		 * by plugin and decide whether to enable it at later point of time. It
+		 * remains enabled if the previous start-up has done so. But we only
+		 * allow the option to be passed in with sufficient version of the
+		 * protocol, and when the output plugin supports it.
+		 */
+		if (!data->sequences)
+			ctx->sequences_opt_given = false;
+		else if (data->protocol_version < LOGICALREP_PROTO_SEQUENCES_VERSION_NUM)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("requested proto_version=%d does not support sequences, need %d or higher",
+							data->protocol_version, LOGICALREP_PROTO_SEQUENCES_VERSION_NUM)));
+		else if (!ctx->sequences)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("sequences requested, but not supported by output plugin")));
+		else
+			ctx->sequences_opt_given = true;
+
 		/* Init publication state. */
 		data->publications = NIL;
 		publications_valid = false;
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 6319f598d8..feca2b02ea 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -83,6 +83,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
 	char		substream;		/* Stream in-progress transactions. See
 								 * LOGICALREP_STREAM_xxx constants. */
 
+	bool		subsequences;	/* True if the subscription wants the
+								 * publisher to send sequence data */
+
 	char		subtwophasestate;	/* Stream two-phase transactions */
 
 	bool		subdisableonerr;	/* True if a worker error should cause the
@@ -130,6 +133,7 @@ typedef struct Subscription
 	char		stream;			/* Allow streaming in-progress transactions.
 								 * See LOGICALREP_STREAM_xxx constants. */
 	char		twophasestate;	/* Allow streaming two-phase transactions */
+	bool		sequences;		/* Allow replication of sequence increments. */
 	bool		disableonerr;	/* Indicates if the subscription should be
 								 * automatically disabled if a worker error
 								 * occurs */
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 5f49554ea0..df31b1276f 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -90,6 +90,12 @@ typedef struct LogicalDecodingContext
 	 */
 	bool		twophase;
 
+	/*
+	 * Does the output pluging support decoding of sequence increments, and
+	 * is it enabled?
+	 */
+	bool		sequences;
+
 	/*
 	 * Is two-phase option given by output plugin?
 	 *
@@ -100,6 +106,15 @@ typedef struct LogicalDecodingContext
 	 */
 	bool		twophase_opt_given;
 
+	/*
+	 * Is sequences option given by output plugin?
+	 *
+	 * This indicates the plugin passed the sequences option as part of the
+	 * START_STREAMING command.  We can't rely solely on the sequences flag
+	 * which only tells whether the plugin provided the necessary callback.
+	 */
+	bool		sequences_opt_given;
+
 	/*
 	 * State for writing output.
 	 */
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 6d82d5b6cc..25de258094 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -42,7 +42,8 @@
 #define LOGICALREP_PROTO_STREAM_VERSION_NUM 2
 #define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM 3
 #define LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM 4
-#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM
+#define LOGICALREP_PROTO_SEQUENCES_VERSION_NUM 5
+#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_SEQUENCES_VERSION_NUM
 
 /*
  * Logical message types
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 281626fa6f..229cb3ca58 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -185,6 +185,7 @@ typedef struct
 			char	   *streaming_str;	/* Streaming of large transactions */
 			bool		twophase;	/* Streaming of two-phase transactions at
 									 * prepare time */
+			bool		sequences;	/* Replication of sequences. */
 			char	   *origin; /* Only publish data originating from the
 								 * specified origin */
 		}			logical;
-- 
2.39.2