From 336ea6007d4c07108d2161cea44ea31f6aa87b45 Mon Sep 17 00:00:00 2001
From: "houzj.fnst" <houzj.fnst@fujitsu.com>
Date: Fri, 26 Nov 2021 10:07:55 +0800
Subject: [PATCH] cache the result of row filter column validation

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

Move the row filter columns invalidation to CheckCmdReplicaIdentity, so that
the invalidation is executed only when actual UPDATE or DELETE executed on the
published relation. It's consistent with the existing check about replica
identity.

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

Temporarily reserved the function GetRelationPublicationActions because it's a
public function.

---
 src/backend/catalog/pg_publication.c      | 131 ++++++---------------
 src/backend/executor/execReplication.c    |  29 ++++-
 src/backend/utils/cache/relcache.c        | 134 +++++++++++++++++-----
 src/include/catalog/pg_publication.h      |  13 +++
 src/include/utils/rel.h                   |   3 +-
 src/include/utils/relcache.h              |   1 +
 src/test/regress/expected/publication.out |  72 +++++++-----
 src/test/regress/sql/publication.sql      |  39 +++++--
 8 files changed, 254 insertions(+), 168 deletions(-)

diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index ae781cd7e0..4bcfdc548c 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -221,12 +221,29 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
 	PG_RETURN_BOOL(result);
 }
 
-/* For rowfilter_walker. */
-typedef struct {
-	Relation	rel;
-	bool		check_replident; /* check if Var is bms_replident member? */
-	Bitmapset  *bms_replident;
-} rf_context;
+/*
+ * Check if all the columns used in the row-filter WHERE clause are part of
+ * REPLICA IDENTITY
+ */
+bool
+check_rowfilter_replident(Node *node, Bitmapset *bms_replident)
+{
+	if (node == NULL)
+		return false;
+
+	if (IsA(node, Var))
+	{
+		Var		   *var = (Var *) node;
+		AttrNumber	attnum = var->varattno;
+
+		if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+						   bms_replident))
+			return true;
+	}
+
+	return expression_tree_walker(node, check_rowfilter_replident,
+								  (void *) bms_replident);
+}
 
 /*
  * The row filter walker checks that the row filter expression is legal.
@@ -242,15 +259,9 @@ typedef struct {
  * - User-defined functions are not allowed.
  * - System functions that are not IMMUTABLE are not allowed.
  * - NULLIF is allowed.
- *
- * Rules: Replica Identity validation
- * -----------------------------------
- * If the flag context.check_replident is true then validate that every variable
- * referenced by the filter expression is a valid member of the allowed set of
- * replica identity columns (context.bms_replindent)
  */
 static bool
-rowfilter_walker(Node *node, rf_context *context)
+rowfilter_walker(Node *node, Relation relation)
 {
 	char *forbidden = NULL;
 	bool too_complex = false;
@@ -258,29 +269,7 @@ rowfilter_walker(Node *node, rf_context *context)
 	if (node == NULL)
 		return false;
 
-	if (IsA(node, Var))
-	{
-		/* Optionally, do replica identify validation of the referenced column. */
-		if (context->check_replident)
-		{
-			Oid			relid = RelationGetRelid(context->rel);
-			Var		   *var = (Var *) node;
-			AttrNumber	attnum = var->varattno;
-
-			if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, context->bms_replident))
-			{
-				const char *colname = get_attname(relid, attnum, false);
-
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
-						errmsg("cannot add relation \"%s\" to publication",
-							   RelationGetRelationName(context->rel)),
-						errdetail("Row filter column \"%s\" is not part of the REPLICA IDENTITY",
-								  colname)));
-			}
-		}
-	}
-	else if (IsA(node, Const) || IsA(node, BoolExpr) || IsA(node, NullIfExpr))
+	else if (IsA(node, Var) || IsA(node, Const) || IsA(node, BoolExpr) || IsA(node, NullIfExpr))
 	{
 		/* OK */
 	}
@@ -320,74 +309,18 @@ rowfilter_walker(Node *node, rf_context *context)
 	if (too_complex)
 		ereport(ERROR,
 				(errmsg("invalid publication WHERE expression for relation \"%s\"",
-						RelationGetRelationName(context->rel)),
+						RelationGetRelationName(relation)),
 				errhint("only simple expressions using columns, constants and immutable system functions are allowed")
 				));
 
 	if (forbidden)
 		ereport(ERROR,
 				(errmsg("invalid publication WHERE expression for relation \"%s\"",
-						RelationGetRelationName(context->rel)),
+						RelationGetRelationName(relation)),
 						errdetail("%s", forbidden)
 				));
 
