From e8fdbdfc76efe63630187e33349686a93d74562b Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Tue, 11 Aug 2026 14:08:25 +0800
Subject: [PATCH v23 05/12] Support dependency tracking via remote replica
 identity

This commit introduces a local hash table on the leader that is used
for tracking dependencies between transactions.

The hash contains the Replica Identity (RI) as a key and the remote XID that
modified the corresponding tuple. The hash entries are inserted when the leader
finds an RI from a replication message, and are removed in the
following cases:

1) When handling a subsequent change that modifies the same row, and the
   stored transaction has already committed.
2) When collecting flush progress of remote transactions, and the stored
   transaction is found to have committed.

When the leader sends replication changes to parallel workers, it checks whether
another transaction has already used the RI associated with the change. If so,
the leader marks the current transaction as dependent on that transaction and
notifies parallel workers via PA_MSG_XACT_DEPENDENCY to wait for that
transaction to finish.

Column values are compared and copied as raw bytes (with their wire length)
rather than as C strings, so that keys remain exact when a subscription uses
the binary option, where values may contain embedded NUL bytes.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Author: Hayato Kuroda <kuroda.hayato@fujitsu.com>
---
 .../replication/logical/applyparallelworker.c |  20 +
 src/backend/replication/logical/relation.c    |  31 +-
 src/backend/replication/logical/worker.c      | 641 +++++++++++++++++-
 src/include/replication/logicalrelation.h     |   9 +
 src/include/replication/worker_internal.h     |   1 +
 src/tools/pgindent/typedefs.list              |   3 +
 6 files changed, 701 insertions(+), 4 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 57366b1fbfb..a2f4ae48a82 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -2103,3 +2103,23 @@ pa_wait_for_depended_transaction(TransactionId xid)
 
 	elog(DEBUG1, "finish waiting for depended xid %u", xid);
 }
+
+/*
+ * Returns true if the given transaction is committed, false if it is still
+ * being applied in parallel. This can only be used by the leader apply worker.
+ */
+bool
+pa_transaction_committed(TransactionId xid)
+{
+	ParallelApplyWorkerEntry *entry;
+
+	Assert(am_leader_apply_worker());
+	Assert(TransactionIdIsValid(xid));
+	Assert(ParallelApplyTxnHash);
+
+	/* Find an entry for the requested transaction */
+	entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, NULL);
+
+	return !entry ||
+		   pa_get_xact_state(entry->winfo->shared) == PARALLEL_TRANS_FINISHED;
+}
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index d12b12c421a..8a0aaf4a217 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -224,7 +224,12 @@ logicalrep_relmap_update(LogicalRepRelation *remoterel)
 	if (found)
 		logicalrep_relmap_free_entry(entry);
 
-	memset(entry, 0, sizeof(LogicalRepRelMapEntry));
+	/*
+	 * Clear all fields except last_depended_xid, which tracks historical
+	 * dependency info that remains valid across schema changes.
+	 */
+	memset(entry, 0, found ? offsetof(LogicalRepRelMapEntry, last_depended_xid)
+		   : sizeof(LogicalRepRelMapEntry));
 
 	/* Make cached copy of the data */
 	oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
@@ -997,3 +1002,27 @@ FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel,
 
 	return InvalidOid;
 }
+
+/*
+ * Get the LogicalRepRelMapEntry corresponding to the given relid without
+ * opening the local relation.
+ */
+LogicalRepRelMapEntry *
+logicalrep_get_relentry(LogicalRepRelId remoteid)
+{
+	LogicalRepRelMapEntry *entry;
+	bool		found;
+
+	if (LogicalRepRelMap == NULL)
+		logicalrep_relmap_init();
+
+	/* Search for existing entry. */
+	entry = hash_search(LogicalRepRelMap, (void *) &remoteid,
+						HASH_FIND, &found);
+
+	if (!found)
+		elog(DEBUG1, "no relation map entry for remote relation ID %u",
+			 remoteid);
+
+	return entry;
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 77445115d11..5a93f5428a8 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -502,6 +502,8 @@ ErrorContextCallback *apply_error_context_stack = NULL;
 MemoryContext ApplyMessageContext = NULL;
 MemoryContext ApplyContext = NULL;
 
+static MemoryContext ParallelApplyContext = NULL;
+
 /* per stream context for streaming transactions */
 static MemoryContext LogicalStreamingContext = NULL;
 
@@ -576,6 +578,62 @@ typedef struct ApplySubXactData
 
 static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL};
 
