diff options
Diffstat (limited to 'src/backend/commands')
| -rw-r--r-- | src/backend/commands/Makefile | 8 | ||||
| -rw-r--r-- | src/backend/commands/alter.c | 10 | ||||
| -rw-r--r-- | src/backend/commands/dbcommands.c | 17 | ||||
| -rw-r--r-- | src/backend/commands/define.c | 28 | ||||
| -rw-r--r-- | src/backend/commands/dropcmds.c | 4 | ||||
| -rw-r--r-- | src/backend/commands/event_trigger.c | 8 | ||||
| -rw-r--r-- | src/backend/commands/publicationcmds.c | 754 | ||||
| -rw-r--r-- | src/backend/commands/subscriptioncmds.c | 643 | ||||
| -rw-r--r-- | src/backend/commands/tablecmds.c | 12 |
9 files changed, 1480 insertions, 4 deletions
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile index 6b3742c0a08..e0fab38cbe1 100644 --- a/src/backend/commands/Makefile +++ b/src/backend/commands/Makefile @@ -17,9 +17,9 @@ OBJS = amcmds.o aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o \ dbcommands.o define.o discard.o dropcmds.o \ event_trigger.o explain.o extension.o foreigncmds.o functioncmds.o \ indexcmds.o lockcmds.o matview.o operatorcmds.o opclasscmds.o \ - policy.o portalcmds.o prepare.o proclang.o \ - schemacmds.o seclabel.o sequence.o tablecmds.o tablespace.o trigger.o \ - tsearchcmds.o typecmds.o user.o vacuum.o vacuumlazy.o \ - variable.o view.o + policy.o portalcmds.o prepare.o proclang.o publicationcmds.o \ + schemacmds.o seclabel.o sequence.o subscriptioncmds.o tablecmds.o \ + tablespace.o trigger.o tsearchcmds.o typecmds.o user.o vacuum.o \ + vacuumlazy.o variable.o view.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c index 8b6f4209096..768fcc82ddc 100644 --- a/src/backend/commands/alter.c +++ b/src/backend/commands/alter.c @@ -45,7 +45,9 @@ #include "commands/extension.h" #include "commands/policy.h" #include "commands/proclang.h" +#include "commands/publicationcmds.h" #include "commands/schemacmds.h" +#include "commands/subscriptioncmds.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" #include "commands/trigger.h" @@ -770,6 +772,14 @@ ExecAlterOwnerStmt(AlterOwnerStmt *stmt) return AlterEventTriggerOwner(strVal(linitial(stmt->object)), newowner); + case OBJECT_PUBLICATION: + return AlterPublicationOwner(strVal(linitial(stmt->object)), + newowner); + + case OBJECT_SUBSCRIPTION: + return AlterSubscriptionOwner(strVal(linitial(stmt->object)), + newowner); + /* Generic cases */ case OBJECT_AGGREGATE: case OBJECT_COLLATION: diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c index 2833f3e8469..6ad8fd77b10 100644 --- a/src/backend/commands/dbcommands.c +++ b/src/backend/commands/dbcommands.c @@ -37,6 +37,7 @@ #include "catalog/pg_authid.h" #include "catalog/pg_database.h" #include "catalog/pg_db_role_setting.h" +#include "catalog/pg_subscription.h" #include "catalog/pg_tablespace.h" #include "commands/comment.h" #include "commands/dbcommands.h" @@ -790,6 +791,7 @@ dropdb(const char *dbname, bool missing_ok) int npreparedxacts; int nslots, nslots_active; + int nsubscriptions; /* * Look up the target database's OID, and get exclusive lock on it. We @@ -875,6 +877,21 @@ dropdb(const char *dbname, bool missing_ok) errdetail_busy_db(notherbackends, npreparedxacts))); /* + * Check if there are subscriptions defined in the target database. + * + * We can't drop them automatically because they might be holding + * resources in other databases/instances. + */ + if ((nsubscriptions = CountDBSubscriptions(db_id)) > 0) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_IN_USE), + errmsg("database \"%s\" is being used by logical replication subscription", + dbname), + errdetail_plural("There is %d subscription.", + "There are %d subscriptions.", + nsubscriptions, nsubscriptions))); + + /* * Remove the database's tuple from pg_database. */ tup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(db_id)); diff --git a/src/backend/commands/define.c b/src/backend/commands/define.c index 714b5252c73..8da924517b9 100644 --- a/src/backend/commands/define.c +++ b/src/backend/commands/define.c @@ -319,3 +319,31 @@ defGetTypeLength(DefElem *def) def->defname, defGetString(def)))); return 0; /* keep compiler quiet */ } + +/* + * Extract a list of string values (otherwise uninterpreted) from a DefElem. + */ +List * +defGetStringList(DefElem *def) +{ + ListCell *cell; + + if (def->arg == NULL) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("%s requires a parameter", + def->defname))); + if (nodeTag(def->arg) != T_List) + elog(ERROR, "unrecognized node type: %d", (int) nodeTag(def->arg)); + + foreach(cell, (List *)def->arg) + { + Node *str = (Node *) lfirst(cell); + + if (!IsA(str, String)) + elog(ERROR, "unexpected node type in name list: %d", + (int) nodeTag(str)); + } + + return (List *) def->arg; +} diff --git a/src/backend/commands/dropcmds.c b/src/backend/commands/dropcmds.c index 96436c06897..8cfbcf43f79 100644 --- a/src/backend/commands/dropcmds.c +++ b/src/backend/commands/dropcmds.c @@ -441,6 +441,10 @@ does_not_exist_skipping(ObjectType objtype, List *objname, List *objargs) } } break; + case OBJECT_PUBLICATION: + msg = gettext_noop("publication \"%s\" does not exist, skipping"); + name = NameListToString(objname); + break; default: elog(ERROR, "unrecognized object type: %d", (int) objtype); break; diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c index c0061e195eb..81255373615 100644 --- a/src/backend/commands/event_trigger.c +++ b/src/backend/commands/event_trigger.c @@ -106,11 +106,13 @@ static event_trigger_support_data event_trigger_support[] = { {"OPERATOR CLASS", true}, {"OPERATOR FAMILY", true}, {"POLICY", true}, + {"PUBLICATION", true}, {"ROLE", false}, {"RULE", true}, {"SCHEMA", true}, {"SEQUENCE", true}, {"SERVER", true}, + {"SUBSCRIPTION", true}, {"TABLE", true}, {"TABLESPACE", false}, {"TRANSFORM", true}, @@ -1103,9 +1105,12 @@ EventTriggerSupportsObjectType(ObjectType obtype) case OBJECT_OPERATOR: case OBJECT_OPFAMILY: case OBJECT_POLICY: + case OBJECT_PUBLICATION: + case OBJECT_PUBLICATION_REL: case OBJECT_RULE: case OBJECT_SCHEMA: case OBJECT_SEQUENCE: + case OBJECT_SUBSCRIPTION: case OBJECT_TABCONSTRAINT: case OBJECT_TABLE: case OBJECT_TRANSFORM: @@ -1168,6 +1173,9 @@ EventTriggerSupportsObjectClass(ObjectClass objclass) case OCLASS_EXTENSION: case OCLASS_POLICY: case OCLASS_AM: + case OCLASS_PUBLICATION: + case OCLASS_PUBLICATION_REL: + case OCLASS_SUBSCRIPTION: return true; } diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c new file mode 100644 index 00000000000..21e523deb08 --- /dev/null +++ b/src/backend/commands/publicationcmds.c @@ -0,0 +1,754 @@ +/*------------------------------------------------------------------------- + * + * publicationcmds.c + * publication manipulation + * + * Copyright (c) 2016, PostgreSQL Global Development Group + * + * IDENTIFICATION + * publicationcmds.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "funcapi.h" +#include "miscadmin.h" + +#include "access/genam.h" +#include "access/hash.h" +#include "access/heapam.h" +#include "access/htup_details.h" +#include "access/xact.h" + +#include "catalog/catalog.h" +#include "catalog/indexing.h" +#include "catalog/namespace.h" +#include "catalog/objectaccess.h" +#include "catalog/objectaddress.h" +#include "catalog/pg_inherits_fn.h" +#include "catalog/pg_type.h" +#include "catalog/pg_publication.h" +#include "catalog/pg_publication_rel.h" + +#include "commands/dbcommands.h" +#include "commands/defrem.h" +#include "commands/event_trigger.h" +#include "commands/publicationcmds.h" + +#include "utils/array.h" +#include "utils/builtins.h" +#include "utils/catcache.h" +#include "utils/fmgroids.h" +#include "utils/inval.h" +#include "utils/lsyscache.h" +#include "utils/rel.h" +#include "utils/syscache.h" + +/* Same as MAXNUMMESSAGES in sinvaladt.c */ +#define MAX_RELCACHE_INVAL_MSGS 4096 + +static List *OpenTableList(List *tables); +static void CloseTableList(List *rels); +static void PublicationAddTables(Oid pubid, List *rels, bool if_not_exists, + AlterPublicationStmt *stmt); +static void PublicationDropTables(Oid pubid, List *rels, bool missing_ok); + +static void +parse_publication_options(List *options, + bool *publish_insert_given, + bool *publish_insert, + bool *publish_update_given, + bool *publish_update, + bool *publish_delete_given, + bool *publish_delete) +{ + ListCell *lc; + + *publish_insert_given = false; + *publish_update_given = false; + *publish_delete_given = false; + + /* Defaults are true */ + *publish_insert = true; + *publish_update = true; + *publish_delete = true; + + /* Parse options */ + foreach (lc, options) + { + DefElem *defel = (DefElem *) lfirst(lc); + + if (strcmp(defel->defname, "publish insert") == 0) + { + if (*publish_insert_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *publish_insert_given = true; + *publish_insert = defGetBoolean(defel); + } + else if (strcmp(defel->defname, "nopublish insert") == 0) + { + if (*publish_insert_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *publish_insert_given = true; + *publish_insert = !defGetBoolean(defel); + } + else if (strcmp(defel->defname, "publish update") == 0) + { + if (*publish_update_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *publish_update_given = true; + *publish_update = defGetBoolean(defel); + } + else if (strcmp(defel->defname, "nopublish update") == 0) + { + if (*publish_update_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *publish_update_given = true; + *publish_update = !defGetBoolean(defel); + } + else if (strcmp(defel->defname, "publish delete") == 0) + { + if (*publish_delete_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *publish_delete_given = true; + *publish_delete = defGetBoolean(defel); + } + else if (strcmp(defel->defname, "nopublish delete") == 0) + { + if (*publish_delete_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *publish_delete_given = true; + *publish_delete = !defGetBoolean(defel); + } + else + elog(ERROR, "unrecognized option: %s", defel->defname); + } +} + +/* + * Create new publication. + */ +ObjectAddress +CreatePublication(CreatePublicationStmt *stmt) +{ + Relation rel; + ObjectAddress myself; + Oid puboid; + bool nulls[Natts_pg_publication]; + Datum values[Natts_pg_publication]; + HeapTuple tup; + bool publish_insert_given; + bool publish_update_given; + bool publish_delete_given; + bool publish_insert; + bool publish_update; + bool publish_delete; + AclResult aclresult; + + /* must have CREATE privilege on database */ + aclresult = pg_database_aclcheck(MyDatabaseId, GetUserId(), ACL_CREATE); + if (aclresult != ACLCHECK_OK) + aclcheck_error(aclresult, ACL_KIND_DATABASE, + get_database_name(MyDatabaseId)); + + /* FOR ALL TABLES requires superuser */ + if (stmt->for_all_tables && !superuser()) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + (errmsg("must be superuser to create FOR ALL TABLES publication")))); + + rel = heap_open(PublicationRelationId, RowExclusiveLock); + + /* Check if name is used */ + puboid = GetSysCacheOid1(PUBLICATIONNAME, CStringGetDatum(stmt->pubname)); + if (OidIsValid(puboid)) + { + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("publication \"%s\" already exists", + stmt->pubname))); + } + + /* Form a tuple. */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + + values[Anum_pg_publication_pubname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(stmt->pubname)); + values[Anum_pg_publication_pubowner - 1] = ObjectIdGetDatum(GetUserId()); + + parse_publication_options(stmt->options, + &publish_insert_given, &publish_insert, + &publish_update_given, &publish_update, + &publish_delete_given, &publish_delete); + + values[Anum_pg_publication_puballtables - 1] = + BoolGetDatum(stmt->for_all_tables); + values[Anum_pg_publication_pubinsert - 1] = + BoolGetDatum(publish_insert); + values[Anum_pg_publication_pubupdate - 1] = + BoolGetDatum(publish_update); + values[Anum_pg_publication_pubdelete - 1] = + BoolGetDatum(publish_delete); + + tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); + + /* Insert tuple into catalog. */ + puboid = simple_heap_insert(rel, tup); + CatalogUpdateIndexes(rel, tup); + heap_freetuple(tup); + + ObjectAddressSet(myself, PublicationRelationId, puboid); + + /* Make the changes visible. */ + CommandCounterIncrement(); + + if (stmt->tables) + { + List *rels; + + Assert(list_length(stmt->tables) > 0); + + rels = OpenTableList(stmt->tables); + PublicationAddTables(puboid, rels, true, NULL); + CloseTableList(rels); + } + + heap_close(rel, RowExclusiveLock); + + InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0); + + return myself; +} + +/* + * Change options of a publication. + */ +static void +AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel, + HeapTuple tup) +{ + bool nulls[Natts_pg_publication]; + bool replaces[Natts_pg_publication]; + Datum values[Natts_pg_publication]; + bool publish_insert_given; + bool publish_update_given; + bool publish_delete_given; + bool publish_insert; + bool publish_update; + bool publish_delete; + ObjectAddress obj; + + parse_publication_options(stmt->options, + &publish_insert_given, &publish_insert, + &publish_update_given, &publish_update, + &publish_delete_given, &publish_delete); + + /* Everything ok, form a new tuple. */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + memset(replaces, false, sizeof(replaces)); + + if (publish_insert_given) + { + values[Anum_pg_publication_pubinsert - 1] = + BoolGetDatum(publish_insert); + replaces[Anum_pg_publication_pubinsert - 1] = true; + } + if (publish_update_given) + { + values[Anum_pg_publication_pubupdate - 1] = + BoolGetDatum(publish_update); + replaces[Anum_pg_publication_pubupdate - 1] = true; + } + if (publish_delete_given) + { + values[Anum_pg_publication_pubdelete - 1] = + BoolGetDatum(publish_delete); + replaces[Anum_pg_publication_pubdelete - 1] = true; + } + + tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls, + replaces); + + /* Update the catalog. */ + simple_heap_update(rel, &tup->t_self, tup); + CatalogUpdateIndexes(rel, tup); + + CommandCounterIncrement(); + + /* Invalidate the relcache. */ + if (((Form_pg_publication) GETSTRUCT(tup))->puballtables) + { + CacheInvalidateRelcacheAll(); + } + else + { + List *relids = GetPublicationRelations(HeapTupleGetOid(tup)); + + /* + * We don't want to send too many individual messages, at some point + * it's cheaper to just reset whole relcache. + */ + if (list_length(relids) < MAX_RELCACHE_INVAL_MSGS) + { + ListCell *lc; + + foreach (lc, relids) + { + Oid relid = lfirst_oid(lc); + + CacheInvalidateRelcacheByRelid(relid); + } + } + else + CacheInvalidateRelcacheAll(); + } + + ObjectAddressSet(obj, PublicationRelationId, HeapTupleGetOid(tup)); + EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress, + (Node *) stmt); + + InvokeObjectPostAlterHook(PublicationRelationId, HeapTupleGetOid(tup), 0); +} + +/* + * Add or remove table to/from publication. + */ +static void +AlterPublicationTables(AlterPublicationStmt *stmt, Relation rel, + HeapTuple tup) +{ + Oid pubid = HeapTupleGetOid(tup); + List *rels = NIL; + Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup); + + /* Check that user is allowed to manipulate the publication tables. */ + if (pubform->puballtables) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("publication \"%s\" is defined as FOR ALL TABLES", + NameStr(pubform->pubname)), + errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications."))); + + Assert(list_length(stmt->tables) > 0); + + rels = OpenTableList(stmt->tables); + + if (stmt->tableAction == DEFELEM_ADD) + PublicationAddTables(pubid, rels, false, stmt); + else if (stmt->tableAction == DEFELEM_DROP) + PublicationDropTables(pubid, rels, false); + else /* DEFELEM_SET */ + { + List *oldrelids = GetPublicationRelations(pubid); + List *delrels = NIL; + ListCell *oldlc; + + /* Calculate which relations to drop. */ + foreach(oldlc, oldrelids) + { + Oid oldrelid = lfirst_oid(oldlc); + ListCell *newlc; + bool found = false; + + foreach(newlc, rels) + { + Relation newrel = (Relation) lfirst(newlc); + + if (RelationGetRelid(newrel) == oldrelid) + { + found = true; + break; + } + } + + if (!found) + { + Relation oldrel = heap_open(oldrelid, + ShareUpdateExclusiveLock); + delrels = lappend(delrels, oldrel); + } + } + + /* And drop them. */ + PublicationDropTables(pubid, delrels, true); + + /* + * Don't bother calculating the difference for adding, we'll catch + * and skip existing ones when doing catalog update. + */ + PublicationAddTables(pubid, rels, true, stmt); + + CloseTableList(delrels); + } + + CloseTableList(rels); +} + +/* + * Alter the existing publication. + * + * This is dispatcher function for AlterPublicationOptions and + * AlterPublicationTables. + */ +void +AlterPublication(AlterPublicationStmt *stmt) +{ + Relation rel; + HeapTuple tup; + + rel = heap_open(PublicationRelationId, RowExclusiveLock); + + tup = SearchSysCacheCopy1(PUBLICATIONNAME, + CStringGetDatum(stmt->pubname)); + + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("publication \"%s\" does not exist", + stmt->pubname))); + + /* must be owner */ + if (!pg_publication_ownercheck(HeapTupleGetOid(tup), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_PUBLICATION, + stmt->pubname); + + if (stmt->options) + AlterPublicationOptions(stmt, rel, tup); + else + AlterPublicationTables(stmt, rel, tup); + + /* Cleanup. */ + heap_freetuple(tup); + heap_close(rel, RowExclusiveLock); +} + +/* + * Drop publication by OID + */ +void +RemovePublicationById(Oid pubid) +{ + Relation rel; + HeapTuple tup; + + rel = heap_open(PublicationRelationId, RowExclusiveLock); + + tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid)); + + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for publication %u", pubid); + + simple_heap_delete(rel, &tup->t_self); + + ReleaseSysCache(tup); + + heap_close(rel, RowExclusiveLock); +} + +/* + * Remove relation from publication by mapping OID. + */ +void +RemovePublicationRelById(Oid proid) +{ + Relation rel; + HeapTuple tup; + Form_pg_publication_rel pubrel; + + rel = heap_open(PublicationRelRelationId, RowExclusiveLock); + + tup = SearchSysCache1(PUBLICATIONREL, ObjectIdGetDatum(proid)); + + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for publication table %u", + proid); + + + pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + /* Invalidate relcache so that publication info is rebuilt. */ + CacheInvalidateRelcacheByRelid(pubrel->prrelid); + + simple_heap_delete(rel, &tup->t_self); + + ReleaseSysCache(tup); + + heap_close(rel, RowExclusiveLock); +} + +/* + * Open relations based om provided by RangeVar list. + * The returned tables are locked in ShareUpdateExclusiveLock mode. + */ +static List * +OpenTableList(List *tables) +{ + List *relids = NIL; + List *rels = NIL; + ListCell *lc; + + /* + * Open, share-lock, and check all the explicitly-specified relations + */ + foreach(lc, tables) + { + RangeVar *rv = lfirst(lc); + Relation rel; + bool recurse = rv->inh; + Oid myrelid; + + CHECK_FOR_INTERRUPTS(); + + rel = heap_openrv(rv, ShareUpdateExclusiveLock); + myrelid = RelationGetRelid(rel); + /* + * filter out duplicates when user specifies "foo, foo" + * Note that this algrithm is know to not be very effective (O(N^2)) + * but given that it only works on list of tables given to us by user + * it's deemed acceptable. + */ + if (list_member_oid(relids, myrelid)) + { + heap_close(rel, ShareUpdateExclusiveLock); + continue; + } + rels = lappend(rels, rel); + relids = lappend_oid(relids, myrelid); + + if (recurse) + { + ListCell *child; + List *children; + + children = find_all_inheritors(myrelid, ShareUpdateExclusiveLock, + NULL); + + foreach(child, children) + { + Oid childrelid = lfirst_oid(child); + + if (list_member_oid(relids, childrelid)) + continue; + + /* + * Skip duplicates if user specified both parent and child + * tables. + */ + if (list_member_oid(relids, childrelid)) + { + heap_close(rel, ShareUpdateExclusiveLock); + continue; + } + + /* find_all_inheritors already got lock */ + rel = heap_open(childrelid, NoLock); + rels = lappend(rels, rel); + relids = lappend_oid(relids, childrelid); + } + } + } + + list_free(relids); + + return rels; +} + +/* + * Close all relations in the list. + */ +static void +CloseTableList(List *rels) +{ + ListCell *lc; + + foreach(lc, rels) + { + Relation rel = (Relation) lfirst(lc); + + heap_close(rel, NoLock); + } +} + +/* + * Add listed tables to the publication. + */ +static void +PublicationAddTables(Oid pubid, List *rels, bool if_not_exists, + AlterPublicationStmt *stmt) +{ + ListCell *lc; + + Assert(!stmt || !stmt->for_all_tables); + + foreach(lc, rels) + { + Relation rel = (Relation) lfirst(lc); + ObjectAddress obj; + + /* Must be owner of the table or superuser. */ + if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS, + RelationGetRelationName(rel)); + + obj = publication_add_relation(pubid, rel, if_not_exists); + if (stmt) + { + EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress, + (Node *) stmt); + + InvokeObjectPostCreateHook(PublicationRelRelationId, + obj.objectId, 0); + } + } +} + +/* + * Remove listed tables from the publication. + */ +static void +PublicationDropTables(Oid pubid, List *rels, bool missing_ok) +{ + ObjectAddress obj; + ListCell *lc; + Oid prid; + + foreach(lc, rels) + { + Relation rel = (Relation) lfirst(lc); + Oid relid = RelationGetRelid(rel); + + prid = GetSysCacheOid2(PUBLICATIONRELMAP, ObjectIdGetDatum(relid), + ObjectIdGetDatum(pubid)); + if (!OidIsValid(prid)) + { + if (missing_ok) + continue; + + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("relation \"%s\" is not part of the publication", + RelationGetRelationName(rel)))); + } + + ObjectAddressSet(obj, PublicationRelRelationId, prid); + performDeletion(&obj, DROP_CASCADE, 0); + } +} + +/* + * Internal workhorse for changing a publication owner + */ + static void +AlterPublicationOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId) +{ + Form_pg_publication form; + + form = (Form_pg_publication) GETSTRUCT(tup); + + if (form->pubowner == newOwnerId) + return; + + if (!pg_publication_ownercheck(HeapTupleGetOid(tup), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_PUBLICATION, + NameStr(form->pubname)); + + /* New owner must be a superuser */ + if (!superuser_arg(newOwnerId)) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("permission denied to change owner of publication \"%s\"", + NameStr(form->pubname)), + errhint("The owner of a publication must be a superuser."))); + + form->pubowner = newOwnerId; + simple_heap_update(rel, &tup->t_self, tup); + CatalogUpdateIndexes(rel, tup); + + /* Update owner dependency reference */ + changeDependencyOnOwner(PublicationRelationId, + HeapTupleGetOid(tup), + newOwnerId); + + InvokeObjectPostAlterHook(PublicationRelationId, + HeapTupleGetOid(tup), 0); +} + +/* + * Change publication owner -- by name + */ +ObjectAddress +AlterPublicationOwner(const char *name, Oid newOwnerId) +{ + Oid subid; + HeapTuple tup; + Relation rel; + ObjectAddress address; + + rel = heap_open(PublicationRelationId, RowExclusiveLock); + + tup = SearchSysCacheCopy1(PUBLICATIONNAME, CStringGetDatum(name)); + + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("publication \"%s\" does not exist", name))); + + subid = HeapTupleGetOid(tup); + + AlterPublicationOwner_internal(rel, tup, newOwnerId); + + ObjectAddressSet(address, PublicationRelationId, subid); + + heap_freetuple(tup); + + heap_close(rel, RowExclusiveLock); + + return address; +} + +/* + * Change publication owner -- by OID + */ +void +AlterPublicationOwner_oid(Oid subid, Oid newOwnerId) +{ + HeapTuple tup; + Relation rel; + + rel = heap_open(PublicationRelationId, RowExclusiveLock); + + tup = SearchSysCacheCopy1(PUBLICATIONOID, ObjectIdGetDatum(subid)); + + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("publication with OID %u does not exist", subid))); + + AlterPublicationOwner_internal(rel, tup, newOwnerId); + + heap_freetuple(tup); + + heap_close(rel, RowExclusiveLock); +} diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c new file mode 100644 index 00000000000..1448ee3beea --- /dev/null +++ b/src/backend/commands/subscriptioncmds.c @@ -0,0 +1,643 @@ +/*------------------------------------------------------------------------- + * + * subscriptioncmds.c + * subscription catalog manipulation functions + * + * Copyright (c) 2015, PostgreSQL Global Development Group + * + * IDENTIFICATION + * subscriptioncmds.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "miscadmin.h" + +#include "access/heapam.h" +#include "access/htup_details.h" + +#include "catalog/indexing.h" +#include "catalog/objectaccess.h" +#include "catalog/objectaddress.h" +#include "catalog/pg_type.h" +#include "catalog/pg_subscription.h" + +#include "commands/defrem.h" +#include "commands/event_trigger.h" +#include "commands/subscriptioncmds.h" + +#include "replication/logicallauncher.h" +#include "replication/origin.h" +#include "replication/walreceiver.h" +#include "replication/worker_internal.h" + +#include "storage/lmgr.h" + +#include "utils/builtins.h" +#include "utils/memutils.h" +#include "utils/syscache.h" + +/* + * Common option parsing function for CREATE and ALTER SUBSCRIPTION commands. + * + * Since not all options can be specified in both commands, this function + * will report an error on options if the target output pointer is NULL to + * accomodate that. + */ +static void +parse_subscription_options(List *options, char **conninfo, + List **publications, bool *enabled_given, + bool *enabled, bool *create_slot, char **slot_name) +{ + ListCell *lc; + bool create_slot_given = false; + + if (conninfo) + *conninfo = NULL; + if (publications) + *publications = NIL; + if (enabled) + { + *enabled_given = false; + *enabled = true; + } + if (create_slot) + *create_slot = true; + if (slot_name) + *slot_name = NULL; + + /* Parse options */ + foreach (lc, options) + { + DefElem *defel = (DefElem *) lfirst(lc); + + if (strcmp(defel->defname, "conninfo") == 0 && conninfo) + { + if (*conninfo) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *conninfo = defGetString(defel); + } + else if (strcmp(defel->defname, "publication") == 0 && publications) + { + if (*publications) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *publications = defGetStringList(defel); + } + else if (strcmp(defel->defname, "enabled") == 0 && enabled) + { + if (*enabled_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *enabled_given = true; + *enabled = defGetBoolean(defel); + } + else if (strcmp(defel->defname, "disabled") == 0 && enabled) + { + if (*enabled_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *enabled_given = true; + *enabled = !defGetBoolean(defel); + } + else if (strcmp(defel->defname, "create slot") == 0 && create_slot) + { + if (create_slot_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + create_slot_given = true; + *create_slot = defGetBoolean(defel); + } + else if (strcmp(defel->defname, "nocreate slot") == 0 && create_slot) + { + if (create_slot_given) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + create_slot_given = true; + *create_slot = !defGetBoolean(defel); + } + else if (strcmp(defel->defname, "slot name") == 0 && slot_name) + { + if (*slot_name) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + + *slot_name = defGetString(defel); + } + else + elog(ERROR, "unrecognized option: %s", defel->defname); + } +} + +/* + * Auxiliary function to return a text array out of a list of String nodes. + */ +static Datum +publicationListToArray(List *publist) +{ + ArrayType *arr; + Datum *datums; + int j = 0; + ListCell *cell; + MemoryContext memcxt; + MemoryContext oldcxt; + + /* Create memory context for temporary allocations. */ + memcxt = AllocSetContextCreate(CurrentMemoryContext, + "publicationListToArray to array", + ALLOCSET_DEFAULT_MINSIZE, + ALLOCSET_DEFAULT_INITSIZE, + ALLOCSET_DEFAULT_MAXSIZE); + oldcxt = MemoryContextSwitchTo(memcxt); + + datums = palloc(sizeof(text *) * list_length(publist)); + foreach(cell, publist) + { + char *name = strVal(lfirst(cell)); + ListCell *pcell; + + /* Check for duplicates. */ + foreach(pcell, publist) + { + char *pname = strVal(lfirst(cell)); + + if (name == pname) + break; + + if (strcmp(name, pname) == 0) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("publication name \"%s\" used more than once", + pname))); + } + + datums[j++] = CStringGetTextDatum(name); + } + + MemoryContextSwitchTo(oldcxt); + + arr = construct_array(datums, list_length(publist), + TEXTOID, -1, false, 'i'); + MemoryContextDelete(memcxt); + + return PointerGetDatum(arr); +} + +/* + * Create new subscription. + */ +ObjectAddress +CreateSubscription(CreateSubscriptionStmt *stmt) +{ + Relation rel; + ObjectAddress myself; + Oid subid; + bool nulls[Natts_pg_subscription]; + Datum values[Natts_pg_subscription]; + HeapTuple tup; + bool enabled_given; + bool enabled; + char *conninfo; + char *slotname; + char originname[NAMEDATALEN]; + bool create_slot; + List *publications; + + if (!superuser()) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + (errmsg("must be superuser to create subscriptions")))); + + rel = heap_open(SubscriptionRelationId, RowExclusiveLock); + + /* Check if name is used */ + subid = GetSysCacheOid2(SUBSCRIPTIONNAME, MyDatabaseId, + CStringGetDatum(stmt->subname)); + if (OidIsValid(subid)) + { + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("subscription \"%s\" already exists", + stmt->subname))); + } + + /* + * Parse and check options. + * Connection and publication should not be specified here. + */ + parse_subscription_options(stmt->options, NULL, NULL, + &enabled_given, &enabled, + &create_slot, &slotname); + if (slotname == NULL) + slotname = stmt->subname; + + conninfo = stmt->conninfo; + publications = stmt->publication; + + /* Load the library providing us libpq calls. */ + load_file("libpqwalreceiver", false); + + /* Check the connection info string. */ + walrcv_check_conninfo(conninfo); + + /* Everything ok, form a new tuple. */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + + values[Anum_pg_subscription_subdbid - 1] = ObjectIdGetDatum(MyDatabaseId); + values[Anum_pg_subscription_subname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(stmt->subname)); + values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(GetUserId()); + values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled); + values[Anum_pg_subscription_subconninfo - 1] = + CStringGetTextDatum(conninfo); + values[Anum_pg_subscription_subslotname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(slotname)); + values[Anum_pg_subscription_subpublications - 1] = + publicationListToArray(publications); + + tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); + + /* Insert tuple into catalog. */ + subid = simple_heap_insert(rel, tup); + CatalogUpdateIndexes(rel, tup); + heap_freetuple(tup); + + snprintf(originname, sizeof(originname), "pg_%u", subid); + replorigin_create(originname); + + /* + * If requested, create the replication slot on remote side for our + * newly created subscription. + */ + if (create_slot) + { + XLogRecPtr lsn; + char *err; + WalReceiverConn *wrconn; + + /* Try to connect to the publisher. */ + wrconn = walrcv_connect(conninfo, true, stmt->subname, &err); + if (!wrconn) + ereport(ERROR, + (errmsg("could not connect to the publisher: %s", err))); + + walrcv_create_slot(wrconn, slotname, false, &lsn); + ereport(NOTICE, + (errmsg("created replication slot \"%s\" on publisher", + slotname))); + + /* And we are done with the remote side. */ + walrcv_disconnect(wrconn); + } + + heap_close(rel, RowExclusiveLock); + + ApplyLauncherWakeupAtCommit(); + + ObjectAddressSet(myself, SubscriptionRelationId, subid); + + InvokeObjectPostCreateHook(SubscriptionRelationId, subid, 0); + + return myself; +} + +/* + * Alter the existing subscription. + */ +ObjectAddress +AlterSubscription(AlterSubscriptionStmt *stmt) +{ + Relation rel; + ObjectAddress myself; + bool nulls[Natts_pg_subscription]; + bool replaces[Natts_pg_subscription]; + Datum values[Natts_pg_subscription]; + HeapTuple tup; + Oid subid; + bool enabled_given; + bool enabled; + char *conninfo; + char *slot_name; + List *publications; + + rel = heap_open(SubscriptionRelationId, RowExclusiveLock); + + /* Fetch the existing tuple. */ + tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId, + CStringGetDatum(stmt->subname)); + + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("subscription \"%s\" does not exist", + stmt->subname))); + + /* must be owner */ + if (!pg_subscription_ownercheck(HeapTupleGetOid(tup), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_SUBSCRIPTION, + stmt->subname); + + subid = HeapTupleGetOid(tup); + + /* Parse options. */ + parse_subscription_options(stmt->options, &conninfo, &publications, + &enabled_given, &enabled, + NULL, &slot_name); + + /* Form a new tuple. */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + memset(replaces, false, sizeof(replaces)); + + if (enabled_given) + { + values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled); + replaces[Anum_pg_subscription_subenabled - 1] = true; + } + if (conninfo) + { + values[Anum_pg_subscription_subconninfo - 1] = + CStringGetTextDatum(conninfo); + replaces[Anum_pg_subscription_subconninfo - 1] = true; + } + if (slot_name) + { + values[Anum_pg_subscription_subslotname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(slot_name)); + replaces[Anum_pg_subscription_subslotname - 1] = true; + } + if (publications != NIL) + { + values[Anum_pg_subscription_subpublications - 1] = + publicationListToArray(publications); + replaces[Anum_pg_subscription_subpublications - 1] = true; + } + + tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls, + replaces); + + /* Update the catalog. */ + simple_heap_update(rel, &tup->t_self, tup); + CatalogUpdateIndexes(rel, tup); + + ObjectAddressSet(myself, SubscriptionRelationId, subid); + + /* Cleanup. */ + heap_freetuple(tup); + heap_close(rel, RowExclusiveLock); + + InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0); + + return myself; +} + +/* + * Drop a subscription + */ +void +DropSubscription(DropSubscriptionStmt *stmt) +{ + Relation rel; + ObjectAddress myself; + HeapTuple tup; + Oid subid; + Datum datum; + bool isnull; + char *subname; + char *conninfo; + char *slotname; + char originname[NAMEDATALEN]; + char *err = NULL; + RepOriginId originid; + WalReceiverConn *wrconn = NULL; + StringInfoData cmd; + + rel = heap_open(SubscriptionRelationId, RowExclusiveLock); + + tup = SearchSysCache2(SUBSCRIPTIONNAME, MyDatabaseId, + CStringGetDatum(stmt->subname)); + + if (!HeapTupleIsValid(tup)) + { + heap_close(rel, NoLock); + + if (!stmt->missing_ok) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("subscription \"%s\" does not exist", + stmt->subname))); + else + ereport(NOTICE, + (errmsg("subscription \"%s\" does not exist, skipping", + stmt->subname))); + + return; + } + + subid = HeapTupleGetOid(tup); + + /* must be owner */ + if (!pg_subscription_ownercheck(subid, GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_SUBSCRIPTION, + stmt->subname); + + /* DROP hook for the subscription being removed */ + InvokeObjectDropHook(SubscriptionRelationId, subid, 0); + + /* + * Lock the subscription so noboby else can do anything with it + * (including the replication workers). + */ + LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock); + + /* Get subname */ + datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, + Anum_pg_subscription_subname, &isnull); + Assert(!isnull); + subname = pstrdup(NameStr(*DatumGetName(datum))); + + /* Get conninfo */ + datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, + Anum_pg_subscription_subconninfo, &isnull); + Assert(!isnull); + conninfo = pstrdup(TextDatumGetCString(datum)); + + /* Get slotname */ + datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, + Anum_pg_subscription_subslotname, &isnull); + Assert(!isnull); + slotname = pstrdup(NameStr(*DatumGetName(datum))); + + ObjectAddressSet(myself, SubscriptionRelationId, subid); + EventTriggerSQLDropAddObject(&myself, true, true); + + /* Remove the tuple from catalog. */ + simple_heap_delete(rel, &tup->t_self); + + ReleaseSysCache(tup); + + /* Protect against launcher restarting the worker. */ + LWLockAcquire(LogicalRepLauncherLock, LW_EXCLUSIVE); + + /* Kill the apply worker so that the slot becomes accessible. */ + logicalrep_worker_stop(subid); + + /* Remove the origin tracking if exists. */ + snprintf(originname, sizeof(originname), "pg_%u", subid); + originid = replorigin_by_name(originname, true); + if (originid != InvalidRepOriginId) + replorigin_drop(originid); + + /* If the user asked to not drop the slot, we are done mow.*/ + if (!stmt->drop_slot) + { + heap_close(rel, NoLock); + return; + } + + /* + * Otherwise drop the replication slot at the publisher node using + * the replication connection. + */ + load_file("libpqwalreceiver", false); + + initStringInfo(&cmd); + appendStringInfo(&cmd, "DROP_REPLICATION_SLOT \"%s\"", slotname); + + wrconn = walrcv_connect(conninfo, true, subname, &err); + if (wrconn == NULL) + ereport(ERROR, + (errmsg("could not connect to publisher when attempting to " + "drop the replication slot \"%s\"", slotname), + errdetail("The error was: %s", err))); + + if (!walrcv_command(wrconn, cmd.data, &err)) + ereport(ERROR, + (errmsg("count not drop the replication slot \"%s\" on publisher", + slotname), + errdetail("The error was: %s", err))); + else + ereport(NOTICE, + (errmsg("dropped replication slot \"%s\" on publisher", + slotname))); + + walrcv_disconnect(wrconn); + + pfree(cmd.data); + + heap_close(rel, NoLock); +} + +/* + * Internal workhorse for changing a subscription owner + */ +static void +AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId) +{ + Form_pg_subscription form; + + form = (Form_pg_subscription) GETSTRUCT(tup); + + if (form->subowner == newOwnerId) + return; + + if (!pg_subscription_ownercheck(HeapTupleGetOid(tup), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_SUBSCRIPTION, + NameStr(form->subname)); + + /* New owner must be a superuser */ + if (!superuser_arg(newOwnerId)) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("permission denied to change owner of subscription \"%s\"", + NameStr(form->subname)), + errhint("The owner of an subscription must be a superuser."))); + + form->subowner = newOwnerId; + simple_heap_update(rel, &tup->t_self, tup); + CatalogUpdateIndexes(rel, tup); + + /* Update owner dependency reference */ + changeDependencyOnOwner(SubscriptionRelationId, + HeapTupleGetOid(tup), + newOwnerId); + + InvokeObjectPostAlterHook(SubscriptionRelationId, + HeapTupleGetOid(tup), 0); +} + +/* + * Change subscription owner -- by name + */ +ObjectAddress +AlterSubscriptionOwner(const char *name, Oid newOwnerId) +{ + Oid subid; + HeapTuple tup; + Relation rel; + ObjectAddress address; + + rel = heap_open(SubscriptionRelationId, RowExclusiveLock); + + tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId, + CStringGetDatum(name)); + + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("subscription \"%s\" does not exist", name))); + + subid = HeapTupleGetOid(tup); + + AlterSubscriptionOwner_internal(rel, tup, newOwnerId); + + ObjectAddressSet(address, SubscriptionRelationId, subid); + + heap_freetuple(tup); + + heap_close(rel, RowExclusiveLock); + + return address; +} + +/* + * Change subscription owner -- by OID + */ +void +AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId) +{ + HeapTuple tup; + Relation rel; + + rel = heap_open(SubscriptionRelationId, RowExclusiveLock); + + tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid)); + + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("subscription with OID %u does not exist", subid))); + + AlterSubscriptionOwner_internal(rel, tup, newOwnerId); + + heap_freetuple(tup); + + heap_close(rel, RowExclusiveLock); +} diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 6ed2a3dc4d1..c4b0011bdd6 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -12056,6 +12056,18 @@ ATPrepChangePersistence(Relation rel, bool toLogged) } /* + * Check that the table is not part any publication when changing to + * UNLOGGED as UNLOGGED tables can't be published. + */ + if (!toLogged && + list_length(GetRelationPublications(RelationGetRelid(rel))) > 0) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot change table \"%s\" to unlogged because it is part of a publication", + RelationGetRelationName(rel)), + errdetail("Unlogged relations cannot be replicated."))); + + /* * Check existing foreign key constraints to preserve the invariant that * permanent tables cannot reference unlogged ones. Self-referencing * foreign keys can safely be ignored. |