v50-0003-Row-filter-validation.patch

From d814336f61c3463d1f3cbc738a981cf159a312f1 Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Fri, 17 Dec 2021 20:37:32 +1100
Subject: [PATCH v50 3/8] Row-filter validation

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

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

Only simple filter expressions are permitted. Specifically:
- no user-defined operators.
- no user-defined functions.
- no user-defined types.
- no system columns.
- no system functions (unless they are immutable). See design decision at [1].
[1] https://www.postgresql.org/message-id/CAA4eK1%2BXoD49bz5-2TtiD0ugq4PHSRX2D1sLPR_X4LNtdMc4OQ%40mail.gmail.com

Permits only simple nodes including:
List, Const, BoolExpr, NullIfExpr, NullTest, BooleanTest, CoalesceExpr,
CaseExpr, CaseTestExpr, MinMaxExpr, ArrayExpr, ScalarArrayOpExpr, XmlExpr.

Author: Peter Smith, Euler Taveira

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

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

Row filter columns invalidation is done in CheckCmdReplicaIdentity, so that
the invalidation is executed only when actual UPDATE or DELETE executed on
the published relation. This is consistent with the existing check about
replica identity and can detect the change related to the row filter in time.

Cache the results of the validation for row filter columns in relcache to
reduce the cost of the validation. It is safe to do this because every
operation that change the row filter and replica identity will invalidate the
relcache.

Author: Hou zj
---
 src/backend/catalog/pg_publication.c        | 177 +++++++++++++-
 src/backend/executor/execReplication.c      |  36 ++-
 src/backend/parser/parse_agg.c              |  10 -
 src/backend/parser/parse_expr.c             |  21 +-
 src/backend/parser/parse_func.c             |   3 -
 src/backend/parser/parse_oper.c             |   7 -
 src/backend/parser/parse_relation.c         |   9 -
 src/backend/replication/pgoutput/pgoutput.c |  27 +--
 src/backend/utils/cache/relcache.c          | 242 +++++++++++++++++---
 src/include/catalog/pg_publication.h        |   2 +-
 src/include/parser/parse_node.h             |   1 -
 src/include/utils/rel.h                     |   7 +
 src/include/utils/relcache.h                |   1 +
 src/test/regress/expected/publication.out   | 227 ++++++++++++++----
 src/test/regress/sql/publication.sql        | 174 ++++++++++++--
 src/test/subscription/t/027_row_filter.pl   |   7 +-
 src/tools/pgindent/typedefs.list            |   1 +
 17 files changed, 783 insertions(+), 169 deletions(-)

diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 0929aa0a35..e3c154e31f 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"
@@ -111,6 +113,137 @@ check_publication_add_schema(Oid schemaid)
 				 errdetail("Temporary schemas cannot be replicated.")));
 }
 
