From 5e4e738f0e50624917bfe1efebb9c51b58c11e01 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <itsajin@gmail.com>
Date: Fri, 18 Oct 2024 03:20:47 -0400
Subject: [PATCH v17 3/4] Conflict resolution for update_exists conflict type.

Supports following resolutions:
- 'apply_remote': Deletes the conflicting local tuple and inserts the remote tuple.
- 'keep_local': Keeps the local tuple and ignores the remote one.
- 'error': Triggers an error.
---
 src/backend/executor/execReplication.c           |  56 +++++--
 src/backend/replication/logical/worker.c         | 104 ++++++++++--
 src/include/executor/executor.h                  |   3 +-
 src/test/subscription/t/034_conflict_resolver.pl | 201 +++++++++++++++++++++++
 4 files changed, 339 insertions(+), 25 deletions(-)

diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 27d6e98..cb6021b 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -486,7 +486,8 @@ retry:
 static bool
 FindConflictTuple(ResultRelInfo *resultRelInfo, EState *estate,
 				  Oid conflictindex, TupleTableSlot *slot,
-				  TupleTableSlot **conflictslot)
+				  TupleTableSlot **conflictslot, ItemPointer tupleid,
+				  LockTupleMode lockmode)
 {
 	Relation	rel = resultRelInfo->ri_RelationDesc;
 	ItemPointerData conflictTid;
@@ -497,7 +498,7 @@ FindConflictTuple(ResultRelInfo *resultRelInfo, EState *estate,
 
 retry:
 	if (ExecCheckIndexConstraints(resultRelInfo, slot, estate,
-								  &conflictTid, &slot->tts_tid,
+								  &conflictTid, tupleid,
 								  list_make1_oid(conflictindex)))
 	{
 		if (*conflictslot)
@@ -514,7 +515,7 @@ retry:
 	res = table_tuple_lock(rel, &conflictTid, GetLatestSnapshot(),
 						   *conflictslot,
 						   GetCurrentCommandId(false),
-						   LockTupleShare,
+						   lockmode,
 						   LockWaitBlock,
 						   0 /* don't follow updates */ ,
 						   &tmfd);
@@ -543,7 +544,8 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,
 
 		if (list_member_oid(recheckIndexes, uniqueidx) &&
 			FindConflictTuple(resultRelInfo, estate, uniqueidx, remoteslot,
-							  &conflictslot))
+							  &conflictslot, &remoteslot->tts_tid,
+							  LockTupleShare))
 		{
 			RepOriginId origin;
 			TimestampTz committs;
@@ -564,17 +566,21 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,
  * tuple information in conflictslot.
  */
 static bool
-has_conflicting_tuple(Oid subid, ConflictType type,
+has_conflicting_tuple(Oid subid, ConflictType type, ItemPointer tupleid,
 					  ResultRelInfo *resultRelInfo, EState *estate,
 					  TupleTableSlot *slot, TupleTableSlot **conflictslot)
 {
 	ConflictResolver resolver;
+	LockTupleMode lockmode;
 	bool		apply_remote = false;
 	Relation	rel = resultRelInfo->ri_RelationDesc;
 	List	   *conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
 
-	/* ASSERT if called for any conflict type other than insert_exists */
-	Assert(type == CT_INSERT_EXISTS);
+	/*
+	 * ASSERT if called for any conflict type other than insert_exists or
+	 * update_exists
+	 */
+	Assert(type == CT_INSERT_EXISTS || type == CT_UPDATE_EXISTS);
 
 	/*
 	 * Get the configured resolver and determine if remote changes should be
@@ -583,6 +589,17 @@ has_conflicting_tuple(Oid subid, ConflictType type,
 	resolver = GetConflictResolver(subid, type, rel, NULL, &apply_remote);
 
 	/*
+	 * Determine the lock mode for the conflicting tuple, if any. Take an
+	 * exclusive lock if the resolution favors modifying the conflicting
+	 * tuple; otherwise, take a shared lock.
+	 */
+
+	if (apply_remote)
+		lockmode = LockTupleExclusive;
+	else
+		lockmode = LockTupleShare;
+
+	/*
 	 * Proceed to find conflict if the resolver is set to a non-default value;
 	 * if the resolver is 'ERROR' (default), the caller will handle it.
 	 */
@@ -594,7 +611,7 @@ has_conflicting_tuple(Oid subid, ConflictType type,
 	{
 		/* Return to caller for resolutions if any conflict is found */
 		if (FindConflictTuple(resultRelInfo, estate, uniqueidx, slot,
-							  &(*conflictslot)))
+							  &(*conflictslot), tupleid, lockmode))
 		{
 			RepOriginId origin;
 			TimestampTz committs;
@@ -651,6 +668,7 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 		List	   *recheckIndexes = NIL;
 		List	   *conflictindexes;
 		bool		conflict = false;
+		ItemPointerData invalidItemPtr;
 
 		/* Compute stored generated columns */
 		if (rel->rd_att->constr &&
@@ -673,8 +691,9 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 		 * inserted tuple if a conflict is detected after insertion with a
 		 * non-default resolution set.
 		 */
-		if (has_conflicting_tuple(subid, CT_INSERT_EXISTS, resultRelInfo,
-								  estate, slot, &(*conflictslot)))
+		if (has_conflicting_tuple(subid, CT_INSERT_EXISTS, &invalidItemPtr,
+								  resultRelInfo, estate, slot,
+								  &(*conflictslot)))
 			return;
 
 		/* OK, store the tuple and create index entries for it */
@@ -732,7 +751,8 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 void
 ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 						 EState *estate, EPQState *epqstate,
-						 TupleTableSlot *searchslot, TupleTableSlot *slot)
+						 TupleTableSlot *searchslot, TupleTableSlot *slot,
+						 TupleTableSlot **conflictslot, Oid subid)
 {
 	bool		skip_tuple = false;
 	Relation	rel = resultRelInfo->ri_RelationDesc;
@@ -775,6 +795,20 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 		if (rel->rd_rel->relispartition)
 			ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
+		/*
+		 * Check for conflict and return to caller for resolution, if found.
+		 *
+		 * XXX In case there are no conflicts, a non-default 'update_exists'
+		 * resolver adds overhead by performing an extra scan here. However,
+		 * this approach avoids the extra work needed to rollback/delete the
+		 * updated tuple if a conflict is detected after update with a
+		 * non-default resolution set.
+		 */
+		if (has_conflicting_tuple(subid, CT_UPDATE_EXISTS, tid,
+								  resultRelInfo, estate, slot,
+								  &(*conflictslot)))
+			return;
+
 		simple_table_tuple_update(rel, tid, slot, estate->es_snapshot,
 								  &update_indexes);
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index abda71a..3579bd0 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -389,7 +389,8 @@ static void apply_handle_update_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
 										 TupleTableSlot *remoteslot,
 										 LogicalRepTupleData *newtup,
-										 Oid localindexoid);
+										 Oid localindexoid, bool found,
+										 TupleTableSlot *localslot);
 static void apply_handle_delete_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
 										 TupleTableSlot *remoteslot,
@@ -2518,8 +2519,9 @@ apply_handle_insert_internal(ApplyExecutionData *edata,
 
 		/* Do the update */
 		TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE);
-		ExecSimpleRelationUpdate(relinfo, estate, &epqstate, conflictslot,
-								 remoteslot);
+		apply_handle_update_internal(edata, relinfo, remoteslot, newtup,
+									 rel_entry->localindexoid, true,
+									 conflictslot);
 
 		EvalPlanQualEnd(&epqstate);
 		ExecDropSingleTupleTableSlot(conflictslot);
@@ -2671,7 +2673,8 @@ apply_handle_update(StringInfo s)
 								   remoteslot, &newtup, CMD_UPDATE);
 	else
 		apply_handle_update_internal(edata, edata->targetRelInfo,
-									 remoteslot, &newtup, rel->localindexoid);
+									 remoteslot, &newtup, rel->localindexoid,
+									 false, NULL);
 
 	finish_edata(edata);
 
@@ -2696,14 +2699,13 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 							 ResultRelInfo *relinfo,
 							 TupleTableSlot *remoteslot,
 							 LogicalRepTupleData *newtup,
-							 Oid localindexoid)
+							 Oid localindexoid, bool found,
+							 TupleTableSlot *localslot)
 {
 	EState	   *estate = edata->estate;
 	LogicalRepRelMapEntry *relmapentry = edata->targetRel;
 	Relation	localrel = relinfo->ri_RelationDesc;
 	EPQState	epqstate;
-	TupleTableSlot *localslot;
-	bool		found;
 	MemoryContext oldctx;
 	bool		apply_remote = true;
 	ConflictResolver resolver;
@@ -2711,10 +2713,11 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 	EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
 	ExecOpenIndices(relinfo, true);
 
-	found = FindReplTupleInLocalRel(edata, localrel,
-									&relmapentry->remoterel,
-									localindexoid,
-									remoteslot, &localslot);
+	if (!found)
+		found = FindReplTupleInLocalRel(edata, localrel,
+										&relmapentry->remoterel,
+										localindexoid,
+										remoteslot, &localslot);
 
 	/*
 	 * Tuple found.
@@ -2756,6 +2759,8 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		 */
 		if (apply_remote)
 		{
+			TupleTableSlot *conflictslot = NULL;
+
 			/* Process and store remote tuple in the slot */
 			oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
 			slot_modify_data(remoteslot, localslot, relmapentry, newtup);
@@ -2768,7 +2773,52 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 			/* Do the actual update. */
 			TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE);
 			ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot,
-									 remoteslot);
+									 remoteslot, &conflictslot,
+									 MySubscription->oid);
+
+			/*
+			 * If an update_exists conflict is detected, delete the found
+			 * tuple and apply the remote update to the conflicting tuple.
+			 *
+			 * If the local table contains multiple unique constraint columns
+			 * and the conflict resolution strategy favors applying the remote
+			 * changes, then a remote INSERT or UPDATE could trigger multiple
+			 * update_exists conflicts. Each conflict will be detected and
+			 * resolved in sequence (recursively), possibly resulting in the
+			 * deletion of multiple local rows.
+			 *
+			 * Consider the scenario: A table where all columns have unique
+			 * indexes, and the first column is the replica identity. The
+			 * publisher has (1, 1, 1) while the subscriber has three rows:
+			 * (1, 1, 1), (2, 2, 2), and (3, 3, 3). The publisher updates (1,
+			 * 1, 1) to (1, 2, 3).
+			 *
+			 * TThe current logic works as follows: We find the row (1, 1, 1)
+			 * and try to update it to (1, 2, 3), but find a conflict with the
+			 * tuple (2, 2, 2). We delete (1, 1, 1), and then try to update
+			 * (2, 2, 2) to (1, 2, 3), but encounter another conflict with (3,
+			 * 3, 3). We then delete (2, 2, 2) and try to update (3, 3, 3) to
+			 * (1, 2, 3).
+			 */
+			if (conflictslot)
+			{
+				/* Process and store remote tuple in the slot */
+				oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+				slot_modify_data(remoteslot, conflictslot, relmapentry, newtup);
+				MemoryContextSwitchTo(oldctx);
+
+				/* Delete the found row */
+				ExecSimpleRelationDelete(relinfo, estate, &epqstate, localslot);
+
+				/* Update the conflicting tuple with remote update */
+				apply_handle_update_internal(edata, relinfo,
+											 remoteslot, newtup,
+											 relmapentry->localindexoid,
+											 true, conflictslot);
+
+				ExecDropSingleTupleTableSlot(conflictslot);
+			}
+
 		}
 	}
 	else
