From 0b6da03293343e390e31967321eb1a810e46be7c Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Thu, 9 Jul 2026 15:43:06 +0800
Subject: [PATCH v22 8/9] Support dependency tracking via local unique indexes

This patch tracks dependencies on transactions that modify the same local unique
key. Even if the replica identity keys differ, unique indexes can still cause
conflicts. This is necessary to prevent unexpected errors. For example:

  TX-1: DELETE row (1,2) with replica identity key (1,2) and unique key (2)
  TX-2: INSERT row (3,2) with replica identity key (3,2) and unique key (2)

If applied in parallel, TX-2's INSERT could be applied before TX-1's DELETE,
leading to a unique index violation error.

We do not track dependencies for INSERT and UPDATE that conflict on a new
unique key value, since such conflicts would cause an error even in serial
mode. Instead, we only track dependencies involving old tuples (from DELETE
or UPDATE) and require INSERT and UPDATE transactions that target the same
unique key to wait for them.

Note that the old tuple of an UPDATE or DELETE may not include the unique key
column if that column is not part of the replica identity columns on the
publisher. In such cases, we only use the unique columns that are part of the
replica identity keys for dependency tracking, which may lead to false
positives. For example, consider a unique index defined as UNIQUE (a, b),
where only b is part of the replica identity keys:

  TX-1: DELETE row (1,2) TX-2: INSERT row (3,2)

If applied in parallel, both transactions will be treated as dependent
because they modify the same unique key value (b=2), even though they
actually modify different unique keys. This is acceptable because it is still
better than completely disallowing parallelism for these transactions.

In the worst case, if none of the unique index columns are part of the
replica identity keys, we treat all transactions that modify the same table
as dependent and disallow parallelism for that table.

XXX We could consider requesting the publisher to include unique key columns
in the old tuple of UPDATE or DELETE when they are not part of the replica
identity keys. This would reduce false positives, but would require changes
on the publisher side and increase disk (WAL size) and network data. For now,
we choose not to implement this. An alternative approach is to provide an
option to skip tracking dependencies on unique keys that are not part of the
replica identity keys. This could be useful for users who prefer higher
parallelism and experience few conflicts.

Note that the local unique key could change after dependency checking and
before applying the change. However, to centralize tracking and keep it
simple, we still perform this check only in the leader apply worker. This is
acceptable because in the worst case, the parallel worker will report an
error and restart the transaction using the latest index information.
---
 .../replication/logical/applyparallelworker.c |  50 +++
 src/backend/replication/logical/relation.c    | 179 ++++++++
 src/backend/replication/logical/worker.c      | 414 +++++++++++++-----
 src/backend/storage/lmgr/deadlock.c           |   1 -
 src/include/replication/logicalrelation.h     |  16 +
 src/test/subscription/t/050_parallel_apply.pl | 124 ++++++
 src/tools/pgindent/typedefs.list              |   2 +
 7 files changed, 673 insertions(+), 113 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 3e436b518bd..1f1a75b5019 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -174,6 +174,56 @@
  * both are allowed to apply in parallel, TX-2's DELETE could be applied before
  * TX-1's INSERT, resulting in a delete_missing conflict.
  *
