v27-0002-publication-parameter-row_filter_on_update.patch

From f00cfd18a7d10254a0539213e38129ed56823869 Mon Sep 17 00:00:00 2001
From: Euler Taveira <euler.taveira@enterprisedb.com>
Date: Wed, 18 Aug 2021 19:32:23 -0300
Subject: [PATCH v27 2/2] publication parameter: row_filter_on_update

This parameter controls which tuple to be used to evaluate the
expression provided by the WHERE clause on UPDATE operations. The
allowed values are new and old. The default is new.
This patch introduces a new List per entry that contains row filter
data. Hence, it might have two row filters with different
row_filter_on_update to test old and new tuples.
---
 doc/src/sgml/ref/create_publication.sgml    |  12 ++
 src/backend/catalog/pg_publication.c        |   1 +
 src/backend/commands/publicationcmds.c      |  44 ++++++-
 src/backend/replication/pgoutput/pgoutput.c | 132 ++++++++++++++------
 src/include/catalog/pg_publication.h        |  11 ++
 src/test/regress/expected/publication.out   |   3 +
 src/test/regress/sql/publication.sql        |   2 +
 src/test/subscription/t/025_row_filter.pl   |  65 +++++++++-
 8 files changed, 224 insertions(+), 46 deletions(-)

diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index 8f78fbbd90..ba1eac08ce 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -146,6 +146,18 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
          </para>
         </listitem>
        </varlistentry>
+
+       <varlistentry>
+        <term><literal>row_filter_on_update</literal> (<type>string</type>)</term>
+        <listitem>
+         <para>
+          This parameter controls which tuple to be used to evaluate the
+          expression for <literal>UPDATE</literal> operations. The allowed
+          values are <literal>new</literal> and <literal>old</literal>. The
+          default is to use the new tuple.
+         </para>
+        </listitem>
+       </varlistentry>
       </variablelist></para>
     </listitem>
    </varlistentry>
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 6057fc3220..77c4f83c7f 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -474,6 +474,7 @@ GetPublication(Oid pubid)
 	pub->pubactions.pubdelete = pubform->pubdelete;
 	pub->pubactions.pubtruncate = pubform->pubtruncate;
 	pub->pubviaroot = pubform->pubviaroot;
+	pub->pubrowfilterupd = pubform->pubrowfilterupd;
 
 	ReleaseSysCache(tup);
 
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 0df7ffbe54..00373db654 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -60,12 +60,15 @@ parse_publication_options(ParseState *pstate,
 						  bool *publish_given,
 						  PublicationActions *pubactions,
 						  bool *publish_via_partition_root_given,
-						  bool *publish_via_partition_root)
+						  bool *publish_via_partition_root,
+						  bool *row_filter_on_update_given,
+						  char *row_filter_on_update)
 {
 	ListCell   *lc;
 
 	*publish_given = false;
 	*publish_via_partition_root_given = false;
+	*row_filter_on_update_given = false;
 
 	/* defaults */
 	pubactions->pubinsert = true;
@@ -73,6 +76,7 @@ parse_publication_options(ParseState *pstate,
 	pubactions->pubdelete = true;
 	pubactions->pubtruncate = true;
 	*publish_via_partition_root = false;
+	*row_filter_on_update = PUB_ROW_FILTER_UPD_NEW_TUPLE;
 
 	/* Parse options */
 	foreach(lc, options)
@@ -131,6 +135,24 @@ parse_publication_options(ParseState *pstate,
 			*publish_via_partition_root_given = true;
 			*publish_via_partition_root = defGetBoolean(defel);
 		}
+		else if (strcmp(defel->defname, "row_filter_on_update") == 0)
+		{
+			char	*rowfilterupd;
+
+			if (*row_filter_on_update_given)
+				errorConflictingDefElem(defel, pstate);
+			*row_filter_on_update_given = true;
+			rowfilterupd = defGetString(defel);
+
+			if (strcmp(rowfilterupd, "new") == 0)
+				*row_filter_on_update = PUB_ROW_FILTER_UPD_NEW_TUPLE;
+			else if (strcmp(rowfilterupd, "old") == 0)
+				*row_filter_on_update = PUB_ROW_FILTER_UPD_OLD_TUPLE;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("unrecognized \"row_filter_on_update\" value: \"%s\"", rowfilterupd)));
+		}
 		else
 			ereport(ERROR,
 					(errcode(ERRCODE_SYNTAX_ERROR),
@@ -154,6 +176,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
 	PublicationActions pubactions;
 	bool		publish_via_partition_root_given;
 	bool		publish_via_partition_root;
+	bool		row_filter_on_update_given;
+	char		row_filter_on_update;
 	AclResult	aclresult;
 
 	/* must have CREATE privilege on database */
@@ -193,7 +217,9 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
 							  stmt->options,
 							  &publish_given, &pubactions,
 							  &publish_via_partition_root_given,
-							  &publish_via_partition_root);
+							  &publish_via_partition_root,
+							  &row_filter_on_update_given,
+							  &row_filter_on_update);
 
 	puboid = GetNewOidWithIndex(rel, PublicationObjectIndexId,
 								Anum_pg_publication_oid);
