public inbox for [email protected]
help / color / mirror / Atom feedFrom: Hayato Kuroda (Fujitsu) <[email protected]>
To: PostgreSQL Hackers <[email protected]>
Cc: Tomas Vondra <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: 'Dilip Kumar' <[email protected]>
Cc: 'Andrei Lepikhov' <[email protected]>
Cc: Zhijie Hou (Fujitsu) <[email protected]>
Cc: 'wenhui qiu' <[email protected]>
Subject: RE: Parallel Apply
Date: Fri, 30 Jan 2026 11:18:07 +0000
Message-ID: <TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <TY7PR01MB145541DD6F5BE4CD403BA39FDF586A@TY7PR01MB14554.jpnprd01.prod.outlook.com>
References: <CAA4eK1+SEus_6vQay9TF_r4ow+E-Q7LYNLfsD78HaOsLSgppxQ@mail.gmail.com>
<CADzfLwXnJ1H4HncFugGPdnm8t+aUAU4E-yfi1j3BbiP5VfXD8g@mail.gmail.com>
<OSCPR01MB1496669CF8792F9D9FAE4D3A4F5D6A@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<CAA4eK1KcfKkm32-owh+Aet_x7Y-XnDVPRaeWpUj-LqpR6Nbbuw@mail.gmail.com>
<OSCPR01MB14966CE2796810D402EBC0964F5D6A@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<[email protected]>
<OSCPR01MB149667D359F65822748A4BC5DF5DBA@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<CAFiTN-sHLhDk2kusac2KWMB1f-xVw8HuM=j8p9njTWOmXjvbqw@mail.gmail.com>
<OSCPR01MB149662F6753E26278F340438CF5D8A@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OSCPR01MB149666AFA68C5D2A46F08BB94F5B4A@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OSCPR01MB14966B464C274665B21FF4320F5B0A@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OSCPR01MB14966585027AE9B53CD85E02FF5B0A@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<TY7PR01MB145541DD6F5BE4CD403BA39FDF586A@TY7PR01MB14554.jpnprd01.prod.outlook.com>
Dear Hackers,
Here is a rebased version, plus 0009 patch.
0009 contains changes to handle dependencies based on the foreign keys. Without
this, parallel apply can violate foreign key constraints if referenced tuples
are committed after referencing tuples, leading to replication failures.
This patch extends the dependency hash to track dependencies via FKs. Since
referenced columns must be unique, information must be already stored to the
hash if they have been modified. Based on the point, the leader apply worker
checks to determine whether values in referencing columns have already been
registered or not if applying tuples refer some columns, and regards that there
is a dependency if found.
Best regards,
Hayato Kuroda
FUJITSU LIMITED
Attachments:
[application/octet-stream] v8-0001-Introduce-new-type-of-logical-replication-message.patch (6.0K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/2-v8-0001-Introduce-new-type-of-logical-replication-message.patch)
download | inline diff:
From e121075c26b1a57c820d2ff57a840f8e3a4f383b Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 1 Dec 2025 10:37:27 +0900
Subject: [PATCH v8 1/9] Introduce new type of logical replication messages to
track dependencies
This patch introduces two logical replication messages,
LOGICAL_REP_MSG_INTERNAL_DEPENDENCY and LOGICAL_REP_MSG_INTERNAL_RELATION.
Apart from other messages, they are not sent by walsnders; the leader worker
sends to parallel workers based on the needs.
LOGICAL_REP_MSG_INTERNAL_DEPENDENCY ensures that dependent transactions are
committed in the correct order. It has a list of transaction IDs that parallel
workers must wait for. The message type would be generated when the leader
detects a dependency between the current and other transactions, or just before
the COMMIT message. The latter one is used to preserve the commit ordering
between the publisher and the subscriber.
LOGICAL_REP_MSG_INTERNAL_RELATION is used to synchronize the relation
information between the leader and parallel workers. It has a list of relations
that the leader already knows, and parallel workers also update the relmap in
response to the message. This type of message is generated when the leader
allocates a new parallel worker to the transaction, or when the publisher sends
additional RELATION messages.
Author: Hou Zhijie <[email protected]>
Author: Hayato Kuroda <[email protected]>
---
.../replication/logical/applyparallelworker.c | 16 ++++++
src/backend/replication/logical/proto.c | 4 ++
src/backend/replication/logical/worker.c | 49 +++++++++++++++++++
src/include/replication/logicalproto.h | 2 +
src/include/replication/worker_internal.h | 4 ++
5 files changed, 75 insertions(+)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 8a01f16a2ca..748fd0c50a3 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -1645,3 +1645,19 @@ pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
pa_free_worker(winfo);
}
+
+/*
+ * Wait for the given transaction to finish.
+ */
+void
+pa_wait_for_depended_transaction(TransactionId xid)
+{
+ elog(DEBUG1, "wait for depended xid %u", xid);
+
+ for (;;)
+ {
+ /* XXX wait until given transaction is finished */
+ }
+
+ elog(DEBUG1, "finish waiting for depended xid %u", xid);
+}
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 3950dd0cf46..66f930c8b06 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -1253,6 +1253,10 @@ logicalrep_message_type(LogicalRepMsgType action)
return "STREAM ABORT";
case LOGICAL_REP_MSG_STREAM_PREPARE:
return "STREAM PREPARE";
+ case LOGICAL_REP_MSG_INTERNAL_DEPENDENCY:
+ return "INTERNAL DEPENDENCY";
+ case LOGICAL_REP_MSG_INTERNAL_RELATION:
+ return "INTERNAL RELATION";
}
/*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 32725c48623..43dce2b5622 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -629,6 +629,47 @@ static TransApplyAction get_transaction_apply_action(TransactionId xid,
static void on_exit_clear_xact_state(int code, Datum arg);
+/*
+ * Handle internal dependency information.
+ *
+ * Wait for all transactions listed in the message to commit.
+ */
+static void
+apply_handle_internal_dependency(StringInfo s)
+{
+ int nxids = pq_getmsgint(s, 4);
+
+ for (int i = 0; i < nxids; i++)
+ {
+ TransactionId xid = pq_getmsgint(s, 4);
+
+ pa_wait_for_depended_transaction(xid);
+ }
+}
+
+/*
+ * Handle internal relation information.
+ *
+ * Update all relation details in the relation map cache.
+ */
+static void
+apply_handle_internal_relation(StringInfo s)
+{
+ int num_rels;
+
+ num_rels = pq_getmsgint(s, 4);
+
+ for (int i = 0; i < num_rels; i++)
+ {
+ LogicalRepRelation *rel = logicalrep_read_rel(s);
+
+ logicalrep_relmap_update(rel);
+
+ elog(DEBUG1, "parallel apply worker worker init relmap for %s",
+ rel->relname);
+ }
+}
+
/*
* Form the origin name for the subscription.
*
@@ -3868,6 +3909,14 @@ apply_dispatch(StringInfo s)
apply_handle_stream_prepare(s);
break;
+ case LOGICAL_REP_MSG_INTERNAL_RELATION:
+ apply_handle_internal_relation(s);
+ break;
+
+ case LOGICAL_REP_MSG_INTERNAL_DEPENDENCY:
+ apply_handle_internal_dependency(s);
+ break;
+
default:
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 058a955e20c..9042470f500 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -75,6 +75,8 @@ typedef enum LogicalRepMsgType
LOGICAL_REP_MSG_STREAM_COMMIT = 'c',
LOGICAL_REP_MSG_STREAM_ABORT = 'A',
LOGICAL_REP_MSG_STREAM_PREPARE = 'p',
+ LOGICAL_REP_MSG_INTERNAL_DEPENDENCY = 'd',
+ LOGICAL_REP_MSG_INTERNAL_RELATION = 'i',
} LogicalRepMsgType;
/*
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index c1285fdd1bc..eb937be15ce 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -359,6 +359,8 @@ extern void pa_decr_and_wait_stream_block(void);
extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
XLogRecPtr remote_lsn);
+extern void pa_wait_for_depended_transaction(TransactionId xid);
+
#define isParallelApplyWorker(worker) ((worker)->in_use && \
(worker)->type == WORKERTYPE_PARALLEL_APPLY)
#define isTableSyncWorker(worker) ((worker)->in_use && \
@@ -366,6 +368,8 @@ extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
#define isSequenceSyncWorker(worker) ((worker)->in_use && \
(worker)->type == WORKERTYPE_SEQUENCESYNC)
+#define PARALLEL_APPLY_INTERNAL_MESSAGE 'i'
+
static inline bool
am_tablesync_worker(void)
{
--
2.47.3
[application/octet-stream] v8-0002-Introduce-a-shared-hash-table-to-store-paralleliz.patch (9.1K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/3-v8-0002-Introduce-a-shared-hash-table-to-store-paralleliz.patch)
download | inline diff:
From 1c460af3d99d74771df3df05632096901bbc0c4a Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 1 Dec 2025 16:28:38 +0900
Subject: [PATCH v8 2/9] Introduce a shared hash table to store parallelized
transactions
This hash table is used for ensuring that parallel workers wait until dependent
transactions are committed.
The shared hash table contains transaction IDs that the leader allocated to
parallel workers. The hash entries are inserted with a remote XID when the
leader bypasses remote transactions to parallel apply workers. Entries are
deleted when parallel workers are committed to corresponding transactions.
When the parallel worker tries to wait for other transactions, it checks the
hash table for the remote XIDs. The process can go ahead only when entries are
removed from the hash.
Author: Hou Zhijie <[email protected]>
Author: Hayato Kuroda <[email protected]>
---
.../replication/logical/applyparallelworker.c | 100 +++++++++++++++++-
.../utils/activity/wait_event_names.txt | 1 +
src/include/replication/worker_internal.h | 4 +
src/include/storage/lwlocklist.h | 1 +
src/tools/pgindent/typedefs.list | 1 +
5 files changed, 106 insertions(+), 1 deletion(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 748fd0c50a3..de2fda72818 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -218,12 +218,35 @@ typedef struct ParallelApplyWorkerEntry
ParallelApplyWorkerInfo *winfo;
} ParallelApplyWorkerEntry;
+/* an entry in the parallelized_txns shared hash table */
+typedef struct ParallelizedTxnEntry
+{
+ TransactionId xid; /* Hash key */
+} ParallelizedTxnEntry;
+
/*
* A hash table used to cache the state of streaming transactions being applied
* by the parallel apply workers.
*/
static HTAB *ParallelApplyTxnHash = NULL;
+/*
+ * A hash table used to track the parallelized transactions that could be
+ * depended on by other transactions.
+ */
+static dsa_area *parallel_apply_dsa_area = NULL;
+static dshash_table *parallelized_txns = NULL;
+
+/* parameters for the parallelized_txns shared hash table */
+static const dshash_parameters dsh_params = {
+ sizeof(TransactionId),
+ sizeof(ParallelizedTxnEntry),
+ dshash_memcmp,
+ dshash_memhash,
+ dshash_memcpy,
+ LWTRANCHE_PARALLEL_APPLY_DSA
+};
+
/*
* A list (pool) of active parallel apply workers. The information for
* the new worker is added to the list after successfully launching it. The
@@ -257,6 +280,8 @@ static List *subxactlist = NIL;
static void pa_free_worker_info(ParallelApplyWorkerInfo *winfo);
static ParallelTransState pa_get_xact_state(ParallelApplyWorkerShared *wshared);
static PartialFileSetState pa_get_fileset_state(void);
+static void pa_attach_parallelized_txn_hash(dsa_handle *pa_dsa_handle,
+ dshash_table_handle *pa_dshash_handle);
/*
* Returns true if it is OK to start a parallel apply worker, false otherwise.
@@ -334,6 +359,15 @@ pa_setup_dsm(ParallelApplyWorkerInfo *winfo)
shm_mq *mq;
Size queue_size = DSM_QUEUE_SIZE;
Size error_queue_size = DSM_ERROR_QUEUE_SIZE;
+ dsa_handle parallel_apply_dsa_handle;
+ dshash_table_handle parallelized_txns_handle;
+
+ pa_attach_parallelized_txn_hash(¶llel_apply_dsa_handle,
+ ¶llelized_txns_handle);
+
+ if (parallel_apply_dsa_handle == DSA_HANDLE_INVALID ||
+ parallelized_txns_handle == DSHASH_HANDLE_INVALID)
+ return false;
/*
* Estimate how much shared memory we need.
@@ -369,6 +403,8 @@ pa_setup_dsm(ParallelApplyWorkerInfo *winfo)
pg_atomic_init_u32(&(shared->pending_stream_count), 0);
shared->last_commit_end = InvalidXLogRecPtr;
shared->fileset_state = FS_EMPTY;
+ shared->parallel_apply_dsa_handle = parallel_apply_dsa_handle;
+ shared->parallelized_txns_handle = parallelized_txns_handle;
shm_toc_insert(toc, PARALLEL_APPLY_KEY_SHARED, shared);
@@ -864,6 +900,8 @@ ParallelApplyWorkerMain(Datum main_arg)
shm_mq *mq;
shm_mq_handle *mqh;
shm_mq_handle *error_mqh;
+ dsa_handle pa_dsa_handle;
+ dshash_table_handle pa_dshash_handle;
ReplOriginId originid;
int worker_slot = DatumGetInt32(main_arg);
char originname[NAMEDATALEN];
@@ -951,6 +989,8 @@ ParallelApplyWorkerMain(Datum main_arg)
InitializingApplyWorker = false;
+ pa_attach_parallelized_txn_hash(&pa_dsa_handle, &pa_dshash_handle);
+
/* Setup replication origin tracking. */
StartTransactionCommand();
ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
@@ -1646,6 +1686,51 @@ pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
pa_free_worker(winfo);
}
+/*
+ * Attach to the shared hash table for parallelized transactions.
+ */
+static void
+pa_attach_parallelized_txn_hash(dsa_handle *pa_dsa_handle,
+ dshash_table_handle *pa_dshash_handle)
+{
+ MemoryContext oldctx;
+
+ if (parallelized_txns)
+ {
+ Assert(parallel_apply_dsa_area);
+ *pa_dsa_handle = dsa_get_handle(parallel_apply_dsa_area);
+ *pa_dshash_handle = dshash_get_hash_table_handle(parallelized_txns);
+ return;
+ }
+
+ /* Be sure any local memory allocated by DSA routines is persistent. */
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ if (am_leader_apply_worker())
+ {
+ /* Initialize dynamic shared hash table for last-start times. */
+ parallel_apply_dsa_area = dsa_create(LWTRANCHE_PARALLEL_APPLY_DSA);
+ dsa_pin(parallel_apply_dsa_area);
+ dsa_pin_mapping(parallel_apply_dsa_area);
+ parallelized_txns = dshash_create(parallel_apply_dsa_area, &dsh_params, NULL);
+
+ /* Store handles in shared memory for other backends to use. */
+ *pa_dsa_handle = dsa_get_handle(parallel_apply_dsa_area);
+ *pa_dshash_handle = dshash_get_hash_table_handle(parallelized_txns);
+ }
+ else if (am_parallel_apply_worker())
+ {
+ /* Attach to existing dynamic shared hash table. */
+ parallel_apply_dsa_area = dsa_attach(MyParallelShared->parallel_apply_dsa_handle);
+ dsa_pin_mapping(parallel_apply_dsa_area);
+ parallelized_txns = dshash_attach(parallel_apply_dsa_area, &dsh_params,
+ MyParallelShared->parallelized_txns_handle,
+ NULL);
+ }
+
+ MemoryContextSwitchTo(oldctx);
+}
+
/*
* Wait for the given transaction to finish.
*/
@@ -1656,7 +1741,20 @@ pa_wait_for_depended_transaction(TransactionId xid)
for (;;)
{
- /* XXX wait until given transaction is finished */
+ ParallelizedTxnEntry *txn_entry;
+
+ txn_entry = dshash_find(parallelized_txns, &xid, false);
+
+ /* The entry is removed only if the transaction is committed */
+ if (txn_entry == NULL)
+ break;
+
+ dshash_release_lock(parallelized_txns, txn_entry);
+
+ pa_lock_transaction(xid, AccessShareLock);
+ pa_unlock_transaction(xid, AccessShareLock);
+
+ CHECK_FOR_INTERRUPTS();
}
elog(DEBUG1, "finish waiting for depended xid %u", xid);
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 5537a2d2530..34ee1a26df5 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -410,6 +410,7 @@ SubtransSLRU "Waiting to access the sub-transaction SLRU cache."
XactSLRU "Waiting to access the transaction status SLRU cache."
ParallelVacuumDSA "Waiting for parallel vacuum dynamic shared memory allocation."
AioUringCompletion "Waiting for another process to complete IO via io_uring."
+ParallelApplyDSA "Waiting for parallel apply dynamic shared memory allocation."
# No "ABI_compatibility" region here as WaitEventLWLock has its own C code.
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index eb937be15ce..411e2c537c7 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -15,6 +15,7 @@
#include "access/xlogdefs.h"
#include "catalog/pg_subscription.h"
#include "datatype/timestamp.h"
+#include "lib/dshash.h"
#include "miscadmin.h"
#include "replication/logicalrelation.h"
#include "replication/walreceiver.h"
@@ -197,6 +198,9 @@ typedef struct ParallelApplyWorkerShared
*/
PartialFileSetState fileset_state;
FileSet fileset;
+
+ dsa_handle parallel_apply_dsa_handle;
+ dshash_table_handle parallelized_txns_handle;
} ParallelApplyWorkerShared;
/*
diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h
index e94ebce95b9..1c138a73a7b 100644
--- a/src/include/storage/lwlocklist.h
+++ b/src/include/storage/lwlocklist.h
@@ -137,3 +137,4 @@ PG_LWLOCKTRANCHE(SUBTRANS_SLRU, SubtransSLRU)
PG_LWLOCKTRANCHE(XACT_SLRU, XactSLRU)
PG_LWLOCKTRANCHE(PARALLEL_VACUUM_DSA, ParallelVacuumDSA)
PG_LWLOCKTRANCHE(AIO_URING_COMPLETION, AioUringCompletion)
+PG_LWLOCKTRANCHE(PARALLEL_APPLY_DSA, ParallelApplyDSA)
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 34374df0d67..d99c1996efd 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2095,6 +2095,7 @@ ParallelHashJoinBatch
ParallelHashJoinBatchAccessor
ParallelHashJoinState
ParallelIndexScanDesc
+ParallelizedTxnEntry
ParallelSlot
ParallelSlotArray
ParallelSlotResultHandler
--
2.47.3
[application/octet-stream] v8-0003-Introduce-a-local-hash-table-to-store-replica-ide.patch (28.3K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/4-v8-0003-Introduce-a-local-hash-table-to-store-replica-ide.patch)
download | inline diff:
From 8b1d0353237933c49a8f53d7f199ace32a07e3c7 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 1 Dec 2025 16:39:02 +0900
Subject: [PATCH v8 3/9] Introduce a local hash table to store replica
identities
This local hash table on the leader is used for detecting dependencies between
transactions.
The hash contains the Replica Identity (RI) as a key and the remote XID that
modified the corresponding tuple. The hash entries are inserted when the leader
finds an RI from a replication message. Entries are deleted when transactions
committed by parallel workers are gathered, or the number of entries exceeds the
limit.
When the leader sends replication changes to parallel workers, it checks whether
other transactions have already used the RI associated with the change. If
something is found, the leader treats it as a dependent transaction and notifies
parallel workers to wait until it finishes via LOGICAL_REP_MSG_INTERNAL_DEPENDENCY.
Author: Hou Zhijie <[email protected]>
Author: Hayato Kuroda <[email protected]>
---
.../replication/logical/applyparallelworker.c | 123 +++-
src/backend/replication/logical/relation.c | 24 +
src/backend/replication/logical/worker.c | 616 +++++++++++++++++-
src/include/replication/logicalrelation.h | 3 +
src/include/replication/worker_internal.h | 8 +-
src/tools/pgindent/typedefs.list | 3 +
6 files changed, 774 insertions(+), 3 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index de2fda72818..d462674b5cd 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -216,6 +216,7 @@ typedef struct ParallelApplyWorkerEntry
{
TransactionId xid; /* Hash key -- must be first */
ParallelApplyWorkerInfo *winfo;
+ XLogRecPtr local_end;
} ParallelApplyWorkerEntry;
/* an entry in the parallelized_txns shared hash table */
@@ -504,7 +505,7 @@ pa_launch_parallel_worker(void)
* streaming changes.
*/
void
-pa_allocate_worker(TransactionId xid)
+pa_allocate_worker(TransactionId xid, bool stream_txn)
{
bool found;
ParallelApplyWorkerInfo *winfo = NULL;
@@ -545,7 +546,9 @@ pa_allocate_worker(TransactionId xid)
winfo->in_use = true;
winfo->serialize_changes = false;
+ winfo->stream_txn = stream_txn;
entry->winfo = winfo;
+ entry->local_end = InvalidXLogRecPtr;
}
/*
@@ -742,6 +745,73 @@ pa_process_spooled_messages_if_required(void)
return true;
}
+/*
+ * Get the local end LSN for a transaction applied by a parallel apply worker.
+ *
+ * Set delete_entry to true if you intend to remove the transaction from the
+ * ParallelApplyTxnHash after collecting its LSN.
+ *
+ * If the parallel apply worker did not write any changes during the transaction
+ * application due to situations like update/delete_missing or a before trigger,
+ * the *skipped_write will be set to true.
+ */
+XLogRecPtr
+pa_get_last_commit_end(TransactionId xid, bool delete_entry, bool *skipped_write)
+{
+ bool found;
+ ParallelApplyWorkerEntry *entry;
+ ParallelApplyWorkerInfo *winfo;
+
+ Assert(TransactionIdIsValid(xid));
+
+ if (skipped_write)
+ *skipped_write = false;
+
+ /* Find an entry for the requested transaction. */
+ entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, &found);
+
+ if (!found)
+ return InvalidXLogRecPtr;
+
+ /*
+ * If worker info is NULL, it indicates that the worker has been reused
+ * for handling other transactions. Consequently, the local end LSN has
+ * already been collected and saved in entry->local_end.
+ */
+ winfo = entry->winfo;
+ if (winfo == NULL)
+ {
+ if (delete_entry &&
+ !hash_search(ParallelApplyTxnHash, &xid, HASH_REMOVE, NULL))
+ elog(ERROR, "hash table corrupted");
+
+ if (skipped_write)
+ *skipped_write = XLogRecPtrIsInvalid(entry->local_end);
+
+ return entry->local_end;
+ }
+
+ /* Return InvalidXLogRecPtr if the transaction is still in progress */
+ if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
+ return InvalidXLogRecPtr;
+
+ /* Collect the local end LSN from the worker's shared memory area */
+ entry->local_end = winfo->shared->last_commit_end;
+ entry->winfo = NULL;
+
+ if (skipped_write)
+ *skipped_write = XLogRecPtrIsInvalid(entry->local_end);
+
+ elog(DEBUG1, "store local commit %X/%X end to txn entry: %u",
+ LSN_FORMAT_ARGS(entry->local_end), xid);
+
+ if (delete_entry &&
+ !hash_search(ParallelApplyTxnHash, &xid, HASH_REMOVE, NULL))
+ elog(ERROR, "hash table corrupted");
+
+ return entry->local_end;
+}
+
/*
* Interrupt handler for main loop of parallel apply worker.
*/
@@ -1686,6 +1756,26 @@ pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
pa_free_worker(winfo);
}
+bool
+pa_transaction_committed(TransactionId xid)
+{
+ bool found;
+ ParallelApplyWorkerEntry *entry;
+
+ Assert(TransactionIdIsValid(xid));
+
+ /* Find an entry for the requested transaction */
+ entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, &found);
+
+ if (!found)
+ return true;
+
+ if (!entry->winfo)
+ return true;
+
+ return pa_get_xact_state(entry->winfo->shared) == PARALLEL_TRANS_FINISHED;
+}
+
/*
* Attach to the shared hash table for parallelized transactions.
*/
@@ -1731,6 +1821,37 @@ pa_attach_parallelized_txn_hash(dsa_handle *pa_dsa_handle,
MemoryContextSwitchTo(oldctx);
}
+/*
+ * Record in-progress transactions from the given list that are being depended
+ * on into the shared hash table.
+ */
+void
+pa_record_dependency_on_transactions(List *depends_on_xids)
+{
+ foreach_xid(xid, depends_on_xids)
+ {
+ bool found;
+ ParallelApplyWorkerEntry *winfo_entry;
+ ParallelApplyWorkerInfo *winfo;
+ ParallelizedTxnEntry *txn_entry;
+
+ winfo_entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, &found);
+ winfo = winfo_entry->winfo;
+
+ txn_entry = dshash_find_or_insert(parallelized_txns, &xid, &found);
+
+ /*
+ * If the transaction has been committed now, remove the entry,
+ * otherwise the parallel apply worker will remove the entry once
+ * committed the transaction.
+ */
+ if (pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED)
+ dshash_delete_entry(parallelized_txns, txn_entry);
+ else
+ dshash_release_lock(parallelized_txns, txn_entry);
+ }
+}
+
/*
* Wait for the given transaction to finish.
*/
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index 0b1d80b5b0f..ab6313eb1bc 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -960,3 +960,27 @@ FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel,
return InvalidOid;
}
+
+/*
+ * Get the LogicalRepRelMapEntry corresponding to the given relid without
+ * opening the local relation.
+ */
+LogicalRepRelMapEntry *
+logicalrep_get_relentry(LogicalRepRelId remoteid)
+{
+ LogicalRepRelMapEntry *entry;
+ bool found;
+
+ if (LogicalRepRelMap == NULL)
+ logicalrep_relmap_init();
+
+ /* Search for existing entry. */
+ entry = hash_search(LogicalRepRelMap, (void *) &remoteid,
+ HASH_FIND, &found);
+
+ if (!found)
+ elog(DEBUG1, "no relation map entry for remote relation ID %u",
+ remoteid);
+
+ return entry;
+}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 43dce2b5622..c2967caf1e5 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -303,6 +303,7 @@ typedef struct FlushPosition
dlist_node node;
XLogRecPtr local_end;
XLogRecPtr remote_end;
+ TransactionId pa_remote_xid;
} FlushPosition;
static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping);
@@ -544,6 +545,49 @@ typedef struct ApplySubXactData
static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL};
+typedef struct ReplicaIdentityKey
+{
+ Oid relid;
+ LogicalRepTupleData *data;
+} ReplicaIdentityKey;
+
+typedef struct ReplicaIdentityEntry
+{
+ ReplicaIdentityKey *keydata;
+ TransactionId remote_xid;
+
+ /* needed for simplehash */
+ uint32 hash;
+ char status;
+} ReplicaIdentityEntry;
+
+#include "common/hashfn.h"
+
+static uint32 hash_replica_identity(ReplicaIdentityKey *key);
+static bool hash_replica_identity_compare(ReplicaIdentityKey *a,
+ ReplicaIdentityKey *b);
+
+/* Define parameters for replica identity hash table code generation. */
+#define SH_PREFIX replica_identity
+#define SH_ELEMENT_TYPE ReplicaIdentityEntry
+#define SH_KEY_TYPE ReplicaIdentityKey *
+#define SH_KEY keydata
+#define SH_HASH_KEY(tb, key) hash_replica_identity(key)
+#define SH_EQUAL(tb, a, b) hash_replica_identity_compare(a, b)
+#define SH_STORE_HASH
+#define SH_GET_HASH(tb, a) (a)->hash
+#define SH_SCOPE static inline
+#define SH_DECLARE
+#define SH_DEFINE
+#include "lib/simplehash.h"
+
+#define REPLICA_IDENTITY_INITIAL_SIZE 128
+#define REPLICA_IDENTITY_CLEANUP_THRESHOLD 1024
+
+static replica_identity_hash *replica_identity_table = NULL;
+
+static void write_internal_dependencies(StringInfo s, List *depends_on_xids);
+
static inline void subxact_filename(char *path, Oid subid, TransactionId xid);
static inline void changes_filename(char *path, Oid subid, TransactionId xid);
@@ -629,6 +673,546 @@ static TransApplyAction get_transaction_apply_action(TransactionId xid,
static void on_exit_clear_xact_state(int code, Datum arg);
+static bool send_internal_dependencies(ParallelApplyWorkerInfo *winfo,
+ StringInfo s);
+
+/*
+ * Compute the hash value for entries in the replica_identity_table.
+ */
+static uint32
+hash_replica_identity(ReplicaIdentityKey *key)
+{
+ int i;
+ uint32 hashkey = 0;
+
+ hashkey = hash_combine(hashkey, hash_uint32(key->relid));
+
+ for (i = 0; i < key->data->ncols; i++)
+ {
+ uint32 hkey;
+
+ if (key->data->colstatus[i] == LOGICALREP_COLUMN_NULL)
+ continue;
+
+ hkey = hash_any((const unsigned char *) key->data->colvalues[i].data,
+ key->data->colvalues[i].len);
+ hashkey = hash_combine(hashkey, hkey);
+ }
+
+ return hashkey;
+}
+
+/*
+ * Compare two entries in the replica_identity_table.
+ */
+static bool
+hash_replica_identity_compare(ReplicaIdentityKey *a, ReplicaIdentityKey *b)
+{
+ if (a->relid != b->relid ||
+ a->data->ncols != b->data->ncols)
+ return false;
+
+ for (int i = 0; i < a->data->ncols; i++)
+ {
+ if (a->data->colstatus[i] != b->data->colstatus[i])
+ return false;
+
+ if (a->data->colvalues[i].len != b->data->colvalues[i].len)
+ return false;
+
+ if (strcmp(a->data->colvalues[i].data, b->data->colvalues[i].data))
+ return false;
+
+ elog(DEBUG1, "conflicting key %s", a->data->colvalues[i].data);
+ }
+
+ return true;
+}
+
+/*
+ * Free resources associated with a replica identity key.
+ */
+static void
+free_replica_identity_key(ReplicaIdentityKey *key)
+{
+ Assert(key);
+
+ pfree(key->data->colvalues);
+ pfree(key->data->colstatus);
+ pfree(key->data);
+ pfree(key);
+}
+
+/*
+ * Clean up hash table entries associated with the given transaction IDs.
+ */
+static void
+cleanup_replica_identity_table(List *committed_xid)
+{
+ replica_identity_iterator i;
+ ReplicaIdentityEntry *rientry;
+
+ if (!committed_xid)
+ return;
+
+ replica_identity_start_iterate(replica_identity_table, &i);
+ while ((rientry = replica_identity_iterate(replica_identity_table, &i)) != NULL)
+ {
+ if (!list_member_xid(committed_xid, rientry->remote_xid))
+ continue;
+
+ /* Clean up the hash entry for committed transaction */
+ free_replica_identity_key(rientry->keydata);
+ replica_identity_delete_item(replica_identity_table, rientry);
+ }
+}
+
+/*
+ * Check committed transactions and clean up corresponding entries in the hash
+ * table.
+ */
+static void
+cleanup_committed_replica_identity_entries(void)
+{
+ dlist_mutable_iter iter;
+ List *committed_xids = NIL;
+
+ dlist_foreach_modify(iter, &lsn_mapping)
+ {
+ FlushPosition *pos =
+ dlist_container(FlushPosition, node, iter.cur);
+ bool skipped_write;
+
+ if (!TransactionIdIsValid(pos->pa_remote_xid) ||
+ !XLogRecPtrIsInvalid(pos->local_end))
+ continue;
+
+ pos->local_end = pa_get_last_commit_end(pos->pa_remote_xid, true,
+ &skipped_write);
+
+ elog(DEBUG1,
+ "got commit end from parallel apply worker, "
+ "txn: %u, remote_end %X/%X, local_end %X/%X",
+ pos->pa_remote_xid, LSN_FORMAT_ARGS(pos->remote_end),
+ LSN_FORMAT_ARGS(pos->local_end));
+
+ if (!skipped_write && XLogRecPtrIsInvalid(pos->local_end))
+ continue;
+
+ committed_xids = lappend_xid(committed_xids, pos->pa_remote_xid);
+ }
+
+ /* cleanup the entries for committed transactions */
+ cleanup_replica_identity_table(committed_xids);
+}
+
+/*
+ * Append a transaction dependency, excluding duplicates and committed
+ * transactions.
+ */
+static List *
+check_and_append_xid_dependency(List *depends_on_xids,
+ TransactionId *depends_on_xid,
+ TransactionId current_xid)
+{
+ Assert(depends_on_xid);
+
+ if (!TransactionIdIsValid(*depends_on_xid))
+ return depends_on_xids;
+
+ if (TransactionIdEquals(*depends_on_xid, current_xid))
+ return depends_on_xids;
+
+ if (list_member_xid(depends_on_xids, *depends_on_xid))
+ return depends_on_xids;
+
+ /*
+ * Return and reset the xid if the transaction has been committed.
+ */
+ if (pa_transaction_committed(*depends_on_xid))
+ {
+ *depends_on_xid = InvalidTransactionId;
+ return depends_on_xids;
+ }
+
+ return lappend_xid(depends_on_xids, *depends_on_xid);
+}
+
+/*
+ * Check for dependencies on preceding transactions that modify the same key.
+ * Returns the dependent transactions in 'depends_on_xids' and records the
+ * current change.
+ */
+static void
+check_dependency_on_replica_identity(Oid relid,
+ LogicalRepTupleData *original_data,
+ TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ LogicalRepRelMapEntry *relentry;
+ LogicalRepTupleData *ridata;
+ ReplicaIdentityKey *rikey;
+ ReplicaIdentityEntry *rientry;
+ MemoryContext oldctx;
+ int n_ri;
+ bool found = false;
+
+ Assert(depends_on_xids);
+
+ /* Search for existing entry */
+ relentry = logicalrep_get_relentry(relid);
+
+ Assert(relentry);
+
+ /*
+ * First search whether any previous transaction has affected the whole
+ * table e.g., truncate or schema change from publisher.
+ */
+ *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
+ &relentry->last_depended_xid,
+ new_depended_xid);
+
+ n_ri = bms_num_members(relentry->remoterel.attkeys);
+
+ /*
+ * Return if there are no replica identity columns, indicating that the
+ * remote relation has neither a replica identity key nor is marked as
+ * replica identity full.
+ */
+ if (!n_ri)
+ return;
+
+ /* Check if the RI key value of the tuple is invalid */
+ for (int i = 0; i < original_data->ncols; i++)
+ {
+ if (!bms_is_member(i, relentry->remoterel.attkeys))
+ continue;
+
+ /*
+ * Return if RI key is NULL or is explicitly marked unchanged. The key
+ * value could be NULL in the new tuple of a update opertaion which
+ * means the RI key is not updated.
+ */
+ if (original_data->colstatus[i] == LOGICALREP_COLUMN_NULL ||
+ original_data->colstatus[i] == LOGICALREP_COLUMN_UNCHANGED)
+ return;
+ }
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ /* Allocate space for replica identity values */
+ ridata = palloc0_object(LogicalRepTupleData);
+ ridata->colvalues = palloc0_array(StringInfoData, n_ri);
+ ridata->colstatus = palloc0_array(char, n_ri);
+ ridata->ncols = n_ri;
+
+ for (int i_original = 0, i_ri = 0; i_original < original_data->ncols; i_original++)
+ {
+ StringInfo original_colvalue = &original_data->colvalues[i_original];
+
+ if (!bms_is_member(i_original, relentry->remoterel.attkeys))
+ continue;
+
+ initStringInfoExt(&ridata->colvalues[i_ri], original_colvalue->len + 1);
+ appendStringInfoString(&ridata->colvalues[i_ri], original_colvalue->data);
+ ridata->colstatus[i_ri] = original_data->colstatus[i_original];
+ i_ri++;
+ }
+
+ rikey = palloc0_object(ReplicaIdentityKey);
+ rikey->relid = relid;
+ rikey->data = ridata;
+
+ if (TransactionIdIsValid(new_depended_xid))
+ {
+ rientry = replica_identity_insert(replica_identity_table, rikey,
+ &found);
+
+ /*
+ * Release the key built to search the entry, if the entry already
+ * exists. Otherwise, initialize the remote_xid.
+ */
+ if (found)
+ {
+ elog(DEBUG1, "found conflicting replica identity change from %u",
+ rientry->remote_xid);
+
+ free_replica_identity_key(rikey);
+ }
+ else
+ rientry->remote_xid = InvalidTransactionId;
+ }
+ else
+ {
+ rientry = replica_identity_lookup(replica_identity_table, rikey);
+ free_replica_identity_key(rikey);
+ }
+
+ MemoryContextSwitchTo(oldctx);
+
+ /* Return if no entry found */
+ if (!rientry)
+ return;
+
+ Assert(!found || TransactionIdIsValid(rientry->remote_xid));
+
+ *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
+ &rientry->remote_xid,
+ new_depended_xid);
+
+ /*
+ * Update the new depended xid into the entry if valid, 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, so no need to be depended on.
+ */
+ if (TransactionIdIsValid(new_depended_xid))
+ rientry->remote_xid = new_depended_xid;
+
+ /*
+ * Remove the entry if the transaction has been committed and no new
+ * dependency needs to be added.
+ */
+ else if (!TransactionIdIsValid(rientry->remote_xid))
+ {
+ free_replica_identity_key(rientry->keydata);
+ replica_identity_delete_item(replica_identity_table, rientry);
+ }
+}
+
+/*
+ * Check for preceding transactions that involve insert, delete, or update
+ * operations on the specified table, and return them in 'depends_on_xids'.
+ */
+static void
+find_all_dependencies_on_rel(LogicalRepRelId relid, TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ replica_identity_iterator i;
+ ReplicaIdentityEntry *rientry;
+
+ Assert(depends_on_xids);
+
+ replica_identity_start_iterate(replica_identity_table, &i);
+ while ((rientry = replica_identity_iterate(replica_identity_table, &i)) != NULL)
+ {
+ Assert(TransactionIdIsValid(rientry->remote_xid));
+
+ if (rientry->keydata->relid != relid)
+ continue;
+
+ /* Clean up the hash entry for committed transaction while on it */
+ if (pa_transaction_committed(rientry->remote_xid))
+ {
+ free_replica_identity_key(rientry->keydata);
+ replica_identity_delete_item(replica_identity_table, rientry);
+
+ continue;
+ }
+
+ *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
+ &rientry->remote_xid,
+ new_depended_xid);
+ }
+}
+
+/*
+ * Check for any preceding transactions that affect the given table and returns
+ * them in 'depends_on_xids'.
+ */
+static void
+check_dependency_on_rel(LogicalRepRelId relid, TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ LogicalRepRelMapEntry *relentry;
+
+ Assert(depends_on_xids);
+
+ find_all_dependencies_on_rel(relid, new_depended_xid, depends_on_xids);
+
+ /* Search for existing entry */
+ relentry = logicalrep_get_relentry(relid);
+
+ /*
+ * The relentry has not been initialized yet, indicating no change has
+ * been applide yet.
+ */
+ if (!relentry)
+ return;
+
+ *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
+ &relentry->last_depended_xid,
+ new_depended_xid);
+
+ if (TransactionIdIsValid(new_depended_xid))
+ relentry->last_depended_xid = new_depended_xid;
+}
+
+/*
+ * Check dependencies related to the current change by determining if the
+ * modification impacts the same row or table as another ongoing transaction. If
+ * needed, instruct parallel apply workers to wait for these preceding
+ * transactions to complete.
+ *
+ * Simultaneously, track the dependency for the current change to ensure that
+ * subsequent transactions address this dependency.
+ */
+static void
+handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
+ TransactionId new_depended_xid,
+ ParallelApplyWorkerInfo *winfo)
+{
+ LogicalRepRelId relid;
+ LogicalRepTupleData oldtup;
+ LogicalRepTupleData newtup;
+ LogicalRepRelation *rel;
+ List *depends_on_xids = NIL;
+ List *remote_relids;
+ bool has_oldtup = false;
+ bool cascade = false;
+ bool restart_seqs = false;
+ StringInfoData dependencies;
+
+ /*
+ * Parse the consume data using a local copy instead of directly consuming
+ * the given remote change as the caller may also read the data from the
+ * remote message.
+ */
+ StringInfoData change = *s;
+
+ /* Compute dependency only for non-streaming transaction */
+ if (in_streamed_transaction || (winfo && winfo->stream_txn))
+ return;
+
+ /* Only the leader checks dependencies and schedules the parallel apply */
+ if (!am_leader_apply_worker())
+ return;
+
+ if (!replica_identity_table)
+ replica_identity_table = replica_identity_create(ApplyContext,
+ REPLICA_IDENTITY_INITIAL_SIZE,
+ NULL);
+
+ if (replica_identity_table->members >= REPLICA_IDENTITY_CLEANUP_THRESHOLD)
+ cleanup_committed_replica_identity_entries();
+
+ switch (action)
+ {
+ case LOGICAL_REP_MSG_INSERT:
+ relid = logicalrep_read_insert(&change, &newtup);
+ check_dependency_on_replica_identity(relid, &newtup,
+ new_depended_xid,
+ &depends_on_xids);
+ break;
+
+ case LOGICAL_REP_MSG_UPDATE:
+ relid = logicalrep_read_update(&change, &has_oldtup, &oldtup,
+ &newtup);
+
+ if (has_oldtup)
+ check_dependency_on_replica_identity(relid, &oldtup,
+ new_depended_xid,
+ &depends_on_xids);
+
+ check_dependency_on_replica_identity(relid, &newtup,
+ new_depended_xid,
+ &depends_on_xids);
+ break;
+
+ case LOGICAL_REP_MSG_DELETE:
+ relid = logicalrep_read_delete(&change, &oldtup);
+ check_dependency_on_replica_identity(relid, &oldtup,
+ new_depended_xid,
+ &depends_on_xids);
+ break;
+
+ case LOGICAL_REP_MSG_TRUNCATE:
+ remote_relids = logicalrep_read_truncate(&change, &cascade,
+ &restart_seqs);
+
+ /*
+ * Truncate affects all rows in a table, so the current
+ * transaction should wait for all preceding transactions that
+ * modified the same table.
+ */
+ foreach_int(truncated_relid, remote_relids)
+ check_dependency_on_rel(truncated_relid, new_depended_xid,
+ &depends_on_xids);
+
+ break;
+
+ case LOGICAL_REP_MSG_RELATION:
+ rel = logicalrep_read_rel(&change);
+
+ /*
+ * The replica identity key could be changed, making existing
+ * entries in the replica identity invalid. In this case, parallel
+ * apply is not allowed on this specific table until all running
+ * transactions that modified it have finished.
+ */
+ check_dependency_on_rel(rel->remoteid, new_depended_xid,
+ &depends_on_xids);
+ break;
+
+ case LOGICAL_REP_MSG_TYPE:
+ case LOGICAL_REP_MSG_MESSAGE:
+
+ /*
+ * Type updates accompany relation updates, so dependencies have
+ * already been checked during relation updates. Logical messages
+ * do not conflict with any changes, so they can be ignored.
+ */
+ break;
+
+ default:
+ Assert(false);
+ break;
+ }
+
+ if (!depends_on_xids)
+ return;
+
+ /*
+ * Notify the transactions that they are dependent on the current
+ * transaction.
+ */
+ pa_record_dependency_on_transactions(depends_on_xids);
+
+ /*
+ * If the leader applies the transaction itself, start waiting for
+ * transactions that depend on the current transaction immediately.
+ */
+ if (winfo == NULL)
+ {
+ foreach_xid(xid, depends_on_xids)
+ pa_wait_for_depended_transaction(xid);
+
+ return;
+ }
+
+ initStringInfo(&dependencies);
+
+ /* Build the dependency message used to send to parallel apply worker */
+ write_internal_dependencies(&dependencies, depends_on_xids);
+
+ (void) send_internal_dependencies(winfo, &dependencies);
+}
+
+/*
+ * Write internal dependency information to the output for the parallel apply
+ * worker.
+ */
+static void
+write_internal_dependencies(StringInfo s, List *depends_on_xids)
+{
+ pq_sendbyte(s, PARALLEL_APPLY_INTERNAL_MESSAGE);
+ pq_sendbyte(s, LOGICAL_REP_MSG_INTERNAL_DEPENDENCY);
+ pq_sendint32(s, list_length(depends_on_xids));
+
+ foreach_xid(xid, depends_on_xids)
+ pq_sendint32(s, xid);
+}
+
/*
* Handle internal dependency information.
*
@@ -826,7 +1410,10 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
/* not in streaming mode */
if (apply_action == TRANS_LEADER_APPLY)
+ {
+ handle_dependency_on_change(action, s, InvalidTransactionId, winfo);
return false;
+ }
Assert(TransactionIdIsValid(stream_xid));
@@ -1268,6 +1855,33 @@ apply_handle_begin(StringInfo s)
pgstat_report_activity(STATE_RUNNING, NULL);
}
+/*
+ * Send an INTERNAL_DEPENDENCY message to a parallel apply worker.
+ *
+ * Returns false if we switched to the serialize mode to send the message,
+ * true otherwise.
+ */
+static bool
+send_internal_dependencies(ParallelApplyWorkerInfo *winfo, StringInfo s)
+{
+ Assert(s->data[0] == PARALLEL_APPLY_INTERNAL_MESSAGE);
+ Assert(s->data[1] == LOGICAL_REP_MSG_INTERNAL_DEPENDENCY);
+
+ if (!winfo->serialize_changes)
+ {
+ if (pa_send_data(winfo, s->len, s->data))
+ return true;
+
+ pa_switch_to_partial_serialize(winfo, true);
+ }
+
+ /* Skip writing the first internal message flag */
+ s->cursor++;
+ stream_write_change(LOGICAL_REP_MSG_INTERNAL_DEPENDENCY, s);
+
+ return false;
+}
+
/*
* Handle COMMIT message.
*
@@ -1795,7 +2409,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, true);
apply_action = get_transaction_apply_action(stream_xid, &winfo);
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index efe0f9d6031..f9aa9e59a6d 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -37,6 +37,8 @@ typedef struct LogicalRepRelMapEntry
/* Sync state. */
char state;
XLogRecPtr statelsn;
+
+ TransactionId last_depended_xid;
} LogicalRepRelMapEntry;
extern void logicalrep_relmap_update(LogicalRepRelation *remoterel);
@@ -50,5 +52,6 @@ extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel,
LOCKMODE lockmode);
extern bool IsIndexUsableForReplicaIdentityFull(Relation idxrel, AttrMap *attrmap);
extern Oid GetRelationIdentityOrPK(Relation rel);
+extern LogicalRepRelMapEntry *logicalrep_get_relentry(LogicalRepRelId remoteid);
#endif /* LOGICALRELATION_H */
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 411e2c537c7..1dbec625c81 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -235,6 +235,8 @@ typedef struct ParallelApplyWorkerInfo
*/
bool in_use;
+ bool stream_txn;
+
ParallelApplyWorkerShared *shared;
} ParallelApplyWorkerInfo;
@@ -332,8 +334,10 @@ 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, 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);
extern void pa_detach_all_error_mq(void);
extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
@@ -362,6 +366,8 @@ extern void pa_decr_and_wait_stream_block(void);
extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
XLogRecPtr remote_lsn);
+extern bool pa_transaction_committed(TransactionId xid);
+extern void pa_record_dependency_on_transactions(List *depends_on_xids);
extern void pa_wait_for_depended_transaction(TransactionId xid);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index d99c1996efd..05936d61a54 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2582,6 +2582,8 @@ ReplOriginXactState
ReplaceVarsFromTargetList_context
ReplaceVarsNoMatchOption
ReplaceWrapOption
+ReplicaIdentityEntry
+ReplicaIdentityKey
ReplicaIdentityStmt
ReplicationKind
ReplicationSlot
@@ -2593,6 +2595,7 @@ ReplicationSlotPersistentData
ReplicationState
ReplicationStateCtl
ReplicationStateOnDisk
+replica_identity_hash
ResTarget
ReservoirState
ReservoirStateData
--
2.47.3
[application/octet-stream] v8-0004-Parallel-apply-non-streaming-transactions.patch (51.3K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/5-v8-0004-Parallel-apply-non-streaming-transactions.patch)
download | inline diff:
From b906303922a0663a53677ca8c1b4082f10a5ce0b Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 1 Dec 2025 12:28:29 +0900
Subject: [PATCH v8 4/9] Parallel apply non-streaming transactions
--
Basic design
--
The leader worker assigns each non-streaming transaction to a parallel apply
worker. Before dispatching changes to a parallel worker, the leader verifies if
the current modification affects the same row (identitied by replica identity
key) as another ongoing transaction. If so, the leader sends a list of dependent
transaction IDs to the parallel worker, indicating that the parallel apply
worker must wait for these transactions to commit before proceeding.
Each parallel apply worker records the local end LSN of the transaction it
applies in shared memory. Subsequently, the leader gathers these local end LSNs
and logs them in the local 'lsn_mapping' for verifying whether they have been
flushed to disk (following the logic in get_flush_position()).
If no parallel apply worker is available, the leader will apply the transaction
independently.
For further details, please refer to the following:
--
dedendency tracking
--
The leader maintains a local hash table, using the remote change's replica
identity column values and relid as keys, with remote transaction IDs as values.
Before sending changes to the parallel apply worker, the leader computes a hash
using RI key values and the relid of the current change to search the hash
table. If an existing entry is found, the leader first updates the hash entry
with the receiving remote xid then tells the parallel worker to wait for it.
If the remote relation lacks a replica identity (RI), it indicates that only
INSERT can be replicated for this table. In such cases, the leader skips
dependency checks, allowing the parallel apply worker to proceed with applying
changes without delay. This is because the only potential conflict could happen
is related to the local unique key or foreign key, which that is yet to be
implemented (see TODO - dependency on local unique key, foreign key.).
In cases of TRUNCATE or remote schema changes affecting the entire table, the
leader retrieves all remote xids touching the same table (via sequential scans
of the hash table) and tells the parallel worker to wait for those transactions
to commit.
Hash entries are cleaned up once the transaction corresponding to the remote xid
in the entry has been committed. Clean-up typically occurs when collecting the
flush position of each transaction, but is forced if the hash table exceeds a
set threshold.
--
dedendency waiting
--
If a transaction is relied upon by others, the leader adds its xid to a shared
hash table. The shared hash table entry is cleared by the parallel apply worker
upon completing the transaction. Workers needing to wait for a transaction check
the shared hash table entry; if present, they lock the transaction ID (using
pa_lock_transaction). If absent, it indicates the transaction has been
committed, negating the need to wait.
--
commit order
--
There is a case where columns have no foreign or primary keys, and integrity is
maintained at the application layer. In this case, the above RI mechanism cannot
detect any dependencies. For safety reasons, parallel apply workers preserve the
commit ordering done on the publisher side. This is done by the leader worker
caching the lastly dispatched transaction ID and adding a dependency between it
and the currently dispatching one.
---
.../replication/logical/applyparallelworker.c | 339 ++++++++++++++++--
src/backend/replication/logical/proto.c | 38 ++
src/backend/replication/logical/relation.c | 31 ++
src/backend/replication/logical/worker.c | 304 ++++++++++++++--
src/include/replication/logicalproto.h | 2 +
src/include/replication/logicalrelation.h | 2 +
src/include/replication/worker_internal.h | 11 +-
src/test/subscription/meson.build | 1 +
src/test/subscription/t/001_rep_changes.pl | 2 +
src/test/subscription/t/010_truncate.pl | 2 +-
src/test/subscription/t/015_stream.pl | 8 +-
src/test/subscription/t/026_stats.pl | 1 +
src/test/subscription/t/027_nosuperuser.pl | 1 +
src/test/subscription/t/050_parallel_apply.pl | 130 +++++++
src/tools/pgindent/typedefs.list | 2 +
15 files changed, 800 insertions(+), 74 deletions(-)
create mode 100644 src/test/subscription/t/050_parallel_apply.pl
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index d462674b5cd..3396ee1cdf1 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -14,6 +14,9 @@
* ParallelApplyWorkerInfo which is required so the leader worker and parallel
* apply workers can communicate with each other.
*
+ * Streaming transactions
+ * ======================
+ *
* The parallel apply workers are assigned (if available) as soon as xact's
* first stream is received for subscriptions that have set their 'streaming'
* option as parallel. The leader apply worker will send changes to this new
@@ -152,6 +155,33 @@
* session-level locks because both locks could be acquired outside the
* transaction, and the stream lock in the leader needs to persist across
* transaction boundaries i.e. until the end of the streaming transaction.
+ *
+ * Non-streaming transactions
+ * ======================
+ * The handling is similar to streaming transactions, but including few
+ * differences:
+ *
+ * Transaction dependency
+ * ----------------------
+ * Before dispatching changes to a parallel worker, the leader verifies if the
+ * current modification affects the same row (identitied by replica identity
+ * key) as another ongoing transaction (see handle_dependency_on_change for
+ * details). If so, the leader sends a list of dependent transaction IDs to the
+ * parallel worker, indicating that the parallel apply worker must wait for
+ * these transactions to commit before proceeding.
+ *
+ * Commit order
+ * ------------
+ * There is a case where columns have no foreign or primary keys, and integrity
+ * is maintained at the application layer. In this case, the above RI mechanism
+ * cannot detect any dependencies. For safety reasons, parallel apply workers
+ * preserve the commit ordering done on the publisher side. This is done by the
+ * leader worker caching the lastly dispatched transaction ID and adding a
+ * dependency between it and the currently dispatching one.
+ * We can extend the parallel apply worker to allow out-of-order commits in the
+ * future: At least we must use a new mechanism to track replication progress
+ * in out-of-order commits. Then we can stop caching the transaction ID and
+ * adding the dependency.
*-------------------------------------------------------------------------
*/
@@ -283,6 +313,7 @@ static ParallelTransState pa_get_xact_state(ParallelApplyWorkerShared *wshared);
static PartialFileSetState pa_get_fileset_state(void);
static void pa_attach_parallelized_txn_hash(dsa_handle *pa_dsa_handle,
dshash_table_handle *pa_dshash_handle);
+static void write_internal_relation(StringInfo s, LogicalRepRelation *rel);
/*
* Returns true if it is OK to start a parallel apply worker, false otherwise.
@@ -400,6 +431,7 @@ pa_setup_dsm(ParallelApplyWorkerInfo *winfo)
shared = shm_toc_allocate(toc, sizeof(ParallelApplyWorkerShared));
SpinLockInit(&shared->mutex);
+ shared->xid = InvalidTransactionId;
shared->xact_state = PARALLEL_TRANS_UNKNOWN;
pg_atomic_init_u32(&(shared->pending_stream_count), 0);
shared->last_commit_end = InvalidXLogRecPtr;
@@ -443,6 +475,8 @@ pa_launch_parallel_worker(void)
MemoryContext oldcontext;
bool launched;
ParallelApplyWorkerInfo *winfo;
+ dsa_handle pa_dsa_handle;
+ dshash_table_handle pa_dshash_handle;
ListCell *lc;
/* Try to get an available parallel apply worker from the worker pool. */
@@ -450,10 +484,33 @@ pa_launch_parallel_worker(void)
{
winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
- if (!winfo->in_use)
+ if (!winfo->stream_txn &&
+ pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED)
+ {
+ /*
+ * Save the local commit LSN of the last transaction applied by
+ * this worker before reusing it for another transaction. This WAL
+ * position is crucial for determining the flush position in
+ * responses to the publisher (see get_flush_position()).
+ */
+ (void) pa_get_last_commit_end(winfo->shared->xid, false, NULL);
+ return winfo;
+ }
+
+ if (winfo->stream_txn && !winfo->in_use)
return winfo;
}
+ pa_attach_parallelized_txn_hash(&pa_dsa_handle, &pa_dshash_handle);
+
+ /*
+ * Return if the number of parallel apply workers has reached the maximum
+ * limit.
+ */
+ if (list_length(ParallelApplyWorkerPool) ==
+ max_parallel_apply_workers_per_subscription)
+ return NULL;
+
/*
* Start a new parallel apply worker.
*
@@ -481,18 +538,32 @@ pa_launch_parallel_worker(void)
dsm_segment_handle(winfo->dsm_seg),
false);
- if (launched)
- {
- ParallelApplyWorkerPool = lappend(ParallelApplyWorkerPool, winfo);
- }
- else
+ if (!launched)
{
+ MemoryContextSwitchTo(oldcontext);
pa_free_worker_info(winfo);
- winfo = NULL;
+ return NULL;
}
+ ParallelApplyWorkerPool = lappend(ParallelApplyWorkerPool, winfo);
+
MemoryContextSwitchTo(oldcontext);
+ /*
+ * Send all existing remote relation information to the parallel apply
+ * worker. This allows the parallel worker to initialize the
+ * LogicalRepRelMapEntry locally before applying remote changes.
+ */
+ if (logicalrep_get_num_rels())
+ {
+ StringInfoData out;
+
+ initStringInfo(&out);
+
+ write_internal_relation(&out, NULL);
+ pa_send_data(winfo, out.len, out.data);
+ }
+
return winfo;
}
@@ -597,7 +668,8 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
{
Assert(!am_parallel_apply_worker());
Assert(winfo->in_use);
- Assert(pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED);
+ Assert(!winfo->stream_txn ||
+ pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED);
if (!hash_search(ParallelApplyTxnHash, &winfo->shared->xid, HASH_REMOVE, NULL))
elog(ERROR, "hash table corrupted");
@@ -613,9 +685,7 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
* been serialized and then letting the parallel apply worker deal with
* the spurious message, we stop the worker.
*/
- if (winfo->serialize_changes ||
- list_length(ParallelApplyWorkerPool) >
- (max_parallel_apply_workers_per_subscription / 2))
+ if (winfo->serialize_changes)
{
logicalrep_pa_worker_stop(winfo);
pa_free_worker_info(winfo);
@@ -812,6 +882,38 @@ pa_get_last_commit_end(TransactionId xid, bool delete_entry, bool *skipped_write
return entry->local_end;
}
+/*
+ * Wait for the remote transaction associated with the specified remote xid to
+ * complete.
+ */
+static void
+pa_wait_for_transaction(TransactionId wait_for_xid)
+{
+ if (!am_leader_apply_worker())
+ return;
+
+ if (!TransactionIdIsValid(wait_for_xid))
+ return;
+
+ elog(DEBUG1, "plan to wait for remote_xid %u to finish",
+ wait_for_xid);
+
+ for (;;)
+ {
+ if (pa_transaction_committed(wait_for_xid))
+ break;
+
+ pa_lock_transaction(wait_for_xid, AccessShareLock);
+ pa_unlock_transaction(wait_for_xid, AccessShareLock);
+
+ /* An interrupt may have occurred while we were waiting. */
+ CHECK_FOR_INTERRUPTS();
+ }
+
+ elog(DEBUG1, "finished wait for remote_xid %u to finish",
+ wait_for_xid);
+}
+
/*
* Interrupt handler for main loop of parallel apply worker.
*/
@@ -887,21 +989,34 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
* parallel apply workers can only be PqReplMsg_WALData.
*/
c = pq_getmsgbyte(&s);
- if (c != PqReplMsg_WALData)
- elog(ERROR, "unexpected message \"%c\"", c);
-
- /*
- * Ignore statistics fields that have been updated by the leader
- * apply worker.
- *
- * XXX We can avoid sending the statistics fields from the leader
- * apply worker but for that, it needs to rebuild the entire
- * message by removing these fields which could be more work than
- * simply ignoring these fields in the parallel apply worker.
- */
- s.cursor += SIZE_STATS_MESSAGE;
+ if (c == PqReplMsg_WALData)
+ {
+ /*
+ * Ignore statistics fields that have been updated by the
+ * leader apply worker.
+ *
+ * XXX We can avoid sending the statistics fields from the
+ * leader apply worker but for that, it needs to rebuild the
+ * entire message by removing these fields which could be more
+ * work than simply ignoring these fields in the parallel
+ * apply worker.
+ */
+ s.cursor += SIZE_STATS_MESSAGE;
- apply_dispatch(&s);
+ apply_dispatch(&s);
+ }
+ else if (c == PARALLEL_APPLY_INTERNAL_MESSAGE)
+ {
+ apply_dispatch(&s);
+ }
+ else
+ {
+ /*
+ * The first byte of messages sent from leader apply worker to
+ * parallel apply workers can only be 'w' or 'i'.
+ */
+ elog(ERROR, "unexpected message \"%c\"", c);
+ }
}
else if (shmq_res == SHM_MQ_WOULD_BLOCK)
{
@@ -918,6 +1033,9 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
if (rc & WL_LATCH_SET)
ResetLatch(MyLatch);
+
+ if (!IsTransactionState())
+ pgstat_report_stat(true);
}
}
else
@@ -955,6 +1073,9 @@ pa_shutdown(int code, Datum arg)
INVALID_PROC_NUMBER);
dsm_detach((dsm_segment *) DatumGetPointer(arg));
+
+ if (parallel_apply_dsa_area)
+ dsa_detach(parallel_apply_dsa_area);
}
/*
@@ -1267,7 +1388,6 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
shm_mq_result result;
TimestampTz startTime = 0;
- Assert(!IsTransactionState());
Assert(!winfo->serialize_changes);
/*
@@ -1319,6 +1439,67 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
}
}
+/*
+ * Distribute remote relation information to all active parallel apply workers
+ * that require it.
+ */
+void
+pa_distribute_schema_changes_to_workers(LogicalRepRelation *rel)
+{
+ List *workers_stopped = NIL;
+ StringInfoData out;
+
+ if (!ParallelApplyWorkerPool)
+ return;
+
+ initStringInfo(&out);
+
+ write_internal_relation(&out, rel);
+
+ foreach_ptr(ParallelApplyWorkerInfo, winfo, ParallelApplyWorkerPool)
+ {
+ /*
+ * Skip the worker responsible for the current transaction, as the
+ * relation information has already been sent to it.
+ */
+ if (winfo == stream_apply_worker)
+ continue;
+
+ /*
+ * Skip the worker that is in serialize mode, as they will soon stop
+ * once they finish applying the transaction.
+ */
+ if (winfo->serialize_changes)
+ continue;
+
+ elog(DEBUG1, "distributing schema changes to pa workers");
+
+ if (pa_send_data(winfo, out.len, out.data))
+ continue;
+
+ elog(DEBUG1, "failed to distribute, will stop that worker instead");
+
+ /*
+ * Distribution to this worker failed due to a sending timeout. Wait
+ * for the worker to complete its transaction and then stop it. This
+ * is consistent with the handling of workers in serialize mode (see
+ * pa_free_worker() for details).
+ */
+ pa_wait_for_transaction(winfo->shared->xid);
+
+ pa_get_last_commit_end(winfo->shared->xid, false, NULL);
+
+ logicalrep_pa_worker_stop(winfo);
+
+ workers_stopped = lappend(workers_stopped, winfo);
+ }
+
+ pfree(out.data);
+
+ foreach_ptr(ParallelApplyWorkerInfo, winfo, workers_stopped)
+ pa_free_worker_info(winfo);
+}
+
/*
* Switch to PARTIAL_SERIALIZE mode for the current transaction -- this means
* that the current data and any subsequent data for this transaction will be
@@ -1401,8 +1582,8 @@ pa_wait_for_xact_finish(ParallelApplyWorkerInfo *winfo)
/*
* Wait for the transaction lock to be released. This is required to
- * detect deadlock among leader and parallel apply workers. Refer to the
- * comments atop this file.
+ * detect detect deadlock among leader and parallel apply workers. Refer
+ * to the comments atop this file.
*/
pa_lock_transaction(winfo->shared->xid, AccessShareLock);
pa_unlock_transaction(winfo->shared->xid, AccessShareLock);
@@ -1479,6 +1660,9 @@ pa_savepoint_name(Oid suboid, TransactionId xid, char *spname, Size szsp)
void
pa_start_subtrans(TransactionId current_xid, TransactionId top_xid)
{
+ if (!TransactionIdIsValid(top_xid))
+ return;
+
if (current_xid != top_xid &&
!list_member_xid(subxactlist, current_xid))
{
@@ -1735,25 +1919,41 @@ pa_decr_and_wait_stream_block(void)
void
pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
{
+ XLogRecPtr local_lsn = InvalidXLogRecPtr;
+ TransactionId pa_remote_xid = winfo->shared->xid;
+
Assert(am_leader_apply_worker());
/*
- * Unlock the shared object lock so that parallel apply worker can
- * continue to receive and apply changes.
+ * Unlock the shared object lock taken for streaming transactions so that
+ * parallel apply worker can continue to receive and apply changes.
*/
- pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);
+ if (winfo->stream_txn)
+ pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);
/*
- * Wait for that worker to finish. This is necessary to maintain commit
- * order which avoids failures due to transaction dependencies and
- * deadlocks.
+ * Wait for that worker for streaming transaction to finish. This is
+ * necessary to maintain commit order which avoids failures due to
+ * transaction dependencies and deadlocks.
+ *
+ * For non-streaming transaction but in partial seralize mode, wait for
+ * stop as well as the worker is anyway cannot be reused anymore (see
+ * pa_free_worker() for details).
*/
- pa_wait_for_xact_finish(winfo);
+ if (winfo->serialize_changes || winfo->stream_txn)
+ {
+ pa_wait_for_xact_finish(winfo);
+
+ local_lsn = winfo->shared->last_commit_end;
+ pa_remote_xid = InvalidTransactionId;
+
+ pa_free_worker(winfo);
+ }
if (XLogRecPtrIsValid(remote_lsn))
- store_flush_position(remote_lsn, winfo->shared->last_commit_end);
+ store_flush_position(remote_lsn, local_lsn, pa_remote_xid);
- pa_free_worker(winfo);
+ pa_set_stream_apply_worker(NULL);
}
bool
@@ -1852,6 +2052,22 @@ pa_record_dependency_on_transactions(List *depends_on_xids)
}
}
+/*
+ * Mark the transaction state as finished and remove the shared hash entry.
+ */
+void
+pa_commit_transaction(void)
+{
+ TransactionId xid = MyParallelShared->xid;
+
+ SpinLockAcquire(&MyParallelShared->mutex);
+ MyParallelShared->xact_state = PARALLEL_TRANS_FINISHED;
+ SpinLockRelease(&MyParallelShared->mutex);
+
+ dshash_delete_key(parallelized_txns, &xid);
+ elog(DEBUG1, "depended xid %u committed", xid);
+}
+
/*
* Wait for the given transaction to finish.
*/
@@ -1860,6 +2076,13 @@ pa_wait_for_depended_transaction(TransactionId xid)
{
elog(DEBUG1, "wait for depended xid %u", xid);
+ /*
+ * Quick exit if parallelized_txns has not been initialized yet. This can
+ * happen when this function is called by the leader worker.
+ */
+ if (!parallelized_txns)
+ return;
+
for (;;)
{
ParallelizedTxnEntry *txn_entry;
@@ -1880,3 +2103,45 @@ pa_wait_for_depended_transaction(TransactionId xid)
elog(DEBUG1, "finish waiting for depended xid %u", xid);
}
+
+/*
+ * Write internal relation description to the output stream.
+ */
+static void
+write_internal_relation(StringInfo s, LogicalRepRelation *rel)
+{
+ pq_sendbyte(s, PARALLEL_APPLY_INTERNAL_MESSAGE);
+ pq_sendbyte(s, LOGICAL_REP_MSG_INTERNAL_RELATION);
+
+ if (rel)
+ {
+ pq_sendint(s, 1, 4);
+ logicalrep_write_internal_rel(s, rel);
+ }
+ else
+ {
+ pq_sendint(s, logicalrep_get_num_rels(), 4);
+ logicalrep_write_all_rels(s);
+ }
+}
+
+/*
+ * Register a transaction to the shared hash table.
+ *
+ * This function is intended to be called during the commit phase of
+ * non-streamed transactions. Other parallel workers would wait,
+ * removing the added entry.
+ */
+void
+pa_add_parallelized_transaction(TransactionId xid)
+{
+ bool found;
+ ParallelizedTxnEntry *txn_entry;
+
+ Assert(parallelized_txns);
+ Assert(TransactionIdIsValid(xid));
+
+ txn_entry = dshash_find_or_insert(parallelized_txns, &xid, &found);
+
+ dshash_release_lock(parallelized_txns, txn_entry);
+}
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 66f930c8b06..d231bfb48a3 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -691,6 +691,44 @@ logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel,
logicalrep_write_attrs(out, rel, columns, include_gencols_type);
}
+/*
+ * Write internal relation description to the output stream.
+ */
+void
+logicalrep_write_internal_rel(StringInfo out, LogicalRepRelation *rel)
+{
+ pq_sendint32(out, rel->remoteid);
+
+ /* Write relation name */
+ pq_sendstring(out, rel->nspname);
+ pq_sendstring(out, rel->relname);
+
+ /* Write the replica identity. */
+ pq_sendbyte(out, rel->replident);
+
+ /* Write attribute description */
+ pq_sendint16(out, rel->natts);
+
+ for (int i = 0; i < rel->natts; i++)
+ {
+ uint8 flags = 0;
+
+ if (bms_is_member(i, rel->attkeys))
+ flags |= LOGICALREP_IS_REPLICA_IDENTITY;
+
+ pq_sendbyte(out, flags);
+
+ /* attribute name */
+ pq_sendstring(out, rel->attnames[i]);
+
+ /* attribute type id */
+ pq_sendint32(out, rel->atttyps[i]);
+
+ /* ignore attribute mode for now */
+ pq_sendint32(out, 0);
+ }
+}
+
/*
* Read the relation info from stream and return as LogicalRepRelation.
*/
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index ab6313eb1bc..e1ce183dfd3 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -961,6 +961,37 @@ FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel,
return InvalidOid;
}
+/*
+ * Get the number of entries in the LogicalRepRelMap.
+ */
+int
+logicalrep_get_num_rels(void)
+{
+ if (LogicalRepRelMap == NULL)
+ return 0;
+
+ return hash_get_num_entries(LogicalRepRelMap);
+}
+
+/*
+ * Write all the remote relation information from the LogicalRepRelMapEntry to
+ * the output stream.
+ */
+void
+logicalrep_write_all_rels(StringInfo out)
+{
+ LogicalRepRelMapEntry *entry;
+ HASH_SEQ_STATUS status;
+
+ if (LogicalRepRelMap == NULL)
+ return;
+
+ hash_seq_init(&status, LogicalRepRelMap);
+
+ while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
+ logicalrep_write_internal_rel(out, &entry->remoterel);
+}
+
/*
* Get the LogicalRepRelMapEntry corresponding to the given relid without
* opening the local relation.
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index c2967caf1e5..d1ba801f96f 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -286,6 +286,7 @@
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/guc.h"
+#include "utils/injection_point.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
@@ -484,6 +485,8 @@ static List *on_commit_wakeup_workers_subids = NIL;
bool in_remote_transaction = false;
static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
+static TransactionId remote_xid = InvalidTransactionId;
+static TransactionId last_remote_xid = InvalidTransactionId;
/* fields valid only when processing streamed transaction */
static bool in_streamed_transaction = false;
@@ -602,11 +605,7 @@ static inline void cleanup_subxact_info(void);
/*
* Serialize and deserialize changes for a toplevel transaction.
*/
-static void stream_open_file(Oid subid, TransactionId xid,
- bool first_segment);
static void stream_write_change(char action, StringInfo s);
-static void stream_open_and_write_change(TransactionId xid, char action, StringInfo s);
-static void stream_close_file(void);
static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
@@ -676,6 +675,8 @@ static void on_exit_clear_xact_state(int code, Datum arg);
static bool send_internal_dependencies(ParallelApplyWorkerInfo *winfo,
StringInfo s);
+static bool build_dependency_with_last_committed_txn(ParallelApplyWorkerInfo *winfo);
+
/*
* Compute the hash value for entries in the replica_identity_table.
*/
@@ -1406,7 +1407,11 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
TransApplyAction apply_action;
StringInfoData original_msg;
- apply_action = get_transaction_apply_action(stream_xid, &winfo);
+ Assert(!in_streamed_transaction || TransactionIdIsValid(stream_xid));
+
+ apply_action = get_transaction_apply_action(in_streamed_transaction
+ ? stream_xid : remote_xid,
+ &winfo);
/* not in streaming mode */
if (apply_action == TRANS_LEADER_APPLY)
@@ -1415,8 +1420,6 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
return false;
}
- Assert(TransactionIdIsValid(stream_xid));
-
/*
* The parallel apply worker needs the xid in this message to decide
* whether to define a savepoint, so save the original message that has
@@ -1427,15 +1430,28 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
/*
* We should have received XID of the subxact as the first part of the
- * message, so extract it.
+ * message in streaming transactions, so extract it.
*/
- current_xid = pq_getmsgint(s, 4);
+ if (in_streamed_transaction)
+ current_xid = pq_getmsgint(s, 4);
+ else
+ current_xid = remote_xid;
if (!TransactionIdIsValid(current_xid))
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid transaction ID in streamed replication transaction")));
+ handle_dependency_on_change(action, s, current_xid, winfo);
+
+ /*
+ * Re-fetch the latest apply action as it might have been changed during
+ * dependency check.
+ */
+ apply_action = get_transaction_apply_action(in_streamed_transaction
+ ? stream_xid : remote_xid,
+ &winfo);
+
switch (apply_action)
{
case TRANS_LEADER_SERIALIZE:
@@ -1839,17 +1855,71 @@ static void
apply_handle_begin(StringInfo s)
{
LogicalRepBeginData begin_data;
+ ParallelApplyWorkerInfo *winfo;
+ TransApplyAction apply_action;
+
+ /* Save the message before it is consumed. */
+ StringInfoData original_msg = *s;
/* There must not be an active streaming transaction. */
Assert(!TransactionIdIsValid(stream_xid));
logicalrep_read_begin(s, &begin_data);
- set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn);
+
+ remote_xid = begin_data.xid;
+
+ set_apply_error_context_xact(remote_xid, begin_data.final_lsn);
remote_final_lsn = begin_data.final_lsn;
maybe_start_skipping_changes(begin_data.final_lsn);
+ pa_allocate_worker(remote_xid, false);
+
+ apply_action = get_transaction_apply_action(remote_xid, &winfo);
+
+ elog(DEBUG1, "new remote_xid %u", remote_xid);
+ switch (apply_action)
+ {
+ case TRANS_LEADER_APPLY:
+ break;
+
+ case TRANS_LEADER_SEND_TO_PARALLEL:
+ Assert(winfo);
+
+ if (pa_send_data(winfo, s->len, s->data))
+ {
+ pa_set_stream_apply_worker(winfo);
+ break;
+ }
+
+ /*
+ * Switch to serialize mode when we are not able to send the
+ * change to parallel apply worker.
+ */
+ pa_switch_to_partial_serialize(winfo, true);
+
+/* fall through */
+ case TRANS_LEADER_PARTIAL_SERIALIZE:
+ Assert(winfo);
+
+ stream_write_change(LOGICAL_REP_MSG_BEGIN, &original_msg);
+
+ /* Cache the parallel apply worker for this transaction. */
+ pa_set_stream_apply_worker(winfo);
+ break;
+
+ case TRANS_PARALLEL_APPLY:
+ /* Hold the lock until the end of the transaction. */
+ pa_lock_transaction(MyParallelShared->xid, AccessExclusiveLock);
+ pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_STARTED);
+ break;
+
+ default:
+ elog(ERROR, "unexpected apply action: %d", (int) apply_action);
+ break;
+ }
+
in_remote_transaction = true;
pgstat_report_activity(STATE_RUNNING, NULL);
@@ -1882,6 +1952,37 @@ send_internal_dependencies(ParallelApplyWorkerInfo *winfo, StringInfo s)
return false;
}
+/*
+ * Make a dependency between this and the lastly committed transaction.
+ *
+ * This function ensures that the commit ordering handled by parallel apply
+ * workers is preserved. Returns false if we switched to the serialize mode to
+ * send the massage, true otherwise.
+ */
+static bool
+build_dependency_with_last_committed_txn(ParallelApplyWorkerInfo *winfo)
+{
+ StringInfoData dependency_msg;
+ bool ret;
+
+ /* Skip if transactions have not been applied yet */
+ if (!TransactionIdIsValid(last_remote_xid))
+ return true;
+
+ /* Build the dependency message used to send to parallel apply worker */
+ initStringInfo(&dependency_msg);
+
+ pq_sendbyte(&dependency_msg, PARALLEL_APPLY_INTERNAL_MESSAGE);
+ pq_sendbyte(&dependency_msg, LOGICAL_REP_MSG_INTERNAL_DEPENDENCY);
+ pq_sendint32(&dependency_msg, 1);
+ pq_sendint32(&dependency_msg, last_remote_xid);
+
+ ret = send_internal_dependencies(winfo, &dependency_msg);
+
+ pfree(dependency_msg.data);
+ return ret;
+}
+
/*
* Handle COMMIT message.
*
@@ -1891,6 +1992,11 @@ static void
apply_handle_commit(StringInfo s)
{
LogicalRepCommitData commit_data;
+ ParallelApplyWorkerInfo *winfo;
+ TransApplyAction apply_action;
+
+ /* Save the message before it is consumed. */
+ StringInfoData original_msg = *s;
logicalrep_read_commit(s, &commit_data);
@@ -1901,7 +2007,98 @@ apply_handle_commit(StringInfo s)
LSN_FORMAT_ARGS(commit_data.commit_lsn),
LSN_FORMAT_ARGS(remote_final_lsn))));
- apply_handle_commit_internal(&commit_data);
+ apply_action = get_transaction_apply_action(remote_xid, &winfo);
+
+ switch (apply_action)
+ {
+ case TRANS_LEADER_APPLY:
+
+ /*
+ * Apart from parallelized transactions, we do not have to
+ * register this transaction to parallelized_txns. The commit
+ * ordering is always preserved.
+ */
+
+ /* Wait until the last transaction finishes */
+ if (TransactionIdIsValid(last_remote_xid))
+ pa_wait_for_depended_transaction(last_remote_xid);
+
+ apply_handle_commit_internal(&commit_data);
+
+ break;
+
+ case TRANS_LEADER_SEND_TO_PARALLEL:
+ Assert(winfo);
+
+ /*
+ * Mark this transaction as parallelized. This ensures that
+ * upcoming transactions wait until this transaction is committed.
+ */
+ pa_add_parallelized_transaction(remote_xid);
+
+ /*
+ * Build a dependency between this transaction and the lastly
+ * committed transaction to preserve the commit order. Then try to
+ * send a COMMIT message if succeeded.
+ */
+ if (build_dependency_with_last_committed_txn(winfo) &&
+ pa_send_data(winfo, s->len, s->data))
+ {
+ /* Finish processing the transaction. */
+ pa_xact_finish(winfo, commit_data.end_lsn);
+ break;
+ }
+
+ /*
+ * Switch to serialize mode when we are not able to send the
+ * change to parallel apply worker.
+ */
+ pa_switch_to_partial_serialize(winfo, true);
+/* fall through */
+ case TRANS_LEADER_PARTIAL_SERIALIZE:
+ Assert(winfo);
+
+ stream_open_and_write_change(remote_xid, LOGICAL_REP_MSG_COMMIT,
+ &original_msg);
+
+ pa_set_fileset_state(winfo->shared, FS_SERIALIZE_DONE);
+
+ /* Finish processing the transaction. */
+ pa_xact_finish(winfo, commit_data.end_lsn);
+ break;
+
+ case TRANS_PARALLEL_APPLY:
+
+ /*
+ * If the parallel apply worker is applying spooled messages then
+ * close the file before committing.
+ */
+ if (stream_fd)
+ stream_close_file();
+
+ INJECTION_POINT("parallel-worker-before-commit", NULL);
+
+ apply_handle_commit_internal(&commit_data);
+
+ MyParallelShared->last_commit_end = XactLastCommitEnd;
+
+ pa_commit_transaction();
+
+ pa_unlock_transaction(remote_xid, AccessExclusiveLock);
+ break;
+
+ default:
+ elog(ERROR, "unexpected apply action: %d", (int) apply_action);
+ break;
+ }
+
+ /* Cache the remote_xid */
+ last_remote_xid = remote_xid;
+
+ remote_xid = InvalidTransactionId;
+ in_remote_transaction = false;
+
+ elog(DEBUG1, "reset remote_xid %u", remote_xid);
/*
* Process any tables that are being synchronized in parallel, as well as
@@ -2024,7 +2221,8 @@ apply_handle_prepare(StringInfo s)
* XactLastCommitEnd, and adding it for this purpose doesn't seems worth
* it.
*/
- store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr);
+ store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr,
+ InvalidTransactionId);
in_remote_transaction = false;
@@ -2084,7 +2282,8 @@ apply_handle_commit_prepared(StringInfo s)
CommitTransactionCommand();
pgstat_report_stat(false);
- store_flush_position(prepare_data.end_lsn, XactLastCommitEnd);
+ store_flush_position(prepare_data.end_lsn, XactLastCommitEnd,
+ InvalidTransactionId);
in_remote_transaction = false;
/*
@@ -2153,7 +2352,8 @@ apply_handle_rollback_prepared(StringInfo s)
* transaction because we always flush the WAL record for it. See
* apply_handle_prepare.
*/
- store_flush_position(rollback_data.rollback_end_lsn, InvalidXLogRecPtr);
+ store_flush_position(rollback_data.rollback_end_lsn, InvalidXLogRecPtr,
+ InvalidTransactionId);
in_remote_transaction = false;
/*
@@ -2215,7 +2415,8 @@ apply_handle_stream_prepare(StringInfo s)
* It is okay not to set the local_end LSN for the prepare because
* we always flush the prepare record. See apply_handle_prepare.
*/
- store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr);
+ store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr,
+ InvalidTransactionId);
in_remote_transaction = false;
@@ -2467,6 +2668,11 @@ apply_handle_stream_start(StringInfo s)
case TRANS_LEADER_PARTIAL_SERIALIZE:
Assert(winfo);
+ /*
+ * TODO, the pa worker could start to wait too soon when
+ * processing some old stream start
+ */
+
/*
* Open the spool file unless it was already opened when switching
* to serialize mode. The transaction started in
@@ -3194,7 +3400,8 @@ apply_handle_commit_internal(LogicalRepCommitData *commit_data)
pgstat_report_stat(false);
- store_flush_position(commit_data->end_lsn, XactLastCommitEnd);
+ store_flush_position(commit_data->end_lsn, XactLastCommitEnd,
+ InvalidTransactionId);
}
else
{
@@ -3227,6 +3434,9 @@ apply_handle_relation(StringInfo s)
/* Also reset all entries in the partition map that refer to remoterel. */
logicalrep_partmap_reset_relmap(rel);
+
+ if (am_leader_apply_worker())
+ pa_distribute_schema_changes_to_workers(rel);
}
/*
@@ -4001,6 +4211,8 @@ FindDeletedTupleInLocalRel(Relation localrel, Oid localidxoid,
/*
* This handles insert, update, delete on a partitioned table.
+ *
+ * TODO, support parallel apply.
*/
static void
apply_handle_tuple_routing(ApplyExecutionData *edata,
@@ -4551,6 +4763,10 @@ apply_dispatch(StringInfo s)
* check which entries on it are already locally flushed. Those we can report
* as having been flushed.
*
+ * For non-streaming transactions managed by a parallel apply worker, we will
+ * get the local commit end from the shared parallel apply worker info once the
+ * transaction has been committed by the worker.
+ *
* The have_pending_txes is true if there are outstanding transactions that
* need to be flushed.
*/
@@ -4560,6 +4776,7 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
{
dlist_mutable_iter iter;
XLogRecPtr local_flush = GetFlushRecPtr(NULL);
+ List *committed_pa_xid = NIL;
*write = InvalidXLogRecPtr;
*flush = InvalidXLogRecPtr;
@@ -4569,6 +4786,36 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
FlushPosition *pos =
dlist_container(FlushPosition, node, iter.cur);
+ if (TransactionIdIsValid(pos->pa_remote_xid) &&
+ XLogRecPtrIsInvalid(pos->local_end))
+ {
+ bool skipped_write;
+
+ pos->local_end = pa_get_last_commit_end(pos->pa_remote_xid, true,
+ &skipped_write);
+
+ elog(DEBUG1,
+ "got commit end from parallel apply worker, "
+ "txn: %u, remote_end %X/%X, local_end %X/%X",
+ pos->pa_remote_xid, LSN_FORMAT_ARGS(pos->remote_end),
+ LSN_FORMAT_ARGS(pos->local_end));
+
+ /*
+ * Break the loop if the worker has not finished applying the
+ * transaction. There's no need to check subsequent transactions,
+ * as they must commit after the current transaction being
+ * examined and thus won't have their commit end available yet.
+ */
+ if (!skipped_write && XLogRecPtrIsInvalid(pos->local_end))
+ break;
+
+ committed_pa_xid = lappend_xid(committed_pa_xid, pos->pa_remote_xid);
+ }
+
+ /*
+ * Worker has finished applying or the transaction was applied in the
+ * leader apply worker
+ */
*write = pos->remote_end;
if (pos->local_end <= local_flush)
@@ -4577,29 +4824,19 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
dlist_delete(iter.cur);
pfree(pos);
}
- else
- {
- /*
- * Don't want to uselessly iterate over the rest of the list which
- * could potentially be long. Instead get the last element and
- * grab the write position from there.
- */
- pos = dlist_tail_element(FlushPosition, node,
- &lsn_mapping);
- *write = pos->remote_end;
- *have_pending_txes = true;
- return;
- }
}
*have_pending_txes = !dlist_is_empty(&lsn_mapping);
+
+ cleanup_replica_identity_table(committed_pa_xid);
}
/*
* Store current remote/local lsn pair in the tracking list.
*/
void
-store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
+store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn,
+ TransactionId remote_xid)
{
FlushPosition *flushpos;
@@ -4617,6 +4854,7 @@ store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
flushpos = palloc_object(FlushPosition);
flushpos->local_end = local_lsn;
flushpos->remote_end = remote_lsn;
+ flushpos->pa_remote_xid = remote_xid;
dlist_push_tail(&lsn_mapping, &flushpos->node);
MemoryContextSwitchTo(ApplyMessageContext);
@@ -6064,7 +6302,7 @@ stream_cleanup_files(Oid subid, TransactionId xid)
* changes for this transaction, create the buffile, otherwise open the
* previously created file.
*/
-static void
+void
stream_open_file(Oid subid, TransactionId xid, bool first_segment)
{
char path[MAXPGPATH];
@@ -6109,7 +6347,7 @@ stream_open_file(Oid subid, TransactionId xid, bool first_segment)
* stream_close_file
* Close the currently open file with streamed changes.
*/
-static void
+void
stream_close_file(void)
{
Assert(stream_fd != NULL);
@@ -6157,7 +6395,7 @@ stream_write_change(char action, StringInfo s)
* target file if not already before writing the message and close the file at
* the end.
*/
-static void
+void
stream_open_and_write_change(TransactionId xid, char action, StringInfo s)
{
Assert(!in_streamed_transaction);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 9042470f500..c0c0b385d40 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -253,6 +253,8 @@ extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecP
extern void logicalrep_write_rel(StringInfo out, TransactionId xid,
Relation rel, Bitmapset *columns,
PublishGencolsType include_gencols_type);
+extern void logicalrep_write_internal_rel(StringInfo out,
+ LogicalRepRelation *rel);
extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
extern void logicalrep_write_typ(StringInfo out, TransactionId xid,
Oid typoid);
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index f9aa9e59a6d..27e365f84c2 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -52,6 +52,8 @@ extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel,
LOCKMODE lockmode);
extern bool IsIndexUsableForReplicaIdentityFull(Relation idxrel, AttrMap *attrmap);
extern Oid GetRelationIdentityOrPK(Relation rel);
+extern int logicalrep_get_num_rels(void);
+extern void logicalrep_write_all_rels(StringInfo out);
extern LogicalRepRelMapEntry *logicalrep_get_relentry(LogicalRepRelId remoteid);
#endif /* LOGICALRELATION_H */
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 1dbec625c81..d5bf4598980 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -314,6 +314,10 @@ extern void apply_dispatch(StringInfo s);
extern void maybe_reread_subscription(void);
extern void stream_cleanup_files(Oid subid, TransactionId xid);
+extern void stream_open_file(Oid subid, TransactionId xid, bool first_segment);
+extern void stream_close_file(void);
+extern void stream_open_and_write_change(TransactionId xid, char action,
+ StringInfo s);
extern void set_stream_options(WalRcvStreamOptions *options,
char *slotname,
@@ -327,7 +331,8 @@ extern void SetupApplyOrSyncWorker(int worker_slot);
extern void DisableSubscriptionAndExit(void);
-extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn);
+extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn,
+ TransactionId remote_xid);
/* Function for apply error callback */
extern void apply_error_callback(void *arg);
@@ -342,6 +347,7 @@ extern void pa_detach_all_error_mq(void);
extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
const void *data);
+extern void pa_distribute_schema_changes_to_workers(LogicalRepRelation *rel);
extern void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
bool stream_locked);
@@ -368,8 +374,9 @@ extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
XLogRecPtr remote_lsn);
extern bool pa_transaction_committed(TransactionId xid);
extern void pa_record_dependency_on_transactions(List *depends_on_xids);
-
+extern void pa_commit_transaction(void);
extern void pa_wait_for_depended_transaction(TransactionId xid);
+extern void pa_add_parallelized_transaction(TransactionId xid);
#define isParallelApplyWorker(worker) ((worker)->in_use && \
(worker)->type == WORKERTYPE_PARALLEL_APPLY)
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index a4c7dbaff59..6251eec2030 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -46,6 +46,7 @@ tests += {
't/034_temporal.pl',
't/035_conflicts.pl',
't/036_sequences.pl',
+ 't/050_parallel_apply.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl
index d7e62e4d488..05d02ef6407 100644
--- a/src/test/subscription/t/001_rep_changes.pl
+++ b/src/test/subscription/t/001_rep_changes.pl
@@ -16,6 +16,8 @@ $node_publisher->start;
# Create subscriber node
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ "max_logical_replication_workers = 10");
$node_subscriber->start;
# Create some preexisting content on publisher
diff --git a/src/test/subscription/t/010_truncate.pl b/src/test/subscription/t/010_truncate.pl
index 945505d0239..e15a6bb2a03 100644
--- a/src/test/subscription/t/010_truncate.pl
+++ b/src/test/subscription/t/010_truncate.pl
@@ -17,7 +17,7 @@ $node_publisher->start;
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
$node_subscriber->append_conf('postgresql.conf',
- qq(max_logical_replication_workers = 6));
+ qq(max_logical_replication_workers = 7));
$node_subscriber->start;
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index ac96bc3f009..705bc83f5ea 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -232,6 +232,12 @@ $node_subscriber->wait_for_log(
$node_publisher->safe_psql('postgres', "INSERT INTO test_tab_2 values(1)");
+# FIXME: Currently, non-streaming transactions are applied in parallel by
+# default. So, the first transaction is handled by a parallel apply worker. To
+# trigger the deadlock, initiate an more transaction to be applied by the
+# leader.
+$node_publisher->safe_psql('postgres', "INSERT INTO test_tab_2 values(1)");
+
$h->query_safe('COMMIT');
$h->quit;
@@ -247,7 +253,7 @@ $node_publisher->wait_for_catchup($appname);
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
-is($result, qq(5001), 'data replicated to subscriber after dropping index');
+is($result, qq(5002), 'data replicated to subscriber after dropping index');
# Clean up test data from the environment.
$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
diff --git a/src/test/subscription/t/026_stats.pl b/src/test/subscription/t/026_stats.pl
index 5d457060a02..911005bde20 100644
--- a/src/test/subscription/t/026_stats.pl
+++ b/src/test/subscription/t/026_stats.pl
@@ -16,6 +16,7 @@ $node_publisher->start;
# Create subscriber node.
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf', "max_logical_replication_workers = 10");
$node_subscriber->start;
diff --git a/src/test/subscription/t/027_nosuperuser.pl b/src/test/subscription/t/027_nosuperuser.pl
index 322f5b4cc6a..fdfc44ac729 100644
--- a/src/test/subscription/t/027_nosuperuser.pl
+++ b/src/test/subscription/t/027_nosuperuser.pl
@@ -86,6 +86,7 @@ $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_publisher->init(allows_streaming => 'logical');
$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf', "max_logical_replication_workers = 10");
$node_publisher->start;
$node_subscriber->start;
$publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl
new file mode 100644
index 00000000000..69cf48cb7ac
--- /dev/null
+++ b/src/test/subscription/t/050_parallel_apply.pl
@@ -0,0 +1,130 @@
+
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# This tests that dependency tracking between transactions can work well
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# Insert initial data
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE regress_tab (id int PRIMARY KEY, value text);");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (generate_series(1, 10), 'test');");
+
+# Create a publication
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION regress_pub FOR ALL TABLES;");
+
+# Initialize subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf', "log_min_messages = debug1");
+$node_subscriber->append_conf('postgresql.conf',
+ "max_logical_replication_workers = 10");
+$node_subscriber->start;
+
+# Check if the extension injection_points is available, as it may be
+# possible that this script is run with installcheck, where the module
+# would not be installed by default.
+if (!$node_subscriber->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+$node_subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+
+# Create a subscription
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE regress_tab (id int PRIMARY KEY, value text);");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION regress_sub CONNECTION '$publisher_connstr' PUBLICATION regress_pub;");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'regress_sub');
+
+# Insert tuples on publisher
+#
+# XXX This may not enough to launch a parallel apply worker, because
+# table_states_not_ready is not discarded yet.
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (generate_series(11, 20), 'test');");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Insert tuples again
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (generate_series(21, 30), 'test');");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# Verify the parallel apply worker is launched
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(1) FROM pg_stat_activity WHERE backend_type = 'logical replication parallel worker'");
+is($result, '1', "parallel apply worker is laucnhed by a non-streamed transaction");
+
+# 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 tuples on publisher
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (generate_series(31, 40), 'test');");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+ 'parallel-worker-before-commit');
+
+my $offset = -s $node_subscriber->logfile;
+
+# Insert tuples on publisher again. This transaction is independent from the
+# previous one, but the parallel worker would wait till it finishes
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (generate_series(41, 50), 'test');");
+
+# Verify the parallel worker waits for the transaction
+my $str = $node_subscriber->wait_for_log(qr/wait for depended xid ([1-9][0-9]+)/, $offset);
+my $xid = $str =~ /wait for depended xid ([1-9][0-9]+)/;
+
+# Update tuples which have not been applied yet on subscriber because the
+# parallel worker stops at the injection point. Newly assigned worker also
+# waits for the same transactions as above.
+$node_publisher->safe_psql('postgres',
+ "UPDATE regress_tab SET value = 'updated' WHERE id BETWEEN 31 AND 35;");
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+# Wakeup the parallel worker. We detach first no to stop other parallel workers
+$node_subscriber->safe_psql('postgres', qq[
+ SELECT injection_points_detach('parallel-worker-before-commit');
+ SELECT injection_points_wakeup('parallel-worker-before-commit');
+]);
+
+# Verify the parallel worker wakes up
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM regress_tab");
+is ($result, 50, 'inserts are replicated to subscriber');
+
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(1) FROM regress_tab WHERE value = 'updated'");
+is ($result, 5, 'updates are also replicated to subscriber');
+
+done_testing();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 05936d61a54..811a23ac03c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2094,6 +2094,7 @@ ParallelHashGrowth
ParallelHashJoinBatch
ParallelHashJoinBatchAccessor
ParallelHashJoinState
+ParallelizedTxnEntry
ParallelIndexScanDesc
ParallelizedTxnEntry
ParallelSlot
@@ -4098,6 +4099,7 @@ rendezvousHashEntry
rep
replace_rte_variables_callback
replace_rte_variables_context
+replica_identity_hash
report_error_fn
ret_type
rewind_source
--
2.47.3
[application/octet-stream] v8-0005-support-2PC.patch (13.9K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/6-v8-0005-support-2PC.patch)
download | inline diff:
From 4e76241460a0707dbbeabb81e36723ddacf90bcb Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Tue, 2 Dec 2025 13:01:26 +0900
Subject: [PATCH v8 5/9] support 2PC
This patch allows the PREPARE transaction to be applied in parallel. Parallel
apply workers are assigned to a transaction when BEGIN_PREPARE is received. This
part and the dependency-waiting mechanism are the same as a normal transaction.
A parallel worker can be freed after it handles a PREPARE message. The prepared
transaction can be deleted from parallelized_txns at that time; the upcoming
transactions will wait until then.
The leader apply worker resolves COMMIT PREPARED/ROLLBACK PREPARED. Since it can
be serialized automatically, it does not add the transaction to parallelized_txns.
---
src/backend/replication/logical/worker.c | 239 ++++++++++++++++--
src/test/subscription/t/050_parallel_apply.pl | 60 +++++
2 files changed, 271 insertions(+), 28 deletions(-)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index d1ba801f96f..41da800b2e4 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -756,6 +756,14 @@ cleanup_replica_identity_table(List *committed_xid)
if (!committed_xid)
return;
+ /*
+ * Skip if the replica_identity_table is not initialized yet. This can
+ * happen if the empty transaction was replicated and a parallel apply
+ * worker was launched. See comments in apply_handle_prepare().
+ */
+ if (!replica_identity_table)
+ return;
+
replica_identity_start_iterate(replica_identity_table, &i);
while ((rientry = replica_identity_iterate(replica_identity_table, &i)) != NULL)
{
@@ -2117,6 +2125,11 @@ static void
apply_handle_begin_prepare(StringInfo s)
{
LogicalRepPreparedTxnData begin_data;
+ ParallelApplyWorkerInfo *winfo;
+ TransApplyAction apply_action;
+
+ /* Save the message before it is consumed. */
+ StringInfoData original_msg = *s;
/* Tablesync should never receive prepare. */
if (am_tablesync_worker())
@@ -2128,12 +2141,61 @@ apply_handle_begin_prepare(StringInfo s)
Assert(!TransactionIdIsValid(stream_xid));
logicalrep_read_begin_prepare(s, &begin_data);
- set_apply_error_context_xact(begin_data.xid, begin_data.prepare_lsn);
+
+ remote_xid = begin_data.xid;
+
+ set_apply_error_context_xact(remote_xid, begin_data.prepare_lsn);
remote_final_lsn = begin_data.prepare_lsn;
maybe_start_skipping_changes(begin_data.prepare_lsn);
+ pa_allocate_worker(remote_xid, false);
+
+ apply_action = get_transaction_apply_action(remote_xid, &winfo);
+
+ elog(DEBUG1, "new remote_xid %u", remote_xid);
+ switch (apply_action)
+ {
+ case TRANS_LEADER_APPLY:
+ break;
+
+ case TRANS_LEADER_SEND_TO_PARALLEL:
+ Assert(winfo);
+
+ if (pa_send_data(winfo, s->len, s->data))
+ {
+ pa_set_stream_apply_worker(winfo);
+ break;
+ }
+
+ /*
+ * Switch to serialize mode when we are not able to send the
+ * change to parallel apply worker.
+ */
+ pa_switch_to_partial_serialize(winfo, true);
+
+/* fall through */
+ case TRANS_LEADER_PARTIAL_SERIALIZE:
+ Assert(winfo);
+
+ stream_write_change(LOGICAL_REP_MSG_BEGIN_PREPARE, &original_msg);
+
+ /* Cache the parallel apply worker for this transaction. */
+ pa_set_stream_apply_worker(winfo);
+ break;
+
+ case TRANS_PARALLEL_APPLY:
+ /* Hold the lock until the end of the transaction. */
+ pa_lock_transaction(MyParallelShared->xid, AccessExclusiveLock);
+ pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_STARTED);
+ break;
+
+ default:
+ elog(ERROR, "unexpected apply action: %d", (int) apply_action);
+ break;
+ }
+
in_remote_transaction = true;
pgstat_report_activity(STATE_RUNNING, NULL);
@@ -2183,6 +2245,11 @@ static void
apply_handle_prepare(StringInfo s)
{
LogicalRepPreparedTxnData prepare_data;
+ ParallelApplyWorkerInfo *winfo;
+ TransApplyAction apply_action;
+
+ /* Save the message before it is consumed. */
+ StringInfoData original_msg = *s;
logicalrep_read_prepare(s, &prepare_data);
@@ -2193,36 +2260,137 @@ apply_handle_prepare(StringInfo s)
LSN_FORMAT_ARGS(prepare_data.prepare_lsn),
LSN_FORMAT_ARGS(remote_final_lsn))));
- /*
- * Unlike commit, here, we always prepare the transaction even though no
- * change has happened in this transaction or all changes are skipped. It
- * is done this way because at commit prepared time, we won't know whether
- * we have skipped preparing a transaction because of those reasons.
- *
- * XXX, We can optimize such that at commit prepared time, we first check
- * whether we have prepared the transaction or not but that doesn't seem
- * worthwhile because such cases shouldn't be common.
- */
- begin_replication_step();
+ apply_action = get_transaction_apply_action(remote_xid, &winfo);
- apply_handle_prepare_internal(&prepare_data);
+ switch (apply_action)
+ {
+ case TRANS_LEADER_APPLY:
- end_replication_step();
- CommitTransactionCommand();
- pgstat_report_stat(false);
+ /*
+ * Unlike commit, here, we always prepare the transaction even
+ * though no change has happened in this transaction or all
+ * changes are skipped. It is done this way because at commit
+ * prepared time, we won't know whether we have skipped preparing
+ * a transaction because of those reasons.
+ *
+ * XXX, We can optimize such that at commit prepared time, we
+ * first check whether we have prepared the transaction or not but
+ * that doesn't seem worthwhile because such cases shouldn't be
+ * common.
+ */
+ begin_replication_step();
- /*
- * It is okay not to set the local_end LSN for the prepare because we
- * always flush the prepare record. So, we can send the acknowledgment of
- * the remote_end LSN as soon as prepare is finished.
- *
- * XXX For the sake of consistency with commit, we could have set it with
- * the LSN of prepare but as of now we don't track that value similar to
- * XactLastCommitEnd, and adding it for this purpose doesn't seems worth
- * it.
- */
- store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr,
- InvalidTransactionId);
+ /* Wait until the last transaction finishes */
+ if (TransactionIdIsValid(last_remote_xid))
+ pa_wait_for_depended_transaction(last_remote_xid);
+
+ apply_handle_prepare_internal(&prepare_data);
+
+ end_replication_step();
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+
+ /*
+ * It is okay not to set the local_end LSN for the prepare because
+ * we always flush the prepare record. So, we can send the
+ * acknowledgment of the remote_end LSN as soon as prepare is
+ * finished.
+ *
+ * XXX For the sake of consistency with commit, we could have set
+ * it with the LSN of prepare but as of now we don't track that
+ * value similar to XactLastCommitEnd, and adding it for this
+ * purpose doesn't seems worth it.
+ */
+ store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr,
+ InvalidTransactionId);
+
+ break;
+
+ case TRANS_LEADER_SEND_TO_PARALLEL:
+ Assert(winfo);
+
+ /*
+ * Mark this transaction as parallelized. This ensures that
+ * upcoming transactions wait until this transaction is committed.
+ */
+ pa_add_parallelized_transaction(remote_xid);
+
+ /*
+ * Build a dependency between this transaction and the lastly
+ * committed transaction to preserve the commit order. Then try to
+ * send a COMMIT message if succeeded.
+ */
+ if (build_dependency_with_last_committed_txn(winfo) &&
+ pa_send_data(winfo, s->len, s->data))
+ {
+ /* Finish processing the transaction. */
+ pa_xact_finish(winfo, prepare_data.end_lsn);
+ break;
+ }
+
+ /*
+ * Switch to serialize mode when we are not able to send the
+ * change to parallel apply worker.
+ */
+ pa_switch_to_partial_serialize(winfo, true);
+/* fall through */
+ case TRANS_LEADER_PARTIAL_SERIALIZE:
+ Assert(winfo);
+
+ stream_open_and_write_change(remote_xid, LOGICAL_REP_MSG_PREPARE,
+ &original_msg);
+
+ pa_set_fileset_state(winfo->shared, FS_SERIALIZE_DONE);
+
+ /* Finish processing the transaction. */
+ pa_xact_finish(winfo, prepare_data.end_lsn);
+ break;
+
+ case TRANS_PARALLEL_APPLY:
+
+ /*
+ * If the parallel apply worker is applying spooled messages then
+ * close the file before committing.
+ */
+ if (stream_fd)
+ stream_close_file();
+
+ begin_replication_step();
+
+ INJECTION_POINT("parallel-worker-before-prepare", NULL);
+
+ /* Mark the transaction as prepared. */
+ apply_handle_prepare_internal(&prepare_data);
+
+ end_replication_step();
+
+ CommitTransactionCommand();
+ pgstat_report_stat(false);
+
+ store_flush_position(prepare_data.end_lsn, InvalidXLogRecPtr,
+ InvalidTransactionId);
+
+ /*
+ * It is okay not to set the local_end LSN for the prepare because
+ * we always flush the prepare record. See apply_handle_prepare.
+ */
+ MyParallelShared->last_commit_end = InvalidXLogRecPtr;
+ pa_commit_transaction();
+
+ pa_unlock_transaction(MyParallelShared->xid, AccessExclusiveLock);
+
+ pa_reset_subtrans();
+ break;
+
+ default:
+ elog(ERROR, "unexpected apply action: %d", (int) apply_action);
+ break;
+ }
+
+ /* Cache the remote_xid */
+ last_remote_xid = remote_xid;
+
+ remote_xid = InvalidTransactionId;
in_remote_transaction = false;
@@ -2270,6 +2438,9 @@ apply_handle_commit_prepared(StringInfo s)
/* There is no transaction when COMMIT PREPARED is called */
begin_replication_step();
+ if (TransactionIdIsValid(last_remote_xid))
+ pa_wait_for_depended_transaction(last_remote_xid);
+
/*
* Update origin state so we can restart streaming from correct position
* in case of crash.
@@ -2282,6 +2453,14 @@ apply_handle_commit_prepared(StringInfo s)
CommitTransactionCommand();
pgstat_report_stat(false);
+ /*
+ * No need to update last_remote_xid because leader worker applied the
+ * message thus upcoming transaction preserves the order automatically.
+ * Let's set the xid to an invalid value to skip sending the
+ * INTERNAL_DEPENDENCY message.
+ */
+ last_remote_xid = InvalidTransactionId;
+
store_flush_position(prepare_data.end_lsn, XactLastCommitEnd,
InvalidTransactionId);
in_remote_transaction = false;
@@ -2338,6 +2517,10 @@ apply_handle_rollback_prepared(StringInfo s)
/* There is no transaction when ABORT/ROLLBACK PREPARED is called */
begin_replication_step();
+
+ if (TransactionIdIsValid(last_remote_xid))
+ pa_wait_for_depended_transaction(last_remote_xid);
+
FinishPreparedTransaction(gid, false);
end_replication_step();
CommitTransactionCommand();
diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl
index 69cf48cb7ac..15973f7d0e0 100644
--- a/src/test/subscription/t/050_parallel_apply.pl
+++ b/src/test/subscription/t/050_parallel_apply.pl
@@ -17,6 +17,8 @@ if ($ENV{enable_injection_points} ne 'yes')
# Initialize publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ "max_prepared_transactions = 10");
$node_publisher->start;
# Insert initial data
@@ -35,6 +37,8 @@ $node_subscriber->init;
$node_subscriber->append_conf('postgresql.conf', "log_min_messages = debug1");
$node_subscriber->append_conf('postgresql.conf',
"max_logical_replication_workers = 10");
+$node_subscriber->append_conf('postgresql.conf',
+ "max_prepared_transactions = 10");
$node_subscriber->start;
# Check if the extension injection_points is available, as it may be
@@ -127,4 +131,60 @@ $result =
"SELECT count(1) FROM regress_tab WHERE value = 'updated'");
is ($result, 5, 'updates are also replicated to subscriber');
+# Ensure PREPAREd transaction also affects the parallel apply
+
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub DISABLE;");
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication apply worker'"
+);
+$node_subscriber->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION regress_sub SET (two_phase = on);
+ ALTER SUBSCRIPTION regress_sub ENABLE;");
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(1) FROM pg_stat_activity WHERE backend_type = 'logical replication parallel worker'");
+is($result, '0', "no parallel apply workers exist after restart");
+
+# Attach an injection_point. Parallel workers would wait before the prepare
+$node_subscriber->safe_psql('postgres',
+ "SELECT injection_points_attach('parallel-worker-before-prepare','wait');"
+);
+
+# PREPARE a transaction on publisher. It would be handled by a parallel apply
+# worker.
+$node_publisher->safe_psql('postgres', qq[
+ BEGIN;
+ INSERT INTO regress_tab VALUES (generate_series(51, 60), 'prepare');
+ PREPARE TRANSACTION 'regress_prepare';
+]);
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+ 'parallel-worker-before-prepare');
+
+$offset = -s $node_subscriber->logfile;
+
+# Insert tuples on publisher again. This transaction waits for the prepared
+# transaction
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (generate_series(61, 70), 'test');");
+
+# Verify the parallel worker waits for the transaction
+$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]+)/;
+
+# Wakeup the parallel worker
+$node_subscriber->safe_psql('postgres', qq[
+ SELECT injection_points_detach('parallel-worker-before-prepare');
+ SELECT injection_points_wakeup('parallel-worker-before-prepare');
+]);
+
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
+# COMMIT the prepared transaction. It is always handled by the leader
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'regress_prepare';");
+$node_publisher->wait_for_catchup('regress_sub');
+
done_testing();
--
2.47.3
[application/octet-stream] v8-0006-Track-dependencies-for-streamed-transactions.patch (10.7K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/7-v8-0006-Track-dependencies-for-streamed-transactions.patch)
download | inline diff:
From 09d21dd6b69c2e07e584227547713a69ebd5d337 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Thu, 4 Dec 2025 20:55:26 +0900
Subject: [PATCH v8 6/9] Track dependencies for streamed transactions
This commit allows tracking dependencies of streamed transactions.
Regarding the streaming=on case, dependency tracking is enabled while applying
spooled changes from files.
In the streaming=parallel case, dependency tracking is performed when the leader
sends changes to parallel workers. Apart from non-streamed transactions, the
leader waits for parallel workers till the assigned transactions are finished at
COMMIT/PREPARE/ABORT; thus, the XID of streamed transactions is not cached as
the lastly handled one. Also, streamed transactions are not recorded as
parallelized transactions because upcoming workers do not have to wait for them.
---
.../replication/logical/applyparallelworker.c | 19 +++++-
src/backend/replication/logical/worker.c | 66 +++++++++++++++++--
src/include/replication/worker_internal.h | 2 +-
src/test/subscription/t/050_parallel_apply.pl | 47 +++++++++++++
4 files changed, 126 insertions(+), 8 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 3396ee1cdf1..173a81f4d9c 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -168,7 +168,14 @@
* key) as another ongoing transaction (see handle_dependency_on_change for
* details). If so, the leader sends a list of dependent transaction IDs to the
* parallel worker, indicating that the parallel apply worker must wait for
- * these transactions to commit before proceeding.
+ * these transactions to commit before proceeding. If transactions are streamed
+ * but leader deciedes no to assign parallel apply workers, dependencies are
+ * verified when the transaction is committed.
+ *
+ * Non-streaming transactions
+ * ======================
+ * The handling is similar to streaming transactions, but including few
+ * differences:
*
* Commit order
* ------------
@@ -1635,6 +1642,12 @@ pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo)
stream_apply_worker = winfo;
}
+bool
+pa_stream_apply_worker_is_null(void)
+{
+ return stream_apply_worker == NULL;
+}
+
/*
* Form a unique savepoint name for the streaming transaction.
*
@@ -1720,6 +1733,10 @@ pa_stream_abort(LogicalRepStreamAbortData *abort_data)
TransactionId xid = abort_data->xid;
TransactionId subxid = abort_data->subxid;
+ /* Streamed transactions won't be registered */
+ Assert(!dshash_find(parallelized_txns, &xid, false) &&
+ !dshash_find(parallelized_txns, &subxid, false));
+
/*
* Update origin state so we can restart streaming from correct position
* in case of crash.
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 41da800b2e4..7f9d99901c0 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -969,13 +969,26 @@ check_dependency_on_replica_identity(Oid relid,
&rientry->remote_xid,
new_depended_xid);
+ /*
+ * Remove the entry if it is registered for the streamed transactions. We
+ * do not have to register an entry for them; The leader worker always
+ * waits until the parallel worker finishes handling streamed
+ * transactions, thus no need to consider the possiblity that upcoming
+ * parallel workers would go ahead.
+ */
+ if (TransactionIdIsValid(stream_xid) && !found)
+ {
+ free_replica_identity_key(rientry->keydata);
+ replica_identity_delete_item(replica_identity_table, rientry);
+ }
+
/*
* Update the new depended xid into the entry if valid, 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, so no need to be depended on.
*/
- if (TransactionIdIsValid(new_depended_xid))
+ else if (TransactionIdIsValid(new_depended_xid))
rientry->remote_xid = new_depended_xid;
/*
@@ -1089,8 +1102,11 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
*/
StringInfoData change = *s;
- /* Compute dependency only for non-streaming transaction */
- if (in_streamed_transaction || (winfo && winfo->stream_txn))
+ /*
+ * Skip if we are handling streaming transactions but changes are not
+ * applied yet.
+ */
+ if (pa_stream_apply_worker_is_null() && in_streamed_transaction)
return;
/* Only the leader checks dependencies and schedules the parallel apply */
@@ -1450,7 +1466,18 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid transaction ID in streamed replication transaction")));
- handle_dependency_on_change(action, s, current_xid, winfo);
+ /*
+ * Check dependencies related to the received change. The XID of the top
+ * transaction is always used to avoid detecting false-positive
+ * dependencies between top and sub transactions. Sub-transactions can be
+ * replicated for streamed transactions, and they won't be marked as
+ * parallelized so that parallel workers won't wait for rolled-back
+ * sub-transactions.
+ */
+ handle_dependency_on_change(action, s,
+ in_streamed_transaction
+ ? stream_xid : remote_xid,
+ winfo);
/*
* Re-fetch the latest apply action as it might have been changed during
@@ -2589,6 +2616,10 @@ apply_handle_stream_prepare(StringInfo s)
apply_spooled_messages(MyLogicalRepWorker->stream_fileset,
prepare_data.xid, prepare_data.prepare_lsn);
+ /* Wait until the last transaction finishes */
+ if (TransactionIdIsValid(last_remote_xid))
+ pa_wait_for_depended_transaction(last_remote_xid);
+
/* Mark the transaction as prepared. */
apply_handle_prepare_internal(&prepare_data);
@@ -2612,7 +2643,8 @@ apply_handle_stream_prepare(StringInfo s)
case TRANS_LEADER_SEND_TO_PARALLEL:
Assert(winfo);
- if (pa_send_data(winfo, s->len, s->data))
+ if (build_dependency_with_last_committed_txn(winfo) &&
+ pa_send_data(winfo, s->len, s->data))
{
/* Finish processing the streaming transaction. */
pa_xact_finish(winfo, prepare_data.end_lsn);
@@ -2678,6 +2710,11 @@ apply_handle_stream_prepare(StringInfo s)
pgstat_report_stat(false);
+ /*
+ * No need to update the last_remote_xid here because leader worker always
+ * wait until streamed transactions finish.
+ */
+
/*
* Process any tables that are being synchronized in parallel, as well as
* any newly added tables or sequences.
@@ -3462,6 +3499,10 @@ apply_handle_stream_commit(StringInfo s)
apply_spooled_messages(MyLogicalRepWorker->stream_fileset, xid,
commit_data.commit_lsn);
+ /* Wait until the last transaction finishes */
+ if (TransactionIdIsValid(last_remote_xid))
+ pa_wait_for_depended_transaction(last_remote_xid);
+
apply_handle_commit_internal(&commit_data);
/* Unlink the files with serialized changes and subxact info. */
@@ -3473,7 +3514,20 @@ apply_handle_stream_commit(StringInfo s)
case TRANS_LEADER_SEND_TO_PARALLEL:
Assert(winfo);
- if (pa_send_data(winfo, s->len, s->data))
+ /*
+ * Apart from non-streaming case, no need to mark this transaction
+ * as parallelized. Because the leader waits until the streamed
+ * transaction is committed thus commit ordering is always
+ * preserved.
+ */
+
+ /*
+ * Build a dependency between this transaction and the lastly
+ * committed transaction to preserve the commit order. Then try to
+ * send a COMMIT message if succeeded.
+ */
+ if (build_dependency_with_last_committed_txn(winfo) &&
+ pa_send_data(winfo, s->len, s->data))
{
/* Finish processing the streaming transaction. */
pa_xact_finish(winfo, commit_data.end_lsn);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index d5bf4598980..5d7143b79b1 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -354,7 +354,7 @@ extern void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
extern void pa_set_xact_state(ParallelApplyWorkerShared *wshared,
ParallelTransState xact_state);
extern void pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo);
-
+extern bool pa_stream_apply_worker_is_null(void);
extern void pa_start_subtrans(TransactionId current_xid,
TransactionId top_xid);
extern void pa_reset_subtrans(void);
diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl
index 15973f7d0e0..9254b85d350 100644
--- a/src/test/subscription/t/050_parallel_apply.pl
+++ b/src/test/subscription/t/050_parallel_apply.pl
@@ -187,4 +187,51 @@ $node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset
$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'regress_prepare';");
$node_publisher->wait_for_catchup('regress_sub');
+# Ensure streamed transactions waits the previous transaction
+
+$node_publisher->append_conf('postgresql.conf',
+ "logical_decoding_work_mem = 64kB");
+$node_publisher->reload;
+# Run a query to make sure that the reload has taken effect.
+$node_publisher->safe_psql('postgres', "SELECT 1");
+
+# Attach the injection_point again
+$node_subscriber->safe_psql('postgres',
+ "SELECT injection_points_attach('parallel-worker-before-commit','wait');"
+);
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (generate_series(71, 80), 'test');");
+
+# Wait until the parallel worker enters the injection point.
+$node_subscriber->wait_for_event('logical replication parallel worker',
+ 'parallel-worker-before-commit');
+
+# Run a transaction which would be streamed
+my $h = $node_publisher->background_psql('postgres', on_error_stop => 0);
+
+$offset = -s $node_subscriber->logfile;
+
+$h->query_safe(
+ q{
+BEGIN;
+UPDATE regress_tab SET value = 'streamed-updated' WHERE id BETWEEN 71 AND 80;
+INSERT INTO regress_tab VALUES (generate_series(100, 5100), 'streamed');
+});
+
+# Verify the parallel worker waits for the transaction
+$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]+)/;
+
+# 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);
+
+$h->query_safe("COMMIT;");
+
done_testing();
--
2.47.3
[application/octet-stream] v8-0007-Wait-applying-transaction-if-one-of-user-defined-.patch (11.8K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/8-v8-0007-Wait-applying-transaction-if-one-of-user-defined-.patch)
download | inline diff:
From 88970abe40b78327f2280089071f2c9c1c71d0e3 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Tue, 23 Dec 2025 17:58:15 +0900
Subject: [PATCH v8 7/9] Wait applying transaction if one of user-defined
triggers is immutable
Since many parallel workers apply transactions, triggers for relations can also
be fired in parallel, which may obtain unexpected results. To make it safe,
parallel apply workers wait for the previously dispatched transaction before
applying changes to the relation that has mutable triggers.
---
src/backend/replication/logical/relation.c | 123 ++++++++++++++++++---
src/backend/replication/logical/worker.c | 68 ++++++++++++
src/include/replication/logicalrelation.h | 20 ++++
3 files changed, 197 insertions(+), 14 deletions(-)
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index e1ce183dfd3..eeb85f8cc5d 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -21,7 +21,9 @@
#include "access/genam.h"
#include "access/table.h"
#include "catalog/namespace.h"
+#include "catalog/pg_proc.h"
#include "catalog/pg_subscription_rel.h"
+#include "commands/trigger.h"
#include "executor/executor.h"
#include "nodes/makefuncs.h"
#include "replication/logicalrelation.h"
@@ -160,6 +162,10 @@ logicalrep_relmap_free_entry(LogicalRepRelMapEntry *entry)
*
* Called when new relation mapping is sent by the publisher to update
* our expected view of incoming data from said publisher.
+ *
+ * Note that we do not check the user-defined constraints here. PostgreSQL has
+ * already assumed that CHECK constraints' conditions are immutable and here
+ * follows the rule.
*/
void
logicalrep_relmap_update(LogicalRepRelation *remoterel)
@@ -209,6 +215,8 @@ logicalrep_relmap_update(LogicalRepRelation *remoterel)
(remoterel->relkind == 0) ? RELKIND_RELATION : remoterel->relkind;
entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
+
+ entry->parallel_safe = LOGICALREP_PARALLEL_UNKNOWN;
MemoryContextSwitchTo(oldctx);
}
@@ -354,27 +362,79 @@ logicalrep_rel_mark_updatable(LogicalRepRelMapEntry *entry)
}
/*
- * Open the local relation associated with the remote one.
+ * Check all local triggers for the relation to see the parallelizability.
*
- * Rebuilds the Relcache mapping if it was invalidated by local DDL.
+ * We regard relations as applicable in parallel if all triggers are immutable.
+ * Result is directly set to LogicalRepRelMapEntry::parallel_safe.
*/
-LogicalRepRelMapEntry *
-logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
+static void
+check_defined_triggers(LogicalRepRelMapEntry *entry)
+{
+ TriggerDesc *trigdesc = entry->localrel->trigdesc;
+
+ /* Quick exit if triffer is not defined */
+ if (trigdesc == NULL)
+ {
+ entry->parallel_safe = LOGICALREP_PARALLEL_SAFE;
+ return;
+ }
+
+ /* Seek triggers one by one to see the volatility */
+ for (int i = 0; i < trigdesc->numtriggers; i++)
+ {
+ Trigger *trigger = &trigdesc->triggers[i];
+
+ Assert(OidIsValid(trigger->tgfoid));
+
+ /* 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)
+ {
+ entry->parallel_safe = LOGICALREP_PARALLEL_RESTRICTED;
+ return;
+ }
+ }
+
+ /* All triggers are immutable, set as parallel safe */
+ entry->parallel_safe = LOGICALREP_PARALLEL_SAFE;
+}
+
+/*
+ * Actual workhorse for logicalrep_rel_open().
+ *
+ * Caller must specify *either* entry or key. If the entry is specified, its
+ * attributes are filled and returned. The logical relation is kept opening.
+ * If the key is given, the corresponding entry is first searched in the hash
+ * table and processed as in the above case. At the end, logical replication is
+ * closed.
+ */
+void
+logicalrep_rel_load(LogicalRepRelMapEntry *entry, LogicalRepRelId remoteid,
+ LOCKMODE lockmode)
{
- LogicalRepRelMapEntry *entry;
- bool found;
LogicalRepRelation *remoterel;
- if (LogicalRepRelMap == NULL)
- logicalrep_relmap_init();
+ Assert((entry && !remoteid) || (!entry && remoteid));
- /* Search for existing entry. */
- entry = hash_search(LogicalRepRelMap, &remoteid,
- HASH_FIND, &found);
+ if (!entry)
+ {
+ bool found;
- if (!found)
- elog(ERROR, "no relation map entry for remote relation ID %u",
- remoteid);
+ if (LogicalRepRelMap == NULL)
+ logicalrep_relmap_init();
+
+ /* Search for existing entry. */
+ entry = hash_search(LogicalRepRelMap, &remoteid,
+ HASH_FIND, &found);
+
+ if (!found)
+ elog(ERROR, "no relation map entry for remote relation ID %u",
+ remoteid);
+ }
remoterel = &entry->remoterel;
@@ -500,6 +560,13 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
entry->localindexoid = FindLogicalRepLocalIndex(entry->localrel, remoterel,
entry->attrmap);
+ /*
+ * Leader must also collect all local unique indexes for dependency
+ * tracking.
+ */
+ if (am_leader_apply_worker())
+ check_defined_triggers(entry);
+
entry->localrelvalid = true;
}
@@ -508,6 +575,34 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
entry->localreloid,
&entry->statelsn);
+ if (remoteid)
+ logicalrep_rel_close(entry, lockmode);
+}
+
+/*
+ * Open the local relation associated with the remote one.
+ *
+ * Rebuilds the Relcache mapping if it was invalidated by local DDL.
+ */
+LogicalRepRelMapEntry *
+logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
+{
+ LogicalRepRelMapEntry *entry;
+ bool found;
+
+ if (LogicalRepRelMap == NULL)
+ logicalrep_relmap_init();
+
+ /* Search for existing entry. */
+ entry = hash_search(LogicalRepRelMap, &remoteid,
+ HASH_FIND, &found);
+
+ if (!found)
+ elog(ERROR, "no relation map entry for remote relation ID %u",
+ remoteid);
+
+ logicalrep_rel_load(entry, 0, lockmode);
+
return entry;
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 7f9d99901c0..c12b83a05d5 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1070,6 +1070,59 @@ check_dependency_on_rel(LogicalRepRelId relid, TransactionId new_depended_xid,
relentry->last_depended_xid = new_depended_xid;
}
+/*
+ * Check the parallelizability of applying changes for the relation.
+ * Append the lastly dispatched transaction in in 'depends_on_xids' if the
+ * relation is parallel unsafe.
+ */
+static void
+check_dependency_for_parallel_safety(LogicalRepRelId relid,
+ TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ LogicalRepRelMapEntry *relentry;
+
+ /* Quick exit if no transactions have been dispatched */
+ if (!TransactionIdIsValid(last_remote_xid))
+ return;
+
+ relentry = logicalrep_get_relentry(relid);
+
+ /*
+ * Gather information for local triggres if not yet. We require to be in a
+ * transaction state because system catalogs are read.
+ */
+ if (relentry->parallel_safe == LOGICALREP_PARALLEL_UNKNOWN)
+ {
+ bool needs_start = !IsTransactionOrTransactionBlock();
+
+ if (needs_start)
+ StartTransactionCommand();
+
+ logicalrep_rel_load(NULL, relid, AccessShareLock);
+
+ /*
+ * Close the transaction if we start here. We must not abort because
+ * it would release all session-level locks, such as the stream lock,
+ * and break the deadlock detection mechanism between LA and PA. The
+ * outcome is the same regardless of the end status, since the
+ * transaction did not modify any tuples.
+ */
+ if (needs_start)
+ CommitTransactionCommand();
+
+ Assert(relentry->parallel_safe != LOGICALREP_PARALLEL_UNKNOWN);
+ }
+
+ /* Do nothing for parallel safe relations */
+ if (relentry->parallel_safe == LOGICALREP_PARALLEL_SAFE)
+ return;
+
+ *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
+ &last_remote_xid,
+ new_depended_xid);
+}
+
/*
* Check dependencies related to the current change by determining if the
* modification impacts the same row or table as another ongoing transaction. If
@@ -1128,6 +1181,8 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_replica_identity(relid, &newtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_for_parallel_safety(relid, new_depended_xid,
+ &depends_on_xids);
break;
case LOGICAL_REP_MSG_UPDATE:
@@ -1135,13 +1190,19 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
&newtup);
if (has_oldtup)
+ {
check_dependency_on_replica_identity(relid, &oldtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_for_parallel_safety(relid, new_depended_xid,
+ &depends_on_xids);
+ }
check_dependency_on_replica_identity(relid, &newtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_for_parallel_safety(relid, new_depended_xid,
+ &depends_on_xids);
break;
case LOGICAL_REP_MSG_DELETE:
@@ -1149,6 +1210,8 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_replica_identity(relid, &oldtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_for_parallel_safety(relid, new_depended_xid,
+ &depends_on_xids);
break;
case LOGICAL_REP_MSG_TRUNCATE:
@@ -1161,8 +1224,13 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
* modified the same table.
*/
foreach_int(truncated_relid, remote_relids)
+ {
check_dependency_on_rel(truncated_relid, new_depended_xid,
&depends_on_xids);
+ check_dependency_for_parallel_safety(truncated_relid,
+ new_depended_xid,
+ &depends_on_xids);
+ }
break;
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 27e365f84c2..916d21c2dbd 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -39,6 +39,20 @@ typedef struct LogicalRepRelMapEntry
XLogRecPtr statelsn;
TransactionId last_depended_xid;
+
+ /*
+ * Whether the relation can be applied in parallel or not. It is
+ * distinglish whether defined triggers are the immutable or not.
+ *
+ * Theoretically, we can determine the parallelizability for each type of
+ * replication messages, INSERT/UPDATE/DELETE/TRUNCATE. But it is not done
+ * yet to reduce the number of attributes.
+ *
+ * Note that we do not check the user-defined constraints here. PostgreSQL
+ * has already assumed that CHECK constraints' conditions are immutable and
+ * here follows the rule.
+ */
+ char parallel_safe;
} LogicalRepRelMapEntry;
extern void logicalrep_relmap_update(LogicalRepRelation *remoterel);
@@ -46,6 +60,8 @@ extern void logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel);
extern LogicalRepRelMapEntry *logicalrep_rel_open(LogicalRepRelId remoteid,
LOCKMODE lockmode);
+extern void logicalrep_rel_load(LogicalRepRelMapEntry *entry,
+ LogicalRepRelId remoteid, LOCKMODE lockmode);
extern LogicalRepRelMapEntry *logicalrep_partition_open(LogicalRepRelMapEntry *root,
Relation partrel, AttrMap *map);
extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel,
@@ -56,4 +72,8 @@ extern int logicalrep_get_num_rels(void);
extern void logicalrep_write_all_rels(StringInfo out);
extern LogicalRepRelMapEntry *logicalrep_get_relentry(LogicalRepRelId remoteid);
+#define LOGICALREP_PARALLEL_SAFE 's'
+#define LOGICALREP_PARALLEL_RESTRICTED 'r'
+#define LOGICALREP_PARALLEL_UNKNOWN 'u'
+
#endif /* LOGICALRELATION_H */
--
2.47.3
[application/octet-stream] v8-0008-Support-dependency-tracking-via-local-unique-inde.patch (23.8K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/9-v8-0008-Support-dependency-tracking-via-local-unique-inde.patch)
download | inline diff:
From 9061a5fe6d79ef729698485c82245ee2e7f5ae0f Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Thu, 11 Dec 2025 22:21:47 +0900
Subject: [PATCH v8 8/9] Support dependency tracking via local unique indexes
Currently, logical replication's parallel apply mechanism tracks dependencies
primarily based on the REPLICA IDENTITY defined on the publisher table.
However, local subscriber tables might have additional unique indexes that
could effectively serve as dependency keys, even if they don't correspond to
the publisher's REPLICA IDENTITY. Failing to track these additional unique
keys can lead to incorrect data and/or deadlocks during parallel application.
This patch extends the parallel apply's dependency tracking to consider
local unique indexes on the subscriber table. This is achieved by extending
the existing Replica Identity hash table to also store dependency information
based on these local unique indexes.
The LogicalRepRelMapEntry structure is extended to store details about these
local unique indexes. This information is collected and cached when
dependency checking is first performed for a remote transaction on a given
relation. This collection process requires to be in a transaction to access
system catalog information.
---
src/backend/replication/logical/relation.c | 161 ++++++++++-
src/backend/replication/logical/worker.c | 270 ++++++++++++++----
src/backend/storage/lmgr/deadlock.c | 1 -
src/include/replication/logicalrelation.h | 14 +
src/test/subscription/t/050_parallel_apply.pl | 57 ++++
src/tools/pgindent/typedefs.list | 2 +
6 files changed, 446 insertions(+), 59 deletions(-)
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index eeb85f8cc5d..6665f5c0cc9 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -128,6 +128,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(LogicalRepSubscriberIdx, 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.
*/
@@ -155,6 +170,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);
}
/*
@@ -361,6 +379,126 @@ logicalrep_rel_mark_updatable(LogicalRepRelMapEntry *entry)
}
}
+/*
+ * Collect all local unique indexes that can be used for dependency tracking.
+ */
+static void
+collect_local_indexes(LogicalRepRelMapEntry *entry)
+{
+ List *idxlist;
+
+ if (entry->local_unique_indexes != NIL)
+ free_local_unique_indexes(entry);
+
+ entry->local_unique_indexes_collected = true;
+
+ idxlist = RelationGetIndexList(entry->localrel);
+
+ /* Quick exit if there are no indexes */
+ if (idxlist == NIL)
+ return;
+
+ /* Iterate indexes to list all usable indexes */
+ foreach_oid(idxoid, idxlist)
+ {
+ Relation idxrel;
+ int indnkeys;
+ AttrMap *attrmap;
+ Bitmapset *indexkeys = NULL;
+ bool suitable = true;
+
+ idxrel = index_open(idxoid, AccessShareLock);
+
+ /*
+ * Check whether the index can be used for the dependency tracking.
+ *
+ * For simplification, the same condition as REPLICA IDENTITY FULL,
+ * plus it must be a unique index.
+ */
+ if (!(idxrel->rd_index->indisunique &&
+ IsIndexUsableForReplicaIdentityFull(idxrel, entry->attrmap)))
+ {
+ index_close(idxrel, AccessShareLock);
+ continue;
+ }
+
+ indnkeys = idxrel->rd_index->indnkeyatts;
+ 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;
+
+ /*
+ * XXX: Mark a relation as parallel-unsafe if it has expression
+ * indexes because we cannot compute the hash value for the
+ * dependency tracking. For safety, transactions that modify such
+ * tables can wait for applications till the lastly dispatched
+ * transaction is committed.
+ */
+ if (!AttributeNumberIsValid(localcol))
+ {
+ entry->parallel_safe = LOGICALREP_PARALLEL_RESTRICTED;
+ suitable = false;
+ break;
+ }
+
+ remotecol = attrmap->attnums[AttrNumberGetAttrOffset(localcol)];
+
+ /*
+ * Skip if the column does not exist on publisher node. In this
+ * case the replicated tuples always have NULL or default value.
+ */
+ if (remotecol < 0)
+ {
+ suitable = false;
+ break;
+ }
+
+ /* Checks are passed, remember the attribute */
+ indexkeys = bms_add_member(indexkeys, remotecol);
+ }
+
+ index_close(idxrel, AccessShareLock);
+
+ /*
+ * Skip using the index if it is not suitable. This can happen if 1)
+ * one of the columns does not exist on the publisher side, or 2)
+ * there is an expression column.
+ */
+ if (!suitable)
+ {
+ if (indexkeys)
+ bms_free(indexkeys);
+
+ continue;
+ }
+
+ /* This index is usable, store on memory */
+ if (indexkeys)
+ {
+ MemoryContext oldctx;
+ LogicalRepSubscriberIdx *idxinfo;
+
+ oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
+ idxinfo = palloc(sizeof(LogicalRepSubscriberIdx));
+ idxinfo->indexoid = idxoid;
+ idxinfo->indexkeys = bms_copy(indexkeys);
+ entry->local_unique_indexes =
+ lappend(entry->local_unique_indexes, idxinfo);
+
+ pfree(indexkeys);
+ MemoryContextSwitchTo(oldctx);
+ }
+ }
+
+ list_free(idxlist);
+}
+
/*
* Check all local triggers for the relation to see the parallelizability.
*
@@ -370,7 +508,16 @@ logicalrep_rel_mark_updatable(LogicalRepRelMapEntry *entry)
static void
check_defined_triggers(LogicalRepRelMapEntry *entry)
{
- TriggerDesc *trigdesc = entry->localrel->trigdesc;
+ TriggerDesc *trigdesc;
+
+ /*
+ * Skip if the parallelizability has already been checked. Possilble if
+ * the relation has expression indexes.
+ */
+ if (entry->parallel_safe != LOGICALREP_PARALLEL_UNKNOWN)
+ return;
+
+ trigdesc = entry->localrel->trigdesc;
/* Quick exit if triffer is not defined */
if (trigdesc == NULL)
@@ -411,7 +558,7 @@ check_defined_triggers(LogicalRepRelMapEntry *entry)
* If the key is given, the corresponding entry is first searched in the hash
* table and processed as in the above case. At the end, logical replication is
* closed.
- */
+ */
void
logicalrep_rel_load(LogicalRepRelMapEntry *entry, LogicalRepRelId remoteid,
LOCKMODE lockmode)
@@ -565,7 +712,11 @@ logicalrep_rel_load(LogicalRepRelMapEntry *entry, LogicalRepRelId remoteid,
* tracking.
*/
if (am_leader_apply_worker())
+ {
+ entry->parallel_safe = LOGICALREP_PARALLEL_UNKNOWN;
+ collect_local_indexes(entry);
check_defined_triggers(entry);
+ }
entry->localrelvalid = true;
}
@@ -867,6 +1018,12 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
entry->localindexoid = FindLogicalRepLocalIndex(partrel, remoterel,
entry->attrmap);
+ /*
+ * TODO: Parallel apply does not support the parallel apply for now. Just
+ * mark local indexes are collected.
+ */
+ entry->local_unique_indexes_collected = true;
+
entry->localrelvalid = true;
return entry;
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index c12b83a05d5..6851596625e 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -548,9 +548,19 @@ 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;
+
typedef struct ReplicaIdentityKey
{
Oid relid;
+ LogicalRepKeyKind kind;
LogicalRepTupleData *data;
} ReplicaIdentityKey;
@@ -710,7 +720,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++)
@@ -718,6 +729,9 @@ hash_replica_identity_compare(ReplicaIdentityKey *a, ReplicaIdentityKey *b)
if (a->data->colstatus[i] != b->data->colstatus[i])
return false;
+ if (a->data->colstatus[i] == LOGICALREP_COLUMN_NULL)
+ continue;
+
if (a->data->colvalues[i].len != b->data->colvalues[i].len)
return false;
@@ -847,6 +861,93 @@ check_and_append_xid_dependency(List *depends_on_xids,
return lappend_xid(depends_on_xids, *depends_on_xid);
}
+/*
+ * Common function for checking dependency by using the key. Used by both
+ * check_dependency_on_replica_identity and check_dependency_on_local_key.
+ */
+static void
+check_dependency_by_key(ReplicaIdentityKey *key, TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ ReplicaIdentityEntry *rientry;
+ bool found = false;
+ MemoryContext oldctx;
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ if (TransactionIdIsValid(new_depended_xid))
+ {
+ rientry = replica_identity_insert(replica_identity_table, key,
+ &found);
+
+ /*
+ * Release the key built to search the entry, if the entry already
+ * exists. Otherwise, initialize the remote_xid.
+ */
+ if (found)
+ {
+ elog(DEBUG1,
+ key->kind == LOGICALREP_KEY_REPLICA_IDENTITY ?
+ "found conflicting replica identity change from %u" :
+ "found conflicting local unique change from %u",
+ rientry->remote_xid);
+
+ free_replica_identity_key(key);
+ }
+ else
+ rientry->remote_xid = InvalidTransactionId;
+ }
+ else
+ {
+ rientry = replica_identity_lookup(replica_identity_table, key);
+ free_replica_identity_key(key);
+ }
+
+ MemoryContextSwitchTo(oldctx);
+
+ /* Return if no entry found */
+ if (!rientry)
+ return;
+
+ Assert(!found || TransactionIdIsValid(rientry->remote_xid));
+
+ *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
+ &rientry->remote_xid,
+ new_depended_xid);
+
+ /*
+ * Remove the entry if it is registered for the streamed transactions. We
+ * do not have to register an entry for them; The leader worker always
+ * waits until the parallel worker finishes handling streamed
+ * transactions, thus no need to consider the possiblity that upcoming
+ * parallel workers would go ahead.
+ */
+ if (TransactionIdIsValid(stream_xid) && !found)
+ {
+ free_replica_identity_key(rientry->keydata);
+ replica_identity_delete_item(replica_identity_table, rientry);
+ }
+
+ /*
+ * Update the new depended xid into the entry if valid, 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, so no need to be depended on.
+ */
+ else if (TransactionIdIsValid(new_depended_xid))
+ rientry->remote_xid = new_depended_xid;
+
+ /*
+ * Remove the entry if the transaction has been committed and no new
+ * dependency needs to be added.
+ */
+ else if (!TransactionIdIsValid(rientry->remote_xid))
+ {
+ free_replica_identity_key(rientry->keydata);
+ replica_identity_delete_item(replica_identity_table, rientry);
+ }
+}
+
/*
* Check for dependencies on preceding transactions that modify the same key.
* Returns the dependent transactions in 'depends_on_xids' and records the
@@ -861,10 +962,8 @@ check_dependency_on_replica_identity(Oid relid,
LogicalRepRelMapEntry *relentry;
LogicalRepTupleData *ridata;
ReplicaIdentityKey *rikey;
- ReplicaIdentityEntry *rientry;
MemoryContext oldctx;
int n_ri;
- bool found = false;
Assert(depends_on_xids);
@@ -930,75 +1029,122 @@ check_dependency_on_replica_identity(Oid relid,
rikey = palloc0_object(ReplicaIdentityKey);
rikey->relid = relid;
+ rikey->kind = LOGICALREP_KEY_REPLICA_IDENTITY;
rikey->data = ridata;
- if (TransactionIdIsValid(new_depended_xid))
+ MemoryContextSwitchTo(oldctx);
+
+ check_dependency_by_key(rikey, new_depended_xid, depends_on_xids);
+}
+
+/*
+ * Mostly same as check_dependency_on_replica_identity() but for local unique
+ * indexes.
+ */
+static void
+check_dependency_on_local_key(Oid relid,
+ LogicalRepTupleData *original_data,
+ TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ LogicalRepRelMapEntry *relentry;
+ LogicalRepTupleData *ridata;
+ ReplicaIdentityKey *rikey;
+ MemoryContext oldctx;
+
+ Assert(depends_on_xids);
+
+ /* Search for existing entry */
+ relentry = logicalrep_get_relentry(relid);
+
+ Assert(relentry);
+
+ /*
+ * Gather information for local indexes if not yet. We require to be in a
+ * transaction state because system catalogs are read.
+ */
+ if (!relentry->local_unique_indexes_collected)
{
- rientry = replica_identity_insert(replica_identity_table, rikey,
- &found);
+ bool needs_start = !IsTransactionOrTransactionBlock();
+
+ if (needs_start)
+ StartTransactionCommand();
+
+ logicalrep_rel_load(NULL, relid, AccessShareLock);
/*
- * Release the key built to search the entry, if the entry already
- * exists. Otherwise, initialize the remote_xid.
+ * Close the transaction if we start here. We must not abort because
+ * it would release all session-level locks, such as the stream lock,
+ * and break the deadlock detection mechanism between LA and PA. The
+ * outcome is the same regardless of the end status, since the
+ * transaction did not modify any tuples.
*/
- if (found)
- {
- elog(DEBUG1, "found conflicting replica identity change from %u",
- rientry->remote_xid);
+ if (needs_start)
+ CommitTransactionCommand();
- free_replica_identity_key(rikey);
- }
- else
- rientry->remote_xid = InvalidTransactionId;
+ Assert(relentry->local_unique_indexes_collected);
}
- else
+
+ foreach_ptr(LogicalRepSubscriberIdx, idxinfo, relentry->local_unique_indexes)
{
- rientry = replica_identity_lookup(replica_identity_table, rikey);
- free_replica_identity_key(rikey);
- }
+ int columns = bms_num_members(idxinfo->indexkeys);
+ bool suitable = true;
- MemoryContextSwitchTo(oldctx);
+ Assert(columns);
- /* Return if no entry found */
- if (!rientry)
- return;
+ for (int i = 0; i < original_data->ncols; i++)
+ {
+ if (!bms_is_member(i, idxinfo->indexkeys))
+ continue;
- Assert(!found || TransactionIdIsValid(rientry->remote_xid));
+ /*
+ * Skip if the column is not changed.
+ *
+ * XXX: NULL is allowed.
+ */
+ if (original_data->colstatus[i] == LOGICALREP_COLUMN_UNCHANGED)
+ {
+ suitable = false;
+ break;
+ }
+ }
- *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
- &rientry->remote_xid,
- new_depended_xid);
+ if (!suitable)
+ continue;
- /*
- * Remove the entry if it is registered for the streamed transactions. We
- * do not have to register an entry for them; The leader worker always
- * waits until the parallel worker finishes handling streamed
- * transactions, thus no need to consider the possiblity that upcoming
- * parallel workers would go ahead.
- */
- if (TransactionIdIsValid(stream_xid) && !found)
- {
- free_replica_identity_key(rientry->keydata);
- replica_identity_delete_item(replica_identity_table, rientry);
- }
+ oldctx = MemoryContextSwitchTo(ApplyContext);
- /*
- * Update the new depended xid into the entry if valid, 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, so no need to be depended on.
- */
- else if (TransactionIdIsValid(new_depended_xid))
- rientry->remote_xid = new_depended_xid;
+ /* Allocate space for replica identity values */
+ ridata = palloc0_object(LogicalRepTupleData);
+ ridata->colvalues = palloc0_array(StringInfoData, columns);
+ ridata->colstatus = palloc0_array(char, columns);
+ ridata->ncols = columns;
- /*
- * Remove the entry if the transaction has been committed and no new
- * dependency needs to be added.
- */
- else if (!TransactionIdIsValid(rientry->remote_xid))
- {
- free_replica_identity_key(rientry->keydata);
- replica_identity_delete_item(replica_identity_table, rientry);
+ for (int i_original = 0, i_key = 0; i_original < original_data->ncols; i_original++)
+ {
+ if (!bms_is_member(i_original, idxinfo->indexkeys))
+ continue;
+
+ if (original_data->colstatus[i_original] != LOGICALREP_COLUMN_NULL)
+ {
+ StringInfo original_colvalue = &original_data->colvalues[i_original];
+
+ initStringInfoExt(&ridata->colvalues[i_key], original_colvalue->len + 1);
+ appendStringInfoString(&ridata->colvalues[i_key], original_colvalue->data);
+ }
+
+ ridata->colstatus[i_key] = original_data->colstatus[i_original];
+ i_key++;
+ }
+
+ rikey = palloc0_object(ReplicaIdentityKey);
+ rikey->relid = relid;
+ rikey->kind = LOGICALREP_KEY_LOCAL_UNIQUE;
+ rikey->data = ridata;
+
+ MemoryContextSwitchTo(oldctx);
+
+ check_dependency_by_key(rikey, new_depended_xid, depends_on_xids);
}
}
@@ -1181,6 +1327,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_replica_identity(relid, &newtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_local_key(relid, &newtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
break;
@@ -1194,6 +1343,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_replica_identity(relid, &oldtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_local_key(relid, &oldtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
}
@@ -1201,6 +1353,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_replica_identity(relid, &newtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_local_key(relid, &newtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
break;
@@ -1210,6 +1365,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_replica_identity(relid, &oldtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_local_key(relid, &oldtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
break;
diff --git a/src/backend/storage/lmgr/deadlock.c b/src/backend/storage/lmgr/deadlock.c
index 8334a887618..b37d872bfd0 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 916d21c2dbd..9152effb2cd 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -16,6 +16,12 @@
#include "catalog/index.h"
#include "replication/logicalproto.h"
+typedef struct LogicalRepSubscriberIdx
+{
+ Oid indexoid; /* OID of the local key */
+ Bitmapset *indexkeys; /* Bitmap of key columns *on remote* */
+} LogicalRepSubscriberIdx;
+
typedef struct LogicalRepRelMapEntry
{
LogicalRepRelation remoterel; /* key is remoterel.remoteid */
@@ -40,6 +46,10 @@ typedef struct LogicalRepRelMapEntry
TransactionId last_depended_xid;
+ /* Local unique indexes. Used for dependency tracking */
+ List *local_unique_indexes;
+ bool local_unique_indexes_collected;
+
/*
* Whether the relation can be applied in parallel or not. It is
* distinglish whether defined triggers are the immutable or not.
@@ -51,6 +61,10 @@ typedef struct LogicalRepRelMapEntry
* Note that we do not check the user-defined constraints here. PostgreSQL
* has already assumed that CHECK constraints' conditions are immutable and
* here follows the rule.
+ *
+ * XXX: Additonally, this can be false if the relation has expression
+ * indexes. Because we cannot compute the hash value for the dependency
+ * tracking.
*/
char parallel_safe;
} LogicalRepRelMapEntry;
diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl
index 9254b85d350..9cb395fa4d8 100644
--- a/src/test/subscription/t/050_parallel_apply.pl
+++ b/src/test/subscription/t/050_parallel_apply.pl
@@ -234,4 +234,61 @@ $node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset
$h->query_safe("COMMIT;");
+# Ensure subscriber-local indexes are also used for the dependency tracking
+
+# 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 INDEX ON regress_tab (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',
+ "INSERT INTO regress_tab VALUES (1, 'would conflict');");
+
+# 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 on publisher again. This transaction is would wait because all
+# parallel workers wait till the previously launched worker commits.
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (2, 'would not conflict');");
+
+# Verify the parallel worker waits for the transaction
+$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]+)/;
+
+# Insert a conflicting tuple on publisher. Leader worker would detect the conflict
+# and wait for the transaction to commit.
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (3, 'would conflict');");
+
+# Verify the parallel worker waits for the same transaction
+$node_subscriber->wait_for_log(qr/wait for depended xid $xid/, $offset);
+
+# 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);
+
+# Cleanup
+$node_subscriber->safe_psql('postgres', "DROP INDEX regress_tab_value_idx;");
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE regress_tab;");
+$node_publisher->wait_for_catchup('regress_sub');
+
done_testing();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 811a23ac03c..18eef00e735 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1642,6 +1642,7 @@ LogicalRepBeginData
LogicalRepCommitData
LogicalRepCommitPreparedTxnData
LogicalRepCtxStruct
+LogicalRepKeyKind
LogicalRepMsgType
LogicalRepPartMapEntry
LogicalRepPreparedTxnData
@@ -1651,6 +1652,7 @@ LogicalRepRelation
LogicalRepRollbackPreparedTxnData
LogicalRepSequenceInfo
LogicalRepStreamAbortData
+LogicalRepSubscriberIdx
LogicalRepTupleData
LogicalRepTyp
LogicalRepWorker
--
2.47.3
[application/octet-stream] v8-0009-Support-dependency-tracking-via-foreign-keys.patch (16.7K, ../TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com/10-v8-0009-Support-dependency-tracking-via-foreign-keys.patch)
download | inline diff:
From e74a96815b0a736bf301f6f1443a7c8d04448724 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Fri, 23 Jan 2026 15:08:15 +0900
Subject: [PATCH v8 9/9] Support dependency tracking via foreign keys
Logical replication's parallel apply mechanism did not account for foreign-key
dependencies. This can cause a failure because the reference column may be
committed before the referenced column.
This patch extends the parallel apply's dependency tracking to account for
foreign keys in the subscriber table. One assumption here is that referenced
columns can be registered to the identity hash table.
The leader apply worker checks to determine whether values in referencing
columns have already been registered or not, and regards that there is a
dependency if found.
---
src/backend/replication/logical/relation.c | 185 ++++++++++++++++++
src/backend/replication/logical/worker.c | 142 ++++++++++++++
src/include/replication/logicalrelation.h | 11 ++
src/test/subscription/t/050_parallel_apply.pl | 50 ++++-
4 files changed, 385 insertions(+), 3 deletions(-)
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index 6665f5c0cc9..83602dcd4cf 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_constraint.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_subscription_rel.h"
#include "commands/trigger.h"
@@ -143,6 +144,21 @@ free_local_unique_indexes(LogicalRepRelMapEntry *entry)
entry->local_unique_indexes = NIL;
}
+/*
+ * Release foreign key list
+ */
+static void
+free_local_fkeys(LogicalRepRelMapEntry *entry)
+{
+ Assert(am_leader_apply_worker());
+
+ foreach_ptr(LogicalRepSubscriberFK, fkinfo, entry->local_fkeys)
+ bms_free(fkinfo->conkeys);
+
+ list_free_deep(entry->local_fkeys);
+ entry->local_fkeys = NIL;
+}
+
/*
* Free the entry of a relation map cache.
*/
@@ -173,6 +189,9 @@ logicalrep_relmap_free_entry(LogicalRepRelMapEntry *entry)
if (entry->local_unique_indexes != NIL)
free_local_unique_indexes(entry);
+
+ if (entry->local_fkeys != NIL)
+ free_local_fkeys(entry);
}
/*
@@ -499,6 +518,171 @@ collect_local_indexes(LogicalRepRelMapEntry *entry)
list_free(idxlist);
}
+/*
+ * Search a relmap entry by local relation OID.
+ */
+static LogicalRepRelMapEntry *
+logicalrep_get_relentry_by_local_oid(Oid localreloid)
+{
+ HASH_SEQ_STATUS status;
+ LogicalRepRelMapEntry *entry = NULL;
+
+ if (LogicalRepRelMap == NULL)
+ return NULL;
+
+ /*
+ * Each entry must be checked individually. Because the key of
+ * LogicalRepRelMap is the "remote" relid but we only have the local one.
+ *
+ * Note: This iteration ignores relations that have not yet been
+ * registered in LogicalRepRelMap. It is OK because such relations have
+ * not been modified yet since the subscriber started receiving changes.
+ */
+ hash_seq_init(&status, LogicalRepRelMap);
+ while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
+ {
+ if (entry->localreloid == localreloid)
+ {
+ hash_seq_term(&status);
+ return entry;
+ }
+ }
+
+ return NULL;
+}
+
+/*
+ * Return true if the FK constraint is always deferred.
+ */
+static bool
+foreign_key_is_always_deferred(Oid conoid)
+{
+ HeapTuple tup;
+ Form_pg_constraint con;
+ bool deferred;
+
+ tup = SearchSysCache1(CONSTROID, ObjectIdGetDatum(conoid));
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "cache lookup failed for foreign key %u", conoid);
+
+ con = (Form_pg_constraint) GETSTRUCT(tup);
+ deferred = con->condeferrable && con->condeferred;
+ ReleaseSysCache(tup);
+
+ return deferred;
+}
+
+/*
+ * Collect all local foreign keys that can be used for dependency tracking.
+ */
+static void
+collect_local_fkeys(LogicalRepRelMapEntry *entry)
+{
+ List *fkeys;
+
+ if (entry->local_fkeys != NIL)
+ free_local_fkeys(entry);
+
+ entry->local_fkeys_collected = true;
+
+ /*
+ * Get the list of foreign keys for the relation.
+ *
+ * XXX: Apart from RelationGetIndexList(), the returned list is a part of
+ * relcache and must be copied before doing anything. See comments atop
+ * RelationGetFKeyList().
+ */
+ fkeys = copyObject(RelationGetFKeyList(entry->localrel));
+
+ /* Quick exit if there are no foreign keys */
+ if (fkeys == NIL)
+ return;
+
+ foreach_ptr(ForeignKeyCacheInfo, fk, fkeys)
+ {
+ LogicalRepRelMapEntry *refentry;
+ Bitmapset *fkkeys = NULL;
+ bool suitable = true;
+
+ /*
+ * Skip NOT ENFORCED constraints because they won't be checked at any
+ * times.
+ */
+ if (!fk->conenforced)
+ continue;
+
+ /*
+ * Skip if the foreign key constraint is always deferred. The commit
+ * ordering is always preserved thus the constraint can be checked in
+ * correct order.
+ */
+ if (foreign_key_is_always_deferred(fk->conoid))
+ continue;
+
+ /* Find the referenced relation by the local OID */
+ refentry = logicalrep_get_relentry_by_local_oid(fk->confrelid);
+
+ /*
+ * Skip if the referenced relation is not the target of this
+ * subscription.
+ */
+ if (!refentry)
+ continue;
+
+ /* Seek each attributes and add to a Bitmap */
+ for (int i = 0; i < fk->nkeys; i++)
+ {
+ AttrNumber localcol = fk->conkey[i];
+ int remotecol =
+ entry->attrmap->attnums[AttrNumberGetAttrOffset(localcol)];
+
+ /* Skip if the column does not exist on publisher node */
+ if (remotecol < 0)
+ {
+ suitable = false;
+ break;
+ }
+
+ /*
+ * XXX: What if the FK column is specified with different order?
+ * E.g., there is a table (a, b) and other table refers like
+ * REFERENCES (b, a). Will it work correctly?
+ */
+ fkkeys = bms_add_member(fkkeys, remotecol);
+ }
+
+ /*
+ * One of the columns does not exist on the publisher side, skip such
+ * a constraint.
+ */
+ if (!suitable)
+ {
+ if (fkkeys)
+ bms_free(fkkeys);
+
+ continue;
+ }
+
+ if (fkkeys)
+ {
+ MemoryContext oldctx;
+ LogicalRepSubscriberFK *fkinfo;
+
+ oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
+ fkinfo = palloc(sizeof(LogicalRepSubscriberFK));
+ fkinfo->conoid = fk->conoid;
+ fkinfo->ref_remoteid = refentry->remoterel.remoteid;
+ fkinfo->conkeys = bms_copy(fkkeys);
+
+ bms_free(fkkeys);
+ entry->local_fkeys = lappend(entry->local_fkeys, fkinfo);
+ MemoryContextSwitchTo(oldctx);
+ }
+ }
+
+ list_free_deep(fkeys);
+}
+
/*
* Check all local triggers for the relation to see the parallelizability.
*
@@ -715,6 +899,7 @@ logicalrep_rel_load(LogicalRepRelMapEntry *entry, LogicalRepRelId remoteid,
{
entry->parallel_safe = LOGICALREP_PARALLEL_UNKNOWN;
collect_local_indexes(entry);
+ collect_local_fkeys(entry);
check_defined_triggers(entry);
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 6851596625e..e36c6dd801b 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1148,6 +1148,136 @@ check_dependency_on_local_key(Oid relid,
}
}
+/*
+ * Lookup-only dependency check by key. Does not register/update entries.
+ */
+static void
+check_dependency_lookup_by_key(ReplicaIdentityKey *key,
+ TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ ReplicaIdentityEntry *rientry;
+ MemoryContext oldctx;
+
+ Assert(replica_identity_table);
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ rientry = replica_identity_lookup(replica_identity_table, key);
+ free_replica_identity_key(key);
+
+ MemoryContextSwitchTo(oldctx);
+
+ if (!rientry)
+ return;
+
+ /*
+ * Clean up committed entries while on it.
+ */
+ if (pa_transaction_committed(rientry->remote_xid))
+ {
+ free_replica_identity_key(rientry->keydata);
+ replica_identity_delete_item(replica_identity_table, rientry);
+ return;
+ }
+
+ *depends_on_xids = check_and_append_xid_dependency(*depends_on_xids,
+ &rientry->remote_xid,
+ new_depended_xid);
+}
+
+/*
+ * Check for dependencies on preceding transactions that modify the referenced
+ * unique key of foreign keys.
+ */
+static void
+check_dependency_on_foreign_key(Oid relid,
+ LogicalRepTupleData *original_data,
+ TransactionId new_depended_xid,
+ List **depends_on_xids)
+{
+ LogicalRepRelMapEntry *relentry;
+ LogicalRepTupleData *ridata;
+ ReplicaIdentityKey *rikey;
+ MemoryContext oldctx;
+
+ Assert(depends_on_xids);
+
+ relentry = logicalrep_get_relentry(relid);
+
+ Assert(relentry);
+
+ /*
+ * We have already checked the local unique indexes and also gathered the
+ * FK info at that time.
+ *
+ * XXX: Should we check and load again?
+ */
+ Assert(relentry->local_fkeys_collected);
+
+ foreach_ptr(LogicalRepSubscriberFK, fkinfo, relentry->local_fkeys)
+ {
+ int columns = bms_num_members(fkinfo->conkeys);
+ bool suitable = true;
+
+ Assert(columns);
+
+ for (int i = 0; i < original_data->ncols; i++)
+ {
+ if (!bms_is_member(i, fkinfo->conkeys))
+ continue;
+
+ /* Skip if the column is NULL or not changed */
+ if (original_data->colstatus[i] == LOGICALREP_COLUMN_NULL ||
+ original_data->colstatus[i] == LOGICALREP_COLUMN_UNCHANGED)
+ {
+ suitable = false;
+ break;
+ }
+ }
+
+ if (!suitable)
+ continue;
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ /* Allocate space for replica identity values */
+ ridata = palloc0_object(LogicalRepTupleData);
+ ridata->colvalues = palloc0_array(StringInfoData, columns);
+ ridata->colstatus = palloc0_array(char, columns);
+ ridata->ncols = columns;
+
+ for (int i_original = 0, i_key = 0; i_original < original_data->ncols; i_original++)
+ {
+ StringInfo original_colvalue;
+
+ if (!bms_is_member(i_original, fkinfo->conkeys))
+ continue;
+
+ original_colvalue = &original_data->colvalues[i_original];
+ initStringInfoExt(&ridata->colvalues[i_key], original_colvalue->len + 1);
+ appendStringInfoString(&ridata->colvalues[i_key], original_colvalue->data);
+
+ ridata->colstatus[i_key] = original_data->colstatus[i_original];
+ i_key++;
+ }
+
+ rikey = palloc0_object(ReplicaIdentityKey);
+ rikey->relid = fkinfo->ref_remoteid;
+
+ /*
+ * XXX: use LOGICALREP_KEY_LOCAL_UNIQUE to compare with local unique
+ * indexes.
+ */
+ rikey->kind = LOGICALREP_KEY_LOCAL_UNIQUE;
+ rikey->data = ridata;
+
+ MemoryContextSwitchTo(oldctx);
+
+ check_dependency_lookup_by_key(rikey, new_depended_xid, depends_on_xids);
+ }
+}
+
/*
* Check for preceding transactions that involve insert, delete, or update
* operations on the specified table, and return them in 'depends_on_xids'.
@@ -1330,6 +1460,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_local_key(relid, &newtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_foreign_key(relid, &newtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
break;
@@ -1346,6 +1479,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_local_key(relid, &oldtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_foreign_key(relid, &oldtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
}
@@ -1356,6 +1492,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_local_key(relid, &newtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_foreign_key(relid, &newtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
break;
@@ -1368,6 +1507,9 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
check_dependency_on_local_key(relid, &oldtup,
new_depended_xid,
&depends_on_xids);
+ check_dependency_on_foreign_key(relid, &oldtup,
+ new_depended_xid,
+ &depends_on_xids);
check_dependency_for_parallel_safety(relid, new_depended_xid,
&depends_on_xids);
break;
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 9152effb2cd..e915fb53246 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -22,6 +22,13 @@ typedef struct LogicalRepSubscriberIdx
Bitmapset *indexkeys; /* Bitmap of key columns *on remote* */
} LogicalRepSubscriberIdx;
+typedef struct LogicalRepSubscriberFK
+{
+ Oid conoid; /* OID of the FK constraint */
+ LogicalRepRelId ref_remoteid; /* referenced remote relid */
+ Bitmapset *conkeys; /* Bitmap of key columns *on remote* */
+} LogicalRepSubscriberFK;
+
typedef struct LogicalRepRelMapEntry
{
LogicalRepRelation remoterel; /* key is remoterel.remoteid */
@@ -50,6 +57,10 @@ typedef struct LogicalRepRelMapEntry
List *local_unique_indexes;
bool local_unique_indexes_collected;
+ /* Local foreign keys. Used for dependency tracking */
+ List *local_fkeys;
+ bool local_fkeys_collected;
+
/*
* Whether the relation can be applied in parallel or not. It is
* distinglish whether defined triggers are the immutable or not.
diff --git a/src/test/subscription/t/050_parallel_apply.pl b/src/test/subscription/t/050_parallel_apply.pl
index 9cb395fa4d8..54352a5f9b9 100644
--- a/src/test/subscription/t/050_parallel_apply.pl
+++ b/src/test/subscription/t/050_parallel_apply.pl
@@ -26,6 +26,8 @@ $node_publisher->safe_psql('postgres',
"CREATE TABLE regress_tab (id int PRIMARY KEY, value text);");
$node_publisher->safe_psql('postgres',
"INSERT INTO regress_tab VALUES (generate_series(1, 10), 'test');");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE regress_tab_fk (id int PRIMARY KEY, fk int);");
# Create a publication
$node_publisher->safe_psql('postgres',
@@ -56,6 +58,8 @@ my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
$node_subscriber->safe_psql('postgres',
"CREATE TABLE regress_tab (id int PRIMARY KEY, value text);");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE regress_tab_fk (id int PRIMARY KEY, fk int REFERENCES regress_tab (id));");
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION regress_sub CONNECTION '$publisher_connstr' PUBLICATION regress_pub;");
@@ -236,8 +240,8 @@ $h->query_safe("COMMIT;");
# Ensure subscriber-local indexes are also used for the dependency tracking
-# Truncate the data for upcoming tests
-$node_publisher->safe_psql('postgres', "TRUNCATE TABLE regress_tab;");
+# delete the data for upcoming tests
+$node_publisher->safe_psql('postgres', "DELETE FROM regress_tab;");
$node_publisher->wait_for_catchup('regress_sub');
# Define an unique index on subscriber
@@ -288,7 +292,47 @@ $node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset
# Cleanup
$node_subscriber->safe_psql('postgres', "DROP INDEX regress_tab_value_idx;");
-$node_publisher->safe_psql('postgres', "TRUNCATE TABLE regress_tab;");
+$node_publisher->safe_psql('postgres', "DELETE FROM regress_tab;");
$node_publisher->wait_for_catchup('regress_sub');
+# Ensure subscriber-local indexes are also used for the dependency tracking
+
+# Insert an initial tuple
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab VALUES (1, 'initial tuple');");
+$node_publisher->wait_for_catchup('regress_sub');
+
+# 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',
+ "INSERT INTO regress_tab VALUES (2, 'tmp');");
+
+# 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 on publisher again. This transaction waits for referring tuple
+# is committed.
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO regress_tab_fk VALUES (1, 1);");
+
+# Verify the parallel worker waits for the transaction
+$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]+)/;
+
+# 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');
+]);
+
+$node_subscriber->wait_for_log(qr/finish waiting for depended xid $xid/, $offset);
+
done_testing();
--
2.47.3
view thread (7+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Parallel Apply
In-Reply-To: <TY7PR01MB1455403538191EE4FA429106FF59FA@TY7PR01MB14554.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox