v41-0004-PS-Combine-multiple-filters-with-OR-instead-of-A.patch

From 77dfa2d4ed8ae230aabbd2add1680b825c84d0b1 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <ajinc@fast.au.fujitsu.com>
Date: Mon, 22 Nov 2021 21:27:36 -0500
Subject: [PATCH v41 4/6] PS - Combine multiple filters with OR instead of AND.

This is a change of behavior requested by Tomas [1]. The subscription now is
treated "as a union of all the publications" so the filters are combined with
OR instead of AND.

If the subscription has several publications in which the same table has been
published with different filters, those expressions get OR'ed together so that
rows satisfying any of the expressions will be replicated.

Notice this means if one of the publications has no filter at all then all other
filters become redundant.

Updated documentation.

Added more test cases.

Author: Peter Smith
[1] https://www.postgresql.org/message-id/574b4e78-2f35-acf3-4bdc-4b872582e739%40enterprisedb.com
---
 doc/src/sgml/ref/create_subscription.sgml   | 27 +++++++----
 src/backend/replication/logical/tablesync.c | 27 +++++++++--
 src/backend/replication/pgoutput/pgoutput.c | 25 ++++++++++-
 src/test/subscription/t/026_row_filter.pl   | 69 ++++++++++++++++++++++++++---
 4 files changed, 128 insertions(+), 20 deletions(-)

diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 5a9430e..42bf8c2 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -206,15 +206,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
          <para>
           Specifies whether to copy pre-existing data in the publications
           that are being subscribed to when the replication starts.
-          The default is <literal>true</literal>. If any table in the
-          publications has a <literal>WHERE</literal> clause, rows that do not
-          satisfy the <replaceable class="parameter">expression</replaceable>
-          will not be copied. If the subscription has several publications in
-          which a table has been published with different
-          <literal>WHERE</literal> clauses, rows must satisfy all expressions
-          to be copied. If the subscriber is a
-          <productname>PostgreSQL</productname> version before 15 then any row
-          filtering is ignored.
+          The default is <literal>true</literal>.
+         </para>
+         <para>
+          Row-filtering may also apply here and will affect what data is
+          copied. Refer to the Notes section below.
          </para>
         </listitem>
        </varlistentry>
@@ -327,6 +323,19 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
    the parameter <literal>create_slot = false</literal>.  This is an
    implementation restriction that might be lifted in a future release.
   </para>
+
+  <para>
+   If any table in the publication has a <literal>WHERE</literal> clause, rows
+   that do not satisfy the <replaceable class="parameter">expression</replaceable>
+   will not be replicated. If the subscription has several publications in
+   which the same table has been published with different filters, those
+   expressions get OR'ed together so that rows satisfying any of the expressions
+   will be replicated. Notice this means if one of the publications has no filter
+   at all then all other filters become redundant. If the subscriber is a
+   <productname>PostgreSQL</productname> version before 15 then any row filtering
+   is ignored.
+  </para>
+
  </refsect1>
 
  <refsect1>
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 9d86a10..e9b7f7c 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -838,6 +838,13 @@ fetch_remote_table_info(char *nspname, char *relname,
 					(errmsg("could not fetch relation qualifications for table \"%s.%s\" from publisher: %s",
 							nspname, relname, res->err)));
 
+		/*
+		 * Multiple row-filter expressions for the same publication will later be
+		 * combined by the COPY using OR, but this means if any of the filters is
+		 * null, then effectively none of the other filters is meaningful. So this
+		 * loop is also checking for null filters and can exit early if any are
+		 * encountered.
+		 */
 		slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
 		while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
 		{
@@ -847,6 +854,20 @@ fetch_remote_table_info(char *nspname, char *relname,
 				*qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
 
 			ExecClearTuple(slot);
+
+			if (isnull)
+			{
+				/*
+				 * A single null filter nullifies the effect of any other filter for this
+				 * table.
+				 */
+				if (*qual)
+				{
+					list_free(*qual);
+					*qual = NIL;
+				}
+				break;
+			}
 		}
 		ExecDropSingleTupleTableSlot(slot);
 
@@ -896,7 +917,7 @@ copy_table(Relation rel)
 		/*
 		 * For non-tables, we need to do COPY (SELECT ...), but we can't just
 		 * do SELECT * because we need to not copy generated columns. For
-		 * tables with any row filters, build a SELECT query with AND'ed row
+		 * tables with any row filters, build a SELECT query with OR'ed row
 		 * filters for COPY.
 		 */
 		appendStringInfoString(&cmd, "COPY (SELECT ");
@@ -908,7 +929,7 @@ copy_table(Relation rel)
 		}
 		appendStringInfo(&cmd, " FROM %s",
 						 quote_qualified_identifier(lrel.nspname, lrel.relname));