+ * Beyond replica identity keys, we also track dependencies on transactions that
+ * modify the same local unique key. Even if the replica identity keys differ,
+ * unique indexes can still cause conflicts. This is necessary to prevent
+ * unexpected errors. For example:
+ *
+ *   TX-1: DELETE row (1,2) with replica identity key (1,2) and unique key (2)
+ *   TX-2: INSERT row (3,2) with replica identity key (3,2) and unique key (2)
+ *
+ * If applied in parallel, TX-2's INSERT could be applied before TX-1's DELETE,
+ * leading to a unique index violation error.
+ *
+ * We do not track dependencies for INSERT and UPDATE that conflict on a new
+ * unique key value, since such conflicts would cause an error even in serial
+ * mode. Instead, we only track dependencies involving old tuples (from DELETE
+ * or UPDATE) and require INSERT and UPDATE transactions that target the same
+ * unique key to wait for them.
+ *
+ * Note that the old tuple of an UPDATE or DELETE may not include the unique key
+ * column if that column is not part of the replica identity columns on the
+ * publisher. In such cases, we only use the unique columns that are part of the
+ * replica identity keys for dependency tracking, which may lead to false
+ * positives. For example, consider a unique index defined as UNIQUE (a, b),
+ * where only b is part of the replica identity keys:
+ *
+ *   TX-1: DELETE row (1,2) TX-2: INSERT row (3,2)
+ *
+ * If applied in parallel, both transactions will be treated as dependent
+ * because they modify the same unique key value (b=2), even though they
+ * actually modify different unique keys. This is acceptable because it is still
+ * better than completely disallowing parallelism for these transactions.
+ *
+ * In the worst case, if none of the unique index columns are part of the
+ * replica identity keys, we treat all transactions that modify the same table
+ * as dependent and disallow parallelism for that table.
+ *
+ * XXX We could consider requesting the publisher to include unique key columns
+ * in the old tuple of UPDATE or DELETE when they are not part of the replica
+ * identity keys. This would reduce false positives, but would require changes
+ * on the publisher side and increase disk (WAL size) and network data. For now,
+ * we choose not to implement this. An alternative approach is to provide an
+ * option to skip tracking dependencies on unique keys that are not part of the
+ * replica identity keys. This could be useful for users who prefer higher
+ * parallelism and experience few conflicts.
+ *
+ * Note that the local unique key could change after dependency checking and
+ * before applying the change. However, to centralize tracking and keep it
+ * simple, we still perform this check only in the leader apply worker. This is
+ * acceptable because in the worst case, the parallel worker will report an
+ * error and restart the transaction using the latest index information.
+ *
  * Commit order
  * ------------
  * We preserve publisher commit order for all transactions for two reasons:
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index 6d3e1827ae2..1572990ee18 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -88,6 +88,7 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid)
 			{
 				entry->localrelvalid = false;
 				entry->parallel_safety_valid = false;
+				entry->local_unique_indexes_valid = false;
 				hash_seq_term(&status);
 				break;
 			}
@@ -104,6 +105,7 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid)
 		{
 			entry->localrelvalid = false;
 			entry->parallel_safety_valid = false;
+			entry->local_unique_indexes_valid = false;
 		}
 	}
 }
@@ -135,6 +137,21 @@ logicalrep_relmap_init(void)
 								  (Datum) 0);
 }
 
+/*
+ * Release local index list
+ */
+static void
+free_local_unique_indexes(LogicalRepRelMapEntry *entry)
+{
+	Assert(am_leader_apply_worker());
+
+	foreach_ptr(LogicalRepSubUniqueIndex, idxinfo, entry->local_unique_indexes)
+		bms_free(idxinfo->indexkeys);
+
+	list_free_deep(entry->local_unique_indexes);
+	entry->local_unique_indexes = NIL;
+}
+
 /*
  * Free the entry of a relation map cache.
  */
@@ -162,6 +179,9 @@ logicalrep_relmap_free_entry(LogicalRepRelMapEntry *entry)
 
 	if (entry->attrmap)
 		free_attrmap(entry->attrmap);
+
+	if (entry->local_unique_indexes != NIL)
+		free_local_unique_indexes(entry);
 }
 
 /*
@@ -218,6 +238,13 @@ logicalrep_relmap_update(LogicalRepRelation *remoterel)
 		(remoterel->relkind == 0) ? RELKIND_RELATION : remoterel->relkind;
 
 	entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
+
+	/*
+	 * Rebuild the key info using the latest replica identity, which may have
+	 * changed.
+	 */
+	entry->local_unique_indexes_valid = false;
+
 	MemoryContextSwitchTo(oldctx);
 }
 
@@ -404,6 +431,7 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 			/* Table was renamed or dropped. */
 			entry->localrelvalid = false;
 			entry->parallel_safety_valid = false;
+			entry->local_unique_indexes_valid = false;
 		}
 		else if (!entry->localrelvalid)
 		{
@@ -614,6 +642,157 @@ logicalrep_rel_check_parallel_safety(LogicalRepRelMapEntry *entry)
 	entry->parallel_safety_valid = true;
 }
 
