From 16b986f8f8544f090d46023f4b7a904b777e7e36 Mon Sep 17 00:00:00 2001
From: Nisha Moond <nisha.moond412@gmail.com>
Date: Wed, 21 Aug 2024 08:51:52 +0530
Subject: [PATCH v9 2/3] Conflict resolvers for insert, update, and delete

This patch introduces support for handling conflicts with the following resolutions:
- For INSERT conflicts:
  - insert_exists: apply_remote, keep_local, error
- For UPDATE conflicts:
  - update_differ: apply_remote, keep_local, error
  - update_missing: apply_or_skip, apply_or_error, skip, error
- For DELETE conflicts:
  - delete_missing: skip, error
  - delete_differ: apply_remote, keep_local, error
---
 src/backend/executor/execReplication.c        |  57 +-
 src/backend/replication/logical/conflict.c    | 210 +++++--
 src/backend/replication/logical/worker.c      | 357 ++++++++---
 src/include/executor/executor.h               |   5 +-
 src/include/replication/conflict.h            |  14 +-
 src/test/subscription/meson.build             |   1 +
 .../subscription/t/034_conflict_resolver.pl   | 581 ++++++++++++++++++
 7 files changed, 1080 insertions(+), 145 deletions(-)
 create mode 100755 src/test/subscription/t/034_conflict_resolver.pl

diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 1086cbc962..c13cfcb77c 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -550,9 +550,9 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,
 			TransactionId xmin;
 
 			GetTupleTransactionInfo(conflictslot, &xmin, &origin, &committs);
-			ReportApplyConflict(estate, resultRelInfo, ERROR, type,
-								searchslot, conflictslot, remoteslot,
-								uniqueidx, xmin, origin, committs);
+			ReportApplyConflict(estate, resultRelInfo, type,
+								CR_ERROR, searchslot, conflictslot, remoteslot,
+								uniqueidx, xmin, origin, committs, false);
 		}
 	}
 }
@@ -565,7 +565,8 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,
  */
 void
 ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
-						 EState *estate, TupleTableSlot *slot)
+						 EState *estate, TupleTableSlot *slot,
+						 TupleTableSlot **conflictslot, Oid subid)
 {
 	bool		skip_tuple = false;
 	Relation	rel = resultRelInfo->ri_RelationDesc;
@@ -601,11 +602,55 @@ ExecSimpleRelationInsert(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)))
+				{
+					RepOriginId origin;
+					TimestampTz committs;
+					TransactionId xmin;
+					ConflictResolver resolver;
+					bool		apply_remote = false;
+
+					GetTupleTransactionInfo(*conflictslot, &xmin, &origin, &committs);
+					resolver = GetConflictResolver(rel, CT_INSERT_EXISTS,
+												   &apply_remote, NULL, subid);
+
+					ReportApplyConflict(estate, resultRelInfo, CT_INSERT_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;
+				}
+			}
+		}
+
 		/* OK, store the tuple and create index entries for it */
 		simple_table_tuple_insert(resultRelInfo->ri_RelationDesc, slot);
 
-		conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
-
 		if (resultRelInfo->ri_NumIndices > 0)
 			recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
 												   slot, estate, false,
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index ad79f816dd..df92c616aa 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -31,6 +31,7 @@
 #include "replication/conflict.h"
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
+#include "replication/logicalproto.h"
 #include "replication/logicalrelation.h"
 #include "storage/lmgr.h"
 #include "utils/lsyscache.h"
@@ -98,12 +99,14 @@ static int	errcode_apply_conflict(ConflictType type);
 static int	errdetail_apply_conflict(EState *estate,
 									 ResultRelInfo *relinfo,
 									 ConflictType type,
+									 ConflictResolver resolver,
 									 TupleTableSlot *searchslot,
 									 TupleTableSlot *localslot,
 									 TupleTableSlot *remoteslot,
 									 Oid indexoid, TransactionId localxmin,
 									 RepOriginId localorigin,
-									 TimestampTz localts);
+									 TimestampTz localts,
+									 bool apply_remote);
 static char *build_tuple_value_details(EState *estate, ResultRelInfo *relinfo,
 									   ConflictType type,
 									   TupleTableSlot *searchslot,
@@ -146,8 +149,8 @@ GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin,
 }
 
 /*
- * This function is used to report a conflict while applying replication
- * changes.
+ * This function is used to report a conflict and resolution applied while
+ * applying replication changes.
  *
  * 'searchslot' should contain the tuple used to search the local tuple to be
  * updated or deleted.
@@ -166,26 +169,36 @@ GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin,
  * that we can fetch and display the conflicting key value.
  */
 void
-ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
-					ConflictType type, TupleTableSlot *searchslot,
-					TupleTableSlot *localslot, TupleTableSlot *remoteslot,
-					Oid indexoid, TransactionId localxmin,
-					RepOriginId localorigin, TimestampTz localts)
+ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
+					ConflictType type, ConflictResolver resolver,
+					TupleTableSlot *searchslot, TupleTableSlot *localslot,
+					TupleTableSlot *remoteslot, Oid indexoid,
+					TransactionId localxmin, RepOriginId localorigin,
+					TimestampTz localts, bool apply_remote)
 {
 	Relation	localrel = relinfo->ri_RelationDesc;
+	int			elevel;
+
+	if (resolver == CR_ERROR ||
+		(resolver == CR_APPLY_OR_ERROR && !apply_remote))
+		elevel = ERROR;
+	else
+		elevel = LOG;
+
 
 	Assert(!OidIsValid(indexoid) ||
 		   CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
 
 	ereport(elevel,
 			errcode_apply_conflict(type),
-			errmsg("conflict detected on relation \"%s.%s\": conflict=%s",
+			errmsg("conflict detected on relation \"%s.%s\": conflict=%s, Resolution=%s.",
 				   get_namespace_name(RelationGetNamespace(localrel)),
 				   RelationGetRelationName(localrel),
-				   ConflictTypeNames[type]),
-			errdetail_apply_conflict(estate, relinfo, type, searchslot,
+				   ConflictTypeNames[type],
+				   ConflictResolverNames[resolver]),
+			errdetail_apply_conflict(estate, relinfo, type, resolver, searchslot,
 									 localslot, remoteslot, indexoid,
-									 localxmin, localorigin, localts));
+									 localxmin, localorigin, localts, apply_remote));
 }
 
 /*
@@ -253,17 +266,25 @@ errcode_apply_conflict(ConflictType type)
  */
 static int
 errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo,
