v44-0002-PS-Row-filter-validation-walker.patch

From beb8a9749a36f4192713acadfdc09cc2e73c94f4 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Thu, 2 Dec 2021 16:29:23 +1100
Subject: [PATCH v44] PS - Row filter validation walker

This patch implements a parse-tree "walker" to validate a row-filter expression.

REPLICA IDENTITY validation
---------------------------

For publish mode "delete" and "update" it validates that any columns referenced
in the filter expression must be part of REPLICA IDENTITY or Primary Key.

Discussion: https://www.postgresql.org/message-id/CAA4eK1Kyax-qnVPcXzODu3JmA4vtgAjUSYPUK1Pm3vBL5gC81g%40mail.gmail.com

Expression Node-kind validation
-------------------------------

Only simple filter expressions are permitted. Specifially:
- no user-defined operators.
- no user-defined functions.
- no user-defined types.
- no system functions (unless they are IMMUTABLE). See design decision at [1].
- no parse nodes of any kind other than Var, OpExpr, Const, BoolExpr, FuncExpr, NullIfExpr, NullTest
[1] https://www.postgresql.org/message-id/CAA4eK1%2BXoD49bz5-2TtiD0ugq4PHSRX2D1sLPR_X4LNtdMc4OQ%40mail.gmail.com
---
 src/backend/catalog/pg_publication.c      | 198 +++++++++++++++++++++++++++++-
 src/backend/parser/parse_agg.c            |  14 ++-
 src/backend/parser/parse_expr.c           |  22 ++--
 src/backend/parser/parse_func.c           |   6 +-
 src/backend/parser/parse_oper.c           |   7 --
 src/test/regress/expected/publication.out | 144 +++++++++++++++++++---
 src/test/regress/sql/publication.sql      | 106 +++++++++++++++-
 src/test/subscription/t/027_row_filter.pl |   7 +-
 src/tools/pgindent/typedefs.list          |   1 +
 9 files changed, 448 insertions(+), 57 deletions(-)

diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 89d00cd..d67023a 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -33,9 +33,11 @@
 #include "catalog/pg_publication_namespace.h"
 #include "catalog/pg_publication_rel.h"
 #include "catalog/pg_type.h"
+#include "catalog/pg_proc.h"
 #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"
@@ -219,10 +221,199 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
 	PG_RETURN_BOOL(result);
 }
 
+/* For rowfilter_walker. */
+typedef struct {
+	Relation	rel;
+	bool		check_replident; /* check if Var is bms_replident member? */
+	Bitmapset  *bms_replident;
+} rf_context;
+
 /*
- * Gets the relations based on the publication partition option for a specified
- * relation.
+ * The row filter walker checks that the row filter expression is legal.
+ *
+ * Rules: Node-type validation
+ * ---------------------------
+ * Allow only simple or compound expressions such as:
+ * - "(Var Op Const)" or
+ * - "(Var Op Var)" or
+ * - "(Var Op Const) Bool (Var Op Const)"
+ * - etc
+ * (where Var is a column of the table this filter belongs to)
+ *
+ * Specifically,
+ * - User-defined operators are not allowed.
+ * - User-defined functions are not allowed.
+ * - User-defined types are not allowed.
+ * - System functions that are not IMMUTABLE are not allowed.
+ * - NULLIF is allowed.
+ * - IS NULL is allowed.
+ *
+ * Notes:
+ *
+ * We don't allow user-defined functions/operators/types because (a) if the user
+ * drops such a user-defnition or if there is any other error via its function,
+ * the walsender won't be able to recover from such an error even if we fix the
+ * function's problem because a historic snapshot is used to access the
+ * row-filter; (b) any other table could be accessed via a function, which won't
+ * work because of historic snapshots in logical decoding environment.
+ *
+ * We don't allow anything other than immutable built-in functions because those
+ * (not immutable ones) can access database and would lead to the problem (b)
+ * mentioned in the previous paragraph.
+ *
+ * Rules: Replica Identity validation
+ * -----------------------------------
+ * If the flag context.check_replident is true then validate that every variable
+ * referenced by the filter expression is a valid member of the allowed set of
+ * replica identity columns (context.bms_replindent)
  */
