From 8ceba9962853637637f74b173253a28c6e9db658 Mon Sep 17 00:00:00 2001 From: Onder Kalaci Date: Tue, 8 Dec 2020 16:44:50 +0300 Subject: [PATCH 5/7] Subject: [PATCH 5/8] Row filtering for logical replication When you define or modify a publication you optionally can filter rows to be published using a WHERE condition. This condition is any expression that evaluates to boolean. Only those rows that satisfy the WHERE condition will be sent to subscribers. --- doc/src/sgml/catalogs.sgml | 9 + doc/src/sgml/ref/alter_publication.sgml | 11 +- doc/src/sgml/ref/create_publication.sgml | 21 +- src/backend/catalog/pg_publication.c | 207 +++++++++++++++++--- src/backend/commands/publicationcmds.c | 95 ++++++--- src/backend/parser/gram.y | 25 ++- src/backend/parser/parse_agg.c | 10 + src/backend/parser/parse_expr.c | 13 ++ src/backend/parser/parse_func.c | 3 + src/backend/replication/logical/tablesync.c | 122 ++++++++++-- src/backend/replication/logical/worker.c | 6 +- src/backend/replication/pgoutput/pgoutput.c | 110 ++++++++++- src/include/catalog/pg_publication.h | 9 +- src/include/catalog/pg_publication_rel.h | 5 + src/include/nodes/nodes.h | 1 + src/include/nodes/parsenodes.h | 11 +- src/include/parser/parse_node.h | 1 + src/include/replication/logicalrelation.h | 2 + src/test/regress/expected/publication.out | 29 +++ src/test/regress/sql/publication.sql | 21 ++ 20 files changed, 626 insertions(+), 85 deletions(-) diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 79069ddfab..b48f97d82e 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -5609,6 +5609,15 @@ SCRAM-SHA-256$<iteration count>:&l The expression tree to be added to the WITH CHECK qualifications for queries that attempt to add rows to the table + + + prqual + pg_node_tree + + Expression tree (in the form of a + nodeToString() representation) for the relation's + qualifying condition + diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml index c2946dfe0f..ae4da00711 100644 --- a/doc/src/sgml/ref/alter_publication.sgml +++ b/doc/src/sgml/ref/alter_publication.sgml @@ -21,8 +21,8 @@ PostgreSQL documentation -ALTER PUBLICATION name ADD TABLE [ ONLY ] table_name [ * ] [, ...] -ALTER PUBLICATION name SET TABLE [ ONLY ] table_name [ * ] [, ...] +ALTER PUBLICATION name ADD TABLE [ ONLY ] table_name [ * ] [ WHERE ( expression ) ] [, ...] +ALTER PUBLICATION name SET TABLE [ ONLY ] table_name [ * ] [ WHERE ( expression ) ] [, ...] ALTER PUBLICATION name DROP TABLE [ ONLY ] table_name [ * ] [, ...] ALTER PUBLICATION name SET ( publication_parameter [= value] [, ... ] ) ALTER PUBLICATION name OWNER TO { new_owner | CURRENT_ROLE | CURRENT_USER | SESSION_USER } @@ -91,7 +91,12 @@ ALTER PUBLICATION name RENAME TO ONLY is not specified, the table and all its descendant tables (if any) are affected. Optionally, * can be specified after the table - name to explicitly indicate that descendant tables are included. + name to explicitly indicate that descendant tables are included. If the + optional WHERE clause is specified, rows that do not + satisfy the expression will + not be published. Note that parentheses are required around the + expression. The expression + is executed with the role used for the replication connection. diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml index ff82fbca55..4b015b37f3 100644 --- a/doc/src/sgml/ref/create_publication.sgml +++ b/doc/src/sgml/ref/create_publication.sgml @@ -22,7 +22,7 @@ PostgreSQL documentation CREATE PUBLICATION name - [ FOR TABLE [ ONLY ] table_name [ * ] [, ...] + [ FOR TABLE [ ONLY ] table_name [ * ] [ WHERE ( expression ) ] [, ...] | FOR ALL TABLES ] [ WITH ( publication_parameter [= value] [, ... ] ) ] @@ -182,6 +182,13 @@ CREATE PUBLICATION name disallowed on those tables. + + Columns used in the WHERE clause must be part of the + primary key or be covered by REPLICA IDENTITY otherwise + UPDATE and DELETE operations will not + be replicated. + + For an INSERT ... ON CONFLICT command, the publication will publish the operation that actually results from the command. So depending @@ -197,6 +204,11 @@ CREATE PUBLICATION name DDL operations are not published. + + + The WHERE clause expression is executed with the role used + for the replication connection. + @@ -209,6 +221,13 @@ CREATE PUBLICATION mypublication FOR TABLE users, departments; + + Create a publication that publishes all changes from active departments: + +CREATE PUBLICATION active_departments FOR TABLE departments WHERE (active IS TRUE); + + + Create a publication that publishes all changes in all tables: diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index 09946be788..3ef427e16c 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -33,6 +33,11 @@ #include "catalog/pg_type.h" #include "funcapi.h" #include "miscadmin.h" + +#include "parser/parse_clause.h" +#include "parser/parse_collate.h" +#include "parser/parse_relation.h" + #include "utils/array.h" #include "utils/builtins.h" #include "utils/catcache.h" @@ -42,6 +47,9 @@ #include "utils/rel.h" #include "utils/syscache.h" +static List * PublicationPartitionedRelationGetRelations(Oid relationId, + PublicationPartOpt pub_partopt); + /* * Check if relation can be in given publication and throws appropriate * error if not. @@ -141,18 +149,22 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS) * Insert new publication / relation mapping. */ ObjectAddress -publication_add_relation(Oid pubid, Relation targetrel, +publication_add_relation(Oid pubid, PublicationRelationQual *targetrel, bool if_not_exists) { Relation rel; HeapTuple tup; Datum values[Natts_pg_publication_rel]; bool nulls[Natts_pg_publication_rel]; - Oid relid = RelationGetRelid(targetrel); + Oid relid = RelationGetRelid(targetrel->relation); Oid prrelid; Publication *pub = GetPublication(pubid); ObjectAddress myself, referenced; + ParseState *pstate; + RangeTblEntry *rte; + Node *whereclause; + ParseNamespaceItem *pitem; rel = table_open(PublicationRelRelationId, RowExclusiveLock); @@ -172,10 +184,41 @@ publication_add_relation(Oid pubid, Relation targetrel, ereport(ERROR, (errcode(ERRCODE_DUPLICATE_OBJECT), errmsg("relation \"%s\" is already member of publication \"%s\"", - RelationGetRelationName(targetrel), pub->name))); + RelationGetRelationName(targetrel->relation), pub->name))); + } + + check_publication_add_relation(targetrel->relation); + + if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE && !pub->pubviaroot && + targetrel->whereClause) + { + table_close(rel, RowExclusiveLock); + + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("cannot create publication \"%s\" with WHERE clause on partitioned table " + "\"%s\" without publish_via_partition_root is true", pub->name, + RelationGetRelationName(targetrel->relation)))); } - check_publication_add_relation(targetrel); + /* Set up a pstate to parse with */ + pstate = make_parsestate(NULL); + pstate->p_sourcetext = nodeToString(targetrel->whereClause); + + pitem = addRangeTableEntryForRelation(pstate, targetrel->relation, + AccessShareLock, + NULL, false, false); + rte = pitem->p_rte; + + addNSItemToQuery(pstate, pitem, false, true, true); + + whereclause = transformWhereClause(pstate, + copyObject(targetrel->whereClause), + EXPR_KIND_PUBLICATION_WHERE, + "PUBLICATION"); + + /* Fix up collation information */ + assign_expr_collations(pstate, whereclause); /* Form a tuple. */ memset(values, 0, sizeof(values)); @@ -189,6 +232,12 @@ publication_add_relation(Oid pubid, Relation targetrel, values[Anum_pg_publication_rel_prrelid - 1] = ObjectIdGetDatum(relid); + /* Add qualifications, if available */ + if (whereclause) + values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(whereclause)); + else + nulls[Anum_pg_publication_rel_prqual - 1] = true; + tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); /* Insert tuple into catalog. */ @@ -205,11 +254,17 @@ publication_add_relation(Oid pubid, Relation targetrel, ObjectAddressSet(referenced, RelationRelationId, relid); recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO); + /* Add dependency on the objects mentioned in the qualifications */ + if (whereclause) + recordDependencyOnExpr(&myself, whereclause, pstate->p_rtable, DEPENDENCY_NORMAL); + + free_parsestate(pstate); + /* Close the table. */ table_close(rel, RowExclusiveLock); /* Invalidate relcache so that publication info is rebuilt. */ - CacheInvalidateRelcache(targetrel); + CacheInvalidateRelcache(targetrel->relation); return myself; } @@ -271,31 +326,136 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); - if (get_rel_relkind(pubrel->prrelid) == RELKIND_PARTITIONED_TABLE && - pub_partopt != PUBLICATION_PART_ROOT) + if (get_rel_relkind(pubrel->prrelid) != RELKIND_PARTITIONED_TABLE) + result = lappend_oid(result, pubrel->prrelid); + else { - List *all_parts = find_all_inheritors(pubrel->prrelid, NoLock, - NULL); + List *all_parts = PublicationPartitionedRelationGetRelations(pubrel->prrelid, pub_partopt); - if (pub_partopt == PUBLICATION_PART_ALL) - result = list_concat(result, all_parts); - else if (pub_partopt == PUBLICATION_PART_LEAF) - { - ListCell *lc; + result = list_concat(result, all_parts); + } + } - foreach(lc, all_parts) - { - Oid partOid = lfirst_oid(lc); + systable_endscan(scan); + table_close(pubrelsrel, AccessShareLock); + + return result; +} + + +/* + * For the input partitionedRelationId and pub_partopt, return list of relations + * that should be used for the publication. + * + */ +static List * +PublicationPartitionedRelationGetRelations(Oid partitionedRelationId, + PublicationPartOpt pub_partopt) +{ + AssertArg(get_rel_relkind(partitionedRelationId) == RELKIND_PARTITIONED_TABLE); + + List *result = NIL; + List *all_parts = NIL; + if (pub_partopt == PUBLICATION_PART_ROOT) + return list_make1_oid(partitionedRelationId); - if (get_rel_relkind(partOid) != RELKIND_PARTITIONED_TABLE) - result = lappend_oid(result, partOid); - } + all_parts = find_all_inheritors(partitionedRelationId, NoLock, NULL); + if (pub_partopt == PUBLICATION_PART_ALL) + result = list_concat(result, all_parts); + else if (pub_partopt == PUBLICATION_PART_LEAF) + { + ListCell *lc; + + foreach(lc, all_parts) + { + Oid partOid = lfirst_oid(lc); + + if (get_rel_relkind(partOid) != RELKIND_PARTITIONED_TABLE) + { + result = lappend_oid(result, partOid); } - else - Assert(false); + } + } + + return result; +} + + +/* + * Gets list of PublicationRelationQuals for a publication. + * + * This should only be used for normal publications, the FOR ALL TABLES + * the WHERE clause cannot be used, hence this function should not be + * called. + */ +List * +GetPublicationRelationQuals(Oid pubid, PublicationPartOpt pub_partopt) +{ + List *result; + Relation pubrelsrel; + ScanKeyData scankey; + SysScanDesc scan; + HeapTuple tup; + + /* Find all publications associated with the relation. */ + pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock); + + ScanKeyInit(&scankey, + Anum_pg_publication_rel_prpubid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(pubid)); + + scan = systable_beginscan(pubrelsrel, PublicationRelPrrelidPrpubidIndexId, + true, NULL, 1, &scankey); + + result = NIL; + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + Form_pg_publication_rel pubrel; + Datum value_datum; + char *qual_value; + Node *qual_expr; + bool isnull; + + pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + value_datum = heap_getattr(tup, Anum_pg_publication_rel_prqual, RelationGetDescr(pubrelsrel), &isnull); + if (!isnull) + { + qual_value = TextDatumGetCString(value_datum); + qual_expr = (Node *) stringToNode(qual_value); } else - result = lappend_oid(result, pubrel->prrelid); + qual_expr = NULL; + + pubrel = (Form_pg_publication_rel) GETSTRUCT(tup); + + if (get_rel_relkind(pubrel->prrelid) != RELKIND_PARTITIONED_TABLE) + { + PublicationRelationQual *relqual = palloc(sizeof(PublicationRelationQual)); + relqual->relation = table_open(pubrel->prrelid, ShareUpdateExclusiveLock); + relqual->whereClause = copyObject(qual_expr); + + result = lappend(result, relqual); + } + else + { + List *all_parts = + PublicationPartitionedRelationGetRelations(pubrel->prrelid, pub_partopt); + ListCell *lc; + + foreach(lc, all_parts) + { + Oid partOid = lfirst_oid(lc); + + PublicationRelationQual *relqual = palloc(sizeof(PublicationRelationQual)); + relqual->relation = table_open(partOid, NoLock); + + /* for all partitions, use the same qual */ + relqual->whereClause = copyObject(qual_expr); + result = lappend(result, relqual); + } + } } systable_endscan(scan); @@ -304,6 +464,7 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) return result; } + /* * Gets list of publication oids for publications marked as FOR ALL TABLES. */ diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index eabbc7473b..ffc1d14ec7 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -372,6 +372,28 @@ AlterPublicationTables(AlterPublicationStmt *stmt, Relation rel, Assert(list_length(stmt->tables) > 0); + /* + * ALTER PUBLICATION ... DROP TABLE cannot contain a WHERE clause. Use + * publication_table_list node (that accepts a WHERE clause) but forbid + * the WHERE clause in it. The use of relation_expr_list node just for + * the DROP TABLE part does not worth the trouble. + */ + if (stmt->tableAction == DEFELEM_DROP) + { + ListCell *lc; + + foreach(lc, stmt->tables) + { + PublicationTable *t = lfirst(lc); + + if (t->whereClause) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot use a WHERE clause for removing table from publication \"%s\"", + NameStr(pubform->pubname)))); + } + } + rels = OpenTableList(stmt->tables); if (stmt->tableAction == DEFELEM_ADD) @@ -380,48 +402,59 @@ AlterPublicationTables(AlterPublicationStmt *stmt, Relation rel, PublicationDropTables(pubid, rels, false); else /* DEFELEM_SET */ { - List *oldrelids = GetPublicationRelations(pubid, - PUBLICATION_PART_ROOT); + List *oldrelquals = GetPublicationRelationQuals(pubid, + PUBLICATION_PART_ROOT); List *delrels = NIL; - ListCell *oldlc; + ListCell *oldrelqualc; /* Calculate which relations to drop. */ - foreach(oldlc, oldrelids) + foreach(oldrelqualc, oldrelquals) { - Oid oldrelid = lfirst_oid(oldlc); + PublicationRelationQual *oldrelqual = lfirst(oldrelqualc); + PublicationRelationQual *newrelqual; ListCell *newlc; bool found = false; foreach(newlc, rels) { - Relation newrel = (Relation) lfirst(newlc); + newrelqual = (PublicationRelationQual *) lfirst(newlc); - if (RelationGetRelid(newrel) == oldrelid) + if (RelationGetRelid(newrelqual->relation) == RelationGetRelid(oldrelqual->relation)) { found = true; break; } } - if (!found) + + /* + * Remove publication / relation mapping iif (i) table is not + * found in the new list or (ii) table is found in the new list, + * however, its qual does not match the old one (in this case, a + * simple tuple update is not enough because of the dependencies). + */ + if (!found || (found && !equal(oldrelqual->whereClause, newrelqual->whereClause))) { - Relation oldrel = table_open(oldrelid, - ShareUpdateExclusiveLock); + PublicationRelationQual *oldrelqual2 = palloc(sizeof(PublicationRelationQual)); - delrels = lappend(delrels, oldrel); + oldrelqual2->relation = table_open(RelationGetRelid(oldrelqual->relation), + ShareUpdateExclusiveLock); + + delrels = lappend(delrels, oldrelqual2); } } /* And drop them. */ PublicationDropTables(pubid, delrels, true); + CloseTableList(oldrelquals); + CloseTableList(delrels); + /* * Don't bother calculating the difference for adding, we'll catch and * skip existing ones when doing catalog update. */ PublicationAddTables(pubid, rels, true, stmt); - - CloseTableList(delrels); } CloseTableList(rels); @@ -509,13 +542,15 @@ OpenTableList(List *tables) List *relids = NIL; List *rels = NIL; ListCell *lc; + PublicationRelationQual *relqual; /* * Open, share-lock, and check all the explicitly-specified relations */ foreach(lc, tables) { - RangeVar *rv = castNode(RangeVar, lfirst(lc)); + PublicationTable *t = lfirst(lc); + RangeVar *rv = castNode(RangeVar, t->relation); bool recurse = rv->inh; Relation rel; Oid myrelid; @@ -538,8 +573,10 @@ OpenTableList(List *tables) table_close(rel, ShareUpdateExclusiveLock); continue; } - - rels = lappend(rels, rel); + relqual = palloc(sizeof(PublicationRelationQual)); + relqual->relation = rel; + relqual->whereClause = t->whereClause; + rels = lappend(rels, relqual); relids = lappend_oid(relids, myrelid); /* @@ -572,7 +609,11 @@ OpenTableList(List *tables) /* find_all_inheritors already got lock */ rel = table_open(childrelid, NoLock); - rels = lappend(rels, rel); + relqual = palloc(sizeof(PublicationRelationQual)); + relqual->relation = rel; + /* child inherits WHERE clause from parent */ + relqual->whereClause = t->whereClause; + rels = lappend(rels, relqual); relids = lappend_oid(relids, childrelid); } } @@ -593,10 +634,12 @@ CloseTableList(List *rels) foreach(lc, rels) { - Relation rel = (Relation) lfirst(lc); + PublicationRelationQual *rel = (PublicationRelationQual *) lfirst(lc); - table_close(rel, NoLock); + table_close(rel->relation, NoLock); } + + list_free_deep(rels); } /* @@ -612,13 +655,13 @@ PublicationAddTables(Oid pubid, List *rels, bool if_not_exists, foreach(lc, rels) { - Relation rel = (Relation) lfirst(lc); + PublicationRelationQual *rel = (PublicationRelationQual *) lfirst(lc); ObjectAddress obj; /* Must be owner of the table or superuser. */ - if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId())) - aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind), - RelationGetRelationName(rel)); + if (!pg_class_ownercheck(RelationGetRelid(rel->relation), GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->relation->rd_rel->relkind), + RelationGetRelationName(rel->relation)); obj = publication_add_relation(pubid, rel, if_not_exists); if (stmt) @@ -644,8 +687,8 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok) foreach(lc, rels) { - Relation rel = (Relation) lfirst(lc); - Oid relid = RelationGetRelid(rel); + PublicationRelationQual *rel = (PublicationRelationQual *) lfirst(lc); + Oid relid = RelationGetRelid(rel->relation); prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid, ObjectIdGetDatum(relid), @@ -658,7 +701,7 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok) ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("relation \"%s\" is not part of the publication", - RelationGetRelationName(rel)))); + RelationGetRelationName(rel->relation)))); } ObjectAddressSet(obj, PublicationRelRelationId, prid); diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index c0bb44a85c..af7cec58e7 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -414,13 +414,14 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); relation_expr_list dostmt_opt_list transform_element_list transform_type_list TriggerTransitions TriggerReferencing + publication_table_list vacuum_relation_list opt_vacuum_relation_list drop_option_list %type group_by_list %type group_by_item empty_grouping_set rollup_clause cube_clause %type grouping_sets_clause -%type opt_publication_for_tables publication_for_tables +%type opt_publication_for_tables publication_for_tables publication_table_elem %type opt_fdw_options fdw_options %type fdw_option @@ -9446,7 +9447,7 @@ opt_publication_for_tables: ; publication_for_tables: - FOR TABLE relation_expr_list + FOR TABLE publication_table_list { $$ = (Node *) $3; } @@ -9477,7 +9478,7 @@ AlterPublicationStmt: n->options = $5; $$ = (Node *)n; } - | ALTER PUBLICATION name ADD_P TABLE relation_expr_list + | ALTER PUBLICATION name ADD_P TABLE publication_table_list { AlterPublicationStmt *n = makeNode(AlterPublicationStmt); n->pubname = $3; @@ -9485,7 +9486,7 @@ AlterPublicationStmt: n->tableAction = DEFELEM_ADD; $$ = (Node *)n; } - | ALTER PUBLICATION name SET TABLE relation_expr_list + | ALTER PUBLICATION name SET TABLE publication_table_list { AlterPublicationStmt *n = makeNode(AlterPublicationStmt); n->pubname = $3; @@ -9493,7 +9494,7 @@ AlterPublicationStmt: n->tableAction = DEFELEM_SET; $$ = (Node *)n; } - | ALTER PUBLICATION name DROP TABLE relation_expr_list + | ALTER PUBLICATION name DROP TABLE publication_table_list { AlterPublicationStmt *n = makeNode(AlterPublicationStmt); n->pubname = $3; @@ -9503,6 +9504,20 @@ AlterPublicationStmt: } ; +publication_table_list: + publication_table_elem { $$ = list_make1($1); } + | publication_table_list ',' publication_table_elem { $$ = lappend($1, $3); } + ; + +publication_table_elem: relation_expr OptWhereClause + { + PublicationTable *n = makeNode(PublicationTable); + n->relation = $1; + n->whereClause = $2; + $$ = (Node *) n; + } + ; + /***************************************************************************** * * CREATE SUBSCRIPTION name ... diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c index 783f3fe8f2..722272f2ba 100644 --- a/src/backend/parser/parse_agg.c +++ b/src/backend/parser/parse_agg.c @@ -544,6 +544,13 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr) err = _("grouping operations are not allowed in COPY FROM WHERE conditions"); break; + case EXPR_KIND_PUBLICATION_WHERE: + if (isAgg) + err = _("aggregate functions are not allowed in publication WHERE expressions"); + else + err = _("grouping operations are not allowed in publication WHERE expressions"); + + break; /* * There is intentionally no default: case here, so that the @@ -933,6 +940,9 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc, case EXPR_KIND_GENERATED_COLUMN: err = _("window functions are not allowed in column generation expressions"); break; + case EXPR_KIND_PUBLICATION_WHERE: + err = _("window functions are not allowed in publication WHERE expressions"); + break; /* * There is intentionally no default: case here, so that the diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c index 36002f059d..c5bc464806 100644 --- a/src/backend/parser/parse_expr.c +++ b/src/backend/parser/parse_expr.c @@ -169,6 +169,13 @@ transformExprRecurse(ParseState *pstate, Node *expr) /* Guard against stack overflow due to overly complex expressions */ check_stack_depth(); + /* Functions are not allowed in publication WHERE clauses */ + if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE && nodeTag(expr) == T_FuncCall) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("functions are not allowed in WHERE"), + parser_errposition(pstate, exprLocation(expr)))); + switch (nodeTag(expr)) { case T_ColumnRef: @@ -567,6 +574,7 @@ transformColumnRef(ParseState *pstate, ColumnRef *cref) case EXPR_KIND_CALL_ARGUMENT: case EXPR_KIND_COPY_WHERE: case EXPR_KIND_GENERATED_COLUMN: + case EXPR_KIND_PUBLICATION_WHERE: /* okay */ break; @@ -1889,6 +1897,9 @@ transformSubLink(ParseState *pstate, SubLink *sublink) case EXPR_KIND_GENERATED_COLUMN: err = _("cannot use subquery in column generation expression"); break; + case EXPR_KIND_PUBLICATION_WHERE: + err = _("cannot use subquery in publication WHERE expression"); + break; /* * There is intentionally no default: case here, so that the @@ -3488,6 +3499,8 @@ ParseExprKindName(ParseExprKind exprKind) return "WHERE"; case EXPR_KIND_GENERATED_COLUMN: return "GENERATED AS"; + case EXPR_KIND_PUBLICATION_WHERE: + return "publication expression"; /* * There is intentionally no default: case here, so that the diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c index 23ac2a2fe6..a793c3bf79 100644 --- a/src/backend/parser/parse_func.c +++ b/src/backend/parser/parse_func.c @@ -2527,6 +2527,9 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location) case EXPR_KIND_GENERATED_COLUMN: err = _("set-returning functions are not allowed in column generation expressions"); break; + case EXPR_KIND_PUBLICATION_WHERE: + err = _("set-returning functions are not allowed in publication WHERE expressions"); + break; /* * There is intentionally no default: case here, so that the diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 8b0d2b13ac..a43a7d011f 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -631,19 +631,26 @@ copy_read_data(void *outbuf, int minread, int maxread) /* * Get information about remote relation in similar fashion the RELATION - * message provides during replication. + * message provides during replication. This function also returns the relation ++ * qualifications to be used in COPY. */ static void fetch_remote_table_info(char *nspname, char *relname, - LogicalRepRelation *lrel) + LogicalRepRelation *lrel, List **qual) { WalRcvExecResult *res; StringInfoData cmd; TupleTableSlot *slot; Oid tableRow[] = {OIDOID, CHAROID, CHAROID}; Oid attrRow[] = {TEXTOID, OIDOID, BOOLOID}; + Oid qualRow[1] = {TEXTOID}; bool isnull; - int natt; + int n; + ListCell *lc; + bool first; + + /* Avoid trashing relation map cache */ + memset(lrel, 0, sizeof(LogicalRepRelation)); lrel->nspname = nspname; lrel->relname = relname; @@ -708,20 +715,20 @@ fetch_remote_table_info(char *nspname, char *relname, lrel->atttyps = palloc0(res->ntuples * sizeof(Oid)); lrel->attkeys = NULL; - natt = 0; + n = 0; slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) { - lrel->attnames[natt] = + lrel->attnames[n] = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); Assert(!isnull); - lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 2, &isnull)); + lrel->atttyps[n] = DatumGetObjectId(slot_getattr(slot, 2, &isnull)); Assert(!isnull); if (DatumGetBool(slot_getattr(slot, 3, &isnull))) - lrel->attkeys = bms_add_member(lrel->attkeys, natt); + lrel->attkeys = bms_add_member(lrel->attkeys, n); /* Should never happen. */ - if (++natt >= MaxTupleAttributeNumber) + if (++n >= MaxTupleAttributeNumber) elog(ERROR, "too many columns in remote table \"%s.%s\"", nspname, relname); @@ -729,12 +736,85 @@ fetch_remote_table_info(char *nspname, char *relname, } ExecDropSingleTupleTableSlot(slot); - lrel->natts = natt; + lrel->natts = n; + + walrcv_clear_result(res); + + /* Get relation qual */ + resetStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT pg_get_expr(prqual, prrelid) " + " FROM pg_publication p " + " INNER JOIN pg_publication_rel pr " + " ON (p.oid = pr.prpubid) " + " WHERE pr.prrelid = %u " + " AND p.pubname IN (", lrel->remoteid); + + first = true; + foreach(lc, MySubscription->publications) + { + char *pubname = strVal(lfirst(lc)); + + if (first) + first = false; + else + appendStringInfoString(&cmd, ", "); + + appendStringInfoString(&cmd, quote_literal_cstr(pubname)); + } + appendStringInfoChar(&cmd, ')'); + + res = walrcv_exec(wrconn, cmd.data, 1, qualRow); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + (errmsg("could not fetch relation qualifications for table \"%s.%s\" from publisher: %s", + nspname, relname, res->err))); + + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + Datum rf = slot_getattr(slot, 1, &isnull); + + if (!isnull) + *qual = lappend(*qual, makeString(TextDatumGetCString(rf))); + + ExecClearTuple(slot); + } + ExecDropSingleTupleTableSlot(slot); walrcv_clear_result(res); pfree(cmd.data); } +static char * +TableQualToText(List *qual) +{ + StringInfoData cmd; + ListCell *lc; + bool first = true; + + if (qual == NIL) + { + return "true"; + } + + initStringInfo(&cmd); + + foreach(lc, qual) + { + char *q = strVal(lfirst(lc)); + + if (first) + first = false; + else + appendStringInfoString(&cmd, " OR "); + appendStringInfo(&cmd, "%s", q); + } + + return cmd.data; +} + /* * Copy existing data of a table from publisher. * @@ -745,6 +825,7 @@ copy_table(Relation rel) { LogicalRepRelMapEntry *relmapentry; LogicalRepRelation lrel; + List *qual = NIL; WalRcvExecResult *res; StringInfoData cmd; CopyFromState cstate; @@ -753,7 +834,7 @@ copy_table(Relation rel) /* Get the publisher relation info. */ fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)), - RelationGetRelationName(rel), &lrel); + RelationGetRelationName(rel), &lrel, &qual); /* Put the relation into relmap. */ logicalrep_relmap_update(&lrel); @@ -762,16 +843,20 @@ copy_table(Relation rel) relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock); Assert(rel == relmapentry->localrel); + /* list of columns for COPY */ + attnamelist = make_copy_attnamelist(relmapentry); + /* Start copy on the publisher. */ initStringInfo(&cmd); - if (lrel.relkind == RELKIND_RELATION) + if (lrel.relkind == RELKIND_RELATION && qual == NIL) appendStringInfo(&cmd, "COPY %s TO STDOUT", quote_qualified_identifier(lrel.nspname, lrel.relname)); else { /* - * For non-tables, we need to do COPY (SELECT ...), but we can't just - * do SELECT * because we need to not copy generated columns. + * For non-tables or tables with quals, we need to do + * COPY (SELECT ...), but we can't just do SELECT * because + * we need to not copy generated columns. */ appendStringInfoString(&cmd, "COPY (SELECT "); for (int i = 0; i < lrel.natts; i++) @@ -780,9 +865,14 @@ copy_table(Relation rel) if (i < lrel.natts - 1) appendStringInfoString(&cmd, ", "); } - appendStringInfo(&cmd, " FROM %s) TO STDOUT", - quote_qualified_identifier(lrel.nspname, lrel.relname)); + appendStringInfo(&cmd, " FROM %s WHERE %s) TO STDOUT", + quote_qualified_identifier(lrel.nspname, lrel.relname), + TableQualToText(qual)); } + + /* we don't need quals anymore */ + list_free_deep(qual); + res = walrcv_exec(wrconn, cmd.data, 0, NULL); pfree(cmd.data); if (res->status != WALRCV_OK_COPY_OUT) @@ -797,7 +887,6 @@ copy_table(Relation rel) (void) addRangeTableEntryForRelation(pstate, rel, AccessShareLock, NULL, false, false); - attnamelist = make_copy_attnamelist(relmapentry); cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL); /* Do the copy */ @@ -806,6 +895,7 @@ copy_table(Relation rel) logicalrep_rel_close(relmapentry, NoLock); } + /* * Start syncing the table in the sync worker. * diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index e742eceb71..29db29e7ba 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -340,7 +340,7 @@ handle_streamed_transaction(LogicalRepMsgType action, StringInfo s) * * This is based on similar code in copy.c */ -static EState * +EState * create_estate_for_relation(Relation rel) { EState *estate; @@ -350,8 +350,8 @@ create_estate_for_relation(Relation rel) rte = makeNode(RangeTblEntry); rte->rtekind = RTE_RELATION; - rte->relid = RelationGetRelid(rel->localrel); - rte->relkind = rel->localrel->rd_rel->relkind; + rte->relid = RelationGetRelid(rel); + rte->relkind = rel->rd_rel->relkind; rte->rellockmode = AccessShareLock; ExecInitRangeTable(estate, list_make1(rte)); diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 9c997aed83..1faa6a224c 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -15,13 +15,23 @@ #include "access/tupconvert.h" #include "catalog/partition.h" #include "catalog/pg_publication.h" +#include "catalog/pg_publication_rel.h" +#include "catalog/pg_type.h" #include "commands/defrem.h" +#include "executor/executor.h" #include "fmgr.h" #include "replication/logical.h" +#include "nodes/execnodes.h" +#include "nodes/nodeFuncs.h" +#include "optimizer/planner.h" +#include "optimizer/optimizer.h" +#include "parser/parse_coerce.h" #include "replication/logicalproto.h" +#include "replication/logicalrelation.h" #include "replication/origin.h" #include "replication/pgoutput.h" #include "utils/int8.h" +#include "utils/builtins.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -98,6 +108,7 @@ typedef struct RelationSyncEntry bool replicate_valid; PublicationActions pubactions; + List *qual; /* * OID of the relation to publish changes as. For a partition, this may @@ -536,6 +547,65 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Assert(false); } + /* ... then check row filter */ + if (list_length(relentry->qual) > 0) + { + HeapTuple old_tuple; + HeapTuple new_tuple; + TupleDesc tupdesc; + EState *estate; + ExprContext *ecxt; + MemoryContext oldcxt; + ListCell *lc; + bool matched = true; + + old_tuple = change->data.tp.oldtuple ? &change->data.tp.oldtuple->tuple : NULL; + new_tuple = change->data.tp.newtuple ? &change->data.tp.newtuple->tuple : NULL; + tupdesc = RelationGetDescr(relation); + estate = create_estate_for_relation(relation); + + /* prepare context per tuple */ + ecxt = GetPerTupleExprContext(estate); + oldcxt = MemoryContextSwitchTo(estate->es_query_cxt); + ecxt->ecxt_scantuple = ExecInitExtraTupleSlot(estate, tupdesc, &TTSOpsHeapTuple); + + ExecStoreHeapTuple(new_tuple ? new_tuple : old_tuple, ecxt->ecxt_scantuple, false); + + foreach(lc, relentry->qual) + { + Node *qual; + ExprState *expr_state; + Expr *expr; + Oid expr_type; + Datum res; + bool isnull; + + qual = (Node *) lfirst(lc); + + /* evaluates row filter */ + expr_type = exprType(qual); + expr = (Expr *) coerce_to_target_type(NULL, qual, expr_type, BOOLOID, -1, COERCION_ASSIGNMENT, COERCE_IMPLICIT_CAST, -1); + expr = expression_planner(expr); + expr_state = ExecInitExpr(expr, NULL); + res = ExecEvalExpr(expr_state, ecxt, &isnull); + + /* if tuple does not match row filter, bail out */ + if (!DatumGetBool(res) || isnull) + { + matched = false; + break; + } + } + + MemoryContextSwitchTo(oldcxt); + + ExecDropSingleTupleTableSlot(ecxt->ecxt_scantuple); + FreeExecutorState(estate); + + if (!matched) + return; + } + /* Avoid leaking memory by using and resetting our own context */ old = MemoryContextSwitchTo(data->context); @@ -960,6 +1030,7 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->replicate_valid = false; entry->pubactions.pubinsert = entry->pubactions.pubupdate = entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false; + entry->qual = NIL; entry->publish_as_relid = InvalidOid; } @@ -990,6 +1061,9 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) foreach(lc, data->publications) { Publication *pub = lfirst(lc); + HeapTuple rf_tuple; + Datum rf_datum; + bool rf_isnull; bool publish = false; if (pub->alltables) @@ -998,11 +1072,11 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) if (pub->pubviaroot && am_partition) publish_as_relid = llast_oid(get_partition_ancestors(relid)); } + bool ancestor_published = false; + Oid ancestorOid = InvalidOid; 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 @@ -1027,13 +1101,19 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) { ancestor_published = true; if (pub->pubviaroot) + { publish_as_relid = ancestor; + } + + ancestorOid = ancestor; } } } if (list_member_oid(pubids, pub->oid) || ancestor_published) + { publish = true; + } } /* @@ -1050,9 +1130,24 @@ get_rel_sync_entry(PGOutputData *data, Oid relid) entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate; } - if (entry->pubactions.pubinsert && entry->pubactions.pubupdate && - entry->pubactions.pubdelete && entry->pubactions.pubtruncate) - break; + /* Cache row filters, if available */ + Oid relToUse = ancestor_published ? ancestorOid : relid; + rf_tuple = SearchSysCache2(PUBLICATIONRELMAP, ObjectIdGetDatum(relToUse), ObjectIdGetDatum(pub->oid)); + if (HeapTupleIsValid(rf_tuple)) + { + rf_datum = SysCacheGetAttr(PUBLICATIONRELMAP, rf_tuple, Anum_pg_publication_rel_prqual, &rf_isnull); + + if (!rf_isnull) + { + MemoryContext oldctx = MemoryContextSwitchTo(CacheMemoryContext); + char *s = TextDatumGetCString(rf_datum); + Node *rf_node = stringToNode(s); + + entry->qual = lappend(entry->qual, rf_node); + MemoryContextSwitchTo(oldctx); + } + ReleaseSysCache(rf_tuple); + } } list_free(pubids); @@ -1173,5 +1268,10 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue) */ hash_seq_init(&status, RelationSyncCache); while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL) + { entry->replicate_valid = false; + if (list_length(entry->qual) > 0) + list_free_deep(entry->qual); + entry->qual = NIL; + } } diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 309d102d7d..3121d93d54 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -85,6 +85,12 @@ typedef struct Publication PublicationActions pubactions; } Publication; +typedef struct PublicationRelationQual +{ + Relation relation; + Node *whereClause; +} PublicationRelationQual; + extern Publication *GetPublication(Oid pubid); extern Publication *GetPublicationByName(const char *pubname, bool missing_ok); extern List *GetRelationPublications(Oid relid); @@ -106,11 +112,12 @@ typedef enum PublicationPartOpt } PublicationPartOpt; extern List *GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt); +extern List * GetPublicationRelationQuals(Oid pubid, PublicationPartOpt pub_partopt); extern List *GetAllTablesPublications(void); extern List *GetAllTablesPublicationRelations(bool pubviaroot); extern bool is_publishable_relation(Relation rel); -extern ObjectAddress publication_add_relation(Oid pubid, Relation targetrel, +extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelationQual *targetrel, bool if_not_exists); extern Oid get_publication_oid(const char *pubname, bool missing_ok); diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h index 652cbcd6cb..47a5a9af43 100644 --- a/src/include/catalog/pg_publication_rel.h +++ b/src/include/catalog/pg_publication_rel.h @@ -31,6 +31,10 @@ CATALOG(pg_publication_rel,6106,PublicationRelRelationId) Oid oid; /* oid */ Oid prpubid; /* Oid of the publication */ Oid prrelid; /* Oid of the relation */ + +#ifdef CATALOG_VARLEN /* variable-length fields start here */ + pg_node_tree prqual; /* qualifications */ +#endif } FormData_pg_publication_rel; /* ---------------- @@ -44,5 +48,6 @@ DECLARE_UNIQUE_INDEX(pg_publication_rel_oid_index, 6112, on pg_publication_rel u #define PublicationRelObjectIndexId 6112 DECLARE_UNIQUE_INDEX(pg_publication_rel_prrelid_prpubid_index, 6113, on pg_publication_rel using btree(prrelid oid_ops, prpubid oid_ops)); #define PublicationRelPrrelidPrpubidIndexId 6113 +DECLARE_TOAST(pg_publication_rel, 8287, 8288); #endif /* PG_PUBLICATION_REL_H */ diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h index 3684f87a88..a336bf219d 100644 --- a/src/include/nodes/nodes.h +++ b/src/include/nodes/nodes.h @@ -479,6 +479,7 @@ typedef enum NodeTag T_PartitionRangeDatum, T_PartitionCmd, T_VacuumRelation, + T_PublicationTable, /* * TAGS FOR REPLICATION GRAMMAR PARSE NODES (replnodes.h) diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index ec14fc2036..06d5d872d6 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3498,12 +3498,19 @@ typedef struct AlterTSConfigurationStmt } AlterTSConfigurationStmt; +typedef struct PublicationTable +{ + NodeTag type; + RangeVar *relation; /* relation to be published */ + Node *whereClause; /* qualifications */ +} PublicationTable; + typedef struct CreatePublicationStmt { NodeTag type; char *pubname; /* Name of the publication */ List *options; /* List of DefElem nodes */ - List *tables; /* Optional list of tables to add */ + List *tables; /* Optional list of PublicationTable to add */ bool for_all_tables; /* Special publication for all tables in db */ } CreatePublicationStmt; @@ -3516,7 +3523,7 @@ typedef struct AlterPublicationStmt List *options; /* List of DefElem nodes */ /* parameters used for ALTER PUBLICATION ... ADD/DROP TABLE */ - List *tables; /* List of tables to add/drop */ + List *tables; /* List of PublicationTable to add/drop */ bool for_all_tables; /* Special publication for all tables in db */ DefElemAction tableAction; /* What action to perform with the tables */ } AlterPublicationStmt; diff --git a/src/include/parser/parse_node.h b/src/include/parser/parse_node.h index d25819aa28..715701a4a7 100644 --- a/src/include/parser/parse_node.h +++ b/src/include/parser/parse_node.h @@ -78,6 +78,7 @@ typedef enum ParseExprKind EXPR_KIND_CALL_ARGUMENT, /* procedure argument in CALL */ EXPR_KIND_COPY_WHERE, /* WHERE condition in COPY FROM */ EXPR_KIND_GENERATED_COLUMN, /* generation expression for a column */ + EXPR_KIND_PUBLICATION_WHERE /* WHERE condition for a table in PUBLICATION */ } ParseExprKind; diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index 62ddd3c7a2..ce4455439d 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -49,4 +49,6 @@ extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel, extern void logicalrep_typmap_update(LogicalRepTyp *remotetyp); extern char *logicalrep_typmap_gettypname(Oid remoteid); +extern EState *create_estate_for_relation(Relation rel); + #endif /* LOGICALRELATION_H */ diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out index 63d6ab7a4e..f2fc6b7ff9 100644 --- a/src/test/regress/expected/publication.out +++ b/src/test/regress/expected/publication.out @@ -156,6 +156,35 @@ Tables: DROP TABLE testpub_parted1; DROP PUBLICATION testpub_forparted, testpub_forparted1; +CREATE TABLE testpub_rf_tbl1 (a integer, b text); +CREATE TABLE testpub_rf_tbl2 (c text, d integer); +CREATE TABLE testpub_rf_tbl3 (e integer); +CREATE TABLE testpub_rf_tbl4 (g text); +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5); +RESET client_min_messages; +ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 2000); +ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2; +-- remove testpub_rf_tbl1 and add testpub_rf_tbl3 again (another WHERE expression) +ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500); +-- fail - functions disallowed +ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6); +ERROR: functions are not allowed in WHERE +LINE 1: ...ICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) ... + ^ +\dRp+ testpub5 + Publication testpub5 + Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root +--------------------------+------------+---------+---------+---------+-----------+---------- + regress_publication_user | f | t | t | t | t | f +Tables: + "public.testpub_rf_tbl3" WHERE ((e > 300) AND (e < 500)) + +DROP TABLE testpub_rf_tbl1; +DROP TABLE testpub_rf_tbl2; +DROP TABLE testpub_rf_tbl3; +DROP TABLE testpub_rf_tbl4; +DROP PUBLICATION testpub5; -- fail - view CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_view; ERROR: "testpub_view" is not a table diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql index d844075368..bad90fbf03 100644 --- a/src/test/regress/sql/publication.sql +++ b/src/test/regress/sql/publication.sql @@ -93,6 +93,27 @@ ALTER PUBLICATION testpub_forparted SET (publish_via_partition_root = true); DROP TABLE testpub_parted1; DROP PUBLICATION testpub_forparted, testpub_forparted1; +CREATE TABLE testpub_rf_tbl1 (a integer, b text); +CREATE TABLE testpub_rf_tbl2 (c text, d integer); +CREATE TABLE testpub_rf_tbl3 (e integer); +CREATE TABLE testpub_rf_tbl4 (g text); +SET client_min_messages = 'ERROR'; +CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5); +RESET client_min_messages; +ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 2000); +ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2; +-- remove testpub_rf_tbl1 and add testpub_rf_tbl3 again (another WHERE expression) +ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500); +-- fail - functions disallowed +ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6); +\dRp+ testpub5 + +DROP TABLE testpub_rf_tbl1; +DROP TABLE testpub_rf_tbl2; +DROP TABLE testpub_rf_tbl3; +DROP TABLE testpub_rf_tbl4; +DROP PUBLICATION testpub5; + -- fail - view CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_view; SET client_min_messages = 'ERROR'; -- 2.19.0