From ba2f9e5a933cfb707ca4415537e38128ed0020ca Mon Sep 17 00:00:00 2001
From: Nisha Moond <nisha.moond412@gmail.com>
Date: Wed, 21 Aug 2024 15:14:02 +0530
Subject: [PATCH v9 3/3] 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        |  60 +++++++++--
 src/backend/replication/logical/worker.c      |  74 ++++++++++---
 src/include/executor/executor.h               |   3 +-
 .../subscription/t/032_subscribe_use_index.pl |   2 +-
 .../subscription/t/034_conflict_resolver.pl   | 100 ++++++++++++++++++
 5 files changed, 217 insertions(+), 22 deletions(-)

diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index c13cfcb77c..56ee179059 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;
@@ -589,6 +589,7 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 		List	   *recheckIndexes = NIL;
 		List	   *conflictindexes;
 		bool		conflict = false;
+		ItemPointerData invalidItemPtr;
 
 		/* Compute stored generated columns */
 		if (rel->rd_att->constr &&
@@ -620,7 +621,7 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 				 * error here; otherwise return to caller for resolutions.
 				 */
 				if (FindConflictTuple(resultRelInfo, estate, uniqueidx,
-									  slot, &(*conflictslot)))
+									  slot, &(*conflictslot), &invalidItemPtr))
 				{
 					RepOriginId origin;
 					TimestampTz committs;
@@ -700,7 +701,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;
@@ -739,11 +741,55 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 		if (rel->rd_rel->relispartition)
 			ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
+		conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
+
+		/*
+		 * If caller has passed non null conflictslot, check all the unique
+		 * indexes for potential conflicts. If the configured resolver is in
+		 * favour of apply, give the conflicted tuple information in
+		 * conflictslot.
+		 */
+		if (conflictslot)
+		{
+			foreach_oid(uniqueidx, conflictindexes)
+			{
+				/*
+				 * Return if any conflict is found other than one with 'ERROR'
+				 * resolver configured. In case of 'ERROR' resolver, emit
+				 * error here; otherwise return to caller for resolutions.
+				 */
+				if (FindConflictTuple(resultRelInfo, estate, uniqueidx,
+									  slot, &(*conflictslot), &searchslot->tts_tid))
+				{
+					RepOriginId origin;
+					TimestampTz committs;
+					TransactionId xmin;
+					ConflictResolver resolver;
+					bool		apply_remote = false;
+
+					GetTupleTransactionInfo(*conflictslot, &xmin, &origin, &committs);
+					resolver = GetConflictResolver(rel, CT_UPDATE_EXISTS,
+												   &apply_remote, NULL, subid);
+
+					ReportApplyConflict(estate, resultRelInfo, CT_UPDATE_EXISTS,
+										resolver, NULL, *conflictslot, slot,
+										uniqueidx, xmin, origin, committs, apply_remote);
+
+					/* Nothing to apply, free the resources */
+					if (!apply_remote)
+					{
+						ExecDropSingleTupleTableSlot(*conflictslot);
+						*conflictslot = NULL;
+					}
+
+					return;
+				}
+			}
+		}
+
 		simple_table_tuple_update(rel, tid, slot, estate->es_snapshot,
 								  &update_indexes);
 
-		conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
-
 		if (resultRelInfo->ri_NumIndices > 0 && (update_indexes != TU_None))
 			recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
 												   slot, estate, true,
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index be68445565..a940d019d7 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,
@@ -2516,8 +2517,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);
@@ -2669,7 +2671,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);
 
@@ -2694,14 +2697,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;
@@ -2709,10 +2711,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.
@@ -2750,6 +2753,8 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 								 * update. */
 		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);
@@ -2762,7 +2767,28 @@ 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);
+
+			/* UPDATE 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 row */
+				apply_handle_update_internal(edata, relinfo,
+											 remoteslot, newtup,
+											 relmapentry->localindexoid,
+											 true, conflictslot);
+
+				ExecDropSingleTupleTableSlot(conflictslot);
+			}
+
 		}
 	}
 	else
@@ -3253,6 +3279,8 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 						 * detected for the found tuple and the resolver is in
 						 * favour of applying the update.
 						 */
+						TupleTableSlot *conflictslot = NULL;
+
 						ExecOpenIndices(partrelinfo, true);
 						InitConflictIndexes(partrelinfo);
 
@@ -3260,7 +3288,27 @@ 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);
+
+						/* UPDATE 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(relinfo, estate, &epqstate, localslot);
+
+							/* update the conflicting row */
+							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/032_subscribe_use_index.pl b/src/test/subscription/t/032_subscribe_use_index.pl
index cc999e33c3..b7f30e32cd 100644
--- a/src/test/subscription/t/032_subscribe_use_index.pl
+++ b/src/test/subscription/t/032_subscribe_use_index.pl
@@ -455,7 +455,7 @@ $node_publisher->safe_psql('postgres',
 # wait until the index is used on the subscriber
 $node_publisher->wait_for_catchup($appname);
 $node_subscriber->poll_query_until('postgres',
-	q{select (idx_scan = 1) from pg_stat_all_indexes where indexrelname = 'test_replica_id_full_idxy';}
+	q{select (idx_scan = 2) from pg_stat_all_indexes where indexrelname = 'test_replica_id_full_idxy';}
   )
   or die
   "Timed out while waiting for check subscriber tap_sub_rep_full updates one row via index";
diff --git a/src/test/subscription/t/034_conflict_resolver.pl b/src/test/subscription/t/034_conflict_resolver.pl
index 6f2ba4c7ba..0c4d808407 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'
 ###################################
-- 
2.34.1

