From 9c1bec1c056e86be64bb77cba7fe853971657d92 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Tue, 11 Aug 2026 14:20:19 +0800
Subject: [PATCH v23 07/12] Bound dependency tracking memory with serial-apply
 fallback

The leader apply worker records one replica identity entry per row
modified by an in-flight parallelized transaction, so a sustained burst
of large transactions could grow the replica_identity_table without
bound.

Add a limit (PARALLEL_DEPENDENCY_MEMORY_LIMIT, 16MB) on the estimated
memory held by dependency tracking entries. The estimate covers the key,
tuple data, column arrays and the hash entry itself, and is accounted at
insert and delete time so it stays exact without scanning the table.

When the estimate exceeds the limit, parallel apply is suspended: the
transaction being processed is first made to wait for the last
parallelized transaction (which, thanks to commit order preservation,
transitively waits for all preceding parallelized transactions), and
dependency checking and recording are skipped from then on. While
suspended, new transactions are applied by the leader itself, each
waiting for the last parallelized transaction before committing, so the
publisher's commit order is still preserved.

Parallel apply resumes once the estimated usage drains to half of the
limit, which happens as committed transactions' entries are reclaimed
while collecting flush positions.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
---
 .../replication/logical/applyparallelworker.c |  17 +++
 src/backend/replication/logical/worker.c      | 101 ++++++++++++++++++
 src/include/replication/worker_internal.h     |   2 +
 3 files changed, 120 insertions(+)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 4649c662c6f..20fe780e2c0 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -379,6 +379,14 @@ static ParallelApplyWorkerInfo *stream_apply_worker = NULL;
 /* A list to maintain subtransactions, if any. */
 static List *subxactlist = NIL;
 
+/*
+ * When true, the leader does not assign new transactions to parallel apply
+ * workers because dependency tracking has hit the memory limit. Cleared
+ * once the estimated usage drops to half of the limit or below and no
+ * transaction with stopped dependency recording is still in flight.
+ */
+bool		parallel_apply_suspended = false;
+
 static void pa_free_worker_info(ParallelApplyWorkerInfo *winfo);
 static ParallelTransState pa_get_xact_state(ParallelApplyWorkerShared *wshared);
 static PartialFileSetState pa_get_fileset_state(void);
@@ -395,6 +403,15 @@ pa_can_start(void)
 	if (!am_leader_apply_worker())
 		return false;
 
+	/*
+	 * Don't assign new transactions to parallel apply workers while the
+	 * dependency tracking memory limit is exceeded; the leader will apply
+	 * such transactions itself until the usage drops (see
+	 * handle_dependency_on_change).
+	 */
+	if (parallel_apply_suspended)
+		return false;
+
 	/*
 	 * It is good to check for any change in the subscription parameter to
 	 * avoid the case where for a very long time the change doesn't get
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 230f8be6d4e..819433c9241 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -585,6 +585,9 @@ typedef struct ReplicaIdentityKey
 {
 	Oid			relid;
 	LogicalRepTupleData *data;
+
+	/* estimated memory usage of this key, set by the key builders */
+	Size		keysize;
 } ReplicaIdentityKey;
 
 /* Hash table entry in replica_identity_table */
@@ -593,6 +596,9 @@ typedef struct ReplicaIdentityEntry
 	ReplicaIdentityKey *keydata;
 	TransactionId remote_xid;
 
+	/* estimated memory usage of this entry (including the key) */
+	Size		entry_size;
+
 	/* needed for simplehash */
 	uint32		hash;
 	char		status;
@@ -620,6 +626,13 @@ static bool hash_replica_identity_compare(ReplicaIdentityKey *a,
 
 #define REPLICA_IDENTITY_INITIAL_SIZE 128
 
+/*
+ * Maximum estimated memory that the leader apply worker may use for dependency
+ * tracking (replica_identity_table) before falling back to serial apply.
+ * Parallel apply resumes once the usage drops below half of this limit.
+ */
+#define PARALLEL_DEPENDENCY_MEMORY_LIMIT	(16 * 1024 * 1024)
+
 /*
  * Hash table storing replica identity values for changes being applied in
  * parallel, along with the latest transaction that modified each row.
@@ -636,6 +649,13 @@ static bool hash_replica_identity_compare(ReplicaIdentityKey *a,
  */
 static replica_identity_hash *replica_identity_table = NULL;
 
+/*
+ * Estimated memory currently held by replica_identity_table entries. Kept
+ * exact by accounting the estimated entry size at insert/delete time (see
+ * check_and_record_ri_dependency and delete_replica_identity_entry).
+ */
+static Size dependency_mem_usage = 0;
+
 static inline void subxact_filename(char *path, Oid subid, TransactionId xid);
 static inline void changes_filename(char *path, Oid subid, TransactionId xid);
 
@@ -786,6 +806,56 @@ hash_replica_identity_compare(ReplicaIdentityKey *a, ReplicaIdentityKey *b)
 	return true;
 }
 
