REL14_v21-0001-Fix-data-replicated-twice-when-specifying-publis_patch

From f7076870670023ea78b6f948120c3de22e226c9e Mon Sep 17 00:00:00 2001
From: wangw <wangw.fnst@fujitsu.com>
Date: Thu, 28 Apr 2022 11:28:35 +0800
Subject: [PATCH v21] Fix data replicated twice when specifying
 publish_via_partition_root option.

If there are two publications - one of them publishing a parent table
(using publish_via_partition_root = true) and the other is publishing one of
the parent's child tables - then subscribing to both publications from one
subscription results in the same initial child data being copied twice. It
should only be copied once.

To fix this, we exclude the partition table whose ancestor belongs to specified
publications when getting the table list from publisher.
---
 src/backend/commands/subscriptioncmds.c  | 35 ++++++++++++++++++------
 src/test/subscription/t/013_partition.pl | 30 ++++++++++++++------
 2 files changed, 48 insertions(+), 17 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 1719f04517..173b29d725 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1477,7 +1477,8 @@ static List *
 fetch_table_list(WalReceiverConn *wrconn, List *publications)
 {
 	WalRcvExecResult *res;
-	StringInfoData cmd;
+	StringInfoData cmd,
+				pub_names;
 	TupleTableSlot *slot;
 	Oid			tableRow[2] = {TEXTOID, TEXTOID};
 	ListCell   *lc;
@@ -1486,10 +1487,7 @@ 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 (");
+	initStringInfo(&pub_names);
 	first = true;
 	foreach(lc, publications)
 	{
@@ -1498,14 +1496,35 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
 		if (first)
 			first = false;
 		else
-			appendStringInfoString(&cmd, ", ");
+			appendStringInfoString(&pub_names, ", ");
 
-		appendStringInfoString(&cmd, quote_literal_cstr(pubname));
+		appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
 	}
-	appendStringInfoChar(&cmd, ')');
+
+	/*
+	 * Get the list of tables from publisher, the partition table whose
+	 * ancestor is also in this list will be ignored, otherwise the initial
+	 * data in the partition table would be replicated twice.
+	 */
+	initStringInfo(&cmd);
+	appendStringInfo(&cmd, "WITH pub_tabs AS(\n"
+					 " SELECT DISTINCT n.nspname, c.oid, c.relname, c.relispartition\n"
+					 "  FROM pg_publication p,\n"
+					 "       LATERAL pg_get_publication_tables(p.pubname) gpt,\n"
+					 "       pg_class c JOIN pg_namespace n ON (n.oid = c.relnamespace)\n"
+					 "  WHERE c.oid = gpt.relid AND p.pubname IN ( %s )\n"
+					 ")\n"
+					 "SELECT DISTINCT pub_tabs.nspname, pub_tabs.relname\n"
+					 " FROM pub_tabs\n"
+					 " WHERE (pub_tabs.relispartition IS FALSE\n"
+					 "  OR NOT EXISTS (SELECT 1 FROM pg_partition_ancestors(pub_tabs.oid) as pa\n"
+					 "                  WHERE pa.relid IN (SELECT pub_tabs.oid FROM pub_tabs)\n"
+					 "                   AND pa.relid != pub_tabs.oid))\n",
+					 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/013_partition.pl b/src/test/subscription/t/013_partition.pl
index 58d78b4292..90b7877bc3 100644
--- a/src/test/subscription/t/013_partition.pl
+++ b/src/test/subscription/t/013_partition.pl
@@ -6,7 +6,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 71;
+use Test::More tests => 73;
 
 # setup
 
@@ -408,9 +408,10 @@ $node_publisher->safe_psql('postgres',
 $node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab4 (a int PRIMARY KEY) PARTITION BY LIST (a)");
 $node_publisher->safe_psql('postgres',
-	"CREATE TABLE tab4_1 PARTITION OF tab4 FOR VALUES IN (0, 1) PARTITION BY LIST (a)");
+	"CREATE TABLE tab4_1 PARTITION OF tab4 FOR VALUES IN (-1, 0, 1) PARTITION BY LIST (a)"
+);
 $node_publisher->safe_psql('postgres',
-	"CREATE TABLE tab4_1_1 PARTITION OF tab4_1 FOR VALUES IN (0, 1)");
+	"CREATE TABLE tab4_1_1 PARTITION OF tab4_1 FOR VALUES IN (-1, 0, 1)");
 $node_publisher->safe_psql('postgres',
 	"ALTER PUBLICATION pub_all SET (publish_via_partition_root = true)");
 # Note: tab3_1's parent is not in the publication, in which case its
@@ -426,6 +427,7 @@ $node_publisher->safe_psql('postgres',
 
 # prepare data for the initial sync
 $node_publisher->safe_psql('postgres', "INSERT INTO tab2 VALUES (1)");
+$node_publisher->safe_psql('postgres', "INSERT INTO tab4 VALUES (-1)");
 
 # subscriber 1
 $node_subscriber1->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
@@ -467,12 +469,14 @@ $node_subscriber2->safe_psql('postgres',
 $node_subscriber2->safe_psql('postgres',
 	"CREATE TABLE tab3_1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab3_1', b text)"
 );
+
+# Note: We create two separate tables, not a partitioned one, so that we can
+# easily identity through which relation were the changes replicated.
 $node_subscriber2->safe_psql('postgres',
-	"CREATE TABLE tab4 (a int PRIMARY KEY)"
-);
+	"CREATE TABLE tab4 (a int PRIMARY KEY)");
 $node_subscriber2->safe_psql('postgres',
-	"CREATE TABLE tab4_1 (a int PRIMARY KEY)"
-);
+	"CREATE TABLE tab4_1 (a int PRIMARY KEY)");
+
 # Publication that sub2 points to now publishes via root, so must update
 # subscription target relations.
 $node_subscriber2->safe_psql('postgres',
@@ -486,6 +490,12 @@ $node_subscriber2->wait_for_subscription_sync;
 $result = $node_subscriber1->safe_psql('postgres',
 	"SELECT c, a FROM tab2");
 is( $result, qq(sub1_tab2|1), 'initial data synced for pub_viaroot');
+$result =
+  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4 ORDER BY 1");
+is($result, qq(-1), 'initial data synced for pub_lower_level and pub_all');
+$result =
+  $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4_1 ORDER BY 1");
+is($result, qq(), 'initial data synced for pub_lower_level and pub_all');
 
 # insert
 $node_publisher->safe_psql('postgres', "INSERT INTO tab1 VALUES (1), (0)");
@@ -538,7 +548,8 @@ sub2_tab3|5), 'inserts into tab3 replicated');
 
 $result = $node_subscriber2->safe_psql('postgres',
 	"SELECT a FROM tab4 ORDER BY 1");
-is( $result, qq(0), 'inserts into tab4 replicated');
+is( $result, qq(-1
+0), 'inserts into tab4 replicated');
 
 $result = $node_subscriber2->safe_psql('postgres',
 	"SELECT a FROM tab4_1 ORDER BY 1");
@@ -564,7 +575,8 @@ $node_publisher->wait_for_catchup('sub2');
 # maps to the tab4 relation on subscriber.
 $result = $node_subscriber2->safe_psql('postgres',
 	"SELECT a FROM tab4 ORDER BY 1");
-is( $result, qq(0
+is( $result, qq(-1
+0
 1), 'inserts into tab4 replicated');
 
 $result = $node_subscriber2->safe_psql('postgres',
-- 
2.39.1.windows.1