0005-Simplify-protocol-versioning-20230720.patch

From c262ef2bee45672eafdfd9fa87b2ca15b0dbc69b Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas.vondra@postgresql.org>
Date: Wed, 12 Jul 2023 20:08:04 +0200
Subject: [PATCH 5/6] Simplify protocol versioning

https://www.postgresql.org/message-id/8046273f-ea88-5c97-5540-0ccd5d244fd4@enterprisedb.com
---
 contrib/test_decoding/test_decoding.c         | 18 ++++----
 src/backend/catalog/pg_subscription.c         |  1 -
 src/backend/commands/subscriptioncmds.c       | 23 +---------
 .../libpqwalreceiver/libpqwalreceiver.c       |  8 ----
 src/backend/replication/logical/logical.c     | 12 -----
 src/backend/replication/logical/worker.c      | 12 -----
 src/backend/replication/pgoutput/pgoutput.c   | 45 ++++---------------
 src/include/catalog/pg_subscription.h         |  4 --
 src/include/replication/logical.h             |  9 ----
 src/include/replication/pgoutput.h            |  1 -
 src/include/replication/walreceiver.h         |  1 -
 11 files changed, 20 insertions(+), 114 deletions(-)

diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
index 60ccbacadf8..9c40702da7e 100644
--- a/contrib/test_decoding/test_decoding.c
+++ b/contrib/test_decoding/test_decoding.c
@@ -29,6 +29,7 @@ typedef struct
 	MemoryContext context;
 	bool		include_xids;
 	bool		include_timestamp;
+	bool		include_sequences;
 	bool		skip_empty_xacts;
 	bool		only_local;
 } TestDecodingData;
@@ -169,7 +170,6 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 	ListCell   *option;
 	TestDecodingData *data;
 	bool		enable_streaming = false;
-	bool		enable_sequences = false;
 
 	data = palloc0(sizeof(TestDecodingData));
 	data->context = AllocSetContextCreate(ctx->context,
@@ -179,6 +179,7 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 	data->include_timestamp = false;
 	data->skip_empty_xacts = false;
 	data->only_local = false;
+	data->include_sequences = false;
 
 	ctx->output_plugin_private = data;
 
@@ -272,10 +273,9 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 		}
 		else if (strcmp(elem->defname, "include-sequences") == 0)
 		{
-
 			if (elem->arg == NULL)
-				continue;
-			else if (!parse_bool(strVal(elem->arg), &enable_sequences))
+				data->include_sequences = true;
+			else if (!parse_bool(strVal(elem->arg), &data->include_sequences))
 				ereport(ERROR,
 						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
@@ -291,11 +291,7 @@ pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 		}
 	}
 
-	/* remember the user explicitly requested sequences, otherwise the */
-	ctx->sequences_opt_given = enable_sequences;
-
 	ctx->streaming &= enable_streaming;
-	ctx->sequences &= enable_sequences;
 }
 
 /* cleanup this plugin's resources */
@@ -797,6 +793,9 @@ pg_decode_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	TestDecodingData *data = ctx->output_plugin_private;
 	TestDecodingTxnData *txndata = txn->output_plugin_private;
 
