database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-11-19 13:42:17 +03:00
Code Activity

Avoid invalidating all RelationSyncCache entries on publication rename.

Browse Source 
On Publication rename, we need to only invalidate the RelationSyncCache
entries corresponding to relations that are part of the publication being
renamed.

As part of this patch, we introduce a new invalidation message to
invalidate the cache maintained by the logical decoding output plugin. We
can't use existing relcache invalidation for this purpose, as that would
unnecessarily cause relcache invalidations in other backends.

This will improve performance by building fewer relation cache entries
during logical replication.

Author: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Author: Shlok Kyal <shlok.kyal.oss@gmail.com>
Reviewed-by: Hou Zhijie <houzj.fnst@fujitsu.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Discussion: https://postgr.es/m/OSCPR01MB14966C09AA201EFFA706576A7F5C92@OSCPR01MB14966.jpnprd01.prod.outlook.com
This commit is contained in:
Amit Kapila
2025-03-13 09:03:45 +05:30
parent 75da2bece6
commit 3abe9dc188
10 changed files with 302 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
src/backend/utils/cache/inval.c vendored
View File

@@ -271,6 +271,7 @@ int debug_discard_caches = 0;
#define MAX_SYSCACHE_CALLBACKS 64
#define MAX_RELCACHE_CALLBACKS 10
#define MAX_RELSYNC_CALLBACKS 10
static struct SYSCACHECALLBACK
{
@@ -292,6 +293,15 @@ static struct RELCACHECALLBACK
static int relcache_callback_count = 0;
static struct RELSYNCCALLBACK
{
RelSyncCallbackFunction function;
Datum arg;
} relsync_callback_list[MAX_RELSYNC_CALLBACKS];
static int relsync_callback_count = 0;
/* ----------------------------------------------------------------
* Invalidation subgroup support functions
* ----------------------------------------------------------------
@@ -484,6 +494,36 @@ AddRelcacheInvalidationMessage(InvalidationMsgsGroup *group,
AddInvalidationMessage(group, RelCacheMsgs, &msg);
}
/*
* Add a relsync inval entry
*
* We put these into the relcache subgroup for simplicity. This message is the
* same as AddRelcacheInvalidationMessage() except that it is for
* RelationSyncCache maintained by decoding plugin pgoutput.
*/
static void
AddRelsyncInvalidationMessage(InvalidationMsgsGroup *group,
Oid dbId, Oid relId)
{
SharedInvalidationMessage msg;
/* Don't add a duplicate item. */
ProcessMessageSubGroup(group, RelCacheMsgs,
if (msg->rc.id == SHAREDINVALRELSYNC_ID &&
(msg->rc.relId == relId ||
msg->rc.relId == InvalidOid))
return);
/* OK, add the item */
msg.rc.id = SHAREDINVALRELSYNC_ID;
msg.rc.dbId = dbId;
msg.rc.relId = relId;
/* check AddCatcacheInvalidationMessage() for an explanation */
VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg));
AddInvalidationMessage(group, RelCacheMsgs, &msg);
}
/*
* Add a snapshot inval entry
*
@@ -611,6 +651,17 @@ RegisterRelcacheInvalidation(InvalidationInfo *info, Oid dbId, Oid relId)
info->RelcacheInitFileInval = true;
}
/*
* RegisterRelsyncInvalidation
*
* As above, but register a relsynccache invalidation event.
*/
static void
RegisterRelsyncInvalidation(InvalidationInfo *info, Oid dbId, Oid relId)
{
AddRelsyncInvalidationMessage(&info->CurrentCmdInvalidMsgs, dbId, relId);
}
/*
* RegisterSnapshotInvalidation
*
@@ -751,6 +802,13 @@ InvalidateSystemCachesExtended(bool debug_discard)
ccitem->function(ccitem->arg, InvalidOid);
}
for (i = 0; i < relsync_callback_count; i++)
{
struct RELSYNCCALLBACK *ccitem = relsync_callback_list + i;
ccitem->function(ccitem->arg, InvalidOid);
}
}
/*
@@ -832,6 +890,12 @@ LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
else if (msg->sn.dbId == MyDatabaseId)
InvalidateCatalogSnapshot();
}
else if (msg->id == SHAREDINVALRELSYNC_ID)
{
/* We only care about our own database */
if (msg->rs.dbId == MyDatabaseId)
CallRelSyncCallbacks(msg->rs.relid);
}
else
elog(FATAL, "unrecognized SI message ID: %d", msg->id);
}
@@ -1621,6 +1685,32 @@ CacheInvalidateRelcacheByRelid(Oid relid)
ReleaseSysCache(tup);
}
/*
* CacheInvalidateRelSync
* Register invalidation of the cache in logical decoding output plugin
* for a database.
*
* This type of invalidation message is used for the specific purpose of output
* plugins. Processes which do not decode WALs would do nothing even when it
* receives the message.
*/
void
CacheInvalidateRelSync(Oid relid)
{
RegisterRelsyncInvalidation(PrepareInvalidationState(),
MyDatabaseId, relid);
}
/*
* CacheInvalidateRelSyncAll
* Register invalidation of the whole cache in logical decoding output
* plugin.
*/
void
CacheInvalidateRelSyncAll(void)
{
CacheInvalidateRelSync(InvalidOid);
}
/*
* CacheInvalidateSmgr
@@ -1763,6 +1853,27 @@ CacheRegisterRelcacheCallback(RelcacheCallbackFunction func,
++relcache_callback_count;
}
/*
* CacheRegisterRelSyncCallback
* Register the specified function to be called for all future
* relsynccache invalidation events.
*
* This function is intended to be call from the logical decoding output
* plugins.
*/
void
CacheRegisterRelSyncCallback(RelSyncCallbackFunction func,
Datum arg)
{
if (relsync_callback_count >= MAX_RELSYNC_CALLBACKS)
elog(FATAL, "out of relsync_callback_list slots");
relsync_callback_list[relsync_callback_count].function = func;
relsync_callback_list[relsync_callback_count].arg = arg;
++relsync_callback_count;
}
/*
* CallSyscacheCallbacks
*
@@ -1788,6 +1899,20 @@ CallSyscacheCallbacks(int cacheid, uint32 hashvalue)
}
}
/*
* CallSyscacheCallbacks
*/
void
CallRelSyncCallbacks(Oid relid)
{
for (int i = 0; i < relsync_callback_count; i++)
{
struct RELSYNCCALLBACK *ccitem = relsync_callback_list + i;
ccitem->function(ccitem->arg, relid);
}
}
/*
* LogLogicalInvalidations
*