public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v1] Add ALTER SUBSCRIPTION...ADD/DROP PUBLICATION... 4+ messages / 4 participants [nested] [flat]
* [PATCH v1] Add ALTER SUBSCRIPTION...ADD/DROP PUBLICATION... @ 2021-01-26 03:51 Japin Li <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Japin Li @ 2021-01-26 03:51 UTC (permalink / raw) --- 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" */ + ©_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" */ + ©_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 --=-=-=-- ^ permalink raw reply [nested|flat] 4+ messages in thread
* pgsql: Harden new test case against force_parallel_mode = regress. @ 2023-03-02 22:47 Tom Lane <[email protected]> 2023-03-03 16:16 ` Re: pgsql: Harden new test case against force_parallel_mode = regress. Robert Haas <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: Tom Lane @ 2023-03-02 22:47 UTC (permalink / raw) To: [email protected] Harden new test case against force_parallel_mode = regress. Per buildfarm: worker processes can't see a role created in the current transaction. Branch ------ master Details ------- https://git.postgresql.org/pg/commitdiff/98a88bc2bcd60e41ca70e2f1e13eee827e23eefb Modified Files -------------- src/test/regress/expected/psql.out | 3 ++- src/test/regress/sql/psql.sql | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: pgsql: Harden new test case against force_parallel_mode = regress. 2023-03-02 22:47 pgsql: Harden new test case against force_parallel_mode = regress. Tom Lane <[email protected]> @ 2023-03-03 16:16 ` Robert Haas <[email protected]> 2023-03-03 16:38 ` Re: pgsql: Harden new test case against force_parallel_mode = regress. Matthias van de Meent <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: Robert Haas @ 2023-03-03 16:16 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: pgsql-hackers On Thu, Mar 2, 2023 at 5:47 PM Tom Lane <[email protected]> wrote: > Harden new test case against force_parallel_mode = regress. > > Per buildfarm: worker processes can't see a role created in > the current transaction. Now why would that happen? Surely the snapshot for each command is passed down from leader to worker, and the worker is not free to invent a snapshot from nothing. -- Robert Haas EDB: http://www.enterprisedb.com ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: pgsql: Harden new test case against force_parallel_mode = regress. 2023-03-02 22:47 pgsql: Harden new test case against force_parallel_mode = regress. Tom Lane <[email protected]> 2023-03-03 16:16 ` Re: pgsql: Harden new test case against force_parallel_mode = regress. Robert Haas <[email protected]> @ 2023-03-03 16:38 ` Matthias van de Meent <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Matthias van de Meent @ 2023-03-03 16:38 UTC (permalink / raw) To: Robert Haas <[email protected]>; +Cc: Tom Lane <[email protected]>; pgsql-hackers On Fri, 3 Mar 2023 at 17:16, Robert Haas <[email protected]> wrote: > > On Thu, Mar 2, 2023 at 5:47 PM Tom Lane <[email protected]> wrote: > > Harden new test case against force_parallel_mode = regress. > > > > Per buildfarm: worker processes can't see a role created in > > the current transaction. > > Now why would that happen? Surely the snapshot for each command is > passed down from leader to worker, and the worker is not free to > invent a snapshot from nothing. Probably because we nitialize which user and database to use in the backend before we load the parent process' snapshot: in ParallelWorkerMain (parallel.c, as of HEAD @ b6a0d469): /* Restore database connection. */ BackgroundWorkerInitializeConnectionByOid(fps->database_id, fps->authenticated_user_id, 0); [...] /* Crank up a transaction state appropriate to a parallel worker. */ tstatespace = shm_toc_lookup(toc, PARALLEL_KEY_TRANSACTION_STATE, false); StartParallelWorkerTransaction(tstatespace); /* Restore combo CID state. */ combocidspace = shm_toc_lookup(toc, PARALLEL_KEY_COMBO_CID, false); RestoreComboCIDState(combocidspace); -Matthias ^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2023-03-03 16:38 UTC | newest] Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-01-26 03:51 [PATCH v1] Add ALTER SUBSCRIPTION...ADD/DROP PUBLICATION... Japin Li <[email protected]> 2023-03-02 22:47 pgsql: Harden new test case against force_parallel_mode = regress. Tom Lane <[email protected]> 2023-03-03 16:16 ` Re: pgsql: Harden new test case against force_parallel_mode = regress. Robert Haas <[email protected]> 2023-03-03 16:38 ` Re: pgsql: Harden new test case against force_parallel_mode = regress. Matthias van de Meent <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox