v45-0002-Validates-a-row-filter-expression-using-a-walker.patch

From f84a3549a238bb5390cbcde5db6e5d37b415ff4d Mon Sep 17 00:00:00 2001
From: Euler Taveira <euler.taveira@enterprisedb.com>
Date: Thu, 2 Dec 2021 22:43:22 -0300
Subject: [PATCH v45 2/2] Validates a row filter expression using a walker
 function

A limited set of expressions is allowed. Check REPLICA IDENTITY if
UPDATE and/or DELETE operations are published.

Discussion: https://postgr.es/m/CAA4eK1Kyax-qnVPcXzODu3JmA4vtgAjUSYPUK1Pm3vBL5gC81g%40mail.gmail.com
Discussion: https://postgr.es/m/CAA4eK1%2BXoD49bz5-2TtiD0ugq4PHSRX2D1sLPR_X4LNtdMc4OQ%40mail.gmail.com
---
 src/backend/catalog/pg_publication.c        | 146 ++++++++++++++++++++
 src/backend/parser/parse_agg.c              |   8 +-
 src/backend/parser/parse_expr.c             |  17 +--
 src/backend/parser/parse_func.c             |   3 +-
 src/backend/parser/parse_oper.c             |   7 -
 src/backend/replication/pgoutput/pgoutput.c |   2 +-
 src/test/regress/expected/publication.out   |  63 ++++++---
 src/test/regress/sql/publication.sql        |  39 ++++--
 src/test/subscription/t/027_row_filter.pl   |  11 +-
 9 files changed, 226 insertions(+), 70 deletions(-)

diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 651e719f81..81ded9d242 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -29,6 +29,7 @@
 #include "catalog/objectaddress.h"
 #include "catalog/pg_inherits.h"
 #include "catalog/pg_namespace.h"
+#include "catalog/pg_proc.h"
 #include "catalog/pg_publication.h"
 #include "catalog/pg_publication_namespace.h"
 #include "catalog/pg_publication_rel.h"
@@ -36,6 +37,7 @@
 #include "commands/publicationcmds.h"
 #include "funcapi.h"
 #include "miscadmin.h"
+#include "nodes/nodeFuncs.h"
 #include "parser/parse_clause.h"
 #include "parser/parse_collate.h"
 #include "parser/parse_relation.h"
@@ -48,6 +50,12 @@
 #include "utils/rel.h"
 #include "utils/syscache.h"
 
