diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index 250806f..406b518 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -47,8 +47,10 @@ CREATE SUBSCRIPTION subscription_name - A logical replication worker will be started to replicate data for the new - subscription at the commit of the transaction where this command is run. + CREATE SUBSCRIPTION cannot be executed inside a + transaction block when CREATE SLOT is specified. + A logical replication worker will be started to replicate data for + the new subscription at the end of this command. diff --git a/doc/src/sgml/ref/drop_subscription.sgml b/doc/src/sgml/ref/drop_subscription.sgml index c89a016..0828257 100644 --- a/doc/src/sgml/ref/drop_subscription.sgml +++ b/doc/src/sgml/ref/drop_subscription.sgml @@ -42,8 +42,8 @@ PostgreSQL documentation - The replication worker associated with the subscription will not stop until - after the transaction that issued this command has committed. + DROP SUBSCRIPTION cannot be executed inside a + transaction block when DROP SLOT is specified. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index c5c8581..bd45acf 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -18,6 +18,7 @@ #include "access/heapam.h" #include "access/htup_details.h" +#include "access/xact.h" #include "catalog/indexing.h" #include "catalog/objectaccess.h" @@ -227,7 +228,7 @@ publicationListToArray(List *publist) * Create new subscription. */ ObjectAddress -CreateSubscription(CreateSubscriptionStmt *stmt) +CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) { Relation rel; ObjectAddress myself; @@ -244,6 +245,24 @@ CreateSubscription(CreateSubscriptionStmt *stmt) bool create_slot; List *publications; + /* + * Parse and check options. + * Connection and publication should not be specified here. + */ + parse_subscription_options(stmt->options, NULL, NULL, + &enabled_given, &enabled, + &create_slot, NULL, &slotname); + + /* + * Since creating replication slot is not transactional, it + * leaves created replication slot even when the transaction + * rollbacks, while there is no corresponding entry in + * pg_subscription. So we cannot run CREATE SUBSCRIPTION inside + * a user transaction block if creating replication slot. + */ + if (create_slot) + PreventTransactionChain(isTopLevel, "CREATE SUBSCRIPTION CREATE SLOT"); + if (!superuser()) ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), @@ -262,13 +281,6 @@ CreateSubscription(CreateSubscriptionStmt *stmt) 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, NULL, &slotname); if (slotname == NULL) slotname = stmt->subname; @@ -447,7 +459,7 @@ AlterSubscription(AlterSubscriptionStmt *stmt) * Drop a subscription */ void -DropSubscription(DropSubscriptionStmt *stmt) +DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) { Relation rel; ObjectAddress myself; @@ -473,6 +485,23 @@ DropSubscription(DropSubscriptionStmt *stmt) NULL, NULL, NULL, &drop_slot, NULL); + /* + * Parse and check options. Other than drop slot option + * should not be specified here. + */ + parse_subscription_options(stmt->options, NULL, NULL, + NULL, NULL, NULL, &drop_slot, + NULL); + + /* + * Since dropping replication slot is not transactional, it drops + * the replication slot even when the transaction rollbacks. + * So we cannot run DROP SUBSCRIPTION inside a user transaction + * block if dropping replication slot. + */ + if (drop_slot) + PreventTransactionChain(isTopLevel, "DROP SUBSCRIPTION DROP SLOT"); + rel = heap_open(SubscriptionRelationId, RowExclusiveLock); tup = SearchSysCache2(SUBSCRIPTIONNAME, MyDatabaseId, diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 3bc0ae5..20b5273 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -1609,7 +1609,8 @@ ProcessUtilitySlow(ParseState *pstate, break; case T_CreateSubscriptionStmt: - address = CreateSubscription((CreateSubscriptionStmt *) parsetree); + address = CreateSubscription((CreateSubscriptionStmt *) parsetree, + isTopLevel); break; case T_AlterSubscriptionStmt: @@ -1617,7 +1618,7 @@ ProcessUtilitySlow(ParseState *pstate, break; case T_DropSubscriptionStmt: - DropSubscription((DropSubscriptionStmt *) parsetree); + DropSubscription((DropSubscriptionStmt *) parsetree, isTopLevel); /* no commands stashed for DROP */ commandCollected = true; break; diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h index 127696c..1765879 100644 --- a/src/include/commands/subscriptioncmds.h +++ b/src/include/commands/subscriptioncmds.h @@ -18,9 +18,10 @@ #include "catalog/objectaddress.h" #include "nodes/parsenodes.h" -extern ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt); +extern ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, + bool isTopLevel); extern ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt); -extern void DropSubscription(DropSubscriptionStmt *stmt); +extern void DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel); extern ObjectAddress AlterSubscriptionOwner(const char *name, Oid newOwnerId); extern void AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId); diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 340180e..9c9dba0 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -14,6 +14,11 @@ ERROR: syntax error at or near "PUBLICATION" LINE 1: CREATE SUBSCRIPTION testsub PUBLICATION foo; ^ set client_min_messages to error; +-- fail - cannot CREATE SUBSCRIPTION CREATE SLOT inside transaction block +BEGIN; +CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub WITH (CREATE SLOT); +ERROR: CREATE SUBSCRIPTION CREATE SLOT cannot run inside a transaction block +COMMIT; CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub; ERROR: invalid connection string syntax: missing "=" after "testconn" in connection info string @@ -61,6 +66,13 @@ ALTER SUBSCRIPTION testsub DISABLE; (1 row) COMMIT; +-- fail - connot DROP SUBSCRIPTION DROP SLOT inside transaction block +BEGIN; +DROP SUBSCRIPTION testsub WITH (DROP SLOT); +ERROR: DROP SUBSCRIPTION DROP SLOT cannot run inside a transaction block +COMMIT; +BEGIN; DROP SUBSCRIPTION testsub WITH (NODROP SLOT); +COMMIT; RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user; diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index e325b0a..5775398 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -12,6 +12,11 @@ CREATE SUBSCRIPTION testsub CONNECTION 'foo'; CREATE SUBSCRIPTION testsub PUBLICATION foo; set client_min_messages to error; +-- fail - cannot CREATE SUBSCRIPTION CREATE SLOT inside transaction block +BEGIN; +CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub WITH (CREATE SLOT); +COMMIT; + CREATE SUBSCRIPTION testsub CONNECTION 'testconn' PUBLICATION testpub; CREATE SUBSCRIPTION testsub CONNECTION 'dbname=doesnotexist' PUBLICATION testpub WITH (DISABLED, NOCREATE SLOT); reset client_min_messages; @@ -38,7 +43,14 @@ ALTER SUBSCRIPTION testsub DISABLE; COMMIT; +-- fail - connot DROP SUBSCRIPTION DROP SLOT inside transaction block +BEGIN; +DROP SUBSCRIPTION testsub WITH (DROP SLOT); +COMMIT; + +BEGIN; DROP SUBSCRIPTION testsub WITH (NODROP SLOT); +COMMIT; RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user;