@@ -3269,6 +3319,8 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 						 * conflict is detected for the found tuple and the
 						 * resolver is in favour of applying the update.
 						 */
+						TupleTableSlot *conflictslot = NULL;
+
 						ExecOpenIndices(partrelinfo, true);
 						InitConflictIndexes(partrelinfo);
 
@@ -3276,7 +3328,33 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 						TargetPrivilegesCheck(partrelinfo->ri_RelationDesc,
 											  ACL_UPDATE);
 						ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate,
-												 localslot, remoteslot_part);
+												 localslot, remoteslot_part,
+												 &conflictslot, MySubscription->oid);
+
+						/*
+						 * If an update_exists conflict is detected, delete
+						 * the found tuple and apply the remote update to the
+						 * conflicting tuple.
+						 */
+						if (conflictslot)
+						{
+							/* Process and store remote tuple in the slot */
+							oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+							slot_modify_data(remoteslot_part, conflictslot, part_entry, newtup);
+							MemoryContextSwitchTo(oldctx);
+
+							/* Delete the found row */
+							ExecSimpleRelationDelete(partrelinfo, estate, &epqstate, localslot);
+
+							/* Update the conflicting tuple with remote update */
+							apply_handle_update_internal(edata, partrelinfo,
+														 remoteslot_part, newtup,
+														 part_entry->localindexoid,
+														 true, conflictslot);
+
+							ExecDropSingleTupleTableSlot(conflictslot);
+						}
+
 						ExecCloseIndices(partrelinfo);
 					}
 					else if (apply_remote)
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index f390975..291d5dd 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -673,7 +673,8 @@ extern void ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 									 TupleTableSlot **conflictsloty, Oid subid);
 extern void ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 									 EState *estate, EPQState *epqstate,
