From 9c9d904c721d4d89ea475ea959f3dc96f16fa48e Mon Sep 17 00:00:00 2001 From: Tomas Vondra Date: Wed, 23 Feb 2022 21:15:18 +0100 Subject: [PATCH 3/3] Allow specifying column filters for logical replication This allows specifying an optional column list when adding a table to logical replication. Columns not included on this list are not sent to the subscriber. The filter is specified as a list of columns after the table name, enclosed in parentheses. For UPDATE/DELETE publications, the column filter needs to cover all REPLICA IDENTITY columns. For INSERT publications, the column list is arbitrary and may omit some REPLICA IDENTITY columns. Furthermore, if the table uses REPLICA IDENTITY FULL, column filter is not allowed. The column filter can contain only simple column references. Complex expressions, function calls etc. are not allowed. This restriction could be relaxed in the future. During the initial table synchronization, only columns specified in the column filter are copied to the subscriber. If the subscription has several publications, containing the same table with different column filters, columns specified in any of the filters will be copied. This means all columns are replicated if the table has no column filter at all (which is treated as column filter with all columns), of when of the publications is defined as FOR ALL TABLES (possibly IN SCHEMA for the schema of the table). For partitioned tables, publish_via_partition_root determines whether the row filter for the root or leaf relation will be used. If the parameter is 'false' (the default), the filter defined for the leaf relation is used. Otherwise, the column filter for the root partition will be used. Psql commands \dRp+ and \d now display any column filters. Author: Rahila Syed Discussion: https://postgr.es/m/CAH2L28vddB_NFdRVpuyRBJEBWjz4BSyTB=_ektNRH8NJ1jf95g@mail.gmail.com --- doc/src/sgml/catalogs.sgml | 13 + doc/src/sgml/protocol.sgml | 4 +- doc/src/sgml/ref/alter_publication.sgml | 23 +- doc/src/sgml/ref/create_publication.sgml | 17 +- src/backend/catalog/pg_publication.c | 260 ++++++ src/backend/commands/publicationcmds.c | 362 ++++++++- src/backend/commands/tablecmds.c | 36 +- src/backend/executor/execReplication.c | 19 +- src/backend/nodes/copyfuncs.c | 1 + src/backend/nodes/equalfuncs.c | 1 + src/backend/parser/gram.y | 60 +- src/backend/replication/logical/proto.c | 55 +- src/backend/replication/logical/tablesync.c | 271 ++++++- src/backend/replication/pgoutput/pgoutput.c | 123 ++- src/backend/utils/cache/relcache.c | 32 + src/bin/pg_dump/pg_dump.c | 47 +- src/bin/pg_dump/pg_dump.h | 1 + src/bin/pg_dump/t/002_pg_dump.pl | 60 ++ src/bin/psql/describe.c | 40 +- src/include/catalog/pg_publication.h | 14 + src/include/catalog/pg_publication_rel.h | 1 + src/include/commands/publicationcmds.h | 2 + src/include/nodes/parsenodes.h | 4 +- src/include/replication/logicalproto.h | 6 +- src/test/regress/expected/publication.out | 375 +++++++++ src/test/regress/sql/publication.sql | 290 +++++++ src/test/subscription/t/029_column_list.pl | 836 ++++++++++++++++++++ 27 files changed, 2872 insertions(+), 81 deletions(-) create mode 100644 src/test/subscription/t/029_column_list.pl diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 83987a99045..2b61f42b71d 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -6263,6 +6263,19 @@ SCRAM-SHA-256$<iteration count>:&l Reference to schema + + + + prattrs int2vector + (references pg_attribute.attnum) + + + This is an array of values that indicates which table columns are + part of the publication. For example a value of 1 3 + would mean that the first and the third table columns are published. + A null value indicates that all columns are published. + + diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index c51c4254a70..496593201b9 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -7005,7 +7005,9 @@ Relation - Next, the following message part appears for each column (except generated columns): + Next, the following message part appears for each column (except + generated columns and other columns that don't appear in the column + filter list, for tables that have one): diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml index 32b75f6c78e..aa6827c977b 100644 --- a/doc/src/sgml/ref/alter_publication.sgml +++ b/doc/src/sgml/ref/alter_publication.sgml @@ -25,12 +25,13 @@ ALTER PUBLICATION name ADD name SET publication_object [, ...] ALTER PUBLICATION name DROP publication_object [, ...] ALTER PUBLICATION name SET ( publication_parameter [= value] [, ... ] ) +ALTER PUBLICATION name ALTER TABLE table_name SET COLUMNS { ( name [, ...] ) | ALL } ALTER PUBLICATION name OWNER TO { new_owner | CURRENT_ROLE | CURRENT_USER | SESSION_USER } ALTER PUBLICATION name RENAME TO new_name where publication_object is one of: - TABLE [ ONLY ] table_name [ * ] [ WHERE ( expression ) ] [, ... ] + TABLE [ ONLY ] table_name [ * ] [ ( column_name [, ... ] ) ] [ WHERE ( expression ) ] [, ... ] ALL TABLES IN SCHEMA { schema_name | CURRENT_SCHEMA } [, ... ] @@ -64,6 +65,12 @@ ALTER PUBLICATION name RENAME TO + + The ALTER TABLE ... SET COLUMNS variant allows changing + the set of columns that are included in the publication. If a column list + is specified, it must include the replica identity columns. + + The remaining variants change the owner and the name of the publication. @@ -112,6 +119,10 @@ ALTER PUBLICATION name RENAME TO * can be specified after the table name to explicitly indicate that descendant tables are included. + + Optionally, a column list can be specified. See for details. + If the optional WHERE clause is specified, rows for which the expression evaluates to false or null will not be published. Note that parentheses @@ -172,9 +183,15 @@ ALTER PUBLICATION noinsert SET (publish = 'update, delete'); - Add some tables to the publication: + Add tables to the publication: + +ALTER PUBLICATION mypublication ADD TABLE users (user_id, firstname), departments; + + + + Change the set of columns published for a table: -ALTER PUBLICATION mypublication ADD TABLE users, departments; +ALTER PUBLICATION mypublication ALTER TABLE users SET COLUMNS (user_id, firstname, lastname); diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml index 4979b9b646d..fb2d013393b 100644 --- a/doc/src/sgml/ref/create_publication.sgml +++ b/doc/src/sgml/ref/create_publication.sgml @@ -28,7 +28,7 @@ CREATE PUBLICATION name where publication_object is one of: - TABLE [ ONLY ] table_name [ * ] [ WHERE ( expression ) ] [, ... ] + TABLE [ ONLY ] table_name [ * ] [ ( column_name [, ... ] ) ] [ WHERE ( expression ) ] [, ... ] ALL TABLES IN SCHEMA { schema_name | CURRENT_SCHEMA } [, ... ] @@ -86,6 +86,13 @@ CREATE PUBLICATION name TRUNCATE commands. + + When a column list is specified, only the named columns are replicated. + If no column list is specified, all columns of the table are replicated + through this publication, including any columns added later. If a column + list is specified, it must include the replica identity columns. + + Only persistent base tables and partitioned tables can be part of a publication. Temporary tables, unlogged tables, foreign tables, @@ -327,6 +334,14 @@ CREATE PUBLICATION production_publication FOR TABLE users, departments, ALL TABL sales: CREATE PUBLICATION sales_publication FOR ALL TABLES IN SCHEMA marketing, sales; + + + + Create a publication that publishes all changes for table users, + but replicates only columns user_id and + firstname: + +CREATE PUBLICATION users_filtered FOR TABLE users (user_id, firstname); diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 25998fbb39b..4dab96265f8 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -45,6 +45,9 @@ #include "utils/rel.h" #include "utils/syscache.h" +static void publication_translate_columns(Relation targetrel, List *columns, + int *natts, AttrNumber **attrs); + /* * Check if relation can be in given publication and throws appropriate * error if not. @@ -328,6 +331,9 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri, Oid relid = RelationGetRelid(targetrel); Oid pubreloid; Publication *pub = GetPublication(pubid); + AttrNumber *attarray; + int natts = 0; + int i; ObjectAddress myself, referenced; List *relids = NIL; @@ -355,6 +361,14 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri, check_publication_add_relation(targetrel); + /* + * Translate column names to attnums and check the column list is valid. + * We also deconstruct the bitmap into an array of attnums, for storing + * in the catalog. + */ + publication_translate_columns(pri->relation, pri->columns, + &natts, &attarray); + /* Form a tuple. */ memset(values, 0, sizeof(values)); memset(nulls, false, sizeof(nulls)); @@ -367,6 +381,17 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri, values[Anum_pg_publication_rel_prrelid - 1] = ObjectIdGetDatum(relid); + /* Add column filter, if available */ + if (pri->columns) + { + int2vector *prattrs; + + prattrs = buildint2vector(attarray, natts); + values[Anum_pg_publication_rel_prattrs - 1] = PointerGetDatum(prattrs); + } + else + nulls[Anum_pg_publication_rel_prattrs - 1] = true; + /* Add qualifications, if available */ if (pri->whereClause != NULL) values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(pri->whereClause)); @@ -382,6 +407,14 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri, /* Register dependencies as needed */ ObjectAddressSet(myself, PublicationRelRelationId, pubreloid); + /* Add dependency on the columns, if any are listed */ + for (i = 0; i < natts; i++) + { + ObjectAddressSubSet(referenced, RelationRelationId, relid, attarray[i]); + recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO); + } + pfree(attarray); + /* Add dependency on the publication */ ObjectAddressSet(referenced, PublicationRelationId, pubid); recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO); @@ -415,6 +448,155 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri, return myself; } +/* + * Update the column list for a relation in a publication. + */ +void +publication_set_table_columns(Relation pubrel, HeapTuple pubreltup, + Relation targetrel, List *columns) +{ + AttrNumber *attarray; + HeapTuple copytup; + int natts; + bool nulls[Natts_pg_publication_rel]; + bool replaces[Natts_pg_publication_rel]; + Datum values[Natts_pg_publication_rel]; + int i; + + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + memset(replaces, false, sizeof(replaces)); + + replaces[Anum_pg_publication_rel_prattrs - 1] = true; + + deleteDependencyRecordsForClass(PublicationRelationId, + ((Form_pg_publication_rel) GETSTRUCT(pubreltup))->oid, + RelationRelationId, + DEPENDENCY_AUTO); + + if (columns == NULL) + { + nulls[Anum_pg_publication_rel_prattrs - 1] = true; + } + else + { + ObjectAddress myself, + referenced; + int2vector *prattrs; + Form_pg_publication_rel pubrel; + + publication_translate_columns(targetrel, columns, &natts, &attarray); + + /* XXX "pub" is leaked here */ + + prattrs = buildint2vector(attarray, natts); + values[Anum_pg_publication_rel_prattrs - 1] = PointerGetDatum(prattrs); + + /* Add dependencies on the new list of columns */ + pubrel = (Form_pg_publication_rel) GETSTRUCT(pubreltup); + ObjectAddressSet(myself, PublicationRelRelationId, pubrel->oid); + + for (i = 0; i < natts; i++) + { + ObjectAddressSubSet(referenced, RelationRelationId, + RelationGetRelid(targetrel), attarray[i]); + recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO); + } + } + + copytup = heap_modify_tuple(pubreltup, RelationGetDescr(pubrel), + values, nulls, replaces); + + CatalogTupleUpdate(pubrel, &pubreltup->t_self, copytup); + + heap_freetuple(copytup); +} + +/* + * qsort comparator for attnums + * + * XXX We already have compare_int16, so maybe let's share that, somehow? + */ +static int +compare_int16(const void *a, const void *b) +{ + int av = *(const int16 *) a; + int bv = *(const int16 *) b; + + /* this can't overflow if int is wider than int16 */ + return (av - bv); +} + +/* + * Translate a list of column names to an array of attribute numbers + * and a Bitmapset with them; verify that each attribute is appropriate + * to have in a publication column list (no system or generated attributes, + * no duplicates). Additional checks with replica identity are done later; + * see check_publication_columns. + * + * Note that the attribute numbers are *not* offset by + * FirstLowInvalidHeapAttributeNumber; system columns are forbidden so this + * is okay. + * + * XXX Should this detect duplicate columns? + */ +static void +publication_translate_columns(Relation targetrel, List *columns, + int *natts, AttrNumber **attrs) +{ + AttrNumber *attarray; + Bitmapset *set = NULL; + ListCell *lc; + int n = 0; + TupleDesc tupdesc = RelationGetDescr(targetrel); + + /* + * Translate list of columns to attnums. We prohibit system attributes and + * make sure there are no duplicate columns. + */ + attarray = palloc(sizeof(AttrNumber) * list_length(columns)); + foreach(lc, columns) + { + char *colname = strVal(lfirst(lc)); + AttrNumber attnum = get_attnum(RelationGetRelid(targetrel), colname); + + if (attnum == InvalidAttrNumber) + ereport(ERROR, + errcode(ERRCODE_UNDEFINED_COLUMN), + errmsg("column \"%s\" of relation \"%s\" does not exist", + colname, RelationGetRelationName(targetrel))); + + if (!AttrNumberIsForUserDefinedAttr(attnum)) + ereport(ERROR, + errcode(ERRCODE_INVALID_COLUMN_REFERENCE), + errmsg("cannot reference system column \"%s\" in publication column list", + colname)); + + if (TupleDescAttr(tupdesc, attnum - 1)->attgenerated) + ereport(ERROR, + errcode(ERRCODE_INVALID_COLUMN_REFERENCE), + errmsg("cannot reference generated column \"%s\" in publication column list", + colname)); + + if (bms_is_member(attnum, set)) + ereport(ERROR, + errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("duplicate column \"%s\" in publication column list", + colname)); + + set = bms_add_member(set, attnum); + attarray[n++] = attnum; + } + + /* Be tidy, so that the catalog representation is always sorted */ + qsort(attarray, n, sizeof(AttrNumber), compare_int16); + + *natts = n; + *attrs = attarray; + + bms_free(set); +} + /* * Insert new publication / schema mapping. */ @@ -522,6 +704,84 @@ GetRelationPublications(Oid relid) return result; } +/* + * Gets a list of OIDs of all partial-column publications of the given + * relation, that is, those that specify a column list. + */ +List * +GetRelationColumnPartialPublications(Oid relid) +{ + CatCList *pubrellist; + List *pubs = NIL; + int i; + + pubrellist = SearchSysCacheList1(PUBLICATIONRELMAP, + ObjectIdGetDatum(relid)); + for (i = 0; i < pubrellist->n_members; i++) + { + HeapTuple tup = &pubrellist->members[i]->tuple; + bool isnull; + Form_pg_publication_rel pubrel; + + (void) SysCacheGetAttr(PUBLICATIONRELMAP, tup, + Anum_pg_publication_rel_prattrs, + &isnull); + + /* no column filter for this publications/relation */ + if (isnull) + continue; + + pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + pubs = lappend_oid(pubs, pubrel->prpubid); + } + + ReleaseSysCacheList(pubrellist); + + return pubs; +} + + +/* + * For a relation in a publication that is known to have a non-null column + * list, return the list of attribute numbers that are in it. + */ +List * +GetRelationColumnListInPublication(Oid relid, Oid pubid) +{ + HeapTuple tup; + Datum adatum; + bool isnull; + ArrayType *arr; + int nelems; + int16 *elems; + List *attnos = NIL; + int i; + + tup = SearchSysCache2(PUBLICATIONRELMAP, + ObjectIdGetDatum(relid), + ObjectIdGetDatum(pubid)); + + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for rel %u of publication %u", relid, pubid); + + adatum = SysCacheGetAttr(PUBLICATIONRELMAP, tup, + Anum_pg_publication_rel_prattrs, &isnull); + if (isnull) + elog(ERROR, "found unexpected null in pg_publication_rel.prattrs"); + + arr = DatumGetArrayTypeP(adatum); + nelems = ARR_DIMS(arr)[0]; + elems = (int16 *) ARR_DATA_PTR(arr); + + for (i = 0; i < nelems; i++) + attnos = lappend_oid(attnos, elems[i]); + + ReleaseSysCache(tup); + + return attnos; +} + /* * Gets list of relation oids for a publication. * diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 16b8661a1b7..fa1462ae546 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -367,6 +367,123 @@ contain_invalid_rfcolumn(Oid pubid, Relation relation, List *ancestors, return result; } +/* + * Check if all columns referenced in the column filter are part of the + * REPLICA IDENTITY index or not. + * + * Returns true if any invalid column is found. + */ +bool +contain_invalid_cfcolumn(Oid pubid, Relation relation, List *ancestors, + bool pubviaroot) +{ + HeapTuple cftuple; + Oid relid = RelationGetRelid(relation); + Oid publish_as_relid = RelationGetRelid(relation); + bool result = false; + Datum cfdatum; + bool cfisnull; + + /* + * For a partition, if pubviaroot is true, find the topmost ancestor that + * is published via this publication as we need to use its column filter + * for the changes. + * + * Note that even though the column filter used is for an ancestor, the + * REPLICA IDENTITY used will be for the actual child table. + */ + if (pubviaroot && relation->rd_rel->relispartition) + { + publish_as_relid = GetTopMostAncestorInPublication(pubid, ancestors); + + if (!OidIsValid(publish_as_relid)) + publish_as_relid = relid; + } + + cftuple = SearchSysCache2(PUBLICATIONRELMAP, + ObjectIdGetDatum(publish_as_relid), + ObjectIdGetDatum(pubid)); + + if (!HeapTupleIsValid(cftuple)) + return false; + + cfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, cftuple, + Anum_pg_publication_rel_prattrs, + &cfisnull); + + if (!cfisnull) + { + int x; + Bitmapset *idattrs; + Bitmapset *columns = NULL; + + /* With REPLICA IDENTITY FULL, no column filter is allowed. */ + if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL) + result = true; + + { + ArrayType *arr; + int nelems; + int16 *elems; + + arr = DatumGetArrayTypeP(cfdatum); + nelems = ARR_DIMS(arr)[0]; + elems = (int16 *) ARR_DATA_PTR(arr); + + /* XXX is there a danger of memory leak here? beware */ + for (int i = 0; i < nelems; i++) + columns = bms_add_member(columns, elems[i]); + } + + /* Remember columns that are part of the REPLICA IDENTITY */ + idattrs = RelationGetIndexAttrBitmap(relation, + INDEX_ATTR_BITMAP_IDENTITY_KEY); + + /* + * Attnums in the bitmap returned by RelationGetIndexAttrBitmap are + * offset (to handle system columns the usual way), while column filter + * does not use offset, so we can't do bms_is_subset(). Instead, we have + * to loop over the idattrs and check all of them are in the filter. + */ + x = -1; + while ((x = bms_next_member(idattrs, x)) >= 0) + { + AttrNumber attnum = (x + FirstLowInvalidHeapAttributeNumber); + + /* + * If pubviaroot is true, we are validating the row filter of the + * parent table, but the bitmap contains the replica identity + * information of the child table. So, get the column number of the + * child table as parent and child column order could be different. + */ + if (pubviaroot) + { + /* attnum is for child table */ + char *colname = get_attname(relid, attnum, false); + + /* + * Determine the matching attnum in parent (because the column + * filter is defined on the parent). + */ + attnum = get_attnum(publish_as_relid, colname); + } + + if (!bms_is_member(attnum, columns)) + { + result = true; + break; + } + } + + bms_free(idattrs); + bms_free(columns); + } + + ReleaseSysCache(cftuple); + + return result; +} + /* check_functions_in_node callback */ static bool contain_mutable_or_user_functions_checker(Oid func_id, void *context) @@ -608,6 +725,45 @@ TransformPubWhereClauses(List *tables, const char *queryString, } } + +/* + * Transform the publication column lists expression for all the relations + * in the list. + * + * XXX The name is a bit misleading, because we don't really transform + * anything here - we merely check the column list is compatible with the + * definition of the publication (with publish_via_partition_root=false) + * we only allow filters on the leaf relations. So maybe rename it? + */ +static void +TransformPubColumnList(List *tables, const char *queryString, + bool pubviaroot) +{ + ListCell *lc; + + foreach(lc, tables) + { + PublicationRelInfo *pri = (PublicationRelInfo *) lfirst(lc); + + if (pri->columns == NIL) + continue; + + /* + * If the publication doesn't publish changes via the root partitioned + * table, the partition's column filter will be used. So disallow using + * the column list on partitioned table in this case. + */ + if (!pubviaroot && + pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot use publication column list for relation \"%s\"", + RelationGetRelationName(pri->relation)), + errdetail("column list cannot be used for a partitioned table when %s is false.", + "publish_via_partition_root"))); + } +} + /* * Create new publication. */ @@ -724,6 +880,9 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) TransformPubWhereClauses(rels, pstate->p_sourcetext, publish_via_partition_root); + TransformPubColumnList(rels, pstate->p_sourcetext, + publish_via_partition_root); + PublicationAddTables(puboid, rels, true, NULL); CloseTableList(rels); } @@ -754,6 +913,52 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) return myself; } +/* + * Change the column list of a relation in a publication + */ +static void +PublicationSetColumns(AlterPublicationStmt *stmt, + Form_pg_publication pubform, PublicationTable *table) +{ + Relation rel, + urel; + HeapTuple tup; + ObjectAddress obj, + secondary; + + rel = table_open(PublicationRelRelationId, RowExclusiveLock); + urel = table_openrv(table->relation, ShareUpdateExclusiveLock); + + tup = SearchSysCache2(PUBLICATIONRELMAP, + ObjectIdGetDatum(RelationGetRelid(urel)), + ObjectIdGetDatum(pubform->oid)); + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + errmsg("relation \"%s\" is not already in publication \"%s\"", + table->relation->relname, + NameStr(pubform->pubname))); + + publication_set_table_columns(rel, tup, urel, table->columns); + + ObjectAddressSet(obj, PublicationRelationId, + ((Form_pg_publication_rel) GETSTRUCT(tup))->oid); + ObjectAddressSet(secondary, RelationRelationId, RelationGetRelid(urel)); + EventTriggerCollectSimpleCommand(obj, secondary, (Node *) stmt); + + ReleaseSysCache(tup); + + /* + * Invalidate relcache for this relation, to force rebuilding the + * publication description. + */ + CacheInvalidateRelcache(urel); + + table_close(rel, RowExclusiveLock); + table_close(urel, NoLock); + + InvokeObjectPostAlterHook(PublicationRelationId, pubform->oid, 0); +} + /* * Change options of a publication. */ @@ -783,8 +988,8 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt, /* * If the publication doesn't publish changes via the root partitioned - * table, the partition's row filter will be used. So disallow using WHERE - * clause on partitioned table in this case. + * table, the partition's row and column filter will be used. So disallow + * using WHERE clause and column filters on partitioned table in this case. */ if (!pubform->puballtables && publish_via_partition_root_given && !publish_via_partition_root) @@ -792,7 +997,8 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt, /* * Lock the publication so nobody else can do anything with it. This * prevents concurrent alter to add partitioned table(s) with WHERE - * clause(s) which we don't allow when not publishing via root. + * clause(s) and/or column filters which we don't allow when not + * publishing via root. */ LockDatabaseObject(PublicationRelationId, pubform->oid, 0, AccessShareLock); @@ -804,13 +1010,21 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt, { HeapTuple rftuple; Oid relid = lfirst_oid(lc); + bool has_column_filter; + bool has_row_filter; rftuple = SearchSysCache2(PUBLICATIONRELMAP, ObjectIdGetDatum(relid), ObjectIdGetDatum(pubform->oid)); + has_row_filter + = !heap_attisnull(rftuple, Anum_pg_publication_rel_prqual, NULL); + + has_column_filter + = !heap_attisnull(rftuple, Anum_pg_publication_rel_prattrs, NULL); + if (HeapTupleIsValid(rftuple) && - !heap_attisnull(rftuple, Anum_pg_publication_rel_prqual, NULL)) + (has_row_filter || has_column_filter)) { HeapTuple tuple; @@ -819,7 +1033,8 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt, { Form_pg_class relform = (Form_pg_class) GETSTRUCT(tuple); - if (relform->relkind == RELKIND_PARTITIONED_TABLE) + if ((relform->relkind == RELKIND_PARTITIONED_TABLE) && + has_row_filter) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("cannot set %s for publication \"%s\"", @@ -830,6 +1045,18 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt, NameStr(relform->relname), "publish_via_partition_root"))); + if ((relform->relkind == RELKIND_PARTITIONED_TABLE) && + has_column_filter) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot set %s for publication \"%s\"", + "publish_via_partition_root = false", + stmt->pubname), + errdetail("The publication contains a column filter for a partitioned table \"%s\" " + "which is not allowed when %s is false.", + NameStr(relform->relname), + "publish_via_partition_root"))); + ReleaseSysCache(tuple); } @@ -838,6 +1065,16 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt, } } + /* + * FIXME check pubactions vs. replica identity, to ensure the replica + * identity is included in the column filter. Only do this for update + * and delete publications. See check_publication_columns. + * + * XXX This is needed because publish_via_partition_root may change, + * in which case the row filters may be invalid (e.g. with pvpr=false + * there must be no filter on partitioned tables). + */ + /* Everything ok, form a new tuple. */ memset(values, 0, sizeof(values)); memset(nulls, false, sizeof(nulls)); @@ -975,10 +1212,20 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup, TransformPubWhereClauses(rels, queryString, pubform->pubviaroot); + TransformPubColumnList(rels, queryString, pubform->pubviaroot); + PublicationAddTables(pubid, rels, false, stmt); } else if (stmt->action == AP_DropObjects) PublicationDropTables(pubid, rels, false); + else if (stmt->action == AP_SetColumns) + { + Assert(schemaidlist == NIL); + Assert(list_length(tables) == 1); + + PublicationSetColumns(stmt, pubform, + linitial_node(PublicationTable, tables)); + } else /* AP_SetObjects */ { List *oldrelids = GetPublicationRelations(pubid, @@ -991,6 +1238,8 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup, TransformPubWhereClauses(rels, queryString, pubform->pubviaroot); + TransformPubColumnList(rels, queryString, pubform->pubviaroot); + /* * To recreate the relation list for the publication, look for * existing relations that do not need to be dropped. @@ -1002,32 +1251,85 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup, PublicationRelInfo *oldrel; bool found = false; HeapTuple rftuple; - bool rfisnull = true; Node *oldrelwhereclause = NULL; + Bitmapset *oldcolumns = NULL; /* look up the cache for the old relmap */ rftuple = SearchSysCache2(PUBLICATIONRELMAP, ObjectIdGetDatum(oldrelid), ObjectIdGetDatum(pubid)); + /* + * See if the existing relation currently has a WHERE clause or a + * column filter. We need to compare those too. + */ if (HeapTupleIsValid(rftuple)) { + bool isnull = true; Datum whereClauseDatum; + Datum columnListDatum; + /* Load the WHERE clause for this table. */ whereClauseDatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple, Anum_pg_publication_rel_prqual, - &rfisnull); - if (!rfisnull) + &isnull); + if (!isnull) oldrelwhereclause = stringToNode(TextDatumGetCString(whereClauseDatum)); + /* Transform the int2vector column list to a bitmap. */ + columnListDatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple, + Anum_pg_publication_rel_prattrs, + &isnull); + + /* + * XXX Maybe make this a separate function. We do this on + * multiple places. + */ + if (!isnull) + { + ArrayType *arr; + int nelems; + int16 *elems; + + arr = DatumGetArrayTypeP(columnListDatum); + nelems = ARR_DIMS(arr)[0]; + elems = (int16 *) ARR_DATA_PTR(arr); + + /* XXX is there a danger of memory leak here? beware */ + for (int i = 0; i < nelems; i++) + oldcolumns = bms_add_member(oldcolumns, elems[i]); + } + ReleaseSysCache(rftuple); } foreach(newlc, rels) { PublicationRelInfo *newpubrel; + Oid newrelid; + Bitmapset *newcolumns = NULL; newpubrel = (PublicationRelInfo *) lfirst(newlc); + newrelid = RelationGetRelid(newpubrel->relation); + + /* + * If the new publication has column filter, transform it to + * a bitmap too. + */ + if (newpubrel->columns) + { + ListCell *lc; + + foreach(lc, newpubrel->columns) + { + char *colname = strVal(lfirst(lc)); + AttrNumber attnum = get_attnum(newrelid, colname); + + /* no checks needed here, that happens elsewhere */ + + newcolumns = bms_add_member(newcolumns, attnum); + } + } /* * Check if any of the new set of relations matches with the @@ -1037,7 +1339,8 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup, */ if (RelationGetRelid(newpubrel->relation) == oldrelid) { - if (equal(oldrelwhereclause, newpubrel->whereClause)) + if (equal(oldrelwhereclause, newpubrel->whereClause) && + bms_equal(oldcolumns, newcolumns)) { found = true; break; @@ -1056,6 +1359,7 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup, { oldrel = palloc(sizeof(PublicationRelInfo)); oldrel->whereClause = NULL; + oldrel->columns = NIL; oldrel->relation = table_open(oldrelid, ShareUpdateExclusiveLock); delrels = lappend(delrels, oldrel); @@ -1117,7 +1421,7 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt, } else if (stmt->action == AP_DropObjects) PublicationDropSchemas(pubform->oid, schemaidlist, false); - else /* AP_SetObjects */ + else if (stmt->action == AP_SetObjects) { List *oldschemaids = GetPublicationSchemas(pubform->oid); List *delschemas = NIL; @@ -1140,6 +1444,10 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt, */ PublicationAddSchemas(pubform->oid, schemaidlist, true, stmt); } + else + { + /* Nothing to do for AP_SetColumns */ + } } /* @@ -1402,6 +1710,7 @@ OpenTableList(List *tables) List *rels = NIL; ListCell *lc; List *relids_with_rf = NIL; + List *relids_with_cf = NIL; /* * Open, share-lock, and check all the explicitly-specified relations @@ -1436,6 +1745,13 @@ OpenTableList(List *tables) errmsg("conflicting or redundant WHERE clauses for table \"%s\"", RelationGetRelationName(rel)))); + /* Disallow duplicate tables if there are any with column filters. */ + if (t->columns || list_member_oid(relids_with_cf, myrelid)) + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("conflicting or redundant column filters for table \"%s\"", + RelationGetRelationName(rel)))); + table_close(rel, ShareUpdateExclusiveLock); continue; } @@ -1443,12 +1759,16 @@ OpenTableList(List *tables) pub_rel = palloc(sizeof(PublicationRelInfo)); pub_rel->relation = rel; pub_rel->whereClause = t->whereClause; + pub_rel->columns = t->columns; rels = lappend(rels, pub_rel); relids = lappend_oid(relids, myrelid); if (t->whereClause) relids_with_rf = lappend_oid(relids_with_rf, myrelid); + if (t->columns) + relids_with_cf = lappend_oid(relids_with_cf, myrelid); + /* * Add children of this rel, if requested, so that they too are added * to the publication. A partitioned table can't have any inheritance @@ -1488,6 +1808,18 @@ OpenTableList(List *tables) errmsg("conflicting or redundant WHERE clauses for table \"%s\"", RelationGetRelationName(rel)))); + /* + * We don't allow to specify column filter for both parent + * and child table at the same time as it is not very + * clear which one should be given preference. + */ + if (childrelid != myrelid && + (t->columns || list_member_oid(relids_with_cf, childrelid))) + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("conflicting or redundant column fiters for table \"%s\"", + RelationGetRelationName(rel)))); + continue; } @@ -1497,11 +1829,16 @@ OpenTableList(List *tables) pub_rel->relation = rel; /* child inherits WHERE clause from parent */ pub_rel->whereClause = t->whereClause; + /* child inherits column list from parent */ + pub_rel->columns = t->columns; rels = lappend(rels, pub_rel); relids = lappend_oid(relids, childrelid); if (t->whereClause) relids_with_rf = lappend_oid(relids_with_rf, childrelid); + + if (t->whereClause) + relids_with_cf = lappend_oid(relids_with_cf, childrelid); } } } @@ -1610,6 +1947,11 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok) Relation rel = pubrel->relation; Oid relid = RelationGetRelid(rel); + if (pubrel->columns) + ereport(ERROR, + errcode(ERRCODE_SYNTAX_ERROR), + errmsg("column list must not be specified in ALTER PUBLICATION ... DROP")); + prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid, ObjectIdGetDatum(relid), ObjectIdGetDatum(pubid)); diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index dc5872f988c..a9fd0f0c895 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -8365,6 +8365,7 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName, bool missing_ok, LOCKMODE lockmode, ObjectAddresses *addrs) { + Oid relid = RelationGetRelid(rel); HeapTuple tuple; Form_pg_attribute targetatt; AttrNumber attnum; @@ -8384,7 +8385,7 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName, /* * get the number of the attribute */ - tuple = SearchSysCacheAttName(RelationGetRelid(rel), colName); + tuple = SearchSysCacheAttName(relid, colName); if (!HeapTupleIsValid(tuple)) { if (!missing_ok) @@ -8438,13 +8439,42 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName, ReleaseSysCache(tuple); + /* + * Also, if the column is used in the column list of a publication, + * disallow the drop if the DROP is RESTRICT. We don't do anything if the + * DROP is CASCADE, which means that the dependency mechanism will remove + * the relation from the publication. + */ + if (behavior == DROP_RESTRICT) + { + List *pubs; + ListCell *lc; + + pubs = GetRelationColumnPartialPublications(relid); + foreach(lc, pubs) + { + Oid pubid = lfirst_oid(lc); + List *published_cols; + + published_cols = + GetRelationColumnListInPublication(relid, pubid); + + if (list_member_oid(published_cols, attnum)) + ereport(ERROR, + errcode(ERRCODE_INVALID_TABLE_DEFINITION), + errmsg("cannot drop column \"%s\" because it is part of publication \"%s\"", + colName, get_publication_name(pubid, false)), + errhint("Specify CASCADE or use ALTER PUBLICATION to remove the column from the publication.")); + } + } + /* * Propagate to children as appropriate. Unlike most other ALTER * routines, we have to do this one level of recursion at a time; we can't * use find_all_inheritors to do it in one pass. */ children = - find_inheritance_children(RelationGetRelid(rel), lockmode); + find_inheritance_children(relid, lockmode); if (children) { @@ -8532,7 +8562,7 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName, /* Add object to delete */ object.classId = RelationRelationId; - object.objectId = RelationGetRelid(rel); + object.objectId = relid; object.objectSubId = attnum; add_exact_object_address(&object, addrs); diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index de106d767d1..d87be2d4775 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -573,9 +573,6 @@ CheckCmdReplicaIdentity(Relation rel, CmdType cmd) if (cmd != CMD_UPDATE && cmd != CMD_DELETE) return; - if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL) - return; - /* * It is only safe to execute UPDATE/DELETE when all columns, referenced * in the row filters from publications which the relation is in, are @@ -595,17 +592,33 @@ CheckCmdReplicaIdentity(Relation rel, CmdType cmd) errmsg("cannot update table \"%s\"", RelationGetRelationName(rel)), errdetail("Column used in the publication WHERE expression is not part of the replica identity."))); + else if (cmd == CMD_UPDATE && !pubdesc.cf_valid_for_update) + ereport(ERROR, + (errcode(ERRCODE_INVALID_COLUMN_REFERENCE), + errmsg("cannot update table \"%s\"", + RelationGetRelationName(rel)), + errdetail("Column filter used by the publication does not cover the replica identity."))); else if (cmd == CMD_DELETE && !pubdesc.rf_valid_for_delete) ereport(ERROR, (errcode(ERRCODE_INVALID_COLUMN_REFERENCE), errmsg("cannot delete from table \"%s\"", RelationGetRelationName(rel)), errdetail("Column used in the publication WHERE expression is not part of the replica identity."))); + else if (cmd == CMD_DELETE && !pubdesc.cf_valid_for_delete) + ereport(ERROR, + (errcode(ERRCODE_INVALID_COLUMN_REFERENCE), + errmsg("cannot delete from table \"%s\"", + RelationGetRelationName(rel)), + errdetail("Column filter used by the publication does not cover the replica identity."))); /* If relation has replica identity we are always good. */ if (OidIsValid(RelationGetReplicaIndex(rel))) return; + /* REPLICA IDENTITY FULL is also good for UPDATE/DELETE. */ + if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL) + return; + /* * This is either UPDATE OR DELETE and there is no replica identity. * diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index d4f8455a2bd..a504437873f 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -4850,6 +4850,7 @@ _copyPublicationTable(const PublicationTable *from) COPY_NODE_FIELD(relation); COPY_NODE_FIELD(whereClause); + COPY_NODE_FIELD(columns); return newnode; } diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index f1002afe7a0..4fc16ce04e3 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2322,6 +2322,7 @@ _equalPublicationTable(const PublicationTable *a, const PublicationTable *b) { COMPARE_NODE_FIELD(relation); COMPARE_NODE_FIELD(whereClause); + COMPARE_NODE_FIELD(columns); return true; } diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index a03b33b53bd..25c9b29afdd 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -9751,13 +9751,14 @@ CreatePublicationStmt: * relation_expr here. */ PublicationObjSpec: - TABLE relation_expr OptWhereClause + TABLE relation_expr opt_column_list OptWhereClause { $$ = makeNode(PublicationObjSpec); $$->pubobjtype = PUBLICATIONOBJ_TABLE; $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation = $2; - $$->pubtable->whereClause = $3; + $$->pubtable->columns = $3; + $$->pubtable->whereClause = $4; } | ALL TABLES IN_P SCHEMA ColId { @@ -9772,11 +9773,12 @@ PublicationObjSpec: $$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA; $$->location = @5; } - | ColId OptWhereClause + | ColId opt_column_list OptWhereClause { $$ = makeNode(PublicationObjSpec); $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION; - if ($2) + /* FIXME this is a bit cumbersome */ + if ($2 || $3) { /* * The OptWhereClause must be stored here but it is @@ -9786,7 +9788,8 @@ PublicationObjSpec: */ $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation = makeRangeVar(NULL, $1, @1); - $$->pubtable->whereClause = $2; + $$->pubtable->columns = $2; + $$->pubtable->whereClause = $3; } else { @@ -9794,23 +9797,25 @@ PublicationObjSpec: } $$->location = @1; } - | ColId indirection OptWhereClause + | ColId indirection opt_column_list OptWhereClause { $$ = makeNode(PublicationObjSpec); $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION; $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation = makeRangeVarFromQualifiedName($1, $2, @1, yyscanner); - $$->pubtable->whereClause = $3; + $$->pubtable->columns = $3; + $$->pubtable->whereClause = $4; $$->location = @1; } /* grammar like tablename * , ONLY tablename, ONLY ( tablename ) */ - | extended_relation_expr OptWhereClause + | extended_relation_expr opt_column_list OptWhereClause { $$ = makeNode(PublicationObjSpec); $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION; $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation = $1; - $$->pubtable->whereClause = $2; + $$->pubtable->columns = $2; + $$->pubtable->whereClause = $3; } | CURRENT_SCHEMA { @@ -9836,6 +9841,10 @@ pub_obj_list: PublicationObjSpec * * ALTER PUBLICATION name SET pub_obj [, ...] * + * ALTER PUBLICATION name SET COLUMNS table_name (column[, ...]) + * + * ALTER PUBLICATION name SET COLUMNS table_name ALL + * * pub_obj is one of: * * TABLE table_name [, ...] @@ -9869,6 +9878,32 @@ AlterPublicationStmt: n->action = AP_SetObjects; $$ = (Node *)n; } + | ALTER PUBLICATION name ALTER TABLE relation_expr SET COLUMNS '(' columnList ')' + { + AlterPublicationStmt *n = makeNode(AlterPublicationStmt); + PublicationObjSpec *obj = makeNode(PublicationObjSpec); + obj->pubobjtype = PUBLICATIONOBJ_TABLE; + obj->pubtable = makeNode(PublicationTable); + obj->pubtable->relation = $6; + obj->pubtable->columns = $10; + n->pubname = $3; + n->pubobjects = list_make1(obj); + n->action = AP_SetColumns; + $$ = (Node *) n; + } + | ALTER PUBLICATION name ALTER TABLE relation_expr SET COLUMNS ALL + { + AlterPublicationStmt *n = makeNode(AlterPublicationStmt); + PublicationObjSpec *obj = makeNode(PublicationObjSpec); + obj->pubobjtype = PUBLICATIONOBJ_TABLE; + obj->pubtable = makeNode(PublicationTable); + obj->pubtable->relation = $6; + obj->pubtable->columns = NIL; + n->pubname = $3; + n->pubobjects = list_make1(obj); + n->action = AP_SetColumns; + $$ = (Node *) n; + } | ALTER PUBLICATION name DROP pub_obj_list { AlterPublicationStmt *n = makeNode(AlterPublicationStmt); @@ -17488,6 +17523,13 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner) errmsg("WHERE clause not allowed for schema"), parser_errposition(pubobj->location)); + /* Column filter is not allowed on a schema object */ + if (pubobj->pubtable && pubobj->pubtable->columns) + ereport(ERROR, + errcode(ERRCODE_SYNTAX_ERROR), + errmsg("column specification not allowed for schema"), + parser_errposition(pubobj->location)); + /* * We can distinguish between the different type of schema * objects based on whether name and pubtable is set. diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index c9b0eeefd7e..1e8785ff9a5 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -29,16 +29,21 @@ #define TRUNCATE_CASCADE (1<<0) #define TRUNCATE_RESTART_SEQS (1<<1) -static void logicalrep_write_attrs(StringInfo out, Relation rel); +static void logicalrep_write_attrs(StringInfo out, Relation rel, + Bitmapset *columns); static void logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot, - bool binary); + bool binary, Bitmapset *columns); static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel); static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple); static void logicalrep_write_namespace(StringInfo out, Oid nspid); static const char *logicalrep_read_namespace(StringInfo in); +#define ColumnInFilter(columns, attnum) \ + (((columns) == NULL) || (bms_is_member((attnum), (columns)))) + + /* * Write BEGIN to the output stream. */ @@ -398,7 +403,7 @@ logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn) */ void logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel, - TupleTableSlot *newslot, bool binary) + TupleTableSlot *newslot, bool binary, Bitmapset *columns) { pq_sendbyte(out, LOGICAL_REP_MSG_INSERT); @@ -410,7 +415,7 @@ logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel, pq_sendint32(out, RelationGetRelid(rel)); pq_sendbyte(out, 'N'); /* new tuple follows */ - logicalrep_write_tuple(out, rel, newslot, binary); + logicalrep_write_tuple(out, rel, newslot, binary, columns); } /* @@ -443,7 +448,7 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup) void logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot, - bool binary) + bool binary, Bitmapset *columns) { pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE); @@ -464,11 +469,11 @@ logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, pq_sendbyte(out, 'O'); /* old tuple follows */ else pq_sendbyte(out, 'K'); /* old key follows */ - logicalrep_write_tuple(out, rel, oldslot, binary); + logicalrep_write_tuple(out, rel, oldslot, binary, columns); } pq_sendbyte(out, 'N'); /* new tuple follows */ - logicalrep_write_tuple(out, rel, newslot, binary); + logicalrep_write_tuple(out, rel, newslot, binary, columns); } /* @@ -537,7 +542,7 @@ logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel, else pq_sendbyte(out, 'K'); /* old key follows */ - logicalrep_write_tuple(out, rel, oldslot, binary); + logicalrep_write_tuple(out, rel, oldslot, binary, NULL); } /* @@ -652,7 +657,8 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, * Write relation description to the output stream. */ void -logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel) +logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel, + Bitmapset *columns) { char *relname; @@ -674,7 +680,7 @@ logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel) pq_sendbyte(out, rel->rd_rel->relreplident); /* send the attribute info */ - logicalrep_write_attrs(out, rel); + logicalrep_write_attrs(out, rel, columns); } /* @@ -751,7 +757,7 @@ logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp) */ static void logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot, - bool binary) + bool binary, Bitmapset *columns) { TupleDesc desc; Datum *values; @@ -763,7 +769,13 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot, for (i = 0; i < desc->natts; i++) { - if (TupleDescAttr(desc, i)->attisdropped || TupleDescAttr(desc, i)->attgenerated) + Form_pg_attribute att = TupleDescAttr(desc, i); + + if (att->attisdropped || att->attgenerated) + continue; + + /* skip columns not included in the column filter */ + if (!ColumnInFilter(columns, att->attnum)) continue; nliveatts++; } @@ -783,6 +795,10 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot, if (att->attisdropped || att->attgenerated) continue; + /* skip columns not included in the column filter */ + if (!ColumnInFilter(columns, att->attnum)) + continue; + if (isnull[i]) { pq_sendbyte(out, LOGICALREP_COLUMN_NULL); @@ -904,7 +920,7 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple) * Write relation attribute metadata to the stream. */ static void -logicalrep_write_attrs(StringInfo out, Relation rel) +logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns) { TupleDesc desc; int i; @@ -917,8 +933,15 @@ logicalrep_write_attrs(StringInfo out, Relation rel) /* send number of live attributes */ for (i = 0; i < desc->natts; i++) { - if (TupleDescAttr(desc, i)->attisdropped || TupleDescAttr(desc, i)->attgenerated) + Form_pg_attribute att = TupleDescAttr(desc, i); + + if (att->attisdropped || att->attgenerated) + continue; + + /* skip columns not included in the column filter */ + if (!ColumnInFilter(columns, att->attnum)) continue; + nliveatts++; } pq_sendint16(out, nliveatts); @@ -937,6 +960,10 @@ logicalrep_write_attrs(StringInfo out, Relation rel) if (att->attisdropped || att->attgenerated) continue; + /* skip columns not included in the column filter */ + if (!ColumnInFilter(columns, att->attnum)) + continue; + /* REPLICA IDENTITY FULL means all columns are sent as part of key. */ if (replidentfull || bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber, diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 1659964571c..42708dcf82e 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -112,6 +112,7 @@ #include "storage/ipc.h" #include "storage/lmgr.h" #include "utils/acl.h" +#include "utils/array.h" #include "utils/builtins.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -700,20 +701,22 @@ fetch_remote_table_info(char *nspname, char *relname, WalRcvExecResult *res; StringInfoData cmd; TupleTableSlot *slot; - Oid tableRow[] = {OIDOID, CHAROID, CHAROID}; - Oid attrRow[] = {TEXTOID, OIDOID, BOOLOID}; + Oid tableRow[] = {OIDOID, CHAROID, CHAROID, BOOLOID}; + Oid attrRow[] = {INT2OID, TEXTOID, OIDOID, BOOLOID}; Oid qualRow[] = {TEXTOID}; bool isnull; int natt; ListCell *lc; bool first; + bool am_partition; + Bitmapset *included_cols = NULL; lrel->nspname = nspname; lrel->relname = relname; /* First fetch Oid and replica identity. */ initStringInfo(&cmd); - appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind" + appendStringInfo(&cmd, "SELECT c.oid, c.relreplident, c.relkind, c.relispartition" " FROM pg_catalog.pg_class c" " INNER JOIN pg_catalog.pg_namespace n" " ON (c.relnamespace = n.oid)" @@ -743,14 +746,225 @@ fetch_remote_table_info(char *nspname, char *relname, Assert(!isnull); lrel->relkind = DatumGetChar(slot_getattr(slot, 3, &isnull)); Assert(!isnull); + am_partition = DatumGetBool(slot_getattr(slot, 4, &isnull)); + Assert(!isnull); ExecDropSingleTupleTableSlot(slot); walrcv_clear_result(res); - /* Now fetch columns. */ + + /* + * Get relation's column filter expressions. + * + * For initial synchronization, column filter can be ignored in following + * cases: + * + * 1) one of the subscribed publications for the table hasn't specified + * any column filter + * + * 2) one of the subscribed publications has puballtables set to true + * + * 3) one of the subscribed publications is declared as ALL TABLES IN + * SCHEMA that includes this relation + */ + if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000) + { + WalRcvExecResult *pubres; + TupleTableSlot *slot; + Oid attrsRow[] = {INT2OID}; + Oid tmpRow[] = {INT4OID}; + StringInfoData publications; + bool first = true; + bool all_columns = false; + + initStringInfo(&publications); + foreach(lc, MySubscription->publications) + { + if (!first) + appendStringInfo(&publications, ", "); + appendStringInfoString(&publications, quote_literal_cstr(strVal(lfirst(lc)))); + first = false; + } + + /* + * First, check if any of the publications FOR ALL TABLES? If yes, we + * should not use any column filter. It's enough to find a single such + * publication. + * + * XXX Maybe we could combine all three steps into a single query, but + * this seems cleaner / easier to understand. + * + * XXX Does this need any handling of partitions / publish_via_part_root? + */ + resetStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT 1\n" + " FROM pg_publication p\n" + " WHERE p.pubname IN ( %s ) AND p.puballtables LIMIT 1\n", + publications.data); + + pubres = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, + lengthof(tmpRow), tmpRow); + + if (pubres->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not fetch publication info for table \"%s.%s\" from publisher: %s", + nspname, relname, pubres->err))); + + slot = MakeSingleTupleTableSlot(pubres->tupledesc, &TTSOpsMinimalTuple); + + if (tuplestore_gettupleslot(pubres->tuplestore, true, false, slot)) + all_columns = true; + + ExecDropSingleTupleTableSlot(slot); + walrcv_clear_result(pubres); + + /* + * If there's no FOR ALL TABLES publication, look for a FOR ALL TABLES + * IN SCHEMA publication, with schema of the remote relation. The logic + * is the same - such publications have no column filters. + * + * XXX Does this need any handling of partitions / publish_via_part_root? + */ + if (!all_columns) + { + resetStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT 1\n" + " FROM pg_publication p\n" + " JOIN pg_publication_namespace pn ON (pn.pnpubid = p.oid)\n" + " JOIN pg_class c ON (pn.pnnspid = c.relnamespace)\n" + " WHERE c.oid = %u AND p.pubname IN ( %s ) LIMIT 1", + lrel->remoteid, + publications.data); + + pubres = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, + lengthof(tmpRow), tmpRow); + + if (pubres->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not fetch publication info for table \"%s.%s\" from publisher: %s", + nspname, relname, pubres->err))); + + slot = MakeSingleTupleTableSlot(pubres->tupledesc, &TTSOpsMinimalTuple); + + if (tuplestore_gettupleslot(pubres->tuplestore, true, false, slot)) + all_columns = true; + + ExecDropSingleTupleTableSlot(slot); + walrcv_clear_result(pubres); + } + + /* + * If we haven't found any FOR ALL TABLES [IN SCHEMA] publications for + * the table, we have to look for the column filters set for relations. + * First, we check if there's a publication with no column filter for + * the relation - which means all columns need to be replicated. + */ + if (!all_columns) + { + resetStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT 1\n" + " FROM pg_catalog.pg_publication p JOIN\n" + " pg_catalog.pg_publication_rel pr ON (p.oid = pr.prpubid)\n" + " WHERE p.pubname IN (%s) AND pr.prattrs IS NULL AND ", + publications.data); + + /* + * For non-partitions, we simply join directly to the catalog. For + * partitions, we need to check all the ancestors, because maybe the + * root was not added to a publication but one of the intermediate + * partitions was. + */ + if (!am_partition) + appendStringInfo(&cmd, "prrelid = %u", lrel->remoteid); + else + appendStringInfo(&cmd, + "prrelid IN (SELECT relid\n" + " FROM pg_catalog.pg_partition_tree(pg_catalog.pg_partition_root(%u)))", + lrel->remoteid); + + pubres = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, + lengthof(tmpRow), tmpRow); + + if (pubres->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not fetch attribute info for table \"%s.%s\" from publisher: %s", + nspname, relname, pubres->err))); + + slot = MakeSingleTupleTableSlot(pubres->tupledesc, &TTSOpsMinimalTuple); + + if (tuplestore_gettupleslot(pubres->tuplestore, true, false, slot)) + all_columns = true; + + ExecDropSingleTupleTableSlot(slot); + walrcv_clear_result(pubres); + } + + /* + * All that + */ + if (!all_columns) + { + resetStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT unnest(pr.prattrs)\n" + " FROM pg_catalog.pg_publication p JOIN\n" + " pg_catalog.pg_publication_rel pr ON (p.oid = pr.prpubid)\n" + " WHERE p.pubname IN (%s) AND pr.prattrs IS NOT NULL AND ", + publications.data); + + /* + * For non-partitions, we simply join directly to the catalog. For + * partitions, we need to check all the ancestors, because maybe the + * root was not added to a publication but one of the intermediate + * partitions was. + */ + if (!am_partition) + appendStringInfo(&cmd, "prrelid = %u", lrel->remoteid); + else + appendStringInfo(&cmd, + "prrelid IN (SELECT relid\n" + " FROM pg_catalog.pg_partition_tree(pg_catalog.pg_partition_root(%u)))", + lrel->remoteid); + + pubres = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, + lengthof(attrsRow), attrsRow); + + if (pubres->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not fetch attribute info for table \"%s.%s\" from publisher: %s", + nspname, relname, pubres->err))); + + slot = MakeSingleTupleTableSlot(pubres->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(pubres->tuplestore, true, false, slot)) + { + AttrNumber attnum; + + attnum = DatumGetInt16(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + included_cols = bms_add_member(included_cols, attnum); + } + ExecDropSingleTupleTableSlot(slot); + walrcv_clear_result(pubres); + } + + pfree(publications.data); + } + + /* + * Now fetch column names and types. + */ resetStringInfo(&cmd); appendStringInfo(&cmd, - "SELECT a.attname," + "SELECT a.attnum," + " a.attname," " a.atttypid," " a.attnum = ANY(i.indkey)" " FROM pg_catalog.pg_attribute a" @@ -778,16 +992,34 @@ fetch_remote_table_info(char *nspname, char *relname, lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid)); lrel->attkeys = NULL; + /* + * Store the columns as a list of names. Ignore those that are not + * present in the column list, if there is one. + */ natt = 0; slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) { - lrel->attnames[natt] = - TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + char *rel_colname; + AttrNumber attnum; + + attnum = DatumGetInt16(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + if (included_cols != NULL && !bms_is_member(attnum, included_cols)) + { + ExecClearTuple(slot); + continue; + } + + rel_colname = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); Assert(!isnull); - lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 2, &isnull)); + + lrel->attnames[natt] = rel_colname; + lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 3, &isnull)); Assert(!isnull); - if (DatumGetBool(slot_getattr(slot, 3, &isnull))) + + if (DatumGetBool(slot_getattr(slot, 4, &isnull))) lrel->attkeys = bms_add_member(lrel->attkeys, natt); /* Should never happen. */ @@ -821,6 +1053,9 @@ fetch_remote_table_info(char *nspname, char *relname, * * 3) one of the subscribed publications is declared as ALL TABLES IN * SCHEMA that includes this relation + * + * XXX Does this actually handle puballtables and schema publications + * correctly? */ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000) { @@ -930,8 +1165,24 @@ copy_table(Relation rel) /* Regular table with no row filter */ if (lrel.relkind == RELKIND_RELATION && qual == NIL) - appendStringInfo(&cmd, "COPY %s TO STDOUT", + { + appendStringInfo(&cmd, "COPY %s (", quote_qualified_identifier(lrel.nspname, lrel.relname)); + + /* + * XXX Do we need to list the columns in all cases? Maybe we're replicating + * all columns? + */ + for (int i = 0; i < lrel.natts; i++) + { + if (i > 0) + appendStringInfoString(&cmd, ", "); + + appendStringInfoString(&cmd, quote_identifier(lrel.attnames[i])); + } + + appendStringInfo(&cmd, ") TO STDOUT"); + } else { /* diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index ea57a0477f0..07cdfc1d8c0 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -29,6 +29,7 @@ #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" +#include "utils/rel.h" #include "utils/syscache.h" #include "utils/varlena.h" @@ -85,7 +86,8 @@ static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); static void send_relation_and_attrs(Relation relation, TransactionId xid, - LogicalDecodingContext *ctx); + LogicalDecodingContext *ctx, + Bitmapset *columns); static void send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, XLogRecPtr origin_lsn, bool send_origin); @@ -93,6 +95,8 @@ static void send_repl_origin(LogicalDecodingContext *ctx, /* * Only 3 publication actions are used for row filtering ("insert", "update", * "delete"). See RelationSyncEntry.exprstate[]. + * + * FIXME Do we need something similar for column filters? */ enum RowFilterPubAction { @@ -164,6 +168,13 @@ typedef struct RelationSyncEntry * having identical TupleDesc. */ AttrMap *attrmap; + + /* + * Columns included in the publication, or NULL if all columns are + * included implicitly. Note that the attnums in this bitmap are not + * shifted by FirstLowInvalidHeapAttributeNumber. + */ + Bitmapset *columns; } RelationSyncEntry; /* Map used to remember which relation schemas we sent. */ @@ -603,11 +614,11 @@ maybe_send_schema(LogicalDecodingContext *ctx, { Relation ancestor = RelationIdGetRelation(relentry->publish_as_relid); - send_relation_and_attrs(ancestor, xid, ctx); + send_relation_and_attrs(ancestor, xid, ctx, relentry->columns); RelationClose(ancestor); } - send_relation_and_attrs(relation, xid, ctx); + send_relation_and_attrs(relation, xid, ctx, relentry->columns); if (in_streaming) set_schema_sent_in_streamed_txn(relentry, topxid); @@ -620,7 +631,8 @@ maybe_send_schema(LogicalDecodingContext *ctx, */ static void send_relation_and_attrs(Relation relation, TransactionId xid, - LogicalDecodingContext *ctx) + LogicalDecodingContext *ctx, + Bitmapset *columns) { TupleDesc desc = RelationGetDescr(relation); int i; @@ -643,13 +655,17 @@ send_relation_and_attrs(Relation relation, TransactionId xid, if (att->atttypid < FirstGenbkiObjectId) continue; + /* Skip this attribute if it's not present in the column list */ + if (columns != NULL && !bms_is_member(att->attnum, columns)) + continue; + OutputPluginPrepareWrite(ctx, false); logicalrep_write_typ(ctx->out, xid, att->atttypid); OutputPluginWrite(ctx, false); } OutputPluginPrepareWrite(ctx, false); - logicalrep_write_rel(ctx->out, xid, relation); + logicalrep_write_rel(ctx->out, xid, relation, columns); OutputPluginWrite(ctx, false); } @@ -1224,7 +1240,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, OutputPluginPrepareWrite(ctx, true); logicalrep_write_insert(ctx->out, xid, targetrel, new_slot, - data->binary); + data->binary, relentry->columns); OutputPluginWrite(ctx, true); break; case REORDER_BUFFER_CHANGE_UPDATE: @@ -1278,11 +1294,13 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, { case REORDER_BUFFER_CHANGE_INSERT: logicalrep_write_insert(ctx->out, xid, targetrel, - new_slot, data->binary); + new_slot, data->binary, + relentry->columns); break; case REORDER_BUFFER_CHANGE_UPDATE: logicalrep_write_update(ctx->out, xid, targetrel, - old_slot, new_slot, data->binary); + old_slot, new_slot, data->binary, + relentry->columns); break; case REORDER_BUFFER_CHANGE_DELETE: logicalrep_write_delete(ctx->out, xid, targetrel, @@ -1731,6 +1749,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) memset(entry->exprstate, 0, sizeof(entry->exprstate)); entry->cache_expr_cxt = NULL; entry->publish_as_relid = InvalidOid; + entry->columns = NULL; entry->attrmap = NULL; } @@ -1775,6 +1794,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) entry->schema_sent = false; list_free(entry->streamed_txns); entry->streamed_txns = NIL; + bms_free(entry->columns); + entry->columns = NULL; entry->pubactions.pubinsert = false; entry->pubactions.pubupdate = false; entry->pubactions.pubdelete = false; @@ -1807,13 +1828,16 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) /* * Build publication cache. We can't use one provided by relcache as - * relcache considers all publications given relation is in, but here - * we only need to consider ones that the subscriber requested. + * relcache considers all publications that the given relation is in, + * but here we only need to consider ones that the subscriber + * requested. */ foreach(lc, data->publications) { Publication *pub = lfirst(lc); bool publish = false; + bool ancestor_published = false; + bool all_columns = false; if (pub->alltables) { @@ -1824,8 +1848,6 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) if (!publish) { - bool ancestor_published = false; - /* * For a partition, check if any of the ancestors are * published. If so, note down the topmost ancestor that is @@ -1855,6 +1877,9 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) } /* + * If the relation is to be published, determine actions to + * publish, and list of columns, if appropriate. + * * Don't publish changes for partitioned tables, because * publishing those of its partitions suffices, unless partition * changes won't be published due to pubviaroot being set. @@ -1867,6 +1892,80 @@ get_rel_sync_entry(PGOutputData *data, Relation relation) entry->pubactions.pubdelete |= pub->pubactions.pubdelete; entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate; + /* + * This might be FOR ALL TABLES or FOR ALL TABLES IN SCHEMA + * publication, in which case there are no column lists, and + * we treat that as all_columns=true. + */ + if (pub->alltables || + list_member_oid(schemaPubids, pub->oid)) + { + all_columns = true; + bms_free(entry->columns); + entry->columns = NULL; + } + + /* + * Obtain columns published by this publication, and add them + * to the list for this rel. Note that if at least one + * publication has an empty column list, that means to publish + * everything; so if we saw a publication that includes all + * columns, skip this. + * + * FIXME This fails to consider column filters defined in + * FOR ALL TABLES and FOR ALL TABLES IN SCHEMA publications. + * We need to check those too. + */ + if (!all_columns) + { + HeapTuple pub_rel_tuple; + + pub_rel_tuple = SearchSysCache2(PUBLICATIONRELMAP, + ObjectIdGetDatum(publish_as_relid), + ObjectIdGetDatum(pub->oid)); + + if (HeapTupleIsValid(pub_rel_tuple)) + { + Datum pub_rel_cols; + bool isnull; + + pub_rel_cols = SysCacheGetAttr(PUBLICATIONRELMAP, + pub_rel_tuple, + Anum_pg_publication_rel_prattrs, + &isnull); + if (isnull) + { + /* + * If we see a publication with no column filter, it + * means we need to publish all columns, so reset the + * list and ignore further ones. + */ + all_columns = true; + bms_free(entry->columns); + entry->columns = NULL; + } + else + { + ArrayType *arr; + int nelems; + int16 *elems; + + arr = DatumGetArrayTypeP(pub_rel_cols); + nelems = ARR_DIMS(arr)[0]; + elems = (int16 *) ARR_DATA_PTR(arr); + + /* XXX is there a danger of memory leak here? beware */ + oldctx = MemoryContextSwitchTo(CacheMemoryContext); + for (int i = 0; i < nelems; i++) + entry->columns = bms_add_member(entry->columns, + elems[i]); + MemoryContextSwitchTo(oldctx); + } + + ReleaseSysCache(pub_rel_tuple); + } + } + rel_publications = lappend(rel_publications, pub); } } diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index fccffce5729..82e595396e3 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -5553,6 +5553,8 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) memset(pubdesc, 0, sizeof(PublicationDesc)); pubdesc->rf_valid_for_update = true; pubdesc->rf_valid_for_delete = true; + pubdesc->cf_valid_for_update = true; + pubdesc->cf_valid_for_delete = true; return; } @@ -5565,6 +5567,8 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) memset(pubdesc, 0, sizeof(PublicationDesc)); pubdesc->rf_valid_for_update = true; pubdesc->rf_valid_for_delete = true; + pubdesc->cf_valid_for_update = true; + pubdesc->cf_valid_for_delete = true; /* Fetch the publication membership info. */ puboids = GetRelationPublications(relid); @@ -5625,6 +5629,24 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) pubdesc->rf_valid_for_delete = false; } + /* + * Check if all columns referenced in the column filter are part of + * the REPLICA IDENTITY index or not. + * + * If the publication is FOR ALL TABLES then it means the table has no + * column filters and we can skip the validation. + */ + if (!pubform->puballtables && + (pubform->pubupdate || pubform->pubdelete) && + contain_invalid_cfcolumn(pubid, relation, ancestors, + pubform->pubviaroot)) + { + if (pubform->pubupdate) + pubdesc->cf_valid_for_update = false; + if (pubform->pubdelete) + pubdesc->cf_valid_for_delete = false; + } + ReleaseSysCache(tup); /* @@ -5636,6 +5658,16 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc) pubdesc->pubactions.pubdelete && pubdesc->pubactions.pubtruncate && !pubdesc->rf_valid_for_update && !pubdesc->rf_valid_for_delete) break; + + /* + * If we know everything is replicated and the column filter is invalid + * for update and delete, there is no point to check for other + * publications. + */ + if (pubdesc->pubactions.pubinsert && pubdesc->pubactions.pubupdate && + pubdesc->pubactions.pubdelete && pubdesc->pubactions.pubtruncate && + !pubdesc->cf_valid_for_update && !pubdesc->cf_valid_for_delete) + break; } if (relation->rd_pubdesc) diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index e69dcf8a484..f208c7a6c59 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -4075,6 +4075,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) int i_prpubid; int i_prrelid; int i_prrelqual; + int i_prattrs; int i, j, ntups; @@ -4088,12 +4089,20 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) if (fout->remoteVersion >= 150000) appendPQExpBufferStr(query, "SELECT tableoid, oid, prpubid, prrelid, " - "pg_catalog.pg_get_expr(prqual, prrelid) AS prrelqual " - "FROM pg_catalog.pg_publication_rel"); + "pg_catalog.pg_get_expr(prqual, prrelid) AS prrelqual, " + "(CASE\n" + " WHEN pr.prattrs IS NOT NULL THEN\n" + " (SELECT array_agg(attname)\n" + " FROM\n" + " pg_catalog.generate_series(0, pg_catalog.array_upper(pr.prattrs::pg_catalog.int2[], 1)) s,\n" + " pg_catalog.pg_attribute\n" + " WHERE attrelid = pr.prrelid AND attnum = prattrs[s])\n" + " ELSE NULL END) prattrs " + "FROM pg_catalog.pg_publication_rel pr"); else appendPQExpBufferStr(query, "SELECT tableoid, oid, prpubid, prrelid, " - "NULL AS prrelqual " + "NULL AS prrelqual, NULL AS prattrs " "FROM pg_catalog.pg_publication_rel"); res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK); @@ -4104,6 +4113,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) i_prpubid = PQfnumber(res, "prpubid"); i_prrelid = PQfnumber(res, "prrelid"); i_prrelqual = PQfnumber(res, "prrelqual"); + i_prattrs = PQfnumber(res, "prattrs"); /* this allocation may be more than we need */ pubrinfo = pg_malloc(ntups * sizeof(PublicationRelInfo)); @@ -4149,6 +4159,28 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables) else pubrinfo[j].pubrelqual = pg_strdup(PQgetvalue(res, i, i_prrelqual)); + if (!PQgetisnull(res, i, i_prattrs)) + { + char **attnames; + int nattnames; + PQExpBuffer attribs; + + if (!parsePGArray(PQgetvalue(res, i, i_prattrs), + &attnames, &nattnames)) + fatal("could not parse %s array", "prattrs"); + attribs = createPQExpBuffer(); + for (int k = 0; k < nattnames; k++) + { + if (k > 0) + appendPQExpBufferStr(attribs, ", "); + + appendPQExpBufferStr(attribs, fmtId(attnames[k])); + } + pubrinfo[j].pubrattrs = attribs->data; + } + else + pubrinfo[j].pubrattrs = NULL; + /* Decide whether we want to dump it */ selectDumpablePublicationObject(&(pubrinfo[j].dobj), fout); @@ -4223,10 +4255,13 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo) query = createPQExpBuffer(); - appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD TABLE ONLY", + appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD TABLE ONLY ", fmtId(pubinfo->dobj.name)); - appendPQExpBuffer(query, " %s", - fmtQualifiedDumpable(tbinfo)); + appendPQExpBufferStr(query, fmtQualifiedDumpable(tbinfo)); + + if (pubrinfo->pubrattrs) + appendPQExpBuffer(query, " (%s)", pubrinfo->pubrattrs); + if (pubrinfo->pubrelqual) { /* diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index 997a3b60719..680b07dcd52 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -632,6 +632,7 @@ typedef struct _PublicationRelInfo PublicationInfo *publication; TableInfo *pubtable; char *pubrelqual; + char *pubrattrs; } PublicationRelInfo; /* diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl index ae8c86a6e88..d4d92465c49 100644 --- a/src/bin/pg_dump/t/002_pg_dump.pl +++ b/src/bin/pg_dump/t/002_pg_dump.pl @@ -2428,6 +2428,28 @@ my %tests = ( unlike => { exclude_dump_test_schema => 1, }, }, + 'ALTER PUBLICATION pub1 ADD TABLE test_sixth_table (col3, col2)' => { + create_order => 52, + create_sql => + 'ALTER PUBLICATION pub1 ADD TABLE dump_test.test_sixth_table (col3, col2);', + regexp => qr/^ + \QALTER PUBLICATION pub1 ADD TABLE ONLY dump_test.test_sixth_table (col2, col3);\E + /xm, + like => { %full_runs, section_post_data => 1, }, + unlike => { exclude_dump_test_schema => 1, }, + }, + + 'ALTER PUBLICATION pub1 ADD TABLE test_seventh_table (col3, col2) WHERE (col1 = 1)' => { + create_order => 52, + create_sql => + 'ALTER PUBLICATION pub1 ADD TABLE dump_test.test_seventh_table (col3, col2) WHERE (col1 = 1);', + regexp => qr/^ + \QALTER PUBLICATION pub1 ADD TABLE ONLY dump_test.test_seventh_table (col2, col3) WHERE ((col1 = 1));\E + /xm, + like => { %full_runs, section_post_data => 1, }, + unlike => { exclude_dump_test_schema => 1, }, + }, + 'ALTER PUBLICATION pub3 ADD ALL TABLES IN SCHEMA dump_test' => { create_order => 51, create_sql => @@ -2778,6 +2800,44 @@ my %tests = ( unlike => { exclude_dump_test_schema => 1, }, }, + 'CREATE TABLE test_sixth_table' => { + create_order => 6, + create_sql => 'CREATE TABLE dump_test.test_sixth_table ( + col1 int, + col2 text, + col3 bytea + );', + regexp => qr/^ + \QCREATE TABLE dump_test.test_sixth_table (\E + \n\s+\Qcol1 integer,\E + \n\s+\Qcol2 text,\E + \n\s+\Qcol3 bytea\E + \n\); + /xm, + like => + { %full_runs, %dump_test_schema_runs, section_pre_data => 1, }, + unlike => { exclude_dump_test_schema => 1, }, + }, + + 'CREATE TABLE test_seventh_table' => { + create_order => 6, + create_sql => 'CREATE TABLE dump_test.test_seventh_table ( + col1 int, + col2 text, + col3 bytea + );', + regexp => qr/^ + \QCREATE TABLE dump_test.test_seventh_table (\E + \n\s+\Qcol1 integer,\E + \n\s+\Qcol2 text,\E + \n\s+\Qcol3 bytea\E + \n\); + /xm, + like => + { %full_runs, %dump_test_schema_runs, section_pre_data => 1, }, + unlike => { exclude_dump_test_schema => 1, }, + }, + 'CREATE TABLE test_table_identity' => { create_order => 3, create_sql => 'CREATE TABLE dump_test.test_table_identity ( diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index e3382933d98..fb18cb82d9f 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -2880,6 +2880,7 @@ describeOneTableDetails(const char *schemaname, printfPQExpBuffer(&buf, "SELECT pubname\n" " , NULL\n" + " , NULL\n" "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" @@ -2887,6 +2888,12 @@ describeOneTableDetails(const char *schemaname, "UNION\n" "SELECT pubname\n" " , pg_get_expr(pr.prqual, c.oid)\n" + " , (CASE WHEN pr.prattrs IS NOT NULL THEN\n" + " (SELECT string_agg(attname, ', ')\n" + " FROM pg_catalog.generate_series(0, pg_catalog.array_upper(pr.prattrs::pg_catalog.int2[], 1)) s,\n" + " pg_catalog.pg_attribute\n" + " WHERE attrelid = pr.prrelid AND attnum = prattrs[s])\n" + " ELSE NULL END) " "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" @@ -2894,6 +2901,7 @@ describeOneTableDetails(const char *schemaname, "UNION\n" "SELECT pubname\n" " , NULL\n" + " , NULL\n" "FROM pg_catalog.pg_publication p\n" "WHERE p.puballtables AND pg_catalog.pg_relation_is_publishable('%s')\n" "ORDER BY 1;", @@ -2904,12 +2912,14 @@ describeOneTableDetails(const char *schemaname, printfPQExpBuffer(&buf, "SELECT pubname\n" " , NULL\n" + " , NULL\n" "FROM pg_catalog.pg_publication p\n" "JOIN pg_catalog.pg_publication_rel pr ON p.oid = pr.prpubid\n" "WHERE pr.prrelid = '%s'\n" "UNION ALL\n" "SELECT pubname\n" " , NULL\n" + " , NULL\n" "FROM pg_catalog.pg_publication p\n" "WHERE p.puballtables AND pg_catalog.pg_relation_is_publishable('%s')\n" "ORDER BY 1;", @@ -2931,6 +2941,11 @@ describeOneTableDetails(const char *schemaname, printfPQExpBuffer(&buf, " \"%s\"", PQgetvalue(result, i, 0)); + /* column filter (if any) */ + if (!PQgetisnull(result, i, 2)) + appendPQExpBuffer(&buf, " (%s)", + PQgetvalue(result, i, 2)); + /* row filter (if any) */ if (!PQgetisnull(result, i, 1)) appendPQExpBuffer(&buf, " WHERE %s", @@ -5867,7 +5882,7 @@ listPublications(const char *pattern) */ static bool addFooterToPublicationDesc(PQExpBuffer buf, char *footermsg, - bool singlecol, printTableContent *cont) + bool as_schema, printTableContent *cont) { PGresult *res; int count = 0; @@ -5884,15 +5899,19 @@ addFooterToPublicationDesc(PQExpBuffer buf, char *footermsg, for (i = 0; i < count; i++) { - if (!singlecol) + if (as_schema) + printfPQExpBuffer(buf, " \"%s\"", PQgetvalue(res, i, 0)); + else { printfPQExpBuffer(buf, " \"%s.%s\"", PQgetvalue(res, i, 0), PQgetvalue(res, i, 1)); + + if (!PQgetisnull(res, i, 3)) + appendPQExpBuffer(buf, " (%s)", PQgetvalue(res, i, 3)); + if (!PQgetisnull(res, i, 2)) appendPQExpBuffer(buf, " WHERE %s", PQgetvalue(res, i, 2)); } - else - printfPQExpBuffer(buf, " \"%s\"", PQgetvalue(res, i, 0)); printTableAddFooter(cont, buf->data); } @@ -6021,11 +6040,22 @@ describePublications(const char *pattern) printfPQExpBuffer(&buf, "SELECT n.nspname, c.relname"); if (pset.sversion >= 150000) + { appendPQExpBufferStr(&buf, ", pg_get_expr(pr.prqual, c.oid)"); + appendPQExpBufferStr(&buf, + ", (CASE WHEN pr.prattrs IS NOT NULL THEN\n" + " pg_catalog.array_to_string(" + " ARRAY(SELECT attname\n" + " FROM\n" + " pg_catalog.generate_series(0, pg_catalog.array_upper(pr.prattrs::pg_catalog.int2[], 1)) s,\n" + " pg_catalog.pg_attribute\n" + " WHERE attrelid = c.oid AND attnum = prattrs[s]), ', ')\n" + " ELSE NULL END)"); + } else appendPQExpBufferStr(&buf, - ", NULL"); + ", NULL, NULL"); appendPQExpBuffer(&buf, "\nFROM pg_catalog.pg_class c,\n" " pg_catalog.pg_namespace n,\n" diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index ba72e62e614..b58f85ede27 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -85,6 +85,14 @@ typedef struct PublicationDesc */ bool rf_valid_for_update; bool rf_valid_for_delete; + + /* + * true if the columns referenced in column filters which are used for UPDATE + * or DELETE are part of the replica identity or the publication actions + * do not include UPDATE or DELETE. + */ + bool cf_valid_for_update; + bool cf_valid_for_delete; } PublicationDesc; typedef struct Publication @@ -100,6 +108,7 @@ typedef struct PublicationRelInfo { Relation relation; Node *whereClause; + List *columns; } PublicationRelInfo; extern Publication *GetPublication(Oid pubid); @@ -123,8 +132,11 @@ typedef enum PublicationPartOpt } PublicationPartOpt; extern List *GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt); +extern List *GetRelationColumnPartialPublications(Oid relid); +extern List *GetRelationColumnListInPublication(Oid relid, Oid pubid); extern List *GetAllTablesPublications(void); extern List *GetAllTablesPublicationRelations(bool pubviaroot); +extern void GetActionsInPublication(Oid pubid, PublicationActions *actions); extern List *GetPublicationSchemas(Oid pubid); extern List *GetSchemaPublications(Oid schemaid); extern List *GetSchemaPublicationRelations(Oid schemaid, @@ -142,6 +154,8 @@ extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *pri bool if_not_exists); extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists); +extern void publication_set_table_columns(Relation pubrel, HeapTuple pubreltup, + Relation targetrel, List *columns); extern Oid get_publication_oid(const char *pubname, bool missing_ok); extern char *get_publication_name(Oid pubid, bool missing_ok); diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h index 0dd0f425db9..4feb581899e 100644 --- a/src/include/catalog/pg_publication_rel.h +++ b/src/include/catalog/pg_publication_rel.h @@ -34,6 +34,7 @@ CATALOG(pg_publication_rel,6106,PublicationRelRelationId) #ifdef CATALOG_VARLEN /* variable-length fields start here */ pg_node_tree prqual; /* qualifications */ + int2vector prattrs; /* columns to replicate */ #endif } FormData_pg_publication_rel; diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h index 7813cbcb6bb..08d14ca7245 100644 --- a/src/include/commands/publicationcmds.h +++ b/src/include/commands/publicationcmds.h @@ -33,5 +33,7 @@ extern void AlterPublicationOwner_oid(Oid pubid, Oid newOwnerId); extern void InvalidatePublicationRels(List *relids); extern bool contain_invalid_rfcolumn(Oid pubid, Relation relation, List *ancestors, bool pubviaroot); +extern bool contain_invalid_cfcolumn(Oid pubid, Relation relation, + List *ancestors, bool pubviaroot); #endif /* PUBLICATIONCMDS_H */ diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 1617702d9d6..1375a173e3b 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3652,6 +3652,7 @@ typedef struct PublicationTable NodeTag type; RangeVar *relation; /* relation to be published */ Node *whereClause; /* qualifications */ + List *columns; /* List of columns in a publication table */ } PublicationTable; /* @@ -3688,7 +3689,8 @@ typedef enum AlterPublicationAction { AP_AddObjects, /* add objects to publication */ AP_DropObjects, /* remove objects from publication */ - AP_SetObjects /* set list of objects */ + AP_SetObjects, /* set list of objects */ + AP_SetColumns /* change list of columns for a table */ } AlterPublicationAction; typedef struct AlterPublicationStmt diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 4d2c881644a..a771ab8ff33 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -209,12 +209,12 @@ extern char *logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn); extern void logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *newslot, - bool binary); + bool binary, Bitmapset *columns); extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup); extern void logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, - TupleTableSlot *newslot, bool binary); + TupleTableSlot *newslot, bool binary, Bitmapset *columns); extern LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup); @@ -231,7 +231,7 @@ extern List *logicalrep_read_truncate(StringInfo in, extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message); extern void logicalrep_write_rel(StringInfo out, TransactionId xid, - Relation rel); + Relation rel, Bitmapset *columns); extern LogicalRepRelation *logicalrep_read_rel(StringInfo in); extern void logicalrep_write_typ(StringInfo out, TransactionId xid, Oid typoid); diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out index 6d16600aaea..152f19fb42b 100644 --- a/src/test/regress/expected/publication.out +++ b/src/test/regress/expected/publication.out @@ -679,6 +679,372 @@ DROP TABLE rf_tbl_abcd_pk; DROP TABLE rf_tbl_abcd_nopk; DROP TABLE rf_tbl_abcd_part_pk; -- ====================================================== +-- fail - duplicate tables are not allowed if that table has any column filters +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_dups FOR TABLE testpub_tbl1 (a), testpub_tbl1 WITH (publish = 'insert'); +ERROR: conflicting or redundant column filters for table "testpub_tbl1" +CREATE PUBLICATION testpub_dups FOR TABLE testpub_tbl1, testpub_tbl1 (a) WITH (publish = 'insert'); +ERROR: conflicting or redundant column filters for table "testpub_tbl1" +RESET client_min_messages; +-- test for column filters +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_fortable FOR TABLE testpub_tbl1; +CREATE PUBLICATION testpub_fortable_insert WITH (publish = 'insert'); +RESET client_min_messages; +CREATE TABLE testpub_tbl5 (a int PRIMARY KEY, b text, c text, + d int generated always as (a + length(b)) stored); +-- error: column "x" does not exist +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, x); +ERROR: column "x" of relation "testpub_tbl5" does not exist +-- error: replica identity "a" not included in the column filter +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (b, c); +UPDATE testpub_tbl5 SET a = 1; +ERROR: cannot update table "testpub_tbl5" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub_fortable DROP TABLE testpub_tbl5; +-- error: generated column "d" can't be in list +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, d); +ERROR: cannot reference generated column "d" in publication column list +-- error: system attributes "ctid" not allowed in column filter +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, ctid); +ERROR: cannot reference system column "ctid" in publication column list +-- ok +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, c); +ALTER TABLE testpub_tbl5 DROP COLUMN c; -- no dice +ERROR: cannot drop column "c" because it is part of publication "testpub_fortable" +HINT: Specify CASCADE or use ALTER PUBLICATION to remove the column from the publication. +-- ok: for insert-only publication, the filter is arbitrary +ALTER PUBLICATION testpub_fortable_insert ADD TABLE testpub_tbl5 (b, c); +/* not all replica identities are good enough */ +CREATE UNIQUE INDEX testpub_tbl5_b_key ON testpub_tbl5 (b, c); +ALTER TABLE testpub_tbl5 ALTER b SET NOT NULL, ALTER c SET NOT NULL; +ALTER TABLE testpub_tbl5 REPLICA IDENTITY USING INDEX testpub_tbl5_b_key; +-- error: replica identity (b,c) is covered by column filter (a, c) +UPDATE testpub_tbl5 SET a = 1; +ERROR: cannot update table "testpub_tbl5" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub_fortable DROP TABLE testpub_tbl5; +-- error: change the replica identity to "b", and then filter (a, c) fails +ALTER TABLE testpub_tbl5 REPLICA IDENTITY USING INDEX testpub_tbl5_b_key; +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, c); +-- error: replica identity (b,c) is not covered by column filter (a,c) +UPDATE testpub_tbl5 SET a = 1; +ERROR: cannot update table "testpub_tbl5" +DETAIL: Column filter used by the publication does not cover the replica identity. +/* But if upd/del are not published, it works OK */ +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_table_ins WITH (publish = 'insert, truncate'); +RESET client_min_messages; +ALTER PUBLICATION testpub_table_ins ADD TABLE testpub_tbl5 (a); -- ok +\dRp+ testpub_table_ins + Publication testpub_table_ins + Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root +--------------------------+------------+---------+---------+---------+-----------+---------- + regress_publication_user | f | t | f | f | t | f +Tables: + "public.testpub_tbl5" (a) + +-- with REPLICA IDENTITY FULL, column filters are not allowed +CREATE TABLE testpub_tbl6 (a int, b text, c text); +ALTER TABLE testpub_tbl6 REPLICA IDENTITY FULL; +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl6 (a, b, c); +UPDATE testpub_tbl6 SET a = 1; +ERROR: cannot update table "testpub_tbl6" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub_fortable DROP TABLE testpub_tbl6; +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl6; -- ok +UPDATE testpub_tbl6 SET a = 1; +ALTER PUBLICATION testpub_fortable ALTER TABLE testpub_tbl6 SET COLUMNS (a, b, c); +UPDATE testpub_tbl6 SET a = 1; +ERROR: cannot update table "testpub_tbl6" +DETAIL: Column filter used by the publication does not cover the replica identity. +-- make sure changing the column filter is updated in SET TABLE +CREATE TABLE testpub_tbl7 (a int primary key, b text, c text); +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl7 (a, b); +\d+ testpub_tbl7 + Table "public.testpub_tbl7" + Column | Type | Collation | Nullable | Default | Storage | Stats target | Description +--------+---------+-----------+----------+---------+----------+--------------+------------- + a | integer | | not null | | plain | | + b | text | | | | extended | | + c | text | | | | extended | | +Indexes: + "testpub_tbl7_pkey" PRIMARY KEY, btree (a) +Publications: + "testpub_fortable" (a, b) + +-- ok: we'll skip this table +ALTER PUBLICATION testpub_fortable SET TABLE testpub_tbl7 (a, b); +\d+ testpub_tbl7 + Table "public.testpub_tbl7" + Column | Type | Collation | Nullable | Default | Storage | Stats target | Description +--------+---------+-----------+----------+---------+----------+--------------+------------- + a | integer | | not null | | plain | | + b | text | | | | extended | | + c | text | | | | extended | | +Indexes: + "testpub_tbl7_pkey" PRIMARY KEY, btree (a) +Publications: + "testpub_fortable" (a, b) + +-- ok: update the column filter +ALTER PUBLICATION testpub_fortable SET TABLE testpub_tbl7 (a, c); +\d+ testpub_tbl7 + Table "public.testpub_tbl7" + Column | Type | Collation | Nullable | Default | Storage | Stats target | Description +--------+---------+-----------+----------+---------+----------+--------------+------------- + a | integer | | not null | | plain | | + b | text | | | | extended | | + c | text | | | | extended | | +Indexes: + "testpub_tbl7_pkey" PRIMARY KEY, btree (a) +Publications: + "testpub_fortable" (a, c) + +-- column filter for partitioned tables has to cover replica identities for +-- all child relations +CREATE TABLE testpub_tbl8 (a int, b text, c text) PARTITION BY HASH (a); +-- first partition has replica identity "a" +CREATE TABLE testpub_tbl8_0 PARTITION OF testpub_tbl8 FOR VALUES WITH (modulus 2, remainder 0); +ALTER TABLE testpub_tbl8_0 ADD PRIMARY KEY (a); +ALTER TABLE testpub_tbl8_0 REPLICA IDENTITY USING INDEX testpub_tbl8_0_pkey; +-- second partition has replica identity "b" +CREATE TABLE testpub_tbl8_1 PARTITION OF testpub_tbl8 FOR VALUES WITH (modulus 2, remainder 1); +ALTER TABLE testpub_tbl8_1 ADD PRIMARY KEY (b); +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; +-- ok: column filter covers both "a" and "b" +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_col_filter FOR TABLE testpub_tbl8 (a, b) WITH (publish_via_partition_root = 'true'); +RESET client_min_messages; +-- ok: the same thing, but try plain ADD TABLE +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, b); +UPDATE testpub_tbl8 SET a = 1; +-- failure: column filter does not cover replica identity for the second partition +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, c); +UPDATE testpub_tbl8 SET a = 1; +ERROR: cannot update table "testpub_tbl8_1" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; +-- failure: one of the partitions has REPLICA IDENTITY FULL +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY FULL; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, c); +UPDATE testpub_tbl8 SET a = 1; +ERROR: cannot update table "testpub_tbl8_1" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; +-- add table and then try changing replica identity +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, b); +-- failure: replica identity full can't be used with a column filter +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY FULL; +UPDATE testpub_tbl8 SET a = 1; +ERROR: cannot update table "testpub_tbl8_1" +DETAIL: Column filter used by the publication does not cover the replica identity. +-- failure: replica identity has to be covered by the column filter +ALTER TABLE testpub_tbl8_1 DROP CONSTRAINT testpub_tbl8_1_pkey; +ALTER TABLE testpub_tbl8_1 ADD PRIMARY KEY (c); +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; +UPDATE testpub_tbl8 SET a = 1; +ERROR: cannot update table "testpub_tbl8_1" +DETAIL: Column filter used by the publication does not cover the replica identity. +DROP TABLE testpub_tbl8; +-- column filter for partitioned tables has to cover replica identities for +-- all child relations +CREATE TABLE testpub_tbl8 (a int, b text, c text) PARTITION BY HASH (a); +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, b); +-- first partition has replica identity "a" +CREATE TABLE testpub_tbl8_0 (a int, b text, c text); +ALTER TABLE testpub_tbl8_0 ADD PRIMARY KEY (a); +ALTER TABLE testpub_tbl8_0 REPLICA IDENTITY USING INDEX testpub_tbl8_0_pkey; +-- second partition has replica identity "b" +CREATE TABLE testpub_tbl8_1 (a int, b text, c text); +ALTER TABLE testpub_tbl8_1 ADD PRIMARY KEY (c); +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; +-- ok: attaching first partition works, because (a) is in column filter +ALTER TABLE testpub_tbl8 ATTACH PARTITION testpub_tbl8_0 FOR VALUES WITH (modulus 2, remainder 0); +-- failure: second partition has replica identity (c), which si not in column filter +ALTER TABLE testpub_tbl8 ATTACH PARTITION testpub_tbl8_1 FOR VALUES WITH (modulus 2, remainder 1); +UPDATE testpub_tbl8 SET a = 1; +ERROR: cannot update table "testpub_tbl8_1" +DETAIL: Column filter used by the publication does not cover the replica identity. +-- failure: changing replica identity to FULL for partition fails, because +-- of the column filter on the parent +ALTER TABLE testpub_tbl8_0 REPLICA IDENTITY FULL; +UPDATE testpub_tbl8 SET a = 1; +ERROR: cannot update table "testpub_tbl8_0" +DETAIL: Column filter used by the publication does not cover the replica identity. +DROP TABLE testpub_tbl5, testpub_tbl6, testpub_tbl7, testpub_tbl8, testpub_tbl8_1; +DROP PUBLICATION testpub_table_ins, testpub_fortable, testpub_fortable_insert, testpub_col_filter; +-- ====================================================== +-- Test combination of column and row filter +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_both_filters; +RESET client_min_messages; +CREATE TABLE testpub_tbl_both_filters (a int, b int, c int, PRIMARY KEY (a,c)); +ALTER TABLE testpub_tbl_both_filters REPLICA IDENTITY USING INDEX testpub_tbl_both_filters_pkey; +ALTER PUBLICATION testpub_both_filters ADD TABLE testpub_tbl_both_filters (a,c) WHERE (c != 1); +\dRp+ testpub_both_filters + Publication testpub_both_filters + Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root +--------------------------+------------+---------+---------+---------+-----------+---------- + regress_publication_user | f | t | t | t | t | f +Tables: + "public.testpub_tbl_both_filters" (a, c) WHERE (c <> 1) + +\d+ testpub_tbl_both_filters + Table "public.testpub_tbl_both_filters" + Column | Type | Collation | Nullable | Default | Storage | Stats target | Description +--------+---------+-----------+----------+---------+---------+--------------+------------- + a | integer | | not null | | plain | | + b | integer | | | | plain | | + c | integer | | not null | | plain | | +Indexes: + "testpub_tbl_both_filters_pkey" PRIMARY KEY, btree (a, c) REPLICA IDENTITY +Publications: + "testpub_both_filters" (a, c) WHERE (c <> 1) + +DROP TABLE testpub_tbl_both_filters; +DROP PUBLICATION testpub_both_filters; +-- ====================================================== +-- More column filter tests for validating column references +CREATE TABLE rf_tbl_abcd_nopk(a int, b int, c int, d int); +CREATE TABLE rf_tbl_abcd_pk(a int, b int, c int, d int, PRIMARY KEY(a,b)); +CREATE TABLE rf_tbl_abcd_part_pk (a int PRIMARY KEY, b int) PARTITION by RANGE (a); +CREATE TABLE rf_tbl_abcd_part_pk_1 (b int, a int PRIMARY KEY); +ALTER TABLE rf_tbl_abcd_part_pk ATTACH PARTITION rf_tbl_abcd_part_pk_1 FOR VALUES FROM (1) TO (10); +-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing) +-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK. +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk (a, b); +RESET client_min_messages; +-- ok - (a,b) coverts all PK cols +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a, b, c); +-- ok - (a,b,c) coverts all PK cols +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a); +-- fail - "b" is missing from the column filter +UPDATE rf_tbl_abcd_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_pk" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (b); +-- fail - "a" is missing from the column filter +UPDATE rf_tbl_abcd_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_pk" +DETAIL: Column filter used by the publication does not cover the replica identity. +-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (a); +-- ok - there's no replica identity, so any column filter works +-- note: it fails anyway, just a bit later because UPDATE requires RI +UPDATE rf_tbl_abcd_nopk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_nopk" because it does not have a replica identity and publishes updates +HINT: To enable updating the table, set REPLICA IDENTITY using ALTER TABLE. +-- Case 2. REPLICA IDENTITY FULL +ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL; +ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (c); +-- fail - with REPLICA IDENTITY FULL no column filter is allowed +UPDATE rf_tbl_abcd_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_pk" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (a, b, c, d); +-- fail - with REPLICA IDENTITY FULL no column filter is allowed +UPDATE rf_tbl_abcd_nopk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_nopk" +DETAIL: Column filter used by the publication does not cover the replica identity. +-- Case 3. REPLICA IDENTITY NOTHING +ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING; +ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a); +-- ok - REPLICA IDENTITY NOTHING means all column filters are valid +-- it still fails later because without RI we can't replicate updates +UPDATE rf_tbl_abcd_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_pk" because it does not have a replica identity and publishes updates +HINT: To enable updating the table, set REPLICA IDENTITY using ALTER TABLE. +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a, b, c, d); +-- ok - REPLICA IDENTITY NOTHING means all column filters are valid +-- it still fails later because without RI we can't replicate updates +UPDATE rf_tbl_abcd_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_pk" because it does not have a replica identity and publishes updates +HINT: To enable updating the table, set REPLICA IDENTITY using ALTER TABLE. +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (d); +-- ok - REPLICA IDENTITY NOTHING means all column filters are valid +-- it still fails later because without RI we can't replicate updates +UPDATE rf_tbl_abcd_nopk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_nopk" because it does not have a replica identity and publishes updates +HINT: To enable updating the table, set REPLICA IDENTITY using ALTER TABLE. +-- Case 4. REPLICA IDENTITY INDEX +ALTER TABLE rf_tbl_abcd_pk ALTER COLUMN c SET NOT NULL; +CREATE UNIQUE INDEX idx_abcd_pk_c ON rf_tbl_abcd_pk(c); +ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY USING INDEX idx_abcd_pk_c; +ALTER TABLE rf_tbl_abcd_nopk ALTER COLUMN c SET NOT NULL; +CREATE UNIQUE INDEX idx_abcd_nopk_c ON rf_tbl_abcd_nopk(c); +ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY USING INDEX idx_abcd_nopk_c; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a); +-- fail - column filter "a" does not cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_pk" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (c); +-- ok - column filter "c" does cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (a); +-- fail - column filter "a" does not cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_nopk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_nopk" +DETAIL: Column filter used by the publication does not cover the replica identity. +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (c); +-- ok - column filter "c" does cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_nopk SET a = 1; +-- Tests for partitioned table +-- set PUBLISH_VIA_PARTITION_ROOT to false and test row filter for partitioned +-- table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0); +-- fail - cannot use row filter for partitioned table +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk (a); +ERROR: cannot use publication column list for relation "rf_tbl_abcd_part_pk" +DETAIL: column list cannot be used for a partitioned table when publish_via_partition_root is false. +-- ok - can use row filter for partition +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk_1 (a); +-- ok - "a" is a PK col +UPDATE rf_tbl_abcd_part_pk SET a = 1; +-- set PUBLISH_VIA_PARTITION_ROOT to true and test row filter for partitioned +-- table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1); +-- ok - can use row filter for partitioned table +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk (a); +-- ok - "a" is a PK col +UPDATE rf_tbl_abcd_part_pk SET a = 1; +-- fail - cannot set PUBLISH_VIA_PARTITION_ROOT to false if any row filter is +-- used for partitioned table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0); +ERROR: cannot set publish_via_partition_root = false for publication "testpub6" +DETAIL: The publication contains a column filter for a partitioned table "rf_tbl_abcd_part_pk" which is not allowed when publish_via_partition_root is false. +-- Now change the root filter to use a column "b" +-- (which is not in the replica identity) +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk_1 (b); +-- ok - we don't have row filter for partitioned table. +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0); +-- fail - "b" is not in REPLICA IDENTITY INDEX +UPDATE rf_tbl_abcd_part_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_part_pk_1" +DETAIL: Column filter used by the publication does not cover the replica identity. +-- set PUBLISH_VIA_PARTITION_ROOT to true +-- can use row filter for partitioned table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1); +-- ok - can use row filter for partitioned table +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk (b); +-- fail - "b" is not in REPLICA IDENTITY INDEX +UPDATE rf_tbl_abcd_part_pk SET a = 1; +ERROR: cannot update table "rf_tbl_abcd_part_pk_1" +DETAIL: Column filter used by the publication does not cover the replica identity. +DROP PUBLICATION testpub6; +DROP TABLE rf_tbl_abcd_pk; +DROP TABLE rf_tbl_abcd_nopk; +DROP TABLE rf_tbl_abcd_part_pk; +-- ====================================================== -- Test cache invalidation FOR ALL TABLES publication SET client_min_messages = 'ERROR'; CREATE TABLE testpub_tbl4(a int); @@ -1124,6 +1490,15 @@ ALTER PUBLICATION testpub1_forschema SET ALL TABLES IN SCHEMA pub_test1, pub_tes Tables from schemas: "pub_test1" +-- Verify that it fails to add a schema with a column specification +ALTER PUBLICATION testpub1_forschema ADD ALL TABLES IN SCHEMA foo (a, b); +ERROR: syntax error at or near "(" +LINE 1: ...TION testpub1_forschema ADD ALL TABLES IN SCHEMA foo (a, b); + ^ +ALTER PUBLICATION testpub1_forschema ADD ALL TABLES IN SCHEMA foo, bar (a, b); +ERROR: column specification not allowed for schema +LINE 1: ... testpub1_forschema ADD ALL TABLES IN SCHEMA foo, bar (a, b)... + ^ -- cleanup pub_test1 schema for invalidation tests ALTER PUBLICATION testpub2_forschema DROP ALL TABLES IN SCHEMA pub_test1; DROP PUBLICATION testpub3_forschema, testpub4_forschema, testpub5_forschema, testpub6_forschema, testpub_fortable; diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql index c135a601a30..2203dc238d2 100644 --- a/src/test/regress/sql/publication.sql +++ b/src/test/regress/sql/publication.sql @@ -381,6 +381,292 @@ DROP TABLE rf_tbl_abcd_nopk; DROP TABLE rf_tbl_abcd_part_pk; -- ====================================================== +-- fail - duplicate tables are not allowed if that table has any column filters +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_dups FOR TABLE testpub_tbl1 (a), testpub_tbl1 WITH (publish = 'insert'); +CREATE PUBLICATION testpub_dups FOR TABLE testpub_tbl1, testpub_tbl1 (a) WITH (publish = 'insert'); +RESET client_min_messages; + +-- test for column filters +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_fortable FOR TABLE testpub_tbl1; +CREATE PUBLICATION testpub_fortable_insert WITH (publish = 'insert'); +RESET client_min_messages; +CREATE TABLE testpub_tbl5 (a int PRIMARY KEY, b text, c text, + d int generated always as (a + length(b)) stored); +-- error: column "x" does not exist +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, x); +-- error: replica identity "a" not included in the column filter +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (b, c); +UPDATE testpub_tbl5 SET a = 1; +ALTER PUBLICATION testpub_fortable DROP TABLE testpub_tbl5; +-- error: generated column "d" can't be in list +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, d); +-- error: system attributes "ctid" not allowed in column filter +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, ctid); +-- ok +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, c); +ALTER TABLE testpub_tbl5 DROP COLUMN c; -- no dice +-- ok: for insert-only publication, the filter is arbitrary +ALTER PUBLICATION testpub_fortable_insert ADD TABLE testpub_tbl5 (b, c); + +/* not all replica identities are good enough */ +CREATE UNIQUE INDEX testpub_tbl5_b_key ON testpub_tbl5 (b, c); +ALTER TABLE testpub_tbl5 ALTER b SET NOT NULL, ALTER c SET NOT NULL; +ALTER TABLE testpub_tbl5 REPLICA IDENTITY USING INDEX testpub_tbl5_b_key; +-- error: replica identity (b,c) is covered by column filter (a, c) +UPDATE testpub_tbl5 SET a = 1; +ALTER PUBLICATION testpub_fortable DROP TABLE testpub_tbl5; + +-- error: change the replica identity to "b", and then filter (a, c) fails +ALTER TABLE testpub_tbl5 REPLICA IDENTITY USING INDEX testpub_tbl5_b_key; +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl5 (a, c); +-- error: replica identity (b,c) is not covered by column filter (a,c) +UPDATE testpub_tbl5 SET a = 1; + +/* But if upd/del are not published, it works OK */ +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_table_ins WITH (publish = 'insert, truncate'); +RESET client_min_messages; +ALTER PUBLICATION testpub_table_ins ADD TABLE testpub_tbl5 (a); -- ok +\dRp+ testpub_table_ins + +-- with REPLICA IDENTITY FULL, column filters are not allowed +CREATE TABLE testpub_tbl6 (a int, b text, c text); +ALTER TABLE testpub_tbl6 REPLICA IDENTITY FULL; + +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl6 (a, b, c); +UPDATE testpub_tbl6 SET a = 1; +ALTER PUBLICATION testpub_fortable DROP TABLE testpub_tbl6; + +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl6; -- ok +UPDATE testpub_tbl6 SET a = 1; + +ALTER PUBLICATION testpub_fortable ALTER TABLE testpub_tbl6 SET COLUMNS (a, b, c); +UPDATE testpub_tbl6 SET a = 1; + +-- make sure changing the column filter is updated in SET TABLE +CREATE TABLE testpub_tbl7 (a int primary key, b text, c text); +ALTER PUBLICATION testpub_fortable ADD TABLE testpub_tbl7 (a, b); +\d+ testpub_tbl7 +-- ok: we'll skip this table +ALTER PUBLICATION testpub_fortable SET TABLE testpub_tbl7 (a, b); +\d+ testpub_tbl7 +-- ok: update the column filter +ALTER PUBLICATION testpub_fortable SET TABLE testpub_tbl7 (a, c); +\d+ testpub_tbl7 + +-- column filter for partitioned tables has to cover replica identities for +-- all child relations +CREATE TABLE testpub_tbl8 (a int, b text, c text) PARTITION BY HASH (a); +-- first partition has replica identity "a" +CREATE TABLE testpub_tbl8_0 PARTITION OF testpub_tbl8 FOR VALUES WITH (modulus 2, remainder 0); +ALTER TABLE testpub_tbl8_0 ADD PRIMARY KEY (a); +ALTER TABLE testpub_tbl8_0 REPLICA IDENTITY USING INDEX testpub_tbl8_0_pkey; +-- second partition has replica identity "b" +CREATE TABLE testpub_tbl8_1 PARTITION OF testpub_tbl8 FOR VALUES WITH (modulus 2, remainder 1); +ALTER TABLE testpub_tbl8_1 ADD PRIMARY KEY (b); +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; + +-- ok: column filter covers both "a" and "b" +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_col_filter FOR TABLE testpub_tbl8 (a, b) WITH (publish_via_partition_root = 'true'); +RESET client_min_messages; + +-- ok: the same thing, but try plain ADD TABLE +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, b); +UPDATE testpub_tbl8 SET a = 1; + +-- failure: column filter does not cover replica identity for the second partition +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, c); +UPDATE testpub_tbl8 SET a = 1; +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; + +-- failure: one of the partitions has REPLICA IDENTITY FULL +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY FULL; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, c); +UPDATE testpub_tbl8 SET a = 1; +ALTER PUBLICATION testpub_col_filter DROP TABLE testpub_tbl8; + +-- add table and then try changing replica identity +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, b); + +-- failure: replica identity full can't be used with a column filter +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY FULL; +UPDATE testpub_tbl8 SET a = 1; + +-- failure: replica identity has to be covered by the column filter +ALTER TABLE testpub_tbl8_1 DROP CONSTRAINT testpub_tbl8_1_pkey; +ALTER TABLE testpub_tbl8_1 ADD PRIMARY KEY (c); +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; +UPDATE testpub_tbl8 SET a = 1; + +DROP TABLE testpub_tbl8; + +-- column filter for partitioned tables has to cover replica identities for +-- all child relations +CREATE TABLE testpub_tbl8 (a int, b text, c text) PARTITION BY HASH (a); +ALTER PUBLICATION testpub_col_filter ADD TABLE testpub_tbl8 (a, b); +-- first partition has replica identity "a" +CREATE TABLE testpub_tbl8_0 (a int, b text, c text); +ALTER TABLE testpub_tbl8_0 ADD PRIMARY KEY (a); +ALTER TABLE testpub_tbl8_0 REPLICA IDENTITY USING INDEX testpub_tbl8_0_pkey; +-- second partition has replica identity "b" +CREATE TABLE testpub_tbl8_1 (a int, b text, c text); +ALTER TABLE testpub_tbl8_1 ADD PRIMARY KEY (c); +ALTER TABLE testpub_tbl8_1 REPLICA IDENTITY USING INDEX testpub_tbl8_1_pkey; + +-- ok: attaching first partition works, because (a) is in column filter +ALTER TABLE testpub_tbl8 ATTACH PARTITION testpub_tbl8_0 FOR VALUES WITH (modulus 2, remainder 0); +-- failure: second partition has replica identity (c), which si not in column filter +ALTER TABLE testpub_tbl8 ATTACH PARTITION testpub_tbl8_1 FOR VALUES WITH (modulus 2, remainder 1); +UPDATE testpub_tbl8 SET a = 1; + +-- failure: changing replica identity to FULL for partition fails, because +-- of the column filter on the parent +ALTER TABLE testpub_tbl8_0 REPLICA IDENTITY FULL; +UPDATE testpub_tbl8 SET a = 1; + +DROP TABLE testpub_tbl5, testpub_tbl6, testpub_tbl7, testpub_tbl8, testpub_tbl8_1; +DROP PUBLICATION testpub_table_ins, testpub_fortable, testpub_fortable_insert, testpub_col_filter; +-- ====================================================== + +-- Test combination of column and row filter +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub_both_filters; +RESET client_min_messages; +CREATE TABLE testpub_tbl_both_filters (a int, b int, c int, PRIMARY KEY (a,c)); +ALTER TABLE testpub_tbl_both_filters REPLICA IDENTITY USING INDEX testpub_tbl_both_filters_pkey; +ALTER PUBLICATION testpub_both_filters ADD TABLE testpub_tbl_both_filters (a,c) WHERE (c != 1); +\dRp+ testpub_both_filters +\d+ testpub_tbl_both_filters + +DROP TABLE testpub_tbl_both_filters; +DROP PUBLICATION testpub_both_filters; +-- ====================================================== + +-- More column filter tests for validating column references +CREATE TABLE rf_tbl_abcd_nopk(a int, b int, c int, d int); +CREATE TABLE rf_tbl_abcd_pk(a int, b int, c int, d int, PRIMARY KEY(a,b)); +CREATE TABLE rf_tbl_abcd_part_pk (a int PRIMARY KEY, b int) PARTITION by RANGE (a); +CREATE TABLE rf_tbl_abcd_part_pk_1 (b int, a int PRIMARY KEY); +ALTER TABLE rf_tbl_abcd_part_pk ATTACH PARTITION rf_tbl_abcd_part_pk_1 FOR VALUES FROM (1) TO (10); + +-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing) + +-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK. +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk (a, b); +RESET client_min_messages; +-- ok - (a,b) coverts all PK cols +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a, b, c); +-- ok - (a,b,c) coverts all PK cols +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a); +-- fail - "b" is missing from the column filter +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (b); +-- fail - "a" is missing from the column filter +UPDATE rf_tbl_abcd_pk SET a = 1; + +-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (a); +-- ok - there's no replica identity, so any column filter works +-- note: it fails anyway, just a bit later because UPDATE requires RI +UPDATE rf_tbl_abcd_nopk SET a = 1; + +-- Case 2. REPLICA IDENTITY FULL +ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL; +ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (c); +-- fail - with REPLICA IDENTITY FULL no column filter is allowed +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (a, b, c, d); +-- fail - with REPLICA IDENTITY FULL no column filter is allowed +UPDATE rf_tbl_abcd_nopk SET a = 1; + +-- Case 3. REPLICA IDENTITY NOTHING +ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING; +ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a); +-- ok - REPLICA IDENTITY NOTHING means all column filters are valid +-- it still fails later because without RI we can't replicate updates +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a, b, c, d); +-- ok - REPLICA IDENTITY NOTHING means all column filters are valid +-- it still fails later because without RI we can't replicate updates +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (d); +-- ok - REPLICA IDENTITY NOTHING means all column filters are valid +-- it still fails later because without RI we can't replicate updates +UPDATE rf_tbl_abcd_nopk SET a = 1; + +-- Case 4. REPLICA IDENTITY INDEX +ALTER TABLE rf_tbl_abcd_pk ALTER COLUMN c SET NOT NULL; +CREATE UNIQUE INDEX idx_abcd_pk_c ON rf_tbl_abcd_pk(c); +ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY USING INDEX idx_abcd_pk_c; +ALTER TABLE rf_tbl_abcd_nopk ALTER COLUMN c SET NOT NULL; +CREATE UNIQUE INDEX idx_abcd_nopk_c ON rf_tbl_abcd_nopk(c); +ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY USING INDEX idx_abcd_nopk_c; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (a); +-- fail - column filter "a" does not cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk (c); +-- ok - column filter "c" does cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_pk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (a); +-- fail - column filter "a" does not cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_nopk SET a = 1; +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk (c); +-- ok - column filter "c" does cover the REPLICA IDENTITY INDEX on "c" +UPDATE rf_tbl_abcd_nopk SET a = 1; + +-- Tests for partitioned table + +-- set PUBLISH_VIA_PARTITION_ROOT to false and test row filter for partitioned +-- table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0); +-- fail - cannot use row filter for partitioned table +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk (a); +-- ok - can use row filter for partition +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk_1 (a); +-- ok - "a" is a PK col +UPDATE rf_tbl_abcd_part_pk SET a = 1; +-- set PUBLISH_VIA_PARTITION_ROOT to true and test row filter for partitioned +-- table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1); +-- ok - can use row filter for partitioned table +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk (a); +-- ok - "a" is a PK col +UPDATE rf_tbl_abcd_part_pk SET a = 1; +-- fail - cannot set PUBLISH_VIA_PARTITION_ROOT to false if any row filter is +-- used for partitioned table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0); +-- Now change the root filter to use a column "b" +-- (which is not in the replica identity) +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk_1 (b); +-- ok - we don't have row filter for partitioned table. +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0); +-- fail - "b" is not in REPLICA IDENTITY INDEX +UPDATE rf_tbl_abcd_part_pk SET a = 1; +-- set PUBLISH_VIA_PARTITION_ROOT to true +-- can use row filter for partitioned table +ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1); +-- ok - can use row filter for partitioned table +ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk (b); +-- fail - "b" is not in REPLICA IDENTITY INDEX +UPDATE rf_tbl_abcd_part_pk SET a = 1; + +DROP PUBLICATION testpub6; +DROP TABLE rf_tbl_abcd_pk; +DROP TABLE rf_tbl_abcd_nopk; +DROP TABLE rf_tbl_abcd_part_pk; +-- ====================================================== + -- Test cache invalidation FOR ALL TABLES publication SET client_min_messages = 'ERROR'; CREATE TABLE testpub_tbl4(a int); @@ -622,6 +908,10 @@ ALTER PUBLICATION testpub1_forschema SET ALL TABLES IN SCHEMA non_existent_schem ALTER PUBLICATION testpub1_forschema SET ALL TABLES IN SCHEMA pub_test1, pub_test1; \dRp+ testpub1_forschema +-- Verify that it fails to add a schema with a column specification +ALTER PUBLICATION testpub1_forschema ADD ALL TABLES IN SCHEMA foo (a, b); +ALTER PUBLICATION testpub1_forschema ADD ALL TABLES IN SCHEMA foo, bar (a, b); + -- cleanup pub_test1 schema for invalidation tests ALTER PUBLICATION testpub2_forschema DROP ALL TABLES IN SCHEMA pub_test1; DROP PUBLICATION testpub3_forschema, testpub4_forschema, testpub5_forschema, testpub6_forschema, testpub_fortable; diff --git a/src/test/subscription/t/029_column_list.pl b/src/test/subscription/t/029_column_list.pl new file mode 100644 index 00000000000..ec2c8a789ad --- /dev/null +++ b/src/test/subscription/t/029_column_list.pl @@ -0,0 +1,836 @@ +# Copyright (c) 2022, PostgreSQL Global Development Group + +# Test partial-column publication of tables +use strict; +use warnings; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More tests => 26; + +# setup + +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init(allows_streaming => 'logical'); +$node_subscriber->append_conf('postgresql.conf', + qq(max_logical_replication_workers = 6)); +$node_subscriber->start; + +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; + +sub wait_for_subscription_sync +{ + my ($node) = @_; + + # 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->poll_query_until('postgres', $synced_query) + or die "Timed out while waiting for subscriber to synchronize data"; +} + +# setup tables on both nodes + +# tab1: simple 1:1 replication +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE tab1 (a int PRIMARY KEY, "B" int, c int) +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE tab1 (a int PRIMARY KEY, "B" int, c int) +)); + +# tab2: replication from regular to table with fewer columns +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE tab2 (a int PRIMARY KEY, b varchar, c int); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE tab2 (a int PRIMARY KEY, b varchar) +)); + +# tab3: simple 1:1 replication with weird column names +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE tab3 ("a'" int PRIMARY KEY, "B" varchar, "c'" int) +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE tab3 ("a'" int PRIMARY KEY, "c'" int) +)); + +# test_part: partitioned tables, with partitioning (including multi-level +# partitioning, and fewer columns on the subscriber) +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_part (a int PRIMARY KEY, b text, c timestamptz) PARTITION BY LIST (a); + CREATE TABLE test_part_1_1 PARTITION OF test_part FOR VALUES IN (1,2,3); + CREATE TABLE test_part_2_1 PARTITION OF test_part FOR VALUES IN (4,5,6) PARTITION BY LIST (a); + CREATE TABLE test_part_2_2 PARTITION OF test_part_2_1 FOR VALUES IN (4,5); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_part (a int PRIMARY KEY, b text) PARTITION BY LIST (a); + CREATE TABLE test_part_1_1 PARTITION OF test_part FOR VALUES IN (1,2,3); + CREATE TABLE test_part_2_1 PARTITION OF test_part FOR VALUES IN (4,5,6) PARTITION BY LIST (a); + CREATE TABLE test_part_2_2 PARTITION OF test_part_2_1 FOR VALUES IN (4,5); +)); + +# tab4: table with user-defined enum types +$node_publisher->safe_psql('postgres', qq( + CREATE TYPE test_typ AS ENUM ('blue', 'red'); + CREATE TABLE tab4 (a INT PRIMARY KEY, b test_typ, c int, d text); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TYPE test_typ AS ENUM ('blue', 'red'); + CREATE TABLE tab4 (a INT PRIMARY KEY, b test_typ, d text); +)); + + +# TEST: create publication and subscription for some of the tables with +# column lists +$node_publisher->safe_psql('postgres', qq( + CREATE PUBLICATION pub1 + FOR TABLE tab1 (a, "B"), tab3 ("a'", "c'"), test_part (a, b), tab4 (a, b, d) + WITH (publish_via_partition_root = 'true'); +)); + +# check that we got the right prattrs values for the publication in the +# pg_publication_rel catalog (order by relname, to get stable ordering) +my $result = $node_publisher->safe_psql('postgres', qq( + SELECT relname, prattrs + FROM pg_publication_rel pb JOIN pg_class pc ON(pb.prrelid = pc.oid) + ORDER BY relname +)); + +is($result, qq(tab1|1 2 +tab3|1 3 +tab4|1 2 4 +test_part|1 2), 'publication relation updated'); + +# create subscription for the publication, wait for sync to complete +$node_subscriber->safe_psql('postgres', qq( + CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1 +)); + +$node_publisher->wait_for_catchup('sub1'); + +# TEST: insert data into the tables, and see we got replication of just +# the filtered columns +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab1 VALUES (1, 2, 3); + INSERT INTO tab1 VALUES (4, 5, 6); +)); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab3 VALUES (1, 2, 3); + INSERT INTO tab3 VALUES (4, 5, 6); +)); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab4 VALUES (1, 'red', 3, 'oh my'); + INSERT INTO tab4 VALUES (2, 'blue', 4, 'hello'); +)); + +# replication of partitioned table +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_part VALUES (1, 'abc', '2021-07-04 12:00:00'); + INSERT INTO test_part VALUES (2, 'bcd', '2021-07-03 11:12:13'); + INSERT INTO test_part VALUES (4, 'abc', '2021-07-04 12:00:00'); + INSERT INTO test_part VALUES (5, 'bcd', '2021-07-03 11:12:13'); +)); + +# wait for catchup before checking the subscriber +$node_publisher->wait_for_catchup('sub1'); + +# tab1: only (a,b) is replicated +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM tab1"); +is($result, qq(1|2| +4|5|), 'insert on column tab1.c is not replicated'); + +# tab3: only (a,c) is replicated +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM tab3"); +is($result, qq(1|3 +4|6), 'insert on column tab3.b is not replicated'); + +# tab4: only (a,b,d) is replicated +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM tab4"); +is($result, qq(1|red|oh my +2|blue|hello), 'insert on column tab4.c is not replicated'); + +# test_part: (a,b) is replicated +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM test_part"); +is($result, qq(1|abc +2|bcd +4|abc +5|bcd), 'insert on column test_part.c columns is not replicated'); + +# TEST: do some updated on some of the tables, both on columns included +# in the column list and other + +# tab1: update of replicated column +$node_publisher->safe_psql('postgres', + qq(UPDATE tab1 SET "B" = 2 * "B" where a = 1)); + +# tab1: update of non-replicated column +$node_publisher->safe_psql('postgres', + qq(UPDATE tab1 SET c = 2*c where a = 4)); + +# tab3: update of non-replicated +$node_publisher->safe_psql('postgres', + qq(UPDATE tab3 SET "B" = "B" || ' updated' where "a'" = 4)); + +# tab3: update of replicated column +$node_publisher->safe_psql('postgres', + qq(UPDATE tab3 SET "c'" = 2 * "c'" where "a'" = 1)); + +# tab4 +$node_publisher->safe_psql('postgres', + qq(UPDATE tab4 SET b = 'blue', c = c * 2, d = d || ' updated' where a = 1)); + +# tab4 +$node_publisher->safe_psql('postgres', + qq(UPDATE tab4 SET b = 'red', c = c * 2, d = d || ' updated' where a = 2)); + +$node_publisher->wait_for_catchup('sub1'); + +$result = $node_subscriber->safe_psql('postgres', + qq(SELECT * FROM tab1 ORDER BY a)); +is($result, +qq(1|4| +4|5|), 'only update on column tab1.b is replicated'); + +$result = $node_subscriber->safe_psql('postgres', + qq(SELECT * FROM tab3 ORDER BY "a'")); +is($result, +qq(1|6 +4|6), 'only update on column tab3.c is replicated'); + +$result = $node_subscriber->safe_psql('postgres', + qq(SELECT * FROM tab4 ORDER BY a)); + +is($result, qq(1|blue|oh my updated +2|red|hello updated), 'update on column tab4.c is not replicated'); + + +# TEST: add table with a column list, insert data, replicate + +# insert some data before adding it to the publication +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab2 VALUES (1, 'abc', 3); +)); + +$node_publisher->safe_psql('postgres', + "ALTER PUBLICATION pub1 ADD TABLE tab2 (a, b)"); + +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION"); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab2 VALUES (2, 'def', 6); +)); + +$node_publisher->wait_for_catchup('sub1'); + +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM tab2 ORDER BY a"); +is($result, qq(1|abc +2|def), 'insert on column tab2.c is not replicated'); + +$node_publisher->safe_psql('postgres', qq( + UPDATE tab2 SET c = 5 where a = 1; + UPDATE tab2 SET b = 'xyz' where a = 2; +)); + +$node_publisher->wait_for_catchup('sub1'); + +$result = $node_subscriber->safe_psql('postgres', + "SELECT * FROM tab2 ORDER BY a"); +is($result, qq(1|abc +2|xyz), 'update on column tab2.c is not replicated'); + + +# TEST: add a table to two publications with different column lists, and +# create a single subscription replicating both publications +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE tab5 (a int PRIMARY KEY, b int, c int, d int); + CREATE PUBLICATION pub2 FOR TABLE tab5 (a, b); + CREATE PUBLICATION pub3 FOR TABLE tab5 (a, d); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE tab5 (a int PRIMARY KEY, b int, d int); +)); + +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub2, pub3 +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->wait_for_catchup('sub1'); + +# TEST: insert data and make sure all the columns (union of the columns lists) +# were replicated +$node_publisher->safe_psql('postgres', "INSERT INTO tab5 VALUES (1, 11, 111, 1111)"); +$node_publisher->safe_psql('postgres', "INSERT INTO tab5 VALUES (2, 22, 222, 2222)"); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM tab5 ORDER BY a"), + qq(1|11|1111 +2|22|2222), + 'overlapping publications with overlapping column lists'); + +# and finally, set the column filter to ALL for one of the publications, +# which means replicating all columns (removing the column filter), but +# first add the missing column to the table on subscriber +$node_publisher->safe_psql('postgres', qq( + ALTER PUBLICATION pub3 ALTER TABLE tab5 SET COLUMNS ALL; +)); + +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION; + ALTER TABLE tab5 ADD COLUMN c INT; +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', "INSERT INTO tab5 VALUES (3, 33, 333, 3333)"); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM tab5 ORDER BY a"), + qq(1|11|1111| +2|22|2222| +3|33|3333|333), + 'overlapping publications with overlapping column lists'); + +# TEST: create a table with a column filter, then change the replica +# identity by replacing a primary key (but use a different column in +# the column filter) +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE tab6 (a int PRIMARY KEY, b int, c int, d int); + CREATE PUBLICATION pub4 FOR TABLE tab6 (a, b); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE tab6 (a int PRIMARY KEY, b int, c int, d int); +)); + +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub4 +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab6 VALUES (1, 22, 333, 4444); + UPDATE tab6 SET b = b * 2, c = c * 3, d = d * 4 WHERE a = 1; +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM tab6 ORDER BY a"), + "1|44||", 'replication with the original primary key'); + +# now redefine the constraint - move the primary key to a different column +# (which is still covered by the column list, though) + +$node_publisher->safe_psql('postgres', qq( + ALTER TABLE tab6 DROP CONSTRAINT tab6_pkey; + ALTER TABLE tab6 ADD PRIMARY KEY (b); +)); + +# we need to do the same thing on the subscriber +# XXX What would happen if this happens before the publisher ALTER? Or +# interleaved, somehow? But that seems unrelated to column lists. +$node_subscriber->safe_psql('postgres', qq( + ALTER TABLE tab6 DROP CONSTRAINT tab6_pkey; + ALTER TABLE tab6 ADD PRIMARY KEY (b); +)); + +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab6 VALUES (2, 55, 666, 8888); + UPDATE tab6 SET b = b * 2, c = c * 3, d = d * 4 WHERE a = 2; +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM tab6 ORDER BY a"), + qq(1|44|| +2|110||), + 'replication with the modified primary key'); + + +# TEST: create a table with a column filter, then change the replica +# identity by replacing a primary key with a key on multiple columns +# (all of them covered by the column filter) +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE tab7 (a int PRIMARY KEY, b int, c int, d int); + CREATE PUBLICATION pub5 FOR TABLE tab7 (a, b); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE tab7 (a int PRIMARY KEY, b int, c int, d int); +)); + +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub5 +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab7 VALUES (1, 22, 333, 4444); + UPDATE tab7 SET b = b * 2, c = c * 3, d = d * 4 WHERE a = 1; +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM tab7 ORDER BY a"), + "1|44||", 'replication with the original primary key'); + +# now redefine the constraint - move the primary key to a different column +# (which is not covered by the column list) +$node_publisher->safe_psql('postgres', qq( + ALTER TABLE tab7 DROP CONSTRAINT tab7_pkey; + ALTER TABLE tab7 ADD PRIMARY KEY (a, b); +)); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO tab7 VALUES (2, 55, 666, 8888); + UPDATE tab7 SET b = b * 2, c = c * 3, d = d * 4 WHERE a = 2; +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM tab7 ORDER BY a"), + qq(1|44|| +2|110||), + 'replication with the modified primary key'); + +# now switch the primary key again to another columns not covered by the +# column filter, but also generate writes between the drop and creation +# of the new constraint + +$node_publisher->safe_psql('postgres', qq( + ALTER TABLE tab7 DROP CONSTRAINT tab7_pkey; + INSERT INTO tab7 VALUES (3, 33, 999, 7777); + -- update/delete is not allowed for tables without RI + ALTER TABLE tab7 ADD PRIMARY KEY (b, a); + UPDATE tab7 SET b = b * 2, c = c * 3, d = d * 4 WHERE a = 3; + DELETE FROM tab7 WHERE a = 1; +)); + +$node_publisher->safe_psql('postgres', qq( +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM tab7 ORDER BY a"), + qq(2|110|| +3|66||), + 'replication with the modified primary key'); + + +# TEST: partitioned tables (with publish_via_partition_root = false) +# and replica identity. The (leaf) partitions may have different RI, so +# we need to check the partition RI (with respect to the column filter) +# while attaching the partition. + +# First, let's create a partitioned table with two partitions, each with +# a different RI, but a column filter not covering all those RI. + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_part_a (a int, b int, c int) PARTITION BY LIST (a); + + CREATE TABLE test_part_a_1 PARTITION OF test_part_a FOR VALUES IN (1); + ALTER TABLE test_part_a_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_a_1 REPLICA IDENTITY USING INDEX test_part_a_1_pkey; + + CREATE TABLE test_part_a_2 PARTITION OF test_part_a FOR VALUES IN (2); + ALTER TABLE test_part_a_2 ADD PRIMARY KEY (b); + ALTER TABLE test_part_a_2 REPLICA IDENTITY USING INDEX test_part_a_2_pkey; +)); + +# do the same thing on the subscriber (with the opposite column order) +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_part_a (b int, a int) PARTITION BY LIST (a); + + CREATE TABLE test_part_a_1 PARTITION OF test_part_a FOR VALUES IN (1); + ALTER TABLE test_part_a_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_a_1 REPLICA IDENTITY USING INDEX test_part_a_1_pkey; + + CREATE TABLE test_part_a_2 PARTITION OF test_part_a FOR VALUES IN (2); + ALTER TABLE test_part_a_2 ADD PRIMARY KEY (b); + ALTER TABLE test_part_a_2 REPLICA IDENTITY USING INDEX test_part_a_2_pkey; +)); + +# create a publication replicating just the column "a", which is not enough +# for the second partition +$node_publisher->safe_psql('postgres', qq( + CREATE PUBLICATION pub6 FOR TABLE test_part_a (b, a) WITH (publish_via_partition_root = true); + ALTER PUBLICATION pub6 ADD TABLE test_part_a_1 (a); + ALTER PUBLICATION pub6 ADD TABLE test_part_a_2 (b); +)); + +# add the publication to our subscription, wait for sync to complete +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub6 +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_part_a VALUES (1, 3); + INSERT INTO test_part_a VALUES (2, 4); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT a, b FROM test_part_a ORDER BY a, b"), + qq(1|3 +2|4), + 'partitions with different replica identities not replicated correctly'); + +# This time start with a column filter covering RI for all partitions, but +# then update the column filter to not cover column "b" (needed by the +# second partition) + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_part_b (a int, b int) PARTITION BY LIST (a); + + CREATE TABLE test_part_b_1 PARTITION OF test_part_b FOR VALUES IN (1,3); + ALTER TABLE test_part_b_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_b_1 REPLICA IDENTITY USING INDEX test_part_b_1_pkey; + + CREATE TABLE test_part_b_2 PARTITION OF test_part_b FOR VALUES IN (2,4); + ALTER TABLE test_part_b_2 ADD PRIMARY KEY (b); + ALTER TABLE test_part_b_2 REPLICA IDENTITY USING INDEX test_part_b_2_pkey; +)); + +# do the same thing on the subscriber +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_part_b (a int, b int) PARTITION BY LIST (a); + + CREATE TABLE test_part_b_1 PARTITION OF test_part_b FOR VALUES IN (1,3); + ALTER TABLE test_part_b_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_b_1 REPLICA IDENTITY USING INDEX test_part_b_1_pkey; + + CREATE TABLE test_part_b_2 PARTITION OF test_part_b FOR VALUES IN (2,4); + ALTER TABLE test_part_b_2 ADD PRIMARY KEY (b); + ALTER TABLE test_part_b_2 REPLICA IDENTITY USING INDEX test_part_b_2_pkey; +)); + +# create a publication replicating both columns, which is sufficient for +# both partitions +$node_publisher->safe_psql('postgres', qq( + CREATE PUBLICATION pub7 FOR TABLE test_part_b (a, b) WITH (publish_via_partition_root = true); +)); + +# add the publication to our subscription, wait for sync to complete +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub7 +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_part_b VALUES (1, 1); + INSERT INTO test_part_b VALUES (2, 2); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_part_b ORDER BY a, b"), + qq(1|1 +2|2), + 'partitions with different replica identities not replicated correctly'); + + +# TEST: This time start with a column filter covering RI for all partitions, +# but then update RI for one of the partitions to not be covered by the +# column filter anymore. + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_part_c (a int, b int, c int) PARTITION BY LIST (a); + + CREATE TABLE test_part_c_1 PARTITION OF test_part_c FOR VALUES IN (1,3); + ALTER TABLE test_part_c_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_c_1 REPLICA IDENTITY USING INDEX test_part_c_1_pkey; + + CREATE TABLE test_part_c_2 PARTITION OF test_part_c FOR VALUES IN (2,4); + ALTER TABLE test_part_c_2 ADD PRIMARY KEY (b); + ALTER TABLE test_part_c_2 REPLICA IDENTITY USING INDEX test_part_c_2_pkey; +)); + +# do the same thing on the subscriber +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_part_c (a int, b int, c int) PARTITION BY LIST (a); + + CREATE TABLE test_part_c_1 PARTITION OF test_part_c FOR VALUES IN (1,3); + ALTER TABLE test_part_c_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_c_1 REPLICA IDENTITY USING INDEX test_part_c_1_pkey; + + CREATE TABLE test_part_c_2 PARTITION OF test_part_c FOR VALUES IN (2,4); + ALTER TABLE test_part_c_2 ADD PRIMARY KEY (b); + ALTER TABLE test_part_c_2 REPLICA IDENTITY USING INDEX test_part_c_2_pkey; +)); + +# create a publication replicating data through partition root, with a column +# filter on the root, and then add the partitions one by one with separate +# column filters (but those are not applied) +$node_publisher->safe_psql('postgres', qq( + CREATE PUBLICATION pub8 FOR TABLE test_part_c WITH (publish_via_partition_root = false); + ALTER PUBLICATION pub8 ADD TABLE test_part_c_1 (a,c); + ALTER PUBLICATION pub8 ADD TABLE test_part_c_2 (a,b); +)); + +# add the publication to our subscription, wait for sync to complete +$node_subscriber->safe_psql('postgres', qq( + DROP SUBSCRIPTION sub1; + CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub8; +)); + +$node_publisher->wait_for_catchup('sub1'); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_part_c VALUES (1, 3, 5); + INSERT INTO test_part_c VALUES (2, 4, 6); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_part_c ORDER BY a, b"), + qq(1||5 +2|4|), + 'partitions with different replica identities not replicated correctly'); + + +# create a publication not replicating data through partition root, without +# a column filter on the root, and then add the partitions one by one with +# separate column filters +$node_publisher->safe_psql('postgres', qq( + DROP PUBLICATION pub8; + CREATE PUBLICATION pub8 FOR TABLE test_part_c WITH (publish_via_partition_root = false); + ALTER PUBLICATION pub8 ADD TABLE test_part_c_1 (a); + ALTER PUBLICATION pub8 ADD TABLE test_part_c_2 (a,b); +)); + +# add the publication to our subscription, wait for sync to complete +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION; + TRUNCATE test_part_c; +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + TRUNCATE test_part_c; + INSERT INTO test_part_c VALUES (1, 3, 5); + INSERT INTO test_part_c VALUES (2, 4, 6); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_part_c ORDER BY a, b"), + qq(1|| +2|4|), + 'partitions with different replica identities not replicated correctly'); + + +# TEST: Start with a single partition, with RI compatible with the column +# filter, and then attach a partition with incompatible RI. + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_part_d (a int, b int) PARTITION BY LIST (a); + + CREATE TABLE test_part_d_1 PARTITION OF test_part_d FOR VALUES IN (1,3); + ALTER TABLE test_part_d_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_d_1 REPLICA IDENTITY USING INDEX test_part_d_1_pkey; +)); + +# do the same thing on the subscriber (in fact, create both partitions right +# away, no need to delay that) +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_part_d (a int, b int) PARTITION BY LIST (a); + + CREATE TABLE test_part_d_1 PARTITION OF test_part_d FOR VALUES IN (1,3); + ALTER TABLE test_part_d_1 ADD PRIMARY KEY (a); + ALTER TABLE test_part_d_1 REPLICA IDENTITY USING INDEX test_part_d_1_pkey; + + CREATE TABLE test_part_d_2 PARTITION OF test_part_d FOR VALUES IN (2,4); + ALTER TABLE test_part_d_2 ADD PRIMARY KEY (a); + ALTER TABLE test_part_d_2 REPLICA IDENTITY USING INDEX test_part_d_2_pkey; +)); + +# create a publication replicating both columns, which is sufficient for +# both partitions +$node_publisher->safe_psql('postgres', qq( + CREATE PUBLICATION pub9 FOR TABLE test_part_d (a) WITH (publish_via_partition_root = true); +)); + +# add the publication to our subscription, wait for sync to complete +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub9 +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_part_d VALUES (1, 1); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_part_d ORDER BY a, b"), + qq(1|), + 'partitions with different replica identities not replicated correctly'); + +# TEST: With a table included in multiple publications, we should use a +# union of the column filters. So with column filters (a,b) and (a,c) we +# should replicate (a,b,c). + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_mix_1 (a int PRIMARY KEY, b int, c int); + CREATE PUBLICATION pub_mix_1 FOR TABLE test_mix_1 (a, b); + CREATE PUBLICATION pub_mix_2 FOR TABLE test_mix_1 (a, c); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_mix_1 (a int PRIMARY KEY, b int, c int); + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub_mix_1, pub_mix_2; +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_mix_1 VALUES (1, 2, 3); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_mix_1"), + qq(1|2|3), + 'a mix of publications should use a union of column filter'); + + +# TEST: With a table included in multiple publications, we should use a +# union of the column filters. If any of the publications is FOR ALL +# TABLES, we should replicate all columns. + +# drop unnecessary tables, so as not to interfere with the FOR ALL TABLES +$node_publisher->safe_psql('postgres', qq( + DROP TABLE tab1, tab2, tab3, tab4, tab5, tab6, tab7, test_mix_1, + test_part, test_part_a, test_part_b, test_part_c, test_part_d; +)); + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_mix_2 (a int PRIMARY KEY, b int, c int); + CREATE PUBLICATION pub_mix_3 FOR TABLE test_mix_2 (a, b); + CREATE PUBLICATION pub_mix_4 FOR ALL TABLES; +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_mix_2 (a int PRIMARY KEY, b int, c int); + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub_mix_3, pub_mix_4; + ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION; +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_mix_2 VALUES (1, 2, 3); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_mix_2"), + qq(1|2|3), + 'a mix of publications should use a union of column filter'); + + +# TEST: With a table included in multiple publications, we should use a +# union of the column filters. If any of the publications is FOR ALL +# TABLES IN SCHEMA, we should replicate all columns. + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_mix_3 (a int PRIMARY KEY, b int, c int); + CREATE PUBLICATION pub_mix_5 FOR TABLE test_mix_3 (a, b); + CREATE PUBLICATION pub_mix_6 FOR ALL TABLES IN SCHEMA public; +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_mix_3 (a int PRIMARY KEY, b int, c int); + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub_mix_5, pub_mix_6; +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_mix_3 VALUES (1, 2, 3); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_mix_3"), + qq(1|2|3), + 'a mix of publications should use a union of column filter'); + + +# TEST: Check handling of publish_via_partition_root - if a partition is +# published through partition root, we should only apply the column filter +# defined for the whole table (not the partitions) - both during the initial +# sync and when replicating changes. This is what we do for row filters. + +$node_publisher->safe_psql('postgres', qq( + CREATE TABLE test_root (a int PRIMARY KEY, b int, c int) PARTITION BY RANGE (a); + CREATE TABLE test_root_1 PARTITION OF test_root FOR VALUES FROM (1) TO (10); + CREATE TABLE test_root_2 PARTITION OF test_root FOR VALUES FROM (10) TO (20); + + CREATE PUBLICATION pub_root_true FOR TABLE test_root (a) WITH (publish_via_partition_root = true); +)); + +$node_subscriber->safe_psql('postgres', qq( + CREATE TABLE test_root (a int PRIMARY KEY, b int, c int) PARTITION BY RANGE (a); + CREATE TABLE test_root_1 PARTITION OF test_root FOR VALUES FROM (1) TO (10); + CREATE TABLE test_root_2 PARTITION OF test_root FOR VALUES FROM (10) TO (20); +)); + +$node_subscriber->safe_psql('postgres', qq( + ALTER SUBSCRIPTION sub1 SET PUBLICATION pub_root_true; +)); + +wait_for_subscription_sync($node_subscriber); + +$node_publisher->safe_psql('postgres', qq( + INSERT INTO test_root VALUES (1, 2, 3); + INSERT INTO test_root VALUES (10, 20, 30); +)); + +$node_publisher->wait_for_catchup('sub1'); + +is($node_subscriber->safe_psql('postgres',"SELECT * FROM test_root ORDER BY a, b, c"), + qq(1|| +10||), + 'publication via partition root applies column filter'); + +$node_subscriber->stop('fast'); +$node_publisher->stop('fast'); + +done_testing(); -- 2.34.1