+/* Hash table key for replica_identity_table */
+typedef struct ReplicaIdentityKey
+{
+	Oid			relid;
+	LogicalRepTupleData *data;
+} ReplicaIdentityKey;
+
+/* Hash table entry in replica_identity_table */
+typedef struct ReplicaIdentityEntry
+{
+	ReplicaIdentityKey *keydata;
+	TransactionId remote_xid;
+
+	/* needed for simplehash */
+	uint32		hash;
+	char		status;
+} ReplicaIdentityEntry;
+
+#include "common/hashfn.h"
+
+static uint32 hash_replica_identity(ReplicaIdentityKey *key);
+static bool hash_replica_identity_compare(ReplicaIdentityKey *a,
+										  ReplicaIdentityKey *b);
+
+/* Define parameters for replica identity hash table code generation. */
+#define SH_PREFIX		replica_identity
+#define SH_ELEMENT_TYPE	ReplicaIdentityEntry
+#define SH_KEY_TYPE		ReplicaIdentityKey *
+#define	SH_KEY			keydata
+#define SH_HASH_KEY(tb, key)	hash_replica_identity(key)
+#define SH_EQUAL(tb, a, b)		hash_replica_identity_compare(a, b)
+#define SH_STORE_HASH
+#define SH_GET_HASH(tb, a) (a)->hash
+#define	SH_SCOPE		static inline
+#define SH_DECLARE
+#define SH_DEFINE
+#include "lib/simplehash.h"
+
+#define REPLICA_IDENTITY_INITIAL_SIZE 128
+
+/*
+ * Hash table storing replica identity values for changes being applied in
+ * parallel, along with the latest transaction that modified each row.
+ *
+ * Entries are removed in the following cases:
+ *
+ * 1) When handling a subsequent change that modifies the same row, and the
+ *    stored transaction has already committed.
+ * 2) When collecting flush progress of remote transactions, and the stored
+ *    transaction is found to have committed.
+ *
+ * We use simplehash for its efficiency (see comments atop of simplehash.h), as
+ * this hash table is accessed for every change in a transaction.
+ */
+static replica_identity_hash *replica_identity_table = NULL;
+
 static inline void subxact_filename(char *path, Oid subid, TransactionId xid);
 static inline void changes_filename(char *path, Oid subid, TransactionId xid);
 
@@ -663,8 +721,574 @@ static void set_wal_receiver_timeout(void);
 
 static void on_exit_clear_xact_state(int code, Datum arg);
 
+static void send_internal_dependencies(ParallelApplyWorkerInfo *winfo, List *depends_on_xids);
+
 static void maintain_commit_order_dependency(ParallelApplyWorkerInfo *winfo);
 
