postgres.git / summary / log / commit / refs

commit    274da088e021a92e2ffd2b8938a7b978b2cf11b3
Author:   Masahiko Sawada <msawada@postgresql.org>
Date:     Fri Sep 18 17:19:07 2026 +0000

    Fix memory leak in pgoutput with streamed prepared transactions.
    
    pgoutput records the toplevel XID in the relation's streamed_txns
    list, allocated in CacheMemoryContext, when it sends a relation's
    schema during a streamed transaction. We clean up that list at stream
    commit and stream abort, but missed stream prepare, so the XIDs stay
    around until a relcache invalidation for the relation or until the
    walsender exits.
    
    Fix by cleaning up the cache after writing the stream prepare message,
    marking the schema as sent as we do at stream commit. The subscriber
    applies the spooled relation messages at PREPARE time and its relation
    map isn't rolled back by ROLLBACK PREPARED, so we don't need to send
    the schema again even if the prepared transaction eventually aborts.
    
    Oversight in commit 63cf61cdeb7b.
    
    Author: Ayush Tiwari <ayushtiwari.slg01@gmail.com>
    Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
    Reviewed-by: Masahiko Sawada <sawada.mshk@gmail.com>
    Discussion: https://postgr.es/m/CAJTYsWWt5aaKHCnV=jNOwABzspd-WbnhWLndDqtESDQRQ-BkKQ@mail.gmail.com
    Backpatch-through: 15


 src/backend/replication/pgoutput/pgoutput.c | 29 ++++++++++++++++-------------
 1 file changed, 16 insertions(+), 13 deletions(-)

diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 7ae31b0acce..bd9b21d8b21 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -220,7 +220,7 @@ typedef struct PGOutputTxnData
 static HTAB *RelationSyncCache = NULL;
 
 static void init_rel_sync_cache(MemoryContext decoding_context);
-static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit);
+static void cleanup_rel_sync_cache(TransactionId xid, bool mark_schema_sent);
 static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data,
 											 Relation relation);
 static void rel_sync_cache_relation_cb(Datum arg, Oid relid);
@@ -1936,6 +1936,9 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
 	OutputPluginPrepareWrite(ctx, true);
 	logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
 	OutputPluginWrite(ctx, true);
+
+	/* Schema cache updates at PREPARE survive ROLLBACK PREPARED. */
+	cleanup_rel_sync_cache(txn->xid, true);
 }
 
 /*
@@ -2307,16 +2310,16 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
 /*
  * Cleanup list of streamed transactions and update the schema_sent flag.
  *
- * When a streamed transaction commits or aborts, we need to remove the
- * toplevel XID from the schema cache. If the transaction aborted, the
- * subscriber will simply throw away the schema records we streamed, so
- * we don't need to do anything else.
+ * When a streamed transaction commits, aborts or prepares, we need to remove
+ * the toplevel XID from the schema cache. If the transaction aborted, the
+ * subscriber will simply throw away the schema records we streamed, so we
+ * don't need to do anything else.
  *
- * If the transaction is committed, the subscriber will update the relation
- * cache - so tweak the schema_sent flag accordingly.
+ * If the transaction is committed or prepared, the subscriber will update
+ * the relation cache - so tweak the schema_sent flag accordingly.
  */
 static void
-cleanup_rel_sync_cache(TransactionId xid, bool is_commit)
+cleanup_rel_sync_cache(TransactionId xid, bool mark_schema_sent)
 {
 	HASH_SEQ_STATUS hash_seq;
 	RelationSyncEntry *entry;
@@ -2328,16 +2331,16 @@ cleanup_rel_sync_cache(TransactionId xid, bool is_commit)
 	while ((entry = hash_seq_search(&hash_seq)) != NULL)
 	{
 		/*
-		 * We can set the schema_sent flag for an entry that has committed xid
-		 * in the list as that ensures that the subscriber would have the
-		 * corresponding schema and we don't need to send it unless there is
-		 * any invalidation for that relation.
+		 * We can set the schema_sent flag for an entry that has a committed or
+		 * prepared xid in the list as that ensures that the subscriber would
+		 * have the corresponding schema and we don't need to send it unless
+		 * there is any invalidation for that relation.
 		 */
 		foreach(lc, entry->streamed_txns)
 		{
 			if (xid == (uint32) lfirst_int(lc))
 			{
-				if (is_commit)
+				if (mark_schema_sent)
 					entry->schema_sent = true;
 
 				entry->streamed_txns =


[parent: 47c43ffdae79]