v6-0002-Support-force-option-for-copy_data-check-and-thro.patch

From 04e6cbca836173f689906c77941bf120e860fe0a Mon Sep 17 00:00:00 2001
From: Vigneshwaran C <vignesh21@gmail.com>
Date: Fri, 1 Apr 2022 12:05:34 +0530
Subject: [PATCH v6 2/2] Support force option for copy_data, check and throw an
 error if publisher tables were also subscribing data in the publisher from
 other publishers.

This patch does couple of things:
1) Added force option for copy_data.
2) Check and throw an error if the publication tables were also subscribing
data in the publisher from other publishers.

Let's consider an existing Multi master logical replication setup between
Node1 and Node2 that is created using the following steps:
a) Node1 - Publication publishing employee table.
b) Node2 - Subscription subscribing from publication pub1 with
subscribe_local_only.
c) Node2 - Publication publishing employee table.
d) Node1 - Subscription subscribing from publication pub2 with
subscribe_local_only. Now when user is trying to add another node Node3 to the
above Multi master logical replication setup, user will have to create one
subscription subscribing from Node1 and another subscription subscribing from
Node2 in Node3 using subscribe_local_only option and copy_data as true, while
the subscription is created, server will identify that Node2 is subscribing
from Node1 and Node1 is subscribing from Node2 and throw an error so
that user can handle the initial copy data.
---
 doc/src/sgml/ref/alter_subscription.sgml   |   8 +-
 doc/src/sgml/ref/create_subscription.sgml  |  59 +++++++-
 src/backend/commands/subscriptioncmds.c    | 159 +++++++++++++++++----
 src/test/regress/expected/subscription.out |   4 +-
 src/test/regress/sql/subscription.sql      |   1 +
 src/test/subscription/t/032_circular.pl    |  24 +++-
 src/tools/pgindent/typedefs.list           |   1 +
 7 files changed, 220 insertions(+), 36 deletions(-)

diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 8f667c916e..7cb485b0ec 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -161,12 +161,14 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
 
       <variablelist>
        <varlistentry>
-        <term><literal>copy_data</literal> (<type>boolean</type>)</term>
+        <term><literal>copy_data</literal> (<type>enum</type>)</term>
         <listitem>
          <para>
           Specifies whether to copy pre-existing data in the publications
-          that are being subscribed to when the replication starts.
-          The default is <literal>true</literal>.
+          that are being subscribed to when the replication starts. This
+          parameter may be either <literal>true</literal>,
+          <literal>false</literal> or <literal>force</literal>. The default is
+          <literal>true</literal>.
          </para>
          <para>
           Previously subscribed tables and sequences are not copied, even if a
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 36dfbc888b..a8e1aedfbe 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -161,6 +161,13 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
           the replicated changes that was generated from other nodes. The
           default is <literal>false</literal>.
          </para>
+         <para>
+          If the tables in the publication were also subscribing to the data in
+          the publisher from other publishers, it will affect the
+          <command>CREATE SUBSCRIPTION</command> based on the value specified
+          for <literal>copy_data</literal> option. Refer to the
+          <xref linkend="sql-createsubscription-notes" /> for details.
+         </para>
         </listitem>
        </varlistentry>
 
@@ -213,18 +220,28 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
        </varlistentry>
 
        <varlistentry>
-        <term><literal>copy_data</literal> (<type>boolean</type>)</term>
+        <term><literal>copy_data</literal> (<type>enum</type>)</term>
         <listitem>
          <para>
           Specifies whether to copy pre-existing data in the publications
-          that are being subscribed to when the replication starts.
-          The default is <literal>true</literal>.
+          that are being subscribed to when the replication starts. This
+          parameter may be either <literal>true</literal>,
+          <literal>false</literal> or <literal>force</literal>. The default is
+          <literal>true</literal>.           
          </para>
          <para>
           If the publications contain <literal>WHERE</literal> clauses, it
           will affect what data is copied. Refer to the
           <xref linkend="sql-createsubscription-notes" /> for details.
          </para>
+
+         <para>
+          If the tables in the publication were also subscribing to the data in
+          the publisher from other publishers, it will affect the
+          <command>CREATE SUBSCRIPTION</command> based on the value specified
+          for <literal>subscribe_local_only</literal> option. Refer to the
+          <xref linkend="sql-createsubscription-notes" /> for details.
+         </para>
         </listitem>
        </varlistentry>
 
@@ -375,6 +392,42 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
    can have non-existent publications.
   </para>
 
