v48-0005-Row-filter-handle-FOR-ALL-TABLES.patch

From 8bbe95240ccdba84d533ca54508ded8ef3d60824 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Fri, 17 Dec 2021 20:41:40 +1100
Subject: [PATCH v48] 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.

If one of the subscriber's publications was created using FOR ALL TABLES IN
SCHEMA and the table belong to that same schmea, then that also implies NO
row-filtering will be applied.

These rules overrides 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
---
 src/backend/replication/logical/tablesync.c |  48 +++++++----
 src/backend/replication/pgoutput/pgoutput.c |  63 ++++++++++++++-
 src/test/subscription/t/027_row_filter.pl   | 118 ++++++++++++++++++++++++++--
 3 files changed, 202 insertions(+), 27 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index c20c221..469aadc 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -802,21 +802,22 @@ fetch_remote_table_info(char *nspname, char *relname,
 	walrcv_clear_result(res);
 
 	/*
+	 * If any publication has puballtables true then all row-filtering is
+	 * ignored.
+	 *
+	 * 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.
+	 *
 	 * 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)
 	{
-		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) "
-						 " WHERE pr.prrelid = %u "
-						 "   AND p.pubname IN (", lrel->remoteid);
+		StringInfoData pub_names;
 
+		/* Build the pubname list. */
+		initStringInfo(&pub_names);
 		first = true;
 		foreach(lc, MySubscription->publications)
 		{
@@ -825,11 +826,28 @@ fetch_remote_table_info(char *nspname, char *relname,
 			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, ')');
+
+		/* 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) "
+						 "    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)",
+						lrel->remoteid,
+						pub_names.data,
+						pub_names.data,
+						lrel->remoteid);
 
 		res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
 
@@ -847,18 +865,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 8a733cb..43d0125 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -912,13 +912,68 @@ 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
+		 *
+		 * NOTE: ALL TABLES IN SCHEMA also implies "use not filters" if the
+		 * table is a member of the same schema.
 		 */
 		foreach(lc, data->publications)
 		{
-			Publication *pub = lfirst(lc);
-			HeapTuple	rftuple;
-			Datum		rfdatum;
-			bool		rfisnull;
+			Publication	   *pub = lfirst(lc);
+			HeapTuple		rftuple;
+			Datum			rfdatum;
+			bool			rfisnull;
+			List		   *schemarelids = NIL;
+
+			/*
+			 * 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).
+			 */
+			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;
+
+				/* Quick exit loop if all pubactions have no row-filter. */
+				if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
+					break;
+
+				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).
+			 */
+			schemarelids = GetAllSchemaPublicationRelations(pub->oid,
+															pub->pubviaroot ?
+															PUBLICATION_PART_ROOT :
+															PUBLICATION_PART_LEAF);
+			if (list_member_oid(schemarelids, entry->relid))
+			{
+				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;
+
+				list_free(schemarelids);
+
+				/* Quick exit loop if all pubactions have no row-filter. */
+				if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
+					break;
+
+				continue;
+			}
+			list_free(schemarelids);
 
 			/*
 			 * Lookup if there is a row-filter, and if yes remember it in a list (per
diff --git a/src/test/subscription/t/027_row_filter.pl b/src/test/subscription/t/027_row_filter.pl
index a2f25f6..73add45 100644
--- a/src/test/subscription/t/027_row_filter.pl
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
-use Test::More tests => 10;
+use Test::More tests => 14;
 
 # create publisher node
 my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
@@ -15,6 +15,116 @@ my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
 $node_subscriber->init(allows_streaming => 'logical');
 $node_subscriber->start;
 
+my $synced_query =
+  "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+my $appname           = 'tap_sub';
+
+# ====================================================================
+# Testcase start: FOR ALL TABLES
+#
+# The FOR ALL TABLES test must come first so that it is not affected by
+# all the other test tables that are later created.
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres', "CREATE TABLE tab_rf_x (x int primary key)");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE tab_rf_x (x int primary key)");
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO tab_rf_x (x) VALUES (0), (5), (10), (15), (20)");
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_x FOR TABLE tab_rf_x WHERE (x > 10)");
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_forall FOR ALL TABLES");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_x, tap_pub_forall");
+
+$node_publisher->wait_for_catchup($appname);
+# wait for initial table synchronization to finish
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+# The subscription of the FOR ALL TABLES publication means there should be no
+# filtering on the tablesync COPY, so all expect all 5 will be present.
+my $result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM tab_rf_x");
+is($result, qq(5), 'check initial data copy from table tab_rf_x should not be filtered');
+
+# Similarly, normal filtering after the initial phase will also have not effect.
+# Expected: 5 initial rows + 2 new rows = 7 rows
+$node_publisher->safe_psql('postgres', "INSERT INTO tab_rf_x (x) VALUES (-99), (99)");
+$node_publisher->wait_for_catchup($appname);
+$result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM tab_rf_x");
+is($result, qq(7), 'check table tab_rf_x should not be filtered');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_forall");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_x");
+$node_publisher->safe_psql('postgres', "DROP TABLE tab_rf_x");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_subscriber->safe_psql('postgres', "DROP TABLE tab_rf_x");
+
+# Testcase end: FOR ALL TABLES
+# ====================================================================
+
+# ====================================================================
+# Testcase start: ALL TABLES IN SCHEMA
+#
+# The ALL TABLES IN SCHEMA test is independent of all other test cases so it
+# cleans up after itself.
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres', "CREATE SCHEMA schema_rf_x");
+$node_publisher->safe_psql('postgres', "CREATE TABLE schema_rf_x.tab_rf_x (x int primary key)");
+$node_subscriber->safe_psql('postgres', "CREATE SCHEMA schema_rf_x");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE schema_rf_x.tab_rf_x (x int primary key)");
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO schema_rf_x.tab_rf_x (x) VALUES (0), (5), (10), (15), (20)");
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_x FOR TABLE schema_rf_x.tab_rf_x WHERE (x > 10)");
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_allinschema FOR ALL TABLES IN SCHEMA schema_rf_x");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_x, tap_pub_allinschema");
+
+$node_publisher->wait_for_catchup($appname);
+# wait for initial table synchronization to finish
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
+# The subscription of the ALL TABLES IN SCHEMA publication means there should be
+# no filtering on the tablesync COPY, so all expect all 5 will be present.
+$result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM schema_rf_x.tab_rf_x");
+is($result, qq(5), 'check initial data copy from table tab_rf_x should not be filtered');
+
+# Similarly, normal filtering after the initial phase will also have not effect.
+# Expected: 5 initial rows + 2 new rows = 7 rows
+$node_publisher->safe_psql('postgres', "INSERT INTO schema_rf_x.tab_rf_x (x) VALUES (-99), (99)");
+$node_publisher->wait_for_catchup($appname);
+$result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM schema_rf_x.tab_rf_x");
+is($result, qq(7), 'check table tab_rf_x should not be filtered');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_allinschema");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_x");
+$node_publisher->safe_psql('postgres', "DROP TABLE schema_rf_x.tab_rf_x");
+$node_publisher->safe_psql('postgres', "DROP SCHEMA schema_rf_x");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_subscriber->safe_psql('postgres', "DROP TABLE schema_rf_x.tab_rf_x");
+$node_subscriber->safe_psql('postgres', "DROP SCHEMA schema_rf_x");
+
+# Testcase end: ALL TABLES IN SCHEMA
+# ====================================================================
+
 # setup structure on publisher
 $node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_1 (a int primary key, b text)");
@@ -127,8 +237,6 @@ $node_publisher->safe_psql('postgres',
 $node_publisher->safe_psql('postgres',
 	"INSERT INTO tab_rowfilter_greater_10k (a, b) VALUES(16000, 103)");
 
-my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
-my $appname           = 'tap_sub';
 $node_subscriber->safe_psql('postgres',
 	"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_1, tap_pub_2, tap_pub_3, tap_pub_4a, tap_pub_4b"
 );
@@ -136,8 +244,6 @@ $node_subscriber->safe_psql('postgres',
 $node_publisher->wait_for_catchup($appname);
 
 # wait for initial table synchronization to finish
-my $synced_query =
-  "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
 $node_subscriber->poll_query_until('postgres', $synced_query)
   or die "Timed out while waiting for subscriber to synchronize data";
 
@@ -148,7 +254,7 @@ $node_subscriber->poll_query_until('postgres', $synced_query)
 # - INSERT (1980, 'not filtered')  YES
 # - generate_series(990,1002)      YES, only for 1001,1002 because a > 1000
 #
-my $result =
+$result =
   $node_subscriber->safe_psql('postgres',
 	"SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2");
 is( $result, qq(1001|test 1001
-- 
1.8.3.1