public inbox for [email protected]
help / color / mirror / Atom feedFrom: Jacob Champion <[email protected]>
To: PostgreSQL Hackers <[email protected]>
Cc: Aleksander Alekseev <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Amit Kapila <[email protected]>
Subject: Re: RFC: logical publication via inheritance root?
Date: Thu, 29 Jun 2023 16:46:44 -0700
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <CAAWbhmiRRJDzbA4fv0vAViu_wfoeYqtT+VYOOu7845WO-yT35w@mail.gmail.com>
<CAAWbhmiQWEGv73En15WRJ5PmhbwwTdBxMiPxunGghW0G-WVAYg@mail.gmail.com>
<CAJ7c6TO+6hs1sMJ5-Qs7DC5HyhhN+hkLU0D8Ci_F-5ruBaHNog@mail.gmail.com>
<CAAWbhmha62Vh9MtsJ3zxRKRQiVSoyxb6sKHzLCRWsTzqwM0bbg@mail.gmail.com>
<[email protected]>
<[email protected]>
<CAHut+PuJq9s8QZ2pn2jKERWr3wa2ABT6McOxR=F-1+cJHtKSiQ@mail.gmail.com>
<CAAWbhmi_sTjbSd-RYSC08DHNbXy9QG1FoHT+qGYxNwoO+MSSDg@mail.gmail.com>
<[email protected]>
On 6/6/23 08:50, Jacob Champion wrote:
> Commit 062a84442 necessitated some rework of the new
> pg_get_publication_rels_to_sync() helper. It now takes a list of
> publications so that we can handle conflicts in the pubviaroot settings.
> This is more complicated than before -- unlike partitions, standard
> inheritance trees can selectively publish tables that aren't leaves. But
> I think I've finally settled on some semantics for it which are
> unsurprising.
The semantics I've picked are wrong :( I've accidentally reintroduced a
bug that's similar to the one discussed upthread, where if you have two
leaf tables published through different roots:
tree pub1 pub2
----- ----- -----
A - A
B C B - - -
D D D
then a subscription on both publications will duplicate the data in the
leaf (D in the above example). I need to choose semantics that are
closer to the current behavior of partitions.
> I wonder if pg_set_logical_root() might be better implemented as part of
> ALTER TABLE. Maybe with a relation option? If it all went through ALTER
> TABLE ONLY ... SET, then we wouldn't have to worry about a user
> modifying roots while reading pg_get_publication_rels_to_sync() in the
> same query. The permissions checks should be more consistent with less
> effort, and there's an existing way to set/clear the option that already
> plays well with pg_dump and pg_upgrade.
I've implemented ALTER TABLE in v3; I like it a lot more. The new
reloption is named publish_via_parent. So now we can unset the flag and
dump/restore.
v3 also fixes a nasty uninitialized stack variable, along with a bad
collation assumption I made.
> The downsides I can see are the
> need to handle simultaneous changes to INHERIT and SET (since we'd be
> manipulating pg_inherits in both),
(This didn't turn out to be as bad as I feared.)
> as well as the fact that ALTER TABLE
> ... SET defaults to altering the entire table hierarchy, which may be
> bad UX for this case.
This was just wrong -- ALTER TABLE ... SET does not recurse. I think the
docs are misleading here, and I'm not sure why we allow an explicit '*'
after a table name if we're going to ignore it. But I also don't really
want to poke at that in this patchset, if I can avoid it.
--Jacob
1: 761b91b3f1 = 1: 7e68f96af6 pgoutput: refactor publication cache construction
2: 0397a52583 ! 2: a48563919e WIP: introduce pg_set_logical_root for use with pubviaroot
@@ Metadata
Author: Jacob Champion <[email protected]>
## Commit message ##
- WIP: introduce pg_set_logical_root for use with pubviaroot
+ WIP: introduce publish_via_parent for logical replication
- Allows regular inherited tables to be published via their root table,
- just like partitions. This works by hijacking pg_inherit's inhseqno
- column, and replacing a (single) existing entry for the child with the
- value zero, indicating that it should be treated as a logical partition
- by the publication machinery.
+ This new relation option allows regular inherited tables to be published
+ via their root table, just like partitions. This works by hijacking
+ pg_inherit's inhseqno column, and replacing a (single) existing entry
+ for the child with the value zero, indicating that it should be treated
+ as a logical partition by the publication machinery.
Initial sync works by asking the publisher for a list of logical
descendants of the published table, then COPYing them one-by-one into
@@ Commit message
roots.
Known bugs/TODOs:
- - The pg_inherits machinery doesn't prohibit changes to inheritance
- after an entry has been marked as a logical root.
+ - If two publications publish the same leaves of a table hierarchy, but
+ have different roots, the shared leaf tables will be incorrectly
+ duplicated when subscribing to both publications simultaneously.
- I haven't given any thought to interactions with row filters, or to
column lists.
- I'm not sure that I'm taking all the necessary locks yet, and those I
do take may be taken in the wrong order.
- - Dump and upgrade aren't supported yet.
+
+ ## src/backend/access/common/reloptions.c ##
+@@ src/backend/access/common/reloptions.c: static relopt_bool boolRelOpts[] =
+ },
+ false
+ },
++ {
++ {
++ "publish_via_parent",
++ "Replicate this table's contents via its parent table when publishing with publish_via_partition_root",
++ RELOPT_KIND_HEAP,
++ AccessExclusiveLock
++ },
++ false
++ },
+ {
+ {
+ "fastupdate",
+@@ src/backend/access/common/reloptions.c: default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
+ offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, analyze_scale_factor)},
+ {"user_catalog_table", RELOPT_TYPE_BOOL,
+ offsetof(StdRdOptions, user_catalog_table)},
++ {"publish_via_parent", RELOPT_TYPE_BOOL,
++ offsetof(StdRdOptions, publish_via_parent)},
+ {"parallel_workers", RELOPT_TYPE_INT,
+ offsetof(StdRdOptions, parallel_workers)},
+ {"vacuum_index_cleanup", RELOPT_TYPE_ENUM,
## src/backend/catalog/pg_inherits.c ##
@@
@@ src/backend/catalog/pg_inherits.c: find_inheritance_children(Oid parentrelId, LO
* committed to the active snapshot. In addition, *detached_xmin (if not null)
* is set to the xmin of the row of the detached partition.
+ *
-+ * If logical_only is true, only tables marked explicitly via
-+ * pg_set_logical_root() are included in the output list.
++ * If logical_only is true, only tables marked explicitly via publish_via_parent
++ * are included in the output list.
*/
List *
find_inheritance_children_extended(Oid parentrelId, bool omit_detached,
@@ src/backend/catalog/pg_inherits.c: PartitionHasPendingDetach(Oid partoid)
+has_logical_parent(Relation inhRel, Oid relid)
+{
+ return (get_logical_parent_worker(inhRel, relid) != InvalidOid);
-+}
-+
-+Datum
-+pg_set_logical_root(PG_FUNCTION_ARGS)
-+{
-+ Oid tableoid = PG_GETARG_OID(0);
-+ Oid rootoid = PG_GETARG_OID(1);
-+ char *tablename;
-+ char *rootname;
-+ Relation inhRel;
-+ ScanKeyData key;
-+ SysScanDesc scan;
-+ Oid parent = InvalidOid;
-+ HeapTuple tuple, copyTuple;
-+ Form_pg_inherits form;
-+
-+ /*
-+ * Check that the tables exist.
-+ * TODO: check identical schemas too? or does replication handle that?
-+ */
-+ tablename = get_rel_name(tableoid);
-+ if (tablename == NULL)
-+ ereport(ERROR,
-+ (errcode(ERRCODE_UNDEFINED_TABLE),
-+ errmsg("OID %u does not refer to a table", tableoid)));
-+ rootname = get_rel_name(rootoid);
-+ if (rootname == NULL)
-+ ereport(ERROR,
-+ (errcode(ERRCODE_UNDEFINED_TABLE),
-+ errmsg("OID %u does not refer to a table", rootoid)));
-+
-+ /* Check ownership. */
-+ if (!object_ownercheck(RelationRelationId, tableoid, GetUserId()))
-+ aclcheck_error(ACLCHECK_NOT_OWNER,
-+ get_relkind_objtype(get_rel_relkind(tableoid)),
-+ tablename);
-+ if (!object_ownercheck(RelationRelationId, rootoid, GetUserId()))
-+ aclcheck_error(ACLCHECK_NOT_OWNER,
-+ get_relkind_objtype(get_rel_relkind(rootoid)),
-+ rootname);
-+
-+ /* Open pg_inherits with RowExclusiveLock so that we can update it. */
-+ inhRel = table_open(InheritsRelationId, RowExclusiveLock);
-+
-+ /*
-+ * We have to make sure that the inheritance relationship already exists,
-+ * and that there is only one existing parent for this table.
-+ *
-+ * TODO: do we have to lock the tables themselves to avoid races?
-+ */
-+ ScanKeyInit(&key,
-+ Anum_pg_inherits_inhrelid,
-+ BTEqualStrategyNumber, F_OIDEQ,
-+ ObjectIdGetDatum(tableoid));
-+
-+ scan = systable_beginscan(inhRel, InheritsRelidSeqnoIndexId, true,
-+ NULL, 1, &key);
-+ tuple = systable_getnext(scan);
-+ if (HeapTupleIsValid(tuple))
-+ {
-+ form = (Form_pg_inherits) GETSTRUCT(tuple);
-+ parent = form->inhparent;
-+ copyTuple = heap_copytuple(tuple);
-+
-+ if (HeapTupleIsValid(systable_getnext(scan)))
-+ ereport(ERROR,
-+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-+ errmsg("table \"%s\" inherits from multiple tables",
-+ tablename)));
-+ }
-+
-+ if (parent != rootoid)
-+ ereport(ERROR,
-+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-+ errmsg("table \"%s\" does not inherit from intended root table \"%s\"",
-+ tablename, rootname)));
-+
-+ systable_endscan(scan);
-+
-+ /* Mark the inheritance as a logical root by setting it to zero. */
-+ form = (Form_pg_inherits) GETSTRUCT(copyTuple);
-+ form->inhseqno = 0;
-+
-+ CatalogTupleUpdate(inhRel, ©Tuple->t_self, copyTuple);
-+
-+ heap_freetuple(copyTuple);
-+ table_close(inhRel, RowExclusiveLock);
-+
-+ PG_RETURN_VOID();
+}
## src/backend/catalog/pg_publication.c ##
@@ src/backend/catalog/pg_publication.c: pg_get_publication_tables(PG_FUNCTION_ARGS
+ * The only time this list consists of anything more than the table itself is
+ * when a publication's publish_via_partition_root is set to true and the table
+ * has inherited child tables in the publications that have been marked with
-+ * pg_set_logical_root().
++ * publish_via_parent.
+ */
+Datum
+pg_get_publication_rels_to_sync(PG_FUNCTION_ARGS)
@@ src/backend/catalog/pg_publication.c: process_relation_publications(Oid relid, c
- * For a partition, check if any of the ancestors are
- * published. If so, note down the topmost ancestor that is
+ * Check if any of the logical ancestors (that is, partition
-+ * parents or tables marked with pg_set_logical_root()) are
++ * parents or tables marked with publish_via_parent) are
+ * published. If so, note down the topmost ancestor that is
* published via this publication, which will be used as the
- * relation via which to publish the partition's changes.
@@ src/backend/catalog/pg_publication.c: process_relation_publications(Oid relid, c
return rel_publications;
}
+@@ src/backend/catalog/pg_publication.c: pg_get_relation_publishing_info(PG_FUNCTION_ARGS)
+
+ {
+ List *rel_publications;
+- PublicationActions pubactions;
++ PublicationActions pubactions = {0};
+ Oid publish_as_relid;
+ ListCell *lc;
+ Datum *puboids;
## src/backend/commands/publicationcmds.c ##
@@ src/backend/commands/publicationcmds.c: contain_invalid_rfcolumn_walker(Node *node, rf_context *context)
@@ src/backend/commands/publicationcmds.c: contain_invalid_rfcolumn_walker(Node *no
* information of the child table. So, get the column number of the
* child table as parent and child column order could be different.
+ *
-+ * TODO: is this applicable to pg_set_logical_root()?
++ * TODO: is this applicable to publish_via_parent?
*/
if (context->pubviaroot)
{
@@ src/backend/commands/publicationcmds.c: pub_rf_contains_invalid_column(Oid pubid
* Note that even though the row filter used is for an ancestor, the
* REPLICA IDENTITY used will be for the actual child table.
+ *
-+ * TODO: is this applicable to pg_set_logical_root()?
++ * TODO: is this applicable to publish_via_parent?
*/
if (pubviaroot && relation->rd_rel->relispartition)
{
@@ src/backend/commands/publicationcmds.c: pub_rf_contains_invalid_column(Oid pubid
*
* Returns true if any replica identity column is not covered by column list.
+ *
-+ * TODO: pg_set_logical_root()?
++ * TODO: publish_via_parent?
*/
bool
pub_collist_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
@@ src/backend/commands/publicationcmds.c: TransformPubWhereClauses(List *tables, c
* table, the partition's row filter will be used. So disallow using
* WHERE clause on partitioned table in this case.
+ *
-+ * TODO: decide how this interacts with pg_set_logical_root
++ * TODO: decide how this interacts with publish_via_parent
*/
if (!pubviaroot &&
pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
@@ src/backend/commands/publicationcmds.c: CheckPubRelationColumnList(char *pubname
* table, the partition's column list will be used. So disallow using
* a column list on the partitioned table in this case.
+ *
-+ * TODO: decide if this interacts with pg_set_logical_root()
++ * TODO: decide if this interacts with publish_via_parent
*/
if (!pubviaroot &&
pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+ ## src/backend/commands/tablecmds.c ##
+@@ src/backend/commands/tablecmds.c: static bool MergeCheckConstraint(List *constraints, char *name, Node *expr);
+ static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel);
+ static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel);
+ static void StoreCatalogInheritance(Oid relationId, List *supers,
+- bool child_is_partition);
++ bool child_is_partition,
++ bool publish_via_parent);
+ static void StoreCatalogInheritance1(Oid relationId, Oid parentOid,
+ int32 seqNumber, Relation inhRelation,
+ bool child_is_partition);
+@@ src/backend/commands/tablecmds.c: static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
+ const char *tablespacename, LOCKMODE lockmode);
+ static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
+ static void ATExecSetTableSpaceNoStorage(Relation rel, Oid newTableSpace);
++static void change_publish_via_parent(Relation rel, bool set);
+ static void ATExecSetRelOptions(Relation rel, List *defList,
+ AlterTableType operation,
+ LOCKMODE lockmode);
+@@ src/backend/commands/tablecmds.c: DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
+ }
+
+ /* Store inheritance information for new rel. */
+- StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL);
++ StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL,
++ RelationIsPublishedViaParent(rel));
+
+ /*
+ * Process the partitioning specification (if any) and store the partition
+@@ src/backend/commands/tablecmds.c: MergeCheckConstraint(List *constraints, char *name, Node *expr)
+ */
+ static void
+ StoreCatalogInheritance(Oid relationId, List *supers,
+- bool child_is_partition)
++ bool child_is_partition, bool publish_via_parent)
+ {
+ Relation relation;
+ int32 seqNumber;
+@@ src/backend/commands/tablecmds.c: StoreCatalogInheritance(Oid relationId, List *supers,
+ */
+ Assert(OidIsValid(relationId));
+
++ /*
++ * publish_via_parent requires exactly one parent. Try to keep this
++ * messaging consistent with ALTER TABLE/change_publish_via_parent().
++ *
++ * Earlier, in MergeAttributes, we already checked that we had ownership of
++ * the parent table, so we don't need to check that again the way that
++ * change_publish_via_parent() does.
++ */
++ if (publish_via_parent && list_length(supers) != 1)
++ ereport(ERROR,
++ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
++ errmsg((list_length(supers) == 0)
++ ? "table \"%s\" does not inherit from any tables"
++ : "table \"%s\" inherits from multiple tables",
++ get_rel_name(relationId))));
++
+ if (supers == NIL)
+ return;
+
+@@ src/backend/commands/tablecmds.c: StoreCatalogInheritance(Oid relationId, List *supers,
+ */
+ relation = table_open(InheritsRelationId, RowExclusiveLock);
+
+- seqNumber = 1;
++ /*
++ * Normally inhseqno starts at one, but for publish_via_parent, it's given a
++ * sentinel value of zero. (In that case, we'll only go through this loop
++ * once, as the list_length() check above enforces.)
++ */
++ seqNumber = publish_via_parent ? 0 : 1;
+ foreach(entry, supers)
+ {
+ Oid parentOid = lfirst_oid(entry);
+@@ src/backend/commands/tablecmds.c: ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, const char *tablespacen
+ tab->newTableSpace = tablespaceId;
+ }
+
++static void
++change_publish_via_parent(Relation rel, bool set)
++{
++ Relation inhRel;
++ ScanKeyData key;
++ SysScanDesc scan;
++ Oid parentOid;
++ Relation parentRel;
++ HeapTuple tuple, copyTuple;
++ Form_pg_inherits form;
++
++ /* Open pg_inherits with RowExclusiveLock so that we can update it. */
++ inhRel = table_open(InheritsRelationId, RowExclusiveLock);
++
++ /*
++ * We have to make sure that an inheritance relationship already exists, and
++ * that there is only one candidate parent for this table.
++ *
++ * TODO: do we have to lock the tables themselves to avoid races?
++ */
++ ScanKeyInit(&key,
++ Anum_pg_inherits_inhrelid,
++ BTEqualStrategyNumber, F_OIDEQ,
++ ObjectIdGetDatum(RelationGetRelid(rel)));
++
++ scan = systable_beginscan(inhRel, InheritsRelidSeqnoIndexId, true,
++ NULL, 1, &key);
++ tuple = systable_getnext(scan);
++
++ if (!HeapTupleIsValid(tuple))
++ ereport(ERROR,
++ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
++ errmsg("table \"%s\" does not inherit from any tables",
++ RelationGetRelationName(rel))));
++
++ form = (Form_pg_inherits) GETSTRUCT(tuple);
++ parentOid = form->inhparent;
++ copyTuple = heap_copytuple(tuple);
++
++ if (HeapTupleIsValid(systable_getnext(scan)))
++ ereport(ERROR,
++ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
++ errmsg("table \"%s\" inherits from multiple tables",
++ RelationGetRelationName(rel))));
++
++ systable_endscan(scan);
++
++ if (set)
++ {
++ /*
++ * Open our parent table and check permissions. Like ATTACH PARTITION,
++ * we require ownership of the parent table in addition to the child
++ * (which has already been checked in ATPrepCmd).
++ *
++ * TODO: what lock strength is needed here? Is nesting it inside the
++ * inhRel lock a risk?
++ */
++ parentRel = table_open(parentOid, AccessExclusiveLock);
++ ATSimplePermissions(AT_SetRelOptions, parentRel, ATT_TABLE);
++ table_close(parentRel, NoLock); /* TODO: do we need to hang onto the lock? */
++ }
++ else
++ {
++ /*
++ * Ownership of the parent isn't needed for unsetting the flag, just
++ * like we wouldn't check ownership during NO INHERIT.
++ */
++ }
++
++ /*
++ * Mark the inheritance as a logical root by setting inhseqno to zero.
++ * Unmark it by setting inhseqno to one (since there's only one parent).
++ */
++ form = (Form_pg_inherits) GETSTRUCT(copyTuple);
++ form->inhseqno = set ? 0 : 1;
++
++ CatalogTupleUpdate(inhRel, ©Tuple->t_self, copyTuple);
++
++ heap_freetuple(copyTuple);
++ table_close(inhRel, RowExclusiveLock);
++}
++
+ /*
+ * Set, reset, or replace reloptions.
+ */
+@@ src/backend/commands/tablecmds.c: ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
+ }
+ }
+
++ /* Extra work for publish_via_parent */
++ if (rel->rd_rel->relkind == RELKIND_RELATION)
++ {
++ List *heap_options = untransformRelOptions(newOptions);
++ ListCell *cell;
++ bool curSetting = RelationIsPublishedViaParent(rel);
++ DefElem *pubElem = NULL;
++
++ foreach (cell, heap_options)
++ {
++ DefElem *defel = lfirst_node(DefElem, cell);
++
++ if (strcmp(defel->defname, "publish_via_parent") == 0)
++ {
++ pubElem = defel;
++ break;
++ }
++ }
++
++ if (pubElem)
++ {
++ bool newSetting = defGetBoolean(pubElem);
++
++ if (newSetting != curSetting)
++ change_publish_via_parent(rel, newSetting);
++ }
++ else if (curSetting)
++ {
++ /* publish_via_parent was RESET */
++ change_publish_via_parent(rel, false);
++ }
++ }
++
+ /*
+ * All we need do here is update the pg_class row; the new options will be
+ * propagated into relcaches during post-commit cache inval.
+@@ src/backend/commands/tablecmds.c: ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode)
+ trigger_name, RelationGetRelationName(child_rel)),
+ errdetail("ROW triggers with transition tables are not supported in inheritance hierarchies.")));
+
++ /* publish_via_parent tables are not allowed to add additional parents. */
++ if (RelationIsPublishedViaParent(child_rel))
++ ereport(ERROR,
++ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
++ errmsg("relation option \"%s\" prevents table \"%s\" from changing inheritance",
++ "publish_via_parent", RelationGetRelationName(child_rel))));
++
+ /* OK to create inheritance */
+ CreateInheritance(child_rel, parent_rel);
+
+@@ src/backend/commands/tablecmds.c: ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot change inheritance of a partition")));
+
++ /* publish_via_parent tables are not allowed to remove their parent. */
++ if (RelationIsPublishedViaParent(rel))
++ ereport(ERROR,
++ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
++ errmsg("relation option \"%s\" prevents table \"%s\" from changing inheritance",
++ "publish_via_parent", RelationGetRelationName(rel))));
++
+ /*
+ * AccessShareLock on the parent is probably enough, seeing that DROP
+ * TABLE doesn't lock parent tables at all. We need some lock since we'll
+
## src/backend/partitioning/partdesc.c ##
@@ src/backend/partitioning/partdesc.c: RelationBuildPartitionDesc(Relation rel, bool omit_detached)
inhoids = find_inheritance_children_extended(RelationGetRelid(rel),
@@ src/backend/replication/logical/tablesync.c: fetch_remote_table_info(char *nspna
+ /*
+ * See if there are any other tables to be copied besides the original. This
+ * happens when a descendant in the inheritance relationship is marked with
-+ * pg_set_logical_root() and is part of a publication with
++ * publish_via_parent and is part of a publication with
+ * publish_via_partition_root = true.
+ *
+ * FIXME: if two publications have conflicting settings for
@@ src/backend/replication/pgoutput/pgoutput.c: pgoutput_change(LogicalDecodingCont
targetrel = ancestor;
}
+ ## src/bin/pg_dump/t/002_pg_dump.pl ##
+@@ src/bin/pg_dump/t/002_pg_dump.pl: my %tests = (
+ no_table_access_method => 1,
+ only_dump_measurement => 1,
+ },
++ },
++
++ 'CREATE TABLE regress_pg_dump_publish_via_parent' => {
++ create_order => 3,
++ create_sql => '
++ CREATE TABLE dump_test.regress_pg_dump_publish_via_parent (col1 int);
++ CREATE TABLE dump_test.regress_pg_dump_publish_via_parent_1() INHERITS (dump_test.regress_pg_dump_publish_via_parent) WITH (publish_via_parent);',
++ regexp => qr/^
++ \QCREATE TABLE dump_test.regress_pg_dump_publish_via_parent (\E
++ \n\s+\Qcol1 integer\E
++ \n\);
++ (\n.*)+?
++ \n\QCREATE TABLE dump_test.regress_pg_dump_publish_via_parent_1 (\E
++ \n\)
++ \n\QINHERITS (dump_test.regress_pg_dump_publish_via_parent)\E
++ \n\QWITH (publish_via_parent='true')\E;/xm,
++ like => {
++ %full_runs, %dump_test_schema_runs, section_pre_data => 1,
++ },
++ unlike => {
++ binary_upgrade => 1,
++ exclude_dump_test_schema => 1,
++ only_dump_measurement => 1,
++ },
+ });
+
+ #########################################
+
+ ## src/bin/psql/tab-complete.c ##
+@@ src/bin/psql/tab-complete.c: static const char *const table_storage_parameters[] = {
+ "fillfactor",
+ "log_autovacuum_min_duration",
+ "parallel_workers",
++ "publish_via_parent",
+ "toast.autovacuum_enabled",
+ "toast.autovacuum_freeze_max_age",
+ "toast.autovacuum_freeze_min_age",
+
## src/include/catalog/pg_inherits.h ##
+@@
+
+ #include "nodes/pg_list.h"
+ #include "storage/lock.h"
++#include "utils/relcache.h"
+
+ /* ----------------
+ * pg_inherits definition. cpp turns this into
@@ src/include/catalog/pg_inherits.h: typedef FormData_pg_inherits *Form_pg_inherits;
DECLARE_UNIQUE_INDEX_PKEY(pg_inherits_relid_seqno_index, 2680, InheritsRelidSeqnoIndexId, on pg_inherits using btree(inhrelid oid_ops, inhseqno int4_ops));
@@ src/include/catalog/pg_proc.dat
{ oid => '8139',
descr => 'get information on how a relation will be published via a list of publications',
proname => 'pg_get_relation_publishing_info', provariadic => 'text',
-@@
- proname => 'pg_partition_root', prorettype => 'regclass',
- proargtypes => 'regclass', prosrc => 'pg_partition_root' },
+
+ ## src/include/utils/rel.h ##
+@@ src/include/utils/rel.h: typedef struct StdRdOptions
+ int toast_tuple_target; /* target for tuple toasting */
+ AutoVacOpts autovacuum; /* autovacuum-related options */
+ bool user_catalog_table; /* use as an additional catalog relation */
++ bool publish_via_parent; /* publish via parent's relid for logrep */
+ int parallel_workers; /* max number of parallel workers */
+ StdRdOptIndexCleanup vacuum_index_cleanup; /* controls index vacuuming */
+ bool vacuum_truncate; /* enables vacuum to truncate a relation */
+@@ src/include/utils/rel.h: typedef struct StdRdOptions
+ (relation)->rd_rel->relkind == RELKIND_MATVIEW) ? \
+ ((StdRdOptions *) (relation)->rd_options)->user_catalog_table : false)
-+{ oid => '8136', descr => 'mark a table root for logical replication',
-+ proname => 'pg_set_logical_root', provolatile => 'v', proparallel => 'u',
-+ prorettype => 'void', proargtypes => 'regclass regclass',
-+ prosrc => 'pg_set_logical_root' },
++/*
++ * RelationIsPublishedViaParent
++ * Returns whether the relation's contents should be logically replicated
++ * via its parent table's relid, when published under the effects of
++ * publish_via_partition_root. Note multiple eval of argument!
++ */
++#define RelationIsPublishedViaParent(relation) \
++ ((relation)->rd_options && \
++ ((relation)->rd_rel->relkind == RELKIND_RELATION) && \
++ ((StdRdOptions *) (relation)->rd_options)->publish_via_parent)
+
- { oid => '4350', descr => 'Unicode normalization',
- proname => 'normalize', prorettype => 'text', proargtypes => 'text text',
- prosrc => 'unicode_normalize_func' },
+ /*
+ * RelationGetParallelWorkers
+ * Returns the relation's parallel_workers reloption setting.
## src/test/regress/expected/publication.out ##
@@ src/test/regress/expected/publication.out: CREATE ROLE regress_publication_user LOGIN SUPERUSER;
@@ src/test/regress/expected/publication.out: SELECT * FROM pg_publication_tables;
pub | sch1 | tbl1 | {a} |
(1 row)
-+-- Sanity check cases for pg_set_logical_root().
++-- Sanity check cases for publish_via_parent.
+CREATE TABLE sch1.iroot (a int);
+CREATE TABLE sch1.ipart1 (a int);
+CREATE TABLE sch1.ipart2 () INHERITS (sch1.iroot);
++-- should do nothing at all
++ALTER TABLE sch1.iroot SET (publish_via_parent = false);
++ALTER TABLE sch1.iroot RESET (publish_via_parent);
+-- marking roots between unrelated tables is not allowed
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
-+ERROR: table "ipart1" does not inherit from intended root table "iroot"
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.tbl1');
-+ERROR: table "ipart2" does not inherit from intended root table "tbl1"
++ALTER TABLE sch1.ipart1 SET (publish_via_parent);
++ERROR: table "ipart1" does not inherit from any tables
++CREATE TABLE fail (a int) WITH (publish_via_parent);
++ERROR: table "fail" does not inherit from any tables
+-- establishing an inheritance relationship fixes the problem
-+ALTER TABLE sch1.ipart1 INHERIT sch1.iroot;
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
++ALTER TABLE sch1.ipart1 INHERIT sch1.iroot,
++ SET (publish_via_parent);
+-- but multiple inheritance is not allowed
-+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1,
++ SET (publish_via_parent);
+ERROR: table "ipart2" inherits from multiple tables
-+ALTER TABLE sch1.ipart2 NO INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
++-- once publish_via_parent is set, inheritance cannot be changed
++ALTER TABLE sch1.ipart2 SET (publish_via_parent);
++ALTER TABLE sch1.ipart2 NO INHERIT sch1.iroot;
++ERROR: relation option "publish_via_parent" prevents table "ipart2" from changing inheritance
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
++ERROR: relation option "publish_via_parent" prevents table "ipart2" from changing inheritance
+-- table ownership must match, like ATTACH PARTITION
+CREATE ROLE regress_test_me;
+CREATE ROLE regress_test_not_me;
@@ src/test/regress/expected/publication.out: SELECT * FROM pg_publication_tables;
+ALTER TABLE root OWNER TO regress_test_me;
+ALTER TABLE part OWNER TO regress_test_not_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should fail
++ALTER TABLE part SET (publish_via_parent); -- should fail
+ERROR: must be owner of table part
+RESET SESSION AUTHORIZATION;
+ALTER TABLE root OWNER TO regress_test_not_me;
+ALTER TABLE part OWNER TO regress_test_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should also fail
++ALTER TABLE part SET (publish_via_parent); -- should also fail
++ERROR: must be owner of table root
++CREATE TABLE fail () INHERITS (root) WITH (publish_via_parent);
+ERROR: must be owner of table root
+RESET SESSION AUTHORIZATION;
+DROP TABLE root, part;
+DROP ROLE regress_test_not_me;
+DROP ROLE regress_test_me;
-+-- TODO: make sure existing logical descendant can't be ALTERed [NO] INHERIT
+-- Mixed publication settings for publish_via_partition_root, at different
+-- levels of the inheritance tree, to pin correct behavior in the worst cases.
-+CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1);
++CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1) WITH (publish_via_parent);
+CREATE TABLE sch1.ipart1_a1 () INHERITS (sch1.ipart1_a);
-+SELECT pg_set_logical_root('sch1.ipart1_a', 'sch1.ipart1');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
-+SELECT pg_set_logical_root('sch1.ipart1_a1', 'sch1.ipart1_a');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
++ALTER TABLE sch1.ipart1_a1 SET (publish_via_parent);
+CREATE PUBLICATION ipub_root FOR TABLE sch1.iroot;
+CREATE PUBLICATION ipub_part1 FOR TABLE ONLY sch1.ipart1, ONLY sch1.ipart1_a1
+ WITH (publish_via_partition_root);
@@ src/test/regress/expected/publication.out: SELECT * FROM pg_publication_tables;
+ ipub_other | sch1.iroot | sch1.ipart1_a
+(10 rows)
+
++-- "Detaching" partitions should change the subscriptions
++ALTER TABLE sch1.ipart2 SET (publish_via_parent = false);
++SELECT pubname, relid::regclass
++ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
++ JOIN pg_publication p ON (p.oid = t.pubid);
++ pubname | relid
++------------+-------------
++ ipub_root | sch1.iroot
++ ipub_root | sch1.ipart1
++ ipub_root | sch1.ipart2
++ ipub_part1 | sch1.ipart1
++ ipub_other | sch1.iroot
++ ipub_other | sch1.ipart2
++(6 rows)
++
++SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
++ published | synced
++-------------+----------------
++ sch1.iroot | sch1.iroot
++ sch1.iroot | sch1.ipart1_a
++ sch1.ipart1 | sch1.ipart1
++ sch1.ipart1 | sch1.ipart1_a1
++ sch1.ipart2 | sch1.ipart2
++(5 rows)
++
++SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
++ pubname | published | synced
++------------+-------------+----------------
++ ipub_root | sch1.iroot | sch1.iroot
++ ipub_root | sch1.iroot | sch1.ipart1_a
++ ipub_root | sch1.ipart1 | sch1.ipart1
++ ipub_root | sch1.ipart1 | sch1.ipart1_a1
++ ipub_root | sch1.ipart2 | sch1.ipart2
++ ipub_part1 | sch1.ipart1 | sch1.ipart1
++ ipub_part1 | sch1.ipart1 | sch1.ipart1_a1
++ ipub_other | sch1.iroot | sch1.iroot
++ ipub_other | sch1.iroot | sch1.ipart1_a
++ ipub_other | sch1.ipart2 | sch1.ipart2
++(10 rows)
++
++ALTER TABLE sch1.ipart1_a1 RESET (publish_via_parent);
++SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
++ relid
++----------------
++ sch1.ipart1
++ sch1.ipart1_a1
++(2 rows)
++
++SELECT * FROM published_sync('ipub_part1');
++ published | synced
++----------------+----------------
++ sch1.ipart1 | sch1.ipart1
++ sch1.ipart1_a1 | sch1.ipart1_a1
++(2 rows)
++
++SELECT * FROM published_stream('ipub_part1');
++ pubname | published | synced
++------------+----------------+----------------
++ ipub_part1 | sch1.ipart1 | sch1.ipart1
++ ipub_part1 | sch1.ipart1_a1 | sch1.ipart1_a1
++(2 rows)
++
+DROP PUBLICATION ipub_other;
+DROP PUBLICATION ipub_part1;
+DROP PUBLICATION ipub_root;
@@ src/test/regress/sql/publication.sql: ALTER TABLE sch1.tbl1 ATTACH PARTITION sch
CREATE PUBLICATION pub FOR TABLES IN SCHEMA sch1 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
SELECT * FROM pg_publication_tables;
-+-- Sanity check cases for pg_set_logical_root().
++-- Sanity check cases for publish_via_parent.
+CREATE TABLE sch1.iroot (a int);
+CREATE TABLE sch1.ipart1 (a int);
+CREATE TABLE sch1.ipart2 () INHERITS (sch1.iroot);
+
++-- should do nothing at all
++ALTER TABLE sch1.iroot SET (publish_via_parent = false);
++ALTER TABLE sch1.iroot RESET (publish_via_parent);
++
+-- marking roots between unrelated tables is not allowed
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.tbl1');
++ALTER TABLE sch1.ipart1 SET (publish_via_parent);
++CREATE TABLE fail (a int) WITH (publish_via_parent);
+
+-- establishing an inheritance relationship fixes the problem
-+ALTER TABLE sch1.ipart1 INHERIT sch1.iroot;
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
++ALTER TABLE sch1.ipart1 INHERIT sch1.iroot,
++ SET (publish_via_parent);
+
+-- but multiple inheritance is not allowed
-+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1,
++ SET (publish_via_parent);
+
-+ALTER TABLE sch1.ipart2 NO INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
++-- once publish_via_parent is set, inheritance cannot be changed
++ALTER TABLE sch1.ipart2 SET (publish_via_parent);
++ALTER TABLE sch1.ipart2 NO INHERIT sch1.iroot;
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
+
+-- table ownership must match, like ATTACH PARTITION
+CREATE ROLE regress_test_me;
+CREATE ROLE regress_test_not_me;
+CREATE TABLE root (a int);
+CREATE TABLE part () INHERITS (root);
++
+ALTER TABLE root OWNER TO regress_test_me;
+ALTER TABLE part OWNER TO regress_test_not_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should fail
++ALTER TABLE part SET (publish_via_parent); -- should fail
+RESET SESSION AUTHORIZATION;
++
+ALTER TABLE root OWNER TO regress_test_not_me;
+ALTER TABLE part OWNER TO regress_test_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should also fail
++ALTER TABLE part SET (publish_via_parent); -- should also fail
++CREATE TABLE fail () INHERITS (root) WITH (publish_via_parent);
+RESET SESSION AUTHORIZATION;
++
+DROP TABLE root, part;
+DROP ROLE regress_test_not_me;
+DROP ROLE regress_test_me;
+
-+-- TODO: make sure existing logical descendant can't be ALTERed [NO] INHERIT
-+
+-- Mixed publication settings for publish_via_partition_root, at different
+-- levels of the inheritance tree, to pin correct behavior in the worst cases.
-+CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1);
++CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1) WITH (publish_via_parent);
+CREATE TABLE sch1.ipart1_a1 () INHERITS (sch1.ipart1_a);
-+
-+SELECT pg_set_logical_root('sch1.ipart1_a', 'sch1.ipart1');
-+SELECT pg_set_logical_root('sch1.ipart1_a1', 'sch1.ipart1_a');
++ALTER TABLE sch1.ipart1_a1 SET (publish_via_parent);
+
+CREATE PUBLICATION ipub_root FOR TABLE sch1.iroot;
+CREATE PUBLICATION ipub_part1 FOR TABLE ONLY sch1.ipart1, ONLY sch1.ipart1_a1
@@ src/test/regress/sql/publication.sql: ALTER TABLE sch1.tbl1 ATTACH PARTITION sch
+SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
+SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
+
++-- "Detaching" partitions should change the subscriptions
++ALTER TABLE sch1.ipart2 SET (publish_via_parent = false);
++SELECT pubname, relid::regclass
++ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
++ JOIN pg_publication p ON (p.oid = t.pubid);
++SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
++SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
++
++ALTER TABLE sch1.ipart1_a1 RESET (publish_via_parent);
++SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
++SELECT * FROM published_sync('ipub_part1');
++SELECT * FROM published_stream('ipub_part1');
++
+DROP PUBLICATION ipub_other;
+DROP PUBLICATION ipub_part1;
+DROP PUBLICATION ipub_root;
@@ src/test/subscription/t/013_partition.pl: $node_subscriber1->append_conf('postgr
$node_subscriber1->reload;
+# Make sure standard inheritance setups aren't broken by the new
-+# pg_set_logical_root() handling.
++# publish_via_parent handling.
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1 (a int, b text)");
+$node_subscriber2->safe_psql('postgres',
@@ src/test/subscription/t/013_partition.pl: $node_subscriber1->append_conf('postgr
+ "CREATE TABLE itab1_2 (CHECK (a = 2)) INHERITS (itab1)");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_1', 'itab1')");
++ "ALTER TABLE itab1_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_2', 'itab1')");
++ "ALTER TABLE itab1_2 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1 VALUES (0, 'itab1')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1_1 VALUES (1, 'itab1')");
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+ FOR EACH ROW EXECUTE FUNCTION itab1_trigger();");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_1', 'itab1')");
++ "ALTER TABLE itab1_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_2', 'itab1')");
++ "ALTER TABLE itab1_2 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab2 (a int, b text)");
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+ FOR EACH ROW EXECUTE FUNCTION itab2_trigger();");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab2_1', 'itab2')");
++ "ALTER TABLE itab2_1 SET (publish_via_parent = true)");
+
+# itab2_1 should be published using its own identity here, since its parent is
+# not included. itab1_1 should be published via its parent, itab1, without
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+ "CREATE TABLE itab3_1_1 () INHERITS (itab3_1)");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab3_1', 'itab3')");
++ "ALTER TABLE itab3_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab3_1_1', 'itab3_1')");
++ "ALTER TABLE itab3_1_1 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO itab3 VALUES (1, 'itab3')");
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+is($result, qq(), 'initial data routed for itab3_1_1 on subscriber 2');
+
+# make sure new data is also correctly routed to the roots
-+$node_publisher->safe_psql('postgres', "INSERT INTO itab3 VALUES (4, 'itab3-new')");
-+$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1 VALUES (4, 'itab3_1-new')");
-+$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1_1 VALUES (4, 'itab3_1_1-new')");
++$node_publisher->safe_psql('postgres', "INSERT INTO itab3 VALUES (4, 'itab3_new')");
++$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1 VALUES (4, 'itab3_1_new')");
++$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1_1 VALUES (4, 'itab3_1_1_new')");
+
+$node_publisher->wait_for_catchup('mixed');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3 ORDER BY 1, 2");
+is($result, qq(1|itab3
-+4|itab3-new), 'new data routed for itab3 on subscriber 2');
++4|itab3_new), 'new data routed for itab3 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1 ORDER BY 1, 2");
+is($result, qq(2|itab3_1
+3|itab3_1_1
-+4|itab3_1-new
-+4|itab3_1_1-new), 'new data routed for itab3_1 on subscriber 2');
++4|itab3_1_1_new
++4|itab3_1_new), 'new data routed for itab3_1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1_1");
Attachments:
[text/plain] since-v2.diff.txt (41.4K, ../[email protected]/2-since-v2.diff.txt)
download | inline:
1: 761b91b3f1 = 1: 7e68f96af6 pgoutput: refactor publication cache construction
2: 0397a52583 ! 2: a48563919e WIP: introduce pg_set_logical_root for use with pubviaroot
@@ Metadata
Author: Jacob Champion <[email protected]>
## Commit message ##
- WIP: introduce pg_set_logical_root for use with pubviaroot
+ WIP: introduce publish_via_parent for logical replication
- Allows regular inherited tables to be published via their root table,
- just like partitions. This works by hijacking pg_inherit's inhseqno
- column, and replacing a (single) existing entry for the child with the
- value zero, indicating that it should be treated as a logical partition
- by the publication machinery.
+ This new relation option allows regular inherited tables to be published
+ via their root table, just like partitions. This works by hijacking
+ pg_inherit's inhseqno column, and replacing a (single) existing entry
+ for the child with the value zero, indicating that it should be treated
+ as a logical partition by the publication machinery.
Initial sync works by asking the publisher for a list of logical
descendants of the published table, then COPYing them one-by-one into
@@ Commit message
roots.
Known bugs/TODOs:
- - The pg_inherits machinery doesn't prohibit changes to inheritance
- after an entry has been marked as a logical root.
+ - If two publications publish the same leaves of a table hierarchy, but
+ have different roots, the shared leaf tables will be incorrectly
+ duplicated when subscribing to both publications simultaneously.
- I haven't given any thought to interactions with row filters, or to
column lists.
- I'm not sure that I'm taking all the necessary locks yet, and those I
do take may be taken in the wrong order.
- - Dump and upgrade aren't supported yet.
+
+ ## src/backend/access/common/reloptions.c ##
+@@ src/backend/access/common/reloptions.c: static relopt_bool boolRelOpts[] =
+ },
+ false
+ },
++ {
++ {
++ "publish_via_parent",
++ "Replicate this table's contents via its parent table when publishing with publish_via_partition_root",
++ RELOPT_KIND_HEAP,
++ AccessExclusiveLock
++ },
++ false
++ },
+ {
+ {
+ "fastupdate",
+@@ src/backend/access/common/reloptions.c: default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
+ offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, analyze_scale_factor)},
+ {"user_catalog_table", RELOPT_TYPE_BOOL,
+ offsetof(StdRdOptions, user_catalog_table)},
++ {"publish_via_parent", RELOPT_TYPE_BOOL,
++ offsetof(StdRdOptions, publish_via_parent)},
+ {"parallel_workers", RELOPT_TYPE_INT,
+ offsetof(StdRdOptions, parallel_workers)},
+ {"vacuum_index_cleanup", RELOPT_TYPE_ENUM,
## src/backend/catalog/pg_inherits.c ##
@@
@@ src/backend/catalog/pg_inherits.c: find_inheritance_children(Oid parentrelId, LO
* committed to the active snapshot. In addition, *detached_xmin (if not null)
* is set to the xmin of the row of the detached partition.
+ *
-+ * If logical_only is true, only tables marked explicitly via
-+ * pg_set_logical_root() are included in the output list.
++ * If logical_only is true, only tables marked explicitly via publish_via_parent
++ * are included in the output list.
*/
List *
find_inheritance_children_extended(Oid parentrelId, bool omit_detached,
@@ src/backend/catalog/pg_inherits.c: PartitionHasPendingDetach(Oid partoid)
+has_logical_parent(Relation inhRel, Oid relid)
+{
+ return (get_logical_parent_worker(inhRel, relid) != InvalidOid);
-+}
-+
-+Datum
-+pg_set_logical_root(PG_FUNCTION_ARGS)
-+{
-+ Oid tableoid = PG_GETARG_OID(0);
-+ Oid rootoid = PG_GETARG_OID(1);
-+ char *tablename;
-+ char *rootname;
-+ Relation inhRel;
-+ ScanKeyData key;
-+ SysScanDesc scan;
-+ Oid parent = InvalidOid;
-+ HeapTuple tuple, copyTuple;
-+ Form_pg_inherits form;
-+
-+ /*
-+ * Check that the tables exist.
-+ * TODO: check identical schemas too? or does replication handle that?
-+ */
-+ tablename = get_rel_name(tableoid);
-+ if (tablename == NULL)
-+ ereport(ERROR,
-+ (errcode(ERRCODE_UNDEFINED_TABLE),
-+ errmsg("OID %u does not refer to a table", tableoid)));
-+ rootname = get_rel_name(rootoid);
-+ if (rootname == NULL)
-+ ereport(ERROR,
-+ (errcode(ERRCODE_UNDEFINED_TABLE),
-+ errmsg("OID %u does not refer to a table", rootoid)));
-+
-+ /* Check ownership. */
-+ if (!object_ownercheck(RelationRelationId, tableoid, GetUserId()))
-+ aclcheck_error(ACLCHECK_NOT_OWNER,
-+ get_relkind_objtype(get_rel_relkind(tableoid)),
-+ tablename);
-+ if (!object_ownercheck(RelationRelationId, rootoid, GetUserId()))
-+ aclcheck_error(ACLCHECK_NOT_OWNER,
-+ get_relkind_objtype(get_rel_relkind(rootoid)),
-+ rootname);
-+
-+ /* Open pg_inherits with RowExclusiveLock so that we can update it. */
-+ inhRel = table_open(InheritsRelationId, RowExclusiveLock);
-+
-+ /*
-+ * We have to make sure that the inheritance relationship already exists,
-+ * and that there is only one existing parent for this table.
-+ *
-+ * TODO: do we have to lock the tables themselves to avoid races?
-+ */
-+ ScanKeyInit(&key,
-+ Anum_pg_inherits_inhrelid,
-+ BTEqualStrategyNumber, F_OIDEQ,
-+ ObjectIdGetDatum(tableoid));
-+
-+ scan = systable_beginscan(inhRel, InheritsRelidSeqnoIndexId, true,
-+ NULL, 1, &key);
-+ tuple = systable_getnext(scan);
-+ if (HeapTupleIsValid(tuple))
-+ {
-+ form = (Form_pg_inherits) GETSTRUCT(tuple);
-+ parent = form->inhparent;
-+ copyTuple = heap_copytuple(tuple);
-+
-+ if (HeapTupleIsValid(systable_getnext(scan)))
-+ ereport(ERROR,
-+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-+ errmsg("table \"%s\" inherits from multiple tables",
-+ tablename)));
-+ }
-+
-+ if (parent != rootoid)
-+ ereport(ERROR,
-+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-+ errmsg("table \"%s\" does not inherit from intended root table \"%s\"",
-+ tablename, rootname)));
-+
-+ systable_endscan(scan);
-+
-+ /* Mark the inheritance as a logical root by setting it to zero. */
-+ form = (Form_pg_inherits) GETSTRUCT(copyTuple);
-+ form->inhseqno = 0;
-+
-+ CatalogTupleUpdate(inhRel, ©Tuple->t_self, copyTuple);
-+
-+ heap_freetuple(copyTuple);
-+ table_close(inhRel, RowExclusiveLock);
-+
-+ PG_RETURN_VOID();
+}
## src/backend/catalog/pg_publication.c ##
@@ src/backend/catalog/pg_publication.c: pg_get_publication_tables(PG_FUNCTION_ARGS
+ * The only time this list consists of anything more than the table itself is
+ * when a publication's publish_via_partition_root is set to true and the table
+ * has inherited child tables in the publications that have been marked with
-+ * pg_set_logical_root().
++ * publish_via_parent.
+ */
+Datum
+pg_get_publication_rels_to_sync(PG_FUNCTION_ARGS)
@@ src/backend/catalog/pg_publication.c: process_relation_publications(Oid relid, c
- * For a partition, check if any of the ancestors are
- * published. If so, note down the topmost ancestor that is
+ * Check if any of the logical ancestors (that is, partition
-+ * parents or tables marked with pg_set_logical_root()) are
++ * parents or tables marked with publish_via_parent) are
+ * published. If so, note down the topmost ancestor that is
* published via this publication, which will be used as the
- * relation via which to publish the partition's changes.
@@ src/backend/catalog/pg_publication.c: process_relation_publications(Oid relid, c
return rel_publications;
}
+@@ src/backend/catalog/pg_publication.c: pg_get_relation_publishing_info(PG_FUNCTION_ARGS)
+
+ {
+ List *rel_publications;
+- PublicationActions pubactions;
++ PublicationActions pubactions = {0};
+ Oid publish_as_relid;
+ ListCell *lc;
+ Datum *puboids;
## src/backend/commands/publicationcmds.c ##
@@ src/backend/commands/publicationcmds.c: contain_invalid_rfcolumn_walker(Node *node, rf_context *context)
@@ src/backend/commands/publicationcmds.c: contain_invalid_rfcolumn_walker(Node *no
* information of the child table. So, get the column number of the
* child table as parent and child column order could be different.
+ *
-+ * TODO: is this applicable to pg_set_logical_root()?
++ * TODO: is this applicable to publish_via_parent?
*/
if (context->pubviaroot)
{
@@ src/backend/commands/publicationcmds.c: pub_rf_contains_invalid_column(Oid pubid
* Note that even though the row filter used is for an ancestor, the
* REPLICA IDENTITY used will be for the actual child table.
+ *
-+ * TODO: is this applicable to pg_set_logical_root()?
++ * TODO: is this applicable to publish_via_parent?
*/
if (pubviaroot && relation->rd_rel->relispartition)
{
@@ src/backend/commands/publicationcmds.c: pub_rf_contains_invalid_column(Oid pubid
*
* Returns true if any replica identity column is not covered by column list.
+ *
-+ * TODO: pg_set_logical_root()?
++ * TODO: publish_via_parent?
*/
bool
pub_collist_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
@@ src/backend/commands/publicationcmds.c: TransformPubWhereClauses(List *tables, c
* table, the partition's row filter will be used. So disallow using
* WHERE clause on partitioned table in this case.
+ *
-+ * TODO: decide how this interacts with pg_set_logical_root
++ * TODO: decide how this interacts with publish_via_parent
*/
if (!pubviaroot &&
pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
@@ src/backend/commands/publicationcmds.c: CheckPubRelationColumnList(char *pubname
* table, the partition's column list will be used. So disallow using
* a column list on the partitioned table in this case.
+ *
-+ * TODO: decide if this interacts with pg_set_logical_root()
++ * TODO: decide if this interacts with publish_via_parent
*/
if (!pubviaroot &&
pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+ ## src/backend/commands/tablecmds.c ##
+@@ src/backend/commands/tablecmds.c: static bool MergeCheckConstraint(List *constraints, char *name, Node *expr);
+ static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel);
+ static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel);
+ static void StoreCatalogInheritance(Oid relationId, List *supers,
+- bool child_is_partition);
++ bool child_is_partition,
++ bool publish_via_parent);
+ static void StoreCatalogInheritance1(Oid relationId, Oid parentOid,
+ int32 seqNumber, Relation inhRelation,
+ bool child_is_partition);
+@@ src/backend/commands/tablecmds.c: static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
+ const char *tablespacename, LOCKMODE lockmode);
+ static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
+ static void ATExecSetTableSpaceNoStorage(Relation rel, Oid newTableSpace);
++static void change_publish_via_parent(Relation rel, bool set);
+ static void ATExecSetRelOptions(Relation rel, List *defList,
+ AlterTableType operation,
+ LOCKMODE lockmode);
+@@ src/backend/commands/tablecmds.c: DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
+ }
+
+ /* Store inheritance information for new rel. */
+- StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL);
++ StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL,
++ RelationIsPublishedViaParent(rel));
+
+ /*
+ * Process the partitioning specification (if any) and store the partition
+@@ src/backend/commands/tablecmds.c: MergeCheckConstraint(List *constraints, char *name, Node *expr)
+ */
+ static void
+ StoreCatalogInheritance(Oid relationId, List *supers,
+- bool child_is_partition)
++ bool child_is_partition, bool publish_via_parent)
+ {
+ Relation relation;
+ int32 seqNumber;
+@@ src/backend/commands/tablecmds.c: StoreCatalogInheritance(Oid relationId, List *supers,
+ */
+ Assert(OidIsValid(relationId));
+
++ /*
++ * publish_via_parent requires exactly one parent. Try to keep this
++ * messaging consistent with ALTER TABLE/change_publish_via_parent().
++ *
++ * Earlier, in MergeAttributes, we already checked that we had ownership of
++ * the parent table, so we don't need to check that again the way that
++ * change_publish_via_parent() does.
++ */
++ if (publish_via_parent && list_length(supers) != 1)
++ ereport(ERROR,
++ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
++ errmsg((list_length(supers) == 0)
++ ? "table \"%s\" does not inherit from any tables"
++ : "table \"%s\" inherits from multiple tables",
++ get_rel_name(relationId))));
++
+ if (supers == NIL)
+ return;
+
+@@ src/backend/commands/tablecmds.c: StoreCatalogInheritance(Oid relationId, List *supers,
+ */
+ relation = table_open(InheritsRelationId, RowExclusiveLock);
+
+- seqNumber = 1;
++ /*
++ * Normally inhseqno starts at one, but for publish_via_parent, it's given a
++ * sentinel value of zero. (In that case, we'll only go through this loop
++ * once, as the list_length() check above enforces.)
++ */
++ seqNumber = publish_via_parent ? 0 : 1;
+ foreach(entry, supers)
+ {
+ Oid parentOid = lfirst_oid(entry);
+@@ src/backend/commands/tablecmds.c: ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, const char *tablespacen
+ tab->newTableSpace = tablespaceId;
+ }
+
++static void
++change_publish_via_parent(Relation rel, bool set)
++{
++ Relation inhRel;
++ ScanKeyData key;
++ SysScanDesc scan;
++ Oid parentOid;
++ Relation parentRel;
++ HeapTuple tuple, copyTuple;
++ Form_pg_inherits form;
++
++ /* Open pg_inherits with RowExclusiveLock so that we can update it. */
++ inhRel = table_open(InheritsRelationId, RowExclusiveLock);
++
++ /*
++ * We have to make sure that an inheritance relationship already exists, and
++ * that there is only one candidate parent for this table.
++ *
++ * TODO: do we have to lock the tables themselves to avoid races?
++ */
++ ScanKeyInit(&key,
++ Anum_pg_inherits_inhrelid,
++ BTEqualStrategyNumber, F_OIDEQ,
++ ObjectIdGetDatum(RelationGetRelid(rel)));
++
++ scan = systable_beginscan(inhRel, InheritsRelidSeqnoIndexId, true,
++ NULL, 1, &key);
++ tuple = systable_getnext(scan);
++
++ if (!HeapTupleIsValid(tuple))
++ ereport(ERROR,
++ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
++ errmsg("table \"%s\" does not inherit from any tables",
++ RelationGetRelationName(rel))));
++
++ form = (Form_pg_inherits) GETSTRUCT(tuple);
++ parentOid = form->inhparent;
++ copyTuple = heap_copytuple(tuple);
++
++ if (HeapTupleIsValid(systable_getnext(scan)))
++ ereport(ERROR,
++ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
++ errmsg("table \"%s\" inherits from multiple tables",
++ RelationGetRelationName(rel))));
++
++ systable_endscan(scan);
++
++ if (set)
++ {
++ /*
++ * Open our parent table and check permissions. Like ATTACH PARTITION,
++ * we require ownership of the parent table in addition to the child
++ * (which has already been checked in ATPrepCmd).
++ *
++ * TODO: what lock strength is needed here? Is nesting it inside the
++ * inhRel lock a risk?
++ */
++ parentRel = table_open(parentOid, AccessExclusiveLock);
++ ATSimplePermissions(AT_SetRelOptions, parentRel, ATT_TABLE);
++ table_close(parentRel, NoLock); /* TODO: do we need to hang onto the lock? */
++ }
++ else
++ {
++ /*
++ * Ownership of the parent isn't needed for unsetting the flag, just
++ * like we wouldn't check ownership during NO INHERIT.
++ */
++ }
++
++ /*
++ * Mark the inheritance as a logical root by setting inhseqno to zero.
++ * Unmark it by setting inhseqno to one (since there's only one parent).
++ */
++ form = (Form_pg_inherits) GETSTRUCT(copyTuple);
++ form->inhseqno = set ? 0 : 1;
++
++ CatalogTupleUpdate(inhRel, ©Tuple->t_self, copyTuple);
++
++ heap_freetuple(copyTuple);
++ table_close(inhRel, RowExclusiveLock);
++}
++
+ /*
+ * Set, reset, or replace reloptions.
+ */
+@@ src/backend/commands/tablecmds.c: ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
+ }
+ }
+
++ /* Extra work for publish_via_parent */
++ if (rel->rd_rel->relkind == RELKIND_RELATION)
++ {
++ List *heap_options = untransformRelOptions(newOptions);
++ ListCell *cell;
++ bool curSetting = RelationIsPublishedViaParent(rel);
++ DefElem *pubElem = NULL;
++
++ foreach (cell, heap_options)
++ {
++ DefElem *defel = lfirst_node(DefElem, cell);
++
++ if (strcmp(defel->defname, "publish_via_parent") == 0)
++ {
++ pubElem = defel;
++ break;
++ }
++ }
++
++ if (pubElem)
++ {
++ bool newSetting = defGetBoolean(pubElem);
++
++ if (newSetting != curSetting)
++ change_publish_via_parent(rel, newSetting);
++ }
++ else if (curSetting)
++ {
++ /* publish_via_parent was RESET */
++ change_publish_via_parent(rel, false);
++ }
++ }
++
+ /*
+ * All we need do here is update the pg_class row; the new options will be
+ * propagated into relcaches during post-commit cache inval.
+@@ src/backend/commands/tablecmds.c: ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode)
+ trigger_name, RelationGetRelationName(child_rel)),
+ errdetail("ROW triggers with transition tables are not supported in inheritance hierarchies.")));
+
++ /* publish_via_parent tables are not allowed to add additional parents. */
++ if (RelationIsPublishedViaParent(child_rel))
++ ereport(ERROR,
++ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
++ errmsg("relation option \"%s\" prevents table \"%s\" from changing inheritance",
++ "publish_via_parent", RelationGetRelationName(child_rel))));
++
+ /* OK to create inheritance */
+ CreateInheritance(child_rel, parent_rel);
+
+@@ src/backend/commands/tablecmds.c: ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot change inheritance of a partition")));
+
++ /* publish_via_parent tables are not allowed to remove their parent. */
++ if (RelationIsPublishedViaParent(rel))
++ ereport(ERROR,
++ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
++ errmsg("relation option \"%s\" prevents table \"%s\" from changing inheritance",
++ "publish_via_parent", RelationGetRelationName(rel))));
++
+ /*
+ * AccessShareLock on the parent is probably enough, seeing that DROP
+ * TABLE doesn't lock parent tables at all. We need some lock since we'll
+
## src/backend/partitioning/partdesc.c ##
@@ src/backend/partitioning/partdesc.c: RelationBuildPartitionDesc(Relation rel, bool omit_detached)
inhoids = find_inheritance_children_extended(RelationGetRelid(rel),
@@ src/backend/replication/logical/tablesync.c: fetch_remote_table_info(char *nspna
+ /*
+ * See if there are any other tables to be copied besides the original. This
+ * happens when a descendant in the inheritance relationship is marked with
-+ * pg_set_logical_root() and is part of a publication with
++ * publish_via_parent and is part of a publication with
+ * publish_via_partition_root = true.
+ *
+ * FIXME: if two publications have conflicting settings for
@@ src/backend/replication/pgoutput/pgoutput.c: pgoutput_change(LogicalDecodingCont
targetrel = ancestor;
}
+ ## src/bin/pg_dump/t/002_pg_dump.pl ##
+@@ src/bin/pg_dump/t/002_pg_dump.pl: my %tests = (
+ no_table_access_method => 1,
+ only_dump_measurement => 1,
+ },
++ },
++
++ 'CREATE TABLE regress_pg_dump_publish_via_parent' => {
++ create_order => 3,
++ create_sql => '
++ CREATE TABLE dump_test.regress_pg_dump_publish_via_parent (col1 int);
++ CREATE TABLE dump_test.regress_pg_dump_publish_via_parent_1() INHERITS (dump_test.regress_pg_dump_publish_via_parent) WITH (publish_via_parent);',
++ regexp => qr/^
++ \QCREATE TABLE dump_test.regress_pg_dump_publish_via_parent (\E
++ \n\s+\Qcol1 integer\E
++ \n\);
++ (\n.*)+?
++ \n\QCREATE TABLE dump_test.regress_pg_dump_publish_via_parent_1 (\E
++ \n\)
++ \n\QINHERITS (dump_test.regress_pg_dump_publish_via_parent)\E
++ \n\QWITH (publish_via_parent='true')\E;/xm,
++ like => {
++ %full_runs, %dump_test_schema_runs, section_pre_data => 1,
++ },
++ unlike => {
++ binary_upgrade => 1,
++ exclude_dump_test_schema => 1,
++ only_dump_measurement => 1,
++ },
+ });
+
+ #########################################
+
+ ## src/bin/psql/tab-complete.c ##
+@@ src/bin/psql/tab-complete.c: static const char *const table_storage_parameters[] = {
+ "fillfactor",
+ "log_autovacuum_min_duration",
+ "parallel_workers",
++ "publish_via_parent",
+ "toast.autovacuum_enabled",
+ "toast.autovacuum_freeze_max_age",
+ "toast.autovacuum_freeze_min_age",
+
## src/include/catalog/pg_inherits.h ##
+@@
+
+ #include "nodes/pg_list.h"
+ #include "storage/lock.h"
++#include "utils/relcache.h"
+
+ /* ----------------
+ * pg_inherits definition. cpp turns this into
@@ src/include/catalog/pg_inherits.h: typedef FormData_pg_inherits *Form_pg_inherits;
DECLARE_UNIQUE_INDEX_PKEY(pg_inherits_relid_seqno_index, 2680, InheritsRelidSeqnoIndexId, on pg_inherits using btree(inhrelid oid_ops, inhseqno int4_ops));
@@ src/include/catalog/pg_proc.dat
{ oid => '8139',
descr => 'get information on how a relation will be published via a list of publications',
proname => 'pg_get_relation_publishing_info', provariadic => 'text',
-@@
- proname => 'pg_partition_root', prorettype => 'regclass',
- proargtypes => 'regclass', prosrc => 'pg_partition_root' },
+
+ ## src/include/utils/rel.h ##
+@@ src/include/utils/rel.h: typedef struct StdRdOptions
+ int toast_tuple_target; /* target for tuple toasting */
+ AutoVacOpts autovacuum; /* autovacuum-related options */
+ bool user_catalog_table; /* use as an additional catalog relation */
++ bool publish_via_parent; /* publish via parent's relid for logrep */
+ int parallel_workers; /* max number of parallel workers */
+ StdRdOptIndexCleanup vacuum_index_cleanup; /* controls index vacuuming */
+ bool vacuum_truncate; /* enables vacuum to truncate a relation */
+@@ src/include/utils/rel.h: typedef struct StdRdOptions
+ (relation)->rd_rel->relkind == RELKIND_MATVIEW) ? \
+ ((StdRdOptions *) (relation)->rd_options)->user_catalog_table : false)
-+{ oid => '8136', descr => 'mark a table root for logical replication',
-+ proname => 'pg_set_logical_root', provolatile => 'v', proparallel => 'u',
-+ prorettype => 'void', proargtypes => 'regclass regclass',
-+ prosrc => 'pg_set_logical_root' },
++/*
++ * RelationIsPublishedViaParent
++ * Returns whether the relation's contents should be logically replicated
++ * via its parent table's relid, when published under the effects of
++ * publish_via_partition_root. Note multiple eval of argument!
++ */
++#define RelationIsPublishedViaParent(relation) \
++ ((relation)->rd_options && \
++ ((relation)->rd_rel->relkind == RELKIND_RELATION) && \
++ ((StdRdOptions *) (relation)->rd_options)->publish_via_parent)
+
- { oid => '4350', descr => 'Unicode normalization',
- proname => 'normalize', prorettype => 'text', proargtypes => 'text text',
- prosrc => 'unicode_normalize_func' },
+ /*
+ * RelationGetParallelWorkers
+ * Returns the relation's parallel_workers reloption setting.
## src/test/regress/expected/publication.out ##
@@ src/test/regress/expected/publication.out: CREATE ROLE regress_publication_user LOGIN SUPERUSER;
@@ src/test/regress/expected/publication.out: SELECT * FROM pg_publication_tables;
pub | sch1 | tbl1 | {a} |
(1 row)
-+-- Sanity check cases for pg_set_logical_root().
++-- Sanity check cases for publish_via_parent.
+CREATE TABLE sch1.iroot (a int);
+CREATE TABLE sch1.ipart1 (a int);
+CREATE TABLE sch1.ipart2 () INHERITS (sch1.iroot);
++-- should do nothing at all
++ALTER TABLE sch1.iroot SET (publish_via_parent = false);
++ALTER TABLE sch1.iroot RESET (publish_via_parent);
+-- marking roots between unrelated tables is not allowed
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
-+ERROR: table "ipart1" does not inherit from intended root table "iroot"
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.tbl1');
-+ERROR: table "ipart2" does not inherit from intended root table "tbl1"
++ALTER TABLE sch1.ipart1 SET (publish_via_parent);
++ERROR: table "ipart1" does not inherit from any tables
++CREATE TABLE fail (a int) WITH (publish_via_parent);
++ERROR: table "fail" does not inherit from any tables
+-- establishing an inheritance relationship fixes the problem
-+ALTER TABLE sch1.ipart1 INHERIT sch1.iroot;
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
++ALTER TABLE sch1.ipart1 INHERIT sch1.iroot,
++ SET (publish_via_parent);
+-- but multiple inheritance is not allowed
-+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1,
++ SET (publish_via_parent);
+ERROR: table "ipart2" inherits from multiple tables
-+ALTER TABLE sch1.ipart2 NO INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
++-- once publish_via_parent is set, inheritance cannot be changed
++ALTER TABLE sch1.ipart2 SET (publish_via_parent);
++ALTER TABLE sch1.ipart2 NO INHERIT sch1.iroot;
++ERROR: relation option "publish_via_parent" prevents table "ipart2" from changing inheritance
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
++ERROR: relation option "publish_via_parent" prevents table "ipart2" from changing inheritance
+-- table ownership must match, like ATTACH PARTITION
+CREATE ROLE regress_test_me;
+CREATE ROLE regress_test_not_me;
@@ src/test/regress/expected/publication.out: SELECT * FROM pg_publication_tables;
+ALTER TABLE root OWNER TO regress_test_me;
+ALTER TABLE part OWNER TO regress_test_not_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should fail
++ALTER TABLE part SET (publish_via_parent); -- should fail
+ERROR: must be owner of table part
+RESET SESSION AUTHORIZATION;
+ALTER TABLE root OWNER TO regress_test_not_me;
+ALTER TABLE part OWNER TO regress_test_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should also fail
++ALTER TABLE part SET (publish_via_parent); -- should also fail
++ERROR: must be owner of table root
++CREATE TABLE fail () INHERITS (root) WITH (publish_via_parent);
+ERROR: must be owner of table root
+RESET SESSION AUTHORIZATION;
+DROP TABLE root, part;
+DROP ROLE regress_test_not_me;
+DROP ROLE regress_test_me;
-+-- TODO: make sure existing logical descendant can't be ALTERed [NO] INHERIT
+-- Mixed publication settings for publish_via_partition_root, at different
+-- levels of the inheritance tree, to pin correct behavior in the worst cases.
-+CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1);
++CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1) WITH (publish_via_parent);
+CREATE TABLE sch1.ipart1_a1 () INHERITS (sch1.ipart1_a);
-+SELECT pg_set_logical_root('sch1.ipart1_a', 'sch1.ipart1');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
-+SELECT pg_set_logical_root('sch1.ipart1_a1', 'sch1.ipart1_a');
-+ pg_set_logical_root
-+---------------------
-+
-+(1 row)
-+
++ALTER TABLE sch1.ipart1_a1 SET (publish_via_parent);
+CREATE PUBLICATION ipub_root FOR TABLE sch1.iroot;
+CREATE PUBLICATION ipub_part1 FOR TABLE ONLY sch1.ipart1, ONLY sch1.ipart1_a1
+ WITH (publish_via_partition_root);
@@ src/test/regress/expected/publication.out: SELECT * FROM pg_publication_tables;
+ ipub_other | sch1.iroot | sch1.ipart1_a
+(10 rows)
+
++-- "Detaching" partitions should change the subscriptions
++ALTER TABLE sch1.ipart2 SET (publish_via_parent = false);
++SELECT pubname, relid::regclass
++ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
++ JOIN pg_publication p ON (p.oid = t.pubid);
++ pubname | relid
++------------+-------------
++ ipub_root | sch1.iroot
++ ipub_root | sch1.ipart1
++ ipub_root | sch1.ipart2
++ ipub_part1 | sch1.ipart1
++ ipub_other | sch1.iroot
++ ipub_other | sch1.ipart2
++(6 rows)
++
++SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
++ published | synced
++-------------+----------------
++ sch1.iroot | sch1.iroot
++ sch1.iroot | sch1.ipart1_a
++ sch1.ipart1 | sch1.ipart1
++ sch1.ipart1 | sch1.ipart1_a1
++ sch1.ipart2 | sch1.ipart2
++(5 rows)
++
++SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
++ pubname | published | synced
++------------+-------------+----------------
++ ipub_root | sch1.iroot | sch1.iroot
++ ipub_root | sch1.iroot | sch1.ipart1_a
++ ipub_root | sch1.ipart1 | sch1.ipart1
++ ipub_root | sch1.ipart1 | sch1.ipart1_a1
++ ipub_root | sch1.ipart2 | sch1.ipart2
++ ipub_part1 | sch1.ipart1 | sch1.ipart1
++ ipub_part1 | sch1.ipart1 | sch1.ipart1_a1
++ ipub_other | sch1.iroot | sch1.iroot
++ ipub_other | sch1.iroot | sch1.ipart1_a
++ ipub_other | sch1.ipart2 | sch1.ipart2
++(10 rows)
++
++ALTER TABLE sch1.ipart1_a1 RESET (publish_via_parent);
++SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
++ relid
++----------------
++ sch1.ipart1
++ sch1.ipart1_a1
++(2 rows)
++
++SELECT * FROM published_sync('ipub_part1');
++ published | synced
++----------------+----------------
++ sch1.ipart1 | sch1.ipart1
++ sch1.ipart1_a1 | sch1.ipart1_a1
++(2 rows)
++
++SELECT * FROM published_stream('ipub_part1');
++ pubname | published | synced
++------------+----------------+----------------
++ ipub_part1 | sch1.ipart1 | sch1.ipart1
++ ipub_part1 | sch1.ipart1_a1 | sch1.ipart1_a1
++(2 rows)
++
+DROP PUBLICATION ipub_other;
+DROP PUBLICATION ipub_part1;
+DROP PUBLICATION ipub_root;
@@ src/test/regress/sql/publication.sql: ALTER TABLE sch1.tbl1 ATTACH PARTITION sch
CREATE PUBLICATION pub FOR TABLES IN SCHEMA sch1 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
SELECT * FROM pg_publication_tables;
-+-- Sanity check cases for pg_set_logical_root().
++-- Sanity check cases for publish_via_parent.
+CREATE TABLE sch1.iroot (a int);
+CREATE TABLE sch1.ipart1 (a int);
+CREATE TABLE sch1.ipart2 () INHERITS (sch1.iroot);
+
++-- should do nothing at all
++ALTER TABLE sch1.iroot SET (publish_via_parent = false);
++ALTER TABLE sch1.iroot RESET (publish_via_parent);
++
+-- marking roots between unrelated tables is not allowed
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.tbl1');
++ALTER TABLE sch1.ipart1 SET (publish_via_parent);
++CREATE TABLE fail (a int) WITH (publish_via_parent);
+
+-- establishing an inheritance relationship fixes the problem
-+ALTER TABLE sch1.ipart1 INHERIT sch1.iroot;
-+SELECT pg_set_logical_root('sch1.ipart1', 'sch1.iroot');
++ALTER TABLE sch1.ipart1 INHERIT sch1.iroot,
++ SET (publish_via_parent);
+
+-- but multiple inheritance is not allowed
-+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1,
++ SET (publish_via_parent);
+
-+ALTER TABLE sch1.ipart2 NO INHERIT sch1.ipart1;
-+SELECT pg_set_logical_root('sch1.ipart2', 'sch1.iroot');
++-- once publish_via_parent is set, inheritance cannot be changed
++ALTER TABLE sch1.ipart2 SET (publish_via_parent);
++ALTER TABLE sch1.ipart2 NO INHERIT sch1.iroot;
++ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
+
+-- table ownership must match, like ATTACH PARTITION
+CREATE ROLE regress_test_me;
+CREATE ROLE regress_test_not_me;
+CREATE TABLE root (a int);
+CREATE TABLE part () INHERITS (root);
++
+ALTER TABLE root OWNER TO regress_test_me;
+ALTER TABLE part OWNER TO regress_test_not_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should fail
++ALTER TABLE part SET (publish_via_parent); -- should fail
+RESET SESSION AUTHORIZATION;
++
+ALTER TABLE root OWNER TO regress_test_not_me;
+ALTER TABLE part OWNER TO regress_test_me;
+SET SESSION AUTHORIZATION regress_test_me;
-+SELECT pg_set_logical_root('part', 'root'); -- should also fail
++ALTER TABLE part SET (publish_via_parent); -- should also fail
++CREATE TABLE fail () INHERITS (root) WITH (publish_via_parent);
+RESET SESSION AUTHORIZATION;
++
+DROP TABLE root, part;
+DROP ROLE regress_test_not_me;
+DROP ROLE regress_test_me;
+
-+-- TODO: make sure existing logical descendant can't be ALTERed [NO] INHERIT
-+
+-- Mixed publication settings for publish_via_partition_root, at different
+-- levels of the inheritance tree, to pin correct behavior in the worst cases.
-+CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1);
++CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1) WITH (publish_via_parent);
+CREATE TABLE sch1.ipart1_a1 () INHERITS (sch1.ipart1_a);
-+
-+SELECT pg_set_logical_root('sch1.ipart1_a', 'sch1.ipart1');
-+SELECT pg_set_logical_root('sch1.ipart1_a1', 'sch1.ipart1_a');
++ALTER TABLE sch1.ipart1_a1 SET (publish_via_parent);
+
+CREATE PUBLICATION ipub_root FOR TABLE sch1.iroot;
+CREATE PUBLICATION ipub_part1 FOR TABLE ONLY sch1.ipart1, ONLY sch1.ipart1_a1
@@ src/test/regress/sql/publication.sql: ALTER TABLE sch1.tbl1 ATTACH PARTITION sch
+SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
+SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
+
++-- "Detaching" partitions should change the subscriptions
++ALTER TABLE sch1.ipart2 SET (publish_via_parent = false);
++SELECT pubname, relid::regclass
++ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
++ JOIN pg_publication p ON (p.oid = t.pubid);
++SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
++SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
++
++ALTER TABLE sch1.ipart1_a1 RESET (publish_via_parent);
++SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
++SELECT * FROM published_sync('ipub_part1');
++SELECT * FROM published_stream('ipub_part1');
++
+DROP PUBLICATION ipub_other;
+DROP PUBLICATION ipub_part1;
+DROP PUBLICATION ipub_root;
@@ src/test/subscription/t/013_partition.pl: $node_subscriber1->append_conf('postgr
$node_subscriber1->reload;
+# Make sure standard inheritance setups aren't broken by the new
-+# pg_set_logical_root() handling.
++# publish_via_parent handling.
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1 (a int, b text)");
+$node_subscriber2->safe_psql('postgres',
@@ src/test/subscription/t/013_partition.pl: $node_subscriber1->append_conf('postgr
+ "CREATE TABLE itab1_2 (CHECK (a = 2)) INHERITS (itab1)");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_1', 'itab1')");
++ "ALTER TABLE itab1_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_2', 'itab1')");
++ "ALTER TABLE itab1_2 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1 VALUES (0, 'itab1')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1_1 VALUES (1, 'itab1')");
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+ FOR EACH ROW EXECUTE FUNCTION itab1_trigger();");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_1', 'itab1')");
++ "ALTER TABLE itab1_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab1_2', 'itab1')");
++ "ALTER TABLE itab1_2 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab2 (a int, b text)");
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+ FOR EACH ROW EXECUTE FUNCTION itab2_trigger();");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab2_1', 'itab2')");
++ "ALTER TABLE itab2_1 SET (publish_via_parent = true)");
+
+# itab2_1 should be published using its own identity here, since its parent is
+# not included. itab1_1 should be published via its parent, itab1, without
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+ "CREATE TABLE itab3_1_1 () INHERITS (itab3_1)");
+
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab3_1', 'itab3')");
++ "ALTER TABLE itab3_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
-+ "SELECT pg_set_logical_root('itab3_1_1', 'itab3_1')");
++ "ALTER TABLE itab3_1_1 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO itab3 VALUES (1, 'itab3')");
@@ src/test/subscription/t/013_partition.pl: $result = $node_subscriber2->safe_psql
+is($result, qq(), 'initial data routed for itab3_1_1 on subscriber 2');
+
+# make sure new data is also correctly routed to the roots
-+$node_publisher->safe_psql('postgres', "INSERT INTO itab3 VALUES (4, 'itab3-new')");
-+$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1 VALUES (4, 'itab3_1-new')");
-+$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1_1 VALUES (4, 'itab3_1_1-new')");
++$node_publisher->safe_psql('postgres', "INSERT INTO itab3 VALUES (4, 'itab3_new')");
++$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1 VALUES (4, 'itab3_1_new')");
++$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1_1 VALUES (4, 'itab3_1_1_new')");
+
+$node_publisher->wait_for_catchup('mixed');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3 ORDER BY 1, 2");
+is($result, qq(1|itab3
-+4|itab3-new), 'new data routed for itab3 on subscriber 2');
++4|itab3_new), 'new data routed for itab3 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1 ORDER BY 1, 2");
+is($result, qq(2|itab3_1
+3|itab3_1_1
-+4|itab3_1-new
-+4|itab3_1_1-new), 'new data routed for itab3_1 on subscriber 2');
++4|itab3_1_1_new
++4|itab3_1_new), 'new data routed for itab3_1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1_1");
[text/x-patch] v3-0001-pgoutput-refactor-publication-cache-construction.patch (18.1K, ../[email protected]/3-v3-0001-pgoutput-refactor-publication-cache-construction.patch)
download | inline diff:
From 7e68f96af6e1811113054b4f238b5ed429f7d628 Mon Sep 17 00:00:00 2001
From: Jacob Champion <[email protected]>
Date: Fri, 7 Apr 2023 09:55:27 -0700
Subject: [PATCH v3 1/2] pgoutput: refactor publication cache construction
Breaking this logic into a standalone helper should help expose the
behavior for testing. Additionally, move the implementation under the
pg_publication catalog helpers, since it seems like it's inherent to the
publication settings (and it must match the behavior of the initial
sync, which doesn't seem to be controlled by the output plugin at all).
---
src/backend/catalog/pg_publication.c | 215 ++++++++++++++++++++
src/backend/replication/pgoutput/pgoutput.c | 137 +------------
src/include/catalog/pg_proc.dat | 8 +
src/include/catalog/pg_publication.h | 4 +
src/test/regress/expected/publication.out | 26 +++
src/test/regress/sql/publication.sql | 15 ++
6 files changed, 272 insertions(+), 133 deletions(-)
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index c488b6370b..df9abf09c3 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -1259,3 +1259,218 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
SRF_RETURN_DONE(funcctx);
}
+
+List *
+process_relation_publications(Oid relid, const List *publications,
+ PublicationActions *pubactions,
+ Oid *publish_as_relid)
+{
+ Oid schemaId = get_rel_namespace(relid);
+ List *pubids = GetRelationPublications(relid);
+
+ /*
+ * We don't acquire a lock on the namespace system table as we build
+ * the cache entry using a historic snapshot and all the later changes
+ * are absorbed while decoding WAL.
+ */
+ List *schemaPubids = GetSchemaPublications(schemaId);
+ ListCell *lc;
+ int publish_ancestor_level = 0;
+ bool am_partition = get_rel_relispartition(relid);
+ char relkind = get_rel_relkind(relid);
+ List *rel_publications = NIL;
+
+ *publish_as_relid = relid;
+
+ foreach(lc, publications)
+ {
+ Publication *pub = lfirst(lc);
+ bool publish = false;
+
+ /*
+ * Under what relid should we publish changes in this publication?
+ * We'll use the top-most relid across all publications. Also
+ * track the ancestor level for this publication.
+ */
+ Oid pub_relid = relid;
+ int ancestor_level = 0;
+
+ /*
+ * If this is a FOR ALL TABLES publication, pick the partition
+ * root and set the ancestor level accordingly.
+ */
+ if (pub->alltables)
+ {
+ publish = true;
+ if (pub->pubviaroot && am_partition)
+ {
+ List *ancestors = get_partition_ancestors(relid);
+
+ pub_relid = llast_oid(ancestors);
+ ancestor_level = list_length(ancestors);
+ }
+ }
+
+ if (!publish)
+ {
+ bool ancestor_published = false;
+
+ /*
+ * For a partition, check if any of the ancestors are
+ * published. If so, note down the topmost ancestor that is
+ * published via this publication, which will be used as the
+ * relation via which to publish the partition's changes.
+ */
+ if (am_partition)
+ {
+ Oid ancestor;
+ int level;
+ List *ancestors = get_partition_ancestors(relid);
+
+ ancestor = GetTopMostAncestorInPublication(pub->oid,
+ ancestors,
+ &level);
+
+ if (ancestor != InvalidOid)
+ {
+ ancestor_published = true;
+ if (pub->pubviaroot)
+ {
+ pub_relid = ancestor;
+ ancestor_level = level;
+ }
+ }
+ }
+
+ if (list_member_oid(pubids, pub->oid) ||
+ list_member_oid(schemaPubids, pub->oid) ||
+ ancestor_published)
+ publish = true;
+ }
+
+ /*
+ * If the relation is to be published, determine actions to
+ * publish, and list of columns, if appropriate.
+ *
+ * Don't publish changes for partitioned tables, because
+ * publishing those of its partitions suffices, unless partition
+ * changes won't be published due to pubviaroot being set.
+ */
+ if (publish &&
+ (relkind != RELKIND_PARTITIONED_TABLE || pub->pubviaroot))
+ {
+ pubactions->pubinsert |= pub->pubactions.pubinsert;
+ pubactions->pubupdate |= pub->pubactions.pubupdate;
+ pubactions->pubdelete |= pub->pubactions.pubdelete;
+ pubactions->pubtruncate |= pub->pubactions.pubtruncate;
+
+ /*
+ * We want to publish the changes as the top-most ancestor
+ * across all publications. So we need to check if the already
+ * calculated level is higher than the new one. If yes, we can
+ * ignore the new value (as it's a child). Otherwise the new
+ * value is an ancestor, so we keep it.
+ */
+ if (publish_ancestor_level > ancestor_level)
+ continue;
+
+ /*
+ * If we found an ancestor higher up in the tree, discard the
+ * list of publications through which we replicate it, and use
+ * the new ancestor.
+ */
+ if (publish_ancestor_level < ancestor_level)
+ {
+ *publish_as_relid = pub_relid;
+ publish_ancestor_level = ancestor_level;
+
+ /* reset the publication list for this relation */
+ rel_publications = NIL;
+ }
+ else
+ {
+ /* Same ancestor level, has to be the same OID. */
+ Assert(*publish_as_relid == pub_relid);
+ }
+
+ /* Track publications for this ancestor. */
+ rel_publications = lappend(rel_publications, pub);
+ }
+ }
+
+ list_free(pubids);
+ list_free(schemaPubids);
+
+ return rel_publications;
+}
+
+Datum
+pg_get_relation_publishing_info(PG_FUNCTION_ARGS)
+{
+ Oid relid = PG_GETARG_OID(0);
+ List *publications = NIL;
+ ArrayType *arr;
+ Datum *elems;
+ int nelems,
+ i;
+ TupleDesc tupdesc;
+ HeapTuple htup;
+
+ if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+ elog(ERROR, "return type must be a row type");
+
+ arr = PG_GETARG_ARRAYTYPE_P(1);
+ deconstruct_array(arr, TEXTOID, -1, false, TYPALIGN_INT, &elems, NULL,
+ &nelems);
+
+ /* Get Oids of tables from each publication. */
+ for (i = 0; i < nelems; i++)
+ {
+ char *pubname = TextDatumGetCString(elems[i]);
+ Publication *pub = GetPublicationByName(pubname, false);
+
+ publications = lappend(publications, pub);
+ }
+
+ {
+ List *rel_publications;
+ PublicationActions pubactions;
+ Oid publish_as_relid;
+ ListCell *lc;
+ Datum *puboids;
+ ArrayType *puboidarray;
+ Datum values[6];
+ bool nulls[6];
+
+ rel_publications =
+ process_relation_publications(relid, publications, &pubactions,
+ &publish_as_relid);
+
+ /* Translate the rel_publications List into an OID array. */
+ puboids = palloc(sizeof(Datum) * list_length(rel_publications));
+
+ foreach(lc, rel_publications)
+ {
+ Publication *pub = lfirst(lc);
+
+ i = foreach_current_index(lc);
+ puboids[i] = ObjectIdGetDatum(pub->oid);
+ }
+
+ puboidarray = construct_array(puboids, list_length(rel_publications),
+ OIDOID, sizeof(Oid), true, TYPALIGN_INT);
+
+ values[0] = PointerGetDatum(puboidarray);
+ values[1] = ObjectIdGetDatum(publish_as_relid);
+ values[2] = BoolGetDatum(pubactions.pubinsert);
+ values[3] = BoolGetDatum(pubactions.pubupdate);
+ values[4] = BoolGetDatum(pubactions.pubdelete);
+ values[5] = BoolGetDatum(pubactions.pubtruncate);
+
+ memset(nulls, false, sizeof(nulls));
+
+ htup = heap_form_tuple(tupdesc, values, nulls);
+ }
+
+ PG_RETURN_DATUM(HeapTupleGetDatum(htup));
+}
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index b08ca55041..640676e7dc 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1986,20 +1986,6 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
/* Validate the entry */
if (!entry->replicate_valid)
{
- Oid schemaId = get_rel_namespace(relid);
- List *pubids = GetRelationPublications(relid);
-
- /*
- * We don't acquire a lock on the namespace system table as we build
- * the cache entry using a historic snapshot and all the later changes
- * are absorbed while decoding WAL.
- */
- List *schemaPubids = GetSchemaPublications(schemaId);
- ListCell *lc;
- Oid publish_as_relid = relid;
- int publish_ancestor_level = 0;
- bool am_partition = get_rel_relispartition(relid);
- char relkind = get_rel_relkind(relid);
List *rel_publications = NIL;
/* Reload publications if needed before use. */
@@ -2063,123 +2049,10 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
* but here we only need to consider ones that the subscriber
* requested.
*/
- foreach(lc, data->publications)
- {
- Publication *pub = lfirst(lc);
- bool publish = false;
-
- /*
- * Under what relid should we publish changes in this publication?
- * We'll use the top-most relid across all publications. Also
- * track the ancestor level for this publication.
- */
- Oid pub_relid = relid;
- int ancestor_level = 0;
-
- /*
- * If this is a FOR ALL TABLES publication, pick the partition
- * root and set the ancestor level accordingly.
- */
- if (pub->alltables)
- {
- publish = true;
- if (pub->pubviaroot && am_partition)
- {
- List *ancestors = get_partition_ancestors(relid);
-
- pub_relid = llast_oid(ancestors);
- ancestor_level = list_length(ancestors);
- }
- }
-
- if (!publish)
- {
- bool ancestor_published = false;
-
- /*
- * For a partition, check if any of the ancestors are
- * published. If so, note down the topmost ancestor that is
- * published via this publication, which will be used as the
- * relation via which to publish the partition's changes.
- */
- if (am_partition)
- {
- Oid ancestor;
- int level;
- List *ancestors = get_partition_ancestors(relid);
-
- ancestor = GetTopMostAncestorInPublication(pub->oid,
- ancestors,
- &level);
-
- if (ancestor != InvalidOid)
- {
- ancestor_published = true;
- if (pub->pubviaroot)
- {
- pub_relid = ancestor;
- ancestor_level = level;
- }
- }
- }
-
- if (list_member_oid(pubids, pub->oid) ||
- list_member_oid(schemaPubids, pub->oid) ||
- ancestor_published)
- publish = true;
- }
-
- /*
- * If the relation is to be published, determine actions to
- * publish, and list of columns, if appropriate.
- *
- * Don't publish changes for partitioned tables, because
- * publishing those of its partitions suffices, unless partition
- * changes won't be published due to pubviaroot being set.
- */
- if (publish &&
- (relkind != RELKIND_PARTITIONED_TABLE || pub->pubviaroot))
- {
- entry->pubactions.pubinsert |= pub->pubactions.pubinsert;
- entry->pubactions.pubupdate |= pub->pubactions.pubupdate;
- entry->pubactions.pubdelete |= pub->pubactions.pubdelete;
- entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate;
-
- /*
- * We want to publish the changes as the top-most ancestor
- * across all publications. So we need to check if the already
- * calculated level is higher than the new one. If yes, we can
- * ignore the new value (as it's a child). Otherwise the new
- * value is an ancestor, so we keep it.
- */
- if (publish_ancestor_level > ancestor_level)
- continue;
-
- /*
- * If we found an ancestor higher up in the tree, discard the
- * list of publications through which we replicate it, and use
- * the new ancestor.
- */
- if (publish_ancestor_level < ancestor_level)
- {
- publish_as_relid = pub_relid;
- publish_ancestor_level = ancestor_level;
-
- /* reset the publication list for this relation */
- rel_publications = NIL;
- }
- else
- {
- /* Same ancestor level, has to be the same OID. */
- Assert(publish_as_relid == pub_relid);
- }
-
- /* Track publications for this ancestor. */
- rel_publications = lappend(rel_publications, pub);
- }
- }
-
- entry->publish_as_relid = publish_as_relid;
+ rel_publications =
+ process_relation_publications(relid, data->publications,
+ &entry->pubactions,
+ &entry->publish_as_relid);
/*
* Initialize the tuple slot, map, and row filter. These are only used
@@ -2198,8 +2071,6 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
pgoutput_column_list_init(data, rel_publications, entry);
}
- list_free(pubids);
- list_free(schemaPubids);
list_free(rel_publications);
entry->replicate_valid = true;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 6996073989..ed67168374 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11833,6 +11833,14 @@
proname => 'pg_relation_is_publishable', provolatile => 's',
prorettype => 'bool', proargtypes => 'regclass',
prosrc => 'pg_relation_is_publishable' },
+{ oid => '8139',
+ descr => 'get information on how a relation will be published via a list of publications',
+ proname => 'pg_get_relation_publishing_info', provariadic => 'text',
+ provolatile => 's', prorettype => 'record', proargtypes => 'regclass _text',
+ proallargtypes => '{regclass,_text,_oid,oid,bool,bool,bool,bool}',
+ proargmodes => '{i,v,o,o,o,o,o,o}',
+ proargnames => '{relid,pubnames,pubids,pubasrelid,pubinsert,pubupdate,pubdelete,pubtruncate}',
+ prosrc => 'pg_get_relation_publishing_info' },
# rls
{ oid => '3298',
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 6ecaa2a01e..d4e3488917 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -155,4 +155,8 @@ extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
extern Bitmapset *pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols,
MemoryContext mcxt);
+extern List *process_relation_publications(Oid relid, const List *publications,
+ PublicationActions *pubactions,
+ Oid *publish_as_relid);
+
#endif /* PG_PUBLICATION_H */
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 69dc6cfd85..dd8fc99111 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -5,6 +5,17 @@ CREATE ROLE regress_publication_user LOGIN SUPERUSER;
CREATE ROLE regress_publication_user2;
CREATE ROLE regress_publication_user_dummy LOGIN NOSUPERUSER;
SET SESSION AUTHORIZATION 'regress_publication_user';
+CREATE FUNCTION published_stream (VARIADIC pubnames text[])
+ RETURNS TABLE (pubname text, published regclass, synced regclass) AS $$
+ -- For each publication, show each published root alongside the tables which
+ -- are published via its OID.
+ SELECT p.pubname, rpi.pubasrelid::regclass, pr.prrelid::regclass
+ FROM pg_publication_rel pr
+ JOIN pg_publication p ON (p.oid = pr.prpubid),
+ pg_get_relation_publishing_info(pr.prrelid, VARIADIC pubnames) rpi
+ WHERE p.pubname = ANY (pubnames)
+ ORDER BY p.oid, 2, 3;
+$$ LANGUAGE sql;
-- suppress warning that depends on wal_level
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_default;
@@ -1686,6 +1697,13 @@ SELECT * FROM pg_publication_tables;
pub | sch1 | tbl1 | {a} |
(1 row)
+SELECT * FROM published_stream('pub');
+ pubname | published | synced
+---------+-----------+-----------------
+ pub | sch1.tbl1 | sch1.tbl1
+ pub | sch1.tbl1 | sch2.tbl1_part1
+(2 rows)
+
DROP PUBLICATION pub;
-- Schema publication that does not include the schema that has the parent table
CREATE PUBLICATION pub FOR TABLES IN SCHEMA sch2 WITH (PUBLISH_VIA_PARTITION_ROOT=0);
@@ -1712,6 +1730,13 @@ SELECT * FROM pg_publication_tables;
pub | sch2 | tbl1_part1 | {a} |
(1 row)
+SELECT * FROM published_stream('pub');
+ pubname | published | synced
+---------+-----------------+-----------------
+ pub | sch1.tbl1 | sch1.tbl1
+ pub | sch2.tbl1_part1 | sch2.tbl1_part1
+(2 rows)
+
DROP PUBLICATION pub;
DROP TABLE sch2.tbl1_part1;
DROP TABLE sch1.tbl1;
@@ -1732,6 +1757,7 @@ DROP PUBLICATION pub;
DROP TABLE sch1.tbl1;
DROP SCHEMA sch1 cascade;
DROP SCHEMA sch2 cascade;
+DROP FUNCTION published_stream;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
DROP ROLE regress_publication_user_dummy;
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index d5051a5e74..6bf9554d48 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -6,6 +6,18 @@ CREATE ROLE regress_publication_user2;
CREATE ROLE regress_publication_user_dummy LOGIN NOSUPERUSER;
SET SESSION AUTHORIZATION 'regress_publication_user';
+CREATE FUNCTION published_stream (VARIADIC pubnames text[])
+ RETURNS TABLE (pubname text, published regclass, synced regclass) AS $$
+ -- For each publication, show each published root alongside the tables which
+ -- are published via its OID.
+ SELECT p.pubname, rpi.pubasrelid::regclass, pr.prrelid::regclass
+ FROM pg_publication_rel pr
+ JOIN pg_publication p ON (p.oid = pr.prpubid),
+ pg_get_relation_publishing_info(pr.prrelid, VARIADIC pubnames) rpi
+ WHERE p.pubname = ANY (pubnames)
+ ORDER BY p.oid, 2, 3;
+$$ LANGUAGE sql;
+
-- suppress warning that depends on wal_level
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_default;
@@ -1064,6 +1076,7 @@ SELECT * FROM pg_publication_tables;
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
SELECT * FROM pg_publication_tables;
+SELECT * FROM published_stream('pub');
DROP PUBLICATION pub;
-- Schema publication that does not include the schema that has the parent table
@@ -1078,6 +1091,7 @@ SELECT * FROM pg_publication_tables;
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
SELECT * FROM pg_publication_tables;
+SELECT * FROM published_stream('pub');
DROP PUBLICATION pub;
DROP TABLE sch2.tbl1_part1;
@@ -1096,6 +1110,7 @@ DROP PUBLICATION pub;
DROP TABLE sch1.tbl1;
DROP SCHEMA sch1 cascade;
DROP SCHEMA sch2 cascade;
+DROP FUNCTION published_stream;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
--
2.25.1
[text/x-patch] v3-0002-WIP-introduce-publish_via_parent-for-logical-repl.patch (83.2K, ../[email protected]/4-v3-0002-WIP-introduce-publish_via_parent-for-logical-repl.patch)
download | inline diff:
From a48563919e0d4a70b7115551a5d061b2e17b5243 Mon Sep 17 00:00:00 2001
From: Jacob Champion <[email protected]>
Date: Mon, 26 Sep 2022 13:23:51 -0700
Subject: [PATCH v3 2/2] WIP: introduce publish_via_parent for logical
replication
This new relation option allows regular inherited tables to be published
via their root table, just like partitions. This works by hijacking
pg_inherit's inhseqno column, and replacing a (single) existing entry
for the child with the value zero, indicating that it should be treated
as a logical partition by the publication machinery.
Initial sync works by asking the publisher for a list of logical
descendants of the published table, then COPYing them one-by-one into
the root. The publisher reuses the existing pubviaroot logic, adding the
new logical roots to code that previously looked only for partition
roots.
Known bugs/TODOs:
- If two publications publish the same leaves of a table hierarchy, but
have different roots, the shared leaf tables will be incorrectly
duplicated when subscribing to both publications simultaneously.
- I haven't given any thought to interactions with row filters, or to
column lists.
- I'm not sure that I'm taking all the necessary locks yet, and those I
do take may be taken in the wrong order.
---
src/backend/access/common/reloptions.c | 11 +
src/backend/catalog/pg_inherits.c | 123 +++++++-
src/backend/catalog/pg_publication.c | 284 +++++++++++++-----
src/backend/commands/publicationcmds.c | 10 +
src/backend/commands/tablecmds.c | 161 +++++++++-
src/backend/partitioning/partdesc.c | 3 +-
src/backend/replication/logical/tablesync.c | 257 ++++++++++------
src/backend/replication/pgoutput/pgoutput.c | 1 -
src/bin/pg_dump/t/002_pg_dump.pl | 24 ++
src/bin/psql/tab-complete.c | 1 +
src/include/catalog/pg_inherits.h | 8 +-
src/include/catalog/pg_proc.dat | 7 +
src/include/utils/rel.h | 12 +
src/test/regress/expected/publication.out | 315 ++++++++++++++++++++
src/test/regress/sql/publication.sql | 140 +++++++++
src/test/subscription/t/013_partition.pl | 312 +++++++++++++++++++
16 files changed, 1490 insertions(+), 179 deletions(-)
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 11cc431677..6d01e2477d 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -122,6 +122,15 @@ static relopt_bool boolRelOpts[] =
},
false
},
+ {
+ {
+ "publish_via_parent",
+ "Replicate this table's contents via its parent table when publishing with publish_via_partition_root",
+ RELOPT_KIND_HEAP,
+ AccessExclusiveLock
+ },
+ false
+ },
{
{
"fastupdate",
@@ -1877,6 +1886,8 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, analyze_scale_factor)},
{"user_catalog_table", RELOPT_TYPE_BOOL,
offsetof(StdRdOptions, user_catalog_table)},
+ {"publish_via_parent", RELOPT_TYPE_BOOL,
+ offsetof(StdRdOptions, publish_via_parent)},
{"parallel_workers", RELOPT_TYPE_INT,
offsetof(StdRdOptions, parallel_workers)},
{"vacuum_index_cleanup", RELOPT_TYPE_ENUM,
diff --git a/src/backend/catalog/pg_inherits.c b/src/backend/catalog/pg_inherits.c
index da969bd2f9..cf90769457 100644
--- a/src/backend/catalog/pg_inherits.c
+++ b/src/backend/catalog/pg_inherits.c
@@ -24,14 +24,22 @@
#include "access/table.h"
#include "catalog/indexing.h"
#include "catalog/pg_inherits.h"
+#include "catalog/partition.h"
+#include "miscadmin.h"
#include "parser/parse_type.h"
#include "storage/lmgr.h"
+#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
+#include "utils/fmgrprotos.h"
+#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
+static List * find_all_inheritors_internal(Oid parentrelId, LOCKMODE lockmode,
+ List **numparents, bool logical_only);
+
/*
* Entry of a hash table used in find_all_inheritors. See below.
*/
@@ -59,14 +67,14 @@ List *
find_inheritance_children(Oid parentrelId, LOCKMODE lockmode)
{
return find_inheritance_children_extended(parentrelId, true, lockmode,
- NULL, NULL);
+ NULL, NULL, false);
}
/*
* find_inheritance_children_extended
*
* As find_inheritance_children, with more options regarding detached
- * partitions.
+ * partitions and logical roots.
*
* If a partition's pg_inherits row is marked "detach pending",
* *detached_exist (if not null) is set true.
@@ -78,16 +86,21 @@ find_inheritance_children(Oid parentrelId, LOCKMODE lockmode)
* whether the transaction that marked those partitions as detached appears
* committed to the active snapshot. In addition, *detached_xmin (if not null)
* is set to the xmin of the row of the detached partition.
+ *
+ * If logical_only is true, only tables marked explicitly via publish_via_parent
+ * are included in the output list.
*/
List *
find_inheritance_children_extended(Oid parentrelId, bool omit_detached,
LOCKMODE lockmode, bool *detached_exist,
- TransactionId *detached_xmin)
+ TransactionId *detached_xmin,
+ bool logical_only)
{
List *list = NIL;
Relation relation;
+ Oid index;
SysScanDesc scan;
- ScanKeyData key[1];
+ ScanKeyData key[2];
HeapTuple inheritsTuple;
Oid inhrelid;
Oid *oidarr;
@@ -110,14 +123,24 @@ find_inheritance_children_extended(Oid parentrelId, bool omit_detached,
numoids = 0;
relation = table_open(InheritsRelationId, AccessShareLock);
+ index = InheritsParentIndexId;
ScanKeyInit(&key[0],
Anum_pg_inherits_inhparent,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(parentrelId));
- scan = systable_beginscan(relation, InheritsParentIndexId, true,
- NULL, 1, key);
+ if (logical_only)
+ {
+ index = InheritsParentSeqnoIndexId;
+ ScanKeyInit(&key[1],
+ Anum_pg_inherits_inhseqno,
+ BTEqualStrategyNumber, F_INT4EQ,
+ Int32GetDatum(0));
+ }
+
+ scan = systable_beginscan(relation, index, true,
+ NULL, logical_only ? 2 : 1, key);
while ((inheritsTuple = systable_getnext(scan)) != NULL)
{
@@ -254,6 +277,20 @@ find_inheritance_children_extended(Oid parentrelId, bool omit_detached,
*/
List *
find_all_inheritors(Oid parentrelId, LOCKMODE lockmode, List **numparents)
+{
+ return find_all_inheritors_internal(parentrelId, lockmode, numparents,
+ false);
+}
+
+List *
+find_all_logical_inheritors(Oid parentrelId)
+{
+ return find_all_inheritors_internal(parentrelId, NoLock, NULL, true);
+}
+
+static List *
+find_all_inheritors_internal(Oid parentrelId, LOCKMODE lockmode,
+ List **numparents, bool logical_only)
{
/* hash table for O(1) rel_oid -> rel_numparents cell lookup */
HTAB *seen_rels;
@@ -290,7 +327,9 @@ find_all_inheritors(Oid parentrelId, LOCKMODE lockmode, List **numparents)
ListCell *lc;
/* Get the direct children of this rel */
- currentchildren = find_inheritance_children(currentrel, lockmode);
+ currentchildren =
+ find_inheritance_children_extended(currentrel, true, lockmode,
+ NULL, NULL, logical_only);
/*
* Add to the queue only those children not already seen. This avoids
@@ -655,3 +694,73 @@ PartitionHasPendingDetach(Oid partoid)
elog(ERROR, "relation %u is not a partition", partoid);
return false; /* keep compiler quiet */
}
+
+static Oid
+get_logical_parent_worker(Relation inhRel, Oid relid)
+{
+ SysScanDesc scan;
+ ScanKeyData key[2];
+ Oid result = InvalidOid;
+ HeapTuple tuple;
+
+ ScanKeyInit(&key[0],
+ Anum_pg_inherits_inhrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relid));
+ ScanKeyInit(&key[1],
+ Anum_pg_inherits_inhseqno,
+ BTEqualStrategyNumber, F_INT4EQ,
+ Int32GetDatum(0));
+
+ scan = systable_beginscan(inhRel, InheritsRelidSeqnoIndexId, true,
+ NULL, 2, key);
+ tuple = systable_getnext(scan);
+ if (HeapTupleIsValid(tuple))
+ {
+ Form_pg_inherits form = (Form_pg_inherits) GETSTRUCT(tuple);
+ result = form->inhparent;
+ }
+
+ systable_endscan(scan);
+
+ return result;
+}
+
+static void
+get_logical_ancestors_worker(Relation inhRel, Oid relid, List **ancestors)
+{
+ Oid parentOid;
+
+ /*
+ * Recursion ends at the topmost level, ie., when there's no parent.
+ */
+ parentOid = get_logical_parent_worker(inhRel, relid);
+ if (parentOid == InvalidOid)
+ return;
+
+ *ancestors = lappend_oid(*ancestors, parentOid);
+ get_logical_ancestors_worker(inhRel, parentOid, ancestors);
+}
+
+List *
+get_logical_ancestors(Oid relid, bool is_partition)
+{
+ List *result = NIL;
+ Relation inhRel;
+
+ /* For partitions, this is identical to get_partition_ancestors(). */
+ if (is_partition)
+ return get_partition_ancestors(relid);
+
+ inhRel = table_open(InheritsRelationId, AccessShareLock);
+ get_logical_ancestors_worker(inhRel, relid, &result);
+ table_close(inhRel, AccessShareLock);
+
+ return result;
+}
+
+bool
+has_logical_parent(Relation inhRel, Oid relid)
+{
+ return (get_logical_parent_worker(inhRel, relid) != InvalidOid);
+}
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index df9abf09c3..873cbf955e 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -51,6 +51,7 @@ typedef struct
Oid relid; /* OID of published table */
Oid pubid; /* OID of publication that publishes this
* table. */
+ bool viaroot; /* does pubid use publish_via_partition_root? */
} published_rel;
static void publication_translate_columns(Relation targetrel, List *columns,
@@ -180,56 +181,46 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
}
/*
- * Returns true if the ancestor is in the list of published relations.
- * Otherwise, returns false.
- */
-static bool
-is_ancestor_member_tableinfos(Oid ancestor, List *table_infos)
-{
- ListCell *lc;
-
- foreach(lc, table_infos)
- {
- Oid relid = ((published_rel *) lfirst(lc))->relid;
-
- if (relid == ancestor)
- return true;
- }
-
- return false;
-}
-
-/*
- * Filter out the partitions whose parent tables are also present in the list.
+ * Filter out the tables who are already being published via a logical root.
+ *
+ * If we only had partitioned tables to check, this would be much easier: it's
+ * impossible for a partition root to be in this list unless pubviaroot=true for
+ * that table, so we would just have to check for the existence of an ancestor,
+ * and if any were found then we'd filter the table.
+ *
+ * Logical roots add another wrinkle, because it's acceptable for a
+ * pubviaroot=false publication to contain a logical root. (Logical roots can
+ * contain data of their own, unlike partition roots.) It's also possible for a
+ * logical root to be included in a pubviaroot publication without any or all of
+ * its children. This is not possible for a partition root, where all of the
+ * leaves are implicitly included.
+ *
+ * So, to keep things relatively consistent, this now uses the same code that
+ * the output plugin uses to decide which relation OID to publish data under. If
+ * the publishing OID is not the same as the relation OID, we remove it from
+ * this list.
+ *
+ * XXX And that's expensive.
*/
static void
-filter_partitions(List *table_infos)
+filter_published_descendants(List *table_infos, List *publications)
{
ListCell *lc;
foreach(lc, table_infos)
{
- bool skip = false;
- List *ancestors = NIL;
- ListCell *lc2;
published_rel *table_info = (published_rel *) lfirst(lc);
+ Oid publish_as_relid;
+ List *rel_publications;
- if (get_rel_relispartition(table_info->relid))
- ancestors = get_partition_ancestors(table_info->relid);
-
- foreach(lc2, ancestors)
- {
- Oid ancestor = lfirst_oid(lc2);
-
- if (is_ancestor_member_tableinfos(ancestor, table_infos))
- {
- skip = true;
- break;
- }
- }
+ rel_publications =
+ process_relation_publications(table_info->relid, publications, NULL,
+ &publish_as_relid);
- if (skip)
+ if (publish_as_relid != table_info->relid)
table_infos = foreach_delete_current(table_infos, lc);
+
+ list_free(rel_publications);
}
}
@@ -805,11 +796,15 @@ List *
GetAllTablesPublicationRelations(bool pubviaroot)
{
Relation classRel;
+ Relation inhRel;
ScanKeyData key[1];
TableScanDesc scan;
HeapTuple tuple;
List *result = NIL;
+ /* TODO: is there a required order to acquire these locks? */
+ if (pubviaroot)
+ inhRel = table_open(InheritsRelationId, AccessShareLock);
classRel = table_open(RelationRelationId, AccessShareLock);
ScanKeyInit(&key[0],
@@ -825,7 +820,8 @@ GetAllTablesPublicationRelations(bool pubviaroot)
Oid relid = relForm->oid;
if (is_publishable_class(relid, relForm) &&
- !(relForm->relispartition && pubviaroot))
+ !(relForm->relispartition && pubviaroot) &&
+ !(pubviaroot && has_logical_parent(inhRel, relid)))
result = lappend_oid(result, relid);
}
@@ -854,6 +850,9 @@ GetAllTablesPublicationRelations(bool pubviaroot)
}
table_close(classRel, AccessShareLock);
+ if (pubviaroot)
+ table_close(inhRel, AccessShareLock);
+
return result;
}
@@ -1070,6 +1069,7 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
int nelems,
i;
bool viaroot = false;
+ List *publications = NIL;
/* create a function context for cross-call persistence */
funcctx = SRF_FIRSTCALL_INIT();
@@ -1093,6 +1093,7 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
ListCell *lc;
pub_elem = GetPublicationByName(TextDatumGetCString(elems[i]), false);
+ publications = lappend(publications, pub_elem);
/*
* Publications support partitioned tables. If
@@ -1132,6 +1133,7 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
table_info->relid = lfirst_oid(lc);
table_info->pubid = pub_elem->oid;
+ table_info->viaroot = pub_elem->pubviaroot;
table_infos = lappend(table_infos, table_info);
}
@@ -1141,15 +1143,14 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
}
/*
- * If the publication publishes partition changes via their respective
- * root partitioned tables, we must exclude partitions in favor of
- * including the root partitioned tables. Otherwise, the function
- * could return both the child and parent tables which could cause
- * data of the child table to be double-published on the subscriber
- * side.
+ * If the publication publishes table changes via their respective
+ * logical root tables, we must exclude partitions in favor of
+ * including the root tables. Otherwise, the function could return
+ * both the child and parent tables which could cause data of the
+ * child table to be double-published on the subscriber side.
*/
if (viaroot)
- filter_partitions(table_infos);
+ filter_published_descendants(table_infos, publications);
/* Construct a tuple descriptor for the result rows. */
tupdesc = CreateTemplateTupleDesc(NUM_PUBLICATION_TABLES_ELEM);
@@ -1165,6 +1166,8 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
funcctx->tuple_desc = BlessTupleDesc(tupdesc);
funcctx->user_fctx = (void *) table_infos;
+ list_free(publications);
+
MemoryContextSwitchTo(oldcontext);
}
@@ -1260,6 +1263,118 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
SRF_RETURN_DONE(funcctx);
}
+/*
+ * Returns a list of tables that should be copied during initial sync for the
+ * given root table and publications.
+ *
+ * The only time this list consists of anything more than the table itself is
+ * when a publication's publish_via_partition_root is set to true and the table
+ * has inherited child tables in the publications that have been marked with
+ * publish_via_parent.
+ */
+Datum
+pg_get_publication_rels_to_sync(PG_FUNCTION_ARGS)
+{
+ FuncCallContext *funcctx;
+ List *tables = NIL;
+
+ /* stuff done only on the first call of the function */
+ if (SRF_IS_FIRSTCALL())
+ {
+ Oid rootid = PG_GETARG_OID(0);
+ MemoryContext oldcontext;
+ ArrayType *arr;
+ Datum *elems;
+ int nelems,
+ i;
+
+ /* create a function context for cross-call persistence */
+ funcctx = SRF_FIRSTCALL_INIT();
+
+ /* switch to memory context appropriate for multiple function calls */
+ oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+ /*
+ * Deconstruct the parameter into elements where each element is a
+ * publication name.
+ */
+ arr = PG_GETARG_ARRAYTYPE_P(1);
+ deconstruct_array(arr, TEXTOID, -1, false, TYPALIGN_INT,
+ &elems, NULL, &nelems);
+
+ /* Get Oids of tables from each publication. */
+ for (i = 0; i < nelems; i++)
+ {
+ char *pubname = TextDatumGetCString(elems[i]);
+ Publication *publication = GetPublicationByName(pubname, false);
+ List *pub_tables = NIL;
+
+ /* TODO: do the tables in this list need to be locked? */
+ if (publication->pubviaroot)
+ pub_tables = find_all_logical_inheritors(rootid);
+ else
+ pub_tables = list_make1_oid(rootid);
+
+ if (!publication->alltables)
+ {
+ List *relids;
+ List *schemarelids;
+ List *published;
+ ListCell *cell;
+
+ relids = GetPublicationRelations(publication->oid,
+ publication->pubviaroot ?
+ PUBLICATION_PART_ROOT :
+ PUBLICATION_PART_ALL);
+ schemarelids = GetAllSchemaPublicationRelations(publication->oid,
+ publication->pubviaroot ?
+ PUBLICATION_PART_ROOT :
+ PUBLICATION_PART_LEAF);
+ published = list_concat_unique_oid(relids, schemarelids);
+
+ /*
+ * First we have to check to make sure the root table is
+ * actually part of the publication; if not, none of its
+ * descendants' contents belong to it.
+ */
+ if (!list_member_oid(published, rootid))
+ continue;
+
+ /*
+ * Now filter out any descendants that aren't part of this
+ * particular publication.
+ */
+ foreach(cell, pub_tables)
+ {
+ Oid current = lfirst_oid(cell);
+
+ if (!list_member_oid(published, current))
+ pub_tables = foreach_delete_current(pub_tables, cell);
+ }
+ }
+
+ tables = list_concat_unique_oid(tables, pub_tables);
+ }
+
+ funcctx->user_fctx = (void *) tables;
+
+ MemoryContextSwitchTo(oldcontext);
+ }
+
+ /* stuff done on every call of the function */
+ funcctx = SRF_PERCALL_SETUP();
+ tables = (List *) funcctx->user_fctx;
+
+ if (funcctx->call_cntr < list_length(tables))
+ {
+ Oid relid = list_nth_oid(tables, funcctx->call_cntr);
+
+ SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid));
+ }
+
+ SRF_RETURN_DONE(funcctx);
+}
+
List *
process_relation_publications(Oid relid, const List *publications,
PublicationActions *pubactions,
@@ -1279,8 +1394,7 @@ process_relation_publications(Oid relid, const List *publications,
bool am_partition = get_rel_relispartition(relid);
char relkind = get_rel_relkind(relid);
List *rel_publications = NIL;
-
- *publish_as_relid = relid;
+ Oid publish_candidate = relid;
foreach(lc, publications)
{
@@ -1296,55 +1410,57 @@ process_relation_publications(Oid relid, const List *publications,
int ancestor_level = 0;
/*
- * If this is a FOR ALL TABLES publication, pick the partition
+ * If this is a FOR ALL TABLES publication, pick the logical
* root and set the ancestor level accordingly.
*/
if (pub->alltables)
{
publish = true;
- if (pub->pubviaroot && am_partition)
+ if (pub->pubviaroot)
{
- List *ancestors = get_partition_ancestors(relid);
+ List *ancestors;
- pub_relid = llast_oid(ancestors);
- ancestor_level = list_length(ancestors);
+ ancestors = get_logical_ancestors(relid, am_partition);
+ if (ancestors != NIL)
+ {
+ pub_relid = llast_oid(ancestors);
+ ancestor_level = list_length(ancestors);
+ }
}
}
if (!publish)
{
bool ancestor_published = false;
+ Oid ancestor;
+ int level;
+ List *ancestors;
/*
- * For a partition, check if any of the ancestors are
- * published. If so, note down the topmost ancestor that is
+ * Check if any of the logical ancestors (that is, partition
+ * parents or tables marked with publish_via_parent) are
+ * published. If so, note down the topmost ancestor that is
* published via this publication, which will be used as the
- * relation via which to publish the partition's changes.
+ * relation via which to publish this table's changes.
*/
- if (am_partition)
- {
- Oid ancestor;
- int level;
- List *ancestors = get_partition_ancestors(relid);
-
- ancestor = GetTopMostAncestorInPublication(pub->oid,
- ancestors,
- &level);
+ ancestors = get_logical_ancestors(relid, am_partition);
+ ancestor = GetTopMostAncestorInPublication(pub->oid,
+ ancestors,
+ &level);
- if (ancestor != InvalidOid)
+ if (ancestor != InvalidOid)
+ {
+ ancestor_published = true;
+ if (pub->pubviaroot)
{
- ancestor_published = true;
- if (pub->pubviaroot)
- {
- pub_relid = ancestor;
- ancestor_level = level;
- }
+ pub_relid = ancestor;
+ ancestor_level = level;
}
}
if (list_member_oid(pubids, pub->oid) ||
list_member_oid(schemaPubids, pub->oid) ||
- ancestor_published)
+ (am_partition && ancestor_published))
publish = true;
}
@@ -1359,10 +1475,13 @@ process_relation_publications(Oid relid, const List *publications,
if (publish &&
(relkind != RELKIND_PARTITIONED_TABLE || pub->pubviaroot))
{
- pubactions->pubinsert |= pub->pubactions.pubinsert;
- pubactions->pubupdate |= pub->pubactions.pubupdate;
- pubactions->pubdelete |= pub->pubactions.pubdelete;
- pubactions->pubtruncate |= pub->pubactions.pubtruncate;
+ if (pubactions)
+ {
+ pubactions->pubinsert |= pub->pubactions.pubinsert;
+ pubactions->pubupdate |= pub->pubactions.pubupdate;
+ pubactions->pubdelete |= pub->pubactions.pubdelete;
+ pubactions->pubtruncate |= pub->pubactions.pubtruncate;
+ }
/*
* We want to publish the changes as the top-most ancestor
@@ -1381,7 +1500,7 @@ process_relation_publications(Oid relid, const List *publications,
*/
if (publish_ancestor_level < ancestor_level)
{
- *publish_as_relid = pub_relid;
+ publish_candidate = pub_relid;
publish_ancestor_level = ancestor_level;
/* reset the publication list for this relation */
@@ -1390,7 +1509,7 @@ process_relation_publications(Oid relid, const List *publications,
else
{
/* Same ancestor level, has to be the same OID. */
- Assert(*publish_as_relid == pub_relid);
+ Assert(publish_candidate == pub_relid);
}
/* Track publications for this ancestor. */
@@ -1401,6 +1520,9 @@ process_relation_publications(Oid relid, const List *publications,
list_free(pubids);
list_free(schemaPubids);
+ if (publish_as_relid)
+ *publish_as_relid = publish_candidate;
+
return rel_publications;
}
@@ -1434,7 +1556,7 @@ pg_get_relation_publishing_info(PG_FUNCTION_ARGS)
{
List *rel_publications;
- PublicationActions pubactions;
+ PublicationActions pubactions = {0};
Oid publish_as_relid;
ListCell *lc;
Datum *puboids;
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index f4ba572697..4cabc9719e 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -238,6 +238,8 @@ contain_invalid_rfcolumn_walker(Node *node, rf_context *context)
* parent table, but the bitmap contains the replica identity
* information of the child table. So, get the column number of the
* child table as parent and child column order could be different.
+ *
+ * TODO: is this applicable to publish_via_parent?
*/
if (context->pubviaroot)
{
@@ -286,6 +288,8 @@ pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
*
* Note that even though the row filter used is for an ancestor, the
* REPLICA IDENTITY used will be for the actual child table.
+ *
+ * TODO: is this applicable to publish_via_parent?
*/
if (pubviaroot && relation->rd_rel->relispartition)
{
@@ -336,6 +340,8 @@ pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
* the column list.
*
* Returns true if any replica identity column is not covered by column list.
+ *
+ * TODO: publish_via_parent?
*/
bool
pub_collist_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
@@ -628,6 +634,8 @@ TransformPubWhereClauses(List *tables, const char *queryString,
* If the publication doesn't publish changes via the root partitioned
* table, the partition's row filter will be used. So disallow using
* WHERE clause on partitioned table in this case.
+ *
+ * TODO: decide how this interacts with publish_via_parent
*/
if (!pubviaroot &&
pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
@@ -715,6 +723,8 @@ CheckPubRelationColumnList(char *pubname, List *tables,
* If the publication doesn't publish changes via the root partitioned
* table, the partition's column list will be used. So disallow using
* a column list on the partitioned table in this case.
+ *
+ * TODO: decide if this interacts with publish_via_parent
*/
if (!pubviaroot &&
pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index d985278ac6..8250640643 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -355,7 +355,8 @@ static bool MergeCheckConstraint(List *constraints, char *name, Node *expr);
static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel);
static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel);
static void StoreCatalogInheritance(Oid relationId, List *supers,
- bool child_is_partition);
+ bool child_is_partition,
+ bool publish_via_parent);
static void StoreCatalogInheritance1(Oid relationId, Oid parentOid,
int32 seqNumber, Relation inhRelation,
bool child_is_partition);
@@ -577,6 +578,7 @@ static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
const char *tablespacename, LOCKMODE lockmode);
static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
static void ATExecSetTableSpaceNoStorage(Relation rel, Oid newTableSpace);
+static void change_publish_via_parent(Relation rel, bool set);
static void ATExecSetRelOptions(Relation rel, List *defList,
AlterTableType operation,
LOCKMODE lockmode);
@@ -1115,7 +1117,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
}
/* Store inheritance information for new rel. */
- StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL);
+ StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL,
+ RelationIsPublishedViaParent(rel));
/*
* Process the partitioning specification (if any) and store the partition
@@ -3218,7 +3221,7 @@ MergeCheckConstraint(List *constraints, char *name, Node *expr)
*/
static void
StoreCatalogInheritance(Oid relationId, List *supers,
- bool child_is_partition)
+ bool child_is_partition, bool publish_via_parent)
{
Relation relation;
int32 seqNumber;
@@ -3229,6 +3232,22 @@ StoreCatalogInheritance(Oid relationId, List *supers,
*/
Assert(OidIsValid(relationId));
+ /*
+ * publish_via_parent requires exactly one parent. Try to keep this
+ * messaging consistent with ALTER TABLE/change_publish_via_parent().
+ *
+ * Earlier, in MergeAttributes, we already checked that we had ownership of
+ * the parent table, so we don't need to check that again the way that
+ * change_publish_via_parent() does.
+ */
+ if (publish_via_parent && list_length(supers) != 1)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg((list_length(supers) == 0)
+ ? "table \"%s\" does not inherit from any tables"
+ : "table \"%s\" inherits from multiple tables",
+ get_rel_name(relationId))));
+
if (supers == NIL)
return;
@@ -3243,7 +3262,12 @@ StoreCatalogInheritance(Oid relationId, List *supers,
*/
relation = table_open(InheritsRelationId, RowExclusiveLock);
- seqNumber = 1;
+ /*
+ * Normally inhseqno starts at one, but for publish_via_parent, it's given a
+ * sentinel value of zero. (In that case, we'll only go through this loop
+ * once, as the list_length() check above enforces.)
+ */
+ seqNumber = publish_via_parent ? 0 : 1;
foreach(entry, supers)
{
Oid parentOid = lfirst_oid(entry);
@@ -14276,6 +14300,88 @@ ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, const char *tablespacen
tab->newTableSpace = tablespaceId;
}
+static void
+change_publish_via_parent(Relation rel, bool set)
+{
+ Relation inhRel;
+ ScanKeyData key;
+ SysScanDesc scan;
+ Oid parentOid;
+ Relation parentRel;
+ HeapTuple tuple, copyTuple;
+ Form_pg_inherits form;
+
+ /* Open pg_inherits with RowExclusiveLock so that we can update it. */
+ inhRel = table_open(InheritsRelationId, RowExclusiveLock);
+
+ /*
+ * We have to make sure that an inheritance relationship already exists, and
+ * that there is only one candidate parent for this table.
+ *
+ * TODO: do we have to lock the tables themselves to avoid races?
+ */
+ ScanKeyInit(&key,
+ Anum_pg_inherits_inhrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(rel)));
+
+ scan = systable_beginscan(inhRel, InheritsRelidSeqnoIndexId, true,
+ NULL, 1, &key);
+ tuple = systable_getnext(scan);
+
+ if (!HeapTupleIsValid(tuple))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("table \"%s\" does not inherit from any tables",
+ RelationGetRelationName(rel))));
+
+ form = (Form_pg_inherits) GETSTRUCT(tuple);
+ parentOid = form->inhparent;
+ copyTuple = heap_copytuple(tuple);
+
+ if (HeapTupleIsValid(systable_getnext(scan)))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("table \"%s\" inherits from multiple tables",
+ RelationGetRelationName(rel))));
+
+ systable_endscan(scan);
+
+ if (set)
+ {
+ /*
+ * Open our parent table and check permissions. Like ATTACH PARTITION,
+ * we require ownership of the parent table in addition to the child
+ * (which has already been checked in ATPrepCmd).
+ *
+ * TODO: what lock strength is needed here? Is nesting it inside the
+ * inhRel lock a risk?
+ */
+ parentRel = table_open(parentOid, AccessExclusiveLock);
+ ATSimplePermissions(AT_SetRelOptions, parentRel, ATT_TABLE);
+ table_close(parentRel, NoLock); /* TODO: do we need to hang onto the lock? */
+ }
+ else
+ {
+ /*
+ * Ownership of the parent isn't needed for unsetting the flag, just
+ * like we wouldn't check ownership during NO INHERIT.
+ */
+ }
+
+ /*
+ * Mark the inheritance as a logical root by setting inhseqno to zero.
+ * Unmark it by setting inhseqno to one (since there's only one parent).
+ */
+ form = (Form_pg_inherits) GETSTRUCT(copyTuple);
+ form->inhseqno = set ? 0 : 1;
+
+ CatalogTupleUpdate(inhRel, ©Tuple->t_self, copyTuple);
+
+ heap_freetuple(copyTuple);
+ table_close(inhRel, RowExclusiveLock);
+}
+
/*
* Set, reset, or replace reloptions.
*/
@@ -14387,6 +14493,39 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
}
}
+ /* Extra work for publish_via_parent */
+ if (rel->rd_rel->relkind == RELKIND_RELATION)
+ {
+ List *heap_options = untransformRelOptions(newOptions);
+ ListCell *cell;
+ bool curSetting = RelationIsPublishedViaParent(rel);
+ DefElem *pubElem = NULL;
+
+ foreach (cell, heap_options)
+ {
+ DefElem *defel = lfirst_node(DefElem, cell);
+
+ if (strcmp(defel->defname, "publish_via_parent") == 0)
+ {
+ pubElem = defel;
+ break;
+ }
+ }
+
+ if (pubElem)
+ {
+ bool newSetting = defGetBoolean(pubElem);
+
+ if (newSetting != curSetting)
+ change_publish_via_parent(rel, newSetting);
+ }
+ else if (curSetting)
+ {
+ /* publish_via_parent was RESET */
+ change_publish_via_parent(rel, false);
+ }
+ }
+
/*
* All we need do here is update the pg_class row; the new options will be
* propagated into relcaches during post-commit cache inval.
@@ -14979,6 +15118,13 @@ ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode)
trigger_name, RelationGetRelationName(child_rel)),
errdetail("ROW triggers with transition tables are not supported in inheritance hierarchies.")));
+ /* publish_via_parent tables are not allowed to add additional parents. */
+ if (RelationIsPublishedViaParent(child_rel))
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("relation option \"%s\" prevents table \"%s\" from changing inheritance",
+ "publish_via_parent", RelationGetRelationName(child_rel))));
+
/* OK to create inheritance */
CreateInheritance(child_rel, parent_rel);
@@ -15395,6 +15541,13 @@ ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot change inheritance of a partition")));
+ /* publish_via_parent tables are not allowed to remove their parent. */
+ if (RelationIsPublishedViaParent(rel))
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("relation option \"%s\" prevents table \"%s\" from changing inheritance",
+ "publish_via_parent", RelationGetRelationName(rel))));
+
/*
* AccessShareLock on the parent is probably enough, seeing that DROP
* TABLE doesn't lock parent tables at all. We need some lock since we'll
diff --git a/src/backend/partitioning/partdesc.c b/src/backend/partitioning/partdesc.c
index 7a2b5e57ff..dfb27990f9 100644
--- a/src/backend/partitioning/partdesc.c
+++ b/src/backend/partitioning/partdesc.c
@@ -162,7 +162,8 @@ RelationBuildPartitionDesc(Relation rel, bool omit_detached)
inhoids = find_inheritance_children_extended(RelationGetRelid(rel),
omit_detached, NoLock,
&detached_exist,
- &detached_xmin);
+ &detached_xmin,
+ false);
nparts = list_length(inhoids);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 6d461654ab..8d0f081dac 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -758,11 +758,12 @@ copy_read_data(void *outbuf, int minread, int maxread)
/*
* Get information about remote relation in similar fashion the RELATION
* message provides during replication. This function also returns the relation
- * qualifications to be used in the COPY command.
+ * qualifications to be used in the COPY command, and the list of tables to COPY
+ * (which for most tables will contain only one entry).
*/
static void
fetch_remote_table_info(char *nspname, char *relname,
- LogicalRepRelation *lrel, List **qual)
+ LogicalRepRelation *lrel, List **qual, List **to_copy)
{
WalRcvExecResult *res;
StringInfoData cmd;
@@ -1073,6 +1074,87 @@ fetch_remote_table_info(char *nspname, char *relname,
walrcv_clear_result(res);
}
+ /*
+ * See if there are any other tables to be copied besides the original. This
+ * happens when a descendant in the inheritance relationship is marked with
+ * publish_via_parent and is part of a publication with
+ * publish_via_partition_root = true.
+ *
+ * FIXME: if two publications have conflicting settings for
+ * publish_via_partition_root, this behaves strangely. It looks like that's
+ * true for regular partitions too...
+ */
+ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 160000)
+ {
+ Oid descRow[] = {TEXTOID, TEXTOID};
+ StringInfoData pub_names;
+
+ /* Build the pubname list. */
+ initStringInfo(&pub_names);
+ foreach(lc, MySubscription->publications)
+ {
+ char *pubname = strVal(lfirst(lc));
+
+ if (foreach_current_index(lc) > 0)
+ appendStringInfoString(&pub_names, ", ");
+
+ appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
+ }
+
+ /*
+ * Ask the server what it wants us to sync.
+ */
+ resetStringInfo(&cmd);
+ appendStringInfo(&cmd,
+ "SELECT DISTINCT n.nspname, c.relname"
+ " FROM pg_catalog.pg_publication p,"
+ " pg_catalog.pg_class c"
+ " JOIN pg_catalog.pg_namespace n"
+ " ON (c.relnamespace = n.oid)"
+ " JOIN LATERAL pg_catalog.pg_get_publication_rels_to_sync(%u::regclass, p.pubname) relid"
+ " ON (c.oid = relid)"
+ " WHERE p.pubname IN ( %s )",
+ lrel->remoteid, pub_names.data);
+
+ res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data,
+ lengthof(descRow), descRow);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ ereport(ERROR,
+ (errmsg("could not fetch logical descendants for table \"%s.%s\" from publisher: %s",
+ nspname, relname, res->err)));
+
+ slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+ while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+ {
+ char *desc_nspname;
+ char *desc_relname;
+ char *quoted;
+
+ desc_nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
+ Assert(!isnull);
+ desc_relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
+ Assert(!isnull);
+
+ quoted = quote_qualified_identifier(desc_nspname, desc_relname);
+ *to_copy = lappend(*to_copy, quoted);
+
+ ExecClearTuple(slot);
+ }
+
+ ExecDropSingleTupleTableSlot(slot);
+ walrcv_clear_result(res);
+
+ pfree(pub_names.data);
+ }
+ else
+ {
+ /* For older servers, we only COPY the table itself. */
+ char *quoted = quote_qualified_identifier(lrel->nspname,
+ lrel->relname);
+ *to_copy = lappend(*to_copy, quoted);
+ }
+
pfree(cmd.data);
}
@@ -1087,6 +1169,8 @@ copy_table(Relation rel)
LogicalRepRelMapEntry *relmapentry;
LogicalRepRelation lrel;
List *qual = NIL;
+ List *to_copy = NIL;
+ ListCell *cur;
WalRcvExecResult *res;
StringInfoData cmd;
CopyFromState cstate;
@@ -1096,7 +1180,8 @@ copy_table(Relation rel)
/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
- RelationGetRelationName(rel), &lrel, &qual);
+ RelationGetRelationName(rel), &lrel, &qual,
+ &to_copy);
/* Put the relation into relmap. */
logicalrep_relmap_update(&lrel);
@@ -1105,105 +1190,109 @@ copy_table(Relation rel)
relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock);
Assert(rel == relmapentry->localrel);
- /* Start copy on the publisher. */
- initStringInfo(&cmd);
-
- /* Regular table with no row filter */
- if (lrel.relkind == RELKIND_RELATION && qual == NIL)
+ foreach(cur, to_copy)
{
- appendStringInfo(&cmd, "COPY %s (",
- quote_qualified_identifier(lrel.nspname, lrel.relname));
-
- /*
- * XXX Do we need to list the columns in all cases? Maybe we're
- * replicating all columns?
- */
- for (int i = 0; i < lrel.natts; i++)
- {
- if (i > 0)
- appendStringInfoString(&cmd, ", ");
+ char *quoted_name = lfirst(cur);
- appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
- }
+ /* Start copy on the publisher. */
+ initStringInfo(&cmd);
- appendStringInfoString(&cmd, ") TO STDOUT");
- }
- else
- {
- /*
- * For non-tables and tables with row filters, we need to do COPY
- * (SELECT ...), but we can't just do SELECT * because we need to not
- * copy generated columns. For tables with any row filters, build a
- * SELECT query with OR'ed row filters for COPY.
- */
- appendStringInfoString(&cmd, "COPY (SELECT ");
- for (int i = 0; i < lrel.natts; i++)
+ /* Regular table with no row filter */
+ if (lrel.relkind == RELKIND_RELATION && qual == NIL)
{
- appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
- if (i < lrel.natts - 1)
- appendStringInfoString(&cmd, ", ");
- }
+ appendStringInfo(&cmd, "COPY %s (", quoted_name);
- appendStringInfoString(&cmd, " FROM ");
+ /*
+ * XXX Do we need to list the columns in all cases? Maybe we're
+ * replicating all columns?
+ */
+ for (int i = 0; i < lrel.natts; i++)
+ {
+ if (i > 0)
+ appendStringInfoString(&cmd, ", ");
- /*
- * For regular tables, make sure we don't copy data from a child that
- * inherits the named table as those will be copied separately.
- */
- if (lrel.relkind == RELKIND_RELATION)
- appendStringInfoString(&cmd, "ONLY ");
+ appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
+ }
- appendStringInfoString(&cmd, quote_qualified_identifier(lrel.nspname, lrel.relname));
- /* list of OR'ed filters */
- if (qual != NIL)
+ appendStringInfoString(&cmd, ") TO STDOUT");
+ }
+ else
{
- ListCell *lc;
- char *q = strVal(linitial(qual));
+ /*
+ * For non-tables and tables with row filters, we need to do COPY
+ * (SELECT ...), but we can't just do SELECT * because we need to not
+ * copy generated columns. For tables with any row filters, build a
+ * SELECT query with OR'ed row filters for COPY.
+ */
+ appendStringInfoString(&cmd, "COPY (SELECT ");
+ for (int i = 0; i < lrel.natts; i++)
+ {
+ appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i]));
+ if (i < lrel.natts - 1)
+ appendStringInfoString(&cmd, ", ");
+ }
+
+ appendStringInfoString(&cmd, " FROM ");
+
+ /*
+ * For regular tables, make sure we don't copy data from a child that
+ * inherits the named table as those will be copied separately.
+ */
+ if (lrel.relkind == RELKIND_RELATION)
+ appendStringInfoString(&cmd, "ONLY ");
- appendStringInfo(&cmd, " WHERE %s", q);
- for_each_from(lc, qual, 1)
+ appendStringInfoString(&cmd, quoted_name);
+ /* list of OR'ed filters */
+ if (qual != NIL)
{
- q = strVal(lfirst(lc));
- appendStringInfo(&cmd, " OR %s", q);
+ ListCell *lc;
+ char *q = strVal(linitial(qual));
+
+ appendStringInfo(&cmd, " WHERE %s", q);
+ for_each_from(lc, qual, 1)
+ {
+ q = strVal(lfirst(lc));
+ appendStringInfo(&cmd, " OR %s", q);
+ }
+ list_free_deep(qual);
}
- list_free_deep(qual);
- }
- appendStringInfoString(&cmd, ") TO STDOUT");
- }
+ appendStringInfoString(&cmd, ") TO STDOUT");
+ }
- /*
- * Prior to v16, initial table synchronization will use text format even
- * if the binary option is enabled for a subscription.
- */
- if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 160000 &&
- MySubscription->binary)
- {
- appendStringInfoString(&cmd, " WITH (FORMAT binary)");
- options = list_make1(makeDefElem("format",
- (Node *) makeString("binary"), -1));
- }
+ /*
+ * Prior to v16, initial table synchronization will use text format even
+ * if the binary option is enabled for a subscription.
+ */
+ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 160000 &&
+ MySubscription->binary)
+ {
+ appendStringInfoString(&cmd, " WITH (FORMAT binary)");
+ options = list_make1(makeDefElem("format",
+ (Node *) makeString("binary"), -1));
+ }
- res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
- pfree(cmd.data);
- if (res->status != WALRCV_OK_COPY_OUT)
- ereport(ERROR,
- (errcode(ERRCODE_CONNECTION_FAILURE),
- errmsg("could not start initial contents copy for table \"%s.%s\": %s",
- lrel.nspname, lrel.relname, res->err)));
- walrcv_clear_result(res);
+ res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
+ pfree(cmd.data);
+ if (res->status != WALRCV_OK_COPY_OUT)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("could not start initial contents copy for table \"%s.%s\" from remote %s: %s",
+ lrel.nspname, lrel.relname, quoted_name, res->err)));
+ walrcv_clear_result(res);
- copybuf = makeStringInfo();
+ copybuf = makeStringInfo();
- pstate = make_parsestate(NULL);
- (void) addRangeTableEntryForRelation(pstate, rel, AccessShareLock,
- NULL, false, false);
+ pstate = make_parsestate(NULL);
+ (void) addRangeTableEntryForRelation(pstate, rel, AccessShareLock,
+ NULL, false, false);
- attnamelist = make_copy_attnamelist(relmapentry);
- cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);
+ attnamelist = make_copy_attnamelist(relmapentry);
+ cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);
- /* Do the copy */
- (void) CopyFrom(cstate);
+ /* Do the copy */
+ (void) CopyFrom(cstate);
+ }
logicalrep_rel_close(relmapentry, NoLock);
}
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 640676e7dc..7c54b46ac7 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1462,7 +1462,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
{
- Assert(relation->rd_rel->relispartition);
ancestor = RelationIdGetRelation(relentry->publish_as_relid);
targetrel = ancestor;
}
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 63bb4689d4..d90745af71 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -4573,6 +4573,30 @@ my %tests = (
no_table_access_method => 1,
only_dump_measurement => 1,
},
+ },
+
+ 'CREATE TABLE regress_pg_dump_publish_via_parent' => {
+ create_order => 3,
+ create_sql => '
+ CREATE TABLE dump_test.regress_pg_dump_publish_via_parent (col1 int);
+ CREATE TABLE dump_test.regress_pg_dump_publish_via_parent_1() INHERITS (dump_test.regress_pg_dump_publish_via_parent) WITH (publish_via_parent);',
+ regexp => qr/^
+ \QCREATE TABLE dump_test.regress_pg_dump_publish_via_parent (\E
+ \n\s+\Qcol1 integer\E
+ \n\);
+ (\n.*)+?
+ \n\QCREATE TABLE dump_test.regress_pg_dump_publish_via_parent_1 (\E
+ \n\)
+ \n\QINHERITS (dump_test.regress_pg_dump_publish_via_parent)\E
+ \n\QWITH (publish_via_parent='true')\E;/xm,
+ like => {
+ %full_runs, %dump_test_schema_runs, section_pre_data => 1,
+ },
+ unlike => {
+ binary_upgrade => 1,
+ exclude_dump_test_schema => 1,
+ only_dump_measurement => 1,
+ },
});
#########################################
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 677847e434..4606272c0c 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1302,6 +1302,7 @@ static const char *const table_storage_parameters[] = {
"fillfactor",
"log_autovacuum_min_duration",
"parallel_workers",
+ "publish_via_parent",
"toast.autovacuum_enabled",
"toast.autovacuum_freeze_max_age",
"toast.autovacuum_freeze_min_age",
diff --git a/src/include/catalog/pg_inherits.h b/src/include/catalog/pg_inherits.h
index ce154ab943..fc539e7401 100644
--- a/src/include/catalog/pg_inherits.h
+++ b/src/include/catalog/pg_inherits.h
@@ -23,6 +23,7 @@
#include "nodes/pg_list.h"
#include "storage/lock.h"
+#include "utils/relcache.h"
/* ----------------
* pg_inherits definition. cpp turns this into
@@ -46,14 +47,17 @@ typedef FormData_pg_inherits *Form_pg_inherits;
DECLARE_UNIQUE_INDEX_PKEY(pg_inherits_relid_seqno_index, 2680, InheritsRelidSeqnoIndexId, on pg_inherits using btree(inhrelid oid_ops, inhseqno int4_ops));
DECLARE_INDEX(pg_inherits_parent_index, 2187, InheritsParentIndexId, on pg_inherits using btree(inhparent oid_ops));
+DECLARE_INDEX(pg_inherits_parent_seqno_index, 8138, InheritsParentSeqnoIndexId, on pg_inherits using btree(inhparent oid_ops, inhseqno int4_ops));
extern List *find_inheritance_children(Oid parentrelId, LOCKMODE lockmode);
extern List *find_inheritance_children_extended(Oid parentrelId, bool omit_detached,
- LOCKMODE lockmode, bool *detached_exist, TransactionId *detached_xmin);
+ LOCKMODE lockmode, bool *detached_exist, TransactionId *detached_xmin,
+ bool logical_only);
extern List *find_all_inheritors(Oid parentrelId, LOCKMODE lockmode,
List **numparents);
+extern List *find_all_logical_inheritors(Oid parentrelId);
extern bool has_subclass(Oid relationId);
extern bool has_superclass(Oid relationId);
extern bool typeInheritsFrom(Oid subclassTypeId, Oid superclassTypeId);
@@ -63,5 +67,7 @@ extern bool DeleteInheritsTuple(Oid inhrelid, Oid inhparent,
bool expect_detach_pending,
const char *childname);
extern bool PartitionHasPendingDetach(Oid partoid);
+extern List *get_logical_ancestors(Oid relid, bool is_partition);
+extern bool has_logical_parent(Relation inhRel, Oid relid);
#endif /* PG_INHERITS_H */
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index ed67168374..a60c6b66d8 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11833,6 +11833,13 @@
proname => 'pg_relation_is_publishable', provolatile => 's',
prorettype => 'bool', proargtypes => 'regclass',
prosrc => 'pg_relation_is_publishable' },
+{ oid => '8137',
+ descr => 'get the relations to copy from all specified publications during a table\'s initial sync',
+ proname => 'pg_get_publication_rels_to_sync', prorows => '10',
+ provariadic => 'text', proretset => 't', provolatile => 's',
+ prorettype => 'regclass', proargtypes => 'regclass _text',
+ proargmodes => '{i,v}', proargnames => '{rootid,pubnames}',
+ prosrc => 'pg_get_publication_rels_to_sync' },
{ oid => '8139',
descr => 'get information on how a relation will be published via a list of publications',
proname => 'pg_get_relation_publishing_info', provariadic => 'text',
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 1426a353cd..3c5ea5a37a 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -339,6 +339,7 @@ typedef struct StdRdOptions
int toast_tuple_target; /* target for tuple toasting */
AutoVacOpts autovacuum; /* autovacuum-related options */
bool user_catalog_table; /* use as an additional catalog relation */
+ bool publish_via_parent; /* publish via parent's relid for logrep */
int parallel_workers; /* max number of parallel workers */
StdRdOptIndexCleanup vacuum_index_cleanup; /* controls index vacuuming */
bool vacuum_truncate; /* enables vacuum to truncate a relation */
@@ -388,6 +389,17 @@ typedef struct StdRdOptions
(relation)->rd_rel->relkind == RELKIND_MATVIEW) ? \
((StdRdOptions *) (relation)->rd_options)->user_catalog_table : false)
+/*
+ * RelationIsPublishedViaParent
+ * Returns whether the relation's contents should be logically replicated
+ * via its parent table's relid, when published under the effects of
+ * publish_via_partition_root. Note multiple eval of argument!
+ */
+#define RelationIsPublishedViaParent(relation) \
+ ((relation)->rd_options && \
+ ((relation)->rd_rel->relkind == RELKIND_RELATION) && \
+ ((StdRdOptions *) (relation)->rd_options)->publish_via_parent)
+
/*
* RelationGetParallelWorkers
* Returns the relation's parallel_workers reloption setting.
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index dd8fc99111..5ee5b8ea3b 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -5,6 +5,17 @@ CREATE ROLE regress_publication_user LOGIN SUPERUSER;
CREATE ROLE regress_publication_user2;
CREATE ROLE regress_publication_user_dummy LOGIN NOSUPERUSER;
SET SESSION AUTHORIZATION 'regress_publication_user';
+CREATE FUNCTION published_sync (VARIADIC pubnames text[])
+ RETURNS TABLE (published regclass, synced regclass) AS $$
+ -- Show each published table alongside the tables to be copied into it during
+ -- initial sync.
+ SELECT t.published, synced
+ FROM (SELECT DISTINCT relid::regclass AS published
+ FROM pg_get_publication_tables(VARIADIC pubnames)) t
+ JOIN LATERAL pg_get_publication_rels_to_sync(t.published, VARIADIC pubnames) synced
+ ON true
+ ORDER BY 1, 2;
+$$ LANGUAGE sql;
CREATE FUNCTION published_stream (VARIADIC pubnames text[])
RETURNS TABLE (pubname text, published regclass, synced regclass) AS $$
-- For each publication, show each published root alongside the tables which
@@ -1697,6 +1708,12 @@ SELECT * FROM pg_publication_tables;
pub | sch1 | tbl1 | {a} |
(1 row)
+SELECT * FROM published_sync('pub');
+ published | synced
+-----------+-----------
+ sch1.tbl1 | sch1.tbl1
+(1 row)
+
SELECT * FROM published_stream('pub');
pubname | published | synced
---------+-----------+-----------------
@@ -1730,6 +1747,12 @@ SELECT * FROM pg_publication_tables;
pub | sch2 | tbl1_part1 | {a} |
(1 row)
+SELECT * FROM published_sync('pub');
+ published | synced
+-----------------+-----------------
+ sch2.tbl1_part1 | sch2.tbl1_part1
+(1 row)
+
SELECT * FROM published_stream('pub');
pubname | published | synced
---------+-----------------+-----------------
@@ -1752,12 +1775,304 @@ SELECT * FROM pg_publication_tables;
pub | sch1 | tbl1 | {a} |
(1 row)
+-- Sanity check cases for publish_via_parent.
+CREATE TABLE sch1.iroot (a int);
+CREATE TABLE sch1.ipart1 (a int);
+CREATE TABLE sch1.ipart2 () INHERITS (sch1.iroot);
+-- should do nothing at all
+ALTER TABLE sch1.iroot SET (publish_via_parent = false);
+ALTER TABLE sch1.iroot RESET (publish_via_parent);
+-- marking roots between unrelated tables is not allowed
+ALTER TABLE sch1.ipart1 SET (publish_via_parent);
+ERROR: table "ipart1" does not inherit from any tables
+CREATE TABLE fail (a int) WITH (publish_via_parent);
+ERROR: table "fail" does not inherit from any tables
+-- establishing an inheritance relationship fixes the problem
+ALTER TABLE sch1.ipart1 INHERIT sch1.iroot,
+ SET (publish_via_parent);
+-- but multiple inheritance is not allowed
+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1,
+ SET (publish_via_parent);
+ERROR: table "ipart2" inherits from multiple tables
+-- once publish_via_parent is set, inheritance cannot be changed
+ALTER TABLE sch1.ipart2 SET (publish_via_parent);
+ALTER TABLE sch1.ipart2 NO INHERIT sch1.iroot;
+ERROR: relation option "publish_via_parent" prevents table "ipart2" from changing inheritance
+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
+ERROR: relation option "publish_via_parent" prevents table "ipart2" from changing inheritance
+-- table ownership must match, like ATTACH PARTITION
+CREATE ROLE regress_test_me;
+CREATE ROLE regress_test_not_me;
+CREATE TABLE root (a int);
+CREATE TABLE part () INHERITS (root);
+ALTER TABLE root OWNER TO regress_test_me;
+ALTER TABLE part OWNER TO regress_test_not_me;
+SET SESSION AUTHORIZATION regress_test_me;
+ALTER TABLE part SET (publish_via_parent); -- should fail
+ERROR: must be owner of table part
+RESET SESSION AUTHORIZATION;
+ALTER TABLE root OWNER TO regress_test_not_me;
+ALTER TABLE part OWNER TO regress_test_me;
+SET SESSION AUTHORIZATION regress_test_me;
+ALTER TABLE part SET (publish_via_parent); -- should also fail
+ERROR: must be owner of table root
+CREATE TABLE fail () INHERITS (root) WITH (publish_via_parent);
+ERROR: must be owner of table root
+RESET SESSION AUTHORIZATION;
+DROP TABLE root, part;
+DROP ROLE regress_test_not_me;
+DROP ROLE regress_test_me;
+-- Mixed publication settings for publish_via_partition_root, at different
+-- levels of the inheritance tree, to pin correct behavior in the worst cases.
+CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1) WITH (publish_via_parent);
+CREATE TABLE sch1.ipart1_a1 () INHERITS (sch1.ipart1_a);
+ALTER TABLE sch1.ipart1_a1 SET (publish_via_parent);
+CREATE PUBLICATION ipub_root FOR TABLE sch1.iroot;
+CREATE PUBLICATION ipub_part1 FOR TABLE ONLY sch1.ipart1, ONLY sch1.ipart1_a1
+ WITH (publish_via_partition_root);
+CREATE PUBLICATION ipub_other FOR TABLE ONLY sch1.iroot, ONLY sch1.ipart1_a,
+ ONLY sch1.ipart2
+ WITH (publish_via_partition_root);
+-- At this point, the published trees look like this:
+--
+-- ipub_root ipub_part1 (pubviaroot) ipub_other (pubviaroot)
+-- ------------------ ----------------------- -----------------------
+-- iroot iroot
+-- +- ipart1 ipart1 |
+-- | +- ipart1_a | +--- ipart1_a
+-- | +- ipart1_a1 +--- ipart1_a1 |
+-- +- ipart2 +- ipart2
+-- What a subscription to only ipub_root should see
+SELECT relid::regclass FROM pg_get_publication_tables('ipub_root');
+ relid
+----------------
+ sch1.iroot
+ sch1.ipart1
+ sch1.ipart2
+ sch1.ipart1_a
+ sch1.ipart1_a1
+(5 rows)
+
+SELECT * FROM published_sync('ipub_root');
+ published | synced
+----------------+----------------
+ sch1.iroot | sch1.iroot
+ sch1.ipart1 | sch1.ipart1
+ sch1.ipart2 | sch1.ipart2
+ sch1.ipart1_a | sch1.ipart1_a
+ sch1.ipart1_a1 | sch1.ipart1_a1
+(5 rows)
+
+SELECT * FROM published_stream('ipub_root');
+ pubname | published | synced
+-----------+----------------+----------------
+ ipub_root | sch1.iroot | sch1.iroot
+ ipub_root | sch1.ipart1 | sch1.ipart1
+ ipub_root | sch1.ipart2 | sch1.ipart2
+ ipub_root | sch1.ipart1_a | sch1.ipart1_a
+ ipub_root | sch1.ipart1_a1 | sch1.ipart1_a1
+(5 rows)
+
+-- What a subscription to only ipub_part1 should see
+SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
+ relid
+-------------
+ sch1.ipart1
+(1 row)
+
+SELECT * FROM published_sync('ipub_part1');
+ published | synced
+-------------+----------------
+ sch1.ipart1 | sch1.ipart1
+ sch1.ipart1 | sch1.ipart1_a1
+(2 rows)
+
+SELECT * FROM published_stream('ipub_part1');
+ pubname | published | synced
+------------+-------------+----------------
+ ipub_part1 | sch1.ipart1 | sch1.ipart1
+ ipub_part1 | sch1.ipart1 | sch1.ipart1_a1
+(2 rows)
+
+-- What a subscription to both ipub_root and ipub_part1 should see
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_root', 'ipub_part1') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+ pubname | relid
+------------+---------------
+ ipub_root | sch1.iroot
+ ipub_root | sch1.ipart1
+ ipub_root | sch1.ipart2
+ ipub_root | sch1.ipart1_a
+ ipub_part1 | sch1.ipart1
+(5 rows)
+
+SELECT * FROM published_sync('ipub_root', 'ipub_part1');
+ published | synced
+---------------+----------------
+ sch1.iroot | sch1.iroot
+ sch1.ipart1 | sch1.ipart1
+ sch1.ipart1 | sch1.ipart1_a1
+ sch1.ipart2 | sch1.ipart2
+ sch1.ipart1_a | sch1.ipart1_a
+(5 rows)
+
+SELECT * FROM published_stream('ipub_root', 'ipub_part1');
+ pubname | published | synced
+------------+---------------+----------------
+ ipub_root | sch1.iroot | sch1.iroot
+ ipub_root | sch1.ipart1 | sch1.ipart1
+ ipub_root | sch1.ipart1 | sch1.ipart1_a1
+ ipub_root | sch1.ipart2 | sch1.ipart2
+ ipub_root | sch1.ipart1_a | sch1.ipart1_a
+ ipub_part1 | sch1.ipart1 | sch1.ipart1
+ ipub_part1 | sch1.ipart1 | sch1.ipart1_a1
+(7 rows)
+
+-- What a subscription to both ipub_part1 and ipub_other should see
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_part1', 'ipub_other') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+ pubname | relid
+------------+-------------
+ ipub_part1 | sch1.ipart1
+ ipub_other | sch1.iroot
+(2 rows)
+
+SELECT * FROM published_sync('ipub_part1', 'ipub_other');
+ published | synced
+-------------+----------------
+ sch1.iroot | sch1.iroot
+ sch1.iroot | sch1.ipart2
+ sch1.iroot | sch1.ipart1_a
+ sch1.ipart1 | sch1.ipart1
+ sch1.ipart1 | sch1.ipart1_a1
+(5 rows)
+
+SELECT * FROM published_stream('ipub_part1', 'ipub_other');
+ pubname | published | synced
+------------+-------------+----------------
+ ipub_part1 | sch1.ipart1 | sch1.ipart1
+ ipub_part1 | sch1.ipart1 | sch1.ipart1_a1
+ ipub_other | sch1.iroot | sch1.iroot
+ ipub_other | sch1.iroot | sch1.ipart2
+ ipub_other | sch1.iroot | sch1.ipart1_a
+(5 rows)
+
+-- What a subscription to all three should see
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+ pubname | relid
+------------+-------------
+ ipub_root | sch1.iroot
+ ipub_root | sch1.ipart1
+ ipub_part1 | sch1.ipart1
+ ipub_other | sch1.iroot
+(4 rows)
+
+SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
+ published | synced
+-------------+----------------
+ sch1.iroot | sch1.iroot
+ sch1.iroot | sch1.ipart2
+ sch1.iroot | sch1.ipart1_a
+ sch1.ipart1 | sch1.ipart1
+ sch1.ipart1 | sch1.ipart1_a1
+(5 rows)
+
+SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
+ pubname | published | synced
+------------+-------------+----------------
+ ipub_root | sch1.iroot | sch1.iroot
+ ipub_root | sch1.iroot | sch1.ipart2
+ ipub_root | sch1.iroot | sch1.ipart1_a
+ ipub_root | sch1.ipart1 | sch1.ipart1
+ ipub_root | sch1.ipart1 | sch1.ipart1_a1
+ ipub_part1 | sch1.ipart1 | sch1.ipart1
+ ipub_part1 | sch1.ipart1 | sch1.ipart1_a1
+ ipub_other | sch1.iroot | sch1.iroot
+ ipub_other | sch1.iroot | sch1.ipart2
+ ipub_other | sch1.iroot | sch1.ipart1_a
+(10 rows)
+
+-- "Detaching" partitions should change the subscriptions
+ALTER TABLE sch1.ipart2 SET (publish_via_parent = false);
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+ pubname | relid
+------------+-------------
+ ipub_root | sch1.iroot
+ ipub_root | sch1.ipart1
+ ipub_root | sch1.ipart2
+ ipub_part1 | sch1.ipart1
+ ipub_other | sch1.iroot
+ ipub_other | sch1.ipart2
+(6 rows)
+
+SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
+ published | synced
+-------------+----------------
+ sch1.iroot | sch1.iroot
+ sch1.iroot | sch1.ipart1_a
+ sch1.ipart1 | sch1.ipart1
+ sch1.ipart1 | sch1.ipart1_a1
+ sch1.ipart2 | sch1.ipart2
+(5 rows)
+
+SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
+ pubname | published | synced
+------------+-------------+----------------
+ ipub_root | sch1.iroot | sch1.iroot
+ ipub_root | sch1.iroot | sch1.ipart1_a
+ ipub_root | sch1.ipart1 | sch1.ipart1
+ ipub_root | sch1.ipart1 | sch1.ipart1_a1
+ ipub_root | sch1.ipart2 | sch1.ipart2
+ ipub_part1 | sch1.ipart1 | sch1.ipart1
+ ipub_part1 | sch1.ipart1 | sch1.ipart1_a1
+ ipub_other | sch1.iroot | sch1.iroot
+ ipub_other | sch1.iroot | sch1.ipart1_a
+ ipub_other | sch1.ipart2 | sch1.ipart2
+(10 rows)
+
+ALTER TABLE sch1.ipart1_a1 RESET (publish_via_parent);
+SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
+ relid
+----------------
+ sch1.ipart1
+ sch1.ipart1_a1
+(2 rows)
+
+SELECT * FROM published_sync('ipub_part1');
+ published | synced
+----------------+----------------
+ sch1.ipart1 | sch1.ipart1
+ sch1.ipart1_a1 | sch1.ipart1_a1
+(2 rows)
+
+SELECT * FROM published_stream('ipub_part1');
+ pubname | published | synced
+------------+----------------+----------------
+ ipub_part1 | sch1.ipart1 | sch1.ipart1
+ ipub_part1 | sch1.ipart1_a1 | sch1.ipart1_a1
+(2 rows)
+
+DROP PUBLICATION ipub_other;
+DROP PUBLICATION ipub_part1;
+DROP PUBLICATION ipub_root;
RESET client_min_messages;
DROP PUBLICATION pub;
DROP TABLE sch1.tbl1;
+DROP TABLE sch1.ipart1_a1;
+DROP TABLE sch1.ipart1_a;
+DROP TABLE sch1.ipart1;
+DROP TABLE sch1.ipart2;
+DROP TABLE sch1.iroot;
DROP SCHEMA sch1 cascade;
DROP SCHEMA sch2 cascade;
DROP FUNCTION published_stream;
+DROP FUNCTION published_sync;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
DROP ROLE regress_publication_user_dummy;
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index 6bf9554d48..d1ea777ef9 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -6,6 +6,18 @@ CREATE ROLE regress_publication_user2;
CREATE ROLE regress_publication_user_dummy LOGIN NOSUPERUSER;
SET SESSION AUTHORIZATION 'regress_publication_user';
+CREATE FUNCTION published_sync (VARIADIC pubnames text[])
+ RETURNS TABLE (published regclass, synced regclass) AS $$
+ -- Show each published table alongside the tables to be copied into it during
+ -- initial sync.
+ SELECT t.published, synced
+ FROM (SELECT DISTINCT relid::regclass AS published
+ FROM pg_get_publication_tables(VARIADIC pubnames)) t
+ JOIN LATERAL pg_get_publication_rels_to_sync(t.published, VARIADIC pubnames) synced
+ ON true
+ ORDER BY 1, 2;
+$$ LANGUAGE sql;
+
CREATE FUNCTION published_stream (VARIADIC pubnames text[])
RETURNS TABLE (pubname text, published regclass, synced regclass) AS $$
-- For each publication, show each published root alongside the tables which
@@ -1076,6 +1088,7 @@ SELECT * FROM pg_publication_tables;
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
SELECT * FROM pg_publication_tables;
+SELECT * FROM published_sync('pub');
SELECT * FROM published_stream('pub');
DROP PUBLICATION pub;
@@ -1091,6 +1104,7 @@ SELECT * FROM pg_publication_tables;
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
SELECT * FROM pg_publication_tables;
+SELECT * FROM published_sync('pub');
SELECT * FROM published_stream('pub');
DROP PUBLICATION pub;
@@ -1105,12 +1119,138 @@ ALTER TABLE sch1.tbl1 ATTACH PARTITION sch1.tbl1_part3 FOR VALUES FROM (20) to (
CREATE PUBLICATION pub FOR TABLES IN SCHEMA sch1 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
SELECT * FROM pg_publication_tables;
+-- Sanity check cases for publish_via_parent.
+CREATE TABLE sch1.iroot (a int);
+CREATE TABLE sch1.ipart1 (a int);
+CREATE TABLE sch1.ipart2 () INHERITS (sch1.iroot);
+
+-- should do nothing at all
+ALTER TABLE sch1.iroot SET (publish_via_parent = false);
+ALTER TABLE sch1.iroot RESET (publish_via_parent);
+
+-- marking roots between unrelated tables is not allowed
+ALTER TABLE sch1.ipart1 SET (publish_via_parent);
+CREATE TABLE fail (a int) WITH (publish_via_parent);
+
+-- establishing an inheritance relationship fixes the problem
+ALTER TABLE sch1.ipart1 INHERIT sch1.iroot,
+ SET (publish_via_parent);
+
+-- but multiple inheritance is not allowed
+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1,
+ SET (publish_via_parent);
+
+-- once publish_via_parent is set, inheritance cannot be changed
+ALTER TABLE sch1.ipart2 SET (publish_via_parent);
+ALTER TABLE sch1.ipart2 NO INHERIT sch1.iroot;
+ALTER TABLE sch1.ipart2 INHERIT sch1.ipart1;
+
+-- table ownership must match, like ATTACH PARTITION
+CREATE ROLE regress_test_me;
+CREATE ROLE regress_test_not_me;
+CREATE TABLE root (a int);
+CREATE TABLE part () INHERITS (root);
+
+ALTER TABLE root OWNER TO regress_test_me;
+ALTER TABLE part OWNER TO regress_test_not_me;
+SET SESSION AUTHORIZATION regress_test_me;
+ALTER TABLE part SET (publish_via_parent); -- should fail
+RESET SESSION AUTHORIZATION;
+
+ALTER TABLE root OWNER TO regress_test_not_me;
+ALTER TABLE part OWNER TO regress_test_me;
+SET SESSION AUTHORIZATION regress_test_me;
+ALTER TABLE part SET (publish_via_parent); -- should also fail
+CREATE TABLE fail () INHERITS (root) WITH (publish_via_parent);
+RESET SESSION AUTHORIZATION;
+
+DROP TABLE root, part;
+DROP ROLE regress_test_not_me;
+DROP ROLE regress_test_me;
+
+-- Mixed publication settings for publish_via_partition_root, at different
+-- levels of the inheritance tree, to pin correct behavior in the worst cases.
+CREATE TABLE sch1.ipart1_a () INHERITS (sch1.ipart1) WITH (publish_via_parent);
+CREATE TABLE sch1.ipart1_a1 () INHERITS (sch1.ipart1_a);
+ALTER TABLE sch1.ipart1_a1 SET (publish_via_parent);
+
+CREATE PUBLICATION ipub_root FOR TABLE sch1.iroot;
+CREATE PUBLICATION ipub_part1 FOR TABLE ONLY sch1.ipart1, ONLY sch1.ipart1_a1
+ WITH (publish_via_partition_root);
+CREATE PUBLICATION ipub_other FOR TABLE ONLY sch1.iroot, ONLY sch1.ipart1_a,
+ ONLY sch1.ipart2
+ WITH (publish_via_partition_root);
+
+-- At this point, the published trees look like this:
+--
+-- ipub_root ipub_part1 (pubviaroot) ipub_other (pubviaroot)
+-- ------------------ ----------------------- -----------------------
+-- iroot iroot
+-- +- ipart1 ipart1 |
+-- | +- ipart1_a | +--- ipart1_a
+-- | +- ipart1_a1 +--- ipart1_a1 |
+-- +- ipart2 +- ipart2
+
+-- What a subscription to only ipub_root should see
+SELECT relid::regclass FROM pg_get_publication_tables('ipub_root');
+SELECT * FROM published_sync('ipub_root');
+SELECT * FROM published_stream('ipub_root');
+
+-- What a subscription to only ipub_part1 should see
+SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
+SELECT * FROM published_sync('ipub_part1');
+SELECT * FROM published_stream('ipub_part1');
+
+-- What a subscription to both ipub_root and ipub_part1 should see
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_root', 'ipub_part1') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+SELECT * FROM published_sync('ipub_root', 'ipub_part1');
+SELECT * FROM published_stream('ipub_root', 'ipub_part1');
+
+-- What a subscription to both ipub_part1 and ipub_other should see
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_part1', 'ipub_other') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+SELECT * FROM published_sync('ipub_part1', 'ipub_other');
+SELECT * FROM published_stream('ipub_part1', 'ipub_other');
+
+-- What a subscription to all three should see
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
+SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
+
+-- "Detaching" partitions should change the subscriptions
+ALTER TABLE sch1.ipart2 SET (publish_via_parent = false);
+SELECT pubname, relid::regclass
+ FROM pg_get_publication_tables('ipub_root', 'ipub_part1', 'ipub_other') t
+ JOIN pg_publication p ON (p.oid = t.pubid);
+SELECT * FROM published_sync('ipub_root', 'ipub_part1', 'ipub_other');
+SELECT * FROM published_stream('ipub_root', 'ipub_part1', 'ipub_other');
+
+ALTER TABLE sch1.ipart1_a1 RESET (publish_via_parent);
+SELECT relid::regclass FROM pg_get_publication_tables('ipub_part1');
+SELECT * FROM published_sync('ipub_part1');
+SELECT * FROM published_stream('ipub_part1');
+
+DROP PUBLICATION ipub_other;
+DROP PUBLICATION ipub_part1;
+DROP PUBLICATION ipub_root;
+
RESET client_min_messages;
DROP PUBLICATION pub;
DROP TABLE sch1.tbl1;
+DROP TABLE sch1.ipart1_a1;
+DROP TABLE sch1.ipart1_a;
+DROP TABLE sch1.ipart1;
+DROP TABLE sch1.ipart2;
+DROP TABLE sch1.iroot;
DROP SCHEMA sch1 cascade;
DROP SCHEMA sch2 cascade;
DROP FUNCTION published_stream;
+DROP FUNCTION published_sync;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_publication_user, regress_publication_user2;
diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl
index 275fb3b525..09e3f2d028 100644
--- a/src/test/subscription/t/013_partition.pl
+++ b/src/test/subscription/t/013_partition.pl
@@ -388,6 +388,59 @@ $node_subscriber1->append_conf('postgresql.conf',
"log_min_messages = warning");
$node_subscriber1->reload;
+# Make sure standard inheritance setups aren't broken by the new
+# publish_via_parent handling.
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1 (a int, b text)");
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1_1 (LIKE itab1)");
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1_2 (LIKE itab1)");
+
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab1 (a int, b text)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab1_1 (CHECK (a = 1)) INHERITS (itab1)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab1_2 (CHECK (a = 2)) INHERITS (itab1)");
+
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE itab1_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE itab1_2 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1 VALUES (0, 'itab1')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1_1 VALUES (1, 'itab1')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1_2 VALUES (2, 'itab1')");
+
+# Regression: Create a publication for an unrelated table and set
+# publish_via_partition_root. This should have no effect at all, since itab1
+# isn't supposed to be using this publication...
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION dummy_pub FOR TABLE tab1 WITH (publish_via_partition_root = true)");
+$node_subscriber2->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub2 ADD PUBLICATION dummy_pub");
+$node_subscriber2->wait_for_subscription_sync;
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab1");
+is($result, qq(0|itab1), 'initial data synced for itab1 on subscriber 2');
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab1_1");
+is($result, qq(1|itab1), 'initial data synced for itab1_1 on subscriber 2');
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab1_2");
+is($result, qq(2|itab1), 'initial data synced for itab1_2 on subscriber 2');
+
+$node_publisher->safe_psql('postgres',
+ "DROP TABLE itab1 CASCADE;");
+$node_subscriber2->safe_psql('postgres',
+ "DROP TABLE itab1 CASCADE;");
+
+$node_subscriber2->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub2 DROP PUBLICATION dummy_pub");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION dummy_pub");
+
# Tests for replication using root table identity and schema
# publisher
@@ -886,4 +939,263 @@ $result = $node_subscriber2->safe_psql('postgres',
"SELECT a, b, c FROM tab5_1 ORDER BY 1");
is($result, qq(4||1), 'updates of tab5 replicated correctly');
+# Test that replication works for older inheritance/trigger setups as well.
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab1 (a int, b text)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab1_1 (CHECK (a = 1)) INHERITS (itab1)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab1_2 (CHECK (a = 2)) INHERITS (itab1)");
+
+$node_publisher->safe_psql('postgres', "
+ CREATE OR REPLACE FUNCTION itab1_trigger()
+ RETURNS TRIGGER AS \$\$
+ BEGIN
+ IF ( NEW.a = 1 ) THEN INSERT INTO itab1_1 VALUES (NEW.*);
+ ELSIF ( NEW.a = 2 ) THEN INSERT INTO itab1_2 VALUES (NEW.*);
+ ELSE RETURN NEW;
+ END IF;
+ RETURN NULL;
+ END;
+ \$\$
+ LANGUAGE plpgsql;");
+$node_publisher->safe_psql('postgres', "
+ CREATE TRIGGER itab1_trigger
+ BEFORE INSERT ON itab1
+ FOR EACH ROW EXECUTE FUNCTION itab1_trigger();");
+
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE itab1_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE itab1_2 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab2 (a int, b text)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab2_1 (CHECK (a = 1)) INHERITS (itab2)");
+
+$node_publisher->safe_psql('postgres', "
+ CREATE OR REPLACE FUNCTION itab2_trigger()
+ RETURNS TRIGGER AS \$\$
+ BEGIN
+ IF ( NEW.a = 1 ) THEN INSERT INTO itab2_1 VALUES (NEW.*);
+ ELSE RETURN NEW;
+ END IF;
+ RETURN NULL;
+ END;
+ \$\$
+ LANGUAGE plpgsql;");
+$node_publisher->safe_psql('postgres', "
+ CREATE TRIGGER itab2_trigger
+ BEFORE INSERT ON itab2
+ FOR EACH ROW EXECUTE FUNCTION itab2_trigger();");
+
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE itab2_1 SET (publish_via_parent = true)");
+
+# itab2_1 should be published using its own identity here, since its parent is
+# not included. itab1_1 should be published via its parent, itab1, without
+# duplicating the rows.
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION pub_viaroot ADD TABLE itab1, itab1_1, itab2_1");
+
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1 VALUES (0, 'itab1')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1 VALUES (1, 'itab1')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1 VALUES (2, 'itab1')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab2 VALUES (0, 'itab2')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab2 VALUES (1, 'itab2')");
+
+# Subscriber 1 only subscribes to some of the partitions, and does not set up
+# partition triggers, to check for the correct routing.
+$node_subscriber1->safe_psql('postgres',
+ "CREATE TABLE itab1 (a int, b text)");
+$node_subscriber1->safe_psql('postgres',
+ "CREATE TABLE itab1_1 (CHECK (a = 1)) INHERITS (itab1)");
+$node_subscriber1->safe_psql('postgres',
+ "CREATE TABLE itab2_1 (a int, b text)");
+
+# Subscriber 2 has different partition names for itab1, and it doesn't partition
+# itab2 at all.
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1 (a int, b text)");
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1_part1 (CHECK (a = 1)) INHERITS (itab1)");
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab1_part2 (CHECK (a = 2)) INHERITS (itab1)");
+
+$node_subscriber2->safe_psql('postgres', "
+ CREATE OR REPLACE FUNCTION itab_trigger()
+ RETURNS TRIGGER AS \$\$
+ BEGIN
+ IF ( NEW.a = 1 ) THEN INSERT INTO public.itab1_part1 VALUES (NEW.*);
+ ELSIF ( NEW.a = 2 ) THEN INSERT INTO public.itab1_part2 VALUES (NEW.*);
+ ELSE RETURN NEW;
+ END IF;
+ RETURN NULL;
+ END;
+ \$\$
+ LANGUAGE plpgsql;");
+$node_subscriber2->safe_psql('postgres', "
+ CREATE TRIGGER itab_trigger
+ BEFORE INSERT ON itab1
+ FOR EACH ROW EXECUTE FUNCTION itab_trigger();");
+$node_subscriber2->safe_psql('postgres', "
+ ALTER TABLE itab1 ENABLE ALWAYS TRIGGER itab_trigger;");
+
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab2 (a int, b text)");
+
+$node_subscriber1->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub_viaroot REFRESH PUBLICATION");
+$node_subscriber2->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub2 REFRESH PUBLICATION");
+
+$node_subscriber1->wait_for_subscription_sync;
+$node_subscriber2->wait_for_subscription_sync;
+
+# check that data is synced correctly
+
+$result = $node_subscriber1->safe_psql('postgres',
+ "SELECT a, b FROM itab1 ORDER BY 1, 2");
+is($result, qq(0|itab1
+1|itab1
+2|itab1), 'initial data synced for itab1 on subscriber 1');
+
+# all of the data should have been routed to itab1 directly (there are no
+# triggers on subscriber 1 to move it elsewhere)
+$result = $node_subscriber1->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab1 ORDER BY 1, 2");
+is($result, qq(0|itab1
+1|itab1
+2|itab1), 'initial data correctly routed for itab1 on subscriber 1');
+
+$result = $node_subscriber1->safe_psql('postgres',
+ "SELECT a, b FROM itab2_1 ORDER BY 1, 2");
+is($result, qq(1|itab2), 'initial data synced for itab2_1 on subscriber 1');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab1 ORDER BY 1, 2");
+is($result, qq(0|itab1
+1|itab1
+2|itab1), 'initial data synced for itab1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab1 ORDER BY 1, 2");
+is($result, qq(0|itab1), 'initial data correctly routed for itab1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab2 ORDER BY 1, 2");
+is($result, qq(0|itab2
+1|itab2), 'initial data synced for itab2 on subscriber 2');
+
+# make sure new data is also correctly routed to the roots
+$node_publisher->safe_psql('postgres', "INSERT INTO itab1 VALUES (1, 'itab1-new')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab2 VALUES (1, 'itab2-new')");
+
+$node_publisher->wait_for_catchup('sub_viaroot');
+$node_publisher->wait_for_catchup('sub2');
+
+$result = $node_subscriber1->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab1 ORDER BY 1, 2");
+is($result, qq(0|itab1
+1|itab1
+1|itab1-new
+2|itab1), 'new data routed for itab1 on subscriber 1');
+
+$result = $node_subscriber1->safe_psql('postgres',
+ "SELECT a, b FROM itab2_1 ORDER BY 1, 2");
+is($result, qq(1|itab2
+1|itab2-new), 'new data routed for itab2_1 on subscriber 1');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab1 ORDER BY 1, 2");
+is($result, qq(0|itab1
+1|itab1
+1|itab1-new
+2|itab1), 'new data routed for itab1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab1_part1 ORDER BY 1, 2");
+is($result, qq(1|itab1
+1|itab1-new), 'new data moved to itab1_part1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM itab2 ORDER BY 1, 2");
+is($result, qq(0|itab2
+1|itab2
+1|itab2-new), 'new data routed for itab2 on subscriber 2');
+
+# Finally, check the effect of mixed publish_via_partition_root settings on
+# logical roots.
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab3 (a int, b text)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab3_1 () INHERITS (itab3)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE itab3_1_1 () INHERITS (itab3_1)");
+
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE itab3_1 SET (publish_via_parent = true)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE itab3_1_1 SET (publish_via_parent = true)");
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO itab3 VALUES (1, 'itab3')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO itab3_1 VALUES (2, 'itab3_1')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO itab3_1_1 VALUES (3, 'itab3_1_1')");
+
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION ipub3 FOR TABLE itab3 *");
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION ipub3_1 FOR TABLE itab3_1 * WITH (publish_via_partition_root = true)");
+
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab3 (a int, b text)");
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab3_1 () INHERITS (itab3)");
+$node_subscriber2->safe_psql('postgres',
+ "CREATE TABLE itab3_1_1 () INHERITS (itab3_1)");
+
+$node_subscriber2->safe_psql('postgres',
+ "CREATE SUBSCRIPTION mixed CONNECTION '$publisher_connstr' PUBLICATION ipub3, ipub3_1");
+$node_subscriber2->wait_for_subscription_sync;
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3");
+is($result, qq(1|itab3), 'initial data routed for itab3 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1 ORDER BY 1, 2");
+is($result, qq(2|itab3_1
+3|itab3_1_1), 'initial data routed for itab3_1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1_1");
+is($result, qq(), 'initial data routed for itab3_1_1 on subscriber 2');
+
+# make sure new data is also correctly routed to the roots
+$node_publisher->safe_psql('postgres', "INSERT INTO itab3 VALUES (4, 'itab3_new')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1 VALUES (4, 'itab3_1_new')");
+$node_publisher->safe_psql('postgres', "INSERT INTO itab3_1_1 VALUES (4, 'itab3_1_1_new')");
+
+$node_publisher->wait_for_catchup('mixed');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3 ORDER BY 1, 2");
+is($result, qq(1|itab3
+4|itab3_new), 'new data routed for itab3 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1 ORDER BY 1, 2");
+is($result, qq(2|itab3_1
+3|itab3_1_1
+4|itab3_1_1_new
+4|itab3_1_new), 'new data routed for itab3_1 on subscriber 2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM ONLY itab3_1_1");
+is($result, qq(), 'new data routed for itab3_1_1 on subscriber 2');
+
done_testing();
--
2.25.1
view thread (18+ 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]
Subject: Re: RFC: logical publication via inheritance root?
In-Reply-To: <[email protected]>
* 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