diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index c9c952a278f..33fe1461e11 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -211,7 +211,7 @@ static HTAB *RelationSyncCache = NULL; static void init_rel_sync_cache(MemoryContext cachectx); static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit); -static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data, +static RelationSyncEntry *get_rel_sync_entry(LogicalDecodingContext *ctx, Relation relation); static void rel_sync_cache_relation_cb(Datum arg, Oid relid); static void rel_sync_cache_publication_cb(Datum arg, int cacheid, @@ -234,6 +234,7 @@ static bool pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot, TupleTableSlot **new_slot_ptr, RelationSyncEntry *entry, ReorderBufferChangeType *action); +static void pgoutput_pubctx_reset_callback(void *arg); /* column list routines */ static void pgoutput_column_list_init(PGOutputData *data, @@ -1426,7 +1427,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (data->in_streaming) xid = change->txn->xid; - relentry = get_rel_sync_entry(data, relation); + relentry = get_rel_sync_entry(ctx, relation); /* First check the table filter */ switch (action) @@ -1598,7 +1599,7 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (!is_publishable_relation(relation)) continue; - relentry = get_rel_sync_entry(data, relation); + relentry = get_rel_sync_entry(ctx, relation); if (!relentry->pubactions.pubtruncate) continue; @@ -1970,8 +1971,9 @@ set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid) * when publishing. */ static RelationSyncEntry * -get_rel_sync_entry(PGOutputData *data, Relation relation) +get_rel_sync_entry(LogicalDecodingContext *ctx, Relation relation) { + PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; RelationSyncEntry *entry; bool found; MemoryContext oldctx; @@ -2024,12 +2026,27 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) /* Reload publications if needed before use. */ if (!publications_valid) { - oldctx = MemoryContextSwitchTo(CacheMemoryContext); - if (data->publications) + static MemoryContext pubctx = NULL; + + if (pubctx == NULL) { - list_free_deep(data->publications); - data->publications = NIL; + MemoryContextCallback *mcallback; + + pubctx = AllocSetContextCreate(ctx->context, + "logical replication publication list context", + ALLOCSET_SMALL_SIZES); + + mcallback = MemoryContextAllocZero(ctx->context, + sizeof(MemoryContextCallback)); + mcallback->func = pgoutput_pubctx_reset_callback; + mcallback->arg = &pubctx; + MemoryContextRegisterResetCallback(ctx->context, mcallback); } + else + MemoryContextReset(pubctx); + + oldctx = MemoryContextSwitchTo(pubctx); + data->publications = LoadPublications(data->publication_names); MemoryContextSwitchTo(oldctx); publications_valid = true; @@ -2402,3 +2419,11 @@ send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, } } } + +static void +pgoutput_pubctx_reset_callback(void *arg) +{ + MemoryContext *pubctx = (MemoryContext *) arg; + + *pubctx = NULL; +}