+  <para>
+   Let's consider an existing Multi master logical replication setup between
+   Node1 and Node2 that is created using the following steps:
+   a) Node1 - Publication publishing employee table.
+   b) Node2 - Subscription subscribing from publication pub1 with
+   <literal>subscribe_local_only</literal>.
+   c) Node2 - Publication publishing employee table.
+   d) Node1 - Subscription subscribing from publication pub2 with
+   <literal>subscribe_local_only</literal>.
+   Now when user is trying to add another node Node3 to the above Multi master
+   logical replication setup, user will have to create one subscription
+   subscribing from Node1 and another subscription subscribing from Node2 in
+   Node3 using <literal>subscribe_local_only</literal> option and
+   <literal>copy_data</literal> as <literal>true</literal>, while
+   the subscription is created, server will identify that Node2 is subscribing
+   from Node1 and Node1 is subscribing from Node2 and throw an error like:
+   <programlisting>
+postgres=# CREATE SUBSCRIPTION mysub CONNECTION 'host=192.168.1.50 port=5432 user=foo dbname=foodb' 
+           PUBLICATION mypublication, insert_only with (subscribe_local_only=on);
+ERROR:  CREATE/ALTER SUBSCRIPTION with subscribe_local_only and copy_data as true is not allowed when the publisher might have replicated data, table:public.t1 might have replicated data in the publisher
+HINT:  Use CREATE/ALTER SUBSCRIPTION with copy_data = off or force
+   </programlisting>
+   In this scenario user can solve this based on one of the 2 possibilities,
+   a) If there are no data present in Node1 and Node2, then the user can create
+   the subscriptions to Node1 and Node2 with
+   <literal>subscribe_local_only</literal> as <literal>true</literal> and
+   <literal>copy_data</literal> as <literal>false</literal>. b) If the data is
+   present, then the user can create subscription with
+   <literal>copy_data</literal> as <literal>force</literal> on Node1 and
+   <literal>copy_data</literal> as <literal>false</literal> on Node2, before
+   allowing any operations on the respective tables of Node1 and Node2, in this
+   case <literal>copy_data</literal> is <literal>false</literal> on Node2
+   because the data will be replicated to each other and available on both the
+   nodes.
+  </para>
+
  </refsect1>
 
  <refsect1>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 55c48af2c9..202ea72189 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -69,6 +69,18 @@
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
 
