From 020e95cdf34e2c9b415b5649a49724b16f21f15a Mon Sep 17 00:00:00 2001
From: Peter Smith <peter.b.smith@fujitsu.com>
Date: Mon, 13 Dec 2021 20:04:42 +1100
Subject: [PATCH v46] 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 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      | 109 ++++++++++++-
 src/backend/executor/execReplication.c    |  36 +++-
 src/backend/parser/parse_agg.c            |   8 +-
 src/backend/parser/parse_expr.c           |  19 +--
 src/backend/parser/parse_func.c           |   3 +-
 src/backend/parser/parse_oper.c           |   7 -
 src/backend/utils/cache/relcache.c        | 262 ++++++++++++++++++++++++++----
 src/include/utils/rel.h                   |   7 +
 src/include/utils/relcache.h              |   1 +
 src/test/regress/expected/publication.out | 223 ++++++++++++++++++++-----
 src/test/regress/sql/publication.sql      | 174 +++++++++++++++++---
 src/test/subscription/t/027_row_filter.pl |   7 +-
 src/tools/pgindent/typedefs.list          |   1 +
 13 files changed, 721 insertions(+), 136 deletions(-)

diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index bc5f6a2..2a4ee71 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"
@@ -112,6 +114,102 @@ check_publication_add_schema(Oid schemaid)
 }
 
 /*
+ * 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.
+ *
+ * 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 (IsA(node, Var))
+	{
+		Var		   *var = (Var *) node;
+
+		/* User-defined types not allowed. */
+		if (var->vartype >= FirstNormalObjectId)
+			errdetail_msg = _("user-defined types are not allowed");
+	}
+	else if (IsA(node, List) || IsA(node, Const) || IsA(node, BoolExpr) || IsA(node, NullIfExpr) ||
+			 IsA(node, NullTest) || IsA(node, BooleanTest) || IsA(node, CoalesceExpr) ||
+			 IsA(node, CaseExpr) || IsA(node, CaseTestExpr) || IsA(node, MinMaxExpr) ||
+			 IsA(node, ArrayExpr) || IsA(node, ScalarArrayOpExpr) || IsA(node, XmlExpr))
+	{
+		/* OK, these nodes are part of simple expressions */
+	}
+	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 +339,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)
@@ -355,6 +449,13 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
 
 		/* Fix up 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 574d7d2..2cbe2aa 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			invalid_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;
+
+	invalid_rfcolnum = RelationGetInvalidRowFilterCol(rel);
+
+	/*
+	 * 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 (invalid_rfcolnum)
+	{
+		const char *colname = get_attname(RelationGetRelid(rel),
+										  invalid_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 193c87d..0d39cfe 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -552,11 +552,7 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr)
 
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			if (isAgg)
-				err = _("aggregate functions are not allowed in publication WHERE expressions");
-			else
-				err = _("grouping operations are not allowed in publication WHERE expressions");
-
+			/* okay (see function rowfilter_walker in pg_publication.c). */
 			break;
 
 		case EXPR_KIND_CYCLE_MARK:
@@ -951,7 +947,7 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
 			errkind = true;
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("window functions are not allowed in publication WHERE expressions");
+			/* okay (see function rowfilter_walker in pg_publication.c). */
 			break;
 
 			/*
diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c
index 3d43839..7933387 100644
--- a/src/backend/parser/parse_expr.c
+++ b/src/backend/parser/parse_expr.c
@@ -200,19 +200,8 @@ transformExprRecurse(ParseState *pstate, Node *expr)
 			break;
 
 		case T_FuncCall:
-			{
-				/*
-				 * Forbid functions in publication WHERE condition
-				 */
-				if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
-					ereport(ERROR,
-							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-							 errmsg("functions are not allowed in publication WHERE expressions"),
-							 parser_errposition(pstate, exprLocation(expr))));
-
-				result = transformFuncCall(pstate, (FuncCall *) expr);
-				break;
-			}
+			result = transformFuncCall(pstate, (FuncCall *) expr);
+			break;
 
 		case T_MultiAssignRef:
 			result = transformMultiAssignRef(pstate, (MultiAssignRef *) expr);
