From 1112b130c3bed0a1338bbdc47b4f7f8fef5119ca Mon Sep 17 00:00:00 2001 From: Ayush Tiwari Date: Thu, 17 Sep 2026 18:53:35 +0530 Subject: [PATCH v3] Clean up pgoutput schema cache when streamed transactions prepare pgoutput remembers the top-level XID for each relation whose schema it sends in a streamed transaction. Stream commit and abort remove these entries, but a streamed transaction that prepares does not take either path. The entries can remain after COMMIT PREPARED or ROLLBACK PREPARED until the relation cache entry is rebuilt or the walsender exits. Clean them up after sending STREAM PREPARE. The subscriber processes the spooled relation messages at prepare time, and the relation mapping cache is not rolled back by ROLLBACK PREPARED. Use the same schema_sent handling as stream commit. This also avoids keeping the entries while the prepared transaction waits for its final outcome. Rename the cleanup helper's boolean argument to describe whether the schema was sent instead of how the transaction finished. Reviewed-by: Hayato Kuroda --- 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 484ffbe2ceeb4abfdfdb11fca94db41851a66815..dada328cd4687b09dffef317c3cd43de8f83a627 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 cachectx); -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 send_relation_and_attrs(Relation relation, TransactionId xid, @@ -1963,6 +1963,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); } /* @@ -2372,16 +2375,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; @@ -2392,16 +2395,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_xid(streamed_txn, entry->streamed_txns) { if (xid == streamed_txn) { - if (is_commit) + if (mark_schema_sent) entry->schema_sent = true; entry->streamed_txns = -- 2.34.1