+/*
+ * Is this a simple Node permitted within a row filter expression?
+ */
+static bool
+IsRowFilterSimpleExpr(Node *node)
+{
+	switch (nodeTag(node))
+	{
+		case T_ArrayExpr:
+		case T_BooleanTest:
+		case T_BoolExpr:
+		case T_CaseExpr:
+		case T_CaseTestExpr:
+		case T_CoalesceExpr:
+		case T_Const:
+		case T_List:
+		case T_MinMaxExpr:
+		case T_NullIfExpr:
+		case T_NullTest:
+		case T_ScalarArrayOpExpr:
+		case T_XmlExpr:
+			return true;
+		default:
+			return false;
+	}
+}
+
+/*
+ * The row filter walker checks if the row filter expression is a "simple
+ * expression".
+ *
+ * It allows 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.
+ * - Non-immutable builtin functions are not allowed.
+ * - System columns are not allowed.
+ *
+ * Notes:
+ *
+ * We don't allow user-defined functions/operators/types because (a) if the user
+ * drops such a user-definition 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
+ * non-immutable functions can access the database and would lead to the problem
+ * (b) mentioned in the previous paragraph.
+ */
+static bool
+rowfilter_walker(Node *node, Relation relation)
+{
+	char *errdetail_msg = NULL;
+
+	if (node == NULL)
+		return false;
+
+
+	if (IsRowFilterSimpleExpr(node))
+	{
+		/* OK, node is part of simple expressions */
+	}
+	else if (IsA(node, Var))
+	{
+		Var		   *var = (Var *) node;
+
+		/* User-defined types not allowed. */
+		if (var->vartype >= FirstNormalObjectId)
+			errdetail_msg = _("User-defined types are not allowed");
+
+		/* System columns not allowed. */
+		else if (var->varattno < InvalidAttrNumber)
+		{
+			Oid			relid = RelationGetRelid(relation);
+			const char *colname = get_attname(relid, var->varattno, false);
+
+			errdetail_msg = psprintf(_("Cannot use system column (%s)."), colname);
+		}
+	}
+	else if (IsA(node, OpExpr))
+	{
+		/* OK, except user-defined operators are not allowed. */
+		if (((OpExpr *) node)->opno >= FirstNormalObjectId)
+			errdetail_msg = _("User-defined operators are not allowed.");
+	}
+	else if (IsA(node, FuncExpr))
+	{
+		Oid			funcid = ((FuncExpr *) node)->funcid;
+		const char *funcname = get_func_name(funcid);
+
+		/*
+		 * User-defined functions are not allowed.
+		 * System-functions that are not IMMUTABLE are not allowed.
+		 */
+		if (funcid >= FirstNormalObjectId)
+			errdetail_msg = psprintf(_("User-defined functions are not allowed (%s)."),
+								 funcname);
+		else if (func_volatile(funcid) != PROVOLATILE_IMMUTABLE)
+			errdetail_msg = psprintf(_("Non-immutable built-in functions are not allowed (%s)."),
+								 funcname);
+	}
+	else
+	{
+		elog(DEBUG1, "the row filter contained something unexpected: %s", nodeToString(node));
+
+		ereport(ERROR,
+				(errmsg("invalid publication WHERE expression for relation \"%s\"",
+						RelationGetRelationName(relation)),
+				errdetail("Expressions only allow columns, constants and some built-in functions and operators.")
+				));
+	}
+
+	if (errdetail_msg)
+		ereport(ERROR,
+				(errmsg("invalid publication WHERE expression for relation \"%s\"",
+						RelationGetRelationName(relation)),
+						errdetail("%s", errdetail_msg)
+				));
+
+	return expression_tree_walker(node, rowfilter_walker, (void *)relation);
+}
+
 /*
  * Returns if relation represented by oid and Form_pg_class entry
  * is publishable.
@@ -241,10 +374,6 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
 	PG_RETURN_BOOL(result);
 }
 
-/*
- * Gets the relations based on the publication partition option for a specified
- * relation.
- */
 List *
 GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
 							   Oid relid)
@@ -298,7 +427,7 @@ GetTransformedWhereClause(ParseState *pstate, PublicationRelInfo *pri,
 	addNSItemToQuery(pstate, nsitem, false, true, true);
 
 	whereclause = transformWhereClause(pstate, copyObject(pri->whereClause),
-									   EXPR_KIND_PUBLICATION_WHERE,
+									   EXPR_KIND_WHERE,
 									   "PUBLICATION WHERE");
 
 	/* Fix up collation information */
@@ -308,6 +437,37 @@ GetTransformedWhereClause(ParseState *pstate, PublicationRelInfo *pri,
 	return whereclause;
 }
 
+/*
+ * Check if any of the ancestors are published in the publication. If so,
+ * return the relid of the topmost ancestor that is published via this
+ * publication, otherwise InvalidOid.
+ */
+Oid
+GetTopMostAncestorInPublication(Oid puboid, List *ancestors)
+{
+	ListCell   *lc;
+	Oid			topmost_relid  = InvalidOid;
+
+	/*
+	 * Find the "topmost" ancestor that is in this
+	 * publication.
+	 */
+	foreach(lc, ancestors)
+	{
+		Oid			ancestor = lfirst_oid(lc);
+
+		if (list_member_oid(GetRelationPublications(ancestor),
+							puboid) ||
+			list_member_oid(GetSchemaPublications(get_rel_namespace(ancestor)),
+							puboid))
+		{
+			topmost_relid = ancestor;
+		}
+	}
+
+	return topmost_relid;
+}
+
 /*
  * Insert new publication / relation mapping.
  */
@@ -362,6 +522,13 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
 		 * collation information.
 		 */
 		whereclause = GetTransformedWhereClause(pstate, pri, true);
+
+		/*
+		 * Walk the parse-tree of this publication row filter expression and
+		 * throw an error if anything not permitted or unexpected is
+		 * encountered.
+		 */
+		rowfilter_walker(whereclause, targetrel);
 	}
 
 	/* Form a tuple. */
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 574d7d27fd..42c5dbe2b9 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -568,14 +568,46 @@ void
 CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
 {
 	PublicationActions *pubactions;
+	AttrNumber			bad_rfcolnum;
 
 	/* We only need to do checks for UPDATE and DELETE. */
 	if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
 		return;
 
+	if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
+		return;
+
+	bad_rfcolnum = GetRelationPublicationInfo(rel, true);
+
+	/*
+	 * It is only safe to execute UPDATE/DELETE when all columns referenced in
+	 * the row filters from publications which the relation is in are valid,
+	 * which means all referenced columns are part of REPLICA IDENTITY, or the
+	 * table do not publish UPDATES or DELETES.
+	 */
+	if (AttributeNumberIsValid(bad_rfcolnum))
+	{
+		const char *colname = get_attname(RelationGetRelid(rel),
+										  bad_rfcolnum, false);
+
+		if (cmd == CMD_UPDATE)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+					 errmsg("cannot update table \"%s\"",
+							RelationGetRelationName(rel)),
+					 errdetail("Column \"%s\" used in the publication WHERE expression is not part of the replica identity.",
+							   colname)));
+		else if (cmd == CMD_DELETE)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+					 errmsg("cannot delete from table \"%s\"",
+							RelationGetRelationName(rel)),
+					 errdetail("Column \"%s\" used in the publication WHERE expression is not part of the replica identity.",
+							   colname)));
+	}
+
 	/* If relation has replica identity we are always good. */
