From b959f17179689938195efde244fbecff4c9d32ba Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Tue, 7 Jul 2026 19:18:05 +0800 Subject: [PATCH v21 9/9] Support dependency tracking via local foreign keys This patch tracks foreign key dependencies to prevent constraint violations. Consider a referencing table FK_TABLE (a INT REFERENCES PK_TABLE) and a referenced table PK_TABLE (a INT PRIMARY KEY): TX-1: INSERT row (1) INTO PK_TABLE TX-2: INSERT row (1) INTO FK_TABLE TX-2 must wait for TX-1; otherwise, a foreign key violation occurs. Similarly, for deletions: TX-1: DELETE row (1) FROM FK_TABLE TX-2: DELETE row (1) FROM PK_TABLE TX-2 must wait for TX-1 to avoid violating the foreign key constraint. Therefore, we record new tuples on the referenced table (PK_TABLE) and check for dependencies when processing new tuples on the referencing table (FK_TABLE). Conversely, we record old tuples deleted from the referencing table and require deletions on the referenced table to depend on them. Similar to unique key dependencies, for delete-delete cases we only track foreign key dependencies on columns that are part of the replica identity keys on the publisher. This may lead to false positives but is acceptable given the trade-off. --- .../replication/logical/applyparallelworker.c | 26 ++ src/backend/replication/logical/relation.c | 384 +++++++++++++++++ src/backend/replication/logical/worker.c | 395 +++++++++++++++++- src/include/replication/logicalrelation.h | 30 ++ src/test/subscription/t/050_parallel_apply.pl | 330 +++++++++++++++ 5 files changed, 1150 insertions(+), 15 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 1f1a75b5019..733879d1edc 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -218,6 +218,32 @@ * replica identity keys. This could be useful for users who prefer higher * parallelism and experience few conflicts. * + * Additionally, foreign key dependencies are also tracked to prevent constraint + * violations. Consider a referencing table FK_TABLE (a INT REFERENCES PK_TABLE) + * and a referenced table PK_TABLE (a INT PRIMARY KEY): + * + * TX-1: INSERT row (1) INTO PK_TABLE + * TX-2: INSERT row (1) INTO FK_TABLE + * + * TX-2 must wait for TX-1; otherwise, a foreign key violation occurs. + * + * Similarly, for deletions: + * + * TX-1: DELETE row (1) FROM FK_TABLE + * TX-2: DELETE row (1) FROM PK_TABLE + * + * TX-2 must wait for TX-1 to avoid violating the foreign key constraint. + * + * Therefore, we record new tuples on the referenced table (PK_TABLE) and check + * for dependencies when processing new tuples on the referencing table + * (FK_TABLE). Conversely, we record old tuples deleted from the referencing + * table and require deletions on the referenced table to depend on them. + * + * Similar to unique key dependencies, for delete-delete cases we only track + * foreign key dependencies on columns that are part of the replica identity + * keys on the publisher. This may lead to false positives but is acceptable + * given the trade-off. + * * 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 diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c index 1572990ee18..cdafa10beb3 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -20,7 +20,11 @@ #include "access/amapi.h" #include "access/genam.h" #include "access/table.h" +#include "catalog/heap.h" #include "catalog/namespace.h" +#include "catalog/partition.h" +#include "catalog/pg_constraint.h" +#include "catalog/pg_inherits.h" #include "catalog/pg_proc.h" #include "catalog/pg_subscription_rel.h" #include "catalog/pg_trigger.h" @@ -31,6 +35,7 @@ #include "replication/logicalrelation.h" #include "replication/worker_internal.h" #include "rewrite/rewriteHandler.h" +#include "utils/fmgroids.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/syscache.h" @@ -62,6 +67,10 @@ typedef struct LogicalRepPartMapEntry static Oid FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel, AttrMap *attrMap); +static void logicalrep_map_attnums_by_name(Oid src_reloid, Oid dst_reloid, + int nkeys, + const AttrNumber *src_attnums, + AttrNumber *dst_attnums); /* * Relcache invalidation callback for our relation map cache. @@ -89,6 +98,7 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid) entry->localrelvalid = false; entry->parallel_safety_valid = false; entry->local_unique_indexes_valid = false; + entry->local_fkeys_valid = false; hash_seq_term(&status); break; } @@ -106,6 +116,7 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid) entry->localrelvalid = false; entry->parallel_safety_valid = false; entry->local_unique_indexes_valid = false; + entry->local_fkeys_valid = false; } } } @@ -152,6 +163,33 @@ free_local_unique_indexes(LogicalRepRelMapEntry *entry) entry->local_unique_indexes = NIL; } +/* + * Release local foreign key lists. + */ +static void +free_local_fkeys(LogicalRepRelMapEntry *entry) +{ + Assert(am_leader_apply_worker()); + + foreach_ptr(LogicalRepSubFKey, fkinfo, entry->local_fkeys) + { + list_free(fkinfo->fkattnums_new); + list_free(fkinfo->fkattnums_old); + } + + foreach_ptr(LogicalRepSubRefKey, refinfo, entry->local_refkeys) + { + list_free(refinfo->refattnums_new); + list_free(refinfo->refattnums_old); + } + + list_free_deep(entry->local_fkeys); + list_free_deep(entry->local_refkeys); + + entry->local_fkeys = NIL; + entry->local_refkeys = NIL; +} + /* * Free the entry of a relation map cache. */ @@ -182,6 +220,9 @@ logicalrep_relmap_free_entry(LogicalRepRelMapEntry *entry) if (entry->local_unique_indexes != NIL) free_local_unique_indexes(entry); + + if (entry->local_fkeys != NIL || entry->local_refkeys != NIL) + free_local_fkeys(entry); } /* @@ -432,6 +473,7 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode) entry->localrelvalid = false; entry->parallel_safety_valid = false; entry->local_unique_indexes_valid = false; + entry->local_fkeys_valid = false; } else if (!entry->localrelvalid) { @@ -793,6 +835,348 @@ logicalrep_build_dependent_unique_indexes(LogicalRepRelMapEntry *entry) entry->local_unique_indexes_valid = true; } +/* + * Return all local relation map entries that match the given list of local + * relation OIDs. + */ +static List * +logicalrep_get_local_relentries(List *localrelids) +{ + HASH_SEQ_STATUS status; + LogicalRepRelMapEntry *entry; + List *entries = NIL; + + if (LogicalRepRelMap == NULL) + return NIL; + + hash_seq_init(&status, LogicalRepRelMap); + while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL) + { + if (!OidIsValid(entry->localreloid)) + continue; + + if (list_member_oid(localrelids, entry->localreloid)) + entries = lappend(entries, entry); + } + + return entries; +} + +/* + * Map key columns from src_reloid to dst_reloid by attribute name. + */ +static void +logicalrep_map_attnums_by_name(Oid src_reloid, Oid dst_reloid, + int nkeys, + const AttrNumber *src_attnums, + AttrNumber *dst_attnums) +{ + if (src_reloid == dst_reloid) + { + memcpy(dst_attnums, src_attnums, nkeys * sizeof(AttrNumber)); + return; + } + + for (int i = 0; i < nkeys; i++) + { + char *attname; + AttrNumber attnum; + + attname = get_attname(src_reloid, src_attnums[i], false); + attnum = get_attnum(dst_reloid, attname); + pfree(attname); + + Assert(AttributeNumberIsValid(attnum)); + + dst_attnums[i] = attnum; + } +} + +/* + * Check whether the table has any foreign key triggers of the given type + * (trig_type) enabled in replica mode. + */ +static bool +fkey_trigger_enabled_in_replica(Relation rel, Oid conoid, int trig_type) +{ + TriggerDesc *trigdesc = rel->trigdesc; + + if (trigdesc == NULL) + return false; + + for (int i = 0; i < trigdesc->numtriggers; i++) + { + Trigger *trig = &trigdesc->triggers[i]; + + /* Keep only PK-side action triggers for this FK constraint */ + if (trig->tgconstraint != conoid || + RI_FKey_trigger_type(trig->tgfoid) != trig_type) + continue; + + /* In replica mode, only REPLICA/ALWAYS triggers can fire. */ + if (trig->tgenabled == TRIGGER_FIRES_ON_REPLICA || + trig->tgenabled == TRIGGER_FIRES_ALWAYS) + return true; + } + + return false; +} + +/* + * Build lists of foreign key and referenced primary key remote columns. Both + * lists are ordered according to the referenced table's remote column order. + */ +static void +build_fk_remote_attnums(LogicalRepRelMapEntry *fkentry, + LogicalRepRelMapEntry *refentry, + int nkeys, + const AttrNumber *fk_conkey, + const AttrNumber *ref_confkey, + int trig_type, + List **fkattnums_new, + List **fkattnums_old) +{ + int pkatt = -1; + + Assert(trig_type == RI_TRIGGER_FK || trig_type == RI_TRIGGER_PK); + + /* + * Traverse the referenced table's remote replica identity columns in order, + * and for each, find the corresponding foreign key column that matches it. + */ + while ((pkatt = bms_next_member(refentry->remoterel.attkeys, pkatt)) >= 0) + { + for (int i = 0; i < nkeys; i++) + { + AttrNumber fk_local_attnum = fk_conkey[i]; + AttrNumber ref_local_attnum = ref_confkey[i]; + int fk_remote_attnum; + int ref_remote_attnum; + int mapped_attnum; + + fk_remote_attnum = fkentry->attrmap->attnums[AttrNumberGetAttrOffset(fk_local_attnum)]; + ref_remote_attnum = refentry->attrmap->attnums[AttrNumberGetAttrOffset(ref_local_attnum)]; + + /* Skip if not the current traversed column */ + if (ref_remote_attnum != pkatt) + continue; + + /* Skip columns that are unavailable in the remote table */ + if (ref_remote_attnum < 0 || fk_remote_attnum < 0) + continue; + + if (trig_type == RI_TRIGGER_FK) + mapped_attnum = fk_remote_attnum + 1; + else if (trig_type == RI_TRIGGER_PK) + mapped_attnum = ref_remote_attnum + 1; + + *fkattnums_new = lappend_int(*fkattnums_new, mapped_attnum); + + /* Old tuple contains only replica identity columns. */ + if (bms_is_member(fk_remote_attnum, fkentry->remoterel.attkeys) && + bms_is_member(ref_remote_attnum, refentry->remoterel.attkeys)) + *fkattnums_old = lappend_int(*fkattnums_old, mapped_attnum); + + break; + } + } +} + +/* + * Collect foreign keys for dependency tracking. + * + * This function collects both foreign keys in the referencing table and the + * primary key in the referenced table, as both are needed for dependency + * tracking (see applyparallelworker.c for details). + * + * For referencing-side foreign keys, we cannot use the referencing column + * numbers directly, because their order may differ from the primary key column + * order in the referenced table. To ensure consistency, we store a list of + * column numbers in the order of the referenced table's remote columns. The + * dependency tracking function will traverse this list to build the hash key. + * For referenced-side primary keys, we preserve the original column order. + * + * For one foreign key constraint on the table, We collect two set of column + * numbers for both the referencing and referenced keys: + * + * - One for INSERT-INSERT dependency tracking, using the new tuple. This + * includes all remote columns referenced by the foreign key, as the new tuple + * contains all remote values. + * + * - One for DELETE-DELETE dependency tracking, using the old tuple. This + * includes only columns that are part of the replica identity key, because + * the old tuple of an UPDATE or DELETE contains only replica identity key + * columns. Any other columns would be missing and unavailable for dependency + * tracking. + * + * If there are multiple foreign keys referencing other tables, or if the + * primary key is referenced by multiple foreign keys, we will have multiple + * column number sets. One might think that for a referenced primary key, we + * only need to store a single column bitmap (since only one primary key + * exists). However, the referencing side may not have all those column values + * available (columns outside the replica identity). Therefore, we store + * multiple primary key column sets, each including only the columns that are + * available on both sides. This ensures that the remote values on both the + * referenced and referencing tables can compute the same hash value. + */ +void +logicalrep_build_dependent_fkeys(LogicalRepRelMapEntry *entry) +{ + int trig_type; + Relation fkeyRel; + SysScanDesc fkeyScan; + HeapTuple tuple; + Oid relid = RelationGetRelid(entry->localrel); + + Assert(OidIsValid(relid)); + + if (entry->local_fkeys_valid) + return; + + free_local_fkeys(entry); + + fkeyRel = table_open(ConstraintRelationId, AccessShareLock); + + fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false, + NULL, 0, NULL); + + while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan))) + { + Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple); + List *fkentries; + AttrNumber conkey[INDEX_MAX_KEYS] = {0}; + AttrNumber confkey[INDEX_MAX_KEYS] = {0}; + AttrNumber *key_to_map; + int numfks; + MemoryContext oldctx; + List *all_parts; + Oid otherrelid; + + /* Not a foreign key */ + if (con->contype != CONSTRAINT_FOREIGN) + continue; + + /* Skip if FK enforcement is disabled */ + if (!con->conenforced) + continue; + + /* + * Deferrable foreign keys do not need tracking, as they won't cause + * constraint violations as long as commit order is preserved. + */ + if (con->condeferrable && con->condeferred) + continue; + + if (con->confrelid == relid) + { + trig_type = RI_TRIGGER_PK; + otherrelid = con->conrelid; + key_to_map = conkey; + } + else if (con->conrelid == relid) + { + trig_type = RI_TRIGGER_FK; + otherrelid = con->confrelid; + key_to_map = confkey; + } + else + continue; + + /* + * Skip when no replica-mode ON DELETE/ON UPDATE violation trigger can + * fire for this FK on the referenced table. + */ + if (!fkey_trigger_enabled_in_replica(entry->localrel, con->oid, + trig_type)) + continue; + + DeconstructFkConstraintRow(tuple, &numfks, conkey, confkey, + NULL, NULL, NULL, NULL, NULL); + + /* + * Collect all partitions of the referenced (or referencing) table if + * it's partitioned, and cache foreign key data between the current + * table and all of its partitions. + * + * This is necessary when both the referencing and referenced sides are + * partitioned tables, as pg_constraint does not record foreign key + * relationships between child partitions. We must search them + * explicitly. + * + * No lock is needed on each partition because we already hold a lock on + * the referencing table, preventing concurrent alterations that would + * affect foreign key dependencies. + */ + all_parts = find_all_inheritors(otherrelid, NoLock, NULL); + fkentries = logicalrep_get_local_relentries(all_parts); + + foreach_ptr(LogicalRepRelMapEntry, fkentry, fkentries) + { + AttrNumber mapped_conkey[INDEX_MAX_KEYS] = {0}; + + /* + * Skip if the referencing table is not published or has not + * replicated any changes. + */ + if (!fkentry->attrmap) + continue; + + /* + * Map column numbers from the parent to the corresponding child + * columns. + */ + logicalrep_map_attnums_by_name(otherrelid, + fkentry->localreloid, + numfks, + key_to_map, + mapped_conkey); + + oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext); + + if (trig_type == RI_TRIGGER_PK) + { + LogicalRepSubRefKey *refinfo; + + refinfo = palloc0_object(LogicalRepSubRefKey); + refinfo->conoid = con->oid; + refinfo->fk_remoteid = fkentry->remoterel.remoteid; + + build_fk_remote_attnums(fkentry, entry, numfks, mapped_conkey, + confkey, trig_type, + &refinfo->refattnums_new, + &refinfo->refattnums_old); + + entry->local_refkeys = lappend(entry->local_refkeys, refinfo); + } + else if (trig_type == RI_TRIGGER_FK) + { + LogicalRepSubFKey *fkinfo; + + fkinfo = palloc0_object(LogicalRepSubFKey); + fkinfo->conoid = con->oid; + fkinfo->ref_remoteid = fkentry->remoterel.remoteid; + + build_fk_remote_attnums(entry, fkentry, numfks, conkey, + mapped_conkey, trig_type, + &fkinfo->fkattnums_new, + &fkinfo->fkattnums_old); + + entry->local_fkeys = lappend(entry->local_fkeys, fkinfo); + } + + MemoryContextSwitchTo(oldctx); + } + + list_free(fkentries); + } + + systable_endscan(fkeyScan); + + table_close(fkeyRel, AccessShareLock); + + entry->local_fkeys_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 3864c95f930..08b4ddb094a 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -255,6 +255,7 @@ #include "access/twophase.h" #include "access/xact.h" #include "catalog/indexing.h" +#include "catalog/namespace.h" #include "catalog/pg_inherits.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" @@ -265,6 +266,7 @@ #include "executor/execPartition.h" #include "libpq/pqformat.h" #include "miscadmin.h" +#include "nodes/makefuncs.h" #include "optimizer/optimizer.h" #include "parser/parse_relation.h" #include "pgstat.h" @@ -584,7 +586,10 @@ static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL}; typedef enum LogicalRepKeyKind { LOGICALREP_KEY_REPLICA_IDENTITY, - LOGICALREP_KEY_LOCAL_UNIQUE + LOGICALREP_KEY_LOCAL_UNIQUE, + LOGICALREP_KEY_FOREIGN_KEY, + LOGICALREP_KEY_REFERENCED_KEY, + LOGICALREP_ALL_KEYS, } LogicalRepKeyKind; /* Hash table key for replica_identity_table */ @@ -950,6 +955,26 @@ append_xid_dependency(TransactionId xid, List **depends_on_xids) *depends_on_xids = lappend_xid(*depends_on_xids, xid); } +static char * +get_dependency_type_str(LogicalRepKeyKind kind) +{ + switch (kind) + { + case LOGICALREP_KEY_REPLICA_IDENTITY: + return "replica identity"; + case LOGICALREP_KEY_LOCAL_UNIQUE: + return "local unique key"; + case LOGICALREP_KEY_FOREIGN_KEY: + return "foreign key"; + case LOGICALREP_KEY_REFERENCED_KEY: + return "referenced key"; + case LOGICALREP_ALL_KEYS: + return "all keys"; + } + + return "???"; +} + /* * Common function for checking dependency by using the key. Used by both * check_and_record_ri_dependency and check_and_record_local_key_dependency. @@ -982,9 +1007,8 @@ check_and_record_key_dependency(ReplicaIdentityKey *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", + "found conflicting %s change on table %u from %u", + get_dependency_type_str(key->kind), key->relid, rientry->remote_xid); existing_xid = rientry->remote_xid; @@ -1007,9 +1031,8 @@ check_and_record_key_dependency(ReplicaIdentityKey *key, 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", + "found conflicting %s change on table %u from %u", + get_dependency_type_str(key->kind), key->relid, rientry->remote_xid); existing_xid = rientry->remote_xid; @@ -1039,6 +1062,23 @@ has_null_key_values(LogicalRepTupleData *data, Bitmapset *indexkeys) return false; } +/* + * Check if any of the attnums have NULL values. + */ +static bool +fk_tuple_has_null(LogicalRepTupleData *data, List *attnums) +{ + foreach_int(attnum, attnums) + { + int attidx = AttrNumberGetAttrOffset(attnum); + + if (data->colstatus[attidx] == 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. @@ -1104,18 +1144,81 @@ build_replica_identity_key(Oid relid, LogicalRepTupleData *original_data, return key; } +/* + * Similar to build_replica_identity_key(), but builds the hash key from + * explicitly ordered columns. + */ +static ReplicaIdentityKey * +build_replica_identity_key_by_attnums(Oid relid, + LogicalRepTupleData *original_data, + List *attnums) +{ + LogicalRepTupleData *keydata; + ReplicaIdentityKey *key; + MemoryContext oldctx; + int i_key = 0; + int nattnums = list_length(attnums); + + oldctx = MemoryContextSwitchTo(ApplyContext); + + keydata = palloc0_object(LogicalRepTupleData); + + if (nattnums) + { + keydata->colvalues = palloc0_array(StringInfoData, nattnums); + keydata->colstatus = palloc0_array(char, nattnums); + } + + keydata->ncols = nattnums; + + foreach_int(attnum, attnums) + { + int attidx = AttrNumberGetAttrOffset(attnum); + StringInfo original_colvalue; + + keydata->colstatus[i_key] = original_data->colstatus[attidx]; + + if (keydata->colstatus[i_key] != LOGICALREP_COLUMN_NULL) + { + Assert(original_data->colstatus[attidx] != LOGICALREP_COLUMN_UNCHANGED || + original_data->colvalues[attidx].len > 0); + + original_colvalue = &original_data->colvalues[attidx]; + + initStringInfoExt(&keydata->colvalues[i_key], original_colvalue->len + 1); + appendStringInfoString(&keydata->colvalues[i_key], original_colvalue->data); + } + + i_key++; + } + + 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. + * When missing_ok is true, we don't throw an error if the local relation + * doesn't exist. + * + * See logicalrep_build_dependent_unique_indexes() and + * logicalrep_build_dependent_fkeys() for details. */ static void -build_local_dependent_key_info(LogicalRepRelMapEntry *relentry) +build_local_dependent_key_info(LogicalRepRelMapEntry *relentry, bool missing_ok) { + Oid localrelid = relentry->localreloid; bool needs_start; - if (relentry->local_unique_indexes_valid) + if (relentry->local_unique_indexes_valid && + relentry->local_fkeys_valid) return; /* @@ -1127,11 +1230,20 @@ build_local_dependent_key_info(LogicalRepRelMapEntry *relentry) if (needs_start) StartTransactionCommand(); - relentry = logicalrep_rel_open(relentry->remoterel.remoteid, AccessShareLock); + if (!OidIsValid(localrelid)) + localrelid = RangeVarGetRelid(makeRangeVar(relentry->remoterel.nspname, + relentry->remoterel.relname, -1), + AccessShareLock, missing_ok); - logicalrep_build_dependent_unique_indexes(relentry); + if (OidIsValid(localrelid)) + { + relentry = logicalrep_rel_open(relentry->remoterel.remoteid, AccessShareLock); + + logicalrep_build_dependent_unique_indexes(relentry); + logicalrep_build_dependent_fkeys(relentry); - logicalrep_rel_close(relentry, AccessShareLock); + logicalrep_rel_close(relentry, AccessShareLock); + } if (needs_start) CommitTransactionCommand(); @@ -1160,7 +1272,7 @@ check_and_record_local_key_dependency(Oid relid, Assert(relentry); - build_local_dependent_key_info(relentry); + build_local_dependent_key_info(relentry, false); foreach_ptr(LogicalRepSubUniqueIndex, idxinfo, relentry->local_unique_indexes) { @@ -1215,6 +1327,135 @@ check_and_record_local_key_dependency(Oid relid, } } +/* + * Check and record dependencies caused by foreign key constraints. + * + * A new value in the referencing table depends on the referenced table's data. + * Conversely, a deletion in the referenced table depends on whether the + * referencing table still contains matching rows. + * + * See applyparallelworker.c for details on why tracking these dependencies is + * necessary. + */ +static void +check_and_record_fkey_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); + + relentry = logicalrep_get_relentry(relid); + + Assert(relentry); + + build_local_dependent_key_info(relentry, false); + + foreach_ptr(LogicalRepSubFKey, fkinfo, relentry->local_fkeys) + { + List *keyattnums = old_tuple ? + fkinfo->fkattnums_old : fkinfo->fkattnums_new; + + /* NULL value won't violate foreign key violation */ + if (fk_tuple_has_null(original_data, keyattnums)) + continue; + + /* + * Old tuples of foreign 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_by_attnums(fkinfo->ref_remoteid, + original_data, + keyattnums); + + /* + * For old tuples, record a dependency that later transactions deleting + * the same key from the referenced table must wait on. No preceding + * transactions are added to the list. + * + * For new tuples, check for existing dependencies on the same key + * inserted into the referenced table and add any dependent transactions + * to the list. + */ + if (old_tuple) + { + rikey->kind = LOGICALREP_KEY_FOREIGN_KEY; + (void) check_and_record_key_dependency(rikey, new_depended_xid); + } + else + { + TransactionId dep_xid; + + rikey->kind = LOGICALREP_KEY_REFERENCED_KEY; + dep_xid = check_and_record_key_dependency(rikey, InvalidTransactionId); + + if (TransactionIdIsValid(dep_xid) && + !TransactionIdEquals(dep_xid, new_depended_xid)) + append_xid_dependency(dep_xid, depends_on_xids); + } + } + + /* Return if no referenced key exists */ + if (relentry->local_refkeys == NIL) + return; + + foreach_ptr(LogicalRepSubRefKey, refinfo, relentry->local_refkeys) + { + List *keyattnums = old_tuple ? + refinfo->refattnums_old : refinfo->refattnums_new; + + /* Shouldn't be real NULL value in referenced key (primary key). */ + Assert(old_tuple || !fk_tuple_has_null(original_data, keyattnums)); + + /* + * New tuples of referenced 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 key entirely. + */ + if (!old_tuple && !TransactionIdIsValid(new_depended_xid)) + continue; + + rikey = build_replica_identity_key_by_attnums(relid, original_data, + keyattnums); + + /* + * For old tuples, check for existing dependencies on the same key + * deleted from the referencing table and add any dependent transactions + * to the list. + * + * For new tuples, record a dependency that later transactions inserting + * the same key into the referencing table must wait on. No preceding + * transactions are added to the list. + */ + if (old_tuple) + { + TransactionId dep_xid; + + rikey->kind = LOGICALREP_KEY_FOREIGN_KEY; + dep_xid = check_and_record_key_dependency(rikey, InvalidTransactionId); + + if (TransactionIdIsValid(dep_xid) && + !TransactionIdEquals(dep_xid, new_depended_xid)) + append_xid_dependency(dep_xid, depends_on_xids); + } + else + { + rikey->kind = LOGICALREP_KEY_REFERENCED_KEY; + (void) check_and_record_key_dependency(rikey, new_depended_xid); + } + } +} + /* * Check for dependencies on preceding transactions that modify the same key. * Returns the dependent transactions in 'depends_on_xids'. @@ -1290,6 +1531,7 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data, static void find_all_dependencies_on_rel(LogicalRepRelId relid, TransactionId new_depended_xid, + LogicalRepKeyKind kind, List **depends_on_xids) { replica_identity_iterator i; @@ -1305,6 +1547,10 @@ find_all_dependencies_on_rel(LogicalRepRelId relid, if (rientry->keydata->relid != relid) continue; + if (kind != LOGICALREP_ALL_KEYS && + rientry->keydata->kind != kind) + continue; + /* Skip entries that do not have a valid and uncommitted transaction */ if (!has_active_key_dependency(rientry, true)) continue; @@ -1337,7 +1583,8 @@ check_and_record_rel_dependency(LogicalRepRelId relid, Assert(depends_on_xids); - find_all_dependencies_on_rel(relid, new_depended_xid, depends_on_xids); + find_all_dependencies_on_rel(relid, new_depended_xid, LOGICALREP_ALL_KEYS, + depends_on_xids); /* Search for existing entry */ relentry = logicalrep_get_relentry(relid); @@ -1363,6 +1610,100 @@ check_and_record_rel_dependency(LogicalRepRelId relid, relentry->last_depended_xid = new_depended_xid; } +/* + * Check for preceding transactions that modified tables referencing or + * referenced by the given table. Any dependent transaction IDs are added to + * 'depends_on_xids'. + * + * This function is called only for TRUNCATE and remote schema change that + * affect the entire table. + * + * For remote schema changes, foreign key information of all referencing and + * referenced side tables is also invalidated. + */ +static void +find_frel_dependency(LogicalRepRelId relid, TransactionId new_depended_xid, + LogicalRepMsgType action, List **depends_on_xids) +{ + LogicalRepRelMapEntry *relentry; + bool schema_change = (action == LOGICAL_REP_MSG_RELATION); + + Assert(depends_on_xids); + Assert(action == LOGICAL_REP_MSG_TRUNCATE || + action == LOGICAL_REP_MSG_RELATION); + + /* Search for existing entry */ + relentry = logicalrep_get_relentry(relid); + + Assert(relentry); + + /* + * The relation message does not always indicate that the corresponding + * local relation exists. For example, the relation message may be sent for + * a partition even if the changes are published as root table and the same + * partition may not exists on subscriber. So missing_ok is true for schema + * change, but false for TRUNCATE. + */ + build_local_dependent_key_info(relentry, schema_change); + + /* + * For both TRUNCATE and schema changes, wait for preceding transactions + * that deleted from the referencing side table. + */ + foreach_ptr(LogicalRepSubRefKey, refinfo, relentry->local_refkeys) + { + find_all_dependencies_on_rel(refinfo->fk_remoteid, new_depended_xid, + LOGICALREP_KEY_FOREIGN_KEY, + depends_on_xids); + + check_and_record_rel_dependency(refinfo->fk_remoteid, + InvalidTransactionId, + depends_on_xids); + + if (schema_change) + { + LogicalRepRelMapEntry *fkentry; + + fkentry = logicalrep_get_relentry(refinfo->fk_remoteid); + + if (fkentry) + fkentry->local_fkeys_valid = false; + } + } + + /* + * TRUNCATE does not need to wait for referenced-side changes, but schema + * changes do. A replica identity change invalidates previously recorded + * referenced-side dependency entries. Without waiting, a later insert on + * the referencing table might miss the correct dependency on the referenced + * table, so it's safer to wait for preceding transactions on the referenced + * side to finish. + */ + if (!schema_change) + return; + + foreach_ptr(LogicalRepSubFKey, fkinfo, relentry->local_fkeys) + { + find_all_dependencies_on_rel(fkinfo->ref_remoteid, new_depended_xid, + LOGICALREP_KEY_REFERENCED_KEY, + depends_on_xids); + + check_and_record_rel_dependency(fkinfo->ref_remoteid, + InvalidTransactionId, + depends_on_xids); + + if (schema_change) + { + LogicalRepRelMapEntry *refentry; + + refentry = logicalrep_get_relentry(fkinfo->ref_remoteid); + + if (refentry) + refentry->local_fkeys_valid = false; + } + } +} + /* * Check the parallel safety of applying the give action for the relation. If * not safe, wait for preceding transactions to finish before proceeding with @@ -1481,6 +1822,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s, check_and_record_local_key_dependency(relid, &newtup, false, new_depended_xid, &depends_on_xids); + check_and_record_fkey_dependency(relid, &newtup, false, + new_depended_xid, + &depends_on_xids); break; case LOGICAL_REP_MSG_UPDATE: @@ -1524,6 +1868,13 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s, check_and_record_local_key_dependency(relid, &oldtup, true, new_depended_xid, &depends_on_xids); + + check_and_record_fkey_dependency(relid, &newtup, false, + new_depended_xid, + &depends_on_xids); + check_and_record_fkey_dependency(relid, &oldtup, true, + new_depended_xid, + &depends_on_xids); break; case LOGICAL_REP_MSG_DELETE: @@ -1533,6 +1884,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s, check_and_record_local_key_dependency(relid, &oldtup, true, new_depended_xid, &depends_on_xids); + check_and_record_fkey_dependency(relid, &oldtup, true, + new_depended_xid, + &depends_on_xids); break; case LOGICAL_REP_MSG_TRUNCATE: @@ -1545,10 +1899,15 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s, * modified the same table. */ foreach_int(truncated_relid, remote_relids) + { check_and_record_rel_dependency(truncated_relid, new_depended_xid, &depends_on_xids); + find_frel_dependency(truncated_relid, new_depended_xid, + action, &depends_on_xids); + } + break; case LOGICAL_REP_MSG_RELATION: @@ -1562,6 +1921,12 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s, */ check_and_record_rel_dependency(rel->remoteid, new_depended_xid, &depends_on_xids); + + logicalrep_relmap_update(rel); + + find_frel_dependency(rel->remoteid, new_depended_xid, action, + &depends_on_xids); + break; case LOGICAL_REP_MSG_TYPE: diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index ec4c9ece571..a917ce2fe86 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -60,6 +60,11 @@ typedef struct LogicalRepRelMapEntry List *local_unique_indexes; bool local_unique_indexes_valid; + /* Local foreign keys. Used for dependency tracking */ + List *local_fkeys; + List *local_refkeys; + bool local_fkeys_valid; + /* * Per-operation safety cache for parallel apply. If * parallel_global_unsafe[action] is true, that action cannot be applied in @@ -89,6 +94,30 @@ typedef struct LogicalRepSubUniqueIndex bool nulls_distinct; /* Whether NULLs are considered distinct */ } LogicalRepSubUniqueIndex; +/* + * Referencing side foreign key information. This is used to track dependencies + * between transactions that modify the same referenced and referencing key + * values. + */ +typedef struct LogicalRepSubFKey +{ + Oid conoid; /* OID of the FK constraint */ + LogicalRepRelId ref_remoteid; /* referenced remote relation */ + List *fkattnums_new; /* new-tuple-safe referencing remote attnums */ + List *fkattnums_old; /* old-tuple-safe referencing remote attnums */ +} LogicalRepSubFKey; + +/* + * Similar to LogicalRepSubFKey, but for referenced-side primary keys. + */ +typedef struct LogicalRepSubRefKey +{ + Oid conoid; /* OID of the FK constraint */ + LogicalRepRelId fk_remoteid; /* referencing remote relation */ + List *refattnums_new; /* new-tuple-safe referenced remote attnums */ + List *refattnums_old; /* old-tuple-safe referenced remote attnums */ +} LogicalRepSubRefKey; + extern void logicalrep_relmap_update(LogicalRepRelation *remoterel); extern void logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel); @@ -100,6 +129,7 @@ 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 void logicalrep_build_dependent_fkeys(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 9e981bab981..9d52cf660f8 100644 --- a/src/test/subscription/t/050_parallel_apply.pl +++ b/src/test/subscription/t/050_parallel_apply.pl @@ -37,6 +37,35 @@ $node_publisher->safe_psql( CREATE UNIQUE INDEX tab_toast_ri_index on tab_toast (a, b); ALTER TABLE tab_toast REPLICA IDENTITY USING INDEX tab_toast_ri_index; INSERT INTO tab_toast(a, b) VALUES(repeat('1234567890', 200), '1234567890'); + + CREATE TABLE regress_tab_pk (id int PRIMARY KEY); + CREATE TABLE regress_tab_fk (id int PRIMARY KEY, fk int REFERENCES regress_tab_pk (id)); + + CREATE TABLE pk_parted ( + id int PRIMARY KEY, + value int + ) PARTITION BY RANGE (id); + + CREATE TABLE pk_parted_1 ( + value int, + id int NOT NULL + ); + + ALTER TABLE pk_parted ATTACH PARTITION pk_parted_1 + FOR VALUES FROM (1) TO (10); + + CREATE TABLE fk_parted ( + id int REFERENCES pk_parted(id), + value int + ) PARTITION BY RANGE (id); + + CREATE TABLE fk_parted_1 ( + value int, + id int + ); + + ALTER TABLE fk_parted ATTACH PARTITION fk_parted_1 + FOR VALUES FROM (1) TO (10); )); $node_publisher->safe_psql('postgres', "INSERT INTO regress_tab VALUES (generate_series(1, 10), 'test');"); @@ -78,6 +107,35 @@ $node_subscriber->safe_psql( ALTER TABLE tab_toast ALTER COLUMN a SET STORAGE EXTERNAL; CREATE UNIQUE INDEX tab_toast_ri_index on tab_toast (a, b); ALTER TABLE tab_toast REPLICA IDENTITY USING INDEX tab_toast_ri_index; + + CREATE TABLE regress_tab_pk (id int PRIMARY KEY); + CREATE TABLE regress_tab_fk (id int PRIMARY KEY, fk int REFERENCES regress_tab_pk (id)); + + CREATE TABLE pk_parted ( + id int PRIMARY KEY, + value int + ) PARTITION BY RANGE (id); + + CREATE TABLE pk_parted_1 ( + value int, + id int NOT NULL + ); + + ALTER TABLE pk_parted ATTACH PARTITION pk_parted_1 + FOR VALUES FROM (1) TO (10); + + CREATE TABLE fk_parted ( + id int REFERENCES pk_parted(id), + value int + ) PARTITION BY RANGE (id); + + CREATE TABLE fk_parted_1 ( + value int, + id int + ); + + ALTER TABLE fk_parted ATTACH PARTITION fk_parted_1 + FOR VALUES FROM (1) TO (10); )); $node_subscriber->safe_psql('postgres', "CREATE SUBSCRIPTION regress_sub CONNECTION '$publisher_connstr' PUBLICATION regress_pub;"); @@ -651,4 +709,276 @@ $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 for foreign keys on +# subscriber during parallel apply. +################################################## + +# Test that when receiving table schema information for a referencing table, the +# subscriber correctly checks for dependencies on the referenced table and waits +# for its changes to complete. +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');" +); + +# Enable foreign triggers on subscriber +my $pk_fk_trigger = $node_subscriber->safe_psql('postgres', qq[ + SELECT tgname + FROM pg_trigger + WHERE tgrelid = 'regress_tab_pk'::regclass + AND tgconstraint > 0 + LIMIT 1 +]); +chomp($pk_fk_trigger); + +my $fk_fk_trigger = $node_subscriber->safe_psql('postgres', qq[ + SELECT tgname + FROM pg_trigger + WHERE tgrelid = 'regress_tab_fk'::regclass + AND tgconstraint > 0 + LIMIT 1 +]); +chomp($fk_fk_trigger); + +$node_subscriber->safe_psql('postgres', + qq[ALTER TABLE regress_tab_pk ENABLE REPLICA TRIGGER "$pk_fk_trigger"; + ALTER TABLE regress_tab_fk ENABLE REPLICA TRIGGER "$fk_fk_trigger";]); + +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_tab_pk VALUES (2);"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_tab_fk VALUES (1, 2);"); + +# Verify the dependency is detected on referenced table schema information +$str = $node_subscriber->wait_for_log(qr/found conflicting change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset); +$xid = $str =~ /found conflicting 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, "referenced table 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_pk"); +is ($result, 1, 'insert is replicated to referenced table on subscriber'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab_fk"); +is ($result, 1, 'insert is replicated to referencing table on subscriber'); + +# INSERT - INSERT case: Tx-1 inserts referenced tuple and Tx-2 inserts +# referencing tuple. + +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');" +); + +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_tab_pk VALUES (3);"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_tab_fk VALUES (2, 3);"); + +# Verify the dependency for referenced key change is detected +$str = $node_subscriber->wait_for_log(qr/found conflicting referenced key change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset); +$xid = $str =~ /found conflicting referenced 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, "referenced 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_pk"); +is ($result, 2, 'insert is replicated to referenced table on subscriber'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab_fk"); +is ($result, 2, 'insert is replicated to referencing table on subscriber'); + +# DELETE - DELETE case: Tx-1 deletes referencing tuple and Tx-2 deletes +# referenced tuple. +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');" +); + +$node_publisher->safe_psql('postgres', + "DELETE FROM regress_tab_fk WHERE id = 1;"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "DELETE FROM regress_tab_pk WHERE id = 2;"); + +$str = $node_subscriber->wait_for_log(qr/found conflicting foreign key change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset); +$xid = $str =~ /found conflicting foreign 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, "foreign key dependency detected for 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'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab_pk"); +is ($result, 1, 'delete is replicated to referenced table on subscriber'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab_fk"); +is ($result, 1, 'delete is replicated to referencing table on subscriber'); + +################################################## +# Test that the dependency tracking works correctly for foreign keys when both +# the referenced and referencing tables are partitioned. +################################################## + +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');" +); + +my $part_pk_trigger = $node_subscriber->safe_psql('postgres', qq[ + SELECT tgname + FROM pg_trigger + WHERE tgrelid = 'pk_parted_1'::regclass + AND tgconstraint > 0 + LIMIT 1 +]); +chomp($part_pk_trigger); + +my $part_fk_trigger = $node_subscriber->safe_psql('postgres', qq[ + SELECT tgname + FROM pg_trigger + WHERE tgrelid = 'fk_parted_1'::regclass + AND tgconstraint > 0 + LIMIT 1 +]); +chomp($part_fk_trigger); + +$node_subscriber->safe_psql('postgres', + qq[ALTER TABLE pk_parted_1 ENABLE REPLICA TRIGGER "$part_pk_trigger"; + ALTER TABLE fk_parted_1 ENABLE REPLICA TRIGGER "$part_fk_trigger";]); + +$node_publisher->safe_psql('postgres', + "INSERT INTO pk_parted_1(id, value) VALUES (1, 10);"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "INSERT INTO fk_parted_1(id, value) VALUES (1, 10);"); + +# Verify the dependency is detected on referenced table schema information +$str = $node_subscriber->wait_for_log(qr/found conflicting change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset); +$xid = $str =~ /found conflicting 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, "partitioned referenced table 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 pk_parted_1"); +is ($result, 1, 'insert is replicated to referenced table on subscriber'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM fk_parted_1"); +is ($result, 1, 'insert is replicated to referencing table on subscriber'); + +# INSERT - INSERT case: Tx-1 inserts referenced tuple and Tx-2 inserts +# referencing tuple. + +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');" +); + +$node_publisher->safe_psql('postgres', + "INSERT INTO pk_parted_1(id, value) VALUES (2, 20);"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "INSERT INTO fk_parted_1(id, value) VALUES (2, 20);"); + +# Verify the dependency for referenced key change is detected +$str = $node_subscriber->wait_for_log(qr/found conflicting referenced key change on table [1-9][0-9]+ from ([1-9][0-9]+)/, $offset); +$xid = $str =~ /found conflicting referenced 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, "partitioned referenced 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 pk_parted_1"); +is ($result, 2, 'insert is replicated to referenced table on subscriber'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM fk_parted_1"); +is ($result, 2, 'insert is replicated to referencing table on subscriber'); + done_testing(); -- 2.43.0