From 3d0023b4f5d507669308be5907169a6916ea2ae8 Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Thu, 9 Jul 2026 15:43:06 +0800 Subject: [PATCH v23 11/12] 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. Author: Zhijie Hou Author: Hayato Kuroda --- .../replication/logical/applyparallelworker.c | 52 +++ src/backend/replication/logical/relation.c | 269 ++++++++++++ src/backend/replication/logical/worker.c | 405 +++++++++++++----- src/backend/storage/lmgr/deadlock.c | 1 - src/include/replication/logicalrelation.h | 16 + src/test/subscription/t/050_parallel_apply.pl | 265 ++++++++++++ src/tools/pgindent/typedefs.list | 2 + 7 files changed, 913 insertions(+), 97 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 78db7b1fb46..da3f811a58c 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -179,6 +179,57 @@ * (DELETE row 1) and TX-2 (INSERT row 1) are applied in parallel, TX-2's INSERT * could be applied before TX-1's DELETE, resulting in a insert_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: @@ -423,6 +474,7 @@ pa_can_start(void) * It is better to do it before the below checks so that the latest values * of subscription can be used for the checks. */ + AcceptInvalidationMessages(); maybe_reread_subscription(); /* diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c index ec1f917071b..93dd113c4a2 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -21,6 +21,7 @@ #include "access/genam.h" #include "access/table.h" #include "catalog/namespace.h" +#include "catalog/pg_inherits.h" #include "catalog/pg_proc.h" #include "catalog/pg_subscription_rel.h" #include "catalog/pg_trigger.h" @@ -131,6 +132,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; } @@ -147,6 +149,7 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid) { entry->localrelvalid = false; entry->parallel_safety_valid = false; + entry->local_unique_indexes_valid = false; } } } @@ -178,6 +181,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. */ @@ -205,6 +223,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); } /* @@ -266,6 +287,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); } @@ -452,6 +480,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) { @@ -673,6 +702,233 @@ logicalrep_rel_check_parallel_safety(LogicalRepRelMapEntry *entry) entry->parallel_safety_valid = true; } +/* + * Append a new unique index to the list if it is not a duplicate of, or a + * superset of, an existing index. If it is a subset of an existing index, + * replace the existing one with the new one. + */ +static void +maybe_append_unique_index(Oid indexoid, Bitmapset *indexkeys, + bool nulls_distinct, List **unique_indexes) +{ + LogicalRepSubUniqueIndex *newinfo; + MemoryContext oldctx; + + foreach_ptr(LogicalRepSubUniqueIndex, oldinfo, *unique_indexes) + { + BMS_Comparison cmp; + + if (oldinfo->nulls_distinct != nulls_distinct) + continue; + + cmp = bms_subset_compare(oldinfo->indexkeys, indexkeys); + + /* Duplicate index, no need to add */ + if (cmp == BMS_EQUAL || cmp == BMS_SUBSET2) + return; + + /* New index is a subset of an existing one, replace the old one */ + if (cmp == BMS_SUBSET1) + { + oldinfo->indexoid = indexoid; + oldinfo->indexkeys = indexkeys; + oldinfo->nulls_distinct = nulls_distinct; + return; + } + } + + oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext); + + newinfo = palloc(sizeof(LogicalRepSubUniqueIndex)); + newinfo->indexoid = indexoid; + newinfo->indexkeys = bms_copy(indexkeys); + newinfo->nulls_distinct = nulls_distinct; + *unique_indexes = lappend(*unique_indexes, newinfo); + + MemoryContextSwitchTo(oldctx); +} + +/* + * 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 or an already collected + * unique index are skipped, as their scope is already covered. + * + * 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. + */ +static void +get_unique_indexes(LogicalRepRelMapEntry *entry, List **local_unique_indexes) +{ + List *idxlist; + + idxlist = RelationGetIndexList(entry->localrel); + + /* Iterate indexes to list all usable indexes */ + foreach_oid(idxoid, idxlist) + { + Relation idxrel; + AttrMap *attrmap; + Bitmapset *indexkeys = NULL; + BMS_Comparison cmp; + int indnkeys; + bool nulls_distinct; + + idxrel = index_open(idxoid, AccessShareLock); + + /* + * Only unique indexes are considered. Indexes backing a deferrable + * unique constraint are skipped: their uniqueness is checked at + * commit time and commit order is preserved, so they cannot cause + * an apply-time conflict. + */ + if (!idxrel->rd_index->indisunique || + !idxrel->rd_index->indimmediate) + { + 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; + + /* + * Skip generated columns. The subscriber recomputes generated + * values using its own expression, so the value sent by the + * publisher (if published at all) does not necessarily match + * the local one and cannot be used for dependency tracking. + * Dropping the column may degenerate the index to table-level + * serialization, which is safe. + */ + if (TupleDescCompactAttr(RelationGetDescr(entry->localrel), + AttrNumberGetAttrOffset(localcol))->attgenerated) + 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. + */ + cmp = bms_subset_compare(entry->remoterel.attkeys, indexkeys); + if (cmp == BMS_EQUAL || cmp == BMS_SUBSET1) + continue; + + maybe_append_unique_index(idxoid, indexkeys, nulls_distinct, + local_unique_indexes); + } + + list_free(idxlist); +} + +/* + * Collect all local unique indexes that can be used for dependency tracking. + * For partitioned tables, this also collects unique indexes from all leaf + * partitions. + * + * See get_unique_indexes() for details on how unique indexes are collected. + */ +void +logicalrep_build_dependent_unique_indexes(LogicalRepRelMapEntry *entry) +{ + List *partitions; + + if (entry->local_unique_indexes_valid) + return; + + free_local_unique_indexes(entry); + + /* For non-partitioned tables, simply get the unique indexes and return */ + if (entry->localrel->rd_rel->relkind == RELKIND_RELATION) + { + get_unique_indexes(entry, &entry->local_unique_indexes); + entry->local_unique_indexes_valid = true; + return; + } + + /* + * Collect unique indexes from all leaf partitions of this partitioned + * table. + */ + partitions = find_all_inheritors(entry->localreloid, AccessShareLock, + NULL); + + foreach_oid(relid, partitions) + { + LogicalRepRelMapEntry *part_entry; + Relation partrel; + AttrMap *root_to_part_attrmap; + + /* Only check leaf partitions */ + if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE) + continue; + + partrel = table_open(relid, AccessShareLock); + + root_to_part_attrmap = + build_attrmap_by_name_if_req(RelationGetDescr(entry->localrel), + RelationGetDescr(partrel), false); + + part_entry = logicalrep_partition_open(entry, partrel, + root_to_part_attrmap); + + if (root_to_part_attrmap) + free_attrmap(root_to_part_attrmap); + + get_unique_indexes(part_entry, &entry->local_unique_indexes); + + table_close(partrel, AccessShareLock); + } + + entry->local_unique_indexes_valid = true; +} + /* * Partition cache: look up partition LogicalRepRelMapEntry's * @@ -704,8 +960,21 @@ logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid) entry = hash_search(LogicalRepPartMap, &reloid, HASH_FIND, NULL); if (entry != NULL) { + LogicalRepRelMapEntry *parent; + entry->relmapentry.localrelvalid = false; entry->relmapentry.parallel_safety_valid = false; + + /* + * Invalidating a leaf partition makes the parent table's collected + * all partition's index information stale. + */ + parent = hash_search(LogicalRepRelMap, + &entry->relmapentry.remoterel.remoteid, + HASH_FIND, NULL); + + if (parent) + parent->local_unique_indexes_valid = false; } } else diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 9bf85a19e4c..330c29f0de9 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -580,10 +580,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; /* estimated memory usage of this key, set by the key builders */ @@ -786,7 +796,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++) @@ -866,8 +877,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); } @@ -972,74 +987,135 @@ append_xid_dependency(TransactionId xid, List **depends_on_xids) } /* - * Check for dependencies on preceding transactions that modify the same key as - * the given tuple. Returns the dependent transactions in 'depends_on_xids'. + * 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. * - * 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. + * Return the existing transaction ID if an active dependency exists for the + * key; otherwise returns InvalidTransactionId. */ -static void -check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data, - TransactionId new_depended_xid, - List **depends_on_xids) +static TransactionId +check_and_record_key_dependency(ReplicaIdentityKey *key, + TransactionId new_depended_xid) { - LogicalRepRelMapEntry *relentry; - LogicalRepTupleData *ridata; - ReplicaIdentityKey *rikey; + TransactionId existing_xid = InvalidTransactionId; ReplicaIdentityEntry *rientry; - MemoryContext oldctx; - int n_ri; bool found = false; - Size keysize; - Assert(depends_on_xids); + /* + * 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); - /* Search for existing entry */ - relentry = logicalrep_get_relentry(relid); + 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); - Assert(relentry); + 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); /* - * First check whether any previous transaction (other than the current one) - * has affected the whole table e.g., truncate or schema change from - * publisher. + * Release the key built to search the entry, if the entry already exists. */ - if (has_active_rel_dependency(relentry) && - !TransactionIdEquals(relentry->last_depended_xid, new_depended_xid)) + if (found) { - elog(DEBUG1, "found table-wide change affecting %u from %u", - relid, relentry->last_depended_xid); + 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); - append_xid_dependency(relentry->last_depended_xid, depends_on_xids); + existing_xid = rientry->remote_xid; + } + + free_replica_identity_key(key); + } + else + { + /* Account the estimated memory usage of the new entry */ + rientry->entry_size = key->keysize + sizeof(ReplicaIdentityEntry); + dependency_mem_usage += rientry->entry_size; } - n_ri = bms_num_members(relentry->remoterel.attkeys); + rientry->remote_xid = new_depended_xid; - /* - * 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; + return existing_xid; +} + +/* + * Check if any of the key columns have NULL values. + */ +static bool +has_null_key_values(LogicalRepTupleData *data, Bitmapset *indexkeys) +{ + for (int i = 0; i < data->ncols; i++) + { + if (bms_is_member(i, indexkeys) && + data->colstatus[i] == LOGICALREP_COLUMN_NULL) + return true; + } + + return false; +} + +/* + * Build a hash key for replica_identity_table using the given relation and + * tuple data, restricted to the specified key columns. + */ +static ReplicaIdentityKey * +build_replica_identity_key(Oid relid, LogicalRepTupleData *original_data, + Bitmapset *keycols) +{ + LogicalRepTupleData *keydata; + ReplicaIdentityKey *key; + MemoryContext oldctx; + int nkeycols = bms_num_members(keycols); + int i_key = 0; + Size keysize; 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; + keydata = palloc0_object(LogicalRepTupleData); + + if (nkeycols) + { + keydata->colvalues = palloc0_array(StringInfoData, nkeycols); + keydata->colstatus = palloc0_array(char, nkeycols); + } + + keydata->ncols = nkeycols; /* Estimated memory usage so far: key, tuple data and column arrays */ keysize = sizeof(ReplicaIdentityKey) + sizeof(LogicalRepTupleData) + - n_ri * (sizeof(StringInfoData) + sizeof(char)); + nkeycols * (sizeof(StringInfoData) + sizeof(char)); - for (int i_original = 0, i_ri = 0; i_original < original_data->ncols; i_original++) + for (int i = 0; i < original_data->ncols; i++) { - if (!bms_is_member(i_original, relentry->remoterel.attkeys)) + if (!bms_is_member(i, keycols)) continue; /* @@ -1052,94 +1128,211 @@ 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); /* * 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) + 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], + initStringInfoExt(&keydata->colvalues[i_key], original_colvalue->len + 1); - appendBinaryStringInfo(&ridata->colvalues[i_ri], + appendBinaryStringInfo(&keydata->colvalues[i_key], original_colvalue->data, original_colvalue->len); keysize += original_colvalue->len + 1; } - 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; - rikey->keysize = keysize; + key = palloc0_object(ReplicaIdentityKey); + key->relid = relid; + key->data = keydata; + key->keysize = keysize; 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; + + /* + * 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 (old_tuple && !TransactionIdIsValid(new_depended_xid)) + continue; + + rikey = build_replica_identity_key(relid, original_data, idxinfo->indexkeys); + rikey->kind = LOGICALREP_KEY_LOCAL_UNIQUE; /* - * Append the dependency to the list if the current transaction was not - * the lastest one to modify the key. + * 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 (has_active_key_dependency(rientry, false) && - !TransactionIdEquals(rientry->remote_xid, new_depended_xid)) + 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); } } - else +} + +/* + * 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)) { - /* Account the estimated memory usage of the new entry */ - rientry->entry_size = rikey->keysize + sizeof(ReplicaIdentityEntry); - dependency_mem_usage += rientry->entry_size; + 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); } - /* Update the new depended xid into the entry */ - rientry->remote_xid = new_depended_xid; + /* + * 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; + + 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); } /* @@ -1351,6 +1544,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: @@ -1380,12 +1576,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 b6356f72a4e..dd80e0a6a89 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 1b2a8e81dba..558edd0bd8a 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 void logicalrep_write_all_internal_rels(StringInfo out, int *num_rels); diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl index fcae6892e35..a869b0fc68f 100644 --- a/src/test/subscription/t/050_parallel_apply.pl +++ b/src/test/subscription/t/050_parallel_apply.pl @@ -637,4 +637,269 @@ $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'); + +################################################## +# Test that the dependency tracking works correctly when a local unique index +# exists only on a child (leaf) partition of a partitioned table. +# +# The publisher has a regular (non-partitioned) table. +# Table design: +# +# (publiser-side) (subscriber-side) +# regress_part_tab ------ regress_part_tab +# | +# +----- regress_part_tab_1 +# (has an unique index on 'value', with +# columns in a different order from root) +################################################## + +# Publisher: plain table. +$node_publisher->safe_psql('postgres', qq[ + CREATE TABLE regress_part_tab (id int PRIMARY KEY, value text, marker text); + ALTER TABLE regress_part_tab REPLICA IDENTITY FULL; +]); + +# Subscriber: partitioned table with a leaf-only unique index on 'value'. The +# leaf's physical column order differs from the root table's column order. +$node_subscriber->safe_psql('postgres', qq[ + CREATE TABLE regress_part_tab (id int PRIMARY KEY, value text, marker text) + PARTITION BY RANGE (id); + CREATE TABLE regress_part_tab_1 (value text, marker text, id int PRIMARY KEY); + ALTER TABLE regress_part_tab ATTACH PARTITION regress_part_tab_1 + FOR VALUES FROM (1) TO (100); + CREATE UNIQUE INDEX regress_part_tab_1_value_idx + ON regress_part_tab_1 (value); +]); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION regress_sub REFRESH PUBLICATION WITH (copy_data = false);"); + +# Insert a row that will become the conflict anchor. +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_part_tab VALUES (1, 'leaf_unique_val', 'leaf_marker_val');"); +$node_publisher->wait_for_catchup('regress_sub'); + +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(1) FROM regress_part_tab"); +is($result, '1', 'initial row inserted into partitioned subscriber table'); + +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');"); + +# Tx-1: Delete the row. The parallel worker records a dependency using the +# leaf partition's local unique index metadata and then pauses at the injection +# point. +$node_publisher->safe_psql('postgres', + "DELETE FROM regress_part_tab WHERE id = 1;"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +# Tx-2: Insert a new row reusing value='leaf_unique_val'. If applied before +# Tx-1 commits, the leaf's unique index would be violated. The dependency +# tracking must detect this conflict and make Tx-2 wait for Tx-1. +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_part_tab VALUES (2, 'leaf_unique_val', 'leaf_marker_val');"); + +# Verify the dependency is detected via the leaf partition's unique index. +$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]+)/; + +$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset); + +ok(1, + "leaf-only unique key dependency with reordered leaf columns detected for partitioned table parallel apply"); + +$node_subscriber->safe_psql('postgres', qq[ + SELECT injection_points_detach('parallel-worker-before-commit'); + SELECT injection_points_wakeup('parallel-worker-before-commit'); +]); + +$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset); +$node_publisher->wait_for_catchup('regress_sub'); + +# Net result: the DELETE (id=1) and INSERT (id=2) both commit; one row remains. +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(1) FROM regress_part_tab"); +is($result, '1', + 'changes to partitioned subscriber table are replicated correctly'); + +# Replace the leaf-only unique index after the root relation's partitioned-table +# information has already been cached. The leaf relcache invalidation must also +# invalidate the root relmap entry so the next dependency check uses the new +# leaf index metadata. +$node_subscriber->safe_psql('postgres', qq[ + DROP INDEX regress_part_tab_1_value_idx; + CREATE UNIQUE INDEX regress_part_tab_1_marker_idx + ON regress_part_tab_1 (marker); +]); + +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');"); + +# Tx-1: Delete the row. With a refreshed root cache, this records a dependency +# using the replacement leaf-only unique index on 'marker'. +$node_publisher->safe_psql('postgres', + "DELETE FROM regress_part_tab WHERE id = 2;"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +# Tx-2: Use a different value but reuse the marker. A stale root cache would +# still track the dropped 'value' index and miss this dependency. +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_part_tab VALUES (3, 'different_leaf_val', 'leaf_marker_val');"); + +$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]+)/; + +$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset); + +ok(1, + "leaf relcache invalidation refreshes partitioned table local unique keys"); + +$node_subscriber->safe_psql('postgres', qq[ + SELECT injection_points_detach('parallel-worker-before-commit'); + SELECT injection_points_wakeup('parallel-worker-before-commit'); +]); + +$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_part_tab"); +is($result, '1', + 'changes after leaf unique index replacement are replicated correctly'); + done_testing(); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index e7849777f9d..e1bc0e8a1d7 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.43.0