-	if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
-		OidIsValid(RelationGetReplicaIndex(rel)))
+	if (OidIsValid(RelationGetReplicaIndex(rel)))
 		return;
 
 	/*
diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index 193c87d8b7..7d829a05a9 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -551,13 +551,6 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr)
 				err = _("grouping operations are not allowed in COPY FROM WHERE conditions");
 
 			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");
-
-			break;
 
 		case EXPR_KIND_CYCLE_MARK:
 			errkind = true;
@@ -950,9 +943,6 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
 		case EXPR_KIND_CYCLE_MARK:
 			errkind = true;
 			break;
-		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("window functions are not allowed in publication WHERE expressions");
-			break;
 
 			/*
 			 * There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c
index 3d43839b35..2d1a477154 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);
@@ -515,7 +504,6 @@ transformColumnRef(ParseState *pstate, ColumnRef *cref)
 		case EXPR_KIND_COPY_WHERE:
 		case EXPR_KIND_GENERATED_COLUMN:
 		case EXPR_KIND_CYCLE_MARK:
-		case EXPR_KIND_PUBLICATION_WHERE:
 			/* okay */
 			break;
 
@@ -1776,9 +1764,6 @@ transformSubLink(ParseState *pstate, SubLink *sublink)
 		case EXPR_KIND_GENERATED_COLUMN:
 			err = _("cannot use subquery in column generation expression");
 			break;
-		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("cannot use subquery in publication WHERE expression");
-			break;
 
 			/*
 			 * There is intentionally no default: case here, so that the
@@ -3099,8 +3084,6 @@ ParseExprKindName(ParseExprKind exprKind)
 			return "GENERATED AS";
 		case EXPR_KIND_CYCLE_MARK:
 			return "CYCLE";
-		case EXPR_KIND_PUBLICATION_WHERE:
-			return "publication 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 29bebb73eb..542f9167aa 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2655,9 +2655,6 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location)
 		case EXPR_KIND_CYCLE_MARK:
 			errkind = true;
 			break;
-		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("set-returning functions are not allowed in publication WHERE expressions");
-			break;
 
 			/*
 			 * There is intentionally no default: case here, so that the
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/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index 036d9c6d26..c5c3f26ecf 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -3538,20 +3538,11 @@ errorMissingRTE(ParseState *pstate, RangeVar *relation)
 						  rte->eref->aliasname)),
 				 parser_errposition(pstate, relation->location)));
 	else
-	{
-		if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
-			ereport(ERROR,
-					(errcode(ERRCODE_UNDEFINED_TABLE),
-					 errmsg("publication WHERE expression invalid reference to table \"%s\"",
-							relation->relname),
-					 parser_errposition(pstate, relation->location)));
-
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_TABLE),
 				 errmsg("missing FROM-clause entry for table \"%s\"",
 						relation->relname),
 				 parser_errposition(pstate, relation->location)));
-	}
 }
 
 /*
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index be5ec1d28f..7192948399 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1572,26 +1572,17 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
 				 */
 				if (am_partition)
 				{
-					List	   *ancestors = get_partition_ancestors(relid);
-					ListCell   *lc2;
+					Oid		ancestor;
+					List   *ancestors = get_partition_ancestors(relid);
 
-					/*
-					 * Find the "topmost" ancestor that is in this
-					 * publication.
-					 */
-					foreach(lc2, ancestors)
+					ancestor = GetTopMostAncestorInPublication(pub->oid,
+															   ancestors);
+
+					if (ancestor != InvalidOid)
 					{
-						Oid			ancestor = lfirst_oid(lc2);
-
-						if (list_member_oid(GetRelationPublications(ancestor),
-											pub->oid) ||
-							list_member_oid(GetSchemaPublications(get_rel_namespace(ancestor)),
-											pub->oid))
-						{
-							ancestor_published = true;
-							if (pub->pubviaroot)
-								publish_as_relid = ancestor;
-						}
+						ancestor_published = true;
+						if (pub->pubviaroot)
+							publish_as_relid = ancestor;
 					}
 				}
 
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 105d8d4601..14184e1337 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -56,6 +56,7 @@
 #include "catalog/pg_opclass.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
 #include "catalog/pg_rewrite.h"
 #include "catalog/pg_shseclabel.h"
 #include "catalog/pg_statistic_ext.h"
