From ef2f8d373e04ebea1f8d057529d1e11642df755b Mon Sep 17 00:00:00 2001
From: Khanna <Shubham.Khanna@fujitsu.com>
Date: Thu, 28 Aug 2025 22:26:04 +0530
Subject: [PATCH v13] Support existing publications in pg_createsubscriber

Allow pg_createsubscriber to reuse existing publications instead of failing
when they already exist on the publisher.

Previously, pg_createsubscriber would fail if any specified publication already
existed. Now, existing publications are reused as-is with their current
configuration, and non-existing publications are created automatically with
FOR ALL TABLES.

This change provides flexibility when working with mixed scenarios of existing
and new publications. Users should verify that existing publications have the
desired configuration before reusing them, and can use --dry-run to see which
publications will be reused and which will be created.

Only publications created by pg_createsubscriber are cleaned up during error
cleanup operations. Pre-existing publications are preserved unless
'--clean=publications' is explicitly specified, which drops all publications.
---
 doc/src/sgml/ref/pg_createsubscriber.sgml     |  11 ++
 src/bin/pg_basebackup/pg_createsubscriber.c   |  80 +++++++--
 .../t/040_pg_createsubscriber.pl              | 162 ++++++++++++++++++
 3 files changed, 234 insertions(+), 19 deletions(-)

diff --git a/doc/src/sgml/ref/pg_createsubscriber.sgml b/doc/src/sgml/ref/pg_createsubscriber.sgml
index bb9cc72576c..75e2edc83fd 100644
--- a/doc/src/sgml/ref/pg_createsubscriber.sgml
+++ b/doc/src/sgml/ref/pg_createsubscriber.sgml
@@ -285,6 +285,17 @@ PostgreSQL documentation
        a generated name is assigned to the publication name. This option cannot
        be used together with <option>--all</option>.
       </para>
+      <para>
+       If a publication with the specified name already exists on the publisher,
+       it will be reused with its current configuration, including its table
+       list, row filters, and column filters.
+       If a publication does not exist, it will be created automatically with
+       <literal>FOR ALL TABLES</literal>.
+      </para>
+      <para>
+       Use <option>--dry-run</option> to safely preview which publications will
+       be reused and which will be created.
+      </para>
      </listitem>
     </varlistentry>
 
diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 3986882f042..ef9e70ceae7 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -114,6 +114,7 @@ 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 bool check_publication_exists(PGconn *conn, const char *pubname, const char *dbname);
 static void drop_publication(PGconn *conn, const char *pubname,
 							 const char *dbname, bool *made_publication);
 static void check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo);
@@ -753,6 +754,31 @@ generate_object_name(PGconn *conn)
 	return objname;
 }
 
+/*
+ * Return whether a specified publication exists in the specified database.
+ */
+static bool
+check_publication_exists(PGconn *conn, const char *pubname, const char *dbname)
+{
+	PGresult   *res;
+	bool		exists;
+	char	   *query;
+
+	query = psprintf("SELECT 1 FROM pg_publication WHERE pubname = %s",
+					 PQescapeLiteral(conn, pubname, strlen(pubname)));
+	res = PQexec(conn, query);
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		pg_fatal("could not check for publication \"%s\" in database \"%s\": %s",
+				 pubname, dbname, PQerrorMessage(conn));
+
+	exists = (PQntuples(res) == 1);
+
+	PQclear(res);
+	pg_free(query);
+	return exists;
+}
+
 /*
  * Create the publications and replication slots in preparation for logical
  * replication. Returns the LSN from latest replication slot. It will be the
@@ -789,13 +815,26 @@ setup_publisher(struct LogicalRepInfo *dbinfo)
 		if (num_replslots == 0)
 			dbinfo[i].replslotname = pg_strdup(dbinfo[i].subname);
 
-		/*
-		 * Create publication on publisher. This step should be executed
-		 * *before* promoting the subscriber to avoid any transactions between
-		 * consistent LSN and the new publication rows (such transactions
-		 * wouldn't see the new publication rows resulting in an error).
-		 */
-		create_publication(conn, &dbinfo[i]);
+		if (check_publication_exists(conn, dbinfo[i].pubname, dbinfo[i].dbname))
+		{
+			/* Reuse existing publication on publisher. */
+			pg_log_info("use existing publication \"%s\" in database \"%s\"",
+						dbinfo[i].pubname, dbinfo[i].dbname);
+			dbinfo[i].made_publication = false;
+		}
+		else
+		{
+			/*
+			 * Create publication on publisher. This step should be executed
+			 * *before* promoting the subscriber to avoid any transactions
+			 * between consistent LSN and the new publication rows (such
+			 * transactions wouldn't see the new publication rows resulting in
+			 * an error).
+			 */
+			create_publication(conn, &dbinfo[i]);
+			pg_log_info("create publication \"%s\" in database \"%s\"",
+						dbinfo[i].pubname, dbinfo[i].dbname);
+		}
 
 		/* Create replication slot on publisher */
 		if (lsn)
