From 9479ff3f1f7360980e06b95a7f5db271e6c07436 Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Tue, 11 Aug 2026 15:15:57 +0800 Subject: [PATCH v23 10/12] 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 or stored generated expressions 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. Author: Zhijie Hou Author: Hayato Kuroda --- .../replication/logical/applyparallelworker.c | 6 +- src/backend/replication/logical/relation.c | 111 ++++++++++++++++++ src/backend/replication/logical/worker.c | 107 +++++++++++++++-- src/include/replication/logicalrelation.h | 28 +++++ src/include/replication/worker_internal.h | 11 +- src/test/subscription/t/050_parallel_apply.pl | 81 +++++++++++++ 6 files changed, 335 insertions(+), 9 deletions(-) diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index efd1d3c177a..78db7b1fb46 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -695,13 +695,16 @@ 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) +pa_allocate_worker(TransactionId xid, TransactionId preceding_xid) { bool found; ParallelApplyWorkerInfo *winfo = NULL; @@ -738,6 +741,7 @@ pa_allocate_worker(TransactionId xid) 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 8a0aaf4a217..ec1f917071b 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -21,12 +21,17 @@ #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 "libpq/pqformat.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" @@ -125,6 +130,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; } @@ -138,7 +144,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; + } } } @@ -442,6 +451,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) { @@ -568,6 +578,101 @@ 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, stored + * generated expressions, 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. + * A multi-event trigger (e.g., BEFORE INSERT OR UPDATE) must mark + * every action it fires for, so these checks are independent ifs + * rather than an else-if chain. + */ + if (func_volatile(trigger->tgfoid) != PROVOLATILE_IMMUTABLE) + { + if (TRIGGER_FOR_INSERT(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_INSERT] = true; + if (TRIGGER_FOR_UPDATE(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_UPDATE] = true; + if (TRIGGER_FOR_DELETE(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_DELETE] = true; + if (TRIGGER_FOR_TRUNCATE(trigger->tgtype)) + entry->parallel_global_unsafe[LRPA_TRUNCATE] = true; + } + } + + /* + * Check column defaults and stored generated expressions for mutable + * functions. + */ + for (int attnum = 0; attnum < desc->natts; attnum++) + { + Form_pg_attribute att = TupleDescAttr(desc, attnum); + Expr *defexpr; + + if (att->attisdropped) + continue; + + if (!att->atthasdef && att->attgenerated != ATTRIBUTE_GENERATED_STORED) + 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 * @@ -598,7 +703,10 @@ logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid) */ entry = hash_search(LogicalRepPartMap, &reloid, HASH_FIND, NULL); if (entry != NULL) + { entry->relmapentry.localrelvalid = false; + entry->relmapentry.parallel_safety_valid = false; + } } else { @@ -608,7 +716,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 6c85d498040..9bf85a19e4c 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -747,6 +747,8 @@ static void send_internal_dependencies(ParallelApplyWorkerInfo *winfo, List *dep static void maintain_commit_order_dependency(ParallelApplyWorkerInfo *winfo); +static TransactionId get_last_parallelized_xid(void); + /* * Compute the hash value for entries in the replica_identity_table. */ @@ -1221,6 +1223,63 @@ 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 all preceding transactions to finish before applying the + * unsafe change. A parallel apply worker waits for the transaction that + * was received immediately before its own, the leader waits for the last + * parallelized transaction; since commit order is preserved, this + * transitively waits for all preceding transactions. The cached XID is + * reset to avoid waiting for the same transaction again for the next + * change. + */ + if (am_parallel_apply_worker()) + { + if (TransactionIdIsValid(MyParallelShared->preceding_xid)) + { + pa_wait_for_depended_transaction(MyParallelShared->preceding_xid); + MyParallelShared->preceding_xid = InvalidTransactionId; + } + } + else + maintain_commit_order_dependency(NULL); +} + /* * Check dependencies related to the current change by determining if the * modification impacts the same row or table as another ongoing transaction. @@ -2085,7 +2144,7 @@ apply_handle_begin(StringInfo s) maybe_start_skipping_changes(begin_data.final_lsn); - pa_allocate_worker(remote_xid); + pa_allocate_worker(remote_xid, get_last_parallelized_xid()); apply_action = get_transaction_apply_action(remote_xid, &winfo); @@ -2277,7 +2336,7 @@ apply_handle_begin_prepare(StringInfo s) maybe_start_skipping_changes(begin_data.prepare_lsn); - pa_allocate_worker(remote_xid); + pa_allocate_worker(remote_xid, get_last_parallelized_xid()); apply_action = get_transaction_apply_action(remote_xid, &winfo); @@ -2891,7 +2950,7 @@ apply_handle_stream_start(StringInfo s) /* Try to allocate a worker for the streaming transaction. */ if (first_segment) - pa_allocate_worker(stream_xid); + pa_allocate_worker(stream_xid, get_last_parallelized_xid()); apply_action = get_transaction_apply_action(stream_xid, &winfo); @@ -3834,6 +3893,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; @@ -3991,6 +4053,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. @@ -4215,6 +4280,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. @@ -4584,22 +4652,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); @@ -4621,6 +4692,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, @@ -4851,6 +4925,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); @@ -7678,6 +7754,23 @@ send_internal_dependencies(ParallelApplyWorkerInfo *winfo, List *depends_on_xids pfree(dependencies.data); } +/* + * Return the remote XID of the last transaction whose commit was dispatched + * to a parallel apply worker, or InvalidTransactionId if there is none. + */ +static TransactionId +get_last_parallelized_xid(void) +{ + FlushPosition *last_txn_node; + + if (dlist_is_empty(&lsn_mapping)) + return InvalidTransactionId; + + last_txn_node = dlist_tail_element(FlushPosition, node, &lsn_mapping); + + return last_txn_node->pa_remote_xid; +} + /* * Ensure commit order is preserved for transactions applied by either a * parallel worker or the leader. diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index 621b1ffb706..1b2a8e81dba 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 void logicalrep_write_all_internal_rels(StringInfo out, int *num_rels); diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 5bcc5b1aa89..d64a937ce64 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -158,6 +158,15 @@ typedef struct ParallelApplyWorkerShared TransactionId xid; + /* + * The remote XID of the transaction received immediately before the + * current one, or InvalidTransactionId if there is none. Parallel apply + * workers use it to wait for all preceding transactions before applying + * a change that is not safe for parallel apply (see + * check_relation_parallel_apply_safety). + */ + TransactionId preceding_xid; + /* * State used to ensure commit ordering. * @@ -351,7 +360,7 @@ 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); +extern void pa_allocate_worker(TransactionId xid, TransactionId preceding_xid); extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid); extern void pa_detach_all_error_mq(void); diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl index 527efe09913..fcae6892e35 100644 --- a/src/test/subscription/t/050_parallel_apply.pl +++ b/src/test/subscription/t/050_parallel_apply.pl @@ -556,4 +556,85 @@ $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'); + +# The TRUNCATE invalidated the publisher's relmap entry for regress_tab, so +# the next transaction touching the table carries a fresh RELATION message. +# A RELATION message records a table-wide dependency, which would make TX-2 +# below wait for TX-1 before even applying its change, hiding the +# unsafe-trigger wait this test exercises. Absorb the RELATION message with +# a dummy transaction (net zero rows) and let it commit fully. +$node_publisher->safe_psql('postgres', + "INSERT INTO regress_tab VALUES (100); DELETE FROM regress_tab WHERE id = 100;"); +$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