@@ -71,6 +72,8 @@
 #include "nodes/makefuncs.h"
 #include "nodes/nodeFuncs.h"
 #include "optimizer/optimizer.h"
+#include "parser/parse_clause.h"
+#include "parser/parse_relation.h"
 #include "rewrite/rewriteDefine.h"
 #include "rewrite/rowsecurity.h"
 #include "storage/lmgr.h"
@@ -84,6 +87,7 @@
 #include "utils/memutils.h"
 #include "utils/relmapper.h"
 #include "utils/resowner_private.h"
+#include "utils/ruleutils.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
 
@@ -267,6 +271,19 @@ typedef struct opclasscacheent
 
 static HTAB *OpClassCache = NULL;
 
+/*
+ * Information used to validate the columns in the row filter expression. see
+ * rowfilter_column_walker for details.
+ */
+typedef struct rf_context
+{
+	AttrNumber	invalid_rfcolnum;	/* invalid column number */
+	Bitmapset  *bms_replident;		/* bitset of replica identity col indexes */
+	bool		pubviaroot;			/* true if we are validating the parent
+									 * relation's row filter */
+	Oid			relid;				/* relid of the relation */
+	Oid			parentid;			/* relid of the parent relation */
+} rf_context;
 
 /* non-export function prototypes */
 
@@ -5521,39 +5538,91 @@ RelationGetExclusionInfo(Relation indexRelation,
 	MemoryContextSwitchTo(oldcxt);
 }
 
+
+
 /*
- * Get publication actions for the given relation.
+ * Check if any columns used in the row-filter WHERE clause are not part of
+ * REPLICA IDENTITY and save the invalid column number in
+ * rf_context::invalid_rfcolnum.
  */
