Re: adding partitioned tables to publications
Amit Langote <[email protected]>
| Newsgroups | gmane.comp.db.postgresql.devel.general |
|---|---|
| Message-ID | <CA+HiwqE2ie6_T01fMzsf7c5=hCGoVdj2yTStCJ78zOmayMVWbw@mail.gmail.com> |
On Mon, Jan 6, 2020 at 8:25 PM Rafia Sabih <[email protected]> wrote: > Hi Amit, > > I went through this patch set once again today and here are my two cents. Thanks Rafia. Rebased and updated to address your comments. Regards, Amit
v8-0001-Support-adding-partitioned-tables-to-publication.patch
(text/plain, 40.1 KB)
From 8a1b409f3217e53d557288fac3c0843e53d0710e Mon Sep 17 00:00:00 2001 From: amit <[email protected]> Date: Thu, 7 Nov 2019 18:19:33 +0900 Subject: [PATCH v8 1/4] Support adding partitioned tables to publication --- doc/src/sgml/logical-replication.sgml | 15 +-- doc/src/sgml/ref/create_publication.sgml | 27 +++-- src/backend/catalog/pg_publication.c | 42 +++++--- src/backend/commands/copy.c | 2 +- src/backend/commands/publicationcmds.c | 12 ++- src/backend/commands/subscriptioncmds.c | 117 +++++++++++++++++--- src/backend/executor/execMain.c | 7 +- src/backend/executor/execPartition.c | 5 +- src/backend/executor/execReplication.c | 47 ++++---- src/backend/executor/nodeModifyTable.c | 6 +- src/backend/replication/logical/tablesync.c | 1 + src/backend/replication/pgoutput/pgoutput.c | 41 +++++-- src/bin/pg_dump/pg_dump.c | 8 +- src/include/catalog/pg_publication.h | 1 + src/include/executor/executor.h | 8 +- src/test/regress/expected/publication.out | 21 +++- src/test/regress/sql/publication.sql | 12 ++- src/test/subscription/t/013_partition.pl | 161 ++++++++++++++++++++++++++++ 18 files changed, 447 insertions(+), 86 deletions(-) create mode 100644 src/test/subscription/t/013_partition.pl diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index f657d1d06e..4584cb82f6 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -402,13 +402,14 @@ <listitem> <para> - Replication is only possible from base tables to base tables. That is, - the tables on the publication and on the subscription side must be normal - tables, not views, materialized views, partition root tables, or foreign - tables. In the case of partitions, you can therefore replicate a - partition hierarchy one-to-one, but you cannot currently replicate to a - differently partitioned setup. Attempts to replicate tables other than - base tables will result in an error. + Replication is only supported by regular and partitioned tables, although + the type of the table must match between the two servers, that is, one + cannot replicate from a regular table into a partitioned able or vice + versa. Also, when replicating between partitioned tables, the actual + replication occurs between leaf partitions, so the partitions on the two + servers must match one-to-one. Attempts to replicate other types of + relations such as views, materialized views, or foreign tables, will + result in an error. </para> </listitem> </itemizedlist> diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml index 99f87ca393..848779a00f 100644 --- a/doc/src/sgml/ref/create_publication.sgml +++ b/doc/src/sgml/ref/create_publication.sgml @@ -68,15 +68,25 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable> that table is added to the publication. If <literal>ONLY</literal> is not specified, the table and all its descendant tables (if any) are added. Optionally, <literal>*</literal> can be specified after the table name to - explicitly indicate that descendant tables are included. + explicitly indicate that descendant tables are included. However, adding + a partitioned table to a publication never explicitly adds its partitions, + because partitions are implicitly published due to the partitioned table + being added to the publication. </para> <para> - Only persistent base tables can be part of a publication. Temporary - tables, unlogged tables, foreign tables, materialized views, regular - views, and partitioned tables cannot be part of a publication. To - replicate a partitioned table, add the individual partitions to the - publication. + Only persistent base tables and partitioned tables can be part of a + publication. Temporary tables, unlogged tables, foreign tables, + materialized views, regular views cannot be part of a publication. + </para> + + <para> + When a partitioned table is added to a publication, all of its existing + and future partitions are also implicitly considered to be part of the + publication. So, any <command>INSERT</command>, <command>UPDATE</update>, + and <command>DELETE</command>, and <command>TRUNCATE</command> operations + that are directly applied to a partition are also published via its + ancestors' publications. </para> </listitem> </varlistentry> @@ -132,6 +142,11 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable> empty set of tables. That is useful if tables are to be added later. </para> + <para> + Partitioned tables are not considered when <literal>FOR ALL TABLES</literal> + is specified. + </para> + <para> The creation of a publication does not start replication. It only defines a grouping and filtering logic for future subscribers. diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index c5eea7af3f..fb369dbe17 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -26,6 +26,7 @@ #include "catalog/namespace.h" #include "catalog/objectaccess.h" #include "catalog/objectaddress.h" +#include "catalog/partition.h" #include "catalog/pg_publication.h" #include "catalog/pg_publication_rel.h" #include "catalog/pg_type.h" @@ -47,17 +48,9 @@ static void check_publication_add_relation(Relation targetrel) { - /* Give more specific error for partitioned tables */ - if (RelationGetForm(targetrel)->relkind == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("\"%s\" is a partitioned table", - RelationGetRelationName(targetrel)), - errdetail("Adding partitioned tables to publications is not supported."), - errhint("You can add the table partitions individually."))); - - /* Must be table */ - if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION) + /* Must be a regular or partitioned table */ + if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION && + RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("\"%s\" is not a table", @@ -103,7 +96,8 @@ check_publication_add_relation(Relation targetrel) static bool is_publishable_class(Oid relid, Form_pg_class reltuple) { - return reltuple->relkind == RELKIND_RELATION && + return (reltuple->relkind == RELKIND_RELATION || + reltuple->relkind == RELKIND_PARTITIONED_TABLE) && !IsCatalogRelationOid(relid) && reltuple->relpersistence == RELPERSISTENCE_PERMANENT && relid >= FirstNormalObjectId; @@ -230,7 +224,7 @@ GetRelationPublications(Oid relid) CatCList *pubrellist; int i; - /* Find all publications associated with the relation. */ + /* Finds all publications associated with the relation. */ pubrellist = SearchSysCacheList1(PUBLICATIONRELMAP, ObjectIdGetDatum(relid)); for (i = 0; i < pubrellist->n_members; i++) @@ -246,6 +240,28 @@ GetRelationPublications(Oid relid) return result; } +/* + * Finds all publications that publish changes to the input relation's + * ancestors. + */ +List * +GetRelationAncestorPublications(Oid relid) +{ + List *ancestors = get_partition_ancestors(relid); + List *ancestor_pubids = NIL; + ListCell *lc; + + foreach(lc, ancestors) + { + Oid ancestor = lfirst_oid(lc); + List *rel_publishers = GetRelationPublications(ancestor); + + ancestor_pubids = list_concat_copy(ancestor_pubids, rel_publishers); + } + + return ancestor_pubids; +} + /* * Gets list of relation oids for a publication. * diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index c93a788798..5a75419caf 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -2837,7 +2837,7 @@ CopyFrom(CopyState cstate) target_resultRelInfo = resultRelInfo; /* Verify the named relation is a valid target for INSERT */ - CheckValidResultRel(resultRelInfo, CMD_INSERT); + CheckValidResultRel(resultRelInfo, NULL, CMD_INSERT); ExecOpenIndices(resultRelInfo, false); diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index f96cb42adc..8f38c63ad2 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -498,7 +498,8 @@ RemovePublicationRelById(Oid proid) /* * Open relations specified by a RangeVar list. - * The returned tables are locked in ShareUpdateExclusiveLock mode. + * The returned tables are locked in ShareUpdateExclusiveLock mode in order to + * add them to a publication. */ static List * OpenTableList(List *tables) @@ -539,8 +540,13 @@ OpenTableList(List *tables) rels = lappend(rels, rel); relids = lappend_oid(relids, myrelid); - /* Add children of this rel, if requested */ - if (recurse) + /* + * Add children of this rel, if requested, so that they too are added + * to the publication. A partitioned table can't have any inheritance + * children other than its partitions, which need not be explicitly + * added to the publication. + */ + if (recurse && rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) { List *children; ListCell *child; diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 95962b4a3e..786b15eb27 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -44,7 +44,19 @@ #include "utils/memutils.h" #include "utils/syscache.h" -static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); +/* + * Structure used by fetch_publication_tables to describe a published table. + * The information is used by the callers of fetch_publication_tables to + * generate a pg_subscription_rel catalog entry for the table. + */ +typedef struct PublishedTable +{ + RangeVar *rv; + + char relkind; +} PublishedTable; + +static List *fetch_publication_tables(WalReceiverConn *wrconn, List *publications); /* * Common option parsing function for CREATE and ALTER SUBSCRIPTION commands. @@ -453,18 +465,42 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) * Get the table list from publisher and build local table status * info. */ - tables = fetch_table_list(wrconn, publications); + tables = fetch_publication_tables(wrconn, publications); foreach(lc, tables) { - RangeVar *rv = (RangeVar *) lfirst(lc); + PublishedTable *pt = (PublishedTable *) lfirst(lc); + RangeVar *rv = pt->rv; Oid relid; + char local_relkind; relid = RangeVarGetRelid(rv, AccessShareLock, false); + local_relkind = get_rel_relkind(relid); /* Check for supported relkind. */ - CheckSubscriptionRelkind(get_rel_relkind(relid), + CheckSubscriptionRelkind(local_relkind, rv->schemaname, rv->relname); + /* + * Currently, partitioned table replication occurs between leaf + * partitions, so both the source and the target tables must be + * partitioned. + */ + if (pt->relkind == RELKIND_RELATION && + local_relkind == RELKIND_PARTITIONED_TABLE) + ereport(ERROR, + (errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("cannot use relation \"%s.%s\" as logical replication target", + rv->schemaname, rv->relname), + errdetail("\"%s.%s\" is a partitioned table whereas it is a regular table on publication server.", + rv->schemaname, rv->relname))); + + /* + * A partitioned table doesn't need local state, because the + * state is managed for individual partitions instead. + */ + if (pt->relkind == RELKIND_PARTITIONED_TABLE) + continue; + AddSubscriptionRelState(subid, relid, table_state, InvalidXLogRecPtr); } @@ -530,7 +566,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) (errmsg("could not connect to the publisher: %s", err))); /* Get the table list from publisher. */ - pubrel_names = fetch_table_list(wrconn, sub->publications); + pubrel_names = fetch_publication_tables(wrconn, sub->publications); /* We are done with the remote side, close connection. */ walrcv_disconnect(wrconn); @@ -565,15 +601,39 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) foreach(lc, pubrel_names) { - RangeVar *rv = (RangeVar *) lfirst(lc); + PublishedTable *pt = (PublishedTable *) lfirst(lc); + RangeVar *rv = pt->rv; Oid relid; + char local_relkind; relid = RangeVarGetRelid(rv, AccessShareLock, false); + local_relkind = get_rel_relkind(relid); /* Check for supported relkind. */ - CheckSubscriptionRelkind(get_rel_relkind(relid), + CheckSubscriptionRelkind(local_relkind, rv->schemaname, rv->relname); + /* + * Currently, partitioned table replication occurs between leaf + * partitions, so both the source and the target tables must be + * partitioned. + */ + if (pt->relkind == RELKIND_RELATION && + local_relkind == RELKIND_PARTITIONED_TABLE) + ereport(ERROR, + (errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("cannot use relation \"%s.%s\" as logical replication target", + rv->schemaname, rv->relname), + errdetail("\"%s.%s\" is a partitioned table whereas it is a regular table on publication server.", + rv->schemaname, rv->relname))); + + /* + * A partitioned table doesn't need local state, because the + * state is managed for individual partitions instead. + */ + if (pt->relkind == RELKIND_PARTITIONED_TABLE) + continue; + pubrel_local_oids[off++] = relid; if (!bsearch(&relid, subrel_local_oids, @@ -1121,15 +1181,17 @@ AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId) /* * Get the list of tables which belong to specified publications on the - * publisher connection. + * publisher connection to create a subscription state (pg_subscription_rel + * entry) for each. For partitioned tables, subscription state is maintained + * per partition, so partitions are fetched too. */ static List * -fetch_table_list(WalReceiverConn *wrconn, List *publications) +fetch_publication_tables(WalReceiverConn *wrconn, List *publications) { WalRcvExecResult *res; StringInfoData cmd; TupleTableSlot *slot; - Oid tableRow[2] = {TEXTOID, TEXTOID}; + Oid tableRow[3] = {TEXTOID, TEXTOID, CHAROID}; ListCell *lc; bool first; List *tablelist = NIL; @@ -1137,9 +1199,30 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) Assert(list_length(publications) > 0); initStringInfo(&cmd); - appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n" + appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.tablename, s.relkind FROM (\n" + " SELECT t.pubname, t.schemaname, t.tablename, c.relkind\n" " FROM pg_catalog.pg_publication_tables t\n" - " WHERE t.pubname IN ("); + " JOIN pg_catalog.pg_class c \n" + " ON t.schemaname = c.relnamespace::pg_catalog.regnamespace::name\n" + " AND t.tablename = c.relname \n"); + + /* + * As of v13, partitioned tables can be published, although their changes + * are published as their partitions', so we will need the partitions in + * the result. + */ + if (walrcv_server_version(wrconn) >= 130000) + appendStringInfoString(&cmd, " UNION\n" + " SELECT t.pubname, s.schemaname, s.tablename, s.relkind\n" + " FROM pg_catalog.pg_publication_tables t,\n" + " LATERAL (SELECT c.relnamespace::regnamespace::name, c.relname, c.relkind\n" + " FROM pg_class c\n" + " JOIN pg_partition_tree(t.schemaname || '.' || t.tablename) p\n" + " ON p.relid = c.oid\n" + " WHERE p.level > 0) AS s(schemaname, tablename, relkind)\n"); + + appendStringInfoString(&cmd, ") s WHERE s.pubname IN ("); + first = true; foreach(lc, publications) { @@ -1154,7 +1237,7 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) } appendStringInfoChar(&cmd, ')'); - res = walrcv_exec(wrconn, cmd.data, 2, tableRow); + res = walrcv_exec(wrconn, cmd.data, 3, tableRow); pfree(cmd.data); if (res->status != WALRCV_OK_TUPLES) @@ -1169,15 +1252,17 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) char *nspname; char *relname; bool isnull; - RangeVar *rv; + PublishedTable *pt = palloc(sizeof(PublishedTable)); nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); Assert(!isnull); relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); Assert(!isnull); + pt->rv = makeRangeVar(pstrdup(nspname), pstrdup(relname), -1); + pt->relkind = DatumGetChar(slot_getattr(slot, 3, &isnull)); + Assert(!isnull); - rv = makeRangeVar(pstrdup(nspname), pstrdup(relname), -1); - tablelist = lappend(tablelist, rv); + tablelist = lappend(tablelist, pt); ExecClearTuple(slot); } diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 4181a7e343..96671ca49e 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -1073,7 +1073,9 @@ InitPlan(QueryDesc *queryDesc, int eflags) * CheckValidRowMarkRel. */ void -CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation) +CheckValidResultRel(ResultRelInfo *resultRelInfo, + ResultRelInfo *rootResultRelInfo, + CmdType operation) { Relation resultRel = resultRelInfo->ri_RelationDesc; TriggerDesc *trigDesc = resultRel->trigdesc; @@ -1083,7 +1085,8 @@ CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation) { case RELKIND_RELATION: case RELKIND_PARTITIONED_TABLE: - CheckCmdReplicaIdentity(resultRel, operation); + CheckCmdReplicaIdentity(resultRelInfo, rootResultRelInfo, + operation); break; case RELKIND_SEQUENCE: ereport(ERROR, diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c index c13b1d3501..2a639011b8 100644 --- a/src/backend/executor/execPartition.c +++ b/src/backend/executor/execPartition.c @@ -384,7 +384,8 @@ ExecFindPartition(ModifyTableState *mtstate, rri = elem->rri; /* Verify this ResultRelInfo allows INSERTs */ - CheckValidResultRel(rri, CMD_INSERT); + CheckValidResultRel(rri, rootResultRelInfo, + CMD_INSERT); /* Set up the PartitionRoutingInfo for it */ ExecInitRoutingInfo(mtstate, estate, proute, dispatch, @@ -529,7 +530,7 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate, * partition-key becomes a DELETE+INSERT operation, so this check is still * required when the operation is CMD_UPDATE. */ - CheckValidResultRel(leaf_part_rri, CMD_INSERT); + CheckValidResultRel(leaf_part_rri, rootResultRelInfo, CMD_INSERT); /* * Open partition indices. The user may have asked to check for conflicts diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index 582b0cb017..65bfb05df5 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -396,10 +396,10 @@ ExecSimpleRelationInsert(EState *estate, TupleTableSlot *slot) ResultRelInfo *resultRelInfo = estate->es_result_relation_info; Relation rel = resultRelInfo->ri_RelationDesc; - /* For now we support only tables. */ + /* For now we support only regular tables. */ Assert(rel->rd_rel->relkind == RELKIND_RELATION); - CheckCmdReplicaIdentity(rel, CMD_INSERT); + CheckCmdReplicaIdentity(resultRelInfo, NULL, CMD_INSERT); /* BEFORE ROW INSERT Triggers */ if (resultRelInfo->ri_TrigDesc && @@ -463,7 +463,7 @@ ExecSimpleRelationUpdate(EState *estate, EPQState *epqstate, /* For now we support only tables. */ Assert(rel->rd_rel->relkind == RELKIND_RELATION); - CheckCmdReplicaIdentity(rel, CMD_UPDATE); + CheckCmdReplicaIdentity(resultRelInfo, NULL, CMD_UPDATE); /* BEFORE ROW UPDATE Triggers */ if (resultRelInfo->ri_TrigDesc && @@ -521,7 +521,7 @@ ExecSimpleRelationDelete(EState *estate, EPQState *epqstate, Relation rel = resultRelInfo->ri_RelationDesc; ItemPointer tid = &searchslot->tts_tid; - CheckCmdReplicaIdentity(rel, CMD_DELETE); + CheckCmdReplicaIdentity(resultRelInfo, NULL, CMD_DELETE); /* BEFORE ROW DELETE Triggers */ if (resultRelInfo->ri_TrigDesc && @@ -544,12 +544,17 @@ ExecSimpleRelationDelete(EState *estate, EPQState *epqstate, } /* - * Check if command can be executed with current replica identity. + * Check if command can be executed on 'target_rel' with its (or the + * ancestor's) current replica identity. */ void -CheckCmdReplicaIdentity(Relation rel, CmdType cmd) +CheckCmdReplicaIdentity(ResultRelInfo *target_rel, + ResultRelInfo *root_target_rel, + CmdType cmd) { PublicationActions *pubactions; + Relation rel = target_rel->ri_RelationDesc; + Relation rootrel = root_target_rel ? root_target_rel->ri_RelationDesc : NULL; /* We only need to do checks for UPDATE and DELETE. */ if (cmd != CMD_UPDATE && cmd != CMD_DELETE) @@ -563,9 +568,18 @@ CheckCmdReplicaIdentity(Relation rel, CmdType cmd) /* * This is either UPDATE OR DELETE and there is no replica identity. * - * Check if the table publishes UPDATES or DELETES. + * Check if the table or its root ancestor publishes UPDATES or DELETES. */ pubactions = GetRelationPublicationActions(rel); + if (rootrel) + { + PublicationActions *root_pubactions; + + root_pubactions = GetRelationPublicationActions(rootrel); + pubactions->pubupdate |= root_pubactions->pubupdate; + pubactions->pubdelete |= root_pubactions->pubdelete; + } + if (cmd == CMD_UPDATE && pubactions->pubupdate) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), @@ -591,17 +605,10 @@ CheckSubscriptionRelkind(char relkind, const char *nspname, const char *relname) { /* - * We currently only support writing to regular tables. However, give a - * more specific error for partitioned and foreign tables. + * We currently only support writing to regular and partitioned tables. + * However, give a more specific error for foreign tables. */ - if (relkind == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - (errcode(ERRCODE_WRONG_OBJECT_TYPE), - errmsg("cannot use relation \"%s.%s\" as logical replication target", - nspname, relname), - errdetail("\"%s.%s\" is a partitioned table.", - nspname, relname))); - else if (relkind == RELKIND_FOREIGN_TABLE) + if (relkind == RELKIND_FOREIGN_TABLE) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("cannot use relation \"%s.%s\" as logical replication target", @@ -609,7 +616,11 @@ CheckSubscriptionRelkind(char relkind, const char *nspname, errdetail("\"%s.%s\" is a foreign table.", nspname, relname))); - if (relkind != RELKIND_RELATION) + /* + * There are some unsupported cases with partitioned tables, but we leave + * it for the caller to report them. + */ + if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("cannot use relation \"%s.%s\" as logical replication target", diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index 59d1a31c97..63e108bb56 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -2268,6 +2268,7 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) int nplans = list_length(node->plans); ResultRelInfo *saved_resultRelInfo; ResultRelInfo *resultRelInfo; + ResultRelInfo *rootResultRelInfo = NULL; Plan *subplan; ListCell *l; int i; @@ -2295,8 +2296,11 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) /* If modifying a partitioned table, initialize the root table info */ if (node->rootResultRelIndex >= 0) + { mtstate->rootResultRelInfo = estate->es_root_result_relations + node->rootResultRelIndex; + rootResultRelInfo = mtstate->rootResultRelInfo; + } mtstate->mt_arowmarks = (List **) palloc0(sizeof(List *) * nplans); mtstate->mt_nplans = nplans; @@ -2330,7 +2334,7 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) /* * Verify result relation is a valid target for the current operation */ - CheckValidResultRel(resultRelInfo, operation); + CheckValidResultRel(resultRelInfo, rootResultRelInfo, operation); /* * If there are indices on the result relation, open them and save diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index f8183cd488..98825f01e9 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -761,6 +761,7 @@ copy_table(Relation rel) /* Map the publisher relation to local one. */ relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock); Assert(rel == relmapentry->localrel); + Assert(relmapentry->localrel->rd_rel->relkind == RELKIND_RELATION); /* Start copy on the publisher. */ initStringInfo(&cmd); diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 752508213a..059d2c9194 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -50,7 +50,12 @@ static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); -/* Entry in the map used to remember which relation schemas we sent. */ +/* + * Entry in the map used to remember which relation schemas we sent. + * + * For partitions, 'pubactions' considers not only the table's own + * publications, but also those of all of its ancestors. + */ typedef struct RelationSyncEntry { Oid relid; /* relation oid */ @@ -63,7 +68,7 @@ typedef struct RelationSyncEntry static HTAB *RelationSyncCache = NULL; static void init_rel_sync_cache(MemoryContext decoding_context); -static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data, Oid relid); +static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data, Relation rel); static void rel_sync_cache_relation_cb(Datum arg, Oid relid); static void rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue); @@ -311,7 +316,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (!is_publishable_relation(relation)) return; - relentry = get_rel_sync_entry(data, RelationGetRelid(relation)); + relentry = get_rel_sync_entry(data, relation); /* First check the table filter */ switch (change->action) @@ -401,7 +406,7 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (!is_publishable_relation(relation)) continue; - relentry = get_rel_sync_entry(data, relid); + relentry = get_rel_sync_entry(data, relation); if (!relentry->pubactions.pubtruncate) continue; @@ -526,8 +531,9 @@ init_rel_sync_cache(MemoryContext cachectx) * Find or create entry in the relation schema cache. */ static RelationSyncEntry * -get_rel_sync_entry(PGOutputData *data, Oid relid) +get_rel_sync_entry(PGOutputData *data, Relation rel) { + Oid relid = RelationGetRelid(rel); RelationSyncEntry *entry; bool found; MemoryContext oldctx; @@ -546,7 +552,9 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) if (!found || !entry->replicate_valid) { List *pubids = GetRelationPublications(relid); - ListCell *lc; + ListCell *lc, + *lc1; + List *ancestor_pubids = NIL; /* Reload publications if needed before use. */ if (!publications_valid) @@ -568,6 +576,11 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->pubactions.pubinsert = entry->pubactions.pubupdate = entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false; + /* For partitions, also consider publications of ancestors. */ + if (rel->rd_rel->relispartition) + ancestor_pubids = + GetRelationAncestorPublications(RelationGetRelid(rel)); + foreach(lc, data->publications) { Publication *pub = lfirst(lc); @@ -580,12 +593,28 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate; } + if (entry->pubactions.pubinsert && entry->pubactions.pubupdate && + entry->pubactions.pubdelete && entry->pubactions.pubtruncate) + break; + + foreach(lc1, ancestor_pubids) + { + if (lfirst_oid(lc1) == pub->oid) + { + entry->pubactions.pubinsert |= pub->pubactions.pubinsert; + entry->pubactions.pubupdate |= pub->pubactions.pubupdate; + entry->pubactions.pubdelete |= pub->pubactions.pubdelete; + entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate; + } + } + if (entry->pubactions.pubinsert && entry->pubactions.pubupdate && entry->pubactions.pubdelete && entry->pubactions.pubtruncate) break; } list_free(pubids); + list_free(ancestor_pubids); entry->replicate_valid = true; } diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 799b6988b7..dc33c20048 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -3969,8 +3969,12 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) { TableInfo *tbinfo = &tblinfo[i]; - /* Only plain tables can be aded to publications. */ - if (tbinfo->relkind != RELKIND_RELATION) + /* + * Only regular and partitioned tables can be added to + * publications. + */ + if (tbinfo->relkind != RELKIND_RELATION && + tbinfo->relkind != RELKIND_PARTITIONED_TABLE) continue; /* diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 6cdc2b1197..3cfb31c2e6 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -80,6 +80,7 @@ typedef struct Publication extern Publication *GetPublication(Oid pubid); extern Publication *GetPublicationByName(const char *pubname, bool missing_ok); extern List *GetRelationPublications(Oid relid); +extern List *GetRelationAncestorPublications(Oid relid); extern List *GetPublicationRelations(Oid pubid); extern List *GetAllTablesPublications(void); extern List *GetAllTablesPublicationRelations(void); diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index 6ef3e1fe06..5b97bb5d57 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -179,7 +179,9 @@ extern void ExecutorEnd(QueryDesc *queryDesc); extern void standard_ExecutorEnd(QueryDesc *queryDesc); extern void ExecutorRewind(QueryDesc *queryDesc); extern bool ExecCheckRTPerms(List *rangeTable, bool ereport_on_violation); -extern void CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation); +extern void CheckValidResultRel(ResultRelInfo *resultRelInfo, + ResultRelInfo *rootResultRelInfo, + CmdType operation); extern void InitResultRelInfo(ResultRelInfo *resultRelInfo, Relation resultRelationDesc, Index resultRelationIndex, @@ -592,7 +594,9 @@ extern void ExecSimpleRelationUpdate(EState *estate, EPQState *epqstate, TupleTableSlot *searchslot, TupleTableSlot *slot); extern void ExecSimpleRelationDelete(EState *estate, EPQState *epqstate, TupleTableSlot *searchslot); -extern void CheckCmdReplicaIdentity(Relation rel, CmdType cmd); +extern void CheckCmdReplicaIdentity(ResultRelInfo *target_rel, + ResultRelInfo *root_target_rel, + CmdType cmd); extern void CheckSubscriptionRelkind(char relkind, const char *nspname, const char *relname); diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out index feb51e4add..e3fabe70f9 100644 --- a/src/test/regress/expected/publication.out +++ b/src/test/regress/expected/publication.out @@ -116,6 +116,22 @@ Tables: DROP TABLE testpub_tbl3, testpub_tbl3a; DROP PUBLICATION testpub3, testpub4; +-- Tests for partitioned tables +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_forparted; +RESET client_min_messages; +-- should add only the parent to publication, not the partition +CREATE TABLE testpub_parted1 PARTITION OF testpub_parted FOR VALUES IN (1); +ALTER PUBLICATION testpub_forparted ADD TABLE testpub_parted; +\dRp+ testpub_forparted + Publication testpub_forparted + Owner | All tables | Inserts | Updates | Deletes | Truncates +--------------------------+------------+---------+---------+---------+----------- + regress_publication_user | f | t | t | t | t +Tables: + "public.testpub_parted" + +DROP PUBLICATION testpub_forparted; -- fail - view CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_view; ERROR: "testpub_view" is not a table @@ -142,11 +158,6 @@ Tables: ALTER PUBLICATION testpub_default ADD TABLE testpub_view; ERROR: "testpub_view" is not a table DETAIL: Only tables can be added to publications. --- fail - partitioned table -ALTER PUBLICATION testpub_fortbl ADD TABLE testpub_parted; -ERROR: "testpub_parted" is a partitioned table -DETAIL: Adding partitioned tables to publications is not supported. -HINT: You can add the table partitions individually. ALTER PUBLICATION testpub_default ADD TABLE testpub_tbl1; ALTER PUBLICATION testpub_default SET TABLE testpub_tbl1; ALTER PUBLICATION testpub_default ADD TABLE pub_test.testpub_nopk; diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql index 5773a755cf..b79a3f8f8f 100644 --- a/src/test/regress/sql/publication.sql +++ b/src/test/regress/sql/publication.sql @@ -69,6 +69,16 @@ RESET client_min_messages; DROP TABLE testpub_tbl3, testpub_tbl3a; DROP PUBLICATION testpub3, testpub4; +-- Tests for partitioned tables +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_forparted; +RESET client_min_messages; +-- should add only the parent to publication, not the partition +CREATE TABLE testpub_parted1 PARTITION OF testpub_parted FOR VALUES IN (1); +ALTER PUBLICATION testpub_forparted ADD TABLE testpub_parted; +\dRp+ testpub_forparted +DROP PUBLICATION testpub_forparted; + -- fail - view CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_view; SET client_min_messages = 'ERROR'; @@ -83,8 +93,6 @@ CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_tbl1; -- fail - view ALTER PUBLICATION testpub_default ADD TABLE testpub_view; --- fail - partitioned table -ALTER PUBLICATION testpub_fortbl ADD TABLE testpub_parted; ALTER PUBLICATION testpub_default ADD TABLE testpub_tbl1; ALTER PUBLICATION testpub_default SET TABLE testpub_tbl1; diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl new file mode 100644 index 0000000000..eb0f1cd6a8 --- /dev/null +++ b/src/test/subscription/t/013_partition.pl @@ -0,0 +1,161 @@ +# Test PARTITION +use strict; +use warnings; +use PostgresNode; +use TestLib; +use Test::More tests => 10; + +# setup + +my $node_publisher = get_new_node('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +my $node_subscriber1 = get_new_node('subscriber1'); +$node_subscriber1->init(allows_streaming => 'logical'); +$node_subscriber1->start; + +my $node_subscriber2 = get_new_node('subscriber2'); +$node_subscriber2->init(allows_streaming => 'logical'); +$node_subscriber2->start; + +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; + +$node_publisher->safe_psql('postgres', + "CREATE TABLE tab1 (a int PRIMARY KEY, b text) PARTITION BY LIST (a)"); + +$node_subscriber1->safe_psql('postgres', + "CREATE TABLE tab1 (a int PRIMARY KEY, b text, c text) PARTITION BY LIST (a)"); + +$node_publisher->safe_psql('postgres', + "CREATE TABLE tab1_1 (b text, a int NOT NULL)"); +$node_publisher->safe_psql('postgres', + "ALTER TABLE tab1 ATTACH PARTITION tab1_1 FOR VALUES IN (1, 2, 3)"); + +$node_subscriber1->safe_psql('postgres', + "CREATE TABLE tab1_1 (b text, c text DEFAULT 'sub1_tab1', a int NOT NULL)"); +$node_subscriber1->safe_psql('postgres', + "ALTER TABLE tab1 ATTACH PARTITION tab1_1 FOR VALUES IN (1, 2, 3, 4)"); + +$node_publisher->safe_psql('postgres', + "CREATE TABLE tab1_2 PARTITION OF tab1 FOR VALUES IN (5, 6)"); + +$node_subscriber1->safe_psql('postgres', + "CREATE TABLE tab1_2 PARTITION OF tab1 (c DEFAULT 'sub1_tab1') FOR VALUES IN (5, 6)"); + +$node_subscriber2->safe_psql('postgres', + "CREATE TABLE tab1_2 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1_2', b text)"); + +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION pub1 FOR TABLE tab1, tab1_1"); +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION pub2 FOR TABLE tab1_2"); + +$node_subscriber1->safe_psql('postgres', + "CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"); + +$node_subscriber2->safe_psql('postgres', + "CREATE SUBSCRIPTION sub2 CONNECTION '$publisher_connstr' PUBLICATION pub2"); + +# Wait for initial sync of all subscriptions +my $synced_query = + "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');"; +$node_subscriber1->poll_query_until('postgres', $synced_query) + or die "Timed out while waiting for subscriber to synchronize data"; +$node_subscriber2->poll_query_until('postgres', $synced_query) + or die "Timed out while waiting for subscriber to synchronize data"; + +# insert data (some into the root parent and some directly into partitions) + +$node_publisher->safe_psql('postgres', + "INSERT INTO tab1 VALUES (1)"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab1_1 (a) VALUES (3)"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab1_2 VALUES (5)"); + +$node_publisher->wait_for_catchup('sub1'); +$node_publisher->wait_for_catchup('sub2'); + +my $result = $node_subscriber1->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1 GROUP BY 1"); +is($result, qq(sub1_tab1|3|1|5), 'insert into tab1_1, tab1_2 replicated'); + +$result = $node_subscriber2->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1_2 GROUP BY 1"); +is($result, qq(sub2_tab1_2|1|5|5), 'inserts into tab1_2 replicated'); + +# update a row (no partition change) + +$node_publisher->safe_psql('postgres', + "UPDATE tab1 SET a = 2 WHERE a = 1"); + +$node_publisher->wait_for_catchup('sub1'); + +$result = $node_subscriber1->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1 GROUP BY 1"); +is($result, qq(sub1_tab1|3|2|5), 'update of tab1_1 replicated'); + +# update a row (partition changes) + +$node_publisher->safe_psql('postgres', + "UPDATE tab1 SET a = 6 WHERE a = 2"); + +$node_publisher->wait_for_catchup('sub1'); +$node_publisher->wait_for_catchup('sub2'); + +$result = $node_subscriber1->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1 GROUP BY 1"); +is($result, qq(sub1_tab1|3|3|6), 'delete from tab1_1 replicated'); + +$result = $node_subscriber2->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1_2 GROUP BY 1"); +is($result, qq(sub2_tab1_2|2|5|6), 'insert into tab1_2 replicated'); + +# delete rows (some from the root parent, some directly from the partition) + +$node_publisher->safe_psql('postgres', + "DELETE FROM tab1 WHERE a IN (3, 5)"); +$node_publisher->safe_psql('postgres', + "DELETE FROM tab1_2"); + +$node_publisher->wait_for_catchup('sub1'); +$node_publisher->wait_for_catchup('sub2'); + +$result = $node_subscriber1->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1"); +is($result, qq(0||), 'delete from tab1_1, tab_2 replicated'); + +$result = $node_subscriber2->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1_2"); +is($result, qq(0||), 'delete from tab1_2 replicated'); + +# truncate (root parent and partition directly) + +$node_subscriber1->safe_psql('postgres', + "INSERT INTO tab1 VALUES (1), (2), (5)"); +$node_subscriber2->safe_psql('postgres', + "INSERT INTO tab1_2 VALUES (5)"); + +$node_publisher->safe_psql('postgres', + "TRUNCATE tab1_2"); + +$node_publisher->wait_for_catchup('sub1'); +$node_publisher->wait_for_catchup('sub2'); + +$result = $node_subscriber1->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1"); +is($result, qq(2|1|2), 'truncate of tab_2 replicated'); + +$result = $node_subscriber2->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1_2"); +is($result, qq(0||), 'truncate of tab1_2 replicated'); + +$node_publisher->safe_psql('postgres', + "TRUNCATE tab1"); + +$node_publisher->wait_for_catchup('sub1'); + +$result = $node_subscriber1->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1"); +is($result, qq(0||), 'truncate of tab1_1 replicated'); -- 2.16.5
v8-0002-Add-publish_using_root_schema-parameter-for-publi.patch
(text/plain, 26.7 KB)
From ccc1d855b4fd62b80fdae745c2dee541608cf740 Mon Sep 17 00:00:00 2001 From: amit <[email protected]> Date: Fri, 29 Nov 2019 17:40:11 +0900 Subject: [PATCH v8 2/4] Add publish_using_root_schema parameter for publications It dictates whether to publish (leaf) partition changes using the the schema of root parent table. --- doc/src/sgml/ref/create_publication.sgml | 15 +++++ src/backend/catalog/pg_publication.c | 1 + src/backend/commands/publicationcmds.c | 94 ++++++++++++++++----------- src/bin/pg_dump/pg_dump.c | 22 ++++++- src/bin/pg_dump/pg_dump.h | 1 + src/bin/psql/describe.c | 17 ++++- src/include/catalog/pg_publication.h | 3 + src/test/regress/expected/publication.out | 103 +++++++++++++++++------------- src/test/regress/sql/publication.sql | 3 + 9 files changed, 171 insertions(+), 88 deletions(-) diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml index 848779a00f..a8cf2c4629 100644 --- a/doc/src/sgml/ref/create_publication.sgml +++ b/doc/src/sgml/ref/create_publication.sgml @@ -124,6 +124,21 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable> </para> </listitem> </varlistentry> + + <varlistentry> + <term><literal>publish_using_root_schema</literal> (<type>boolean</type>)</term> + <listitem> + <para> + This parameter determines whether DML operations on a partitioned + table contained in the publication will be published using its own + schema rather than of the individual partitions which are actually + changed; the latter is the default. Setting it to + <literal>true</literal> allows the changes to be replicated into a + non-partitioned table or a partitioned table consisting of a + a different set of partitions. + </para> + </listitem> + </varlistentry> </variablelist> </para> diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index fb369dbe17..6d2911d18f 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -403,6 +403,7 @@ GetPublication(Oid pubid) pub->pubactions.pubupdate = pubform->pubupdate; pub->pubactions.pubdelete = pubform->pubdelete; pub->pubactions.pubtruncate = pubform->pubtruncate; + pub->publish_using_root_schema = pubform->pubasroot; ReleaseSysCache(tup); diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 8f38c63ad2..e48815534c 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -55,20 +55,23 @@ static void PublicationDropTables(Oid pubid, List *rels, bool missing_ok); static void parse_publication_options(List *options, bool *publish_given, - bool *publish_insert, - bool *publish_update, - bool *publish_delete, - bool *publish_truncate) + PublicationActions *pubactions, + bool *publish_using_root_schema_given, + bool *publish_using_root_schema) { ListCell *lc; + *publish_using_root_schema_given = false; *publish_given = false; /* Defaults are true */ - *publish_insert = true; - *publish_update = true; - *publish_delete = true; - *publish_truncate = true; + pubactions->pubinsert = true; + pubactions->pubupdate = true; + pubactions->pubdelete = true; + pubactions->pubtruncate = true; + + /* Relation changes published as of itself by default. */ + *publish_using_root_schema = false; /* Parse options */ foreach(lc, options) @@ -90,10 +93,10 @@ parse_publication_options(List *options, * If publish option was given only the explicitly listed actions * should be published. */ - *publish_insert = false; - *publish_update = false; - *publish_delete = false; - *publish_truncate = false; + pubactions->pubinsert = false; + pubactions->pubupdate = false; + pubactions->pubdelete = false; + pubactions->pubtruncate = false; *publish_given = true; publish = defGetString(defel); @@ -109,19 +112,28 @@ parse_publication_options(List *options, char *publish_opt = (char *) lfirst(lc); if (strcmp(publish_opt, "insert") == 0) - *publish_insert = true; + pubactions->pubinsert = true; else if (strcmp(publish_opt, "update") == 0) - *publish_update = true; + pubactions->pubupdate = true; else if (strcmp(publish_opt, "delete") == 0) - *publish_delete = true; + pubactions->pubdelete = true; else if (strcmp(publish_opt, "truncate") == 0) - *publish_truncate = true; + pubactions->pubtruncate = true; else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("unrecognized \"publish\" value: \"%s\"", publish_opt))); } } + else if (strcmp(defel->defname, "publish_using_root_schema") == 0) + { + if (*publish_using_root_schema_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + *publish_using_root_schema_given = true; + *publish_using_root_schema = defGetBoolean(defel); + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -142,10 +154,9 @@ CreatePublication(CreatePublicationStmt *stmt) Datum values[Natts_pg_publication]; HeapTuple tup; bool publish_given; - bool publish_insert; - bool publish_update; - bool publish_delete; - bool publish_truncate; + PublicationActions pubactions; + bool publish_using_root_schema_given; + bool publish_using_root_schema; AclResult aclresult; /* must have CREATE privilege on database */ @@ -182,9 +193,9 @@ CreatePublication(CreatePublicationStmt *stmt) values[Anum_pg_publication_pubowner - 1] = ObjectIdGetDatum(GetUserId()); parse_publication_options(stmt->options, - &publish_given, &publish_insert, - &publish_update, &publish_delete, - &publish_truncate); + &publish_given, &pubactions, + &publish_using_root_schema_given, + &publish_using_root_schema); puboid = GetNewOidWithIndex(rel, PublicationObjectIndexId, Anum_pg_publication_oid); @@ -192,13 +203,15 @@ CreatePublication(CreatePublicationStmt *stmt) values[Anum_pg_publication_puballtables - 1] = BoolGetDatum(stmt->for_all_tables); values[Anum_pg_publication_pubinsert - 1] = - BoolGetDatum(publish_insert); + BoolGetDatum(pubactions.pubinsert); values[Anum_pg_publication_pubupdate - 1] = - BoolGetDatum(publish_update); + BoolGetDatum(pubactions.pubupdate); values[Anum_pg_publication_pubdelete - 1] = - BoolGetDatum(publish_delete); + BoolGetDatum(pubactions.pubdelete); values[Anum_pg_publication_pubtruncate - 1] = - BoolGetDatum(publish_truncate); + BoolGetDatum(pubactions.pubtruncate); + values[Anum_pg_publication_pubasroot - 1] = + BoolGetDatum(publish_using_root_schema); tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); @@ -250,17 +263,16 @@ AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel, bool replaces[Natts_pg_publication]; Datum values[Natts_pg_publication]; bool publish_given; - bool publish_insert; - bool publish_update; - bool publish_delete; - bool publish_truncate; + PublicationActions pubactions; + bool publish_using_root_schema_given; + bool publish_using_root_schema; ObjectAddress obj; Form_pg_publication pubform; parse_publication_options(stmt->options, - &publish_given, &publish_insert, - &publish_update, &publish_delete, - &publish_truncate); + &publish_given, &pubactions, + &publish_using_root_schema_given, + &publish_using_root_schema); /* Everything ok, form a new tuple. */ memset(values, 0, sizeof(values)); @@ -269,19 +281,25 @@ AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel, if (publish_given) { - values[Anum_pg_publication_pubinsert - 1] = BoolGetDatum(publish_insert); + values[Anum_pg_publication_pubinsert - 1] = BoolGetDatum(pubactions.pubinsert); replaces[Anum_pg_publication_pubinsert - 1] = true; - values[Anum_pg_publication_pubupdate - 1] = BoolGetDatum(publish_update); + values[Anum_pg_publication_pubupdate - 1] = BoolGetDatum(pubactions.pubupdate); replaces[Anum_pg_publication_pubupdate - 1] = true; - values[Anum_pg_publication_pubdelete - 1] = BoolGetDatum(publish_delete); + values[Anum_pg_publication_pubdelete - 1] = BoolGetDatum(pubactions.pubdelete); replaces[Anum_pg_publication_pubdelete - 1] = true; - values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(publish_truncate); + values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate); replaces[Anum_pg_publication_pubtruncate - 1] = true; } + if (publish_using_root_schema_given) + { + values[Anum_pg_publication_pubasroot - 1] = BoolGetDatum(publish_using_root_schema); + replaces[Anum_pg_publication_pubasroot - 1] = true; + } + tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls, replaces); diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index dc33c20048..bdbd1f823b 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -3780,6 +3780,7 @@ getPublications(Archive *fout) int i_pubupdate; int i_pubdelete; int i_pubtruncate; + int i_pubasroot; int i, ntups; @@ -3791,11 +3792,18 @@ getPublications(Archive *fout) resetPQExpBuffer(query); /* Get the publications. */ - if (fout->remoteVersion >= 110000) + if (fout->remoteVersion >= 130000) + appendPQExpBuffer(query, + "SELECT p.tableoid, p.oid, p.pubname, " + "(%s p.pubowner) AS rolname, " + "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubasroot " + "FROM pg_publication p", + username_subquery); + else if (fout->remoteVersion >= 110000) appendPQExpBuffer(query, "SELECT p.tableoid, p.oid, p.pubname, " "(%s p.pubowner) AS rolname, " - "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate " + "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false as pubasroot " "FROM pg_publication p", username_subquery); else @@ -3819,6 +3827,7 @@ getPublications(Archive *fout) i_pubupdate = PQfnumber(res, "pubupdate"); i_pubdelete = PQfnumber(res, "pubdelete"); i_pubtruncate = PQfnumber(res, "pubtruncate"); + i_pubasroot = PQfnumber(res, "pubasroot"); pubinfo = pg_malloc(ntups * sizeof(PublicationInfo)); @@ -3841,6 +3850,8 @@ getPublications(Archive *fout) (strcmp(PQgetvalue(res, i, i_pubdelete), "t") == 0); pubinfo[i].pubtruncate = (strcmp(PQgetvalue(res, i, i_pubtruncate), "t") == 0); + pubinfo[i].pubasroot = + (strcmp(PQgetvalue(res, i, i_pubasroot), "t") == 0); if (strlen(pubinfo[i].rolname) == 0) pg_log_warning("owner of publication \"%s\" appears to be invalid", @@ -3917,7 +3928,12 @@ dumpPublication(Archive *fout, PublicationInfo *pubinfo) first = false; } - appendPQExpBufferStr(query, "');\n"); + appendPQExpBufferStr(query, "'"); + + if (pubinfo->pubasroot) + appendPQExpBufferStr(query, ", publish_using_root_schema = true"); + + appendPQExpBufferStr(query, ");\n"); ArchiveEntry(fout, pubinfo->dobj.catId, pubinfo->dobj.dumpId, ARCHIVE_OPTS(.tag = pubinfo->dobj.name, diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index 21004e5078..90e47dd1f3 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -600,6 +600,7 @@ typedef struct _PublicationInfo bool pubupdate; bool pubdelete; bool pubtruncate; + bool pubasroot; } PublicationInfo; /* diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index f3c7eb96fa..3f6ce713af 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -5706,7 +5706,7 @@ listPublications(const char *pattern) PQExpBufferData buf; PGresult *res; printQueryOpt myopt = pset.popt; - static const bool translate_columns[] = {false, false, false, false, false, false, false}; + static const bool translate_columns[] = {false, false, false, false, false, false, false, false}; if (pset.sversion < 100000) { @@ -5737,6 +5737,10 @@ listPublications(const char *pattern) appendPQExpBuffer(&buf, ",\n pubtruncate AS \"%s\"", gettext_noop("Truncates")); + if (pset.sversion >= 130000) + appendPQExpBuffer(&buf, + ",\n pubasroot AS \"%s\"", + gettext_noop("Publishes Using Root Schema")); appendPQExpBufferStr(&buf, "\nFROM pg_catalog.pg_publication\n"); @@ -5778,6 +5782,7 @@ describePublications(const char *pattern) int i; PGresult *res; bool has_pubtruncate; + bool has_pubasroot; if (pset.sversion < 100000) { @@ -5790,6 +5795,7 @@ describePublications(const char *pattern) } has_pubtruncate = (pset.sversion >= 110000); + has_pubasroot = (pset.sversion >= 130000); initPQExpBuffer(&buf); @@ -5800,6 +5806,9 @@ describePublications(const char *pattern) if (has_pubtruncate) appendPQExpBufferStr(&buf, ", pubtruncate"); + if (has_pubasroot) + appendPQExpBufferStr(&buf, + ", pubasroot"); appendPQExpBufferStr(&buf, "\nFROM pg_catalog.pg_publication\n"); @@ -5849,6 +5858,8 @@ describePublications(const char *pattern) if (has_pubtruncate) ncols++; + if (has_pubasroot) + ncols++; initPQExpBuffer(&title); printfPQExpBuffer(&title, _("Publication %s"), pubname); @@ -5861,6 +5872,8 @@ describePublications(const char *pattern) printTableAddHeader(&cont, gettext_noop("Deletes"), true, align); if (has_pubtruncate) printTableAddHeader(&cont, gettext_noop("Truncates"), true, align); + if (has_pubasroot) + printTableAddHeader(&cont, gettext_noop("Publishes Using Root Schema"), true, align); printTableAddCell(&cont, PQgetvalue(res, i, 2), false, false); printTableAddCell(&cont, PQgetvalue(res, i, 3), false, false); @@ -5869,6 +5882,8 @@ describePublications(const char *pattern) printTableAddCell(&cont, PQgetvalue(res, i, 6), false, false); if (has_pubtruncate) printTableAddCell(&cont, PQgetvalue(res, i, 7), false, false); + if (has_pubasroot) + printTableAddCell(&cont, PQgetvalue(res, i, 8), false, false); if (!puballtables) { diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 3cfb31c2e6..9d13e5c735 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -52,6 +52,8 @@ CATALOG(pg_publication,6104,PublicationRelationId) /* true if truncates are published */ bool pubtruncate; + /* true if partition changes are published using root schema */ + bool pubasroot; } FormData_pg_publication; /* ---------------- @@ -74,6 +76,7 @@ typedef struct Publication Oid oid; char *name; bool alltables; + bool publish_using_root_schema; PublicationActions pubactions; } Publication; diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out index e3fabe70f9..da22ca3c6a 100644 --- a/src/test/regress/expected/publication.out +++ b/src/test/regress/expected/publication.out @@ -25,21 +25,23 @@ CREATE PUBLICATION testpub_xxx WITH (foo); ERROR: unrecognized publication parameter: "foo" CREATE PUBLICATION testpub_xxx WITH (publish = 'cluster, vacuum'); ERROR: unrecognized "publish" value: "cluster" +CREATE PUBLICATION testpub_xxx WITH (publish_using_root_schema = 'true', publish_using_root_schema = '0'); +ERROR: conflicting or redundant options \dRp - List of publications - Name | Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------+--------------------------+------------+---------+---------+---------+----------- - testpib_ins_trunct | regress_publication_user | f | t | f | f | f - testpub_default | regress_publication_user | f | f | t | f | f + List of publications + Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------+--------------------------+------------+---------+---------+---------+-----------+----------------------------- + testpib_ins_trunct | regress_publication_user | f | t | f | f | f | f + testpub_default | regress_publication_user | f | f | t | f | f | f (2 rows) ALTER PUBLICATION testpub_default SET (publish = 'insert, update, delete'); \dRp - List of publications - Name | Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------+--------------------------+------------+---------+---------+---------+----------- - testpib_ins_trunct | regress_publication_user | f | t | f | f | f - testpub_default | regress_publication_user | f | t | t | t | f + List of publications + Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------+--------------------------+------------+---------+---------+---------+-----------+----------------------------- + testpib_ins_trunct | regress_publication_user | f | t | f | f | f | f + testpub_default | regress_publication_user | f | t | t | t | f | f (2 rows) --- adding tables @@ -83,10 +85,10 @@ Publications: "testpub_foralltables" \dRp+ testpub_foralltables - Publication testpub_foralltables - Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------+------------+---------+---------+---------+----------- - regress_publication_user | t | t | t | f | f + Publication testpub_foralltables + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | t | t | t | f | f | f (1 row) DROP TABLE testpub_tbl2; @@ -98,19 +100,19 @@ CREATE PUBLICATION testpub3 FOR TABLE testpub_tbl3; CREATE PUBLICATION testpub4 FOR TABLE ONLY testpub_tbl3; RESET client_min_messages; \dRp+ testpub3 - Publication testpub3 - Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------+------------+---------+---------+---------+----------- - regress_publication_user | f | t | t | t | t + Publication testpub3 + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | f | t | t | t | t | f Tables: "public.testpub_tbl3" "public.testpub_tbl3a" \dRp+ testpub4 - Publication testpub4 - Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------+------------+---------+---------+---------+----------- - regress_publication_user | f | t | t | t | t + Publication testpub4 + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | f | t | t | t | t | f Tables: "public.testpub_tbl3" @@ -124,10 +126,19 @@ RESET client_min_messages; CREATE TABLE testpub_parted1 PARTITION OF testpub_parted FOR VALUES IN (1); ALTER PUBLICATION testpub_forparted ADD TABLE testpub_parted; \dRp+ testpub_forparted - Publication testpub_forparted - Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------+------------+---------+---------+---------+----------- - regress_publication_user | f | t | t | t | t + Publication testpub_forparted + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | f | t | t | t | t | f +Tables: + "public.testpub_parted" + +ALTER PUBLICATION testpub_forparted SET (publish_using_root_schema = true); +\dRp+ testpub_forparted + Publication testpub_forparted + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | f | t | t | t | t | t Tables: "public.testpub_parted" @@ -146,10 +157,10 @@ ERROR: relation "testpub_tbl1" is already member of publication "testpub_fortbl CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_tbl1; ERROR: publication "testpub_fortbl" already exists \dRp+ testpub_fortbl - Publication testpub_fortbl - Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------+------------+---------+---------+---------+----------- - regress_publication_user | f | t | t | t | t + Publication testpub_fortbl + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | f | t | t | t | t | f Tables: "pub_test.testpub_nopk" "public.testpub_tbl1" @@ -187,10 +198,10 @@ Publications: "testpub_fortbl" \dRp+ testpub_default - Publication testpub_default - Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------+------------+---------+---------+---------+----------- - regress_publication_user | f | t | t | t | f + Publication testpub_default + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | f | t | t | t | f | f Tables: "pub_test.testpub_nopk" "public.testpub_tbl1" @@ -234,10 +245,10 @@ DROP TABLE testpub_parted; DROP VIEW testpub_view; DROP TABLE testpub_tbl1; \dRp+ testpub_default - Publication testpub_default - Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------+------------+---------+---------+---------+----------- - regress_publication_user | f | t | t | t | f + Publication testpub_default + Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +--------------------------+------------+---------+---------+---------+-----------+----------------------------- + regress_publication_user | f | t | t | t | f | f (1 row) -- fail - must be owner of publication @@ -247,20 +258,20 @@ ERROR: must be owner of publication testpub_default RESET ROLE; ALTER PUBLICATION testpub_default RENAME TO testpub_foo; \dRp testpub_foo - List of publications - Name | Owner | All tables | Inserts | Updates | Deletes | Truncates --------------+--------------------------+------------+---------+---------+---------+----------- - testpub_foo | regress_publication_user | f | t | t | t | f + List of publications + Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +-------------+--------------------------+------------+---------+---------+---------+-----------+----------------------------- + testpub_foo | regress_publication_user | f | t | t | t | f | f (1 row) -- rename back to keep the rest simple ALTER PUBLICATION testpub_foo RENAME TO testpub_default; ALTER PUBLICATION testpub_default OWNER TO regress_publication_user2; \dRp testpub_default - List of publications - Name | Owner | All tables | Inserts | Updates | Deletes | Truncates ------------------+---------------------------+------------+---------+---------+---------+----------- - testpub_default | regress_publication_user2 | f | t | t | t | f + List of publications + Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Publishes Using Root Schema +-----------------+---------------------------+------------+---------+---------+---------+-----------+----------------------------- + testpub_default | regress_publication_user2 | f | t | t | t | f | f (1 row) DROP PUBLICATION testpub_default; diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql index b79a3f8f8f..7ddca1b974 100644 --- a/src/test/regress/sql/publication.sql +++ b/src/test/regress/sql/publication.sql @@ -23,6 +23,7 @@ ALTER PUBLICATION testpub_default SET (publish = update); -- error cases CREATE PUBLICATION testpub_xxx WITH (foo); CREATE PUBLICATION testpub_xxx WITH (publish = 'cluster, vacuum'); +CREATE PUBLICATION testpub_xxx WITH (publish_using_root_schema = 'true', publish_using_root_schema = '0'); \dRp @@ -77,6 +78,8 @@ RESET client_min_messages; CREATE TABLE testpub_parted1 PARTITION OF testpub_parted FOR VALUES IN (1); ALTER PUBLICATION testpub_forparted ADD TABLE testpub_parted; \dRp+ testpub_forparted +ALTER PUBLICATION testpub_forparted SET (publish_using_root_schema = true); +\dRp+ testpub_forparted DROP PUBLICATION testpub_forparted; -- fail - view -- 2.16.5
v8-0003-Some-refactoring-of-logical-worker.c.patch
(text/plain, 13.7 KB)
From 8126a6bb784506180ba1d9c4985aabe124ffc63e Mon Sep 17 00:00:00 2001 From: amit <[email protected]> Date: Thu, 5 Dec 2019 09:17:06 +0900 Subject: [PATCH v8 3/4] Some refactoring of logical/worker.c --- src/backend/replication/logical/worker.c | 291 ++++++++++++++++++------------- 1 file changed, 170 insertions(+), 121 deletions(-) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 7a5471f95c..34b0ac78cc 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -90,7 +90,8 @@ static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping); typedef struct SlotErrCallbackArg { - LogicalRepRelMapEntry *rel; + LogicalRepRelation *remoterel; + Oid local_reloid; int local_attnum; int remote_attnum; } SlotErrCallbackArg; @@ -268,7 +269,6 @@ static void slot_store_error_callback(void *arg) { SlotErrCallbackArg *errarg = (SlotErrCallbackArg *) arg; - LogicalRepRelMapEntry *rel; char *remotetypname; Oid remotetypoid, localtypoid; @@ -277,19 +277,18 @@ slot_store_error_callback(void *arg) if (errarg->remote_attnum < 0) return; - rel = errarg->rel; - remotetypoid = rel->remoterel.atttyps[errarg->remote_attnum]; + remotetypoid = errarg->remoterel->atttyps[errarg->remote_attnum]; /* Fetch remote type name from the LogicalRepTypMap cache */ remotetypname = logicalrep_typmap_gettypname(remotetypoid); /* Fetch local type OID from the local sys cache */ - localtypoid = get_atttype(rel->localreloid, errarg->local_attnum + 1); + localtypoid = get_atttype(errarg->local_reloid, errarg->local_attnum + 1); errcontext("processing remote data for replication target relation \"%s.%s\" column \"%s\", " "remote type %s, local type %s", - rel->remoterel.nspname, rel->remoterel.relname, - rel->remoterel.attnames[errarg->remote_attnum], + errarg->remoterel->nspname, errarg->remoterel->relname, + errarg->remoterel->attnames[errarg->remote_attnum], remotetypname, format_type_be(localtypoid)); } @@ -311,7 +310,8 @@ slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, ExecClearTuple(slot); /* Push callback + info on the error context stack */ - errarg.rel = rel; + errarg.remoterel = &rel->remoterel; + errarg.local_reloid = rel->localreloid; errarg.local_attnum = -1; errarg.remote_attnum = -1; errcallback.callback = slot_store_error_callback; @@ -375,8 +375,9 @@ slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, */ static void slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot, - LogicalRepRelMapEntry *rel, - char **values, bool *replaces) + char **values, bool *replaces, + AttrMap *attrmap, LogicalRepRelation *remoterel, + Oid local_reloid) { int natts = slot->tts_tupleDescriptor->natts; int i; @@ -396,7 +397,8 @@ slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot, memcpy(slot->tts_isnull, srcslot->tts_isnull, natts * sizeof(bool)); /* For error reporting, push callback + info on the error context stack */ - errarg.rel = rel; + errarg.remoterel = remoterel; + errarg.local_reloid = local_reloid; errarg.local_attnum = -1; errarg.remote_attnum = -1; errcallback.callback = slot_store_error_callback; @@ -405,11 +407,11 @@ slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot, error_context_stack = &errcallback; /* Call the "in" function for each replaced attribute */ - Assert(natts == rel->attrmap->maplen); + Assert(natts == attrmap->maplen); for (i = 0; i < natts; i++) { Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i); - int remoteattnum = rel->attrmap->attnums[i]; + int remoteattnum = attrmap->attnums[i]; if (remoteattnum < 0) continue; @@ -578,6 +580,148 @@ GetRelationIdentityOrPK(Relation rel) return idxoid; } +/* Workhorse for apply_handle_insert() */ +static void +apply_handle_do_insert(ResultRelInfo *relinfo, + EState *estate, TupleTableSlot *localslot) +{ + ExecOpenIndices(relinfo, false); + + /* Do the insert. */ + ExecSimpleRelationInsert(estate, localslot); + + /* Cleanup. */ + ExecCloseIndices(relinfo); +} + +/* Workhorse for apply_handle_update() */ +static void +apply_handle_do_update(ResultRelInfo *relinfo, + EState *estate, TupleTableSlot *remoteslot, + LogicalRepTupleData *newtup, + AttrMap *attrmap, LogicalRepRelation *remoterel) +{ + Relation rel = relinfo->ri_RelationDesc; + Oid idxoid; + EPQState epqstate; + TupleTableSlot *localslot; + bool found; + MemoryContext oldctx; + + localslot = table_slot_create(rel, &estate->es_tupleTable); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + + ExecOpenIndices(relinfo, false); + + /* + * Try to find tuple using either replica identity index, primary key or + * if needed, sequential scan. + */ + idxoid = GetRelationIdentityOrPK(rel); + Assert(OidIsValid(idxoid) || + (remoterel->replident == REPLICA_IDENTITY_FULL)); + + if (OidIsValid(idxoid)) + found = RelationFindReplTupleByIndex(rel, idxoid, + LockTupleExclusive, + remoteslot, localslot); + else + found = RelationFindReplTupleSeq(rel, LockTupleExclusive, + remoteslot, localslot); + + ExecClearTuple(remoteslot); + + /* + * Tuple found. + * + * Note this will fail if there are other conflicting unique indexes. + */ + if (found) + { + /* Process and store remote tuple in the slot */ + oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); + slot_modify_cstrings(remoteslot, localslot, + newtup->values, newtup->changed, + attrmap, remoterel, RelationGetRelid(rel)); + MemoryContextSwitchTo(oldctx); + + EvalPlanQualSetSlot(&epqstate, remoteslot); + + /* Do the actual update. */ + ExecSimpleRelationUpdate(estate, &epqstate, localslot, remoteslot); + } + else + { + /* + * The tuple to be updated could not be found. + * + * TODO what to do here, change the log level to LOG perhaps? + */ + elog(DEBUG1, + "logical replication did not find row for update " + "in replication target relation \"%s\"", + RelationGetRelationName(rel)); + } + + /* Cleanup. */ + ExecCloseIndices(relinfo); + EvalPlanQualEnd(&epqstate); +} + +/* Workhorse for apply_handle_delete() */ +static void +apply_handle_do_delete(ResultRelInfo *relinfo, EState *estate, + TupleTableSlot *remoteslot, + LogicalRepRelation *remoterel) +{ + Relation rel = relinfo->ri_RelationDesc; + Oid idxoid; + EPQState epqstate; + TupleTableSlot *localslot; + bool found; + + localslot = table_slot_create(rel, &estate->es_tupleTable); + EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + + /* + * Try to find tuple using either replica identity index, primary key or + * if needed, sequential scan. + */ + idxoid = GetRelationIdentityOrPK(rel); + Assert(OidIsValid(idxoid) || + (remoterel->replident == REPLICA_IDENTITY_FULL)); + + if (OidIsValid(idxoid)) + found = RelationFindReplTupleByIndex(rel, idxoid, + LockTupleExclusive, + remoteslot, localslot); + else + found = RelationFindReplTupleSeq(rel, LockTupleExclusive, + remoteslot, localslot); + ExecOpenIndices(relinfo, false); + + /* If found delete it. */ + if (found) + { + EvalPlanQualSetSlot(&epqstate, localslot); + + /* Do the actual delete. */ + ExecSimpleRelationDelete(estate, &epqstate, localslot); + } + else + { + /* The tuple to be deleted could not be found. */ + elog(DEBUG1, + "logical replication could not find row for delete " + "in replication target relation \"%s\"", + RelationGetRelationName(rel)); + } + + /* Cleanup. */ + ExecCloseIndices(relinfo); + EvalPlanQualEnd(&epqstate); +} + /* * Handle INSERT message. */ @@ -620,13 +764,10 @@ apply_handle_insert(StringInfo s) slot_fill_defaults(rel, estate, remoteslot); MemoryContextSwitchTo(oldctx); - ExecOpenIndices(estate->es_result_relation_info, false); - - /* Do the insert. */ - ExecSimpleRelationInsert(estate, remoteslot); + Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION); + apply_handle_do_insert(estate->es_result_relation_info, estate, + remoteslot); - /* Cleanup. */ - ExecCloseIndices(estate->es_result_relation_info); PopActiveSnapshot(); /* Handle queued AFTER triggers. */ @@ -683,16 +824,12 @@ apply_handle_update(StringInfo s) { LogicalRepRelMapEntry *rel; LogicalRepRelId relid; - Oid idxoid; EState *estate; - EPQState epqstate; LogicalRepTupleData oldtup; LogicalRepTupleData newtup; bool has_oldtup; - TupleTableSlot *localslot; TupleTableSlot *remoteslot; RangeTblEntry *target_rte; - bool found; MemoryContext oldctx; ensure_transaction(); @@ -718,9 +855,6 @@ apply_handle_update(StringInfo s) remoteslot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel->localrel), &TTSOpsVirtual); - localslot = table_slot_create(rel->localrel, - &estate->es_tupleTable); - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); /* * Populate updatedCols so that per-column triggers can fire. This could @@ -738,7 +872,6 @@ apply_handle_update(StringInfo s) } PushActiveSnapshot(GetTransactionSnapshot()); - ExecOpenIndices(estate->es_result_relation_info, false); /* Build the search tuple. */ oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); @@ -746,63 +879,16 @@ apply_handle_update(StringInfo s) has_oldtup ? oldtup.values : newtup.values); MemoryContextSwitchTo(oldctx); - /* - * Try to find tuple using either replica identity index, primary key or - * if needed, sequential scan. - */ - idxoid = GetRelationIdentityOrPK(rel->localrel); - Assert(OidIsValid(idxoid) || - (rel->remoterel.replident == REPLICA_IDENTITY_FULL && has_oldtup)); - - if (OidIsValid(idxoid)) - found = RelationFindReplTupleByIndex(rel->localrel, idxoid, - LockTupleExclusive, - remoteslot, localslot); - else - found = RelationFindReplTupleSeq(rel->localrel, LockTupleExclusive, - remoteslot, localslot); - - ExecClearTuple(remoteslot); - - /* - * Tuple found. - * - * Note this will fail if there are other conflicting unique indexes. - */ - if (found) - { - /* Process and store remote tuple in the slot */ - oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - slot_modify_cstrings(remoteslot, localslot, rel, - newtup.values, newtup.changed); - MemoryContextSwitchTo(oldctx); - - EvalPlanQualSetSlot(&epqstate, remoteslot); - - /* Do the actual update. */ - ExecSimpleRelationUpdate(estate, &epqstate, localslot, remoteslot); - } - else - { - /* - * The tuple to be updated could not be found. - * - * TODO what to do here, change the log level to LOG perhaps? - */ - elog(DEBUG1, - "logical replication did not find row for update " - "in replication target relation \"%s\"", - RelationGetRelationName(rel->localrel)); - } + Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION); + apply_handle_do_update(estate->es_result_relation_info, estate, + remoteslot, &newtup, rel->attrmap, + &rel->remoterel); - /* Cleanup. */ - ExecCloseIndices(estate->es_result_relation_info); PopActiveSnapshot(); /* Handle queued AFTER triggers. */ AfterTriggerEndQuery(estate); - EvalPlanQualEnd(&epqstate); ExecResetTupleTable(estate->es_tupleTable, false); FreeExecutorState(estate); @@ -822,12 +908,8 @@ apply_handle_delete(StringInfo s) LogicalRepRelMapEntry *rel; LogicalRepTupleData oldtup; LogicalRepRelId relid; - Oid idxoid; EState *estate; - EPQState epqstate; TupleTableSlot *remoteslot; - TupleTableSlot *localslot; - bool found; MemoryContext oldctx; ensure_transaction(); @@ -852,58 +934,25 @@ apply_handle_delete(StringInfo s) remoteslot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel->localrel), &TTSOpsVirtual); - localslot = table_slot_create(rel->localrel, - &estate->es_tupleTable); - EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1); + /* Input functions may need an active snapshot, so get one */ PushActiveSnapshot(GetTransactionSnapshot()); - ExecOpenIndices(estate->es_result_relation_info, false); - /* Find the tuple using the replica identity index. */ + /* Build the search tuple. */ oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); slot_store_cstrings(remoteslot, rel, oldtup.values); + slot_fill_defaults(rel, estate, remoteslot); MemoryContextSwitchTo(oldctx); - /* - * Try to find tuple using either replica identity index, primary key or - * if needed, sequential scan. - */ - idxoid = GetRelationIdentityOrPK(rel->localrel); - Assert(OidIsValid(idxoid) || - (rel->remoterel.replident == REPLICA_IDENTITY_FULL)); - - if (OidIsValid(idxoid)) - found = RelationFindReplTupleByIndex(rel->localrel, idxoid, - LockTupleExclusive, - remoteslot, localslot); - else - found = RelationFindReplTupleSeq(rel->localrel, LockTupleExclusive, - remoteslot, localslot); - /* If found delete it. */ - if (found) - { - EvalPlanQualSetSlot(&epqstate, localslot); - - /* Do the actual delete. */ - ExecSimpleRelationDelete(estate, &epqstate, localslot); - } - else - { - /* The tuple to be deleted could not be found. */ - elog(DEBUG1, - "logical replication could not find row for delete " - "in replication target relation \"%s\"", - RelationGetRelationName(rel->localrel)); - } + Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION); + apply_handle_do_delete(estate->es_result_relation_info, estate, + remoteslot, &rel->remoterel); - /* Cleanup. */ - ExecCloseIndices(estate->es_result_relation_info); PopActiveSnapshot(); /* Handle queued AFTER triggers. */ AfterTriggerEndQuery(estate); - EvalPlanQualEnd(&epqstate); ExecResetTupleTable(estate->es_tupleTable, false); FreeExecutorState(estate); -- 2.16.5
v8-0004-Publish-partitioned-table-inserts-as-its-own.patch
(text/plain, 41.9 KB)
From 365b81efed48fb0b1cdb6708fc3f4a9b82a84a22 Mon Sep 17 00:00:00 2001 From: amit <[email protected]> Date: Wed, 13 Nov 2019 17:18:51 +0900 Subject: [PATCH v8 4/4] Publish partitioned table inserts as its own --- doc/src/sgml/logical-replication.sgml | 11 +- src/backend/catalog/pg_publication.c | 11 +- src/backend/commands/subscriptioncmds.c | 103 +++++----- src/backend/executor/nodeModifyTable.c | 2 + src/backend/replication/logical/tablesync.c | 28 ++- src/backend/replication/logical/worker.c | 289 ++++++++++++++++++++++++++-- src/backend/replication/pgoutput/pgoutput.c | 191 ++++++++++++++---- src/include/catalog/pg_publication.h | 2 +- src/test/subscription/t/013_partition.pl | 48 ++++- 9 files changed, 558 insertions(+), 127 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 4584cb82f6..1a4d5a9d25 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -402,14 +402,9 @@ <listitem> <para> - Replication is only supported by regular and partitioned tables, although - the type of the table must match between the two servers, that is, one - cannot replicate from a regular table into a partitioned able or vice - versa. Also, when replicating between partitioned tables, the actual - replication occurs between leaf partitions, so the partitions on the two - servers must match one-to-one. Attempts to replicate other types of - relations such as views, materialized views, or foreign tables, will - result in an error. + Replication is only supported by regular and partitioned tables. + Attempts to replicate other types of relations such as + views, materialized views, or foreign tables, will result in an error. </para> </listitem> </itemizedlist> diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 6d2911d18f..d47461f763 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -243,20 +243,29 @@ GetRelationPublications(Oid relid) /* * Finds all publications that publish changes to the input relation's * ancestors. + * + * *published_ancestors will contain one OID for each publication returned, + * of the ancestor which belongs to it. Values in this list can be repeated, + * because a given ancestor may belong to multiple publications. */ List * -GetRelationAncestorPublications(Oid relid) +GetRelationAncestorPublications(Oid relid, List **published_ancestors) { List *ancestors = get_partition_ancestors(relid); List *ancestor_pubids = NIL; ListCell *lc; + *published_ancestors = NIL; foreach(lc, ancestors) { Oid ancestor = lfirst_oid(lc); List *rel_publishers = GetRelationPublications(ancestor); + int n = list_length(rel_publishers), + i; ancestor_pubids = list_concat_copy(ancestor_pubids, rel_publishers); + for (i = 0; i < n; i++) + *published_ancestors = lappend_oid(*published_ancestors, ancestor); } return ancestor_pubids; diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 786b15eb27..2a45aff445 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -54,6 +54,15 @@ typedef struct PublishedTable RangeVar *rv; char relkind; + + /* + * If the published table is partitioned, the following being true means + * its changes are published using own schema rather than the schema of + * its individual partitions. In the latter case, a separate + * PublicationTable instance (and hence pg_subscription_rel entry) for + * each partition will be needed. + */ + bool published_using_root_schema; } PublishedTable; static List *fetch_publication_tables(WalReceiverConn *wrconn, List *publications); @@ -481,24 +490,13 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) rv->schemaname, rv->relname); /* - * Currently, partitioned table replication occurs between leaf - * partitions, so both the source and the target tables must be - * partitioned. + * A partitioned table doesn't need local state if the state + * is managed for individual partitions, which is the case if + * the partitioned table is published using the schema of its + * partitions. */ - if (pt->relkind == RELKIND_RELATION && - local_relkind == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - (errcode(ERRCODE_WRONG_OBJECT_TYPE), - errmsg("cannot use relation \"%s.%s\" as logical replication target", - rv->schemaname, rv->relname), - errdetail("\"%s.%s\" is a partitioned table whereas it is a regular table on publication server.", - rv->schemaname, rv->relname))); - - /* - * A partitioned table doesn't need local state, because the - * state is managed for individual partitions instead. - */ - if (pt->relkind == RELKIND_PARTITIONED_TABLE) + if (pt->relkind == RELKIND_PARTITIONED_TABLE && + !pt->published_using_root_schema) continue; AddSubscriptionRelState(subid, relid, table_state, @@ -614,24 +612,12 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) rv->schemaname, rv->relname); /* - * Currently, partitioned table replication occurs between leaf - * partitions, so both the source and the target tables must be - * partitioned. + * A partitioned table doesn't need local state if the state is + * managed for individual partitions, which is the case if the + * partitioned table is published using the schema of its partitions. */ - if (pt->relkind == RELKIND_RELATION && - local_relkind == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - (errcode(ERRCODE_WRONG_OBJECT_TYPE), - errmsg("cannot use relation \"%s.%s\" as logical replication target", - rv->schemaname, rv->relname), - errdetail("\"%s.%s\" is a partitioned table whereas it is a regular table on publication server.", - rv->schemaname, rv->relname))); - - /* - * A partitioned table doesn't need local state, because the - * state is managed for individual partitions instead. - */ - if (pt->relkind == RELKIND_PARTITIONED_TABLE) + if (pt->relkind == RELKIND_PARTITIONED_TABLE && + !pt->published_using_root_schema) continue; pubrel_local_oids[off++] = relid; @@ -1191,7 +1177,7 @@ fetch_publication_tables(WalReceiverConn *wrconn, List *publications) WalRcvExecResult *res; StringInfoData cmd; TupleTableSlot *slot; - Oid tableRow[3] = {TEXTOID, TEXTOID, CHAROID}; + Oid tableRow[4] = {TEXTOID, TEXTOID, CHAROID, BOOLOID}; ListCell *lc; bool first; List *tablelist = NIL; @@ -1199,27 +1185,41 @@ fetch_publication_tables(WalReceiverConn *wrconn, List *publications) Assert(list_length(publications) > 0); initStringInfo(&cmd); - appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.tablename, s.relkind FROM (\n" - " SELECT t.pubname, t.schemaname, t.tablename, c.relkind\n" - " FROM pg_catalog.pg_publication_tables t\n" - " JOIN pg_catalog.pg_class c \n" - " ON t.schemaname = c.relnamespace::pg_catalog.regnamespace::name\n" - " AND t.tablename = c.relname \n"); + appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.tablename, s.relkind, s.pubasroot FROM (\n"); /* * As of v13, partitioned tables can be published, although their changes - * are published as their partitions', so we will need the partitions in - * the result. + * may be published either as their own or as their partitions', which is + * checked with pg_publication.pubasroot (whether the publication publishes + * using root partitioned table's schema). + */ + if (walrcv_server_version(wrconn) >= 130000) + appendStringInfoString(&cmd, " SELECT t.pubname, t.schemaname, t.tablename, c.relkind, p.pubasroot\n"); + else + appendStringInfoString(&cmd, " SELECT t.pubname, t.schemaname, t.tablename, c.relkind, false AS pubasroot\n"); + + appendStringInfoString(&cmd, " FROM pg_catalog.pg_publication_tables t\n" + " JOIN pg_catalog.pg_publication p ON t.pubname = p.pubname\n" + " JOIN pg_catalog.pg_class c\n" + " ON t.schemaname = c.relnamespace::pg_catalog.regnamespace::pg_catalog.name\n" + " AND t.tablename = c.relname\n"); + + /* + * If publication doesn't publish using the root table's schema, we will + * need partitions in the result. */ if (walrcv_server_version(wrconn) >= 130000) appendStringInfoString(&cmd, " UNION\n" - " SELECT t.pubname, s.schemaname, s.tablename, s.relkind\n" - " FROM pg_catalog.pg_publication_tables t,\n" - " LATERAL (SELECT c.relnamespace::regnamespace::name, c.relname, c.relkind\n" - " FROM pg_class c\n" - " JOIN pg_partition_tree(t.schemaname || '.' || t.tablename) p\n" - " ON p.relid = c.oid\n" - " WHERE p.level > 0) AS s(schemaname, tablename, relkind)\n"); + " SELECT DISTINCT t.pubname, s.schemaname, s.tablename, c.relkind, false AS pubasroot\n" + " FROM pg_catalog.pg_publication_tables t\n" + " JOIN pg_catalog.pg_publication p ON t.pubname = p.pubname AND NOT p.pubasroot,\n" + " LATERAL (SELECT c.relnamespace::pg_catalog.regnamespace::pg_catalog.name, c.relname\n" + " FROM pg_catalog.pg_class c\n" + " JOIN pg_catalog.pg_partition_tree(t.schemaname || '.' || t.tablename) p\n" + " ON p.relid = c.oid\n" + " WHERE p.level > 0) AS s(schemaname, tablename)\n" + " JOIN pg_catalog.pg_class c ON s.schemaname = c.relnamespace::pg_catalog.regnamespace::pg_catalog.name\n" + " AND s.tablename = c.relname\n"); appendStringInfoString(&cmd, ") s WHERE s.pubname IN ("); @@ -1237,7 +1237,7 @@ fetch_publication_tables(WalReceiverConn *wrconn, List *publications) } appendStringInfoChar(&cmd, ')'); - res = walrcv_exec(wrconn, cmd.data, 3, tableRow); + res = walrcv_exec(wrconn, cmd.data, 4, tableRow); pfree(cmd.data); if (res->status != WALRCV_OK_TUPLES) @@ -1260,6 +1260,7 @@ fetch_publication_tables(WalReceiverConn *wrconn, List *publications) Assert(!isnull); pt->rv = makeRangeVar(pstrdup(nspname), pstrdup(relname), -1); pt->relkind = DatumGetChar(slot_getattr(slot, 3, &isnull)); + pt->published_using_root_schema = DatumGetBool(slot_getattr(slot, 4, &isnull)); Assert(!isnull); tablelist = lappend(tablelist, pt); diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index 63e108bb56..5b7265939f 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -2299,6 +2299,8 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) { mtstate->rootResultRelInfo = estate->es_root_result_relations + node->rootResultRelIndex; + CheckValidResultRel(mtstate->rootResultRelInfo, + mtstate->rootResultRelInfo, operation); rootResultRelInfo = mtstate->rootResultRelInfo; } diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 98825f01e9..6a18b78f22 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -630,16 +630,17 @@ copy_read_data(void *outbuf, int minread, int maxread) /* * Get information about remote relation in similar fashion the RELATION - * message provides during replication. + * message provides during replication. XXX - while we fetch relkind too + * here, the RELATION message doesn't provide it */ static void fetch_remote_table_info(char *nspname, char *relname, - LogicalRepRelation *lrel) + LogicalRepRelation *lrel, char *relkind) { WalRcvExecResult *res; StringInfoData cmd; TupleTableSlot *slot; - Oid tableRow[2] = {OIDOID, CHAROID}; + Oid tableRow[3] = {OIDOID, CHAROID, CHAROID}; Oid attrRow[4] = {TEXTOID, OIDOID, INT4OID, BOOLOID}; bool isnull; int natt; @@ -649,16 +650,16 @@ fetch_remote_table_info(char *nspname, char *relname, /* First fetch Oid and replica identity. */ initStringInfo(&cmd); - appendStringInfo(&cmd, "SELECT c.oid, c.relreplident" + appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind" " FROM pg_catalog.pg_class c" " INNER JOIN pg_catalog.pg_namespace n" " ON (c.relnamespace = n.oid)" " WHERE n.nspname = %s" " AND c.relname = %s" - " AND c.relkind = 'r'", + " AND pg_relation_is_publishable(c.oid)", quote_literal_cstr(nspname), quote_literal_cstr(relname)); - res = walrcv_exec(wrconn, cmd.data, 2, tableRow); + res = walrcv_exec(wrconn, cmd.data, 3, tableRow); if (res->status != WALRCV_OK_TUPLES) ereport(ERROR, @@ -675,6 +676,8 @@ fetch_remote_table_info(char *nspname, char *relname, Assert(!isnull); lrel->replident = DatumGetChar(slot_getattr(slot, 2, &isnull)); Assert(!isnull); + *relkind = DatumGetChar(slot_getattr(slot, 3, &isnull)); + Assert(!isnull); ExecDropSingleTupleTableSlot(slot); walrcv_clear_result(res); @@ -750,10 +753,12 @@ copy_table(Relation rel) CopyState cstate; List *attnamelist; ParseState *pstate; + char remote_relkind; /* Get the publisher relation info. */ fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)), - RelationGetRelationName(rel), &lrel); + RelationGetRelationName(rel), &lrel, + &remote_relkind); /* Put the relation into relmap. */ logicalrep_relmap_update(&lrel); @@ -761,12 +766,15 @@ copy_table(Relation rel) /* Map the publisher relation to local one. */ relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock); Assert(rel == relmapentry->localrel); - Assert(relmapentry->localrel->rd_rel->relkind == RELKIND_RELATION); /* Start copy on the publisher. */ initStringInfo(&cmd); - appendStringInfo(&cmd, "COPY %s TO STDOUT", - quote_qualified_identifier(lrel.nspname, lrel.relname)); + if (remote_relkind == RELKIND_PARTITIONED_TABLE) + appendStringInfo(&cmd, "COPY (SELECT * FROM %s) TO STDOUT", + quote_qualified_identifier(lrel.nspname, lrel.relname)); + else + appendStringInfo(&cmd, "COPY %s TO STDOUT", + quote_qualified_identifier(lrel.nspname, lrel.relname)); res = walrcv_exec(wrconn, cmd.data, 0, NULL); pfree(cmd.data); if (res->status != WALRCV_OK_COPY_OUT) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 34b0ac78cc..ec34418f75 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -29,11 +29,14 @@ #include "access/xlog_internal.h" #include "catalog/catalog.h" #include "catalog/namespace.h" +#include "catalog/partition.h" +#include "catalog/pg_inherits.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "executor/executor.h" +#include "executor/execPartition.h" #include "executor/nodeModifyTable.h" #include "funcapi.h" #include "libpq/pqformat.h" @@ -722,6 +725,180 @@ apply_handle_do_delete(ResultRelInfo *relinfo, EState *estate, EvalPlanQualEnd(&epqstate); } +/* + * This handles insert, update, delete on a partitioned table. + */ +static void +apply_handle_tuple_routing(ResultRelInfo *relinfo, + LogicalRepRelMapEntry *relmapentry, + EState *estate, CmdType operation, + TupleTableSlot *remoteslot, + LogicalRepTupleData *newtup) +{ + Relation rel = relinfo->ri_RelationDesc; + ModifyTableState *mtstate = NULL; + PartitionTupleRouting *proute = NULL; + ResultRelInfo *partrelinfo; + TupleTableSlot *localslot; + PartitionRoutingInfo *partinfo; + TupleConversionMap *map; + MemoryContext oldctx; + + /* ModifyTableState is needed for ExecFindPartition(). */ + mtstate = makeNode(ModifyTableState); + mtstate->ps.plan = NULL; + mtstate->ps.state = estate; + mtstate->operation = operation; + mtstate->resultRelInfo = relinfo; + proute = ExecSetupPartitionTupleRouting(estate, mtstate, rel); + + /* + * Find a partition for the tuple contained in remoteslot. + * + * For insert, remoteslot is tuple to insert. For update and delete, it + * is the tuple to be replaced and deleted, respectively. + */ + Assert(remoteslot != NULL); + oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); + /* The following throws error if a suitable partition is not found. */ + partrelinfo = ExecFindPartition(mtstate, relinfo, proute, + remoteslot, estate); + Assert(partrelinfo != NULL); + /* Convert the tuple to match the partition's rowtype. */ + partinfo = partrelinfo->ri_PartitionInfo; + map = partinfo->pi_RootToPartitionMap; + if (map != NULL) + { + TupleTableSlot *part_slot = partinfo->pi_PartitionTupleSlot; + + remoteslot = execute_attr_map_slot(map->attrMap, remoteslot, + part_slot); + } + MemoryContextSwitchTo(oldctx); + + switch (operation) + { + case CMD_INSERT: + /* Just insert into the partition. */ + estate->es_result_relation_info = partrelinfo; + apply_handle_do_insert(partrelinfo, estate, remoteslot); + break; + + case CMD_DELETE: + /* Just delete from the partition. */ + estate->es_result_relation_info = partrelinfo; + apply_handle_do_delete(partrelinfo, estate, remoteslot, + &relmapentry->remoterel); + break; + + case CMD_UPDATE: + { + ResultRelInfo *partrelinfo_new; + + /* + * partrelinfo computed above is the partition which might + * contain the search tuple. Now find the partition for the + * replacement tuple, which might not be the same as + * partrelinfo. + */ + localslot = table_slot_create(rel, &estate->es_tupleTable); + oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); + slot_modify_cstrings(localslot, remoteslot, + newtup->values, newtup->changed, + relmapentry->attrmap, + &relmapentry->remoterel, + RelationGetRelid(rel)); + partrelinfo_new = ExecFindPartition(mtstate, relinfo, proute, + localslot, estate); + MemoryContextSwitchTo(oldctx); + + /* + * If both search and replacement tuple would be in the same + * partition, we can apply this as an UPDATE on the parttion. + */ + if (partrelinfo == partrelinfo_new) + { + AttrMap *attrmap = relmapentry->attrmap, + *new_attrmap = NULL; + + /* + * If the partition's attributes don't match the root + * relation's, we'll need to make a new attrmap which maps + * partition attribute numbers to remoterel's, instead + * the original which maps root relation's attribute + * numbers to remoterel's. + */ + if (map) + { + TupleDesc partdesc = RelationGetDescr(partrelinfo->ri_RelationDesc); + TupleDesc rootdesc = RelationGetDescr(rel); + AttrMap *partToRootMap; + AttrNumber attno; + + /* Need the reverse map here */ + partToRootMap = build_attrmap_by_name(partdesc, rootdesc); + new_attrmap = make_attrmap(partdesc->natts); + memset(new_attrmap->attnums, -1, + new_attrmap->maplen * sizeof(AttrNumber)); + for (attno = 0; attno < new_attrmap->maplen; attno++) + { + AttrNumber root_attno = partToRootMap->attnums[attno]; + + new_attrmap->attnums[attno] = attrmap->attnums[root_attno - 1]; + } + attrmap = new_attrmap; + } + + /* UPDATE partition. */ + estate->es_result_relation_info = partrelinfo; + apply_handle_do_update(partrelinfo, estate, remoteslot, + newtup, attrmap, + &relmapentry->remoterel); + if (new_attrmap) + free_attrmap(new_attrmap); + } + else + { + /* + * Different, so handle this as DELETE followed by INSERT. + */ + + /* DELETE from partition partrelinfo. */ + estate->es_result_relation_info = partrelinfo; + apply_handle_do_delete(partrelinfo, estate, remoteslot, + &relmapentry->remoterel); + + /* + * Convert the replacement tuple to match the destination + * partition rowtype. + */ + oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); + partinfo = partrelinfo_new->ri_PartitionInfo; + map = partinfo->pi_RootToPartitionMap; + if (map != NULL) + { + TupleTableSlot *part_slot = partinfo->pi_PartitionTupleSlot; + + localslot = execute_attr_map_slot(map->attrMap, localslot, + part_slot); + } + MemoryContextSwitchTo(oldctx); + /* INSERT into partition partrelinfo_new. */ + estate->es_result_relation_info = partrelinfo_new; + apply_handle_do_insert(partrelinfo_new, estate, + localslot); + } + } + break; + + default: + elog(ERROR, "unrecognized CmdType: %d", (int) operation); + break; + } + + ExecCleanupTupleRouting(mtstate, proute); +} + /* * Handle INSERT message. */ @@ -764,9 +941,13 @@ apply_handle_insert(StringInfo s) slot_fill_defaults(rel, estate, remoteslot); MemoryContextSwitchTo(oldctx); - Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION); - apply_handle_do_insert(estate->es_result_relation_info, estate, - remoteslot); + /* For a partitioned table, insert the tuple into a partition. */ + if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + apply_handle_tuple_routing(estate->es_result_relation_info, rel, + estate, CMD_INSERT, remoteslot, NULL); + else + apply_handle_do_insert(estate->es_result_relation_info, estate, + remoteslot); PopActiveSnapshot(); @@ -879,10 +1060,14 @@ apply_handle_update(StringInfo s) has_oldtup ? oldtup.values : newtup.values); MemoryContextSwitchTo(oldctx); - Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION); - apply_handle_do_update(estate->es_result_relation_info, estate, - remoteslot, &newtup, rel->attrmap, - &rel->remoterel); + /* For a partitioned table, apply update to correct partition. */ + if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + apply_handle_tuple_routing(estate->es_result_relation_info, rel, + estate, CMD_UPDATE, remoteslot, &newtup); + else + apply_handle_do_update(estate->es_result_relation_info, estate, + remoteslot, &newtup, rel->attrmap, + &rel->remoterel); PopActiveSnapshot(); @@ -944,9 +1129,13 @@ apply_handle_delete(StringInfo s) slot_fill_defaults(rel, estate, remoteslot); MemoryContextSwitchTo(oldctx); - Assert(rel->localrel->rd_rel->relkind == RELKIND_RELATION); - apply_handle_do_delete(estate->es_result_relation_info, estate, - remoteslot, &rel->remoterel); + /* For a partitioned table, apply delete to correct partition. */ + if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + apply_handle_tuple_routing(estate->es_result_relation_info, rel, + estate, CMD_DELETE, remoteslot, NULL); + else + apply_handle_do_delete(estate->es_result_relation_info, estate, + remoteslot, &rel->remoterel); PopActiveSnapshot(); @@ -988,14 +1177,43 @@ apply_handle_truncate(StringInfo s) LogicalRepRelMapEntry *rel; rel = logicalrep_rel_open(relid, RowExclusiveLock); + if (!should_apply_changes_for_rel(rel)) { + bool really_skip = true; + + /* + * If we seem to have gotten sent a leaf partition because an + * ancestor was truncated, confirm before proceeding with + * truncating the partition that an ancestor indeed has a valid + * subscription state. + */ + if (rel->state == SUBREL_STATE_UNKNOWN && + rel->localrel->rd_rel->relispartition) + { + List *ancestors = get_partition_ancestors(rel->localreloid); + ListCell *lc1; + + foreach(lc1, ancestors) + { + Oid anc_oid = lfirst_oid(lc1); + LogicalRepRelMapEntry *anc_rel; + + anc_rel = logicalrep_rel_open(anc_oid, RowExclusiveLock); + really_skip &= !should_apply_changes_for_rel(anc_rel); + logicalrep_rel_close(anc_rel, RowExclusiveLock); + } + } + /* * The relation can't become interesting in the middle of the * transaction so it's safe to unlock it. */ - logicalrep_rel_close(rel, RowExclusiveLock); - continue; + if (really_skip) + { + logicalrep_rel_close(rel, RowExclusiveLock); + continue; + } } remote_rels = lappend(remote_rels, rel); @@ -1003,6 +1221,47 @@ apply_handle_truncate(StringInfo s) relids = lappend_oid(relids, rel->localreloid); if (RelationIsLogicallyLogged(rel->localrel)) relids_logged = lappend_oid(relids_logged, rel->localreloid); + + if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + { + ListCell *child; + List *children = find_all_inheritors(rel->localreloid, + RowExclusiveLock, + NULL); + + foreach(child, children) + { + Oid childrelid = lfirst_oid(child); + Relation childrel; + + if (list_member_oid(relids, childrelid)) + continue; + + /* find_all_inheritors already got lock */ + childrel = table_open(childrelid, NoLock); + + /* + * It is possible that the parent table has children that are + * temp tables of other backends. We cannot safely access + * such tables (because of buffering issues), and the best + * thing to do is to silently ignore them. Note that this + * check is the same as one of the checks done in + * truncate_check_activity() called below, still it is kept + * here for simplicity. + */ + if (RELATION_IS_OTHER_TEMP(childrel)) + { + table_close(childrel, RowExclusiveLock); + continue; + } + + rels = lappend(rels, childrel); + relids = lappend_oid(relids, childrelid); + /* Log this relation only if needed for logical decoding */ + if (RelationIsLogicallyLogged(childrel)) + relids_logged = lappend_oid(relids_logged, childrelid); + } + } } /* @@ -1012,11 +1271,11 @@ apply_handle_truncate(StringInfo s) */ ExecuteTruncateGuts(rels, relids, relids_logged, DROP_RESTRICT, restart_seqs); - foreach(lc, remote_rels) + foreach(lc, rels) { - LogicalRepRelMapEntry *rel = lfirst(lc); + Relation rel = lfirst(lc); - logicalrep_rel_close(rel, NoLock); + table_close(rel, NoLock); } CommandCounterIncrement(); diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 059d2c9194..99ceae0d5f 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -12,6 +12,7 @@ */ #include "postgres.h" +#include "access/tupconvert.h" #include "catalog/pg_publication.h" #include "fmgr.h" #include "replication/logical.h" @@ -49,6 +50,7 @@ static bool publications_valid; static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); +static void send_relation_and_attrs(Relation relation, LogicalDecodingContext *ctx); /* * Entry in the map used to remember which relation schemas we sent. @@ -59,9 +61,22 @@ static void publication_invalidation_cb(Datum arg, int cacheid, typedef struct RelationSyncEntry { Oid relid; /* relation oid */ - bool schema_sent; /* did we send the schema? */ + + /* + * Did we send the schema? If ancestor relid is set, its schema must also + * have been sent for this to be true. + */ + bool schema_sent; bool replicate_valid; PublicationActions pubactions; + + /* + * Valid if publishing relation's changes as changes to some ancestor, + * that is, if relation is a partition. The map, if any, will be used to + * convert the tuples from partition's type to the ancestor's. + */ + Oid replicate_as_relid; + TupleConversionMap *map; } RelationSyncEntry; /* Map used to remember which relation schemas we sent. */ @@ -259,47 +274,72 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } /* - * Write the relation schema if the current schema hasn't been sent yet. + * Write the current schema of the relation and its ancestor (if any) if not + * done yet. */ static void maybe_send_schema(LogicalDecodingContext *ctx, Relation relation, RelationSyncEntry *relentry) { - if (!relentry->schema_sent) + if (relentry->schema_sent) + return; + + /* If needed, send the ancestor's schema first. */ + if (OidIsValid(relentry->replicate_as_relid)) { - TupleDesc desc; - int i; + Relation ancestor = + RelationIdGetRelation(relentry->replicate_as_relid); + TupleDesc indesc = RelationGetDescr(relation); + TupleDesc outdesc = RelationGetDescr(ancestor); + MemoryContext oldctx; + + /* Map must live as long as the session does. */ + oldctx = MemoryContextSwitchTo(CacheMemoryContext); + relentry->map = convert_tuples_by_name(indesc, outdesc); + MemoryContextSwitchTo(oldctx); + send_relation_and_attrs(ancestor, ctx); + RelationClose(ancestor); + } - desc = RelationGetDescr(relation); + send_relation_and_attrs(relation, ctx); + relentry->schema_sent = true; +} - /* - * Write out type info if needed. We do that only for user-created - * types. We use FirstGenbkiObjectId as the cutoff, so that we only - * consider objects with hand-assigned OIDs to be "built in", not for - * instance any function or type defined in the information_schema. - * This is important because only hand-assigned OIDs can be expected - * to remain stable across major versions. - */ - for (i = 0; i < desc->natts; i++) - { - Form_pg_attribute att = TupleDescAttr(desc, i); +/* + * Sends a relation + */ +static void +send_relation_and_attrs(Relation relation, LogicalDecodingContext *ctx) +{ + TupleDesc desc = RelationGetDescr(relation); + int i; - if (att->attisdropped || att->attgenerated) - continue; + /* + * Write out type info if needed. We do that only for user-created types. + * We use FirstGenbkiObjectId as the cutoff, so that we only consider + * objects with hand-assigned OIDs to be "built in", not for instance any + * function or type defined in the information_schema. This is important + * because only hand-assigned OIDs can be expected to remain stable across + * major versions. + */ + for (i = 0; i < desc->natts; i++) + { + Form_pg_attribute att = TupleDescAttr(desc, i); - if (att->atttypid < FirstGenbkiObjectId) - continue; + if (att->attisdropped || att->attgenerated) + continue; - OutputPluginPrepareWrite(ctx, false); - logicalrep_write_typ(ctx->out, att->atttypid); - OutputPluginWrite(ctx, false); - } + if (att->atttypid < FirstGenbkiObjectId) + continue; OutputPluginPrepareWrite(ctx, false); - logicalrep_write_rel(ctx->out, relation); + logicalrep_write_typ(ctx->out, att->atttypid); OutputPluginWrite(ctx, false); - relentry->schema_sent = true; } + + OutputPluginPrepareWrite(ctx, false); + logicalrep_write_rel(ctx->out, relation); + OutputPluginWrite(ctx, false); } /* @@ -346,28 +386,65 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, switch (change->action) { case REORDER_BUFFER_CHANGE_INSERT: - OutputPluginPrepareWrite(ctx, true); - logicalrep_write_insert(ctx->out, relation, - &change->data.tp.newtuple->tuple); - OutputPluginWrite(ctx, true); - break; + { + HeapTuple tuple = &change->data.tp.newtuple->tuple; + + /* Publish as root relation change if requested. */ + if (OidIsValid(relentry->replicate_as_relid)) + { + Assert(relation->rd_rel->relispartition); + relation = RelationIdGetRelation(relentry->replicate_as_relid); + /* Convert tuple if needed. */ + if (relentry->map) + tuple = execute_attr_map_tuple(tuple, relentry->map); + } + + OutputPluginPrepareWrite(ctx, true); + logicalrep_write_insert(ctx->out, relation, tuple); + OutputPluginWrite(ctx, true); + break; + } case REORDER_BUFFER_CHANGE_UPDATE: { HeapTuple oldtuple = change->data.tp.oldtuple ? &change->data.tp.oldtuple->tuple : NULL; + HeapTuple newtuple = &change->data.tp.newtuple->tuple; + + /* Publish as root relation change if requested. */ + if (OidIsValid(relentry->replicate_as_relid)) + { + Assert(relation->rd_rel->relispartition); + relation = RelationIdGetRelation(relentry->replicate_as_relid); + /* Convert tuples if needed. */ + if (relentry->map) + { + oldtuple = execute_attr_map_tuple(oldtuple, relentry->map); + newtuple = execute_attr_map_tuple(newtuple, relentry->map); + } + } OutputPluginPrepareWrite(ctx, true); - logicalrep_write_update(ctx->out, relation, oldtuple, - &change->data.tp.newtuple->tuple); + logicalrep_write_update(ctx->out, relation, oldtuple, newtuple); OutputPluginWrite(ctx, true); break; } case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) { + HeapTuple oldtuple = &change->data.tp.oldtuple->tuple; + + /* Publish as root relation change if requested. */ + if (OidIsValid(relentry->replicate_as_relid)) + { + Assert(relation->rd_rel->relispartition); + relation = RelationIdGetRelation(relentry->replicate_as_relid); + /* Convert tuple if needed. */ + if (relentry->map) + oldtuple = execute_attr_map_tuple(oldtuple, relentry->map); + } + OutputPluginPrepareWrite(ctx, true); - logicalrep_write_delete(ctx->out, relation, - &change->data.tp.oldtuple->tuple); + logicalrep_write_delete(ctx->out, relation, oldtuple); OutputPluginWrite(ctx, true); } else @@ -411,6 +488,28 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (!relentry->pubactions.pubtruncate) continue; + /* + * If this partition was not *directly* truncated, don't bother + * sending it to the subscriber. + */ + if (OidIsValid(relentry->replicate_as_relid)) + { + int j; + bool can_skip_part_trunc = false; + + for (j = 0; j < nrelids; j++) + { + if (relentry->replicate_as_relid == relids[j]) + { + can_skip_part_trunc = true; + break; + } + } + + if (can_skip_part_trunc) + continue; + } + relids[nrelids++] = relid; maybe_send_schema(ctx, relation, relentry); } @@ -529,6 +628,11 @@ init_rel_sync_cache(MemoryContext cachectx) /* * Find or create entry in the relation schema cache. + * + * For a partition, the schema of the top-most ancestor that is published + * will be used in some cases, instead of that of the partition itself, so + * the information about ancestor's publications is looked up here and saved in + * the schema cache entry. */ static RelationSyncEntry * get_rel_sync_entry(PGOutputData *data, Relation rel) @@ -553,8 +657,11 @@ get_rel_sync_entry(PGOutputData *data, Relation rel) { List *pubids = GetRelationPublications(relid); ListCell *lc, - *lc1; + *lc1, + *lc2; List *ancestor_pubids = NIL; + List *published_ancestors = NIL; + Oid topmost_published_ancestor = InvalidOid; /* Reload publications if needed before use. */ if (!publications_valid) @@ -579,7 +686,9 @@ get_rel_sync_entry(PGOutputData *data, Relation rel) /* For partitions, also consider publications of ancestors. */ if (rel->rd_rel->relispartition) ancestor_pubids = - GetRelationAncestorPublications(RelationGetRelid(rel)); + GetRelationAncestorPublications(RelationGetRelid(rel), + &published_ancestors); + Assert(list_length(ancestor_pubids) == list_length(published_ancestors)); foreach(lc, data->publications) { @@ -597,7 +706,7 @@ get_rel_sync_entry(PGOutputData *data, Relation rel) entry->pubactions.pubdelete && entry->pubactions.pubtruncate) break; - foreach(lc1, ancestor_pubids) + forboth(lc1, ancestor_pubids, lc2, published_ancestors) { if (lfirst_oid(lc1) == pub->oid) { @@ -605,6 +714,8 @@ get_rel_sync_entry(PGOutputData *data, Relation rel) entry->pubactions.pubupdate |= pub->pubactions.pubupdate; entry->pubactions.pubdelete |= pub->pubactions.pubdelete; entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate; + if (pub->publish_using_root_schema) + topmost_published_ancestor = lfirst_oid(lc2); } } @@ -615,7 +726,9 @@ get_rel_sync_entry(PGOutputData *data, Relation rel) list_free(pubids); list_free(ancestor_pubids); + list_free(published_ancestors); + entry->replicate_as_relid = topmost_published_ancestor; entry->replicate_valid = true; } diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 9d13e5c735..0a45c11d7d 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -83,7 +83,7 @@ typedef struct Publication extern Publication *GetPublication(Oid pubid); extern Publication *GetPublicationByName(const char *pubname, bool missing_ok); extern List *GetRelationPublications(Oid relid); -extern List *GetRelationAncestorPublications(Oid relid); +extern List *GetRelationAncestorPublications(Oid relid, List **published_ancestors); extern List *GetPublicationRelations(Oid pubid); extern List *GetAllTablesPublications(void); extern List *GetAllTablesPublicationRelations(void); diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl index eb0f1cd6a8..957c7b4be1 100644 --- a/src/test/subscription/t/013_partition.pl +++ b/src/test/subscription/t/013_partition.pl @@ -3,7 +3,7 @@ use strict; use warnings; use PostgresNode; use TestLib; -use Test::More tests => 10; +use Test::More tests => 16; # setup @@ -41,21 +41,38 @@ $node_publisher->safe_psql('postgres', "CREATE TABLE tab1_2 PARTITION OF tab1 FOR VALUES IN (5, 6)"); $node_subscriber1->safe_psql('postgres', - "CREATE TABLE tab1_2 PARTITION OF tab1 (c DEFAULT 'sub1_tab1') FOR VALUES IN (5, 6)"); + "CREATE TABLE tab1_2 PARTITION OF tab1 (c DEFAULT 'sub1_tab1') FOR VALUES IN (5, 6) PARTITION BY LIST (a)"); +$node_subscriber1->safe_psql('postgres', + "CREATE TABLE tab1_2_1 PARTITION OF tab1_2 FOR VALUES IN (5)"); +$node_subscriber1->safe_psql('postgres', + "CREATE TABLE tab1_2_2 PARTITION OF tab1_2 FOR VALUES IN (6)"); $node_subscriber2->safe_psql('postgres', "CREATE TABLE tab1_2 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1_2', b text)"); +$node_subscriber2->safe_psql('postgres', + "CREATE TABLE tab1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab1', b text) PARTITION BY HASH (a)"); +$node_subscriber2->safe_psql('postgres', + "CREATE TABLE tab1_part1 (b text, c text, a int NOT NULL)"); +$node_subscriber2->safe_psql('postgres', + "ALTER TABLE tab1 ATTACH PARTITION tab1_part1 FOR VALUES WITH (MODULUS 2, REMAINDER 0)"); +$node_subscriber2->safe_psql('postgres', + "CREATE TABLE tab1_part2 PARTITION OF tab1 FOR VALUES WITH (MODULUS 2, REMAINDER 1)"); + $node_publisher->safe_psql('postgres', "CREATE PUBLICATION pub1 FOR TABLE tab1, tab1_1"); $node_publisher->safe_psql('postgres', "CREATE PUBLICATION pub2 FOR TABLE tab1_2"); +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION pub3 FOR TABLE tab1 WITH (publish_using_root_schema = true)"); $node_subscriber1->safe_psql('postgres', "CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"); $node_subscriber2->safe_psql('postgres', "CREATE SUBSCRIPTION sub2 CONNECTION '$publisher_connstr' PUBLICATION pub2"); +$node_subscriber2->safe_psql('postgres', + "CREATE SUBSCRIPTION sub3 CONNECTION '$publisher_connstr' PUBLICATION pub3"); # Wait for initial sync of all subscriptions my $synced_query = @@ -85,17 +102,26 @@ $result = $node_subscriber2->safe_psql('postgres', "SELECT c, count(*), min(a), max(a) FROM tab1_2 GROUP BY 1"); is($result, qq(sub2_tab1_2|1|5|5), 'inserts into tab1_2 replicated'); +$result = $node_subscriber2->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1 GROUP BY 1"); +is($result, qq(sub2_tab1|3|1|5), 'inserts into tab1_2 replicated'); + # update a row (no partition change) $node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 2 WHERE a = 1"); $node_publisher->wait_for_catchup('sub1'); +$node_publisher->wait_for_catchup('sub2'); $result = $node_subscriber1->safe_psql('postgres', "SELECT c, count(*), min(a), max(a) FROM tab1 GROUP BY 1"); is($result, qq(sub1_tab1|3|2|5), 'update of tab1_1 replicated'); +$result = $node_subscriber2->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1 GROUP BY 1"); +is($result, qq(sub2_tab1|3|2|5), 'update of tab1_1 replicated'); + # update a row (partition changes) $node_publisher->safe_psql('postgres', @@ -112,6 +138,10 @@ $result = $node_subscriber2->safe_psql('postgres', "SELECT c, count(*), min(a), max(a) FROM tab1_2 GROUP BY 1"); is($result, qq(sub2_tab1_2|2|5|6), 'insert into tab1_2 replicated'); +$result = $node_subscriber2->safe_psql('postgres', + "SELECT c, count(*), min(a), max(a) FROM tab1 GROUP BY 1"); +is($result, qq(sub2_tab1|3|3|6), 'delete from tab1_1 replicated'); + # delete rows (some from the root parent, some directly from the partition) $node_publisher->safe_psql('postgres', @@ -130,12 +160,18 @@ $result = $node_subscriber2->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab1_2"); is($result, qq(0||), 'delete from tab1_2 replicated'); +$result = $node_subscriber2->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1"); +is($result, qq(0||), 'delete from tab1_1, tab_2 replicated'); + # truncate (root parent and partition directly) $node_subscriber1->safe_psql('postgres', "INSERT INTO tab1 VALUES (1), (2), (5)"); $node_subscriber2->safe_psql('postgres', "INSERT INTO tab1_2 VALUES (5)"); +$node_subscriber2->safe_psql('postgres', + "INSERT INTO tab1 VALUES (1), (2), (5)"); $node_publisher->safe_psql('postgres', "TRUNCATE tab1_2"); @@ -151,6 +187,10 @@ $result = $node_subscriber2->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab1_2"); is($result, qq(0||), 'truncate of tab1_2 replicated'); +$result = $node_subscriber2->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1"); +is($result, qq(3|1|5), 'no change, because only truncate of tab1 will be replicated'); + $node_publisher->safe_psql('postgres', "TRUNCATE tab1"); @@ -159,3 +199,7 @@ $node_publisher->wait_for_catchup('sub1'); $result = $node_subscriber1->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab1"); is($result, qq(0||), 'truncate of tab1_1 replicated'); + +$result = $node_subscriber2->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab1"); +is($result, qq(0||), 'truncate of tab1_1 replicated'); -- 2.16.5