From 59f959ff6a9f81e1c152ca7f2351d2f6e4e7c886 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Mon, 13 Dec 2021 20:39:33 +1100
Subject: [PATCH v46] Row filter handle FOR ALL TABLES

If one of the subscriber's publications was created using FOR ALL TABLES then
that implies NO row-filtering will be applied.

This overides any other row-filters from other subscribed publications.

Note that the initial COPY does not take publication operations into account.

Author: Peter Smith
Reported By: Tang
Discussion: https://www.postgresql.org/message-id/OS0PR01MB6113D82113AA081ACF710D0CFB6E9%40OS0PR01MB6113.jpnprd01.prod.outlook.com

TODO
- PG docs
- TAP test cases
- Similar case for FOR ALL TABLES IN SCHEMA?
---
 src/backend/replication/logical/tablesync.c | 64 ++++++++++++++++++++++++++---
 src/backend/replication/pgoutput/pgoutput.c | 18 ++++++++
 2 files changed, 76 insertions(+), 6 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 971e037..1f80261 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -700,6 +700,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	TupleTableSlot *slot;
 	Oid			tableRow[] = {OIDOID, CHAROID, CHAROID};
 	Oid			attrRow[] = {TEXTOID, OIDOID, BOOLOID};
+	Oid			alltablesRow[] = {BOOLOID};
 	Oid			qualRow[] = {TEXTOID};
 	bool		isnull;
 	int			natt;
@@ -802,12 +803,67 @@ fetch_remote_table_info(char *nspname, char *relname,
 	walrcv_clear_result(res);
 
 	/*
+	 * If any publication has puballtable true then all row-filtering is
+	 * ignored.
+	 *
 	 * Get relation qual. DISTINCT avoids the same expression of a table in
 	 * multiple publications from being included multiple times in the final
 	 * expression.
 	 */
 	if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
 	{
+		bool puballtables = false;
+
+		/*
+		 * Check for puballtables flag
+		 */
+		resetStringInfo(&cmd);
+		appendStringInfo(&cmd,
+						 "SELECT p.puballtables "
+						 "  FROM pg_publication p "
+						 " WHERE p.puballtables IS TRUE "
+						 "   AND p.pubname IN (");
+
+		first = true;
+		foreach(lc, MySubscription->publications)
+		{
+			char	   *pubname = strVal(lfirst(lc));
+
+			if (first)
+				first = false;
+			else
+				appendStringInfoString(&cmd, ", ");
+
+			appendStringInfoString(&cmd, quote_literal_cstr(pubname));
+		}
+		appendStringInfoChar(&cmd, ')');
+
+		res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, alltablesRow);
+
+		if (res->status != WALRCV_OK_TUPLES)
+			ereport(ERROR,
+					(errmsg("could not fetch puballtables flag for table \"%s.%s\" from publisher: %s",
+							nspname, relname, res->err)));
+
+		slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+		puballtables = tuplestore_gettupleslot(res->tuplestore, true, false, slot);
+		ExecDropSingleTupleTableSlot(slot);
+		walrcv_clear_result(res);
+
+		if (puballtables)
+		{
+			if (*qual)
+			{
+				list_free_deep(*qual);
+				*qual = NIL;
+			}
+			pfree(cmd.data);
+			return;
+		}
+
+		/*
+		 * Check for row-filters
+		 */
 		resetStringInfo(&cmd);
 		appendStringInfo(&cmd,
 						 "SELECT DISTINCT pg_get_expr(prqual, prrelid) "
@@ -847,18 +903,14 @@ 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)));
 
 			ExecClearTuple(slot);
 
-			/*
-			 * One entry without a row filter expression means clean up
-			 * previous expressions (if there are any) and return with no
-			 * expressions.
-			 */
+			/* Ignore filters and cleanup as necessary. */
 			if (isnull)
 			{
 				if (*qual)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index c072b25..1a13d70 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -917,6 +917,8 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
 		 * relation. Since row filter usage depends on the DML operation,
 		 * 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
 		 */
 		foreach(lc, data->publications)
 		{
@@ -926,6 +928,22 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
 			bool		rfisnull;
 
 			/*
+			 * If the publication is FOR ALL TABLES then it is treated same as if this
+			 * table has no filters (even if for some other publication it does).
+			*/
+			if (pub->alltables)
+			{
+				if (pub->pubactions.pubinsert)
+					no_filter[idx_ins] = true;
+				if (pub->pubactions.pubupdate)
+					no_filter[idx_upd] = true;
+				if (pub->pubactions.pubdelete)
+					no_filter[idx_del] = true;
+
+				continue;
+			}
+
+			/*
 			 * Lookup if there is a row-filter, and if yes remember it in a list (per
 			 * pubaction). If no, then remember there was no filter for this pubaction.
 			 * Code following this 'publications' loop will combine all filters.
-- 
1.8.3.1