From 8d8a69c1b77d9d0d569c4ac94c7ed8ffc54d4860 Mon Sep 17 00:00:00 2001
From: Nisha Moond <nisha.moond412@gmail.com>
Date: Thu, 10 Oct 2024 15:47:18 +0530
Subject: [PATCH v16 4/4] Implements last_update_wins conflict resolver.

This resolver is applicable for conflict types: insert_exists, update_exists,
update_origin_differs and delete_origin_differs.

For these conflicts, when the resolver is set to last_update_wins,
the timestamps of the remote and local conflicting tuple are compared to
determine whether to apply or ignore the remote changes.
The GUC track_commit_timestamp must be enabled to support this resolver.

Since conflict resolution for two phase commit transactions using
prepare-timestamp can result in data divergence, this patch restricts
enabling both two_phase and the last_update_wins resolver together
for a subscription.

The patch also restrict starting a parallel apply worker if resolver is set
to last_update_wins for any conflict type.
---
 src/backend/commands/subscriptioncmds.c       |  26 +++-
 src/backend/executor/execReplication.c        |  10 +-
 .../replication/logical/applyparallelworker.c |  13 ++
 src/backend/replication/logical/conflict.c    | 134 ++++++++++++++++-
 src/backend/replication/logical/origin.c      |   1 +
 src/backend/replication/logical/worker.c      |  32 ++--
 src/include/replication/conflict.h            |   9 +-
 src/include/replication/origin.h              |   1 +
 .../subscription/t/034_conflict_resolver.pl   | 138 +++++++++++++++++-
 9 files changed, 333 insertions(+), 31 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 54d8ad9554..b21387ff58 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -601,7 +601,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 
 	/* Parse and get conflict resolvers list. */
 	conflict_resolvers =
