v50-0008-fixes-0005.patch

From 06578a09b57558a1fce0327c748ecd074415bd63 Mon Sep 17 00:00:00 2001
From: Euler Taveira <euler.taveira@enterprisedb.com>
Date: Mon, 20 Dec 2021 10:05:23 -0300
Subject: [PATCH v50 8/8] fixes 0005

---
 src/backend/replication/logical/tablesync.c | 45 ++++++++++++---------
 src/backend/replication/pgoutput/pgoutput.c | 24 ++++++-----
 2 files changed, 39 insertions(+), 30 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 469aadc1c4..51271a6a00 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -802,15 +802,20 @@ fetch_remote_table_info(char *nspname, char *relname,
 	walrcv_clear_result(res);
 
 	/*
-	 * If any publication has puballtables true then all row-filtering is
-	 * ignored.
+	 * Get relation row filter expressions. DISTINCT avoids the same expression
+	 * of a table in multiple publications from being included multiple times
+	 * in the final expression.
 	 *
-	 * If the relation is a member of a schema of a subscribed publication that
-	 * said ALL TABLES IN SCHEMA then all row-filtering is ignored.
+	 * For initial synchronization, row filtering can be ignored in 2 cases:
 	 *
-	 * Get relation qual. DISTINCT avoids the same expression of a table in
-	 * multiple publications from being included multiple times in the final
-	 * expression.
+	 * 1) one of the subscribed publications has puballtables set to true
+	 * 2) one of the subscribed publications is declared as ALL TABLES IN
+	 * SCHEMA that includes this relation
+	 *
+	 * These 2 cases don't allow row filter expressions, so a absence of
+	 * relation row filter expressions is a sufficient condition to copy the
+	 * entire table, although other publications contain a row filter for this
+	 * relation.
 	 */
 	if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
 	{
@@ -831,23 +836,23 @@ fetch_remote_table_info(char *nspname, char *relname,
 			appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
 		}
 
-		/* Check for row-filters */
+		/* Check for row filters. */
 		resetStringInfo(&cmd);
 		appendStringInfo(&cmd,
-						 "SELECT DISTINCT pg_get_expr(prqual, prrelid) "
-						 "  FROM pg_publication p "
-						 "  INNER JOIN pg_publication_rel pr ON (p.oid = pr.prpubid) "
+						 "SELECT DISTINCT pg_catalog.pg_get_expr(pr.prqual, pr.prrelid) "
+						 "  FROM pg_catalog.pg_publication p "
+						 "  INNER JOIN pg_catalog.pg_publication_rel pr ON (p.oid = pr.prpubid) "
 						 "    WHERE pr.prrelid = %u AND p.pubname IN ( %s ) "
-						 "    AND NOT (select bool_or(puballtables) "
-						 "      FROM pg_publication "
-						 "      WHERE pubname in ( %s )) "
-						 "    AND (SELECT count(1)=0 "
-						 "      FROM pg_publication_namespace pn, pg_class c "
-						 "      WHERE c.oid = %u AND c.relnamespace = pn.pnnspid)",
+						 "    AND NOT (SELECT pg_catalog.bool_or(b.puballtables) "
+						 "      FROM pg_catalog.pg_publication b "
+						 "      WHERE b.pubname IN ( %s )) "
+						 "    AND NOT EXISTS( "
+						 "      SELECT 1 FROM pg_catalog.pg_publication_namespace pn "
+						 "      INNER JOIN pg_catalog.pg_class c ON (pn.pnnspid = c.relnamespace) "
+						 "      WHERE c.oid = pr.prrelid)",
 						lrel->remoteid,
 						pub_names.data,
-						pub_names.data,
-						lrel->remoteid);
+						pub_names.data);
 
 		res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
 
@@ -865,7 +870,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 		slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
 		while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
 		{
-			Datum rf = slot_getattr(slot, 1, &isnull);
+			Datum		rf = slot_getattr(slot, 1, &isnull);
 
 			if (!isnull)
 				*qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index a926ffc566..30a6adb2d2 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -805,10 +805,11 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
 		 * there are multiple lists (one for each operation) which row filters
 		 * will be appended.
 		 *
-		 * NOTE: FOR ALL TABLES implies "use no filters" so it takes precedence
+		 * NOTE: FOR ALL TABLES implies "don't use row filter expression" so it
+		 * takes precedence.
 		 *
-		 * NOTE: ALL TABLES IN SCHEMA also implies "use not filters" if the
-		 * table is a member of the same schema.
+		 * NOTE: ALL TABLES IN SCHEMA implies "don't use row filter expression"
+		 * if the schema is the same as the table schema.
 		 */
 		foreach(lc, data->publications)
 		{
@@ -820,8 +821,8 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
 
 			/*
 			 * If the publication is FOR ALL TABLES then it is treated the same
-			 * as if this table has no filters (even if for some other
-			 * publication it does).
+			 * as if this table has no row filters (even if for other
+			 * publications it does).
 			 */
 			if (pub->alltables)
 			{
@@ -832,17 +833,19 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
 				if (pub->pubactions.pubdelete)
 					no_filter[idx_del] = true;
 
-				/* Quick exit loop if all pubactions have no row-filter. */
+				/* Quick exit loop if all pubactions have no row filter. */
 				if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
 					break;
 
+				/* No additional work for this publication. Next one. */
 				continue;
 			}
 
 			/*
-			 * If the publication is FOR ALL TABLES IN SCHEMA and it overlaps with the
-			 * current relation in the same schema then this is also treated same as if
-			 * this table has no filters (even if for some other publication it does).
+			 * If the publication is FOR ALL TABLES IN SCHEMA and it overlaps
+			 * with the current relation in the same schema then this is also
+			 * treated same as if this table has no row filters (even if for
+			 * other publications it does).
 			 */
 			schemarelids = GetAllSchemaPublicationRelations(pub->oid,
 															pub->pubviaroot ?
@@ -859,10 +862,11 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
 
 				list_free(schemarelids);
 
-				/* Quick exit loop if all pubactions have no row-filter. */
+				/* Quick exit loop if all pubactions have no row filter. */
 				if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
 					break;
 
+				/* No additional work for this publication. Next one. */
 				continue;
 			}
 			list_free(schemarelids);
-- 
2.20.1