From ad1cdc213f4ac7764295d2de1038077ac96ea326 Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Thu, 9 Jul 2026 15:43:06 +0800 Subject: [PATCH v21 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 c9b24dbfa59..987489f1ef1 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1704,6 +1704,7 @@ LogicalRepBeginData LogicalRepCommitData LogicalRepCommitPreparedTxnData LogicalRepCtxStruct +LogicalRepKeyKind LogicalRepMsgType LogicalRepPartMapEntry LogicalRepPreparedTxnData @@ -1713,6 +1714,7 @@ LogicalRepRelation LogicalRepRollbackPreparedTxnData LogicalRepSequenceInfo LogicalRepStreamAbortData +LogicalRepSubscriberIdx LogicalRepTupleData LogicalRepTyp LogicalRepWorker -- 2.43.0