Fix replica identity check for MERGE.

When executing a MERGE, check that the target relation supports all
actions mentioned in the MERGE command. Specifically, check that it
has a REPLICA IDENTITY if it publishes updates or deletes and the
MERGE command contains update or delete actions. Failing to do this
can silently break replication.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Reviewed-by: Ashutosh Bapat <ashutosh.bapat.oss@gmail.com>
Reviewed-by: Dean Rasheed <dean.a.rasheed@gmail.com>
Tested-by: Chao Li <li.evan.chao@gmail.com>
Discussion: https://postgr.es/m/OS3PR01MB57180C87E43A679A730482DF94B62@OS3PR01MB5718.jpnprd01.prod.outlook.com
Backpatch-through: 15

src/backend/executor/execMain.c

@@ -1013,12 +1013,16 @@ InitPlan(QueryDesc *queryDesc, int eflags)
  * For INSERT ON CONFLICT, the result relation is required to support the
  * onConflictAction, regardless of whether a conflict actually occurs.
  *
+ * For MERGE, mergeActions is the list of actions that may be performed.  The
+ * result relation is required to support every action, regardless of whether
+ * or not they are all executed.
+ *
  * Note: when changing this function, you probably also need to look at
  * CheckValidRowMarkRel.
  */
 void
 CheckValidResultRelNew(ResultRelInfo *resultRelInfo, CmdType operation,
-					   OnConflictAction onConflictAction)
+					   OnConflictAction onConflictAction, List *mergeActions)
 {
 	Relation	resultRel = resultRelInfo->ri_RelationDesc;
 	TriggerDesc *trigDesc = resultRel->trigdesc;
@@ -1032,7 +1036,24 @@ CheckValidResultRelNew(ResultRelInfo *resultRelInfo, CmdType operation,
 	{
 		case RELKIND_RELATION:
 		case RELKIND_PARTITIONED_TABLE:
-			CheckCmdReplicaIdentity(resultRel, operation);
+
+			/*
+			 * For MERGE, check that the target relation supports each action.
+			 * For other operations, just check the operation itself.
+			 */
+			if (operation == CMD_MERGE)
+			{
+				ListCell   *lc;
+
+				foreach(lc, mergeActions)
+				{
+					MergeAction *action = (MergeAction *) lfirst(lc);
+
+					CheckCmdReplicaIdentity(resultRel, action->commandType);
+				}
+			}
+			else
+				CheckCmdReplicaIdentity(resultRel, operation);
 
 			/*
 			 * For INSERT ON CONFLICT DO UPDATE, additionally check that the
@@ -1165,7 +1186,7 @@ CheckValidResultRelNew(ResultRelInfo *resultRelInfo, CmdType operation,
 void
 CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation)
 {
-	return CheckValidResultRelNew(resultRelInfo, operation, ONCONFLICT_NONE);
+	return CheckValidResultRelNew(resultRelInfo, operation, ONCONFLICT_NONE, NIL);
 }
 
 /*

src/backend/executor/execPartition.c

@@ -364,7 +364,8 @@ ExecFindPartition(ModifyTableState *mtstate,
 
 					/* Verify this ResultRelInfo allows INSERTs */
 					CheckValidResultRelNew(rri, CMD_INSERT,
-										   node ? node->onConflictAction : ONCONFLICT_NONE);
+										   node ? node->onConflictAction : ONCONFLICT_NONE,
+										   NIL);
 
 					/*
 					 * Initialize information needed to insert this and
@@ -531,7 +532,8 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 	 * required when the operation is CMD_UPDATE.
 	 */
 	CheckValidResultRelNew(leaf_part_rri, CMD_INSERT,
-						   node ? node->onConflictAction : ONCONFLICT_NONE);
+						   node ? node->onConflictAction : ONCONFLICT_NONE,
+						   NIL);
 
 	/*
 	 * Open partition indices.  The user may have asked to check for conflicts

src/backend/executor/nodeModifyTable.c

@@ -4240,6 +4240,10 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
 	foreach(l, node->resultRelations)
 	{
 		Index		resultRelation = lfirst_int(l);
+		List	   *mergeActions = NIL;
+
+		if (node->mergeActionLists)
+			mergeActions = list_nth(node->mergeActionLists, i);
 
 		if (resultRelInfo != mtstate->rootResultRelInfo)
 		{
@@ -4262,7 +4266,7 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
 		 * Verify result relation is a valid target for the current operation
 		 */
 		CheckValidResultRelNew(resultRelInfo, operation,
-							   node->onConflictAction);
+							   node->onConflictAction, mergeActions);
 
 		resultRelInfo++;
 		i++;

src/include/executor/executor.h

