From 4407334edd1a7276cbc0fab449912c879552971f Mon Sep 17 00:00:00 2001
From: Japin Li <japinli@hotmail.com>
Date: Tue, 26 Jan 2021 11:51:49 +0800
Subject: [PATCH v1] Add ALTER SUBSCRIPTION...ADD/DROP PUBLICATION...

---
 doc/src/sgml/ref/alter_subscription.sgml |  68 ++++++++++
 src/backend/commands/subscriptioncmds.c  | 156 +++++++++++++++++++++++
 src/backend/parser/gram.y                |  20 +++
 src/bin/psql/tab-complete.c              |   2 +-
 src/include/nodes/parsenodes.h           |   2 +
 5 files changed, 247 insertions(+), 1 deletion(-)

diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index db5e59f707..97c427e0f4 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,72 @@ 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>.
+     </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>.
+     </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>REFRESH PUBLICATION</literal></term>
     <listitem>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 082f7855b8..c014a227e9 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -857,6 +857,162 @@ AlterSubscription(AlterSubscriptionStmt *stmt)
 				break;
 			}
 
+		case ALTER_SUBSCRIPTION_ADD_PUBLICATION:
+			{
+				int			i;
+				int			noldsubpublications;
+				bool		copy_data;
+				bool		refresh;
+				List	   *publications = NIL;
+				Datum	   *oldsubpublications;
+				ArrayType  *array;
+
+				/* deconstruct the publications */
+				heap_deform_tuple(tup, RelationGetDescr(rel), values, nulls);
+				array = DatumGetArrayTypeP(values[Anum_pg_subscription_subpublications - 1]);
+				deconstruct_array(array, TEXTOID, -1, false, TYPALIGN_INT,
+								  &oldsubpublications, NULL, &noldsubpublications);
+
+				publications = list_copy(stmt->publication);
+				for (i = 0; i < noldsubpublications; i++)
+				{
+					char		*oldname = VARDATA(oldsubpublications[i]);
+					ListCell	*cell;
+
+					foreach(cell, stmt->publication)
+					{
+						char	*name = strVal(lfirst(cell));
+
+						if (strcmp(name, oldname) == 0)
+							ereport(ERROR,
+									(errcode(ERRCODE_SYNTAX_ERROR),
+									 errmsg("publication name \"%s\" is already in subscription",
+											name)));
+					}
+
+					publications = lappend(publications, makeString(oldname));
+				}
+
+				parse_subscription_options(stmt->options,
+										   NULL,	/* no "connect" */
+										   NULL, NULL,	/* no "enabled" */
+										   NULL,	/* no "create_slot" */
+										   NULL, NULL,	/* no "slot_name" */
+										   &copy_data,
+										   NULL,	/* no "synchronous_commit" */
+										   &refresh,
+										   NULL, NULL,	/* no "binary" */
+										   NULL, NULL); /* no "streaming" */
+				values[Anum_pg_subscription_subpublications - 1] =
+					publicationListToArray(publications);
+				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).")));
+
+					/* Make sure refresh sees the new list of publications. */
+					sub->publications = stmt->publication;
+
+					AlterSubscription_refresh(sub, copy_data);
+				}
+
+				break;
+			}
+
+		case ALTER_SUBSCRIPTION_DROP_PUBLICATION:
+			{
+				int			i;
+				int			noldsubpublications;
+				bool		copy_data;
+				bool		refresh;
+				List	   *publications = NIL;
+				ListCell   *cell;
+				Datum	   *oldsubpublications;
+				ArrayType  *array;
+
+				/* deconstruct the publications */
+				heap_deform_tuple(tup, RelationGetDescr(rel), values, nulls);
+				array = DatumGetArrayTypeP(values[Anum_pg_subscription_subpublications - 1]);
+				deconstruct_array(array, TEXTOID, -1, false, TYPALIGN_INT,
+								  &oldsubpublications, NULL, &noldsubpublications);
+
+				for (i = 0; i < noldsubpublications; i++)
+				{
+					char	*name = VARDATA(oldsubpublications[i]);
+
+					publications = lappend(publications, makeString(name));
+				}
+
+				foreach(cell, stmt->publication)
+				{
+					char		*name = strVal(lfirst(cell));
+					ListCell	*lc = NULL;
+
+					foreach(lc, publications)
+					{
+						char	*oldname = strVal(lfirst(lc));
+
+						if (strcmp(name, oldname) == 0)
+						{
+							publications = list_delete_cell(publications, lc);
+							break;
+						}
+					}
+
+					if (lc == NULL)
+						ereport(ERROR,
+								(errcode(ERRCODE_SYNTAX_ERROR),
+								 errmsg("publication name \"%s\" do not in subscription",
+										name)));
+				}
+
+				if (publications == NIL)
+					ereport(ERROR,
+							(errcode(ERRCODE_SYNTAX_ERROR),
+							 errmsg("subscription must be have at least one publication")));
+
+				parse_subscription_options(stmt->options,
+										   NULL,	/* no "connect" */
+										   NULL, NULL,	/* no "enabled" */
+										   NULL,	/* no "create_slot" */
+										   NULL, NULL,	/* no "slot_name" */
+										   &copy_data,
+										   NULL,	/* no "synchronous_commit" */
+										   &refresh,
+										   NULL, NULL,	/* no "binary" */
+										   NULL, NULL); /* no "streaming" */
+				values[Anum_pg_subscription_subpublications - 1] =
+					publicationListToArray(publications);
+				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).")));
+
+					/* Make sure refresh sees the new list of publications. */
+					sub->publications = stmt->publication;
+
+					AlterSubscription_refresh(sub, copy_data);
+				}
+
+				break;
+			}
+
 		case ALTER_SUBSCRIPTION_REFRESH:
 			{
 				bool		copy_data;
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/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 17f7265038..49b6b96428 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1634,7 +1634,7 @@ 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"))
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

