v66-0001-Allow-specifying-row-filter-for-logical-replication-.patch
application/octet-stream
Filename: v66-0001-Allow-specifying-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 v66-0001
Subject: Allow specifying row filter for logical replication of tables.
| File | + | − |
|---|---|---|
| doc/src/sgml/catalogs.sgml | 9 | 0 |
| doc/src/sgml/ref/alter_publication.sgml | 10 | 3 |
| doc/src/sgml/ref/alter_subscription.sgml | 5 | 2 |
| doc/src/sgml/ref/create_publication.sgml | 32 | 1 |
| doc/src/sgml/ref/create_subscription.sgml | 25 | 1 |
| src/backend/catalog/pg_publication.c | 44 | 4 |
| src/backend/commands/publicationcmds.c | 433 | 16 |
| src/backend/executor/execReplication.c | 33 | 2 |
| src/backend/nodes/copyfuncs.c | 1 | 0 |
| src/backend/nodes/equalfuncs.c | 1 | 0 |
| src/backend/parser/gram.y | 32 | 6 |
| src/backend/replication/logical/proto.c | 18 | 18 |
| src/backend/replication/logical/tablesync.c | 124 | 7 |
| src/backend/replication/pgoutput/pgoutput.c | 723 | 64 |
| src/backend/utils/cache/relcache.c | 79 | 22 |
| src/bin/psql/describe.c | 24 | 2 |
| src/include/catalog/pg_publication.h | 3 | 2 |
| src/include/catalog/pg_publication_rel.h | 6 | 0 |
| src/include/commands/publicationcmds.h | 3 | 0 |
| src/include/nodes/parsenodes.h | 1 | 0 |
| src/include/replication/logicalproto.h | 7 | 4 |
| src/include/replication/reorderbuffer.h | 3 | 3 |
| src/include/utils/relcache.h | 1 | 0 |
| src/include/utils/rel.h | 7 | 0 |
| src/test/regress/expected/publication.out | 301 | 0 |
| src/test/regress/sql/publication.sql | 206 | 0 |
| src/test/subscription/t/027_row_filter.pl | 509 | 0 |
| src/tools/pgindent/typedefs.list | 2 | 0 |
From e84fde61265d261658ed7a5838e7773392723e5b Mon Sep 17 00:00:00 2001
From: "houzj.fnst" <houzj.fnst@cn.fujitsu.com>
Date: Thu, 13 Jan 2022 17:26:52 +0800
Subject: [PATCH] Allow specifying row filter for logical replication of
tables.
This feature adds row filtering 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 row filter WHERE clause for a table added to a publication that
publishes UPDATE and/or DELETE operations must contain only columns that
are covered by REPLICA IDENTITY. The row filter WHERE clause for a table
added to a publication that publishes INSERT can use any column. If the
row filter evaluates to NULL, it returns false. The WHERE clause allows
simple expressions that don't have user-defined functions, operators, or
non-immutable built-in functions. These restrictions could possibly be
addressed in the future.
If you choose to do the initial table synchronization, only data that
satisfies the row filters is pulled by the subscriber. If the subscription
has several publications in which a table has been published with
different WHERE clauses, rows that satisfy ANY of the expressions will be
copied. If a subscriber is a pre-15 version, the initial table
synchronization won't use row filters even if they are defined in the
publisher.
The row filters are applied before publishing the changes. If the
subscription has several publications in which the same table has been
published with different filters, those expressions get OR'ed together so
that rows satisfying any of the expressions will be replicated.
This means all the other filters become redundant if (a) one of the
publications have no filter at all, (b) one of the publications was
created using FOR ALL TABLES, (c) one of the publications was created
using FOR ALL TABLES IN SCHEMA and the table belongs to that same schema.
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.
Psql commands \dRp+ and \d+ will display any row filters.
Author: Euler Taveira, Peter Smith, Hou Zhijie, Ajin Cherian
Reviewed-by: Greg Nancarrow, Haiying Tang, Amit Kapila, Tomas Vondra, Dilip Kumar, Vignesh C, Alvaro Herrera
Discussion: https://www.postgresql.org/message-id/flat/CAHE3wggb715X%2BmK_DitLXF25B%3DjE6xyNCH4YOwM860JR7HarGQ%40mail.gmail.com
---
doc/src/sgml/catalogs.sgml | 9 +
doc/src/sgml/ref/alter_publication.sgml | 13 +-
doc/src/sgml/ref/alter_subscription.sgml | 7 +-
doc/src/sgml/ref/create_publication.sgml | 33 +-
doc/src/sgml/ref/create_subscription.sgml | 26 +-
src/backend/catalog/pg_publication.c | 48 +-
src/backend/commands/publicationcmds.c | 449 +++++++++++++++-
src/backend/executor/execReplication.c | 35 +-
src/backend/nodes/copyfuncs.c | 1 +
src/backend/nodes/equalfuncs.c | 1 +
src/backend/parser/gram.y | 38 +-
src/backend/replication/logical/proto.c | 36 +-
src/backend/replication/logical/tablesync.c | 131 ++++-
src/backend/replication/pgoutput/pgoutput.c | 787 +++++++++++++++++++++++++---
src/backend/utils/cache/relcache.c | 101 +++-
src/bin/psql/describe.c | 26 +-
src/include/catalog/pg_publication.h | 5 +-
src/include/catalog/pg_publication_rel.h | 6 +
src/include/commands/publicationcmds.h | 3 +
src/include/nodes/parsenodes.h | 1 +
src/include/replication/logicalproto.h | 11 +-
src/include/replication/reorderbuffer.h | 6 +-
src/include/utils/rel.h | 7 +
src/include/utils/relcache.h | 1 +
src/test/regress/expected/publication.out | 301 +++++++++++
src/test/regress/sql/publication.sql | 206 ++++++++
src/test/subscription/t/027_row_filter.pl | 509 ++++++++++++++++++
src/tools/pgindent/typedefs.list | 2 +
28 files changed, 2642 insertions(+), 157 deletions(-)
create mode 100644 src/test/subscription/t/027_row_filter.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 2aeb2ef..17cb730 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -6311,6 +6311,15 @@ 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. Null if
+ there is no 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 bb4ef5e..bb58e76 100644
--- a/doc/src/sgml/ref/alter_publication.sgml
+++ b/doc/src/sgml/ref/alter_publication.sgml
@@ -30,7 +30,7 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
<phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase>
- TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ... ]
+ TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ... ]
ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
</synopsis>
</refsynopsisdiv>
@@ -52,7 +52,9 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
remove one or more tables/schemas from the publication. Note that adding
tables/schemas to a publication that is already subscribed to will require a
<literal>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</literal> action on the
- subscribing side in order to become effective.
+ subscribing side in order to become effective. Note also that the combination
+ of <literal>DROP</literal> with <literal>WHERE</literal> clause is not
+ allowed.
</para>
<para>
@@ -109,7 +111,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 for which the
+ <replaceable class="parameter">expression</replaceable> returns false or null 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/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 0b027cc..27442b1 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -163,8 +163,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<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>. (Previously-subscribed
- tables are not copied.)
+ The default is <literal>true</literal>.
+ </para>
+ <para>
+ Previously-subscribed tables are not copied, even if a table's row
+ filter <literal>WHERE</literal> clause had been modified.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index d805e8e..7eb12ee 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -28,7 +28,7 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
<phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase>
- TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [, ... ]
+ TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ... ]
ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
</synopsis>
</refsynopsisdiv>
@@ -79,6 +79,14 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
</para>
<para>
+ If the optional <literal>WHERE</literal> clause is specified, rows for
+ which the <replaceable class="parameter">expression</replaceable> returns
+ false or null will not be published. Note that parentheses are required
+ around the expression. It has no effect on <literal>TRUNCATE</literal>
+ commands.
+ </para>
+
+ <para>
Only persistent base tables and partitioned tables can be part of a
publication. Temporary tables, unlogged tables, foreign tables,
materialized views, and regular views cannot be part of a publication.
@@ -226,6 +234,17 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
</para>
<para>
+ A <literal>WHERE</literal> clause must contain only columns that are
+ part of the primary key or are covered by the <literal>REPLICA
+ IDENTITY</literal>, in order for <command>UPDATE</command> and
+ <command>DELETE</command> operations to be published.
+ For publication of <command>INSERT</command> operations, any column
+ may be used in the <literal>WHERE</literal> clause.
+ The <literal>WHERE</literal> clause allows simple expressions that don't have
+ user-defined functions, operators, or non-immutable built-in functions.
+ </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
@@ -247,6 +266,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>
@@ -260,6 +284,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..3473b13 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -208,6 +208,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
that are being subscribed to when the replication starts.
The default is <literal>true</literal>.
</para>
+ <para>
+ If the publications contain conditional expressions, it will affect
+ what data is copied. Refer to the
+ <xref linkend="sql-createsubscription-notes" /> for details.
+ </para>
</listitem>
</varlistentry>
@@ -293,7 +298,7 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</variablelist>
</refsect1>
- <refsect1>
+ <refsect1 id="sql-createsubscription-notes" xreflabel="Notes">
<title>Notes</title>
<para>
@@ -319,6 +324,25 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
the parameter <literal>create_slot = false</literal>. This is an
implementation restriction that might be lifted in a future release.
</para>
+
+ <para>
+ If any table in the publication has a <literal>WHERE</literal> clause, rows
+ for which the <replaceable class="parameter">expression</replaceable>
+ returns false or null will not be published. If the subscription has several
+ publications in which the same table has been published with different
+ <literal>WHERE</literal> clauses, a row will be published if any of the
+ expressions (referring to that publish operation) are satisfied. In the case
+ of different <literal>WHERE</literal> clauses, if one of the publications
+ has no <literal>WHERE</literal> clause (referring to that publish operation)
+ or the publication is declared as <literal>FOR ALL TABLES</literal> or
+ <literal>FOR ALL TABLES IN SCHEMA</literal>, rows are always published
+ regardless of the definition of the other expressions.
+ If the subscriber is a <productname>PostgreSQL</productname> version before
+ 15 then any row filtering is ignored during the initial data synchronization
+ phase. For this case, the user might want to consider deleting any initially
+ copied data that would be incompatible with subsequent filtering.
+ </para>
+
</refsect1>
<refsect1>
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index cf0700f..cabcf09 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -276,17 +276,45 @@ GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
}
/*
+ * Returns the relid of the topmost ancestor that is published via this
+ * publication if any, otherwise return InvalidOid.
+ */
+Oid
+GetTopMostAncestorInPublication(Oid puboid, List *ancestors)
+{
+ ListCell *lc;
+ Oid topmost_relid = InvalidOid;
+
+ /*
+ * Find the "topmost" ancestor that is in this publication.
+ */
+ foreach(lc, ancestors)
+ {
+ Oid ancestor = lfirst_oid(lc);
+
+ if (list_member_oid(GetRelationPublications(ancestor),
+ puboid) ||
+ list_member_oid(GetSchemaPublications(get_rel_namespace(ancestor)),
+ puboid))
+ topmost_relid = ancestor;
+ }
+
+ return topmost_relid;
+}
+
+/*
* 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 = RelationGetRelid(targetrel);
Oid pubreloid;
Publication *pub = GetPublication(pubid);
ObjectAddress myself,
@@ -311,10 +339,10 @@ 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);
/* Form a tuple. */
memset(values, 0, sizeof(values));
@@ -328,6 +356,12 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
values[Anum_pg_publication_rel_prrelid - 1] =
ObjectIdGetDatum(relid);
+ /* Add qualifications, if available */
+ if (pri->whereClause != NULL)
+ values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(pri->whereClause));
+ else
+ nulls[Anum_pg_publication_rel_prqual - 1] = true;
+
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
/* Insert tuple into catalog. */
@@ -345,6 +379,12 @@ 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 (pri->whereClause)
+ recordDependencyOnSingleRelExpr(&myself, pri->whereClause, relid,
+ DEPENDENCY_NORMAL, DEPENDENCY_NORMAL,
+ false);
+
/* Close the table. */
table_close(rel, RowExclusiveLock);
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 3ab1bde..072d608 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -26,6 +26,7 @@
#include "catalog/partition.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
+#include "catalog/pg_proc.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_namespace.h"
#include "catalog/pg_publication_rel.h"
@@ -36,6 +37,10 @@
#include "commands/publicationcmds.h"
#include "funcapi.h"
#include "miscadmin.h"
+#include "nodes/nodeFuncs.h"
+#include "parser/parse_clause.h"
+#include "parser/parse_collate.h"
+#include "parser/parse_relation.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
#include "utils/array.h"
@@ -48,6 +53,20 @@
#include "utils/syscache.h"
#include "utils/varlena.h"
+/*
+ * Information used to validate the columns in the row filter expression. See
+ * contain_invalid_rfcolumn_walker for details.
+ */
+typedef struct rf_context
+{
+ AttrNumber invalid_rfcolnum; /* invalid column number */
+ Bitmapset *bms_replident; /* bitset of replica identity columns */
+ bool pubviaroot; /* true if we are validating the parent
+ * relation's row filter */
+ Oid relid; /* relid of the relation */
+ Oid parentid; /* relid of the parent relation */
+} rf_context;
+
static List *OpenRelIdList(List *relids);
static List *OpenTableList(List *tables);
static void CloseTableList(List *rels);
@@ -235,6 +254,337 @@ CheckObjSchemaNotAlreadyInPublication(List *rels, List *schemaidlist,
}
/*
+ * Returns true, if any of the columns used in the row filter WHERE clause are
+ * not part of REPLICA IDENTITY, false, otherwise.
+ *
+ * Remember the invalid column number if there is any, to be later used in
+ * error message.
+ */
+static bool
+contain_invalid_rfcolumn_walker(Node *node, rf_context *context)
+{
+ if (node == NULL)
+ return false;
+
+ if (IsA(node, Var))
+ {
+ Var *var = (Var *) node;
+ AttrNumber attnum = var->varattno;
+
+ /*
+ * If pubviaroot is true, we are validating the row filter of the
+ * parent table, but the bitmap contains the replica identity
+ * information of the child table. So, get the column number of child
+ * table as parent and child column order could be different.
+ */
+ if (context->pubviaroot)
+ {
+ char *colname = get_attname(context->parentid, attnum, false);
+
+ attnum = get_attnum(context->relid, colname);
+ }
+
+ if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
+ context->bms_replident))
+ {
+ context->invalid_rfcolnum = attnum;
+ return true;
+ }
+ }
+
+ return expression_tree_walker(node, contain_invalid_rfcolumn_walker,
+ (void *) context);
+}
+
+/*
+ * Check, if all columns referenced in the filter expression are part of the
+ * REPLICA IDENTITY index or not.
+ *
+ * Returns true if any invalid column is found and store the invalid column
+ * number in invalid_rfcolumn.
+ */
+bool
+contain_invalid_rfcolumn(Oid pubid, Relation relation, List *ancestors,
+ AttrNumber *invalid_rfcolumn)
+{
+ HeapTuple rftuple;
+ Oid relid = RelationGetRelid(relation);
+ Oid publish_as_relid = RelationGetRelid(relation);
+ bool result = false;
+ Datum rfdatum;
+ bool rfisnull;
+ Publication *pub;
+
+ /*
+ * FULL means all cols are in the REPLICA IDENTITY, so all cols are
+ * allowed in the row filter and we can skip the validation.
+ */
+ if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
+ return false;
+
+ pub = GetPublication(pubid);
+
+ /*
+ * For a partition, if pubviaroot is true, find the topmost
+ * ancestor that is published via this publication as we need to
+ * use its row filter expression to filter the partition's changes.
+ */
+ if (pub->pubviaroot && relation->rd_rel->relispartition)
+ {
+ if (pub->alltables)
+ publish_as_relid = llast_oid(ancestors);
+ else
+ {
+ publish_as_relid = GetTopMostAncestorInPublication(pubid,
+ ancestors);
+ if (publish_as_relid == InvalidOid)
+ publish_as_relid = relid;
+ }
+ }
+
+ rftuple = SearchSysCache2(PUBLICATIONRELMAP,
+ ObjectIdGetDatum(publish_as_relid),
+ ObjectIdGetDatum(pubid));
+
+ if (!HeapTupleIsValid(rftuple))
+ return false;
+
+ rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
+ Anum_pg_publication_rel_prqual,
+ &rfisnull);
+
+ if (!rfisnull)
+ {
+ rf_context context = {0};
+ Node *rfnode;
+ Bitmapset *bms = NULL;
+
+ context.invalid_rfcolnum = InvalidAttrNumber;
+ context.pubviaroot = pub->pubviaroot;
+ context.parentid = publish_as_relid;
+ context.relid = relid;
+
+ /*
+ * Remember columns that are part of the REPLICA IDENTITY. Note that
+ * REPLICA IDENTITY DEFAULT means primary key or nothing.
+ */
+ if (relation->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT)
+ bms = RelationGetIndexAttrBitmap(relation,
+ INDEX_ATTR_BITMAP_PRIMARY_KEY);
+ else if (relation->rd_rel->relreplident == REPLICA_IDENTITY_INDEX)
+ bms = RelationGetIndexAttrBitmap(relation,
+ INDEX_ATTR_BITMAP_IDENTITY_KEY);
+
+ context.bms_replident = bms;
+ rfnode = stringToNode(TextDatumGetCString(rfdatum));
+ result = contain_invalid_rfcolumn_walker(rfnode, &context);
+
+ *invalid_rfcolumn = context.invalid_rfcolnum;
+
+ bms_free(bms);
+ pfree(rfnode);
+ }
+
+ ReleaseSysCache(rftuple);
+
+ return result;
+}
+
+/*
+ * Is this a simple Node permitted within a row filter expression?
+ */
+static bool
+IsRowFilterSimpleExpr(Node *node)
+{
+ switch (nodeTag(node))
+ {
+ case T_ArrayExpr:
+ case T_BooleanTest:
+ case T_BoolExpr:
+ case T_CaseExpr:
+ case T_CaseTestExpr:
+ case T_CoalesceExpr:
+ case T_Const:
+ case T_List:
+ case T_MinMaxExpr:
+ case T_NullIfExpr:
+ case T_NullTest:
+ case T_ScalarArrayOpExpr:
+ case T_XmlExpr:
+ return true;
+ default:
+ return false;
+ }
+}
+
+/*
+ * The row filter walker checks if the row filter expression is a "simple
+ * expression".
+ *
+ * It allows only simple or compound expressions such as:
+ * - (Var Op Const)
+ * - (Var Op Var)
+ * - (Var Op Const) Bool (Var Op Const)
+ * - etc
+ * (where Var is a column of the table this filter belongs to)
+ *
+ * The simple expression contains the following restrictions:
+ * - User-defined operators are not allowed;
+ * - User-defined functions are not allowed;
+ * - User-defined types are not allowed;
+ * - Non-immutable built-in functions are not allowed;
+ * - System columns are not allowed.
+ *
+ * NOTES
+ *
+ * We don't allow user-defined functions/operators/types because
+ * (a) if a user drops a user-defined object used in a row filter expression or
+ * if there is any other error while using it, the logical decoding
+ * infrastructure won't be able to recover from such an error even if the
+ * object is recreated again because a historic snapshot is used to evaluate
+ * the row filter;
+ * (b) a user-defined function can be used to access tables which could have
+ * unpleasant results because a historic snapshot is used. That's why only
+ * immutable built-in functions are allowed in row filter expressions.
+ */
+static bool
+check_simple_rowfilter_expr_walker(Node *node, Relation relation)
+{
+ char *errdetail_msg = NULL;
+
+ if (node == NULL)
+ return false;
+
+ if (IsRowFilterSimpleExpr(node))
+ {
+ /* OK, node is part of simple expressions */
+ }
+ else if (IsA(node, Var))
+ {
+ Var *var = (Var *) node;
+
+ /* User-defined types are not allowed. */
+ if (var->vartype >= FirstNormalObjectId)
+ errdetail_msg = _("User-defined types are not allowed.");
+
+ /* System columns are not allowed. */
+ else if (var->varattno < InvalidAttrNumber)
+ {
+ Oid relid = RelationGetRelid(relation);
+ const char *colname = get_attname(relid, var->varattno, false);
+
+ errdetail_msg = psprintf(_("Cannot use system column (%s)."), colname);
+ }
+ }
+ else if (IsA(node, OpExpr))
+ {
+ /* OK, except user-defined operators are not allowed. */
+ if (((OpExpr *) node)->opno >= FirstNormalObjectId)
+ errdetail_msg = _("User-defined operators are not allowed.");
+ }
+ else if (IsA(node, FuncExpr))
+ {
+ Oid funcid = ((FuncExpr *) node)->funcid;
+ const char *funcname = get_func_name(funcid);
+
+ /*
+ * User-defined functions are not allowed. System-functions that are
+ * not IMMUTABLE are not allowed.
+ */
+ if (funcid >= FirstNormalObjectId)
+ errdetail_msg = psprintf(_("User-defined functions are not allowed (%s)."),
+ funcname);
+ else if (func_volatile(funcid) != PROVOLATILE_IMMUTABLE)
+ errdetail_msg = psprintf(_("Non-immutable built-in functions are not allowed (%s)."),
+ funcname);
+ }
+ else
+ {
+ elog(DEBUG3, "row filter contains an unexpected expression component: %s", nodeToString(node));
+
+ ereport(ERROR,
+ (errmsg("invalid publication WHERE expression for relation \"%s\"",
+ RelationGetRelationName(relation)),
+ errdetail("Expressions only allow columns, constants, built-in operators, built-in data types and non-immutable built-in functions.")
+ ));
+ }
+
+ if (errdetail_msg)
+ ereport(ERROR,
+ (errmsg("invalid publication WHERE expression for relation \"%s\"",
+ RelationGetRelationName(relation)),
+ errdetail("%s", errdetail_msg)
+ ));
+
+ return expression_tree_walker(node, check_simple_rowfilter_expr_walker,
+ (void *) relation);
+}
+
+/*
+ * Check, if the row filter expression is a "simple expression".
+ *
+ * See check_simple_rowfilter_expr_walker for details.
+ */
+static bool
+check_simple_rowfilter_expr(Node *node, Relation relation)
+{
+ return check_simple_rowfilter_expr_walker(node, relation);
+}
+
+/*
+ * Transform the publication WHERE clause for all the relations in list,
+ * ensuring it is coerced to boolean and necessary collation information is
+ * added if required, and add a new nsitem/RTE for the associated relation to
+ * the ParseState's namespace list.
+ *
+ * Also check the publication row filter expression and throw an error if
+ * anything not permitted or unexpected is encountered.
+ */
+static void
+TransformPubWhereClauses(List *tables, const char *queryString)
+{
+ ListCell *lc;
+
+ foreach(lc, tables)
+ {
+ ParseNamespaceItem *nsitem;
+ Node *whereclause = NULL;
+ ParseState *pstate;
+ PublicationRelInfo *pri = (PublicationRelInfo *) lfirst(lc);
+
+ if (pri->whereClause == NULL)
+ continue;
+
+ pstate = make_parsestate(NULL);
+ pstate->p_sourcetext = queryString;
+
+ nsitem = addRangeTableEntryForRelation(pstate, pri->relation,
+ AccessShareLock, NULL,
+ false, false);
+
+ addNSItemToQuery(pstate, nsitem, false, true, true);
+
+ whereclause = transformWhereClause(pstate,
+ copyObject(pri->whereClause),
+ EXPR_KIND_WHERE,
+ "PUBLICATION WHERE");
+
+ /* Fix up collation information */
+ assign_expr_collations(pstate, whereclause);
+
+ /*
+ * We allow only simple expressions in row filters. See
+ * check_simple_rowfilter_expr_walker.
+ */
+ check_simple_rowfilter_expr(whereclause, pri->relation);
+
+ free_parsestate(pstate);
+
+ pri->whereClause = whereclause;
+ }
+}
+
+/*
* Create new publication.
*/
ObjectAddress
@@ -344,8 +694,12 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
List *rels;
rels = OpenTableList(relations);
+
CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
PUBLICATIONOBJ_TABLE);
+
+ TransformPubWhereClauses(rels, pstate->p_sourcetext);
+
PublicationAddTables(puboid, rels, true, NULL);
CloseTableList(rels);
}
@@ -492,7 +846,8 @@ InvalidatePublicationRels(List *relids)
*/
static void
AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
- List *tables, List *schemaidlist)
+ List *tables, List *schemaidlist,
+ const char *queryString)
{
List *rels = NIL;
Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
@@ -519,6 +874,9 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
schemas = list_concat_copy(schemaidlist, GetPublicationSchemas(pubid));
CheckObjSchemaNotAlreadyInPublication(rels, schemas,
PUBLICATIONOBJ_TABLE);
+
+ TransformPubWhereClauses(rels, queryString);
+
PublicationAddTables(pubid, rels, false, stmt);
}
else if (stmt->action == AP_DropObjects)
@@ -533,37 +891,73 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
PUBLICATIONOBJ_TABLE);
- /* Calculate which relations to drop. */
+ TransformPubWhereClauses(rels, queryString);
+
+ /*
+ * In order to recreate the relation list for the publication, look
+ * for existing relations that do not need to be dropped.
+ */
foreach(oldlc, oldrelids)
{
Oid oldrelid = lfirst_oid(oldlc);
ListCell *newlc;
+ PublicationRelInfo *oldrel;
bool found = false;
+ HeapTuple rftuple;
+ bool rfisnull = true;
+ Node *oldrelwhereclause = NULL;
+
+ /* look up the cache for the old relmap */
+ rftuple = SearchSysCache2(PUBLICATIONRELMAP, ObjectIdGetDatum(oldrelid),
+ ObjectIdGetDatum(pubid));
+ if (HeapTupleIsValid(rftuple))
+ {
+ Datum whereClauseDatum;
+
+ whereClauseDatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple, Anum_pg_publication_rel_prqual,
+ &rfisnull);
+ if (!rfisnull)
+ oldrelwhereclause = stringToNode(TextDatumGetCString(whereClauseDatum));
+
+ ReleaseSysCache(rftuple);
+ }
foreach(newlc, rels)
{
PublicationRelInfo *newpubrel;
newpubrel = (PublicationRelInfo *) lfirst(newlc);
+
+ /*
+ * Check if any of the new set of relations match with the
+ * existing relations in the publication. Additionally, if the
+ * relation has an associated where-clause, check the
+ * where-clauses also match. Drop the rest.
+ */
if (RelationGetRelid(newpubrel->relation) == oldrelid)
{
- found = true;
- break;
+ if (equal(oldrelwhereclause, newpubrel->whereClause))
+ {
+ 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;
+ if (oldrelwhereclause)
+ pfree(oldrelwhereclause);
- delrels = lappend(delrels, pubrel);
+ /*
+ * Add the non-matched relations to a list so that they can be
+ * dropped.
+ */
+ if (!found)
+ {
+ oldrel = palloc(sizeof(PublicationRelInfo));
+ oldrel->whereClause = NULL;
+ oldrel->relation = table_open(oldrelid,
+ ShareUpdateExclusiveLock);
+ delrels = lappend(delrels, oldrel);
}
}
@@ -749,7 +1143,8 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
errmsg("publication \"%s\" does not exist",
stmt->pubname));
- AlterPublicationTables(stmt, tup, relations, schemaidlist);
+ AlterPublicationTables(stmt, tup, relations, schemaidlist,
+ pstate->p_sourcetext);
AlterPublicationSchemas(stmt, tup, schemaidlist);
}
@@ -901,6 +1296,7 @@ OpenTableList(List *tables)
List *relids = NIL;
List *rels = NIL;
ListCell *lc;
+ List *relids_with_rf = NIL;
/*
* Open, share-lock, and check all the explicitly-specified relations
@@ -928,15 +1324,26 @@ OpenTableList(List *tables)
*/
if (list_member_oid(relids, myrelid))
{
+ /* Disallow duplicate tables if there are any with row filters. */
+ if (t->whereClause || list_member_oid(relids_with_rf, myrelid))
+ ereport(ERROR,
+ (errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
+ RelationGetRelationName(rel))));
+
table_close(rel, ShareUpdateExclusiveLock);
continue;
}
pub_rel = palloc(sizeof(PublicationRelInfo));
pub_rel->relation = rel;
+ pub_rel->whereClause = t->whereClause;
rels = lappend(rels, pub_rel);
relids = lappend_oid(relids, myrelid);
+ if (t->whereClause)
+ relids_with_rf = lappend_oid(relids_with_rf, myrelid);
+
/*
* Add children of this rel, if requested, so that they too are added
* to the publication. A partitioned table can't have any inheritance
@@ -969,6 +1376,8 @@ OpenTableList(List *tables)
rel = table_open(childrelid, NoLock);
pub_rel = palloc(sizeof(PublicationRelInfo));
pub_rel->relation = rel;
+ /* child inherits WHERE clause from parent */
+ pub_rel->whereClause = t->whereClause;
rels = lappend(rels, pub_rel);
relids = lappend_oid(relids, childrelid);
}
@@ -976,6 +1385,7 @@ OpenTableList(List *tables)
}
list_free(relids);
+ list_free(relids_with_rf);
return rels;
}
@@ -995,6 +1405,8 @@ CloseTableList(List *rels)
pub_rel = (PublicationRelInfo *) lfirst(lc);
table_close(pub_rel->relation, NoLock);
}
+
+ list_free_deep(rels);
}
/*
@@ -1090,6 +1502,11 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
RelationGetRelationName(rel))));
}
+ if (pubrel->whereClause)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("cannot use a WHERE clause when removing a table from a publication")));
+
ObjectAddressSet(obj, PublicationRelRelationId, prid);
performDeletion(&obj, DROP_CASCADE, 0);
}
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 313c873..5e760af 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -568,14 +568,45 @@ void
CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
{
PublicationActions *pubactions;
+ AttrNumber invalid_rf_column;
/* We only need to do checks for UPDATE and DELETE. */
if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
return;
+ if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
+ return;
+
+ /*
+ * It is only safe to execute UPDATE/DELETE when all columns, referenced in
+ * the row filters from publications which the relation is in, are valid -
+ * i.e. when all referenced columns are part of REPLICA IDENTITY or the
+ * table does not publish UPDATES or DELETES.
+ */
+ invalid_rf_column = GetRelationPublicationInfo(rel, true);
+ if (AttributeNumberIsValid(invalid_rf_column))
+ {
+ const char *colname = get_attname(RelationGetRelid(rel),
+ invalid_rf_column, false);
+
+ if (cmd == CMD_UPDATE)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+ errmsg("cannot update table \"%s\"",
+ RelationGetRelationName(rel)),
+ errdetail("Column \"%s\" used in the publication WHERE expression is not part of the replica identity.",
+ colname)));
+ else if (cmd == CMD_DELETE)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+ errmsg("cannot delete from table \"%s\"",
+ RelationGetRelationName(rel)),
+ errdetail("Column \"%s\" used in the publication WHERE expression is not part of the replica identity.",
+ colname)));
+ }
+
/* If relation has replica identity we are always good. */
- if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
- OidIsValid(RelationGetReplicaIndex(rel)))
+ if (OidIsValid(RelationGetReplicaIndex(rel)))
return;
/*
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index b105c26..77a2a2e 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4834,6 +4834,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 ae37ea9..342c007 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2312,6 +2312,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 bb015a8..413515c 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9740,12 +9740,13 @@ CreatePublicationStmt:
* relation_expr here.
*/
PublicationObjSpec:
- TABLE relation_expr
+ TABLE relation_expr OptWhereClause
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_TABLE;
$$->pubtable = makeNode(PublicationTable);
$$->pubtable->relation = $2;
+ $$->pubtable->whereClause = $3;
}
| ALL TABLES IN_P SCHEMA ColId
{
@@ -9760,28 +9761,45 @@ PublicationObjSpec:
$$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA;
$$->location = @5;
}
- | ColId
+ | ColId OptWhereClause
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
- $$->name = $1;
+ if ($2)
+ {
+ /*
+ * The OptWhereClause must be stored here but it is
+ * valid only for tables. For non-table objects, an
+ * error will be thrown later via
+ * preprocess_pubobj_list().
+ */
+ $$->pubtable = makeNode(PublicationTable);
+ $$->pubtable->relation = makeRangeVar(NULL, $1, @1);
+ $$->pubtable->whereClause = $2;
+ }
+ else
+ {
+ $$->name = $1;
+ }
$$->location = @1;
}
- | ColId indirection
+ | ColId indirection OptWhereClause
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
$$->pubtable = makeNode(PublicationTable);
$$->pubtable->relation = makeRangeVarFromQualifiedName($1, $2, @1, yyscanner);
+ $$->pubtable->whereClause = $3;
$$->location = @1;
}
/* grammar like tablename * , ONLY tablename, ONLY ( tablename ) */
- | extended_relation_expr
+ | extended_relation_expr OptWhereClause
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
$$->pubtable = makeNode(PublicationTable);
$$->pubtable->relation = $1;
+ $$->pubtable->whereClause = $2;
}
| CURRENT_SCHEMA
{
@@ -17430,7 +17448,8 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("invalid table name at or near"),
parser_errposition(pubobj->location));
- else if (pubobj->name)
+
+ if (pubobj->name)
{
/* convert it to PublicationTable */
PublicationTable *pubtable = makeNode(PublicationTable);
@@ -17444,6 +17463,13 @@ preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t yyscanner)
else if (pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_SCHEMA ||
pubobj->pubobjtype == PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA)
{
+ /* WHERE clause is not allowed on a schema object */
+ if (pubobj->pubtable && pubobj->pubtable->whereClause)
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("WHERE clause not allowed for schema"),
+ parser_errposition(pubobj->location));
+
/*
* We can distinguish between the different type of schema
* objects based on whether name and pubtable is set.
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 9539426..c9b0eee 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -31,8 +31,8 @@
static void logicalrep_write_attrs(StringInfo out, Relation rel);
static void logicalrep_write_tuple(StringInfo out, Relation rel,
- HeapTuple tuple, bool binary);
-
+ TupleTableSlot *slot,
+ bool binary);
static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel);
static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple);
@@ -398,7 +398,7 @@ logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn)
*/
void
logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel,
- HeapTuple newtuple, bool binary)
+ TupleTableSlot *newslot, bool binary)
{
pq_sendbyte(out, LOGICAL_REP_MSG_INSERT);
@@ -410,7 +410,7 @@ logicalrep_write_insert(StringInfo out, TransactionId xid, Relation rel,
pq_sendint32(out, RelationGetRelid(rel));
pq_sendbyte(out, 'N'); /* new tuple follows */
- logicalrep_write_tuple(out, rel, newtuple, binary);
+ logicalrep_write_tuple(out, rel, newslot, binary);
}
/*
@@ -442,7 +442,8 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
*/
void
logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
- HeapTuple oldtuple, HeapTuple newtuple, bool binary)
+ TupleTableSlot *oldslot, TupleTableSlot *newslot,
+ bool binary)
{
pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE);
@@ -457,17 +458,17 @@ logicalrep_write_update(StringInfo out, TransactionId xid, Relation rel,
/* use Oid as relation identifier */
pq_sendint32(out, RelationGetRelid(rel));
- if (oldtuple != NULL)
+ if (oldslot != NULL)
{
if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
pq_sendbyte(out, 'O'); /* old tuple follows */
else
pq_sendbyte(out, 'K'); /* old key follows */
- logicalrep_write_tuple(out, rel, oldtuple, binary);
+ logicalrep_write_tuple(out, rel, oldslot, binary);
}
pq_sendbyte(out, 'N'); /* new tuple follows */
- logicalrep_write_tuple(out, rel, newtuple, binary);
+ logicalrep_write_tuple(out, rel, newslot, binary);
}
/*
@@ -516,7 +517,7 @@ logicalrep_read_update(StringInfo in, bool *has_oldtuple,
*/
void
logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel,
- HeapTuple oldtuple, bool binary)
+ TupleTableSlot *oldslot, bool binary)
{
Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
@@ -536,7 +537,7 @@ logicalrep_write_delete(StringInfo out, TransactionId xid, Relation rel,
else
pq_sendbyte(out, 'K'); /* old key follows */
- logicalrep_write_tuple(out, rel, oldtuple, binary);
+ logicalrep_write_tuple(out, rel, oldslot, binary);
}
/*
@@ -749,11 +750,12 @@ logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp)
* Write a tuple to the outputstream, in the most efficient format possible.
*/
static void
-logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary)
+logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot,
+ bool binary)
{
TupleDesc desc;
- Datum values[MaxTupleAttributeNumber];
- bool isnull[MaxTupleAttributeNumber];
+ Datum *values;
+ bool *isnull;
int i;
uint16 nliveatts = 0;
@@ -767,11 +769,9 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binar
}
pq_sendint16(out, nliveatts);
- /* try to allocate enough memory from the get-go */
- enlargeStringInfo(out, tuple->t_len +
- nliveatts * (1 + 4));
-
- heap_deform_tuple(tuple, desc, values, isnull);
+ slot_getallattrs(slot);
+ values = slot->tts_values;
+ isnull = slot->tts_isnull;
/* Write the values */
for (i = 0; i < desc->natts; i++)
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e596b69..886ef32 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -690,19 +690,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 the 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;
@@ -798,6 +802,98 @@ fetch_remote_table_info(char *nspname, char *relname,
lrel->natts = natt;
walrcv_clear_result(res);
+
+ /*
+ * Get relation row filter expressions. DISTINCT avoids the same
+ * expression of a table in multiple publications from being included
+ * multiple times in the final expression.
+ *
+ * We do need to copy the row even if it matches one of the publications,
+ * so, we later combine all the quals with OR.
+ *
+ * For initial synchronization, row filtering can be ignored in following
+ * cases:
+ *
+ * 1) one of the subscribed publications for the table hasn't specified any
+ * row filter
+ *
+ * 2) one of the subscribed publications has puballtables set to true
+ *
+ * 3) one of the subscribed publications is declared as ALL TABLES IN
+ * SCHEMA that includes this relation
+ */
+ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
+ {
+ StringInfoData pub_names;
+
+ /* Build the pubname list. */
+ initStringInfo(&pub_names);
+ first = true;
+ foreach(lc, MySubscription->publications)
+ {
+ char *pubname = strVal(lfirst(lc));
+
+ if (first)
+ first = false;
+ else
+ appendStringInfoString(&pub_names, ", ");
+
+ appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
+ }
+
+ /* Check for row filters. */
+ resetStringInfo(&cmd);
+ appendStringInfo(&cmd,
+ "SELECT DISTINCT pg_get_expr(pr.prqual, pr.prrelid)"
+ " FROM pg_publication p"
+ " LEFT OUTER JOIN pg_publication_rel pr"
+ " ON (p.oid = pr.prpubid AND pr.prrelid = %u),"
+ " LATERAL pg_get_publication_tables(p.pubname) gpt"
+ " WHERE gpt.relid = %u"
+ " AND p.pubname IN ( %s )",
+ lrel->remoteid,
+ lrel->remoteid,
+ pub_names.data);
+
+ 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)));
+
+ /*
+ * Multiple row filter expressions for the same table will be combined
+ * by COPY using OR. If any of the filter expressions for this table
+ * are null, it means the whole table will be copied. In this case it
+ * is not necessary to construct a unified row filter expression at
+ * all.
+ */
+ 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)));
+ else
+ {
+ /* Ignore filters and cleanup as necessary. */
+ if (*qual)
+ {
+ list_free_deep(*qual);
+ *qual = NIL;
+ }
+ break;
+ }
+
+ ExecClearTuple(slot);
+ }
+ ExecDropSingleTupleTableSlot(slot);
+
+ walrcv_clear_result(res);
+ }
+
pfree(cmd.data);
}
@@ -811,6 +907,7 @@ copy_table(Relation rel)
{
LogicalRepRelMapEntry *relmapentry;
LogicalRepRelation lrel;
+ List *qual = NIL;
WalRcvExecResult *res;
StringInfoData cmd;
CopyFromState cstate;
@@ -819,7 +916,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);
@@ -830,14 +927,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.
+ * For non-tables and tables with row filters, we need to do COPY
+ * (SELECT ...), but we can't just do SELECT * because we need to not
+ * copy generated columns. For tables with any row filters, build a
+ * SELECT query with OR'ed row filters for COPY.
*/
appendStringInfoString(&cmd, "COPY (SELECT ");
for (int i = 0; i < lrel.natts; i++)
@@ -846,8 +947,24 @@ 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 OR'ed filters */
+ if (qual != NIL)
+ {
+ ListCell *lc;
+ char *q = strVal(linitial(qual));
+
+ appendStringInfo(&cmd, " WHERE %s", q);
+ for_each_from(lc, qual, 1)
+ {
+ q = strVal(lfirst(lc));
+ appendStringInfo(&cmd, " OR %s", 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 af8d51a..9589c59 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -15,16 +15,22 @@
#include "access/tupconvert.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/makefuncs.h"
+#include "optimizer/optimizer.h"
#include "replication/logical.h"
#include "replication/logicalproto.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"
@@ -87,6 +93,19 @@ static void send_repl_origin(LogicalDecodingContext *ctx,
bool send_origin);
/*
+ * Only 3 publication actions are used for row filtering ("insert", "update",
+ * "delete"). See RelationSyncEntry.exprstate[].
+ */
+enum RowFilterPubAction
+{
+ PUBACTION_INSERT,
+ PUBACTION_UPDATE,
+ PUBACTION_DELETE
+};
+
+#define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)
+
+/*
* Entry in the map used to remember which relation schemas we sent.
*
* The schema_sent flag determines if the current schema record for the
@@ -116,6 +135,26 @@ typedef struct RelationSyncEntry
PublicationActions pubactions;
/*
+ * ExprState cannot be used to indicate no cache, invalid cache and valid
+ * cache, so the flag exprstate_valid indicates if the current cache is
+ * valid.
+ *
+ * Multiple ExprState entries might be used if there are multiple
+ * publications for a single table. Different publication actions don't
+ * allow multiple expressions to always be combined into one, so there is
+ * one ExprState per publication action. The exprstate array is indexed by
+ * ReorderBufferChangeType.
+ */
+ bool exprstate_valid;
+
+ /* ExprState array for row filter. One per publication action. */
+ ExprState *exprstate[NUM_ROWFILTER_PUBACTIONS];
+ TupleTableSlot *new_slot; /* slot for storing new tuple */
+ TupleTableSlot *old_slot; /* slot for storing old tuple */
+ MemoryContext cache_expr_cxt; /* private context for table slot and
+ * exprstate, if any */
+
+ /*
* 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
@@ -129,7 +168,7 @@ typedef struct RelationSyncEntry
* same as 'relid' or if unnecessary due to partition and the ancestor
* having identical TupleDesc.
*/
- TupleConversionMap *map;
+ AttrMap *attrmap;
} RelationSyncEntry;
/* Map used to remember which relation schemas we sent. */
@@ -146,6 +185,16 @@ 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 void pgoutput_row_filter_init(PGOutputData *data, Relation relation,
+ RelationSyncEntry *entry);
+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(Relation relation, TupleTableSlot *old_slot,
+ TupleTableSlot *new_slot, RelationSyncEntry *entry,
+ ReorderBufferChangeType *action);
+
/*
* Specify output plugin callbacks
*/
@@ -555,19 +604,7 @@ maybe_send_schema(LogicalDecodingContext *ctx,
/* Map must live as long as the session does. */
oldctx = MemoryContextSwitchTo(CacheMemoryContext);
- /*
- * Make copies of the TupleDescs that will live as long as the map
- * does before putting into the map.
- */
- indesc = CreateTupleDescCopy(indesc);
- outdesc = CreateTupleDescCopy(outdesc);
- relentry->map = convert_tuples_by_name(indesc, outdesc);
- if (relentry->map == NULL)
- {
- /* Map not necessary, so free the TupleDescs too. */
- FreeTupleDesc(indesc);
- FreeTupleDesc(outdesc);
- }
+ relentry->attrmap = build_attrmap_by_name_if_req(indesc, outdesc);
MemoryContextSwitchTo(oldctx);
send_relation_and_attrs(ancestor, xid, ctx);
@@ -621,6 +658,528 @@ 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;
+ Expr *expr;
+
+ /*
+ * This is the same code as ExecPrepareExpr() but that is not used because
+ * we want to cache the expression. There should probably be another
+ * function in the executor to handle the execution outside a normal Plan
+ * tree context.
+ */
+ expr = expression_planner((Expr *) rfnode);
+ exprstate = ExecInitExpr(expr, NULL);
+
+ 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 ? "false" : "true");
+
+ if (isnull)
+ return false;
+
+ return DatumGetBool(ret);
+}
+
+/*
+ * Initialize the row filter, the first time.
+ */
+static void
+pgoutput_row_filter_init(PGOutputData *data, Relation relation,
+ RelationSyncEntry *entry)
+{
+ ListCell *lc;
+ List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */
+ bool no_filter[] = {false, false, false}; /* One per pubaction */
+ MemoryContext oldctx;
+ int idx;
+ Node *rfnode;
+ TupleDesc tupdesc;
+ Oid schemaId;
+ List *schemaPubids;
+
+ /*
+ * 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.
+ *
+ * NOTE: The ExprState cache could have been created up-front in the
+ * function get_rel_sync_entry() instead of the deferred on-the-fly
+ * assignment below. The reason for choosing to do it here is because
+ * there are some scenarios where the get_rel_sync_entry() is called but
+ * where a row will not be published. For example, for truncate, we may
+ * not need any row evaluation, so there is no need to compute it. It
+ * would also be a waste if any error happens before actually evaluating
+ * the filter. And tomorrow there could be other operations (which use
+ * get_rel_sync_entry) but which don't need to build ExprState.
+ * Furthermore, because the decision to publish or not is made AFTER the
+ * call to get_rel_sync_entry it may be that the filter evaluation is not
+ * necessary at all. So the decision was to defer this logic to last
+ * moment when we know it will be needed.
+ */
+ if (entry->exprstate_valid)
+ return;
+
+ if (entry->cache_expr_cxt == NULL)
+ entry->cache_expr_cxt = AllocSetContextCreate(CacheMemoryContext,
+ "Row filter expressions",
+ ALLOCSET_DEFAULT_SIZES);
+ else
+ MemoryContextReset(entry->cache_expr_cxt);
+
+ entry->old_slot = NULL;
+ entry->new_slot = NULL;
+ memset(entry->exprstate, 0, sizeof(entry->exprstate));
+
+ schemaId = get_rel_namespace(entry->publish_as_relid);
+ schemaPubids = GetSchemaPublications(schemaId);
+
+ /*
+ * Find if there are any row filters for this relation. If there are,
+ * then prepare the necessary ExprState and cache it in
+ * entry->exprstate. To build an expression state, we need to ensure
+ * the following:
+ *
+ * All publication-table mappings must be checked.
+ *
+ * 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.
+ *
+ * Multiple publications might have multiple row filters for this
+ * relation. Since row filter usage depends on the DML operation, there
+ * are multiple lists (one for each operation) to which row filters
+ * will be appended.
+ *
+ * FOR ALL TABLES implies "don't use row filter expression" so it takes
+ * precedence.
+ *
+ * ALL TABLES IN SCHEMA implies "don't use row filter expression" if
+ * the schema is the same as the table schema.
+ */
+ foreach(lc, data->publications)
+ {
+ Publication *pub = lfirst(lc);
+ HeapTuple rftuple = NULL;
+ Datum rfdatum = 0;
+ bool pub_no_filter = false;
+
+ if (pub->alltables)
+ {
+ /*
+ * If the publication is FOR ALL TABLES then it is treated the
+ * same as if this table has no row filters (even if for other
+ * publications it does).
+ */
+ pub_no_filter = true;
+ }
+ else if (list_member_oid(schemaPubids, pub->oid))
+ {
+ /*
+ * If the publication is FOR ALL TABLES IN SCHEMA and it overlaps
+ * with the current relation in the same schema then this is also
+ * treated same as if this table has no row filters (even if for
+ * other publications it does).
+ */
+ pub_no_filter = true;
+ }
+ else
+ {
+ bool rfisnull;
+
+ /*
+ * Lookup if there is a row filter, If no, then remember there
+ * was no filter for this pubaction.
+ */
+ rftuple = SearchSysCache2(PUBLICATIONRELMAP,
+ ObjectIdGetDatum(entry->publish_as_relid),
+ ObjectIdGetDatum(pub->oid));
+
+ if (HeapTupleIsValid(rftuple))
+ {
+ rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
+ Anum_pg_publication_rel_prqual,
+ &rfisnull);
+ pub_no_filter = rfisnull;
+ }
+
+ /*
+ * If no record in publication, check if the table is the partition
+ * of a published partitioned table. If so, the table has no row
+ * filter.
+ */
+ else if (!pub->pubviaroot)
+ {
+ List *schemarelids;
+ List *relids;
+
+ schemarelids = GetAllSchemaPublicationRelations(pub->oid,
+ PUBLICATION_PART_LEAF);
+ relids = GetPublicationRelations(pub->oid,
+ PUBLICATION_PART_LEAF);
+
+ if (list_member_oid(schemarelids, entry->publish_as_relid) ||
+ list_member_oid(relids, entry->publish_as_relid))
+ pub_no_filter = true;
+
+ list_free(schemarelids);
+ list_free(relids);
+
+ if (!pub_no_filter)
+ continue;
+ }
+ else
+ {
+ /* Table is not published in this publication. */
+ continue;
+ }
+ }
+
+ if (pub_no_filter)
+ {
+ if (rftuple)
+ ReleaseSysCache(rftuple);
+
+ no_filter[PUBACTION_INSERT] |= pub->pubactions.pubinsert;
+ no_filter[PUBACTION_UPDATE] |= pub->pubactions.pubupdate;
+ no_filter[PUBACTION_DELETE] |= pub->pubactions.pubdelete;
+
+ /* Quick exit loop if all pubactions have no row filter. */
+ if (no_filter[PUBACTION_INSERT] &&
+ no_filter[PUBACTION_UPDATE] &&
+ no_filter[PUBACTION_DELETE])
+ break;
+
+ /* No additional work for this publication. Next one. */
+ continue;
+ }
+
+ /*
+ * If row filter exists remember it in a list (per pubaction).
+ * Code following this 'publications' loop will combine all
+ * filters.
+ */
+ if (pub->pubactions.pubinsert && !no_filter[PUBACTION_INSERT])
+ rfnodes[PUBACTION_INSERT] = lappend(rfnodes[PUBACTION_INSERT],
+ TextDatumGetCString(rfdatum));
+ if (pub->pubactions.pubupdate && !no_filter[PUBACTION_UPDATE])
+ rfnodes[PUBACTION_UPDATE] = lappend(rfnodes[PUBACTION_UPDATE],
+ TextDatumGetCString(rfdatum));
+ if (pub->pubactions.pubdelete && !no_filter[PUBACTION_DELETE])
+ rfnodes[PUBACTION_DELETE] = lappend(rfnodes[PUBACTION_DELETE],
+ TextDatumGetCString(rfdatum));
+
+ ReleaseSysCache(rftuple);
+ } /* loop all subscribed publications */
+
+ list_free(schemaPubids);
+
+ /* Clean the row filter */
+ for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
+ {
+ if (no_filter[idx])
+ {
+ list_free_deep(rfnodes[idx]);
+ rfnodes[idx] = NIL;
+ }
+ }
+
+ /*
+ * Now all the filters for all pubactions are known. Combine them when
+ * their pubactions are same.
+ *
+ * All row filter expressions will be discarded if there is one
+ * publication-relation entry without a row filter. That's because all
+ * expressions are aggregated by the OR operator. The row filter
+ * absence means replicate all rows so a single valid expression means
+ * publish this row.
+ */
+ oldctx = MemoryContextSwitchTo(entry->cache_expr_cxt);
+ for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
+ {
+ List *filters = NIL;
+
+ if (rfnodes[idx] == NIL)
+ continue;
+
+ foreach(lc, rfnodes[idx])
+ filters = lappend(filters, stringToNode((char *) lfirst(lc)));
+
+ /* combine the row filter and cache the ExprState */
+ rfnode = (Node *) make_orclause(filters);
+ entry->exprstate[idx] = pgoutput_row_filter_init_expr(rfnode);
+ } /* for each pubaction */
+
+ /*
+ * Create tuple table slots for row filter. Create a copy of the
+ * TupleDesc as it needs to live as long as the cache remains.
+ */
+ tupdesc = CreateTupleDescCopy(RelationGetDescr(relation));
+ entry->old_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+ entry->new_slot = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+ MemoryContextSwitchTo(oldctx);
+
+ entry->exprstate_valid = true;
+}
+
+/*
+ * Change is checked against the row filter, if any.
+ *
+ * Returns true if the change is to be replicated, else false.
+ *
+ * For inserts: evaluates the row filter for new tuple.
+ * For deletes: evaluates the row filter for old tuple.
+ * For updates: evaluates the row filter for old and new tuple. If both
+ * evaluations are true, it sends the UPDATE. If both evaluations are false, it
+ * doesn't send the UPDATE. If only one of the tuples matches the row filter
+ * expression, there is a data consistency issue. Fixing this issue requires a
+ * transformation.
+ *
+ * Transformations:
+ * Updates are transformed to inserts and deletes based on the
+ * old tuple and new tuple. The new action is updated in the
+ * action parameter. If not updated, action remains as update.
+ *
+ * Case 1: old-row (no match) new-row (no match) -> (drop change)
+ * Case 2: old-row (no match) new row (match) -> INSERT
+ * Case 3: old-row (match) new-row (no match) -> DELETE
+ * Case 4: old-row (match) new row (match) -> UPDATE
+ *
+ * Examples:
+ * Let's say the old tuple satisfies the row filter but the new tuple doesn't.
+ * Since the old tuple satisfies, the initial table synchronization copied this
+ * row (or another method was used to guarantee that there is data
+ * consistency). However, after the UPDATE the new tuple doesn't satisfy the
+ * row filter, so from a data consistency perspective, that row should be
+ * removed on the subscriber. The UPDATE should be transformed into a DELETE
+ * statement and be sent to the subscriber. Keep this row on the subscriber is
+ * undesirable because it doesn't reflect what was defined in the row filter
+ * expression on the publisher. This row on the subscriber would likely not be
+ * modified by replication again. If someone inserted a new row with the same
+ * old identifier, replication could stop due to a constraint violation.
+ *
+ * Let's say the old tuple doesn't match the row filter but the new tuple does.
+ * Since the old tuple doesn't satisfy, the initial table synchronization
+ * probably didn't copy this row. However, after the UPDATE the new tuple does
+ * satisfy the row filter, so from a data consistency perspective, that row
+ * should be inserted on the subscriber. Otherwise, subsequent UPDATE or DELETE
+ * statements have no effect (it matches no row -- see
+ * apply_handle_update_internal()). So, The UPDATE should be transformed into a
+ * INSERT statement and be sent to the subscriber. However, this might surprise
+ * someone who expects the data set to satisfy the row filter expression on the
+ * provider.
+ */
+static bool
+pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
+ TupleTableSlot *new_slot, RelationSyncEntry *entry,
+ ReorderBufferChangeType *action)
+{
+ TupleDesc desc = RelationGetDescr(relation);
+ int i;
+ bool old_matched,
+ new_matched,
+ result;
+ TupleTableSlot *tmp_new_slot;
+ EState *estate;
+ ExprContext *ecxt;
+ ExprState *filter_exprstate;
+
+ static int map_changetype_pubaction[] = {
+ [REORDER_BUFFER_CHANGE_INSERT] = PUBACTION_INSERT,
+ [REORDER_BUFFER_CHANGE_UPDATE] = PUBACTION_UPDATE,
+ [REORDER_BUFFER_CHANGE_DELETE] = PUBACTION_DELETE
+ };
+
+ Assert(*action == REORDER_BUFFER_CHANGE_INSERT ||
+ *action == REORDER_BUFFER_CHANGE_UPDATE ||
+ *action == REORDER_BUFFER_CHANGE_DELETE);
+
+ Assert(new_slot || old_slot);
+
+ /* Get the corresponding row filter */
+ filter_exprstate = entry->exprstate[map_changetype_pubaction[*action]];
+
+ /* Bail out if there is no row filter */
+ if (!filter_exprstate)
+ return true;
+
+ elog(DEBUG3, "table \"%s.%s\" has row filter",
+ get_namespace_name(RelationGetNamespace(relation)),
+ RelationGetRelationName(relation));
+
+ estate = create_estate_for_relation(relation);
+ ecxt = GetPerTupleExprContext(estate);
+
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /*
+ * For the following occasions where there is only one tuple, we can
+ * evaluates the row filter for that tuple and return.
+ *
+ * For inserts we only have the new tuple.
+ *
+ * For updates if no old tuple, it means none of the replica identity
+ * columns changed and this would reduce to a simple update. We only need
+ * to evaluate the row filter for the new tuple.
+ *
+ * For deletes we only have the old tuple.
+ */
+ if (!new_slot || !old_slot)
+ {
+ ecxt->ecxt_scantuple = new_slot ? new_slot : old_slot;
+ result = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
+
+ FreeExecutorState(estate);
+ PopActiveSnapshot();
+
+ return result;
+ }
+
+ /*
+ * For updates, if both the new tuple and old tuple are not null, then both
+ * of them need to be checked against the row filter.
+ */
+ tmp_new_slot = new_slot;
+ slot_getallattrs(new_slot);
+ slot_getallattrs(old_slot);
+
+ /*
+ * The new tuple might not have all the replica identity columns, in which
+ * case it needs to be copied over from the old tuple.
+ */
+ for (i = 0; i < desc->natts; i++)
+ {
+ Form_pg_attribute att = TupleDescAttr(desc, i);
+
+ /*
+ * if the column in the new tuple or old tuple is null, nothing to do
+ */
+ if (tmp_new_slot->tts_isnull[i] || old_slot->tts_isnull[i])
+ continue;
+
+ /*
+ * Unchanged toasted replica identity columns are only detoasted in the
+ * old tuple, copy this over to the new tuple.
+ */
+ if (att->attlen == -1 &&
+ VARATT_IS_EXTERNAL_ONDISK(tmp_new_slot->tts_values[i]) &&
+ !VARATT_IS_EXTERNAL_ONDISK(old_slot->tts_values[i]))
+ {
+ if (tmp_new_slot == new_slot)
+ {
+ tmp_new_slot = MakeSingleTupleTableSlot(desc, &TTSOpsVirtual);
+ ExecClearTuple(tmp_new_slot);
+ ExecCopySlot(tmp_new_slot, new_slot);
+ }
+
+ tmp_new_slot->tts_values[i] = old_slot->tts_values[i];
+ tmp_new_slot->tts_isnull[i] = old_slot->tts_isnull[i];
+ }
+ }
+
+ ecxt->ecxt_scantuple = old_slot;
+ old_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
+
+ ecxt->ecxt_scantuple = tmp_new_slot;
+ new_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
+
+ FreeExecutorState(estate);
+ PopActiveSnapshot();
+
+ /*
+ * Case 1: if both tuples don't match the row filter, bail out. Send
+ * nothing.
+ */
+ if (!old_matched && !new_matched)
+ return false;
+
+ /*
+ * Case 2: if the old tuple doesn't satisfy the row filter but the new
+ * tuple does, transform the UPDATE into INSERT.
+ *
+ * This transformation requires another tuple. This transformed tuple will
+ * be used for INSERT. The new tuple is the base for the transformed tuple.
+ * However, the new tuple might not have column values from the replica
+ * identity. In this case, copy these values from the old tuple.
+ */
+ if (!old_matched && new_matched)
+ {
+ *action = REORDER_BUFFER_CHANGE_INSERT;
+
+ if (new_slot != tmp_new_slot)
+ {
+ ExecClearTuple(new_slot);
+ ExecCopySlot(new_slot, tmp_new_slot);
+ }
+ }
+
+ /*
+ * Case 3: if the old tuple satisfies the row filter but the new tuple
+ * doesn't, transform the UPDATE into DELETE.
+ *
+ * This transformation does not require another tuple. Old tuple will be
+ * used for DELETE.
+ */
+ else if (old_matched && !new_matched)
+ *action = REORDER_BUFFER_CHANGE_DELETE;
+
+ /*
+ * Case 4: if both tuples matches the row filter, transformation isn't
+ * required. (*action is default UPDATE).
+ */
+
+ return true;
+}
+
+/*
* Sends the decoded DML over wire.
*
* This is called both in streaming and non-streaming modes.
@@ -634,6 +1193,9 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
RelationSyncEntry *relentry;
TransactionId xid = InvalidTransactionId;
Relation ancestor = NULL;
+ ReorderBufferChangeType action = change->action;
+ TupleTableSlot *old_slot = NULL;
+ TupleTableSlot *new_slot = NULL;
if (!is_publishable_relation(relation))
return;
@@ -671,14 +1233,25 @@ 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);
+ /* Initialize the row_filter */
+ pgoutput_row_filter_init(data, relation, relentry);
/* Send the data */
switch (change->action)
{
case REORDER_BUFFER_CHANGE_INSERT:
{
- HeapTuple tuple = &change->data.tp.newtuple->tuple;
+ new_slot = relentry->new_slot;
+
+ ExecClearTuple(relentry->new_slot);
+ ExecStoreHeapTuple(&change->data.tp.newtuple->tuple,
+ relentry->new_slot, false);
+
+ /*
+ * 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))
@@ -687,21 +1260,46 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
ancestor = RelationIdGetRelation(relentry->publish_as_relid);
relation = ancestor;
/* Convert tuple if needed. */
- if (relentry->map)
- tuple = execute_attr_map_tuple(tuple, relentry->map);
+ if (relentry->attrmap)
+ {
+ TupleDesc tupdesc = RelationGetDescr(relation);
+ TupleTableSlot *tmp_slot = MakeTupleTableSlot(tupdesc,
+ &TTSOpsVirtual);
+
+ new_slot = execute_attr_map_slot(relentry->attrmap,
+ new_slot,
+ tmp_slot);
+ }
}
+ /* Check row filter */
+ if (!pgoutput_row_filter(relation, NULL, new_slot, relentry,
+ &action))
+ break;
+
OutputPluginPrepareWrite(ctx, true);
- logicalrep_write_insert(ctx->out, xid, relation, tuple,
+ logicalrep_write_insert(ctx->out, xid, relation, new_slot,
data->binary);
OutputPluginWrite(ctx, true);
break;
}
case REORDER_BUFFER_CHANGE_UPDATE:
{
- HeapTuple oldtuple = change->data.tp.oldtuple ?
- &change->data.tp.oldtuple->tuple : NULL;
- HeapTuple newtuple = &change->data.tp.newtuple->tuple;
+ if (change->data.tp.oldtuple)
+ {
+ old_slot = relentry->old_slot;
+
+ ExecClearTuple(old_slot);
+ ExecStoreHeapTuple(&change->data.tp.oldtuple->tuple,
+ old_slot, false);
+ }
+
+ new_slot = relentry->new_slot;
+ ExecClearTuple(new_slot);
+ ExecStoreHeapTuple(&change->data.tp.newtuple->tuple,
+ new_slot, false);
+
+ maybe_send_schema(ctx, change, relation, relentry);
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
@@ -710,26 +1308,71 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
ancestor = RelationIdGetRelation(relentry->publish_as_relid);
relation = ancestor;
/* Convert tuples if needed. */
- if (relentry->map)
+ if (relentry->attrmap)
{
- if (oldtuple)
- oldtuple = execute_attr_map_tuple(oldtuple,
- relentry->map);
- newtuple = execute_attr_map_tuple(newtuple,
- relentry->map);
+ TupleDesc tupdesc = RelationGetDescr(relation);
+ TupleTableSlot *tmp_slot;
+
+ if (old_slot)
+ {
+ tmp_slot = MakeTupleTableSlot(tupdesc,
+ &TTSOpsVirtual);
+ old_slot = execute_attr_map_slot(relentry->attrmap,
+ old_slot,
+ tmp_slot);
+ }
+
+ tmp_slot = MakeTupleTableSlot(tupdesc, &TTSOpsVirtual);
+ new_slot = execute_attr_map_slot(relentry->attrmap,
+ new_slot,
+ tmp_slot);
}
}
+ /* Check row filter */
+ if (!pgoutput_row_filter(relation, old_slot, new_slot,
+ relentry, &action))
+ break;
+
OutputPluginPrepareWrite(ctx, true);
- logicalrep_write_update(ctx->out, xid, relation, oldtuple,
- newtuple, data->binary);
+
+ /*
+ * Updates could be transformed to inserts or deletes based on
+ * the results of the row filter for old and new tuple.
+ */
+ switch (action)
+ {
+ case REORDER_BUFFER_CHANGE_INSERT:
+ logicalrep_write_insert(ctx->out, xid, relation,
+ new_slot, data->binary);
+ break;
+ case REORDER_BUFFER_CHANGE_UPDATE:
+ logicalrep_write_update(ctx->out, xid, relation,
+ old_slot, new_slot,
+ data->binary);
+ break;
+ case REORDER_BUFFER_CHANGE_DELETE:
+ logicalrep_write_delete(ctx->out, xid, relation,
+ old_slot,
+ data->binary);
+ break;
+ default:
+ Assert(false);
+ }
+
OutputPluginWrite(ctx, true);
break;
}
case REORDER_BUFFER_CHANGE_DELETE:
if (change->data.tp.oldtuple)
{
- HeapTuple oldtuple = &change->data.tp.oldtuple->tuple;
+ old_slot = relentry->old_slot;
+
+ ExecClearTuple(old_slot);
+ ExecStoreHeapTuple(&change->data.tp.oldtuple->tuple,
+ old_slot, false);
+
+ maybe_send_schema(ctx, change, relation, relentry);
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
@@ -738,13 +1381,26 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
ancestor = RelationIdGetRelation(relentry->publish_as_relid);
relation = ancestor;
/* Convert tuple if needed. */
- if (relentry->map)
- oldtuple = execute_attr_map_tuple(oldtuple, relentry->map);
+ if (relentry->attrmap)
+ {
+ TupleDesc tupdesc = RelationGetDescr(relation);
+ TupleTableSlot *tmp_slot = MakeTupleTableSlot(tupdesc,
+ &TTSOpsVirtual);
+
+ old_slot = execute_attr_map_slot(relentry->attrmap,
+ old_slot,
+ tmp_slot);
+ }
}
+ /* Check row filter */
+ if (!pgoutput_row_filter(relation, old_slot, NULL,
+ relentry, &action))
+ break;
+
OutputPluginPrepareWrite(ctx, true);
- logicalrep_write_delete(ctx->out, xid, relation, oldtuple,
- data->binary);
+ logicalrep_write_delete(ctx->out, xid, relation,
+ old_slot, data->binary);
OutputPluginWrite(ctx, true);
}
else
@@ -877,6 +1533,16 @@ pgoutput_shutdown(LogicalDecodingContext *ctx)
{
if (RelationSyncCache)
{
+ HASH_SEQ_STATUS hash_seq;
+ RelationSyncEntry *entry;
+
+ hash_seq_init(&hash_seq, RelationSyncCache);
+ while ((entry = hash_seq_search(&hash_seq)) != NULL)
+ {
+ if (entry->cache_expr_cxt != NULL)
+ MemoryContextDelete(entry->cache_expr_cxt);
+ }
+
hash_destroy(RelationSyncCache);
RelationSyncCache = NULL;
}
@@ -1137,10 +1803,15 @@ get_rel_sync_entry(PGOutputData *data, Oid relid)
entry->schema_sent = false;
entry->streamed_txns = NIL;
entry->replicate_valid = false;
+ entry->exprstate_valid = false;
entry->pubactions.pubinsert = entry->pubactions.pubupdate =
entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
+ entry->new_slot = NULL;
+ entry->old_slot = NULL;
+ memset(entry->exprstate, 0, sizeof(entry->exprstate));
+ entry->cache_expr_cxt = NULL;
entry->publish_as_relid = InvalidOid;
- entry->map = NULL; /* will be set by maybe_send_schema() if
+ entry->attrmap = NULL; /* will be set by maybe_send_schema() if
* needed */
}
@@ -1202,26 +1873,17 @@ get_rel_sync_entry(PGOutputData *data, Oid relid)
*/
if (am_partition)
{
+ Oid ancestor;
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);
+ ancestor = GetTopMostAncestorInPublication(pub->oid,
+ ancestors);
- if (list_member_oid(GetRelationPublications(ancestor),
- pub->oid) ||
- list_member_oid(GetSchemaPublications(get_rel_namespace(ancestor)),
- pub->oid))
- {
- ancestor_published = true;
- if (pub->pubviaroot)
- publish_as_relid = ancestor;
- }
+ if (ancestor != InvalidOid)
+ {
+ ancestor_published = true;
+ if (pub->pubviaroot)
+ publish_as_relid = ancestor;
}
}
@@ -1343,17 +2005,14 @@ rel_sync_cache_relation_cb(Datum arg, Oid relid)
entry->schema_sent = false;
list_free(entry->streamed_txns);
entry->streamed_txns = NIL;
- if (entry->map)
- {
- /*
- * Must free the TupleDescs contained in the map explicitly,
- * because free_conversion_map() doesn't.
- */
- FreeTupleDesc(entry->map->indesc);
- FreeTupleDesc(entry->map->outdesc);
- free_conversion_map(entry->map);
- }
- entry->map = NULL;
+ if (entry->attrmap)
+ free_attrmap(entry->attrmap);
+ entry->attrmap = NULL;
+
+ /*
+ * Row filter cache cleanups. (Will be rebuilt later if needed).
+ */
+ entry->exprstate_valid = false;
}
}
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 2e760e8..6e7dc86 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -66,6 +66,7 @@
#include "catalog/schemapg.h"
#include "catalog/storage.h"
#include "commands/policy.h"
+#include "commands/publicationcmds.h"
#include "commands/trigger.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
@@ -267,7 +268,6 @@ typedef struct opclasscacheent
static HTAB *OpClassCache = NULL;
-
/* non-export function prototypes */
static void RelationDestroyRelation(Relation relation, bool remember_tupdesc);
@@ -5522,38 +5522,55 @@ RelationGetExclusionInfo(Relation indexRelation,
}
/*
- * Get publication actions for the given relation.
+ * Get the publication information for the given relation.
+ *
+ * Traverse all the publications which the relation is in to get the
+ * publication actions.
+ *
+ * If validate_rowfilter is true and the publication actions include UPDATE or
+ * DELETE, then validate that if all columns referenced in the row filter
+ * expression are part of REPLICA IDENTITY. If any invalid column found, return
+ * the invalid column number immediately.
+ *
+ * If validate_rowfilter is false or all columns are valid, set the
+ * relation->rd_rfcol_valid to true, cache the collected publication actions in
+ * relation->rd_pubactions, and return InvalidAttrNumber.
*/
-struct PublicationActions *
-GetRelationPublicationActions(Relation relation)
+AttrNumber
+GetRelationPublicationInfo(Relation relation, bool validate_rowfilter)
{
List *puboids;
ListCell *lc;
MemoryContext oldcxt;
Oid schemaid;
- PublicationActions *pubactions = palloc0(sizeof(PublicationActions));
+ List *ancestors = NIL;
+ Oid relid = RelationGetRelid(relation);
+ PublicationActions pubactions = {0};
+ AttrNumber invalid_rfcolnum = InvalidAttrNumber;
/*
* If not publishable, it publishes no actions. (pgoutput_change() will
* ignore it.)
*/
if (!is_publishable_relation(relation))
- return pubactions;
+ return invalid_rfcolnum;
- if (relation->rd_pubactions)
- return memcpy(pubactions, relation->rd_pubactions,
- sizeof(PublicationActions));
+ if (relation->rd_rfcol_valid)
+ {
+ /* publication actions must be valid */
+ Assert(relation->rd_pubactions);
+ return invalid_rfcolnum;
+ }
/* Fetch the publication membership info. */
- puboids = GetRelationPublications(RelationGetRelid(relation));
+ puboids = GetRelationPublications(relid);
schemaid = RelationGetNamespace(relation);
puboids = list_concat_unique_oid(puboids, GetSchemaPublications(schemaid));
if (relation->rd_rel->relispartition)
{
/* Add publications that the ancestors are in too. */
- List *ancestors = get_partition_ancestors(RelationGetRelid(relation));
- ListCell *lc;
+ ancestors = get_partition_ancestors(relid);
foreach(lc, ancestors)
{
@@ -5581,19 +5598,30 @@ GetRelationPublicationActions(Relation relation)
pubform = (Form_pg_publication) GETSTRUCT(tup);
- pubactions->pubinsert |= pubform->pubinsert;
- pubactions->pubupdate |= pubform->pubupdate;
- pubactions->pubdelete |= pubform->pubdelete;
- pubactions->pubtruncate |= pubform->pubtruncate;
+ pubactions.pubinsert |= pubform->pubinsert;
+ pubactions.pubupdate |= pubform->pubupdate;
+ pubactions.pubdelete |= pubform->pubdelete;
+ pubactions.pubtruncate |= pubform->pubtruncate;
ReleaseSysCache(tup);
/*
- * If we know everything is replicated, there is no point to check for
- * other publications.
+ * Check, if all columns referenced in the filter expression are part
+ * of the REPLICA IDENTITY index or not. Return the invalid column
+ * number immediately if found.
*/
- if (pubactions->pubinsert && pubactions->pubupdate &&
- pubactions->pubdelete && pubactions->pubtruncate)
+ if (validate_rowfilter && (pubform->pubupdate || pubform->pubdelete) &&
+ contain_invalid_rfcolumn(pubid, relation, ancestors, &invalid_rfcolnum))
+ return invalid_rfcolnum;
+
+ /*
+ * If we know everything is replicated and we are not validating the
+ * row filter column, there is no point to check for other
+ * publications.
+ */
+ if (pubactions.pubinsert && pubactions.pubupdate &&
+ pubactions.pubdelete && pubactions.pubtruncate &&
+ !validate_rowfilter)
break;
}
@@ -5603,13 +5631,41 @@ GetRelationPublicationActions(Relation relation)
relation->rd_pubactions = NULL;
}
+ if (validate_rowfilter)
+ relation->rd_rfcol_valid = true;
+
/* Now save copy of the actions in the relcache entry. */
oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
relation->rd_pubactions = palloc(sizeof(PublicationActions));
- memcpy(relation->rd_pubactions, pubactions, sizeof(PublicationActions));
+ memcpy(relation->rd_pubactions, &pubactions, sizeof(PublicationActions));
MemoryContextSwitchTo(oldcxt);
- return pubactions;
+ return invalid_rfcolnum;
+}
+
+/*
+ * Get publication actions for the given relation.
+ */
+struct PublicationActions *
+GetRelationPublicationActions(Relation relation)
+{
+ PublicationActions *pubactions = palloc0(sizeof(PublicationActions));
+
+ /*
+ * If not publishable, it publishes no actions. (pgoutput_change() will
+ * ignore it.)
+ */
+ if (!is_publishable_relation(relation))
+ return pubactions;
+
+ if (!relation->rd_pubactions)
+ {
+ (void) GetRelationPublicationInfo(relation, false);
+ Assert(relation->rd_pubactions);
+ }
+
+ return memcpy(pubactions, relation->rd_pubactions,
+ sizeof(PublicationActions));
}
/*
@@ -6163,6 +6219,7 @@ load_relcache_init_file(bool shared)
rel->rd_idattr = NULL;
rel->rd_hotblockingattr = NULL;
rel->rd_pubactions = NULL;
+ rel->rd_rfcol_valid = false;
rel->rd_statvalid = false;
rel->rd_statlist = NIL;
rel->rd_fkeyvalid = false;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 8587b19..9b1171d 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -2868,17 +2868,21 @@ describeOneTableDetails(const char *schemaname,
{
printfPQExpBuffer(&buf,
"SELECT pubname\n"
+ " , NULL\n"
"FROM pg_catalog.pg_publication p\n"
" JOIN pg_catalog.pg_publication_namespace pn ON p.oid = pn.pnpubid\n"
" JOIN pg_catalog.pg_class pc ON pc.relnamespace = pn.pnnspid\n"
"WHERE pc.oid ='%s' and pg_catalog.pg_relation_is_publishable('%s')\n"
"UNION\n"
"SELECT pubname\n"
+ " , pg_get_expr(pr.prqual, c.oid)\n"
"FROM pg_catalog.pg_publication p\n"
" JOIN pg_catalog.pg_publication_rel pr ON p.oid = pr.prpubid\n"
+ " JOIN pg_catalog.pg_class c ON c.oid = pr.prrelid\n"
"WHERE pr.prrelid = '%s'\n"
"UNION\n"
"SELECT pubname\n"
+ " , NULL\n"
"FROM pg_catalog.pg_publication p\n"
"WHERE p.puballtables AND pg_catalog.pg_relation_is_publishable('%s')\n"
"ORDER BY 1;",
@@ -2914,6 +2918,13 @@ describeOneTableDetails(const char *schemaname,
printfPQExpBuffer(&buf, " \"%s\"",
PQgetvalue(result, i, 0));
+ /* row filter (if any) */
+ if (pset.sversion >= 150000)
+ {
+ if (!PQgetisnull(result, i, 1))
+ appendPQExpBuffer(&buf, " WHERE %s", PQgetvalue(result, i, 1));
+ }
+
printTableAddFooter(&cont, buf.data);
}
PQclear(result);
@@ -5841,8 +5852,12 @@ addFooterToPublicationDesc(PQExpBuffer buf, char *footermsg,
for (i = 0; i < count; i++)
{
if (!singlecol)
+ {
printfPQExpBuffer(buf, " \"%s.%s\"", PQgetvalue(res, i, 0),
PQgetvalue(res, i, 1));
+ if (!PQgetisnull(res, i, 2))
+ appendPQExpBuffer(buf, " WHERE %s", PQgetvalue(res, i, 2));
+ }
else
printfPQExpBuffer(buf, " \"%s\"", PQgetvalue(res, i, 0));
@@ -5971,8 +5986,15 @@ describePublications(const char *pattern)
{
/* Get the tables for the specified publication */
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"
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 841b9b6..ec60b41 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -86,6 +86,7 @@ typedef struct Publication
typedef struct PublicationRelInfo
{
Relation relation;
+ Node *whereClause;
} PublicationRelInfo;
extern Publication *GetPublication(Oid pubid);
@@ -120,10 +121,11 @@ extern List *GetAllSchemaPublicationRelations(Oid puboid,
extern List *GetPubPartitionOptionRelations(List *result,
PublicationPartOpt pub_partopt,
Oid relid);
+extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors);
extern bool is_publishable_relation(Relation rel);
extern bool is_schema_publication(Oid pubid);
-extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
+extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *pri,
bool if_not_exists);
extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
bool if_not_exists);
@@ -131,5 +133,4 @@ extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
extern Oid get_publication_oid(const char *pubname, bool missing_ok);
extern char *get_publication_name(Oid pubid, bool missing_ok);
-
#endif /* PG_PUBLICATION_H */
diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h
index 117a1d6..0dd0f42 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));
DECLARE_INDEX(pg_publication_rel_prpubid_index, 6116, PublicationRelPrpubidIndexId, on pg_publication_rel using btree(prpubid oid_ops));
diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h
index cec7525..b00839a 100644
--- a/src/include/commands/publicationcmds.h
+++ b/src/include/commands/publicationcmds.h
@@ -31,5 +31,8 @@ extern void RemovePublicationSchemaById(Oid psoid);
extern ObjectAddress AlterPublicationOwner(const char *name, Oid newOwnerId);
extern void AlterPublicationOwner_oid(Oid pubid, Oid newOwnerId);
extern void InvalidatePublicationRels(List *relids);
+extern bool contain_invalid_rfcolumn(Oid pubid, Relation relation,
+ List *ancestors,
+ AttrNumber *invalid_rfcolumn);
#endif /* PUBLICATIONCMDS_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 413e7c8..d52b5b2 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3642,6 +3642,7 @@ typedef struct PublicationTable
{
NodeTag type;
RangeVar *relation; /* relation to be published */
+ Node *whereClause; /* qualifications */
} PublicationTable;
/*
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 22fffac..1d0024d 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -16,6 +16,7 @@
#include "access/xact.h"
#include "replication/reorderbuffer.h"
#include "utils/rel.h"
+#include "executor/tuptable.h"
/*
* Protocol capabilities
@@ -206,17 +207,19 @@ extern void logicalrep_write_origin(StringInfo out, const char *origin,
XLogRecPtr origin_lsn);
extern char *logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn);
extern void logicalrep_write_insert(StringInfo out, TransactionId xid,
- Relation rel, HeapTuple newtuple,
+ Relation rel,
+ TupleTableSlot *newslot,
bool binary);
extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup);
extern void logicalrep_write_update(StringInfo out, TransactionId xid,
- Relation rel, HeapTuple oldtuple,
- HeapTuple newtuple, bool binary);
+ Relation rel,
+ TupleTableSlot *oldslot,
+ TupleTableSlot *newslot, bool binary);
extern LogicalRepRelId logicalrep_read_update(StringInfo in,
bool *has_oldtuple, LogicalRepTupleData *oldtup,
LogicalRepTupleData *newtup);
extern void logicalrep_write_delete(StringInfo out, TransactionId xid,
- Relation rel, HeapTuple oldtuple,
+ Relation rel, TupleTableSlot *oldtuple,
bool binary);
extern LogicalRepRelId logicalrep_read_delete(StringInfo in,
LogicalRepTupleData *oldtup);
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index aa0a733..f12e75d 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -51,7 +51,7 @@ typedef struct ReorderBufferTupleBuf
* respectively. They're used by INSERT .. ON CONFLICT .. UPDATE. Users of
* logical decoding don't have to care about these.
*/
-enum ReorderBufferChangeType
+typedef enum ReorderBufferChangeType
{
REORDER_BUFFER_CHANGE_INSERT,
REORDER_BUFFER_CHANGE_UPDATE,
@@ -65,7 +65,7 @@ enum ReorderBufferChangeType
REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM,
REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT,
REORDER_BUFFER_CHANGE_TRUNCATE
-};
+} ReorderBufferChangeType;
/* forward declaration */
struct ReorderBufferTXN;
@@ -82,7 +82,7 @@ typedef struct ReorderBufferChange
XLogRecPtr lsn;
/* The type of change. */
- enum ReorderBufferChangeType action;
+ ReorderBufferChangeType action;
/* Transaction this change belongs to. */
struct ReorderBufferTXN *txn;
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 6da1b22..2723e3e 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -164,6 +164,13 @@ typedef struct RelationData
PublicationActions *rd_pubactions; /* publication actions */
/*
+ * true if the columns referenced in row filters from all the publications
+ * the relation is in are part of the replica identity, or the publication
+ * actions do not include UPDATE or DELETE.
+ */
+ bool rd_rfcol_valid;
+
+ /*
* rd_options is set whenever rd_rel is loaded into the relcache entry.
* Note that you can NOT look into rd_rel for this data. NULL means "use
* defaults".
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 84d6afe..82308ae 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -76,6 +76,7 @@ extern void RelationInitIndexAccessInfo(Relation relation);
/* caller must include pg_publication.h */
struct PublicationActions;
extern struct PublicationActions *GetRelationPublicationActions(Relation relation);
+extern AttrNumber GetRelationPublicationInfo(Relation relation, bool validate_rowfilter);
extern void RelationInitTableAccessMethod(Relation relation);
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 12c5f67..c998c0f 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -239,6 +239,307 @@ 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;
+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);
+CREATE TABLE testpub_rf_tbl5 (a xml);
+CREATE SCHEMA testpub_rf_schema1;
+CREATE TABLE testpub_rf_schema1.testpub_rf_tbl5 (h integer);
+CREATE SCHEMA testpub_rf_schema2;
+CREATE TABLE testpub_rf_schema2.testpub_rf_tbl6 (i integer);
+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))
+
+-- test \d+ (now it displays filter information)
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_dplus_rf_yes FOR TABLE testpub_rf_tbl1 WHERE (a > 1) WITH (publish = 'insert');
+CREATE PUBLICATION testpub_dplus_rf_no FOR TABLE testpub_rf_tbl1;
+RESET client_min_messages;
+\d+ testpub_rf_tbl1
+ Table "public.testpub_rf_tbl1"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+----------+--------------+-------------
+ a | integer | | | | plain | |
+ b | text | | | | extended | |
+Publications:
+ "testpub_dplus_rf_no"
+ "testpub_dplus_rf_yes" WHERE (a > 1)
+
+DROP PUBLICATION testpub_dplus_rf_yes, testpub_dplus_rf_no;
+-- some more syntax tests to exercise other parser pathways
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999) WITH (publish = 'insert');
+RESET client_min_messages;
+\dRp+ testpub_syntax1
+ Publication testpub_syntax1
+ 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 < 999)
+
+DROP PUBLICATION testpub_syntax1;
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = 'insert');
+RESET client_min_messages;
+\dRp+ testpub_syntax2
+ Publication testpub_syntax2
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | f | f | f | f
+Tables:
+ "public.testpub_rf_tbl1"
+ "testpub_rf_schema1.testpub_rf_tbl5" WHERE (h < 999)
+
+DROP PUBLICATION testpub_syntax2;
+-- fail - schemas don't allow WHERE clause
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a = 123);
+ERROR: syntax error at or near "WHERE"
+LINE 1: ...ntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a =...
+ ^
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1, testpub_rf_schema1 WHERE (a = 123);
+ERROR: WHERE clause not allowed for schema
+LINE 1: ...tax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1, testpub_rf...
+ ^
+RESET client_min_messages;
+-- fail - duplicate tables are not allowed if that table has any WHERE clause
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1 WHERE (a = 1), testpub_rf_tbl1 WITH (publish = 'insert');
+ERROR: conflicting or redundant WHERE clauses for table "testpub_rf_tbl1"
+CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1, testpub_rf_tbl1 WHERE (a = 2) WITH (publish = 'insert');
+ERROR: conflicting or redundant WHERE clauses for table "testpub_rf_tbl1"
+RESET client_min_messages;
+-- fail - publication WHERE clause must be boolean
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (1234);
+ERROR: argument of PUBLICATION WHERE must be type boolean, not type integer
+LINE 1: ...PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (1234);
+ ^
+-- fail - aggregate functions not allowed in WHERE clause
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
+ERROR: aggregate functions are not allowed in WHERE
+LINE 1: ...ATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
+ ^
+-- fail - user-defined operators are not allowed
+CREATE FUNCTION testpub_rf_func1(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func1, 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.
+-- fail - user-defined functions are not allowed
+CREATE FUNCTION testpub_rf_func2() RETURNS integer AS $$ BEGIN RETURN 123; END; $$ LANGUAGE plpgsql;
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a >= testpub_rf_func2());
+ERROR: invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL: User-defined functions are not allowed (testpub_rf_func2).
+-- fail - non-immutable functions are not allowed. random() is volatile.
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a < random());
+ERROR: invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL: Non-immutable built-in functions are not allowed (random).
+-- ok - NULLIF is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1,2) = a);
+-- ok - built-in operators are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE ((a > 5) IS FALSE);
+-- ok - immutable built-in functions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- fail - user-defined types disallowed
+CREATE TYPE rf_bug_status AS ENUM ('new', 'open', 'closed');
+CREATE TABLE rf_bug (id serial, description text, status rf_bug_status);
+CREATE PUBLICATION testpub6 FOR TABLE rf_bug WHERE (status = 'open') WITH (publish = 'insert');
+ERROR: invalid publication WHERE expression for relation "rf_bug"
+DETAIL: User-defined types are not allowed.
+DROP TABLE rf_bug;
+DROP TYPE rf_bug_status;
+-- fail - row filter expression is not simple
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+ERROR: invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL: Expressions only allow columns, constants, built-in operators, built-in data types and non-immutable built-in functions.
+-- fail - system columns are not allowed
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl1 WHERE ('(0,1)'::tid = ctid);
+ERROR: invalid publication WHERE expression for relation "testpub_rf_tbl1"
+DETAIL: Cannot use system column (ctid).
+-- ok - conditional expressions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (a IS DOCUMENT);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (xmlexists('//foo[text() = ''bar'']' PASSING BY VALUE a));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1, 2) = a);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (CASE a WHEN 5 THEN true ELSE false END);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (COALESCE(b, 'foo') = 'foo');
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (GREATEST(a, 10) > 10);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IN (2, 4, 6));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (ARRAY[a] <@ ARRAY[2, 4, 6]);
+-- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl1 WHERE (e < 27);
+ERROR: cannot use a WHERE clause when removing a table from a publication
+-- fail - cannot ALTER SET table which is a member of a pre-existing schema
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR ALL TABLES IN SCHEMA testpub_rf_schema2;
+ALTER PUBLICATION testpub6 SET ALL TABLES IN SCHEMA testpub_rf_schema2, TABLE testpub_rf_schema2.testpub_rf_tbl6 WHERE (i < 99);
+ERROR: cannot add relation "testpub_rf_schema2.testpub_rf_tbl6" to publication
+DETAIL: Table's schema "testpub_rf_schema2" is already part of the publication or part of the specified schema list.
+RESET client_min_messages;
+DROP TABLE testpub_rf_tbl1;
+DROP TABLE testpub_rf_tbl2;
+DROP TABLE testpub_rf_tbl3;
+DROP TABLE testpub_rf_tbl4;
+DROP TABLE testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema1.testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema2.testpub_rf_tbl6;
+DROP SCHEMA testpub_rf_schema1;
+DROP SCHEMA testpub_rf_schema2;
+DROP PUBLICATION testpub5;
+DROP PUBLICATION testpub6;
+DROP OPERATOR =#>(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
+-- ======================================================
+-- 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));
+CREATE TABLE rf_tbl_abcd_part_pk (a int PRIMARY KEY, b int) PARTITION by RANGE (a);
+CREATE TABLE rf_tbl_abcd_part_pk_1 (b int, a int PRIMARY KEY);
+ALTER TABLE rf_tbl_abcd_part_pk ATTACH PARTITION rf_tbl_abcd_part_pk_1 FOR VALUES FROM (1) TO (10);
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+-- ok - "a" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (b > 99);
+-- ok - "b" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_pk"
+DETAIL: Column "c" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (d > 99);
+-- fail - "d" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_pk"
+DETAIL: Column "d" used in the publication WHERE expression is not part of the replica identity.
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not part of REPLICA IDENTITY
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_nopk"
+DETAIL: Column "a" used in the publication WHERE expression 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;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- ok - "a" is in REPLICA IDENTITY now
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_pk"
+DETAIL: Column "a" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_pk"
+DETAIL: Column "c" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_nopk"
+DETAIL: Column "a" used in the publication WHERE expression 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;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_pk"
+DETAIL: Column "a" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_nopk"
+DETAIL: Column "a" used in the publication WHERE expression is not part of the replica identity.
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+-- Tests for partitioned table
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (a > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- ok - "a" is a OK col
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (b > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- fail - "b" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ERROR: cannot update table "rf_tbl_abcd_part_pk_1"
+DETAIL: Column "b" used in the publication WHERE expression is not part of the replica identity.
+DROP PUBLICATION testpub6;
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+DROP TABLE rf_tbl_abcd_part_pk;
+-- ======================================================
-- Test cache invalidation FOR ALL TABLES publication
SET client_min_messages = 'ERROR';
CREATE TABLE testpub_tbl4(a int);
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index 56dd358..12648d7 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -134,6 +134,212 @@ UPDATE testpub_parted2 SET a = 2;
DROP TABLE testpub_parted1, testpub_parted2;
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);
+CREATE TABLE testpub_rf_tbl5 (a xml);
+CREATE SCHEMA testpub_rf_schema1;
+CREATE TABLE testpub_rf_schema1.testpub_rf_tbl5 (h integer);
+CREATE SCHEMA testpub_rf_schema2;
+CREATE TABLE testpub_rf_schema2.testpub_rf_tbl6 (i integer);
+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
+-- test \d+ (now it displays filter information)
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_dplus_rf_yes FOR TABLE testpub_rf_tbl1 WHERE (a > 1) WITH (publish = 'insert');
+CREATE PUBLICATION testpub_dplus_rf_no FOR TABLE testpub_rf_tbl1;
+RESET client_min_messages;
+\d+ testpub_rf_tbl1
+DROP PUBLICATION testpub_dplus_rf_yes, testpub_dplus_rf_no;
+-- some more syntax tests to exercise other parser pathways
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999) WITH (publish = 'insert');
+RESET client_min_messages;
+\dRp+ testpub_syntax1
+DROP PUBLICATION testpub_syntax1;
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = 'insert');
+RESET client_min_messages;
+\dRp+ testpub_syntax2
+DROP PUBLICATION testpub_syntax2;
+-- fail - schemas don't allow WHERE clause
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1, testpub_rf_schema1 WHERE (a = 123);
+RESET client_min_messages;
+-- fail - duplicate tables are not allowed if that table has any WHERE clause
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1 WHERE (a = 1), testpub_rf_tbl1 WITH (publish = 'insert');
+CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1, testpub_rf_tbl1 WHERE (a = 2) WITH (publish = 'insert');
+RESET client_min_messages;
+-- fail - publication WHERE clause must be boolean
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (1234);
+-- fail - aggregate functions not allowed in WHERE clause
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
+-- fail - user-defined operators are not allowed
+CREATE FUNCTION testpub_rf_func1(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func1, LEFTARG = integer, RIGHTARG = integer);
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+-- fail - user-defined functions are not allowed
+CREATE FUNCTION testpub_rf_func2() RETURNS integer AS $$ BEGIN RETURN 123; END; $$ LANGUAGE plpgsql;
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a >= testpub_rf_func2());
+-- fail - non-immutable functions are not allowed. random() is volatile.
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a < random());
+-- ok - NULLIF is allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1,2) = a);
+-- ok - built-in operators are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IS NULL);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE ((a > 5) IS FALSE);
+-- ok - immutable built-in functions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- fail - user-defined types disallowed
+CREATE TYPE rf_bug_status AS ENUM ('new', 'open', 'closed');
+CREATE TABLE rf_bug (id serial, description text, status rf_bug_status);
+CREATE PUBLICATION testpub6 FOR TABLE rf_bug WHERE (status = 'open') WITH (publish = 'insert');
+DROP TABLE rf_bug;
+DROP TYPE rf_bug_status;
+-- fail - row filter expression is not simple
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl1 WHERE (a IN (SELECT generate_series(1,5)));
+-- fail - system columns are not allowed
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl1 WHERE ('(0,1)'::tid = ctid);
+-- ok - conditional expressions are allowed
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (a IS DOCUMENT);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl5 WHERE (xmlexists('//foo[text() = ''bar'']' PASSING BY VALUE a));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (NULLIF(1, 2) = a);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (CASE a WHEN 5 THEN true ELSE false END);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (COALESCE(b, 'foo') = 'foo');
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (GREATEST(a, 10) > 10);
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (a IN (2, 4, 6));
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl1 WHERE (ARRAY[a] <@ ARRAY[2, 4, 6]);
+-- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl1 WHERE (e < 27);
+-- fail - cannot ALTER SET table which is a member of a pre-existing schema
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR ALL TABLES IN SCHEMA testpub_rf_schema2;
+ALTER PUBLICATION testpub6 SET ALL TABLES IN SCHEMA testpub_rf_schema2, TABLE testpub_rf_schema2.testpub_rf_tbl6 WHERE (i < 99);
+RESET client_min_messages;
+
+DROP TABLE testpub_rf_tbl1;
+DROP TABLE testpub_rf_tbl2;
+DROP TABLE testpub_rf_tbl3;
+DROP TABLE testpub_rf_tbl4;
+DROP TABLE testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema1.testpub_rf_tbl5;
+DROP TABLE testpub_rf_schema2.testpub_rf_tbl6;
+DROP SCHEMA testpub_rf_schema1;
+DROP SCHEMA testpub_rf_schema2;
+DROP PUBLICATION testpub5;
+DROP PUBLICATION testpub6;
+DROP OPERATOR =#>(integer, integer);
+DROP FUNCTION testpub_rf_func1(integer, integer);
+DROP FUNCTION testpub_rf_func2();
+
+-- ======================================================
+-- 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));
+CREATE TABLE rf_tbl_abcd_part_pk (a int PRIMARY KEY, b int) PARTITION by RANGE (a);
+CREATE TABLE rf_tbl_abcd_part_pk_1 (b int, a int PRIMARY KEY);
+ALTER TABLE rf_tbl_abcd_part_pk ATTACH PARTITION rf_tbl_abcd_part_pk_1 FOR VALUES FROM (1) TO (10);
+
+-- Case 1. REPLICA IDENTITY DEFAULT (means use primary key or nothing)
+-- 1a. REPLICA IDENTITY is DEFAULT and table has a PK.
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub6 FOR TABLE rf_tbl_abcd_pk WHERE (a > 99);
+RESET client_min_messages;
+-- ok - "a" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (b > 99);
+-- ok - "b" is a PK col
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (d > 99);
+-- fail - "d" is not part of the PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+-- 1b. REPLICA IDENTITY is DEFAULT and table has no PK
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not part of REPLICA IDENTITY
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- Case 2. REPLICA IDENTITY FULL
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY FULL;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY FULL;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is in REPLICA IDENTITY now even though not in PK
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- ok - "a" is in REPLICA IDENTITY now
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- Case 3. REPLICA IDENTITY NOTHING
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY NOTHING;
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY NOTHING;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- fail - "c" is not in PK and not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY NOTHING
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- Case 4. REPLICA IDENTITY INDEX
+ALTER TABLE rf_tbl_abcd_pk ALTER COLUMN c SET NOT NULL;
+CREATE UNIQUE INDEX idx_abcd_pk_c ON rf_tbl_abcd_pk(c);
+ALTER TABLE rf_tbl_abcd_pk REPLICA IDENTITY USING INDEX idx_abcd_pk_c;
+ALTER TABLE rf_tbl_abcd_nopk ALTER COLUMN c SET NOT NULL;
+CREATE UNIQUE INDEX idx_abcd_nopk_c ON rf_tbl_abcd_nopk(c);
+ALTER TABLE rf_tbl_abcd_nopk REPLICA IDENTITY USING INDEX idx_abcd_nopk_c;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (a > 99);
+-- fail - "a" is in PK but it is not part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_pk WHERE (c > 99);
+-- ok - "c" is not in PK but it is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (a > 99);
+-- fail - "a" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_nopk WHERE (c > 99);
+-- ok - "c" is part of REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_nopk SET a = 1;
+
+-- Tests for partitioned table
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (a > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- ok - "a" is a OK col
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET TABLE rf_tbl_abcd_part_pk WHERE (b > 99);
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=0);
+-- ok - partition does not have row filter
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+ALTER PUBLICATION testpub6 SET (PUBLISH_VIA_PARTITION_ROOT=1);
+-- fail - "b" is not in REPLICA IDENTITY INDEX
+UPDATE rf_tbl_abcd_part_pk SET a = 1;
+
+DROP PUBLICATION testpub6;
+DROP TABLE rf_tbl_abcd_pk;
+DROP TABLE rf_tbl_abcd_nopk;
+DROP TABLE rf_tbl_abcd_part_pk;
+-- ======================================================
+
-- Test cache invalidation FOR ALL TABLES publication
SET client_min_messages = 'ERROR';
CREATE TABLE testpub_tbl4(a int);
diff --git a/src/test/subscription/t/027_row_filter.pl b/src/test/subscription/t/027_row_filter.pl
new file mode 100644
index 0000000..81a1374
--- /dev/null
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -0,0 +1,509 @@
+# Test logical replication behavior with row filtering
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More tests => 15;
+
+# create publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+my $synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+my $appname = 'tap_sub';
+
+# ====================================================================
+# Testcase start: FOR ALL TABLES
+#
+# The FOR ALL TABLES test must come first so that it is not affected by
+# all the other test tables that are later created.
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres', "CREATE TABLE tab_rf_x (x int primary key)");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE tab_rf_x (x int primary key)");
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rf_x (x) VALUES (0), (5), (10), (15), (20)");
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_x FOR TABLE tab_rf_x WHERE (x > 10)");
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_forall FOR ALL TABLES");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_x, tap_pub_forall");
+
+$node_publisher->wait_for_catchup($appname);
+# wait for initial table synchronization to finish
+$node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# The subscription of the FOR ALL TABLES publication means there should be no
+# filtering on the tablesync COPY, so all expect all 5 will be present.
+my $result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM tab_rf_x");
+is($result, qq(5), 'check initial data copy from table tab_rf_x should not be filtered');
+
+# Similarly, normal filtering after the initial phase will also have not effect.
+# Expected: 5 initial rows + 2 new rows = 7 rows
+$node_publisher->safe_psql('postgres', "INSERT INTO tab_rf_x (x) VALUES (-99), (99)");
+$node_publisher->wait_for_catchup($appname);
+$result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM tab_rf_x");
+is($result, qq(7), 'check table tab_rf_x should not be filtered');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_forall");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_x");
+$node_publisher->safe_psql('postgres', "DROP TABLE tab_rf_x");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_subscriber->safe_psql('postgres', "DROP TABLE tab_rf_x");
+
+# Testcase end: FOR ALL TABLES
+# ====================================================================
+
+# ====================================================================
+# Testcase start: ALL TABLES IN SCHEMA
+#
+# The ALL TABLES IN SCHEMA test is independent of all other test cases so it
+# cleans up after itself.
+
+# create tables pub and sub
+$node_publisher->safe_psql('postgres', "CREATE SCHEMA schema_rf_x");
+$node_publisher->safe_psql('postgres', "CREATE TABLE schema_rf_x.tab_rf_x (x int primary key)");
+$node_subscriber->safe_psql('postgres', "CREATE SCHEMA schema_rf_x");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE schema_rf_x.tab_rf_x (x int primary key)");
+
+# insert some initial data
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO schema_rf_x.tab_rf_x (x) VALUES (0), (5), (10), (15), (20)");
+
+# create pub/sub
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_x FOR TABLE schema_rf_x.tab_rf_x WHERE (x > 10)");
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_allinschema FOR ALL TABLES IN SCHEMA schema_rf_x");
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_x, tap_pub_allinschema");
+
+$node_publisher->wait_for_catchup($appname);
+# wait for initial table synchronization to finish
+$node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# The subscription of the ALL TABLES IN SCHEMA publication means there should be
+# no filtering on the tablesync COPY, so all expect all 5 will be present.
+$result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM schema_rf_x.tab_rf_x");
+is($result, qq(5), 'check initial data copy from table tab_rf_x should not be filtered');
+
+# Similarly, normal filtering after the initial phase will also have not effect.
+# Expected: 5 initial rows + 2 new rows = 7 rows
+$node_publisher->safe_psql('postgres', "INSERT INTO schema_rf_x.tab_rf_x (x) VALUES (-99), (99)");
+$node_publisher->wait_for_catchup($appname);
+$result = $node_subscriber->safe_psql('postgres', "SELECT count(x) FROM schema_rf_x.tab_rf_x");
+is($result, qq(7), 'check table tab_rf_x should not be filtered');
+
+# cleanup pub
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_allinschema");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_x");
+$node_publisher->safe_psql('postgres', "DROP TABLE schema_rf_x.tab_rf_x");
+$node_publisher->safe_psql('postgres', "DROP SCHEMA schema_rf_x");
+# cleanup sub
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+$node_subscriber->safe_psql('postgres', "DROP TABLE schema_rf_x.tab_rf_x");
+$node_subscriber->safe_psql('postgres', "DROP SCHEMA schema_rf_x");
+
+# Testcase end: ALL TABLES IN SCHEMA
+# ====================================================================
+
+# 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_4 (c int primary key)");
+$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)"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_toast (a text PRIMARY KEY, b text)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE tab_rowfilter_toast ALTER COLUMN a SET STORAGE EXTERNAL");
+
+# 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_4 (c int primary key)");
+$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)"
+);
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE tab_rowfilter_toast (a text PRIMARY KEY, b text)");
+
+# 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)"
+);
+
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_4a FOR TABLE tab_rowfilter_4 WHERE (c % 2 = 0)"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_4b FOR TABLE tab_rowfilter_4"
+);
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_toast FOR TABLE tab_rowfilter_toast WHERE (a = repeat('1234567890', 200))");
+
+#
+# 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");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_4 (c) SELECT generate_series(1, 10)");
+
+# 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)");
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_toast(a, b) VALUES(repeat('1234567890', 200), '1234567890')");
+
+$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, tap_pub_4a, tap_pub_4b, tap_pub_toast"
+);
+
+$node_publisher->wait_for_catchup($appname);
+
+# wait for initial table synchronization to finish
+$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
+#
+$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, the filters
+# expressions are OR'ed together. In this case, rows are replicated if
+# c value is divided by 2 OR 3 (2, 3, 4, 6, 8, 9, 10, 12, 14, 15, 16, 18, 20)
+#
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(c), min(c), max(c) FROM tab_rowfilter_2");
+is($result, qq(13|2|20), 'check initial data copy from table tab_rowfilter_2');
+
+# Check expected replicated rows for tab_rowfilter_4
+# (same table in two publications but only one has a filter).
+# tap_pub_4a filter is: (c % 2 = 0)
+# tap_pub_4b filter is: <no filter>
+# Expressions are OR'ed together but when there is no filter it just means
+# OR everything - e.g. same as no filter at all.
+# Expect all rows: (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(c), min(c), max(c) FROM tab_rowfilter_4");
+is($result, qq(10|1|10), 'check initial data copy from table tab_rowfilter_4');
+
+# 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');
+
+# Check expected replicated rows for tab_rowfilter_toast
+# tab_rowfilter_toast filter: (a = repeat('1234567890', 200))
+# INSERT (repeat('1234567890', 200) ,'1234567890') YES
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a = repeat('1234567890', 200), b FROM tab_rowfilter_toast");
+is($result, qq(t|1234567890), 'check initial data copy from table tab_rowfilter_toast');
+
+# 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 (1602, 'filtered')");
+$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',
+ "UPDATE tab_rowfilter_1 SET b = 'test 1602 updated' WHERE a = 1602");
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM tab_rowfilter_1 WHERE a = 1700");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_2 (c) VALUES (21), (22), (23), (24), (25)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_4 (c) VALUES (0), (11), (12)");
+
+$node_publisher->wait_for_catchup($appname);
+
+# 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, the filters
+# expressions are OR'ed together. In this case, rows are replicated if
+# c value is divided by 2 OR 3.
+#
+# Expect original rows (2, 3, 4, 6, 8, 9, 10, 12, 14, 15, 16, 18, 20)
+# Plus (21, 22, 24)
+#
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(c), min(c), max(c) FROM tab_rowfilter_2");
+is($result, qq(16|2|24), 'check replicated rows to tab_rowfilter_2');
+
+# Check expected replicated rows for tab_rowfilter_4
+# (same table in two publications but only one has a filter).
+# tap_pub_4a filter is: (c % 2 = 0)
+# tap_pub_4b filter is: <no filter>
+# Expressions are OR'ed together but when there is no filter it just means
+# OR everything - e.g. same as no filter at all.
+# Expect all rows from initial copy: (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+# And also (0, 11, 12)
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(c), min(c), max(c) FROM tab_rowfilter_4");
+is($result, qq(13|0|12), 'check replicated rows to tab_rowfilter_4');
+
+# 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',
+# but row deleted after the update below.
+# - INSERT (1601, 'test 1601') YES, because 1601 > 1000 and 'test 1601' <> 'filtered'
+# - INSERT (1602, 'filtered') NO, because b == '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'
+# - UPDATE (1602, 'test 1602 updated') YES, because 1602 > 1000 and 'test 1602 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
+1601|test 1601 updated
+1602|test 1602 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');
+
+# FIXME: Currently, If replica identity is set to key and the key is not
+# modified we don't log key seperately, because it should be logged along with
+# the updated tuple. But if the key is stored externally we must have to
+# detoast and log it separately. The patch to fix the bug is still pending[1],
+# the following tests for unchanged toasted key column would fail without
+# applying the bug fix patch. So temporarily keep the the following tests
+# commented before the bug fix patch is committed.
+# [1] https://postgr.es/m/OS0PR01MB611342D0A92D4F4BF26C0F47FB229@OS0PR01MB6113.jpnprd01.prod.outlook.com
+
+# UPDATE the non-key column for table tab_rowfilter_toast
+#$node_publisher->safe_psql('postgres',
+# "UPDATE tab_rowfilter_toast SET b = '1'");
+
+# Check expected replicated rows for tab_rowfilter_toast
+# tab_rowfilter_toast filter: (a = repeat('1234567890', 200))
+# UPDATE (repeat('1234567890', 200) ,'1') YES
+#$result =
+# $node_subscriber->safe_psql('postgres',
+# "SELECT a = repeat('1234567890', 200), b FROM tab_rowfilter_toast");
+#is($result, qq(t|1), 'check replicated rows to tab_rowfilter_toast');
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 89249ec..e4ae106 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2198,6 +2198,7 @@ ReorderBufferApplyChangeCB
ReorderBufferApplyTruncateCB
ReorderBufferBeginCB
ReorderBufferChange
+ReorderBufferChangeType
ReorderBufferCommitCB
ReorderBufferCommitPreparedCB
ReorderBufferDiskChange
@@ -3504,6 +3505,7 @@ replace_rte_variables_context
ret_type
rewind_source
rewrite_event
+rf_context
rijndael_ctx
rm_detail_t
role_auth_extra
--
2.7.2.windows.1