+/*
+ * Compute the hash value for entries in the replica_identity_table.
+ */
+static uint32
+hash_replica_identity(ReplicaIdentityKey *key)
+{
+	int			i;
+	uint32		hashkey = 0;
+
+	hashkey = hash_combine(hashkey, hash_uint32(key->relid));
+
+	for (i = 0; i < key->data->ncols; i++)
+	{
+		uint32		hkey;
+
+		if (key->data->colstatus[i] == LOGICALREP_COLUMN_NULL)
+			continue;
+
+		hkey = hash_any((const unsigned char *) key->data->colvalues[i].data,
+						key->data->colvalues[i].len);
+		hashkey = hash_combine(hashkey, hkey);
+	}
+
+	return hashkey;
+}
+
+/*
+ * Compare two entries in the replica_identity_table.
+ *
+ * Column values are compared as raw bytes rather than as strings, because
+ * with the binary subscription option they are in the type's binary send
+ * format and may contain embedded NUL bytes.
+ */
+static bool
+hash_replica_identity_compare(ReplicaIdentityKey *a, ReplicaIdentityKey *b)
+{
+	if (a->relid != b->relid ||
+		a->data->ncols != b->data->ncols)
+		return false;
+
+	for (int i = 0; i < a->data->ncols; i++)
+	{
+		if (a->data->colstatus[i] != b->data->colstatus[i])
+			return false;
+
+		if (a->data->colstatus[i] == LOGICALREP_COLUMN_NULL)
+			continue;
+
+		if (a->data->colvalues[i].len != b->data->colvalues[i].len)
+			return false;
+
+		if (memcmp(a->data->colvalues[i].data, b->data->colvalues[i].data,
+				   a->data->colvalues[i].len) != 0)
+			return false;
+	}
+
+	return true;
+}
+
+/*
+ * Free resources associated with a replica identity key.
+ */
+static void
+free_replica_identity_key(ReplicaIdentityKey *key)
+{
+	Assert(key);
+
+	pfree(key->data->colvalues);
+	pfree(key->data->colstatus);
+	pfree(key->data);
+	pfree(key);
+}
+
+/*
+ * Delete an entry from replica_identity_table.
+ */
+static void
+delete_replica_identity_entry(ReplicaIdentityEntry *rientry)
+{
+	free_replica_identity_key(rientry->keydata);
+	replica_identity_delete_item(replica_identity_table, rientry);
+}
+
+/*
+ * Clean up hash table entries associated with the given transaction IDs.
+ */
+static void
+delete_replica_identity_entries_for_txns(List *committed_xids)
+{
+	replica_identity_iterator i;
+	ReplicaIdentityEntry *rientry;
+
+	if (!committed_xids)
+		return;
+
+	replica_identity_start_iterate(replica_identity_table, &i);
+	while ((rientry = replica_identity_iterate(replica_identity_table, &i)) != NULL)
+	{
+		if (!list_member_xid(committed_xids, rientry->remote_xid))
+			continue;
+
+		/* Clean up the hash entry for committed transaction */
+		delete_replica_identity_entry(rientry);
+	}
+}
+
+/*
+ * Check whether the given relation entry contains an uncommitted transaction
+ * that affected the whole table.
+ */
+static bool
+has_active_rel_dependency(LogicalRepRelMapEntry *relentry)
+{
+	if (!TransactionIdIsValid(relentry->last_depended_xid))
+		return false;
+
+	if (pa_transaction_committed(relentry->last_depended_xid))
+	{
+		relentry->last_depended_xid = InvalidTransactionId;
+		return false;
+	}
+
+	return true;
+}
+
+/*
+ * Check whether the given key dependency entry contains an uncommitted
+ * transaction. If cleanup is true, also clean up the entry if the transaction
+ * has already committed.
+ */
+static bool
+has_active_key_dependency(ReplicaIdentityEntry *rientry, bool cleanup)
+{
+	Assert(TransactionIdIsValid(rientry->remote_xid));
+
+	if (pa_transaction_committed(rientry->remote_xid))
+	{
+		if (cleanup)
+			delete_replica_identity_entry(rientry);
+
+		return false;
+	}
+
+	return true;
+}
+
+/*
+ * Append the given transaction ID to the dependency list if it is not already
+ * present.
+ */
+static void
+append_xid_dependency(TransactionId xid, List **depends_on_xids)
+{
+	Assert(TransactionIdIsValid(xid));
+
+	if (list_member_xid(*depends_on_xids, xid))
+		return;
+
+	*depends_on_xids = lappend_xid(*depends_on_xids, xid);
+}
+
+/*
+ * Check for dependencies on preceding transactions that modify the same key as
+ * the given tuple. Returns the dependent transactions in 'depends_on_xids'.
+ *
+ * Additionally, if new_depended_xid is valid, record the current change and the
+ * transaction as a new dependency for the replica identity key modification,
+ * allowing subsequent transactions that modify the same key to be dependent on
+ * it.
+ */
+static void
+check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
+							   TransactionId new_depended_xid,
+							   List **depends_on_xids)
+{
+	LogicalRepRelMapEntry *relentry;
+	LogicalRepTupleData *ridata;
+	ReplicaIdentityKey *rikey;
+	ReplicaIdentityEntry *rientry;
+	MemoryContext oldctx;
+	int			n_ri;
+	bool		found = false;
+
+	Assert(depends_on_xids);
+
+	/* Search for existing entry */
+	relentry = logicalrep_get_relentry(relid);
+
+	Assert(relentry);
+
+	/*
+	 * First check whether any previous transaction (other than the current one)
+	 * has affected the whole table e.g., truncate or schema change from
+	 * publisher.
+	 */
+	if (has_active_rel_dependency(relentry) &&
+		!TransactionIdEquals(relentry->last_depended_xid, new_depended_xid))
+	{
+		elog(DEBUG1, "found table-wide change affecting %u from %u",
+			 relid, relentry->last_depended_xid);
+
+		append_xid_dependency(relentry->last_depended_xid, depends_on_xids);
+	}
+
+	n_ri = bms_num_members(relentry->remoterel.attkeys);
+
+	/*
+	 * Return if there are no replica identity columns, indicating that the
+	 * remote relation has neither a replica identity key nor is marked as
+	 * replica identity full.
+	 */
+	if (!n_ri)
+		return;
+
+	oldctx = MemoryContextSwitchTo(ParallelApplyContext);
+
+	/* Allocate space for replica identity values */
+	ridata = palloc0_object(LogicalRepTupleData);
+	ridata->colvalues = palloc0_array(StringInfoData, n_ri);
+	ridata->colstatus = palloc0_array(char, n_ri);
+	ridata->ncols = n_ri;
+
+	for (int i_original = 0, i_ri = 0; i_original < original_data->ncols; i_original++)
+	{
+		StringInfo	original_colvalue = &original_data->colvalues[i_original];
+
+		if (!bms_is_member(i_original, relentry->remoterel.attkeys))
+			continue;
+
+		/*
+		 * LOGICALREP_COLUMN_UNCHANGED only indicates that a TOAST column in the
+		 * replica identity key hasn't changed. However, other columns may have
+		 * changed, so we still need to check the dependency for this column.
+		 *
+		 * Before calling this function, unchanged TOAST column values should
+		 * have been copied from the old tuple to the new tuple. So, we should
+		 * see the complete replica identity key value in original_data and
+		 * correctly check the dependency.
+		 */
+		Assert(original_data->colstatus[i_original] != LOGICALREP_COLUMN_UNCHANGED ||
+			   original_data->colvalues[i_original].len > 0);
+
+		/*
+		 * Copy the raw value bytes; the value may contain embedded NULs in
+		 * binary mode, so a plain string copy would silently truncate it.
+		 * NULL columns have no value to copy.
+		 */
+		if (original_data->colstatus[i_original] != LOGICALREP_COLUMN_NULL)
+		{
+			initStringInfoExt(&ridata->colvalues[i_ri],
+							  original_colvalue->len + 1);
+			appendBinaryStringInfo(&ridata->colvalues[i_ri],
+								   original_colvalue->data,
+								   original_colvalue->len);
+		}
+
+		ridata->colstatus[i_ri] = original_data->colstatus[i_original];
+		i_ri++;
+	}
+
+	rikey = palloc0_object(ReplicaIdentityKey);
+	rikey->relid = relid;
+	rikey->data = ridata;
+
+	MemoryContextSwitchTo(oldctx);
+
+	/*
+	 * The new xid could be invalid if the transaction will be applied by the
+	 * leader itself which means all the changes will be committed before
+	 * processing next transaction. In this case, we only need to check for
+	 * dependencies on preceding transactions, there is no need to record a new
+	 * dependency for subsequent transactions to wait on.
+	 */
+	if (!TransactionIdIsValid(new_depended_xid))
+	{
+		rientry = replica_identity_lookup(replica_identity_table, rikey);
+		free_replica_identity_key(rikey);
+
+		if (rientry && has_active_key_dependency(rientry, true))
+		{
+			elog(DEBUG1, "found conflicting replica identity change on table %u from %u",
+				 relid, rientry->remote_xid);
+
+			append_xid_dependency(rientry->remote_xid, depends_on_xids);
+		}
+
+		return;
+	}
+
+	/* Record a new dependency for subsequent transactions to wait on */
+	rientry = replica_identity_insert(replica_identity_table, rikey,
+									  &found);
+
+	/*
+	 * Release the key built to search the entry, if the entry already exists.
+	 */
+	if (found)
+	{
+		free_replica_identity_key(rikey);
+
+		/*
+		 * Append the dependency to the list if the current transaction was not
+		 * the lastest one to modify the key.
+		 */
+		if (has_active_key_dependency(rientry, false) &&
+			!TransactionIdEquals(rientry->remote_xid, new_depended_xid))
+		{
+			elog(DEBUG1, "found conflicting replica identity change on table %u from %u",
+				 relid, rientry->remote_xid);
+
+			append_xid_dependency(rientry->remote_xid, depends_on_xids);
+		}
+	}
+
+	/* Update the new depended xid into the entry */
+	rientry->remote_xid = new_depended_xid;
+}
+
+/*
+ * Check for preceding transactions (other than 'current_xid') that involve
+ * insert, delete, or update operations on the specified table, and return them
+ * in 'depends_on_xids'.
+ */
+static void
+find_all_dependencies_on_rel(LogicalRepRelId relid,
+							 TransactionId new_depended_xid,
+							 List **depends_on_xids)
+{
+	replica_identity_iterator i;
+	ReplicaIdentityEntry *rientry;
+
+	Assert(depends_on_xids);
+
+	replica_identity_start_iterate(replica_identity_table, &i);
+	while ((rientry = replica_identity_iterate(replica_identity_table, &i)) != NULL)
+	{
+		Assert(TransactionIdIsValid(rientry->remote_xid));
+
+		if (rientry->keydata->relid != relid)
+			continue;
+
+		/* Skip entries that do not have a valid and uncommitted transaction */
+		if (!has_active_key_dependency(rientry, true))
+			continue;
+
+		/* Skip self-dependency */
+		if (TransactionIdEquals(rientry->remote_xid, new_depended_xid))
+			continue;
+
+		elog(DEBUG1, "found conflicting change on table %u from %u",
+			 relid, rientry->remote_xid);
+
+		append_xid_dependency(rientry->remote_xid, depends_on_xids);
+	}
+}
+
+/*
+ * Check for any preceding transactions that affect the given table and returns
+ * them in 'depends_on_xids'.
+ *
+ * Additionally, if new_depended_xid is valid, record the current change and its
+ * transaction as a new table-level dependency, allowing subsequent transactions
+ * that modify the same table to be dependent on it.
+ */
+static void
+check_and_record_rel_dependency(LogicalRepRelId relid,
+								TransactionId new_depended_xid,
+								List **depends_on_xids)
+{
+	LogicalRepRelMapEntry *relentry;
+
+	Assert(depends_on_xids);
+
+	find_all_dependencies_on_rel(relid, new_depended_xid, depends_on_xids);
+
+	/* Search for existing entry */
+	relentry = logicalrep_get_relentry(relid);
+
+	/* First time seeing this relation, no entry exists. */
+	if (!relentry)
+		return;
+
+	/*
+	 * Check whether any previous transaction (other than the current one) has
+	 * affected the whole table e.g., truncate or schema change from publisher.
+	 */
+	if (has_active_rel_dependency(relentry) &&
+		!TransactionIdEquals(relentry->last_depended_xid, new_depended_xid))
+	{
+		elog(DEBUG1, "found table-wide change affecting %u from %u",
+			 relid, relentry->last_depended_xid);
+
+		append_xid_dependency(relentry->last_depended_xid, depends_on_xids);
+	}
+
+	if (TransactionIdIsValid(new_depended_xid))
+		relentry->last_depended_xid = new_depended_xid;
+}
+
+/*
+ * Check dependencies related to the current change by determining if the
+ * modification impacts the same row or table as another ongoing transaction.
+ *
+ * If a dependency on preceding transactions is found, this function instructs
+ * the appropriate worker (parallel apply or leader) to wait for those
+ * transactions to complete.
+ *
+ * Simultaneously, if the current change will be dispatched to a parallel apply
+ * worker (indicated by a valid new_depended_xid and a non-NULL winfo), it
+ * records a new dependency so that subsequent transactions will wait on it.
+ */
+static void
+handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
+							TransactionId new_depended_xid,
+							ParallelApplyWorkerInfo *winfo)
+{
+	LogicalRepRelId relid;
+	LogicalRepTupleData oldtup;
+	LogicalRepTupleData newtup;
+	LogicalRepRelation *rel;
+	List	   *depends_on_xids = NIL;
+	List	   *remote_relids;
+	bool		has_oldtup = false;
+	bool		cascade = false;
+	bool		restart_seqs = false;
+
+	/*
+	 * Parse the consume data using a local copy instead of directly consuming
+	 * the given remote change as the caller may also read the data from the
+	 * remote message.
+	 */
+	StringInfoData change = *s;
+
+	/* Only the leader checks dependencies and schedules the parallel apply */
+	if (!am_leader_apply_worker())
+		return;
+
+	if (!ParallelApplyContext)
+	{
+		Assert(!replica_identity_table);
+
+		/*
+		 * Create a permanent memory context for dependency information that
+		 * persists across transactions. Using a dedicated context makes memory
+		 * consumption more visible and easier to track, especially when
+		 * handling a large number of change entries for transactions being
+		 * applied in parallel.
+		 */
+		ParallelApplyContext = AllocSetContextCreate(ApplyContext,
+													 "ParallelApplyContext",
+													 ALLOCSET_DEFAULT_SIZES);
+
+		replica_identity_table = replica_identity_create(ParallelApplyContext,
+														 REPLICA_IDENTITY_INITIAL_SIZE,
+														 NULL);
+	}
+
+	switch (action)
+	{
+		case LOGICAL_REP_MSG_INSERT:
+			relid = logicalrep_read_insert(&change, &newtup);
+			check_and_record_ri_dependency(relid, &newtup, new_depended_xid,
+										   &depends_on_xids);
+			break;
+
+		case LOGICAL_REP_MSG_UPDATE:
+			relid = logicalrep_read_update(&change, &has_oldtup, &oldtup,
+										   &newtup);
+
+			if (has_oldtup)
+			{
+				check_and_record_ri_dependency(relid, &oldtup, new_depended_xid,
+											   &depends_on_xids);
+
+				/*
+				 * Copy unchanged column values from the old tuple to the new
+				 * tuple for replica identity dependency checking. See
+				 * check_and_record_ri_dependency() for details. Also adjust the
+				 * column status so that hash comparisons match correctly.
+				 */
+				for (int i = 0; i < oldtup.ncols; i++)
+				{
+					if (newtup.colstatus[i] == LOGICALREP_COLUMN_UNCHANGED)
+					{
+						newtup.colvalues[i] = oldtup.colvalues[i];
+						newtup.colstatus[i] = oldtup.colstatus[i];
+					}
+				}
+			}
+
+			check_and_record_ri_dependency(relid, &newtup, new_depended_xid,
+										   &depends_on_xids);
+			break;
+
+		case LOGICAL_REP_MSG_DELETE:
+			relid = logicalrep_read_delete(&change, &oldtup);
+			check_and_record_ri_dependency(relid, &oldtup, new_depended_xid,
+										   &depends_on_xids);
+			break;
+
+		case LOGICAL_REP_MSG_TRUNCATE:
+			remote_relids = logicalrep_read_truncate(&change, &cascade,
+													 &restart_seqs);
+
+			/*
+			 * Truncate affects all rows in a table, so the current
+			 * transaction should wait for all preceding transactions that
+			 * modified the same table.
+			 */
+			foreach_int(truncated_relid, remote_relids)
+				check_and_record_rel_dependency(truncated_relid,
+												new_depended_xid,
+												&depends_on_xids);
+
+			break;
+
+		case LOGICAL_REP_MSG_RELATION:
+			rel = logicalrep_read_rel(&change);
+
+			/*
+			 * The replica identity key could be changed, making existing
+			 * entries in the replica identity invalid. In this case, parallel
+			 * apply is not allowed on this specific table until all running
+			 * transactions that modified it have finished.
+			 */
+			check_and_record_rel_dependency(rel->remoteid, new_depended_xid,
+											&depends_on_xids);
+			break;
+
+		case LOGICAL_REP_MSG_TYPE:
+		case LOGICAL_REP_MSG_MESSAGE:
+
+			/*
+			 * Type updates accompany relation updates, so dependencies have
+			 * already been checked during relation updates. Logical messages
+			 * do not conflict with any changes, so they can be ignored.
+			 */
+			break;
+
+		default:
+			Assert(false);
+			break;
+	}
+
+	/* Return early if the current change has no dependencies. */
+	if (!depends_on_xids)
+		return;
+
+	/*
+	 * If the leader applies the transaction itself, start waiting for
+	 * transactions that the current change depends on to finish. Otherwise,
+	 * instruct the parallel apply worker to wait for them.
+	 */
+	if (winfo == NULL)
+	{
+		foreach_xid(xid, depends_on_xids)
+			pa_wait_for_depended_transaction(xid);
+	}
+	else
+	{
+		send_internal_dependencies(winfo, depends_on_xids);
+	}
+}
+
 /*
  * Form the origin name for the subscription.
  *
@@ -829,7 +1453,10 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
 
 	/* not in streaming mode */
 	if (apply_action == TRANS_LEADER_APPLY)