-struct PublicationActions *
-GetRelationPublicationActions(Relation relation)
+static bool
+rowfilter_column_walker(Node *node, rf_context *context)
 {
-	List	   *puboids;
-	ListCell   *lc;
-	MemoryContext oldcxt;
-	Oid			schemaid;
-	PublicationActions *pubactions = palloc0(sizeof(PublicationActions));
+	if (node == NULL)
+		return false;
+
+	if (IsA(node, Var))
+	{
+		Var		   *var = (Var *) node;
+		AttrNumber	attnum = var->varattno;
+
+		/*
+		 * If pubviaroot is true, we are validating the row filter of the
+		 * parent table, but we can only use the bitset of replica identity col
+		 * indexes in the child table to check. So, we need to convert the
+		 * column number in the row filter expression to the child table's in
+		 * case the column order of the parent table is different from the
+		 * child table's.
+		 */
+		if (context->pubviaroot)
+		{
+			char *colname = get_attname(context->parentid, attnum, false);
+			attnum = get_attnum(context->relid, colname);
+		}
+
+		if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+						   context->bms_replident))
+		{
+			context->invalid_rfcolnum = attnum;
+			return true;
+		}
+	}
+
+	return expression_tree_walker(node, rowfilter_column_walker,
+								  (void *) context);
+}
+
+/*
+ * Get the publication information for the given relation.
+ *
+ * Traverse all the publications which the relation is in to get the
+ * publication actions. If the publication actions include UPDATE or DELETE and
+ * validate_rowfilter is true, then validate that if all columns referenced in
+ * the row filter expression are part of REPLICA IDENTITY.
+ *
+ * If not all the row filter columns are part of REPLICA IDENTITY, return the
+ * invalid column number, otherwise InvalidAttrNumber.
+ */
+AttrNumber
+GetRelationPublicationInfo(Relation relation, bool validate_rowfilter)
+{
+	List		   *puboids;
+	ListCell	   *lc;
+	MemoryContext	oldcxt;
+	Oid				schemaid;
+	List		   *ancestors = NIL;
+	Oid				relid = RelationGetRelid(relation);
+	rf_context		context = { 0 };
+	PublicationActions pubactions = { 0 };
+	bool			rfcol_valid = true;
+	AttrNumber		invalid_rfcolnum = InvalidAttrNumber;
 
 	/*
 	 * If not publishable, it publishes no actions.  (pgoutput_change() will
 	 * ignore it.)
 	 */
-	if (!is_publishable_relation(relation))
-		return pubactions;
-
-	if (relation->rd_pubactions)
-		return memcpy(pubactions, relation->rd_pubactions,
-					  sizeof(PublicationActions));
+	if (!is_publishable_relation(relation) || relation->rd_rfcol_valid)
+		return invalid_rfcolnum;
 
 	/* Fetch the publication membership info. */
-	puboids = GetRelationPublications(RelationGetRelid(relation));
+	puboids = GetRelationPublications(relid);
 	schemaid = RelationGetNamespace(relation);
 	puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
 
 	if (relation->rd_rel->relispartition)
 	{
 		/* Add publications that the ancestors are in too. */
-		List	   *ancestors = get_partition_ancestors(RelationGetRelid(relation));
-		ListCell   *lc;
+		ancestors = get_partition_ancestors(relid);
 
 		foreach(lc, ancestors)
 		{
@@ -5568,6 +5637,20 @@ GetRelationPublicationActions(Relation relation)
 	}
 	puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
 
+	/*
+	 * Find what are the cols that are part of the REPLICA IDENTITY.
+	 * Note that REPLICA IDENTITY DEFAULT means primary key or nothing.
+	 */
+	if (validate_rowfilter)
+	{
+		if (relation->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT)
+			context.bms_replident = RelationGetIndexAttrBitmap(relation,
+															   INDEX_ATTR_BITMAP_PRIMARY_KEY);
+		else if (relation->rd_rel->relreplident == REPLICA_IDENTITY_INDEX)
+			context.bms_replident = RelationGetIndexAttrBitmap(relation,
+															   INDEX_ATTR_BITMAP_IDENTITY_KEY);
+	}
+
 	foreach(lc, puboids)
 	{
 		Oid			pubid = lfirst_oid(lc);
@@ -5581,34 +5664,140 @@ GetRelationPublicationActions(Relation relation)
 
 		pubform = (Form_pg_publication) GETSTRUCT(tup);
 
-		pubactions->pubinsert |= pubform->pubinsert;
-		pubactions->pubupdate |= pubform->pubupdate;
-		pubactions->pubdelete |= pubform->pubdelete;
-		pubactions->pubtruncate |= pubform->pubtruncate;
+		pubactions.pubinsert |= pubform->pubinsert;
+		pubactions.pubupdate |= pubform->pubupdate;
+		pubactions.pubdelete |= pubform->pubdelete;
+		pubactions.pubtruncate |= pubform->pubtruncate;
 
 		ReleaseSysCache(tup);
 
 		/*
-		 * If we know everything is replicated, there is no point to check for
-		 * other publications.
+		 * If the publication action include UPDATE and DELETE and
+		 * validate_rowfilter flag is true, validates that any columns
+		 * referenced in the filter expression are part of REPLICA IDENTITY
+		 * index.
+		 *
+		 * FULL means all cols are in the REPLICA IDENTITY, so all cols are
+		 * allowed in the row-filter and we can skip the validation.
+		 *
+		 * If we already found the column in row filter which is not part of
+		 * REPLICA IDENTITY index, skip the validation too.
 		 */
-		if (pubactions->pubinsert && pubactions->pubupdate &&
-			pubactions->pubdelete && pubactions->pubtruncate)
+		if (validate_rowfilter &&
+			(pubform->pubupdate || pubform->pubdelete) &&
+			relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL &&
+			rfcol_valid)
+		{
+			HeapTuple	rftuple;
+			Oid			publish_as_relid = InvalidOid;
+
+			/*
+			 * For a partition, if pubviaroot is true, check if any of the
+			 * ancestors are published. If so, note down the topmost ancestor
+			 * that is published via this publication, the row filter
+			 * expression on which will be used to filter the partition's
+			 * changes. We could have got the topmost ancestor when collecting
+			 * the publication oids, but that will make the code more
+			 * complicated.
+			 */
+			if (pubform->pubviaroot && relation->rd_rel->relispartition)
+			{
+				if (pubform->puballtables)
+					publish_as_relid = llast_oid(ancestors);
+				else
+					publish_as_relid = GetTopMostAncestorInPublication(pubform->oid,
+																	   ancestors);
+			}
+
+			if (publish_as_relid == InvalidOid)
+				publish_as_relid = relid;
+
+			rftuple = SearchSysCache2(PUBLICATIONRELMAP,
+									  ObjectIdGetDatum(publish_as_relid),
+									  ObjectIdGetDatum(pubid));
+
+			if (HeapTupleIsValid(rftuple))
+			{
+				Datum		rfdatum;
+				bool		rfisnull;
+				Node	   *rfnode;
+
+				context.pubviaroot = pubform->pubviaroot;
+				context.parentid = publish_as_relid;
+				context.relid = relid;
+
+				rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
+										  Anum_pg_publication_rel_prqual,
+										  &rfisnull);
+
+				if (!rfisnull)
+				{
+					rfnode = stringToNode(TextDatumGetCString(rfdatum));
+					rfcol_valid = !rowfilter_column_walker(rfnode, &context);
+					invalid_rfcolnum = context.invalid_rfcolnum;
+					pfree(rfnode);
+				}
+
+				ReleaseSysCache(rftuple);
+			}
+		}
+
+		/*
+		 * If we know everything is replicated and some columns are not part of
+		 * replica identity, there is no point to check for other publications.
+		 */
+		if (pubactions.pubinsert && pubactions.pubupdate &&
+			pubactions.pubdelete && pubactions.pubtruncate &&
+			(!validate_rowfilter || !rfcol_valid))
 			break;
 	}
 
+	bms_free(context.bms_replident);
+
 	if (relation->rd_pubactions)
 	{
 		pfree(relation->rd_pubactions);
 		relation->rd_pubactions = NULL;
 	}
 
+	if (validate_rowfilter)
+		relation->rd_rfcol_valid = rfcol_valid;
+
 	/* Now save copy of the actions in the relcache entry. */
 	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
 	relation->rd_pubactions = palloc(sizeof(PublicationActions));
-	memcpy(relation->rd_pubactions, pubactions, sizeof(PublicationActions));
+	memcpy(relation->rd_pubactions, &pubactions, sizeof(PublicationActions));
 	MemoryContextSwitchTo(oldcxt);
 
+	return invalid_rfcolnum;
+}
+
+/*
+ * Get publication actions for the given relation.
+ */
+struct PublicationActions *
+GetRelationPublicationActions(Relation relation)
+{
+	PublicationActions *pubactions = palloc0(sizeof(PublicationActions));
+
+	/*
+	 * If not publishable, it publishes no actions.  (pgoutput_change() will
+	 * ignore it.)
+	 */
+	if (!is_publishable_relation(relation))
+		return pubactions;
+
+	if (relation->rd_pubactions)
+	{
+		memcpy(pubactions, relation->rd_pubactions,
+			   sizeof(PublicationActions));
+		return pubactions;
+	}
+
+	(void) GetRelationPublicationInfo(relation, false);
+	memcpy(pubactions, relation->rd_pubactions,
+		   sizeof(PublicationActions));
+
 	return pubactions;
 }
 