@@ -1777,7 +1766,7 @@ transformSubLink(ParseState *pstate, SubLink *sublink)
 			err = _("cannot use subquery in column generation expression");
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("cannot use subquery in publication WHERE expression");
+			/* okay (see function rowfilter_walker in pg_publication.c). */
 			break;
 
 			/*
@@ -3100,7 +3089,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..2b2486a 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2656,7 +2656,8 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location)
 			errkind = true;
 			break;
 		case EXPR_KIND_PUBLICATION_WHERE:
-			err = _("set-returning functions are not allowed in publication WHERE expressions");
+			/* okay (see function rowfilter_walker in pg_publication.c). */
+			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/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 105d8d4..ed04881 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"
 
@@ -5521,57 +5525,169 @@ RelationGetExclusionInfo(Relation indexRelation,
 	MemoryContextSwitchTo(oldcxt);
 }
 
+/* For invalid_rowfilter_column_walker. */
+typedef struct {
+	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;
+
 /*
- * 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
+invalid_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 need to convert the column number of
+		 * parent to the column number of child relation first.
+		 */
+		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, invalid_rowfilter_column_walker,
+								  (void *) context);
+}
+
+/*
+ * Append to cur_puboids each member of add_puboids that isn't already in
+ * cur_puboids.
+ *
+ * Also update the top most parent relation's relid in the publication.
+ */
+static void
+concat_publication_oid(Oid relid,
+					   List **cur_puboids,
+					   List **toprelid_in_pub,
+					   const List *add_puboids)
+{
+	ListCell   *lc1,
+			   *lc2,
+			   *lc3;
+
+	foreach(lc1, add_puboids)
+	{
+		bool		is_member = false;
+
+		forboth(lc2, *cur_puboids, lc3, *toprelid_in_pub)
+		{
+			if (lfirst_oid(lc2) == lfirst_oid(lc1))
+			{
+				is_member = true;
+				lfirst_oid(lc3) = relid;
+			}
+		}
+
+		if (!is_member)
+		{
+			*cur_puboids = lappend_oid(*cur_puboids, lfirst_oid(lc1));
+			*toprelid_in_pub = lappend_oid(*toprelid_in_pub, relid);
+		}
+	}
+}
+
+/*
+ * Get the invalid row filter column number 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,
+ * 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
+RelationGetInvalidRowFilterCol(Relation relation)
+{
+	List		   *puboids,
+				   *toprelid_in_pub;
+	ListCell	   *lc;
+	MemoryContext	oldcxt;
+	Oid				schemaid;
+	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));
+	toprelid_in_pub = puboids = NIL;
+	concat_publication_oid(relid, &puboids, &toprelid_in_pub,
+						   GetRelationPublications(relid));
 	schemaid = RelationGetNamespace(relation);
-	puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
+	concat_publication_oid(relid, &puboids, &toprelid_in_pub,
+						   GetSchemaPublications(schemaid));
 
 	if (relation->rd_rel->relispartition)
 	{
 		/* Add publications that the ancestors are in too. */
-		List	   *ancestors = get_partition_ancestors(RelationGetRelid(relation));
+		List	   *ancestors = get_partition_ancestors(relid);
 		ListCell   *lc;
 
 		foreach(lc, ancestors)
 		{
 			Oid			ancestor = lfirst_oid(lc);
 
-			puboids = list_concat_unique_oid(puboids,
-											 GetRelationPublications(ancestor));
+			concat_publication_oid(ancestor, &puboids, &toprelid_in_pub,
+								   GetRelationPublications(ancestor));
 			schemaid = get_rel_namespace(ancestor);
-			puboids = list_concat_unique_oid(puboids,
-											 GetSchemaPublications(schemaid));
+			concat_publication_oid(ancestor, &puboids, &toprelid_in_pub,
+								   GetSchemaPublications(schemaid));
 		}
+
+		relid = llast_oid(ancestors);
 	}
-	puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+	concat_publication_oid(relid, &puboids, &toprelid_in_pub,
+						   GetAllTablesPublications());
+
+	/*
+	 * Find what are the cols that are part of the REPLICA IDENTITY.
+	 * Note that REPLICA IDENTITY DEFAULT means primary key or nothing.
+	 */
+	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);
 		HeapTuple	tup;
+
 		Form_pg_publication pubform;
 
 		tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
