From 105b1be8b65132bda38d01aa364deb6c063b0b48 Mon Sep 17 00:00:00 2001
From: Khanna <Shubham.Khanna@fujitsu.com>
Date: Thu, 13 Feb 2025 09:25:11 +0530
Subject: [PATCH v10] Support for dropping all publications in
 'pg_createsubscriber'

This patch introduces a new '--cleanup-existing-publications' option in the
'pg_createsubscriber' utility.

This feature ensures a clean and streamlined setup of logical replication by
removing publications on the subscriber that were originally replicated from
the primary server during streaming replication.
These publications become redundant once the setup transitions to logical
replication and serve no further purpose.

This cleanup process removes all publications from the subscriber, regardless
of their origin. Users should back up any manually created publications before
running this command.

The --cleanup-existing-publications option allows users to remove replicated
publications from the subscriber when setting up logical replication.
This is particularly useful when publications created in a previous setup are
no longer needed.
By default, publications are preserved to avoid unintended data loss.
---
 doc/src/sgml/ref/pg_createsubscriber.sgml     |  11 ++
 src/bin/pg_basebackup/pg_createsubscriber.c   |  97 +++++++---
 .../t/040_pg_createsubscriber.pl              | 180 +++++++++++++-----
 3 files changed, 208 insertions(+), 80 deletions(-)

diff --git a/doc/src/sgml/ref/pg_createsubscriber.sgml b/doc/src/sgml/ref/pg_createsubscriber.sgml
index d56487fe2c..a98b583537 100644
--- a/doc/src/sgml/ref/pg_createsubscriber.sgml
+++ b/doc/src/sgml/ref/pg_createsubscriber.sgml
@@ -87,6 +87,17 @@ PostgreSQL documentation
    command-line arguments:
 
    <variablelist>
+    <varlistentry>
+     <term><option>-c</option></term>
+     <term><option>--cleanup-existing-publications</option></term>
+     <listitem>
+      <para>
+       Remove all existing publications from specified databases on the target
+       server.
+      </para>
+     </listitem>
+    </varlistentry>
+
     <varlistentry>
      <term><option>-d <replaceable class="parameter">dbname</replaceable></option></term>
      <term><option>--database=<replaceable class="parameter">dbname</replaceable></option></term>
diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 9fdf15e5ac..d1aa39c99e 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -43,6 +43,7 @@ struct CreateSubscriberOptions
 	SimpleStringList sub_names; /* list of subscription names */
 	SimpleStringList replslot_names;	/* list of replication slot names */
 	int			recovery_timeout;	/* stop recovery after this time */
+	bool		drop_publications;	/* drop all publications */
 };
 
 struct LogicalRepInfo
@@ -79,7 +80,8 @@ static void check_publisher(const struct LogicalRepInfo *dbinfo);
 static char *setup_publisher(struct LogicalRepInfo *dbinfo);
 static void check_subscriber(const struct LogicalRepInfo *dbinfo);
 static void setup_subscriber(struct LogicalRepInfo *dbinfo,
-							 const char *consistent_lsn);
+							 const char *consistent_lsn,
+							 bool drop_publications);
 static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir,
 						   const char *lsn);
 static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo,
@@ -97,7 +99,11 @@ static void stop_standby_server(const char *datadir);
 static void wait_for_end_recovery(const char *conninfo,
 								  const struct CreateSubscriberOptions *opt);
 static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
-static void drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
+static void drop_publication_by_name(PGconn *conn, const char *pubname,
+									 const char *dbname);
+static void check_and_drop_existing_publications(PGconn *conn,
+												 const char *dbname,
+												 bool drop_publications);
 static void create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo);
 static void set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo,
 									 const char *lsn);
@@ -180,7 +186,8 @@ cleanup_objects_atexit(void)
 			if (conn != NULL)
 			{
 				if (dbinfo[i].made_publication)
-					drop_publication(conn, &dbinfo[i]);
+					drop_publication_by_name(conn, dbinfo[i].pubname,
+											 dbinfo[i].dbname);
 				if (dbinfo[i].made_replslot)
 					drop_replication_slot(conn, &dbinfo[i], dbinfo[i].replslotname);
 				disconnect_database(conn, false);
@@ -220,6 +227,8 @@ usage(void)
 	printf(_("Usage:\n"));
 	printf(_("  %s [OPTION]...\n"), progname);
 	printf(_("\nOptions:\n"));
+	printf(_("  -c  --cleanup-existing-publications\n"
+			 "                                  drop all publications on the subscriber\n"));
 	printf(_("  -d, --database=DBNAME           database in which to create a subscription\n"));
 	printf(_("  -D, --pgdata=DATADIR            location for the subscriber data directory\n"));
 	printf(_("  -n, --dry-run                   dry run, just show what would be done\n"));
@@ -1154,10 +1163,12 @@ check_and_drop_existing_subscriptions(PGconn *conn,
 /*
  * Create the subscriptions, adjust the initial location for logical
  * replication and enable the subscriptions. That's the last step for logical
- * replication setup.
+ * replication setup. If 'drop_publications' is true, existing publications on
+ * the subscriber will be dropped before creating new subscriptions.
  */
 static void
-setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
+setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn,
+				 bool cleanup_existing_publications)
 {
 	for (int i = 0; i < num_dbs; i++)
 	{
@@ -1178,8 +1189,14 @@ setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
 		 * Since the publication was created before the consistent LSN, it is
 		 * available on the subscriber when the physical replica is promoted.
 		 * Remove publications from the subscriber because it has no use.
+		 * Additionally, drop publications existed before this command if
+		 * requested.
 		 */
-		drop_publication(conn, &dbinfo[i]);
+		if (cleanup_existing_publications)
+			check_and_drop_existing_publications(conn, dbinfo[i].dbname,
+												 cleanup_existing_publications);
+		else
+			drop_publication_by_name(conn, dbinfo[i].pubname, dbinfo[i].dbname);
 
 		create_subscription(conn, &dbinfo[i]);
 
@@ -1643,25 +1660,47 @@ create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
 }
 
 /*
- * Remove publication if it couldn't finish all steps.
+ * Check and drop existing publications on the subscriber if requested.
  */
 static void
-drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
+check_and_drop_existing_publications(PGconn *conn, const char *dbname,
+									 bool drop_publications)
 {
-	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
-	char	   *pubname_esc;
 
-	Assert(conn != NULL);
+	pg_log_info("dropping all existing publications in database \"%s\"",
+				dbname);
 
-	pubname_esc = PQescapeIdentifier(conn, dbinfo->pubname, strlen(dbinfo->pubname));
+	/* Fetch all publication names */
+	res = PQexec(conn, "SELECT pubname FROM pg_catalog.pg_publication;");
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not obtain publication information: %s",
+					 PQresultErrorMessage(res));
+		PQclear(res);
+		return;
+	}
 
-	pg_log_info("dropping publication \"%s\" in database \"%s\"",
-				dbinfo->pubname, dbinfo->dbname);
+	/* Drop each publication */
+	for (int i = 0; i < PQntuples(res); i++)
+		drop_publication_by_name(conn, PQgetvalue(res, i, 0), dbname);
 
-	appendPQExpBuffer(str, "DROP PUBLICATION %s", pubname_esc);
+	PQclear(res);
+	pg_log_info("dropped all publications in database \"%s\"", dbname);
+}
 
-	PQfreemem(pubname_esc);
+/* Drop a single publication, given in dbinfo. */
+static void
+drop_publication_by_name(PGconn *conn, const char *pubname, const char *dbname)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res;
+
+	char	   *pubname_esc = PQescapeIdentifier(conn, pubname, strlen(pubname));
+
+	pg_log_info("dropping publication %s in database \"%s\"", pubname, dbname);
+
+	appendPQExpBuffer(str, "DROP PUBLICATION %s", pubname_esc);
 
 	pg_log_debug("command is: %s", str->data);
 
@@ -1670,21 +1709,16 @@ drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
 		res = PQexec(conn, str->data);
 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
 		{
-			pg_log_error("could not drop publication \"%s\" in database \"%s\": %s",
-						 dbinfo->pubname, dbinfo->dbname, PQresultErrorMessage(res));
-			dbinfo->made_publication = false;	/* don't try again. */
-
-			/*
-			 * Don't disconnect and exit here. This routine is used by primary
-			 * (cleanup publication / replication slot due to an error) and
-			 * subscriber (remove the replicated publications). In both cases,
-			 * it can continue and provide instructions for the user to remove
-			 * it later if cleanup fails.
-			 */
+			pg_log_error("could not drop publication %s in database \"%s\": %s",
+						 pubname, dbname, PQresultErrorMessage(res));
 		}
 		PQclear(res);
 	}
 
