From f1e2c053c50248595dc6539a606665336449e0e0 Mon Sep 17 00:00:00 2001 From: Vignesh Date: Thu, 26 Dec 2024 15:11:09 +0530 Subject: [PATCH v2_PG14-] Fix memory leak in pgoutput relation attribute map The pgoutput module caches relation attribute map and frees it upon invalidation. However, this was not getting freed incase of SQL functions like pg_logical_slot_{get,peek}_changes() which would bloat its cache memory usage. Every pg_logical_slot_{get,peek}_changes() call for changes on partition table creates more bloat. To address this, this commit adds a new memory context within the logical decoding context. This ensures that the lifespan of the relation entry attribute map aligns with that of the logical decoding context. The context is tracked with a static variable whose state is reset with a MemoryContext reset callback attached to PGOutputData->context, so as ABI compatibility is preserved in stable branches. --- src/backend/replication/pgoutput/pgoutput.c | 33 +++++++++++++++++++-- 1 file changed, 30 insertions(+), 3 deletions(-) diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 9fd879ee0c..577db5909d 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -72,6 +72,13 @@ static bool in_streaming; */ static MemoryContext pubctx = NULL; +/* + * Private memory context for relation attribute map, created in + * PGOutputData->context when starting pgoutput, and set to NULL when its + * parent context is reset via a dedicated MemoryContextCallback. + */ +static MemoryContext cachectx = NULL; + static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); @@ -268,6 +275,15 @@ pgoutput_pubctx_reset_callback(void *arg) pubctx = NULL; } +/* + * Callback of PGOutputData->context in charge of cleaning cachectx. + */ +static void +pgoutput_cachectx_reset_callback(void *arg) +{ + cachectx = NULL; +} + /* * Initialize this plugin */ @@ -278,6 +294,7 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, PGOutputData *data = palloc0(sizeof(PGOutputData)); static bool publication_callback_registered = false; MemoryContextCallback *mcallback; + MemoryContextCallback *cachectx_mcallback; /* Create our memory context for private allocations. */ data->context = AllocSetContextCreate(ctx->context, @@ -293,6 +310,15 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, mcallback->func = pgoutput_pubctx_reset_callback; MemoryContextRegisterResetCallback(ctx->context, mcallback); + Assert(cachectx == NULL); + cachectx = AllocSetContextCreate(ctx->context, + "logical replication cache context", + ALLOCSET_SMALL_SIZES); + + cachectx_mcallback = palloc0(sizeof(MemoryContextCallback)); + cachectx_mcallback->func = pgoutput_cachectx_reset_callback; + MemoryContextRegisterResetCallback(ctx->context, cachectx_mcallback); + ctx->output_plugin_private = data; /* This plugin uses binary protocol. */ @@ -490,7 +516,7 @@ maybe_send_schema(LogicalDecodingContext *ctx, MemoryContext oldctx; /* Map must live as long as the session does. */ - oldctx = MemoryContextSwitchTo(CacheMemoryContext); + oldctx = MemoryContextSwitchTo(cachectx); /* * Make copies of the TupleDescs that will live as long as the map @@ -812,8 +838,8 @@ pgoutput_origin_filter(LogicalDecodingContext *ctx, /* * Shutdown the output plugin. * - * Note, we don't need to clean the data->context and pubctx as they are - * child contexts of the ctx->context so they will be cleaned up by logical + * Note, we don't need to clean the data->context, pubctx, and cachectx as they + * are child contexts of the ctx->context so they will be cleaned up by logical * decoding machinery. */ static void @@ -827,6 +853,7 @@ pgoutput_shutdown(LogicalDecodingContext *ctx) /* Better safe than sorry */ pubctx = NULL; + cachectx = NULL; } /* -- 2.43.0