+/*
+ * Collect all local unique indexes that can be used for dependency tracking
+ *
+ * This function collects all types of unique indexes, including those with
+ * index expressions and partial indexes. However, to avoid the overhead and
+ * complexity of executing expressions, we do not evaluate them during
+ * dependency tracking.
+ *
+ * For indexes with expressions, only the non-expression columns are recorded in
+ * the bitmap. The dependency tracking function will use only these columns,
+ * which may lead to false dependency detection. For example, consider a unique
+ * index defined as UNIQUE (a, func(b)), where b is an expression column. Rows
+ * (1, 2) and (1, 3) will be treated as dependent even though they are not. This
+ * is acceptable, as it is still better than disabling parallelism for all
+ * relations that have expression indexes.
+ *
+ * Similarly, partial indexes may also cause false dependencies due to predicate
+ * expressions. For the same reason, we consider this acceptable as well.
+ *
+ * To avoid redundant dependency tracking, indexes whose key columns are the
+ * same as, or a superset of, the replica identity key columns are skipped,
+ * since tracking the replica identity keys already covers their scope.
+ *
+ * Columns not in the replica identity key are excluded from the unique column
+ * set. Since the old tuple of an UPDATE or DELETE contains only replica
+ * identity key columns, any other columns would be missing and thus unavailable
+ * for dependency tracking.
+ */
+void
+logicalrep_build_dependent_unique_indexes(LogicalRepRelMapEntry *entry)
+{
+	List	   *idxlist;
+
+	if (entry->local_unique_indexes_valid)
+		return;
+
+	free_local_unique_indexes(entry);
+
+	/*
+	 * XXX For partitioned tables, we must collect unique indexes from leaf
+	 * partitions, which are the actual replication targets. This is because
+	 * leaf partitions can have unique indexes that are not present on the
+	 * partitioned table, and those indexes can be used for dependency tracking.
+	 * However, collecting unique indexes from leaf partitions requires building
+	 * the tuple, and executing partition pruning expressions, which could be
+	 * expensive for each change on a partitioned table. For now, we skip
+	 * collecting local unique indexes for partitioned tables and create a dummy
+	 * entry, ensuring that changes on partitioned tables are not applied in
+	 * parallel.
+	 */
+	if (entry->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+	{
+		MemoryContext oldctx;
+		LogicalRepSubUniqueIndex *indexinfo;
+
+		oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
+		indexinfo = palloc(sizeof(LogicalRepSubUniqueIndex));
+		indexinfo->indexoid = InvalidOid;
+		indexinfo->indexkeys = NULL;
+		indexinfo->nulls_distinct = false;
+		entry->local_unique_indexes = lappend(entry->local_unique_indexes,
+											  indexinfo);
+		MemoryContextSwitchTo(oldctx);
+
+		entry->local_unique_indexes_valid = true;
+
+		return;
+	}
+
+	idxlist = RelationGetIndexList(entry->localrel);
+
+	/* Iterate indexes to list all usable indexes */
+	foreach_oid(idxoid, idxlist)
+	{
+		Relation	idxrel;
+		int			indnkeys;
+		AttrMap    *attrmap;
+		MemoryContext oldctx;
+		LogicalRepSubUniqueIndex *indexinfo;
+		Bitmapset  *indexkeys = NULL;
+		bool		nulls_distinct;
+
+		idxrel = index_open(idxoid, AccessShareLock);
+
+		/* Only unique indexes are considered */
+		if (!idxrel->rd_index->indisunique)
+		{
+			index_close(idxrel, AccessShareLock);
+			continue;
+		}
+
+		indnkeys = idxrel->rd_index->indnkeyatts;
+		nulls_distinct = !idxrel->rd_index->indnullsnotdistinct;
+		attrmap = entry->attrmap;
+
+		Assert(indnkeys);
+
+		/* Seek each attributes and add to a Bitmap */
+		for (int i = 0; i < indnkeys; i++)
+		{
+			AttrNumber	localcol = idxrel->rd_index->indkey.values[i];
+			AttrNumber	remotecol;
+
+			/* Skip expression */
+			if (!AttributeNumberIsValid(localcol))
+				continue;
+
+			remotecol = attrmap->attnums[AttrNumberGetAttrOffset(localcol)];
+
+			/* Skip if the column does not exist on publisher node */
+			if (remotecol < 0)
+				continue;
+
+			/* Skip columns that are not part of the replica identity key */
+			if (!bms_is_member(remotecol, entry->remoterel.attkeys))
+				continue;
+
+			/* Checks are passed, remember the attribute */
+			indexkeys = bms_add_member(indexkeys, remotecol);
+		}
+
+		index_close(idxrel, AccessShareLock);
+
+		/*
+		 * Skip indexes whose key columns are a superset of the replica identity
+		 * key.
+		 */
+		if (bms_equal(entry->remoterel.attkeys, indexkeys) ||
+			bms_is_subset(entry->remoterel.attkeys, indexkeys))
+		{
+			bms_free(indexkeys);
+			continue;
+		}
+
+		oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
+		indexinfo = palloc(sizeof(LogicalRepSubUniqueIndex));
+		indexinfo->indexoid = idxoid;
+		indexinfo->indexkeys = bms_copy(indexkeys);
+		indexinfo->nulls_distinct = nulls_distinct;
+		entry->local_unique_indexes = lappend(entry->local_unique_indexes,
+											  indexinfo);
+		MemoryContextSwitchTo(oldctx);
+
+		bms_free(indexkeys);
+	}
+
+	list_free(idxlist);
+
+	entry->local_unique_indexes_valid = true;
+}
+
 /*
  * Partition cache: look up partition LogicalRepRelMapEntry's
  *
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 18d71ceabe6..3864c95f930 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -578,10 +578,20 @@ typedef struct ApplySubXactData
 
 static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL};
 
+/*
+ * Type of key used for dependency tracking.
+ */
+typedef enum LogicalRepKeyKind
+{
+	LOGICALREP_KEY_REPLICA_IDENTITY,
+	LOGICALREP_KEY_LOCAL_UNIQUE
+} LogicalRepKeyKind;
+
 /* Hash table key for replica_identity_table */
 typedef struct ReplicaIdentityKey
 {
 	Oid			relid;
+	LogicalRepKeyKind kind;
 	LogicalRepTupleData *data;
 } ReplicaIdentityKey;
 
@@ -761,7 +771,8 @@ static bool
 hash_replica_identity_compare(ReplicaIdentityKey *a, ReplicaIdentityKey *b)
 {
 	if (a->relid != b->relid ||
-		a->data->ncols != b->data->ncols)
+		a->data->ncols != b->data->ncols ||
+		a->kind != b->kind)
 		return false;
 
 	for (int i = 0; i < a->data->ncols; i++)
@@ -792,8 +803,12 @@ free_replica_identity_key(ReplicaIdentityKey *key)
 {
 	Assert(key);
 
-	pfree(key->data->colvalues);
-	pfree(key->data->colstatus);
+	if (key->data->colvalues)
+		pfree(key->data->colvalues);
+
+	if (key->data->colstatus)
+		pfree(key->data->colstatus);
+
 	pfree(key->data);
 	pfree(key);
 }
@@ -935,6 +950,79 @@ append_xid_dependency(TransactionId xid, List **depends_on_xids)
 	*depends_on_xids = lappend_xid(*depends_on_xids, xid);
 }
 
