public inbox for [email protected]
help / color / mirror / Atom feedFrom: Hayato Kuroda (Fujitsu) <[email protected]>
To: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Zhijie Hou (Fujitsu) <[email protected]>
Cc: 'Amit Kapila' <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Subject: RE: Selectively invalidate caches in pgoutput module
Date: Tue, 11 Mar 2025 12:17:14 +0000
Message-ID: <OSCPR01MB149665015BC2F5DC260E4E184F5D12@OSCPR01MB14966.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <TYAPR01MB5724414167645263B733A86394D12@TYAPR01MB5724.jpnprd01.prod.outlook.com>
References: <OSCPR01MB14966C09AA201EFFA706576A7F5C92@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<CAA4eK1LHoHjZ1mbuf1Kya5eHZXyeO7Xi+XEiRFVf=-YMdPAY+g@mail.gmail.com>
<OSCPR01MB14966206D8490B0EC5E7B6AAEF5C82@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OS0PR01MB5716BCFACE982F17D1F7554994CB2@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OSCPR01MB1496622457D213FA8681750BEF5CB2@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OSCPR01MB149666315C734F781ED70D0AAF5CB2@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<CAA4eK1LdT80+ASA4c_pvqoxzK=ZOCn8DtMLK5t8++62aL5LeNw@mail.gmail.com>
<OSCPR01MB1496638A193D30795EC24386FF5CA2@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<CAA4eK1LMR7anY0Rv7sUFYFcqxetzV-M9BVsQYAm=OAp=3nL4CQ@mail.gmail.com>
<OSCPR01MB149661B4D4A4E912E8D85CB43F5D52@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<CAA4eK1Jw4uSFv2zu2tT6GN4DVoW54kV=sM2iHj+h+b6xyVF02w@mail.gmail.com>
<OSCPR01MB149669AF61681064D2A75FAE9F5D42@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OSCPR01MB1496609BD298A6660B1F8AF28F5D62@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OSCPR01MB14966B4602D103050DC91837DF5D62@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<OS3PR01MB5718B8EFFBB856C8B5749E8B94D62@OS3PR01MB5718.jpnprd01.prod.outlook.com>
<OSCPR01MB14966DBBE4666C701FB8943E0F5D62@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<TYAPR01MB5724414167645263B733A86394D12@TYAPR01MB5724.jpnprd01.prod.outlook.com>
Dear Amit, Hou,
Thanks for giving comments!
> For patch 0002, I think the implementation could be improved. The
> current patch introduces a new function, RenamePublication, to replace the
> existing generic approach in ExecRenameStmt->AlterObjectRename_internal.
> However, this creates inconsistency because the original code uses
> AccessExclusiveLock for publication renaming, making concurrent renaming
> impossible. The current patch seems to overlook this aspect.
Oh, I missed the point, thanks.
> Additionally, introducing a new function results in code duplication, which can
> be avoided. After further consideration, handling the publication rename
> separately seems unnecessary, given it requires only sending a few extra
> invalidation messages. Therefore, I suggest reusing the generic handling and
> simply extending AlterObjectRename_internal to include the additional
> invalidation messages.
>
> I've attached a diff with the proposed changes for 0002.
Hmm, possible. previously I introduced new function to preserve existing codes as
much as possible. But this introduced some duplicy. Your approach which extends
the common function looks nice, apart from two points;
1. Relsync cache invalidations should be done after the catalog update, but
proposed one did before that. I preferred to add new if-statementfor post catalog-update.
2. Also, common check for the name conflict can be reused.
Attached patch address comments by you and Amit [1]. What's new;
* Skip adding new relsync messages when message for InvalidOid has already exist.
Per comment from Amit [1].
* Update some code comments. Some of them are pointed out by [1].
* Stop using RenamePublication() and extend the common function.
Best regards,
Hayato Kuroda
FUJITSU LIMITED
Attachments:
[application/octet-stream] v10-0001-Introduce-a-new-invalidation-message-to-invalida.patch (11.3K, ../OSCPR01MB149665015BC2F5DC260E4E184F5D12@OSCPR01MB14966.jpnprd01.prod.outlook.com/2-v10-0001-Introduce-a-new-invalidation-message-to-invalida.patch)
download | inline diff:
From 68caed94ad297cdc8a82d9926c39db5ff96ed876 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Tue, 4 Mar 2025 16:51:19 +0900
Subject: [PATCH v10 1/2] Introduce a new invalidation message to invalidate
caches in output plugins
A new invalidation message is generated when either ALTER PUBLICATION RENAME TO
or is executed. The primal use-case of the message is to invalidate caches on
the logical decoding output plugin. Plugins can register callback functions for
the message via CacheRegisterRelSyncCallback(), and the function can invalidate
the cache for the specified relation.
A new invalidation message is transactional, and decoder processes should
recognize the message and invalidate specified caches. Thus, the messages are
stored in InvalMessageArray and serialized at the end of the transaction.
---
src/backend/access/rmgrdesc/standbydesc.c | 2 +
src/backend/utils/cache/inval.c | 125 ++++++++++++++++++++++
src/include/pg_config_manual.h | 8 +-
src/include/storage/sinval.h | 23 ++--
src/include/utils/inval.h | 10 ++
5 files changed, 158 insertions(+), 10 deletions(-)
diff --git a/src/backend/access/rmgrdesc/standbydesc.c b/src/backend/access/rmgrdesc/standbydesc.c
index d849f8e54b..81eff5f31c 100644
--- a/src/backend/access/rmgrdesc/standbydesc.c
+++ b/src/backend/access/rmgrdesc/standbydesc.c
@@ -132,6 +132,8 @@ standby_desc_invalidations(StringInfo buf,
appendStringInfo(buf, " relmap db %u", msg->rm.dbId);
else if (msg->id == SHAREDINVALSNAPSHOT_ID)
appendStringInfo(buf, " snapshot %u", msg->sn.relId);
+ else if (msg->id == SHAREDINVALRELSYNC_ID)
+ appendStringInfo(buf, " relsync %u", msg->rs.relid);
else
appendStringInfo(buf, " unrecognized id %d", msg->id);
}
diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index 700ccb6df9..4eb6772073 100644
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -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
*
diff --git a/src/include/pg_config_manual.h b/src/include/pg_config_manual.h
index 449e50bd78..125d3eb5ff 100644
--- a/src/include/pg_config_manual.h
+++ b/src/include/pg_config_manual.h
@@ -282,10 +282,10 @@
/*
* For cache-invalidation debugging, define DISCARD_CACHES_ENABLED to enable
- * use of the debug_discard_caches GUC to aggressively flush syscache/relcache
- * entries whenever it's possible to deliver invalidations. See
- * AcceptInvalidationMessages() in src/backend/utils/cache/inval.c for
- * details.
+ * use of the debug_discard_caches GUC to aggressively flush
+ * syscache/relcache/relsynccache entries whenever it's possible to deliver
+ * invalidations. See AcceptInvalidationMessages() in
+ * src/backend/utils/cache/inval.c for details.
*
* USE_ASSERT_CHECKING builds default to enabling this. It's possible to use
* DISCARD_CACHES_ENABLED without a cassert build and the implied
diff --git a/src/include/storage/sinval.h b/src/include/storage/sinval.h
index 2463c0f9fa..f168b5fbf8 100644
--- a/src/include/storage/sinval.h
+++ b/src/include/storage/sinval.h
@@ -27,6 +27,7 @@
* * invalidate an smgr cache entry for a specific physical relation
* * invalidate the mapped-relation mapping for a given database
* * invalidate any saved snapshot that might be used to scan a given relation
+ * * invalidate a RelationSyncCache entry for a specific relation
* More types could be added if needed. The message type is identified by
* the first "int8" field of the message struct. Zero or positive means a
* specific-catcache inval message (and also serves as the catcache ID field).
@@ -46,12 +47,12 @@
* catcache inval messages must be generated for each of its caches, since
* the hash keys will generally be different.
*
- * Catcache, relcache, and snapshot invalidations are transactional, and so
- * are sent to other backends upon commit. Internally to the generating
- * backend, they are also processed at CommandCounterIncrement so that later
- * commands in the same transaction see the new state. The generating backend
- * also has to process them at abort, to flush out any cache state it's loaded
- * from no-longer-valid entries.
+ * Catcache, relcache, relsynccache, and snapshot invalidations are
+ * transactional, and so are sent to other backends upon commit. Internally
+ * to the generating backend, they are also processed at
+ * CommandCounterIncrement so that later commands in the same transaction see
+ * the new state. The generating backend also has to process them at abort,
+ * to flush out any cache state it's loaded from no-longer-valid entries.
*
* smgr and relation mapping invalidations are non-transactional: they are
* sent immediately when the underlying file change is made.
@@ -110,6 +111,15 @@ typedef struct
Oid relId; /* relation ID */
} SharedInvalSnapshotMsg;
+#define SHAREDINVALRELSYNC_ID (-6)
+
+typedef struct
+{
+ int8 id; /* type field --- must be first */
+ Oid dbId; /* database ID */
+ Oid relid; /* relation ID, or 0 if whole RelationSyncCache */
+} SharedInvalRelSyncMsg;
+
typedef union
{
int8 id; /* type field --- must be first */
@@ -119,6 +129,7 @@ typedef union
SharedInvalSmgrMsg sm;
SharedInvalRelmapMsg rm;
SharedInvalSnapshotMsg sn;
+ SharedInvalRelSyncMsg rs;
} SharedInvalidationMessage;
diff --git a/src/include/utils/inval.h b/src/include/utils/inval.h
index 40658ba2ff..9b871caef6 100644
--- a/src/include/utils/inval.h
+++ b/src/include/utils/inval.h
@@ -22,6 +22,7 @@ extern PGDLLIMPORT int debug_discard_caches;
typedef void (*SyscacheCallbackFunction) (Datum arg, int cacheid, uint32 hashvalue);
typedef void (*RelcacheCallbackFunction) (Datum arg, Oid relid);
+typedef void (*RelSyncCallbackFunction) (Datum arg, Oid relid);
extern void AcceptInvalidationMessages(void);
@@ -55,6 +56,10 @@ extern void CacheInvalidateRelcacheByTuple(HeapTuple classTuple);
extern void CacheInvalidateRelcacheByRelid(Oid relid);
+extern void CacheInvalidateRelSync(Oid relid);
+
+extern void CacheInvalidateRelSyncAll(void);
+
extern void CacheInvalidateSmgr(RelFileLocatorBackend rlocator);
extern void CacheInvalidateRelmap(Oid databaseId);
@@ -66,8 +71,13 @@ extern void CacheRegisterSyscacheCallback(int cacheid,
extern void CacheRegisterRelcacheCallback(RelcacheCallbackFunction func,
Datum arg);
+extern void CacheRegisterRelSyncCallback(RelSyncCallbackFunction func,
+ Datum arg);
+
extern void CallSyscacheCallbacks(int cacheid, uint32 hashvalue);
+extern void CallRelSyncCallbacks(Oid relid);
+
extern void InvalidateSystemCaches(void);
extern void InvalidateSystemCachesExtended(bool debug_discard);
--
2.43.5
[application/octet-stream] v10-0002-Invalidate-Relcaches-while-ALTER-PUBLICATION-REN.patch (4.6K, ../OSCPR01MB149665015BC2F5DC260E4E184F5D12@OSCPR01MB14966.jpnprd01.prod.outlook.com/3-v10-0002-Invalidate-Relcaches-while-ALTER-PUBLICATION-REN.patch)
download | inline diff:
From f1ecaa5e9c08886f0eab2c038e13d25a195e7d33 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 3 Mar 2025 19:41:04 +0900
Subject: [PATCH v10 2/2] Invalidate Relcaches while ALTER PUBLICATION RENAME
TO
---
src/backend/commands/alter.c | 16 +++++++
src/backend/commands/publicationcmds.c | 49 +++++++++++++++++++++
src/backend/replication/pgoutput/pgoutput.c | 8 +---
src/include/commands/publicationcmds.h | 2 +
4 files changed, 69 insertions(+), 6 deletions(-)
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index 78c1d4e1b8..2187d55a2e 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -338,6 +338,22 @@ AlterObjectRename_internal(Relation rel, Oid objectId, const char *new_name)
InvokeObjectPostAlterHook(classId, objectId, 0);
+ /* Do post catalog-update tasks */
+ if (classId == PublicationRelationId)
+ {
+ Form_pg_publication pub = (Form_pg_publication) GETSTRUCT(oldtup);
+
+ /*
+ * Invalidate relsynccaches.
+ *
+ * Unlike ALTER PUBLICATION ADD/SET/DROP commands, renaming a
+ * publication does not impact the publication status of tables.
+ * Therefore, we do not need to invalidate relcache to rebuild the
+ * rd_pubdesc. Instead, we invalidate only the relsyncache.
+ */
+ InvalidatePubRelSyncCaches(pub->oid, pub->puballtables);
+ }
+
/* Release memory */
pfree(values);
pfree(nulls);
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 150a768d16..2ae0f807f2 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -491,6 +491,43 @@ pub_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
return *invalid_column_list || *invalid_gen_col;
}
+/*
+ * Invalidate entries in the RelationSyncCache for relations included in the
+ * specified publication, either via FOR TABLE or FOR TABLES IN SCHEMA.
+ *
+ * If 'puballtables' is true, invalidate all cache entries.
+ */
+void
+InvalidatePubRelSyncCaches(Oid pubid, bool puballtables)
+{
+ if (puballtables)
+ {
+ CacheInvalidateRelSyncAll();
+ }
+ else
+ {
+ List *relids = NIL;
+ List *schemarelids = NIL;
+
+ /*
+ * For partitioned tables, we must invalidate all partitions and
+ * itself. WAL records for INSERT/UPDATE/DELETE specify leaf
+ * tables as a target. However, WAL records for TRUNCATE specify
+ * both a root and its leaves.
+ */
+ relids = GetPublicationRelations(pubid,
+ PUBLICATION_PART_ALL);
+ schemarelids = GetAllSchemaPublicationRelations(pubid,
+ PUBLICATION_PART_ALL);
+
+ relids = list_concat_unique_oid(relids, schemarelids);
+
+ InvalidateRelSyncCaches(relids);
+ }
+
+ return;
+}
+
/* check_functions_in_node callback */
static bool
contain_mutable_or_user_functions_checker(Oid func_id, void *context)
@@ -2096,3 +2133,15 @@ defGetGeneratedColsOption(DefElem *def)
return PUBLISH_GENCOLS_NONE; /* keep compiler quiet */
}
+
+/*
+ * Invalidate the relsyncaches.
+ */
+void
+InvalidateRelSyncCaches(List *relids)
+{
+ ListCell *lc;
+
+ foreach(lc, relids)
+ CacheInvalidateRelSync(lfirst_oid(lc));
+}
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 9063af6e1d..ed806c5430 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -531,6 +531,8 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
CacheRegisterSyscacheCallback(PUBLICATIONOID,
publication_invalidation_cb,
(Datum) 0);
+ CacheRegisterRelSyncCallback(rel_sync_cache_relation_cb,
+ (Datum) 0);
publication_callback_registered = true;
}
@@ -1789,12 +1791,6 @@ static void
publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue)
{
publications_valid = false;
-
- /*
- * Also invalidate per-relation cache so that next time the filtering info
- * is checked it will be updated with the new publication settings.
- */
- rel_sync_cache_publication_cb(arg, cacheid, hashvalue);
}
/*
diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h
index e11a942ea0..8eb1795d47 100644
--- a/src/include/commands/publicationcmds.h
+++ b/src/include/commands/publicationcmds.h
@@ -38,5 +38,7 @@ extern bool pub_contains_invalid_column(Oid pubid, Relation relation,
char pubgencols_type,
bool *invalid_column_list,
bool *invalid_gen_col);
+extern void InvalidatePubRelSyncCaches(Oid pubid, bool puballtables);
+extern void InvalidateRelSyncCaches(List *relids);
#endif /* PUBLICATIONCMDS_H */
--
2.43.5
view thread (23+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Selectively invalidate caches in pgoutput module
In-Reply-To: <OSCPR01MB149665015BC2F5DC260E4E184F5D12@OSCPR01MB14966.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox