From 69d3ac557246355c3999c98ebe349dd6190cbf40 Mon Sep 17 00:00:00 2001
From: Nisha Moond <nisha.moond412@gmail.com>
Date: Fri, 30 Aug 2024 08:55:09 +0530
Subject: [PATCH v11 4/4] Manage Clock skew and implement last_update_wins

This patch attempts to manage clock skew between nodes by
introducing two new GUCs:
a) max_logical_rep_clock_skew
b) max_logical_rep_clock_skew_action
c) max_logical_rep_clock_skew_wait

If the timestamp of the currently replayed transaction is in the future
compared to the current time on the subscriber and the difference is
larger than 'max_logical_rep_clock_skew', then the action configured
in 'max_logical_rep_clock_skew_action' is performed by the apply worker.

If user configures 'wait' in 'max_logical_rep_clock_skew_action' and
actual clock skew is large while 'max_logical_rep_clock_skew' is small,
the apply worker may have to wait for a longer period to manage the clock
skew. To control this maximum wait time, a new GUC,
'max_logical_rep_clock_skew_wait', is provided.  This allows the user to
set a cap on how long the apply worker should wait. If the computed wait
time exceeds this value, the apply worker will error out without waiting.

This patch also implements last_update_wins resolver.

Since conflict resolution for two phase commit transactions
using prepare-timestamp can result in data divergence, this patch
also restricts enabling two_phase and detect_conflict together
for a subscription.
---
 src/backend/commands/subscriptioncmds.c       |  30 +++-
 src/backend/executor/execReplication.c        |  68 +++-----
 .../replication/logical/applyparallelworker.c |  23 ++-
 src/backend/replication/logical/conflict.c    | 128 +++++++++++++-
 src/backend/replication/logical/origin.c      |   1 +
 src/backend/replication/logical/worker.c      | 162 ++++++++++++++++--
 .../utils/activity/wait_event_names.txt       |   1 +
 src/backend/utils/misc/guc_tables.c           |  40 +++++
 src/backend/utils/misc/postgresql.conf.sample |   9 +-
 src/include/replication/conflict.h            |  12 +-
 src/include/replication/logicalworker.h       |  18 ++
 src/include/replication/origin.h              |   1 +
 src/include/replication/worker_internal.h     |   2 +-
 src/include/utils/timestamp.h                 |   1 +
 .../subscription/t/034_conflict_resolver.pl   | 136 ++++++++++++++-
 src/tools/pgindent/typedefs.list              |   1 +
 16 files changed, 558 insertions(+), 75 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 38dd67487e..5f4e601363 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -448,7 +448,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;
 
@@ -466,6 +467,16 @@ parse_subscription_conflict_resolvers(List *stmtresolvers,
 
 		/* update the corresponding resolver for the given conflict type */
 		resolvers[type].resolver = defGetString(defel);
+
+		/*
+		 * Time based conflict resolution for two phase transactions can
+		 * result in data divergence, so disallow enabling both together.
+		 */
+		if (twophase && strcmp(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")));
 	}
 }
 
@@ -635,7 +646,10 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 		check_conflict_detection();
 
 	SetDefaultResolvers(conflictResolvers);
-	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
@@ -1374,6 +1388,15 @@ 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.
+					 */
+					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 ?
@@ -1631,7 +1654,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				check_conflict_detection();
 
 				/* get 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 f944df761c..5b031a462a 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 e7f7d4c5e4..da2857b59a 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,15 @@ pa_can_start(void)
 	if (!AllTablesyncsReady())
 		return false;
 
+	/*
+	 * Don't start a new parallel worker if user has either configured max
+	 * clock skew or if 'last_update_wins' is configured for any conflict
+	 * type. In both cases we need commit timestamp in the beginning.
+	 */
+	if ((max_logical_rep_clock_skew > LR_CLOCK_SKEW_DEFAULT) ||
+		CheckIfSubHasTimeStampResolver(MySubscription->oid))
+		return false;
+
 	return true;
 }
 