-									 TupleTableSlot *searchslot, TupleTableSlot *slot);
+									 TupleTableSlot *searchslot, TupleTableSlot *slot,
+									 TupleTableSlot **conflictslot, Oid subid);
 extern void ExecSimpleRelationDelete(ResultRelInfo *resultRelInfo,
 									 EState *estate, EPQState *epqstate,
 									 TupleTableSlot *searchslot);
diff --git a/src/test/subscription/t/034_conflict_resolver.pl b/src/test/subscription/t/034_conflict_resolver.pl
index 86e3ad1..f7c8abc 100644
--- a/src/test/subscription/t/034_conflict_resolver.pl
+++ b/src/test/subscription/t/034_conflict_resolver.pl
@@ -156,6 +156,106 @@ $node_subscriber->wait_for_log(
 # Truncate table on subscriber to get rid of the error
 $node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
 
+# Truncate the table on the publisher
+$node_publisher->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# Drop the subscriber to remove error
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub;");
+
+# Create the subscription
+$node_subscriber->safe_psql(
+	'postgres',
+	"CREATE SUBSCRIPTION tap_sub
+	 CONNECTION '$publisher_connstr application_name=$appname'
+	 PUBLICATION tap_pub");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+
+############################################
+# Test 'apply_remote' for 'update_exists'
+############################################
+# Change CONFLICT RESOLVER of update_exists to apply_remote
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_exists = 'apply_remote');"
+);
+
+# Insert data in the subscriber
+$node_subscriber->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (1,'fromsub');
+	 INSERT INTO conf_tab(a, data) VALUES (2,'fromsub');
+	 INSERT INTO conf_tab(a, data) VALUES (3,'fromsub');");
+
+# Insert data in the publisher
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (4,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (5,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (6,'frompub');");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Update on publisher which already exists on subscriber
+$node_publisher->safe_psql('postgres', "UPDATE conf_tab SET a=1 WHERE a=4;");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_exists, resolution=apply_remote/,
+	$log_offset);
+
+# Confirm that remote update is applied.
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data FROM conf_tab WHERE (a=1);");
+
+is($result, 'frompub', "remote data is kept");
+
+########################################
+# Test 'keep_local' for 'update_exists'
+########################################
+
+# Change CONFLICT RESOLVER of update_exists to keep_local
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_exists = 'keep_local');"
+);
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Update on publisher which already exists on subscriber
+$node_publisher->safe_psql('postgres', "UPDATE conf_tab SET a=2 WHERE a=5;");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_exists, resolution=keep_local/,
+	$log_offset);
+
+# Confirm that remote insert is ignored and the local row is kept
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data FROM conf_tab WHERE (a=2);");
+
+is($result, 'fromsub', "data from local is kept");
+
+###################################
+# Test 'error' for 'update_exists'
+###################################
+
+# Change CONFLICT RESOLVER of update_exists to error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_exists = 'error');"
+);
+
+# Update on publisher which already exists on subscriber
+$node_publisher->safe_psql('postgres', "UPDATE conf_tab SET a=3 WHERE a=6;");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Confirm that this causes an error on the subscriber
+$node_subscriber->wait_for_log(
+	qr/ERROR:  conflict detected on relation \"public.conf_tab\": conflict=update_exists, resolution=error/,
+	$log_offset);
+
+# Truncate table on subscriber to get rid of the error
+$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
+
 ###################################
 # Test 'skip' for 'delete_missing'
 ###################################