+typedef struct RowFilterCheckInfo
+{
+	Relation	rel;
+	PublicationActions pubactions;
+} RowFilterCheckInfo;
+
 /*
  * Check if relation can be in given publication and throws appropriate
  * error if not.
@@ -111,6 +119,141 @@ check_publication_add_schema(Oid schemaid)
 				 errdetail("Temporary schemas cannot be replicated.")));
 }
 
+/*
+ * This routine checks if the row filter expression is a "simple expression".
+ * By "simple expression" it means:
+ *
+ * - simple or compound expressions;
+ *   Examples:
+ *     (Var Op Const)
+ *     (Var Op Var)
+ *     (Var Op Const) Bool (Var Op Const)
+ * - user-defined operators are not allowed;
+ * - user-defined types are not allowed;
+ * - user-defined functions are not allowed;
+ * - non-immutable builtin functions are not allowed.
+ *
+ * NOTES:
+ *
+ * User-defined functions, operators or types are not allowed because
+ * (a) if a user drops a user-defined object used in a row filter expression,
+ * the logical decoding infrastructure won't be able to recover from such pilot
+ * error even if the object is recreated again because a historic snapshot is
+ * used to execute the row filter.
+ * (b) a user-defined function can be used to access tables which could have
+ * unpleasant results because a historic snapshot is used. That's why only
+ * non-immutable functions are allowed in row filter expressions.
+ */
+static bool
+publication_row_filter_walker(Node *node, RowFilterCheckInfo *rfinfo)
+{
+	char	   *errormsg = NULL;
+
+	if (node == NULL)
+		return false;
+
+	if (IsA(node, Var))
+	{
+		Var		   *var = (Var *) node;
+		AttrNumber	attnum = var->varattno;
+		char		replident = rfinfo->rel->rd_rel->relreplident;
+
+		/*
+		 * Check replica identity if the publication allows UPDATE and/or
+		 * DELETE as DML operations. REPLICA IDENTITY FULL is OK since it
+		 * includes all columns in the old tuple.
+		 */
+		if ((rfinfo->pubactions.pubupdate || rfinfo->pubactions.pubdelete) &&
+			replident != REPLICA_IDENTITY_FULL)
+		{
+			Bitmapset  *bms_replident = NULL;
+
+			if (replident == REPLICA_IDENTITY_DEFAULT)
+				bms_replident = RelationGetIndexAttrBitmap(rfinfo->rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
+			else if (replident == REPLICA_IDENTITY_INDEX)
+				bms_replident = RelationGetIndexAttrBitmap(rfinfo->rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+
+			/*
+			 * REPLICA IDENTITY NOTHING does not contain columns in the old
+			 * tuple so it is not supported. The referenced column must be
+			 * contained by REPLICA IDENTITY DEFAULT (primary key) or REPLICA
+			 * IDENTITY INDEX (index columns).
+			 */
+			if (replident == REPLICA_IDENTITY_NOTHING ||
+				!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, bms_replident))
+			{
+				const char *colname = get_attname(RelationGetRelid(rfinfo->rel), attnum, false);
+
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+						 errmsg("cannot add relation \"%s\" to publication",
+								RelationGetRelationName(rfinfo->rel)),
+						 errdetail("Column \"%s\" used in the WHERE expression is not part of the replica identity.",
+								   colname)));
+			}
+
+			bms_free(bms_replident);
+		}
+		else if (var->vartype >= FirstNormalObjectId)
+		{
+			errormsg = _("User-defined types are not allowed.");
+		}
+	}
+	else if (IsA(node, List) || IsA(node, Const) || IsA(node, BoolExpr) || IsA(node, NullIfExpr) || IsA(node, NullTest) || IsA(node, BooleanTest) || IsA(node, CoalesceExpr) || IsA(node, CaseExpr) || IsA(node, CaseTestExpr) || IsA(node, MinMaxExpr) || IsA(node, ArrayExpr) || IsA(node, ScalarArrayOpExpr) || IsA(node, XmlExpr))
+	{
+		/* nodes are part of simple expressions */
+	}
+	else if (IsA(node, OpExpr))
+	{
+		if (((OpExpr *) node)->opno >= FirstNormalObjectId)
+			errormsg = _("User-defined operators are not allowed.");
+	}
+	else if (IsA(node, FuncExpr))
+	{
+		Oid			funcid = ((FuncExpr *) node)->funcid;
+		const char *funcname = get_func_name(funcid);
+
+		if (funcid >= FirstNormalObjectId)
+			errormsg = psprintf(_("User-defined functions are not allowed (%s)."), funcname);
+		else if (func_volatile(funcid) != PROVOLATILE_IMMUTABLE)
+			errormsg = psprintf(_("Non-immutable functions are not allowed (%s)."), funcname);
+	}
+	else
+	{
+		elog(WARNING, "unexpected node: %s", nodeToString(node));
+		ereport(ERROR,
+				(errmsg("invalid publication WHERE expression for relation \"%s\"",
+						RelationGetRelationName(rfinfo->rel)),
+				 errdetail("Expressions only allows columns, constants and some builtin functions and operators.")));
+	}
+
+	if (errormsg)
+		ereport(ERROR,
+				(errmsg("invalid publication WHERE expression for relation \"%s\"",
+						RelationGetRelationName(rfinfo->rel)),
+				 errdetail("%s", errormsg)));
+
+	return expression_tree_walker(node, publication_row_filter_walker, (void *) rfinfo);
+}
+
+/*
+ * Validate the row filter with the following rules:
+ * (a) few node types are allowed in the expression. See the function
+ * publication_row_filter_walker for details.
+ * (b) If the publication publishes UPDATE and/or DELETE operations, all
+ * columns used in the row filter must be contained in the replica identity.
+ */
+static void
+check_publication_row_filter(PublicationActions pubactions, Relation rel, Node *rfnode)
+{
+	RowFilterCheckInfo rfinfo = {0};
+
+	rfinfo.rel = rel;
+	rfinfo.pubactions = pubactions;
+
+	publication_row_filter_walker(rfnode, &rfinfo);
+}
+
 /*
  * Returns if relation represented by oid and Form_pg_class entry
  * is publishable.
@@ -317,6 +460,9 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
 										   EXPR_KIND_PUBLICATION_WHERE,
 										   "PUBLICATION");
 
+		/* Validate row filter expression */
+		check_publication_row_filter(pub->pubactions, targetrel, whereclause);
+
 		/* Fix up collation information */
 		assign_expr_collations(pstate, whereclause);
 	}
diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index 193c87d8b7..7388bbdbd4 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -552,11 +552,7 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr)
 
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			if (isAgg)
-				err = _("aggregate functions are not allowed in publication WHERE expressions");
-			else
-				err = _("grouping operations are not allowed in publication WHERE expressions");
-
+			/* okay (see function publication_row_filter_walker) */
 			break;
 
 		case EXPR_KIND_CYCLE_MARK:
@@ -951,7 +947,7 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
 			errkind = true;
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("window functions are not allowed in publication WHERE expressions");
+			/* okay (see function publication_row_filter_walker) */
 			break;
 
 			/*
diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c
index 3d43839b35..f1bb01c80d 100644
--- a/src/backend/parser/parse_expr.c
+++ b/src/backend/parser/parse_expr.c
@@ -200,19 +200,8 @@ transformExprRecurse(ParseState *pstate, Node *expr)
 			break;
 
 		case T_FuncCall:
-			{
-				/*
-				 * Forbid functions in publication WHERE condition
-				 */
-				if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
-					ereport(ERROR,
-							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-							 errmsg("functions are not allowed in publication WHERE expressions"),
-							 parser_errposition(pstate, exprLocation(expr))));
-
-				result = transformFuncCall(pstate, (FuncCall *) expr);
-				break;
-			}
+			result = transformFuncCall(pstate, (FuncCall *) expr);
+			break;
 
 		case T_MultiAssignRef:
 			result = transformMultiAssignRef(pstate, (MultiAssignRef *) expr);
@@ -1777,7 +1766,7 @@ transformSubLink(ParseState *pstate, SubLink *sublink)
 			err = _("cannot use subquery in column generation expression");
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("cannot use subquery in publication WHERE expression");
+			/* okay (see function publication_row_filter_walker) */
 			break;
 
 			/*
diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c
index 29bebb73eb..ffdf86fc7e 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2656,7 +2656,8 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location)
 			errkind = true;
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("set-returning functions are not allowed in publication WHERE expressions");
+			/* okay (see function publication_row_filter_walker) */
+			pstate->p_hasTargetSRFs = true;
 			break;
 
 			/*
diff --git a/src/backend/parser/parse_oper.c b/src/backend/parser/parse_oper.c
index 29f8835ce1..bc34a23afc 100644
--- a/src/backend/parser/parse_oper.c
+++ b/src/backend/parser/parse_oper.c
@@ -718,13 +718,6 @@ make_op(ParseState *pstate, List *opname, Node *ltree, Node *rtree,
 											opform->oprright)),
 				 parser_errposition(pstate, location)));
 
-	/* Check it's not a custom operator for publication WHERE expressions */
-	if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE && opform->oid >= FirstNormalObjectId)
-		ereport(ERROR,
-				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-				 errmsg("user-defined operators are not allowed in publication WHERE expressions"),
-				 parser_errposition(pstate, location)));
-
 	/* Do typecasting and build the expression tree */
 	if (ltree == NULL)
 	{
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index c97e3da642..0f704792e4 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -857,7 +857,7 @@ pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple, H
 			/*
 			 * All row filter expressions will be discarded if there is one
 			 * publication-relation entry without a row filter. That's because
-			 * all expressionsare aggregated by the OR operator. The row
+			 * all expressions are aggregated by the OR operator. The row
 			 * filter absence means replicate all rows so a single valid
 			 * expression means publish this row.
 			 */
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index aa784350a2..d978ee7014 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -243,18 +243,19 @@ CREATE TABLE testpub_rf_tbl1 (a integer, b text);
 CREATE TABLE testpub_rf_tbl2 (c text, d integer);
 CREATE TABLE testpub_rf_tbl3 (e integer);
 CREATE TABLE testpub_rf_tbl4 (g text);
+CREATE TABLE testpub_rf_tbl5 (a xml);
 CREATE SCHEMA testpub_rf_schema1;
 CREATE TABLE testpub_rf_schema1.testpub_rf_tbl5 (h integer);
 CREATE SCHEMA testpub_rf_schema2;
 CREATE TABLE testpub_rf_schema2.testpub_rf_tb16 (i integer);
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5);
+CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5) WITH (publish = 'insert');
 RESET client_min_messages;
 \dRp+ testpub5
                                     Publication testpub5
           Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
 --------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f          | t       | t       | t       | t         | f
