On Thu, 28 Jan 2021 at 12:22, Bharath Rupireddy
<bharath.rupireddyforpostg...@gmail.com> wrote:
> On Wed, Jan 27, 2021 at 7:35 PM Li Japin <japi...@hotmail.com> wrote:
>> > I don't see any problem if ALTER SUBSCRIPTION ... ADD PUBLICATION with
>> > refresh true refreshes only the newly added publications, because what
>> > we do in AlterSubscription_refresh() is that we fetch the tables
>> > associated with the publications from the publisher, compare them with
>> > the previously fetched tables from that publication and add the new
>> > tables or remove the table that don't exist in that publication
>> > anymore.
>> >
>> > For ALTER SUBSCRIPTION ... DROP PUBLICATION, also we can do the same
>> > thing i.e. refreshes only the dropped publications.
>> >
>> > Thoughts?
>>
>> Agreed. We just only need to refresh the added/dropped publications.
>> Furthermore, for publications that will be dropped, we do not need the
>> “copy_data” option, right?
>
> I think you are right. The copy_data option doesn't make sense for
> ALTER SUBSCRIPTION ... DROP PUBLICATION, maybe we should throw an
> error if the user specifies it. Of course, we need that option for
> ALTER SUBSCRIPTION ... ADD PUBLICATION.
>
Thanks for your confirm. Attached v3 patch fix it.
* v3-0001
Only refresh the publications that will be added/dropped, also remove
"copy_data"
option from DROP PUBLICATION.
* v3-0002
Add a new testcase for DROP PUBLICATION WITH (copy_data).
* v3-0003
Remove the reference of REFRESH PUBLICATION in DROP PUBLICATION.
* v3-0004
Do not change.
Attaching v3 patches, please consider these for further review.
--
Regrads,
Japin Li.
ChengDu WenWu Information Technology Co.,Ltd.
>From bff45768b066ccf94de6dae9f687cf54212900b1 Mon Sep 17 00:00:00 2001
From: Japin Li <japi...@hotmail.com>
Date: Tue, 26 Jan 2021 15:43:52 +0800
Subject: [PATCH v3 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 | 121 ++++++++++++++++++++++++
src/backend/parser/gram.y | 20 ++++
src/include/nodes/parsenodes.h | 2 +
3 files changed, 143 insertions(+)
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 082f7855b8..88fa7f1b3f 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -46,6 +46,8 @@
#include "utils/syscache.h"
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
+static List *merge_subpublications(HeapTuple tuple, TupleDesc tupledesc,
+ List *publications, bool addpub);
/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
@@ -857,6 +859,51 @@ AlterSubscription(AlterSubscriptionStmt *stmt)
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_subpublications(tup, RelationGetDescr(rel),
+ 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,
+ 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).")));
+
+ /* 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;
@@ -1278,3 +1325,77 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
return tablelist;
}
+
+/*
+ * Merge current subscription's publications and user specified publications
+ * by ADD/DROP PUBLICATIONS.
+ *
+ * If isadd == true, we will add the list of publications into current
+ * subscription's publications. Otherwise, we will delete the list of
+ * publications from current subscription's publications.
+ */
+static List *
+merge_subpublications(HeapTuple tuple, TupleDesc tupledesc,
+ List *newpublist, bool isadd)
+{
+ int i;
+ int npublications;
+ Datum *publications;
+ bool nulls[Natts_pg_subscription];
+ Datum values[Natts_pg_subscription];
+ List *publist = NIL;
+ ListCell *lc;
+ ArrayType *array;
+
+ /* deconstruct the subpublications */
+ heap_deform_tuple(tuple, tupledesc, values, nulls);
+ array = DatumGetArrayTypeP(values[Anum_pg_subscription_subpublications - 1]);
+ deconstruct_array(array, TEXTOID, -1, false, TYPALIGN_INT,
+ &publications, NULL, &npublications);
+
+ for (i = 0; i < npublications; i++)
+ publist = lappend(publist,
+ makeString(TextDatumGetCString((publications[i]))));
+
+ foreach(lc, newpublist)
+ {
+ char *name = strVal(lfirst(lc));
+ ListCell *cell = NULL;
+
+ foreach(cell, publist)
+ {
+ char *pubname = strVal(lfirst(cell));
+
+ if (strcmp(name, pubname) == 0)
+ {
+ if (isadd)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("publication name \"%s\" is already in subscription",
+ name)));
+ }
+ else
+ {
+ publist = list_delete_cell(publist, cell);
+ break;
+ }
+ }
+ }
+
+ if (isadd)
+ publist = lappend(publist, makeString(name));
+ else if (cell == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("publication name \"%s\" do not in subscription",
+ name)));
+ }
+
+ if (publist == NIL)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("subscription must contain at least one publication")));
+
+ return publist;
+}
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 7574d545e0..d20e513518 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9615,6 +9615,26 @@ AlterSubscriptionStmt:
n->options = $7;
$$ = (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 ENABLE_P
{
AlterSubscriptionStmt *n =
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index dc2bb40926..9148ca9888 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3553,6 +3553,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.30.0
>From 0e996b62bed021fd5eed369eb85ecb9a8a5b5373 Mon Sep 17 00:00:00 2001
From: Japin Li <japi...@hotmail.com>
Date: Tue, 26 Jan 2021 17:43:11 +0800
Subject: [PATCH v3 2/4] Test ALTER SUBSCRIPTION ... ADD/DROP PUBLICATION
---
src/test/regress/expected/subscription.out | 30 ++++++++++++++++++++++
src/test/regress/sql/subscription.sql | 22 ++++++++++++++++
2 files changed, 52 insertions(+)
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 2fa9bce66a..2b83beed11 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -200,6 +200,36 @@ 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 name "testpub" is already in subscription
+-- ok - add two publications into subscription
+ALTER SUBSCRIPTION regress_testsub ADD 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,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 publications do not in subscription
+ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub3 WITH (refresh = false);
+ERROR: publication name "testpub3" do not in 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;
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 14fa0b247e..a0178cfb7e 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -145,6 +145,28 @@ 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);
+
+\dRs+
+
+-- fail - all publications are deleted
+ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub, testpub1, testpub2 WITH (refresh = false);
+
+-- fail - the deleted publications do not 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;
RESET SESSION AUTHORIZATION;
--
2.30.0
>From c6bf33b6c0a078b0996e09b704997032b810001b Mon Sep 17 00:00:00 2001
From: Japin Li <japi...@hotmail.com>
Date: Tue, 26 Jan 2021 17:54:44 +0800
Subject: [PATCH v3 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 db5e59f707..e6e81ce7a3 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
@@ -107,6 +109,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.30.0
>From 3366cb9329752b4e193852990534d35e5ecd4f79 Mon Sep 17 00:00:00 2001
From: Japin Li <japi...@hotmail.com>
Date: Tue, 26 Jan 2021 18:25:10 +0800
Subject: [PATCH v3 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 17f7265038..dd178c48e2 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1634,7 +1634,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"))
@@ -1658,6 +1659,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.30.0