-						 ConflictType type, TupleTableSlot *searchslot,
-						 TupleTableSlot *localslot, TupleTableSlot *remoteslot,
-						 Oid indexoid, TransactionId localxmin,
-						 RepOriginId localorigin, TimestampTz localts)
+						 ConflictType type, ConflictResolver resolver,
+						 TupleTableSlot *searchslot, TupleTableSlot *localslot,
+						 TupleTableSlot *remoteslot, Oid indexoid,
+						 TransactionId localxmin, RepOriginId localorigin,
+						 TimestampTz localts, bool apply_remote)
 {
 	StringInfoData err_detail;
 	char	   *val_desc;
 	char	   *origin_name;
+	char	   *applymsg;
 
 	initStringInfo(&err_detail);
 
+	if (apply_remote)
+		applymsg = "applying the remote changes.";
+	else
+		applymsg = "ignoring the remote changes.";
+
+
 	/* First, construct a detailed message describing the type of conflict */
 	switch (type)
 	{
@@ -274,13 +295,13 @@ errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo,
 			if (localts)
 			{
 				if (localorigin == InvalidRepOriginId)
-					appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified locally in transaction %u at %s."),
+					appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified locally in transaction %u at %s, %s"),
 									 get_rel_name(indexoid),
-									 localxmin, timestamptz_to_str(localts));
+									 localxmin, timestamptz_to_str(localts), applymsg);
 				else if (replorigin_by_oid(localorigin, true, &origin_name))
-					appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by origin \"%s\" in transaction %u at %s."),
+					appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by origin \"%s\" in transaction %u at %s, %s"),
 									 get_rel_name(indexoid), origin_name,
-									 localxmin, timestamptz_to_str(localts));
+									 localxmin, timestamptz_to_str(localts), applymsg);
 
 				/*
 				 * The origin that modified this row has been removed. This
@@ -290,47 +311,55 @@ errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo,
 				 * manually dropped by the user.
 				 */
 				else
-					appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by a non-existent origin in transaction %u at %s."),
+					appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by a non-existent origin in transaction %u at %s, %s"),
 									 get_rel_name(indexoid),
-									 localxmin, timestamptz_to_str(localts));
+									 localxmin, timestamptz_to_str(localts), applymsg);
 			}
 			else
-				appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified in transaction %u."),
-								 get_rel_name(indexoid), localxmin);
+				appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified in transaction %u, %s"),
+								 get_rel_name(indexoid), localxmin, applymsg);
 
 			break;
 
 		case CT_UPDATE_DIFFER:
 			if (localorigin == InvalidRepOriginId)
-				appendStringInfo(&err_detail, _("Updating the row that was modified locally in transaction %u at %s."),
-								 localxmin, timestamptz_to_str(localts));
+				appendStringInfo(&err_detail, _("Updating the row that was modified locally in transaction %u at %s, %s"),
+								 localxmin, timestamptz_to_str(localts), applymsg);
 			else if (replorigin_by_oid(localorigin, true, &origin_name))
-				appendStringInfo(&err_detail, _("Updating the row that was modified by a different origin \"%s\" in transaction %u at %s."),
-								 origin_name, localxmin, timestamptz_to_str(localts));
+				appendStringInfo(&err_detail, _("Updating the row that was modified by a different origin \"%s\" in transaction %u at %s, %s"),
+								 origin_name, localxmin, timestamptz_to_str(localts), applymsg);
 
 			/* The origin that modified this row has been removed. */
 			else
-				appendStringInfo(&err_detail, _("Updating the row that was modified by a non-existent origin in transaction %u at %s."),
-								 localxmin, timestamptz_to_str(localts));
+				appendStringInfo(&err_detail, _("Updating the row that was modified by a non-existent origin in transaction %u at %s, %s"),
+								 localxmin, timestamptz_to_str(localts), applymsg);
 
 			break;
 
 		case CT_UPDATE_MISSING:
-			appendStringInfo(&err_detail, _("Could not find the row to be updated."));
+			if (resolver == CR_APPLY_OR_SKIP && !apply_remote)
+				appendStringInfo(&err_detail, _("Could not find the row to be updated, UPDATE can not be converted to INSERT, hence SKIP the update."));
+			else if (resolver == CR_APPLY_OR_ERROR && !apply_remote)
+				appendStringInfo(&err_detail, _("Could not find the row to be updated, UPDATE can not be converted to INSERT, hence ERROR out."));
+			else if (apply_remote)
+				appendStringInfo(&err_detail, _("Could not find the row to be updated, Convert UPDATE to INSERT and %s"),
+								 applymsg);
+			else
+				appendStringInfo(&err_detail, _("Could not find the row to be updated, %s"), applymsg);
 			break;
 
 		case CT_DELETE_DIFFER:
 			if (localorigin == InvalidRepOriginId)
-				appendStringInfo(&err_detail, _("Deleting the row that was modified locally in transaction %u at %s."),
-								 localxmin, timestamptz_to_str(localts));
+				appendStringInfo(&err_detail, _("Deleting the row that was modified locally in transaction %u at %s, %s"),
+								 localxmin, timestamptz_to_str(localts), applymsg);
 			else if (replorigin_by_oid(localorigin, true, &origin_name))
-				appendStringInfo(&err_detail, _("Deleting the row that was modified by a different origin \"%s\" in transaction %u at %s."),
-								 origin_name, localxmin, timestamptz_to_str(localts));
+				appendStringInfo(&err_detail, _("Deleting the row that was modified by a different origin \"%s\" in transaction %u at %s, %s"),
+								 origin_name, localxmin, timestamptz_to_str(localts), applymsg);
 
 			/* The origin that modified this row has been removed. */
 			else