+ regress_publication_user | f          | t       | f       | f       | f         | f
 Tables:
     "public.testpub_rf_tbl1"
     "public.testpub_rf_tbl2" WHERE (((c <> 'test'::text) AND (d < 5)))
@@ -264,7 +265,7 @@ ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 200
                                     Publication testpub5
           Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
 --------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f          | t       | t       | t       | t         | f
+ regress_publication_user | f          | t       | f       | f       | f         | f
 Tables:
     "public.testpub_rf_tbl1"
     "public.testpub_rf_tbl2" WHERE (((c <> 'test'::text) AND (d < 5)))
@@ -275,7 +276,7 @@ ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2;
                                     Publication testpub5
           Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
 --------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f          | t       | t       | t       | t         | f
+ regress_publication_user | f          | t       | f       | f       | f         | f
 Tables:
     "public.testpub_rf_tbl1"
     "public.testpub_rf_tbl3" WHERE (((e > 1000) AND (e < 2000)))
@@ -286,7 +287,7 @@ ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500)
                                     Publication testpub5
           Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
 --------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f          | t       | t       | t       | t         | f
+ regress_publication_user | f          | t       | f       | f       | f         | f
 Tables:
     "public.testpub_rf_tbl3" WHERE (((e > 300) AND (e < 500)))
 
@@ -310,26 +311,26 @@ Publications:
 DROP PUBLICATION testpub5a, testpub5b, testpub5c;
 -- some more syntax tests to exercise other parser pathways
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999);
+CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999) WITH (publish = 'insert');
 RESET client_min_messages;
 \dRp+ testpub_syntax1
                                 Publication testpub_syntax1
           Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
 --------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f          | t       | t       | t       | t         | f
+ regress_publication_user | f          | t       | f       | f       | f         | f
 Tables:
     "public.testpub_rf_tbl1"
     "public.testpub_rf_tbl3" WHERE ((e < 999))
 
 DROP PUBLICATION testpub_syntax1;
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999);
+CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = 'insert');
 RESET client_min_messages;
 \dRp+ testpub_syntax2
                                 Publication testpub_syntax2
           Owner           | All tables | Inserts | Updates | Deletes | Truncates | Via root 
 --------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f          | t       | t       | t       | t         | f
+ regress_publication_user | f          | t       | f       | f       | f         | f
 Tables:
     "public.testpub_rf_tbl1"
     "testpub_rf_schema1.testpub_rf_tbl5" WHERE ((h < 999))
@@ -353,20 +354,36 @@ ERROR:  conflicting or redundant WHERE clauses for table "testpub_rf_tbl1"
 CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1, testpub_rf_tbl1 WHERE (a = 2) WITH (publish = 'insert');
 ERROR:  conflicting or redundant WHERE clauses for table "testpub_rf_tbl1"
 RESET client_min_messages;
--- fail - functions disallowed
-ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
-ERROR:  functions are not allowed in publication WHERE expressions
-LINE 1: ...ICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) ...
-                                                             ^
--- fail - user-defined operators disallowed
-CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
-CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+-- fail - user-defined operators are not allowed
+CREATE FUNCTION testpub_rf_func1(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func1, LEFTARG = integer, RIGHTARG = integer);
 CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
