Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtp (Exim 4.84_2) (envelope-from ) id 1d6gbw-00022t-K3 for pgsql-hackers@arkaria.postgresql.org; Fri, 05 May 2017 17:01:44 +0000 Received: from localhost ([127.0.0.1] helo=postgresql.org) by malur.postgresql.org with smtp (Exim 4.84_2) (envelope-from ) id 1d6gbw-0007i9-3M for pgsql-hackers@arkaria.postgresql.org; Fri, 05 May 2017 17:01:44 +0000 Received: from makus.postgresql.org ([2001:4800:1501:1::229]) by malur.postgresql.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_CBC_SHA384:256) (Exim 4.84_2) (envelope-from ) id 1d6gbm-0007V0-3l for pgsql-hackers@postgresql.org; Fri, 05 May 2017 17:01:34 +0000 Received: from mail-wr0-x232.google.com ([2a00:1450:400c:c0c::232]) by makus.postgresql.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_CBC_SHA1:256) (Exim 4.84_2) (envelope-from ) id 1d6gbi-0006QN-QY for pgsql-hackers@postgresql.org; Fri, 05 May 2017 17:01:32 +0000 Received: by mail-wr0-x232.google.com with SMTP id l50so7175915wrc.3 for ; Fri, 05 May 2017 10:01:30 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=2ndquadrant-com.20150623.gappssmtp.com; s=20150623; h=subject:to:references:cc:from:message-id:date:user-agent :mime-version:in-reply-to; bh=hOAuI2OX3DA4e34D7rUUTtRa6p9AvxXKgqMWOf/eOTc=; b=c5c4u0nPiBkXt81qtXBPqKsWdtbL6CzxGv4BgFZUAKxORQHlocatEcgzxc6XZ1z8XQ 1mxQyxNBpQshSD3QzR2Dh5+2DDJzroG7/ZYWQTbeOTI6/SXUuNnw6k3ezD+dTGOT0v4l aIRSN9gVkI6URQ3GnyhBIhLjvSbdX+M2M8j40CpxGQ3PntPf9uV5kDJuiXxOhMvAJeUw EmeBM4LJDS5YrcQ5qpmhhBZvfzdDJqZcVF3S7uWD11M4CX/b0NZWUK1c3TQe1+EXbnRg jS3X7GMgwI1jshuQWmy+afUUPa78NzJj0Uqlg9S5EEbN+sUblEtpwLD/rnan3TJfd37+ uMww== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:subject:to:references:cc:from:message-id:date :user-agent:mime-version:in-reply-to; bh=hOAuI2OX3DA4e34D7rUUTtRa6p9AvxXKgqMWOf/eOTc=; b=poOY06ftzsrjVhldPF4U4D0oNAV3g43bejEGCp9j5tOdL236E4RNLRGC8HlJoKG4UU 2KxFrnp2+CsQapTNv4K7oIcZjKRG6AgZFc4t2skM/MxPsYODCBMOJ2JjYBU4d4budHiE Ejn1Mt9872+rrJjfb42R9A+SDgjPxU6GLVed5xMv9jTt+7e0zvgYpZ4m617Qz89POiak wAhmaf9UciHLh5SiZtNWpI6rGSigxADQCPeI3YmnNYQAjYaW34q2ZtU+1vx04v0xQqc2 ySbQxdJijvHmzlpLTfYqdwfGCMAAyRIkedaCmHvIGOUNEZwe6q3a9IvgLf+Mq5zVMa4K PASQ== X-Gm-Message-State: AN3rC/7yu4NnS9yXZdyssXNf31G6ti4uQ03lJwGSuJYdrDOM39lSenHV NJiwtT7o4FrfSClcpok/ZALLAB2JEbqR5cvS2BvQF7iKwJG0KUaVBAWhZLIjzDtr0ZX/Bfh5Qhb Wg4zdg7DGe1E+d4tslpc4J8ADziG9o26XlFb037s2HluoU1Tmqs4Ii0w6qO1dkc5mpMtdqxcr X-Received: by 10.223.130.201 with SMTP id 67mr30709854wrc.106.1494003689091; Fri, 05 May 2017 10:01:29 -0700 (PDT) Received: from [192.168.0.105] (150.123.broadband11.iol.cz. [90.178.123.150]) by smtp.gmail.com with ESMTPSA id l1sm6871445wrl.15.2017.05.05.10.01.27 (version=TLS1_2 cipher=ECDHE-RSA-AES128-GCM-SHA256 bits=128/128); Fri, 05 May 2017 10:01:27 -0700 (PDT) Subject: Re: logical replication syntax (was DROP SUBSCRIPTION, query cancellations and slot handling) To: Tom Lane References: <27478.1493727233@sss.pgh.pa.us> <29431.1493730652@sss.pgh.pa.us> <360a1932-7af4-609b-94f1-2ad419430869@2ndquadrant.com> <9174dc31-787a-a2f6-818f-cb4a5c1dfddb@2ndquadrant.com> <30166.1493731904@sss.pgh.pa.us> Cc: Robert Haas , Michael Paquier , PostgreSQL mailing lists From: Petr Jelinek Message-ID: Date: Fri, 5 May 2017 19:01:27 +0200 User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:45.0) Gecko/20100101 Thunderbird/45.8.0 MIME-Version: 1.0 In-Reply-To: <30166.1493731904@sss.pgh.pa.us> Content-Type: multipart/mixed; boundary="------------06EC69927D8D5233C2F60AB9" X-Pg-Spam-Score: -1.9 (-) List-Archive: List-Help: List-ID: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: X-Mailing-List: pgsql-hackers Precedence: bulk Sender: pgsql-hackers-owner@postgresql.org This is a multi-part message in MIME format. --------------06EC69927D8D5233C2F60AB9 Content-Type: text/plain; charset=windows-1252 Content-Transfer-Encoding: 7bit On 02/05/17 15:31, Tom Lane wrote: > Petr Jelinek writes: >> Let me expand, if we don't drop the slot by default when dropping >> subscription, we'll have a lot of users with dead slots laying around >> holding back WAL/catalog_xmin, that's really bad. If we do drop by >> default like now, we need option to not do that, neither RESTRICT, nor >> CASCADE fit that. > > I'm not sure which part of "you can't have an option in DROP" isn't > clear to you. Whatever the default behavior is always has to work, > because that is what's going to happen during DROP OWNED BY, or > DROP DATABASE. If you want more than one behavior during DROP, > you need to drive that off something represented as a persistent > property of the object, not off an option in the DROP command. > > I agree that RESTRICT/CASCADE don't fit this, unless the model > is that the slot is always owned by the subscription, which might > be too restrictive. > What do you think of attached. I actually did add RESTRICT/CASCADE, in a way that if there is slot RESTRICT will refuse to drop subscription and CASCADE will try to drop it. Still all errors. The new way to not drop slot is to set slot_name to NONE which is new value that I invented (inspiration from OWNED BY sequences) which basically disassociates the subscription from slot. It's slightly less automatic this way but perhaps that's not a bad thing, at least nobody will drop slots by mistake while we still make sure that slots are not left over without anybody noticing. Note that this would need catalog version bump as it removes NOT NULL constraint from subslotname. -- Petr Jelinek http://www.2ndQuadrant.com/ PostgreSQL Development, 24x7 Support, Training & Services --------------06EC69927D8D5233C2F60AB9 Content-Type: binary/octet-stream; name="Remove-the-NODROP-SLOT-option-from-DROP-SUBSCRIPTION.patch" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename*0="Remove-the-NODROP-SLOT-option-from-DROP-SUBSCRIPTION.patch" From d242be3dc95443f226a906f108c52d32b9ac3e1e Mon Sep 17 00:00:00 2001 From: Petr Jelinek Date: Fri, 5 May 2017 18:14:00 +0200 Subject: [PATCH] Remove the NODROP SLOT option from DROP SUBSCRIPTION --- doc/src/sgml/ref/create_subscription.sgml | 7 + doc/src/sgml/ref/drop_subscription.sgml | 22 +-- src/backend/catalog/pg_subscription.c | 9 +- src/backend/commands/subscriptioncmds.c | 195 ++++++++++++++++----- src/backend/nodes/copyfuncs.c | 2 +- src/backend/nodes/equalfuncs.c | 2 +- src/backend/parser/gram.y | 43 ++--- src/backend/replication/logical/worker.c | 35 ++-- src/backend/tcop/utility.c | 3 + src/include/catalog/pg_subscription.h | 2 +- src/include/nodes/parsenodes.h | 2 +- .../dummy_seclabel/expected/dummy_seclabel.out | 4 +- .../modules/dummy_seclabel/sql/dummy_seclabel.sql | 4 +- src/test/regress/expected/object_address.out | 4 +- src/test/regress/expected/subscription.out | 13 +- src/test/regress/sql/object_address.sql | 4 +- src/test/regress/sql/subscription.sql | 12 +- src/test/subscription/t/001_rep_changes.pl | 2 +- src/test/subscription/t/004_sync.pl | 8 +- 19 files changed, 241 insertions(+), 132 deletions(-) diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index fcec254..c9256e7 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -129,6 +129,13 @@ CREATE SUBSCRIPTION subscription_namesubscription_name for slot name. + + When slot_name is set to + NONE there will be no slot associated with the + subscription. Such subscriptions must also have both + enabled and create_slot set + to false. + diff --git a/doc/src/sgml/ref/drop_subscription.sgml b/doc/src/sgml/ref/drop_subscription.sgml index f1ac125..8b00c45 100644 --- a/doc/src/sgml/ref/drop_subscription.sgml +++ b/doc/src/sgml/ref/drop_subscription.sgml @@ -21,7 +21,7 @@ PostgreSQL documentation -DROP SUBSCRIPTION [ IF EXISTS ] name [ DROP SLOT | NODROP SLOT ] +DROP SUBSCRIPTION [ IF EXISTS ] name [ CASCADE | RESTRICT ] @@ -57,20 +57,16 @@ DROP SUBSCRIPTION [ IF EXISTS ] name - DROP SLOT - NODROP SLOT - - - Specifies whether to drop the replication slot on the publisher. The - default is - DROP SLOT. - + CASCADE + RESTRICT + - If the publisher is not reachable when the subscription is to be - dropped, then it is useful to specify NODROP SLOT. - But the replication slot on the publisher will then have to be removed - manually. + These key words are used to determine what to do with when there is a + replication slot associated with the subscription. The + RESTRICT will refuse to drop the subscription in + such case, while CASCADE will drop the associated + slot. RESTRICT is the default. diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 22587a4..7dc21f1 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -82,8 +82,10 @@ GetSubscription(Oid subid, bool missing_ok) tup, Anum_pg_subscription_subslotname, &isnull); - Assert(!isnull); - sub->slotname = pstrdup(NameStr(*DatumGetName(datum))); + if (!isnull) + sub->slotname = pstrdup(NameStr(*DatumGetName(datum))); + else + sub->slotname = NULL; /* Get synccommit */ datum = SysCacheGetAttr(SUBSCRIPTIONOID, @@ -147,7 +149,8 @@ FreeSubscription(Subscription *sub) { pfree(sub->name); pfree(sub->conninfo); - pfree(sub->slotname); + if (sub->slotname) + pfree(sub->slotname); list_free_deep(sub->publications); pfree(sub); } diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index c00981e..afb116a 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -60,7 +60,8 @@ static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); */ static void parse_subscription_options(List *options, bool *connect, bool *enabled_given, - bool *enabled, bool *create_slot, char **slot_name, + bool *enabled, bool *create_slot, + bool *slot_name_given, char **slot_name, bool *copy_data, char **synchronous_commit) { ListCell *lc; @@ -78,7 +79,10 @@ parse_subscription_options(List *options, bool *connect, bool *enabled_given, 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) @@ -121,12 +125,17 @@ parse_subscription_options(List *options, bool *connect, bool *enabled_given, } else if (strcmp(defel->defname, "slot_name") == 0 && slot_name) { - if (*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) { @@ -164,26 +173,43 @@ parse_subscription_options(List *options, bool *connect, bool *enabled_given, if (connect && !*connect) { /* Check for incompatible options from the user. */ - if (*enabled_given && *enabled) + if (enabled && *enabled_given && *enabled) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("noconnect and enabled are mutually exclusive options"))); + errmsg("connect = false and enabled are mutually exclusive options"))); - if (create_slot_given && *create_slot) + if (create_slot && create_slot_given && *create_slot) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("noconnect and create slot are mutually exclusive options"))); + errmsg("connect = false and create_slot are mutually exclusive options"))); - if (copy_data_given && *copy_data) + if (copy_data && copy_data_given && *copy_data) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("noconnect and copy data are mutually exclusive options"))); + errmsg("connect = false and copy_data 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 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 are mutually exclusive options"))); + } } /* @@ -260,6 +286,7 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) char *synchronous_commit; char *conninfo; char *slotname; + bool slotname_given; char originname[NAMEDATALEN]; bool create_slot; List *publications; @@ -269,8 +296,8 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) * Connection and publication should not be specified here. */ parse_subscription_options(stmt->options, &connect, &enabled_given, - &enabled, &create_slot, &slotname, ©_data, - &synchronous_commit); + &enabled, &create_slot, &slotname_given, + &slotname, ©_data, &synchronous_commit); /* * Since creating a replication slot is not transactional, rolling back @@ -299,8 +326,9 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) stmt->subname))); } - if (slotname == NULL) + if (!slotname_given && slotname == NULL) slotname = stmt->subname; + /* The default for synchronous_commit of subscriptions is off. */ if (synchronous_commit == NULL) synchronous_commit = "off"; @@ -325,8 +353,11 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) 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)); + 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] = @@ -396,6 +427,8 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) */ if (create_slot) { + Assert(slotname); + walrcv_create_slot(wrconn, slotname, false, CRS_NOEXPORT_SNAPSHOT, &lsn); ereport(NOTICE, @@ -548,6 +581,7 @@ AlterSubscription(AlterSubscriptionStmt *stmt) HeapTuple tup; Oid subid; bool update_tuple = false; + Subscription *sub; rel = heap_open(SubscriptionRelationId, RowExclusiveLock); @@ -567,6 +601,7 @@ AlterSubscription(AlterSubscriptionStmt *stmt) stmt->subname); subid = HeapTupleGetOid(tup); + sub = GetSubscription(subid, false); /* Form a new tuple. */ memset(values, 0, sizeof(values)); @@ -577,19 +612,29 @@ AlterSubscription(AlterSubscriptionStmt *stmt) { case ALTER_SUBSCRIPTION_OPTIONS: { - char *slot_name; - char *synchronous_commit; + char *slotname; + bool slotname_given; + char *synchronous_commit; parse_subscription_options(stmt->options, NULL, NULL, NULL, - NULL, &slot_name, NULL, - &synchronous_commit); + NULL, &slotname_given, &slotname, + NULL, &synchronous_commit); - if (slot_name) + if (slotname_given) { - values[Anum_pg_subscription_subslotname - 1] = - DirectFunctionCall1(namein, CStringGetDatum(slot_name)); + 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] = @@ -608,9 +653,14 @@ AlterSubscription(AlterSubscriptionStmt *stmt) parse_subscription_options(stmt->options, NULL, &enabled_given, &enabled, NULL, - NULL, NULL, NULL); + NULL, NULL, NULL, NULL); Assert(enabled_given); + if (!sub->slotname && enabled) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("cannot enable subscription which does not have a slot_name"))); + values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled); replaces[Anum_pg_subscription_subenabled - 1] = true; @@ -633,10 +683,10 @@ AlterSubscription(AlterSubscriptionStmt *stmt) case ALTER_SUBSCRIPTION_PUBLICATION_REFRESH: { bool copy_data; - Subscription *sub = GetSubscription(subid, false); parse_subscription_options(stmt->options, NULL, NULL, NULL, - NULL, NULL, ©_data, NULL); + NULL, NULL, NULL, ©_data, + NULL); values[Anum_pg_subscription_subpublications - 1] = publicationListToArray(stmt->publication); @@ -647,6 +697,11 @@ AlterSubscription(AlterSubscriptionStmt *stmt) /* Refresh if user asked us to. */ if (stmt->kind == ALTER_SUBSCRIPTION_PUBLICATION_REFRESH) { + if (!sub->enabled) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions"))); + /* Make sure refresh sees the new list of publications. */ sub->publications = stmt->publication; @@ -659,10 +714,15 @@ AlterSubscription(AlterSubscriptionStmt *stmt) case ALTER_SUBSCRIPTION_REFRESH: { bool copy_data; - Subscription *sub = GetSubscription(subid, false); + + 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, ©_data, NULL); + NULL, NULL, NULL, ©_data, + NULL); AlterSubscription_refresh(sub, copy_data); @@ -721,8 +781,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) * run DROP SUBSCRIPTION inside a transaction block if dropping the * replication slot. */ - if (stmt->drop_slot) - PreventTransactionChain(isTopLevel, "DROP SUBSCRIPTION ... DROP SLOT"); + if (stmt->behavior == DROP_CASCADE) + PreventTransactionChain(isTopLevel, "DROP SUBSCRIPTION ... CASCADE"); /* * Lock pg_subscription with AccessExclusiveLock to ensure @@ -782,8 +842,10 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) /* Get slotname */ datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subslotname, &isnull); - Assert(!isnull); - slotname = pstrdup(NameStr(*DatumGetName(datum))); + if (!isnull) + slotname = pstrdup(NameStr(*DatumGetName(datum))); + else + slotname = NULL; ObjectAddressSet(myself, SubscriptionRelationId, subid); EventTriggerSQLDropAddObject(&myself, true, true); @@ -808,43 +870,84 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) if (originid != InvalidRepOriginId) replorigin_drop(originid); - /* If the user asked to not drop the slot, we are done mow.*/ - if (!stmt->drop_slot) + /* If there is no slot associated with subscription we can finish here. */ + if (!slotname) { heap_close(rel, NoLock); return; } /* - * Otherwise drop the replication slot at the publisher node using + * Otherwise check for 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))); + (errmsg("could not connect to publisher when attempting to fetch " + "information about replication slot \"%s\"", slotname), + errdetail("The error was: %s", err), + errhint("Use ALTER SUBSCRIPTION ... WITH (slot_name = NONE) " + "to disassociate the subscription from slot."))); PG_TRY(); { WalRcvExecResult *res; - res = walrcv_exec(wrconn, cmd.data, 0, NULL); - - if (res->status != WALRCV_OK_COMMAND) + TupleTableSlot *tupslot; + Oid slot_row_desc[1] = {BOOLOID}; + bool found; + + initStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT true FROM pg_catalog.pg_replication_slots WHERE slot_name = %s", + quote_literal_cstr(slotname)); + res = walrcv_exec(wrconn, cmd.data, 1, slot_row_desc); + if (res->status != WALRCV_OK_TUPLES) ereport(ERROR, - (errmsg("could not drop the replication slot \"%s\" on publisher", - slotname), + (errmsg("could not fetch the replication slot info from publisher"), errdetail("The error was: %s", res->err))); + + tupslot = MakeSingleTupleTableSlot(res->tupledesc); + found = tuplestore_gettupleslot(res->tuplestore, true, false, + tupslot); + ExecDropSingleTupleTableSlot(tupslot); + walrcv_clear_result(res); + + /* If slot was not found on publisher, we are done. */ + if (!found) + { + walrcv_disconnect(wrconn); + pfree(cmd.data); + heap_close(rel, NoLock); + return; + } + + /* Otherwise the next action depends on the drop_behavior. */ + if (stmt->behavior == DROP_CASCADE) + { + resetStringInfo(&cmd); + appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s", + quote_identifier(slotname)); + 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))); + } else - ereport(NOTICE, - (errmsg("dropped replication slot \"%s\" on publisher", - slotname))); + ereport(ERROR, + (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST), + errmsg("cannot drop subscription \"%s\" because there is still replication slot associated with it", + subname), + errhint("Use DROP ... CASCADE to drop the slot too."))); walrcv_clear_result(res); } diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 35a237a..2d2a9d0 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -4537,8 +4537,8 @@ _copyDropSubscriptionStmt(const DropSubscriptionStmt *from) DropSubscriptionStmt *newnode = makeNode(DropSubscriptionStmt); COPY_STRING_FIELD(subname); - COPY_SCALAR_FIELD(drop_slot); COPY_SCALAR_FIELD(missing_ok); + COPY_SCALAR_FIELD(behavior); return newnode; } diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index 21dfbb0..b5459cd 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2246,8 +2246,8 @@ _equalDropSubscriptionStmt(const DropSubscriptionStmt *a, const DropSubscriptionStmt *b) { COMPARE_STRING_FIELD(subname); - COMPARE_SCALAR_FIELD(drop_slot); COMPARE_SCALAR_FIELD(missing_ok); + COMPARE_SCALAR_FIELD(behavior); return true; } diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 732647b..bdd2f6d 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -415,7 +415,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type arg_class %type func_return func_type -%type opt_trusted opt_restart_seqs opt_drop_slot +%type opt_trusted opt_restart_seqs %type OptTemp %type OptNoLog %type OnCommitOption @@ -467,7 +467,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type def_arg columnElem where_clause where_or_current_clause a_expr b_expr c_expr AexprConst indirection_el opt_slice_bound columnref in_expr having_clause func_table xmltable array_expr - ExclusionWhereClause + ExclusionWhereClause operator_def_arg %type rowsfrom_item rowsfrom_list opt_col_def_list %type opt_ordinality %type ExclusionConstraintList ExclusionConstraintElem @@ -5684,6 +5684,7 @@ def_arg: func_type { $$ = (Node *)$1; } | qual_all_Op { $$ = (Node *)$1; } | NumericOnly { $$ = (Node *)$1; } | Sconst { $$ = (Node *)makeString($1); } + | NONE { $$ = (Node *)makeString(pstrdup($1)); } ; old_aggr_definition: '(' old_aggr_list ')' { $$ = $2; } @@ -8923,8 +8924,16 @@ operator_def_list: operator_def_elem { $$ = list_make1($1); } operator_def_elem: ColLabel '=' NONE { $$ = makeDefElem($1, NULL, @1); } - | ColLabel '=' def_arg - { $$ = makeDefElem($1, (Node *) $3, @1); } + | ColLabel '='operator_def_arg + { $$ = makeDefElem($1, (Node *) $3, @1); } + ; + +operator_def_arg: + func_type { $$ = (Node *)$1; } + | reserved_keyword { $$ = (Node *)makeString(pstrdup($1)); } + | qual_all_Op { $$ = (Node *)$1; } + | NumericOnly { $$ = (Node *)$1; } + | Sconst { $$ = (Node *)makeString($1); } ; /***************************************************************************** @@ -9315,42 +9324,24 @@ AlterSubscriptionStmt: * *****************************************************************************/ -DropSubscriptionStmt: DROP SUBSCRIPTION name opt_drop_slot +DropSubscriptionStmt: DROP SUBSCRIPTION name opt_drop_behavior { DropSubscriptionStmt *n = makeNode(DropSubscriptionStmt); n->subname = $3; - n->drop_slot = $4; n->missing_ok = false; + n->behavior = $4; $$ = (Node *) n; } - | DROP SUBSCRIPTION IF_P EXISTS name opt_drop_slot + | DROP SUBSCRIPTION IF_P EXISTS name opt_drop_behavior { DropSubscriptionStmt *n = makeNode(DropSubscriptionStmt); n->subname = $5; - n->drop_slot = $6; n->missing_ok = true; + n->behavior = $6; $$ = (Node *) n; } ; -opt_drop_slot: - DROP SLOT - { - $$ = TRUE; - } - | IDENT SLOT - { - if (strcmp($1, "nodrop") == 0) - $$ = FALSE; - else - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("unrecognized option \"%s\"", $1), - parser_errposition(@1))); - } - | /*EMPTY*/ { $$ = TRUE; } - ; - /***************************************************************************** * * QUERY: Define Rewrite Rule diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 2d7770d..a4b7dcb 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -1329,6 +1329,22 @@ reread_subscription(void) } /* + * Exit if the subscription was disabled. + * This normally should not happen as the worker gets killed + * during ALTER SUBSCRIPTION ... DISABLE. + */ + if (!newsub->enabled) + { + ereport(LOG, + (errmsg("logical replication worker for subscription \"%s\" will " + "stop because the subscription was disabled", + MySubscription->name))); + + walrcv_disconnect(wrconn); + proc_exit(0); + } + + /* * Exit if connection string was changed. The launcher will start * new worker. */ @@ -1358,6 +1374,9 @@ reread_subscription(void) proc_exit(0); } + /* !slotname should never happen when enabled is true. */ + Assert(newsub->slotname); + /* * We need to make new connection to new slot if slot name has changed * so exit here as well if that's the case. @@ -1388,22 +1407,6 @@ reread_subscription(void) proc_exit(0); } - /* - * Exit if the subscription was disabled. - * This normally should not happen as the worker gets killed - * during ALTER SUBSCRIPTION ... DISABLE. - */ - if (!newsub->enabled) - { - ereport(LOG, - (errmsg("logical replication worker for subscription \"%s\" will " - "stop because the subscription was disabled", - MySubscription->name))); - - walrcv_disconnect(wrconn); - proc_exit(0); - } - /* Check for other changes that should never happen too. */ if (newsub->dbid != MySubscription->dbid) { diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 24e5c42..d4fa5a7 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -2303,6 +2303,9 @@ CreateCommandTag(Node *parsetree) case OBJECT_PUBLICATION: tag = "DROP PUBLICATION"; break; + case OBJECT_SUBSCRIPTION: + tag = "DROP SUBSCRIPTION"; + break; case OBJECT_STATISTIC_EXT: tag = "DROP STATISTICS"; break; diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 5550f19..d4f3979 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -45,7 +45,7 @@ CATALOG(pg_subscription,6100) BKI_SHARED_RELATION BKI_ROWTYPE_OID(6101) BKI_SCHE text subconninfo BKI_FORCE_NOT_NULL; /* Slot name on publisher */ - NameData subslotname BKI_FORCE_NOT_NULL; + NameData subslotname; /* Synchronous commit setting for worker */ text subsynccommit BKI_FORCE_NOT_NULL; diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index e1d454a..46c23c2 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3393,8 +3393,8 @@ typedef struct DropSubscriptionStmt { NodeTag type; char *subname; /* Name of of the subscription */ - bool drop_slot; /* Should we drop the slot on remote side? */ bool missing_ok; /* Skip error if missing? */ + DropBehavior behavior; /* RESTRICT or CASCADE behavior */ } DropSubscriptionStmt; #endif /* PARSENODES_H */ diff --git a/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out b/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out index f6fc29a..77bdc93 100644 --- a/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out +++ b/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out @@ -69,7 +69,7 @@ CREATE SCHEMA dummy_seclabel_test; SECURITY LABEL ON SCHEMA dummy_seclabel_test IS 'unclassified'; -- OK SET client_min_messages = error; CREATE PUBLICATION dummy_pub; -CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false); +CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false, slot_name = NONE); RESET client_min_messages; SECURITY LABEL ON PUBLICATION dummy_pub IS 'classified'; SECURITY LABEL ON SUBSCRIPTION dummy_sub IS 'classified'; @@ -111,7 +111,7 @@ NOTICE: event ddl_command_end: SECURITY LABEL DROP EVENT TRIGGER always_start, always_end, always_drop, always_rewrite; DROP VIEW dummy_seclabel_view1; DROP TABLE dummy_seclabel_tbl1, dummy_seclabel_tbl2; -DROP SUBSCRIPTION dummy_sub NODROP SLOT; +DROP SUBSCRIPTION dummy_sub; DROP PUBLICATION dummy_pub; DROP ROLE regress_dummy_seclabel_user1; DROP ROLE regress_dummy_seclabel_user2; diff --git a/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql b/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql index d7795bd..8c347b6 100644 --- a/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql +++ b/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql @@ -73,7 +73,7 @@ SECURITY LABEL ON SCHEMA dummy_seclabel_test IS 'unclassified'; -- OK SET client_min_messages = error; CREATE PUBLICATION dummy_pub; -CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false); +CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false, slot_name = NONE); RESET client_min_messages; SECURITY LABEL ON PUBLICATION dummy_pub IS 'classified'; SECURITY LABEL ON SUBSCRIPTION dummy_sub IS 'classified'; @@ -108,7 +108,7 @@ DROP EVENT TRIGGER always_start, always_end, always_drop, always_rewrite; DROP VIEW dummy_seclabel_view1; DROP TABLE dummy_seclabel_tbl1, dummy_seclabel_tbl2; -DROP SUBSCRIPTION dummy_sub NODROP SLOT; +DROP SUBSCRIPTION dummy_sub; DROP PUBLICATION dummy_pub; DROP ROLE regress_dummy_seclabel_user1; diff --git a/src/test/regress/expected/object_address.out b/src/test/regress/expected/object_address.out index 0849715..700f261 100644 --- a/src/test/regress/expected/object_address.out +++ b/src/test/regress/expected/object_address.out @@ -37,7 +37,7 @@ CREATE TRANSFORM FOR int LANGUAGE SQL ( FROM SQL WITH FUNCTION varchar_transform(internal), TO SQL WITH FUNCTION int4recv(internal)); CREATE PUBLICATION addr_pub FOR TABLE addr_nsp.gentable; -CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false); +CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false, slot_name = NONE); WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables CREATE STATISTICS addr_nsp.gentable_stat ON (a,b) FROM addr_nsp.gentable; -- test some error cases @@ -477,7 +477,7 @@ SELECT (pg_identify_object(addr1.classid, addr1.objid, addr1.objsubid)).*, SET client_min_messages TO 'warning'; DROP FOREIGN DATA WRAPPER addr_fdw CASCADE; DROP PUBLICATION addr_pub; -DROP SUBSCRIPTION addr_sub NODROP SLOT; +DROP SUBSCRIPTION addr_sub; DROP SCHEMA addr_nsp CASCADE; DROP OWNED BY regress_addr_user; DROP USER regress_addr_user; diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 33d53de..fcf5646 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -109,17 +109,18 @@ HINT: The owner of a subscription must be a superuser. ALTER ROLE regress_subscription_user2 SUPERUSER; -- now it works ALTER SUBSCRIPTION testsub OWNER TO regress_subscription_user2; --- fail - cannot do DROP SUBSCRIPTION DROP SLOT inside transaction block +ALTER SUBSCRIPTION testsub WITH (slot_name = NONE); +-- fail - cannot do DROP SUBSCRIPTION CASCADE inside transaction block BEGIN; -DROP SUBSCRIPTION testsub DROP SLOT; -ERROR: DROP SUBSCRIPTION ... DROP SLOT cannot run inside a transaction block +DROP SUBSCRIPTION testsub CASCADE; +ERROR: DROP SUBSCRIPTION ... CASCADE cannot run inside a transaction block COMMIT; BEGIN; -DROP SUBSCRIPTION testsub NODROP SLOT; +DROP SUBSCRIPTION testsub; COMMIT; -DROP SUBSCRIPTION IF EXISTS testsub NODROP SLOT; +DROP SUBSCRIPTION IF EXISTS testsub; NOTICE: subscription "testsub" does not exist, skipping -DROP SUBSCRIPTION testsub NODROP SLOT; -- fail +DROP SUBSCRIPTION testsub; -- fail ERROR: subscription "testsub" does not exist RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user; diff --git a/src/test/regress/sql/object_address.sql b/src/test/regress/sql/object_address.sql index c698a63..8a738e2 100644 --- a/src/test/regress/sql/object_address.sql +++ b/src/test/regress/sql/object_address.sql @@ -40,7 +40,7 @@ CREATE TRANSFORM FOR int LANGUAGE SQL ( FROM SQL WITH FUNCTION varchar_transform(internal), TO SQL WITH FUNCTION int4recv(internal)); CREATE PUBLICATION addr_pub FOR TABLE addr_nsp.gentable; -CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false); +CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false, slot_name = NONE); CREATE STATISTICS addr_nsp.gentable_stat ON (a,b) FROM addr_nsp.gentable; -- test some error cases @@ -205,7 +205,7 @@ SET client_min_messages TO 'warning'; DROP FOREIGN DATA WRAPPER addr_fdw CASCADE; DROP PUBLICATION addr_pub; -DROP SUBSCRIPTION addr_sub NODROP SLOT; +DROP SUBSCRIPTION addr_sub; DROP SCHEMA addr_nsp CASCADE; diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 5454420..c27fee5 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -80,17 +80,19 @@ ALTER ROLE regress_subscription_user2 SUPERUSER; -- now it works ALTER SUBSCRIPTION testsub OWNER TO regress_subscription_user2; --- fail - cannot do DROP SUBSCRIPTION DROP SLOT inside transaction block +ALTER SUBSCRIPTION testsub WITH (slot_name = NONE); + +-- fail - cannot do DROP SUBSCRIPTION CASCADE inside transaction block BEGIN; -DROP SUBSCRIPTION testsub DROP SLOT; +DROP SUBSCRIPTION testsub CASCADE; COMMIT; BEGIN; -DROP SUBSCRIPTION testsub NODROP SLOT; +DROP SUBSCRIPTION testsub; COMMIT; -DROP SUBSCRIPTION IF EXISTS testsub NODROP SLOT; -DROP SUBSCRIPTION testsub NODROP SLOT; -- fail +DROP SUBSCRIPTION IF EXISTS testsub; +DROP SUBSCRIPTION testsub; -- fail RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user; diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl index b4c5539..be55279 100644 --- a/src/test/subscription/t/001_rep_changes.pl +++ b/src/test/subscription/t/001_rep_changes.pl @@ -191,7 +191,7 @@ $node_publisher->poll_query_until('postgres', or die "Timed out while waiting for apply to restart"; # check all the cleanup -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_renamed DROP SLOT"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_renamed CASCADE"); $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_subscription"); diff --git a/src/test/subscription/t/004_sync.pl b/src/test/subscription/t/004_sync.pl index 8ece7dd..779c668 100644 --- a/src/test/subscription/t/004_sync.pl +++ b/src/test/subscription/t/004_sync.pl @@ -52,7 +52,7 @@ my $result = is($result, qq(10), 'initial data synced for first sub'); # drop subscription so that there is unreplicated data -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub CASCADE"); $node_publisher->safe_psql('postgres', "INSERT INTO tab_rep SELECT generate_series(11,20)"); @@ -89,8 +89,8 @@ $node_subscriber->poll_query_until('postgres', "SELECT pid IS NOT NULL FROM pg_s or die "Timed out while waiting for subscriber to start"; # and drop both subscriptions -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub2"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub CASCADE"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub2 CASCADE"); # check subscriptions are removed $result = @@ -154,7 +154,7 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_rep_next"); is($result, qq(20), 'changes for table added after subscription initialized replicated'); -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub CASCADE"); $node_subscriber->stop('fast'); $node_publisher->stop('fast'); -- 2.7.4 --------------06EC69927D8D5233C2F60AB9 Content-Type: text/plain Content-Disposition: inline Content-Transfer-Encoding: 8bit MIME-Version: 1.0 -- Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org) To make changes to your subscription: http://www.postgresql.org/mailpref/pgsql-hackers --------------06EC69927D8D5233C2F60AB9--