From 604db8ac5ce06da98ab6aeb19915b239a7541a83 Mon Sep 17 00:00:00 2001 From: Vigneshwaran C Date: Tue, 12 Apr 2022 10:55:34 +0530 Subject: [PATCH v1 2/2] Skip publishing the tables. A new option "SKIP TABLE" in Create/Alter Publication allows one or more skip tables to be specified, publisher will skip sending the data of the tables present in the skip table to the subscriber. The new syntax allows specifying schemas. For example: CREATE PUBLICATION pub1 FOR ALL TABLES SKIP TABLE t1,t2; OR ALTER PUBLICATION pub1 ADD SKIP TABLE t1,t2; A new column pnskip is added to table "pg_publication_rel", to maintain the relations 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 table publication. Updates pg_dump to identify and dump skip table publications. Updates the \d family of commands to display skip table publications and \dRp+ variant will now display associated skip tables if any. Bump catalog version. --- doc/src/sgml/catalogs.sgml | 13 ++- doc/src/sgml/logical-replication.sgml | 10 +- doc/src/sgml/ref/alter_publication.sgml | 22 +++-- doc/src/sgml/ref/create_publication.sgml | 29 +++++- doc/src/sgml/ref/psql-ref.sgml | 4 +- src/backend/catalog/pg_publication.c | 31 ++++-- src/backend/commands/publicationcmds.c | 62 +++++++----- src/backend/commands/tablecmds.c | 4 +- src/backend/parser/gram.y | 45 +++++---- src/backend/replication/pgoutput/pgoutput.c | 8 +- src/backend/utils/cache/relcache.c | 18 ++-- 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 | 22 ++++- src/bin/psql/tab-complete.c | 17 ++-- src/include/catalog/pg_publication.h | 3 +- src/include/catalog/pg_publication_rel.h | 1 + src/include/nodes/parsenodes.h | 1 + src/test/regress/expected/publication.out | 87 ++++++++++++++++- src/test/regress/sql/publication.sql | 41 +++++++- .../t/033_rep_changes_skip_table.pl | 96 +++++++++++++++++++ 23 files changed, 481 insertions(+), 99 deletions(-) create mode 100644 src/test/subscription/t/033_rep_changes_skip_table.pl diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 1b02af1c03..fb1446cb07 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -6354,6 +6354,15 @@ SCRAM-SHA-256$<iteration count>:&l Reference to schema + + + + pnskip bool + + + True if the schema is skip schema + + @@ -6441,10 +6450,10 @@ SCRAM-SHA-256$<iteration count>:&l - pnskip bool + prskip bool - True if the schema is skip schema + True if the table is skip table diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index e2a4b89226..e16d4b2f86 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -598,11 +598,11 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER - To add tables to a publication, the user must have ownership rights on the - 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. + To add tables or skip tables to a publication, the user must have ownership + rights on the 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. diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml index 6f915d8c5d..f87a46d1e6 100644 --- a/doc/src/sgml/ref/alter_publication.sgml +++ b/doc/src/sgml/ref/alter_publication.sgml @@ -30,7 +30,7 @@ ALTER PUBLICATION name RENAME TO where publication_object is one of: - TABLE [ ONLY ] table_name [ * ] [ ( column_name [, ... ] ) ] [ WHERE ( expression ) ] [, ... ] + [SKIP] TABLE [ ONLY ] table_name [ * ] [ ( column_name [, ... ] ) ] [ WHERE ( expression ) ] [, ... ] [SKIP] ALL TABLES IN SCHEMA { schema_name | CURRENT_SCHEMA } [, ... ] @@ -70,8 +70,8 @@ ALTER PUBLICATION name RENAME TO You must own the publication to use ALTER PUBLICATION. - Adding a table to a publication additionally requires owning that table. - The ADD [SKIP] ALL TABLES IN SCHEMA and + Adding a table or a skip table to a publication additionally requires owning + that table. The ADD [SKIP] ALL TABLES IN SCHEMA and SET [SKIP] ALL TABLES IN SCHEMA 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 @@ -90,10 +90,11 @@ ALTER PUBLICATION name RENAME TO - The ADD SKIP ALL TABLES IN SCHEMA and - SET SKIP ALL TABLES IN SCHEMA can be specified only for + Adding/Dropping/Setting SKIP ALL TABLES IN SCHEMA, and + SKIP TABLE can be specified only for FOR ALL TABLES publication. It is not supported for - FOR ALL TABLES IN SCHEMA publication and + FOR ALL TABLES IN SCHEMA publication, + FOR ALL SEQUENCES IN SCHEMA publication and FOR TABLE publication. @@ -232,6 +233,15 @@ ALTER PUBLICATION production_publication ADD TABLE users, departments, ALL TABLE mypublication: ALTER PUBLICATION mypublication ADD SKIP ALL TABLES IN SCHEMA sales_june, sales_july; + + + + + Add skip tables users and + departments to the publication + mypublication: + +ALTER PUBLICATION mypublication ADD SKIP TABLE users, departments; diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml index 5acf80452a..43a1a1d718 100644 --- a/doc/src/sgml/ref/create_publication.sgml +++ b/doc/src/sgml/ref/create_publication.sgml @@ -22,7 +22,7 @@ PostgreSQL documentation CREATE PUBLICATION name - [ FOR ALL TABLES [SKIP ALL TABLES IN SCHEMA { schema_name | CURRENT_SCHEMA }] + [ FOR ALL TABLES [SKIP ALL TABLES IN SCHEMA { schema_name | CURRENT_SCHEMA }] [SKIP TABLE [ ONLY ] table_name [ * ]] | FOR publication_object [, ... ] ] [ WITH ( publication_parameter [= value] [, ... ] ) ] @@ -124,6 +124,7 @@ CREATE PUBLICATION name + SKIP ALL TABLES IN SCHEMA @@ -141,6 +142,24 @@ CREATE PUBLICATION name + + + SKIP TABLE + + + Marks the publication as one that skips replicating changes for the + specified tables. + + + + SKIP TABLE can be specified only for + FOR ALL TABLES publication. It is not supported for + FOR ALL TABLES IN SCHEMA publication and + FOR TABLE publication. + + + + FOR ALL TABLES IN SCHEMA @@ -360,6 +379,14 @@ CREATE PUBLICATION sales_publication FOR ALL TABLES IN SCHEMA marketing, sales; marketing and sales: CREATE PUBLICATION mypublication FOR ALL TABLE SKIP ALL TABLES IN SCHEMA marketing, sales; + + + + Create a publication that publishes all changes in all the tables except for + the changes of users and + departments table; + +CREATE PUBLICATION mypublication FOR ALL TABLE SKIP TABLE users, departments; diff --git a/doc/src/sgml/ref/psql-ref.sgml b/doc/src/sgml/ref/psql-ref.sgml index b097b863cd..d2ae2f757f 100644 --- a/doc/src/sgml/ref/psql-ref.sgml +++ b/doc/src/sgml/ref/psql-ref.sgml @@ -1869,8 +1869,8 @@ testdb=> specified, only those publications whose names match the pattern are listed. If + is appended to the command name, the tables, - schemas and the skip schema associated with each publication are shown - as well. + the skip tables, schemas and the skip schema associated with each + publication are shown as well. diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 3d2cab47a6..291e9746c9 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -303,9 +303,10 @@ GetTopMostAncestorInPublication(Oid puboid, List *ancestors, foreach(lc, ancestors) { Oid ancestor = lfirst_oid(lc); - List *apubids = GetRelationPublications(ancestor); + List *apubids = GetRelationPublications(ancestor, false); List *aschemaPubids = NIL; List *askipschemaPubids = NIL; + List *askippubids; level++; @@ -320,9 +321,11 @@ GetTopMostAncestorInPublication(Oid puboid, List *ancestors, { aschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor), false); askipschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor), true); + askippubids = GetRelationPublications(ancestor, true); if (list_member_oid(aschemaPubids, puboid) || - (puballtables && !list_member_oid(askipschemaPubids, puboid))) + (puballtables && !list_member_oid(askipschemaPubids, puboid) && + !list_member_oid(askippubids, puboid))) { topmost_relid = ancestor; @@ -401,6 +404,9 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri, ObjectIdGetDatum(pubid); values[Anum_pg_publication_rel_prrelid - 1] = ObjectIdGetDatum(relid); + values[Anum_pg_publication_rel_prskip - 1] = + BoolGetDatum(pri->skip); + /* Add qualifications, if available */ if (pri->whereClause != NULL) @@ -674,7 +680,7 @@ publication_add_schema(Oid pubid, PublicationSchInfo *pubsch, bool if_not_exists /* Gets list of publication oids for a relation */ List * -GetRelationPublications(Oid relid) +GetRelationPublications(Oid relid, bool bskip) { List *result = NIL; CatCList *pubrellist; @@ -688,7 +694,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 (bskip == ((Form_pg_publication_rel) GETSTRUCT(tup))->prskip) + result = lappend_oid(result, pubid); } ReleaseSysCacheList(pubrellist); @@ -728,6 +735,7 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) Form_pg_publication_rel pubrel; pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + result = GetPubPartitionOptionRelations(result, pub_partopt, pubrel->prrelid); } @@ -794,8 +802,15 @@ GetAllTablesPublicationRelations(Oid pubid, bool pubviaroot) TableScanDesc scan; HeapTuple tuple; List *result = NIL; - List *skipschemaidlist = NIL; + + /* + * pg_publication_rel and pg_publication_namespace will only have skip + * tables and skip namespaces in case of all tables publication, no need + * to pass skip flag to get the relations. + */ List *pubschemalist = GetPublicationSchemas(pubid); + List *skippubtablelist = GetPublicationRelations(pubid, PUBLICATION_PART_ALL); + List *skipschemaidlist = NIL; ListCell *cell; foreach(cell, pubschemalist) @@ -822,7 +837,8 @@ GetAllTablesPublicationRelations(Oid pubid, bool pubviaroot) if (is_publishable_class(relid, relForm) && !(relForm->relispartition && pubviaroot) && - !list_member_oid(skipschemaidlist, schid)) + !list_member_oid(skipschemaidlist, schid) && + !list_member_oid(skippubtablelist, relid)) result = lappend_oid(result, relid); } @@ -845,7 +861,8 @@ GetAllTablesPublicationRelations(Oid pubid, bool pubviaroot) if (is_publishable_class(relid, relForm) && !relForm->relispartition && - !list_member_oid(skipschemaidlist, schid)) + !list_member_oid(skipschemaidlist, schid) && + !list_member_oid(skippubtablelist, relid)) result = lappend_oid(result, relid); } diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 6863900bd9..b26e606c7e 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -849,37 +849,32 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) /* Associate objects with the publication. */ if (stmt->for_all_tables) { - Assert(!relations); - /* Invalidate relcache so that publication info is rebuilt. */ CacheInvalidateRelcacheAll(); } - else - { - /* FOR [SKIP] 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 [SKIP] 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); } /* tables added through a schema */ @@ -1371,6 +1366,8 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, bool nonskipschema = false; bool skipschema = false; + bool nonskiptable = false; + bool skiptable = false; foreach(lc, schemaidlist) { @@ -1382,6 +1379,16 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, skipschema = true; } + foreach(lc, tables) + { + PublicationTable *pub_table = lfirst_node(PublicationTable, lc); + + if (!pub_table->skip) + nonskiptable = true; + else + skiptable = true; + } + if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) && schemaidlist && !superuser()) ereport(ERROR, @@ -1407,12 +1414,19 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup, 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 tables. */ - if (tables && pubform->puballtables) + if (nonskiptable && 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 (skiptable && !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 table cannot be added to, dropped from, or set on NON ALL TABLES publications."))); } /* @@ -1689,6 +1703,7 @@ OpenTableList(List *tables) pub_rel->relation = rel; pub_rel->whereClause = t->whereClause; pub_rel->columns = t->columns; + pub_rel->skip = t->skip; rels = lappend(rels, pub_rel); relids = lappend_oid(relids, myrelid); @@ -1761,6 +1776,7 @@ OpenTableList(List *tables) /* child inherits column list from parent */ pub_rel->columns = t->columns; + pub_rel->skip = t->skip; 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 141a2eabf8..f9655c140f 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -16276,7 +16276,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", @@ -16413,7 +16413,7 @@ AlterTableNamespace(AlterObjectSchemaStmt *stmt, Oid *oldschema) { ListCell *lc; List *schemaPubids = GetSchemaPublications(nspOid, false); - 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 7fd8d194ce..be3d49f607 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -498,7 +498,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type opt_interval interval_second %type unicode_normal_form -%type opt_instead +%type opt_instead opt_skip %type opt_unique opt_concurrently opt_verbose opt_full %type opt_freeze opt_analyze opt_default opt_recheck %type opt_binary copy_delimiter @@ -9923,14 +9923,16 @@ CreatePublicationStmt: * relation_expr here. */ PublicationObjSpec: - TABLE relation_expr opt_column_list OptWhereClause + opt_skip TABLE relation_expr opt_column_list OptWhereClause { $$ = makeNode(PublicationObjSpec); $$->pubobjtype = PUBLICATIONOBJ_TABLE; $$->pubtable = makeNode(PublicationTable); - $$->pubtable->relation = $2; - $$->pubtable->columns = $3; - $$->pubtable->whereClause = $4; + $$->skip = $1; + $$->pubtable->skip = $1; + $$->pubtable->relation = $3; + $$->pubtable->columns = $4; + $$->pubtable->whereClause = $5; } | ALL TABLES IN_P SCHEMA ColId { @@ -10024,6 +10026,17 @@ pub_obj_list: PublicationObjSpec { $$ = lappend($1, $3); } ; + skip_pub_obj_list: pub_obj_list + { $$ = $1; } + | /*EMPTY*/ + { $$ = NULL; } + ; + +opt_skip: + SKIP { $$ = true; } + | /*EMPTY*/ { $$ = false; } + ; + /***************************************************************************** * * ALTER PUBLICATION name SET ( options ) @@ -10285,13 +10298,6 @@ opt_instead: | /*EMPTY*/ { $$ = false; } ; - - skip_pub_obj_list: pub_obj_list - { $$ = $1; } - | /*EMPTY*/ - { $$ = NULL; } - ; - /***************************************************************************** * * QUERY: @@ -18789,6 +18795,8 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner) pubobj->pubtable = pubtable; pubobj->name = NULL; } + + pubobj->pubtable->skip = pubobj->skip; } else if (pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA || pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA) @@ -18846,11 +18854,10 @@ preprocess_alltables_pubobj_list(List *pubobjspec_list, int location, 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) + if (!pubobj->skip) ereport(ERROR, errcode(ERRCODE_SYNTAX_ERROR), - errmsg("only SKIP ALL TABLES IN SCHEMA can be specified with ALL TABLES option"), + errmsg("only SKIP ALL TABLES IN SCHEMA or SKIP TABLE can be specified with ALL TABLES option"), parser_errposition(pubobj->location)); } } @@ -18872,12 +18879,12 @@ check_skip_in_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner) { pubobj = (PublicationObjSpec *) lfirst(cell); - /* Only SKIP ALL TABLES IN SCHEMA option supported with ALL TABLES */ - if (pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA && - pubobj->skip) + /* SKIP ALL TABLES IN SCHEMA/SKIP TABLE option supported only with ALL TABLES */ + if ((pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA || + pubobj->pubobjtype == PUBLICATIONOBJ_TABLE) && pubobj->skip) ereport(ERROR, errcode(ERRCODE_SYNTAX_ERROR), - errmsg("SKIP ALL TABLES IN SCHEMA can be specified only with ALL TABLES option"), + errmsg("SKIP ALL TABLES IN SCHEMA/SKIP 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 19181297af..b81ac797b8 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -1989,7 +1989,7 @@ 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); /* * We don't acquire a lock on the namespace system table as we build @@ -1998,6 +1998,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) */ List *schemaPubids = GetSchemaPublications(schemaId, false); List *skipSchemaPubids = GetSchemaPublications(schemaId, true); + List *skipTablePubids = GetRelationPublications(relid, true); ListCell *lc; Oid publish_as_relid = relid; int publish_ancestor_level = 0; @@ -2113,7 +2114,9 @@ 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)) || + (pub->alltables && + !list_member_oid(skipSchemaPubids, pub->oid) && + !list_member_oid(skipTablePubids, pub->oid)) || ancestor_published) publish = true; } @@ -2190,6 +2193,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) list_free(pubids); list_free(schemaPubids); list_free(skipSchemaPubids); + list_free(skipTablePubids); 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 02c0a47ca1..5b51185656 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5563,7 +5563,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) { List *puboids; List *alltablespuboids; - List *skipschemapuboids; + List *skippuboids; ListCell *lc; MemoryContext oldcxt; Oid schemaid; @@ -5597,10 +5597,10 @@ 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, false)); - skipschemapuboids = GetSchemaPublications(schemaid, true); + skippuboids = GetSchemaPublications(schemaid, true); if (relation->rd_rel->relispartition) { @@ -5612,20 +5612,22 @@ 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, false)); - skipschemapuboids = list_concat_unique_oid(skipschemapuboids, - GetSchemaPublications(schemaid, - true)); + skippuboids = list_concat_unique_oid(skippuboids, + GetSchemaPublications(schemaid, + true)); + skippuboids = list_concat_unique_oid(skippuboids, + GetRelationPublications(ancestor, true)); } } alltablespuboids = GetAllTablesPublications(); puboids = list_concat_unique_oid(puboids, list_difference_oid(alltablespuboids, - skipschemapuboids)); + skippuboids)); foreach(lc, puboids) { Oid pubid = lfirst_oid(lc); diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 296527bde5..ee57de9890 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -282,7 +282,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 bskip); static void dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo); static void dumpDatabase(Archive *AH); static void dumpDatabaseConfig(Archive *AH, PQExpBuffer outbuf, @@ -4099,6 +4100,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) int i_prrelid; int i_prrelqual; int i_prattrs; + int i_prskip; int i, j, ntups; @@ -4111,7 +4113,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, prskip," "pg_catalog.pg_get_expr(prqual, prrelid) AS prrelqual, " "(CASE\n" " WHEN pr.prattrs IS NOT NULL THEN\n" @@ -4137,6 +4139,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) i_prrelid = PQfnumber(res, "prrelid"); i_prrelqual = PQfnumber(res, "prrelqual"); i_prattrs = PQfnumber(res, "prattrs"); + i_prskip = PQfnumber(res, "prskip"); /* this allocation may be more than we need */ pubrinfo = pg_malloc(ntups * sizeof(PublicationRelInfo)); @@ -4148,6 +4151,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) Oid prrelid = atooid(PQgetvalue(res, i, i_prrelid)); PublicationInfo *pubinfo; TableInfo *tbinfo; + char *prskip = pg_strdup(PQgetvalue(res, i, i_prskip)); /* * Ignore any entries for which we aren't interested in either the @@ -4168,7 +4172,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(prskip, "f") == 0) + pubrinfo[j].dobj.objType = DO_PUBLICATION_REL; + else + pubrinfo[j].dobj.objType = DO_PUBLICATION_SKIP_REL; + pubrinfo[j].dobj.catId.tableoid = atooid(PQgetvalue(res, i, i_tableoid)); pubrinfo[j].dobj.catId.oid = atooid(PQgetvalue(res, i, i_oid)); @@ -4267,13 +4275,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 bskip) { 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) @@ -4283,8 +4293,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 (bskip) + appendPQExpBufferStr(query, "SKIP "); + + appendPQExpBufferStr(query, "TABLE ONLY"); + description = (bskip) ? "SKIP PUBLICATION TABLE" : "PUBLICATION TABLE"; + appendPQExpBuffer(query, " %s", fmtQualifiedDumpable(tbinfo)); @@ -4313,7 +4330,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)); @@ -9889,7 +9906,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_SKIP_REL: + dumpPublicationTable(fout, (const PublicationRelInfo *) dobj, true); break; case DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA: dumpPublicationNamespace(fout, @@ -17828,6 +17848,7 @@ addBoundaryDependencies(DumpableObject **dobjs, int numObjs, case DO_POLICY: case DO_PUBLICATION: case DO_PUBLICATION_REL: + case DO_PUBLICATION_SKIP_REL: case DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA: case DO_PUBLICATION_TABLE_IN_SCHEMA: case DO_SUBSCRIPTION: diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index cb9e5e164b..7dbd0f6538 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_REL, DO_PUBLICATION_SKIP_TABLE_IN_SCHEMA, 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 47d4baecb3..4336f4f674 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_REL, PRIO_PUBLICATION_SKIP_TABLE_IN_SCHEMA, PRIO_PUBLICATION_TABLE_IN_SCHEMA, PRIO_SUBSCRIPTION, @@ -146,6 +147,7 @@ static const int dbObjectTypePriority[] = PRIO_POLICY, /* DO_POLICY */ PRIO_PUBLICATION, /* DO_PUBLICATION */ PRIO_PUBLICATION_REL, /* DO_PUBLICATION_REL */ + PRIO_PUBLICATION_SKIP_REL, /* DO_PUBLICATION_SKIP_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 */ @@ -1490,6 +1492,11 @@ describeDumpableObject(DumpableObject *obj, char *buf, int bufsize) "PUBLICATION TABLE (ID %d OID %u)", obj->dumpId, obj->catId.oid); return; + case DO_PUBLICATION_SKIP_REL: + snprintf(buf, bufsize, + "PUBLICATION SKIP 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)", diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl index 2dbb43c30e..b902cca44d 100644 --- a/src/bin/pg_dump/t/002_pg_dump.pl +++ b/src/bin/pg_dump/t/002_pg_dump.pl @@ -2463,6 +2463,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');\E + /xm, + like => { %full_runs, section_post_data => 1, }, + }, + 'CREATE SUBSCRIPTION sub1' => { create_order => 50, create_sql => 'CREATE SUBSCRIPTION sub1 @@ -2588,6 +2597,20 @@ my %tests = ( like => { %full_runs, section_post_data => 1, }, }, + 'ALTER PUBLICATION pub6 ADD SKIP TABLE test_table' => { + create_order => 52, + create_sql => + 'ALTER PUBLICATION pub6 ADD SKIP TABLE dump_test.test_table;', + regexp => qr/^ + \QALTER PUBLICATION pub6 ADD SKIP 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 dd6b63e7e2..3d61317e9a 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -2911,7 +2911,7 @@ 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.prskip = 'f'\n" "UNION\n" "SELECT pubname\n" " , NULL\n" @@ -2923,8 +2923,13 @@ describeOneTableDetails(const char *schemaname, " JOIN pg_catalog.pg_class pc\n" " ON pc.relnamespace = pn.pnnspid\n" " WHERE pc.oid ='%s' AND pn.pnpubid = p.oid)\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, oid, oid); } else { @@ -6159,6 +6164,7 @@ describePublications(const char *pattern) "WHERE c.relnamespace = n.oid\n" " AND c.oid = pr.prrelid\n" " AND pr.prpubid = '%s'\n" + " AND pr.prskip = 'f'\n" "ORDER BY 1,2", pubid); if (!addFooterToPublicationDesc(&buf, _("Tables:"), false, &cont)) goto error_return; @@ -6191,6 +6197,18 @@ describePublications(const char *pattern) if (!addFooterToPublicationDesc(&buf, "Skip tables from schemas:", true, &cont)) goto error_return; + + /* Get the skip 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.prskip = 't'\n" + "ORDER BY 1", pubid); + if (!addFooterToPublicationDesc(&buf, "Skip tables:", + true, &cont)) + goto error_return; } printTable(&cont, pset.queryFout, false, pset.logfile); diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index 402ab90b58..7d4c3ef405 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1828,10 +1828,11 @@ psql_completion(const char *text, int start, int end) COMPLETE_WITH("ADD", "DROP", "OWNER TO", "RENAME TO", "SET"); /* ALTER PUBLICATION ADD */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD")) - COMPLETE_WITH("ALL TABLES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "TABLE"); + COMPLETE_WITH("ALL TABLES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "SKIP TABLE", "TABLE"); else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD", "SKIP")) - COMPLETE_WITH("ALL TABLES IN SCHEMA"); + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") || + Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "SKIP", "TABLE") || (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") && ends_with(prev_wd, ','))) COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables); @@ -1853,14 +1854,14 @@ psql_completion(const char *text, int start, int end) COMPLETE_WITH(","); /* ALTER PUBLICATION DROP */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP")) - COMPLETE_WITH("ALL TABLES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "TABLE"); + COMPLETE_WITH("ALL TABLES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "SKIP TABLE", "TABLE"); else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP", "SKIP")) - COMPLETE_WITH("ALL TABLES IN SCHEMA"); + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); /* ALTER PUBLICATION SET */ else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET")) - COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "TABLE"); + COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "SKIP ALL TABLES IN SCHEMA", "SKIP TABLE", "TABLE"); else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET", "SKIP")) - COMPLETE_WITH("ALL TABLES IN SCHEMA"); + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES", "IN", "SCHEMA") || Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "SKIP", "ALL", "TABLES", "IN", "SCHEMA")) COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas @@ -2998,7 +2999,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 (", "SKIP ALL TABLES IN SCHEMA"); + COMPLETE_WITH("IN SCHEMA", "WITH (", "SKIP ALL TABLES IN SCHEMA", "SKIP TABLE"); + else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES", "SKIP")) + COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE"); else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE", MatchAny) && !ends_with(prev_wd, ',')) COMPLETE_WITH("WHERE (", "WITH ("); /* Complete "CREATE PUBLICATION FOR TABLE" with ", ..." */ diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 30a2fcb974..12fca8ea2b 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -108,6 +108,7 @@ typedef struct PublicationRelInfo Relation relation; Node *whereClause; List *columns; + bool skip; } PublicationRelInfo; typedef struct PublicationSchInfo @@ -119,7 +120,7 @@ typedef struct PublicationSchInfo 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 bskip); /*--------- * Expected values for pub_partopt parameter of GetRelationPublications(), diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h index 4feb581899..ab6efc5b30 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 prskip BKI_DEFAULT(f); /* skip the relation */ #ifdef CATALOG_VARLEN /* variable-length fields start here */ pg_node_tree prqual; /* qualifications */ diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 188285c99d..86372a1531 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 skip; /* skip relation */ } PublicationTable; /* diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out index e7e46d0841..0a36ec06b9 100644 --- a/src/test/regress/expected/publication.out +++ b/src/test/regress/expected/publication.out @@ -110,6 +110,35 @@ ALTER PUBLICATION testpub_foralltables DROP SKIP ALL TABLES IN SCHEMA public; regress_publication_user | t | t | t | f | f | f (1 row) +-- should be able to add skip table to 'FOR ALL TABLES' publication +ALTER PUBLICATION testpub_foralltables ADD SKIP 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 +Skip tables: + "public.testpub_tbl1" + +-- should be able to set skip table to 'FOR ALL TABLES' publication +ALTER PUBLICATION testpub_foralltables SET SKIP 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 +Skip tables: + "public.testpub_tbl2" + +-- should be able to drop skip table from 'FOR ALL TABLES' publication +ALTER PUBLICATION testpub_foralltables DROP SKIP 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_fortable FOR TABLE testpub_tbl1; RESET client_min_messages; @@ -157,6 +186,18 @@ DETAIL: Skip tables from schema cannot be added to, dropped from, or set on NON 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. +-- fail - can't add skip table to 'FOR TABLE' publication +ALTER PUBLICATION testpub_fortable ADD SKIP TABLE testpub_tbl1; +ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES +DETAIL: Skip table cannot be added to, dropped from, or set on NON ALL TABLES publications. +-- fail - can't drop skip table from 'FOR TABLE' publication +ALTER PUBLICATION testpub_fortable DROP SKIP TABLE testpub_tbl1; +ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES +DETAIL: Skip table cannot be added to, dropped from, or set on NON ALL TABLES publications. +-- fail - can't set skip table to 'FOR TABLE' publication +ALTER PUBLICATION testpub_fortable SET SKIP TABLE testpub_tbl1; +ERROR: publication "testpub_fortable" is not defined as FOR ALL TABLES +DETAIL: Skip table 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; @@ -194,6 +235,18 @@ DETAIL: Skip tables from schema cannot be added to, dropped from, or set on NON 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. +-- fail - can't add skip table to schema publication +ALTER PUBLICATION testpub_forschema ADD SKIP TABLE testpub_tbl1; +ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES +DETAIL: Skip table cannot be added to, dropped from, or set on NON ALL TABLES publications. +-- fail - can't drop skip table from schema publication +ALTER PUBLICATION testpub_forschema DROP SKIP TABLE testpub_tbl1; +ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES +DETAIL: Skip table cannot be added to, dropped from, or set on NON ALL TABLES publications. +-- fail - can't set skip table to schema publication +ALTER PUBLICATION testpub_forschema SET SKIP TABLE testpub_tbl1; +ERROR: publication "testpub_forschema" is not defined as FOR ALL TABLES +DETAIL: Skip 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 ----------------------+-------------- @@ -231,21 +284,47 @@ Skip tables from schemas: -- 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 +ERROR: SKIP ALL TABLES IN SCHEMA/SKIP TABLE 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 +ERROR: SKIP ALL TABLES IN SCHEMA/SKIP TABLE 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 +ERROR: SKIP ALL TABLES IN SCHEMA/SKIP TABLE can be specified only with ALL TABLES option LINE 1: ...N testpub_skipschema FOR SKIP ALL TABLES IN SCHEMA pub_test; ^ +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_foralltables_skiptable FOR ALL TABLES SKIP TABLE testpub_tbl1; +RESET client_min_messages; +\dRp+ testpub_foralltables_skiptable + Publication testpub_foralltables_skiptable + Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root +--------------------------+------------+---------+---------+---------+-----------+---------- + regress_publication_user | t | t | t | t | t | f +Skip tables: + "public.testpub_tbl1" + +-- fail - can't specify skip table along with table publication +CREATE PUBLICATION testpub_fortable_skiptable FOR TABLE pub_test.testpub_nopk, SKIP TABLE testpub_tbl1; +ERROR: SKIP ALL TABLES IN SCHEMA/SKIP TABLE can be specified only with ALL TABLES option +LINE 1: CREATE PUBLICATION testpub_fortable_skiptable FOR TABLE pub_... + ^ +-- fail - can't specify skip table along with schema publication +CREATE PUBLICATION testpub_fortable_skiptable FOR ALL TABLES IN SCHEMA pub_test, SKIP TABLE testpub_tbl1; +ERROR: SKIP ALL TABLES IN SCHEMA/SKIP TABLE can be specified only with ALL TABLES option +LINE 1: CREATE PUBLICATION testpub_fortable_skiptable FOR ALL TABLES... + ^ +-- fail - can't specify only skip table while create publication +CREATE PUBLICATION testpub_fortable_skiptable FOR SKIP TABLE testpub_tbl1; +ERROR: SKIP ALL TABLES IN SCHEMA/SKIP TABLE can be specified only with ALL TABLES option +LINE 1: CREATE PUBLICATION testpub_fortable_skiptable FOR SKIP TABLE... + ^ DROP TABLE testpub_tbl2; -DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_skipschema; +DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_skipschema, testpub_foralltables_skiptable; 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 8d8a522c76..4a1d2f7013 100644 --- a/src/test/regress/sql/publication.sql +++ b/src/test/regress/sql/publication.sql @@ -68,6 +68,16 @@ ALTER PUBLICATION testpub_foralltables SET SKIP ALL TABLES IN SCHEMA public; ALTER PUBLICATION testpub_foralltables DROP SKIP ALL TABLES IN SCHEMA public; \dRp+ testpub_foralltables +-- should be able to add skip table to 'FOR ALL TABLES' publication +ALTER PUBLICATION testpub_foralltables ADD SKIP TABLE testpub_tbl1; +\dRp+ testpub_foralltables +-- should be able to set skip table to 'FOR ALL TABLES' publication +ALTER PUBLICATION testpub_foralltables SET SKIP TABLE testpub_tbl2; +\dRp+ testpub_foralltables +-- should be able to drop skip table from 'FOR ALL TABLES' publication +ALTER PUBLICATION testpub_foralltables DROP SKIP TABLE testpub_tbl2; +\dRp+ testpub_foralltables + SET client_min_messages = 'ERROR'; CREATE PUBLICATION testpub_fortable FOR TABLE testpub_tbl1; RESET client_min_messages; @@ -88,6 +98,13 @@ 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; +-- fail - can't add skip table to 'FOR TABLE' publication +ALTER PUBLICATION testpub_fortable ADD SKIP TABLE testpub_tbl1; +-- fail - can't drop skip table from 'FOR TABLE' publication +ALTER PUBLICATION testpub_fortable DROP SKIP TABLE testpub_tbl1; +-- fail - can't set skip table to 'FOR TABLE' publication +ALTER PUBLICATION testpub_fortable SET SKIP TABLE testpub_tbl1; + SET client_min_messages = 'ERROR'; CREATE PUBLICATION testpub_forschema FOR ALL TABLES IN SCHEMA pub_test; RESET client_min_messages; @@ -109,6 +126,13 @@ 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; +-- fail - can't add skip table to schema publication +ALTER PUBLICATION testpub_forschema ADD SKIP TABLE testpub_tbl1; +-- fail - can't drop skip table from schema publication +ALTER PUBLICATION testpub_forschema DROP SKIP TABLE testpub_tbl1; +-- fail - can't set skip table to schema publication +ALTER PUBLICATION testpub_forschema SET SKIP TABLE testpub_tbl1; + SELECT pubname, puballtables FROM pg_publication WHERE pubname = 'testpub_foralltables'; \d+ testpub_tbl2 \dRp+ testpub_foralltables @@ -128,8 +152,23 @@ CREATE PUBLICATION testpub_forschema_skipschema FOR ALL TABLES IN SCHEMA pub_tes -- fail - can't specify only skip schema while create publication CREATE PUBLICATION testpub_skipschema FOR SKIP ALL TABLES IN SCHEMA pub_test; +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_foralltables_skiptable FOR ALL TABLES SKIP TABLE testpub_tbl1; +RESET client_min_messages; + +\dRp+ testpub_foralltables_skiptable + +-- fail - can't specify skip table along with table publication +CREATE PUBLICATION testpub_fortable_skiptable FOR TABLE pub_test.testpub_nopk, SKIP TABLE testpub_tbl1; + +-- fail - can't specify skip table along with schema publication +CREATE PUBLICATION testpub_fortable_skiptable FOR ALL TABLES IN SCHEMA pub_test, SKIP TABLE testpub_tbl1; + +-- fail - can't specify only skip table while create publication +CREATE PUBLICATION testpub_fortable_skiptable FOR SKIP TABLE testpub_tbl1; + DROP TABLE testpub_tbl2; -DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_skipschema; +DROP PUBLICATION testpub_foralltables, testpub_fortable, testpub_forschema, testpub_foralltables_skipschema, testpub_foralltables_skiptable; 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_skip_table.pl b/src/test/subscription/t/033_rep_changes_skip_table.pl new file mode 100644 index 0000000000..6c1dc6f382 --- /dev/null +++ b/src/test/subscription/t/033_rep_changes_skip_table.pl @@ -0,0 +1,96 @@ + +# Copyright (c) 2021-2022, PostgreSQL Global Development Group + +# Logical replication tests for skip 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 SKIP 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 SKIP 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 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.tab1 and verify that subscriber does not get +# the new table data. +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_schema add SKIP 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 skip table public.tab1 and verify that subscriber gets +# the new table data. +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_schema drop SKIP 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