@@ -210,6 +236,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
 		BoolGetDatum(pubactions.pubtruncate);
 	values[Anum_pg_publication_pubviaroot - 1] =
 		BoolGetDatum(publish_via_partition_root);
+	values[Anum_pg_publication_pubrowfilterupd - 1] =
+		CharGetDatum(row_filter_on_update);
 
 	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
@@ -264,6 +292,8 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
 	PublicationActions pubactions;
 	bool		publish_via_partition_root_given;
 	bool		publish_via_partition_root;
+	bool		row_filter_on_update_given;
+	char		row_filter_on_update;
 	ObjectAddress obj;
 	Form_pg_publication pubform;
 
@@ -271,7 +301,9 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
 							  stmt->options,
 							  &publish_given, &pubactions,
 							  &publish_via_partition_root_given,
-							  &publish_via_partition_root);
+							  &publish_via_partition_root,
+							  &row_filter_on_update_given,
+							  &row_filter_on_update);
 
 	/* Everything ok, form a new tuple. */
 	memset(values, 0, sizeof(values));
@@ -299,6 +331,12 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
 		replaces[Anum_pg_publication_pubviaroot - 1] = true;
 	}
 
+	if (row_filter_on_update_given)
+	{
+		values[Anum_pg_publication_pubrowfilterupd - 1] = CharGetDatum(row_filter_on_update);
+		replaces[Anum_pg_publication_pubrowfilterupd - 1] = true;
+	}
+
 	tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
 							replaces);
 
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 1220203af7..5313cd0b78 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -95,6 +95,19 @@ static void send_repl_origin(LogicalDecodingContext *ctx,
 							 RepOriginId origin_id, XLogRecPtr origin_lsn,
 							 bool send_origin);
 
+/*
+ * One relation can have multiple row filters. This structure has data for each
+ * row filter including an ExprState and TupleTableSlot for cache purposes and
+ * also a variable that indicates which tuple the row filter uses for UPDATE
+ * actions.
+ */
+typedef struct RowFilterState
+{
+	ExprState		*exprstate;
+	TupleTableSlot  *scantuple;
+	char			row_filter_on_update;
+} RowFilterState;
+
 /*
  * Entry in the map used to remember which relation schemas we sent.
  *
@@ -123,8 +136,7 @@ typedef struct RelationSyncEntry
 
 	bool		replicate_valid;
 	PublicationActions pubactions;
-	List	   *exprstate;			/* ExprState for row filter */
-	TupleTableSlot	*scantuple;		/* tuple table slot for row filter */
+	List	   *rfstate;		/* row filter list */
 
 	/*
 	 * OID of the relation to publish changes as.  For a partition, this may
@@ -161,7 +173,7 @@ static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
 static EState *create_estate_for_relation(Relation rel);
 static ExprState *pgoutput_row_filter_init_expr(Node *rfnode);
 static bool pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext);
-static bool pgoutput_row_filter(Relation relation, HeapTuple oldtuple,
+static bool pgoutput_row_filter(Relation relation, int action, HeapTuple oldtuple,
 								HeapTuple newtuple, RelationSyncEntry *entry);
 
 /*
@@ -731,15 +743,14 @@ pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
  * If it returns true, the change is replicated, otherwise, it is not.
  */
 static bool
