database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-09-08 00:47:37 +03:00
Code Activity

Invalidate all partitions for a partitioned table in publication.

Browse Source 
Updates/Deletes on a partition were allowed even without replica identity
after the parent table was added to a publication. This would later lead
to an error on subscribers. The reason was that we were not invalidating
the partition's relcache and the publication information for partitions
was not getting rebuilt. Similarly, we were not invalidating the
partitions' relcache after dropping a partitioned table from a publication
which will prohibit Updates/Deletes on its partition without replica
identity even without any publication.

Reported-by: Haiying Tang
Author: Hou Zhijie and Vignesh C
Reviewed-by: Vignesh C and Amit Kapila
Backpatch-through: 13
Discussion: https://postgr.es/m/OS0PR01MB6113D77F583C922F1CEAA1C3FBD29@OS0PR01MB6113.jpnprd01.prod.outlook.com
This commit is contained in:
Amit Kapila
2021-09-22 08:24:20 +05:30
parent 583e15af99
commit f09a81f1c0
6 changed files with 116 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
src/backend/commands/publicationcmds.c
View File

@@ -45,9 +45,6 @@
#include "utils/syscache.h"
#include "utils/varlena.h"
/* Same as MAXNUMMESSAGES in sinvaladt.c */
#define MAX_RELCACHE_INVAL_MSGS 4096
static List *OpenTableList(List *tables);
static void CloseTableList(List *rels);
static void PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
@@ -330,23 +327,7 @@ AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel,
List *relids = GetPublicationRelations(pubform->oid,
PUBLICATION_PART_ALL);
/*
* 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)
{
Oid relid = lfirst_oid(lc);
CacheInvalidateRelcacheByRelid(relid);
}
}
else
CacheInvalidateRelcacheAll();
InvalidatePublicationRels(relids);
}
ObjectAddressSet(obj, PublicationRelationId, pubform->oid);
@@ -356,6 +337,27 @@ AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel,
InvokeObjectPostAlterHook(PublicationRelationId, pubform->oid, 0);
}
/*
* Invalidate the relations.
*/
void
InvalidatePublicationRels(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)
CacheInvalidateRelcacheByRelid(lfirst_oid(lc));
}
else
CacheInvalidateRelcacheAll();
}
/*
* Add or remove table to/from publication.
*/
@@ -512,6 +514,7 @@ RemovePublicationRelById(Oid proid)
Relation rel;
HeapTuple tup;
Form_pg_publication_rel pubrel;
List *relids = NIL;
rel = table_open(PublicationRelRelationId, RowExclusiveLock);
@@ -523,8 +526,18 @@ RemovePublicationRelById(Oid proid)
pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
/* Invalidate relcache so that publication info is rebuilt. */
CacheInvalidateRelcacheByRelid(pubrel->prrelid);
/*
* Invalidate relcache so that publication info is rebuilt.
*
* For the partitioned tables, we must invalidate all partitions contained
* in the respective partition hierarchies, not just the one explicitly
* mentioned in the publication. This is required because we implicitly
* publish the child tables when the parent table is published.
*/
relids = GetPubPartitionOptionRelations(relids, PUBLICATION_PART_ALL,
pubrel->prrelid);
InvalidatePublicationRels(relids);
CatalogTupleDelete(rel, &tup->t_self);