+/*
+ * Common function for checking dependency by using the key. Used by both
+ * check_and_record_ri_dependency and check_and_record_local_key_dependency.
+ *
+ * Check whether the given key has an active dependency. If new_depended_xid is
+ * valid, also records a new dependency for that transaction.
+ *
+ * Return the existing transaction ID if an active dependency exists for the
+ * key; otherwise returns InvalidTransactionId.
+ */
+static TransactionId
+check_and_record_key_dependency(ReplicaIdentityKey *key,
+								TransactionId new_depended_xid)
+{
+	TransactionId existing_xid = InvalidTransactionId;
+	ReplicaIdentityEntry *rientry;
+	bool		found = false;
+
+	/*
+	 * 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, key);
+
+		if (rientry && has_active_key_dependency(rientry, true))
+		{
+			elog(DEBUG1,
+				 key->kind == LOGICALREP_KEY_REPLICA_IDENTITY ?
+				 "found conflicting replica identity change on table %u from %u" :
+				 "found conflicting local unique key change on table %u from %u",
+				 key->relid, rientry->remote_xid);
+
+			existing_xid = rientry->remote_xid;
+		}
+
+		free_replica_identity_key(key);
+
+		return existing_xid;
+	}
+
+	/* Record a new dependency for subsequent transactions to wait on */
+	rientry = replica_identity_insert(replica_identity_table, key,
+									  &found);
+
+	/*
+	 * Release the key built to search the entry, if the entry already exists.
+	 */
+	if (found)
+	{
+		if (has_active_key_dependency(rientry, false))
+		{
+			elog(DEBUG1,
+				 key->kind == LOGICALREP_KEY_REPLICA_IDENTITY ?
+				 "found conflicting replica identity change on table %u from %u" :
+				 "found conflicting local unique key change on table %u from %u",
+				 key->relid, rientry->remote_xid);
+
+			existing_xid = rientry->remote_xid;
+		}
+
+		free_replica_identity_key(key);
+	}
+
+	rientry->remote_xid = new_depended_xid;
+
+	return existing_xid;
+}
+
 /*
  * Check if any of the key columns have NULL values.
  */
@@ -952,81 +1040,34 @@ has_null_key_values(LogicalRepTupleData *data, Bitmapset *indexkeys)
 }
 
 /*
- * 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.
+ * Build a hash key for replica_identity_table using the given relation and
+ * tuple data, restricted to the specified key columns.
  */