-	return expression_tree_walker(node, rowfilter_walker, (void *)context);
-}
-
-/*
- * Check if the row-filter is valid according to the following rules:
- *
- * 1. Only certain simple node types are permitted in the expression. See
- * function rowfilter_walker for details.
- *
- * 2. If the publish operation contains "delete" or "update" then only columns
- * that are allowed by the REPLICA IDENTITY rules are permitted to be used in
- * the row-filter WHERE clause.
- */
-static void
-rowfilter_expr_checker(Publication *pub, Relation rel, Node *rfnode)
-{
-	rf_context	context = {0};
-
-	context.rel = rel;
-
-	/*
-	 * For "delete" or "update", check that filter cols are also valid replica
-	 * identity cols.
-	 */
-	if (pub->pubactions.pubdelete || pub->pubactions.pubupdate)
-	{
-		char replica_identity = rel->rd_rel->relreplident;
-
-		if (replica_identity == REPLICA_IDENTITY_FULL)
-		{
-			/*
-			 * FULL means all cols are in the REPLICA IDENTITY, so all cols are
-			 * allowed in the row-filter too.
-			 */
-		}
-		else
-		{
-			context.check_replident = true;
-
-			/*
-			 * Find what are the cols that are part of the REPLICA IDENTITY.
-			 * Note that REPLICA IDENTIY DEFAULT means primary key or nothing.
-			 */
-			if (replica_identity == REPLICA_IDENTITY_DEFAULT)
-				context.bms_replident = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
-			else
-				context.bms_replident = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
-		}
-	}
-
-	/*
-	 * Walk the parse-tree of this publication row filter expression and throw an
-	 * error if anything not permitted or unexpected is encountered.
-	 */
-	rowfilter_walker(rfnode, &context);
-
-	bms_free(context.bms_replident);
+	return expression_tree_walker(node, rowfilter_walker, (void *) relation);
 }
 
 List *
@@ -487,8 +420,12 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
 		/* Fix up collation information */
 		assign_expr_collations(pstate, whereclause);
 
-		/* Validate the row-filter. */
-		rowfilter_expr_checker(pub, targetrel, whereclause);
+		/*
+		 * 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..c917466e7e 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -567,12 +567,34 @@ ExecSimpleRelationDelete(ResultRelInfo *resultRelInfo,
 void
 CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
 {
-	PublicationActions *pubactions;
+	PublicationInfo	   *pubinfo;
 
 	/* We only need to do checks for UPDATE and DELETE. */
 	if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
 		return;
 
+	pubinfo = RelationGetPublicationInfo(rel);
+
+	/*
+	 * if not all columns in the publication row filter are part of the REPLICA
+	 * IDENTITY, then it's unsafe to execute it for UPDATE and DELETE.
+	 */
+	if (!pubinfo->rfcol_valid_for_replid)
+	{
+		if (cmd == CMD_UPDATE)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+					 errmsg("cannot update table \"%s\"",
+							RelationGetRelationName(rel)),
+					 errdetail("Not all row filter columns are not part of the REPLICA IDENTITY")));
+		else if (cmd == CMD_DELETE)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+					 errmsg("cannot delete from table \"%s\"",
+							RelationGetRelationName(rel)),
+					 errdetail("Not all row filter columns are not part of the REPLICA IDENTITY")));
+	}
+
 	/* If relation has replica identity we are always good. */
 	if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
 		OidIsValid(RelationGetReplicaIndex(rel)))
@@ -583,14 +605,13 @@ CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
 	 *
 	 * Check if the table publishes UPDATES or DELETES.
 	 */
-	pubactions = GetRelationPublicationActions(rel);
-	if (cmd == CMD_UPDATE && pubactions->pubupdate)
+	if (cmd == CMD_UPDATE && pubinfo->pubactions.pubupdate)
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("cannot update table \"%s\" because it does not have a replica identity and publishes updates",
 						RelationGetRelationName(rel)),
 				 errhint("To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.")));