-pgoutput_row_filter(Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry)
+pgoutput_row_filter(Relation relation, int action, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry)
 {
 	EState	   *estate;
-	ExprContext *ecxt;
 	ListCell   *lc;
 	bool		result = true;
 
 	/* Bail out if there is no row filter */
-	if (entry->exprstate == NIL)
+	if (entry->rfstate == NIL)
 		return true;
 
 	elog(DEBUG3, "table \"%s.%s\" has row filter",
@@ -750,31 +761,62 @@ pgoutput_row_filter(Relation relation, HeapTuple oldtuple, HeapTuple newtuple, R
 
 	estate = create_estate_for_relation(relation);
 
-	/* Prepare context per tuple */
-	ecxt = GetPerTupleExprContext(estate);
-	ecxt->ecxt_scantuple = entry->scantuple;
-
-	ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
-
 	/*
 	 * If the subscription has multiple publications and the same table has a
 	 * different row filter in these publications, all row filters must be
 	 * matched in order to replicate this change.
 	 */
-	foreach(lc, entry->exprstate)
+	foreach(lc, entry->rfstate)
 	{
-		ExprState  *exprstate = (ExprState *) lfirst(lc);
+		ExprContext		*ecxt;
+		RowFilterState	*rfstate = (RowFilterState *) lfirst(lc);
+
+		/* Bail out if row_filter_on_update = old and old tuple is NULL */
+		if (action == REORDER_BUFFER_CHANGE_UPDATE && oldtuple == NULL &&
+				rfstate->row_filter_on_update == PUB_ROW_FILTER_UPD_OLD_TUPLE)
+			return false;
+
+		/* Prepare context per tuple */
+		ecxt = GetPerTupleExprContext(estate);
+		ecxt->ecxt_scantuple = rfstate->scantuple;
+
+		/*
+		 * Choose which tuple to use for row filter.
+		 * - INSERT: uses new tuple.
+		 * - UPDATE: it can use new tuple or old tuple. The behavior is controlled
+		 *   by the publication parameter row_filter_on_update. The default is new
+		 *   tuple.
+		 * - DELETE: uses old tuple.
+		 */
+		switch (action)
+		{
+			case REORDER_BUFFER_CHANGE_INSERT:
+				ExecStoreHeapTuple(newtuple, ecxt->ecxt_scantuple, false);
+				break;
+			case REORDER_BUFFER_CHANGE_UPDATE:
+				if (rfstate->row_filter_on_update == PUB_ROW_FILTER_UPD_NEW_TUPLE)
+					ExecStoreHeapTuple(newtuple, ecxt->ecxt_scantuple, false);
+				else if (rfstate->row_filter_on_update == PUB_ROW_FILTER_UPD_OLD_TUPLE)
+					ExecStoreHeapTuple(oldtuple, ecxt->ecxt_scantuple, false);
+				else
+					Assert(false);
+				break;
+			case REORDER_BUFFER_CHANGE_DELETE:
+				ExecStoreHeapTuple(oldtuple, ecxt->ecxt_scantuple, false);
+				break;
+		}
 
 		/* Evaluates row filter */
-		result = pgoutput_row_filter_exec_expr(exprstate, ecxt);
+		result = pgoutput_row_filter_exec_expr(rfstate->exprstate, ecxt);
 
 		/* If the tuple does not match one of the row filters, bail out */
 		if (!result)
 			break;
+
+		ResetExprContext(ecxt);
 	}
 
 	/* Cleanup allocated resources */
-	ResetExprContext(ecxt);
 	FreeExecutorState(estate);
 	PopActiveSnapshot();
 
@@ -840,7 +882,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 				HeapTuple	tuple = &change->data.tp.newtuple->tuple;
 
 				/* Check row filter. */
-				if (!pgoutput_row_filter(relation, NULL, tuple, relentry))
+				if (!pgoutput_row_filter(relation, change->action, NULL, tuple, relentry))
 					break;
 
 				/*
@@ -873,7 +915,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 				HeapTuple	newtuple = &change->data.tp.newtuple->tuple;
 
 				/* Check row filter. */
-				if (!pgoutput_row_filter(relation, oldtuple, newtuple, relentry))
+				if (!pgoutput_row_filter(relation, change->action, oldtuple, newtuple, relentry))
 					break;
 
 				maybe_send_schema(ctx, change, relation, relentry);
@@ -907,7 +949,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 				HeapTuple	oldtuple = &change->data.tp.oldtuple->tuple;
 
 				/* Check row filter. */
-				if (!pgoutput_row_filter(relation, oldtuple, NULL, relentry))
+				if (!pgoutput_row_filter(relation, change->action, oldtuple, NULL, relentry))
 					break;
 
 				maybe_send_schema(ctx, change, relation, relentry);
@@ -1320,8 +1362,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
 		entry->replicate_valid = false;
 		entry->pubactions.pubinsert = entry->pubactions.pubupdate =
 			entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
-		entry->scantuple = NULL;
-		entry->exprstate = NIL;
+		entry->rfstate = NIL;
 		entry->publish_as_relid = InvalidOid;
 		entry->map = NULL;		/* will be set by maybe_send_schema() if
 								 * needed */
@@ -1347,21 +1388,26 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
 			publications_valid = true;
 		}
 
-		/* Release tuple table slot */
-		if (entry->scantuple != NULL)
+		foreach(lc, entry->rfstate)
 		{
-			ExecDropSingleTupleTableSlot(entry->scantuple);
-			entry->scantuple = NULL;
-		}
+			RowFilterState	*rfstate = (RowFilterState *) lfirst(lc);
 
-		/*
-		 * Create a tuple table slot for row filter. TupleDesc must live as
-		 * long as the cache remains.
-		 */
-		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
-		tupdesc = CreateTupleDescCopy(tupdesc);
-		entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
-		MemoryContextSwitchTo(oldctx);
+			/* Release tuple table slot */
+			if (rfstate->scantuple != NULL)
+			{
+				ExecDropSingleTupleTableSlot(rfstate->scantuple);
+				rfstate->scantuple = NULL;
+			}
+
+			/*
+			 * Create a tuple table slot for row filter. TupleDesc must live as
+			 * long as the cache remains.
+			 */
+			oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+			tupdesc = CreateTupleDescCopy(tupdesc);
+			rfstate->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+			MemoryContextSwitchTo(oldctx);
+		}
 
 		/*
 		 * Build publication cache. We can't use one provided by relcache as
@@ -1447,15 +1493,19 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
 
 				if (!rfisnull)
 				{
-					Node	   *rfnode;
-					ExprState  *exprstate;
+					Node		   *rfnode;
+					RowFilterState *rfstate;
 
 					oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+					rfstate = palloc0(sizeof(RowFilterState));
 					rfnode = stringToNode(TextDatumGetCString(rfdatum));
 
 					/* Prepare for expression execution */
-					exprstate = pgoutput_row_filter_init_expr(rfnode);
-					entry->exprstate = lappend(entry->exprstate, exprstate);
+					rfstate->exprstate = pgoutput_row_filter_init_expr(rfnode);
+					rfstate->row_filter_on_update = pub->pubrowfilterupd;
+					tupdesc = CreateTupleDescCopy(tupdesc);
+					rfstate->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+					entry->rfstate = lappend(entry->rfstate, rfstate);
 					MemoryContextSwitchTo(oldctx);
 				}
 
