v2-0001-Fix-data-replicated-twice-when-specifying-PUBLISH.patch

From 8f302f5dd254853f7d0fd855ac8328a79abad38a Mon Sep 17 00:00:00 2001
From: "shiy.fnst" <shiy.fnst@fujitsu.com>
Date: Tue, 19 Apr 2022 11:33:33 +0800
Subject: [PATCH v2] Fix data replicated twice when specifying
 PUBLISH_VIA_PARTITION_ROOT option.

If there are two publications that publish the parent table and the child table
separately, and both specify the option PUBLISH_VIA_PARTITION_ROOT, when
subscribing to both publications using one subscription, the data is replicated
twice in inital copy. What we expect is to be copied only once.

To fix this, we exclude the partitioned table whose ancestor belongs to
specified publications when getting the table list from publisher.
---
 src/backend/commands/subscriptioncmds.c | 26 +++++++---
 src/test/subscription/t/100_bugs.pl     | 69 +++++++++++++++++++++++++
 2 files changed, 89 insertions(+), 6 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index b94236f74d..8f63dfe047 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1759,20 +1759,34 @@ static List *
 fetch_table_list(WalReceiverConn *wrconn, List *publications)
 {
 	WalRcvExecResult *res;
-	StringInfoData cmd;
+	StringInfoData cmd,
+				   pub_names;
 	TupleTableSlot *slot;
 	Oid			tableRow[2] = {TEXTOID, TEXTOID};
 	List	   *tablelist = NIL;
 
+	initStringInfo(&pub_names);
+	get_publications_str(publications, &pub_names, true);
+
+	/*
+	 * Get the list of tables from publisher, the partitioned table whose
+	 * ancestor is also in this list should be ignored, otherwise the initial
+	 * date in the partitioned table would be replicated twice.
+	 */
 	initStringInfo(&cmd);
-	appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n"
-						   "  FROM pg_catalog.pg_publication_tables t\n"
-						   " WHERE t.pubname IN (");
-	get_publications_str(publications, &cmd, true);
-	appendStringInfoChar(&cmd, ')');
+	appendStringInfo(&cmd, "SELECT DISTINCT ns.nspname, c.relname\n"
+					" FROM pg_catalog.pg_publication_tables t\n"
+					"      JOIN pg_catalog.pg_namespace ns ON ns.nspname = t.schemaname\n"
+					"      JOIN pg_catalog.pg_class c ON c.relname = t.tablename AND c.relnamespace = ns.oid\n"
+					" WHERE t.pubname IN (%s)\n"
+					"   AND (c.relispartition IS FALSE OR NOT EXISTS (SELECT 1 FROM pg_partition_ancestors(c.oid)\n"
+					" WHERE relid IN ( SELECT DISTINCT (schemaname || '.' || tablename)::regclass::oid\n"
+					" FROM pg_catalog.pg_publication_tables t\n"
+					" WHERE t.pubname IN (%s) ) AND relid != c.oid))", pub_names.data, pub_names.data);
 
 	res = walrcv_exec(wrconn, cmd.data, 2, tableRow);
 	pfree(cmd.data);
+	pfree(pub_names.data);
 
 	if (res->status != WALRCV_OK_TUPLES)
 		ereport(ERROR,
diff --git a/src/test/subscription/t/100_bugs.pl b/src/test/subscription/t/100_bugs.pl
index 11ba473715..5b11e06f02 100644
--- a/src/test/subscription/t/100_bugs.pl
+++ b/src/test/subscription/t/100_bugs.pl
@@ -307,4 +307,73 @@ is( $node_subscriber->safe_psql(
 $node_publisher->stop('fast');
 $node_subscriber->stop('fast');
 
+# https://www.postgresql.org/message-id/OS0PR01MB5716DC2982CC735FDE388804940B9%40OS0PR01MB5716.jpnprd01.prod.outlook.com
+
+# The bug was that if there are two publications that publish the parent table
+# and the child table separately, and both specify the option
+# PUBLISH_VIA_PARTITION_ROOT, when subscribing to both publications with one
+# subscription, the data is replicated twice. What we expect is to be copied
+# only once.
+$node_publisher = PostgreSQL::Test::Cluster->new('publisher4');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+$node_subscriber = PostgreSQL::Test::Cluster->new('subscriber4');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE viaroot_partition(a int) PARTITION BY RANGE(a)"
+);
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE viaroot_partitioned(LIKE viaroot_partition)"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE viaroot_partition ATTACH PARTITION viaroot_partitioned DEFAULT"
+);
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE viaroot_partition(a int) PARTITION BY RANGE(a)"
+);
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE viaroot_partitioned(LIKE viaroot_partition)"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE viaroot_partition ATTACH PARTITION viaroot_partitioned DEFAULT"
+);
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO viaroot_partition (a) VALUES (1), (2), (3)"
+);
+
+# create pub/sub
+$publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION pub_viaroot1 FOR TABLE viaroot_partition WITH (PUBLISH_VIA_PARTITION_ROOT=TRUE)"
+);
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION pub_viaroot2 FOR TABLE viaroot_partitioned WITH (PUBLISH_VIA_PARTITION_ROOT=TRUE)"
+);
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION sub_viaroot CONNECTION '$publisher_connstr' PUBLICATION pub_viaroot1, pub_viaroot2"
+);
+
+$node_publisher->wait_for_catchup('sub_viaroot');
+
+# Also wait for initial table sync to finish
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check expected replicated result.
+is( $node_subscriber->safe_psql(
+		'postgres', "SELECT * FROM viaroot_partition"),
+	qq(1
+2
+3),
+	"check initial data copy from table viaroot_partition");
+
+$node_publisher->stop('fast');
+$node_subscriber->stop('fast');
+
 done_testing();
-- 
2.18.4