-ERROR:  user-defined operators are not allowed in publication WHERE expressions
-LINE 1: ...ICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
-                                                               ^
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl3"
+DETAIL:  User-defined operators are not allowed.
+-- fail - user-defined functions are not allowed
+CREATE FUNCTION testpub_rf_func2() RETURNS integer AS $$ BEGIN RETURN 123; END; $$ LANGUAGE plpgsql;
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a >= testpub_rf_func2());
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL:  User-defined functions are not allowed (testpub_rf_func2).
+-- fail - non-immutable functions are not allowed. random() is volatile.
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a < random());
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL:  Non-immutable functions are not allowed (random).
+-- ok - builtin operators are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
+-- ok - immutable builtin functions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- ok - conditional expressions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (a IS DOCUMENT);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (xmlexists('//foo[text() = ''bar'']' PASSING BY VALUE a));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1, 2) = a);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (CASE a WHEN 5 THEN true ELSE false END);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (COALESCE(b, 'foo') = 'foo');
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (GREATEST(a, 10) > 10);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IN (2, 4, 6));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (ARRAY[a] <@ ARRAY[2, 4, 6]);
 -- fail - WHERE not allowed in DROP
-ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl1 WHERE (e < 27);
 ERROR:  cannot use a WHERE clause when removing a table from publication
 -- fail - cannot ALTER SET table which is a member of a pre-existing schema
 SET client_min_messages = 'ERROR';
@@ -379,6 +396,7 @@ DROP TABLE testpub_rf_tbl1;
 DROP TABLE testpub_rf_tbl2;
 DROP TABLE testpub_rf_tbl3;
 DROP TABLE testpub_rf_tbl4;
+DROP TABLE testpub_rf_tbl5;
 DROP TABLE testpub_rf_schema1.testpub_rf_tbl5;
 DROP TABLE testpub_rf_schema2.testpub_rf_tb16;
 DROP SCHEMA testpub_rf_schema1;
@@ -386,7 +404,8 @@ DROP SCHEMA testpub_rf_schema2;
 DROP PUBLICATION testpub5;
 DROP PUBLICATION testpub7;
 DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
 -- Test cache invalidation FOR ALL TABLES publication
 SET client_min_messages = 'ERROR';
 CREATE TABLE testpub_tbl4(a int);
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index f8d70be24c..7ab0ef7e63 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -138,12 +138,13 @@ CREATE TABLE testpub_rf_tbl1 (a integer, b text);
 CREATE TABLE testpub_rf_tbl2 (c text, d integer);
 CREATE TABLE testpub_rf_tbl3 (e integer);
 CREATE TABLE testpub_rf_tbl4 (g text);
+CREATE TABLE testpub_rf_tbl5 (a xml);
 CREATE SCHEMA testpub_rf_schema1;
 CREATE TABLE testpub_rf_schema1.testpub_rf_tbl5 (h integer);
 CREATE SCHEMA testpub_rf_schema2;
 CREATE TABLE testpub_rf_schema2.testpub_rf_tb16 (i integer);
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5);
+CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5) WITH (publish = 'insert');
 RESET client_min_messages;
 \dRp+ testpub5
 ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 2000);
@@ -163,12 +164,12 @@ RESET client_min_messages;
 DROP PUBLICATION testpub5a, testpub5b, testpub5c;
 -- some more syntax tests to exercise other parser pathways
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999);
+CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999) WITH (publish = 'insert');
 RESET client_min_messages;
 \dRp+ testpub_syntax1
 DROP PUBLICATION testpub_syntax1;
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999);
+CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = 'insert');
 RESET client_min_messages;
 \dRp+ testpub_syntax2
 DROP PUBLICATION testpub_syntax2;
@@ -182,14 +183,30 @@ SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1 WHERE (a = 1), testpub_rf_tbl1 WITH (publish = 'insert');
 CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1, testpub_rf_tbl1 WHERE (a = 2) WITH (publish = 'insert');
 RESET client_min_messages;
--- fail - functions disallowed
-ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
--- fail - user-defined operators disallowed
-CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
-CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+-- fail - user-defined operators are not allowed
+CREATE FUNCTION testpub_rf_func1(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func1, LEFTARG = integer, RIGHTARG = integer);
 CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+-- fail - user-defined functions are not allowed