+static bool
+rowfilter_walker(Node *node, rf_context *context)
+{
+	char *forbidden = NULL;
+	bool too_complex = false;
+
+	if (node == NULL)
+		return false;
+
+	if (IsA(node, Var))
+	{
+		Var		   *var = (Var *) node;
+
+		/* User-defined types not allowed. */
+		if (var->vartype >= FirstNormalObjectId)
+			forbidden = _("user-defined types are not allowed");
+
+		/* Optionally, do replica identify validation of the referenced column. */
+		if (context->check_replident)
+		{
+			Oid			relid = RelationGetRelid(context->rel);
+			AttrNumber	attnum = var->varattno;
+
+			if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, context->bms_replident))
+			{
+				const char *colname = get_attname(relid, attnum, false);
+
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+						errmsg("cannot add relation \"%s\" to publication",
+							   RelationGetRelationName(context->rel)),
+						errdetail("Row filter column \"%s\" is not part of the REPLICA IDENTITY",
+								  colname)));
+			}
+		}
+	}
+	else if (IsA(node, Const) || IsA(node, BoolExpr) || IsA(node, NullIfExpr)
+			 || IsA(node, NullTest))
+	{
+		/* OK */
+	}
+	else if (IsA(node, OpExpr))
+	{
+		/* OK, except user-defined operators are not allowed. */
+		if (((OpExpr *)node)->opno >= FirstNormalObjectId)
+			forbidden = _("user-defined operators are not allowed");
+	}
+	else if (IsA(node, FuncExpr))
+	{
+		Oid funcid = ((FuncExpr *)node)->funcid;
+		char *funcname = get_func_name(funcid);
+
+		/*
+		 * User-defined functions are not allowed.
+		 * System-functions that are not IMMUTABLE are not allowed.
+		 */
+		if (funcid >= FirstNormalObjectId)
+		{
+			forbidden = psprintf("user-defined functions are not allowed: %s",
+								 funcname);
+		}
+		else
+		{
+			if (func_volatile(funcid) != PROVOLATILE_IMMUTABLE)
+				forbidden = psprintf("system functions that are not IMMUTABLE are not allowed: %s",
+									 funcname);
+		}
+	}
+	else
+	{
+		elog(DEBUG1, "the row filter contained something unexpected: %s", nodeToString(node));
+		too_complex = true;
+	}
+
+	if (too_complex)
+		ereport(ERROR,
+				(errmsg("invalid publication WHERE expression for relation \"%s\"",
+						RelationGetRelationName(context->rel)),
+				errhint("only simple expressions using columns, constants and immutable system functions are allowed")
+				));
+
+	if (forbidden)
+		ereport(ERROR,
+				(errmsg("invalid publication WHERE expression for relation \"%s\"",
+						RelationGetRelationName(context->rel)),
+						errdetail("%s", forbidden)
+				));
+
+	return expression_tree_walker(node, rowfilter_walker, (void *)context);
+}
+
+/*
+ * Check if the row-filter is valid according to the following rules:
+ *
+ * 1. Only certain simple node types are permitted in the expression. See
+ * function rowfilter_walker for details.
+ *
+ * 2. If the publish operation contains "delete" or "update" then only columns
+ * that are allowed by the REPLICA IDENTITY rules are permitted to be used in
+ * the row-filter WHERE clause.
+ */
+static void
+rowfilter_expr_checker(Publication *pub, Relation rel, Node *rfnode)
+{
+	rf_context	context = {0};
+
+	context.rel = rel;
+
+	/*
+	 * For "delete" or "update", check that filter cols are also valid replica
+	 * identity cols.
+	 */
+	if (pub->pubactions.pubdelete || pub->pubactions.pubupdate)
+	{
+		char replica_identity = rel->rd_rel->relreplident;
+
+		if (replica_identity == REPLICA_IDENTITY_FULL)
+		{
+			/*
+			 * FULL means all cols are in the REPLICA IDENTITY, so all cols are
+			 * allowed in the row-filter too.
+			 */
+		}
+		else
+		{
+			context.check_replident = true;
+
+			/*
+			 * Find what are the cols that are part of the REPLICA IDENTITY.
+			 * Note that REPLICA IDENTIY DEFAULT means primary key or nothing.
+			 */
+			if (replica_identity == REPLICA_IDENTITY_DEFAULT)
+				context.bms_replident = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
+			else
+				context.bms_replident = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+		}
+	}
+
+	/*
+	 * Walk the parse-tree of this publication row filter expression and throw an
+	 * error if anything not permitted or unexpected is encountered.
+	 */
+	rowfilter_walker(rfnode, &context);
+
+	bms_free(context.bms_replident);
+}
+
 List *
 GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
 							   Oid relid)
@@ -333,6 +524,9 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
 
 		/* Fix up collation information */
 		whereclause = GetTransformedWhereClause(pstate, pri, true);
+
+		/* Validate the row-filter. */
+		rowfilter_expr_checker(pub, targetrel, whereclause);
 	}
 
 	/* Form a tuple. */
diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index 193c87d..f65a86f 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -552,11 +552,10 @@ 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");
-
+			/*
+			 * OK for now. The row-filter validation is done later by a walker
+			 * function (see pg_publication).
+			 */
 			break;
 
 		case EXPR_KIND_CYCLE_MARK:
@@ -951,7 +950,10 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
 			errkind = true;
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("window functions are not allowed in publication WHERE expressions");
+			/*
+			 * OK for now. The row-filter validation is done later by a walker
+			 * function (see pg_publication).
+			 */
 			break;
 
 			/*
diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c
index 3d43839..d8627b9 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,10 @@ 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");
+			/*
+			 * OK for now. The row-filter validation is done later by a walker
+			 * function (see pg_publication).
+			 */
 			break;
 
 			/*
@@ -3100,7 +3092,7 @@ ParseExprKindName(ParseExprKind exprKind)
 		case EXPR_KIND_CYCLE_MARK:
 			return "CYCLE";
 		case EXPR_KIND_PUBLICATION_WHERE:
-			return "publication expression";
+			return "publication WHERE expression";
 
 			/*
 			 * There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c
index 29bebb7..4e4557f 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2656,7 +2656,11 @@ 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");
+			/*
+			 * OK for now. The row-filter validation is done later by a walker
+			 * function (see pg_publication).
+			 */
+			pstate->p_hasTargetSRFs = true;
 			break;
 
 			/*
diff --git a/src/backend/parser/parse_oper.c b/src/backend/parser/parse_oper.c
index 29f8835..bc34a23 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/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 6959675..d9ee9ff 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -248,13 +248,15 @@ CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
 CREATE SCHEMA testpub_rf_myschema1;
 CREATE TABLE testpub_rf_myschema1.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);
+-- Firstly, test using the option publish="insert" because the row filter
+-- validation of referenced columns is less strict than for delete/update.
+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 +266,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 +277,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 +288,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 +312,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_myschema.testpub_rf_tbl5 WHERE (h < 999);
+CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_myschema.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_myschema.testpub_rf_tbl5" WHERE ((h < 999))
@@ -353,19 +355,41 @@ ERROR:  conflicting or redundant row-filters for "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 row-filters for "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 functions are not allowed
+CREATE FUNCTION testpub_rf_func99() RETURNS integer AS $$ BEGIN RETURN 99; END; $$ LANGUAGE plpgsql;
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a < testpub_rf_func99());
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL:  user-defined functions are not allowed: testpub_rf_func99
+-- fail - system functions that are not IMMUTABLE are not allowed; random() is a "volatile" function.
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a < random());
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL:  system functions that are not IMMUTABLE are not allowed: random
+-- ok - system functions that are IMMUTABLE are allowed; int8inc() is an "immutable" function.
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a < int8inc(999));
+-- ok - NULLIF is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1,2) = a);
+-- ok - IS NULL is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
 -- 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);
 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 types disallowed
+CREATE TYPE rf_bug_status AS ENUM ('new', 'open', 'closed');
+CREATE TABLE rf_bug (id serial, description text, status rf_bug_status);
+CREATE PUBLICATION testpub6 FOR TABLE rf_bug WHERE (status = 'open') WITH (publish="insert");
+ERROR:  invalid publication WHERE expression for relation "rf_bug"
+DETAIL:  user-defined types are not allowed
+DROP TABLE rf_bug;
+DROP TYPE rf_bug_status;
+-- fail - row-filter expression is not simple
+CREATE PUBLICATION testpub7 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl1"
+HINT:  only simple expressions using columns, constants and immutable system functions are allowed
 -- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3;
 ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
 ERROR:  invalid use of WHERE row-filter in ALTER PUBLICATION ... DROP TABLE
 -- fail - cannot ALTER SET table which is a member of a pre-existing schema
@@ -387,6 +411,92 @@ DROP PUBLICATION testpub5;
 DROP PUBLICATION testpub7;
 DROP OPERATOR =#>(integer, integer);
 DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func99();
+-- ======================================================
+-- More row filter tests for validating column references
+CREATE TABLE rf_tbl_abcd_nopk(a int, b int, c int, d int);
+CREATE TABLE rf_tbl_abcd_pk(a int, b int, c int, d int, PRIMARY KEY(a,b));
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+-- ok - "a" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- ok - "b" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (b > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- fail - "c" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL:  Row filter column "c" is not part of the REPLICA IDENTITY
+-- fail - "d" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (d > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL:  Row filter column "d" is not part of the REPLICA IDENTITY
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+-- fail - "a" is not part of REPLICA IDENTITY
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_nopk" to publication
+DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- Case 2. REPLICA IDENTITY FULL
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL;
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- ok - "a" is in REPLICA IDENTITY now
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL:  Row filter column "c" is not part of the REPLICA IDENTITY
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_nopk" to publication
+DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- Case 4. REPLICA IDENTITY INDEX
+ALTER TABLE rf_tbl_abcd_pk ALTER COLUMN c SET NOT NULL;
+CREATE UNIQUE INDEX idx_abcd_pk_c ON rf_tbl_abcd_pk(c);
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY USING INDEX idx_abcd_pk_c;
+ALTER TABLE rf_tbl_abcd_nopk ALTER COLUMN c SET NOT NULL;
+CREATE UNIQUE INDEX idx_abcd_nopk_c ON rf_tbl_abcd_nopk(c);
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY USING INDEX idx_abcd_nopk_c;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+ERROR:  cannot add relation "rf_tbl_abcd_nopk" to publication
+DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+-- ======================================================
 -- 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 40198fc..fcc09b1 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -143,7 +143,9 @@ CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
 CREATE SCHEMA testpub_rf_myschema1;
 CREATE TABLE testpub_rf_myschema1.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);
+-- Firstly, test using the option publish="insert" because the row filter
+-- validation of referenced columns is less strict than for delete/update.
+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 +165,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_myschema.testpub_rf_tbl5 WHERE (h < 999);
+CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_myschema.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = "insert");
 RESET client_min_messages;
 \dRp+ testpub_syntax2
 DROP PUBLICATION testpub_syntax2;
@@ -182,13 +184,31 @@ 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 functions are not allowed
+CREATE FUNCTION testpub_rf_func99() RETURNS integer AS $$ BEGIN RETURN 99; END; $$ LANGUAGE plpgsql;
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a < testpub_rf_func99());
+-- fail - system functions that are not IMMUTABLE are not allowed; random() is a "volatile" function.
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a < random());
+-- ok - system functions that are IMMUTABLE are allowed; int8inc() is an "immutable" function.
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a < int8inc(999));
+-- ok - NULLIF is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1,2) = a);
+-- ok - IS NULL is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
 -- 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);
 CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+-- fail - user-defined types disallowed
+CREATE TYPE rf_bug_status AS ENUM ('new', 'open', 'closed');
+CREATE TABLE rf_bug (id serial, description text, status rf_bug_status);
+CREATE PUBLICATION testpub6 FOR TABLE rf_bug WHERE (status = 'open') WITH (publish="insert");
+DROP TABLE rf_bug;
+DROP TYPE rf_bug_status;
+-- fail - row-filter expression is not simple
+CREATE PUBLICATION testpub7 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
 -- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3;
 ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
 -- fail - cannot ALTER SET table which is a member of a pre-existing schema
 SET client_min_messages = 'ERROR';
@@ -208,6 +228,82 @@ DROP PUBLICATION testpub5;
 DROP PUBLICATION testpub7;
 DROP OPERATOR =#>(integer, integer);
 DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func99();
+
+-- ======================================================
+-- More row filter tests for validating column references
+CREATE TABLE rf_tbl_abcd_nopk(a int, b int, c int, d int);
+CREATE TABLE rf_tbl_abcd_pk(a int, b int, c int, d int, PRIMARY KEY(a,b));
+
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+-- ok - "a" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- ok - "b" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (b > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- fail - "c" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "d" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (d > 99);
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+-- fail - "a" is not part of REPLICA IDENTITY
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+
+-- Case 2. REPLICA IDENTITY FULL
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL;
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- ok - "a" is in REPLICA IDENTITY now
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+
+-- Case 4. REPLICA IDENTITY INDEX
+ALTER TABLE rf_tbl_abcd_pk ALTER COLUMN c SET NOT NULL;
+CREATE UNIQUE INDEX idx_abcd_pk_c ON rf_tbl_abcd_pk(c);
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY USING INDEX idx_abcd_pk_c;
+ALTER TABLE rf_tbl_abcd_nopk ALTER COLUMN c SET NOT NULL;
+CREATE UNIQUE INDEX idx_abcd_nopk_c ON rf_tbl_abcd_nopk(c);
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY USING INDEX idx_abcd_nopk_c;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+-- ======================================================
 
 -- 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 64e71d0..de6b73d 100644
--- a/src/test/subscription/t/027_row_filter.pl
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -19,6 +19,8 @@ $node_subscriber->start;
 $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',
 	"CREATE TABLE tab_rowfilter_3 (a int primary key, b boolean)");
@@ -280,9 +282,7 @@ is($result, qq(13|0|12), 'check replicated rows to tab_rowfilter_4');
 # - 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
+# - DELETE (1700)              YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
 #
 $result =
   $node_subscriber->safe_psql('postgres',
@@ -291,7 +291,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
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index f41ef0d..575969c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3501,6 +3501,7 @@ replace_rte_variables_context
 ret_type
 rewind_source
 rewrite_event
+rf_context
 rijndael_ctx
 rm_detail_t
 role_auth_extra
-- 
1.8.3.1