postgresql/src/backend/commands/subscriptioncmds.c

1175 lines
31 KiB
C
Raw Normal View History

/*-------------------------------------------------------------------------
*
* subscriptioncmds.c
* subscription catalog manipulation functions
*
* Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* subscriptioncmds.c
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "miscadmin.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/xact.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/objectaddress.h"
#include "catalog/pg_type.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/subscriptioncmds.h"
#include "executor/executor.h"
#include "nodes/makefuncs.h"
#include "replication/logicallauncher.h"
#include "replication/origin.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "replication/worker_internal.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/syscache.h"
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
/*
* 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
* accommodate that.
*/
static void
parse_subscription_options(List *options, bool *connect, bool *enabled_given,
bool *enabled, bool *create_slot,
bool *slot_name_given, char **slot_name,
bool *copy_data, char **synchronous_commit,
bool *refresh)
{
ListCell *lc;
bool connect_given = false;
bool create_slot_given = false;
bool copy_data_given = false;
bool refresh_given = false;
2017-05-15 19:59:58 +02:00
/* If connect is specified, the others also need to be. */
Assert(!connect || (enabled && create_slot && copy_data));
if (connect)
*connect = true;
if (enabled)
{
*enabled_given = false;
*enabled = true;
}
if (create_slot)
*create_slot = true;
if (slot_name)
{
*slot_name_given = false;
*slot_name = NULL;
}
if (copy_data)
*copy_data = true;
if (synchronous_commit)
*synchronous_commit = NULL;
if (refresh)
*refresh = true;
/* Parse options */
foreach(lc, options)
{
DefElem *defel = (DefElem *) lfirst(lc);
if (strcmp(defel->defname, "connect") == 0 && connect)
{
if (connect_given)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting or redundant options")));
connect_given = true;
*connect = defGetBoolean(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, "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, "slot_name") == 0 && slot_name)
{
if (*slot_name_given)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting or redundant options")));
*slot_name_given = true;
*slot_name = defGetString(defel);
/* Setting slot_name = NONE is treated as no slot name. */
if (strcmp(*slot_name, "none") == 0)
*slot_name = NULL;
}
else if (strcmp(defel->defname, "copy_data") == 0 && copy_data)
{
if (copy_data_given)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting or redundant options")));
copy_data_given = true;
*copy_data = defGetBoolean(defel);
}
else if (strcmp(defel->defname, "synchronous_commit") == 0 &&
synchronous_commit)
{
if (*synchronous_commit)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting or redundant options")));
*synchronous_commit = defGetString(defel);
/* Test if the given value is valid for synchronous_commit GUC. */
(void) set_config_option("synchronous_commit", *synchronous_commit,
PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
false, 0, false);
}
else if (strcmp(defel->defname, "refresh") == 0 && refresh)
{
if (refresh_given)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("conflicting or redundant options")));
refresh_given = true;
*refresh = defGetBoolean(defel);
}
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("unrecognized subscription parameter: %s", defel->defname)));
}
/*
* We've been explicitly asked to not connect, that requires some
* additional processing.
*/
if (connect && !*connect)
{
/* Check for incompatible options from the user. */
if (enabled && *enabled_given && *enabled)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("connect = false and enabled = true are mutually exclusive options")));
if (create_slot && create_slot_given && *create_slot)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("connect = false and create_slot = true are mutually exclusive options")));
if (copy_data && copy_data_given && *copy_data)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("connect = false and copy_data = true are mutually exclusive options")));
/* Change the defaults of other options. */
*enabled = false;
*create_slot = false;
*copy_data = false;
}
/*
* Do additional checking for disallowed combination when slot_name = NONE
* was used.
*/
if (slot_name && *slot_name_given && !*slot_name)
{
if (enabled && *enabled_given && *enabled)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("slot_name = NONE and enabled = true are mutually exclusive options")));
if (create_slot && create_slot_given && *create_slot)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("slot_name = NONE and create_slot = true are mutually exclusive options")));
if (enabled && !*enabled_given && *enabled)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("subscription with slot_name = NONE must also set enabled = false")));
if (create_slot && !create_slot_given && *create_slot)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("subscription with slot_name = NONE must also set create_slot = false")));
}
}
/*
* Auxiliary function to build 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",
Rethink MemoryContext creation to improve performance. This patch makes a number of interrelated changes to reduce the overhead involved in creating/deleting memory contexts. The key ideas are: * Include the AllocSetContext header of an aset.c context in its first malloc request, rather than allocating it separately in TopMemoryContext. This means that we now always create an initial or "keeper" block in an aset, even if it never receives any allocation requests. * Create freelists in which we can save and recycle recently-destroyed asets (this idea is due to Robert Haas). * In the common case where the name of a context is a constant string, just store a pointer to it in the context header, rather than copying the string. The first change eliminates a palloc/pfree cycle per context, and also avoids bloat in TopMemoryContext, at the price that creating a context now involves a malloc/free cycle even if the context never receives any allocations. That would be a loser for some common usage patterns, but recycling short-lived contexts via the freelist eliminates that pain. Avoiding copying constant strings not only saves strlen() and strcpy() overhead, but is an essential part of the freelist optimization because it makes the context header size constant. Currently we make no attempt to use the freelist for contexts with non-constant names. (Perhaps someday we'll need to think harder about that, but in current usage, most contexts with custom names are long-lived anyway.) The freelist management in this initial commit is pretty simplistic, and we might want to refine it later --- but in common workloads that will never matter because the freelists will never get full anyway. To create a context with a non-constant name, one is now required to call AllocSetContextCreateExtended and specify the MEMCONTEXT_COPY_NAME option. AllocSetContextCreate becomes a wrapper macro, and it includes a test that will complain about non-string-literal context name parameters on gcc and similar compilers. An unfortunate side effect of making AllocSetContextCreate a macro is that one is now *required* to use the size parameter abstraction macros (ALLOCSET_DEFAULT_SIZES and friends) with it; the pre-9.6 habit of writing out individual size parameters no longer works unless you switch to AllocSetContextCreateExtended. Internally to the memory-context-related modules, the context creation APIs are simplified, removing the rather baroque original design whereby a context-type module called mcxt.c which then called back into the context-type module. That saved a bit of code duplication, but not much, and it prevented context-type modules from exercising control over the allocation of context headers. In passing, I converted the test-and-elog validation of aset size parameters into Asserts to save a few more cycles. The original thought was that callers might compute size parameters on the fly, but in practice nobody does that, so it's useless to expend cycles on checking those numbers in production builds. Also, mark the memory context method-pointer structs "const", just for cleanliness. Discussion: https://postgr.es/m/2264.1512870796@sss.pgh.pa.us
2017-12-13 19:55:12 +01:00
ALLOCSET_DEFAULT_SIZES);
oldcxt = MemoryContextSwitchTo(memcxt);
datums = (Datum *) palloc(sizeof(Datum) * list_length(publist));
foreach(cell, publist)
{
char *name = strVal(lfirst(cell));
ListCell *pcell;
/* Check for duplicates. */
foreach(pcell, publist)
{
char *pname = strVal(lfirst(pcell));
if (pcell == cell)
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, bool isTopLevel)
{
Relation rel;
ObjectAddress myself;
Oid subid;
bool nulls[Natts_pg_subscription];
Datum values[Natts_pg_subscription];
Oid owner = GetUserId();
HeapTuple tup;
bool connect;
bool enabled_given;
bool enabled;
bool copy_data;
char *synchronous_commit;
char *conninfo;
char *slotname;
bool slotname_given;
char originname[NAMEDATALEN];
bool create_slot;
List *publications;
/*
* Parse and check options.
*
* Connection and publication should not be specified here.
*/
parse_subscription_options(stmt->options, &connect, &enabled_given,
&enabled, &create_slot, &slotname_given,
&slotname, &copy_data, &synchronous_commit,
NULL);
/*
* Since creating a replication slot is not transactional, rolling back
* the transaction leaves the created replication slot. So we cannot run
* CREATE SUBSCRIPTION inside a transaction block if creating a
* replication slot.
*/
if (create_slot)
PreventInTransactionBlock(isTopLevel, "CREATE SUBSCRIPTION ... WITH (create_slot = true)");
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)));
}
if (!slotname_given && slotname == NULL)
slotname = stmt->subname;
/* The default for synchronous_commit of subscriptions is off. */
if (synchronous_commit == NULL)
synchronous_commit = "off";
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(owner);
values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (slotname)
values[Anum_pg_subscription_subslotname - 1] =
DirectFunctionCall1(namein, CStringGetDatum(slotname));
else
nulls[Anum_pg_subscription_subslotname - 1] = true;
values[Anum_pg_subscription_subsynccommit - 1] =
CStringGetTextDatum(synchronous_commit);
values[Anum_pg_subscription_subpublications - 1] =
publicationListToArray(publications);
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
/* Insert tuple into catalog. */
subid = CatalogTupleInsert(rel, tup);
heap_freetuple(tup);
recordDependencyOnOwner(SubscriptionRelationId, subid, owner);
snprintf(originname, sizeof(originname), "pg_%u", subid);
replorigin_create(originname);
/*
* Connect to remote side to execute requested commands and fetch table
* info.
*/
if (connect)
{
XLogRecPtr lsn;
char *err;
WalReceiverConn *wrconn;
List *tables;
ListCell *lc;
char table_state;
/* 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)));
PG_TRY();
{
/*
* Set sync state based on if we were asked to do data copy or
* not.
*/
table_state = copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
/*
* Get the table list from publisher and build local table status
* info.
*/
tables = fetch_table_list(wrconn, publications);
foreach(lc, tables)
{
RangeVar *rv = (RangeVar *) lfirst(lc);
Oid relid;
relid = RangeVarGetRelid(rv, AccessShareLock, false);
/* Check for supported relkind. */
CheckSubscriptionRelkind(get_rel_relkind(relid),
rv->schemaname, rv->relname);
AddSubscriptionRelState(subid, relid, table_state,
InvalidXLogRecPtr);
}
/*
* If requested, create permanent slot for the subscription. We
* won't use the initial snapshot for anything, so no need to
* export it.
*/
if (create_slot)
{
Assert(slotname);
walrcv_create_slot(wrconn, slotname, false,
CRS_NOEXPORT_SNAPSHOT, &lsn);
ereport(NOTICE,
(errmsg("created replication slot \"%s\" on publisher",
slotname)));
}
}
PG_CATCH();
{
/* Close the connection in case of failure. */
walrcv_disconnect(wrconn);
PG_RE_THROW();
}
PG_END_TRY();
/* And we are done with the remote side. */
walrcv_disconnect(wrconn);
}
else
ereport(WARNING,
(errmsg("tables were not subscribed, you will have to run "
"ALTER SUBSCRIPTION ... REFRESH PUBLICATION to "
"subscribe the tables")));
heap_close(rel, RowExclusiveLock);
if (enabled)
ApplyLauncherWakeupAtCommit();
ObjectAddressSet(myself, SubscriptionRelationId, subid);
InvokeObjectPostCreateHook(SubscriptionRelationId, subid, 0);
return myself;
}
static void
AlterSubscription_refresh(Subscription *sub, bool copy_data)
{
char *err;
List *pubrel_names;
List *subrel_states;
Oid *subrel_local_oids;
Oid *pubrel_local_oids;
ListCell *lc;
int off;
/* Load the library providing us libpq calls. */
load_file("libpqwalreceiver", false);
/* Try to connect to the publisher. */
wrconn = walrcv_connect(sub->conninfo, true, sub->name, &err);
if (!wrconn)
ereport(ERROR,
(errmsg("could not connect to the publisher: %s", err)));
/* Get the table list from publisher. */
pubrel_names = fetch_table_list(wrconn, sub->publications);
/* We are done with the remote side, close connection. */
walrcv_disconnect(wrconn);
/* Get local table list. */
subrel_states = GetSubscriptionRelations(sub->oid);
/*
* Build qsorted array of local table oids for faster lookup. This can
* potentially contain all tables in the database so speed of lookup is
* important.
*/
subrel_local_oids = palloc(list_length(subrel_states) * sizeof(Oid));
off = 0;
foreach(lc, subrel_states)
{
SubscriptionRelState *relstate = (SubscriptionRelState *) lfirst(lc);
subrel_local_oids[off++] = relstate->relid;
}
qsort(subrel_local_oids, list_length(subrel_states),
sizeof(Oid), oid_cmp);
/*
* Walk over the remote tables and try to match them to locally known
* tables. If the table is not known locally create a new state for it.
*
* Also builds array of local oids of remote tables for the next step.
*/
off = 0;
pubrel_local_oids = palloc(list_length(pubrel_names) * sizeof(Oid));
foreach(lc, pubrel_names)
{
RangeVar *rv = (RangeVar *) lfirst(lc);
Oid relid;
relid = RangeVarGetRelid(rv, AccessShareLock, false);
/* Check for supported relkind. */
CheckSubscriptionRelkind(get_rel_relkind(relid),
rv->schemaname, rv->relname);
pubrel_local_oids[off++] = relid;
if (!bsearch(&relid, subrel_local_oids,
list_length(subrel_states), sizeof(Oid), oid_cmp))
{
AddSubscriptionRelState(sub->oid, relid,
copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
InvalidXLogRecPtr);
ereport(DEBUG1,
(errmsg("table \"%s.%s\" added to subscription \"%s\"",
rv->schemaname, rv->relname, sub->name)));
}
}
/*
* Next remove state for tables we should not care about anymore using the
* data we collected above
*/
qsort(pubrel_local_oids, list_length(pubrel_names),
sizeof(Oid), oid_cmp);
for (off = 0; off < list_length(subrel_states); off++)
{
Oid relid = subrel_local_oids[off];
if (!bsearch(&relid, pubrel_local_oids,
list_length(pubrel_names), sizeof(Oid), oid_cmp))
{
RemoveSubscriptionRel(sub->oid, relid);
logicalrep_worker_stop_at_commit(sub->oid, relid);
ereport(DEBUG1,
(errmsg("table \"%s.%s\" removed from subscription \"%s\"",
get_namespace_name(get_rel_namespace(relid)),
get_rel_name(relid),
sub->name)));
}
}
}
/*
* 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 update_tuple = false;
Subscription *sub;
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, OBJECT_SUBSCRIPTION,
stmt->subname);
subid = HeapTupleGetOid(tup);
sub = GetSubscription(subid, false);
/* Lock the subscription so nobody else can do anything with it. */
LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock);
/* Form a new tuple. */
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
memset(replaces, false, sizeof(replaces));
switch (stmt->kind)
{
case ALTER_SUBSCRIPTION_OPTIONS:
{
char *slotname;
bool slotname_given;
char *synchronous_commit;
parse_subscription_options(stmt->options, NULL, NULL, NULL,
NULL, &slotname_given, &slotname,
NULL, &synchronous_commit, NULL);
if (slotname_given)
{
if (sub->enabled && !slotname)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("cannot set slot_name = NONE for enabled subscription")));
if (slotname)
values[Anum_pg_subscription_subslotname - 1] =
DirectFunctionCall1(namein, CStringGetDatum(slotname));
else
nulls[Anum_pg_subscription_subslotname - 1] = true;
replaces[Anum_pg_subscription_subslotname - 1] = true;
}
if (synchronous_commit)
{
values[Anum_pg_subscription_subsynccommit - 1] =
CStringGetTextDatum(synchronous_commit);
replaces[Anum_pg_subscription_subsynccommit - 1] = true;
}
update_tuple = true;
break;
}
case ALTER_SUBSCRIPTION_ENABLED:
{
bool enabled,
enabled_given;
parse_subscription_options(stmt->options, NULL,
&enabled_given, &enabled, NULL,
NULL, NULL, NULL, NULL, NULL);
Assert(enabled_given);
if (!sub->slotname && enabled)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("cannot enable subscription that does not have a slot name")));
values[Anum_pg_subscription_subenabled - 1] =
BoolGetDatum(enabled);
replaces[Anum_pg_subscription_subenabled - 1] = true;
if (enabled)
ApplyLauncherWakeupAtCommit();
update_tuple = true;
break;
}
case ALTER_SUBSCRIPTION_CONNECTION:
/* Load the library providing us libpq calls. */
load_file("libpqwalreceiver", false);
/* Check the connection info string. */
walrcv_check_conninfo(stmt->conninfo);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(stmt->conninfo);
replaces[Anum_pg_subscription_subconninfo - 1] = true;
update_tuple = true;
break;
case ALTER_SUBSCRIPTION_PUBLICATION:
{
bool copy_data;
bool refresh;
parse_subscription_options(stmt->options, NULL, NULL, NULL,
NULL, NULL, NULL, &copy_data,
NULL, &refresh);
values[Anum_pg_subscription_subpublications - 1] =
publicationListToArray(stmt->publication);
replaces[Anum_pg_subscription_subpublications - 1] = true;
update_tuple = true;
/* Refresh if user asked us to. */
if (refresh)
{
if (!sub->enabled)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("ALTER SUBSCRIPTION with refresh is not allowed for disabled subscriptions"),
errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION ... WITH (refresh = false).")));
/* Make sure refresh sees the new list of publications. */
sub->publications = stmt->publication;
AlterSubscription_refresh(sub, copy_data);
}
break;
}
case ALTER_SUBSCRIPTION_REFRESH:
{
bool copy_data;
if (!sub->enabled)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));
parse_subscription_options(stmt->options, NULL, NULL, NULL,
NULL, NULL, NULL, &copy_data,
NULL, NULL);
AlterSubscription_refresh(sub, copy_data);
break;
}
default:
elog(ERROR, "unrecognized ALTER SUBSCRIPTION kind %d",
stmt->kind);
}
/* Update the catalog if needed. */
if (update_tuple)
{
tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
replaces);
CatalogTupleUpdate(rel, &tup->t_self, tup);
heap_freetuple(tup);
}
heap_close(rel, RowExclusiveLock);
ObjectAddressSet(myself, SubscriptionRelationId, subid);
InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0);
return myself;
}
/*
* Drop a subscription
*/
void
DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
{
Relation rel;
ObjectAddress myself;
HeapTuple tup;
Oid subid;
Datum datum;
bool isnull;
char *subname;
char *conninfo;
char *slotname;
List *subworkers;
ListCell *lc;
char originname[NAMEDATALEN];
char *err = NULL;
RepOriginId originid;
WalReceiverConn *wrconn = NULL;
StringInfoData cmd;
/*
* Lock pg_subscription with AccessExclusiveLock to ensure that the
* launcher doesn't restart new worker during dropping the subscription
*/
rel = heap_open(SubscriptionRelationId, AccessExclusiveLock);
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, OBJECT_SUBSCRIPTION,
stmt->subname);
/* DROP hook for the subscription being removed */
InvokeObjectDropHook(SubscriptionRelationId, subid, 0);
/*
* Lock the subscription so nobody 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 = TextDatumGetCString(datum);
/* Get slotname */
datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup,
Anum_pg_subscription_subslotname, &isnull);
if (!isnull)
slotname = pstrdup(NameStr(*DatumGetName(datum)));
else
slotname = NULL;
/*
* Since dropping a replication slot is not transactional, the replication
* slot stays dropped even if the transaction rolls back. So we cannot
* run DROP SUBSCRIPTION inside a transaction block if dropping the
* replication slot.
*
* XXX The command name should really be something like "DROP SUBSCRIPTION
* of a subscription that is associated with a replication slot", but we
* don't have the proper facilities for that.
*/
if (slotname)
PreventInTransactionBlock(isTopLevel, "DROP SUBSCRIPTION");
ObjectAddressSet(myself, SubscriptionRelationId, subid);
EventTriggerSQLDropAddObject(&myself, true, true);
/* Remove the tuple from catalog. */
CatalogTupleDelete(rel, &tup->t_self);
ReleaseSysCache(tup);
/*
* Stop all the subscription workers immediately.
*
* This is necessary if we are dropping the replication slot, so that the
* slot becomes accessible.
*
* It is also necessary if the subscription is disabled and was disabled
* in the same transaction. Then the workers haven't seen the disabling
* yet and will still be running, leading to hangs later when we want to
* drop the replication origin. If the subscription was disabled before
* this transaction, then there shouldn't be any workers left, so this
* won't make a difference.
*
* New workers won't be started because we hold an exclusive lock on the
* subscription till the end of the transaction.
*/
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
subworkers = logicalrep_workers_find(subid, false);
LWLockRelease(LogicalRepWorkerLock);
2017-08-14 23:29:33 +02:00
foreach(lc, subworkers)
{
LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
2017-08-14 23:29:33 +02:00
logicalrep_worker_stop(w->subid, w->relid);
}
list_free(subworkers);
/* Clean up dependencies */
deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
/* Remove any associated relation synchronization states. */
RemoveSubscriptionRel(subid, InvalidOid);
/* 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, false);
/*
* If there is no slot associated with the subscription, we can finish
* here.
*/
if (!slotname)
{
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 WAIT", quote_identifier(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),
errhint("Use ALTER SUBSCRIPTION ... SET (slot_name = NONE) "
"to disassociate the subscription from the slot.")));
PG_TRY();
{
WalRcvExecResult *res;
res = walrcv_exec(wrconn, cmd.data, 0, NULL);
if (res->status != WALRCV_OK_COMMAND)
ereport(ERROR,
(errmsg("could not drop the replication slot \"%s\" on publisher",
slotname),
errdetail("The error was: %s", res->err)));
else
ereport(NOTICE,
(errmsg("dropped replication slot \"%s\" on publisher",
slotname)));
walrcv_clear_result(res);
}
PG_CATCH();
{
/* Close the connection in case of failure */
2017-02-21 21:44:07 +01:00
walrcv_disconnect(wrconn);
PG_RE_THROW();
}
PG_END_TRY();
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, OBJECT_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 a subscription must be a superuser.")));
form->subowner = newOwnerId;
CatalogTupleUpdate(rel, &tup->t_self, 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);
}
/*
* Get the list of tables which belong to specified publications on the
* publisher connection.
*/
static List *
fetch_table_list(WalReceiverConn *wrconn, List *publications)
{
WalRcvExecResult *res;
StringInfoData cmd;
TupleTableSlot *slot;
Oid tableRow[2] = {TEXTOID, TEXTOID};
ListCell *lc;
bool first;
List *tablelist = NIL;
Assert(list_length(publications) > 0);
initStringInfo(&cmd);
appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n"
" FROM pg_catalog.pg_publication_tables t\n"
" WHERE t.pubname IN (");
first = true;
foreach(lc, publications)
{
char *pubname = strVal(lfirst(lc));
if (first)
first = false;
else
appendStringInfoString(&cmd, ", ");
appendStringInfoString(&cmd, quote_literal_cstr(pubname));
}
appendStringInfoChar(&cmd, ')');
res = walrcv_exec(wrconn, cmd.data, 2, tableRow);
pfree(cmd.data);
if (res->status != WALRCV_OK_TUPLES)
ereport(ERROR,
(errmsg("could not receive list of replicated tables from the publisher: %s",
res->err)));
/* Process tables. */
Introduce notion of different types of slots (without implementing them). Upcoming work intends to allow pluggable ways to introduce new ways of storing table data. Accessing those table access methods from the executor requires TupleTableSlots to be carry tuples in the native format of such storage methods; otherwise there'll be a significant conversion overhead. Different access methods will require different data to store tuples efficiently (just like virtual, minimal, heap already require fields in TupleTableSlot). To allow that without requiring additional pointer indirections, we want to have different structs (embedding TupleTableSlot) for different types of slots. Thus different types of slots are needed, which requires adapting creators of slots. The slot that most efficiently can represent a type of tuple in an executor node will often depend on the type of slot a child node uses. Therefore we need to track the type of slot is returned by nodes, so parent slots can create slots based on that. Relatedly, JIT compilation of tuple deforming needs to know which type of slot a certain expression refers to, so it can create an appropriate deforming function for the type of tuple in the slot. But not all nodes will only return one type of slot, e.g. an append node will potentially return different types of slots for each of its subplans. Therefore add function that allows to query the type of a node's result slot, and whether it'll always be the same type (whether it's fixed). This can be queried using ExecGetResultSlotOps(). The scan, result, inner, outer type of slots are automatically inferred from ExecInitScanTupleSlot(), ExecInitResultSlot(), left/right subtrees respectively. If that's not correct for a node, that can be overwritten using new fields in PlanState. This commit does not introduce the actually abstracted implementation of different kind of TupleTableSlots, that will be left for a followup commit. The different types of slots introduced will, for now, still use the same backing implementation. While this already partially invalidates the big comment in tuptable.h, it seems to make more sense to update it later, when the different TupleTableSlot implementations actually exist. Author: Ashutosh Bapat and Andres Freund, with changes by Amit Khandekar Discussion: https://postgr.es/m/20181105210039.hh4vvi4vwoq5ba2q@alap3.anarazel.de
2018-11-16 07:00:30 +01:00
slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
{
char *nspname;
char *relname;
bool isnull;
RangeVar *rv;
nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
Assert(!isnull);
relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
Assert(!isnull);
rv = makeRangeVar(pstrdup(nspname), pstrdup(relname), -1);
tablelist = lappend(tablelist, rv);
ExecClearTuple(slot);
}
ExecDropSingleTupleTableSlot(slot);
walrcv_clear_result(res);
return tablelist;
}