@@ -874,4 +974,105 @@ $node_subscriber->wait_for_log(
 	qr/ERROR:  conflict detected on relation \"public.conf_tab_part_2\": conflict=update_missing, resolution=error/,
 	$log_offset);
 
+# Drop the subscriber to remove error
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub_part;");
+
+# Truncate the table on the publisher
+$node_publisher->safe_psql('postgres', "TRUNCATE conf_tab_part;");
+
+# Truncate the table on the subscriber
+$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab_part;");
+
+$node_subscriber->safe_psql(
+	'postgres',
+	"CREATE SUBSCRIPTION sub_part
+	CONNECTION '$publisher_connstr application_name=$appname'
+	PUBLICATION pub_part;");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+############################################
+# Test 'apply_remote' for 'update_exists'
+############################################
+# Change CONFLICT RESOLVER of update_exists to apply_remote
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION sub_part CONFLICT RESOLVER (update_exists = 'apply_remote');"
+);
+
+# Insert data in the subscriber
+$node_subscriber->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab_part VALUES (1,1,'fromsub');
+	 INSERT INTO conf_tab_part VALUES (2,1,'fromsub');
+	 INSERT INTO conf_tab_part VALUES (3,1,'fromsub');");
+
+# Insert data in the publisher
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab_part VALUES (4,1,'frompub');
+	 INSERT INTO conf_tab_part VALUES (5,1,'frompub');
+	 INSERT INTO conf_tab_part VALUES (6,1,'frompub');");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Update on publisher which already exists on subscriber
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab_part SET a=1 WHERE a=4;");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab_part_1\": conflict=update_exists, resolution=apply_remote/,
+	$log_offset);
+
+# Confirm that remote update is applied.
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data FROM conf_tab_part WHERE (a=1);");
+
+is($result, 'frompub', "update from remote on partition is kept");
+
+########################################
+# Test 'keep_local' for 'update_exists'
+########################################
+
+# Change CONFLICT RESOLVER of update_exists to keep_local
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION sub_part CONFLICT RESOLVER (update_exists = 'keep_local');"
+);
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Update on publisher which already exists on subscriber
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab_part SET a=2 WHERE a=5;");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab_part_1\": conflict=update_exists, resolution=keep_local/,
+	$log_offset);
+
+# Confirm that remote insert is ignored and the local row is kept
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data FROM conf_tab_part WHERE (a=2);");
+
+is($result, 'fromsub', "update from remote on partition is skipped");
+
+###################################
+# Test 'error' for 'update_exists'
+###################################
+
+# Change CONFLICT RESOLVER of update_exists to error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION sub_part CONFLICT RESOLVER (update_exists = 'error');"
+);
+
+# Update on publisher which already exists on subscriber
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab_part SET a=3 WHERE a=6;");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Confirm that this causes an error on the subscriber
+$node_subscriber->wait_for_log(
+	qr/ERROR:  conflict detected on relation \"public.conf_tab_part_1\": conflict=update_exists, resolution=error/,
+	$log_offset);
+
 done_testing();
-- 
1.8.3.1