@@ -1608,10 +1658,10 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
 		entry->pubactions.pubdelete = false;
 		entry->pubactions.pubtruncate = false;
 
-		if (entry->exprstate != NIL)
+		if (entry->rfstate != NIL)
 		{
-			list_free_deep(entry->exprstate);
-			entry->exprstate = NIL;
+			list_free_deep(entry->rfstate);
+			entry->rfstate = NIL;
 		}
 	}
 
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 2703b9c3fe..e68591c1fd 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -54,6 +54,12 @@ CATALOG(pg_publication,6104,PublicationRelationId)
 
 	/* true if partition changes are published using root schema */
 	bool		pubviaroot;
+
+	/*
+	 * Choose which tuple to use for row filter on UPDATE actions.
+	 * See constants below.
+	 */
+	char		pubrowfilterupd;
 } FormData_pg_publication;
 
 /* ----------------
@@ -81,8 +87,13 @@ typedef struct Publication
 	bool		alltables;
 	bool		pubviaroot;
 	PublicationActions pubactions;
+	char		pubrowfilterupd;
 } Publication;
 
+/* pubrowfilterupd values */
+#define	PUB_ROW_FILTER_UPD_NEW_TUPLE	'n'		/* use new tuple */
+#define	PUB_ROW_FILTER_UPD_OLD_TUPLE	'o'		/* use old tuple */
+
 typedef struct PublicationRelationInfo
 {
 	Oid			relid;
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 319c6bc7d9..1a4ba78033 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -29,6 +29,8 @@ CREATE PUBLICATION testpub_xxx WITH (publish_via_partition_root = 'true', publis
 ERROR:  conflicting or redundant options
 LINE 1: ...ub_xxx WITH (publish_via_partition_root = 'true', publish_vi...
                                                              ^
+CREATE PUBLICATION testpub_xxx WITH (row_filter_on_update = 'foo');
+ERROR:  unrecognized "row_filter_on_update" value: "foo"
 \dRp
                                               List of publications
         Name        |          Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
@@ -205,6 +207,7 @@ ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500)
 Tables:
     "public.testpub_rf_tbl3" WHERE (((e > 300) AND (e < 500)))
 
+ALTER PUBLICATION testpub5 SET (row_filter_on_update = 'new');
 -- fail - functions disallowed
 ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
 ERROR:  functions are not allowed in publication WHERE expressions
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index b1606cce7e..352d4482b1 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -24,6 +24,7 @@ ALTER PUBLICATION testpub_default SET (publish = update);
 CREATE PUBLICATION testpub_xxx WITH (foo);
 CREATE PUBLICATION testpub_xxx WITH (publish = 'cluster, vacuum');
 CREATE PUBLICATION testpub_xxx WITH (publish_via_partition_root = 'true', publish_via_partition_root = '0');
+CREATE PUBLICATION testpub_xxx WITH (row_filter_on_update = 'foo');
 
 \dRp
 
@@ -108,6 +109,7 @@ ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2;
 -- remove testpub_rf_tbl1 and add testpub_rf_tbl3 again (another WHERE expression)
 ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500);
 \dRp+ testpub5
+ALTER PUBLICATION testpub5 SET (row_filter_on_update = 'new');
 -- fail - functions disallowed
 ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
 -- fail - user-defined operators disallowed
diff --git a/src/test/subscription/t/025_row_filter.pl b/src/test/subscription/t/025_row_filter.pl
index 6428f0da00..3ad5de024c 100644
--- a/src/test/subscription/t/025_row_filter.pl
+++ b/src/test/subscription/t/025_row_filter.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 7;
+use Test::More tests => 10;
 
 # create publisher node
 my $node_publisher = PostgresNode->new('publisher');
@@ -22,6 +22,8 @@ $node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_2 (c int primary key)");
 $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 (a int, b integer, primary key(a, b))");
 $node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_partitioned (a int primary key, b integer) PARTITION BY RANGE(a)"
 );
