On Tue, Mar 22, 2022 at 12:38 PM vignesh C <vignes...@gmail.com> wrote:
>
> Hi,
>
> This feature adds an option to skip changes of all tables in specified
> schema while creating publication.
> 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
> schemas.
> Ex:
> CREATE PUBLICATION pub1 FOR ALL TABLES SKIP ALL TABLES IN SCHEMA s1,s2;
> OR
> ALTER PUBLICATION pub1 ADD SKIP ALL TABLES IN SCHEMA s1,s2;
>
> A new column pnskip is added to table "pg_publication_namespace", to
> maintain the schemas that the user wants to skip publishing through
> the publication. Modified the output plugin (pgoutput) to skip
> publishing the changes if the relation is part of skip schema
> publication.
> As a continuation to this, I will work on implementing skipping tables
> from all tables in schema and skipping tables from all tables
> publication.
>
> Attached patch has the implementation for this.
The patch was not applying on top of HEAD because of the recent
commits, attached patch is rebased on top of HEAD.
Regards,
Vignesh
From 1524254629c72149ae99d13aae283d3aa6a70253 Mon Sep 17 00:00:00 2001
From: Vigneshwaran C <vignes...@gmail.com>
Date: Sat, 26 Mar 2022 19:19:31 +0530
Subject: [PATCH v1] Skip publishing the tables of schema.
A new option "SKIP ALL TABLES IN SCHEMA" in Create/Alter Publication allows
one or more skip schemas to be specified, publisher will skip sending the data
of the tables present in the skip schema to the subscriber.
The new syntax allows specifying schemas. For example:
CREATE PUBLICATION pub1 FOR ALL TABLES SKIP ALL TABLES IN SCHEMA s1,s2;
OR
ALTER PUBLICATION pub1 ADD SKIP ALL TABLES IN SCHEMA s1,s2;
A new column pnskip is added to table "pg_publication_namespace", to maintain
the schemas that the user wants to skip publishing through the publication.
Modified the output plugin (pgoutput) to skip publishing the changes if the
relation is part of skip schema publication.
Updates pg_dump to identify and dump skip schema publications. Updates the \d
family of commands to display skip schema publications and \dRp+ variant will
now display associated skip schemas if any.
---
doc/src/sgml/catalogs.sgml | 9 +
doc/src/sgml/logical-replication.sgml | 7 +-
doc/src/sgml/ref/alter_publication.sgml | 28 ++-
doc/src/sgml/ref/create_publication.sgml | 28 ++-
doc/src/sgml/ref/psql-ref.sgml | 5 +-
src/backend/catalog/pg_publication.c | 70 +++++--
src/backend/commands/publicationcmds.c | 186 +++++++++++-------
src/backend/commands/tablecmds.c | 6 +-
src/backend/nodes/copyfuncs.c | 14 ++
src/backend/nodes/equalfuncs.c | 14 ++
src/backend/parser/gram.y | 119 ++++++++++-
src/backend/replication/pgoutput/pgoutput.c | 25 +--
src/backend/utils/cache/relcache.c | 21 +-
src/bin/pg_dump/pg_dump.c | 33 +++-
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 | 30 +++
src/bin/psql/describe.c | 19 +-
src/bin/psql/tab-complete.c | 26 ++-
src/include/catalog/pg_publication.h | 20 +-
.../catalog/pg_publication_namespace.h | 1 +
src/include/commands/publicationcmds.h | 6 +-
src/include/nodes/nodes.h | 1 +
src/include/nodes/parsenodes.h | 1 +
src/test/regress/expected/publication.out | 81 +++++++-
src/test/regress/sql/publication.sql | 41 +++-
.../t/032_rep_changes_skip_schema.pl | 96 +++++++++
src/tools/pgindent/typedefs.list | 1 +
28 files changed, 742 insertions(+), 154 deletions(-)
create mode 100644 src/test/subscription/t/032_rep_changes_skip_schema.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 560e205b95..59b0126c18 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -6304,6 +6304,15 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
A null value indicates that all columns are published.
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>pnskip</structfield> <type>bool</type>
+ </para>
+ <para>
+ True if the schema is skip schema
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 555fbd749c..e2a4b89226 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -599,9 +599,10 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
<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.
+ table. To add all tables in schema or skip 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 40366a10fe..c854722bda 100644
--- a/doc/src/sgml/ref/alter_publication.sgml
+++ b/doc/src/sgml/ref/alter_publication.sgml
@@ -32,7 +32,7 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ ( <replaceable class="parameter">column_name</replaceable> [, ... ] ) ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ... ]
SEQUENCE <replaceable class="parameter">sequence_name</replaceable> [, ... ]
- ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
+ [SKIP] ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
ALL SEQUENCES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
</synopsis>
</refsynopsisdiv>
@@ -73,12 +73,12 @@ 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
- <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
+ The <literal>ADD [SKIP] ALL TABLES IN SCHEMA</literal> and
+ <literal>SET [SKIP] 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
<literal>CREATE</literal> privilege on the database. Also, the new owner
- of a <literal>FOR ALL TABLES</literal> or <literal>FOR ALL TABLES IN
+ of a <literal>FOR ALL TABLES</literal> or <literal>FOR [SKIP] ALL TABLES IN
SCHEMA</literal> publication must be a superuser. However, a superuser can
change the ownership of a publication regardless of these restrictions.
</para>
@@ -90,6 +90,14 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
adding/setting a table to a publication that already has a table's schema as
part of the specified schema is not supported.
</para>
+
+ <para>
+ The <literal>ADD SKIP ALL TABLES IN SCHEMA</literal> and
+ <literal>SET SKIP ALL TABLES IN SCHEMA</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>
</refsect1>
<refsect1>
@@ -220,6 +228,16 @@ ALTER PUBLICATION sales_publication ADD ALL TABLES IN SCHEMA marketing, sales;
ALTER PUBLICATION production_publication ADD TABLE users, departments, ALL TABLES IN SCHEMA production;
</programlisting>
</para>
+
+ <para>
+ Add skip schemas <structname>sales_june</structname> and
+ <structname>sales_july</structname> to the publication
+ <structname>mypublication</structname>:
+<programlisting>
+ALTER PUBLICATION mypublication ADD SKIP ALL TABLES IN SCHEMA sales_june, sales_july;
+</programlisting>
+ </para>
+
</refsect1>
<refsect1>
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index d2739968d9..e83864d175 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -28,7 +28,7 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
<phrase>where <replaceable class="parameter">object type</replaceable> is one of:</phrase>
- TABLES
+ TABLES [SKIP ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA }]
SEQUENCES
<phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase>
@@ -146,6 +146,23 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
</listitem>
</varlistentry>
+ <varlistentry>
+ <term><literal>SKIP ALL TABLES IN SCHEMA</literal></term>
+ <listitem>
+ <para>
+ Marks the publication as one that skips replicating changes for all
+ tables in the specified list of schemas.
+ </para>
+
+ <para>
+ <literal>SKIP ALL TABLES IN SCHEMA</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>FOR ALL TABLES IN SCHEMA</literal></term>
<term><literal>FOR ALL SEQUENCES IN SCHEMA</literal></term>
@@ -355,6 +372,15 @@ CREATE PUBLICATION production_publication FOR TABLE users, departments, ALL TABL
<structname>sales</structname>:
<programlisting>
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 all the tables present in the schema
+ <structname>marketing</structname> and <structname>sales</structname>:
+<programlisting>
+CREATE PUBLICATION mypublication FOR ALL TABLE SKIP ALL TABLES IN SCHEMA marketing, sales;
</programlisting></para>
<para>
diff --git a/doc/src/sgml/ref/psql-ref.sgml b/doc/src/sgml/ref/psql-ref.sgml
index caabb06c53..4ba4140933 100644
--- a/doc/src/sgml/ref/psql-ref.sgml
+++ b/doc/src/sgml/ref/psql-ref.sgml
@@ -1856,8 +1856,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,
+ schemas and the skip schema 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 a5a54e676e..8a6af83ef9 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -338,7 +338,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;
@@ -352,6 +353,7 @@ GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level
Oid ancestor = lfirst_oid(lc);
List *apubids = GetRelationPublications(ancestor);
List *aschemaPubids = NIL;
+ List *askipschemaPubids = NIL;
level++;
@@ -366,8 +368,11 @@ GetTopMostAncestorInPublication(Oid puboid, List *ancestors, int *ancestor_level
{
/* we only search for ancestors of tables, so PUB_OBJTYPE_TABLE */
aschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor),
- PUB_OBJTYPE_TABLE);
- if (list_member_oid(aschemaPubids, puboid))
+ PUB_OBJTYPE_TABLE, false);
+ askipschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor),
+ PUB_OBJTYPE_TABLE, true);
+ if (list_member_oid(aschemaPubids, puboid) ||
+ (puballtables && !list_member_oid(askipschemaPubids, puboid)))
{
topmost_relid = ancestor;
@@ -631,13 +636,14 @@ pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols, MemoryContext mcxt)
* Insert new publication / schema mapping.
*/
ObjectAddress
-publication_add_schema(Oid pubid, Oid schemaid, char objectType, bool if_not_exists)
+publication_add_schema(Oid pubid, PublicationSchInfo *pubsch, char objectType, bool if_not_exists)
{
Relation rel;
HeapTuple tup;
Datum values[Natts_pg_publication_namespace];
bool nulls[Natts_pg_publication_namespace];
Oid psschid;
+ Oid schemaid = pubsch->oid;
Publication *pub = GetPublication(pubid);
List *schemaRels = NIL;
ObjectAddress myself,
@@ -683,6 +689,8 @@ publication_add_schema(Oid pubid, Oid schemaid, char objectType, bool if_not_exi
ObjectIdGetDatum(schemaid);
values[Anum_pg_publication_namespace_pntype - 1] =
CharGetDatum(objectType);
+ values[Anum_pg_publication_namespace_pnskip - 1] =
+ BoolGetDatum(pubsch->skip);
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
@@ -890,13 +898,23 @@ GetAllSequencesPublications(void)
* root partitioned tables.
*/
List *
-GetAllTablesPublicationRelations(bool pubviaroot)
+GetAllTablesPublicationRelations(Oid pubid, bool pubviaroot)
{
Relation classRel;
ScanKeyData key[1];
TableScanDesc scan;
HeapTuple tuple;
List *result = NIL;
+ List *skipschemaidlist = NIL;
+ List *pubschemalist = GetPublicationSchemas(pubid, PUB_OBJTYPE_TABLE);
+ ListCell *cell;
+
+ foreach(cell, pubschemalist)
+ {
+ PublicationSchInfo *pubsch = (PublicationSchInfo *) lfirst(cell);
+
+ skipschemaidlist = lappend_oid(result, pubsch->oid);
+ }
classRel = table_open(RelationRelationId, AccessShareLock);
@@ -911,9 +929,11 @@ GetAllTablesPublicationRelations(bool pubviaroot)
{
Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
Oid relid = relForm->oid;
+ Oid schid = get_rel_namespace(relid);
if (is_publishable_class(relid, relForm) &&
- !(relForm->relispartition && pubviaroot))
+ !(relForm->relispartition && pubviaroot) &&
+ !list_member_oid(skipschemaidlist, schid))
result = lappend_oid(result, relid);
}
@@ -932,9 +952,11 @@ GetAllTablesPublicationRelations(bool pubviaroot)
{
Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
Oid relid = relForm->oid;
+ Oid schid = get_rel_namespace(relid);
if (is_publishable_class(relid, relForm) &&
- !relForm->relispartition)
+ !relForm->relispartition &&
+ !list_member_oid(skipschemaidlist, schid))
result = lappend_oid(result, relid);
}
@@ -983,10 +1005,14 @@ GetPublicationSchemas(Oid pubid, char objectType)
while (HeapTupleIsValid(tup = systable_getnext(scan)))
{
Form_pg_publication_namespace pubsch;
+ PublicationSchInfo *schinfo = makeNode(PublicationSchInfo);
+
pubsch = (Form_pg_publication_namespace) GETSTRUCT(tup);
+ schinfo->oid = pubsch->pnnspid;
+ schinfo->skip = pubsch->pnskip;
- result = lappend_oid(result, pubsch->pnnspid);
+ result = lappend(result, schinfo);
}
systable_endscan(scan);
@@ -1005,7 +1031,7 @@ GetPublicationSchemas(Oid pubid, char objectType)
* Which is why we handle the PUB_OBJTYPE_UNSUPPORTED object type too.
*/
List *
-GetSchemaPublications(Oid schemaid, char objectType)
+GetSchemaPublications(Oid schemaid, char objectType, bool skippub)
{
List *result = NIL;
CatCList *pubschlist;
@@ -1030,7 +1056,8 @@ GetSchemaPublications(Oid schemaid, char objectType)
if (pntype != objectType)
continue;
- result = lappend_oid(result, pubid);
+ if (skippub == ((Form_pg_publication_namespace) GETSTRUCT(tup))->pnskip)
+ result = lappend_oid(result, pubid);
}
ReleaseSysCacheList(pubschlist);
@@ -1116,7 +1143,7 @@ GetSchemaPublicationRelations(Oid schemaid, char objectType,
*/
List *
GetAllSchemaPublicationRelations(Oid pubid, char objectType,
- PublicationPartOpt pub_partopt)
+ PublicationPartOpt pub_partopt, bool bskip)
{
List *result = NIL;
List *pubschemalist = GetPublicationSchemas(pubid, objectType);
@@ -1126,12 +1153,16 @@ GetAllSchemaPublicationRelations(Oid pubid, char objectType,
foreach(cell, pubschemalist)
{
- Oid schemaid = lfirst_oid(cell);
+ PublicationSchInfo *pubsch = (PublicationSchInfo *) lfirst(cell);
List *schemaRels = NIL;
- schemaRels = GetSchemaPublicationRelations(schemaid, objectType,
- pub_partopt);
- result = list_concat(result, schemaRels);
+ /* Skip the skip schemas if bskip is true */
+ if (bskip && !pubsch->skip)
+ {
+ schemaRels = GetSchemaPublicationRelations(pubsch->oid, objectType,
+ pub_partopt);
+ result = list_concat(result, schemaRels);
+ }
}
return result;
@@ -1303,7 +1334,8 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
*/
if (publication->alltables)
{
- tables = GetAllTablesPublicationRelations(publication->pubviaroot);
+ tables = GetAllTablesPublicationRelations(publication->oid,
+ publication->pubviaroot);
}
else
{
@@ -1319,7 +1351,8 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
PUB_OBJTYPE_TABLE,
publication->pubviaroot ?
PUBLICATION_PART_ROOT :
- PUBLICATION_PART_LEAF);
+ PUBLICATION_PART_LEAF,
+ true);
tables = list_concat_unique_oid(relids, schemarelids);
/*
@@ -1398,7 +1431,8 @@ pg_get_publication_sequences(PG_FUNCTION_ARGS)
PUB_OBJTYPE_SEQUENCE,
publication->pubviaroot ?
PUBLICATION_PART_ROOT :
- PUBLICATION_PART_LEAF);
+ PUBLICATION_PART_LEAF,
+ true);
sequences = list_concat_unique_oid(relids, schemarelids);
}
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 84e37df783..cbc805bbdd 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -185,8 +185,8 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
foreach(cell, pubobjspec_list)
{
- Oid schemaid;
List *search_path;
+ PublicationSchInfo *pubsch = makeNode(PublicationSchInfo);
pubobj = (PublicationObjSpec *) lfirst(cell);
@@ -199,16 +199,18 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
*sequences = lappend(*sequences, pubobj->pubtable);
break;
case PUBLICATIONOBJ_TABLES_IN_SCHEMA:
- schemaid = get_namespace_oid(pubobj->name, false);
+ pubsch->oid = get_namespace_oid(pubobj->name, false);
+ pubsch->skip = pubobj->skip;
/* Filter out duplicates if user specifies "sch1, sch1" */
- *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid);
+ *tables_schemas = list_append_unique(*tables_schemas, pubsch);
break;
case PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA:
- schemaid = get_namespace_oid(pubobj->name, false);
+ pubsch->oid = get_namespace_oid(pubobj->name, false);
+ pubsch->skip = pubobj->skip;
/* Filter out duplicates if user specifies "sch1, sch1" */
- *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid);
+ *sequences_schemas = list_append_unique(*sequences_schemas, pubsch);
break;
case PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA:
search_path = fetch_search_path(false);
@@ -217,11 +219,12 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
errcode(ERRCODE_UNDEFINED_SCHEMA),
errmsg("no schema has been selected for CURRENT_SCHEMA"));
- schemaid = linitial_oid(search_path);
+ pubsch->oid = linitial_oid(search_path);
list_free(search_path);
+ pubsch->skip = pubobj->skip;
/* Filter out duplicates if user specifies "sch1, sch1" */
- *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid);
+ *tables_schemas = list_append_unique(*tables_schemas, pubsch);
break;
case PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA:
search_path = fetch_search_path(false);
@@ -230,11 +233,12 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
errcode(ERRCODE_UNDEFINED_SCHEMA),
errmsg("no schema has been selected for CURRENT_SCHEMA"));
- schemaid = linitial_oid(search_path);
+ pubsch->oid = linitial_oid(search_path);
list_free(search_path);
+ pubsch->skip = pubobj->skip;
/* Filter out duplicates if user specifies "sch1, sch1" */
- *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid);
+ *sequences_schemas = list_append_unique(*sequences_schemas, pubsch);
break;
default:
/* shouldn't happen */
@@ -260,40 +264,45 @@ CheckObjSchemaNotAlreadyInPublication(List *rels, List *schemaidlist,
Relation rel = pub_rel->relation;
Oid relSchemaId = RelationGetNamespace(rel);
- if (list_member_oid(schemaidlist, relSchemaId))
+ foreach(lc, schemaidlist)
{
- if (checkobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA)
- ereport(ERROR,
- errcode(ERRCODE_INVALID_PARAMETER_VALUE),
- errmsg("cannot add schema \"%s\" to publication",
- get_namespace_name(relSchemaId)),
- errdetail("Table \"%s\" in schema \"%s\" is already part of the publication, adding the same schema is not supported.",
- RelationGetRelationName(rel),
- get_namespace_name(relSchemaId)));
- else if (checkobjtype == PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA)
- ereport(ERROR,
- errcode(ERRCODE_INVALID_PARAMETER_VALUE),
- errmsg("cannot add schema \"%s\" to publication",
- get_namespace_name(relSchemaId)),
- errdetail("Sequence \"%s\" in schema \"%s\" is already part of the publication, adding the same schema is not supported.",
- RelationGetRelationName(rel),
- get_namespace_name(relSchemaId)));
- else if (checkobjtype == PUBLICATIONOBJ_TABLE)
- ereport(ERROR,
- errcode(ERRCODE_INVALID_PARAMETER_VALUE),
- errmsg("cannot add relation \"%s.%s\" to publication",
- get_namespace_name(relSchemaId),
- RelationGetRelationName(rel)),
- errdetail("Table's schema \"%s\" is already part of the publication or part of the specified schema list.",
- get_namespace_name(relSchemaId)));
- else if (checkobjtype == PUBLICATIONOBJ_SEQUENCE)
- ereport(ERROR,
- errcode(ERRCODE_INVALID_PARAMETER_VALUE),
- errmsg("cannot add relation \"%s.%s\" to publication",
- get_namespace_name(relSchemaId),
- RelationGetRelationName(rel)),
- errdetail("Sequence's schema \"%s\" is already part of the publication or part of the specified schema list.",
- get_namespace_name(relSchemaId)));
+ PublicationSchInfo *pub_sch = (PublicationSchInfo *) lfirst(lc);
+
+ if (pub_sch->oid == relSchemaId)
+ {
+ if (checkobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add schema \"%s\" to publication",
+ get_namespace_name(relSchemaId)),
+ errdetail("Table \"%s\" in schema \"%s\" is already part of the publication, adding the same schema is not supported.",
+ RelationGetRelationName(rel),
+ get_namespace_name(relSchemaId)));
+ else if (checkobjtype == PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add schema \"%s\" to publication",
+ get_namespace_name(relSchemaId)),
+ errdetail("Sequence \"%s\" in schema \"%s\" is already part of the publication, adding the same schema is not supported.",
+ RelationGetRelationName(rel),
+ get_namespace_name(relSchemaId)));
+ else if (checkobjtype == PUBLICATIONOBJ_TABLE)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add relation \"%s.%s\" to publication",
+ get_namespace_name(relSchemaId),
+ RelationGetRelationName(rel)),
+ errdetail("Table's schema \"%s\" is already part of the publication or part of the specified schema list.",
+ get_namespace_name(relSchemaId)));
+ else if (checkobjtype == PUBLICATIONOBJ_SEQUENCE)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add relation \"%s.%s\" to publication",
+ get_namespace_name(relSchemaId),
+ RelationGetRelationName(rel)),
+ errdetail("Sequence's schema \"%s\" is already part of the publication or part of the specified schema list.",
+ get_namespace_name(relSchemaId)));
+ }
}
}
}
@@ -343,7 +352,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);
@@ -370,7 +379,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;
@@ -419,7 +429,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);
@@ -438,7 +448,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;
@@ -907,9 +917,14 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
/* Make the changes visible. */
CommandCounterIncrement();
+ ObjectsInPublicationToOids(stmt->pubobjects, pstate, &tables, &sequences,
+ &tables_schemaidlist, &sequences_schemaidlist);
+
/* Associate objects with the publication. */
if (for_all_tables || for_all_sequences)
{
+ Assert(!tables);
+
/* Invalidate relcache so that publication info is rebuilt. */
CacheInvalidateRelcacheAll();
}
@@ -920,12 +935,7 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
*/
if (!for_all_tables || !for_all_sequences)
{
- ObjectsInPublicationToOids(stmt->pubobjects, pstate,
- &tables, &sequences,
- &tables_schemaidlist,
- &sequences_schemaidlist);
-
- /* FOR ALL TABLES IN SCHEMA requires superuser */
+ /* FOR [SKIP] ALL TABLES IN SCHEMA requires superuser */
if (list_length(tables_schemaidlist) > 0 && !superuser())
ereport(ERROR,
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
@@ -968,19 +978,6 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
CloseRelationList(rels);
}
- /* tables added through a schema */
- if (list_length(tables_schemaidlist) > 0)
- {
- /*
- * Schema lock is held until the publication is created to prevent
- * concurrent schema deletion.
- */
- LockSchemaList(tables_schemaidlist);
- PublicationAddSchemas(puboid,
- tables_schemaidlist, PUB_OBJTYPE_TABLE,
- true, NULL);
- }
-
/* sequences added through a schema */
if (list_length(sequences_schemaidlist) > 0)
{
@@ -995,6 +992,19 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
}
}
+ /* tables added through a schema */
+ if (list_length(tables_schemaidlist) > 0)
+ {
+ /*
+ * Schema lock is held until the publication is created to prevent
+ * concurrent schema deletion.
+ */
+ LockSchemaList(tables_schemaidlist);
+ PublicationAddSchemas(puboid,
+ tables_schemaidlist, PUB_OBJTYPE_TABLE,
+ true, NULL);
+ }
+
table_close(rel, RowExclusiveLock);
InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0);
@@ -1190,7 +1200,8 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
/* tables */
schemarelids = GetAllSchemaPublicationRelations(pubform->oid,
PUB_OBJTYPE_TABLE,
- PUBLICATION_PART_ALL);
+ PUBLICATION_PART_ALL,
+ false);
relids = list_concat_unique_oid(relids, schemarelids);
/* sequences */
@@ -1201,7 +1212,8 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
schemarelids = GetAllSchemaPublicationRelations(pubform->oid,
PUB_OBJTYPE_SEQUENCE,
- PUBLICATION_PART_ALL);
+ PUBLICATION_PART_ALL,
+ false);
relids = list_concat_unique_oid(relids, schemarelids);
InvalidatePublicationRels(relids);
@@ -1265,9 +1277,8 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
* Check if the relation is member of the existing schema in the
* publication or member of the schema list specified.
*/
- schemas = list_concat_copy(schemaidlist,
- GetPublicationSchemas(pubid,
- PUB_OBJTYPE_TABLE));
+ schemas = list_concat(schemaidlist,
+ GetPublicationSchemas(pubid, PUB_OBJTYPE_TABLE));
CheckObjSchemaNotAlreadyInPublication(rels, schemas,
PUBLICATIONOBJ_TABLE);
@@ -1462,7 +1473,7 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt,
List *delschemas = NIL;
/* Identify which schemas should be dropped */
- delschemas = list_difference_oid(oldschemaids, schemaidlist);
+ delschemas = list_difference(oldschemaids, schemaidlist);
/*
* Schema lock is held until the publication is altered to prevent
@@ -1491,6 +1502,20 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
List *sequences, List *sequences_schemaidlist)
{
Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
+ ListCell *lc;
+
+ bool nonskipschema = false;
+ bool skipschema = false;
+
+ foreach(lc, tables_schemaidlist)
+ {
+ PublicationSchInfo *pub_sch = (PublicationSchInfo *) lfirst(lc);
+
+ if (!pub_sch->skip)
+ nonskipschema = true;
+ else
+ skipschema = true;
+ }
if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) &&
(tables_schemaidlist || sequences_schemaidlist) && !superuser())
@@ -1502,13 +1527,20 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
* Check that user is allowed to manipulate the publication tables in
* schema
*/
- if (tables_schemaidlist && pubform->puballtables)
+ if (nonskipschema && 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 from schema cannot be added to, dropped from, or set on FOR ALL TABLES publications.")));
+ if (skipschema && !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("Skip tables from schema cannot be added to, dropped from, or set on NON ALL TABLES publications.")));
+
/*
* Check that user is allowed to manipulate the publication sequences in
* schema
@@ -2054,7 +2086,8 @@ LockSchemaList(List *schemalist)
foreach(lc, schemalist)
{
- Oid schemaid = lfirst_oid(lc);
+ PublicationSchInfo *pubsch = (PublicationSchInfo *) lfirst(lc);
+ Oid schemaid = pubsch->oid;
/* Allow query cancel in case this takes a long time */
CHECK_FOR_INTERRUPTS();
@@ -2164,10 +2197,10 @@ PublicationAddSchemas(Oid pubid, List *schemas, char objectType,
foreach(lc, schemas)
{
- Oid schemaid = lfirst_oid(lc);
+ PublicationSchInfo *pubsch = (PublicationSchInfo *) lfirst(lc);
ObjectAddress obj;
- obj = publication_add_schema(pubid, schemaid, objectType, if_not_exists);
+ obj = publication_add_schema(pubid, pubsch, objectType, if_not_exists);
if (stmt)
{
EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
@@ -2191,7 +2224,8 @@ PublicationDropSchemas(Oid pubid, List *schemas, char objectType, bool missing_o
foreach(lc, schemas)
{
- Oid schemaid = lfirst_oid(lc);
+ PublicationSchInfo *pubsch = (PublicationSchInfo *) lfirst(lc);
+ Oid schemaid = pubsch->oid;
psid = GetSysCacheOid3(PUBLICATIONNAMESPACEMAP,
Anum_pg_publication_namespace_oid,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 124b9961dc..2b1e29be10 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -16389,7 +16389,9 @@ AlterTableNamespace(AlterObjectSchemaStmt *stmt, Oid *oldschema)
if (stmt->objectType == OBJECT_TABLE)
{
ListCell *lc;
- List *schemaPubids = GetSchemaPublications(nspOid, PUB_OBJTYPE_TABLE);
+ List *schemaPubids = GetSchemaPublications(nspOid,
+ PUB_OBJTYPE_TABLE,
+ false);
List *relPubids = GetRelationPublications(RelationGetRelid(rel));
foreach(lc, relPubids)
@@ -16410,7 +16412,7 @@ AlterTableNamespace(AlterObjectSchemaStmt *stmt, Oid *oldschema)
else if (stmt->objectType == OBJECT_SEQUENCE)
{
ListCell *lc;
- List *schemaPubids = GetSchemaPublications(nspOid, PUB_OBJTYPE_SEQUENCE);
+ List *schemaPubids = GetSchemaPublications(nspOid, PUB_OBJTYPE_SEQUENCE, false);
List *relPubids = GetRelationPublications(RelationGetRelid(rel));
foreach(lc, relPubids)
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index e38ff4000f..89c111e18c 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4855,6 +4855,17 @@ _copyPublicationTable(const PublicationTable *from)
return newnode;
}
+static PublicationSchInfo *
+_copyPublicationSchInfo(const PublicationSchInfo *from)
+{
+ PublicationSchInfo *newnode = makeNode(PublicationSchInfo);
+
+ COPY_SCALAR_FIELD(oid);
+ COPY_SCALAR_FIELD(skip);
+
+ return newnode;
+}
+
static CreatePublicationStmt *
_copyCreatePublicationStmt(const CreatePublicationStmt *from)
{
@@ -5941,6 +5952,9 @@ copyObjectImpl(const void *from)
case T_PublicationObjSpec:
retval = _copyPublicationObject(from);
break;
+ case T_PublicationSchInfo:
+ retval = _copyPublicationSchInfo(from);
+ break;
case T_PublicationTable:
retval = _copyPublicationTable(from);
break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 0f330e3c70..11706e7816 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -33,6 +33,7 @@
#include "nodes/extensible.h"
#include "nodes/pathnodes.h"
#include "utils/datum.h"
+#include "catalog/pg_publication.h"
/*
@@ -2327,6 +2328,16 @@ _equalPublicationTable(const PublicationTable *a, const PublicationTable *b)
return true;
}
+static bool
+_equalPublicationSchema(const PublicationSchInfo *a,
+ const PublicationSchInfo *b)
+{
+ COMPARE_SCALAR_FIELD(oid);
+ COMPARE_SCALAR_FIELD(skip);
+
+ return true;
+}
+
static bool
_equalCreatePublicationStmt(const CreatePublicationStmt *a,
const CreatePublicationStmt *b)
@@ -3936,6 +3947,9 @@ equal(const void *a, const void *b)
case T_PublicationObjSpec:
retval = _equalPublicationObject(a, b);
break;
+ case T_PublicationSchInfo:
+ retval = _equalPublicationSchema(a, b);
+ break;
case T_PublicationTable:
retval = _equalPublicationTable(a, b);
break;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 945a9ada8b..adddb33666 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -219,6 +219,12 @@ 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,
+ List *for_all_objects,
+ int location,
+ core_yyscan_t yyscanner);
+static void check_skip_in_pubobj_list(List *pubobjspec_list,
+ core_yyscan_t yyscanner);
static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
%}
@@ -447,6 +453,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
TriggerTransitions TriggerReferencing
vacuum_relation_list opt_vacuum_relation_list
drop_option_list pub_obj_list pub_obj_type_list
+ skip_pub_obj_list
%type <node> opt_routine_body
%type <groupclause> group_clause
@@ -9716,12 +9723,18 @@ CreatePublicationStmt:
n->options = $4;
$$ = (Node *)n;
}
- | CREATE PUBLICATION name FOR ALL pub_obj_type_list opt_definition
+ | CREATE PUBLICATION name FOR ALL pub_obj_type_list skip_pub_obj_list opt_definition
{
CreatePublicationStmt *n = makeNode(CreatePublicationStmt);
n->pubname = $3;
- n->options = $7;
+ n->options = $8;
n->for_all_objects = $6;
+ n->pubobjects = (List *)$7;
+ preprocess_pubobj_list(n->pubobjects, yyscanner);
+ preprocess_alltables_pubobj_list(n->pubobjects,
+ n->for_all_objects,
+ @6,
+ yyscanner);
$$ = (Node *)n;
}
| CREATE PUBLICATION name FOR pub_obj_list opt_definition
@@ -9731,6 +9744,7 @@ CreatePublicationStmt:
n->options = $6;
n->pubobjects = (List *)$5;
preprocess_pubobj_list(n->pubobjects, yyscanner);
+ check_skip_in_pubobj_list(n->pubobjects, yyscanner);
$$ = (Node *)n;
}
;
@@ -9763,18 +9777,36 @@ PublicationObjSpec:
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_SCHEMA;
$$->name = $5;
+ $$->skip = false;
$$->location = @5;
}
| ALL TABLES IN_P SCHEMA CURRENT_SCHEMA
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA;
+ $$->skip = false;
$$->location = @5;
}
+ | SKIP ALL TABLES IN_P SCHEMA ColId
+ {
+ $$ = makeNode(PublicationObjSpec);
+ $$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_SCHEMA;
+ $$->name = $6;
+ $$->skip = true;
+ $$->location = @6;
+ }
+ | SKIP ALL TABLES IN_P SCHEMA CURRENT_SCHEMA
+ {
+ $$ = makeNode(PublicationObjSpec);
+ $$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA;
+ $$->skip = true;
+ $$->location = @6;
+ }
| SEQUENCE relation_expr
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_SEQUENCE;
+ $$->skip = false;
$$->pubtable = makeNode(PublicationTable);
$$->pubtable->relation = $2;
}
@@ -9783,12 +9815,14 @@ PublicationObjSpec:
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA;
$$->name = $5;
+ $$->skip = false;
$$->location = @5;
}
| ALL SEQUENCES IN_P SCHEMA CURRENT_SCHEMA
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA;
+ $$->skip = false;
$$->location = @5;
}
| ColId opt_column_list OptWhereClause
@@ -9865,6 +9899,12 @@ pub_obj_type_list: pub_obj_type
;
+ skip_pub_obj_list: pub_obj_list
+ { $$ = $1; }
+ | /*EMPTY*/
+ { $$ = NULL; }
+ ;
+
/*****************************************************************************
*
* ALTER PUBLICATION name SET ( options )
@@ -17482,6 +17522,7 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
ListCell *cell;
PublicationObjSpec *pubobj;
PublicationObjSpecType prevobjtype = PUBLICATIONOBJ_CONTINUATION;
+ bool prevskipobj = false;
if (!pubobjspec_list)
return;
@@ -17499,7 +17540,10 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
pubobj = (PublicationObjSpec *) lfirst(cell);
if (pubobj->pubobjtype == PUBLICATIONOBJ_CONTINUATION)
+ {
pubobj->pubobjtype = prevobjtype;
+ pubobj->skip = prevskipobj;
+ }
if (pubobj->pubobjtype == PUBLICATIONOBJ_TABLE ||
pubobj->pubobjtype == PUBLICATIONOBJ_SEQUENCE)
@@ -17579,6 +17623,77 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
}
prevobjtype = pubobj->pubobjtype;
+ prevskipobj = pubobj->skip;
+ }
+}
+
+/*
+ * Process pubobjspec_list to check if any other option other that
+ * "SKIP ALL TABLES IN SCHEMA" is specified with "ALL TABLES" and throw an
+ * error.
+ */
+static void
+preprocess_alltables_pubobj_list(List *pubobjspec_list, List *for_all_objects,
+ int location, core_yyscan_t yyscanner)
+{
+ ListCell *cell;
+ bool for_all_tables = false;
+
+ if (!pubobjspec_list)
+ return;
+
+ /* Translate the list of object types (represented by strings) to bool flags. */
+ foreach (cell, for_all_objects)
+ {
+ char *val = strVal(lfirst(cell));
+ if (strcmp(val, "tables") == 0)
+ for_all_tables = true;
+ }
+
+ if (!for_all_tables)
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("SKIP ALL TABLES IN SCHEMA can be specified only with ALL TABLES option"),
+ parser_errposition(location));
+
+ foreach(cell, pubobjspec_list)
+ {
+ PublicationObjSpec *pubobj = (PublicationObjSpec *) lfirst(cell);
+
+ /* Only SKIP ALL TABLES IN SCHEMA option supported with ALL TABLES */
+ if (pubobj->pubobjtype != PUBLICATIONOBJ_TABLES_IN_SCHEMA ||
+ !pubobj->skip)
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("only SKIP ALL TABLES IN SCHEMA can be specified with ALL TABLES option"),
+ parser_errposition(pubobj->location));
+ }
+}
+
+/*
+ * Process pubobjspec_list to check if "SKIP ALL TABLES IN SCHEMA" is specified
+ * with "ALL TABLES" and throw an error.
+ */
+static void
+check_skip_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);
+
+ /* Only SKIP ALL TABLES IN SCHEMA option supported with ALL TABLES */
+ if (pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA &&
+ pubobj->skip)
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("SKIP ALL TABLES IN SCHEMA 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 893833ea83..35b28f7bcb 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1949,7 +1949,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
* the cache entry using a historic snapshot and all the later changes
* are absorbed while decoding WAL.
*/
- List *schemaPubids = GetSchemaPublications(schemaId, objectType);
+ List *schemaPubids = GetSchemaPublications(schemaId, objectType, false);
+ List *skipSchemaPubids = GetSchemaPublications(schemaId, objectType, true);
ListCell *lc;
Oid publish_as_relid = relid;
int publish_ancestor_level = 0;
@@ -2032,22 +2033,11 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
int ancestor_level = 0;
/*
- * If this is a FOR ALL TABLES publication, pick the partition root
- * and set the ancestor level accordingly. If this is a FOR ALL
- * SEQUENCES publication, we publish it too but we don't need to
- * pick the partition root etc.
+ * If this is a FOR ALL SEQUENCES publication, we publish it too
+ * but we don't need to pick the partition root etc.
*/
- if (pub->alltables || pub->allsequences)
- {
+ if (pub->allsequences)
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)
{
@@ -2067,7 +2057,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
ancestor = GetTopMostAncestorInPublication(pub->oid,
ancestors,
- &level);
+ &level,
+ pub->alltables);
if (ancestor != InvalidOid)
{
@@ -2082,6 +2073,7 @@ 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(skipSchemaPubids, pub->oid)) ||
ancestor_published)
publish = true;
}
@@ -2158,6 +2150,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
list_free(pubids);
list_free(schemaPubids);
+ list_free(skipSchemaPubids);
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 d47fac7bb9..da39ebfcc3 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -5558,6 +5558,8 @@ void
RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
{
List *puboids;
+ List *alltablespuboids;
+ List *skipschemapuboids;
ListCell *lc;
MemoryContext oldcxt;
Oid schemaid;
@@ -5598,7 +5600,8 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
objType = pub_get_object_type_for_relkind(relkind);
puboids = list_concat_unique_oid(puboids,
- GetSchemaPublications(schemaid, objType));
+ GetSchemaPublications(schemaid, objType, false));
+ skipschemapuboids = GetSchemaPublications(schemaid, objType, true);
/*
* If this is a partion (and thus a table), lookup all ancestors and track
@@ -5620,10 +5623,15 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
schemaid = get_rel_namespace(ancestor);
puboids = list_concat_unique_oid(puboids,
GetSchemaPublications(schemaid,
- PUB_OBJTYPE_TABLE));
+ PUB_OBJTYPE_TABLE, false));
+ skipschemapuboids = list_concat_unique_oid(skipschemapuboids,
+ GetSchemaPublications(schemaid,
+ PUB_OBJTYPE_TABLE, true));
}
}
+ alltablespuboids = GetAllTablesPublications();
+
/*
* Consider also FOR ALL TABLES and FOR ALL SEQUENCES publications,
* depending on the relkind of the relation.
@@ -5631,7 +5639,9 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
if (relation->rd_rel->relkind == RELKIND_SEQUENCE)
puboids = list_concat_unique_oid(puboids, GetAllSequencesPublications());
else
- puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+ puboids = list_concat_unique_oid(puboids,
+ list_difference_oid(alltablespuboids,
+ skipschemapuboids));
foreach(lc, puboids)
{
@@ -5662,7 +5672,8 @@ 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;
@@ -5679,7 +5690,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 535b160165..8bed12d9b2 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4040,6 +4040,7 @@ getPublicationNamespaces(Archive *fout)
int i_pnpubid;
int i_pnnspid;
int i_pntype;
+ int i_pnskip;
int i,
j,
ntups;
@@ -4051,7 +4052,7 @@ getPublicationNamespaces(Archive *fout)
/* Collect all publication membership info. */
appendPQExpBufferStr(query,
- "SELECT tableoid, oid, pnpubid, pnnspid, pntype "
+ "SELECT tableoid, oid, pnpubid, pnnspid, pntype, pnskip "
"FROM pg_catalog.pg_publication_namespace");
res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
@@ -4062,6 +4063,7 @@ getPublicationNamespaces(Archive *fout)
i_pnpubid = PQfnumber(res, "pnpubid");
i_pnnspid = PQfnumber(res, "pnnspid");
i_pntype = PQfnumber(res, "pntype");
+ i_pnskip = PQfnumber(res, "pnskip");
/* this allocation may be more than we need */
pubsinfo = pg_malloc(ntups * sizeof(PublicationSchemaInfo));
@@ -4072,6 +4074,7 @@ getPublicationNamespaces(Archive *fout)
Oid pnpubid = atooid(PQgetvalue(res, i, i_pnpubid));
Oid pnnspid = atooid(PQgetvalue(res, i, i_pnnspid));
char pntype = PQgetvalue(res, i, i_pntype)[0];
+ char *pnskip = pg_strdup(PQgetvalue(res, i, i_pnskip));
PublicationInfo *pubinfo;
NamespaceInfo *nspinfo;
@@ -4094,7 +4097,10 @@ getPublicationNamespaces(Archive *fout)
continue;
/* OK, make a DumpableObject for this relationship */
- pubsinfo[j].dobj.objType = DO_PUBLICATION_TABLE_IN_SCHEMA;
+ if (strcmp(pnskip, "t") == 0)
+ pubsinfo[j].dobj.objType = DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA;
+ else
+ pubsinfo[j].dobj.objType = DO_PUBLICATION_TABLE_IN_SCHEMA;
pubsinfo[j].dobj.catId.tableoid =
atooid(PQgetvalue(res, i, i_tableoid));
pubsinfo[j].dobj.catId.oid = atooid(PQgetvalue(res, i, i_oid));
@@ -4252,13 +4258,15 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
* dump the definition of the given publication schema mapping.
*/
static void
-dumpPublicationNamespace(Archive *fout, const PublicationSchemaInfo *pubsinfo)
+dumpPublicationNamespace(Archive *fout, const PublicationSchemaInfo *pubsinfo,
+ bool bskip)
{
DumpOptions *dopt = fout->dopt;
NamespaceInfo *schemainfo = pubsinfo->pubschema;
PublicationInfo *pubinfo = pubsinfo->publication;
PQExpBuffer query;
char *tag;
+ char *description = (bskip) ? "PUBLICATION SKIP TABLES IN SCHEMA" : "PUBLICATION TABLES IN SCHEMA";
/* Do nothing in data-only dump */
if (dopt->dataOnly)
@@ -4271,7 +4279,13 @@ dumpPublicationNamespace(Archive *fout, const PublicationSchemaInfo *pubsinfo)
appendPQExpBuffer(query, "ALTER PUBLICATION %s ", fmtId(pubinfo->dobj.name));
if (pubsinfo->pubtype == 't')
- appendPQExpBuffer(query, "ADD ALL TABLES IN SCHEMA %s;\n", fmtId(schemainfo->dobj.name));
+ {
+ appendPQExpBufferStr(query, "ADD ");
+ if (bskip)
+ appendPQExpBufferStr(query, "SKIP ");
+
+ appendPQExpBuffer(query, "ALL TABLES IN SCHEMA %s;\n", fmtId(schemainfo->dobj.name));
+ }
else
appendPQExpBuffer(query, "ADD ALL SEQUENCES IN SCHEMA %s;\n", fmtId(schemainfo->dobj.name));
@@ -4284,7 +4298,7 @@ dumpPublicationNamespace(Archive *fout, const PublicationSchemaInfo *pubsinfo)
ARCHIVE_OPTS(.tag = tag,
.namespace = schemainfo->dobj.name,
.owner = pubinfo->rolname,
- .description = "PUBLICATION TABLES IN SCHEMA",
+ .description = description,
.section = SECTION_POST_DATA,
.createStmt = query->data));
@@ -9935,9 +9949,15 @@ dumpDumpableObject(Archive *fout, DumpableObject *dobj)
case DO_PUBLICATION_REL:
dumpPublicationTable(fout, (const PublicationRelInfo *) dobj);
break;
+ case DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA:
+ dumpPublicationNamespace(fout,
+ (const PublicationSchemaInfo *) dobj,
+ true);
+ break;
case DO_PUBLICATION_TABLE_IN_SCHEMA:
dumpPublicationNamespace(fout,
- (const PublicationSchemaInfo *) dobj);
+ (const PublicationSchemaInfo *) dobj,
+ false);
break;
case DO_SUBSCRIPTION:
dumpSubscription(fout, (const SubscriptionInfo *) dobj);
@@ -17868,6 +17888,7 @@ addBoundaryDependencies(DumpableObject **dobjs, int numObjs,
case DO_POLICY:
case DO_PUBLICATION:
case DO_PUBLICATION_REL:
+ case DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA:
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 688093c55e..24b809a8fc 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -81,6 +81,7 @@ typedef enum
DO_POLICY,
DO_PUBLICATION,
DO_PUBLICATION_REL,
+ DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA,
DO_PUBLICATION_TABLE_IN_SCHEMA,
DO_SUBSCRIPTION
} DumpableObjectType;
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index 1592090839..d024936b14 100644
--- a/src/bin/pg_dump/pg_dump_sort.c
+++ b/src/bin/pg_dump/pg_dump_sort.c
@@ -91,6 +91,7 @@ enum dbObjectTypePriorities
PRIO_POLICY,
PRIO_PUBLICATION,
PRIO_PUBLICATION_REL,
+ PRIO_PUBLICATION_SKIP_TABLE_IN_SCHEMA,
PRIO_PUBLICATION_TABLE_IN_SCHEMA,
PRIO_SUBSCRIPTION,
PRIO_DEFAULT_ACL, /* done in ACL pass */
@@ -145,6 +146,7 @@ static const int dbObjectTypePriority[] =
PRIO_POLICY, /* DO_POLICY */
PRIO_PUBLICATION, /* DO_PUBLICATION */
PRIO_PUBLICATION_REL, /* DO_PUBLICATION_REL */
+ PRIO_PUBLICATION_SKIP_TABLE_IN_SCHEMA, /* DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA */
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_SKIP_TABLE_IN_SCHEMA:
+ snprintf(buf, bufsize,
+ "PUBLICATION SKIP TABLES IN SCHEMA (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 af5d6fa5a3..b14864648e 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -2402,6 +2402,15 @@ my %tests = (
like => { %full_runs, section_post_data => 1, },
},
+ 'CREATE PUBLICATION pub6' => {
+ create_order => 50,
+ create_sql => 'CREATE PUBLICATION pub6 FOR ALL TABLES;',
+ regexp => qr/^
+ \QCREATE PUBLICATION pub6 FOR ALL TABLES WITH (publish = 'insert, update, delete, truncate, sequence');\E
+ /xm,
+ like => { %full_runs, section_post_data => 1, },
+ },
+
'CREATE SUBSCRIPTION sub1' => {
create_order => 50,
create_sql => 'CREATE SUBSCRIPTION sub1
@@ -2527,6 +2536,27 @@ my %tests = (
like => { %full_runs, section_post_data => 1, },
},
+ 'ALTER PUBLICATION pub6 ADD SKIP ALL TABLES IN SCHEMA dump_test' => {
+ create_order => 51,
+ create_sql =>
+ 'ALTER PUBLICATION pub6 ADD SKIP ALL TABLES IN SCHEMA dump_test;',
+ regexp => qr/^
+ \QALTER PUBLICATION pub6 ADD SKIP ALL TABLES IN SCHEMA dump_test;\E
+ /xm,
+ like => { %full_runs, section_post_data => 1, },
+ unlike => { exclude_dump_test_schema => 1, },
+ },
+
+ 'ALTER PUBLICATION pub6 ADD SKIP ALL TABLES IN SCHEMA public' => {
+ create_order => 52,
+ create_sql =>
+ 'ALTER PUBLICATION pub6 ADD SKIP ALL TABLES IN SCHEMA public;',
+ regexp => qr/^
+ \QALTER PUBLICATION pub6 ADD SKIP ALL TABLES IN SCHEMA public;\E
+ /xm,
+ like => { %full_runs, section_post_data => 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 4dddf08789..5510c190db 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -2964,7 +2964,7 @@ describeOneTableDetails(const char *schemaname,
"FROM pg_catalog.pg_publication p\n"
" JOIN pg_catalog.pg_publication_namespace pn ON p.oid = pn.pnpubid\n"
" JOIN pg_catalog.pg_class pc ON pc.relnamespace = pn.pnnspid\n"
- "WHERE pc.oid ='%s' and pn.pntype = 't' and pg_catalog.pg_relation_is_publishable('%s')\n"
+ "WHERE pc.oid ='%s' and pn.pntype = 't' AND pn.pnskip = 'f' and pg_catalog.pg_relation_is_publishable('%s')\n"
"UNION\n"
"SELECT pubname\n"
" , pg_get_expr(pr.prqual, c.oid)\n"
@@ -6209,7 +6209,7 @@ describePublications(const char *pattern)
"SELECT n.nspname\n"
"FROM pg_catalog.pg_namespace n\n"
" JOIN pg_catalog.pg_publication_namespace pn ON n.oid = pn.pnnspid\n"
- "WHERE pn.pnpubid = '%s' AND pn.pntype = 't'\n"
+ "WHERE pn.pnpubid = '%s' AND pn.pntype = 't' AND pn.pnskip = 'f'\n"
"ORDER BY 1", pubid);
if (!addFooterToPublicationDesc(&buf, "Tables from schemas:",
true, &cont))
@@ -6248,6 +6248,21 @@ describePublications(const char *pattern)
}
}
+ if (pset.sversion >= 150000)
+ {
+ /* Get the skip schemas for the specified publication */
+ printfPQExpBuffer(&buf,
+ "SELECT n.nspname\n"
+ "FROM pg_catalog.pg_namespace n\n"
+ " JOIN pg_catalog.pg_publication_namespace pn ON n.oid = pn.pnnspid\n"
+ "WHERE pn.pnpubid = '%s'\n"
+ " AND pn.pnskip = 't'\n"
+ "ORDER BY 1", pubid);
+ if (!addFooterToPublicationDesc(&buf, "Skip tables from schemas:",
+ 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 63bfdf11c6..2b7fcaca16 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1815,7 +1815,9 @@ 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", "ALL SEQUENCES IN SCHEMA", "TABLE", "SEQUENCE");
+ COMPLETE_WITH("ALL TABLES IN SCHEMA", "ALL SEQUENCES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "TABLE", "SEQUENCE");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD", "SKIP"))
+ COMPLETE_WITH("ALL TABLES IN SCHEMA");
else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") ||
(HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") &&
ends_with(prev_wd, ',')))
@@ -1842,11 +1844,16 @@ 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", "ALL SEQUENCES IN SCHEMA", "TABLE", "SEQUENCE");
+ COMPLETE_WITH("ALL TABLES IN SCHEMA", "ALL SEQUENCES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "TABLE", "SEQUENCE");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP", "SKIP"))
+ COMPLETE_WITH("ALL TABLES IN SCHEMA");
/* ALTER PUBLICATION <name> SET */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET"))
- COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "ALL SEQUENCES IN SCHEMA", "TABLE", "SEQUENCE");
- else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA"))
+ COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "ALL SEQUENCES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "TABLE", "SEQUENCE");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET", "SKIP"))
+ COMPLETE_WITH("ALL TABLES IN SCHEMA");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA") ||
+ Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "SKIP", "ALL", "TABLES", "IN", "SCHEMA"))
COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas
" AND nspname NOT LIKE E'pg\\\\_%%'",
"CURRENT_SCHEMA");
@@ -2979,7 +2986,9 @@ psql_completion(const char *text, int start, int end)
"SEQUENCE", "ALL SEQUENCES", "ALL SEQUENCES IN SCHEMA");
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL"))
COMPLETE_WITH("TABLES", "TABLES IN SCHEMA", "SEQUENCES", "SEQUENCES IN SCHEMA");
- else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES|SEQUENCES"))
+ else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES"))
+ COMPLETE_WITH("IN SCHEMA", "WITH (", "SKIP ALL TABLES IN SCHEMA");
+ else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "SEQUENCES"))
COMPLETE_WITH("IN SCHEMA", "WITH (");
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE|SEQUENCE", MatchAny) && !ends_with(prev_wd, ','))
COMPLETE_WITH("WHERE (", "WITH (");
@@ -3005,11 +3014,14 @@ psql_completion(const char *text, int start, int end)
* Complete "CREATE PUBLICATION <name> FOR ALL TABLES|SEQUENCES IN SCHEMA <schema>,
* ..."
*/
- else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA"))
+ else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA") ||
+ Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "SKIP", "ALL", "TABLES", "IN", "SCHEMA"))
COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas
" AND nspname NOT LIKE E'pg\\\\_%%'",
"CURRENT_SCHEMA");
- else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA", MatchAny) && (!ends_with(prev_wd, ',')))
+ else if ((Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA", MatchAny) ||
+ Matches("CREATE", "PUBLICATION", MatchAny, "SKIP", "FOR", "ALL", "TABLES", "IN", "SCHEMA", MatchAny)) &&
+ (!ends_with(prev_wd, ',')))
COMPLETE_WITH("WITH (");
/* Complete "CREATE PUBLICATION <name> [...] WITH" */
else if (HeadMatches("CREATE", "PUBLICATION") && TailMatches("WITH", "("))
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 186d8ea74b..3c09120620 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -121,6 +121,13 @@ typedef struct PublicationRelInfo
List *columns;
} PublicationRelInfo;
+typedef struct PublicationSchInfo
+{
+ NodeTag type;
+ Oid oid;
+ bool skip;
+} PublicationSchInfo;
+
extern Publication *GetPublication(Oid pubid);
extern Publication *GetPublicationByName(const char *pubname, bool missing_ok);
extern List *GetRelationPublications(Oid relid);
@@ -144,19 +151,21 @@ typedef enum PublicationPartOpt
extern List *GetPublicationRelations(Oid pubid, char objectType,
PublicationPartOpt pub_partopt);
extern List *GetAllTablesPublications(void);
-extern List *GetAllTablesPublicationRelations(bool pubviaroot);
+extern List *GetAllTablesPublicationRelations(Oid pubid, bool pubviaroot);
extern void GetActionsInPublication(Oid pubid, PublicationActions *actions);
extern List *GetPublicationSchemas(Oid pubid, char objectType);
-extern List *GetSchemaPublications(Oid schemaid, char objectType);
+extern List *GetSchemaPublications(Oid schemaid, char objectType, bool skippub);
extern List *GetSchemaPublicationRelations(Oid schemaid, char objectType,
PublicationPartOpt pub_partopt);
extern List *GetAllSchemaPublicationRelations(Oid puboid, char objectType,
- PublicationPartOpt pub_partopt);
+ PublicationPartOpt pub_partopt,
+ bool bskip);
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 List *GetAllSequencesPublications(void);
extern List *GetAllSequencesPublicationRelations(void);
@@ -165,7 +174,8 @@ extern bool is_publishable_relation(Relation rel);
extern bool is_schema_publication(Oid pubid);
extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *pri,
bool if_not_exists);
-extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
+extern ObjectAddress publication_add_schema(Oid pubid,
+ PublicationSchInfo *pubsch,
char objectType,
bool if_not_exists);
diff --git a/src/include/catalog/pg_publication_namespace.h b/src/include/catalog/pg_publication_namespace.h
index 7340a1ec64..5709efa69e 100644
--- a/src/include/catalog/pg_publication_namespace.h
+++ b/src/include/catalog/pg_publication_namespace.h
@@ -33,6 +33,7 @@ CATALOG(pg_publication_namespace,8901,PublicationNamespaceRelationId)
Oid pnpubid BKI_LOOKUP(pg_publication); /* Oid of the publication */
Oid pnnspid BKI_LOOKUP(pg_namespace); /* Oid of the schema */
char pntype; /* object type to include */
+ bool pnskip BKI_DEFAULT(f); /* skip objects */
} FormData_pg_publication_namespace;
/* ----------------
diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h
index ae87caf089..0c8ad24939 100644
--- a/src/include/commands/publicationcmds.h
+++ b/src/include/commands/publicationcmds.h
@@ -32,8 +32,10 @@ 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 puballtables);
extern bool pub_collist_contains_invalid_column(Oid pubid, Relation relation,
- List *ancestors, bool pubviaroot);
+ List *ancestors, bool pubviaroot,
+ bool puballtables);
#endif /* PUBLICATIONCMDS_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 5d075f0c34..18bba3bf3b 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -490,6 +490,7 @@ typedef enum NodeTag
T_PartitionCmd,
T_VacuumRelation,
T_PublicationObjSpec,
+ T_PublicationSchInfo,
T_PublicationTable,
/*
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 5a458c42e5..4ed0121fd8 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3679,6 +3679,7 @@ typedef struct PublicationObjSpec
PublicationObjSpecType pubobjtype; /* type of this publication object */
char *name;
PublicationTable *pubtable;
+ bool skip;
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 0308e40ba6..8844f2619d 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -84,6 +84,35 @@ DETAIL: Tables from schema cannot be added to, dropped from, or set on FOR ALL
ALTER PUBLICATION testpub_foralltables SET ALL TABLES IN SCHEMA pub_test;
ERROR: publication "testpub_foralltables" is defined as FOR ALL TABLES
DETAIL: Tables from schema cannot be added to, dropped from, or set on FOR ALL TABLES publications.
+-- should be able to add skip schema to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables ADD SKIP ALL TABLES IN SCHEMA pub_test;
+\dRp+ testpub_foralltables
+ Publication testpub_foralltables
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | t | f | t | t | f | f | f | f
+Skip tables from schemas:
+ "pub_test"
+
+-- should be able to set skip schema to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables SET SKIP ALL TABLES IN SCHEMA public;
+\dRp+ testpub_foralltables
+ Publication testpub_foralltables
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | t | f | t | t | f | f | f | f
+Skip tables from schemas:
+ "public"
+
+-- should be able to drop skip schema from 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables DROP SKIP ALL TABLES IN SCHEMA public;
+\dRp+ testpub_foralltables
+ Publication testpub_foralltables
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | t | f | t | t | f | f | f | f
+(1 row)
+
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_fortable FOR TABLE testpub_tbl1;
RESET client_min_messages;
@@ -119,6 +148,18 @@ ALTER PUBLICATION testpub_fortable SET ALL TABLES IN SCHEMA pub_test;
Tables from schemas:
"pub_test"
+-- fail - can't add skip schema to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable ADD SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES
+DETAIL: Skip tables from schema cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't drop skip schema from 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable DROP SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES
+DETAIL: Skip tables from schema cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't set skip schema to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable SET SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES
+DETAIL: Skip tables from schema cannot be added to, dropped from, or set on NON ALL TABLES publications.
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_forschema FOR ALL TABLES IN SCHEMA pub_test;
RESET client_min_messages;
@@ -144,6 +185,18 @@ ALTER PUBLICATION testpub_forschema SET TABLE pub_test.testpub_nopk;
Tables:
"pub_test.testpub_nopk"
+-- fail - can't add skip schema to schema publication
+ALTER PUBLICATION testpub_forschema ADD SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES
+DETAIL: Skip tables from schema cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't drop skip schema from schema publication
+ALTER PUBLICATION testpub_forschema DROP SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES
+DETAIL: Skip tables from schema cannot be added to, dropped from, or set on NON ALL TABLES publications.
+-- fail - can't set skip schema to schema publication
+ALTER PUBLICATION testpub_forschema SET SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES
+DETAIL: Skip tables from schema 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
----------------------+--------------
@@ -168,8 +221,34 @@ Publications:
regress_publication_user | t | f | t | t | f | f | f | f
(1 row)
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_foralltables_skipschema FOR ALL TABLES SKIP ALL TABLES IN SCHEMA pub_test;
+RESET client_min_messages;
+\dRp+ testpub_foralltables_skipschema
+ Publication testpub_foralltables_skipschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | t | f | t | t | t | t | t | f
+Skip tables from schemas:
+ "pub_test"
+
+-- fail - can't specify skip schema along with table publication
+CREATE PUBLICATION testpub_fortable_skipschema FOR TABLE pub_test.testpub_nopk, SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: SKIP ALL TABLES IN SCHEMA can be specified only with ALL TABLES option
+LINE 1: ...E pub_test.testpub_nopk, SKIP ALL TABLES IN SCHEMA pub_test;
+ ^
+-- fail - can't specify skip schema along with schema publication
+CREATE PUBLICATION testpub_forschema_skipschema FOR ALL TABLES IN SCHEMA pub_test, SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: SKIP ALL TABLES IN SCHEMA can be specified only with ALL TABLES option
+LINE 1: ...BLES IN SCHEMA pub_test, SKIP ALL TABLES IN SCHEMA pub_test;
+ ^
+-- fail - can't specify only skip schema while create publication
+CREATE PUBLICATION testpub_skipschema FOR SKIP ALL TABLES IN SCHEMA pub_test;
+ERROR: SKIP ALL TABLES IN SCHEMA can be specified only with ALL TABLES option
+LINE 1: ...N testpub_skipschema FOR SKIP ALL TABLES IN SCHEMA pub_test;
+ ^
DROP TABLE testpub_tbl2;
-DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema;
+DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_skipschema;
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 96b02947fa..e2031b4ac6 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -60,6 +60,16 @@ ALTER PUBLICATION testpub_foralltables DROP ALL TABLES IN SCHEMA pub_test;
-- fail - can't set schema to 'FOR ALL TABLES' publication
ALTER PUBLICATION testpub_foralltables SET ALL TABLES IN SCHEMA pub_test;
+-- should be able to add skip schema to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables ADD SKIP ALL TABLES IN SCHEMA pub_test;
+\dRp+ testpub_foralltables
+-- should be able to set skip schema to 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables SET SKIP ALL TABLES IN SCHEMA public;
+\dRp+ testpub_foralltables
+-- should be able to drop skip schema from 'FOR ALL TABLES' publication
+ALTER PUBLICATION testpub_foralltables DROP SKIP ALL TABLES IN SCHEMA public;
+\dRp+ testpub_foralltables
+
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_fortable FOR TABLE testpub_tbl1;
RESET client_min_messages;
@@ -73,6 +83,13 @@ 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
+-- fail - can't add skip schema to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable ADD SKIP ALL TABLES IN SCHEMA pub_test;
+-- fail - can't drop skip schema from 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable DROP SKIP ALL TABLES IN SCHEMA pub_test;
+-- fail - can't set skip schema to 'FOR TABLE' publication
+ALTER PUBLICATION testpub_fortable SET SKIP ALL TABLES IN SCHEMA pub_test;
+
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_forschema FOR ALL TABLES IN SCHEMA pub_test;
RESET client_min_messages;
@@ -87,12 +104,34 @@ 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 skip schema to schema publication
+ALTER PUBLICATION testpub_forschema ADD SKIP ALL TABLES IN SCHEMA pub_test;
+-- fail - can't drop skip schema from schema publication
+ALTER PUBLICATION testpub_forschema DROP SKIP ALL TABLES IN SCHEMA pub_test;
+-- fail - can't set skip schema to schema publication
+ALTER PUBLICATION testpub_forschema SET SKIP ALL TABLES IN SCHEMA pub_test;
+
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_skipschema FOR ALL TABLES SKIP ALL TABLES IN SCHEMA pub_test;
+RESET client_min_messages;
+
+\dRp+ testpub_foralltables_skipschema
+
+-- fail - can't specify skip schema along with table publication
+CREATE PUBLICATION testpub_fortable_skipschema FOR TABLE pub_test.testpub_nopk, SKIP ALL TABLES IN SCHEMA pub_test;
+
+-- fail - can't specify skip schema along with schema publication
+CREATE PUBLICATION testpub_forschema_skipschema FOR ALL TABLES IN SCHEMA pub_test, SKIP ALL TABLES IN SCHEMA pub_test;
+
+-- fail - can't specify only skip schema while create publication
+CREATE PUBLICATION testpub_skipschema FOR SKIP ALL TABLES IN SCHEMA pub_test;
+
DROP TABLE testpub_tbl2;
-DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema;
+DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_skipschema;
CREATE TABLE testpub_tbl3 (a int);
CREATE TABLE testpub_tbl3a (b text) INHERITS (testpub_tbl3);
diff --git a/src/test/subscription/t/032_rep_changes_skip_schema.pl b/src/test/subscription/t/032_rep_changes_skip_schema.pl
new file mode 100644
index 0000000000..7a16ad350c
--- /dev/null
+++ b/src/test/subscription/t/032_rep_changes_skip_schema.pl
@@ -0,0 +1,96 @@
+
+# Copyright (c) 2021-2022, PostgreSQL Global Development Group
+
+# Logical replication tests for skip schema 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 SKIP ALL TABLES IN SCHEMA
+# 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 SKIP ALL TABLES IN SCHEMA sch1");
+
+$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 schema table data does not sync for skip schemas
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), min(a), max(a) FROM sch1.tab1");
+is($result, qq(0||), 'check tablesync is skipped for skip schemas');
+
+# 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 skip data changes in public and verify that subscriber does not get
+# the new table data.
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_schema add SKIP ALL TABLES IN SCHEMA public");
+$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 skip schema public and verify that subscriber gets
+# the new table data.
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_schema drop SKIP ALL TABLES IN SCHEMA public");
+$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();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 85c808af90..7627b46547 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2065,6 +2065,7 @@ PublicationObjSpecType
PublicationPartOpt
PublicationRelInfo
PublicationSchemaInfo
+PublicationSchInfo
PublicationTable
PullFilter
PullFilterOps
--
2.32.0