+	/* For cleanup purposes */
+	dbinfo->made_publication = true;
+
+	PQfreemem(pubname_esc);
 	destroyPQExpBuffer(str);
 }
 
@@ -1888,6 +1922,7 @@ main(int argc, char **argv)
 {
 	static struct option long_options[] =
 	{
+		{"cleanup-existing-publications", no_argument, NULL, 'c'},
 		{"database", required_argument, NULL, 'd'},
 		{"pgdata", required_argument, NULL, 'D'},
 		{"dry-run", no_argument, NULL, 'n'},
@@ -1955,6 +1990,7 @@ main(int argc, char **argv)
 		0
 	};
 	opt.recovery_timeout = 0;
+	opt.drop_publications = false;
 
 	/*
 	 * Don't allow it to be run as root. It uses pg_ctl which does not allow
@@ -1972,11 +2008,14 @@ main(int argc, char **argv)
 
 	get_restricted_token();
 
-	while ((c = getopt_long(argc, argv, "d:D:np:P:s:t:U:v",
+	while ((c = getopt_long(argc, argv, "cd:D:np:P:s:t:U:v",
 							long_options, &option_index)) != -1)
 	{
 		switch (c)
 		{
+			case 'c':
+				opt.drop_publications = true;
+				break;
 			case 'd':
 				if (!simple_string_list_member(&opt.database_names, optarg))
 				{
@@ -2252,7 +2291,7 @@ main(int argc, char **argv)
 	 * point to the LSN reported by setup_publisher().  It also cleans up
 	 * publications created by this tool and replication to the standby.
 	 */
-	setup_subscriber(dbinfo, consistent_lsn);
+	setup_subscriber(dbinfo, consistent_lsn, opt.drop_publications);
 
 	/* Remove primary_slot_name if it exists on primary */
 	drop_primary_replication_slot(dbinfo, primary_slot_name);
diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
index c8dbdb7e9b..801392b4bb 100644
--- a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
+++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
@@ -152,16 +152,16 @@ $node_p->safe_psql($db2,
 
 # Set up node S as standby linking to node P
 $node_p->backup('backup_1');
-my $node_s = PostgreSQL::Test::Cluster->new('node_s');
-$node_s->init_from_backup($node_p, 'backup_1', has_streaming => 1);
-$node_s->append_conf(
+my $node_s1 = PostgreSQL::Test::Cluster->new('node_s1');
+$node_s1->init_from_backup($node_p, 'backup_1', has_streaming => 1);
+$node_s1->append_conf(
 	'postgresql.conf', qq[
 primary_slot_name = '$slotname'
 primary_conninfo = '$pconnstr dbname=postgres'
 hot_standby_feedback = on
 ]);
-$node_s->set_standby_mode();
-$node_s->start;
+$node_s1->set_standby_mode();
+$node_s1->start;
 
 # Set up node T as standby linking to node P then promote it
 my $node_t = PostgreSQL::Test::Cluster->new('node_t');
@@ -192,10 +192,10 @@ command_fails(
 		'pg_createsubscriber',
 		'--verbose',
 		'--dry-run',
-		'--pgdata' => $node_s->data_dir,
+		'--pgdata' => $node_s1->data_dir,
 		'--publisher-server' => $node_p->connstr($db1),
-		'--socketdir' => $node_s->host,
-		'--subscriber-port' => $node_s->port,
+		'--socketdir' => $node_s1->host,
+		'--subscriber-port' => $node_s1->port,
 		'--database' => $db1,
 		'--database' => $db2,
 	],
@@ -216,9 +216,9 @@ command_fails(
 	'subscriber data directory is not a copy of the source database cluster');
 
 # Set up node C as standby linking to node S
-$node_s->backup('backup_2');
+$node_s1->backup('backup_2');
 my $node_c = PostgreSQL::Test::Cluster->new('node_c');
-$node_c->init_from_backup($node_s, 'backup_2', has_streaming => 1);
+$node_c->init_from_backup($node_s1, 'backup_2', has_streaming => 1);
 $node_c->adjust_conf('postgresql.conf', 'primary_slot_name', undef);
 $node_c->set_standby_mode();
 
@@ -229,7 +229,7 @@ command_fails(
 		'--verbose',
 		'--dry-run',
 		'--pgdata' => $node_c->data_dir,
-		'--publisher-server' => $node_s->connstr($db1),
+		'--publisher-server' => $node_s1->connstr($db1),
 		'--socketdir' => $node_c->host,
 		'--subscriber-port' => $node_c->port,
 		'--database' => $db1,
@@ -246,16 +246,16 @@ max_wal_senders = 1
 max_worker_processes = 2
 });
 $node_p->restart;
-$node_s->stop;
+$node_s1->stop;
 command_fails(
 	[
 		'pg_createsubscriber',
 		'--verbose',
 		'--dry-run',
-		'--pgdata' => $node_s->data_dir,
+		'--pgdata' => $node_s1->data_dir,
 		'--publisher-server' => $node_p->connstr($db1),
-		'--socketdir' => $node_s->host,
-		'--subscriber-port' => $node_s->port,
+		'--socketdir' => $node_s1->host,
+		'--subscriber-port' => $node_s1->port,
 		'--database' => $db1,
 		'--database' => $db2,
 
@@ -272,7 +272,7 @@ max_worker_processes = 8
 });
 
 # Check some unmet conditions on node S
-$node_s->append_conf(
+$node_s1->append_conf(
 	'postgresql.conf', q{
 max_replication_slots = 1
 max_logical_replication_workers = 1
@@ -283,15 +283,15 @@ command_fails(
 		'pg_createsubscriber',
 		'--verbose',
 		'--dry-run',
-		'--pgdata' => $node_s->data_dir,
+		'--pgdata' => $node_s1->data_dir,
 		'--publisher-server' => $node_p->connstr($db1),
-		'--socketdir' => $node_s->host,
-		'--subscriber-port' => $node_s->port,
+		'--socketdir' => $node_s1->host,
+		'--subscriber-port' => $node_s1->port,
 		'--database' => $db1,
 		'--database' => $db2,
 	],
 	'standby contains unmet conditions on node S');
-$node_s->append_conf(
+$node_s1->append_conf(
 	'postgresql.conf', q{
 max_replication_slots = 10
 max_logical_replication_workers = 4
@@ -305,12 +305,12 @@ my $fslotname = 'failover_slot';
 $node_p->safe_psql($db1,
 	"SELECT pg_create_logical_replication_slot('$fslotname', 'pgoutput', false, false, true)"
 );
-$node_s->start;
+$node_s1->start;
 # Wait for the standby to catch up so that the standby is not lagging behind
 # the failover slot.
-$node_p->wait_for_replay_catchup($node_s);
-$node_s->safe_psql('postgres', "SELECT pg_sync_replication_slots()");
-my $result = $node_s->safe_psql('postgres',
+$node_p->wait_for_replay_catchup($node_s1);
+$node_s1->safe_psql('postgres', "SELECT pg_sync_replication_slots()");
+my $result = $node_s1->safe_psql('postgres',
 	"SELECT slot_name FROM pg_replication_slots WHERE slot_name = '$fslotname' AND synced AND NOT temporary"
 );
 is($result, 'failover_slot', 'failover slot is synced');
@@ -321,15 +321,30 @@ is($result, 'failover_slot', 'failover slot is synced');
 # slot) xmin on standby could be ahead of the remote slot leading
 # to failure in synchronization.
 $node_p->safe_psql($db1, "INSERT INTO tbl1 VALUES('second row')");
-$node_p->wait_for_replay_catchup($node_s);
+$node_p->wait_for_replay_catchup($node_s1);
 
 # Create subscription to test its removal
 my $dummy_sub = 'regress_sub_dummy';
 $node_p->safe_psql($db1,
 	"CREATE SUBSCRIPTION $dummy_sub CONNECTION 'dbname=dummy' PUBLICATION pub_dummy WITH (connect=false)"
 );
-$node_p->wait_for_replay_catchup($node_s);
-$node_s->stop;
+$node_p->wait_for_replay_catchup($node_s1);
+
+# Create user-defined publications, wait for streaming replication to sync them
+# to the standby, then verify that '--cleanup-existing-publications'
+# removes them.
+$node_p->safe_psql(
+	$db1, qq(
+	CREATE PUBLICATION test_pub1 FOR ALL TABLES;
+	CREATE PUBLICATION test_pub2 FOR ALL TABLES;
+));
+
+$node_p->wait_for_replay_catchup($node_s1);
+
+ok( $node_s1->safe_psql($db1, "SELECT COUNT(*) = 2 FROM pg_publication"),
+	'two publications created before --cleanup-existing-publications is run');
+
+$node_s1->stop;
 
 # dry run mode on node S
 command_ok(
@@ -338,10 +353,10 @@ command_ok(
 		'--verbose',
 		'--dry-run',
 		'--recovery-timeout' => $PostgreSQL::Test::Utils::timeout_default,
-		'--pgdata' => $node_s->data_dir,
+		'--pgdata' => $node_s1->data_dir,
 		'--publisher-server' => $node_p->connstr($db1),
-		'--socketdir' => $node_s->host,
-		'--subscriber-port' => $node_s->port,
+		'--socketdir' => $node_s1->host,
+		'--subscriber-port' => $node_s1->port,
 		'--publication' => 'pub1',
 		'--publication' => 'pub2',
 		'--subscription' => 'sub1',
@@ -352,10 +367,10 @@ command_ok(
 	'run pg_createsubscriber --dry-run on node S');
 
 # Check if node S is still a standby
-$node_s->start;
-is($node_s->safe_psql('postgres', 'SELECT pg_catalog.pg_is_in_recovery()'),
+$node_s1->start;
+is($node_s1->safe_psql('postgres', 'SELECT pg_catalog.pg_is_in_recovery()'),
 	't', 'standby is in recovery');
-$node_s->stop;
+$node_s1->stop;
 
 # pg_createsubscriber can run without --databases option
 command_ok(
@@ -363,31 +378,32 @@ command_ok(
 		'pg_createsubscriber',
 		'--verbose',
 		'--dry-run',
-		'--pgdata' => $node_s->data_dir,
+		'--pgdata' => $node_s1->data_dir,
 		'--publisher-server' => $node_p->connstr($db1),
-		'--socketdir' => $node_s->host,
-		'--subscriber-port' => $node_s->port,
+		'--socketdir' => $node_s1->host,
+		'--subscriber-port' => $node_s1->port,
 		'--replication-slot' => 'replslot1',
 	],
 	'run pg_createsubscriber without --databases');
 
-# Run pg_createsubscriber on node S.  --verbose is used twice
-# to show more information.
+# Run pg_createsubscriber on node S using '--cleanup-existing-publications'.
+# --verbose is used twice to show more information.
 command_ok(
 	[
 		'pg_createsubscriber',
 		'--verbose', '--verbose',
 		'--recovery-timeout' => $PostgreSQL::Test::Utils::timeout_default,
-		'--pgdata' => $node_s->data_dir,
+		'--pgdata' => $node_s1->data_dir,
 		'--publisher-server' => $node_p->connstr($db1),
-		'--socketdir' => $node_s->host,
-		'--subscriber-port' => $node_s->port,
+		'--socketdir' => $node_s1->host,
+		'--subscriber-port' => $node_s1->port,
 		'--publication' => 'pub1',
 		'--publication' => 'pub2',
 		'--replication-slot' => 'replslot1',
 		'--replication-slot' => 'replslot2',
 		'--database' => $db1,
 		'--database' => $db2,
+		'--cleanup-existing-publications',
 	],
 	'run pg_createsubscriber on node S');
 
@@ -404,53 +420,115 @@ $node_p->safe_psql($db1, "INSERT INTO tbl1 VALUES('third row')");
 $node_p->safe_psql($db2, "INSERT INTO tbl2 VALUES('row 1')");
 
 # Start subscriber
-$node_s->start;
+$node_s1->start;
+
+# Confirm publications are removed from the subscriber node
+is($node_s1->safe_psql($db1, "SELECT COUNT(*) FROM pg_publication;"),
+	'0',
+	'all publications dropped after --cleanup-existing-publications is run');
 
 # Confirm the pre-existing subscription has been removed
-$result = $node_s->safe_psql(
+$result = $node_s1->safe_psql(
 	'postgres', qq(
 	SELECT count(*) FROM pg_subscription WHERE subname = '$dummy_sub'
 ));
 is($result, qq(0), 'pre-existing subscription was dropped');
 
 # Get subscription names
-$result = $node_s->safe_psql(
+$result = $node_s1->safe_psql(
 	'postgres', qq(
 	SELECT subname FROM pg_subscription WHERE subname ~ '^pg_createsubscriber_'
 ));
 my @subnames = split("\n", $result);
 
 # Wait subscriber to catch up
-$node_s->wait_for_subscription_sync($node_p, $subnames[0]);
-$node_s->wait_for_subscription_sync($node_p, $subnames[1]);
+$node_s1->wait_for_subscription_sync($node_p, $subnames[0]);
+$node_s1->wait_for_subscription_sync($node_p, $subnames[1]);
 
 # Confirm the failover slot has been removed
-$result = $node_s->safe_psql($db1,
+$result = $node_s1->safe_psql($db1,
 	"SELECT count(*) FROM pg_replication_slots WHERE slot_name = '$fslotname'"
 );
 is($result, qq(0), 'failover slot was removed');
 
 # Check result in database $db1
-$result = $node_s->safe_psql($db1, 'SELECT * FROM tbl1');
+$result = $node_s1->safe_psql($db1, 'SELECT * FROM tbl1');
 is( $result, qq(first row
 second row
 third row),
 	"logical replication works in database $db1");
 
 # Check result in database $db2
-$result = $node_s->safe_psql($db2, 'SELECT * FROM tbl2');
+$result = $node_s1->safe_psql($db2, 'SELECT * FROM tbl2');
 is($result, qq(row 1), "logical replication works in database $db2");
 
 # Different system identifier?
 my $sysid_p = $node_p->safe_psql('postgres',
 	'SELECT system_identifier FROM pg_control_system()');
-my $sysid_s = $node_s->safe_psql('postgres',
+my $sysid_s = $node_s1->safe_psql('postgres',
 	'SELECT system_identifier FROM pg_control_system()');
 ok($sysid_p != $sysid_s, 'system identifier was changed');
 
+# Reuse P as primary
+# Set up node S2 as standby linking to node P
+$node_p->backup('backup_3');
+my $node_s2 = PostgreSQL::Test::Cluster->new('node_s2');
+$node_s2->init_from_backup($node_p, 'backup_3', has_streaming => 1);
+$node_s2->append_conf(
+	'postgresql.conf', qq[
+       primary_conninfo = '$pconnstr'
+       hot_standby_feedback = on
+       max_logical_replication_workers = 5
+       ]);
+$node_s2->set_standby_mode();
+$node_s2->start;
+
+# Ensure there is a user database on the publisher
+my $db3 = generate_db($node_p, 'regression', 91, 127, '');
+
+# Create user-defined publications
+$node_p->safe_psql($db3, "CREATE PUBLICATION test_pub3 FOR ALL TABLES;");
+$node_p->safe_psql($db3, "CREATE PUBLICATION test_pub4 FOR ALL TABLES;");
+
+$node_p->wait_for_replay_catchup($node_s2);
+
+# Verify the existing publications
+ok( $node_s2->safe_psql($db3, "SELECT COUNT(*) = 2 FROM pg_publication"),
+	'two publications are created before running pg_createsubscriber');
+
+$node_s2->stop;
+
+# Run pg_createsubscriber on node A without using
+# '--cleanup-existing-publications'.
+# --verbose is used twice to show more information.
+command_ok(
+	[
+		'pg_createsubscriber',
+		'--verbose', '--verbose',
+		'--recovery-timeout' => $PostgreSQL::Test::Utils::timeout_default,
+		'--pgdata' => $node_s2->data_dir,
+		'--publisher-server' => $node_p->connstr($db3),
+		'--socketdir' => $node_s2->host,
+		'--subscriber-port' => $node_s2->port,
+		'--database' => $db3,
+	],
+	'run pg_createsubscriber without --cleanup-existing-publications on node A'
+);
+
+$node_s2->start;
+
+# Confirm publications still remain after running 'pg_createsubscriber'
+is($node_s2->safe_psql($db3, "SELECT COUNT(*) FROM pg_publication;"),
+	'2', 'all publications remain after running pg_createsubscriber');
+
+# Drop the newly created publications
+$node_p->safe_psql($db3, "DROP PUBLICATION IF EXISTS test_pub3;");
+$node_p->safe_psql($db3, "DROP PUBLICATION IF EXISTS test_pub4;");
+
 # clean up
 $node_p->teardown_node;
-$node_s->teardown_node;
+$node_s1->teardown_node;
+$node_s2->teardown_node;
 $node_t->teardown_node;
 $node_f->teardown_node;
 
-- 
2.34.1