-		ParseAndGetSubConflictResolvers(pstate, stmt->resolvers, true);
+		ParseAndGetSubConflictResolvers(pstate, stmt->resolvers, true, opts.twophase);
 
 	/*
 	 * Since creating a replication slot is not transactional, rolling back
@@ -1340,6 +1340,20 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 								 errmsg("cannot disable two_phase when prepared transactions are present"),
 								 errhint("Resolve these transactions and try again.")));
 
+					/*
+					 * two_phase cannot be enabled if sub has a time based
+					 * resolver set, as it may result in data divergence.
+					 *
+					 * XXX: This restriction may be removed if the solution in
+					 * ParseAndGetSubConflictResolvers() comments is
+					 * implemented.
+					 */
+					if (opts.twophase && CheckIfSubHasTimeStampResolver(subid))
+						ereport(ERROR,
+								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+								 errmsg("cannot enable \"%s\" when a time based resolver is configured",
+										"two_phase")));
+
 					/* Change system catalog accordingly */
 					values[Anum_pg_subscription_subtwophasestate - 1] =
 						CharGetDatum(opts.twophase ?
@@ -1593,15 +1607,19 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 		case ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS:
 			{
 				List	   *conflict_resolvers = NIL;
+				bool		sub_twophase = false;
+
+				if (sub->twophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED)
+					sub_twophase = true;
 
 				/*
 				 * Get the list of conflict types and resolvers and validate
 				 * them.
 				 */
-				conflict_resolvers = ParseAndGetSubConflictResolvers(
-																	 pstate,
+				conflict_resolvers = ParseAndGetSubConflictResolvers(pstate,
 																	 stmt->resolvers,
-																	 false);
+																	 false,
+																	 sub_twophase);
 
 				/*
 				 * Update the conflict resolvers for the corresponding
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 216e57e49f..8fae4078e6 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -583,7 +583,7 @@ has_conflicting_tuple(Oid subid, ConflictType type, ItemPointer tupleid,
 	 * Get the configured resolver and determine if remote changes should be
 	 * applied.
 	 */
-	resolver = GetConflictResolver(subid, type, rel, NULL, &apply_remote);
+	resolver = GetConflictResolver(subid, type, rel, NULL, NULL, NULL);
 
 	/*
 	 * Proceed to find conflict if the resolver is set to a non-default value;
@@ -604,6 +604,14 @@ has_conflicting_tuple(Oid subid, ConflictType type, ItemPointer tupleid,
 			TransactionId xmin;
 
 			GetTupleTransactionInfo(*conflictslot, &xmin, &origin, &committs);
+
+			/*
+			 * Get the configured resolver and determine if remote changes
+			 * should be applied.
+			 */
+			resolver = GetConflictResolver(subid, type, rel, *conflictslot,
+										   NULL, &apply_remote);
+
 			ReportApplyConflict(estate, resultRelInfo, type, resolver,
 								NULL, *conflictslot, slot, uniqueidx,
 								xmin, origin, committs, apply_remote);
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index e7f7d4c5e4..6cbfab0097 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -161,6 +161,7 @@
 #include "libpq/pqmq.h"
 #include "pgstat.h"
 #include "postmaster/interrupt.h"
+#include "replication/conflict.h"
 #include "replication/logicallauncher.h"
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
@@ -312,6 +313,18 @@ pa_can_start(void)
 	if (!AllTablesyncsReady())
 		return false;
 
+	/*
+	 * Do not start a new parallel worker if 'last_update_wins' is configured
+	 * for any conflict type, as we need the commit timestamp in the
+	 * beginning.
+	 *
+	 * XXX: To lift this restriction, we could write the changes to a file
+	 * when a conflict is detected, and then at the commit time, let the
+	 * remaining changes be applied by the apply worker.
+	 */
+	if (CheckIfSubHasTimeStampResolver(MySubscription->oid))
+		return false;
+
 	return true;
 }
 
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 8309a70883..61d009ade7 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -43,6 +43,7 @@ static const char *const ConflictTypeNames[] = {
 static const char *const ConflictResolverNames[] = {
 	[CR_APPLY_REMOTE] = "apply_remote",
 	[CR_KEEP_LOCAL] = "keep_local",
+	[CR_LAST_UPDATE_WINS] = "last_update_wins",
 	[CR_APPLY_OR_SKIP] = "apply_or_skip",
 	[CR_APPLY_OR_ERROR] = "apply_or_error",
 	[CR_SKIP] = "skip",
@@ -73,12 +74,12 @@ static const char *const ConflictResolverNames[] = {
  * friendly name for a resolver and thus has been added here.
  */
 static const int ConflictTypeResolverMap[][CONFLICT_NUM_RESOLVERS + 1] = {
-	[CT_INSERT_EXISTS] = {CR_ERROR, CR_APPLY_REMOTE, CR_KEEP_LOCAL, -1},
-	[CT_UPDATE_EXISTS] = {CR_ERROR, CR_APPLY_REMOTE, CR_KEEP_LOCAL, -1},
-	[CT_UPDATE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR, -1},
+	[CT_INSERT_EXISTS] = {CR_ERROR, CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_LAST_UPDATE_WINS, -1},
+	[CT_UPDATE_EXISTS] = {CR_ERROR, CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_LAST_UPDATE_WINS, -1},
+	[CT_UPDATE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR, CR_LAST_UPDATE_WINS, -1},
 	[CT_UPDATE_MISSING] = {CR_SKIP, CR_APPLY_OR_SKIP, CR_APPLY_OR_ERROR, CR_ERROR, -1},
 	[CT_DELETE_MISSING] = {CR_SKIP, CR_ERROR, -1},
-	[CT_DELETE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR, -1}
+	[CT_DELETE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR, CR_LAST_UPDATE_WINS, -1}
 };
 
 StaticAssertDecl(lengthof(ConflictTypeResolverMap) == CONFLICT_NUM_TYPES,
@@ -382,6 +383,11 @@ errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo,
 	if (val_desc)
 		appendStringInfo(&err_detail, "\n%s", val_desc);
 
+	/* append remote tuple timestamp details if resolver is last_update_wins */
+	if (resolver == CR_LAST_UPDATE_WINS)
+		appendStringInfo(&err_detail, _(" The remote tuple is from orign-id=%u, modified at %s."),
+						 replorigin_session_origin,
+						 timestamptz_to_str(replorigin_session_origin_timestamp));
 	return errdetail_internal("%s", err_detail.data);
 }
 
@@ -682,6 +688,14 @@ ValidateConflictTypeAndResolver(const char *conflict_type,
 					   conflict_resolver,
 					   conflict_type));
 
+	if ((resolver == CR_LAST_UPDATE_WINS) && !track_commit_timestamp)
+		ereport(ERROR,
+				errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("resolver %s requires \"%s\" to be enabled",
+					   conflict_resolver, "track_commit_timestamp"),
+				errhint("Make sure the configuration parameter \"%s\" is set.",
+						"track_commit_timestamp"));
+
 	return type;
 }
 
@@ -725,6 +739,42 @@ get_conflict_resolver_internal(ConflictType type, Oid subid)
 	return resolver;
 }
 
+/*
+ * Compare the timestamps of given local tuple and the remote tuple to
+ * resolve the conflict.
+ *
+ * Returns true if remote tuple has the latest timestamp, false otherwise.
+ */
+static bool
+resolve_by_timestamp(TupleTableSlot *localslot)
+{
+	TransactionId local_xmin;
+	TimestampTz local_ts;
+	RepOriginId local_origin;
+	int			ts_cmp;
+	uint64		local_system_identifier;
+
+	/* Get origin and timestamp info of the local tuple */
+	GetTupleTransactionInfo(localslot, &local_xmin, &local_origin, &local_ts);
+
+	/* Compare the timestamps of remote & local tuple to decide the winner */
+	ts_cmp = timestamptz_cmp_internal(replorigin_session_origin_timestamp,
+									  local_ts);
+
+	if (ts_cmp == 0)
+	{
+		elog(LOG, "Timestamps of remote and local tuple are equal, comparing remote and local system identifiers");
+
+		/* Get current system's identifier */
+		local_system_identifier = GetSystemIdentifier();
+
+		return local_system_identifier <= replorigin_session_origin_sysid;
+	}
+	else
+		return (ts_cmp > 0);
+
+}
+
 /*
  * Check if a full tuple can be created from the new tuple.
  * Return true if yes, false otherwise.
@@ -763,8 +813,8 @@ can_create_full_tuple(Relation localrel,
  * conflict types that have not been explicitly defined by the user.
  */
 List *
-ParseAndGetSubConflictResolvers(ParseState *pstate,
-								List *stmtresolvers, bool add_defaults)
+ParseAndGetSubConflictResolvers(ParseState *pstate, List *stmtresolvers,
+								bool add_defaults, bool sub_twophase)
 {
 	List	   *SeenTypes = NIL;
 	List	   *res = NIL;
@@ -786,6 +836,23 @@ ParseAndGetSubConflictResolvers(ParseState *pstate,
 		ValidateConflictTypeAndResolver(defel->defname,
 										resolver);
 
+		/*
+		 * Time based conflict resolution for two phase transactions can
+		 * result in data divergence, so disallow last_update_wins resolver
+		 * when two_phase is enabled.
+		 *
+		 * XXX: An alternative solution idea is that if a conflict is detected
+		 * and the resolution strategy is last_update_wins, then start writing
+		 * all the changes to a file similar to what we do for streaming mode.
+		 * Once commit_prepared arrives, we will read and apply the changes.
+		 */
+		if ((pg_strcasecmp(resolver, "last_update_wins") == 0) &&
+			sub_twophase)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("cannot set %s resolver when \"%s\" is enabled; these options are mutually exclusive",
+							"last_update_wins", "two_phase")));
+
 		/* Add the conflict type to the list of seen types */
 		SeenTypes = lappend(SeenTypes, makeString(defel->defname));
 
@@ -1028,14 +1095,30 @@ RemoveSubConflictResolvers(Oid subid)
  */
 ConflictResolver
 GetConflictResolver(Oid subid, ConflictType type, Relation localrel,
-					LogicalRepTupleData *newtup, bool *apply_remote)
+					TupleTableSlot *conflictslot, LogicalRepTupleData *newtup,
+					bool *apply_remote)
 {
 	ConflictResolver resolver;
 
 	resolver = get_conflict_resolver_internal(type, subid);
 
+	/* If caller has given apply_remote as NULL, simply return the resolver */
+	if (!apply_remote)
+		return resolver;
+
 	switch (resolver)
 	{
+		case CR_LAST_UPDATE_WINS:
+			if (!track_commit_timestamp)
+				ereport(ERROR,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("resolver %s requires \"%s\" to be enabled",
+							   ConflictResolverNames[resolver], "track_commit_timestamp"),
+						errhint("Make sure the configuration parameter \"%s\" is set.",
+								"track_commit_timestamp"));
+			else
+				*apply_remote = resolve_by_timestamp(conflictslot);
+			break;
 		case CR_APPLY_REMOTE:
 			*apply_remote = true;
 			break;
@@ -1063,3 +1146,40 @@ GetConflictResolver(Oid subid, ConflictType type, Relation localrel,
 
 	return resolver;
 }
+
+/*
+ * Check if the "last_update_wins" resolver is configured
+ * for any conflict type in the given subscription.
+ * Returns true if set, false otherwise.
+ */
+bool
+CheckIfSubHasTimeStampResolver(Oid subid)
+{
+	bool		found = false;
+	bool		started_tx = false;
+	ConflictType type;
+	ConflictResolver resolver;
+
+	/* This function might be called inside or outside of transaction */
+	if (!IsTransactionState())
+	{
+		StartTransactionCommand();
+		started_tx = true;
+	}
+
+	/* Check if any conflict type has resolver set to last_update_wins */
+	for (type = 0; type < CONFLICT_NUM_TYPES; type++)
+	{
+		resolver = get_conflict_resolver_internal(type, subid);
+		if (resolver == CR_LAST_UPDATE_WINS)
+		{
+			found = true;
+			break;
+		}
+	}
+
+	if (started_tx)
+		CommitTransactionCommand();
+
+	return found;
+}
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index baf696d8e6..3094030103 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -159,6 +159,7 @@ typedef struct ReplicationStateCtl
 RepOriginId replorigin_session_origin = InvalidRepOriginId; /* assumed identity */
 XLogRecPtr	replorigin_session_origin_lsn = InvalidXLogRecPtr;
 TimestampTz replorigin_session_origin_timestamp = 0;
+uint64		replorigin_session_origin_sysid = 0;
 
 /*
  * Base address into a shared memory array of replication states of size
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 3579bd04f3..291e79a6ac 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1006,6 +1006,12 @@ apply_handle_begin(StringInfo s)
 	in_remote_transaction = true;
 
 	pgstat_report_activity(STATE_RUNNING, NULL);
+
+	/*
+	 * Capture the commit timestamp of the remote transaction for time based
+	 * conflict resolution purpose.
+	 */
+	replorigin_session_origin_timestamp = begin_data.committime;
 }
 
 /*
@@ -2741,7 +2747,7 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 
 			resolver = GetConflictResolver(MySubscription->oid,
 										   CT_UPDATE_ORIGIN_DIFFERS,
-										   localrel, NULL,
+										   localrel, localslot, NULL,
 										   &apply_remote);
 
 			/* Store the new tuple for conflict reporting */
@@ -2834,7 +2840,8 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		 * UPDATE to INSERT and apply the change.
 		 */
 		resolver = GetConflictResolver(MySubscription->oid, CT_UPDATE_MISSING,
-									   localrel, newtup, &apply_remote);
+									   localrel, localslot, newtup,
+									   &apply_remote);
 
 		ReportApplyConflict(estate, relinfo, CT_UPDATE_MISSING, resolver,
 							remoteslot, NULL, newslot, InvalidOid,
@@ -2989,7 +2996,8 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 		{
 			resolver = GetConflictResolver(MySubscription->oid,
 										   CT_DELETE_ORIGIN_DIFFERS,
-										   localrel, NULL, &apply_remote);
+										   localrel, localslot, NULL,
+										   &apply_remote);
 
 			ReportApplyConflict(estate, relinfo, CT_DELETE_ORIGIN_DIFFERS,
 								resolver, remoteslot, localslot, NULL,
@@ -3017,7 +3025,8 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 		 * configured, either skip and log a message or emit an error.
 		 */
 		resolver = GetConflictResolver(MySubscription->oid, CT_DELETE_MISSING,
-									   localrel, NULL, &apply_remote);
+									   localrel, localslot, NULL,
+									   &apply_remote);
 
 		/* Resolver is set to skip, thus report the conflict and skip */
 		if (!apply_remote)
@@ -3214,8 +3223,8 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 					/* Store the new tuple for conflict reporting */
 					slot_store_data(newslot, part_entry, newtup);
 					resolver = GetConflictResolver(MySubscription->oid,
-												   CT_UPDATE_MISSING,
-												   partrel, newtup,
+												   CT_UPDATE_MISSING, partrel,
+												   localslot, newtup,
 												   &apply_remote);
 
 					/*
@@ -3259,7 +3268,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 
 						resolver = GetConflictResolver(MySubscription->oid,
 													   CT_UPDATE_ORIGIN_DIFFERS,
-													   partrel, NULL,
+													   partrel, localslot, NULL,
 													   &apply_remote);
 
 						ReportApplyConflict(estate, partrelinfo, CT_UPDATE_ORIGIN_DIFFERS,
@@ -4790,6 +4799,7 @@ run_apply_worker()
 	TimeLineID	startpointTLI;
 	char	   *err;
 	bool		must_use_password;
+	char	   *replorigin_sysid;
 
 	slotname = MySubscription->slotname;
 
@@ -4830,10 +4840,12 @@ run_apply_worker()
 						MySubscription->name, err)));
 
 	/*
-	 * We don't really use the output identify_system for anything but it does
-	 * some initializations on the upstream so let's still call it.
+	 * Call identify_system to do some initializations on the upstream and
+	 * store the output as system identifier of the replication origin node.
 	 */
-	(void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI);
+	replorigin_sysid = walrcv_identify_system(LogRepWorkerWalRcvConn,
+											  &startpointTLI);
+	replorigin_session_origin_sysid = strtoul(replorigin_sysid, NULL, 10);
 
 	set_apply_error_context_origin(originname);
 
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 0680ef1d4e..9eca7fccbe 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -9,6 +9,7 @@
 #ifndef CONFLICT_H
 #define CONFLICT_H
 
+#include "catalog/pg_subscription.h"
 #include "nodes/execnodes.h"
 #include "parser/parse_node.h"
 #include "replication/logicalrelation.h"
@@ -66,6 +67,9 @@ typedef enum ConflictResolver
 	/* Keep the local change */
 	CR_KEEP_LOCAL,
 
+	/* Apply the change with latest timestamp */
+	CR_LAST_UPDATE_WINS,
+
 	/* Apply the remote change; skip if it cannot be applied */
 	CR_APPLY_OR_SKIP,
 
@@ -105,7 +109,8 @@ extern void SetSubConflictResolvers(Oid subId, List *resolvers);
 extern void RemoveSubConflictResolvers(Oid confid);
 extern List *ParseAndGetSubConflictResolvers(ParseState *pstate,
 											 List *stmtresolvers,
-											 bool add_defaults);
+											 bool add_defaults,
+											 bool sub_twophase);
 extern void UpdateSubConflictResolvers(List *conflict_resolvers, Oid subid);
 extern ConflictType ValidateConflictType(const char *conflict_type);
 extern ConflictType ValidateConflictTypeAndResolver(const char *conflict_type,
@@ -114,6 +119,8 @@ extern List *GetDefaultConflictResolvers(void);
 extern void ResetSubConflictResolver(Oid subid, char *conflict_type);
 extern ConflictResolver GetConflictResolver(Oid subid, ConflictType type,
 											Relation localrel,
+											TupleTableSlot *localslot,
 											LogicalRepTupleData *newtup,
 											bool *apply_remote);
+extern bool CheckIfSubHasTimeStampResolver(Oid subid);
 #endif
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 7189ba9e76..dcbbbdf6ea 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -36,6 +36,7 @@ typedef struct xl_replorigin_drop
 extern PGDLLIMPORT RepOriginId replorigin_session_origin;
 extern PGDLLIMPORT XLogRecPtr replorigin_session_origin_lsn;
 extern PGDLLIMPORT TimestampTz replorigin_session_origin_timestamp;
+extern PGDLLIMPORT uint64 replorigin_session_origin_sysid;
 
 /* API for querying & manipulating replication origins */
 extern RepOriginId replorigin_by_name(const char *roname, bool missing_ok);
diff --git a/src/test/subscription/t/034_conflict_resolver.pl b/src/test/subscription/t/034_conflict_resolver.pl
index f7c8abcecf..f6a7258aa4 100644
--- a/src/test/subscription/t/034_conflict_resolver.pl
+++ b/src/test/subscription/t/034_conflict_resolver.pl
@@ -156,6 +156,34 @@ $node_subscriber->wait_for_log(
 # Truncate table on subscriber to get rid of the error
 $node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
 
+#############################################
+# Test 'last_update_wins' for 'insert_exists'
+#############################################
+
+# Change CONFLICT RESOLVER of insert_exists to last_update_wins
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (insert_exists = 'last_update_wins');"
+);
+
+# Create local data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (4,'fromsub')");
+
+# Create conflicting data on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (4,'frompub')");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=insert_exists, resolution=last_update_wins/,
+	$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=4);");
+
+is($result, 'frompub', "remote data wins");
+
 # Truncate the table on the publisher
 $node_publisher->safe_psql('postgres', "TRUNCATE conf_tab;");
 
@@ -256,6 +284,36 @@ $node_subscriber->wait_for_log(
 # Truncate table on subscriber to get rid of the error
 $node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
 
+############################################
+# Test 'last_update_wins' for 'update_exists'
+############################################
+
+# Change CONFLICT RESOLVER of update_exists to last_update_wins
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_exists = 'last_update_wins');"
+);
+
+# Create local data on the subscriber
+$node_subscriber->safe_psql(
+	'postgres',
+	"INSERT INTO conf_tab(a, data) VALUES (3,'fromsub');
+	 INSERT INTO conf_tab(a, data) VALUES (5,'fromsub')");
+
+# Create conflicting data on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres', "UPDATE conf_tab SET a=5 WHERE a=3;");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_exists, resolution=last_update_wins/,
+	$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=5);");
+
+is($result, 'frompub', "remote data wins");
+
+
 ###################################
 # Test 'skip' for 'delete_missing'
 ###################################
@@ -309,16 +367,48 @@ $node_subscriber->safe_psql(
 $node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
 
 
-#################################################
-# Test 'apply_remote' for 'delete_origin_differs'
-#################################################
+#####################################################
+# Test 'last_update_wins' for 'delete_origin_differs'
+#####################################################
 
+# Change CONFLICT RESOLVER of delete_origin_differs to last_update_wins
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (delete_origin_differs = 'last_update_wins');"
+);
 # 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');");
