From 4bab5951866751a7e834628e5256f651d09dd92e Mon Sep 17 00:00:00 2001
From: Nisha Moond <nisha.moond412@gmail.com>
Date: Thu, 19 Sep 2024 18:06:41 +0530
Subject: [PATCH v14 3/5] 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        |  44 +++-
 src/backend/replication/logical/worker.c      |  84 ++++++--
 src/include/executor/executor.h               |   3 +-
 .../subscription/t/034_conflict_resolver.pl   | 201 ++++++++++++++++++
 4 files changed, 310 insertions(+), 22 deletions(-)

diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 4059707a27..f944df761c 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -486,7 +486,7 @@ retry:
 static bool
 FindConflictTuple(ResultRelInfo *resultRelInfo, EState *estate,
 				  Oid conflictindex, TupleTableSlot *slot,
-				  TupleTableSlot **conflictslot)
+				  TupleTableSlot **conflictslot, ItemPointer tupleid)
 {
 	Relation	rel = resultRelInfo->ri_RelationDesc;
 	ItemPointerData conflictTid;
@@ -497,7 +497,7 @@ FindConflictTuple(ResultRelInfo *resultRelInfo, EState *estate,
 
 retry:
 	if (ExecCheckIndexConstraints(resultRelInfo, slot, estate,
-								  &conflictTid, &slot->tts_tid,
+								  &conflictTid, tupleid,
 								  list_make1_oid(conflictindex)))
 	{
 		if (*conflictslot)
@@ -543,7 +543,7 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,
 
 		if (list_member_oid(recheckIndexes, uniqueidx) &&
 			FindConflictTuple(resultRelInfo, estate, uniqueidx, remoteslot,
-							  &conflictslot))
+							  &conflictslot, &remoteslot->tts_tid))
 		{
 			RepOriginId origin;
 			TimestampTz committs;
@@ -566,7 +566,7 @@ static bool
 has_conflicting_tuple(EState *estate, ResultRelInfo *resultRelInfo,
 					  TupleTableSlot **conflictslot, ConflictType type,
 					  ConflictResolver resolver, TupleTableSlot *slot,
-					  Oid subid, bool apply_remote)
+					  Oid subid, bool apply_remote, ItemPointer tupleid)
 {
 	List	   *conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
 
@@ -579,7 +579,7 @@ has_conflicting_tuple(EState *estate, ResultRelInfo *resultRelInfo,
 		 * otherwise return to caller for resolutions.
 		 */
 		if (FindConflictTuple(resultRelInfo, estate, uniqueidx, slot,
-							  &(*conflictslot)))
+							  &(*conflictslot), tupleid))
 		{
 			RepOriginId origin;
 			TimestampTz committs;
@@ -636,6 +636,7 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 		bool		conflict = false;
 		ConflictResolver resolver;
 		bool		apply_remote = false;
+		ItemPointerData invalidItemPtr;
 
 		/* Compute stored generated columns */
 		if (rel->rd_att->constr &&
@@ -656,11 +657,16 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 		resolver = GetConflictResolver(rel, CT_INSERT_EXISTS, &apply_remote,
 									   NULL, subid);
 
-		/* Check for conflict and return to caller for resolution if found */
+		/*
+		 * Check for conflict and return to caller for resolution if found.
+		 * Only perform this check if the 'insert_exists' resolver is not set
+		 * to 'ERROR'; if the resolver is 'ERROR', the upcoming
+		 * CheckAndReportConflict() call will handle it.
+		 */
 		if (resolver != CR_ERROR &&
 			has_conflicting_tuple(estate, resultRelInfo, &(*conflictslot),
 								  CT_INSERT_EXISTS, resolver, slot, subid,
-								  apply_remote))
+								  apply_remote, &invalidItemPtr))
 			return;
 
 		/* OK, store the tuple and create index entries for it */
@@ -717,7 +723,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;
@@ -743,6 +750,8 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 		TU_UpdateIndexes update_indexes;
 		List	   *conflictindexes;
 		bool		conflict = false;
+		ConflictResolver resolver;
+		bool		apply_remote = false;
 
 		/* Compute stored generated columns */
 		if (rel->rd_att->constr &&
@@ -756,6 +765,25 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 		if (rel->rd_rel->relispartition)
 			ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
+		/*
+		 * Get the configured resolver and determine if remote changes should
+		 * be applied.
+		 */
+		resolver = GetConflictResolver(rel, CT_UPDATE_EXISTS, &apply_remote,
+									   NULL, subid);
+
+		/*
+		 * Check for conflict and return to caller for resolution if found.
+		 * Only perform this check if the 'update_exists' resolver is not set
+		 * to 'ERROR'; if the resolver is 'ERROR', the upcoming
+		 * CheckAndReportConflict() call will handle it.
+		 */
+		if (resolver != CR_ERROR &&
+			has_conflicting_tuple(estate, resultRelInfo, &(*conflictslot),
+								  CT_UPDATE_EXISTS, resolver, slot, subid,
+								  apply_remote, tid))
+			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 fb9a99f9a8..bfb9c37071 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,
@@ -2517,8 +2518,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);
@@ -2670,7 +2672,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);
 
@@ -2695,14 +2698,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;
@@ -2710,10 +2712,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.
@@ -2754,6 +2757,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);
@@ -2766,7 +2771,32 @@ 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 (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
@@ -3265,6 +3295,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);
 
@@ -3272,7 +3304,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 f3909759f8..291d5dde36 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 0091443224..9a9c75b835 100755
--- a/src/test/subscription/t/034_conflict_resolver.pl
+++ b/src/test/subscription/t/034_conflict_resolver.pl
@@ -152,6 +152,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/,
+	$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/,
+	$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/,
+	$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'
 ###################################
@@ -870,4 +970,105 @@ $node_subscriber->wait_for_log(
 	qr/ERROR:  conflict detected on relation \"public.conf_tab_part_1\": conflict=update_missing/,
 	$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/,
+	$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/,
+	$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/,
+	$log_offset);
+
 done_testing();
-- 
2.34.1

