public inbox for [email protected]
help / color / mirror / Atom feedFrom: Hayato Kuroda (Fujitsu) <[email protected]>
To: 'Amit Kapila' <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Zhijie Hou (Fujitsu) <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Subject: RE: Selectively invalidate caches in pgoutput module
Date: Tue, 4 Mar 2025 11:43:33 +0000
Message-ID: <OSCPR01MB14966206D8490B0EC5E7B6AAEF5C82@OSCPR01MB14966.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1LHoHjZ1mbuf1Kya5eHZXyeO7Xi+XEiRFVf=-YMdPAY+g@mail.gmail.com>
References: <OSCPR01MB14966C09AA201EFFA706576A7F5C92@OSCPR01MB14966.jpnprd01.prod.outlook.com>
<CAA4eK1LHoHjZ1mbuf1Kya5eHZXyeO7Xi+XEiRFVf=-YMdPAY+g@mail.gmail.com>
Dear Amit,
> > When the ALTER PUBLICATION command is executed, all entries in
> RelationSyncCache
> > will be discarded anyway. This mechanism works well but is sometimes not
> efficient.
> > For example, when the ALTER PUBLICATION DROP TABLE is executed,
> > 1) the specific entry in RelationSyncCache will be removed, and then
> > 2) all entries will be discarded twice.
> >
>
> In (2) Why twice? Is it because we call
> rel_sync_cache_publication_cb() first for PUBLICATIONRELMAP and then
> for PUBLICATIONOID via publication_invalidation_cb()?
This part was not correct. I considered that ALTER PUBLICATION DROP TABLE also
invalidated the pg_publication syscache, but it would not do. So...the command
firstly invalidate a specific entry, and then invalidate all entries once.
> > This happens because the pgoutput plugin registers both RelcacheCallback
> > (rel_sync_cache_relation_cb) and SyscacheCallback
> (publication_invalidation_cb,
> > rel_sync_cache_publication_cb). Then, when ALTER PUBLICATION
> ADD/SET/DROP is executed,
> > both the relation cache of added tables and the syscache of pg_publication_rel
> and
> > pg_publication are invalidated.
> > The callback for the relation cache will remove an entry from the hash table, and
> > syscache callbacks will look up all entries and invalidate them. However,
> AFAICS
> > does not need to invalidate all of them.
> >
>
> IIUC, you want to say in the above case only (1) would be sufficient, right?
Right, this is a main aim of this proposal.
> Is it possible to see the impact? I guess if there are a large number
> of relations (or partitions), then all will get invalidated even if
> one relation is added/dropped from the publication; if so, this should
> impact the performance.
Agreed. I'm planning to do and share results. Please wait...
> For Rename/Owner, can we use a new invalidation instead of using
> relcache invalidation, as that can impact other sessions for no
> benefit? IIUC, changing the other properties of publication like
> dropping the table requires us to invalidate even the corresponding
> relcache entry because it contains the publication descriptor
> (rd_pubdesc). See RelationBuildPublicationDesc().
Attached patchset implemented with the approach.
0001 contains only part to avoid whole cache invalidation, for ADD/SET/DROP command.
0002 introduces new type of invalidation message which only invalidates caches on
the decoding plugin. Better name is very welcome.
0003 uses the mechanism to avoid discarding relcache for RENAME/OWNER case.
Best regards,
Hayato Kuroda
FUJITSU LIMITED
Attachments:
[application/octet-stream] v2-0001-Avoid-Invalidating-all-entries-when-ALTER-PUBLICA.patch (1.7K, ../OSCPR01MB14966206D8490B0EC5E7B6AAEF5C82@OSCPR01MB14966.jpnprd01.prod.outlook.com/2-v2-0001-Avoid-Invalidating-all-entries-when-ALTER-PUBLICA.patch)
download | inline diff:
From 025ecfd3b105508c50936be170b2ccbc3a25c994 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 3 Mar 2025 19:40:10 +0900
Subject: [PATCH v2 1/3] Avoid Invalidating all entries when ALTER PUBLICATION
is executed
When the ALTER PUBLICATION command is executed, all entries in RelationSyncCache
will be discarded anyway. This mechanism works well but is sometimes not
efficient. For example, when the ALTER PUBLICATION DROP TABLE is executed,
1) the specific entry in RelationSyncCache will be removed, and then 2) all
entries will be discarded.
This patch avoids dropping all entries in RelationSyncCache when ALTER
PUBLICATION is executed. Theoretically, it is enough to invalidate the related
relacache since RelacheCallback has already been registered.
Author: Shlok Kyal and Hayato Kuroda
---
src/backend/replication/pgoutput/pgoutput.c | 12 ------------
1 file changed, 12 deletions(-)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 7d464f656a..64aea3af83 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1970,18 +1970,6 @@ init_rel_sync_cache(MemoryContext cachectx)
rel_sync_cache_publication_cb,
(Datum) 0);
- /*
- * Flush all cache entries after any publication changes. (We need no
- * callback entry for pg_publication, because publication_invalidation_cb
- * will take care of it.)
- */
- CacheRegisterSyscacheCallback(PUBLICATIONRELMAP,
- rel_sync_cache_publication_cb,
- (Datum) 0);
- CacheRegisterSyscacheCallback(PUBLICATIONNAMESPACEMAP,
- rel_sync_cache_publication_cb,
- (Datum) 0);
-
relation_callbacks_registered = true;
}
--
2.43.5
[application/octet-stream] v2-0002-Introduce-a-new-invalidation-message-to-invalidat.patch (6.4K, ../OSCPR01MB14966206D8490B0EC5E7B6AAEF5C82@OSCPR01MB14966.jpnprd01.prod.outlook.com/3-v2-0002-Introduce-a-new-invalidation-message-to-invalidat.patch)
download | inline diff:
From 07f9825641c6f64fb1fc475121df79ad361867be Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Tue, 4 Mar 2025 16:51:19 +0900
Subject: [PATCH v2 2/3] Introduce a new invalidation message to invalidate
caches in output plugins
---
src/backend/utils/cache/inval.c | 87 +++++++++++++++++++++++++++++++++
src/include/storage/sinval.h | 11 +++++
src/include/utils/inval.h | 10 ++++
3 files changed, 108 insertions(+)
diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index 700ccb6df9..d793bc9281 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
* ----------------------------------------------------------------
@@ -832,6 +842,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);
}
@@ -1695,6 +1711,42 @@ CacheInvalidateRelmap(Oid databaseId)
}
+/*
+ * RelationCacheInvalidate
+ * 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)
+{
+ SharedInvalidationMessage msg;
+
+ msg.rs.id = SHAREDINVALRELSYNC_ID;
+ msg.rs.dbId = MyDatabaseId;
+ msg.rs.relid = relid;
+ /* check AddCatcacheInvalidationMessage() for an explanation */
+ VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg));
+
+ SendSharedInvalidMessages(&msg, 1);
+}
+
+
+/*
+ * CacheInvalidateRelSyncAll
+ * Register invalidation of the whole cache in logical decoding output
+ * plugin.
+ */
+void
+CacheInvalidateRelSyncAll(void)
+{
+ CacheInvalidateRelSync(InvalidOid);
+}
+
+
/*
* CacheRegisterSyscacheCallback
* Register the specified function to be called for all future
@@ -1763,6 +1815,27 @@ CacheRegisterRelcacheCallback(RelcacheCallbackFunction func,
++relcache_callback_count;
}
+/*
+ * CacheRegisterRelSyncCallback
+ * Register the specified function to be called for all future
+ * decoding-cache 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 +1861,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/storage/sinval.h b/src/include/storage/sinval.h
index 2463c0f9fa..90a5af4ed8 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 specific entry for specific output plugin
* 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).
@@ -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 relcache */
+} 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..5922306c11 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);
@@ -59,6 +60,10 @@ extern void CacheInvalidateSmgr(RelFileLocatorBackend rlocator);
extern void CacheInvalidateRelmap(Oid databaseId);
+extern void CacheInvalidateRelSync(Oid relid);
+
+extern void CacheInvalidateRelSyncAll(void);
+
extern void CacheRegisterSyscacheCallback(int cacheid,
SyscacheCallbackFunction func,
Datum arg);
@@ -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] v2-0003-Invalidate-Relcaches-while-ALTER-PUBLICATION-OWNE.patch (7.8K, ../OSCPR01MB14966206D8490B0EC5E7B6AAEF5C82@OSCPR01MB14966.jpnprd01.prod.outlook.com/4-v2-0003-Invalidate-Relcaches-while-ALTER-PUBLICATION-OWNE.patch)
download | inline diff:
From 5a49160c929b84ad7a62f8e1bccf215a53c0fb93 Mon Sep 17 00:00:00 2001
From: Hayato Kuroda <[email protected]>
Date: Mon, 3 Mar 2025 19:41:04 +0900
Subject: [PATCH v2 3/3] Invalidate Relcaches while ALTER PUBLICATION OWNER
TO/RENAME TO
---
src/backend/commands/alter.c | 4 +-
src/backend/commands/publicationcmds.c | 139 ++++++++++++++++++++
src/backend/parser/gram.y | 2 +-
src/backend/replication/pgoutput/pgoutput.c | 8 +-
src/include/commands/publicationcmds.h | 2 +
5 files changed, 147 insertions(+), 8 deletions(-)
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index 78c1d4e1b8..a79329acc1 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -400,6 +400,9 @@ ExecRenameStmt(RenameStmt *stmt)
case OBJECT_TYPE:
return RenameType(stmt);
+ case OBJECT_PUBLICATION:
+ return RenamePublication(stmt->subname, stmt->newname);
+
case OBJECT_AGGREGATE:
case OBJECT_COLLATION:
case OBJECT_CONVERSION:
@@ -417,7 +420,6 @@ ExecRenameStmt(RenameStmt *stmt)
case OBJECT_TSDICTIONARY:
case OBJECT_TSPARSER:
case OBJECT_TSTEMPLATE:
- case OBJECT_PUBLICATION:
case OBJECT_SUBSCRIPTION:
{
ObjectAddress address;
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 150a768d16..c1cab00ddb 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -491,6 +491,95 @@ pub_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
return *invalid_column_list || *invalid_gen_col;
}
+/*
+ * Execute ALTER PUBLICATION RENAME
+ */
+ObjectAddress
+RenamePublication(const char *oldname, const char *newname)
+{
+ Relation rel;
+ HeapTuple tup;
+ ObjectAddress address;
+ Form_pg_publication pubform;
+ bool replaces[Natts_pg_publication];
+ bool nulls[Natts_pg_publication];
+ Datum values[Natts_pg_publication];
+
+ rel = table_open(PublicationRelationId, RowExclusiveLock);
+
+ tup = SearchSysCacheCopy1(PUBLICATIONNAME,
+ CStringGetDatum(oldname));
+
+ if (!HeapTupleIsValid(tup))
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("publication \"%s\" does not exist",
+ oldname)));
+
+ pubform = (Form_pg_publication) GETSTRUCT(tup);
+
+ /* must be owner */
+ if (!object_ownercheck(PublicationRelationId, pubform->oid, GetUserId()))
+ aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
+ NameStr(pubform->pubname));
+
+ /* Everything ok, form a new tuple. */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replaces, false, sizeof(replaces));
+
+ /* Only update the pubname */
+ values[Anum_pg_publication_pubname - 1] =
+ DirectFunctionCall1(namein, CStringGetDatum(newname));
+ replaces[Anum_pg_publication_pubname - 1] = true;
+
+ tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+ replaces);
+
+ /*
+ * Invalidate caches on the logical decoding output plugin.
+ *
+ * Apart from the ALTER PUBLICATION ADD/SET/DROP commands, we do not have
+ * to invalidate relcaches. They are required to refresh the publication's
+ * descriptor. The publication's name is not recorded in the attribute, so
+ * the RENAME commands do not require a refresh. Instead, we only
+ * invalidate a cache on the output plugin to rebuild its cache.
+ */
+ if (pubform->puballtables)
+ {
+ CacheInvalidateRelSyncAll();
+ }
+ else
+ {
+ List *relids = NIL;
+ List *schemarelids = NIL;
+
+ /*
+ * For partition table, when we insert data, get_rel_sync_entry is
+ * called and a hash entry is created for the corresponding leaf table.
+ * So invalidating the leaf nodes would be sufficient here.
+ */
+ relids = GetPublicationRelations(pubform->oid,
+ PUBLICATION_PART_LEAF);
+ schemarelids = GetAllSchemaPublicationRelations(pubform->oid,
+ PUBLICATION_PART_LEAF);
+
+ relids = list_concat_unique_oid(relids, schemarelids);
+
+ InvalidateRelSyncCaches(relids);
+ }
+
+ CatalogTupleUpdate(rel, &tup->t_self, tup);
+
+ ObjectAddressSet(address, PublicationRelationId, pubform->oid);
+
+ heap_freetuple(tup);
+
+ table_close(rel, RowExclusiveLock);
+
+ return address;
+}
+
/* check_functions_in_node callback */
static bool
contain_mutable_or_user_functions_checker(Oid func_id, void *context)
@@ -1996,6 +2085,37 @@ AlterPublicationOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
}
form->pubowner = newOwnerId;
+
+ /*
+ * Invalidate caches on the logical decoding output plugin.
+ *
+ * No need to invalidate relcache as the same reason as RENAME command.
+ * Please see comments in RenamePublication().
+ */
+ if (form->puballtables)
+ {
+ CacheInvalidateRelSyncAll();
+ }
+ else
+ {
+ List *relids = NIL;
+ List *schemarelids = NIL;
+
+ /*
+ * For partition table, when we insert data, get_rel_sync_entry is
+ * called and a hash entry is created for the corresponding leaf table.
+ * So invalidating the leaf nodes would be sufficient here.
+ */
+ relids = GetPublicationRelations(form->oid,
+ PUBLICATION_PART_LEAF);
+ schemarelids = GetAllSchemaPublicationRelations(form->oid,
+ PUBLICATION_PART_LEAF);
+
+ relids = list_concat_unique_oid(relids, schemarelids);
+
+ InvalidateRelSyncCaches(relids);
+ }
+
CatalogTupleUpdate(rel, &tup->t_self, tup);
/* Update owner dependency reference */
@@ -2096,3 +2216,22 @@ defGetGeneratedColsOption(DefElem *def)
return PUBLISH_GENCOLS_NONE; /* keep compiler quiet */
}
+
+
+void
+InvalidateRelSyncCaches(List *relids)
+{
+ /*
+ * We don't want to send too many individual messages, at some point it's
+ * cheaper to just reset whole relcache.
+ */
+ if (list_length(relids) < MAX_RELCACHE_INVAL_MSGS)
+ {
+ ListCell *lc;
+
+ foreach(lc, relids)
+ CacheInvalidateRelSync(lfirst_oid(lc));
+ }
+ else
+ CacheInvalidateRelSync(InvalidOid);
+}
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 7d99c9355c..49fe0567c5 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9513,7 +9513,7 @@ RenameStmt: ALTER AGGREGATE aggregate_with_argtypes RENAME TO name
RenameStmt *n = makeNode(RenameStmt);
n->renameType = OBJECT_PUBLICATION;
- n->object = (Node *) makeString($3);
+ n->subname = $3;
n->newname = $6;
n->missing_ok = false;
$$ = (Node *) n;
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 64aea3af83..527165dc1a 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..f130ea3090 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 ObjectAddress RenamePublication(const char *oldname, const char *newname);
+extern void InvalidateRelSyncCaches(List *relids);
#endif /* PUBLICATIONCMDS_H */
--
2.43.5
view thread (2+ messages)
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: <OSCPR01MB14966206D8490B0EC5E7B6AAEF5C82@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