-				appendStringInfo(&err_detail, _("Deleting the row that was modified by a non-existent origin in transaction %u at %s."),
-								 localxmin, timestamptz_to_str(localts));
+				appendStringInfo(&err_detail, _("Deleting the row that was modified by a non-existent origin in transaction %u at %s, %s"),
+								 localxmin, timestamptz_to_str(localts), applymsg);
 
 			break;
 
@@ -634,6 +663,69 @@ validate_conflict_type_and_resolver(const char *conflict_type,
 
 }
 
+/*
+ * Get the conflict resolver configured at subscription level for
+ * for the given conflict type.
+ */
+static ConflictResolver
+get_conflict_resolver_internal(ConflictType type, Oid subid)
+{
+
+	ConflictResolver resolver;
+	HeapTuple	tuple;
+	Datum		datum;
+	char	   *conflict_res;
+
+	/*
+	 * XXX: Currently, we fetch the conflict resolver from cache for each
+	 * conflict detection. If needed, we can keep the info in global variable
+	 * and fetch from cache only once after cache invalidation.
+	 */
+	tuple = SearchSysCache2(SUBSCRIPTIONCONFLICTSUBOID,
+							ObjectIdGetDatum(subid),
+							CStringGetTextDatum(ConflictTypeNames[type]));
+
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for conflict type %u", type);
+
+	datum = SysCacheGetAttrNotNull(SUBSCRIPTIONCONFLICTSUBOID,
+								   tuple, Anum_pg_subscription_conflict_confrres);
+
+	conflict_res = TextDatumGetCString(datum);
+
+	for (resolver = CR_MIN; resolver <= CR_MAX; resolver++)
+	{
+		if (strcmp(ConflictResolverNames[resolver], conflict_res) == 0)
+			break;
+	}
+
+	ReleaseSysCache(tuple);
+	return resolver;
+}
+
+/*
+ * Check if a full tuple can be created from the new tuple.
+ * Return true if yes, false otherwise.
+ */
+static bool
+can_create_full_tuple(Relation localrel,
+					  LogicalRepTupleData *newtup)
+{
+	int			i;
+	int			local_att = RelationGetNumberOfAttributes(localrel);
+
+	if (newtup->ncols != local_att)
+		return false;
+
+	for (i = 0; i < newtup->ncols; i++)
+	{
+		if (newtup->colstatus[i] == LOGICALREP_COLUMN_UNCHANGED)
+			return false;
+	}
+
+	return true;
+}
+
 /*
  * Extract the conflict type and conflict resolvers from the
  * ALTER SUBSCRIPTION command and return a list of ConflictTypeResolver nodes.
@@ -794,3 +886,47 @@ RemoveSubscriptionConflictBySubid(Oid subid)
 	table_endscan(scan);
 	table_close(rel, RowExclusiveLock);
 }
+
+/*
+ * Find the resolver of the conflict type set under the given subscription.
+ *
+ * Set 'apply_remote' to true if remote tuple should be applied,
+ * false otherwise.
+ */
+ConflictResolver
+GetConflictResolver(Relation localrel, ConflictType type, bool *apply_remote,
+					LogicalRepTupleData *newtup, Oid subid)
+{
+	ConflictResolver resolver;
+
+	resolver = get_conflict_resolver_internal(type, subid);
+
+	switch (resolver)
+	{
+		case CR_APPLY_REMOTE:
+			*apply_remote = true;
+			break;
+		case CR_APPLY_OR_SKIP:
+			if (can_create_full_tuple(localrel, newtup))
+				*apply_remote = true;
+			else
+				*apply_remote = false;
+			break;
+		case CR_APPLY_OR_ERROR:
+			if (can_create_full_tuple(localrel, newtup))
+				*apply_remote = true;
+			else
+				*apply_remote = false;
+			break;
+		case CR_KEEP_LOCAL:
+		case CR_SKIP:
+		case CR_ERROR:
+			*apply_remote = false;
+			break;
+		default:
+			elog(ERROR, "Conflict %s is detected! Unrecogonized conflict resolution method",
+				 ConflictTypeNames[type]);
+	}
+
+	return resolver;
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 38c2895307..be68445565 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -382,7 +382,9 @@ static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
 static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
 static void apply_handle_insert_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
-										 TupleTableSlot *remoteslot);
+										 TupleTableSlot *remoteslot,
+										 LogicalRepTupleData *newtup,
+										 LogicalRepRelMapEntry *rel_entry);
 static void apply_handle_update_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
 										 TupleTableSlot *remoteslot,
@@ -2451,10 +2453,10 @@ apply_handle_insert(StringInfo s)
 	/* For a partitioned table, insert the tuple into a partition. */
 	if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
 		apply_handle_tuple_routing(edata,
-								   remoteslot, NULL, CMD_INSERT);
+								   remoteslot, &newtup, CMD_INSERT);
 	else
 		apply_handle_insert_internal(edata, edata->targetRelInfo,
-									 remoteslot);
+									 remoteslot, &newtup, rel);
 
 	finish_edata(edata);
 
@@ -2477,9 +2479,13 @@ apply_handle_insert(StringInfo s)
 static void
 apply_handle_insert_internal(ApplyExecutionData *edata,
 							 ResultRelInfo *relinfo,
-							 TupleTableSlot *remoteslot)
+							 TupleTableSlot *remoteslot,
+							 LogicalRepTupleData *newtup,
+							 LogicalRepRelMapEntry *rel_entry)
 {
 	EState	   *estate = edata->estate;
+	MemoryContext oldctx;
+	TupleTableSlot *conflictslot = NULL;
 
 	/* We must open indexes here. */
 	ExecOpenIndices(relinfo, true);
@@ -2487,7 +2493,35 @@ apply_handle_insert_internal(ApplyExecutionData *edata,
 
 	/* Do the insert. */
 	TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_INSERT);
-	ExecSimpleRelationInsert(relinfo, estate, remoteslot);
+
+	ExecSimpleRelationInsert(relinfo, estate, remoteslot, &conflictslot,
+							 MySubscription->oid);
+
+	/*
+	 * If conflict is detected and resolver is in favor of applying the remote
+	 * changes, then, apply remote tuple by converting INSERT to UPDATE.
+	 */
+	if (conflictslot)
+	{
+		EPQState	epqstate;
+
+		EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
+
+		/* Process and store remote tuple in the slot */
+		oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+		slot_modify_data(remoteslot, conflictslot, rel_entry, newtup);
+		MemoryContextSwitchTo(oldctx);
+
+		EvalPlanQualSetSlot(&epqstate, remoteslot);
+
+		/* Do the update */
+		TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE);
+		ExecSimpleRelationUpdate(relinfo, estate, &epqstate, conflictslot,
+								 remoteslot);
+
+		EvalPlanQualEnd(&epqstate);
+		ExecDropSingleTupleTableSlot(conflictslot);
+	}
 
 	/* Cleanup. */
 	ExecCloseIndices(relinfo);
@@ -2669,6 +2703,8 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 	TupleTableSlot *localslot;
 	bool		found;
 	MemoryContext oldctx;
+	bool		apply_remote = true;
+	ConflictResolver resolver;
 
 	EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
 	ExecOpenIndices(relinfo, true);
@@ -2690,36 +2726,44 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		TimestampTz localts;
 
 		/*
-		 * Report the conflict if the tuple was modified by a different
-		 * origin.
+		 * Report the conflict  and configured resolver if the tuple was
+		 * modified by a different origin.
 		 */
 		if (GetTupleTransactionInfo(localslot, &localxmin, &localorigin, &localts) &&
 			localorigin != replorigin_session_origin)
 		{
 			TupleTableSlot *newslot;
 
+			resolver = GetConflictResolver(localrel, CT_UPDATE_DIFFER,
+										   &apply_remote, NULL,
+										   MySubscription->oid);
 			/* Store the new tuple for conflict reporting */
 			newslot = table_slot_create(localrel, &estate->es_tupleTable);
 			slot_store_data(newslot, relmapentry, newtup);
 
-			ReportApplyConflict(estate, relinfo, LOG, CT_UPDATE_DIFFER,
-								remoteslot, localslot, newslot,
-								InvalidOid, localxmin, localorigin, localts);
-		}
-
-		/* Process and store remote tuple in the slot */
-		oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-		slot_modify_data(remoteslot, localslot, relmapentry, newtup);
-		MemoryContextSwitchTo(oldctx);
+			ReportApplyConflict(estate, relinfo, CT_UPDATE_DIFFER,
+								resolver, remoteslot, localslot, newslot,
+								InvalidOid, localxmin, localorigin, localts,
+								apply_remote);
+		}						/* Apply the change if configured resolver is
+								 * in favor of that, else ignore the remote
+								 * update. */
+		if (apply_remote)
+		{
+			/* Process and store remote tuple in the slot */
+			oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+			slot_modify_data(remoteslot, localslot, relmapentry, newtup);
+			MemoryContextSwitchTo(oldctx);
 
-		EvalPlanQualSetSlot(&epqstate, remoteslot);
+			EvalPlanQualSetSlot(&epqstate, remoteslot);
 
-		InitConflictIndexes(relinfo);
+			InitConflictIndexes(relinfo);
 
-		/* Do the actual update. */
-		TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE);
-		ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot,
-								 remoteslot);
+			/* Do the actual update. */
+			TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE);
+			ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot,
+									 remoteslot);
+		}
 	}
 	else
 	{
@@ -2729,13 +2773,27 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		slot_store_data(newslot, relmapentry, newtup);
 
 		/*
-		 * The tuple to be updated could not be found.  Do nothing except for
-		 * emitting a log message.
+		 * The tuple to be updated could not be found. Report the conflict. If
+		 * the configured resolver is in favor of applying the change, convert
+		 * UPDATE to INSERT and apply the change.
 		 */
-		ReportApplyConflict(estate, relinfo, LOG, CT_UPDATE_MISSING,
-							remoteslot, NULL, newslot,
+		resolver = GetConflictResolver(localrel, CT_UPDATE_MISSING,
+									   &apply_remote, newtup,
+									   MySubscription->oid);
+		ReportApplyConflict(estate, relinfo, CT_UPDATE_MISSING,
+							resolver, remoteslot, NULL, newslot,
 							InvalidOid, InvalidTransactionId,
-							InvalidRepOriginId, 0);
+							InvalidRepOriginId, 0, apply_remote);
+
+		/* Process and store remote tuple in the slot */
+		oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+		slot_store_data(remoteslot, relmapentry, newtup);
+		slot_fill_defaults(relmapentry, estate, remoteslot);
+		MemoryContextSwitchTo(oldctx);
+
+		if (apply_remote)
+			apply_handle_insert_internal(edata, relinfo, remoteslot,
+										 newtup, relmapentry);
 	}
 
 	/* Cleanup. */
@@ -2848,6 +2906,8 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 	EPQState	epqstate;
 	TupleTableSlot *localslot;
 	bool		found;
+	bool		apply_remote = true;
+	ConflictResolver resolver;
 
 	EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
 	ExecOpenIndices(relinfo, false);
@@ -2868,26 +2928,41 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 		 */
 		if (GetTupleTransactionInfo(localslot, &localxmin, &localorigin, &localts) &&
 			localorigin != replorigin_session_origin)
-			ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_DIFFER,
-								remoteslot, localslot, NULL,
-								InvalidOid, localxmin, localorigin, localts);
+		{
+			resolver = GetConflictResolver(localrel, CT_DELETE_DIFFER,
+										   &apply_remote, NULL,
+										   MySubscription->oid);
+			ReportApplyConflict(estate, relinfo, CT_DELETE_DIFFER,
+								resolver, remoteslot, localslot, NULL,
+								InvalidOid, localxmin, localorigin, localts,
+								apply_remote);
+		}
 
-		EvalPlanQualSetSlot(&epqstate, localslot);
+		if (apply_remote)
+		{
+			EvalPlanQualSetSlot(&epqstate, localslot);
 
-		/* Do the actual delete. */
-		TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_DELETE);
-		ExecSimpleRelationDelete(relinfo, estate, &epqstate, localslot);
+			/* Do the actual delete. */
+			TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_DELETE);
+			ExecSimpleRelationDelete(relinfo, estate, &epqstate, localslot);
+		}
 	}
 	else
 	{
 		/*
-		 * The tuple to be deleted could not be found.  Do nothing except for
-		 * emitting a log message.
+		 * The tuple to be deleted could not be found. Based on resolver
+		 * configured, either skip and log a message or emit an error.
 		 */
-		ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_MISSING,
-							remoteslot, NULL, NULL,
-							InvalidOid, InvalidTransactionId,
-							InvalidRepOriginId, 0);
+		resolver = GetConflictResolver(localrel, CT_DELETE_MISSING,
+									   &apply_remote, NULL,
+									   MySubscription->oid);
+
+		/* Resolver is set to skip, thus report the conflict and skip */
+		if (!apply_remote)
+			ReportApplyConflict(estate, relinfo, CT_DELETE_MISSING,
+								resolver, remoteslot, NULL, NULL,
+								InvalidOid, InvalidTransactionId,
+								InvalidRepOriginId, 0, apply_remote);
 	}
 
 	/* Cleanup. */
@@ -3020,19 +3095,21 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 	}
 	MemoryContextSwitchTo(oldctx);
 
-	/* Check if we can do the update or delete on the leaf partition. */
-	if (operation == CMD_UPDATE || operation == CMD_DELETE)
+	if (operation == CMD_INSERT || operation == CMD_UPDATE ||
+		operation == CMD_DELETE)
 	{
-		part_entry = logicalrep_partition_open(relmapentry, partrel,
-											   attrmap);
-		check_relation_updatable(part_entry);
+		part_entry = logicalrep_partition_open(relmapentry, partrel, attrmap);
+
+		/* Check if we can do the update or delete on the leaf partition */
+		if (operation != CMD_INSERT)
+			check_relation_updatable(part_entry);
 	}
 
 	switch (operation)
 	{
 		case CMD_INSERT:
 			apply_handle_insert_internal(edata, partrelinfo,
-										 remoteslot_part);
+										 remoteslot_part, newtup, part_entry);
 			break;
 
 		case CMD_DELETE:
@@ -3058,6 +3135,9 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 				RepOriginId localorigin;
 				TransactionId localxmin;
 				TimestampTz localts;
+				LogicalRepRelMapEntry *part_entry_new = NULL;
+				ConflictResolver resolver;
+				bool		apply_remote = true;
 
 				/* Get the matching local tuple from the partition. */
 				found = FindReplTupleInLocalRel(edata, partrel,
@@ -3070,47 +3150,84 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 
 					/* Store the new tuple for conflict reporting */
 					slot_store_data(newslot, part_entry, newtup);
+					resolver = GetConflictResolver(partrel, CT_UPDATE_MISSING,
+												   &apply_remote, newtup,
+												   MySubscription->oid);
 
 					/*
-					 * The tuple to be updated could not be found.  Do nothing
-					 * except for emitting a log message.
+					 * The tuple to be updated could not be found. Report the
+					 * conflict and resolver. And take action based on the
+					 * configured resolver.
 					 */
 					ReportApplyConflict(estate, partrelinfo,
-										LOG, CT_UPDATE_MISSING,
+										CT_UPDATE_MISSING, resolver,
 										remoteslot_part, NULL, newslot,
 										InvalidOid, InvalidTransactionId,
-										InvalidRepOriginId, 0);
+										InvalidRepOriginId, 0, apply_remote);
 
-					return;
+					if (apply_remote)
+					{
+						/*
+						 * Resolver is in favour of applying the remote
+						 * changes. Prepare the slot for the INSERT.
+						 */
+						oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+						slot_store_data(remoteslot_part, part_entry, newtup);
+						slot_fill_defaults(part_entry, estate, remoteslot_part);
+						MemoryContextSwitchTo(oldctx);
+					}
 				}
-
-				/*
-				 * Report the conflict if the tuple was modified by a
-				 * different origin.
-				 */
-				if (GetTupleTransactionInfo(localslot, &localxmin, &localorigin, &localts) &&
-					localorigin != replorigin_session_origin)
+				else
 				{
-					TupleTableSlot *newslot;
+					/*
+					 * The tuple to be updated is found. Report and resolve
+					 * the conflict if the tuple was modified by a different
+					 * origin.
+					 */
+					if (GetTupleTransactionInfo(localslot, &localxmin, &localorigin, &localts) &&
+						localorigin != replorigin_session_origin)
+					{
+						TupleTableSlot *newslot;
 
-					/* Store the new tuple for conflict reporting */
-					newslot = table_slot_create(partrel, &estate->es_tupleTable);
-					slot_store_data(newslot, part_entry, newtup);
+						/* Store the new tuple for conflict reporting */
+						newslot = table_slot_create(partrel, &estate->es_tupleTable);
+						slot_store_data(newslot, part_entry, newtup);
 
-					ReportApplyConflict(estate, partrelinfo, LOG, CT_UPDATE_DIFFER,
-										remoteslot_part, localslot, newslot,
-										InvalidOid, localxmin, localorigin,
-										localts);
-				}
+						resolver = GetConflictResolver(partrel, CT_UPDATE_DIFFER,
+													   &apply_remote, NULL,
+													   MySubscription->oid);
 
-				/*
-				 * Apply the update to the local tuple, putting the result in
-				 * remoteslot_part.
-				 */
-				oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
-				slot_modify_data(remoteslot_part, localslot, part_entry,
-								 newtup);
-				MemoryContextSwitchTo(oldctx);
+						ReportApplyConflict(estate, partrelinfo, CT_UPDATE_DIFFER,
+											resolver, remoteslot_part, localslot, newslot,
+											InvalidOid, localxmin, localorigin,
+											localts, apply_remote);
+					}
+					if (apply_remote)
+					{
+						/*
+						 * We can reach here in two cases:
+						 *
+						 * 1. If we found a tuple and no conflict is detected
+						 *
+						 * 2. If we found a tuple and resolver is in favor of
+						 * applying the change when conflict is detected
+						 *
+						 * Putting the result in remoteslot_part.
+						 */
+						oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+						slot_modify_data(remoteslot_part, localslot, part_entry,
+										 newtup);
+						MemoryContextSwitchTo(oldctx);
+					}
+					else
+
+						/*
+						 * apply_remote can be toggled if resolver for
+						 * update_differ is set to skip. Ignore remote update.
+						 */
+						return;
+
+				}
 
 				EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
 
@@ -3122,23 +3239,52 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 					ExecPartitionCheck(partrelinfo, remoteslot_part, estate,
 									   false))
 				{
-					/*
-					 * Yes, so simply UPDATE the partition.  We don't call
-					 * apply_handle_update_internal() here, which would
-					 * normally do the following work, to avoid repeating some
-					 * work already done above to find the local tuple in the
-					 * partition.
-					 */
-					ExecOpenIndices(partrelinfo, true);
-					InitConflictIndexes(partrelinfo);
-
-					EvalPlanQualSetSlot(&epqstate, remoteslot_part);
-					TargetPrivilegesCheck(partrelinfo->ri_RelationDesc,
-										  ACL_UPDATE);
-					ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate,
-											 localslot, remoteslot_part);
+					if (found && apply_remote)
+					{
+						/*
+						 * Yes, so simply UPDATE the partition.  We don't call
+						 * apply_handle_update_internal() here, which would
+						 * normally do the following work, to avoid repeating
+						 * some work already done above to find the local
+						 * tuple in the partition.
+						 *
+						 * Do the update in cases - 1. found a tuple and no
+						 * conflict is detected  2. update_differ conflict is
+						 * detected for the found tuple and the resolver is in
+						 * favour of applying the update.
+						 */
+						ExecOpenIndices(partrelinfo, true);
+						InitConflictIndexes(partrelinfo);
+
+						EvalPlanQualSetSlot(&epqstate, remoteslot_part);
+						TargetPrivilegesCheck(partrelinfo->ri_RelationDesc,
+											  ACL_UPDATE);
+						ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate,
+												 localslot, remoteslot_part);
+						ExecCloseIndices(partrelinfo);
+					}
+					else if (apply_remote)
+					{
+						/*
+						 * Tuple is not found but update_missing resolver is
+						 * in favour of applying the change as INSERT.
+						 */
+						apply_handle_insert_internal(edata, partrelinfo,
+													 remoteslot_part, newtup,
+													 part_entry);
+					}
 				}
-				else
+
+				/*
+				 * Updated tuple doesn't satisfy the current partition's
+				 * constraint.
+				 *
+				 * Proceed by always applying the update (as 'apply_remote' is
+				 * by default true). The 'apply_remote' can be OFF as well if
+				 * the resolver for update_missing conflict conveys to skip
+				 * the update.
+				 */
+				else if (apply_remote)
 				{
 					/* Move the tuple into the new partition. */
 
@@ -3179,12 +3325,20 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 											 get_namespace_name(RelationGetNamespace(partrel_new)),
 											 RelationGetRelationName(partrel_new));
 
-					ExecOpenIndices(partrelinfo, false);
-
-					/* DELETE old tuple found in the old partition. */
-					EvalPlanQualSetSlot(&epqstate, localslot);
-					TargetPrivilegesCheck(partrelinfo->ri_RelationDesc, ACL_DELETE);
-					ExecSimpleRelationDelete(partrelinfo, estate, &epqstate, localslot);
+					/*
+					 * If tuple is found, delete it from old partition. We can
+					 * reach this flow even for the case when the 'found' flag
+					 * is false for 'update_missing' conflict and resolver is
+					 * in favor of inserting the tuple.
+					 */
+					if (found)
+					{
+						ExecOpenIndices(partrelinfo, false);
+						EvalPlanQualSetSlot(&epqstate, localslot);
+						TargetPrivilegesCheck(partrelinfo->ri_RelationDesc, ACL_DELETE);
+						ExecSimpleRelationDelete(partrelinfo, estate, &epqstate, localslot);
+						ExecCloseIndices(partrelinfo);
+					}
 
 					/* INSERT new tuple into the new partition. */
 
@@ -3200,22 +3354,27 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 					map = ExecGetRootToChildMap(partrelinfo_new, estate);
 					if (map != NULL)
 					{
+						attrmap = map->attrMap;
 						remoteslot_part = execute_attr_map_slot(map->attrMap,
 																remoteslot,
 																remoteslot_part);
 					}
 					else
 					{
+						attrmap = NULL;
 						remoteslot_part = ExecCopySlot(remoteslot_part,
 													   remoteslot);
 						slot_getallattrs(remoteslot);
 					}
 					MemoryContextSwitchTo(oldctx);
+					part_entry_new = logicalrep_partition_open(part_entry,
+															   partrel_new,
+															   attrmap);
 					apply_handle_insert_internal(edata, partrelinfo_new,
-												 remoteslot_part);
+												 remoteslot_part, newtup,
+												 part_entry_new);
 				}
 
-				ExecCloseIndices(partrelinfo);
 				EvalPlanQualEnd(&epqstate);
 			}
 			break;
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 69c3ebff00..f3909759f8 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -18,6 +18,8 @@
 #include "fmgr.h"
 #include "nodes/lockoptions.h"
 #include "nodes/parsenodes.h"
+#include "replication/logicalproto.h"
+#include "replication/logicalrelation.h"
 #include "utils/memutils.h"
 
 
@@ -667,7 +669,8 @@ extern bool RelationFindReplTupleSeq(Relation rel, LockTupleMode lockmode,
 									 TupleTableSlot *searchslot, TupleTableSlot *outslot);
 
 extern void ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
-									 EState *estate, TupleTableSlot *slot);
+									 EState *estate, TupleTableSlot *slot,
+									 TupleTableSlot **conflictsloty, Oid subid);
 extern void ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 									 EState *estate, EPQState *epqstate,
 									 TupleTableSlot *searchslot, TupleTableSlot *slot);
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index fac3f8335e..0ddbfbe9c2 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -10,6 +10,8 @@
 #define CONFLICT_H
 
 #include "nodes/execnodes.h"
+#include "replication/logicalproto.h"
+#include "replication/logicalrelation.h"
 #include "utils/timestamp.h"
 
 /*
@@ -88,12 +90,14 @@ extern bool GetTupleTransactionInfo(TupleTableSlot *localslot,
 									RepOriginId *localorigin,
 									TimestampTz *localts);
 extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
-								int elevel, ConflictType type,
+								ConflictType type,
+								ConflictResolver resolver,
 								TupleTableSlot *searchslot,
 								TupleTableSlot *localslot,
 								TupleTableSlot *remoteslot,
 								Oid indexoid, TransactionId localxmin,
-								RepOriginId localorigin, TimestampTz localts);
+								RepOriginId localorigin, TimestampTz localts,
+								bool apply_remote);
 extern void InitConflictIndexes(ResultRelInfo *relInfo);
 extern void SetSubConflictResolver(Oid subId, ConflictTypeResolver * resolvers, int max_types);
 extern void RemoveSubscriptionConflictById(Oid confid);
@@ -103,5 +107,11 @@ extern void UpdateSubConflictResolvers(List *conflict_resolvers, Oid subid);
 extern ConflictType validate_conflict_type_and_resolver(const char *conflict_type,
 														const char *conflict_resolver);
 extern void SetDefaultResolvers(ConflictTypeResolver * conflictResolvers);
+extern ConflictResolver GetConflictResolver(Relation localrel,
+											ConflictType type,
+											bool *apply_remote,
+											LogicalRepTupleData *newtup,
+											Oid subid);
+extern bool CanCreateFullTuple(Relation localrel, LogicalRepTupleData *newtup);
 
 #endif
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index c591cd7d61..00ade29b02 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -40,6 +40,7 @@ tests += {
       't/031_column_list.pl',
       't/032_subscribe_use_index.pl',
       't/033_run_as_table_owner.pl',
+      't/034_conflict_resolver.pl',
       't/100_bugs.pl',
     ],
   },
diff --git a/src/test/subscription/t/034_conflict_resolver.pl b/src/test/subscription/t/034_conflict_resolver.pl
new file mode 100755
index 0000000000..6f2ba4c7ba
--- /dev/null
+++ b/src/test/subscription/t/034_conflict_resolver.pl
@@ -0,0 +1,581 @@
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+###############################
+# Setup
+###############################
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+	qq(max_prepared_transactions = 10));
+$node_publisher->start;
+
+# Create subscriber node with track_commit_timestamp_set
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+	qq(track_commit_timestamp = on));
+$node_subscriber->start;
+
+# Create table on publisher
+$node_publisher->safe_psql(
+	'postgres',
+	"CREATE TABLE conf_tab(a int PRIMARY key, data text, comments text);
+	 ALTER TABLE conf_tab ALTER COLUMN comments SET STORAGE EXTERNAL;");
+
+# Create similar table on subscriber
+$node_subscriber->safe_psql(
+	'postgres',
+	"CREATE TABLE conf_tab(a int PRIMARY key, data text, comments text);
+	 ALTER TABLE conf_tab ALTER COLUMN comments SET STORAGE EXTERNAL;");
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub FOR TABLE conf_tab");
+
+# Create the subscription
+my $appname = 'tap_sub';
+$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);
+
+my $result = $node_subscriber->safe_psql('postgres',
+	"SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype"
+);
+is( $result, qq(delete_differ|apply_remote
+delete_missing|skip
+insert_exists|error
+update_differ|apply_remote
+update_exists|error
+update_missing|skip),
+	"confirm that the default conflict resolvers are in place");
+
+############################################
+# Test 'apply_remote' for 'insert_exists'
+############################################
+
+# Change CONFLICT RESOLVER of insert_exists to apply_remote
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (insert_exists = 'apply_remote');"
+);
+
+# Create local data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (1,'fromsub')");
+
+# Create conflicting data on the publisher
+my $log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (1,'frompub')");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=insert_exists/,
+	$log_offset);
+
+# Confirm that remote insert is converted to an update and the remote data is updated.
+$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 'insert_exists'
+########################################
+
+# Change CONFLICT RESOLVER of insert_exists to keep_local
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (insert_exists = 'keep_local');"
+);
+
+# Create local data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (2,'fromsub')");
+
+# Confirm that row is updated
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data FROM conf_tab WHERE (a=2);");
+
+is($result, 'fromsub', "data 2 from local is inserted");
+
+$log_offset = -s $node_subscriber->logfile;
+
+# Create conflicting data on the publisher
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (2,'frompub')");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=insert_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 'insert_exists'
+###################################
+
+# Change CONFLICT RESOLVER of insert_exists to error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (insert_exists = 'error');"
+);
+
+# Create local data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (3,'fromsub')");
+
+# Create conflicting data on the publisher
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (3,'frompub')");
+
+$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=insert_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'
+###################################
+
+# Delete row on publisher that is not present on the subscriber and confirm that it is skipped
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=2);");
+
+# Confirm that the missing row is skipped because 'delete_missing' is set to 'skip'
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=delete_missing/,
+	$log_offset);
+
+####################################
+# Test 'error' for 'delete_missing'
+####################################
+
+# Change CONFLICT RESOLVER of delete_missing to error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (delete_missing = 'error');"
+);
+
+# Capture the log offset before performing the delete on the publisher
+$log_offset = -s $node_subscriber->logfile;
+
+# Perform the delete on the publisher
+$node_publisher->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=1);");
+
+# Confirm that this causes an error on the subscriber
+$node_subscriber->wait_for_log(
+	qr/ERROR:  conflict detected on relation \"public.conf_tab\": conflict=delete_missing./,
+	$log_offset);
+
+# Drop the subscriber to remove error
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub;");
+
+# Truncate table on subscriber
+$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# Truncate the table on the publisher
+$node_publisher->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# 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 'delete_differ'
+#########################################
+
+# Insert data in the publisher
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (1,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (2,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (3,'frompub');");
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub' WHERE (a=1);");
+
+# Create a conflicting delete on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=1);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=delete_differ/,
+	$log_offset);
+
+# Confirm that the remote delete the local updated row
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=1);");
+
+is($result, '', "delete from remote is applied");
+
+#########################################
+# Test 'keep_local' for 'delete_differ'
+#########################################
+
+# Change CONFLICT RESOLVER of delete_differ to keep_local
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (delete_differ = 'keep_local');"
+);
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub' WHERE (a=2);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=2);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=delete_differ/,
+	$log_offset);
+
+# Confirm that the local data is untouched by the remote update
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=2);");
+
+is($result, 'fromsub', "update from local is kept");
+
+#########################################
+# Test 'error' for 'delete_differ'
+#########################################
+
+# Change CONFLICT RESOLVER of delete_differ to error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (delete_differ = 'error');"
+);
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub' WHERE (a=3);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=3);");
+
+$node_subscriber->wait_for_log(
+	qr/ERROR:  conflict detected on relation \"public.conf_tab\": conflict=delete_differ./,
+	$log_offset);
+
+# Drop the subscriber to remove error
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub;");
+
+# Truncate the table on the publisher
+$node_publisher->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# Truncate the table on the subscriber
+$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# 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_differ'
+#########################################
+
+# Insert data in the publisher
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (1,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (2,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (3,'frompub');");
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub' WHERE (a=1);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=1);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_differ/,
+	$log_offset);
+
+# Confirm that the remote update overrides the local update
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=1);");
+
+is($result, 'frompubnew', "update from remote is kept");
+
+#########################################
+# Test 'keep_local' for 'update_differ'
+#########################################
+
+# Change CONFLICT RESOLVER of update_differ to keep_local
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_differ = 'keep_local');"
+);
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub' WHERE (a=2);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=2);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_differ/,
+	$log_offset);
+
+# Confirm that the local data is untouched by the remote update
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=2);");
+
+is($result, 'fromsub', "update from local is kept");
+
+#########################################
+# Test 'error' for 'update_differ'
+#########################################
+
+# Change CONFLICT RESOLVER of update_differ to error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_differ = 'error');"
+);
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub' WHERE (a=3);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=3);");
+
+$node_subscriber->wait_for_log(
+	qr/ERROR:  conflict detected on relation \"public.conf_tab\": conflict=update_differ./,
+	$log_offset);
+
+# Drop the subscriber to remove error
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub;");
+
+# Truncate the table on the publisher
+$node_publisher->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# Truncate the table on the subscriber
+$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# 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);
+
+# Insert data in the publisher
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (1,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (2,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (3,'frompub');");
+
+###########################################
+# Test 'apply_or_skip' for 'update_missing'
+###########################################
+
+#Change CONFLICT RESOLVER of update_missing to apply_or_skip
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_missing = 'apply_or_skip');"
+);
+
+#test the apply part
+
+# Delete the row on the subscriber
+$node_subscriber->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=3);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=3);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_missing/,
+	$log_offset);
+
+# Confirm that the remote update is converted to an insert and new row applied
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=3);");
+
+is($result, 'frompubnew', "update from remote is converted to insert");
+
+#test the skip part
+
+# Create new row on publisher with toast data
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab (a, data, comments) VALUES(4,'frompub',repeat('abcdefghij', 200));"
+);
+
+# Delete the row on the subscriber
+$node_subscriber->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=4);");
+
+# Update the row on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=4);");
+
+$node_subscriber->wait_for_log(
+	qr/DETAIL:  Could not find the row to be updated, UPDATE can not be converted to INSERT, hence SKIP the update./,
+	$log_offset);
+
+###########################################
+# Test 'apply_or_error' for 'update_missing'
+###########################################
+
+# test the apply part
+
+#Change CONFLICT RESOLVER of update_missing to apply_or_error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_missing = 'apply_or_error');"
+);
+
+# Create new row on the publisher
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (5,'frompub');");
+
+# Delete the row on the subscriber
+$node_subscriber->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=5);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=5);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_missing/,
+	$log_offset);
+
+# Confirm that the remote update is converted to an insert and new row applied
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=5);");
+
+is($result, 'frompubnew', "update from remote is converted to insert");
+
+#test the error part
+
+# Create new row on publisher with toast data
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab (a, data, comments) VALUES(6,'frompub',repeat('abcdefghij', 200));"
+);
+
+# Delete the row on the subscriber
+$node_subscriber->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=6);");
+
+# Update the row on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=6);");
+
+$node_subscriber->wait_for_log(
+	qr/ERROR:  conflict detected on relation \"public.conf_tab\": conflict=update_missing/,
+	$log_offset);
+
+# Drop the subscriber to remove error
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub;");
+
+# Truncate the table on the publisher
+$node_publisher->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# Truncate the table on the subscriber
+$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
+
+# 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);
+
+# Insert data in the publisher
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (1,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (2,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (3,'frompub');");
+
+###################################
+# Test 'skip' for 'update_missing'
+###################################
+
+#Change CONFLICT RESOLVER of update_missing to skip
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_missing = 'skip');"
+);
+
+# Delete the row on the subscriber
+$node_subscriber->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=2);");
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnew' WHERE (a=2);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_missing/,
+	$log_offset);
+
+# Confirm that the update does not change anything on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=2);");
+
+is($result, '', "update from remote is skipped on the subscriber");
+
+###################################
+# Test 'error' for 'update_missing'
+###################################
+
+#Change CONFLICT RESOLVER of update_missing to error
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_missing = 'error');"
+);
+
+# Create a conflicting update on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'frompubnewer' WHERE (a=2);");
+
+$node_subscriber->wait_for_log(
+	qr/ERROR:  conflict detected on relation \"public.conf_tab\": conflict=update_missing/,
+	$log_offset);
+
+done_testing();
-- 
2.34.1