-	else if (cmd == CMD_DELETE && pubactions->pubdelete)
+	else if (cmd == CMD_DELETE && pubinfo->pubactions.pubdelete)
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("cannot delete from table \"%s\" because it does not have a replica identity and publishes deletes",
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 9fa9e671a1..542da49fe2 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"
@@ -2432,8 +2433,8 @@ RelationDestroyRelation(Relation relation, bool remember_tupdesc)
 	bms_free(relation->rd_keyattr);
 	bms_free(relation->rd_pkattr);
 	bms_free(relation->rd_idattr);
-	if (relation->rd_pubactions)
-		pfree(relation->rd_pubactions);
+	if (relation->rd_pubinfo)
+		pfree(relation->rd_pubinfo);
 	if (relation->rd_options)
 		pfree(relation->rd_options);
 	if (relation->rd_indextuple)
@@ -5547,22 +5548,45 @@ RelationGetExclusionInfo(Relation indexRelation,
 struct PublicationActions *
 GetRelationPublicationActions(Relation relation)
 {
-	List	   *puboids;
-	ListCell   *lc;
-	MemoryContext oldcxt;
-	Oid			schemaid;
-	PublicationActions *pubactions = palloc0(sizeof(PublicationActions));
+	PublicationInfo	   *pubinfo;
+	PublicationActions *pubactions = palloc0(sizeof(PublicationInfo));
+
+	pubinfo = RelationGetPublicationInfo(relation);
+
+	pubactions = memcpy(pubactions, relation->rd_pubinfo,
+						sizeof(PublicationActions));
+
+	pfree(pubinfo);
+
+	return pubactions;
+}
+
+
+
+/*
+ * Get publication information for the given relation.
+ */
+struct PublicationInfo *
+RelationGetPublicationInfo(Relation relation)
+{
+	List		   *puboids;
+	ListCell	   *lc;
+	MemoryContext	oldcxt;
+	Oid				schemaid;
+	Bitmapset	   *bms_replident = NULL;
+	PublicationInfo *pubinfo = palloc0(sizeof(PublicationInfo));
+
+	pubinfo->rfcol_valid_for_replid = true;
 
 	/*
 	 * If not publishable, it publishes no actions.  (pgoutput_change() will
 	 * ignore it.)
 	 */
 	if (!is_publishable_relation(relation))
-		return pubactions;
+		return pubinfo;
 
-	if (relation->rd_pubactions)
-		return memcpy(pubactions, relation->rd_pubactions,
-					  sizeof(PublicationActions));
+	if (relation->rd_pubinfo)
+		return memcpy(pubinfo, relation->rd_pubinfo, sizeof(PublicationInfo));
 
 	/* Fetch the publication membership info. */
 	puboids = GetRelationPublications(RelationGetRelid(relation));
@@ -5586,12 +5610,25 @@ GetRelationPublicationActions(Relation relation)
 											 GetSchemaPublications(schemaid));
 		}
 	}
+
 	puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
 
+	/*
+	 * Find what are the cols that are part of the REPLICA IDENTITY.
+	 * Note that REPLICA IDENTIY DEFAULT means primary key or nothing.
+	 */
+	if (relation->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT)
+		bms_replident = RelationGetIndexAttrBitmap(relation,
+												   INDEX_ATTR_BITMAP_PRIMARY_KEY);
+	else if (relation->rd_rel->relreplident == REPLICA_IDENTITY_INDEX)
+		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));
@@ -5601,35 +5638,80 @@ 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;
+		pubinfo->pubactions.pubinsert |= pubform->pubinsert;
+		pubinfo->pubactions.pubupdate |= pubform->pubupdate;
+		pubinfo->pubactions.pubdelete |= pubform->pubdelete;
+		pubinfo->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 UDDATE 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 ((pubform->pubupdate || pubform->pubdelete) &&
+			relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL)
+		{
+			HeapTuple	rftuple;
+
+			rftuple = SearchSysCache2(PUBLICATIONRELMAP,
+									  ObjectIdGetDatum(RelationGetRelid(relation)),
+									  ObjectIdGetDatum(pubid));
+
+			if (HeapTupleIsValid(rftuple))
+			{
+				Datum		rfdatum;
+				bool		rfisnull;
+				Node	   *rfnode;
+
+				rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
+										  Anum_pg_publication_rel_prqual,
+										  &rfisnull);
+
+				if (!rfisnull)
+				{
+					rfnode = stringToNode(TextDatumGetCString(rfdatum));
+					if (check_rowfilter_replident(rfnode, bms_replident))
+					{
+						pubinfo->rfcol_valid_for_replid = false;
+						ReleaseSysCache(rftuple);
+						break;
+					}
+				}
+
+				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)
+		if (pubinfo->pubactions.pubinsert && pubinfo->pubactions.pubupdate &&
+			pubinfo->pubactions.pubdelete && pubinfo->pubactions.pubtruncate &&
+			!pubinfo->rfcol_valid_for_replid)
 			break;
 	}
 