+/*
+ * Suspend parallel apply if dependency memory usage exceeds the limit.
+ *
+ * Return true if parallel apply is already suspended, or if it has just been
+ * suspended due to exceeding the memory limit, false otherwise.
+ */
+static bool
+maybe_suspend_parallel_apply(ParallelApplyWorkerInfo *winfo)
+{
+	if (parallel_apply_suspended)
+		return true;
+
+	if (dependency_mem_usage < PARALLEL_DEPENDENCY_MEMORY_LIMIT)
+		return false;
+
+	parallel_apply_suspended = true;
+
+	/*
+	 * Wait for preceding transactions to finish before processing remaining
+	 * changes, as dependencies will not be tracked for them.
+	 */
+	maintain_commit_order_dependency(winfo);
+
+	ereport(LOG,
+			errmsg("parallel apply suspended: dependency memory limit of %d bytes exceeded",
+				   PARALLEL_DEPENDENCY_MEMORY_LIMIT));
+
+	return true;
+}
+
+/*
+ * Resume parallel apply if the estimated dependency tracking memory has
+ * dropped to half of the limit or below.
+ */
+static void
+maybe_resume_parallel_apply(void)
+{
+	if (!parallel_apply_suspended)
+		return;
+
+	if (dependency_mem_usage > PARALLEL_DEPENDENCY_MEMORY_LIMIT / 2)
+		return;
+
+	parallel_apply_suspended = false;
+
+	ereport(LOG,
+			errmsg("parallel apply resumed: dependency memory usage drained to %zu bytes",
+				   dependency_mem_usage));
+}
+
 /*
  * Free resources associated with a replica identity key.
  */
@@ -808,6 +878,7 @@ delete_replica_identity_entry(ReplicaIdentityEntry *rientry)
 {
 	free_replica_identity_key(rientry->keydata);
 	replica_identity_delete_item(replica_identity_table, rientry);
+	dependency_mem_usage -= rientry->entry_size;
 }
 
 /*
@@ -831,6 +902,8 @@ delete_replica_identity_entries_for_txns(List *committed_xids)
 		/* Clean up the hash entry for committed transaction */
 		delete_replica_identity_entry(rientry);
 	}
+
+	maybe_resume_parallel_apply();
 }
 
 /*
@@ -909,6 +982,7 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 	MemoryContext oldctx;
 	int			n_ri;
 	bool		found = false;
+	Size		keysize;
 
 	Assert(depends_on_xids);
 
@@ -949,6 +1023,10 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 	ridata->colstatus = palloc0_array(char, n_ri);
 	ridata->ncols = n_ri;
 
+	/* Estimated memory usage so far: key, tuple data and column arrays */
+	keysize = sizeof(ReplicaIdentityKey) + sizeof(LogicalRepTupleData) +
+		n_ri * (sizeof(StringInfoData) + sizeof(char));
+
 	for (int i_original = 0, i_ri = 0; i_original < original_data->ncols; i_original++)
 	{
 		if (!bms_is_member(i_original, relentry->remoterel.attkeys))
@@ -981,6 +1059,8 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 			appendBinaryStringInfo(&ridata->colvalues[i_ri],
 								   original_colvalue->data,
 								   original_colvalue->len);
+
+			keysize += original_colvalue->len + 1;
 		}
 
 		ridata->colstatus[i_ri] = original_data->colstatus[i_original];
@@ -990,6 +1070,7 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 	rikey = palloc0_object(ReplicaIdentityKey);
 	rikey->relid = relid;
 	rikey->data = ridata;
+	rikey->keysize = keysize;
 
 	MemoryContextSwitchTo(oldctx);
 
@@ -1040,6 +1121,12 @@ check_and_record_ri_dependency(Oid relid, LogicalRepTupleData *original_data,
 			append_xid_dependency(rientry->remote_xid, depends_on_xids);
 		}
 	}
+	else
+	{
+		/* Account the estimated memory usage of the new entry */
+		rientry->entry_size = rikey->keysize + sizeof(ReplicaIdentityEntry);
+		dependency_mem_usage += rientry->entry_size;
+	}
 
 	/* Update the new depended xid into the entry */
 	rientry->remote_xid = new_depended_xid;
@@ -1184,6 +1271,13 @@ handle_dependency_on_change(LogicalRepMsgType action, StringInfo s,
 														 NULL);
 	}
 
+	/*
+	 * If the estimated memory usage of dependency tracking exceeds the limit,
+	 * suspend parallel apply until the usage drops below half of the limit.
+	 */
+	if (maybe_suspend_parallel_apply(winfo))
+		return;
+
 	switch (action)
 	{
 		case LOGICAL_REP_MSG_INSERT:
@@ -1982,6 +2076,13 @@ apply_handle_begin(StringInfo s)
 	switch (apply_action)
 	{
 		case TRANS_LEADER_APPLY:
+			/*
+			 * Parallel apply has been suspended, meaning we may have skipped
+			 * dependency tracking for the last parallelized transaction. Wait
+			 * for it to finish to avoid any dependency violation.
+			 */
+			if (parallel_apply_suspended)
+				maintain_commit_order_dependency(NULL);
 			break;
 
 		case TRANS_LEADER_SEND_TO_PARALLEL:
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index aab6c04cd14..699b2198ac8 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -274,6 +274,8 @@ extern PGDLLIMPORT bool InitializingApplyWorker;
 
 extern PGDLLIMPORT List *table_states_not_ready;
 
+extern bool parallel_apply_suspended;
+
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype,
 												Oid subid, Oid relid,
-- 
2.43.0