@@ -696,9 +706,20 @@ pa_process_spooled_messages_if_required(void)
 	}
 	else if (fileset_state == FS_READY)
 	{
+		/*
+		 * Currently we do not support starting parallel apply worker when
+		 * either clock skew is configured or conflict resolution is
+		 * configured to last_update_wins, thus it is okay to pass 0 as
+		 * origin-timestamp here.
+		 *
+		 * XXX: If in future, we support starting pa worker even with
+		 * last_update_wins resolver, then here we need to pass remote's
+		 * commit/prepare/abort timestamp; we can get that info from leader
+		 * worker in shared memory.
+		 */
 		apply_spooled_messages(&MyParallelShared->fileset,
 							   MyParallelShared->xid,
-							   InvalidXLogRecPtr);
+							   InvalidXLogRecPtr, 0);
 		pa_set_fileset_state(MyParallelShared, FS_EMPTY);
 	}
 
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 9fd96d4979..e7e7571427 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -49,6 +49,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",
@@ -74,12 +75,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}
 };
 
 /*
@@ -387,6 +388,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);
 }
 
@@ -666,6 +672,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;
 
 }
@@ -710,6 +725,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.
@@ -738,7 +789,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 *CTR = NULL;
@@ -758,6 +809,18 @@ GetAndValidateSubsConflictResolverList(List *stmtresolvers)
 		 */
 		validate_conflict_type_and_resolver(CTR->conflict_type, CTR->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.
+		 */
+		if ((strcmp(CTR->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")));
+
 		res = lappend(res, CTR);
 	}
 
@@ -977,15 +1040,31 @@ RemoveSubscriptionConflictBySubid(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;
@@ -1013,3 +1092,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 419e4814f0..bd8e6f0024 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -155,6 +155,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 95157438ce..18dc3aa332 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -318,6 +318,20 @@ static uint32 parallel_stream_nchanges = 0;
 /* Are we initializing an apply worker? */
 bool		InitializingApplyWorker = false;
 
+/*
+ * GUC support
+ */
+const struct config_enum_entry logical_rep_clock_skew_action_options[] = {
+	{"error", LR_CLOCK_SKEW_ACTION_ERROR, false},
+	{"wait", LR_CLOCK_SKEW_ACTION_WAIT, false},
+	{NULL, 0, false}
+};
+
+/* GUCs */
+int			max_logical_rep_clock_skew = LR_CLOCK_SKEW_DEFAULT;
+int			max_logical_rep_clock_skew_action = LR_CLOCK_SKEW_ACTION_ERROR;
+int			max_logical_rep_clock_skew_wait = 300;	/* 5 mins */
+
 /*
  * We enable skipping all data modification changes (INSERT, UPDATE, etc.) for
  * the subscription if the remote transaction's finish LSN matches the subskiplsn.
@@ -985,6 +999,95 @@ slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot,
 	ExecStoreVirtualTuple(slot);
 }
 
+/*
+ * Manage clock skew between nodes.
+ *
+ * It checks if the remote timestamp is ahead of the local clock
+ * and if the difference exceeds max_logical_rep_clock_skew, it performs
+ * the action specified by the max_logical_rep_clock_skew_action.
+ */
+static void
+manage_clock_skew(TimestampTz origin_timestamp)
+{
+	TimestampTz current;
+	TimestampTz delayUntil;
+	long		msecs;
+	int			rc;
+
+	/* nothing to do if no max clock skew configured */
+	if (max_logical_rep_clock_skew == LR_CLOCK_SKEW_DEFAULT)
+		return;
+
+	current = GetCurrentTimestamp();
+
+	/*
+	 * If the timestamp of the currently replayed transaction is in the future
+	 * compared to the current time on the subscriber and the difference is
+	 * larger than max_logical_rep_clock_skew, then perform the action
+	 * specified by the max_logical_rep_clock_skew_action setting.
+	 */
+	if (origin_timestamp > current &&
+		TimestampDifferenceExceeds(current, origin_timestamp,
+								   max_logical_rep_clock_skew * 1000))
+	{
+		if (max_logical_rep_clock_skew_action == LR_CLOCK_SKEW_ACTION_ERROR)
+			ereport(ERROR,
+					(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+					 errmsg_internal("clock skew exceeds max_logical_rep_clock_skew (%d seconds)",
+									 max_logical_rep_clock_skew)));
+
+		/* Perform the wait */
+		while (true)
+		{
+			delayUntil =
+				TimestampTzMinusSeconds(origin_timestamp,
+										max_logical_rep_clock_skew);
+
+			/* Exit without waiting if it's already past 'delayUntil' time */
+			msecs = TimestampDifferenceMilliseconds(GetCurrentTimestamp(),
+													delayUntil);
+			if (msecs <= 0)
+				break;
+
+			/* The wait time should not exceed max_logical_rep_clock_skew_wait */
+			if (msecs > (max_logical_rep_clock_skew_wait * 1000L))
+				ereport(ERROR,
+						(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+						 errmsg_internal("clock skew wait time exceeds max_logical_rep_clock_skew_wait (%d seconds)",
+										 max_logical_rep_clock_skew_wait)));
+
+			elog(DEBUG2, "delaying apply for %ld milliseconds to manage clock skew",
+				 msecs);
+
+			/* Sleep until we are signaled or msecs have elapsed */
+			rc = WaitLatch(MyLatch,
+						   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+						   msecs,
+						   WAIT_EVENT_LOGICAL_CLOCK_SKEW);
+
+			/* Exit the loop if msecs have elapsed */
+			if (rc & WL_TIMEOUT)
+				break;
+
+			if (rc & WL_LATCH_SET)
+			{
+				ResetLatch(MyLatch);
+				CHECK_FOR_INTERRUPTS();
+			}
+
+			/*
+			 * This might change max_logical_rep_clock_skew and
+			 * max_logical_rep_clock_skew_wait.
+			 */
+			if (ConfigReloadPending)
+			{
+				ConfigReloadPending = false;
+				ProcessConfigFile(PGC_SIGHUP);
+			}
+		}
+	}
+}
+
 /*
  * Handle BEGIN message.
  */
@@ -1006,6 +1109,15 @@ apply_handle_begin(StringInfo s)
 	in_remote_transaction = true;
 
 	pgstat_report_activity(STATE_RUNNING, NULL);
+
+	/* Check if there is any clock skew and perform configured action */
+	manage_clock_skew(begin_data.committime);
+
+	/*
+	 * Capture the commit timestamp of the remote transaction for time based
+	 * conflict resolution purpose.
+	 */
+	replorigin_session_origin_timestamp = begin_data.committime;
 }
 
 /*
@@ -1063,6 +1175,9 @@ apply_handle_begin_prepare(StringInfo s)
 	in_remote_transaction = true;
 
 	pgstat_report_activity(STATE_RUNNING, NULL);
+
+	/* Check if there is any clock skew and perform configured action */
+	manage_clock_skew(begin_data.prepare_time);
 }
 
 /*
@@ -1318,7 +1433,8 @@ apply_handle_stream_prepare(StringInfo s)
 			 * spooled operations.
 			 */
 			apply_spooled_messages(MyLogicalRepWorker->stream_fileset,
-								   prepare_data.xid, prepare_data.prepare_lsn);
+								   prepare_data.xid, prepare_data.prepare_lsn,
+								   prepare_data.prepare_time);
 
 			/* Mark the transaction as prepared. */
 			apply_handle_prepare_internal(&prepare_data);
@@ -2023,7 +2139,8 @@ ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno,
  */
 void
 apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
-					   XLogRecPtr lsn)
+					   XLogRecPtr lsn,
+					   TimestampTz origin_timestamp)
 {
 	int			nchanges;
 	char		path[MAXPGPATH];
@@ -2076,6 +2193,16 @@ apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
 
 	end_replication_step();
 
+	/*
+	 * If origin_timestamp is provided by caller, then check clock skew with
+	 * respect to the passed time and take configured action.
+	 */
+	if (origin_timestamp)
+		manage_clock_skew(origin_timestamp);
+
+	/* Capture the timestamp (prepare or commit) of the remote transaction */
+	replorigin_session_origin_timestamp = origin_timestamp;
+
 	/*
 	 * Read the entries one by one and pass them through the same logic as in
 	 * apply_dispatch.
@@ -2181,7 +2308,8 @@ apply_handle_stream_commit(StringInfo s)
 			 * spooled operations.
 			 */
 			apply_spooled_messages(MyLogicalRepWorker->stream_fileset, xid,
-								   commit_data.commit_lsn);
+								   commit_data.commit_lsn,
+								   commit_data.committime);
 
 			apply_handle_commit_internal(&commit_data);
 
@@ -2507,7 +2635,7 @@ apply_handle_insert_internal(ApplyExecutionData *edata,
 	{
 		EPQState	epqstate;
 
-		EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
+		/* EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); */
 
 		/* Process and store remote tuple in the slot */
 		oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
@@ -2522,7 +2650,7 @@ apply_handle_insert_internal(ApplyExecutionData *edata,
 									 rel_entry->localindexoid, true,
 									 conflictslot);
 
-		EvalPlanQualEnd(&epqstate);
+		/* EvalPlanQualEnd(&epqstate); */
 		ExecDropSingleTupleTableSlot(conflictslot);
 	}
 
@@ -2738,7 +2866,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 +2940,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 +3093,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 +3123,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);
 
@@ -3190,7 +3320,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);
 
@@ -3233,7 +3363,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);
 
@@ -4765,6 +4896,7 @@ run_apply_worker()
 	TimeLineID	startpointTLI;
 	char	   *err;
 	bool		must_use_password;
+	char	   *replorigin_sysid;
 
 	slotname = MySubscription->slotname;
 
@@ -4805,10 +4937,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/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 8efb4044d6..0ebad6fcab 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -59,6 +59,7 @@ CHECKPOINTER_MAIN	"Waiting in main loop of checkpointer process."
 LOGICAL_APPLY_MAIN	"Waiting in main loop of logical replication apply process."
 LOGICAL_LAUNCHER_MAIN	"Waiting in main loop of logical replication launcher process."
 LOGICAL_PARALLEL_APPLY_MAIN	"Waiting in main loop of logical replication parallel apply process."
