From 28cff2370f49d34ca742254656e778dbef1e1fb5 Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Tue, 9 Jun 2026 18:45:02 +0800 Subject: [PATCH v21 7/9] Handle unsafe parallel apply cases due to non-immutable functions This patch adds table-level parallel safety checks that mark operations as unsafe when a table has triggers executing non-immutable functions, or when subscriber-only column defaults contain non-immutable functions. Safety is determined by the worker that actually applies the changes, rather than solely by the leader. Leader-only evaluation is unsafe because the table could be altered before changes are dispatched to a parallel worker. Without re-validation by the parallel worker, it might incorrectly assume the relation is safe for parallel apply. When an unsafe operation is encountered, the worker waits for the last parallelized transaction to commit before proceeding. --- .../replication/logical/applyparallelworker.c | 7 +- src/backend/replication/logical/relation.c | 98 +++++++++++++++++++ src/backend/replication/logical/worker.c | 85 ++++++++++++++-- src/include/replication/logicalrelation.h | 28 ++++++ src/include/replication/worker_internal.h | 6 +- src/test/subscription/t/050_parallel_apply.pl | 71 ++++++++++++++ 6 files changed, 286 insertions(+), 9 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index 6e478a41e73..3e436b518bd 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -667,13 +667,17 @@ pa_launch_parallel_worker(void) /* * Allocate a parallel apply worker that will be used for the specified xid. * + * preceding_xid is the transaction received before the current one; pass + * InvalidTransactionId if none exists. + * * We first try to get an available worker from the pool, if any and then try * to launch a new worker. On successful allocation, remember the worker * information in the hash table so that we can get it later for processing the * streaming changes. */ void -pa_allocate_worker(TransactionId xid, bool stream_txn) +pa_allocate_worker(TransactionId xid, TransactionId preceding_xid, + bool stream_txn) { bool found; ParallelApplyWorkerInfo *winfo = NULL; @@ -710,6 +714,7 @@ pa_allocate_worker(TransactionId xid, bool stream_txn) SpinLockAcquire(&winfo->shared->mutex); winfo->shared->xact_state = PARALLEL_TRANS_UNKNOWN; winfo->shared->xid = xid; + winfo->shared->preceding_xid = preceding_xid; SpinLockRelease(&winfo->shared->mutex); winfo->in_use = true; diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c index c5a83b095fa..6d3e1827ae2 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -21,11 +21,16 @@ #include "access/genam.h" #include "access/table.h" #include "catalog/namespace.h" +#include "catalog/pg_proc.h" #include "catalog/pg_subscription_rel.h" +#include "catalog/pg_trigger.h" +#include "commands/trigger.h" #include "executor/executor.h" #include "nodes/makefuncs.h" +#include "optimizer/optimizer.h" #include "replication/logicalrelation.h" #include "replication/worker_internal.h" +#include "rewrite/rewriteHandler.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/syscache.h" @@ -82,6 +87,7 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid) if (entry->localreloid == reloid) { entry->localrelvalid = false; + entry->parallel_safety_valid = false; hash_seq_term(&status); break; } @@ -95,7 +101,10 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid) hash_seq_init(&status, LogicalRepRelMap); while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL) + { entry->localrelvalid = false; + entry->parallel_safety_valid = false; + } } } @@ -394,6 +403,7 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode) { /* Table was renamed or dropped. */ entry->localrelvalid = false; + entry->parallel_safety_valid = false; } else if (!entry->localrelvalid) { @@ -520,6 +530,90 @@ logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode) rel->localrel = NULL; } +/* + * Check whether changes on this relation can be applied in parallel. + * + * Parallel apply is unsafe if the table has any user-defined functions that may + * be executed during change application (e.g., in triggers, defaults, or + * constraint expressions). In theory, we should check the volatility of all + * such functions. + * + * Note that we do not check user-defined CHECK constraints here, as PostgreSQL + * already assumes they are immutable; we follow that same rule. + * + * If any mutable function is found, the relation is marked as globally unsafe, + * and the result is cached. + */ +void +logicalrep_rel_check_parallel_safety(LogicalRepRelMapEntry *entry) +{ + Relation localrel = entry->localrel; + TriggerDesc *trigdesc = localrel->trigdesc; + int ntriggers = trigdesc ? trigdesc->numtriggers : 0; + TupleDesc desc = RelationGetDescr(localrel); + + if (entry->parallel_safety_valid) + return; + + memset(entry->parallel_global_unsafe, 0, + sizeof(entry->parallel_global_unsafe)); + + /* Seek triggers one by one to see the volatility */ + for (int i = 0; i < ntriggers; i++) + { + Trigger *trigger = &trigdesc->triggers[i]; + + Assert(OidIsValid(trigger->tgfoid)); + + /* Skip if the trigger is internal (e.g., fkey triggers) */ + if (trigger->tgisinternal) + continue; + + /* Skip if the trigger is not enabled for logical replication */ + if (trigger->tgenabled == TRIGGER_DISABLED || + trigger->tgenabled == TRIGGER_FIRES_ON_ORIGIN) + continue; + + /* Check the volatility of the trigger. Exit if it is not immutable */ + if (func_volatile(trigger->tgfoid) != PROVOLATILE_IMMUTABLE) + { + if (TRIGGER_FOR_INSERT(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_INSERT] = true; + else if (TRIGGER_FOR_UPDATE(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_UPDATE] = true; + else if (TRIGGER_FOR_DELETE(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_DELETE] = true; + else if (TRIGGER_FOR_TRUNCATE(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_TRUNCATE] = true; + } + } + + /* Check column defaults for mutable functions */ + for (int attnum = 0; attnum < desc->natts; attnum++) + { + CompactAttribute *cattr = TupleDescCompactAttr(desc, attnum); + Expr *defexpr; + + if (cattr->attisdropped || cattr->attgenerated) + continue; + + /* Skip columns whose values come from remote change */ + if (entry->attrmap->attnums[attnum] >= 0) + continue; + + defexpr = (Expr *) build_column_default(entry->localrel, attnum + 1); + + if (contain_mutable_functions_after_planning(defexpr)) + { + /* Only INSERT operations are affected by column defaults */ + entry->parallel_global_unsafe[LRPA_INSERT] = true; + break; + } + } + + entry->parallel_safety_valid = true; +} + /* * Partition cache: look up partition LogicalRepRelMapEntry's * @@ -553,6 +647,7 @@ logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid) if (entry->relmapentry.localreloid == reloid) { entry->relmapentry.localrelvalid = false; + entry->relmapentry.parallel_safety_valid = false; hash_seq_term(&status); break; } @@ -566,7 +661,10 @@ logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid) hash_seq_init(&status, LogicalRepPartMap); while ((entry = (LogicalRepPartMapEntry *) hash_seq_search(&status)) != NULL) + { entry->relmapentry.localrelvalid = false; + entry->relmapentry.parallel_safety_valid = false; + } } } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 0afba345713..18d71ceabe6 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -1193,6 +1193,54 @@ check_and_record_rel_dependency(LogicalRepRelId relid, relentry->last_depended_xid = new_depended_xid; } +/* + * Check the parallel safety of applying the give action for the relation. If + * not safe, wait for preceding transactions to finish before proceeding with + * the current change. + * + * See logicalrep_rel_check_parallel_safety for details on how the safety is + * determined. + */ +static void +check_relation_parallel_apply_safety(LogicalRepRelMapEntry *relentry, + LogicalRepParallelAction action) +{ + /* Do not check parallel apply safety for streamed transactions */ + if (in_streamed_transaction) + return; + + /* Parallel apply only involves the leader and parallel apply workers */ + if (!am_leader_apply_worker() && !am_parallel_apply_worker()) + return; + + /* + * For partitioned tables, we only need to care if the target partition is + * parallel apply safe or not. + */ + if (relentry->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + return; + + logicalrep_rel_check_parallel_safety(relentry); + + /* Return if the action is safe for parallel apply */ + if (!relentry->parallel_global_unsafe[action]) + return; + + elog(DEBUG1, "found parallel unsafe change on table %u for action %d", + relentry->remoterel.remoteid, action); + + /* + * Wait for preceding transactions to finish before proceeding and reset the + * preceding_xid to InvalidTransactionId to avoid waiting for the same + * transaction again in the next change. + */ + if (TransactionIdIsValid(last_parallelized_remote_xid)) + { + pa_wait_for_depended_transaction(last_parallelized_remote_xid); + last_parallelized_remote_xid = InvalidTransactionId; + } +} + /* * Check dependencies related to the current change by determining if the * modification impacts the same row or table as another ongoing transaction. @@ -2048,7 +2096,7 @@ apply_handle_begin(StringInfo s) maybe_start_skipping_changes(begin_data.final_lsn); - pa_allocate_worker(remote_xid, false); + pa_allocate_worker(remote_xid, last_parallelized_remote_xid, false); apply_action = get_transaction_apply_action(remote_xid, &winfo); @@ -2080,6 +2128,8 @@ apply_handle_begin(StringInfo s) /* Hold the lock until the end of the transaction. */ pa_lock_transaction(MyParallelShared->xid, AccessExclusiveLock); pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_STARTED); + + last_parallelized_remote_xid = MyParallelShared->preceding_xid; break; default: @@ -2232,6 +2282,8 @@ apply_handle_commit(StringInfo s) pa_commit_transaction(); pa_unlock_transaction(remote_xid, AccessExclusiveLock); + + last_parallelized_remote_xid = InvalidTransactionId; break; default: @@ -2283,7 +2335,7 @@ apply_handle_begin_prepare(StringInfo s) maybe_start_skipping_changes(begin_data.prepare_lsn); - pa_allocate_worker(remote_xid, false); + pa_allocate_worker(remote_xid, last_parallelized_remote_xid, false); apply_action = get_transaction_apply_action(remote_xid, &winfo); @@ -2500,6 +2552,8 @@ apply_handle_prepare(StringInfo s) pa_unlock_transaction(MyParallelShared->xid, AccessExclusiveLock); pa_reset_subtrans(); + + last_parallelized_remote_xid = InvalidTransactionId; break; default: @@ -2926,7 +2980,7 @@ apply_handle_stream_start(StringInfo s) /* Try to allocate a worker for the streaming transaction. */ if (first_segment) - pa_allocate_worker(stream_xid, true); + pa_allocate_worker(stream_xid, last_parallelized_remote_xid, true); apply_action = get_transaction_apply_action(stream_xid, &winfo); @@ -3873,6 +3927,9 @@ apply_handle_insert(StringInfo s) /* Set relation for error callback */ apply_error_callback_arg.rel = rel; + /* Check if the relation is safe for parallel apply */ + check_relation_parallel_apply_safety(rel, LRPA_INSERT); + /* Initialize the executor state. */ edata = create_edata_for_relation(rel); estate = edata->estate; @@ -4030,6 +4087,9 @@ apply_handle_update(StringInfo s) /* Check if we can do the update. */ check_relation_updatable(rel); + /* Check if the relation is safe for parallel apply */ + check_relation_parallel_apply_safety(rel, LRPA_UPDATE); + /* * Make sure that any user-supplied code runs as the table owner, unless * the user has opted out of that behavior. @@ -4254,6 +4314,9 @@ apply_handle_delete(StringInfo s) /* Check if we can do the delete. */ check_relation_updatable(rel); + /* Check if the relation is safe for parallel apply */ + check_relation_parallel_apply_safety(rel, LRPA_DELETE); + /* * Make sure that any user-supplied code runs as the table owner, unless * the user has opted out of that behavior. @@ -4623,22 +4686,25 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, } MemoryContextSwitchTo(oldctx); + part_entry = logicalrep_partition_open(relmapentry, partrel, + attrmap); + /* Check if we can do the update or delete on the leaf partition. */ if (operation == CMD_UPDATE || operation == CMD_DELETE) - { - part_entry = logicalrep_partition_open(relmapentry, partrel, - attrmap); check_relation_updatable(part_entry); - } switch (operation) { case CMD_INSERT: + check_relation_parallel_apply_safety(part_entry, + LRPA_INSERT); apply_handle_insert_internal(edata, partrelinfo, remoteslot_part); break; case CMD_DELETE: + check_relation_parallel_apply_safety(part_entry, + LRPA_DELETE); apply_handle_delete_internal(edata, partrelinfo, remoteslot_part, part_entry->localindexoid); @@ -4660,6 +4726,9 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, EPQState epqstate; ConflictTupleInfo conflicttuple = {0}; + check_relation_parallel_apply_safety(part_entry, + LRPA_UPDATE); + /* Get the matching local tuple from the partition. */ found = FindReplTupleInLocalRel(edata, partrel, &part_entry->remoterel, @@ -4890,6 +4959,8 @@ apply_handle_truncate(StringInfo s) continue; } + check_relation_parallel_apply_safety(rel, LRPA_TRUNCATE); + remote_rels = lappend(remote_rels, rel); TargetPrivilegesCheck(rel->localrel, ACL_TRUNCATE); rels = lappend(rels, rel->localrel); diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index cd852337165..439243be742 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -16,6 +16,16 @@ #include "catalog/index.h" #include "replication/logicalproto.h" +typedef enum LogicalRepParallelAction +{ + LRPA_INSERT, + LRPA_UPDATE, + LRPA_DELETE, + LRPA_TRUNCATE +} LogicalRepParallelAction; + +#define LRPA_ACTION_COUNT (LRPA_TRUNCATE + 1) + typedef struct LogicalRepRelMapEntry { LogicalRepRelation remoterel; /* key is remoterel.remoteid */ @@ -45,6 +55,23 @@ typedef struct LogicalRepRelMapEntry * applying (see check_dependency_on_rel). */ TransactionId last_depended_xid; + + /* + * Per-operation safety cache for parallel apply. If + * parallel_global_unsafe[action] is true, that action cannot be applied in + * parallel for this relation. + * + * This cache must be computed by the worker that actually applies changes + * (via logicalrep_rel_check_parallel_safety), rather than solely by the + * leader. + * + * Relying on the leader alone is unsafe because the table could be altered + * before changes are dispatched to a parallel worker. Without re-validation + * by the parallel worker, it might incorrectly assume that parallel apply + * is safe for this relation. + */ + bool parallel_safety_valid; + bool parallel_global_unsafe[LRPA_ACTION_COUNT]; } LogicalRepRelMapEntry; extern void logicalrep_relmap_update(LogicalRepRelation *remoterel); @@ -56,6 +83,7 @@ extern LogicalRepRelMapEntry *logicalrep_partition_open(LogicalRepRelMapEntry *r Relation partrel, AttrMap *map); extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode); +extern void logicalrep_rel_check_parallel_safety(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/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 3dde899f465..4f6d699492f 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -205,6 +205,9 @@ typedef struct ParallelApplyWorkerShared */ dsa_handle parallel_apply_dsa_handle; dshash_table_handle parallelized_txns_handle; + + /* The remote transaction ID received before the current transaction */ + TransactionId preceding_xid; } ParallelApplyWorkerShared; /* @@ -353,7 +356,8 @@ extern void apply_error_callback(void *arg); extern void set_apply_error_context_origin(char *originname); /* Parallel apply worker setup and interactions */ -extern void pa_allocate_worker(TransactionId xid, bool stream_txn); +extern void pa_allocate_worker(TransactionId xid, TransactionId preceding_xid, + bool stream_txn); extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid); extern XLogRecPtr pa_get_last_commit_end(TransactionId xid, bool delete_entry, bool *skipped_write); diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl index a6343d57bc5..c90a72682e5 100644 --- a/src/test/subscription/t/050_parallel_apply.pl +++ b/src/test/subscription/t/050_parallel_apply.pl @@ -456,4 +456,75 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab"); is ($result, 5011, 'inserts are replicated to subscriber'); +################################################## +# Test that mutable user-defined triggers force apply-time waiting so +# conflicting trigger side effects are serialized. +################################################## + +# Truncate the data for upcoming tests +$node_publisher->safe_psql('postgres', "TRUNCATE TABLE regress_tab;"); +$node_publisher->wait_for_catchup('regress_sub'); + +$node_subscriber->safe_psql('postgres', qq[ + CREATE OR REPLACE FUNCTION regress_trigger_guard_fn() + RETURNS trigger + LANGUAGE plpgsql + AS \$\$ + BEGIN + IF NOT EXISTS (SELECT 1 FROM public.regress_tab WHERE id = 1) THEN + UPDATE public.regress_tab SET id = 2 WHERE id = 1; + END IF; + RETURN NEW; + END; + \$\$; + + CREATE TRIGGER regress_trigger_guard_tg + BEFORE INSERT ON regress_tab + FOR EACH ROW + EXECUTE FUNCTION regress_trigger_guard_fn(); +]); + +# Ensure trigger fires during logical replication apply. +$node_subscriber->safe_psql('postgres', + "ALTER TABLE regress_tab ENABLE REPLICA TRIGGER regress_trigger_guard_tg;" +); + +# Hold the first parallel worker just before commit. +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('parallel-worker-before-commit','wait');" +); + +# TX-1: first insert; worker pauses at before-commit injection point. +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_tab VALUES (1);"); + +$node_subscriber->wait_for_event('logical replication parallel worker', + 'parallel-worker-before-commit'); + +$offset = -s $node_subscriber->logfile; + +# TX-2: second insert. For unsafe trigger tables, this must wait before apply. +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_tab VALUES (2);"); + +$node_subscriber->wait_for_log(qr/found parallel unsafe change on table [1-9][0-9]* for action [0-9]+/, $offset); + +$str = $node_subscriber->wait_for_log(qr/wait for depended xid ([1-9][0-9]+)/, $offset); +$xid = $str =~ /wait for depended xid ([1-9][0-9]+)/; + +ok(1, "mutable trigger relation waits before apply in parallel mode"); + +# Resume workers. +$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"); +is ($result, 2, 'inserts are replicated to subscriber'); + done_testing(); -- 2.43.0