@@ -44,6 +46,8 @@ $node_subscriber->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_2 (c int primary key)");
 $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 (a int, b integer, primary key(a, b))");
 $node_subscriber->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_partitioned (a int primary key, b integer) PARTITION BY RANGE(a)"
 );
@@ -82,6 +86,9 @@ $node_publisher->safe_psql('postgres',
 $node_publisher->safe_psql('postgres',
 	"ALTER PUBLICATION tap_pub_3 ADD TABLE tab_rowfilter_less_10k WHERE (a < 6000)"
 );
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_4 FOR TABLE tab_rowfilter_4 WHERE (a < 10 AND b < 40)"
+);
 $node_publisher->safe_psql('postgres',
 	"CREATE PUBLICATION tap_pub_not_used FOR TABLE tab_rowfilter_1 WHERE (a < 0)"
 );
@@ -103,6 +110,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 (a, b) SELECT i, i + 30 FROM generate_series(1, 6) i");
 
 # insert data into partitioned table and directly on the partition
 $node_publisher->safe_psql('postgres',
@@ -115,7 +124,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_4"
 );
 
 $node_publisher->wait_for_catchup($appname);
@@ -159,6 +168,13 @@ $result =
 	"SELECT count(a) FROM tab_rowfilter_3");
 is($result, qq(10), 'check initial data copy from table tab_rowfilter_3');
 