@@ -212,7 +212,8 @@ extern bool ExecCheckPermissions(List *rangeTable,
 								 List *rteperminfos, bool ereport_on_violation);
 extern bool ExecCheckOneRelPerms(RTEPermissionInfo *perminfo);
 extern void CheckValidResultRelNew(ResultRelInfo *resultRelInfo, CmdType operation,
-								   OnConflictAction onConflictAction);
+								   OnConflictAction onConflictAction,
+								   List *mergeActions);
 extern void CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation);
 extern void InitResultRelInfo(ResultRelInfo *resultRelInfo,
 							  Relation resultRelationDesc,

src/test/regress/expected/publication.out

@@ -1755,6 +1755,34 @@ INSERT INTO testpub_insert_onconfl_parted VALUES (1, 1) ON CONFLICT DO NOTHING;
 DROP PUBLICATION pub1;
 DROP TABLE testpub_insert_onconfl_no_ri;
 DROP TABLE testpub_insert_onconfl_parted;
+-- Test that the MERGE command correctly checks REPLICA IDENTITY when the
+-- target table is published.
+CREATE TABLE testpub_merge_no_ri (a int, b int);
+CREATE TABLE testpub_merge_pk (a int primary key, b int);
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION pub1 FOR ALL TABLES;
+RESET client_min_messages;
+-- fail - missing REPLICA IDENTITY
+MERGE INTO testpub_merge_no_ri USING testpub_merge_pk s ON s.a >= 1
+ WHEN MATCHED THEN UPDATE SET b = s.b;
+ERROR:  cannot update table "testpub_merge_no_ri" because it does not have a replica identity and publishes updates
+HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.
+-- fail - missing REPLICA IDENTITY
+MERGE INTO testpub_merge_no_ri USING testpub_merge_pk s ON s.a >= 1
+ WHEN MATCHED THEN DELETE;
+ERROR:  cannot delete from table "testpub_merge_no_ri" because it does not have a replica identity and publishes deletes
+HINT:  To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.
+-- ok - insert and do nothing are not restricted
+MERGE INTO testpub_merge_no_ri USING testpub_merge_pk s ON s.a >= 1
+ WHEN MATCHED THEN DO NOTHING
+ WHEN NOT MATCHED THEN INSERT (a, b) VALUES (0, 0);
+-- ok - REPLICA IDENTITY is DEFAULT and table has a PK
+MERGE INTO testpub_merge_pk USING testpub_merge_no_ri s ON s.a >= 1
+ WHEN MATCHED AND s.a > 0 THEN UPDATE SET b = s.b
+ WHEN MATCHED THEN DELETE;
+DROP PUBLICATION pub1;
+DROP TABLE testpub_merge_no_ri;
+DROP TABLE testpub_merge_pk;
 RESET SESSION AUTHORIZATION;
 DROP ROLE regress_publication_user, regress_publication_user2;
 DROP ROLE regress_publication_user_dummy;

src/test/regress/sql/publication.sql

@@ -1123,6 +1123,37 @@ DROP PUBLICATION pub1;
 DROP TABLE testpub_insert_onconfl_no_ri;
 DROP TABLE testpub_insert_onconfl_parted;
 
+-- Test that the MERGE command correctly checks REPLICA IDENTITY when the
+-- target table is published.
+CREATE TABLE testpub_merge_no_ri (a int, b int);
+CREATE TABLE testpub_merge_pk (a int primary key, b int);
+
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION pub1 FOR ALL TABLES;
+RESET client_min_messages;
+
+-- fail - missing REPLICA IDENTITY
+MERGE INTO testpub_merge_no_ri USING testpub_merge_pk s ON s.a >= 1
+ WHEN MATCHED THEN UPDATE SET b = s.b;
+
+-- fail - missing REPLICA IDENTITY
+MERGE INTO testpub_merge_no_ri USING testpub_merge_pk s ON s.a >= 1
+ WHEN MATCHED THEN DELETE;
+
+-- ok - insert and do nothing are not restricted
+MERGE INTO testpub_merge_no_ri USING testpub_merge_pk s ON s.a >= 1
+ WHEN MATCHED THEN DO NOTHING
+ WHEN NOT MATCHED THEN INSERT (a, b) VALUES (0, 0);
+
+-- ok - REPLICA IDENTITY is DEFAULT and table has a PK
+MERGE INTO testpub_merge_pk USING testpub_merge_no_ri s ON s.a >= 1
+ WHEN MATCHED AND s.a > 0 THEN UPDATE SET b = s.b
+ WHEN MATCHED THEN DELETE;
+
+DROP PUBLICATION pub1;
+DROP TABLE testpub_merge_no_ri;
+DROP TABLE testpub_merge_pk;
+
 RESET SESSION AUTHORIZATION;
 DROP ROLE regress_publication_user, regress_publication_user2;
 DROP ROLE regress_publication_user_dummy;