+	 INSERT INTO conf_tab(a, data) VALUES (3,'frompub');
+	 INSERT INTO conf_tab(a, data) VALUES (4,'frompub');");
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub' WHERE (a=4);");
+
+# Create a conflicting delete on the publisher
+$log_offset = -s $node_subscriber->logfile;
+$node_publisher->safe_psql('postgres', "DELETE FROM conf_tab WHERE (a=4);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=delete_origin_differs, resolution=last_update_wins/,
+	$log_offset);
+
+# Confirm that the remote delete the local updated row
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT data from conf_tab WHERE (a=4);");
+
+is($result, '', "delete from remote wins");
+
+#################################################
+# Test 'apply_remote' for 'delete_origin_differs'
+#################################################
+
+# Change CONFLICT RESOLVER of delete_origin_differs to apply_remote
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (delete_origin_differs = 'apply_remote');"
+);
 
 # Modify data on the subscriber
 $node_subscriber->safe_psql('postgres',
@@ -405,10 +495,14 @@ $node_subscriber->safe_psql(
 # Wait for initial table sync to finish
 $node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
 
-#################################################
-# Test 'apply_remote' for 'update_origin_differs'
-#################################################
 
+#####################################################
+# Test 'last_update_wins' for 'update_origin_differs'
+#####################################################
+# Change CONFLICT RESOLVER of update_origin_differs to last_update_wins
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_origin_differs = 'last_update_wins');"
+);
 # Insert data in the publisher
 $node_publisher->safe_psql(
 	'postgres',
@@ -426,7 +520,7 @@ $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_origin_differs, resolution=apply_remote/,
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_origin_differs, resolution=last_update_wins/,
 	$log_offset);
 
 # Confirm that the remote update overrides the local update
@@ -435,6 +529,34 @@ $result = $node_subscriber->safe_psql('postgres',
 
 is($result, 'frompubnew', "update from remote is kept");
 
+#################################################
+# Test 'apply_remote' for 'update_origin_differs'
+#################################################
+
+# Change CONFLICT RESOLVER of update_origin_differs to apply_remote
+$node_subscriber->safe_psql('postgres',
+	"ALTER SUBSCRIPTION tap_sub CONFLICT RESOLVER (update_origin_differs = 'apply_remote');"
+);
+
+# Modify data on the subscriber
+$node_subscriber->safe_psql('postgres',
+	"UPDATE conf_tab SET data = 'fromsub2' 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 = 'frompubnew2' WHERE (a=1);");
+
+$node_subscriber->wait_for_log(
+	qr/LOG:  conflict detected on relation \"public.conf_tab\": conflict=update_origin_differs, resolution=apply_remote/,
+	$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, 'frompubnew2', "update from remote is kept");
+
 ###############################################
 # Test 'keep_local' for 'update_origin_differs'
 ###############################################
-- 
2.34.1