@@ -6163,6 +6352,7 @@ load_relcache_init_file(bool shared)
 		rel->rd_idattr = NULL;
 		rel->rd_hotblockingattr = NULL;
 		rel->rd_pubactions = NULL;
+		rel->rd_rfcol_valid = false;
 		rel->rd_statvalid = false;
 		rel->rd_statlist = NIL;
 		rel->rd_fkeyvalid = false;
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 96c55f627d..9e197defc0 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -135,6 +135,6 @@ extern char *get_publication_name(Oid pubid, bool missing_ok);
 extern Node *GetTransformedWhereClause(ParseState *pstate,
 									   PublicationRelInfo *pri,
 									   bool bfixupcollation);
-
+extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors);
 
 #endif							/* PG_PUBLICATION_H */
diff --git a/src/include/parser/parse_node.h b/src/include/parser/parse_node.h
index d58ae6a63f..ee179082ce 100644
--- a/src/include/parser/parse_node.h
+++ b/src/include/parser/parse_node.h
@@ -80,7 +80,6 @@ typedef enum ParseExprKind
 	EXPR_KIND_COPY_WHERE,		/* WHERE condition in COPY FROM */
 	EXPR_KIND_GENERATED_COLUMN, /* generation expression for a column */
 	EXPR_KIND_CYCLE_MARK,		/* cycle mark value */
-	EXPR_KIND_PUBLICATION_WHERE /* WHERE condition for a table in PUBLICATION */
 } ParseExprKind;
 
 
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 31281279cf..27cec813c0 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -163,6 +163,13 @@ typedef struct RelationData
 
 	PublicationActions *rd_pubactions;	/* publication actions */
 
+	/*
+	 * true if the columns referenced in row filters from all the publications
+	 * the relation is in are part of replica identity, or the publication
+	 * actions do not include UPDATE and DELETE.
+	 */
+	bool		rd_rfcol_valid;
+
 	/*
 	 * rd_options is set whenever rd_rel is loaded into the relcache entry.
 	 * Note that you can NOT look into rd_rel for this data.  NULL means "use
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 82316bba54..9cc4a380f8 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -76,6 +76,7 @@ extern void RelationInitIndexAccessInfo(Relation relation);
 /* caller must include pg_publication.h */
 struct PublicationActions;
 extern struct PublicationActions *GetRelationPublicationActions(Relation relation);
+extern AttrNumber GetRelationPublicationInfo(Relation relation, bool validate_rowfilter);
 
 extern void RelationInitTableAccessMethod(Relation relation);
 
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 02491c4b6b..e6651bdaae 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -243,18 +243,21 @@ 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 SCHEMA testpub_rf_myschema;
-CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
-CREATE SCHEMA testpub_rf_myschema1;
-CREATE TABLE testpub_rf_myschema1.testpub_rf_tbl6(i integer);
+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_tbl6 (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 +267,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 +278,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 +289,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))
 