+CREATE FUNCTION testpub_rf_func2() RETURNS integer AS $$ BEGIN RETURN 123; END; $$ LANGUAGE plpgsql;
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a >= testpub_rf_func2());
+-- fail - non-immutable functions are not allowed. random() is volatile.
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a < random());
+-- ok - builtin operators are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
+-- ok - immutable builtin functions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- ok - conditional expressions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (a IS DOCUMENT);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (xmlexists('//foo[text() = ''bar'']' PASSING BY VALUE a));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1, 2) = a);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (CASE a WHEN 5 THEN true ELSE false END);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (COALESCE(b, 'foo') = 'foo');
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (GREATEST(a, 10) > 10);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IN (2, 4, 6));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (ARRAY[a] <@ ARRAY[2, 4, 6]);
 -- fail - WHERE not allowed in DROP
-ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl1 WHERE (e < 27);
 -- fail - cannot ALTER SET table which is a member of a pre-existing schema
 SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_schema2;
@@ -200,6 +217,7 @@ DROP TABLE testpub_rf_tbl1;
 DROP TABLE testpub_rf_tbl2;
 DROP TABLE testpub_rf_tbl3;
 DROP TABLE testpub_rf_tbl4;
+DROP TABLE testpub_rf_tbl5;
 DROP TABLE testpub_rf_schema1.testpub_rf_tbl5;
 DROP TABLE testpub_rf_schema2.testpub_rf_tb16;
 DROP SCHEMA testpub_rf_schema1;
@@ -207,7 +225,8 @@ DROP SCHEMA testpub_rf_schema2;
 DROP PUBLICATION testpub5;
 DROP PUBLICATION testpub7;
 DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
 
 -- Test cache invalidation FOR ALL TABLES publication
 SET client_min_messages = 'ERROR';
diff --git a/src/test/subscription/t/027_row_filter.pl b/src/test/subscription/t/027_row_filter.pl
index a2aa05f1e9..affd206e8a 100644
--- a/src/test/subscription/t/027_row_filter.pl
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -18,6 +18,8 @@ $node_subscriber->start;
 # setup structure on publisher
 $node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_1 (a int primary key, b text)");
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE tab_rowfilter_1 REPLICA IDENTITY FULL");
 $node_publisher->safe_psql('postgres',
 	"CREATE TABLE tab_rowfilter_2 (c int primary key)");
 $node_publisher->safe_psql('postgres',
@@ -231,14 +233,10 @@ $node_publisher->safe_psql('postgres',
 	"INSERT INTO tab_rowfilter_1 (a, b) VALUES (1600, 'test 1600')");
 $node_publisher->safe_psql('postgres',
 	"INSERT INTO tab_rowfilter_1 (a, b) VALUES (1601, 'test 1601')");
-$node_publisher->safe_psql('postgres',
-	"INSERT INTO tab_rowfilter_1 (a, b) VALUES (1700, 'test 1700')");
 $node_publisher->safe_psql('postgres',
 	"UPDATE tab_rowfilter_1 SET b = NULL WHERE a = 1600");
 $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',
@@ -281,12 +279,8 @@ is($result, qq(13|0|12), 'check replicated rows to tab_rowfilter_4');
 # - INSERT (800, 'test 800')   NO, because 800 is not > 1000
 # - INSERT (1600, 'test 1600') YES, because 1600 > 1000 and 'test 1600' <> 'filtered'
 # - INSERT (1601, 'test 1601') YES, because 1601 > 1000 and 'test 1601' <> 'filtered'
-# - INSERT (1700, 'test 1700') YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
 # - UPDATE (1600, NULL)        NO, row filter evaluates to false because NULL is not <> 'filtered'
 # - UPDATE (1601, 'test 1601 updated') YES, because 1601 > 1000 and 'test 1601 updated' <> 'filtered'
-# - DELETE (1700)              NO, row filter contains column b that is not part of
-# the PK or REPLICA IDENTITY and old tuple contains b = NULL, hence, row filter
-# evaluates to false
 #
 $result =
   $node_subscriber->safe_psql('postgres',
@@ -295,7 +289,6 @@ is( $result, qq(1001|test 1001
 1002|test 1002
 1600|test 1600
 1601|test 1601 updated
-1700|test 1700
 1980|not filtered), 'check replicated rows to table tab_rowfilter_1');
 
 # Publish using root partitioned table
-- 
2.20.1