From c9d9d19168f00fe0cef1d046bf2d0583e38ef999 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <itsajin@gmail.com>
Date: Thu, 12 Sep 2024 03:34:45 -0400
Subject: [PATCH v13 4/5] 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            |  37 +++++-
 src/backend/executor/execReplication.c             |  68 +++++------
 .../replication/logical/applyparallelworker.c      |  13 ++
 src/backend/replication/logical/conflict.c         | 131 +++++++++++++++++++-
 src/backend/replication/logical/origin.c           |   1 +
 src/backend/replication/logical/worker.c           |  30 +++--
 src/bin/pg_dump/pg_dump.c                          |  38 +++---
 src/bin/pg_dump/pg_dump.h                          |  36 +++---
 src/include/replication/conflict.h                 |  13 +-
 src/include/replication/origin.h                   |   1 +
 src/test/subscription/t/034_conflict_resolver.pl   | 136 +++++++++++++++++++--
 11 files changed, 399 insertions(+), 105 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 131e4f5..6b00cd8 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -450,7 +450,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
  */
 static void
 parse_subscription_conflict_resolvers(List *stmtresolvers,
-									  ConflictTypeResolver *resolvers)
+									  ConflictTypeResolver *resolvers,
+									  bool twophase)
 {
 	ListCell   *lc;
 	List	   *SeenTypes = NIL;
@@ -484,6 +485,21 @@ parse_subscription_conflict_resolvers(List *stmtresolvers,
 
 		/* Update the corresponding resolver for the given conflict type. */
 		resolvers[type].resolver = downcase_truncate_identifier(resolver, strlen(resolver), false);
+
+		/*
+		 * Time based conflict resolution for two phase transactions can
+		 * result in data divergence, so disallow enabling both together.
+		 *
+		 * 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 (twophase && pg_strcasecmp(resolvers[type].resolver, "last_update_wins") == 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("Setting any resolver to last_update_wins and %s are mutually exclusive options",
+						 "two_phase = true")));
 	}
 }
 
@@ -647,7 +663,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/* Parse and check conflict resolvers. */
-	parse_subscription_conflict_resolvers(stmt->resolvers, conflictResolvers);
+	parse_subscription_conflict_resolvers(stmt->resolvers, conflictResolvers,
+										  opts.twophase);
 
 	/*
 	 * Since creating a replication slot is not transactional, rolling back
@@ -1386,6 +1403,19 @@ 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
+					 * parse_subscription_conflict_resolvers() comments is
+					 * implemented.
+					 */
+					if (opts.twophase && CheckIfSubHasTimeStampResolver(subid))
+						ereport(ERROR,
+								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+								 errmsg("cannot enable two_phase when a time based resolver is configured")));
+
 					/* Change system catalog accordingly */
 					values[Anum_pg_subscription_subtwophasestate - 1] =
 						CharGetDatum(opts.twophase ?
@@ -1643,7 +1673,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				conf_detection_check_prerequisites();
 
 				/* Get the list of conflict types and resolvers and validate them. */
-				conflict_resolvers = GetAndValidateSubsConflictResolverList(stmt->resolvers);
+				conflict_resolvers = GetAndValidateSubsConflictResolverList(stmt->resolvers,
+																			sub);
 
 				/*
 				 * Update the conflict resolvers for the corresponding
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index f944df7..5b031a4 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -565,10 +565,20 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,
 static bool
 has_conflicting_tuple(EState *estate, ResultRelInfo *resultRelInfo,
 					  TupleTableSlot **conflictslot, ConflictType type,
-					  ConflictResolver resolver, TupleTableSlot *slot,
-					  Oid subid, bool apply_remote, ItemPointer tupleid)
+					  TupleTableSlot *slot, Oid subid, ItemPointer tupleid)
 {
+	ConflictResolver resolver;
 	List	   *conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
+	Relation	rel = resultRelInfo->ri_RelationDesc;
+
+	/*
+	 * Proceed only if the resolver is not set to 'ERROR'; if the resolver is
+	 * 'ERROR', the caller will handle it.
+	 */
+	resolver = GetConflictResolver(*conflictslot, rel, type, NULL, NULL,
+								   subid);
+	if (resolver == CR_ERROR)
+		return false;
 
 	/* Check all the unique indexes for a conflict */
 	foreach_oid(uniqueidx, conflictindexes)
@@ -584,8 +594,17 @@ has_conflicting_tuple(EState *estate, ResultRelInfo *resultRelInfo,
 			RepOriginId origin;
 			TimestampTz committs;
 			TransactionId xmin;
+			bool		apply_remote = false;
 
 			GetTupleTransactionInfo(*conflictslot, &xmin, &origin, &committs);
+
+			/*
+			 * Get the configured resolver and determine if remote changes
+			 * should be applied.
+			 */
+			resolver = GetConflictResolver(*conflictslot, rel, type,
+										   &apply_remote, NULL, subid);
+
 			ReportApplyConflict(estate, resultRelInfo, type, resolver,
 								NULL, *conflictslot, slot, uniqueidx,
 								xmin, origin, committs, apply_remote);
@@ -634,8 +653,6 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 		List	   *recheckIndexes = NIL;
 		List	   *conflictindexes;
 		bool		conflict = false;
-		ConflictResolver resolver;
-		bool		apply_remote = false;
 		ItemPointerData invalidItemPtr;
 
 		/* Compute stored generated columns */
@@ -650,23 +667,10 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
 		if (rel->rd_rel->relispartition)
 			ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
-		/*
-		 * Get the configured resolver and determine if remote changes should
-		 * be applied.
-		 */
-		resolver = GetConflictResolver(rel, CT_INSERT_EXISTS, &apply_remote,
-									   NULL, subid);
-
-		/*
-		 * Check for conflict and return to caller for resolution if found.
-		 * Only perform this check if the 'insert_exists' resolver is not set
-		 * to 'ERROR'; if the resolver is 'ERROR', the upcoming
-		 * CheckAndReportConflict() call will handle it.
-		 */
-		if (resolver != CR_ERROR &&
-			has_conflicting_tuple(estate, resultRelInfo, &(*conflictslot),
-								  CT_INSERT_EXISTS, resolver, slot, subid,
-								  apply_remote, &invalidItemPtr))
+		/* Check for conflict and return to caller for resolution if found */
+		if (has_conflicting_tuple(estate, resultRelInfo, &(*conflictslot),
+								  CT_INSERT_EXISTS, slot, subid,
+								  &invalidItemPtr))
 			return;
 
 		/* OK, store the tuple and create index entries for it */
@@ -750,8 +754,6 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 		TU_UpdateIndexes update_indexes;
 		List	   *conflictindexes;
 		bool		conflict = false;
-		ConflictResolver resolver;
-		bool		apply_remote = false;
 
 		/* Compute stored generated columns */
 		if (rel->rd_att->constr &&
@@ -765,23 +767,9 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
 		if (rel->rd_rel->relispartition)
 			ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
-		/*
-		 * Get the configured resolver and determine if remote changes should
-		 * be applied.
-		 */
-		resolver = GetConflictResolver(rel, CT_UPDATE_EXISTS, &apply_remote,
-									   NULL, subid);
-
-		/*
-		 * Check for conflict and return to caller for resolution if found.
-		 * Only perform this check if the 'update_exists' resolver is not set
-		 * to 'ERROR'; if the resolver is 'ERROR', the upcoming
-		 * CheckAndReportConflict() call will handle it.
-		 */
-		if (resolver != CR_ERROR &&
-			has_conflicting_tuple(estate, resultRelInfo, &(*conflictslot),
-								  CT_UPDATE_EXISTS, resolver, slot, subid,
-								  apply_remote, tid))
+		/* Check for conflict and return to caller for resolution if found */
+		if (has_conflicting_tuple(estate, resultRelInfo, &(*conflictslot),
+								  CT_UPDATE_EXISTS, slot, subid, tid))
 			return;
 
 		simple_table_tuple_update(rel, tid, slot, estate->es_snapshot,
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index e7f7d4c..6cbfab0 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 3db399c..20107d9 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -51,6 +51,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",
@@ -76,12 +77,12 @@ static const char *const ConflictResolverNames[] = {
  * friendly name for a resolver and thus has been added here.
  */
 static const int ConflictTypeResolverMap[][CONFLICT_TYPE_MAX_RESOLVERS] = {
-	[CT_INSERT_EXISTS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR},
-	[CT_UPDATE_EXISTS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR},
-	[CT_UPDATE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR},
+	[CT_INSERT_EXISTS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_LAST_UPDATE_WINS, CR_ERROR},
+	[CT_UPDATE_EXISTS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_LAST_UPDATE_WINS, CR_ERROR},
+	[CT_UPDATE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_LAST_UPDATE_WINS, CR_ERROR},
 	[CT_UPDATE_MISSING] = {CR_APPLY_OR_SKIP, CR_APPLY_OR_ERROR, CR_SKIP, CR_ERROR},
 	[CT_DELETE_MISSING] = {CR_SKIP, CR_ERROR},
-	[CT_DELETE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR}
+	[CT_DELETE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_LAST_UPDATE_WINS, CR_ERROR}
 };
 
 /*
@@ -391,6 +392,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);
 }
 
@@ -672,6 +678,15 @@ validate_conflict_type_and_resolver(const char *conflict_type,
 				errmsg("%s is not a valid conflict resolver for conflict type %s",
 					   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;
 
 }
@@ -717,6 +732,42 @@ get_conflict_resolver_internal(ConflictType type, Oid subid)
 }
 
 /*
+ * 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.
  */
@@ -744,7 +795,7 @@ can_create_full_tuple(Relation localrel,
  * ALTER SUBSCRIPTION command and return a list of ConflictTypeResolver nodes.
  */
 List *
-GetAndValidateSubsConflictResolverList(List *stmtresolvers)
+GetAndValidateSubsConflictResolverList(List *stmtresolvers, Subscription *sub)
 {
 	ListCell   *lc;
 	ConflictTypeResolver *conftyperesolver = NULL;
@@ -779,6 +830,21 @@ GetAndValidateSubsConflictResolverList(List *stmtresolvers)
 											conftyperesolver->conflict_type,
 											conftyperesolver->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: This restriction may be removed if the solution in
+		 * parse_subscription_conflict_resolvers() comments is implemented.
+		 */
+		if ((pg_strcasecmp(conftyperesolver->resolver, "last_update_wins") == 0) &&
+			sub->twophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					errmsg("cannot set %s resolver for a subscription that has two_phase enabled",
+						   "last_update_wins")));
+
 		/* Add the conflict type to the list of seen types */
 		conflictTypes = lappend(conflictTypes,
 								makeString((char *) conftyperesolver->conflict_type));
@@ -1018,15 +1084,31 @@ RemoveSubscriptionConflictResolvers(Oid subid)
  * false otherwise.
  */
 ConflictResolver
-GetConflictResolver(Relation localrel, ConflictType type, bool *apply_remote,
+GetConflictResolver(TupleTableSlot *localslot, Relation localrel,
+					ConflictType type, bool *apply_remote,
 					LogicalRepTupleData *newtup, Oid subid)
 {
 	ConflictResolver resolver;
 
 	resolver = get_conflict_resolver_internal(type, subid);
 
+	/* If caller has given apply_remote as NULL, 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(localslot);
+			break;
 		case CR_APPLY_REMOTE:
 			*apply_remote = true;
 			break;
@@ -1054,3 +1136,40 @@ GetConflictResolver(Relation localrel, ConflictType type, bool *apply_remote,
 
 	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 = CT_MIN; type <= CT_MAX; 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 baf696d..3094030 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 9e9f4b4..c2186f8 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;
 }
 
 /*
@@ -2738,7 +2744,8 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		{
 			TupleTableSlot *newslot;
 
-			resolver = GetConflictResolver(localrel, CT_UPDATE_ORIGIN_DIFFERS,
+			resolver = GetConflictResolver(localslot, localrel,
+										   CT_UPDATE_ORIGIN_DIFFERS,
 										   &apply_remote, NULL,
 										   MySubscription->oid);
 
@@ -2811,7 +2818,7 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		 * the configured resolver is in favor of applying the change, convert
 		 * UPDATE to INSERT and apply the change.
 		 */
-		resolver = GetConflictResolver(localrel, CT_UPDATE_MISSING,
+		resolver = GetConflictResolver(localslot, localrel, CT_UPDATE_MISSING,
 									   &apply_remote, newtup,
 									   MySubscription->oid);
 
@@ -2964,7 +2971,8 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 		if (GetTupleTransactionInfo(localslot, &localxmin, &localorigin, &localts) &&
 			localorigin != replorigin_session_origin)
 		{
-			resolver = GetConflictResolver(localrel, CT_DELETE_ORIGIN_DIFFERS,
+			resolver = GetConflictResolver(localslot, localrel,
+										   CT_DELETE_ORIGIN_DIFFERS,
 										   &apply_remote, NULL,
 										   MySubscription->oid);
 
@@ -2993,7 +3001,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 		 * The tuple to be deleted could not be found. Based on resolver
 		 * configured, either skip and log a message or emit an error.
 		 */
-		resolver = GetConflictResolver(localrel, CT_DELETE_MISSING,
+		resolver = GetConflictResolver(localslot, localrel, CT_DELETE_MISSING,
 									   &apply_remote, NULL,
 									   MySubscription->oid);
 
@@ -3191,7 +3199,7 @@ 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,
+					resolver = GetConflictResolver(localslot, partrel, CT_UPDATE_MISSING,
 												   &apply_remote, newtup,
 												   MySubscription->oid);
 
@@ -3234,7 +3242,8 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 						newslot = table_slot_create(partrel, &estate->es_tupleTable);
 						slot_store_data(newslot, part_entry, newtup);
 
-						resolver = GetConflictResolver(partrel, CT_UPDATE_ORIGIN_DIFFERS,
+						resolver = GetConflictResolver(localslot, partrel,
+													   CT_UPDATE_ORIGIN_DIFFERS,
 													   &apply_remote, NULL,
 													   MySubscription->oid);
 
@@ -4766,6 +4775,7 @@ run_apply_worker()
 	TimeLineID	startpointTLI;
 	char	   *err;
 	bool		must_use_password;
+	char	   *replorigin_sysid;
 
 	slotname = MySubscription->slotname;
 
@@ -4806,10 +4816,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/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 27f9f30..a5afbe3 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4828,8 +4828,8 @@ getSubscriptions(Archive *fout)
 	DumpOptions *dopt = fout->dopt;
 	PQExpBuffer query;
 	PQExpBuffer confQuery;
-	PGresult    *res;
-	PGresult    *confRes;
+	PGresult   *res;
+	PGresult   *confRes;
 	SubscriptionInfo *subinfo;
 	int			i_tableoid;
 	int			i_oid;
@@ -5023,50 +5023,50 @@ getSubscriptions(Archive *fout)
 		ntuples = PQntuples(confRes);
 		for (j = 0; j < ntuples; j++)
 		{
-			char *confType = PQgetvalue(confRes, j, 0);
-			char *confResVal = PQgetvalue(confRes, j, 1);
+			char	   *confType = PQgetvalue(confRes, j, 0);
+			char	   *confResVal = PQgetvalue(confRes, j, 1);
 
 			if (strcmp(confType, "delete_missing") == 0)
 			{
 				subinfo[i].conflict_resolver[CT_DELETE_MISSING].resolver =
-																pg_strdup(confResVal);
+					pg_strdup(confResVal);
 				subinfo[i].conflict_resolver[CT_DELETE_MISSING].conflict_type =
-																pg_strdup(confType);
+					pg_strdup(confType);
 			}
 			else if (strcmp(confType, "delete_origin_differs") == 0)
 			{
 				subinfo[i].conflict_resolver[CT_DELETE_ORIGIN_DIFFERS].resolver =
-																pg_strdup(confResVal);
+					pg_strdup(confResVal);
 				subinfo[i].conflict_resolver[CT_DELETE_ORIGIN_DIFFERS].conflict_type =
-																pg_strdup(confType);
+					pg_strdup(confType);
 			}
 			else if (strcmp(confType, "insert_exists") == 0)
 			{
 				subinfo[i].conflict_resolver[CT_INSERT_EXISTS].resolver =
-																pg_strdup(confResVal);
+					pg_strdup(confResVal);
 				subinfo[i].conflict_resolver[CT_INSERT_EXISTS].conflict_type =
-																pg_strdup(confType);
+					pg_strdup(confType);
 			}
 			else if (strcmp(confType, "update_exists") == 0)
 			{
 				subinfo[i].conflict_resolver[CT_UPDATE_EXISTS].resolver =
-																pg_strdup(confResVal);
+					pg_strdup(confResVal);
 				subinfo[i].conflict_resolver[CT_UPDATE_EXISTS].conflict_type =
-																pg_strdup(confType);
+					pg_strdup(confType);
 			}
 			else if (strcmp(confType, "update_missing") == 0)
 			{
 				subinfo[i].conflict_resolver[CT_UPDATE_MISSING].resolver =
-																pg_strdup(confResVal);
+					pg_strdup(confResVal);
 				subinfo[i].conflict_resolver[CT_UPDATE_MISSING].conflict_type =
-																pg_strdup(confType);
+					pg_strdup(confType);
 			}
 			else if (strcmp(confType, "update_origin_differs") == 0)
 			{
 				subinfo[i].conflict_resolver[CT_UPDATE_ORIGIN_DIFFERS].resolver =
-																pg_strdup(confResVal);
+					pg_strdup(confResVal);
 				subinfo[i].conflict_resolver[CT_UPDATE_ORIGIN_DIFFERS].conflict_type =
-																pg_strdup(confType);
+					pg_strdup(confType);
 			}
 
 		}
@@ -5323,12 +5323,12 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 	if (fout->remoteVersion >= 180000)
 	{
 		ConflictType idx;
-		bool		 first_resolver = true;
+		bool		first_resolver = true;
 
 		for (idx = 0; idx < CONFLICT_NUM_TYPES; idx++)
 		{
 			if (strcmp(subinfo->conflict_resolver[idx].resolver,
-					ConflictResolverNames[ConflictTypeDefaultResolvers[idx]]) != 0)
+					   ConflictResolverNames[ConflictTypeDefaultResolvers[idx]]) != 0)
 			{
 				if (first_resolver)
 				{
@@ -5338,7 +5338,7 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 					first_resolver = false;
 				}
 				else
-					appendPQExpBuffer(query,", %s = '%s'",
+					appendPQExpBuffer(query, ", %s = '%s'",
 									  subinfo->conflict_resolver[idx].conflict_type,
 									  subinfo->conflict_resolver[idx].resolver);
 			}
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 13625c6..9822d2e 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -711,7 +711,7 @@ typedef enum ConflictResolver
 
 	/* Error out */
 	CR_ERROR,
-}ConflictResolver;
+}			ConflictResolver;
 
 static const char *const ConflictResolverNames[] = {
 	[CR_APPLY_REMOTE] = "apply_remote",
@@ -739,27 +739,27 @@ typedef struct ConflictTypeResolver
 {
 	const char *conflict_type;
 	const char *resolver;
-}ConflictTypeResolver;
+}			ConflictTypeResolver;
 
 typedef struct _SubscriptionInfo
 {
 	DumpableObject dobj;
-	const char				*rolname;
-	char					*subenabled;
-	char					*subbinary;
-	char					*substream;
-	char					*subtwophasestate;
-	char					*subdisableonerr;
-	char					*subpasswordrequired;
-	char					*subrunasowner;
-	char					*subconninfo;
-	char					*subslotname;
-	char					*subsynccommit;
-	char					*subpublications;
-	char					*suborigin;
-	char					*suboriginremotelsn;
-	char					*subfailover;
-	ConflictTypeResolver	 conflict_resolver[CONFLICT_NUM_TYPES];
+	const char *rolname;
+	char	   *subenabled;
+	char	   *subbinary;
+	char	   *substream;
+	char	   *subtwophasestate;
+	char	   *subdisableonerr;
+	char	   *subpasswordrequired;
+	char	   *subrunasowner;
+	char	   *subconninfo;
+	char	   *subslotname;
+	char	   *subsynccommit;
+	char	   *subpublications;
+	char	   *suborigin;
+	char	   *suboriginremotelsn;
+	char	   *subfailover;
+	ConflictTypeResolver conflict_resolver[CONFLICT_NUM_TYPES];
 } SubscriptionInfo;
 
 /*
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 2c9b61c..f9262c5 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 "replication/logicalproto.h"
 #include "replication/logicalrelation.h"
@@ -70,6 +71,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 can not be applied */
 	CR_APPLY_OR_SKIP,
 
@@ -81,7 +85,7 @@ typedef enum ConflictResolver
 
 	/* Error out */
 	CR_ERROR,
-} ConflictResolver;
+}			ConflictResolver;
 
 /* Min and max conflict resolver */
 #define CR_MIN CR_APPLY_REMOTE
@@ -110,17 +114,20 @@ extern void InitConflictIndexes(ResultRelInfo *relInfo);
 extern void SetSubConflictResolver(Oid subId, ConflictTypeResolver * resolvers, int max_types);
 extern void RemoveSubscriptionConflictById(Oid confid);
 extern void RemoveSubscriptionConflictResolvers(Oid confid);
-extern List *GetAndValidateSubsConflictResolverList(List *stmtresolvers);
+extern List *GetAndValidateSubsConflictResolverList(List *stmtresolvers,
+													Subscription *sub);
 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 void ResetConflictResolver(Oid subid, char *conflict_type);
 extern void conf_detection_check_prerequisites(void);
-extern ConflictResolver GetConflictResolver(Relation localrel,
+extern ConflictResolver GetConflictResolver(TupleTableSlot *localslot,
+											Relation localrel,
 											ConflictType type,
 											bool *apply_remote,
 											LogicalRepTupleData *newtup,
 											Oid subid);
+extern bool CheckIfSubHasTimeStampResolver(Oid subid);
 
 #endif
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 7189ba9..dcbbbdf 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 d9333dc..6955e91 100755
--- a/src/test/subscription/t/034_conflict_resolver.pl
+++ b/src/test/subscription/t/034_conflict_resolver.pl
@@ -152,6 +152,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/,
+	$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;");
 
@@ -252,6 +280,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/,
+	$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'
 ###################################
@@ -305,16 +363,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/,
+	$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',
@@ -401,10 +491,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',
@@ -431,6 +525,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/,
+	$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'
 ###############################################
-- 
1.8.3.1