-		/* list of AND'ed filters */
+		/* list of OR'ed filters */
 		if (qual != NIL)
 		{
 			ListCell   *lc;
@@ -922,7 +943,7 @@ copy_table(Relation rel)
 				if (first)
 					first = false;
 				else
-					appendStringInfoString(&cmd, " AND ");
+					appendStringInfoString(&cmd, " OR ");
 				appendStringInfoString(&cmd, q);
 			}
 			list_free_deep(qual);
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index fd024d4..b332057 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -797,6 +797,11 @@ pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple, H
 		 * NOTE: If the relation is a partition and pubviaroot is true, use
 		 * the row filter of the topmost partitioned table instead of the row
 		 * filter of its own partition.
+		 *
+		 * NOTE: Multiple row-filters for the same table are combined by OR-ing
+		 * them together, but this means that if (in any of the publications)
+		 * there is *no* filter then effectively none of the other filters have
+		 * any meaning either.
 		 */
 		foreach(lc, data->publications)
 		{
@@ -825,12 +830,28 @@ pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple, H
 				}
 
 				ReleaseSysCache(rftuple);
+
+				if (rfisnull)
+				{
+					/*
+					 * If there is no row-filter, then any other row-filters for this table
+					 * also have no effect (because filters get OR-ed together) so we can
+					 * just discard anything found so far and exit early from the publications
+					 * loop.
+					 */
+					if (rfnodes)
+					{
+						list_free_deep(rfnodes);
+						rfnodes = NIL;
+					}
+					break;
+				}
 			}
 
 		} /* loop all subscribed publications */
 
 		/*
-		 * Combine all the row-filters (if any) into a single filter, and then build the ExprState for it
+		 * Combine using all the row-filters (if any) into a single filter, and then build the ExprState for it
 		 */
 		n_filters = list_length(rfnodes);
 		if (n_filters > 0)
@@ -838,7 +859,7 @@ pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple, H
 			Node	   *rfnode;
 
 			oldctx = MemoryContextSwitchTo(CacheMemoryContext);
-			rfnode = n_filters > 1 ? makeBoolExpr(AND_EXPR, rfnodes, -1) : linitial(rfnodes);
+			rfnode = n_filters > 1 ? makeBoolExpr(OR_EXPR, rfnodes, -1) : linitial(rfnodes);
 			entry->exprstate = pgoutput_row_filter_init_expr(rfnode);
 			MemoryContextSwitchTo(oldctx);
 		}
diff --git a/src/test/subscription/t/026_row_filter.pl b/src/test/subscription/t/026_row_filter.pl
index e806b5d..64e71d0 100644
--- a/src/test/subscription/t/026_row_filter.pl
+++ b/src/test/subscription/t/026_row_filter.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
-use Test::More tests => 7;
+use Test::More tests => 10;
 
 # create publisher node
 my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
@@ -23,6 +23,8 @@ $node_publisher->safe_psql('postgres',
 $node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_3 (a int primary key, b boolean)");
 $node_publisher->safe_psql('postgres',
+	"CREATE TABLE tab_rowfilter_4 (c int primary key)");
+$node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_partitioned (a int primary key, b integer) PARTITION BY RANGE(a)"
 );
 $node_publisher->safe_psql('postgres',
@@ -45,6 +47,8 @@ $node_subscriber->safe_psql('postgres',
 $node_subscriber->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_3 (a int primary key, b boolean)");
 $node_subscriber->safe_psql('postgres',
+	"CREATE TABLE tab_rowfilter_4 (c int primary key)");
+$node_subscriber->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_partitioned (a int primary key, b integer) PARTITION BY RANGE(a)"
 );
 $node_subscriber->safe_psql('postgres',
@@ -86,6 +90,13 @@ $node_publisher->safe_psql('postgres',
 	"CREATE PUBLICATION tap_pub_not_used FOR TABLE tab_rowfilter_1 WHERE (a < 0)"
 );
 
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_4a FOR TABLE tab_rowfilter_4 WHERE (c % 2 = 0)"
+);
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_4b FOR TABLE tab_rowfilter_4"
+);
+
 #
 # The following INSERTs are executed before the CREATE SUBSCRIPTION, so these
 # SQL commands are for testing the initial data copy using logical replication.