+#define IS_COPY_DATA_VALID(copy_data) (copy_data != COPY_DATA_OFF)
+
+/*
+ * Represents whether copy_data option is specified with on, off or force.
+ */
+typedef enum CopyData
+{
+	COPY_DATA_OFF,
+	COPY_DATA_ON,
+	COPY_DATA_FORCE
+} CopyData;
+
 /*
  * Structure to hold a bitmap representing the user-provided CREATE/ALTER
  * SUBSCRIPTION command options and the parsed/default values of each of them.
@@ -81,7 +93,7 @@ typedef struct SubOpts
 	bool		connect;
 	bool		enabled;
 	bool		create_slot;
-	bool		copy_data;
+	CopyData	copy_data;
 	bool		refresh;
 	bool		binary;
 	bool		streaming;
@@ -91,12 +103,69 @@ typedef struct SubOpts
 	XLogRecPtr	lsn;
 } SubOpts;
 
-static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
+static List *fetch_table_list(WalReceiverConn *wrconn, List *publications,
+							  CopyData copydata, bool only_local);
 static List *fetch_sequence_list(WalReceiverConn *wrconn, List *publications);
 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);
 
+/*
+ * Validate the value specified for copy_data option.
+ */
+static CopyData
+DefGetCopyData(DefElem *def)
+{
+	/*
+	 * If no parameter given, assume "true" is meant.
+	 */
+	if (def->arg == NULL)
+		return COPY_DATA_ON;
+
+	/*
+	 * Allow 0, 1, "true", "false", "on", "off" or "force".
+	 */
+	switch (nodeTag(def->arg))
+	{
+		case T_Integer:
+			switch (intVal(def->arg))
+			{
+				case 0:
+					return COPY_DATA_OFF;
+				case 1:
+					return COPY_DATA_ON;
+				default:
+					/* otherwise, error out below */
+					break;
+			}
+			break;
+		default:
+		{
+			char    *sval = defGetString(def);
+
+			/*
+			 * The set of strings accepted here should match up with
+			 * the grammar's opt_boolean_or_string production.
+			 */
+			if (pg_strcasecmp(sval, "true") == 0)
+				return COPY_DATA_ON;
+			if (pg_strcasecmp(sval, "false") == 0)
+				return COPY_DATA_OFF;
+			if (pg_strcasecmp(sval, "on") == 0)
+				return COPY_DATA_ON;
+			if (pg_strcasecmp(sval, "off") == 0)
+				return COPY_DATA_OFF;
+			if (pg_strcasecmp(sval, "force") == 0)
+				return COPY_DATA_FORCE;
+		}
+		break;
+	}
+
+	ereport(ERROR,
+			errcode(ERRCODE_SYNTAX_ERROR),
+			errmsg("%s requires a boolean or \"force\"", def->defname));
+	return COPY_DATA_OFF;                                           /* keep compiler quiet */
+}
 
 /*
  * Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
@@ -129,7 +198,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 	if (IsSet(supported_opts, SUBOPT_CREATE_SLOT))
 		opts->create_slot = true;
 	if (IsSet(supported_opts, SUBOPT_COPY_DATA))
-		opts->copy_data = true;
+		opts->copy_data = COPY_DATA_ON;
 	if (IsSet(supported_opts, SUBOPT_REFRESH))
 		opts->refresh = true;
 	if (IsSet(supported_opts, SUBOPT_BINARY))
@@ -197,7 +266,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 				errorConflictingDefElem(defel, pstate);
 
 			opts->specified_opts |= SUBOPT_COPY_DATA;
-			opts->copy_data = defGetBoolean(defel);
+			opts->copy_data = DefGetCopyData(defel);
 		}
 		else if (IsSet(supported_opts, SUBOPT_SYNCHRONOUS_COMMIT) &&
 				 strcmp(defel->defname, "synchronous_commit") == 0)
@@ -334,17 +403,17 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 					 errmsg("%s and %s are mutually exclusive options",
 							"connect = false", "create_slot = true")));
 
-		if (opts->copy_data &&
+		if (IS_COPY_DATA_VALID(opts->copy_data) &&
 			IsSet(opts->specified_opts, SUBOPT_COPY_DATA))
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR),
 					 errmsg("%s and %s are mutually exclusive options",
-							"connect = false", "copy_data = true")));
+							"connect = false", "copy_data = true/force")));
 
 		/* Change the defaults of other options. */
 		opts->enabled = false;
 		opts->create_slot = false;
-		opts->copy_data = false;
+		opts->copy_data = COPY_DATA_OFF;
 	}
 
 	/*
@@ -672,13 +741,15 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 			 * Set sync state based on if we were asked to do data copy or
 			 * not.
 			 */
-			sync_state = opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
+			sync_state = IS_COPY_DATA_VALID(opts.copy_data) ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
 
 			/*
 			 * Get the table and sequence list from publisher and build
 			 * local relation sync status info.
 			 */
-			relations = fetch_table_list(wrconn, publications);
+			relations = fetch_table_list(wrconn, publications, opts.copy_data,
+										 opts.only_local);
+
 			relations = list_concat(relations,
 									fetch_sequence_list(wrconn, publications));
 
@@ -724,7 +795,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 				 * as PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
 				 * PUBLICATION to work.
 				 */
-				if (opts.twophase && !opts.copy_data && relations != NIL)
+				if (opts.twophase && IS_COPY_DATA_VALID(opts.copy_data) &&
+					relations != NIL)
 					twophase_enabled = true;
 
 				walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
@@ -763,8 +835,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 }
 
 static void
