From 8d1c9e4fd21b14d0f5ca9c10e89d706d504d7295 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Sat, 3 Apr 2021 10:16:44 +0530 Subject: [PATCH v9] Introduce a new syntax to add/drop publications At present, if we want to update publications in subscription, we can use SET PUBLICATION, however, it requires supply all publications that exists and the new publications if we want to add new publications, it's inconvenient. The new syntax only supply the new publications. When the refresh is true, it only refresh the new publications. --- doc/src/sgml/ref/alter_subscription.sgml | 65 +++++++++ src/backend/commands/subscriptioncmds.c | 162 ++++++++++++++++++--- src/backend/parser/gram.y | 20 +++ src/bin/psql/tab-complete.c | 11 +- src/include/nodes/parsenodes.h | 2 + src/test/regress/expected/subscription.out | 39 +++++ src/test/regress/sql/subscription.sql | 31 ++++ 7 files changed, 305 insertions(+), 25 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 5aed269435..1b3fd3a391 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -23,6 +23,8 @@ PostgreSQL documentation ALTER SUBSCRIPTION name CONNECTION 'conninfo' ALTER SUBSCRIPTION name SET PUBLICATION publication_name [, ...] [ WITH ( set_publication_option [= value] [, ... ] ) ] +ALTER SUBSCRIPTION name ADD PUBLICATION publication_name [, ...] [ WITH ( add_publication_option [= value] [, ... ] ) ] +ALTER SUBSCRIPTION name DROP PUBLICATION publication_name [, ...] [ WITH ( drop_publication_option [= value] [, ... ] ) ] ALTER SUBSCRIPTION name REFRESH PUBLICATION [ WITH ( refresh_option [= value] [, ... ] ) ] ALTER SUBSCRIPTION name ENABLE ALTER SUBSCRIPTION name DISABLE @@ -125,6 +127,69 @@ ALTER SUBSCRIPTION name RENAME TO < + + ADD PUBLICATION publication_name + + + Add list of publications to subscription. See + for more information. + By default this command will also act like REFRESH + PUBLICATION, except it only affect on added publications. + + + + add_publication_option specifies additional + options for this operation. The supported options are: + + + + refresh (boolean) + + + When false, the command will not try to refresh table information. + REFRESH PUBLICATION should then be executed separately. + The default is true. + + + + + + Additionally, refresh options as described + under REFRESH PUBLICATION may be specified. + + + + + + DROP PUBLICATION publication_name + + + Drop list of publications from subscription. See + for more information. + By default this command will also act like REFRESH + PUBLICATION, except it only affect on dropped publications. + + + + drop_publication_option specifies additional + options for this operation. The supported options are: + + + + refresh (boolean) + + + When false, the command will not try to refresh table information. + REFRESH PUBLICATION should then be executed separately. + The default is true. + + + + + + + + REFRESH PUBLICATION diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 5282b79735..7d3c1a1e28 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -47,6 +47,8 @@ #include "utils/syscache.h" static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); +static void check_duplicates_in_publist(List *publist, Datum *datums); +static List *merge_publications(List *oldpublist, List *newpublist, bool addpub); static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); @@ -293,8 +295,6 @@ publicationListToArray(List *publist) { ArrayType *arr; Datum *datums; - int j = 0; - ListCell *cell; MemoryContext memcxt; MemoryContext oldcxt; @@ -306,28 +306,7 @@ publicationListToArray(List *publist) 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); - } + check_duplicates_in_publist(publist, datums); MemoryContextSwitchTo(oldcxt); @@ -964,6 +943,53 @@ AlterSubscription(AlterSubscriptionStmt *stmt, bool isTopLevel) break; } + case ALTER_SUBSCRIPTION_ADD_PUBLICATION: + case ALTER_SUBSCRIPTION_DROP_PUBLICATION: + { + bool copy_data = false; + bool isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION; + bool refresh; + List *publist = NULL; + + publist = merge_publications(sub->publications, stmt->publication, isadd); + + parse_subscription_options(stmt->options, + NULL, /* no "connect" */ + NULL, NULL, /* no "enabled" */ + NULL, /* no "create_slot" */ + NULL, NULL, /* no "slot_name" */ + isadd ? ©_data : NULL, /* for drop, no "copy_data" */ + NULL, /* no "synchronous_commit" */ + &refresh, + NULL, NULL, /* no "binary" */ + NULL, NULL); /* no "streaming" */ + + values[Anum_pg_subscription_subpublications - 1] = + publicationListToArray(publist); + 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)."))); + + PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION with refresh"); + + /* Only refresh the added/dropped list of publications. */ + sub->publications = stmt->publication; + + AlterSubscription_refresh(sub, copy_data); + } + + break; + } + case ALTER_SUBSCRIPTION_REFRESH: { bool copy_data; @@ -1548,3 +1574,91 @@ ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err) errhint("Use %s to disassociate the subscription from the slot.", "ALTER SUBSCRIPTION ... SET (slot_name = NONE)"))); } + +/* + * Check for duplicates in the given list of publications and error out if + * found one. + */ +static void +check_duplicates_in_publist(List *publist, Datum *datums) +{ + ListCell *cell; + int j = 0; + + foreach(cell, publist) + { + char *name = strVal(lfirst(cell)); + ListCell *pcell; + + 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))); + } + + if (datums) + datums[j++] = CStringGetTextDatum(name); + } +} + +/* + * Merge current subscription's publications and user specified publications + * by ADD/DROP PUBLICATIONS. + * + * If addpub is true, we will add the list of publications into oldpublist. + * Otherwise, we will delete the list of publications from oldpublist. + */ +static List * +merge_publications(List *oldpublist, List *newpublist, bool addpub) +{ + ListCell *lc; + + check_duplicates_in_publist(newpublist, NULL); + + foreach(lc, newpublist) + { + char *name = strVal(lfirst(lc)); + ListCell *cell; + + foreach(cell, oldpublist) + { + char *pubname = strVal(lfirst(cell)); + + if (strcmp(name, pubname) == 0) + { + if (addpub) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("publication \"%s\" is already present in the subscription", + name))); + else + oldpublist = list_delete_cell(oldpublist, cell); + + break; + } + } + + if (addpub && !cell) + oldpublist = lappend(oldpublist, makeString(name)); + else if (!addpub && !cell) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("publication \"%s\" doesn't exist in the subscription", + name))); + } + + if (!oldpublist) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("subscription must contain at least one publication"))); + + return oldpublist; +} diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 8b1bad0d79..0b1c28f25e 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -9687,6 +9687,26 @@ AlterSubscriptionStmt: n->options = $6; $$ = (Node *)n; } + | ALTER SUBSCRIPTION name ADD_P PUBLICATION name_list opt_definition + { + AlterSubscriptionStmt *n = + makeNode(AlterSubscriptionStmt); + n->kind = ALTER_SUBSCRIPTION_ADD_PUBLICATION; + n->subname = $3; + n->publication = $6; + n->options = $7; + $$ = (Node *)n; + } + | ALTER SUBSCRIPTION name DROP PUBLICATION name_list opt_definition + { + AlterSubscriptionStmt *n = + makeNode(AlterSubscriptionStmt); + n->kind = ALTER_SUBSCRIPTION_DROP_PUBLICATION; + n->subname = $3; + n->publication = $6; + n->options = $7; + $$ = (Node *)n; + } | ALTER SUBSCRIPTION name SET PUBLICATION name_list opt_definition { AlterSubscriptionStmt *n = diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index a053bc1e45..6f16720102 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1652,7 +1652,8 @@ psql_completion(const char *text, int start, int end) /* ALTER SUBSCRIPTION */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny)) COMPLETE_WITH("CONNECTION", "ENABLE", "DISABLE", "OWNER TO", - "RENAME TO", "REFRESH PUBLICATION", "SET"); + "RENAME TO", "REFRESH PUBLICATION", "SET", + "ADD PUBLICATION", "DROP PUBLICATION"); /* ALTER SUBSCRIPTION REFRESH PUBLICATION */ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("REFRESH", "PUBLICATION")) @@ -1676,6 +1677,14 @@ psql_completion(const char *text, int start, int end) else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "PUBLICATION", MatchAny)) COMPLETE_WITH("WITH ("); + /* ALTER SUBSCRIPTION ADD PUBLICATION */ + else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && + TailMatches("ADD", "PUBLICATION", MatchAny)) + COMPLETE_WITH("WITH ("); + /* ALTER SUBSCRIPTION DROP PUBLICATION */ + else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && + TailMatches("DROP", "PUBLICATION", MatchAny)) + COMPLETE_WITH("WITH ("); /* ALTER SUBSCRIPTION SET PUBLICATION WITH ( */ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "PUBLICATION", MatchAny, "WITH", "(")) diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 7960cfe1a8..8bda755392 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3619,6 +3619,8 @@ typedef enum AlterSubscriptionType ALTER_SUBSCRIPTION_OPTIONS, ALTER_SUBSCRIPTION_CONNECTION, ALTER_SUBSCRIPTION_PUBLICATION, + ALTER_SUBSCRIPTION_ADD_PUBLICATION, + ALTER_SUBSCRIPTION_DROP_PUBLICATION, ALTER_SUBSCRIPTION_REFRESH, ALTER_SUBSCRIPTION_ENABLED } AlterSubscriptionType; diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 14a430221d..f3b640c632 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -200,6 +200,45 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); regress_testsub | regress_subscription_user | f | {testpub} | f | f | off | dbname=regress_doesnotexist (1 row) +-- fail - publication already exists +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub WITH (refresh = false); +ERROR: publication "testpub" is already present in the subscription +-- fail - publication used more than once +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub1 WITH (refresh = false); +ERROR: publication name "testpub1" used more than once +-- ok - add two publications into subscription +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); +-- fail - publications already exist +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); +ERROR: publication "testpub1" is already present in the subscription +\dRs+ + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Synchronous commit | Conninfo +-----------------+---------------------------+---------+-----------------------------+--------+-----------+--------------------+----------------------------- + regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | f | off | dbname=regress_doesnotexist +(1 row) + +-- fail - publication used more then once +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub1 WITH (refresh = false); +ERROR: publication name "testpub1" used more than once +-- fail - all publications are deleted +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub, testpub1, testpub2 WITH (refresh = false); +ERROR: subscription must contain at least one publication +-- fail - publication does not exist in subscription +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub3 WITH (refresh = false); +ERROR: publication "testpub3" doesn't exist in the subscription +-- fail - do not support copy_data option +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1 WITH (refresh = false, copy_data = true); +ERROR: unrecognized subscription parameter: "copy_data" +-- ok - delete publications +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); +\dRs+ + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Synchronous commit | Conninfo +-----------------+---------------------------+---------+-------------+--------+-----------+--------------------+----------------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | f | off | dbname=regress_doesnotexist +(1 row) + DROP SUBSCRIPTION regress_testsub; CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION mypub WITH (connect = false, create_slot = false, copy_data = false); diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 81e65e5e64..308c098c14 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -145,6 +145,37 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ +-- fail - publication already exists +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub WITH (refresh = false); + +-- fail - publication used more than once +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub1 WITH (refresh = false); + +-- ok - add two publications into subscription +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); + +-- fail - publications already exist +ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); + +\dRs+ + +-- fail - publication used more then once +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub1 WITH (refresh = false); + +-- fail - all publications are deleted +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub, testpub1, testpub2 WITH (refresh = false); + +-- fail - publication does not exist in subscription +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub3 WITH (refresh = false); + +-- fail - do not support copy_data option +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1 WITH (refresh = false, copy_data = true); + +-- ok - delete publications +ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); + +\dRs+ + DROP SUBSCRIPTION regress_testsub; CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION mypub -- 2.25.1