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

From 5b3264a4c368d1a06098c525be39f75ef3680059 Mon Sep 17 00:00:00 2001
From: Vigneshwaran C <vignesh21@gmail.com>
Date: Sat, 12 Mar 2022 09:31:22 +0530
Subject: [PATCH v4 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
publish_local_only.

If a subscription is created to Node1 from Node3 with publish_local_only
and copy_data as ON, then throw an error so that user can handle
creation of subscription with table having consistent data.
---
 doc/src/sgml/ref/alter_subscription.sgml   |   2 +-
 doc/src/sgml/ref/create_subscription.sgml  |  54 +++++++-
 src/backend/commands/subscriptioncmds.c    | 150 +++++++++++++++++----
 src/test/regress/expected/subscription.out |   4 +-
 src/test/regress/sql/subscription.sql      |   1 +
 src/test/subscription/t/029_circular.pl    |  24 +++-
 src/tools/pgindent/typedefs.list           |   1 +
 7 files changed, 205 insertions(+), 31 deletions(-)

diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 67bffd1555..74f06a1617 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -158,7 +158,7 @@ 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>boolean</type> | <literal>force</literal>)</term>
         <listitem>
          <para>
           Specifies whether to copy pre-existing data in the publications
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index c69bb16d39..58f5d88235 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
           changes and the replicated changes that was generated from other
           nodes in the publisher. The default is <literal>false</literal>.
          </para>
+         <para>
+          If the publication tables were also subscribing 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,7 +220,7 @@ 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>boolean</type> | <literal>force</literal>)</term>
         <listitem>
          <para>
           Specifies whether to copy pre-existing data in the publications
@@ -225,6 +232,14 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
           will affect what data is copied. Refer to the
           <xref linkend="sql-createsubscription-notes" /> for details.
          </para>
+
+         <para>
+          If the publication tables were also subscribing data in the publisher
+          from other publishers, it will affect the
+          <command>CREATE SUBSCRIPTION</command> based on the value specified
+          for <literal>publish_local_only</literal> option. Refer to the
+          <xref linkend="sql-createsubscription-notes" /> for details.
+         </para>
         </listitem>
        </varlistentry>
 
@@ -356,6 +371,43 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
    copied data that would be incompatible with subsequent filtering.
   </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>publish_local_only</literal>.
+   c) Node2 - Publication publishing employee table.
+   d) Node1 - Subscription subscribing from publication pub2 with
+   <literal>publish_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>publish_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 sub2 CONNECTION 'dbname =postgres port = 9999' publication pub2 with (publish_local_only=on);
+ERROR:  CREATE/ALTER SUBSCRIPTION with publish_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 3 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>publish_local_only</literal> as <literal>true</literal> and
+   <literal>copy_data</literal> as <literal>false</literal>. b) If there is
+   data present only in one Node let's say Node1, then the user can create the
+   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 operation on the respective tables of Node2.
+   c) If there is data present both on Node1 and Node2, 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.
+  </para>
+
  </refsect1>
 
  <refsect1>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 78136050b8..25c0c85e93 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -66,6 +66,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.
@@ -78,7 +90,7 @@ typedef struct SubOpts
 	bool		connect;
 	bool		enabled;
 	bool		create_slot;
-	bool		copy_data;
+	CopyData	copy_data;
 	bool		refresh;
 	bool		binary;
 	bool		streaming;
@@ -86,11 +98,69 @@ typedef struct SubOpts
 	bool		only_local;
 } 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 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.
@@ -123,7 +193,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))
@@ -189,7 +259,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)
@@ -290,17 +360,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;
 	}
 
 	/*
@@ -527,13 +597,14 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 			 * Set sync state based on if we were asked to do data copy or
 			 * not.
 			 */
-			table_state = opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
+			table_state = IS_COPY_DATA_VALID(opts.copy_data) ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
 
 			/*
 			 * Get the table list from publisher and build local table status
 			 * info.
 			 */