@@ -103,6 +114,8 @@ $node_publisher->safe_psql('postgres',
 	"INSERT INTO tab_rowfilter_2 (c) SELECT generate_series(1, 20)");
 $node_publisher->safe_psql('postgres',
 	"INSERT INTO tab_rowfilter_3 (a, b) SELECT x, (x % 3 = 0) FROM generate_series(1, 10) x");
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO tab_rowfilter_4 (c) SELECT generate_series(1, 10)");
 
 # insert data into partitioned table and directly on the partition
 $node_publisher->safe_psql('postgres',
@@ -115,7 +128,7 @@ $node_publisher->safe_psql('postgres',
 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"
+	"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"
 );
 
 $node_publisher->wait_for_catchup($appname);
@@ -143,14 +156,26 @@ is( $result, qq(1001|test 1001
 # Check expected replicated rows for tab_rowfilter_2
 # tap_pub_1 filter is: (c % 2 = 0)
 # tap_pub_2 filter is: (c % 3 = 0)
-# When there are multiple publications for the same table, all filter
-# expressions should succeed. In this case, rows are replicated if c value is
-# divided by 2 AND 3 (6, 12, 18).
+# When there are multiple publications for the same table, the filters
+# expressions are OR'ed together. In this case, rows are replicated if
+# c value is divided by 2 OR 3 (2, 3, 4, 6, 8, 9, 10, 12, 14, 15, 16, 18, 20)
 #
 $result =
   $node_subscriber->safe_psql('postgres',
 	"SELECT count(c), min(c), max(c) FROM tab_rowfilter_2");
-is($result, qq(3|6|18), 'check initial data copy from table tab_rowfilter_2');
+is($result, qq(13|2|20), 'check initial data copy from table tab_rowfilter_2');
+
+# Check expected replicated rows for tab_rowfilter_4
+# (same table in two publications but only one has a filter).
+# tap_pub_4a filter is: (c % 2 = 0)
+# tap_pub_4b filter is: <no filter>
+# Expressions are OR'ed together but when there is no filter it just means
+# OR everything - e.g. same as no filter at all.
+# Expect all rows: (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(c), min(c), max(c) FROM tab_rowfilter_4");
+is($result, qq(10|1|10), 'check initial data copy from table tab_rowfilter_4');
 
 # Check expected replicated rows for tab_rowfilter_3
 # There is no filter. 10 rows are inserted, so 10 rows are replicated.
@@ -210,9 +235,41 @@ $node_publisher->safe_psql('postgres',
 	"UPDATE tab_rowfilter_1 SET b = 'test 1601 updated' WHERE a = 1601");
 $node_publisher->safe_psql('postgres',
 	"DELETE FROM tab_rowfilter_1 WHERE a = 1700");
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO tab_rowfilter_2 (c) VALUES (21), (22), (23), (24), (25)");
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO tab_rowfilter_4 (c) VALUES (0), (11), (12)");
 
 $node_publisher->wait_for_catchup($appname);
 
+# Check expected replicated rows for tab_rowfilter_2
+# tap_pub_1 filter is: (c % 2 = 0)
+# tap_pub_2 filter is: (c % 3 = 0)
+# When there are multiple publications for the same table, the filters
+# expressions are OR'ed together. In this case, rows are replicated if
+# c value is divided by 2 OR 3.
+#
+# Expect original rows (2, 3, 4, 6, 8, 9, 10, 12, 14, 15, 16, 18, 20)
+# Plus (21, 22, 24)
+#
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(c), min(c), max(c) FROM tab_rowfilter_2");
+is($result, qq(16|2|24), 'check replicated rows to tab_rowfilter_2');
+
+# Check expected replicated rows for tab_rowfilter_4
+# (same table in two publications but only one has a filter).
+# tap_pub_4a filter is: (c % 2 = 0)
+# tap_pub_4b filter is: <no filter>
+# Expressions are OR'ed together but when there is no filter it just means
+# OR everything - e.g. same as no filter at all.
+# Expect all rows from initial copy: (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+# And also (0, 11, 12)
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(c), min(c), max(c) FROM tab_rowfilter_4");
+is($result, qq(13|0|12), 'check replicated rows to tab_rowfilter_4');
+
 # Check expected replicated rows for tab_rowfilter_1
 # tap_pub_1 filter is: (a > 1000 AND b <> 'filtered')
 #
-- 
1.8.3.1