-static void
-check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
-							   TransactionId new_depended_xid,
-							   List **depends_on_xids)
+static ReplicaIdentityKey *
+build_replica_identity_key(Oid relid, LogicalRepTupleData *original_data,
+						   Bitmapset *keycols)
 {
-	LogicalRepRelMapEntry *relentry;
-	LogicalRepTupleData *ridata;
-	ReplicaIdentityKey *rikey;
-	ReplicaIdentityEntry *rientry;
+	LogicalRepTupleData *keydata;
+	ReplicaIdentityKey *key;
 	MemoryContext oldctx;
-	int			n_ri;
-	bool		found = false;
-
-	Assert(depends_on_xids);
+	int		nkeycols = bms_num_members(keycols);
 
-	/* Search for existing entry */
-	relentry = logicalrep_get_relentry(relid);
+	oldctx = MemoryContextSwitchTo(ApplyContext);
 
-	Assert(relentry);
+	/* Allocate space for replica identity values */
+	keydata = palloc0_object(LogicalRepTupleData);
 
-	/*
-	 * 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))
+	if (nkeycols)
 	{
-		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);
+		keydata->colvalues = palloc0_array(StringInfoData, nkeycols);
+		keydata->colstatus = palloc0_array(char, nkeycols);
 	}
 
-	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;
-
-	/*
-	 * For replica identity FULL, the tuple contains all columns including
-	 * NULLs, so we always check for existing dependencies.
-	 *
-	 * For other cases, NULL in the new tuple means the replica identity key
-	 * hasn't changed, so no new dependency needs to be recorded. The dependency
-	 * should have been recorded when processing the old tuple.
-	 */
-	if (relentry->remoterel.replident != REPLICA_IDENTITY_FULL &&
-		has_null_key_values(original_data, relentry->remoterel.attkeys))
-		return;
-
-	oldctx = MemoryContextSwitchTo(ApplyContext);
+	keydata->ncols = nkeycols;
 
-	/* 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++)
+	for (int i = 0, i_key = 0; i < original_data->ncols; i++)
 	{
-		if (!bms_is_member(i_original, relentry->remoterel.attkeys))
+		if (!bms_is_member(i, keycols))
 			continue;
 
 		/*
@@ -1039,77 +1080,206 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 		 * 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);
+		Assert(original_data->colstatus[i] != LOGICALREP_COLUMN_UNCHANGED ||
+			   original_data->colvalues[i].len > 0);
 
-		if (original_data->colstatus[i_original] != LOGICALREP_COLUMN_NULL)
+		if (original_data->colstatus[i] != LOGICALREP_COLUMN_NULL)
 		{
-			StringInfo	original_colvalue = &original_data->colvalues[i_original];
+			StringInfo	original_colvalue = &original_data->colvalues[i];
 
-			initStringInfoExt(&ridata->colvalues[i_ri], original_colvalue->len + 1);
-			appendStringInfoString(&ridata->colvalues[i_ri], original_colvalue->data);
+			initStringInfoExt(&keydata->colvalues[i_key], original_colvalue->len + 1);
+			appendStringInfoString(&keydata->colvalues[i_key], original_colvalue->data);
 		}
 
-		ridata->colstatus[i_ri] = original_data->colstatus[i_original];
-		i_ri++;
+		keydata->colstatus[i_key] = original_data->colstatus[i];
+		i_key++;
 	}
 
-	rikey = palloc0_object(ReplicaIdentityKey);
-	rikey->relid = relid;
-	rikey->data = ridata;
+	key = palloc0_object(ReplicaIdentityKey);
+	key->relid = relid;
+	key->data = keydata;
 
 	MemoryContextSwitchTo(oldctx);
 
+	return key;
+}
+
+/*
+ * Build dependency key information for the given relation entry if not already
+ * collected.
+ *
+ * See logicalrep_build_dependent_unique_indexes() for details.
+ */
+static void
+build_local_dependent_key_info(LogicalRepRelMapEntry *relentry)
+{
+	bool		needs_start;
+
+	if (relentry->local_unique_indexes_valid)
+		return;
+
 	/*
-	 * 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.
+	 * Gather information for local indexes if not yet. We require to be in a
+	 * transaction state to collect indexes info from system catalogs.
 	 */
-	if (!TransactionIdIsValid(new_depended_xid))
-	{
-		rientry = replica_identity_lookup(replica_identity_table, rikey);
-		free_replica_identity_key(rikey);
+	needs_start = !IsTransactionState();
 
-		if (rientry && has_active_key_dependency(rientry, true))
-		{
-			elog(DEBUG1, "found conflicting replica identity change on table %u from %u",
-				 relid, rientry->remote_xid);
+	if (needs_start)
+		StartTransactionCommand();
 
-			append_xid_dependency(rientry->remote_xid, depends_on_xids);
-		}
+	relentry = logicalrep_rel_open(relentry->remoterel.remoteid, AccessShareLock);
 
-		return;
-	}
+	logicalrep_build_dependent_unique_indexes(relentry);
 
-	/* Record a new dependency for subsequent transactions to wait on */
-	rientry = replica_identity_insert(replica_identity_table, rikey,
-									  &found);
+	logicalrep_rel_close(relentry, AccessShareLock);
 
-	/*
-	 * Release the key built to search the entry, if the entry already exists.
-	 */
-	if (found)
+	if (needs_start)
+		CommitTransactionCommand();
+}
+
+/*
+ * Mostly same as check_and_record_ri_dependency() but for local unique indexes.
+ *
+ * See the comments in applyparallelworker.c for details on why tracking these
+ * dependencies is necessary.
+ */
+static void
+check_and_record_local_key_dependency(Oid relid,
+									  LogicalRepTupleData *original_data,
+									  bool old_tuple,
+									  TransactionId new_depended_xid,
+									  List **depends_on_xids)
+{
+	LogicalRepRelMapEntry *relentry;
+	ReplicaIdentityKey *rikey;
+
+	Assert(depends_on_xids);
+
+	/* Search for existing entry */
+	relentry = logicalrep_get_relentry(relid);
+
+	Assert(relentry);
+
+	build_local_dependent_key_info(relentry);
+
+	foreach_ptr(LogicalRepSubUniqueIndex, idxinfo, relentry->local_unique_indexes)
 	{
-		free_replica_identity_key(rikey);
+		/*
+		 * NULL values in the new tuple represent true NULLs in a unique index.
+		 * If NULLs are treated as distinct (nulls_distinct = true), they never
+		 * cause conflicts. Therefore, we can skip dependency checking if any
+		 * key column is NULL in this case.
+		 *
+		 * However, for old tuples in UPDATE or DELETE operations, a NULL key
+		 * simply indicate the column lies outside the replica identity key
+		 * rather than a true NULL. In such cases, the remote old tuple could
+		 * still conflict with a local tuple, so we must not skip the check.
+		 */
+		if (!old_tuple && idxinfo->nulls_distinct &&
+			has_null_key_values(original_data, idxinfo->indexkeys))
+			continue;
 
 		/*
-		 * Append the dependency to the list if the current transaction was not
-		 * the lastest one to modify the key.
+		 * Old tuples of unique keys do not conflict with any preceding
+		 * transaction (see the comments in applyparallelworker.c for details on
+		 * conflicting cases). When we don't need to record a new dependency, we
+		 * can skip processing this index entirely.
 		 */
-		if (has_active_key_dependency(rientry, false) &&
-			!TransactionIdEquals(rientry->remote_xid, new_depended_xid))
+		if (old_tuple && !TransactionIdIsValid(new_depended_xid))
+			continue;
+
+		rikey = build_replica_identity_key(relid, original_data, idxinfo->indexkeys);
+		rikey->kind = LOGICALREP_KEY_LOCAL_UNIQUE;
+
+		/*
+		 * For old tuples, record a dependency for subsequent transactions to
+		 * wait on; no preceding transactions are added to the list.
+		 *
+		 * For new tuples in INSERT or UPDATE, check for existing key
+		 * dependencies and add any dependent transactions to the list.
+		 */
+		if (old_tuple)
+		{
+			(void) check_and_record_key_dependency(rikey, new_depended_xid);
+		}
+		else
 		{
-			elog(DEBUG1, "found conflicting replica identity change on table %u from %u",
-				 relid, rientry->remote_xid);
+			TransactionId	xid;
+
+			xid = check_and_record_key_dependency(rikey, InvalidTransactionId);
 
-			append_xid_dependency(rientry->remote_xid, depends_on_xids);
+			if (TransactionIdIsValid(xid) &&
+				!TransactionIdEquals(xid, new_depended_xid))
+				append_xid_dependency(xid, depends_on_xids);
 		}
 	}
+}
 
-	/* Update the new depended xid into the entry */
-	rientry->remote_xid = new_depended_xid;
+/*
+ * Check for dependencies on preceding transactions that modify the same key.
+ * Returns the dependent transactions in 'depends_on_xids'.
+ *
+ * Additionally, if new_depended_xid is valid, record it as a 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;
+	ReplicaIdentityKey *rikey;
+	TransactionId xid;
+
+	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))
+		append_xid_dependency(relentry->last_depended_xid, depends_on_xids);
+
+	/*
+	 * 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 (!bms_num_members(relentry->remoterel.attkeys))
+		return;
+
+	/*
+	 * For replica identity FULL, the tuple contains all columns including
+	 * NULLs, so we always check for existing dependencies.
+	 *
+	 * For other cases, NULL in the new tuple means the replica identity key
+	 * hasn't changed, so no new dependency needs to be recorded. The dependency
+	 * should have been recorded when processing the old tuple.
+	 */
+	if (REPLICA_IDENTITY_FULL != relentry->remoterel.replident &&
+		has_null_key_values(original_data, relentry->remoterel.attkeys))
+		return;
+
+	rikey = build_replica_identity_key(relid, original_data, relentry->remoterel.attkeys);
+	rikey->kind = LOGICALREP_KEY_REPLICA_IDENTITY;
+
+	xid = check_and_record_key_dependency(rikey, new_depended_xid);
+
+	/*
+	 * Append the dependency to the list if the current transaction was not the
+	 * lastest one to modify the key.
+	 */
+	if (TransactionIdIsValid(xid) &&
+		!TransactionIdEquals(xid, new_depended_xid))
+		append_xid_dependency(xid, depends_on_xids);
 }
 
 /*
@@ -1308,6 +1478,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
 			relid = logicalrep_read_insert(&change, &newtup);
 			check_and_record_ri_dependency(relid, &newtup, new_depended_xid,
 										   &depends_on_xids);
+			check_and_record_local_key_dependency(relid, &newtup, false,
+												  new_depended_xid,
+												  &depends_on_xids);
 			break;
 
 		case LOGICAL_REP_MSG_UPDATE:
@@ -1337,12 +1510,29 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
 
 			check_and_record_ri_dependency(relid, &newtup, new_depended_xid,
 										   &depends_on_xids);
+
+			/*
+			 * Check the new tuple first to detect dependencies on preceding
+			 * transactions that modified the same key. If we processed the old
+			 * tuple first, it might update the same hash entry with the current
+			 * transaction ID, causing the new tuple check to miss any preceding
+			 * transaction.
+			 */
+			check_and_record_local_key_dependency(relid, &newtup, false,
+												  new_depended_xid,
+												  &depends_on_xids);
+			check_and_record_local_key_dependency(relid, &oldtup, true,
+												  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);
+			check_and_record_local_key_dependency(relid, &oldtup, true,
+												  new_depended_xid,
+												  &depends_on_xids);
 			break;
 
 		case LOGICAL_REP_MSG_TRUNCATE:
