v30-0001-Row-filter-for-logical-replication.patch
application/octet-stream
Filename: v30-0001-Row-filter-for-logical-replication.patch
Type: application/octet-stream
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: format-patch
Series: patch v30-0001
Subject: Row filter for logical replication
| File | + | − |
|---|---|---|
| doc/src/sgml/catalogs.sgml | 8 | 0 |
| doc/src/sgml/ref/alter_publication.sgml | 8 | 3 |
| doc/src/sgml/ref/create_publication.sgml | 32 | 1 |
| doc/src/sgml/ref/create_subscription.sgml | 9 | 1 |
| src/backend/catalog/dependency.c | 126 | 0 |
| src/backend/catalog/pg_publication.c | 129 | 4 |
| src/backend/commands/publicationcmds.c | 63 | 41 |
| src/backend/nodes/copyfuncs.c | 1 | 0 |
| src/backend/nodes/equalfuncs.c | 1 | 0 |
| src/backend/parser/gram.y | 3 | 2 |
| src/backend/parser/parse_agg.c | 13 | 0 |
| src/backend/parser/parse_expr.c | 23 | 2 |
| src/backend/parser/parse_func.c | 6 | 0 |
| src/backend/parser/parse_oper.c | 9 | 0 |
| src/backend/replication/logical/tablesync.c | 89 | 6 |
| src/backend/replication/pgoutput/pgoutput.c | 320 | 9 |
| src/bin/pg_dump/pg_dump.c | 20 | 4 |
| src/bin/pg_dump/pg_dump.h | 1 | 0 |
| src/bin/psql/describe.c | 13 | 2 |
| src/bin/psql/tab-complete.c | 8 | 2 |
| src/include/catalog/dependency.h | 8 | 1 |
| src/include/catalog/pg_publication.h | 3 | 1 |
| src/include/catalog/pg_publication_rel.h | 6 | 0 |
| src/include/nodes/parsenodes.h | 3 | 2 |
| src/include/parser/parse_node.h | 1 | 0 |
| src/test/regress/expected/publication.out | 164 | 0 |
| src/test/regress/sql/publication.sql | 112 | 0 |
| src/test/subscription/t/025_row_filter.pl | 299 | 0 |
From 946ba28d0d3d95d2d8c15066ff86b514930cb7a9 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <ajinc@fast.au.fujitsu.com>
Date: Sat, 2 Oct 2021 01:25:16 -0400
Subject: [PATCH v30] Row filter for logical replication
This feature adds row filter for publication tables. When a publication
is defined or modified, rows that don't satisfy an optional WHERE clause
will be filtered out. This allows a database or set of tables to be
partially replicated. The row filter is per table. A new row filter can
be added simply by specifying a WHERE clause after the table name. The
WHERE clause must be enclosed by parentheses.
The WHERE clause should probably contain only columns that are part of the
primary key or that are covered by REPLICA IDENTITY. Otherwise, any DELETEs
won't be replicated. DELETE uses the old row version (that is limited to
primary key or REPLICA IDENTITY) to evaluate the row filter. INSERT and UPDATE
use the new row version to evaluate the row filter, hence, you can use any
column. If the row filter evaluates to NULL, it returns false. For simplicity,
functions are not allowed; it could possibly be addressed in a future patch.
If you choose to do the initial table synchronization, only data that satisfies
the row filters is sent. If the subscription has several publications in which
a table has been published with different WHERE clauses, rows must satisfy all
expressions to be copied. If subscriber is a pre-15 version, data
synchronization won't use row filters if they are defined in the publisher.
Previous versions cannot handle row filters.
If your publication contains a partitioned table, the publication parameter
publish_via_partition_root determines if it uses the partition row filter (if
the parameter is false, the default) or the root partitioned table row filter.
---
doc/src/sgml/catalogs.sgml | 8 +
doc/src/sgml/ref/alter_publication.sgml | 11 +-
doc/src/sgml/ref/create_publication.sgml | 33 ++-
doc/src/sgml/ref/create_subscription.sgml | 10 +-
src/backend/catalog/dependency.c | 126 +++++++++++
src/backend/catalog/pg_publication.c | 133 ++++++++++-
src/backend/commands/publicationcmds.c | 104 +++++----
src/backend/nodes/copyfuncs.c | 1 +
src/backend/nodes/equalfuncs.c | 1 +
src/backend/parser/gram.y | 5 +-
src/backend/parser/parse_agg.c | 13 ++
src/backend/parser/parse_expr.c | 25 ++-
src/backend/parser/parse_func.c | 6 +
src/backend/parser/parse_oper.c | 9 +
src/backend/replication/logical/tablesync.c | 95 +++++++-
src/backend/replication/pgoutput/pgoutput.c | 329 +++++++++++++++++++++++++++-
src/bin/pg_dump/pg_dump.c | 24 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 15 +-
src/bin/psql/tab-complete.c | 10 +-
src/include/catalog/dependency.h | 9 +-
src/include/catalog/pg_publication.h | 4 +-
src/include/catalog/pg_publication_rel.h | 6 +
src/include/nodes/parsenodes.h | 5 +-
src/include/parser/parse_node.h | 1 +
src/test/regress/expected/publication.out | 164 ++++++++++++++
src/test/regress/sql/publication.sql | 112 ++++++++++
src/test/subscription/t/025_row_filter.pl | 299 +++++++++++++++++++++++++
28 files changed, 1478 insertions(+), 81 deletions(-)
create mode 100644 src/test/subscription/t/025_row_filter.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 384e6ea..d117652 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -6231,6 +6231,14 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
Reference to relation
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>prqual</structfield> <type>pg_node_tree</type>
+ </para>
+ <para>Expression tree (in <function>nodeToString()</function>
+ representation) for the relation's qualifying condition</para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml
index faa114b..4bb4314 100644
--- a/doc/src/sgml/ref/alter_publication.sgml
+++ b/doc/src/sgml/ref/alter_publication.sgml
@@ -21,8 +21,8 @@ PostgreSQL documentation
<refsynopsisdiv>
<synopsis>
-ALTER PUBLICATION <replaceable class="parameter">name</replaceable> ADD TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ...]
-ALTER PUBLICATION <replaceable class="parameter">name</replaceable> SET TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ...]
+ALTER PUBLICATION <replaceable class="parameter">name</replaceable> ADD TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ...]
+ALTER PUBLICATION <replaceable class="parameter">name</replaceable> SET TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ...]
ALTER PUBLICATION <replaceable class="parameter">name</replaceable> DROP TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ...]
ALTER PUBLICATION <replaceable class="parameter">name</replaceable> SET ( <replaceable class="parameter">publication_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] )
ALTER PUBLICATION <replaceable class="parameter">name</replaceable> OWNER TO { <replaceable>new_owner</replaceable> | CURRENT_ROLE | CURRENT_USER | SESSION_USER }
@@ -92,7 +92,12 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
table name, only that table is affected. If <literal>ONLY</literal> is not
specified, the table and all its descendant tables (if any) are
affected. Optionally, <literal>*</literal> 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 <literal>WHERE</literal> clause is specified, rows that do not
+ satisfy the <replaceable class="parameter">expression</replaceable> will
+ not be published. Note that parentheses are required around the
+ expression. The <replaceable class="parameter">expression</replaceable>
+ is evaluated with the role used for the replication connection.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index ff82fbc..8f78fbb 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -22,7 +22,7 @@ PostgreSQL documentation
<refsynopsisdiv>
<synopsis>
CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
- [ FOR TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ...]
+ [ FOR TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ...]
| FOR ALL TABLES ]
[ WITH ( <replaceable class="parameter">publication_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
</synopsis>
@@ -71,6 +71,10 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
This does not apply to a partitioned table, however. The partitions of
a partitioned table are always implicitly considered part of the
publication, so they are never explicitly added to the publication.
+ If the optional <literal>WHERE</literal> clause is specified, rows that do
+ not satisfy the <replaceable class="parameter">expression</replaceable>
+ will not be published. Note that parentheses are required around the
+ expression. It has no effect on <literal>TRUNCATE</literal> commands.
</para>
<para>
@@ -183,6 +187,21 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
</para>
<para>
+ The <literal>WHERE</literal> clause should contain only columns that are
+ part of the primary key or be covered by <literal>REPLICA
+ IDENTITY</literal> otherwise, <command>DELETE</command> operations will not
+ be replicated. That's because old row is used and it only contains primary
+ key or columns that are part of the <literal>REPLICA IDENTITY</literal>; the
+ remaining columns are <literal>NULL</literal>. For <command>INSERT</command>
+ and <command>UPDATE</command> operations, any column might be used in the
+ <literal>WHERE</literal> clause. New row is used and it contains all
+ columns. A <literal>NULL</literal> value causes the expression to evaluate
+ to false; avoid using columns without not-null constraints in the
+ <literal>WHERE</literal> clause. The <literal>WHERE</literal> clause does
+ not allow functions or user-defined operators.
+ </para>
+
+ <para>
For an <command>INSERT ... ON CONFLICT</command> command, the publication will
publish the operation that actually results from the command. So depending
of the outcome, it may be published as either <command>INSERT</command> or
@@ -197,6 +216,11 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
<para>
<acronym>DDL</acronym> operations are not published.
</para>
+
+ <para>
+ The <literal>WHERE</literal> clause expression is executed with the role used
+ for the replication connection.
+ </para>
</refsect1>
<refsect1>
@@ -210,6 +234,13 @@ CREATE PUBLICATION mypublication FOR TABLE users, departments;
</para>
<para>
+ Create a publication that publishes all changes from active departments:
+<programlisting>
+CREATE PUBLICATION active_departments FOR TABLE departments WHERE (active IS TRUE);
+</programlisting>
+ </para>
+
+ <para>
Create a publication that publishes all changes in all tables:
<programlisting>
CREATE PUBLICATION alltables FOR ALL TABLES;
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 990a41f..5a9430e 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -206,7 +206,15 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<para>
Specifies whether to copy pre-existing data in the publications
that are being subscribed to when the replication starts.
- The default is <literal>true</literal>.
+ The default is <literal>true</literal>. If any table in the
+ publications has a <literal>WHERE</literal> clause, rows that do not
+ satisfy the <replaceable class="parameter">expression</replaceable>
+ will not be copied. If the subscription has several publications in
+ which a table has been published with different
+ <literal>WHERE</literal> clauses, rows must satisfy all expressions
+ to be copied. If the subscriber is a
+ <productname>PostgreSQL</productname> version before 15 then any row
+ filtering is ignored.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 91c3e97..405b3cd 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -132,6 +132,12 @@ typedef struct
int subflags; /* flags to pass down when recursing to obj */
} ObjectAddressAndFlags;
+/* for rowfilter_walker */
+typedef struct
+{
+ char *relname;
+} rf_context;
+
/* for find_expr_references_walker */
typedef struct
{
@@ -1554,6 +1560,126 @@ ReleaseDeletionLock(const ObjectAddress *object)
}
/*
+ * Walker checks that the row filter extression is legal. Allow only simple or
+ * or compound expressions like:
+ *
+ * "(Var Op Const)" or
+ * "(Var Op Const) Bool (Var Op Const)"
+ *
+ * Nothing more complicated is permitted. Specifically, no functions of any kind
+ * and no user-defined operators.
+ */
+static bool
+rowfilter_walker(Node *node, rf_context *context)
+{
+ char *forbidden = NULL;
+
+ if (node == NULL)
+ return false;
+
+ if (IsA(node, Var) || IsA(node, Const) || IsA(node, BoolExpr))
+ {
+ /* OK */
+ }
+ else if (IsA(node, OpExpr))
+ {
+ /* OK, except user-defined operators are not allowed. */
+ if (((OpExpr *)node)->opno >= FirstNormalObjectId)
+ forbidden = _("user-defined operators are not allowed");
+ }
+ else if (IsA(node, FuncExpr))
+ {
+ forbidden = _("function calls are not allowed");
+ }
+ else
+ {
+ elog(DEBUG1, "row filter contained something unexpected: %s", nodeToString(node));
+ forbidden = _("too complex");
+ }
+
+ if (forbidden)
+ ereport(ERROR,
+ (errmsg("invalid publication WHERE expression for relation \"%s\"",
+ context->relname),
+ errdetail("%s", forbidden),
+ errhint("only simple expressions using columns and constants are allowed")
+ ));
+
+ return expression_tree_walker(node, rowfilter_walker, (void *)context);
+}
+
+/*
+ * Walk the parse-tree of this publication row filter expression and throw an
+ * error if it encounters anything not permitted or unexpected.
+ */
+void
+rowfilter_validator(char *relname, Node *expr)
+{
+ rf_context context = {0};
+
+ context.relname = relname;
+ rowfilter_walker(expr, &context);
+}
+
+/*
+ * Find all the columns referenced by the row-filter expression and return what
+ * is found as a list of RfCol. This list is used for row-filter validation.
+ */
+List *
+rowfilter_find_cols(Node *expr, Oid relId)
+{
+ find_expr_references_context context;
+ RangeTblEntry rte;
+ int ref;
+ List *rfcol_list = NIL;
+
+ context.addrs = new_object_addresses();
+
+ /* We gin up a rather bogus rangetable list to handle Vars */
+ MemSet(&rte, 0, sizeof(rte));
+ rte.type = T_RangeTblEntry;
+ rte.rtekind = RTE_RELATION;
+ rte.relid = relId;
+ rte.relkind = RELKIND_RELATION; /* no need for exactness here */
+ rte.rellockmode = AccessShareLock;
+
+ context.rtables = list_make1(list_make1(&rte));
+
+ /* Scan the expression tree for referenceable objects */
+ find_expr_references_walker(expr, &context);
+
+ /* Remove any duplicates */
+ eliminate_duplicate_dependencies(context.addrs);
+
+ /* Build/Return the list of columns referenced by this Row Filter */
+ for (ref = 0; ref < context.addrs->numrefs; ref++)
+ {
+ ObjectAddress *thisobj = context.addrs->refs + ref;
+
+ if (thisobj->classId == RelationRelationId)
+ {
+ RfCol *rfcol;
+
+ /*
+ * The parser already took care of ensuring columns must be from
+ * the correct table.
+ */
+ Assert(thisobj->objectId == relId);
+
+ rfcol = palloc(sizeof(RfCol));
+ rfcol->name = get_attname(thisobj->objectId, thisobj->objectSubId, false);
+ rfcol->attnum = thisobj->objectSubId;
+
+ rfcol_list = lappend(rfcol_list, rfcol);
+ }
+ }
+
+ free_object_addresses(context.addrs);
+
+ return rfcol_list;
+}
+
+/*
* recordDependencyOnExpr - find expression dependencies
*
* This is used to find the dependencies of rules, constraint expressions,
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 9cd0c82..7b94192 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -34,6 +34,9 @@
#include "commands/publicationcmds.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"
@@ -138,6 +141,86 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
}
/*
+ * Walk the parse-tree to decide if the row-filter is valid or not.
+ */
+static void
+rowfilter_expr_checker(Publication *pub, ParseState *pstate, Node *rfnode, Relation rel)
+{
+ Oid relid = RelationGetRelid(rel);
+ char *relname = RelationGetRelationName(rel);
+
+ /*
+ * Rule:
+ *
+ * Walk the parse-tree and reject anything more complicated than a very
+ * simple expression.
+ */
+ rowfilter_validator(relname, rfnode);
+
+ /*
+ * Rule:
+ *
+ * If the publish operation contains "delete" then only columns that
+ * are allowed by the REPLICA IDENTITY rules are permitted to be used in
+ * the row-filter WHERE clause.
+ *
+ * TODO - check later for publish "update" case.
+ */
+ if (pub->pubactions.pubdelete)
+ {
+ char replica_identity = rel->rd_rel->relreplident;
+
+ if (replica_identity == REPLICA_IDENTITY_FULL)
+ {
+ /*
+ * FULL means all cols are in the REPLICA IDENTITY, so all cols are
+ * allowed in the row-filter too.
+ */
+ }
+ else
+ {
+ List *rfcols;
+ ListCell *lc;
+ Bitmapset *bms_okcols;
+
+ /*
+ * Find what are the cols that are part of the REPLICA IDENTITY.
+ * Note that REPLICA IDENTIY DEFAULT means primary key or nothing.
+ */
+ if (replica_identity == REPLICA_IDENTITY_DEFAULT)
+ bms_okcols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_PRIMARY_KEY);
+ else
+ bms_okcols = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+
+ /*
+ * Find what cols are referenced in the row filter WHERE clause,
+ * and validate that each of those referenced cols is allowed.
+ */
+ rfcols = rowfilter_find_cols(rfnode, relid);
+ foreach(lc, rfcols)
+ {
+ RfCol *rfcol = lfirst(lc);
+ char *colname = rfcol->name;
+ int attnum = rfcol->attnum;
+
+ if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber, bms_okcols))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+ errmsg("cannot add relation \"%s\" to publication",
+ relname),
+ errdetail("Row filter column \"%s\" is not part of the REPLICA IDENTITY",
+ colname)));
+ }
+ }
+
+ bms_free(bms_okcols);
+ list_free_deep(rfcols);
+ }
+ }
+}
+
+/*
* Gets the relations based on the publication partition option for a specified
* relation.
*/
@@ -178,21 +261,26 @@ GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
* Insert new publication / relation mapping.
*/
ObjectAddress
-publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
+publication_add_relation(Oid pubid, PublicationRelInfo *pri,
bool if_not_exists)
{
Relation rel;
HeapTuple tup;
Datum values[Natts_pg_publication_rel];
bool nulls[Natts_pg_publication_rel];
- Oid relid = RelationGetRelid(targetrel->relation);
+ Relation targetrel = pri->relation;
+ Oid relid;
Oid prrelid;
Publication *pub = GetPublication(pubid);
ObjectAddress myself,
referenced;
+ ParseState *pstate;
+ ParseNamespaceItem *nsitem;
+ Node *whereclause = NULL;
List *relids = NIL;
rel = table_open(PublicationRelRelationId, RowExclusiveLock);
+ relid = RelationGetRelid(targetrel);
/*
* Check for duplicates. Note that this does not really prevent
@@ -210,10 +298,33 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("relation \"%s\" is already member of publication \"%s\"",
- RelationGetRelationName(targetrel->relation), pub->name)));
+ RelationGetRelationName(targetrel), pub->name)));
}
- check_publication_add_relation(targetrel->relation);
+ check_publication_add_relation(targetrel);
+
+ if (pri->whereClause != NULL)
+ {
+ /* Set up a pstate to parse with */
+ pstate = make_parsestate(NULL);
+ pstate->p_sourcetext = nodeToString(pri->whereClause);
+
+ nsitem = addRangeTableEntryForRelation(pstate, targetrel,
+ AccessShareLock,
+ NULL, false, false);
+ addNSItemToQuery(pstate, nsitem, false, true, true);
+
+ whereclause = transformWhereClause(pstate,
+ copyObject(pri->whereClause),
+ EXPR_KIND_PUBLICATION_WHERE,
+ "PUBLICATION WHERE");
+
+ /* Fix up collation information */
+ assign_expr_collations(pstate, whereclause);
+
+ /* Validate the row-filter. */
+ rowfilter_expr_checker(pub, pstate, whereclause, targetrel);
+ }
/* Form a tuple. */
memset(values, 0, sizeof(values));
@@ -227,6 +338,12 @@ publication_add_relation(Oid pubid, PublicationRelInfo *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. */
@@ -243,6 +360,14 @@ publication_add_relation(Oid pubid, PublicationRelInfo *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);
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 9c7f916..747f388 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -391,38 +391,24 @@ AlterPublicationTables(AlterPublicationStmt *stmt, Relation rel,
List *delrels = NIL;
ListCell *oldlc;
- /* Calculate which relations to drop. */
+ /*
+ * Remove all publication-table mappings. We could possibly remove (i)
+ * tables that are not found in the new table list and (ii) tables that
+ * are being re-added with a different qual expression. For (ii),
+ * simply updating the existing tuple is not enough, because of qual
+ * expression dependencies.
+ */
foreach(oldlc, oldrelids)
{
Oid oldrelid = lfirst_oid(oldlc);
- ListCell *newlc;
- bool found = false;
-
- foreach(newlc, rels)
- {
- PublicationRelInfo *newpubrel;
-
- newpubrel = (PublicationRelInfo *) lfirst(newlc);
- if (RelationGetRelid(newpubrel->relation) == oldrelid)
- {
- found = true;
- break;
- }
- }
- /* Not yet in the list, open it and add to the list */
- if (!found)
- {
- Relation oldrel;
- PublicationRelInfo *pubrel;
-
- /* Wrap relation into PublicationRelInfo */
- oldrel = table_open(oldrelid, ShareUpdateExclusiveLock);
-
- pubrel = palloc(sizeof(PublicationRelInfo));
- pubrel->relation = oldrel;
-
- delrels = lappend(delrels, pubrel);
- }
+ PublicationRelInfo *oldrel;
+
+ oldrel = palloc(sizeof(PublicationRelInfo));
+ oldrel->relid = oldrelid;
+ oldrel->whereClause = NULL;
+ oldrel->relation = table_open(oldrel->relid,
+ ShareUpdateExclusiveLock);
+ delrels = lappend(delrels, oldrel);
}
/* And drop them. */
@@ -552,9 +538,10 @@ RemovePublicationById(Oid pubid)
}
/*
- * Open relations specified by a PublicationTable list.
- * In the returned list of PublicationRelInfo, tables are locked
- * in ShareUpdateExclusiveLock mode in order to add them to a publication.
+ * Open relations specified by a RangeVar list (PublicationTable or Relation).
+ *
+ * The returned tables are locked in ShareUpdateExclusiveLock mode in order to
+ * add them to a publication.
*/
static List *
OpenTableList(List *tables)
@@ -562,22 +549,46 @@ OpenTableList(List *tables)
List *relids = NIL;
List *rels = NIL;
ListCell *lc;
+ PublicationRelInfo *pub_rel;
/*
* Open, share-lock, and check all the explicitly-specified relations
*/
foreach(lc, tables)
{
- PublicationTable *t = lfirst_node(PublicationTable, lc);
- bool recurse = t->relation->inh;
+ PublicationTable *t = NULL;
+ RangeVar *rv;
+ bool recurse;
Relation rel;
Oid myrelid;
- PublicationRelInfo *pub_rel;
+ bool whereclause;
+
+ /*
+ * ALTER PUBLICATION ... ADD TABLE provides a PublicationTable List
+ * (Relation, Where clause). ALTER PUBLICATION ... DROP TABLE provides
+ * a Relation List. Check the List element to be used.
+ */
+ if (IsA(lfirst(lc), PublicationTable))
+ whereclause = true;
+ else
+ whereclause = false;
+
+ if (whereclause)
+ {
+ t = lfirst(lc);
+ rv = castNode(RangeVar, t->relation);
+ }
+ else
+ {
+ rv = lfirst_node(RangeVar, lc);
+ }
+
+ recurse = rv->inh;
/* Allow query cancel in case this takes a long time */
CHECK_FOR_INTERRUPTS();
- rel = table_openrv(t->relation, ShareUpdateExclusiveLock);
+ rel = table_openrv(rv, ShareUpdateExclusiveLock);
myrelid = RelationGetRelid(rel);
/*
@@ -594,7 +605,12 @@ OpenTableList(List *tables)
}
pub_rel = palloc(sizeof(PublicationRelInfo));
+ pub_rel->relid = myrelid;
pub_rel->relation = rel;
+ if (whereclause)
+ pub_rel->whereClause = t->whereClause;
+ else
+ pub_rel->whereClause = NULL;
rels = lappend(rels, pub_rel);
relids = lappend_oid(relids, myrelid);
@@ -629,7 +645,13 @@ OpenTableList(List *tables)
/* find_all_inheritors already got lock */
rel = table_open(childrelid, NoLock);
pub_rel = palloc(sizeof(PublicationRelInfo));
+ pub_rel->relid = childrelid;
pub_rel->relation = rel;
+ /* child inherits WHERE clause from parent */
+ if (whereclause)
+ pub_rel->whereClause = t->whereClause;
+ else
+ pub_rel->whereClause = NULL;
rels = lappend(rels, pub_rel);
relids = lappend_oid(relids, childrelid);
}
@@ -656,6 +678,8 @@ CloseTableList(List *rels)
pub_rel = (PublicationRelInfo *) lfirst(lc);
table_close(pub_rel->relation, NoLock);
}
+
+ list_free_deep(rels);
}
/*
@@ -676,7 +700,7 @@ PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
ObjectAddress obj;
/* Must be owner of the table or superuser. */
- if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId()))
+ if (!pg_class_ownercheck(pub_rel->relid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
RelationGetRelationName(rel));
@@ -705,11 +729,9 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
foreach(lc, rels)
{
PublicationRelInfo *pubrel = (PublicationRelInfo *) lfirst(lc);
- Relation rel = pubrel->relation;
- Oid relid = RelationGetRelid(rel);
prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid,
- ObjectIdGetDatum(relid),
+ ObjectIdGetDatum(pubrel->relid),
ObjectIdGetDatum(pubid));
if (!OidIsValid(prid))
{
@@ -719,7 +741,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(pubrel->relation))));
}
ObjectAddressSet(obj, PublicationRelRelationId, prid);
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 228387e..a69e131 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4964,6 +4964,7 @@ _copyPublicationTable(const PublicationTable *from)
PublicationTable *newnode = makeNode(PublicationTable);
COPY_NODE_FIELD(relation);
+ COPY_NODE_FIELD(whereClause);
return newnode;
}
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 800f588..7a33695 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -3137,6 +3137,7 @@ static bool
_equalPublicationTable(const PublicationTable *a, const PublicationTable *b)
{
COMPARE_NODE_FIELD(relation);
+ COMPARE_NODE_FIELD(whereClause);
return true;
}
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index e3068a3..9765aeb 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9637,10 +9637,11 @@ publication_table_list:
{ $$ = lappend($1, $3); }
;
-publication_table: relation_expr
+publication_table: relation_expr OptWhereClause
{
PublicationTable *n = makeNode(PublicationTable);
n->relation = $1;
+ n->whereClause = $2;
$$ = (Node *) n;
}
;
@@ -9681,7 +9682,7 @@ AlterPublicationStmt:
n->tableAction = DEFELEM_SET;
$$ = (Node *)n;
}
- | ALTER PUBLICATION name DROP TABLE publication_table_list
+ | ALTER PUBLICATION name DROP TABLE relation_expr_list
{
AlterPublicationStmt *n = makeNode(AlterPublicationStmt);
n->pubname = $3;
diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index 7d829a0..5e0c391 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -551,6 +551,14 @@ 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 0
+ if (isAgg)
+ err = _("aggregate functions are not allowed in publication WHERE expressions");
+ else
+ err = _("grouping operations are not allowed in publication WHERE expressions");
+#endif
+ break;
case EXPR_KIND_CYCLE_MARK:
errkind = true;
@@ -943,6 +951,11 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
+ case EXPR_KIND_PUBLICATION_WHERE:
+#if 0
+ err = _("window functions are not allowed in publication WHERE expressions");
+#endif
+ 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 2d1a477..3519e62 100644
--- a/src/backend/parser/parse_expr.c
+++ b/src/backend/parser/parse_expr.c
@@ -200,8 +200,21 @@ transformExprRecurse(ParseState *pstate, Node *expr)
break;
case T_FuncCall:
- result = transformFuncCall(pstate, (FuncCall *) expr);
- break;
+ {
+#if 0
+ /*
+ * Forbid functions in publication WHERE condition
+ */
+ if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("functions are not allowed in publication WHERE expressions"),
+ parser_errposition(pstate, exprLocation(expr))));
+#endif
+
+ result = transformFuncCall(pstate, (FuncCall *) expr);
+ break;
+ }
case T_MultiAssignRef:
result = transformMultiAssignRef(pstate, (MultiAssignRef *) expr);
@@ -504,6 +517,7 @@ transformColumnRef(ParseState *pstate, ColumnRef *cref)
case EXPR_KIND_COPY_WHERE:
case EXPR_KIND_GENERATED_COLUMN:
case EXPR_KIND_CYCLE_MARK:
+ case EXPR_KIND_PUBLICATION_WHERE:
/* okay */
break;
@@ -1764,6 +1778,11 @@ transformSubLink(ParseState *pstate, SubLink *sublink)
case EXPR_KIND_GENERATED_COLUMN:
err = _("cannot use subquery in column generation expression");
break;
+ case EXPR_KIND_PUBLICATION_WHERE:
+#if 0
+ err = _("cannot use subquery in publication WHERE expression");
+#endif
+ break;
/*
* There is intentionally no default: case here, so that the
@@ -3084,6 +3103,8 @@ ParseExprKindName(ParseExprKind exprKind)
return "GENERATED AS";
case EXPR_KIND_CYCLE_MARK:
return "CYCLE";
+ case EXPR_KIND_PUBLICATION_WHERE:
+ return "publication WHERE 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 3cec8de..de9600f 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2655,6 +2655,12 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location)
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
+ case EXPR_KIND_PUBLICATION_WHERE:
+ pstate->p_hasTargetSRFs = true;
+#if 0
+ err = _("set-returning functions are not allowed in publication WHERE expressions");
+#endif
+ break;
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_oper.c b/src/backend/parser/parse_oper.c
index bc34a23..b3588df 100644
--- a/src/backend/parser/parse_oper.c
+++ b/src/backend/parser/parse_oper.c
@@ -718,6 +718,15 @@ make_op(ParseState *pstate, List *opname, Node *ltree, Node *rtree,
opform->oprright)),
parser_errposition(pstate, location)));
+#if 0
+ /* Check it's not a custom operator for publication WHERE expressions */
+ if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE && opform->oid >= FirstNormalObjectId)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("user-defined operators are not allowed in publication WHERE expressions"),
+ parser_errposition(pstate, location)));
+#endif
+
/* Do typecasting and build the expression tree */
if (ltree == NULL)
{
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a..9d86a10 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -688,19 +688,23 @@ 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 command.
*/
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[] = {TEXTOID};
bool isnull;
int natt;
+ ListCell *lc;
+ bool first;
lrel->nspname = nspname;
lrel->relname = relname;
@@ -796,6 +800,59 @@ fetch_remote_table_info(char *nspname, char *relname,
lrel->natts = natt;
walrcv_clear_result(res);
+
+ /*
+ * Get relation qual. DISTINCT avoids the same expression of a table in
+ * multiple publications from being included multiple times in the final
+ * expression.
+ */
+ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
+ {
+ resetStringInfo(&cmd);
+ appendStringInfo(&cmd,
+ "SELECT DISTINCT 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(LogRepWorkerWalRcvConn, 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);
}
@@ -809,6 +866,7 @@ copy_table(Relation rel)
{
LogicalRepRelMapEntry *relmapentry;
LogicalRepRelation lrel;
+ List *qual = NIL;
WalRcvExecResult *res;
StringInfoData cmd;
CopyFromState cstate;
@@ -817,7 +875,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);
@@ -828,14 +886,18 @@ copy_table(Relation rel)
/* Start copy on the publisher. */
initStringInfo(&cmd);
- if (lrel.relkind == RELKIND_RELATION)
+
+ /* Regular table with no row filter */
+ 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.
+ * do SELECT * because we need to not copy generated columns. For
+ * tables with any row filters, build a SELECT query with AND'ed row
+ * filters for COPY.
*/
appendStringInfoString(&cmd, "COPY (SELECT ");
for (int i = 0; i < lrel.natts; i++)
@@ -844,8 +906,29 @@ copy_table(Relation rel)
if (i < lrel.natts - 1)
appendStringInfoString(&cmd, ", ");
}
- appendStringInfo(&cmd, " FROM %s) TO STDOUT",
+ appendStringInfo(&cmd, " FROM %s",
quote_qualified_identifier(lrel.nspname, lrel.relname));
+ /* list of AND'ed filters */
+ if (qual != NIL)
+ {
+ ListCell *lc;
+ bool first = true;
+
+ appendStringInfoString(&cmd, " WHERE ");
+ foreach(lc, qual)
+ {
+ char *q = strVal(lfirst(lc));
+
+ if (first)
+ first = false;
+ else
+ appendStringInfoString(&cmd, " AND ");
+ appendStringInfoString(&cmd, q);
+ }
+ list_free_deep(qual);
+ }
+
+ appendStringInfoString(&cmd, ") TO STDOUT");
}
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 14d737f..ce5e1c5 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -13,18 +13,27 @@
#include "postgres.h"
#include "access/tupconvert.h"
+#include "access/xact.h"
#include "catalog/partition.h"
#include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
#include "commands/defrem.h"
+#include "executor/executor.h"
#include "fmgr.h"
+#include "nodes/nodeFuncs.h"
+#include "optimizer/optimizer.h"
+#include "parser/parse_coerce.h"
#include "replication/logical.h"
#include "replication/logicalproto.h"
+#include "replication/logicalrelation.h"
#include "replication/origin.h"
#include "replication/pgoutput.h"
+#include "utils/builtins.h"
#include "utils/int8.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
+#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/varlena.h"
@@ -116,6 +125,16 @@ typedef struct RelationSyncEntry
PublicationActions pubactions;
/*
+ * Row-filter related members:
+ * The flag 'rowfilter_valid' only means that exprstate_list is correct -
+ * It doesn't mean that there actual is any row filter present for the
+ * current relid.
+ */
+ bool rowfilter_valid;
+ List *exprstate_list; /* ExprState for row filter(s) */
+ TupleTableSlot *scantuple; /* tuple table slot for row filter */
+
+ /*
* OID of the relation to publish changes as. For a partition, this may
* be set to one of its ancestors whose schema will be used when
* replicating changes, if publish_via_partition_root is set for the
@@ -137,7 +156,7 @@ static HTAB *RelationSyncCache = NULL;
static void init_rel_sync_cache(MemoryContext decoding_context);
static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit);
-static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data, Oid relid);
+static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data, Relation relation);
static void rel_sync_cache_relation_cb(Datum arg, Oid relid);
static void rel_sync_cache_publication_cb(Datum arg, int cacheid,
uint32 hashvalue);
@@ -146,6 +165,13 @@ static void set_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
TransactionId xid);
+/* row filter routines */
+static EState *create_estate_for_relation(Relation rel);
+static ExprState *pgoutput_row_filter_init_expr(Node *rfnode);
+static bool pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext);
+static bool pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple,
+ HeapTuple newtuple, RelationSyncEntry *entry);
+
/*
* Specify output plugin callbacks
*/
@@ -621,6 +647,250 @@ send_relation_and_attrs(Relation relation, TransactionId xid,
}
/*
+ * Executor state preparation for evaluation of row filter expressions for the
+ * specified relation.
+ */
+static EState *
+create_estate_for_relation(Relation rel)
+{
+ EState *estate;
+ RangeTblEntry *rte;
+
+ estate = CreateExecutorState();
+
+ rte = makeNode(RangeTblEntry);
+ rte->rtekind = RTE_RELATION;
+ rte->relid = RelationGetRelid(rel);
+ rte->relkind = rel->rd_rel->relkind;
+ rte->rellockmode = AccessShareLock;
+ ExecInitRangeTable(estate, list_make1(rte));
+
+ estate->es_output_cid = GetCurrentCommandId(false);
+
+ return estate;
+}
+
+/*
+ * Initialize for row filter expression execution.
+ */
+static ExprState *
+pgoutput_row_filter_init_expr(Node *rfnode)
+{
+ ExprState *exprstate;
+ Oid exprtype;
+ Expr *expr;
+ MemoryContext oldctx;
+
+ /* Prepare expression for execution */
+ exprtype = exprType(rfnode);
+ expr = (Expr *) coerce_to_target_type(NULL, rfnode, exprtype, BOOLOID, -1, COERCION_ASSIGNMENT, COERCE_IMPLICIT_CAST, -1);
+
+ if (expr == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_CANNOT_COERCE),
+ errmsg("row filter returns type %s that cannot be coerced to the expected type %s",
+ format_type_be(exprtype),
+ format_type_be(BOOLOID)),
+ errhint("You will need to rewrite the row filter.")));
+
+ /*
+ * Cache ExprState using CacheMemoryContext. This is the same code as
+ * ExecPrepareExpr() but that is not used because it doesn't use an EState.
+ * It should probably be another function in the executor to handle the
+ * execution outside a normal Plan tree context.
+ */
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ expr = expression_planner(expr);
+ exprstate = ExecInitExpr(expr, NULL);
+ MemoryContextSwitchTo(oldctx);
+
+ return exprstate;
+}
+
+/*
+ * Evaluates row filter.
+ *
+ * If the row filter evaluates to NULL, it is taken as false i.e. the change
+ * isn't replicated.
+ */
+static bool
+pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
+{
+ Datum ret;
+ bool isnull;
+
+ Assert(state != NULL);
+
+ ret = ExecEvalExprSwitchContext(state, econtext, &isnull);
+
+ elog(DEBUG3, "row filter evaluates to %s (isnull: %s)",
+ DatumGetBool(ret) ? "true" : "false",
+ isnull ? "true" : "false");
+
+ if (isnull)
+ return false;
+
+ return DatumGetBool(ret);
+}
+
+/*
+ * Change is checked against the row filter, if any.
+ *
+ * If it returns true, the change is replicated, otherwise, it is not.
+ */
+static bool
+pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry)
+{
+ EState *estate;
+ ExprContext *ecxt;
+ ListCell *lc;
+ bool result = true;
+ Oid relid = RelationGetRelid(relation);
+
+ /*
+ * If the row filter caching is currently flagged "invalid" then it means we
+ * don't know yet if there is/isn't any row filters for this relation.
+ *
+ * This code is usually one-time execution.
+ */
+ if (!entry->rowfilter_valid)
+ {
+ bool am_partition = get_rel_relispartition(relid);
+ MemoryContext oldctx;
+ TupleDesc tupdesc = RelationGetDescr(relation);
+
+ /*
+ * Create a tuple table slot for row filter. TupleDesc must live as
+ * long as the cache remains. Release the tuple table slot if it
+ * already exists.
+ */
+ if (entry->scantuple != NULL)
+ {
+ ExecDropSingleTupleTableSlot(entry->scantuple);
+ entry->scantuple = NULL;
+ }
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ tupdesc = CreateTupleDescCopy(tupdesc);
+ entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+ MemoryContextSwitchTo(oldctx);
+
+ /*
+ * Find if there are any row filters for this relation. If there are,
+ * then prepare the necessary ExprState(s) and cache then in the
+ * entry->exprstate_list.
+ *
+ * NOTE: All publication-table mappings must be checked.
+ *
+ * NOTE: If the relation is a partition and pubviaroot is true, use
+ * the row filter of the topmost partitioned table instead of the row
+ * filter of its own partition.
+ */
+ foreach(lc, data->publications)
+ {
+ Publication *pub = lfirst(lc);
+ HeapTuple rftuple;
+ Datum rfdatum;
+ bool rfisnull;
+ Oid pub_relid = relid;
+
+ if (pub->pubviaroot && am_partition)
+ {
+ if (pub->alltables)
+ pub_relid = llast_oid(get_partition_ancestors(relid));
+ else
+ {
+ List *ancestors = get_partition_ancestors(relid);
+ ListCell *lc2;
+
+ /*
+ * Find the "topmost" ancestor that is in this
+ * publication.
+ */
+ foreach(lc2, ancestors)
+ {
+ Oid ancestor = lfirst_oid(lc2);
+
+ if (list_member_oid(GetRelationPublications(ancestor),
+ pub->oid))
+ {
+ pub_relid = ancestor;
+ }
+ }
+ }
+ }
+
+ /*
+ * Lookup if there is a row-filter, and if so build the ExprState for it.
+ */
+ rftuple = SearchSysCache2(PUBLICATIONRELMAP, ObjectIdGetDatum(pub_relid), ObjectIdGetDatum(pub->oid));
+ if (HeapTupleIsValid(rftuple))
+ {
+ rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple, Anum_pg_publication_rel_prqual, &rfisnull);
+
+ if (!rfisnull)
+ {
+ Node *rfnode;
+ ExprState *exprstate;
+
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ rfnode = stringToNode(TextDatumGetCString(rfdatum));
+ exprstate = pgoutput_row_filter_init_expr(rfnode);
+ entry->exprstate_list = lappend(entry->exprstate_list, exprstate);
+ MemoryContextSwitchTo(oldctx);
+ }
+
+ ReleaseSysCache(rftuple);
+ }
+
+ } /* loop all subscribed publications */
+
+ entry->rowfilter_valid = true;
+ }
+
+ /* Bail out if there is no row filter */
+ if (entry->exprstate_list == NIL)
+ return true;
+
+ elog(DEBUG3, "table \"%s.%s\" has row filter",
+ get_namespace_name(get_rel_namespace(relid)),
+ get_rel_name(relid));
+
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ estate = create_estate_for_relation(relation);
+
+ /* Prepare context per tuple */
+ ecxt = GetPerTupleExprContext(estate);
+ ecxt->ecxt_scantuple = entry->scantuple;
+
+ ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
+
+ /*
+ * If the subscription has multiple publications and the same table has a
+ * different row filter in these publications, all row filters must be
+ * matched in order to replicate this change.
+ */
+ foreach(lc, entry->exprstate_list)
+ {
+ ExprState *exprstate = (ExprState *) lfirst(lc);
+
+ /* Evaluates row filter */
+ result = pgoutput_row_filter_exec_expr(exprstate, ecxt);
+
+ /* If the tuple does not match one of the row filters, bail out */
+ if (!result)
+ break;
+ }
+
+ /* Cleanup allocated resources */
+ ResetExprContext(ecxt);
+ FreeExecutorState(estate);
+ PopActiveSnapshot();
+
+ return result;
+}
+
+/*
* Sends the decoded DML over wire.
*
* This is called both in streaming and non-streaming modes.
@@ -647,7 +917,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
if (in_streaming)
xid = change->txn->xid;
- relentry = get_rel_sync_entry(data, RelationGetRelid(relation));
+ relentry = get_rel_sync_entry(data, relation);
/* First check the table filter */
switch (change->action)
@@ -671,8 +941,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Avoid leaking memory by using and resetting our own context */
old = MemoryContextSwitchTo(data->context);
- maybe_send_schema(ctx, change, relation, relentry);
-
/* Send the data */
switch (change->action)
{
@@ -680,6 +948,16 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
{
HeapTuple tuple = &change->data.tp.newtuple->tuple;
+ /* Check row filter. */
+ if (!pgoutput_row_filter(data, relation, NULL, tuple, relentry))
+ break;
+
+ /*
+ * Schema should be sent before the logic that replaces the
+ * relation because it also sends the ancestor's relation.
+ */
+ maybe_send_schema(ctx, change, relation, relentry);
+
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
{
@@ -703,6 +981,12 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
&change->data.tp.oldtuple->tuple : NULL;
HeapTuple newtuple = &change->data.tp.newtuple->tuple;
+ /* Check row filter. */
+ if (!pgoutput_row_filter(data, relation, oldtuple, newtuple, relentry))
+ break;
+
+ maybe_send_schema(ctx, change, relation, relentry);
+
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
{
@@ -731,6 +1015,12 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
{
HeapTuple oldtuple = &change->data.tp.oldtuple->tuple;
+ /* Check row filter. */
+ if (!pgoutput_row_filter(data, relation, oldtuple, NULL, relentry))
+ break;
+
+ maybe_send_schema(ctx, change, relation, relentry);
+
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
{
@@ -794,7 +1084,7 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
if (!is_publishable_relation(relation))
continue;
- relentry = get_rel_sync_entry(data, relid);
+ relentry = get_rel_sync_entry(data, relation);
if (!relentry->pubactions.pubtruncate)
continue;
@@ -1113,9 +1403,10 @@ set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
* when publishing.
*/
static RelationSyncEntry *
-get_rel_sync_entry(PGOutputData *data, Oid relid)
+get_rel_sync_entry(PGOutputData *data, Relation relation)
{
RelationSyncEntry *entry;
+ Oid relid = RelationGetRelid(relation);
bool am_partition = get_rel_relispartition(relid);
char relkind = get_rel_relkind(relid);
bool found;
@@ -1136,8 +1427,11 @@ get_rel_sync_entry(PGOutputData *data, Oid relid)
entry->schema_sent = false;
entry->streamed_txns = NIL;
entry->replicate_valid = false;
+ entry->rowfilter_valid = false;
entry->pubactions.pubinsert = entry->pubactions.pubupdate =
entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
+ entry->scantuple = NULL;
+ entry->exprstate_list = NIL;
entry->publish_as_relid = InvalidOid;
entry->map = NULL; /* will be set by maybe_send_schema() if
* needed */
@@ -1230,9 +1524,6 @@ 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;
}
list_free(pubids);
@@ -1339,6 +1630,21 @@ rel_sync_cache_relation_cb(Datum arg, Oid relid)
free_conversion_map(entry->map);
}
entry->map = NULL;
+
+ /*
+ * Row filter cache cleanups. (Will be rebuilt later if needed).
+ */
+ entry->rowfilter_valid = false;
+ if (entry->scantuple != NULL)
+ {
+ ExecDropSingleTupleTableSlot(entry->scantuple);
+ entry->scantuple = NULL;
+ }
+ if (entry->exprstate_list != NIL)
+ {
+ list_free_deep(entry->exprstate_list);
+ entry->exprstate_list = NIL;
+ }
}
}
@@ -1350,6 +1656,7 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
{
HASH_SEQ_STATUS status;
RelationSyncEntry *entry;
+ MemoryContext oldctx;
/*
* We can get here if the plugin was used in SQL interface as the
@@ -1359,6 +1666,8 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
if (RelationSyncCache == NULL)
return;
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+
/*
* There is no way to find which entry in our cache the hash belongs to so
* mark the whole cache as invalid.
@@ -1377,6 +1686,8 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
entry->pubactions.pubdelete = false;
entry->pubactions.pubtruncate = false;
}
+
+ MemoryContextSwitchTo(oldctx);
}
/* Send Replication origin */
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index a485fb2..0f4892c 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4141,6 +4141,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
int i_oid;
int i_prpubid;
int i_prrelid;
+ int i_prrelqual;
int i,
j,
ntups;
@@ -4151,9 +4152,16 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
query = createPQExpBuffer();
/* Collect all publication membership info. */
- appendPQExpBufferStr(query,
- "SELECT tableoid, oid, prpubid, prrelid "
- "FROM pg_catalog.pg_publication_rel");
+ 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");
+ else
+ appendPQExpBufferStr(query,
+ "SELECT tableoid, oid, prpubid, prrelid, "
+ "NULL AS prrelqual "
+ "FROM pg_catalog.pg_publication_rel");
res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
ntups = PQntuples(res);
@@ -4162,6 +4170,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
i_oid = PQfnumber(res, "oid");
i_prpubid = PQfnumber(res, "prpubid");
i_prrelid = PQfnumber(res, "prrelid");
+ i_prrelqual = PQfnumber(res, "prrelqual");
/* this allocation may be more than we need */
pubrinfo = pg_malloc(ntups * sizeof(PublicationRelInfo));
@@ -4202,6 +4211,10 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
pubrinfo[j].dobj.name = tbinfo->dobj.name;
pubrinfo[j].publication = pubinfo;
pubrinfo[j].pubtable = tbinfo;
+ if (PQgetisnull(res, i, i_prrelqual))
+ pubrinfo[j].pubrelqual = NULL;
+ else
+ pubrinfo[j].pubrelqual = pg_strdup(PQgetvalue(res, i, i_prrelqual));
/* Decide whether we want to dump it */
selectDumpablePublicationTable(&(pubrinfo[j].dobj), fout);
@@ -4234,8 +4247,11 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD TABLE ONLY",
fmtId(pubinfo->dobj.name));
- appendPQExpBuffer(query, " %s;\n",
+ appendPQExpBuffer(query, " %s",
fmtQualifiedDumpable(tbinfo));
+ if (pubrinfo->pubrelqual)
+ appendPQExpBuffer(query, " WHERE (%s)", pubrinfo->pubrelqual);
+ appendPQExpBufferStr(query, ";\n");
/*
* There is no point in creating a drop query as the drop is done by table
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 29af845..f932a70 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -629,6 +629,7 @@ typedef struct _PublicationRelInfo
DumpableObject dobj;
PublicationInfo *publication;
TableInfo *pubtable;
+ char *pubrelqual;
} PublicationRelInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index a33d77c..83249e8 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6329,8 +6329,15 @@ describePublications(const char *pattern)
if (!puballtables)
{
printfPQExpBuffer(&buf,
- "SELECT n.nspname, c.relname\n"
- "FROM pg_catalog.pg_class c,\n"
+ "SELECT n.nspname, c.relname");
+ if (pset.sversion >= 150000)
+ appendPQExpBufferStr(&buf,
+ ", pg_get_expr(pr.prqual, c.oid)");
+ else
+ appendPQExpBufferStr(&buf,
+ ", NULL");
+ appendPQExpBuffer(&buf,
+ "\nFROM pg_catalog.pg_class c,\n"
" pg_catalog.pg_namespace n,\n"
" pg_catalog.pg_publication_rel pr\n"
"WHERE c.relnamespace = n.oid\n"
@@ -6359,6 +6366,10 @@ describePublications(const char *pattern)
PQgetvalue(tabres, j, 0),
PQgetvalue(tabres, j, 1));
+ if (!PQgetisnull(tabres, j, 2))
+ appendPQExpBuffer(&buf, " WHERE (%s)",
+ PQgetvalue(tabres, j, 2));
+
printTableAddFooter(&cont, buf.data);
}
PQclear(tabres);
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 5cd5838..8686ec6 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1648,6 +1648,11 @@ psql_completion(const char *text, int start, int end)
/* ALTER PUBLICATION <name> SET */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET"))
COMPLETE_WITH("(", "TABLE");
+ /* ALTER PUBLICATION <name> SET TABLE <name> */
+ /* ALTER PUBLICATION <name> ADD TABLE <name> */
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET", "TABLE", MatchAny)
+ || Matches("ALTER", "PUBLICATION", MatchAny, "ADD", "TABLE", MatchAny))
+ COMPLETE_WITH("WHERE (");
/* ALTER PUBLICATION <name> SET ( */
else if (HeadMatches("ALTER", "PUBLICATION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("publish", "publish_via_partition_root");
@@ -2693,9 +2698,10 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("TABLE", "ALL TABLES");
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL"))
COMPLETE_WITH("TABLES");
- else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES")
- || Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE", MatchAny))
+ else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES"))
COMPLETE_WITH("WITH (");
+ else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE", MatchAny))
+ COMPLETE_WITH("WHERE (", "WITH (");
/* Complete "CREATE PUBLICATION <name> FOR TABLE" with "<table>, ..." */
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE"))
COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables, NULL);
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index 2885f35..ec1cb75 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -16,7 +16,6 @@
#include "catalog/objectaddress.h"
-
/*
* Precise semantics of a dependency relationship are specified by the
* DependencyType code (which is stored in a "char" field in pg_depend,
@@ -151,6 +150,14 @@ extern void performDeletion(const ObjectAddress *object,
extern void performMultipleDeletions(const ObjectAddresses *objects,
DropBehavior behavior, int flags);
+typedef struct RfCol {
+ char *name;
+ int attnum;
+} RfCol;
+extern List *rowfilter_find_cols(Node *expr, Oid relId);
+
+extern void rowfilter_validator(char *relname, Node *expr);
+
extern void recordDependencyOnExpr(const ObjectAddress *depender,
Node *expr, List *rtable,
DependencyType behavior);
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 82f2536..e5df91e 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -85,7 +85,9 @@ typedef struct Publication
typedef struct PublicationRelInfo
{
+ Oid relid;
Relation relation;
+ Node *whereClause;
} PublicationRelInfo;
extern Publication *GetPublication(Oid pubid);
@@ -116,7 +118,7 @@ extern List *GetPubPartitionOptionRelations(List *result,
Oid relid);
extern bool is_publishable_relation(Relation rel);
-extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
+extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *pri,
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 b5d5504..154bb61 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 BKI_LOOKUP(pg_publication); /* Oid of the publication */
Oid prrelid BKI_LOOKUP(pg_class); /* Oid of the relation */
+
+#ifdef CATALOG_VARLEN /* variable-length fields start here */
+ pg_node_tree prqual; /* qualifications */
+#endif
} FormData_pg_publication_rel;
/* ----------------
@@ -40,6 +44,8 @@ CATALOG(pg_publication_rel,6106,PublicationRelRelationId)
*/
typedef FormData_pg_publication_rel *Form_pg_publication_rel;
+DECLARE_TOAST(pg_publication_rel, 8287, 8288);
+
DECLARE_UNIQUE_INDEX_PKEY(pg_publication_rel_oid_index, 6112, PublicationRelObjectIndexId, on pg_publication_rel using btree(oid oid_ops));
DECLARE_UNIQUE_INDEX(pg_publication_rel_prrelid_prpubid_index, 6113, PublicationRelPrrelidPrpubidIndexId, on pg_publication_rel using btree(prrelid oid_ops, prpubid oid_ops));
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 3138877..1dfdaf3 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3640,6 +3640,7 @@ typedef struct PublicationTable
{
NodeTag type;
RangeVar *relation; /* relation to be published */
+ Node *whereClause; /* qualifications */
} PublicationTable;
typedef struct CreatePublicationStmt
@@ -3647,7 +3648,7 @@ 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;
@@ -3660,7 +3661,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 ee17908..d58ae6a 100644
--- a/src/include/parser/parse_node.h
+++ b/src/include/parser/parse_node.h
@@ -80,6 +80,7 @@ typedef enum ParseExprKind
EXPR_KIND_COPY_WHERE, /* WHERE condition in COPY FROM */
EXPR_KIND_GENERATED_COLUMN, /* generation expression for a column */
EXPR_KIND_CYCLE_MARK, /* cycle mark value */
+ EXPR_KIND_PUBLICATION_WHERE /* WHERE condition for a table in PUBLICATION */
} ParseExprKind;
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 82bce9b..0443823 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -167,6 +167,170 @@ ALTER PUBLICATION testpub_forparted DROP TABLE testpub_parted;
UPDATE testpub_parted2 SET a = 2;
DROP TABLE testpub_parted1, testpub_parted2;
DROP PUBLICATION testpub_forparted, testpub_forparted1;
+-- Test row filter for publications
+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';
+-- Firstly, test using the option publish="insert" because the row filter
+-- validation of referenced columns is less strict than for delete/update.
+CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5) WITH (publish = "insert");
+RESET client_min_messages;
+\dRp+ testpub5
+ Publication testpub5
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | f | f | f | f
+Tables:
+ "public.testpub_rf_tbl1"
+ "public.testpub_rf_tbl2" WHERE (((c <> 'test'::text) AND (d < 5)))
+
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 2000);
+\dRp+ testpub5
+ Publication testpub5
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | f | f | f | f
+Tables:
+ "public.testpub_rf_tbl1"
+ "public.testpub_rf_tbl2" WHERE (((c <> 'test'::text) AND (d < 5)))
+ "public.testpub_rf_tbl3" WHERE (((e > 1000) AND (e < 2000)))
+
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2;
+\dRp+ testpub5
+ Publication testpub5
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | f | f | f | f
+Tables:
+ "public.testpub_rf_tbl1"
+ "public.testpub_rf_tbl3" WHERE (((e > 1000) AND (e < 2000)))
+
+-- 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);
+\dRp+ testpub5
+ Publication testpub5
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | f | f | f | f
+Tables:
+ "public.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: invalid publication WHERE expression for relation "testpub_rf_tbl4"
+DETAIL: function calls are not allowed
+HINT: only simple expressions using columns and constants are allowed
+-- fail - user-defined operators disallowed
+CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+ERROR: invalid publication WHERE expression for relation "testpub_rf_tbl3"
+DETAIL: user-defined operators are not allowed
+HINT: only simple expressions using columns and constants are allowed
+-- fail - row-filter expression is not simple
+CREATE PUBLICATION testpub7 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+ERROR: invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL: too complex
+HINT: only simple expressions using columns and constants are allowed
+-- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
+ERROR: syntax error at or near "WHERE"
+LINE 1: ...R PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e <...
+ ^
+DROP TABLE testpub_rf_tbl1;
+DROP TABLE testpub_rf_tbl2;
+DROP TABLE testpub_rf_tbl3;
+DROP TABLE testpub_rf_tbl4;
+DROP PUBLICATION testpub5;
+DROP OPERATOR =#>(integer, integer);
+DROP FUNCTION testpub_rf_func(integer, integer);
+-- ======================================================
+-- More row 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));
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+-- ok - "a" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- ok - "b" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (b > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- fail - "c" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+ERROR: cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL: Row filter column "c" is not part of the REPLICA IDENTITY
+-- fail - "d" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (d > 99);
+ERROR: cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL: Row filter column "d" is not part of the REPLICA IDENTITY
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+-- fail - "a" is not part of REPLICA IDENTITY
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+ERROR: cannot add relation "rf_tbl_abcd_nopk" to publication
+DETAIL: Row filter column "a" is not part of the REPLICA IDENTITY
+-- Case 2. REPLICA IDENTITY FULL
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL;
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- ok - "a" is in REPLICA IDENTITY now
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+ERROR: cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL: Row filter column "a" is not part of the REPLICA IDENTITY
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+ERROR: cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL: Row filter column "c" is not part of the REPLICA IDENTITY
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+ERROR: cannot add relation "rf_tbl_abcd_nopk" to publication
+DETAIL: Row filter column "a" is not part of the REPLICA IDENTITY
+-- 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;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+ERROR: cannot add relation "rf_tbl_abcd_pk" to publication
+DETAIL: Row filter column "a" is not part of the REPLICA IDENTITY
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+ERROR: cannot add relation "rf_tbl_abcd_nopk" to publication
+DETAIL: Row filter column "a" is not part of the REPLICA IDENTITY
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+-- ======================================================
-- Test cache invalidation FOR ALL TABLES publication
SET client_min_messages = 'ERROR';
CREATE TABLE testpub_tbl4(a int);
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index e5745d5..209eab7 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -100,6 +100,118 @@ UPDATE testpub_parted2 SET a = 2;
DROP TABLE testpub_parted1, testpub_parted2;
DROP PUBLICATION testpub_forparted, testpub_forparted1;
+-- Test row filter for publications
+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';
+-- Firstly, test using the option publish="insert" because the row filter
+-- validation of referenced columns is less strict than for delete/update.
+CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5) WITH (publish = "insert");
+RESET client_min_messages;
+\dRp+ testpub5
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 2000);
+\dRp+ testpub5
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2;
+\dRp+ testpub5
+-- 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);
+\dRp+ testpub5
+-- fail - functions disallowed
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- fail - user-defined operators disallowed
+CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+-- fail - row-filter expression is not simple
+CREATE PUBLICATION testpub7 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+-- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
+
+DROP TABLE testpub_rf_tbl1;
+DROP TABLE testpub_rf_tbl2;
+DROP TABLE testpub_rf_tbl3;
+DROP TABLE testpub_rf_tbl4;
+DROP PUBLICATION testpub5;
+DROP OPERATOR =#>(integer, integer);
+DROP FUNCTION testpub_rf_func(integer, integer);
+
+-- ======================================================
+-- More row 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));
+
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+-- ok - "a" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- ok - "b" is a PK col
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (b > 99);
+RESET client_min_messages;
+DROP PUBLICATION testpub6;
+-- fail - "c" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "d" is not part of the PK
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (d > 99);
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+-- fail - "a" is not part of REPLICA IDENTITY
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+
+-- Case 2. REPLICA IDENTITY FULL
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL;
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- ok - "a" is in REPLICA IDENTITY now
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+
+-- 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;
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+DROP PUBLICATION testpub6;
+RESET client_min_messages;
+
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+-- ======================================================
+
-- Test cache invalidation FOR ALL TABLES publication
SET client_min_messages = 'ERROR';
CREATE TABLE testpub_tbl4(a int);
diff --git a/src/test/subscription/t/025_row_filter.pl b/src/test/subscription/t/025_row_filter.pl
new file mode 100644
index 0000000..dc9becc
--- /dev/null
+++ b/src/test/subscription/t/025_row_filter.pl
@@ -0,0 +1,299 @@
+# Test logical replication behavior with row filtering
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 7;
+
+# create publisher node
+my $node_publisher = PostgresNode->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# create subscriber node
+my $node_subscriber = PostgresNode->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# setup structure on publisher
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_1 (a int primary key, b text)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE tab_rowfilter_1 REPLICA IDENTITY FULL;");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_2 (c int primary key)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_3 (a int primary key, b boolean)");
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_partitioned (a int primary key, b integer) PARTITION BY RANGE(a)"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_less_10k (LIKE tab_rowfilter_partitioned)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE tab_rowfilter_partitioned ATTACH PARTITION tab_rowfilter_less_10k FOR VALUES FROM (MINVALUE) TO (10000)"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_greater_10k (LIKE tab_rowfilter_partitioned)"
+);
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE tab_rowfilter_partitioned ATTACH PARTITION tab_rowfilter_greater_10k FOR VALUES FROM (10000) TO (MAXVALUE)"
+);
+
+# setup structure on subscriber
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_1 (a int primary key, b text)");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_2 (c int primary key)");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_3 (a int primary key, b boolean)");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_partitioned (a int primary key, b integer) PARTITION BY RANGE(a)"
+);
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_less_10k (LIKE tab_rowfilter_partitioned)");
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE tab_rowfilter_partitioned ATTACH PARTITION tab_rowfilter_less_10k FOR VALUES FROM (MINVALUE) TO (10000)"
+);
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_greater_10k (LIKE tab_rowfilter_partitioned)"
+);
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE tab_rowfilter_partitioned ATTACH PARTITION tab_rowfilter_greater_10k FOR VALUES FROM (10000) TO (MAXVALUE)"
+);
+
+# setup logical replication
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_1 FOR TABLE tab_rowfilter_1 WHERE (a > 1000 AND b <> 'filtered')"
+);
+
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_1 ADD TABLE tab_rowfilter_2 WHERE (c % 7 = 0)"
+);
+
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_1 SET TABLE tab_rowfilter_1 WHERE (a > 1000 AND b <> 'filtered'), tab_rowfilter_2 WHERE (c % 2 = 0), tab_rowfilter_3"
+);
+
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_2 FOR TABLE tab_rowfilter_2 WHERE (c % 3 = 0)"
+);
+
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_3 FOR TABLE tab_rowfilter_partitioned WHERE (a < 5000)"
+);
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_3 ADD TABLE tab_rowfilter_less_10k WHERE (a < 6000)"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_not_used FOR TABLE tab_rowfilter_1 WHERE (a < 0)"
+);
+
+#
+# The following INSERTs are executed before the CREATE SUBSCRIPTION, so these
+# SQL commands are for testing the initial data copy using logical replication.
+#
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1, 'not replicated')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1500, 'filtered')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1980, 'not filtered')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) SELECT x, 'test ' || x FROM generate_series(990,1002) x"
+);
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_2 (c) SELECT generate_series(1, 20)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_3 (a, b) SELECT x, (x % 3 = 0) FROM generate_series(1, 10) x");
+
+# insert data into partitioned table and directly on the partition
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_partitioned (a, b) VALUES(1, 100),(7000, 101),(15000, 102),(5500, 300)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_less_10k (a, b) VALUES(2, 200),(6005, 201)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_greater_10k (a, b) VALUES(16000, 103)");
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_1, tap_pub_2, tap_pub_3"
+);
+
+$node_publisher->wait_for_catchup($appname);
+
+# wait for initial table synchronization to finish
+my $synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check expected replicated rows for tab_rowfilter_1
+# tap_pub_1 filter is: (a > 1000 AND b <> 'filtered')
+# - INSERT (1, 'not replicated') NO, because a is not > 1000
+# - INSERT (1500, 'filtered') NO, because b == 'filtered'
+# - INSERT (1980, 'not filtered') YES
+# - generate_series(990,1002) YES, only for 1001,1002 because a > 1000
+#
+my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2");
+is( $result, qq(1001|test 1001
+1002|test 1002
+1980|not filtered), 'check initial data copy from table tab_rowfilter_1');
+
+# Check expected replicated rows for tab_rowfilter_2
+# tap_pub_1 filter is: (c % 2 = 0)
+# tap_pub_2 filter is: (c % 3 = 0)
+# When there are multiple publications for the same table, all filter
+# expressions should succeed. In this case, rows are replicated if c value is
+# divided by 2 AND 3 (6, 12, 18).
+#
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(c), min(c), max(c) FROM tab_rowfilter_2");
+is($result, qq(3|6|18), 'check initial data copy from table tab_rowfilter_2');
+
+# Check expected replicated rows for tab_rowfilter_3
+# There is no filter. 10 rows are inserted, so 10 rows are replicated.
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(a) FROM tab_rowfilter_3");
+is($result, qq(10), 'check initial data copy from table tab_rowfilter_3');
+
+# Check expected replicated rows for partitions
+# publication option publish_via_partition_root is false so use the row filter
+# from a partition
+# tab_rowfilter_partitioned filter: (a < 5000)
+# tab_rowfilter_less_10k filter: (a < 6000)
+# tab_rowfilter_greater_10k filter: no filter
+#
+# INSERT into tab_rowfilter_partitioned:
+# - INSERT (1,100) YES, because 1 < 6000
+# - INSERT (7000, 101) NO, because 7000 is not < 6000
+# - INSERT (15000, 102) YES, because tab_rowfilter_greater_10k has no filter
+# - INSERT (5500, 300) YES, because 5500 < 6000
+#
+# INSERT directly into tab_rowfilter_less_10k:
+# - INSERT (2, 200) YES, because 2 < 6000
+# - INSERT (6005, 201) NO, because 6005 is not < 6000
+#
+# INSERT directly into tab_rowfilter_greater_10k:
+# - INSERT (16000, 103) YES, because tab_rowfilter_greater_10k has no filter
+#
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_less_10k ORDER BY 1, 2");
+is($result, qq(1|100
+2|200
+5500|300), 'check initial data copy from partition tab_rowfilter_less_10k');
+
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_greater_10k ORDER BY 1, 2");
+is($result, qq(15000|102
+16000|103), 'check initial data copy from partition tab_rowfilter_greater_10k');
+
+# The following commands are executed after CREATE SUBSCRIPTION, so these SQL
+# commands are for testing normal logical replication behavior.
+#
+# test row filter (INSERT, UPDATE, DELETE)
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (800, 'test 800')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1600, 'test 1600')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1601, 'test 1601')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1700, 'test 1700')");
+$node_publisher->safe_psql('postgres',
+ "UPDATE tab_rowfilter_1 SET b = NULL WHERE a = 1600");
+$node_publisher->safe_psql('postgres',
+ "UPDATE tab_rowfilter_1 SET b = 'test 1601 updated' WHERE a = 1601");
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM tab_rowfilter_1 WHERE a = 1700");
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check expected replicated rows for tab_rowfilter_1
+# tap_pub_1 filter is: (a > 1000 AND b <> 'filtered')
+#
+# - 1001, 1002, 1980 already exist from initial data copy
+# - INSERT (800, 'test 800') NO, because 800 is not > 1000
+# - INSERT (1600, 'test 1600') YES, because 1600 > 1000 and 'test 1600' <> 'filtered'
+# - INSERT (1601, 'test 1601') YES, because 1601 > 1000 and 'test 1601' <> 'filtered'
+# - INSERT (1700, 'test 1700') YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
+# - UPDATE (1600, NULL) NO, row filter evaluates to false because NULL is not <> 'filtered'
+# - UPDATE (1601, 'test 1601 updated') YES, because 1601 > 1000 and 'test 1601 updated' <> 'filtered'
+# - DELETE (1700) YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
+#
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2");
+is($result, qq(1001|test 1001
+1002|test 1002
+1600|test 1600
+1601|test 1601 updated
+1980|not filtered), 'check replicated rows to table tab_rowfilter_1');
+
+# Publish using root partitioned table
+# Use a different partitioned table layout (exercise publish_via_partition_root)
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_3 SET (publish_via_partition_root = true)");
+$node_subscriber->safe_psql('postgres',
+ "TRUNCATE TABLE tab_rowfilter_partitioned");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION WITH (copy_data = true)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_partitioned (a, b) VALUES(4000, 400),(4001, 401),(4002, 402)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_less_10k (a, b) VALUES(4500, 450)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_less_10k (a, b) VALUES(5600, 123)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_greater_10k (a, b) VALUES(14000, 1950)");
+$node_publisher->safe_psql('postgres',
+ "UPDATE tab_rowfilter_less_10k SET b = 30 WHERE a = 4001");
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM tab_rowfilter_less_10k WHERE a = 4002");
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check expected replicated rows for partitions
+# publication option publish_via_partition_root is true so use the row filter
+# from the root partitioned table
+# tab_rowfilter_partitioned filter: (a < 5000)
+# tab_rowfilter_less_10k filter: (a < 6000)
+# tab_rowfilter_greater_10k filter: no filter
+#
+# After TRUNCATE, REFRESH PUBLICATION, the initial data copy will apply the
+# partitioned table row filter.
+# - INSERT (1, 100) YES, 1 < 5000
+# - INSERT (7000, 101) NO, 7000 is not < 5000
+# - INSERT (15000, 102) NO, 15000 is not < 5000
+# - INSERT (5500, 300) NO, 5500 is not < 5000
+# - INSERT (2, 200) YES, 2 < 5000
+# - INSERT (6005, 201) NO, 6005 is not < 5000
+# - INSERT (16000, 103) NO, 16000 is not < 5000
+#
+# Execute SQL commands after initial data copy for testing the logical
+# replication behavior.
+# - INSERT (4000, 400) YES, 4000 < 5000
+# - INSERT (4001, 401) YES, 4001 < 5000
+# - INSERT (4002, 402) YES, 4002 < 5000
+# - INSERT (4500, 450) YES, 4500 < 5000
+# - INSERT (5600, 123) NO, 5600 is not < 5000
+# - INSERT (14000, 1950) NO, 16000 is not < 5000
+# - UPDATE (4001) YES, 4001 < 5000
+# - DELETE (4002) YES, 4002 < 5000
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_partitioned ORDER BY 1, 2");
+is( $result, qq(1|100
+2|200
+4000|400
+4001|30
+4500|450), 'check publish_via_partition_root behavior');
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
--
1.8.3.1