@@ -5581,35 +5697,116 @@ 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, 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 ((pubform->pubupdate || pubform->pubdelete) &&
+			relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL &&
+			rfcol_valid)
+		{
+			HeapTuple	rftuple;
+
+			if (pubform->pubviaroot)
+				relid = list_nth_oid(toprelid_in_pub,
+									 foreach_current_index(lc));
+			else
+				relid = RelationGetRelid(relation);
+
+			rftuple = SearchSysCache2(PUBLICATIONRELMAP,
+									  ObjectIdGetDatum(relid),
+									  ObjectIdGetDatum(pubid));
+
+			if (HeapTupleIsValid(rftuple))
+			{
+				Datum		rfdatum;
+				bool		rfisnull;
+				Node	   *rfnode;
+
+				context.pubviaroot = pubform->pubviaroot;
+				context.parentid = relid;
+				context.relid = RelationGetRelid(relation);
+
+				rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
+										  Anum_pg_publication_rel_prqual,
+										  &rfisnull);
+
+				if (!rfisnull)
+				{
+					rfnode = stringToNode(TextDatumGetCString(rfdatum));
+					rfcol_valid = !invalid_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 &&
+			!rfcol_valid)
 			break;
 	}
 
+	bms_free(context.bms_replident);
+
 	if (relation->rd_pubactions)
 	{
 		pfree(relation->rd_pubactions);
 		relation->rd_pubactions = NULL;
 	}
 
+	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 pubactions;
+	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)
+		(void) RelationGetInvalidRowFilterCol(relation);
+
+	return memcpy(pubactions, relation->rd_pubactions,
+				  sizeof(PublicationActions));
 }
 
 /*
@@ -6163,6 +6360,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/utils/rel.h b/src/include/utils/rel.h
index 3128127..27cec81 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -164,6 +164,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
 	 * defaults".
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 82316bb..25c759f 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 RelationGetInvalidRowFilterCol(Relation relation);
 
 extern void RelationInitTableAccessMethod(Relation relation);
 
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index f93a63d..69ed2c8 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_tb16(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_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 +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,83 +311,221 @@ 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 are not allowed WHERE row-filter
+-- 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 row-filters
+-- fail - duplicate tables are not allowed if that table has any WHERE clause
 SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1 WHERE (a = 1), testpub_rf_tbl1 WITH (publish = 'insert');
 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 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 testpub7 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.
+-- 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:  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
 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_tb16;
-ERROR:  cannot add relation "testpub_rf_myschema1.testpub_rf_tb16" to publication
-DETAIL:  Table's schema "testpub_rf_myschema1" is already part of the publication or part of the specified schema list.
+CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_schema2;
+ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_schema2, TABLE testpub_rf_schema2.testpub_rf_tb16;
+ERROR:  cannot add relation "testpub_rf_schema2.testpub_rf_tb16" 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_tb16;
-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_tb16;
+DROP SCHEMA testpub_rf_schema1;
+DROP SCHEMA testpub_rf_schema2;
 DROP PUBLICATION testpub5;
 DROP PUBLICATION testpub7;
 DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
+-- ======================================================
+-- 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 c13ccc5..0985681 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_tb16(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_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);
@@ -162,51 +165,174 @@ 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 are not allowed WHERE row-filter
+-- 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 row-filters
+-- fail - duplicate tables are not allowed if that table has any WHERE clause
 SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1 WHERE (a = 1), testpub_rf_tbl1 WITH (publish = 'insert');
 CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1, testpub_rf_tbl1 WHERE (a = 2) WITH (publish = 'insert');
 RESET client_min_messages;
--- fail - functions disallowed
-ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
--- fail - user-defined operators disallowed
-CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
-CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+-- fail - user-defined operators are not allowed
+CREATE FUNCTION testpub_rf_func1(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func1, LEFTARG = integer, RIGHTARG = integer);
 CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+-- fail - user-defined functions are not allowed
+CREATE FUNCTION testpub_rf_func2() RETURNS integer AS $$ BEGIN RETURN 123; END; $$ LANGUAGE plpgsql;
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a >= testpub_rf_func2());
+-- fail - non-immutable functions are not allowed. random() is volatile.
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a < random());
+-- ok - 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 testpub7 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+-- 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_tb16;
+CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_schema2;
+ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_schema2, TABLE testpub_rf_schema2.testpub_rf_tb16;
 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_tb16;
-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_tb16;
+DROP SCHEMA testpub_rf_schema1;
+DROP SCHEMA testpub_rf_schema2;
 DROP PUBLICATION testpub5;
 DROP PUBLICATION testpub7;
 DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
+
+-- ======================================================
+-- 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 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