-AlterSubscription_refresh(Subscription *sub, bool copy_data,
-						  List *validate_publications)
+AlterSubscription_refresh(Subscription *sub, CopyData copy_data,
+						  List *validate_publications, bool only_local)
 {
 	char	   *err;
 	List	   *pubrel_names;
@@ -799,7 +871,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 			check_publications(wrconn, validate_publications);
 
 		/* Get the list of relations from publisher. */
-		pubrel_names = fetch_table_list(wrconn, sub->publications);
+		pubrel_names = fetch_table_list(wrconn, sub->publications, copy_data,
+										only_local);
 		pubrel_names = list_concat(pubrel_names,
 								   fetch_sequence_list(wrconn, sub->publications));
 
@@ -855,7 +928,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 						 list_length(subrel_states), sizeof(Oid), oid_cmp))
 			{
 				AddSubscriptionRelState(sub->oid, relid,
-										copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
+										IS_COPY_DATA_VALID(copy_data) ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
 										InvalidXLogRecPtr);
 				ereport(DEBUG1,
 						(errmsg_internal("table \"%s.%s\" added to subscription \"%s\"",
@@ -1138,7 +1211,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 		case ALTER_SUBSCRIPTION_SET_PUBLICATION:
 			{
-				supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
+				supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH | SUBOPT_LOCAL_ONLY;
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
 
@@ -1161,7 +1234,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					 * See ALTER_SUBSCRIPTION_REFRESH for details why this is
 					 * not allowed.
 					 */
-					if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
+					if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && IS_COPY_DATA_VALID(opts.copy_data))
 						ereport(ERROR,
 								(errcode(ERRCODE_SYNTAX_ERROR),
 								 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
@@ -1174,7 +1247,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					sub->publications = stmt->publication;
 
 					AlterSubscription_refresh(sub, opts.copy_data,
-											  stmt->publication);
+											  stmt->publication,
+											  opts.only_local);
 				}
 
 				break;
@@ -1186,7 +1260,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				List	   *publist;
 				bool		isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;
 
-				supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
+				supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA | SUBOPT_LOCAL_ONLY;
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
 
@@ -1226,7 +1300,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					sub->publications = publist;
 
 					AlterSubscription_refresh(sub, opts.copy_data,
-											  validate_publications);
+											  validate_publications,
+											  opts.only_local);
 				}
 
 				break;
@@ -1240,7 +1315,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 							 errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));
 
 				parse_subscription_options(pstate, stmt->options,
-										   SUBOPT_COPY_DATA, &opts);
+										   SUBOPT_COPY_DATA | SUBOPT_LOCAL_ONLY,
+										   &opts);
 
 				/*
 				 * The subscription option "two_phase" requires that
@@ -1259,7 +1335,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				 *
 				 * For more details see comments atop worker.c.
 				 */
-				if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
+				if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+					IS_COPY_DATA_VALID(opts.copy_data))
 					ereport(ERROR,
 							(errcode(ERRCODE_SYNTAX_ERROR),
 							 errmsg("ALTER SUBSCRIPTION ... REFRESH with copy_data is not allowed when two_phase is enabled"),
@@ -1268,7 +1345,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 				PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH");
 
-				AlterSubscription_refresh(sub, opts.copy_data, NULL);
+				AlterSubscription_refresh(sub, opts.copy_data, NULL,
+										  opts.only_local);
 
 				break;
 			}
@@ -1783,22 +1861,28 @@ AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId)
  * publisher connection.
  */
 static List *
-fetch_table_list(WalReceiverConn *wrconn, List *publications)
+fetch_table_list(WalReceiverConn *wrconn, List *publications, CopyData copydata,
+				 bool only_local)
 {
 	WalRcvExecResult *res;
 	StringInfoData cmd;
 	TupleTableSlot *slot;
-	Oid			tableRow[2] = {TEXTOID, TEXTOID};
+	Oid			tableRow[3] = {TEXTOID, TEXTOID, CHAROID};
 	List	   *tablelist = NIL;
 
 	initStringInfo(&cmd);
-	appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n"
-						   "  FROM pg_catalog.pg_publication_tables t\n"
-						   " WHERE t.pubname IN (");
+	appendStringInfoString(&cmd,
+						   "SELECT DISTINCT N.nspname AS schemaname, C.relname AS tablename, PS.srsubstate as replicated\n"
+						   "FROM pg_publication P,\n"
+						   "LATERAL pg_get_publication_tables(P.pubname) GPT\n"
+						   "LEFT JOIN pg_subscription_rel PS ON (GPT.relid = PS.srrelid),\n"
+						   "pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
+						   "WHERE C.oid = GPT.relid AND P.pubname in (");
+
 	get_publications_str(publications, &cmd, true);
 	appendStringInfoChar(&cmd, ')');
 
-	res = walrcv_exec(wrconn, cmd.data, 2, tableRow);
+	res = walrcv_exec(wrconn, cmd.data, 3, tableRow);
 	pfree(cmd.data);
 
 	if (res->status != WALRCV_OK_TUPLES)
@@ -1824,6 +1908,25 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
 		rv = makeRangeVar(nspname, relname, -1);
 		tablelist = lappend(tablelist, rv);
 
