On Mon, Apr 18, 2022 at 12:32 PM Amit Kapila <amit.kapil...@gmail.com> wrote:
>
> On Fri, Apr 15, 2022 at 1:26 AM Euler Taveira <eu...@eulerto.com> wrote:
> >
> > On Thu, Apr 14, 2022, at 10:47 AM, Peter Eisentraut wrote:
> >
> > On 12.04.22 08:23, vignesh C wrote:
> > > I have also included the implementation for skipping a few tables from
> > > all tables publication, the 0002 patch has the implementation for the
> > > same.
> > > This feature is helpful for use cases where the user wants to
> > > subscribe to all the changes except for the changes present in a few
> > > tables.
> > > Ex:
> > > CREATE PUBLICATION pub1 FOR ALL TABLES SKIP TABLE t1,t2;
> > > OR
> > > ALTER PUBLICATION pub1 ADD SKIP TABLE t1,t2;
> >
> > We have already allocated the "skip" terminology for skipping
> > transactions, which is a dynamic run-time action. We are also using the
> > term "skip" elsewhere to skip locked rows, which is similarly a run-time
> > action. I think it would be confusing to use the term SKIP for DDL
> > construction.
> >
> > I didn't like the SKIP choice too. We already have EXCEPT for IMPORT FOREIGN
> > SCHEMA and if I were to suggest a keyword, it would be EXCEPT.
> >
>
> +1 for EXCEPT.
Updated patch by changing the syntax to use EXCEPT instead of SKIP.
Regards,
Vignesh
From 289b5acfc33f70f488f45e2cb55714d20097ac4c Mon Sep 17 00:00:00 2001
From: Vigneshwaran C <vignes...@gmail.com>
Date: Wed, 20 Apr 2022 11:19:50 +0530
Subject: [PATCH v2] Skip publishing the tables specified in EXCEPT TABLE.
A new option "EXCEPT TABLE" in Create/Alter Publication allows
one or more tables to be excluded, publisher will exclude sending the data
of the tables present in the except table to the subscriber.
The new syntax allows specifying schemas. For example:
CREATE PUBLICATION pub1 FOR ALL TABLES EXCEPT TABLE t1,t2;
OR
ALTER PUBLICATION pub1 ADD EXCEPT TABLE t1,t2;
A new column prexcept is added to table "pg_publication_rel", to maintain
the relations that the user wants to exclude publishing through the publication.
Modified the output plugin (pgoutput) to exclude publishing the changes if the
relation is part of except table publication.
Updates pg_dump to identify and dump except table publications. Updates the \d
family of commands to display except table publications and \dRp+ variant will
now display associated except tables if any.
Bump catalog version.
---
doc/src/sgml/catalogs.sgml | 9 ++
doc/src/sgml/logical-replication.sgml | 8 +-
doc/src/sgml/ref/alter_publication.sgml | 14 ++-
doc/src/sgml/ref/create_publication.sgml | 29 ++++-
doc/src/sgml/ref/psql-ref.sgml | 5 +-
src/backend/catalog/pg_publication.c | 36 ++++--
src/backend/commands/publicationcmds.c | 106 +++++++++++-------
src/backend/commands/tablecmds.c | 4 +-
src/backend/parser/gram.y | 102 +++++++++++++++--
src/backend/replication/pgoutput/pgoutput.c | 25 ++---
src/backend/utils/cache/relcache.c | 17 ++-
src/bin/pg_dump/pg_dump.c | 35 ++++--
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/pg_dump/pg_dump_sort.c | 7 ++
src/bin/pg_dump/t/002_pg_dump.pl | 23 ++++
src/bin/psql/describe.c | 25 ++++-
src/bin/psql/tab-complete.c | 15 ++-
src/include/catalog/pg_publication.h | 7 +-
src/include/catalog/pg_publication_rel.h | 1 +
src/include/commands/publicationcmds.h | 4 +-
src/include/nodes/parsenodes.h | 2 +
src/test/regress/expected/publication.out | 81 ++++++++++++-
src/test/regress/sql/publication.sql | 40 ++++++-
.../t/033_rep_changes_except_table.pl | 97 ++++++++++++++++
24 files changed, 580 insertions(+), 113 deletions(-)
create mode 100644 src/test/subscription/t/033_rep_changes_except_table.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index a533a2153e..78e8c22a59 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -6426,6 +6426,15 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
if there is no publication qualifying condition.</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>prexcept</structfield> <type>bool</type>
+ </para>
+ <para>
+ True if the table must be excluded
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>prattrs</structfield> <type>int2vector</type>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index a0f9cecd01..36e943d305 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1160,10 +1160,10 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
</para>
<para>
- To add tables to a publication, the user must have ownership rights on the
- table. To add all tables in schema to a publication, the user must be a
- superuser. To create a publication that publishes all tables or all tables in
- schema automatically, the user must be a superuser.
+ To add tables or exclude tables to a publication, the user must have
+ ownership rights on the table. To add all tables in schema to a publication,
+ the user must be a superuser. To create a publication that publishes all
+ tables or all tables in schema automatically, the user must be a superuser.
</para>
<para>
diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml
index e2cce49471..2a8e4e041b 100644
--- a/doc/src/sgml/ref/alter_publication.sgml
+++ b/doc/src/sgml/ref/alter_publication.sgml
@@ -30,7 +30,7 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
<phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase>
- TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ... ]
+ [EXCEPT] TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ... ]
ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
</synopsis>
</refsynopsisdiv>
@@ -70,8 +70,8 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
<para>
You must own the publication to use <command>ALTER PUBLICATION</command>.
- Adding a table to a publication additionally requires owning that table.
- The <literal>ADD ALL TABLES IN SCHEMA</literal> and
+ Adding a table or excluding a table to a publication additionally requires
+ owning that table. The <literal>ADD ALL TABLES IN SCHEMA</literal> and
<literal>SET ALL TABLES IN SCHEMA</literal> to a publication requires the
invoking user to be a superuser. To alter the owner, you must also be a
direct or indirect member of the new owning role. The new owner must have
@@ -200,6 +200,14 @@ ALTER PUBLICATION sales_publication ADD ALL TABLES IN SCHEMA marketing, sales;
</programlisting>
</para>
+ <para>
+ Adding tables <structname>users</structname> and
+ <structname>departments</structname> that must be excluded from the
+ publication <structname>production_publication</structname>:
+<programlisting>
+ALTER PUBLICATION production_publication ADD EXCEPT TABLE users, departments production;
+</programlisting></para>
+
<para>
Add tables <structname>users</structname>,
<structname>departments</structname> and schema
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index 23d883c158..a0379fb285 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -22,7 +22,7 @@ PostgreSQL documentation
<refsynopsisdiv>
<synopsis>
CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
- [ FOR ALL TABLES
+ [ FOR ALL TABLES [EXCEPT TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ]]
| FOR <replaceable class="parameter">publication_object</replaceable> [, ... ] ]
[ WITH ( <replaceable class="parameter">publication_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
@@ -155,6 +155,24 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>EXCEPT TABLE</literal></term>
+ <listitem>
+ <para>
+ Marks the publication as one that excludes replicating changes for the
+ specified tables.
+ </para>
+
+ <para>
+ <literal>EXCEPT TABLE</literal> can be specified only for
+ <literal>FOR ALL TABLES</literal> publication. It is not supported for
+ <literal>FOR ALL TABLES IN SCHEMA </literal> publication and
+ <literal>FOR TABLE</literal> publication.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><literal>WITH ( <replaceable class="parameter">publication_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] )</literal></term>
<listitem>
@@ -338,6 +356,15 @@ CREATE PUBLICATION production_publication FOR TABLE users, departments, ALL TABL
CREATE PUBLICATION sales_publication FOR ALL TABLES IN SCHEMA marketing, sales;
</programlisting></para>
+ <para>
+ Create a publication that publishes all changes in all the tables except for
+ the changes of <structname>users</structname> and
+ <structname>departments</structname> table;
+<programlisting>
+CREATE PUBLICATION mypublication FOR ALL TABLE EXCEPT TABLE users, departments;
+</programlisting>
+ </para>
+
<para>
Create a publication that publishes all changes for table <structname>users</structname>,
but replicates only columns <structname>user_id</structname> and
diff --git a/doc/src/sgml/ref/psql-ref.sgml b/doc/src/sgml/ref/psql-ref.sgml
index 5fc6b1034a..4d123d1e5d 100644
--- a/doc/src/sgml/ref/psql-ref.sgml
+++ b/doc/src/sgml/ref/psql-ref.sgml
@@ -1868,8 +1868,9 @@ testdb=>
If <replaceable class="parameter">pattern</replaceable> is
specified, only those publications whose names match the pattern are
listed.
- If <literal>+</literal> is appended to the command name, the tables and
- schemas associated with each publication are shown as well.
+ If <literal>+</literal> is appended to the command name, the tables,
+ except tables and schemas associated with each publication are shown as
+ well.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 2631558ff1..e55ae4211a 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -290,7 +290,8 @@ GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
* ancestor is at the end of the list.
*/
Oid
-GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level)
+GetTopMostAncestorInPublication(Oid puboid, List *ancestors,
+ int *ancestor_level, bool puballtables)
{
ListCell *lc;
Oid topmost_relid = InvalidOid;
@@ -302,8 +303,9 @@ GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level
foreach(lc, ancestors)
{
Oid ancestor = lfirst_oid(lc);
- List *apubids = GetRelationPublications(ancestor);
+ List *apubids = GetRelationPublications(ancestor, false);
List *aschemaPubids = NIL;
+ List *aexceptpubids;
level++;
@@ -317,7 +319,9 @@ GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level
else
{
aschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor));
- if (list_member_oid(aschemaPubids, puboid))
+ aexceptpubids = GetRelationPublications(ancestor, true);
+ if (list_member_oid(aschemaPubids, puboid) ||
+ (puballtables && !list_member_oid(aexceptpubids, puboid)))
{
topmost_relid = ancestor;
@@ -396,6 +400,9 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
ObjectIdGetDatum(pubid);
values[Anum_pg_publication_rel_prrelid - 1] =
ObjectIdGetDatum(relid);
+ values[Anum_pg_publication_rel_prexcept - 1] =
+ BoolGetDatum(pri->except);
+
/* Add qualifications, if available */
if (pri->whereClause != NULL)
@@ -666,7 +673,7 @@ publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
/* Gets list of publication oids for a relation */
List *
-GetRelationPublications(Oid relid)
+GetRelationPublications(Oid relid, bool bexcept)
{
List *result = NIL;
CatCList *pubrellist;
@@ -680,7 +687,8 @@ GetRelationPublications(Oid relid)
HeapTuple tup = &pubrellist->members[i]->tuple;
Oid pubid = ((Form_pg_publication_rel) GETSTRUCT(tup))->prpubid;
- result = lappend_oid(result, pubid);
+ if (bexcept == ((Form_pg_publication_rel) GETSTRUCT(tup))->prexcept)
+ result = lappend_oid(result, pubid);
}
ReleaseSysCacheList(pubrellist);
@@ -779,7 +787,7 @@ GetAllTablesPublications(void)
* root partitioned tables.
*/
List *
-GetAllTablesPublicationRelations(bool pubviaroot)
+GetAllTablesPublicationRelations(Oid pubid, bool pubviaroot)
{
Relation classRel;
ScanKeyData key[1];
@@ -787,6 +795,13 @@ GetAllTablesPublicationRelations(bool pubviaroot)
HeapTuple tuple;
List *result = NIL;
+ /*
+ * pg_publication_rel and pg_publication_namespace will only have except
+ * tables in case of all tables publication, no need to pass except flag
+ * to get the relations.
+ */
+ List *exceptpubtablelist = GetPublicationRelations(pubid, PUBLICATION_PART_ALL);
+
classRel = table_open(RelationRelationId, AccessShareLock);
ScanKeyInit(&key[0],
@@ -802,7 +817,8 @@ GetAllTablesPublicationRelations(bool pubviaroot)
Oid relid = relForm->oid;
if (is_publishable_class(relid, relForm) &&
- !(relForm->relispartition && pubviaroot))
+ !(relForm->relispartition && pubviaroot) &&
+ !list_member_oid(exceptpubtablelist, relid))
result = lappend_oid(result, relid);
}
@@ -823,7 +839,8 @@ GetAllTablesPublicationRelations(bool pubviaroot)
Oid relid = relForm->oid;
if (is_publishable_class(relid, relForm) &&
- !relForm->relispartition)
+ !relForm->relispartition &&
+ !list_member_oid(exceptpubtablelist, relid))
result = lappend_oid(result, relid);
}
@@ -1107,7 +1124,8 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
*/
if (publication->alltables)
{
- tables = GetAllTablesPublicationRelations(publication->pubviaroot);
+ tables = GetAllTablesPublicationRelations(publication->oid,
+ publication->pubviaroot);
}
else
{
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 6df0e6670f..257e669e43 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -297,7 +297,7 @@ contain_invalid_rfcolumn_walker(Node *node, rf_context *context)
*/
bool
pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
- bool pubviaroot)
+ bool pubviaroot, bool puballtables)
{
HeapTuple rftuple;
Oid relid = RelationGetRelid(relation);
@@ -324,7 +324,8 @@ pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
if (pubviaroot && relation->rd_rel->relispartition)
{
publish_as_relid
- = GetTopMostAncestorInPublication(pubid, ancestors, NULL);
+ = GetTopMostAncestorInPublication(pubid, ancestors, NULL,
+ puballtables);
if (!OidIsValid(publish_as_relid))
publish_as_relid = relid;
@@ -373,7 +374,7 @@ pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
*/
bool
pub_collist_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
- bool pubviaroot)
+ bool pubviaroot, bool puballtables)
{
HeapTuple tuple;
Oid relid = RelationGetRelid(relation);
@@ -392,7 +393,7 @@ pub_collist_contains_invalid_column(Oid pubid, Relation relation, List *ancestor
*/
if (pubviaroot && relation->rd_rel->relispartition)
{
- publish_as_relid = GetTopMostAncestorInPublication(pubid, ancestors, NULL);
+ publish_as_relid = GetTopMostAncestorInPublication(pubid, ancestors, NULL, puballtables);
if (!OidIsValid(publish_as_relid))
publish_as_relid = relid;
@@ -835,54 +836,53 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
/* Make the changes visible. */
CommandCounterIncrement();
- /* Associate objects with the publication. */
- if (stmt->for_all_tables)
- {
- /* Invalidate relcache so that publication info is rebuilt. */
- CacheInvalidateRelcacheAll();
- }
- else
- {
- ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
- &schemaidlist);
+ ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
+ &schemaidlist);
- /* FOR ALL TABLES IN SCHEMA requires superuser */
- if (list_length(schemaidlist) > 0 && !superuser())
- ereport(ERROR,
- errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
- errmsg("must be superuser to create FOR ALL TABLES IN SCHEMA publication"));
+ /* FOR ALL TABLES IN SCHEMA requires superuser */
+ if (list_length(schemaidlist) > 0 && !superuser())
+ ereport(ERROR,
+ errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("must be superuser to create FOR ALL TABLES IN SCHEMA publication"));
- if (list_length(relations) > 0)
- {
- List *rels;
+ if (list_length(relations) > 0)
+ {
+ List *rels;
- rels = OpenTableList(relations);
- CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
- PUBLICATIONOBJ_TABLE);
+ rels = OpenTableList(relations);
+ CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
+ PUBLICATIONOBJ_TABLE);
- TransformPubWhereClauses(rels, pstate->p_sourcetext,
- publish_via_partition_root);
+ TransformPubWhereClauses(rels, pstate->p_sourcetext,
+ publish_via_partition_root);
- CheckPubRelationColumnList(rels, pstate->p_sourcetext,
- publish_via_partition_root);
+ CheckPubRelationColumnList(rels, pstate->p_sourcetext,
+ publish_via_partition_root);
- PublicationAddTables(puboid, rels, true, NULL);
- CloseTableList(rels);
- }
+ PublicationAddTables(puboid, rels, true, NULL);
+ CloseTableList(rels);
+ }
- if (list_length(schemaidlist) > 0)
- {
- /*
- * Schema lock is held until the publication is created to prevent
- * concurrent schema deletion.
- */
- LockSchemaList(schemaidlist);
- PublicationAddSchemas(puboid, schemaidlist, true, NULL);
- }
+ /* tables added through a schema */
+ if (list_length(schemaidlist) > 0)
+ {
+ /*
+ * Schema lock is held until the publication is created to prevent
+ * concurrent schema deletion.
+ */
+ LockSchemaList(schemaidlist);
+ PublicationAddSchemas(puboid, schemaidlist, true, NULL);
}
table_close(rel, RowExclusiveLock);
+ /* Associate objects with the publication. */
+ if (stmt->for_all_tables)
+ {
+ /* Invalidate relcache so that publication info is rebuilt. */
+ CacheInvalidateRelcacheAll();
+ }
+
InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0);
if (wal_level != WAL_LEVEL_LOGICAL)
@@ -1355,6 +1355,19 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
List *tables, List *schemaidlist)
{
Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
+ ListCell *lc;
+ bool nonexcepttable = false;
+ bool excepttable = false;
+
+ foreach(lc, tables)
+ {
+ PublicationTable *pub_table = lfirst_node(PublicationTable, lc);
+
+ if (!pub_table->except)
+ nonexcepttable = true;
+ else
+ excepttable = true;
+ }
if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) &&
schemaidlist && !superuser())
@@ -1374,12 +1387,19 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
errdetail("Tables from schema cannot be added to, dropped from, or set on FOR ALL TABLES publications.")));
/* Check that user is allowed to manipulate the publication tables. */
- if (tables && pubform->puballtables)
+ if (nonexcepttable && tables && pubform->puballtables)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("publication \"%s\" is defined as FOR ALL TABLES",
NameStr(pubform->pubname)),
errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications.")));
+
+ if (excepttable && !pubform->puballtables)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("publication \"%s\" is not defined as FOR ALL TABLES",
+ NameStr(pubform->pubname)),
+ errdetail("except table cannot be added to, dropped from, or set on NON ALL TABLES publications.")));
}
/*
@@ -1656,6 +1676,7 @@ OpenTableList(List *tables)
pub_rel->relation = rel;
pub_rel->whereClause = t->whereClause;
pub_rel->columns = t->columns;
+ pub_rel->except = t->except;
rels = lappend(rels, pub_rel);
relids = lappend_oid(relids, myrelid);
@@ -1728,6 +1749,7 @@ OpenTableList(List *tables)
/* child inherits column list from parent */
pub_rel->columns = t->columns;
+ pub_rel->except = t->except;
rels = lappend(rels, pub_rel);
relids = lappend_oid(relids, childrelid);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 2cd8546d47..4d660de55e 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -16274,7 +16274,7 @@ ATPrepChangePersistence(Relation rel, bool toLogged)
* UNLOGGED as UNLOGGED tables can't be published.
*/
if (!toLogged &&
- list_length(GetRelationPublications(RelationGetRelid(rel))) > 0)
+ list_length(GetRelationPublications(RelationGetRelid(rel), false)) > 0)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("cannot change table \"%s\" to unlogged because it is part of a publication",
@@ -16411,7 +16411,7 @@ AlterTableNamespace(AlterObjectSchemaStmt *stmt, Oid *oldschema)
{
ListCell *lc;
List *schemaPubids = GetSchemaPublications(nspOid);
- List *relPubids = GetRelationPublications(RelationGetRelid(rel));
+ List *relPubids = GetRelationPublications(RelationGetRelid(rel), false);
foreach(lc, relPubids)
{
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index c9941d9cb4..25639f6b02 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -219,6 +219,11 @@ static void processCASbits(int cas_bits, int location, const char *constrType,
bool *no_inherit, core_yyscan_t yyscanner);
static void preprocess_pubobj_list(List *pubobjspec_list,
core_yyscan_t yyscanner);
+static void preprocess_alltables_pubobj_list(List *pubobjspec_list,
+ int location,
+ core_yyscan_t yyscanner);
+static void check_except_in_pubobj_list(List *pubobjspec_list,
+ core_yyscan_t yyscanner);
static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
%}
@@ -455,7 +460,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
transform_element_list transform_type_list
TriggerTransitions TriggerReferencing
vacuum_relation_list opt_vacuum_relation_list
- drop_option_list pub_obj_list
+ drop_option_list pub_obj_list except_pub_obj_list
%type <node> opt_routine_body
%type <groupclause> group_clause
@@ -493,7 +498,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
%type <list> opt_interval interval_second
%type <str> unicode_normal_form
-%type <boolean> opt_instead
+%type <boolean> opt_instead opt_except
%type <boolean> opt_unique opt_concurrently opt_verbose opt_full
%type <boolean> opt_freeze opt_analyze opt_default opt_recheck
%type <defelt> opt_binary copy_delimiter
@@ -9879,12 +9884,17 @@ CreatePublicationStmt:
n->options = $4;
$$ = (Node *)n;
}
- | CREATE PUBLICATION name FOR ALL TABLES opt_definition
+ | CREATE PUBLICATION name FOR ALL TABLES except_pub_obj_list opt_definition
{
CreatePublicationStmt *n = makeNode(CreatePublicationStmt);
n->pubname = $3;
- n->options = $7;
+ n->options = $8;
+ n->pubobjects = (List *)$7;
n->for_all_tables = true;
+ preprocess_pubobj_list(n->pubobjects, yyscanner);
+ preprocess_alltables_pubobj_list(n->pubobjects,
+ @6,
+ yyscanner);
$$ = (Node *)n;
}
| CREATE PUBLICATION name FOR pub_obj_list opt_definition
@@ -9894,6 +9904,7 @@ CreatePublicationStmt:
n->options = $6;
n->pubobjects = (List *)$5;
preprocess_pubobj_list(n->pubobjects, yyscanner);
+ check_except_in_pubobj_list(n->pubobjects, yyscanner);
$$ = (Node *)n;
}
;
@@ -9912,26 +9923,30 @@ CreatePublicationStmt:
* relation_expr here.
*/
PublicationObjSpec:
- TABLE relation_expr opt_column_list OptWhereClause
+ opt_except TABLE relation_expr opt_column_list OptWhereClause
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_TABLE;
$$->pubtable = makeNode(PublicationTable);
- $$->pubtable->relation = $2;
- $$->pubtable->columns = $3;
- $$->pubtable->whereClause = $4;
+ $$->except = $1;
+ $$->pubtable->relation = $3;
+ $$->pubtable->columns = $4;
+ $$->pubtable->whereClause = $5;
+ $$->location = @1;
}
| ALL TABLES IN_P SCHEMA ColId
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_SCHEMA;
$$->name = $5;
+ $$->except = false;
$$->location = @5;
}
| ALL TABLES IN_P SCHEMA CURRENT_SCHEMA
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA;
+ $$->except = false;
$$->location = @5;
}
| ColId opt_column_list OptWhereClause
@@ -9995,6 +10010,17 @@ pub_obj_list: PublicationObjSpec
{ $$ = lappend($1, $3); }
;
+ except_pub_obj_list: pub_obj_list
+ { $$ = $1; }
+ | /*EMPTY*/
+ { $$ = NULL; }
+ ;
+
+opt_except:
+ EXCEPT { $$ = true; }
+ | /*EMPTY*/ { $$ = false; }
+ ;
+
/*****************************************************************************
*
* ALTER PUBLICATION name SET ( options )
@@ -18712,6 +18738,7 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
ListCell *cell;
PublicationObjSpec *pubobj;
PublicationObjSpecType prevobjtype = PUBLICATIONOBJ_CONTINUATION;
+ bool prevexceptobj = false;
if (!pubobjspec_list)
return;
@@ -18729,7 +18756,10 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
pubobj = (PublicationObjSpec *) lfirst(cell);
if (pubobj->pubobjtype == PUBLICATIONOBJ_CONTINUATION)
+ {
pubobj->pubobjtype = prevobjtype;
+ pubobj->except = prevexceptobj;
+ }
if (pubobj->pubobjtype == PUBLICATIONOBJ_TABLE)
{
@@ -18750,6 +18780,8 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
pubobj->pubtable = pubtable;
pubobj->name = NULL;
}
+
+ pubobj->pubtable->except = pubobj->except;
}
else if (pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA ||
pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA)
@@ -18784,6 +18816,60 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
}
prevobjtype = pubobj->pubobjtype;
+ prevexceptobj = pubobj->except;
+ }
+}
+
+/*
+ * Process pubobjspec_list to check if any other option other that
+ * "EXCEPT TABLE" is specified with "ALL TABLES" and throw an
+ * error.
+ */
+static void
+preprocess_alltables_pubobj_list(List *pubobjspec_list, int location,
+ core_yyscan_t yyscanner)
+{
+ ListCell *cell;
+
+ if (!pubobjspec_list)
+ return;
+
+ foreach(cell, pubobjspec_list)
+ {
+ PublicationObjSpec *pubobj = (PublicationObjSpec *) lfirst(cell);
+
+ /* Only EXCEPT TABLE option supported with ALL TABLES */
+ if (!pubobj->except)
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("only EXCEPT TABLE can be specified with ALL TABLES option"),
+ parser_errposition(pubobj->location));
+ }
+}
+
+/*
+ * Process pubobjspec_list to check if "EXCEPT TABLES" is specified
+ * with "ALL TABLES" and throw an error.
+ */
+static void
+check_except_in_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
+{
+ ListCell *cell;
+ PublicationObjSpec *pubobj;
+
+ if (!pubobjspec_list)
+ return;
+
+ foreach(cell, pubobjspec_list)
+ {
+ pubobj = (PublicationObjSpec *) lfirst(cell);
+
+ /* EXCEPT TABLE option supported only with ALL TABLES */
+ if (pubobj->except)
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("EXCEPT TABLE can be specified only with ALL TABLES option"),
+ parser_errposition(pubobj->location));
}
}
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index b197bfd565..4dcd35d1f5 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1988,7 +1988,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
if (!entry->replicate_valid)
{
Oid schemaId = get_rel_namespace(relid);
- List *pubids = GetRelationPublications(relid);
+ List *pubids = GetRelationPublications(relid, false);
+ List *exceptTablePubids = GetRelationPublications(relid, true);
/*
* We don't acquire a lock on the namespace system table as we build
@@ -2077,22 +2078,6 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
Oid pub_relid = relid;
int ancestor_level = 0;
- /*
- * If this is a FOR ALL TABLES publication, pick the partition root
- * and set the ancestor level accordingly.
- */
- if (pub->alltables)
- {
- publish = true;
- if (pub->pubviaroot && am_partition)
- {
- List *ancestors = get_partition_ancestors(relid);
-
- pub_relid = llast_oid(ancestors);
- ancestor_level = list_length(ancestors);
- }
- }
-
if (!publish)
{
bool ancestor_published = false;
@@ -2111,7 +2096,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
ancestor = GetTopMostAncestorInPublication(pub->oid,
ancestors,
- &level);
+ &level,
+ pub->alltables);
if (ancestor != InvalidOid)
{
@@ -2126,6 +2112,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
if (list_member_oid(pubids, pub->oid) ||
list_member_oid(schemaPubids, pub->oid) ||
+ (pub->alltables &&
+ !list_member_oid(exceptTablePubids, pub->oid)) ||
ancestor_published)
publish = true;
}
@@ -2201,6 +2189,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
list_free(pubids);
list_free(schemaPubids);
+ list_free(exceptTablePubids);
list_free(rel_publications);
entry->replicate_valid = true;
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 43f14c233d..56592afac1 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -5562,6 +5562,8 @@ void
RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
{
List *puboids;
+ List *alltablespuboids;
+ List *exceptpuboids = NIL;
ListCell *lc;
MemoryContext oldcxt;
Oid schemaid;
@@ -5595,7 +5597,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
pubdesc->cols_valid_for_delete = true;
/* Fetch the publication membership info. */
- puboids = GetRelationPublications(relid);
+ puboids = GetRelationPublications(relid, false);
schemaid = RelationGetNamespace(relation);
puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
@@ -5609,14 +5611,19 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
Oid ancestor = lfirst_oid(lc);
puboids = list_concat_unique_oid(puboids,
- GetRelationPublications(ancestor));
+ GetRelationPublications(ancestor, false));
schemaid = get_rel_namespace(ancestor);
puboids = list_concat_unique_oid(puboids,
GetSchemaPublications(schemaid));
+ exceptpuboids = list_concat_unique_oid(exceptpuboids,
+ GetRelationPublications(ancestor, true));
}
}
- puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+ alltablespuboids = GetAllTablesPublications();
+ puboids = list_concat_unique_oid(puboids,
+ list_difference_oid(alltablespuboids,
+ exceptpuboids));
foreach(lc, puboids)
{
Oid pubid = lfirst_oid(lc);
@@ -5645,7 +5652,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
if (!pubform->puballtables &&
(pubform->pubupdate || pubform->pubdelete) &&
pub_rf_contains_invalid_column(pubid, relation, ancestors,
- pubform->pubviaroot))
+ pubform->pubviaroot, pubform->puballtables))
{
if (pubform->pubupdate)
pubdesc->rf_valid_for_update = false;
@@ -5662,7 +5669,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
if (!pubform->puballtables &&
(pubform->pubupdate || pubform->pubdelete) &&
pub_collist_contains_invalid_column(pubid, relation, ancestors,
- pubform->pubviaroot))
+ pubform->pubviaroot, pubform->puballtables))
{
if (pubform->pubupdate)
pubdesc->cols_valid_for_update = false;
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index d3588607e7..d22a81e790 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -285,7 +285,8 @@ static void dumpBlob(Archive *fout, const BlobInfo *binfo);
static int dumpBlobs(Archive *fout, const void *arg);
static void dumpPolicy(Archive *fout, const PolicyInfo *polinfo);
static void dumpPublication(Archive *fout, const PublicationInfo *pubinfo);
-static void dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo);
+static void dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo,
+ bool bexcept);
static void dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo);
static void dumpDatabase(Archive *AH);
static void dumpDatabaseConfig(Archive *AH, PQExpBuffer outbuf,
@@ -4150,6 +4151,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
int i_prrelid;
int i_prrelqual;
int i_prattrs;
+ int i_prexcept;
int i,
j,
ntups;
@@ -4162,7 +4164,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
/* Collect all publication membership info. */
if (fout->remoteVersion >= 150000)
appendPQExpBufferStr(query,
- "SELECT tableoid, oid, prpubid, prrelid, "
+ "SELECT tableoid, oid, prpubid, prrelid, prexcept,"
"pg_catalog.pg_get_expr(prqual, prrelid) AS prrelqual, "
"(CASE\n"
" WHEN pr.prattrs IS NOT NULL THEN\n"
@@ -4188,6 +4190,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
i_prrelid = PQfnumber(res, "prrelid");
i_prrelqual = PQfnumber(res, "prrelqual");
i_prattrs = PQfnumber(res, "prattrs");
+ i_prexcept = PQfnumber(res, "prexcept");
/* this allocation may be more than we need */
pubrinfo = pg_malloc(ntups * sizeof(PublicationRelInfo));
@@ -4199,6 +4202,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
Oid prrelid = atooid(PQgetvalue(res, i, i_prrelid));
PublicationInfo *pubinfo;
TableInfo *tbinfo;
+ char *prexcept = pg_strdup(PQgetvalue(res, i, i_prexcept));
/*
* Ignore any entries for which we aren't interested in either the
@@ -4219,7 +4223,11 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
continue;
/* OK, make a DumpableObject for this relationship */
- pubrinfo[j].dobj.objType = DO_PUBLICATION_REL;
+ if (strcmp(prexcept, "f") == 0)
+ pubrinfo[j].dobj.objType = DO_PUBLICATION_REL;
+ else
+ pubrinfo[j].dobj.objType = DO_PUBLICATION_EXCEPT_REL;
+
pubrinfo[j].dobj.catId.tableoid =
atooid(PQgetvalue(res, i, i_tableoid));
pubrinfo[j].dobj.catId.oid = atooid(PQgetvalue(res, i, i_oid));
@@ -4313,13 +4321,15 @@ dumpPublicationNamespace(Archive *fout, const PublicationSchemaInfo *pubsinfo)
* dump the definition of the given publication table mapping
*/
static void
-dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
+dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo,
+ bool bexcept)
{
DumpOptions *dopt = fout->dopt;
PublicationInfo *pubinfo = pubrinfo->publication;
TableInfo *tbinfo = pubrinfo->pubtable;
PQExpBuffer query;
char *tag;
+ char *description;
/* Do nothing in data-only dump */
if (dopt->dataOnly)
@@ -4329,8 +4339,15 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
query = createPQExpBuffer();
- appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD TABLE ONLY",
+ appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD ",
fmtId(pubinfo->dobj.name));
+
+ if (bexcept)
+ appendPQExpBufferStr(query, "EXCEPT ");
+
+ appendPQExpBufferStr(query, "TABLE ONLY");
+ description = (bexcept) ? "PUBLICATION EXCEPT TABLE" : "PUBLICATION TABLE";
+
appendPQExpBuffer(query, " %s",
fmtQualifiedDumpable(tbinfo));
@@ -4359,7 +4376,7 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
ARCHIVE_OPTS(.tag = tag,
.namespace = tbinfo->dobj.namespace->dobj.name,
.owner = pubinfo->rolname,
- .description = "PUBLICATION TABLE",
+ .description = description,
.section = SECTION_POST_DATA,
.createStmt = query->data));
@@ -9935,7 +9952,10 @@ dumpDumpableObject(Archive *fout, DumpableObject *dobj)
dumpPublication(fout, (const PublicationInfo *) dobj);
break;
case DO_PUBLICATION_REL:
- dumpPublicationTable(fout, (const PublicationRelInfo *) dobj);
+ dumpPublicationTable(fout, (const PublicationRelInfo *) dobj, false);
+ break;
+ case DO_PUBLICATION_EXCEPT_REL:
+ dumpPublicationTable(fout, (const PublicationRelInfo *) dobj, true);
break;
case DO_PUBLICATION_TABLE_IN_SCHEMA:
dumpPublicationNamespace(fout,
@@ -17868,6 +17888,7 @@ addBoundaryDependencies(DumpableObject **dobjs, int numObjs,
case DO_POLICY:
case DO_PUBLICATION:
case DO_PUBLICATION_REL:
+ case DO_PUBLICATION_EXCEPT_REL:
case DO_PUBLICATION_TABLE_IN_SCHEMA:
case DO_SUBSCRIPTION:
/* Post-data objects: must come after the post-data boundary */
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 1d21c2906f..07a50ece68 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -80,6 +80,7 @@ typedef enum
DO_REFRESH_MATVIEW,
DO_POLICY,
DO_PUBLICATION,
+ DO_PUBLICATION_EXCEPT_REL,
DO_PUBLICATION_REL,
DO_PUBLICATION_TABLE_IN_SCHEMA,
DO_SUBSCRIPTION
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index 5de3241eb4..a116f4da97 100644
--- a/src/bin/pg_dump/pg_dump_sort.c
+++ b/src/bin/pg_dump/pg_dump_sort.c
@@ -90,6 +90,7 @@ enum dbObjectTypePriorities
PRIO_FK_CONSTRAINT,
PRIO_POLICY,
PRIO_PUBLICATION,
+ PRIO_PUBLICATION_EXCEPT_REL,
PRIO_PUBLICATION_REL,
PRIO_PUBLICATION_TABLE_IN_SCHEMA,
PRIO_SUBSCRIPTION,
@@ -144,6 +145,7 @@ static const int dbObjectTypePriority[] =
PRIO_REFRESH_MATVIEW, /* DO_REFRESH_MATVIEW */
PRIO_POLICY, /* DO_POLICY */
PRIO_PUBLICATION, /* DO_PUBLICATION */
+ PRIO_PUBLICATION_EXCEPT_REL, /* DO_PUBLICATION_EXCEPT_REL */
PRIO_PUBLICATION_REL, /* DO_PUBLICATION_REL */
PRIO_PUBLICATION_TABLE_IN_SCHEMA, /* DO_PUBLICATION_TABLE_IN_SCHEMA */
PRIO_SUBSCRIPTION /* DO_SUBSCRIPTION */
@@ -1488,6 +1490,11 @@ describeDumpableObject(DumpableObject *obj, char *buf, int bufsize)
"PUBLICATION TABLE (ID %d OID %u)",
obj->dumpId, obj->catId.oid);
return;
+ case DO_PUBLICATION_EXCEPT_REL:
+ snprintf(buf, bufsize,
+ "PUBLICATION EXCEPT TABLE (ID %d OID %u)",
+ obj->dumpId, obj->catId.oid);
+ return;
case DO_PUBLICATION_TABLE_IN_SCHEMA:
snprintf(buf, bufsize,
"PUBLICATION TABLES IN SCHEMA (ID %d OID %u)",
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 1ecfd7ae23..1837c1878d 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -2454,6 +2454,15 @@ my %tests = (
like => { %full_runs, section_post_data => 1, },
},
+ 'CREATE PUBLICATION pub5' => {
+ create_order => 50,
+ create_sql => 'CREATE PUBLICATION pub5 FOR ALL TABLES;',
+ regexp => qr/^
+ \QCREATE PUBLICATION pub5 FOR ALL TABLES WITH (publish = 'insert, update, delete, truncate');\E
+ /xm,
+ like => { %full_runs, section_post_data => 1, },
+ },
+
'CREATE SUBSCRIPTION sub1' => {
create_order => 50,
create_sql => 'CREATE SUBSCRIPTION sub1
@@ -2558,6 +2567,20 @@ my %tests = (
unlike => { exclude_dump_test_schema => 1, },
},
+ 'ALTER PUBLICATION pub5 ADD EXCEPT TABLE test_table' => {
+ create_order => 52,
+ create_sql =>
+ 'ALTER PUBLICATION pub5 ADD EXCEPT TABLE dump_test.test_table;',
+ regexp => qr/^
+ \QALTER PUBLICATION pub5 ADD EXCEPT TABLE ONLY dump_test.test_table;\E
+ /xm,
+ like => { %full_runs, section_post_data => 1, },
+ unlike => {
+ exclude_dump_test_schema => 1,
+ exclude_test_table => 1,
+ },
+ },
+
'CREATE SCHEMA public' => {
regexp => qr/^CREATE SCHEMA public;/m,
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 4369f2235b..89db708713 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -2952,15 +2952,20 @@ describeOneTableDetails(const char *schemaname,
"FROM pg_catalog.pg_publication p\n"
" JOIN pg_catalog.pg_publication_rel pr ON p.oid = pr.prpubid\n"
" JOIN pg_catalog.pg_class c ON c.oid = pr.prrelid\n"
- "WHERE pr.prrelid = '%s'\n"
+ "WHERE pr.prrelid = '%s' AND pr.prexcept = 'f'\n"
"UNION\n"
"SELECT pubname\n"
" , NULL\n"
" , NULL\n"
"FROM pg_catalog.pg_publication p\n"
"WHERE p.puballtables AND pg_catalog.pg_relation_is_publishable('%s')\n"
+ " AND NOT EXISTS (SELECT 1\n"
+ " FROM pg_catalog.pg_publication_rel pr\n"
+ " JOIN pg_catalog.pg_class pc\n"
+ " ON pr.prrelid = pc.oid\n"
+ " WHERE pr.prrelid = '%s' AND pr.prpubid = p.oid)\n"
"ORDER BY 1;",
- oid, oid, oid, oid);
+ oid, oid, oid, oid, oid);
}
else
{
@@ -6303,6 +6308,7 @@ describePublications(const char *pattern)
"WHERE c.relnamespace = n.oid\n"
" AND c.oid = pr.prrelid\n"
" AND pr.prpubid = '%s'\n"
+ " AND pr.prexcept = 'f'\n"
"ORDER BY 1,2", pubid);
if (!addFooterToPublicationDesc(&buf, _("Tables:"), false, &cont))
goto error_return;
@@ -6322,6 +6328,21 @@ describePublications(const char *pattern)
}
}
+ if (pset.sversion >= 150000)
+ {
+ /* Get the except tables for the specified publication */
+ printfPQExpBuffer(&buf,
+ "SELECT concat(c.relnamespace::regnamespace, '.', c.relname)\n"
+ "FROM pg_catalog.pg_class c\n"
+ " JOIN pg_catalog.pg_publication_rel pr ON c.oid = pr.prrelid\n"
+ "WHERE pr.prpubid = '%s'\n"
+ " AND pr.prexcept = 't'\n"
+ "ORDER BY 1", pubid);
+ if (!addFooterToPublicationDesc(&buf, "Except tables:",
+ true, &cont))
+ goto error_return;
+ }
+
printTable(&cont, pset.queryFout, false, pset.logfile);
printTableCleanup(&cont);
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 588c0841fe..7870e16acd 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1822,8 +1822,11 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("ADD", "DROP", "OWNER TO", "RENAME TO", "SET");
/* ALTER PUBLICATION <name> ADD */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD"))
+ COMPLETE_WITH("ALL TABLES IN SCHEMA", "EXCEPT TABLE", "TABLE");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD", "EXCEPT"))
COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE");
else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") ||
+ Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "EXCEPT", "TABLE") ||
(HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") &&
ends_with(prev_wd, ',')))
COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables);
@@ -1845,10 +1848,14 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH(",");
/* ALTER PUBLICATION <name> DROP */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP"))
- COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE");
+ COMPLETE_WITH("ALL TABLES IN SCHEMA", "EXCEPT TABLE", "TABLE");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP", "EXCEPT"))
+ COMPLETE_WITH("TABLE");
/* ALTER PUBLICATION <name> SET */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET"))
- COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE");
+ COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "EXCEPT TABLE", "TABLE");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET", "EXCEPT"))
+ COMPLETE_WITH("TABLE");
else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES", "IN", "SCHEMA"))
COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas
" AND nspname NOT LIKE E'pg\\\\_%%'",
@@ -2985,7 +2992,9 @@ psql_completion(const char *text, int start, int end)
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL"))
COMPLETE_WITH("TABLES", "TABLES IN SCHEMA");
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES"))
- COMPLETE_WITH("IN SCHEMA", "WITH (");
+ COMPLETE_WITH("IN SCHEMA", "WITH (", "EXCEPT TABLE");
+ else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES", "EXCEPT"))
+ COMPLETE_WITH("TABLE");
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE", MatchAny) && !ends_with(prev_wd, ','))
COMPLETE_WITH("WHERE (", "WITH (");
/* Complete "CREATE PUBLICATION <name> FOR TABLE" with "<table>, ..." */
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 29b1856665..eeee96f42d 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -108,11 +108,12 @@ typedef struct PublicationRelInfo
Relation relation;
Node *whereClause;
List *columns;
+ bool except;
} PublicationRelInfo;
extern Publication *GetPublication(Oid pubid);
extern Publication *GetPublicationByName(const char *pubname, bool missing_ok);
-extern List *GetRelationPublications(Oid relid);
+extern List *GetRelationPublications(Oid relid, bool bexcept);
/*---------
* Expected values for pub_partopt parameter of GetRelationPublications(),
@@ -132,7 +133,7 @@ typedef enum PublicationPartOpt
extern List *GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt);
extern List *GetAllTablesPublications(void);
-extern List *GetAllTablesPublicationRelations(bool pubviaroot);
+extern List *GetAllTablesPublicationRelations(Oid pubid, bool pubviaroot);
extern List *GetPublicationSchemas(Oid pubid);
extern List *GetSchemaPublications(Oid schemaid);
extern List *GetSchemaPublicationRelations(Oid schemaid,
@@ -143,7 +144,7 @@ extern List *GetPubPartitionOptionRelations(List *result,
PublicationPartOpt pub_partopt,
Oid relid);
extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors,
- int *ancestor_level);
+ int *ancestor_level, bool puballtables);
extern bool is_publishable_relation(Relation rel);
extern bool is_schema_publication(Oid pubid);
diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h
index 4feb581899..2eb1fbeabd 100644
--- a/src/include/catalog/pg_publication_rel.h
+++ b/src/include/catalog/pg_publication_rel.h
@@ -31,6 +31,7 @@ CATALOG(pg_publication_rel,6106,PublicationRelRelationId)
Oid oid; /* oid */
Oid prpubid BKI_LOOKUP(pg_publication); /* Oid of the publication */
Oid prrelid BKI_LOOKUP(pg_class); /* Oid of the relation */
+ bool prexcept BKI_DEFAULT(f); /* except the relation */
#ifdef CATALOG_VARLEN /* variable-length fields start here */
pg_node_tree prqual; /* qualifications */
diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h
index ae87caf089..a515cdb802 100644
--- a/src/include/commands/publicationcmds.h
+++ b/src/include/commands/publicationcmds.h
@@ -32,8 +32,8 @@ extern ObjectAddress AlterPublicationOwner(const char *name, Oid newOwnerId);
extern void AlterPublicationOwner_oid(Oid pubid, Oid newOwnerId);
extern void InvalidatePublicationRels(List *relids);
extern bool pub_rf_contains_invalid_column(Oid pubid, Relation relation,
- List *ancestors, bool pubviaroot);
+ List *ancestors, bool pubviaroot, bool alltables);
extern bool pub_collist_contains_invalid_column(Oid pubid, Relation relation,
- List *ancestors, bool pubviaroot);
+ List *ancestors, bool pubviaroot, bool alltables);
#endif /* PUBLICATIONCMDS_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index da02658c81..308d3c07a8 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -4003,6 +4003,7 @@ typedef struct PublicationTable
RangeVar *relation; /* relation to be published */
Node *whereClause; /* qualifications */
List *columns; /* List of columns in a publication table */
+ bool except; /* except relation */
} PublicationTable;
/*
@@ -4023,6 +4024,7 @@ typedef struct PublicationObjSpec
PublicationObjSpecType pubobjtype; /* type of this publication object */
char *name;
PublicationTable *pubtable;
+ bool except;
int location; /* token location, or -1 if unknown */
} PublicationObjSpec;
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 398c0f38f6..5d213309e4 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -116,6 +116,35 @@ ALTER PUBLICATION testpub_fortable SET ALL TABLES IN SCHEMA pub_test;
Tables from schemas:
"pub_test"
+-- should be able to add except table to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables ADD EXCEPT TABLE testpub_tbl1;
+\dRp+ testpub_foralltables
+ Publication testpub_foralltables
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | t | t | t | f | f | f
+Except tables:
+ "public.testpub_tbl1"
+
+-- should be able to set except table to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables SET EXCEPT TABLE testpub_tbl2;
+\dRp+ testpub_foralltables
+ Publication testpub_foralltables
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | t | t | t | f | f | f
+Except tables:
+ "public.testpub_tbl2"
+
+-- should be able to drop except table from 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables DROP EXCEPT TABLE testpub_tbl2;
+\dRp+ testpub_foralltables
+ Publication testpub_foralltables
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | t | t | t | f | f | f
+(1 row)
+
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_forschema FOR ALL TABLES IN SCHEMA pub_test;
RESET client_min_messages;
@@ -141,6 +170,30 @@ ALTER PUBLICATION testpub_forschema SET TABLE pub_test.testpub_nopk;
Tables:
"pub_test.testpub_nopk"
+-- fail - can't add except table to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable ADD EXCEPT TABLE testpub_tbl1;
+ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES
+DETAIL: except table cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't drop except table from 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable DROP EXCEPT TABLE testpub_tbl1;
+ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES
+DETAIL: except table cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't set except table to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable SET EXCEPT TABLE testpub_tbl1;
+ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES
+DETAIL: except table cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't add except table to schema publication
+ALTER PUBLICATION testpub_forschema ADD EXCEPT TABLE testpub_tbl1;
+ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES
+DETAIL: except table cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't drop except table from schema publication
+ALTER PUBLICATION testpub_forschema DROP EXCEPT TABLE testpub_tbl1;
+ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES
+DETAIL: except table cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't set except table to schema publication
+ALTER PUBLICATION testpub_forschema SET EXCEPT TABLE testpub_tbl1;
+ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES
+DETAIL: except table cannot be added to, dropped from, or set on NON ALL TABLES publications.
SELECT pubname, puballtables FROM pg_publication WHERE pubname = 'testpub_foralltables';
pubname | puballtables
----------------------+--------------
@@ -165,8 +218,34 @@ Publications:
regress_publication_user | t | t | t | f | f | f
(1 row)
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_foralltables_excepttable FOR ALL TABLES EXCEPT TABLE testpub_tbl1;
+RESET client_min_messages;
+\dRp+ testpub_foralltables_excepttable
+ Publication testpub_foralltables_excepttable
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | t | t | t | t | t | f
+Except tables:
+ "public.testpub_tbl1"
+
+-- fail - can't specify except table along with table publication
+CREATE PUBLICATION testpub_fortable_excepttable FOR TABLE pub_test.testpub_nopk, EXCEPT TABLE testpub_tbl1;
+ERROR: EXCEPT TABLE can be specified only with ALL TABLES option
+LINE 1: ...able_excepttable FOR TABLE pub_test.testpub_nopk, EXCEPT TAB...
+ ^
+-- fail - can't specify except table along with schema publication
+CREATE PUBLICATION testpub_fortable_excepttable FOR ALL TABLES IN SCHEMA pub_test, EXCEPT TABLE testpub_tbl1;
+ERROR: EXCEPT TABLE can be specified only with ALL TABLES option
+LINE 1: ...le_excepttable FOR ALL TABLES IN SCHEMA pub_test, EXCEPT TAB...
+ ^
+-- fail - can't specify only except table while create publication
+CREATE PUBLICATION testpub_fortable_excepttable FOR EXCEPT TABLE testpub_tbl1;
+ERROR: EXCEPT TABLE can be specified only with ALL TABLES option
+LINE 1: ...EATE PUBLICATION testpub_fortable_excepttable FOR EXCEPT TAB...
+ ^
DROP TABLE testpub_tbl2;
-DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema;
+DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_excepttable;
CREATE TABLE testpub_tbl3 (a int);
CREATE TABLE testpub_tbl3a (b text) INHERITS (testpub_tbl3);
SET client_min_messages = 'ERROR';
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index 9eb86fd54f..a6e6543d71 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -71,6 +71,16 @@ ALTER PUBLICATION testpub_fortable DROP ALL TABLES IN SCHEMA pub_test;
ALTER PUBLICATION testpub_fortable SET ALL TABLES IN SCHEMA pub_test;
\dRp+ testpub_fortable
+-- should be able to add except table to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables ADD EXCEPT TABLE testpub_tbl1;
+\dRp+ testpub_foralltables
+-- should be able to set except table to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables SET EXCEPT TABLE testpub_tbl2;
+\dRp+ testpub_foralltables
+-- should be able to drop except table from 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables DROP EXCEPT TABLE testpub_tbl2;
+\dRp+ testpub_foralltables
+
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_forschema FOR ALL TABLES IN SCHEMA pub_test;
RESET client_min_messages;
@@ -85,12 +95,40 @@ ALTER PUBLICATION testpub_forschema DROP TABLE pub_test.testpub_nopk;
ALTER PUBLICATION testpub_forschema SET TABLE pub_test.testpub_nopk;
\dRp+ testpub_forschema
+-- fail - can't add except table to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable ADD EXCEPT TABLE testpub_tbl1;
+-- fail - can't drop except table from 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable DROP EXCEPT TABLE testpub_tbl1;
+-- fail - can't set except table to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable SET EXCEPT TABLE testpub_tbl1;
+
+-- fail - can't add except table to schema publication
+ALTER PUBLICATION testpub_forschema ADD EXCEPT TABLE testpub_tbl1;
+-- fail - can't drop except table from schema publication
+ALTER PUBLICATION testpub_forschema DROP EXCEPT TABLE testpub_tbl1;
+-- fail - can't set except table to schema publication
+ALTER PUBLICATION testpub_forschema SET EXCEPT TABLE testpub_tbl1;
SELECT pubname, puballtables FROM pg_publication WHERE pubname = 'testpub_foralltables';
\d+ testpub_tbl2
\dRp+ testpub_foralltables
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_foralltables_excepttable FOR ALL TABLES EXCEPT TABLE testpub_tbl1;
+RESET client_min_messages;
+
+\dRp+ testpub_foralltables_excepttable
+
+-- fail - can't specify except table along with table publication
+CREATE PUBLICATION testpub_fortable_excepttable FOR TABLE pub_test.testpub_nopk, EXCEPT TABLE testpub_tbl1;
+
+-- fail - can't specify except table along with schema publication
+CREATE PUBLICATION testpub_fortable_excepttable FOR ALL TABLES IN SCHEMA pub_test, EXCEPT TABLE testpub_tbl1;
+
+-- fail - can't specify only except table while create publication
+CREATE PUBLICATION testpub_fortable_excepttable FOR EXCEPT TABLE testpub_tbl1;
+
DROP TABLE testpub_tbl2;
-DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema;
+DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_excepttable;
CREATE TABLE testpub_tbl3 (a int);
CREATE TABLE testpub_tbl3a (b text) INHERITS (testpub_tbl3);
diff --git a/src/test/subscription/t/033_rep_changes_except_table.pl b/src/test/subscription/t/033_rep_changes_except_table.pl
new file mode 100644
index 0000000000..1f6141ceca
--- /dev/null
+++ b/src/test/subscription/t/033_rep_changes_except_table.pl
@@ -0,0 +1,97 @@
+
+# Copyright (c) 2021-2022, PostgreSQL Global Development Group
+
+# Logical replication tests for except table publications
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# Test replication with publications created using FOR ALL TABLES EXCEPT TABLE
+# option.
+# Create schemas and tables on publisher
+$node_publisher->safe_psql('postgres', "CREATE SCHEMA sch1");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE sch1.tab1 AS SELECT generate_series(1,10) AS a");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE public.tab1(a int)");
+
+# Create schemas and tables on subscriber
+$node_subscriber->safe_psql('postgres', "CREATE SCHEMA sch1");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE sch1.tab1 (a int)");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE public.tab1 (a int)");
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_schema FOR ALL TABLES EXCEPT TABLE sch1.tab1");
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_schema CONNECTION '$publisher_connstr' PUBLICATION tap_pub_schema"
+);
+
+$node_publisher->wait_for_catchup('tap_sub_schema');
+
+# Also wait for initial table sync to finish
+my $synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check the table data does not sync for excluded table
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), min(a), max(a) FROM sch1.tab1");
+is($result, qq(0||), 'check tablesync is excluded for excluded tables');
+
+# Insert some data into few tables and verify that inserted data is not
+# replicated
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO sch1.tab1 VALUES(generate_series(11,20))");
+
+$node_publisher->wait_for_catchup('tap_sub_schema');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), min(a), max(a) FROM sch1.tab1");
+is($result, qq(0||), 'check replicated inserts on subscriber');
+
+# Alter publication to exclude data changes in public.tab1 and verify that
+# subscriber does not get the new table data.
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_schema add EXCEPT TABLE public.tab1");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO public.tab1 VALUES(generate_series(1,10))");
+
+$node_publisher->wait_for_catchup('tap_sub_schema');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), min(a), max(a) FROM public.tab1");
+is($result, qq(0||), 'check rows on subscriber catchup');
+
+# Alter publication to drop except table public.tab1 and verify that subscriber
+# gets the new table data.
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_schema drop EXCEPT TABLE public.tab1");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO public.tab1 VALUES(generate_series(1,10))");
+
+$node_publisher->wait_for_catchup('tap_sub_schema');
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), min(a), max(a) FROM public.tab1");
+is($result, qq(10|1|10), 'check rows on subscriber catchup');
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
+
+done_testing();
--
2.32.0