+	if (!data->include_sequences)
+		return;
+
 	/* output BEGIN if we haven't yet, but only for the transactional case */
 	if (transactional)
 	{
@@ -1025,6 +1024,9 @@ pg_decode_stream_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	TestDecodingData *data = ctx->output_plugin_private;
 	TestDecodingTxnData *txndata = txn->output_plugin_private;
 
+	if (!data->include_sequences)
+		return;
+
 	/* output BEGIN if we haven't yet, but only for the transactional case */
 	if (transactional)
 	{
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 49aa3b9f533..d07f88ce280 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -68,7 +68,6 @@ 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 7078b61c773..8be85c8659c 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -71,7 +71,6 @@
 #define SUBOPT_RUN_AS_OWNER			0x00001000
 #define SUBOPT_LSN					0x00002000
 #define SUBOPT_ORIGIN				0x00004000
-#define SUBOPT_SEQUENCES			0x00004000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -93,7 +92,6 @@ typedef struct SubOpts
 	bool		binary;
 	char		streaming;
 	bool		twophase;
-	bool		sequences;
 	bool		disableonerr;
 	bool		passwordrequired;
 	bool		runasowner;
@@ -148,8 +146,6 @@ 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 = true;
 	if (IsSet(supported_opts, SUBOPT_STREAMING))
 		opts->streaming = LOGICALREP_STREAM_OFF;
 	if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
@@ -260,15 +256,6 @@ 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)
 		{
 			/*
@@ -605,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_RUN_AS_OWNER | SUBOPT_ORIGIN | SUBOPT_SEQUENCES);
+					  SUBOPT_RUN_AS_OWNER | SUBOPT_ORIGIN);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/*
@@ -704,7 +691,6 @@ 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 :
@@ -1194,13 +1180,6 @@ 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 f8420ec4b4b..60d5c1fc403 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -461,14 +461,6 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
 			appendStringInfo(&cmd, ", streaming '%s'",
 							 options->proto.logical.streaming_str);
 
-		if (PQserverVersion(conn->streamConn) >= 170000)
-		{
-			if (options->proto.logical.sequences)
-				appendStringInfoString(&cmd, ", sequences 'on'");
-			else
-				appendStringInfoString(&cmd, ", sequences 'off'");
-		}
-
 		if (options->proto.logical.twophase &&
 			PQserverVersion(conn->streamConn) >= 150000)
 			appendStringInfoString(&cmd, ", two_phase 'on'");
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index d648a126283..74ba6fd861e 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -605,18 +605,6 @@ 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? That is,
-	 * if the client requests sequences for pluging that does not support
-	 * sequences, maybe we should error-out instead of silently ignoring it.
-	 */
-	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 d5220398d25..7ebd13604c5 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -4741,21 +4741,9 @@ 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.
-	 *
-	 * XXX Isn't this redundant with the version check in libpqwalreceiver.c, using
-	 * PQserverVersion(conn->streamConn)?
-	 */
-	if (server_version >= 170000 &&
-		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 afa420a88cc..80262adf3e5 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -290,7 +290,6 @@ parse_output_parameters(List *options, PGOutputData *data)
 	bool		publication_names_given = false;
 	bool		binary_option_given = false;
 	bool		messages_option_given = false;
-	bool		sequences_option_given = false;
 	bool		streaming_given = false;
 	bool		two_phase_option_given = false;
 	bool		origin_option_given = false;
@@ -299,7 +298,6 @@ parse_output_parameters(List *options, PGOutputData *data)
 	data->streaming = LOGICALREP_STREAM_OFF;
 	data->messages = false;
 	data->two_phase = false;
-	data->sequences = false;
 
 	foreach(lc, options)
 	{
@@ -368,16 +366,6 @@ parse_output_parameters(List *options, PGOutputData *data)
 
 			data->messages = defGetBoolean(defel);
 		}
-		else if (strcmp(defel->defname, "sequences") == 0)
-		{
-			if (sequences_option_given)
-				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("conflicting or redundant options")));
-			sequences_option_given = true;
-
-			data->sequences = defGetBoolean(defel);
-		}
 		else if (strcmp(defel->defname, "streaming") == 0)
 		{
 			if (streaming_given)
@@ -522,27 +510,6 @@ 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;
@@ -1716,12 +1683,18 @@ pgoutput_sequence(LogicalDecodingContext *ctx,
 	TransactionId xid = InvalidTransactionId;
 	RelationSyncEntry *relentry;
 
-	if (!data->sequences)
-		return;
-
 	if (!is_publishable_relation(relation))
 		return;
 
+	/*
+	 * If the negotiated protocol version does not support sequences, yet we
+	 * found a sequence in the publication, error out.
+	 */
+	if (data->protocol_version < LOGICALREP_PROTO_SEQUENCES_VERSION_NUM)
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				errmsg("protocol version does not support sequence replication")));
+
 	/*
 	 * Remember the xid for the message in streaming mode. See
 	 * pgoutput_change.
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 047e86fcd9b..1d40eebc789 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -83,9 +83,6 @@ 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
@@ -136,7 +133,6 @@ 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 df31b1276f6..c62ac4247b0 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -106,15 +106,6 @@ 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/pgoutput.h b/src/include/replication/pgoutput.h
index 56dfa2a7417..b4a8015403b 100644
--- a/src/include/replication/pgoutput.h
+++ b/src/include/replication/pgoutput.h
@@ -30,7 +30,6 @@ typedef struct PGOutputData
 	bool		messages;
 	bool		two_phase;
 	char	   *origin;
-	bool		sequences;
 } PGOutputData;
 
 #endif							/* PGOUTPUT_H */
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 229cb3ca589..281626fa6f5 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -185,7 +185,6 @@ 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.41.0