diff --git a/src/backend/storage/lmgr/deadlock.c b/src/backend/storage/lmgr/deadlock.c
index b8962d875b6..31bbfcf1971 100644
--- a/src/backend/storage/lmgr/deadlock.c
+++ b/src/backend/storage/lmgr/deadlock.c
@@ -33,7 +33,6 @@
 #include "storage/procnumber.h"
 #include "utils/memutils.h"
 
-
 /*
  * One edge in the waits-for graph.
  *
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 439243be742..ec4c9ece571 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -56,6 +56,10 @@ typedef struct LogicalRepRelMapEntry
 	 */
 	TransactionId last_depended_xid;
 
+	/* Local unique indexes. Used for dependency tracking */
+	List	   *local_unique_indexes;
+	bool		local_unique_indexes_valid;
+
 	/*
 	 * Per-operation safety cache for parallel apply. If
 	 * parallel_global_unsafe[action] is true, that action cannot be applied in
@@ -74,6 +78,17 @@ typedef struct LogicalRepRelMapEntry
 	bool		parallel_global_unsafe[LRPA_ACTION_COUNT];
 } LogicalRepRelMapEntry;
 
+/*
+ * Subscriber side unique index information.  This is used to track dependencies
+ * between transactions that modify the same unique key value.
+ */
+typedef struct LogicalRepSubUniqueIndex
+{
+	Oid			indexoid;	/* OID of the local key */
+	Bitmapset  *indexkeys;	/* Bitmap of key columns *on remote* */
+	bool		nulls_distinct;	/* Whether NULLs are considered distinct */
+} LogicalRepSubUniqueIndex;
+
 extern void logicalrep_relmap_update(LogicalRepRelation *remoterel);
 extern void logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel);
 