+LOGICAL_CLOCK_SKEW	"Waiting in apply-begin of logical replication apply process to bring clock skew in permissible range."
 RECOVERY_WAL_STREAM	"Waiting in main loop of startup process for WAL to arrive, during streaming recovery."
 REPLICATION_SLOTSYNC_MAIN	"Waiting in main loop of slot sync worker."
 REPLICATION_SLOTSYNC_SHUTDOWN	"Waiting for slot sync worker to shut down."
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 880d76aae0..4a4f54a0c7 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -68,6 +68,7 @@
 #include "postmaster/walsummarizer.h"
 #include "postmaster/walwriter.h"
 #include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
 #include "replication/slot.h"
 #include "replication/slotsync.h"
 #include "replication/syncrep.h"
@@ -482,6 +483,7 @@ extern const struct config_enum_entry archive_mode_options[];
 extern const struct config_enum_entry recovery_target_action_options[];
 extern const struct config_enum_entry wal_sync_method_options[];
 extern const struct config_enum_entry dynamic_shared_memory_options[];
+extern const struct config_enum_entry logical_rep_clock_skew_action_options[];
 
 /*
  * GUC option variables that are exported from this module
@@ -3714,6 +3716,33 @@ struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"max_logical_rep_clock_skew", PGC_SIGHUP, REPLICATION_SUBSCRIBERS,
+			gettext_noop("Sets maximum clock skew tolerance between logical "
+						 "replication nodes beyond which action configured "
+						 "in max_logical_rep_clock_skew_action is triggered."),
+			gettext_noop("-1 turns this check off."),
+			GUC_UNIT_S
+		},
+		&max_logical_rep_clock_skew,
+		LR_CLOCK_SKEW_DEFAULT, LR_CLOCK_SKEW_DEFAULT, INT_MAX,
+		NULL, NULL, NULL
+	},
+
+	{
+		{"max_logical_rep_clock_skew_wait", PGC_SIGHUP, REPLICATION_SUBSCRIBERS,
+			gettext_noop("Sets max limit on how long apply worker shall wait to "
+						 "bring clock skew within permissible range of max_logical_rep_clock_skew. "
+						 "If the computed wait time is more than this value, "
+						 "apply worker will error out without waiting."),
+			gettext_noop("0 turns this limit off."),
+			GUC_UNIT_S
+		},
+		&max_logical_rep_clock_skew_wait,
+		300, 0, 3600,
+		NULL, NULL, NULL
+	},
+
 	/* End-of-list marker */
 	{
 		{NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL
@@ -4991,6 +5020,17 @@ struct config_enum ConfigureNamesEnum[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"max_logical_rep_clock_skew_action", PGC_POSTMASTER, REPLICATION_SUBSCRIBERS,
+			gettext_noop("Sets the action to perform if a clock skew higher "
+						 "than max_logical_rep_clock_skew is detected."),
+			NULL
+		},
+		&max_logical_rep_clock_skew_action,
+		LR_CLOCK_SKEW_ACTION_ERROR, logical_rep_clock_skew_action_options,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"track_functions", PGC_SUSET, STATS_CUMULATIVE,
 			gettext_noop("Collects function-level statistics on database activity."),
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 667e0dc40a..6424432362 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -383,7 +383,14 @@
 					# (change requires restart)
 #max_sync_workers_per_subscription = 2	# taken from max_logical_replication_workers
 #max_parallel_apply_workers_per_subscription = 2	# taken from max_logical_replication_workers
-
+#max_logical_rep_clock_skew = -1	# maximum clock skew tolerance between logical
+					# replication nodes beyond which action configured in
+					# 'max_logical_rep_clock_skew_action' is triggered.
+#max_logical_rep_clock_skew_action = error # error or wait
+					   # (change requires restart)
+#max_logical_rep_clock_skew_wait = 300 # max limit on how long apply worker
+					# shall wait to bring clock skew within permissible
+					# range of max_logical_rep_clock_skew.
 
 #------------------------------------------------------------------------------
 # QUERY TUNING
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index c0b7b91063..3182350e64 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"
@@ -62,6 +63,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,
 
@@ -102,15 +106,19 @@ extern void InitConflictIndexes(ResultRelInfo *relInfo);
 extern void SetSubConflictResolver(Oid subId, ConflictTypeResolver * resolvers, int max_types);
 extern void RemoveSubscriptionConflictById(Oid confid);
 extern void RemoveSubscriptionConflictBySubid(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 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/logicalworker.h b/src/include/replication/logicalworker.h
index a18d79d1b2..7cb03062ac 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -14,7 +14,25 @@
 
 #include <signal.h>
 
+/*
+ * The default for max_logical_rep_clock_skew is -1, which means ignore clock
+ * skew (the check is turned off).
+ */
+#define LR_CLOCK_SKEW_DEFAULT -1
+
+/*
+ * Worker Clock Skew Action.
+ */
+typedef enum
+{
+	LR_CLOCK_SKEW_ACTION_ERROR,
+	LR_CLOCK_SKEW_ACTION_WAIT,
+} LogicalRepClockSkewAction;
+
 extern PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending;
+extern PGDLLIMPORT int max_logical_rep_clock_skew;
+extern PGDLLIMPORT int max_logical_rep_clock_skew_action;
+extern PGDLLIMPORT int max_logical_rep_clock_skew_wait;
 
 extern void ApplyWorkerMain(Datum main_arg);
 extern void ParallelApplyWorkerMain(Datum main_arg);
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/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 9646261d7e..95b2a5286d 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -268,7 +268,7 @@ extern void stream_stop_internal(TransactionId xid);
 
 /* Common streaming function to apply all the spooled messages */
 extern void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
-								   XLogRecPtr lsn);
+								   XLogRecPtr lsn, TimestampTz origin_timestamp);
 
 extern void apply_dispatch(StringInfo s);
 
diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.h
index a6ce03ed46..53b828d89d 100644
--- a/src/include/utils/timestamp.h
+++ b/src/include/utils/timestamp.h
@@ -84,6 +84,7 @@ IntervalPGetDatum(const Interval *X)
 /* Macros for doing timestamp arithmetic without assuming timestamp's units */
 #define TimestampTzPlusMilliseconds(tz,ms) ((tz) + ((ms) * (int64) 1000))
 #define TimestampTzPlusSeconds(tz,s) ((tz) + ((s) * (int64) 1000000))
+#define TimestampTzMinusSeconds(tz,s) ((tz) - ((s) * (int64) 1000000))
 
 
 /* Set at postmaster start */
diff --git a/src/test/subscription/t/034_conflict_resolver.pl b/src/test/subscription/t/034_conflict_resolver.pl
index d9333dc962..6955e91a1f 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'
 ###############################################
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 9e951a9e6f..304a22e55a 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1565,6 +1565,7 @@ LogicalOutputPluginWriterPrepareWrite
 LogicalOutputPluginWriterUpdateProgress
 LogicalOutputPluginWriterWrite
 LogicalRepBeginData
+LogicalRepClockSkewAction
 LogicalRepCommitData
 LogicalRepCommitPreparedTxnData
 LogicalRepCtxStruct
-- 
2.34.1