-	if (relation->rd_pubactions)
+	bms_free(bms_replident);
+
+	if (relation->rd_pubinfo)
 	{
-		pfree(relation->rd_pubactions);
-		relation->rd_pubactions = NULL;
+		pfree(relation->rd_pubinfo);
+		relation->rd_pubinfo = NULL;
 	}
 
 	/* 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));
+	relation->rd_pubinfo = palloc(sizeof(PublicationInfo));
+	memcpy(relation->rd_pubinfo, pubinfo, sizeof(PublicationInfo));
 	MemoryContextSwitchTo(oldcxt);
 
-	return pubactions;
+	return pubinfo;
 }
 
 /*
@@ -6184,7 +6266,7 @@ load_relcache_init_file(bool shared)
 		rel->rd_keyattr = NULL;
 		rel->rd_pkattr = NULL;
 		rel->rd_idattr = NULL;
-		rel->rd_pubactions = NULL;
+		rel->rd_pubinfo = NULL;
 		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 bd0d4cec05..f698049633 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -74,6 +74,18 @@ typedef struct PublicationActions
 	bool		pubtruncate;
 } PublicationActions;
 
+typedef struct PublicationInfo
+{
+	PublicationActions	pubactions;
+
+	/*
+	 * True if pubactions don't include UPDATE and DELETE or
+	 * all the columns in the row filter expression are part
+	 * of replica identity.
+	 */
+	bool				rfcol_valid_for_replid;
+} PublicationInfo;
+
 typedef struct Publication
 {
 	Oid			oid;
@@ -131,5 +143,6 @@ extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
 extern Oid	get_publication_oid(const char *pubname, bool missing_ok);
 extern char *get_publication_name(Oid pubid, bool missing_ok);
 
+extern bool check_rowfilter_replident(Node *node, Bitmapset *bms_replident);
 
 #endif							/* PG_PUBLICATION_H */
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index b4faa1c123..57cebde965 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -160,7 +160,8 @@ typedef struct RelationData
 	Bitmapset  *rd_pkattr;		/* cols included in primary key */
 	Bitmapset  *rd_idattr;		/* included in replica identity index */
 
-	PublicationActions *rd_pubactions;	/* publication actions */
+	/* data managed by RelationGetPublicationInfo: */
+	PublicationInfo *rd_pubinfo;	/* publication information */
 
 	/*
 	 * rd_options is set whenever rd_rel is loaded into the relcache entry.
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index aa060ef115..54f9825ebb 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 struct PublicationInfo *RelationGetPublicationInfo(Relation relation);
 
 extern void RelationInitTableAccessMethod(Relation relation);
 
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index fdf7659e82..d9e42e5924 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -416,21 +416,27 @@ DROP PUBLICATION testpub6;
 -- ok - "b" is a PK col
 SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (b > 99);
-RESET client_min_messages;
 DROP PUBLICATION testpub6;
--- fail - "c" is not part of the PK
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
-DETAIL:  Row filter column "c" is not part of the REPLICA IDENTITY
--- fail - "d" is not part of the PK
+-- 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:  Not all row filter columns are not part of the REPLICA IDENTITY
+DROP PUBLICATION testpub6;
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (d > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
-DETAIL:  Row filter column "d" is not part of the REPLICA IDENTITY
+-- 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:  Not all row filter columns are not part of the REPLICA IDENTITY
+DROP PUBLICATION testpub6;
 -- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
--- fail - "a" is not part of REPLICA IDENTITY
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_nopk" to publication
-DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- 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:  Not all row filter columns are not part of the REPLICA IDENTITY
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
 -- Case 2. REPLICA IDENTITY FULL
 ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
 ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL;
@@ -444,21 +450,29 @@ SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
 DROP PUBLICATION testpub6;
 RESET client_min_messages;
+SET client_min_messages = 'ERROR';
 -- Case 3. REPLICA IDENTITY NOTHING
 ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
 ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
--- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
-DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
--- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+-- 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:  Not all row filter columns are not part of the REPLICA IDENTITY
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
-DETAIL:  Row filter column "c" is not part of the REPLICA IDENTITY
--- fail - "a" is not in REPLICA IDENTITY NOTHING
+ERROR:  publication "testpub6" already exists
+-- 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:  Not all row filter columns are not part of the REPLICA IDENTITY
+DROP PUBLICATION testpub6;
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_nopk" to publication
-DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- 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:  Not all row filter columns are not part of the REPLICA IDENTITY
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
 -- 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);
@@ -466,21 +480,23 @@ ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY USING INDEX idx_abcd_pk_c;
 ALTER TABLE rf_tbl_abcd_nopk ALTER COLUMN c SET NOT NULL;
 CREATE UNIQUE INDEX idx_abcd_nopk_c ON rf_tbl_abcd_nopk(c);
 ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY USING INDEX idx_abcd_nopk_c;
--- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_pk" to publication
-DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- fail - "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:  Not all row filter columns are not part of the REPLICA IDENTITY
+DROP PUBLICATION testpub6;
 -- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
-SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
 DROP PUBLICATION testpub6;
-RESET client_min_messages;
--- fail - "a" is not in REPLICA IDENTITY INDEX
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
-ERROR:  cannot add relation "rf_tbl_abcd_nopk" to publication
-DETAIL:  Row filter column "a" is not part of the REPLICA IDENTITY
+-- 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:  Not all row filter columns are not part of the REPLICA IDENTITY
+DROP PUBLICATION testpub6;
 -- ok - "c" is part of REPLICA IDENTITY INDEX
-SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (c > 99);
 DROP PUBLICATION testpub6;
 RESET client_min_messages;
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index c7160bd457..8f6fd65057 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -237,15 +237,21 @@ DROP PUBLICATION testpub6;
 -- ok - "b" is a PK col
 SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (b > 99);
-RESET client_min_messages;
 DROP PUBLICATION testpub6;
--- fail - "c" is not part of the PK
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
--- fail - "d" is not part of the PK
+-- fail - "c" is not part of the PK
+UPDATE rf_tbl_abcd_pk set a = 1;
+DROP PUBLICATION testpub6;
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (d > 99);
+-- fail - "d" is not part of the PK
+UPDATE rf_tbl_abcd_pk set a = 1;
+DROP PUBLICATION testpub6;
 -- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
--- fail - "a" is not part of REPLICA IDENTITY
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not part of REPLICA IDENTITY
+UPDATE rf_tbl_abcd_nopk set a = 1;
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
 
 -- Case 2. REPLICA IDENTITY FULL
 ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
@@ -261,15 +267,22 @@ CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
 DROP PUBLICATION testpub6;
 RESET client_min_messages;
 
+SET client_min_messages = 'ERROR';
 -- Case 3. REPLICA IDENTITY NOTHING
 ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
 ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
--- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
--- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk set a = 1;
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
--- fail - "a" is not in REPLICA IDENTITY NOTHING
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk set a = 1;
+DROP PUBLICATION testpub6;
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_nopk set a = 1;
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
 
 -- Case 4. REPLICA IDENTITY INDEX
 ALTER TABLE rf_tbl_abcd_pk ALTER COLUMN c SET NOT NULL;
@@ -278,17 +291,19 @@ ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY USING INDEX idx_abcd_pk_c;
 ALTER TABLE rf_tbl_abcd_nopk ALTER COLUMN c SET NOT NULL;
 CREATE UNIQUE INDEX idx_abcd_nopk_c ON rf_tbl_abcd_nopk(c);
 ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY USING INDEX idx_abcd_nopk_c;
--- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR 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;
+DROP PUBLICATION testpub6;
 -- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
-SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
 DROP PUBLICATION testpub6;
-RESET client_min_messages;
--- fail - "a" is not in REPLICA IDENTITY INDEX
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+update rf_tbl_abcd_nopk set a = 1;
+DROP PUBLICATION testpub6;
 -- ok - "c" is part of REPLICA IDENTITY INDEX
-SET client_min_messages = 'ERROR';
 CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (c > 99);
 DROP PUBLICATION testpub6;
 RESET client_min_messages;
-- 
2.18.4