@@ -84,6 +99,7 @@ extern LogicalRepRelMapEntry *logicalrep_partition_open(LogicalRepRelMapEntry *r
 extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel,
 								 LOCKMODE lockmode);
 extern void logicalrep_rel_check_parallel_safety(LogicalRepRelMapEntry *entry);
+extern void logicalrep_build_dependent_unique_indexes(LogicalRepRelMapEntry *entry);
 extern bool IsIndexUsableForReplicaIdentityFull(Relation idxrel, AttrMap *attrmap);
 extern Oid	GetRelationIdentityOrPK(Relation rel);
 extern int	logicalrep_get_num_rels(void);
diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl
index c90a72682e5..9e981bab981 100644
--- a/src/test/subscription/t/050_parallel_apply.pl
+++ b/src/test/subscription/t/050_parallel_apply.pl
@@ -527,4 +527,128 @@ $result =
   $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab");
 is ($result, 2, 'inserts are replicated to subscriber');
 
+##################################################
+# Test that the dependency tracking works correctly for local unique indexes on
+# subscriber during parallel apply.
+##################################################
+
+# Truncate the data for upcoming tests
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE regress_tab;");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Define an unique index on subscriber
+$node_subscriber->safe_psql('postgres',
+    "CREATE UNIQUE INDEX local_unique_idx ON regress_tab (value);");
+
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (1, 'would conflict');");
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab");
+is ($result, 1, 'the insert is replicated to subscriber');
+
+# Attach an injection_point. Parallel workers would wait before the commit
+$node_subscriber->safe_psql('postgres',
+	"SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+# Delete the tuple on publisher.
+$node_publisher->safe_psql('postgres',
+    "DELETE FROM regress_tab WHERE id = 1;");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+	'parallel-worker-before-commit');
+
+$offset = -s $node_subscriber->logfile;
+
+# Insert tuples. This should conflict with the DELETE transaction, as both
+# transactions modify the same key. The parallel worker will wait for the
+# preceding transaction to finish.
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (2, 'would conflict');");
+
+# Verify the dependency is detected for the insert
+$str = $node_subscriber->wait_for_log(qr/found conflicting local unique key change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset);
+$xid = $str =~ /found conflicting local unique key change on table [1-9][0-9]+ from ([1-9][0-9]+)/;
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+ok(1, "local unique key dependency detected for parallel apply");
+
+# Wakeup the parallel worker
+$node_subscriber->safe_psql('postgres', qq[
+    SELECT injection_points_detach('parallel-worker-before-commit');
+    SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+# Verify the streamed transaction can be applied
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab");
+is ($result, 1, 'inserts are replicated to subscriber');
+
+# Test that the dependency tracking works correctly for local unique indexes on
+# subscriber during parallel apply when the unique index has expression.
+$node_subscriber->safe_psql('postgres', "DROP INDEX local_unique_idx;");
+$node_subscriber->safe_psql('postgres',
+    "CREATE UNIQUE INDEX local_unique_idx_expr ON regress_tab ((LOWER(value)));");
+
+# Attach an injection_point. Parallel workers would wait before the commit
+$node_subscriber->safe_psql('postgres',
+	"SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+# Insert a tuple on publisher. Parallel worker would wait at the injection
+# point
+$node_publisher->safe_psql('postgres',
+    "DELETE FROM regress_tab WHERE id = 2;");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+	'parallel-worker-before-commit');
+
+$offset = -s $node_subscriber->logfile;
+
+# Insert tuples. This should conflict with the DELETE transaction, as both
+# transactions modify the same key value. The parallel worker will wait for the
+# preceding transaction to finish.
+$node_publisher->safe_psql('postgres',
+    "INSERT INTO regress_tab VALUES (3, 'WOULD CONFLICT');");
+
+# Verify the dependency is detected for the insert
+$str = $node_subscriber->wait_for_log(qr/found conflicting local unique key change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset);
+$xid = $str =~ /found conflicting local unique key change on table [1-9][0-9]+ from ([1-9][0-9]+)/;
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+ok(1, "local unique key dependency from index expression detected for parallel apply");
+
+# Wakeup the parallel worker
+$node_subscriber->safe_psql('postgres', qq[
+    SELECT injection_points_detach('parallel-worker-before-commit');
+    SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+# Verify the streamed transaction can be applied
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+$node_publisher->wait_for_catchup('regress_sub');
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab");
+is ($result, 1, 'inserts are replicated to subscriber');
+
+# Cleanup
+$node_subscriber->safe_psql('postgres', "DROP INDEX local_unique_idx_expr;");
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE regress_tab;");
+$node_publisher->wait_for_catchup('regress_sub');
+
 done_testing();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 06d4bf7123a..9a061ebbc6c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1707,6 +1707,7 @@ LogicalRepBeginData
 LogicalRepCommitData
 LogicalRepCommitPreparedTxnData
 LogicalRepCtxStruct
+LogicalRepKeyKind
 LogicalRepMsgType
 LogicalRepPartMapEntry
 LogicalRepPreparedTxnData
@@ -1716,6 +1717,7 @@ LogicalRepRelation
 LogicalRepRollbackPreparedTxnData
 LogicalRepSequenceInfo
 LogicalRepStreamAbortData
+LogicalRepSubscriberIdx
 LogicalRepTupleData
 LogicalRepTyp
 LogicalRepWorker
-- 
2.52.0