+	{
+		handle_dependency_on_change(action, s, InvalidTransactionId, winfo);
 		return false;
+	}
 
 	Assert(TransactionIdIsValid(stream_xid));
 
@@ -3969,6 +4596,7 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
 {
 	dlist_mutable_iter iter;
 	XLogRecPtr	local_flush = GetFlushRecPtr(NULL);
+	List	   *committed_xids = NIL;
 
 	*write = InvalidXLogRecPtr;
 	*flush = InvalidXLogRecPtr;
@@ -3983,9 +4611,13 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
 		 * worker. Transactions are committed in the publisher's commit order,
 		 * so no later transaction can have been committed or flushed either.
 		 */
-		if (TransactionIdIsValid(pos->pa_remote_xid) &&
-			!pa_get_last_commit_end(pos->pa_remote_xid, &pos->local_end))
-			break;
+		if (TransactionIdIsValid(pos->pa_remote_xid))
+		{
+			if (!pa_get_last_commit_end(pos->pa_remote_xid, &pos->local_end))
+				break;
+
+			committed_xids = lappend_xid(committed_xids, pos->pa_remote_xid);
+		}
 
 		*write = pos->remote_end;
 
@@ -4003,6 +4635,9 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
 		 */
 	}
 
+	/* cleanup the entries for committed transactions */
+	delete_replica_identity_entries_for_txns(committed_xids);
+
 	*have_pending_txes = !dlist_is_empty(&lsn_mapping);
 }
 
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 8fe6ec8ed78..621b1ffb706 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -37,6 +37,14 @@ typedef struct LogicalRepRelMapEntry
 	/* Sync state. */
 	char		state;
 	XLogRecPtr	statelsn;
+
+	/*
+	 * The last remote transaction that modified the relation's schema or
+	 * truncated the relation. Used in dependency tracking to ensure subsequent
+	 * transactions modifying the same table wait for this transaction to finish
+	 * applying (see check_dependency_on_rel).
+	 */
+	TransactionId last_depended_xid;
 } LogicalRepRelMapEntry;
 
 extern void logicalrep_relmap_update(LogicalRepRelation *remoterel);
@@ -52,5 +60,6 @@ extern bool IsIndexUsableForReplicaIdentityFull(Relation idxrel, AttrMap *attrma
 extern Oid	GetRelationIdentityOrPK(Relation rel);
 extern void logicalrep_write_all_internal_rels(StringInfo out, int *num_rels);
 extern void logicalrep_write_one_internal_rel(StringInfo out, LogicalRepRelation *rel);
+extern LogicalRepRelMapEntry *logicalrep_get_relentry(LogicalRepRelId remoteid);
 
 #endif							/* LOGICALRELATION_H */
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 471e37f1382..d895fc0284d 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -385,6 +385,7 @@ extern void pa_distribute_remote_rel_to_workers(LogicalRepRelation *rel);
 extern void pa_bind_flush_position(TransactionId xid, XLogRecPtr *local_end);
 extern bool pa_get_last_commit_end(TransactionId xid, XLogRecPtr *local_end);
 extern void pa_wait_for_depended_transaction(TransactionId xid);
+extern bool pa_transaction_committed(TransactionId xid);
 
 #define isParallelApplyWorker(worker) ((worker)->in_use && \
 									   (worker)->type == WORKERTYPE_PARALLEL_APPLY)
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index d19b46e16f6..0461e9b51b1 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2679,6 +2679,8 @@ ReplOriginXactState
 ReplaceVarsFromTargetList_context
 ReplaceVarsNoMatchOption
 ReplaceWrapOption
+ReplicaIdentityEntry
+ReplicaIdentityKey
 ReplicaIdentityStmt
 ReplicationKind
 ReplicationSlot
@@ -2690,6 +2692,7 @@ ReplicationSlotPersistentData
 ReplicationState
 ReplicationStateCtl
 ReplicationStateOnDisk
+replica_identity_hash
 ResTarget
 ReservoirState
 ReservoirStateData
-- 
2.43.0