@@ -308,40 +311,40 @@ Publications:
 DROP PUBLICATION testpub_dplus_rf_yes, testpub_dplus_rf_no;
 -- 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_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_myschema.testpub_rf_tbl5" WHERE (h < 999)
+    "testpub_rf_schema1.testpub_rf_tbl5" WHERE (h < 999)
 
 DROP PUBLICATION testpub_syntax2;
 -- fail - schemas don't allow WHERE clause
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a = 123);
 ERROR:  syntax error at or near "WHERE"
-LINE 1: ...tax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a =...
+LINE 1: ...ntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a =...
                                                              ^
-CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf_myschema WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1, testpub_rf_schema1 WHERE (a = 123);
 ERROR:  WHERE clause for schema not allowed
-LINE 1: ...ax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf...
+LINE 1: ...tax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1, testpub_rf...
                                                              ^
 RESET client_min_messages;
 -- fail - duplicate tables are not allowed if that table has any WHERE clause
@@ -353,43 +356,185 @@ ERROR:  conflicting or redundant WHERE clauses for table "testpub_rf_tbl1"
 RESET client_min_messages;
 -- fail - aggregate functions not allowed in WHERE clause
 ALTER PUBLICATION  testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
-ERROR:  functions are not allowed in publication WHERE expressions
+ERROR:  aggregate functions are not allowed in WHERE
 LINE 1: ...TION  testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
                                                                ^
--- 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 built-in functions are not allowed (random).
+-- ok - NULLIF is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1,2) = a);
+-- ok - builtin operators are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE ((a > 5) IS FALSE);
+-- ok - immutable builtin functions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- 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 testpub6 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL:  Expressions only allow columns, constants and some built-in functions and operators.
+-- fail - system columns are not allowed
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl1 WHERE ('(0,1)'::tid = ctid);
+ERROR:  invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL:  Cannot use system column (ctid).
+-- 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';
-CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_myschema1;
-ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_myschema1, TABLE testpub_rf_myschema1.testpub_rf_tbl6 WHERE (i < 99);
-ERROR:  cannot add relation "testpub_rf_myschema1.testpub_rf_tbl6" to publication
-DETAIL:  Table's schema "testpub_rf_myschema1" is already part of the publication or part of the specified schema list.
+CREATE PUBLICATION testpub6 FOR ALL TABLES IN SCHEMA testpub_rf_schema2;
+ALTER PUBLICATION testpub6 SET ALL TABLES IN SCHEMA testpub_rf_schema2, TABLE testpub_rf_schema2.testpub_rf_tbl6 WHERE (i < 99);
+ERROR:  cannot add relation "testpub_rf_schema2.testpub_rf_tbl6" to publication
+DETAIL:  Table's schema "testpub_rf_schema2" is already part of the publication or part of the specified schema list.
 RESET client_min_messages;
 DROP TABLE testpub_rf_tbl1;
 DROP TABLE testpub_rf_tbl2;
 DROP TABLE testpub_rf_tbl3;
 DROP TABLE testpub_rf_tbl4;
-DROP TABLE testpub_rf_myschema.testpub_rf_tbl5;
-DROP TABLE testpub_rf_myschema1.testpub_rf_tbl6;
-DROP SCHEMA testpub_rf_myschema;
-DROP SCHEMA testpub_rf_myschema1;
+DROP TABLE testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema1.testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema2.testpub_rf_tbl6;
+DROP SCHEMA testpub_rf_schema1;
+DROP SCHEMA testpub_rf_schema2;
 DROP PUBLICATION testpub5;
-DROP PUBLICATION testpub7;
+DROP PUBLICATION testpub6;
 DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
