On Sun, 07 Mar 2021 at 19:43, Bharath Rupireddy
<bharath.rupireddyforpostg...@gmail.com> wrote:
> I'm reading through the v6 patches again, here are some comments.
>
Thanks for your review again.
> 1) IMO, we can merge 0001 into 0002
> 2) A typo, it's "current" not "ccurrent" - + * Merge ccurrent
> subscription's publications and user specified publications
Fixed.
> 3) In merge_subpublications, do we need to error out or do something
> instead of Assert(!isnull); as in the release build we don't reach
> assert. So, if at all catalogue search returns a null tuple, we don't
> surprise users.
> 4) Can we have a better name for the function merge_subpublications
> say merge_publications (because it's a local function to
> subscriptioncmds.c we don't need "sub" in function name) or any other
> better name?
Rename merge_subpublications to merge_publications as you suggested.
> 5) Instead of doing catalogue look up for the subscriber publications
> in merge_subpublications, why can't we pass in the list from sub =
> GetSubscription(subid, false); (being called in AlterSubscription)
> ---> sub->publications. Do you see any problems in doing so? If done
> that, we can discard the 0001 patch and comments (1) and (3) becomes
> irrelevant.
Thank you point out this. Fixed it in v7 patch set.
Please consider the v7 patch for futher review.
--
Regrads,
Japin Li.
ChengDu WenWu Information Technology Co.,Ltd.
>From a6ec682b2badb1d6dcb5e41f27bd3d7851353be3 Mon Sep 17 00:00:00 2001
From: Japin Li <japi...@hotmail.com>
Date: Sun, 7 Mar 2021 12:56:55 +0000
Subject: [PATCH v7 1/4] 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.
---
src/backend/commands/subscriptioncmds.c | 132 ++++++++++++++++++++++++
src/backend/parser/gram.y | 20 ++++
src/include/nodes/parsenodes.h | 2 +
3 files changed, 154 insertions(+)
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index bfd3514546..bf1e579e6f 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -47,6 +47,7 @@
#include "utils/syscache.h"
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
+static List *merge_publications(List *oldpublist, List *newpublist, bool addpub);
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
@@ -964,6 +965,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 = NIL;
+
+ 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;
@@ -1551,3 +1599,87 @@ 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)")));
}
+
+/*
+ * 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)
+{
+ StringInfoData errstr;
+ int errstrcnt = 0;
+ ListCell *lc;
+
+ initStringInfo(&errstr);
+
+ foreach(lc, newpublist)
+ {
+ char *name = strVal(lfirst(lc));
+ ListCell *cell = NULL;
+
+ foreach(cell, oldpublist)
+ {
+ char *pubname = strVal(lfirst(cell));
+
+ if (strcmp(name, pubname) == 0)
+ {
+ if (addpub)
+ {
+ errstrcnt++;
+
+ if (errstrcnt == 1)
+ appendStringInfo(&errstr, _("\"%s\""), name);
+ else
+ appendStringInfo(&errstr, _(", \"%s\""), name);
+ }
+ else
+ oldpublist = list_delete_cell(oldpublist, cell);
+
+ break;
+ }
+ }
+
+ if (addpub && cell == NULL)
+ oldpublist = lappend(oldpublist, makeString(name));
+ else if (!addpub && cell == NULL)
+ {
+ errstrcnt++;
+
+ if (errstrcnt == 1)
+ appendStringInfo(&errstr, _("\"%s\""), name);
+ else
+ appendStringInfo(&errstr, _(", \"%s\""), name);
+ }
+ }
+
+ if (errstrcnt >= 1)
+ {
+ if (addpub)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg_plural("publication %s is already present in the subscription",
+ "publications %s are already present in the subscription",
+ errstrcnt, errstr.data)));
+ }
+ else
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg_plural("publication %s doesn't exist in the subscription",
+ "publications %s do not exist in the subscription",
+ errstrcnt, errstr.data)));
+ }
+ }
+
+ if (oldpublist == NIL)
+ 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 652be0b96d..6d75e82096 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9609,6 +9609,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/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 236832a2ca..e109607936 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3580,6 +3580,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;
--
2.25.1
>From a752fb0c118b748dda03c7983e3c8f16539d40f3 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupire...@enterprisedb.com>
Date: Tue, 16 Feb 2021 07:20:36 +0530
Subject: [PATCH v7 2/4] Test ALTER SUBSCRIPTION ... ADD/DROP PUBLICATION
---
src/test/regress/expected/subscription.out | 36 ++++++++++++++++++++++
src/test/regress/sql/subscription.sql | 28 +++++++++++++++++
2 files changed, 64 insertions(+)
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 14a430221d..d7df579069 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -200,6 +200,42 @@ 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
+-- 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: publications "testpub1", "testpub2" are 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 - 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 - the deleted publication does not in subscription
+ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub3 WITH (refresh = false);
+ERROR: publication "testpub3" doesn't exist in the subscription
+-- fail - the deleted publications do not in subscription
+ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub3, testpub4 WITH (refresh = false);
+ERROR: publications "testpub3", "testpub4" do not 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..746a589a8e 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -145,6 +145,34 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
+-- fail - publication already exists
+ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub 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 - all publications are deleted
+ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub, testpub1, testpub2 WITH (refresh = false);
+
+-- fail - the deleted publication does not in subscription
+ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub3 WITH (refresh = false);
+
+-- fail - the deleted publications do not in subscription
+ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub3, testpub4 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
>From 23e9b98332ba3e26635bc0baf42b94f4a14e0703 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupire...@enterprisedb.com>
Date: Tue, 16 Feb 2021 07:21:36 +0530
Subject: [PATCH v7 3/4] Add documentation for ALTER SUBSCRIPTION...ADD/DROP
PUBLICATION
---
doc/src/sgml/ref/alter_subscription.sgml | 65 ++++++++++++++++++++++++
1 file changed, 65 insertions(+)
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 0adf68ecca..aa181b94c5 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -23,6 +23,8 @@ PostgreSQL documentation
<synopsis>
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> CONNECTION '<replaceable>conninfo</replaceable>'
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> SET PUBLICATION <replaceable class="parameter">publication_name</replaceable> [, ...] [ WITH ( <replaceable class="parameter">set_publication_option</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
+ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> ADD PUBLICATION <replaceable class="parameter">publication_name</replaceable> [, ...] [ WITH ( <replaceable class="parameter">set_publication_option</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
+ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> DROP PUBLICATION <replaceable class="parameter">publication_name</replaceable> [, ...] [ WITH ( <replaceable class="parameter">set_publication_option</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> REFRESH PUBLICATION [ WITH ( <replaceable class="parameter">refresh_option</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> ENABLE
ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> DISABLE
@@ -125,6 +127,69 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
</listitem>
</varlistentry>
+ <varlistentry>
+ <term><literal>ADD PUBLICATION <replaceable class="parameter">publication_name</replaceable></literal></term>
+ <listitem>
+ <para>
+ Add list of publications to subscription. See
+ <xref linkend="sql-createsubscription"/> for more information.
+ By default this command will also act like <literal>REFRESH
+ PUBLICATION</literal>, except it only affect on added publications.
+ </para>
+
+ <para>
+ <replaceable>set_publication_option</replaceable> specifies additional
+ options for this operation. The supported options are:
+
+ <variablelist>
+ <varlistentry>
+ <term><literal>refresh</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ When false, the command will not try to refresh table information.
+ <literal>REFRESH PUBLICATION</literal> should then be executed separately.
+ The default is <literal>true</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+
+ Additionally, refresh options as described
+ under <literal>REFRESH PUBLICATION</literal> may be specified.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
+ <term><literal>DROP PUBLICATION <replaceable class="parameter">publication_name</replaceable></literal></term>
+ <listitem>
+ <para>
+ Drop list of publications from subscription. See
+ <xref linkend="sql-createsubscription"/> for more information.
+ By default this command will also act like <literal>REFRESH
+ PUBLICATION</literal>, except it only affect on dropped publications.
+ </para>
+
+ <para>
+ <replaceable>set_publication_option</replaceable> specifies additional
+ options for this operation. The supported options are:
+
+ <variablelist>
+ <varlistentry>
+ <term><literal>refresh</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ When false, the command will not try to refresh table information.
+ <literal>REFRESH PUBLICATION</literal> should then be executed separately.
+ The default is <literal>true</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><literal>REFRESH PUBLICATION</literal></term>
<listitem>
--
2.25.1
>From 45c25845e6e317aff62ec0813a54066370159c5c Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupire...@enterprisedb.com>
Date: Tue, 16 Feb 2021 07:22:41 +0530
Subject: [PATCH v7 4/4] Add tab-complete for ALTER SUBSCRIPTION...ADD/DROP
---
src/bin/psql/tab-complete.c | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 9f0208ac49..a16f1e7dfe 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 <name> */
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 <name> 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 <name> ADD PUBLICATION <name> */
+ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
+ TailMatches("ADD", "PUBLICATION", MatchAny))
+ COMPLETE_WITH("WITH (");
+ /* ALTER SUBSCRIPTION <name> DROP PUBLICATION <name> */
+ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
+ TailMatches("DROP", "PUBLICATION", MatchAny))
+ COMPLETE_WITH("WITH (");
/* ALTER SUBSCRIPTION <name> SET PUBLICATION <name> WITH ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
TailMatches("SET", "PUBLICATION", MatchAny, "WITH", "("))
--
2.25.1