-			tables = fetch_table_list(wrconn, publications);
+			tables = fetch_table_list(wrconn, publications, opts.copy_data,
+									  opts.only_local);
 			foreach(lc, tables)
 			{
 				RangeVar   *rv = (RangeVar *) lfirst(lc);
@@ -576,7 +647,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 				 * PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
 				 * PUBLICATION to work.
 				 */
-				if (opts.twophase && !opts.copy_data && tables != NIL)
+				if (opts.twophase && IS_COPY_DATA_VALID(opts.copy_data) &&
+					tables != NIL)
 					twophase_enabled = true;
 
 				walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
@@ -615,7 +687,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 }
 
 static void
-AlterSubscription_refresh(Subscription *sub, bool copy_data)
+AlterSubscription_refresh(Subscription *sub, CopyData copy_data,
+						  bool only_local)
 {
 	char	   *err;
 	List	   *pubrel_names;
@@ -647,7 +720,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
 	PG_TRY();
 	{
 		/* Get the table list from publisher. */
-		pubrel_names = fetch_table_list(wrconn, sub->publications);
+		pubrel_names = fetch_table_list(wrconn, sub->publications, copy_data,
+										only_local);
 
 		/* Get local table list. */
 		subrel_states = GetSubscriptionRelations(sub->oid);
@@ -701,7 +775,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\"",
@@ -975,7 +1049,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_PUBLISH_LOCAL_ONLY;
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
 
@@ -998,7 +1072,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"),
@@ -1010,7 +1084,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					/* Make sure refresh sees the new list of publications. */
 					sub->publications = stmt->publication;
 
-					AlterSubscription_refresh(sub, opts.copy_data);
+					AlterSubscription_refresh(sub, opts.copy_data, opts.only_local);
 				}
 
 				break;
@@ -1022,7 +1096,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_PUBLISH_LOCAL_ONLY;
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
 
@@ -1058,7 +1132,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					/* Refresh the new list of publications. */
 					sub->publications = publist;
 
-					AlterSubscription_refresh(sub, opts.copy_data);
+					AlterSubscription_refresh(sub, opts.copy_data, opts.only_local);
 				}
 
 				break;
@@ -1072,7 +1146,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_PUBLISH_LOCAL_ONLY,
+										   &opts);
 
 				/*
 				 * The subscription option "two_phase" requires that
@@ -1091,7 +1166,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"),
@@ -1100,7 +1176,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 				PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH");
 
-				AlterSubscription_refresh(sub, opts.copy_data);
+				AlterSubscription_refresh(sub, opts.copy_data, opts.only_local);
 
 				break;
 			}
@@ -1573,12 +1649,13 @@ 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};
 	ListCell   *lc;
 	bool		first;
 	List	   *tablelist = NIL;
@@ -1586,9 +1663,13 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
 	Assert(list_length(publications) > 0);
 
 	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 (");
 	first = true;
 	foreach(lc, publications)
 	{
@@ -1603,7 +1684,7 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
 	}
 	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)
@@ -1626,6 +1707,21 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
 		relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
 		Assert(!isnull);
 
+		/*
+		 * XXX: 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. Throw an error so that the user can
+		 * take care of the initial data copying and then create subscription
+		 * with copy_data off.
+		 */
+		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 publish_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"));
+
 		rv = makeRangeVar(nspname, relname, -1);
 		tablelist = lappend(tablelist, rv);
 
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 535f6da660..0882e0996d 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 313c4ada25..18858728dc 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/029_circular.pl b/src/test/subscription/t/029_circular.pl
index e4b9e58c39..7bce6acf23 100644
--- a/src/test/subscription/t/029_circular.pl
+++ b/src/test/subscription/t/029_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 (publish_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 (publish_local_only = on, copy_data = on)");
+like(
+        $stderr,
+        qr/ERROR:  CREATE\/ALTER SUBSCRIPTION with publish_local_only and copy_data as true is not allowed when the publisher might have replicated data/,
+        "Create subscription with publish_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 (publish_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 321e083d43..ffae65fdde 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -441,6 +441,7 @@ ConvProcInfo
 ConversionLocation
 ConvertRowtypeExpr
 CookedConstraint
+CopyData
 CopyDest
 CopyFormatOptions
 CopyFromState
-- 
2.32.0