+-- ======================================================
+-- 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));
+CREATE TABLE rf_tbl_abcd_part_pk (a int PRIMARY KEY, b int) PARTITION by RANGE (a);
+CREATE TABLE rf_tbl_abcd_part_pk_1 (b int, a int PRIMARY KEY);
+ALTER TABLE rf_tbl_abcd_part_pk ATTACH PARTITION rf_tbl_abcd_part_pk_1 FOR VALUES FROM (1) TO (10);
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+-- ok - "a" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (b > 99);
+-- ok - "b" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_pk"
+DETAIL:  Column "c" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (d > 99);
+-- fail - "d" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_pk"
+DETAIL:  Column "d" used in the publication WHERE expression is not part of the replica identity.
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not part of REPLICA IDENTITY
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_nopk"
+DETAIL:  Column "a" used in the publication WHERE expression 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;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- ok - "a" is in REPLICA IDENTITY now
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_pk"
+DETAIL:  Column "a" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_pk"
+DETAIL:  Column "c" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_nopk"
+DETAIL:  Column "a" used in the publication WHERE expression 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;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_pk"
+DETAIL:  Column "a" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_nopk"
+DETAIL:  Column "a" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+-- Tests for partitioned table
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (a > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- ok - "a" is a OK col
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (b > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- fail - "b" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ERROR:  cannot update table "rf_tbl_abcd_part_pk_1"
+DETAIL:  Column "b" used in the publication WHERE expression is not part of the replica identity.
+DROP PUBLICATION testpub6;
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+DROP TABLE rf_tbl_abcd_part_pk;
+-- ======================================================
 -- 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 9185d5a1d2..a95c71b1cb 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -138,12 +138,15 @@ 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 SCHEMA testpub_rf_myschema;
-CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
-CREATE SCHEMA testpub_rf_myschema1;
-CREATE TABLE testpub_rf_myschema1.testpub_rf_tbl6(i integer);
+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_tbl6 (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);
@@ -162,19 +165,19 @@ RESET client_min_messages;
 DROP PUBLICATION testpub_dplus_rf_yes, testpub_dplus_rf_no;
 -- 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_schema1.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = 'insert');
 RESET client_min_messages;
 \dRp+ testpub_syntax2
 DROP PUBLICATION testpub_syntax2;
 -- fail - schemas don't allow WHERE clause
 SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a = 123);
-CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf_myschema WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1, testpub_rf_schema1 WHERE (a = 123);
 RESET client_min_messages;
 -- fail - duplicate tables are not allowed if that table has any WHERE clause
 SET client_min_messages = 'ERROR';
@@ -183,32 +186,157 @@ CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1, testpub_rf_tbl1 WHERE
 RESET client_min_messages;
 -- fail - aggregate functions not allowed in WHERE clause
 ALTER PUBLICATION  testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
--- 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 - NULLIF is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1,2) = a);
+-- ok - builtin operators are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE ((a > 5) IS FALSE);
+-- ok - immutable builtin functions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- 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 testpub6 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+-- fail - system columns are not allowed
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl1 WHERE ('(0,1)'::tid = ctid);
+-- 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_myschema1;
-ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_myschema1, TABLE testpub_rf_myschema1.testpub_rf_tbl6 WHERE (i < 99);
+CREATE PUBLICATION testpub6 FOR ALL TABLES IN SCHEMA testpub_rf_schema2;
+ALTER PUBLICATION testpub6 SET ALL TABLES IN SCHEMA testpub_rf_schema2, TABLE testpub_rf_schema2.testpub_rf_tbl6 WHERE (i < 99);
 RESET client_min_messages;
 
 DROP TABLE testpub_rf_tbl1;
 DROP TABLE testpub_rf_tbl2;
 DROP TABLE testpub_rf_tbl3;
 DROP TABLE testpub_rf_tbl4;
-DROP TABLE testpub_rf_myschema.testpub_rf_tbl5;
-DROP TABLE testpub_rf_myschema1.testpub_rf_tbl6;
-DROP SCHEMA testpub_rf_myschema;
-DROP SCHEMA testpub_rf_myschema1;
+DROP TABLE testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema1.testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema2.testpub_rf_tbl6;
+DROP SCHEMA testpub_rf_schema1;
+DROP SCHEMA testpub_rf_schema2;
 DROP PUBLICATION testpub5;
-DROP PUBLICATION testpub7;
+DROP PUBLICATION testpub6;
 DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
+
+-- ======================================================
+-- 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));
+CREATE TABLE rf_tbl_abcd_part_pk (a int PRIMARY KEY, b int) PARTITION by RANGE (a);
+CREATE TABLE rf_tbl_abcd_part_pk_1 (b int, a int PRIMARY KEY);
+ALTER TABLE rf_tbl_abcd_part_pk ATTACH PARTITION rf_tbl_abcd_part_pk_1 FOR VALUES FROM (1) TO (10);
+
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+-- ok - "a" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (b > 99);
+-- ok - "b" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (d > 99);
+-- fail - "d" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not part of REPLICA IDENTITY
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- Case 2. REPLICA IDENTITY FULL
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- ok - "a" is in REPLICA IDENTITY now
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- 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;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- Tests for partitioned table
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (a > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- ok - "a" is a OK col
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (b > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- fail - "b" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+
+DROP PUBLICATION testpub6;
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+DROP TABLE rf_tbl_abcd_part_pk;
+-- ======================================================
 
 -- 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 64e71d0adb..de6b73d26a 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',
@@ -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 0c61ccbdd0..89f3917352 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3503,6 +3503,7 @@ replace_rte_variables_context
 ret_type
 rewind_source
 rewrite_event
+rf_context
 rijndael_ctx
 rm_detail_t
 role_auth_extra
-- 
2.20.1