+		/*
+		 * XXX: During initial table sync we cannot differentiate between the
+		 * local and non-local data that is present in the HEAP. Identification
+		 * of local data can be done only from the WAL by using the origin id.
+		 * Throw an error so that the user can take care of the initial data
+		 * copying and then create subscription with copy_data off.
+		 *
+		 * It is quite possible that subscriber has not yet pulled data to
+		 * the tables, but in ideal cases the table data will be subscribed.
+		 * Too keep the code simple it is not checked if the subscriber table
+		 * has pulled the data or not.
+		 */
+		if (copydata == COPY_DATA_ON && only_local && !slot_attisnull(slot, 3))
+			ereport(ERROR,
+					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("CREATE/ALTER SUBSCRIPTION with subscribe_local_only and copy_data as true is not allowed when the publisher might have replicated data, table:%s.%s might have replicated data in the publisher",
+						   nspname, relname),
+					errhint("Use CREATE/ALTER SUBSCRIPTION with copy_data = off or force"));
+
 		ExecClearTuple(slot);
 	}
 	ExecDropSingleTupleTableSlot(slot);
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 17b991b1fc..fad2e6b3aa 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -47,7 +47,9 @@ ERROR:  must be superuser to create subscriptions
 SET SESSION AUTHORIZATION 'regress_subscription_user';
 -- fail - invalid option combinations
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = true);
-ERROR:  connect = false and copy_data = true are mutually exclusive options
+ERROR:  connect = false and copy_data = true/force are mutually exclusive options
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = force);
+ERROR:  connect = false and copy_data = true/force are mutually exclusive options
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, enabled = true);
 ERROR:  connect = false and enabled = true are mutually exclusive options
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, create_slot = true);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index badc00b7c7..dfe36e000e 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -40,6 +40,7 @@ SET SESSION AUTHORIZATION 'regress_subscription_user';
 
 -- fail - invalid option combinations
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = true);
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = force);
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, enabled = true);
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, create_slot = true);
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, enabled = true);
diff --git a/src/test/subscription/t/032_circular.pl b/src/test/subscription/t/032_circular.pl
index 30e14bf727..6852372f8d 100644
--- a/src/test/subscription/t/032_circular.pl
+++ b/src/test/subscription/t/032_circular.pl
@@ -42,6 +42,10 @@ $node_A->safe_psql('postgres',
 $node_B->safe_psql('postgres',
 	"CREATE TABLE tab_full (a int PRIMARY KEY)");
 
+my $result;
+my $stdout;
+my $stderr;
+
 # Setup logical replication
 # node_A (pub) -> node_B (sub)
 my $node_A_connstr = $node_A->connstr . ' dbname=postgres';
@@ -65,6 +69,25 @@ $node_A->safe_psql('postgres',	"
 	PUBLICATION tap_pub_B
 	WITH (subscribe_local_only = on, copy_data = off)");
 
+($result, $stdout, $stderr) = $node_A->psql('postgres',  "
+        CREATE SUBSCRIPTION tap_sub_A1
+        CONNECTION '$node_B_connstr application_name=$appname_A'
+        PUBLICATION tap_pub_B
+        WITH (subscribe_local_only = on, copy_data = on)");
+like(
+        $stderr,
+        qr/ERROR:  CREATE\/ALTER SUBSCRIPTION with subscribe_local_only and copy_data as true is not allowed when the publisher might have replicated data/,
+        "Create subscription with subscribe_local_only and copy_data having replicated table in publisher");
+
+$node_A->safe_psql('postgres',  "
+        CREATE SUBSCRIPTION tap_sub_A2
+        CONNECTION '$node_B_connstr application_name=$appname_A'
+        PUBLICATION tap_pub_B
+        WITH (subscribe_local_only = on, copy_data = force)");
+
+$node_A->safe_psql('postgres',  "
+        DROP SUBSCRIPTION tap_sub_A2");
+
 # Wait for subscribers to finish initialization
 $node_A->wait_for_catchup($appname_B);
 $node_B->wait_for_catchup($appname_A);
@@ -79,7 +102,6 @@ $node_B->poll_query_until('postgres', $synced_query)
 
 is(1,1, "Circular replication setup is complete");
 
-my $result;
 
 ##########################################################################
 # check that circular replication setup does not cause infinite recursive
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 72fafb795b..bd2e84a91b 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -442,6 +442,7 @@ ConvProcInfo
 ConversionLocation
 ConvertRowtypeExpr
 CookedConstraint
+CopyData
 CopyDest
 CopyFormatOptions
 CopyFromState
-- 
2.32.0