+# Check expected replicated rows for tab_rowfilter_4
+# tap_pub_4 filter is: (a < 10 AND b < 40)
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT count(a), min(a), max(a), min(b), max(b) FROM tab_rowfilter_4");
+is($result, qq(6|1|6|31|36), 'check initial data copy from table tab_rowfilter_4');
+
 # Check expected replicated rows for partitions
 # publication option publish_via_partition_root is false so use the row filter
 # from a partition
@@ -210,6 +226,11 @@ $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");
+# publication parameter: row_filter_on_update = new
+$node_publisher->safe_psql('postgres',
+	"UPDATE tab_rowfilter_4 SET a = 7, b = 37 WHERE a = 1 AND b = 31");
+$node_publisher->safe_psql('postgres',
+	"UPDATE tab_rowfilter_4 SET a = 102, b = 132 WHERE a = 2 AND b = 32");
 
 $node_publisher->wait_for_catchup($appname);
 
@@ -237,6 +258,46 @@ is($result, qq(1001|test 1001
 1700|test 1700
 1980|not filtered), 'check replicated rows to table tab_rowfilter_1');
 
+# Check expected replicated rows for tab_rowfilter_4
+# tap_pub_4 filter is: (a < 10 AND b < 40)
+#
+# - UPDATE (7, 37)     YES, uses new tuple for row filter
+# - UPDATE (102, 132)  NO, uses new tuple for row filter
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT a, b FROM tab_rowfilter_4 ORDER BY 1, 2");
+is($result, qq(2|32
+3|33
+4|34
+5|35
+6|36
+7|37), 'check replicated rows to table tab_rowfilter_4');
+
+# publication parameter: row_filter_on_update = old
+$node_publisher->safe_psql('postgres',
+	"ALTER PUBLICATION tap_pub_4 SET (row_filter_on_update = old)");
+$node_publisher->safe_psql('postgres',
+	"UPDATE tab_rowfilter_4 SET a = 8, b = 38 WHERE a = 3 AND b = 33");
+$node_publisher->safe_psql('postgres',
+	"UPDATE tab_rowfilter_4 SET a = 104, b = 134 WHERE a = 4 AND b = 34");
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check expected replicated rows for tab_rowfilter_4
+# tap_pub_4 filter is: (a < 10 AND b < 40)
+#
+# - UPDATE (8, 38)     YES, uses old tuple for row filter
+# - UPDATE (104, 134)  YES, uses old tuple for row filter
+$result =
+  $node_subscriber->safe_psql('postgres',
+	"SELECT a, b FROM tab_rowfilter_4 ORDER BY 1, 2");
+is($result, qq(2|32
+5|35
+6|36
+7|37
+8|38
+104|134), 'check replicated rows to table tab_rowfilter_4');
+
 # Publish using root partitioned table
 # Use a different partitioned table layout (exercise publish_via_partition_root)
 $node_publisher->safe_psql('postgres',
-- 
2.20.1