@@ -1728,12 +1767,6 @@ drop_publication(PGconn *conn, const char *pubname, const char *dbname,
 
 /*
  * Retrieve and drop the publications.
- *
- * Since the publications were created before the consistent LSN, they
- * remain on the subscriber even after the physical replica is
- * promoted. Remove these publications from the subscriber because
- * they have no use. Additionally, if requested, drop all pre-existing
- * publications.
  */
 static void
 check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo)
@@ -1766,13 +1799,22 @@ check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo)
 		PQclear(res);
 	}
 
-	/*
-	 * In dry-run mode, we don't create publications, but we still try to drop
-	 * those to provide necessary information to the user.
-	 */
 	if (!drop_all_pubs || dry_run)
-		drop_publication(conn, dbinfo->pubname, dbinfo->dbname,
-						 &dbinfo->made_publication);
+	{
+		/*
+		 * Since the publications were created before the consistent LSN, they
+		 * remain on the subscriber even after the physical replica is
+		 * promoted. Only drop publications that were created by
+		 * pg_createsubscriber during this operation. Pre-existing
+		 * publications are preserved.
+		 */
+		if (!drop_all_pubs && dbinfo->made_publication)
+			drop_publication(conn, dbinfo->pubname, dbinfo->dbname,
+							 &dbinfo->made_publication);
+		else if (!drop_all_pubs)
+			pg_log_info("preserve existing publication \"%s\" in database \"%s\"",
+						dbinfo->pubname, dbinfo->dbname);
+	}
 }
 
 /*
diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
index 229fef5b3b5..c4f10fdee7b 100644
--- a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
+++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
@@ -537,9 +537,171 @@ my $sysid_s = $node_s->safe_psql('postgres',
 	'SELECT system_identifier FROM pg_control_system()');
 ok($sysid_p != $sysid_s, 'system identifier was changed');
 
+# Create user-defined publications.
+$node_p->safe_psql($db1,
+	"CREATE PUBLICATION test_pub_existing FOR TABLE tbl1");
+
+# Initialize node_s2 and node_s3 as a fresh standby of node_p for existing/new
+# publication test.
+$node_p->backup('backup_tablepub');
+my $node_s2 = PostgreSQL::Test::Cluster->new('node_s2');
+$node_s2->init_from_backup($node_p, 'backup_tablepub', has_streaming => 1);
+$node_s2->start;
+$node_s2->stop;
+
+my $node_s3 = PostgreSQL::Test::Cluster->new('node_s3');
+$node_s3->init_from_backup($node_p, 'backup_tablepub', has_streaming => 1);
+$node_s3->start;
+$node_s3->stop;
+
+# Test publication reuse and creation behavior with --dry-run.
+# This should reuse existing 'test_pub_existing' and create new 'test_pub_new',
+# demonstrating mixed publication handling without actual changes.
+($stdout, $stderr) = run_command(
+	[
+		'pg_createsubscriber',
+		'--verbose',
+		'--dry-run',
+		'--pgdata' => $node_s2->data_dir,
+		'--publisher-server' => $node_p->connstr($db1),
+		'--socketdir' => $node_s2->host,
+		'--subscriber-port' => $node_s2->port,
+		'--database' => $db1,
+		'--database' => $db2,
+		'--publication' => 'test_pub_existing',
+		'--publication' => 'test_pub_new',
+	],
+	'run pg_createsubscriber --dry-run on node S2');
+
+like(
+	$stderr,
+	qr/use existing publication "test_pub_existing"/,
+	'dry-run logs reuse of existing publication');
+like(
+	$stderr,
+	qr/create publication "test_pub_new"/,
+	'dry-run logs creation of new publication');
+
+# Verify dry-run did not modify publisher state
+my $pub_names_db1 = $node_p->safe_psql($db1,
+	"SELECT pubname FROM pg_publication ORDER BY pubname");
+is( $pub_names_db1, qq(pub1
+test_pub1
+test_pub2
+test_pub_existing),
+	"existing publication remains unchanged after dry-run");
+
+my $pub_names_db2 = $node_p->safe_psql($db2,
+	"SELECT pubname FROM pg_publication ORDER BY pubname");
+is($pub_names_db2, 'pub2',
+	"dry-run did not actually create publications in db2");
+
+# Run pg_createsubscriber to test publication reuse and creation behavior.
+# This should reuse the existing 'test_pub_existing' publication in db1 and
+# create a new 'test_pub_new' publication in db2, demonstrating how the tool
+# handles mixed existing/new publication scenarios.
+command_ok(
+	[
+		'pg_createsubscriber',
+		'--verbose', '--verbose',
+		'--recovery-timeout' => $PostgreSQL::Test::Utils::timeout_default,
+		'--pgdata' => $node_s2->data_dir,
+		'--publisher-server' => $node_p->connstr($db1),
+		'--socketdir' => $node_s2->host,
+		'--subscriber-port' => $node_s2->port,
+		'--database' => $db1,
+		'--database' => $db2,
+		'--publication' => 'test_pub_existing',
+		'--publication' => 'test_pub_new',
+	],
+	'run pg_createsubscriber on node S2');
+
+# Start subscriber
+$node_s2->start;
+
+# Verify that test_pub_new was created in db2
+$result = $node_p->safe_psql($db2,
+	"SELECT COUNT(*) FROM pg_publication WHERE pubname = 'test_pub_new'");
+is($result, '1', 'test_pub_new publication was created in db2');
+
+# Insert rows on P
+$node_p->safe_psql($db1, "INSERT INTO tbl1 VALUES('fourth row')");
+$node_p->safe_psql($db2, "INSERT INTO tbl2 VALUES('row 2')");
+
+# Get subscription names and publications
+$result = $node_s2->safe_psql(
+	'postgres', qq(
+    SELECT subname, subpublications FROM pg_subscription WHERE subname ~ '^pg_createsubscriber_'
+));
+@subnames = split("\n", $result);
+
+# Check result in database $db1
+$result = $node_s2->safe_psql($db1, 'SELECT * FROM tbl1');
+is( $result, qq(first row
+second row
+third row
+fourth row),
+	"logical replication works in database $db1");
+
+# Check result in database $db2
+$result = $node_s2->safe_psql($db2, 'SELECT * FROM tbl2');
+is( $result, qq(row 1
+row 2),
+	"logical replication works in database $db2");
+
+# Verify that the correct publications are being used
+$result = $node_s2->safe_psql(
+	'postgres', qq(
+        SELECT s.subpublications
+        FROM pg_subscription s
+        WHERE s.subname ~ '^pg_createsubscriber_'
+        ORDER BY s.subdbid
+    )
+);
+
+is( $result, qq({test_pub_existing}
+{test_pub_new}),
+	"subscriptions use the correct publications");
+
+# Run pg_createsubscriber with --clean=publications to test that ALL
+# publications (both pre-existing and pg_createsubscriber-created) are dropped
+# when the --clean=publications option is used.
+command_ok(
+	[
+		'pg_createsubscriber',
+		'--verbose', '--verbose',
+		'--recovery-timeout' => $PostgreSQL::Test::Utils::timeout_default,
+		'--pgdata' => $node_s3->data_dir,
+		'--publisher-server' => $node_p->connstr($db1),
+		'--socketdir' => $node_s3->host,
+		'--subscriber-port' => $node_s3->port,
+		'--database' => $db1,
+		'--database' => $db2,
+		'--publication' => 'test_pub_existing',
+		'--publication' => 'test_pub_new',
+		'--clean' => 'publications',
+	],
+	'run pg_createsubscriber on node S3');
+
+# Start subscriber
+$node_s3->start;
+
+# Confirm ALL publications were removed by --clean=publications
+my $pub_count_db1 =
+  $node_s3->safe_psql($db1, "SELECT COUNT(*) FROM pg_publication");
+is($pub_count_db1, '0',
+	'all publications were removed from db1 by --clean=publications');
+
+my $pub_count_db2 =
+  $node_s3->safe_psql($db2, "SELECT COUNT(*) FROM pg_publication");
+is($pub_count_db2, '0',
+	'all publications were removed from db2 by --clean=publications');
+
 # clean up
 $node_p->teardown_node;
 $node_s->teardown_node;
+$node_s2->teardown_node;
+$node_s3->teardown_node;
 $node_t->teardown_node;
 $node_f->teardown_node;
 
-- 
2.34.1

