public inbox for [email protected]
help / color / mirror / Atom feedFrom: Euler Taveira <[email protected]>
To: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Ajin Cherian <[email protected]>
Cc: Greg Nancarrow <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: Rahila Syed <[email protected]>
Cc: Peter Eisentraut <[email protected]>
Cc: Önder Kalacı <[email protected]>
Cc: japin <[email protected]>
Cc: Michael Paquier <[email protected]>
Cc: David Steele <[email protected]>
Cc: Craig Ringer <[email protected]>
Cc: Amit Langote <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: row filtering for logical replication
Date: Mon, 20 Dec 2021 15:57:45 -0300
Message-ID: <[email protected]> (raw)
In-Reply-To: <OS0PR01MB5716BBB443019ED97FC48A0C947B9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
References: <CAHut+PvSHjUsrrgVeE43hN8AZVPa8VqQ=GH1LYJFSubp2W_dhA@mail.gmail.com>
<CAHut+PtRdXzPpm3qv3cEYWWfVUkGT84EopEHxwt95eo_cG_3eQ@mail.gmail.com>
<OS0PR01MB571625D4A5CC1DAB4045B2BB94919@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAHut+Pu1AMDUwBNKeevLKOO2=M6ZBXpddi-s7fhqBpMxMWUQpA@mail.gmail.com>
<CAFPTHDZgxAbLA98HZkN-JNHoHXSBGtK4eAGQrXam5JQY6Zx7wg@mail.gmail.com>
<OS0PR01MB61132C0E4FFEE73D34AE9823FB999@OS0PR01MB6113.jpnprd01.prod.outlook.com>
<CAJcOf-d0=Pnv8qpBaN5WM5wtmhT3EJ64crWtfPBtiwtK8iUEwg@mail.gmail.com>
<CAHut+Pv-D4rQseRO_OzfEz2dQsTKEnKjBCET9Z-iJppyT1XNMQ@mail.gmail.com>
<CALDaNm2bq-Zab3i5pvuA3UTxHvo3BqPwmgXbyznpw5vz4=fxpA@mail.gmail.com>
<OS0PR01MB57162EB465A0E6BCFDF9B3F394609@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KZ+N=vfgqeUuZfRLRC+WyNBFvU9Ei7kF-7OPXnTX2WrA@mail.gmail.com>
<OS0PR01MB5716F2C3904CB807050010A394619@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KVUs2s3PTcTTRj3qdmUzEDG7PQ5Xq_b1J7zMsi2OvE5w@mail.gmail.com>
<OS0PR01MB57168FD9932E3F42406EB13B94629@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAJcOf-dz0srExG0NPPgXh5X8eL2uxk7C=cZoGTbf8cNqoRUY6w@mail.gmail.com>
<CAA4eK1+8UqT5b3Ha42Wt07668yxs9KkKvV5r==vT9KfRUGCs-g@mail.gmail.com>
<[email protected]>
<CAA4eK1+hVXfOSScbf5LUB=5is=wYaC6NBhLxuvetbWQnZRnsVQ@mail.gmail.com>
<CAHut+PtJnnM8MYQDf7xCyFAp13U_0Ya2dv-UQeFD=ghixFLZiw@mail.gmail.com>
<[email protected]>
<CAA4eK1+C2Z8bmGT_7xo4T+fZqqTr1i56Hidv9krmaEDUACwGjA@mail.gmail.com>
<[email protected]>
<CAA4eK1K-+GhHqG+8yJb-neWxR_B+yeXTj7cCg9OqxTmPK3seRA@mail.gmail.com>
<CAHut+PugCTDwoS6Ysaocj+AjqSUuxB5qL6F=7_E2v59E9MZm7A@mail.gmail.com>
<CAFPTHDYB4nbxCMAFQGowJtDf7E6uBc==_HupBKy7MaMhM+9QQA@mail.gmail.com>
<CAHut+Ptoxjo6hpDFTya6WYH-zdspKQ5j+wZHBRc6EZkAkq7Nfw@mail.gmail.com>
<CAHut+Ptjsj_OVMWEdYp2Wq19=H5D4Vgta43FbFVDYr2LuS_djg@mail.gmail.com>
<CAA4eK1KbzN-WYXNpYdHheaB+eEQN+Rq4=1BQy=6TW2qLEw139g@mail.gmail.com>
<OS0PR01MB5716BBB443019ED97FC48A0C947B9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
On Mon, Dec 20, 2021, at 12:10 AM, [email protected] wrote:
> Attach the V49 patch set, which addressed all the above comments on the 0002
> patch.
I've been testing the latest versions of this patch set. I'm attaching a new
patch set based on v49. The suggested fixes are in separate patches after the
current one so it is easier to integrate them into the related patch. The
majority of these changes explains some decision to improve readability IMO.
row-filter x row filter. I'm not a native speaker but "row filter" is widely
used in similar contexts so I suggest to use it. (I didn't adjust the commit
messages)
An ancient patch use the term coerce but it was changed to cast. Coercion
implies an implicit conversion [1]. If you look at a few lines above you will
see that this expression expects an implicit conversion.
I modified the query to obtain the row filter expressions to (i) add the schema
pg_catalog to some objects and (ii) use NOT EXISTS instead of subquery (it
reads better IMO).
A detail message requires you to capitalize the first word of sentences and
includes a period at the end.
It seems all server messages and documentation use the terminology "WHERE
clause". Let's adopt it instead of "row filter".
I reviewed 0003. It uses TupleTableSlot instead of HeapTuple. I probably missed
the explanation but it requires more changes (logicalrep_write_tuple and 3 new
entries into RelationSyncEntry). I replaced this patch with a slightly
different one (0005 in this patch set) that uses HeapTuple instead. I didn't
only simple tests and it requires tests. I noticed that this patch does not
include a test to cover the case where TOASTed values are not included in the
new tuple. We should probably add one.
I agree with Amit that it is a good idea to merge 0001, 0002, and 0005. I would
probably merge 0004 because it is just isolated changes.
[1] https://en.wikipedia.org/wiki/Type_conversion
--
Euler Taveira
EDB https://www.enterprisedb.com/
Attachments:
[text/x-patch] v50-0001-Row-filter-for-logical-replication.patch (91.2K, ../[email protected]/3-v50-0001-Row-filter-for-logical-replication.patch)
download | inline diff:
From 4d22158a63a05e990b04af1d9c12992e31aa08e2 Mon Sep 17 00:00:00 2001
From: Peter Smith <[email protected]>
Date: Fri, 17 Dec 2021 18:59:42 +1100
Subject: [PATCH v50 1/8] Row-filter for logical replication.
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. Simple expressions
cannot contain any aggregate or window functions, non-immutable functions,
user-defined types, operators or functions. This restriction 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 which 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.
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
Discussion: https://www.postgresql.org/message-id/flat/CAHE3wggb715X%2BmK_DitLXF25B%3DjE6xyNCH4YOwM860JR7HarGQ%40mail.gmail.com
Combining multiple row-filters
==============================
The subscription is treated "as a union of all the publications" [1], so the
row-filters are combined with OR.
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.
Notice this means if one of the publications has no filter at all then all other
filters become redundant.
Author: Peter Smith
[1] https://www.postgresql.org/message-id/574b4e78-2f35-acf3-4bdc-4b872582e739%40enterprisedb.com
Row-filter caching
==================
The cached row-filters (e.g. ExprState *) are invalidated only in function
rel_sync_cache_relation_cb, so it means the ALTER PUBLICATION for one table
will not cause row-filters of other tables to also become invalidated.
The code related to caching row-filters is done just before they are needed
(in the pgoutput_row_filter function).
If there are multiple publication row-filters for a given table these are all
combined/flattened into a single filter.
Author: Peter Smith, Greg Nancarrow
The filter caching is based on a suggestions from Amit [1] [2], and Houz [3]
[1] https://www.postgresql.org/message-id/CAA4eK1%2BxQb06NGs6Y7OzwMtKYYixEqR8tdWV5THAVE4SAqNrDg%40mail.gmail.com
[2] https://www.postgresql.org/message-id/CAA4eK1%2Btio46goUKBUfAKFsFVxtgk8nOty%3DTxKoKH-gdLzHD2g%40mail.gmail.com
[3] https://www.postgresql.org/message-id/OS0PR01MB5716090A70A73ADF58C58950948D9%40OS0PR01MB5716.jpnprd01.prod.outlook.com
Cache ExprState per pubaction.
If a subscriber has multiple publications and these publications include the
same table then there can be multiple filters that apply to that table.
These filters are stored per-pubactions of the publications. There are 4 kinds
of pubaction ("insert", "update", "delete", "truncate"), but row-filters are
not applied for "truncate".
Filters for the same pubaction are all combined (OR'ed) and cached as one, so
at the end there are at most 3 cached filters per table.
The appropriate (pubaction) filter is executed according to the DML operation.
Author: Peter Smith
Discussion: https://www.postgresql.org/message-id/CAA4eK1%2BhVXfOSScbf5LUB%3D5is%3DwYaC6NBhLxuvetbWQnZRnsVQ%40mail.gmail.com
---
doc/src/sgml/catalogs.sgml | 8 +
doc/src/sgml/ref/alter_publication.sgml | 13 +-
doc/src/sgml/ref/create_publication.sgml | 37 +-
doc/src/sgml/ref/create_subscription.sgml | 24 +-
src/backend/catalog/pg_publication.c | 69 +++-
src/backend/commands/publicationcmds.c | 108 +++++-
src/backend/nodes/copyfuncs.c | 1 +
src/backend/nodes/equalfuncs.c | 1 +
src/backend/parser/gram.y | 38 +-
src/backend/parser/parse_agg.c | 10 +
src/backend/parser/parse_expr.c | 21 +-
src/backend/parser/parse_func.c | 3 +
src/backend/parser/parse_oper.c | 7 +
src/backend/parser/parse_relation.c | 9 +
src/backend/replication/logical/tablesync.c | 118 +++++-
src/backend/replication/pgoutput/pgoutput.c | 410 +++++++++++++++++++-
src/bin/psql/describe.c | 26 +-
src/include/catalog/pg_publication.h | 7 +-
src/include/catalog/pg_publication_rel.h | 6 +
src/include/nodes/parsenodes.h | 1 +
src/include/parser/parse_node.h | 1 +
src/test/regress/expected/publication.out | 151 +++++++
src/test/regress/sql/publication.sql | 76 ++++
src/test/subscription/t/027_row_filter.pl | 357 +++++++++++++++++
24 files changed, 1451 insertions(+), 51 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 03e2537b07..2f1f9132c6 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -6311,6 +6311,14 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
Reference to relation
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>prqual</structfield> <type>pg_node_tree</type>
+ </para>
+ <para>Expression tree (in <function>nodeToString()</function>
+ representation) for the relation's qualifying condition</para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml
index bb4ef5e5e2..5d9869c4f6 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 that do not
+ satisfy the <replaceable class="parameter">expression</replaceable> will
+ not be published. Note that parentheses are required around the
+ expression. The <replaceable class="parameter">expression</replaceable>
+ is evaluated with the role used for the replication connection.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index d805e8e77a..5aeee2309d 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>
@@ -78,6 +78,13 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
publication, so they are never explicitly added to the publication.
</para>
+ <para>
+ If the optional <literal>WHERE</literal> clause is specified, only rows
+ that satisfy the <replaceable class="parameter">expression</replaceable>
+ will 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,
@@ -225,6 +232,22 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
disallowed on those tables.
</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.
+ If nullable columns are present in the <literal>WHERE</literal> clause,
+ possible NULL values should be accounted for in expressions, to avoid
+ unexpected results, because <literal>NULL</literal> values can cause
+ those expressions to evaluate to false.
+ A <literal>WHERE</literal> clause allows simple expressions. The simple
+ expression cannot contain any aggregate or window functions, non-immutable
+ functions, user-defined types, operators or 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
@@ -247,6 +270,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>
@@ -259,6 +287,13 @@ CREATE PUBLICATION mypublication FOR TABLE users, departments;
</programlisting>
</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>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 990a41f1a1..db255f323a 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,23 @@ 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
+ that do not satisfy the <replaceable class="parameter">expression</replaceable>
+ will not be published (i.e. they will be filtered out).
+ If the subscription has several publications in which the same table has been
+ published with different <literal>WHERE</literal> clauses, those expressions
+ (for the same publish operation) get OR'ed together so that rows satisfying any
+ of the expressions will be published. Also, if one of the publications for the
+ same table has no <literal>WHERE</literal> clause at all, or is a <literal>FOR
+ ALL TABLES</literal> or <literal>FOR ALL TABLES IN SCHEMA</literal> publication,
+ then all other <literal>WHERE</literal> clauses (for the same publish operation)
+ become redundant.
+ If the subscriber is a <productname>PostgreSQL</productname> version before 15
+ then any row filtering is ignored during the initial data synchronization phase.
+ </para>
+
</refsect1>
<refsect1>
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 62f10bcbd2..0929aa0a35 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -36,6 +36,9 @@
#include "commands/publicationcmds.h"
#include "funcapi.h"
#include "miscadmin.h"
+#include "parser/parse_clause.h"
+#include "parser/parse_collate.h"
+#include "parser/parse_relation.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
@@ -275,22 +278,55 @@ GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
return result;
}
+/*
+ * Transform a publication WHERE clause, 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.
+ */
+Node *
+GetTransformedWhereClause(ParseState *pstate, PublicationRelInfo *pri,
+ bool fixup_collation)
+{
+ ParseNamespaceItem *nsitem;
+ Node *whereclause = NULL;
+
+ pstate->p_sourcetext = nodeToString(pri->whereClause);
+
+ nsitem = addRangeTableEntryForRelation(pstate, pri->relation,
+ AccessShareLock, NULL, false, false);
+
+ addNSItemToQuery(pstate, nsitem, false, true, true);
+
+ whereclause = transformWhereClause(pstate, copyObject(pri->whereClause),
+ EXPR_KIND_PUBLICATION_WHERE,
+ "PUBLICATION WHERE");
+
+ /* Fix up collation information */
+ if (fixup_collation)
+ assign_expr_collations(pstate, whereclause);
+
+ return whereclause;
+}
+
/*
* 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 prrelid;
Publication *pub = GetPublication(pubid);
ObjectAddress myself,
referenced;
+ ParseState *pstate;
+ Node *whereclause = NULL;
List *relids = NIL;
rel = table_open(PublicationRelRelationId, RowExclusiveLock);
@@ -311,10 +347,22 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("relation \"%s\" is already member of publication \"%s\"",
- RelationGetRelationName(targetrel->relation), pub->name)));
+ RelationGetRelationName(targetrel), pub->name)));
}
- check_publication_add_relation(targetrel->relation);
+ check_publication_add_relation(targetrel);
+
+ if (pri->whereClause != NULL)
+ {
+ /* Set up a ParseState to parse with */
+ pstate = make_parsestate(NULL);
+
+ /*
+ * Get the transformed WHERE clause, of boolean type, with necessary
+ * collation information.
+ */
+ whereclause = GetTransformedWhereClause(pstate, pri, true);
+ }
/* Form a tuple. */
memset(values, 0, sizeof(values));
@@ -328,6 +376,12 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
values[Anum_pg_publication_rel_prrelid - 1] =
ObjectIdGetDatum(relid);
+ /* Add qualifications, if available */
+ if (whereclause)
+ values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(whereclause));
+ else
+ nulls[Anum_pg_publication_rel_prqual - 1] = true;
+
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
/* Insert tuple into catalog. */
@@ -344,6 +398,13 @@ publication_add_relation(Oid pubid, PublicationRelInfo *targetrel,
ObjectAddressSet(referenced, RelationRelationId, relid);
recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
+ /* Add dependency on the objects mentioned in the qualifications */
+ if (whereclause)
+ {
+ recordDependencyOnExpr(&myself, whereclause, pstate->p_rtable, DEPENDENCY_NORMAL);
+ free_parsestate(pstate);
+ }
+
/* Close the table. */
table_close(rel, RowExclusiveLock);
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 404bb5d0c8..9ca743c6d2 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -529,40 +529,96 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
List *delrels = NIL;
ListCell *oldlc;
+ /*
+ * Check if the relation is member of the existing schema in the
+ * publication or member of the schema list specified.
+ */
CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
PUBLICATIONOBJ_TABLE);
- /* Calculate which relations to drop. */
+ /*
+ * In order to recreate the relation list for the publication,
+ * look for existing relations that need not be dropped.
+ */
foreach(oldlc, oldrelids)
{
Oid oldrelid = lfirst_oid(oldlc);
- ListCell *newlc;
+ 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);
+
+ /*
+ * Look 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 (rfisnull && !newpubrel->whereClause)
+ {
+ found = true;
+ break;
+ }
+
+ if (!rfisnull && newpubrel->whereClause)
+ {
+ ParseState *pstate = make_parsestate(NULL);
+ Node *whereclause;
+
+ whereclause = GetTransformedWhereClause(pstate,
+ newpubrel,
+ false);
+ if (equal(oldrelwhereclause, whereclause))
+ {
+ free_parsestate(pstate);
+ found = true;
+ break;
+ }
+
+ free_parsestate(pstate);
+ }
}
}
- /* Not yet in the list, open it and add to the list */
+
+ if (oldrelwhereclause)
+ pfree(oldrelwhereclause);
+
+ /*
+ * Add the non-matched relations to a list so that they can
+ * be dropped.
+ */
if (!found)
{
- Relation oldrel;
- PublicationRelInfo *pubrel;
-
- /* Wrap relation into PublicationRelInfo */
- oldrel = table_open(oldrelid, ShareUpdateExclusiveLock);
-
- pubrel = palloc(sizeof(PublicationRelInfo));
- pubrel->relation = oldrel;
-
- delrels = lappend(delrels, pubrel);
+ oldrel = palloc(sizeof(PublicationRelInfo));
+ oldrel->whereClause = NULL;
+ oldrel->relation = table_open(oldrelid,
+ ShareUpdateExclusiveLock);
+ delrels = lappend(delrels, oldrel);
}
}
@@ -899,6 +955,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
@@ -926,15 +983,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 row-filters for \"%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
@@ -967,6 +1035,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);
}
@@ -974,6 +1044,7 @@ OpenTableList(List *tables)
}
list_free(relids);
+ list_free(relids_with_rf);
return rels;
}
@@ -993,6 +1064,8 @@ CloseTableList(List *rels)
pub_rel = (PublicationRelInfo *) lfirst(lc);
table_close(pub_rel->relation, NoLock);
}
+
+ list_free_deep(rels);
}
/*
@@ -1088,6 +1161,11 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
RelationGetRelationName(rel))));
}
+ if (pubrel->whereClause)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("invalid use of WHERE row-filter in ALTER PUBLICATION ... DROP TABLE")));
+
ObjectAddressSet(obj, PublicationRelRelationId, prid);
performDeletion(&obj, DROP_CASCADE, 0);
}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index df0b747883..bd55ea667f 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4833,6 +4833,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 cb7ddd463c..028b8e5dc0 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 3d4dd43e47..9da93a01a1 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9742,12 +9742,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
{
@@ -9762,28 +9763,45 @@ PublicationObjSpec:
$$->pubobjtype = PUBLICATIONOBJ_TABLE_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. If the ColId was mistakenly
+ * not a table this will be detected later in
+ * preprocess_pubobj_list() and an error is thrown.
+ */
+ $$->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
{
@@ -17431,7 +17449,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_TABLE_IN_SCHEMA ||
pubobj->pubobjtype == PUBLICATIONOBJ_TABLE_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 for schema not allowed"),
+ 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/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index 7d829a05a9..193c87d8b7 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -551,6 +551,13 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr)
err = _("grouping operations are not allowed in COPY FROM WHERE conditions");
break;
+ case EXPR_KIND_PUBLICATION_WHERE:
+ if (isAgg)
+ err = _("aggregate functions are not allowed in publication WHERE expressions");
+ else
+ err = _("grouping operations are not allowed in publication WHERE expressions");
+
+ break;
case EXPR_KIND_CYCLE_MARK:
errkind = true;
@@ -943,6 +950,9 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
+ case EXPR_KIND_PUBLICATION_WHERE:
+ err = _("window functions are not allowed in publication WHERE expressions");
+ break;
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c
index 2d1a477154..3d43839b35 100644
--- a/src/backend/parser/parse_expr.c
+++ b/src/backend/parser/parse_expr.c
@@ -200,8 +200,19 @@ transformExprRecurse(ParseState *pstate, Node *expr)
break;
case T_FuncCall:
- result = transformFuncCall(pstate, (FuncCall *) expr);
- break;
+ {
+ /*
+ * Forbid functions in publication WHERE condition
+ */
+ if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("functions are not allowed in publication WHERE expressions"),
+ parser_errposition(pstate, exprLocation(expr))));
+
+ result = transformFuncCall(pstate, (FuncCall *) expr);
+ break;
+ }
case T_MultiAssignRef:
result = transformMultiAssignRef(pstate, (MultiAssignRef *) expr);
@@ -504,6 +515,7 @@ transformColumnRef(ParseState *pstate, ColumnRef *cref)
case EXPR_KIND_COPY_WHERE:
case EXPR_KIND_GENERATED_COLUMN:
case EXPR_KIND_CYCLE_MARK:
+ case EXPR_KIND_PUBLICATION_WHERE:
/* okay */
break;
@@ -1764,6 +1776,9 @@ transformSubLink(ParseState *pstate, SubLink *sublink)
case EXPR_KIND_GENERATED_COLUMN:
err = _("cannot use subquery in column generation expression");
break;
+ case EXPR_KIND_PUBLICATION_WHERE:
+ err = _("cannot use subquery in publication WHERE expression");
+ break;
/*
* There is intentionally no default: case here, so that the
@@ -3084,6 +3099,8 @@ ParseExprKindName(ParseExprKind exprKind)
return "GENERATED AS";
case EXPR_KIND_CYCLE_MARK:
return "CYCLE";
+ case EXPR_KIND_PUBLICATION_WHERE:
+ return "publication expression";
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c
index 542f9167aa..29bebb73eb 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2655,6 +2655,9 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location)
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
+ case EXPR_KIND_PUBLICATION_WHERE:
+ err = _("set-returning functions are not allowed in publication WHERE expressions");
+ break;
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_oper.c b/src/backend/parser/parse_oper.c
index bc34a23afc..29f8835ce1 100644
--- a/src/backend/parser/parse_oper.c
+++ b/src/backend/parser/parse_oper.c
@@ -718,6 +718,13 @@ make_op(ParseState *pstate, List *opname, Node *ltree, Node *rtree,
opform->oprright)),
parser_errposition(pstate, location)));
+ /* Check it's not a custom operator for publication WHERE expressions */
+ if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE && opform->oid >= FirstNormalObjectId)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("user-defined operators are not allowed in publication WHERE expressions"),
+ parser_errposition(pstate, location)));
+
/* Do typecasting and build the expression tree */
if (ltree == NULL)
{
diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index c5c3f26ecf..036d9c6d26 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -3538,11 +3538,20 @@ errorMissingRTE(ParseState *pstate, RangeVar *relation)
rte->eref->aliasname)),
parser_errposition(pstate, relation->location)));
else
+ {
+ if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_TABLE),
+ errmsg("publication WHERE expression invalid reference to table \"%s\"",
+ relation->relname),
+ parser_errposition(pstate, relation->location)));
+
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_TABLE),
errmsg("missing FROM-clause entry for table \"%s\"",
relation->relname),
parser_errposition(pstate, relation->location)));
+ }
}
/*
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a43c..c20c2219fc 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -687,20 +687,24 @@ copy_read_data(void *outbuf, int minread, int maxread)
/*
- * Get information about remote relation in similar fashion the RELATION
- * message provides during replication.
+ * Get information about a remote relation, in a similar fashion to how the
+ * RELATION message provides information 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;
@@ -796,6 +800,80 @@ fetch_remote_table_info(char *nspname, char *relname,
lrel->natts = natt;
walrcv_clear_result(res);
+
+ /*
+ * Get relation qual. DISTINCT avoids the same expression of a table in
+ * multiple publications from being included multiple times in the final
+ * expression.
+ */
+ if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
+ {
+ resetStringInfo(&cmd);
+ appendStringInfo(&cmd,
+ "SELECT DISTINCT pg_get_expr(prqual, prrelid) "
+ " FROM pg_publication p "
+ " INNER JOIN pg_publication_rel pr "
+ " ON (p.oid = pr.prpubid) "
+ " WHERE pr.prrelid = %u "
+ " AND p.pubname IN (", lrel->remoteid);
+
+ first = true;
+ foreach(lc, MySubscription->publications)
+ {
+ char *pubname = strVal(lfirst(lc));
+
+ if (first)
+ first = false;
+ else
+ appendStringInfoString(&cmd, ", ");
+
+ appendStringInfoString(&cmd, quote_literal_cstr(pubname));
+ }
+ appendStringInfoChar(&cmd, ')');
+
+ res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ ereport(ERROR,
+ (errmsg("could not fetch relation qualifications for table \"%s.%s\" from publisher: %s",
+ nspname, relname, res->err)));
+
+ /*
+ * 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)));
+
+ ExecClearTuple(slot);
+
+ /*
+ * One entry without a row filter expression means clean up
+ * previous expressions (if there are any) and return with no
+ * expressions.
+ */
+ if (isnull)
+ {
+ if (*qual)
+ {
+ list_free_deep(*qual);
+ *qual = NIL;
+ }
+ break;
+ }
+ }
+ ExecDropSingleTupleTableSlot(slot);
+
+ walrcv_clear_result(res);
+ }
+
pfree(cmd.data);
}
@@ -809,6 +887,7 @@ copy_table(Relation rel)
{
LogicalRepRelMapEntry *relmapentry;
LogicalRepRelation lrel;
+ List *qual = NIL;
WalRcvExecResult *res;
StringInfoData cmd;
CopyFromState cstate;
@@ -817,7 +896,7 @@ copy_table(Relation rel)
/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
- RelationGetRelationName(rel), &lrel);
+ RelationGetRelationName(rel), &lrel, &qual);
/* Put the relation into relmap. */
logicalrep_relmap_update(&lrel);
@@ -828,14 +907,18 @@ copy_table(Relation rel)
/* Start copy on the publisher. */
initStringInfo(&cmd);
- if (lrel.relkind == RELKIND_RELATION)
+
+ /* Regular table with no row filter */
+ if (lrel.relkind == RELKIND_RELATION && qual == NIL)
appendStringInfo(&cmd, "COPY %s TO STDOUT",
quote_qualified_identifier(lrel.nspname, lrel.relname));
else
{
/*
* For non-tables, we need to do COPY (SELECT ...), but we can't just
- * do SELECT * because we need to not copy generated columns.
+ * do SELECT * because we need to not copy generated columns. For
+ * tables with any row filters, build a SELECT query with OR'ed row
+ * filters for COPY.
*/
appendStringInfoString(&cmd, "COPY (SELECT ");
for (int i = 0; i < lrel.natts; i++)
@@ -844,8 +927,29 @@ copy_table(Relation rel)
if (i < lrel.natts - 1)
appendStringInfoString(&cmd, ", ");
}
- appendStringInfo(&cmd, " FROM %s) TO STDOUT",
+ appendStringInfo(&cmd, " FROM %s",
quote_qualified_identifier(lrel.nspname, lrel.relname));
+ /* list of OR'ed filters */
+ if (qual != NIL)
+ {
+ ListCell *lc;
+ bool first = true;
+
+ appendStringInfoString(&cmd, " WHERE ");
+ foreach(lc, qual)
+ {
+ char *q = strVal(lfirst(lc));
+
+ if (first)
+ first = false;
+ else
+ appendStringInfoString(&cmd, " OR ");
+ appendStringInfoString(&cmd, q);
+ }
+ list_free_deep(qual);
+ }
+
+ appendStringInfoString(&cmd, ") TO STDOUT");
}
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 6f6a203dea..2fa08e7278 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -15,16 +15,24 @@
#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/nodeFuncs.h"
+#include "nodes/makefuncs.h"
+#include "optimizer/optimizer.h"
+#include "parser/parse_coerce.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"
@@ -115,6 +123,24 @@ typedef struct RelationSyncEntry
bool replicate_valid;
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. Only 3 publication actions are used
+ * for row filtering ("insert", "update", "delete"). The exprstate array is
+ * indexed by ReorderBufferChangeType.
+ */
+ bool exprstate_valid;
+#define IDX_PUBACTION_n 3
+ ExprState *exprstate[IDX_PUBACTION_n]; /* ExprState array for row filter.
+ One per publication action. */
+ TupleTableSlot *scantuple; /* tuple table slot for row filter */
+
/*
* OID of the relation to publish changes as. For a partition, this may
* be set to one of its ancestors whose schema will be used when
@@ -137,7 +163,7 @@ static HTAB *RelationSyncCache = NULL;
static void init_rel_sync_cache(MemoryContext decoding_context);
static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit);
-static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data, Oid relid);
+static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data, Relation relation);
static void rel_sync_cache_relation_cb(Datum arg, Oid relid);
static void rel_sync_cache_publication_cb(Datum arg, int cacheid,
uint32 hashvalue);
@@ -146,6 +172,14 @@ static void set_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
TransactionId xid);
+/* row filter routines */
+static EState *create_estate_for_relation(Relation rel);
+static ExprState *pgoutput_row_filter_init_expr(Node *rfnode);
+static bool pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext);
+static bool pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
+ Relation relation, HeapTuple oldtuple,
+ HeapTuple newtuple, RelationSyncEntry *entry);
+
/*
* Specify output plugin callbacks
*/
@@ -620,6 +654,316 @@ send_relation_and_attrs(Relation relation, TransactionId xid,
OutputPluginWrite(ctx, false);
}
+/*
+ * Executor state preparation for evaluation of row filter expressions for the
+ * specified relation.
+ */
+static EState *
+create_estate_for_relation(Relation rel)
+{
+ EState *estate;
+ RangeTblEntry *rte;
+
+ estate = CreateExecutorState();
+
+ rte = makeNode(RangeTblEntry);
+ rte->rtekind = RTE_RELATION;
+ rte->relid = RelationGetRelid(rel);
+ rte->relkind = rel->rd_rel->relkind;
+ rte->rellockmode = AccessShareLock;
+ ExecInitRangeTable(estate, list_make1(rte));
+
+ estate->es_output_cid = GetCurrentCommandId(false);
+
+ return estate;
+}
+
+/*
+ * Initialize for row filter expression execution.
+ */
+static ExprState *
+pgoutput_row_filter_init_expr(Node *rfnode)
+{
+ ExprState *exprstate;
+ Oid exprtype;
+ Expr *expr;
+
+ /* Cache ExprState using CacheMemoryContext. */
+ Assert(CurrentMemoryContext = CacheMemoryContext);
+
+ /* Prepare expression for execution */
+ exprtype = exprType(rfnode);
+ expr = (Expr *) coerce_to_target_type(NULL, rfnode, exprtype, BOOLOID, -1, COERCION_ASSIGNMENT, COERCE_IMPLICIT_CAST, -1);
+
+ if (expr == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_CANNOT_COERCE),
+ errmsg("row filter returns type %s that cannot be cast to the expected type %s",
+ format_type_be(exprtype),
+ format_type_be(BOOLOID)),
+ errhint("You will need to rewrite the row filter.")));
+
+ /*
+ * This is the same code as ExecPrepareExpr() but that is not used because
+ * we have no EState to pass it. There should probably be another function
+ * in the executor to handle the execution outside a normal Plan tree
+ * context.
+ */
+ expr = expression_planner(expr);
+ 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 ? "true" : "false");
+
+ if (isnull)
+ return false;
+
+ return DatumGetBool(ret);
+}
+
+/*
+ * Change is checked against the row filter, if any.
+ *
+ * If it returns true, the change is replicated, otherwise, it is not.
+ */
+static bool
+pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
+ Relation relation, HeapTuple oldtuple, HeapTuple newtuple,
+ RelationSyncEntry *entry)
+{
+ EState *estate;
+ ExprContext *ecxt;
+ ListCell *lc;
+ bool result = true;
+ Oid relid = RelationGetRelid(relation);
+ List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */
+ bool no_filter[] = {false, false, false}; /* One per pubaction */
+
+ Assert(changetype == REORDER_BUFFER_CHANGE_INSERT ||
+ changetype == REORDER_BUFFER_CHANGE_UPDATE ||
+ changetype == REORDER_BUFFER_CHANGE_DELETE);
+ /*
+ * 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)
+ {
+ MemoryContext oldctx;
+ int idx;
+ bool found_filters = false;
+ int idx_ins = REORDER_BUFFER_CHANGE_INSERT;
+ int idx_upd = REORDER_BUFFER_CHANGE_UPDATE;
+ int idx_del = REORDER_BUFFER_CHANGE_DELETE;
+
+ /*
+ * Find if there are any row filters for this relation. If there are,
+ * then prepare the necessary ExprState and cache it in entry->exprstate.
+ *
+ * NOTE: All publication-table mappings must be checked.
+ *
+ * NOTE: If the relation is a partition and pubviaroot is true, use
+ * the row filter of the topmost partitioned table instead of the row
+ * filter of its own partition.
+ *
+ * NOTE: 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) which row filters
+ * will be appended.
+ */
+ foreach(lc, data->publications)
+ {
+ Publication *pub = lfirst(lc);
+ HeapTuple rftuple;
+ Datum rfdatum;
+ bool rfisnull;
+
+ /*
+ * Lookup if there is a row-filter, and if yes remember it in a list (per
+ * pubaction). If no, then remember there was no filter for this pubaction.
+ * Code following this 'publications' loop will combine all filters.
+ */
+ rftuple = SearchSysCache2(PUBLICATIONRELMAP, ObjectIdGetDatum(entry->publish_as_relid), ObjectIdGetDatum(pub->oid));
+ if (HeapTupleIsValid(rftuple))
+ {
+ rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple, Anum_pg_publication_rel_prqual, &rfisnull);
+
+ if (!rfisnull)
+ {
+ Node *rfnode;
+
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ /* Gather the rfnodes per pubaction of this publiaction. */
+ if (pub->pubactions.pubinsert)
+ {
+ rfnode = stringToNode(TextDatumGetCString(rfdatum));
+ rfnodes[idx_ins] = lappend(rfnodes[idx_ins], rfnode);
+ }
+ if (pub->pubactions.pubupdate)
+ {
+ rfnode = stringToNode(TextDatumGetCString(rfdatum));
+ rfnodes[idx_upd] = lappend(rfnodes[idx_upd], rfnode);
+ }
+ if (pub->pubactions.pubdelete)
+ {
+ rfnode = stringToNode(TextDatumGetCString(rfdatum));
+ rfnodes[idx_del] = lappend(rfnodes[idx_del], rfnode);
+ }
+ MemoryContextSwitchTo(oldctx);
+ }
+ else
+ {
+ /* Remember which pubactions have no row-filter. */
+ if (pub->pubactions.pubinsert)
+ no_filter[idx_ins] = true;
+ if (pub->pubactions.pubupdate)
+ no_filter[idx_upd] = true;
+ if (pub->pubactions.pubdelete)
+ no_filter[idx_del] = true;
+
+ /* Quick exit loop if all pubactions have no row-filter. */
+ if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
+ {
+ ReleaseSysCache(rftuple);
+ break;
+ }
+ }
+
+ ReleaseSysCache(rftuple);
+ }
+
+ } /* loop all subscribed publications */
+
+ /*
+ * 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.
+ */
+ for (idx = 0; idx < IDX_PUBACTION_n; idx++)
+ {
+ int n_filters;
+
+ if (no_filter[idx])
+ {
+ if (rfnodes[idx])
+ {
+ list_free_deep(rfnodes[idx]);
+ rfnodes[idx] = NIL;
+ }
+ }
+
+ /*
+ * If there was one or more filter for this pubaction then combine them
+ * (if necessary) and cache the ExprState.
+ */
+ n_filters = list_length(rfnodes[idx]);
+ if (n_filters > 0)
+ {
+ Node *rfnode;
+
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ rfnode = n_filters > 1 ? makeBoolExpr(OR_EXPR, rfnodes[idx], -1) : linitial(rfnodes[idx]);
+ entry->exprstate[idx] = pgoutput_row_filter_init_expr(rfnode);
+ MemoryContextSwitchTo(oldctx);
+
+ found_filters = true; /* flag that we will need slots made */
+ }
+ } /* for each pubaction */
+
+ if (found_filters)
+ {
+ TupleDesc tupdesc = RelationGetDescr(relation);
+
+ /*
+ * Create tuple table slots for row filter. Create a copy of the
+ * TupleDesc as it needs to live as long as the cache remains.
+ */
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ tupdesc = CreateTupleDescCopy(tupdesc);
+ entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+ MemoryContextSwitchTo(oldctx);
+ }
+
+ entry->exprstate_valid = true;
+ }
+
+ /* Bail out if there is no row filter */
+ if (!entry->exprstate[changetype])
+ return true;
+
+ if (message_level_is_interesting(DEBUG3))
+ elog(DEBUG3, "table \"%s.%s\" has row filter",
+ get_namespace_name(get_rel_namespace(relid)),
+ get_rel_name(relid));
+
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ estate = create_estate_for_relation(relation);
+
+ /* Prepare context per tuple */
+ ecxt = GetPerTupleExprContext(estate);
+ ecxt->ecxt_scantuple = entry->scantuple;
+
+ ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
+
+ /*
+ * NOTE: Multiple publication row-filters have already been combined to a
+ * single exprstate (for this pubaction).
+ */
+ if (entry->exprstate[changetype])
+ {
+ /* Evaluates row filter */
+ result = pgoutput_row_filter_exec_expr(entry->exprstate[changetype], ecxt);
+ }
+
+ /* Cleanup allocated resources */
+ ResetExprContext(ecxt);
+ FreeExecutorState(estate);
+ PopActiveSnapshot();
+
+ return result;
+}
+
/*
* Sends the decoded DML over wire.
*
@@ -647,7 +991,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
if (in_streaming)
xid = change->txn->xid;
- relentry = get_rel_sync_entry(data, RelationGetRelid(relation));
+ relentry = get_rel_sync_entry(data, relation);
/* First check the table filter */
switch (change->action)
@@ -671,8 +1015,6 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Avoid leaking memory by using and resetting our own context */
old = MemoryContextSwitchTo(data->context);
- maybe_send_schema(ctx, change, relation, relentry);
-
/* Send the data */
switch (change->action)
{
@@ -680,6 +1022,16 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
{
HeapTuple tuple = &change->data.tp.newtuple->tuple;
+ /* Check row filter. */
+ if (!pgoutput_row_filter(change->action, data, relation, NULL, tuple, relentry))
+ break;
+
+ /*
+ * Schema should be sent before the logic that replaces the
+ * relation because it also sends the ancestor's relation.
+ */
+ maybe_send_schema(ctx, change, relation, relentry);
+
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
{
@@ -703,6 +1055,12 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
&change->data.tp.oldtuple->tuple : NULL;
HeapTuple newtuple = &change->data.tp.newtuple->tuple;
+ /* Check row filter. */
+ if (!pgoutput_row_filter(change->action, data, relation, oldtuple, newtuple, relentry))
+ break;
+
+ maybe_send_schema(ctx, change, relation, relentry);
+
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
{
@@ -731,6 +1089,12 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
{
HeapTuple oldtuple = &change->data.tp.oldtuple->tuple;
+ /* Check row filter. */
+ if (!pgoutput_row_filter(change->action, data, relation, oldtuple, NULL, relentry))
+ break;
+
+ maybe_send_schema(ctx, change, relation, relentry);
+
/* Switch relation if publishing via root. */
if (relentry->publish_as_relid != RelationGetRelid(relation))
{
@@ -794,7 +1158,7 @@ pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
if (!is_publishable_relation(relation))
continue;
- relentry = get_rel_sync_entry(data, relid);
+ relentry = get_rel_sync_entry(data, relation);
if (!relentry->pubactions.pubtruncate)
continue;
@@ -1116,9 +1480,10 @@ set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
* when publishing.
*/
static RelationSyncEntry *
-get_rel_sync_entry(PGOutputData *data, Oid relid)
+get_rel_sync_entry(PGOutputData *data, Relation relation)
{
RelationSyncEntry *entry;
+ Oid relid = RelationGetRelid(relation);
bool am_partition = get_rel_relispartition(relid);
char relkind = get_rel_relkind(relid);
bool found;
@@ -1139,8 +1504,13 @@ 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->scantuple = NULL;
+ entry->exprstate[REORDER_BUFFER_CHANGE_INSERT] = NULL;
+ entry->exprstate[REORDER_BUFFER_CHANGE_UPDATE] = NULL;
+ entry->exprstate[REORDER_BUFFER_CHANGE_DELETE] = NULL;
entry->publish_as_relid = InvalidOid;
entry->map = NULL; /* will be set by maybe_send_schema() if
* needed */
@@ -1245,9 +1615,6 @@ get_rel_sync_entry(PGOutputData *data, Oid relid)
entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate;
}
- if (entry->pubactions.pubinsert && entry->pubactions.pubupdate &&
- entry->pubactions.pubdelete && entry->pubactions.pubtruncate)
- break;
}
list_free(pubids);
@@ -1310,6 +1677,7 @@ static void
rel_sync_cache_relation_cb(Datum arg, Oid relid)
{
RelationSyncEntry *entry;
+ int idx;
/*
* We can get here if the plugin was used in SQL interface as the
@@ -1354,6 +1722,25 @@ rel_sync_cache_relation_cb(Datum arg, Oid relid)
free_conversion_map(entry->map);
}
entry->map = NULL;
+
+ /*
+ * Row filter cache cleanups. (Will be rebuilt later if needed).
+ */
+ entry->exprstate_valid = false;
+ if (entry->scantuple != NULL)
+ {
+ ExecDropSingleTupleTableSlot(entry->scantuple);
+ entry->scantuple = NULL;
+ }
+ /* Cleanup the ExprState for each of the pubactions. */
+ for (idx = 0; idx < IDX_PUBACTION_n; idx++)
+ {
+ if (entry->exprstate[idx] != NULL)
+ {
+ pfree(entry->exprstate[idx]);
+ entry->exprstate[idx] = NULL;
+ }
+ }
}
}
@@ -1365,6 +1752,7 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
{
HASH_SEQ_STATUS status;
RelationSyncEntry *entry;
+ MemoryContext oldctx;
/*
* We can get here if the plugin was used in SQL interface as the
@@ -1374,6 +1762,8 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
if (RelationSyncCache == NULL)
return;
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+
/*
* There is no way to find which entry in our cache the hash belongs to so
* mark the whole cache as invalid.
@@ -1392,6 +1782,8 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue)
entry->pubactions.pubdelete = false;
entry->pubactions.pubtruncate = false;
}
+
+ MemoryContextSwitchTo(oldctx);
}
/* Send Replication origin */
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index c28788e84f..929b2f5388 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);
@@ -5833,8 +5844,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));
@@ -5963,8 +5978,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 902f2f2f0d..96c55f627d 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -20,6 +20,7 @@
#include "catalog/genbki.h"
#include "catalog/objectaddress.h"
#include "catalog/pg_publication_d.h"
+#include "parser/parse_node.h"
/* ----------------
* pg_publication definition. cpp turns this into
@@ -86,6 +87,7 @@ typedef struct Publication
typedef struct PublicationRelInfo
{
Relation relation;
+ Node *whereClause;
} PublicationRelInfo;
extern Publication *GetPublication(Oid pubid);
@@ -123,13 +125,16 @@ extern List *GetPubPartitionOptionRelations(List *result,
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);
extern Oid get_publication_oid(const char *pubname, bool missing_ok);
extern char *get_publication_name(Oid pubid, bool missing_ok);
+extern Node *GetTransformedWhereClause(ParseState *pstate,
+ PublicationRelInfo *pri,
+ bool bfixupcollation);
#endif /* PG_PUBLICATION_H */
diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h
index b5d5504cbb..154bb61777 100644
--- a/src/include/catalog/pg_publication_rel.h
+++ b/src/include/catalog/pg_publication_rel.h
@@ -31,6 +31,10 @@ CATALOG(pg_publication_rel,6106,PublicationRelRelationId)
Oid oid; /* oid */
Oid prpubid BKI_LOOKUP(pg_publication); /* Oid of the publication */
Oid prrelid BKI_LOOKUP(pg_class); /* Oid of the relation */
+
+#ifdef CATALOG_VARLEN /* variable-length fields start here */
+ pg_node_tree prqual; /* qualifications */
+#endif
} FormData_pg_publication_rel;
/* ----------------
@@ -40,6 +44,8 @@ CATALOG(pg_publication_rel,6106,PublicationRelRelationId)
*/
typedef FormData_pg_publication_rel *Form_pg_publication_rel;
+DECLARE_TOAST(pg_publication_rel, 8287, 8288);
+
DECLARE_UNIQUE_INDEX_PKEY(pg_publication_rel_oid_index, 6112, PublicationRelObjectIndexId, on pg_publication_rel using btree(oid oid_ops));
DECLARE_UNIQUE_INDEX(pg_publication_rel_prrelid_prpubid_index, 6113, PublicationRelPrrelidPrpubidIndexId, on pg_publication_rel using btree(prrelid oid_ops, prpubid oid_ops));
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 4c5a8a39bf..e437a55bb2 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/parser/parse_node.h b/src/include/parser/parse_node.h
index ee179082ce..d58ae6a63f 100644
--- a/src/include/parser/parse_node.h
+++ b/src/include/parser/parse_node.h
@@ -80,6 +80,7 @@ typedef enum ParseExprKind
EXPR_KIND_COPY_WHERE, /* WHERE condition in COPY FROM */
EXPR_KIND_GENERATED_COLUMN, /* generation expression for a column */
EXPR_KIND_CYCLE_MARK, /* cycle mark value */
+ EXPR_KIND_PUBLICATION_WHERE /* WHERE condition for a table in PUBLICATION */
} ParseExprKind;
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 5ac2d666a2..5a49003ae4 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -239,6 +239,157 @@ 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 SCHEMA testpub_rf_myschema;
+CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
+CREATE SCHEMA testpub_rf_myschema1;
+CREATE TABLE testpub_rf_myschema1.testpub_rf_tbl6(i integer);
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5);
+RESET client_min_messages;
+\dRp+ testpub5
+ Publication testpub5
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | t | t | t | f
+Tables:
+ "public.testpub_rf_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 | t | t | t | 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 | t | t | t | 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 | t | t | t | 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);
+RESET client_min_messages;
+\dRp+ testpub_syntax1
+ Publication testpub_syntax1
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | t | t | t | 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_myschema.testpub_rf_tbl5 WHERE (h < 999);
+RESET client_min_messages;
+\dRp+ testpub_syntax2
+ Publication testpub_syntax2
+ Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | t | t | t | t | f
+Tables:
+ "public.testpub_rf_tbl1"
+ "testpub_rf_myschema.testpub_rf_tbl5" WHERE (h < 999)
+
+DROP PUBLICATION testpub_syntax2;
+-- fail - schemas are not allowed WHERE row-filter
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a = 123);
+ERROR: syntax error at or near "WHERE"
+LINE 1: ...tax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a =...
+ ^
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf_myschema WHERE (a = 123);
+ERROR: WHERE clause for schema not allowed
+LINE 1: ...ax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf...
+ ^
+RESET client_min_messages;
+-- fail - duplicate tables are not allowed if that table has any WHERE row-filters
+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 row-filters for "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 row-filters for "testpub_rf_tbl1"
+RESET client_min_messages;
+-- fail - aggregate functions not allowed in WHERE clause
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
+ERROR: functions are not allowed in publication WHERE expressions
+LINE 1: ...TION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
+ ^
+-- fail - functions disallowed
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+ERROR: functions are not allowed in publication WHERE expressions
+LINE 1: ...ICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) ...
+ ^
+-- fail - user-defined operators disallowed
+CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+ERROR: user-defined operators are not allowed in publication WHERE expressions
+LINE 1: ...ICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+ ^
+-- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
+ERROR: invalid use of WHERE row-filter in ALTER PUBLICATION ... DROP TABLE
+-- fail - cannot ALTER SET table which is a member of a pre-existing schema
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_myschema1;
+ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_myschema1, TABLE testpub_rf_myschema1.testpub_rf_tbl6 WHERE (i < 99);
+ERROR: cannot add relation "testpub_rf_myschema1.testpub_rf_tbl6" to publication
+DETAIL: Table's schema "testpub_rf_myschema1" 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_myschema.testpub_rf_tbl5;
+DROP TABLE testpub_rf_myschema1.testpub_rf_tbl6;
+DROP SCHEMA testpub_rf_myschema;
+DROP SCHEMA testpub_rf_myschema1;
+DROP PUBLICATION testpub5;
+DROP PUBLICATION testpub7;
+DROP OPERATOR =#>(integer, integer);
+DROP FUNCTION testpub_rf_func(integer, integer);
-- 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 56dd358554..47bdba86ac 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -134,6 +134,82 @@ 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 SCHEMA testpub_rf_myschema;
+CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
+CREATE SCHEMA testpub_rf_myschema1;
+CREATE TABLE testpub_rf_myschema1.testpub_rf_tbl6(i integer);
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5);
+RESET client_min_messages;
+\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);
+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_myschema.testpub_rf_tbl5 WHERE (h < 999);
+RESET client_min_messages;
+\dRp+ testpub_syntax2
+DROP PUBLICATION testpub_syntax2;
+-- fail - schemas are not allowed WHERE row-filter
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf_myschema WHERE (a = 123);
+RESET client_min_messages;
+-- fail - duplicate tables are not allowed if that table has any WHERE row-filters
+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 - aggregate functions not allowed in WHERE clause
+ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
+-- fail - functions disallowed
+ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
+-- fail - user-defined operators disallowed
+CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
+CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+CREATE PUBLICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
+-- fail - WHERE not allowed in DROP
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
+-- fail - cannot ALTER SET table which is a member of a pre-existing schema
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_myschema1;
+ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_myschema1, TABLE testpub_rf_myschema1.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_myschema.testpub_rf_tbl5;
+DROP TABLE testpub_rf_myschema1.testpub_rf_tbl6;
+DROP SCHEMA testpub_rf_myschema;
+DROP SCHEMA testpub_rf_myschema1;
+DROP PUBLICATION testpub5;
+DROP PUBLICATION testpub7;
+DROP OPERATOR =#>(integer, integer);
+DROP FUNCTION testpub_rf_func(integer, integer);
+
-- 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 0000000000..64e71d0adb
--- /dev/null
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -0,0 +1,357 @@
+# Test logical replication behavior with row filtering
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More tests => 10;
+
+# 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;
+
+# 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',
+ "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)"
+);
+
+# 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)"
+);
+
+# 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"
+);
+
+#
+# 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)");
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_1, tap_pub_2, tap_pub_3, tap_pub_4a, tap_pub_4b"
+);
+
+$node_publisher->wait_for_catchup($appname);
+
+# wait for initial table synchronization to finish
+my $synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check expected replicated rows for tab_rowfilter_1
+# tap_pub_1 filter is: (a > 1000 AND b <> 'filtered')
+# - INSERT (1, 'not replicated') NO, because a is not > 1000
+# - INSERT (1500, 'filtered') NO, because b == 'filtered'
+# - INSERT (1980, 'not filtered') YES
+# - generate_series(990,1002) YES, only for 1001,1002 because a > 1000
+#
+my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2");
+is( $result, qq(1001|test 1001
+1002|test 1002
+1980|not filtered), 'check initial data copy from table tab_rowfilter_1');
+
+# Check expected replicated rows for tab_rowfilter_2
+# tap_pub_1 filter is: (c % 2 = 0)
+# tap_pub_2 filter is: (c % 3 = 0)
+# When there are multiple publications for the same table, 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');
+
+# The following commands are executed after CREATE SUBSCRIPTION, so these SQL
+# commands are for testing normal logical replication behavior.
+#
+# test row filter (INSERT, UPDATE, DELETE)
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (800, 'test 800')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1600, 'test 1600')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1601, 'test 1601')");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_1 (a, b) VALUES (1700, 'test 1700')");
+$node_publisher->safe_psql('postgres',
+ "UPDATE tab_rowfilter_1 SET b = NULL WHERE a = 1600");
+$node_publisher->safe_psql('postgres',
+ "UPDATE tab_rowfilter_1 SET b = 'test 1601 updated' WHERE a = 1601");
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM tab_rowfilter_1 WHERE a = 1700");
+$node_publisher->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'
+# - INSERT (1601, 'test 1601') YES, because 1601 > 1000 and 'test 1601' <> 'filtered'
+# - INSERT (1700, 'test 1700') YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
+# - UPDATE (1600, NULL) NO, row filter evaluates to false because NULL is not <> 'filtered'
+# - UPDATE (1601, 'test 1601 updated') YES, because 1601 > 1000 and 'test 1601 updated' <> 'filtered'
+# - DELETE (1700) NO, row filter contains column b that is not part of
+# the PK or REPLICA IDENTITY and old tuple contains b = NULL, hence, row filter
+# evaluates to false
+#
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2");
+is($result, qq(1001|test 1001
+1002|test 1002
+1600|test 1600
+1601|test 1601 updated
+1700|test 1700
+1980|not filtered), 'check replicated rows to table tab_rowfilter_1');
+
+# Publish using root partitioned table
+# Use a different partitioned table layout (exercise publish_via_partition_root)
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_3 SET (publish_via_partition_root = true)");
+$node_subscriber->safe_psql('postgres',
+ "TRUNCATE TABLE tab_rowfilter_partitioned");
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION WITH (copy_data = true)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_partitioned (a, b) VALUES(4000, 400),(4001, 401),(4002, 402)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_less_10k (a, b) VALUES(4500, 450)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_less_10k (a, b) VALUES(5600, 123)");
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_greater_10k (a, b) VALUES(14000, 1950)");
+$node_publisher->safe_psql('postgres',
+ "UPDATE tab_rowfilter_less_10k SET b = 30 WHERE a = 4001");
+$node_publisher->safe_psql('postgres',
+ "DELETE FROM tab_rowfilter_less_10k WHERE a = 4002");
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check expected replicated rows for partitions
+# publication option publish_via_partition_root is true so use the row filter
+# from the root partitioned table
+# tab_rowfilter_partitioned filter: (a < 5000)
+# tab_rowfilter_less_10k filter: (a < 6000)
+# tab_rowfilter_greater_10k filter: no filter
+#
+# After TRUNCATE, REFRESH PUBLICATION, the initial data copy will apply the
+# partitioned table row filter.
+# - INSERT (1, 100) YES, 1 < 5000
+# - INSERT (7000, 101) NO, 7000 is not < 5000
+# - INSERT (15000, 102) NO, 15000 is not < 5000
+# - INSERT (5500, 300) NO, 5500 is not < 5000
+# - INSERT (2, 200) YES, 2 < 5000
+# - INSERT (6005, 201) NO, 6005 is not < 5000
+# - INSERT (16000, 103) NO, 16000 is not < 5000
+#
+# Execute SQL commands after initial data copy for testing the logical
+# replication behavior.
+# - INSERT (4000, 400) YES, 4000 < 5000
+# - INSERT (4001, 401) YES, 4001 < 5000
+# - INSERT (4002, 402) YES, 4002 < 5000
+# - INSERT (4500, 450) YES, 4500 < 5000
+# - INSERT (5600, 123) NO, 5600 is not < 5000
+# - INSERT (14000, 1950) NO, 16000 is not < 5000
+# - UPDATE (4001) YES, 4001 < 5000
+# - DELETE (4002) YES, 4002 < 5000
+$result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT a, b FROM tab_rowfilter_partitioned ORDER BY 1, 2");
+is( $result, qq(1|100
+2|200
+4000|400
+4001|30
+4500|450), 'check publish_via_partition_root behavior');
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
--
2.20.1
[text/x-patch] v50-0002-fixes-0001.patch (10.9K, ../[email protected]/4-v50-0002-fixes-0001.patch)
download | inline diff:
From a71668b85eac7b59b33a0fdd57e242f278b61107 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email protected]>
Date: Fri, 17 Dec 2021 16:06:51 -0300
Subject: [PATCH v50 2/8] fixes 0001
---
doc/src/sgml/ref/create_publication.sgml | 7 +++----
doc/src/sgml/ref/create_subscription.sgml | 23 +++++++++++----------
src/backend/commands/publicationcmds.c | 6 +++---
src/backend/replication/pgoutput/pgoutput.c | 10 ++++-----
src/test/regress/expected/publication.out | 10 ++++-----
src/test/regress/sql/publication.sql | 4 ++--
6 files changed, 30 insertions(+), 30 deletions(-)
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index 5aeee2309d..1c0d6111ca 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -239,10 +239,9 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
<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.
- If nullable columns are present in the <literal>WHERE</literal> clause,
- possible NULL values should be accounted for in expressions, to avoid
- unexpected results, because <literal>NULL</literal> values can cause
- those expressions to evaluate to false.
+ A nullable column in the <literal>WHERE</literal> clause could cause the
+ expression to evaluate to false; avoid using columns without not-null
+ constraints in the <literal>WHERE</literal> clause.
A <literal>WHERE</literal> clause allows simple expressions. The simple
expression cannot contain any aggregate or window functions, non-immutable
functions, user-defined types, operators or functions.
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index db255f323a..17c4606785 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -328,17 +328,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<para>
If any table in the publication has a <literal>WHERE</literal> clause, rows
that do not satisfy the <replaceable class="parameter">expression</replaceable>
- will not be published (i.e. they will be filtered out).
- If the subscription has several publications in which the same table has been
- published with different <literal>WHERE</literal> clauses, those expressions
- (for the same publish operation) get OR'ed together so that rows satisfying any
- of the expressions will be published. Also, if one of the publications for the
- same table has no <literal>WHERE</literal> clause at all, or is a <literal>FOR
- ALL TABLES</literal> or <literal>FOR ALL TABLES IN SCHEMA</literal> publication,
- then all other <literal>WHERE</literal> clauses (for the same publish operation)
- become redundant.
- If the subscriber is a <productname>PostgreSQL</productname> version before 15
- then any row filtering is ignored during the initial data synchronization phase.
+ 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) is satisfied. In this 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.
</para>
</refsect1>
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 9ca743c6d2..a43ad41c6b 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -983,11 +983,11 @@ OpenTableList(List *tables)
*/
if (list_member_oid(relids, myrelid))
{
- /* Disallow duplicate tables if there are any with row-filters. */
+ /* 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 row-filters for \"%s\"",
+ errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
RelationGetRelationName(rel))));
table_close(rel, ShareUpdateExclusiveLock);
@@ -1164,7 +1164,7 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
if (pubrel->whereClause)
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
- errmsg("invalid use of WHERE row-filter in ALTER PUBLICATION ... DROP TABLE")));
+ errmsg("cannot use a WHERE clause when removing a table from publication")));
ObjectAddressSet(obj, PublicationRelRelationId, prid);
performDeletion(&obj, DROP_CASCADE, 0);
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 2fa08e7278..be5ec1d28f 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -698,7 +698,7 @@ pgoutput_row_filter_init_expr(Node *rfnode)
if (expr == NULL)
ereport(ERROR,
(errcode(ERRCODE_CANNOT_COERCE),
- errmsg("row filter returns type %s that cannot be cast to the expected type %s",
+ errmsg("row filter returns type %s that cannot be coerced to the expected type %s",
format_type_be(exprtype),
format_type_be(BOOLOID)),
errhint("You will need to rewrite the row filter.")));
@@ -814,7 +814,7 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
bool rfisnull;
/*
- * Lookup if there is a row-filter, and if yes remember it in a list (per
+ * Lookup if there is a row filter, and if yes remember it in a list (per
* pubaction). If no, then remember there was no filter for this pubaction.
* Code following this 'publications' loop will combine all filters.
*/
@@ -848,7 +848,7 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
}
else
{
- /* Remember which pubactions have no row-filter. */
+ /* Remember which pubactions have no row filter. */
if (pub->pubactions.pubinsert)
no_filter[idx_ins] = true;
if (pub->pubactions.pubupdate)
@@ -856,7 +856,7 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
if (pub->pubactions.pubdelete)
no_filter[idx_del] = true;
- /* Quick exit loop if all pubactions have no row-filter. */
+ /* Quick exit loop if all pubactions have no row filter. */
if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
{
ReleaseSysCache(rftuple);
@@ -947,7 +947,7 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
/*
- * NOTE: Multiple publication row-filters have already been combined to a
+ * NOTE: Multiple publication row filters have already been combined to a
* single exprstate (for this pubaction).
*/
if (entry->exprstate[changetype])
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 5a49003ae4..02491c4b6b 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -333,7 +333,7 @@ Tables:
"testpub_rf_myschema.testpub_rf_tbl5" WHERE (h < 999)
DROP PUBLICATION testpub_syntax2;
--- fail - schemas are not allowed WHERE row-filter
+-- fail - schemas don't allow WHERE clause
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a = 123);
ERROR: syntax error at or near "WHERE"
@@ -344,12 +344,12 @@ ERROR: WHERE clause for schema not allowed
LINE 1: ...ax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf...
^
RESET client_min_messages;
--- fail - duplicate tables are not allowed if that table has any WHERE row-filters
+-- 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 row-filters for "testpub_rf_tbl1"
+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 row-filters for "testpub_rf_tbl1"
+ERROR: conflicting or redundant WHERE clauses for table "testpub_rf_tbl1"
RESET client_min_messages;
-- fail - aggregate functions not allowed in WHERE clause
ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
@@ -370,7 +370,7 @@ LINE 1: ...ICATION testpub6 FOR TABLE testpub_rf_tbl3 WHERE (e =#> 27);
^
-- fail - WHERE not allowed in DROP
ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl3 WHERE (e < 27);
-ERROR: invalid use of WHERE row-filter in ALTER PUBLICATION ... DROP TABLE
+ERROR: cannot use a WHERE clause when removing a table from publication
-- fail - cannot ALTER SET table which is a member of a pre-existing schema
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_myschema1;
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index 47bdba86ac..9185d5a1d2 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -171,12 +171,12 @@ CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_myschem
RESET client_min_messages;
\dRp+ testpub_syntax2
DROP PUBLICATION testpub_syntax2;
--- fail - schemas are not allowed WHERE row-filter
+-- fail - schemas don't allow WHERE clause
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a = 123);
CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf_myschema WHERE (a = 123);
RESET client_min_messages;
--- fail - duplicate tables are not allowed if that table has any WHERE row-filters
+-- 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');
--
2.20.1
[text/x-patch] v50-0003-Row-filter-validation.patch (62.7K, ../[email protected]/5-v50-0003-Row-filter-validation.patch)
download | inline diff:
From d814336f61c3463d1f3cbc738a981cf159a312f1 Mon Sep 17 00:00:00 2001
From: Peter Smith <[email protected]>
Date: Fri, 17 Dec 2021 20:37:32 +1100
Subject: [PATCH v50 3/8] Row-filter validation
This patch implements parse-tree "walkers" to validate a row-filter.
Expression Node-kind validation
-------------------------------
Only simple filter expressions are permitted. Specifically:
- no user-defined operators.
- no user-defined functions.
- no user-defined types.
- no system columns.
- no system functions (unless they are immutable). See design decision at [1].
[1] https://www.postgresql.org/message-id/CAA4eK1%2BXoD49bz5-2TtiD0ugq4PHSRX2D1sLPR_X4LNtdMc4OQ%40mail.gmail.com
Permits only simple nodes including:
List, Const, BoolExpr, NullIfExpr, NullTest, BooleanTest, CoalesceExpr,
CaseExpr, CaseTestExpr, MinMaxExpr, ArrayExpr, ScalarArrayOpExpr, XmlExpr.
Author: Peter Smith, Euler Taveira
REPLICA IDENTITY validation
---------------------------
For publish mode "delete" "update", validate that any columns referenced
in the filter expression must be part of REPLICA IDENTITY or Primary Key.
Row filter columns invalidation is done in CheckCmdReplicaIdentity, so that
the invalidation is executed only when actual UPDATE or DELETE executed on
the published relation. This is consistent with the existing check about
replica identity and can detect the change related to the row filter in time.
Cache the results of the validation for row filter columns in relcache to
reduce the cost of the validation. It is safe to do this because every
operation that change the row filter and replica identity will invalidate the
relcache.
Author: Hou zj
---
src/backend/catalog/pg_publication.c | 177 +++++++++++++-
src/backend/executor/execReplication.c | 36 ++-
src/backend/parser/parse_agg.c | 10 -
src/backend/parser/parse_expr.c | 21 +-
src/backend/parser/parse_func.c | 3 -
src/backend/parser/parse_oper.c | 7 -
src/backend/parser/parse_relation.c | 9 -
src/backend/replication/pgoutput/pgoutput.c | 27 +--
src/backend/utils/cache/relcache.c | 242 +++++++++++++++++---
src/include/catalog/pg_publication.h | 2 +-
src/include/parser/parse_node.h | 1 -
src/include/utils/rel.h | 7 +
src/include/utils/relcache.h | 1 +
src/test/regress/expected/publication.out | 227 ++++++++++++++----
src/test/regress/sql/publication.sql | 174 ++++++++++++--
src/test/subscription/t/027_row_filter.pl | 7 +-
src/tools/pgindent/typedefs.list | 1 +
17 files changed, 783 insertions(+), 169 deletions(-)
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 0929aa0a35..e3c154e31f 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -33,9 +33,11 @@
#include "catalog/pg_publication_namespace.h"
#include "catalog/pg_publication_rel.h"
#include "catalog/pg_type.h"
+#include "catalog/pg_proc.h"
#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"
@@ -111,6 +113,137 @@ check_publication_add_schema(Oid schemaid)
errdetail("Temporary schemas cannot be replicated.")));
}
+/*
+ * 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)" or
+ * - "(Var Op Var)" or
+ * - "(Var Op Const) Bool (Var Op Const)"
+ * - etc
+ * (where Var is a column of the table this filter belongs to)
+ *
+ * Specifically,
+ * - User-defined operators are not allowed.
+ * - User-defined functions are not allowed.
+ * - User-defined types are not allowed.
+ * - Non-immutable builtin functions are not allowed.
+ * - System columns are not allowed.
+ *
+ * Notes:
+ *
+ * We don't allow user-defined functions/operators/types because (a) if the user
+ * drops such a user-definition or if there is any other error via its function,
+ * the walsender won't be able to recover from such an error even if we fix the
+ * function's problem because a historic snapshot is used to access the
+ * row-filter; (b) any other table could be accessed via a function, which won't
+ * work because of historic snapshots in logical decoding environment.
+ *
+ * We don't allow anything other than immutable built-in functions because
+ * non-immutable functions can access the database and would lead to the problem
+ * (b) mentioned in the previous paragraph.
+ */
+static bool
+rowfilter_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 not allowed. */
+ if (var->vartype >= FirstNormalObjectId)
+ errdetail_msg = _("User-defined types are not allowed");
+
+ /* System columns 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(DEBUG1, "the row filter contained something unexpected: %s", nodeToString(node));
+
+ ereport(ERROR,
+ (errmsg("invalid publication WHERE expression for relation \"%s\"",
+ RelationGetRelationName(relation)),
+ errdetail("Expressions only allow columns, constants and some built-in functions and operators.")
+ ));
+ }
+
+ if (errdetail_msg)
+ ereport(ERROR,
+ (errmsg("invalid publication WHERE expression for relation \"%s\"",
+ RelationGetRelationName(relation)),
+ errdetail("%s", errdetail_msg)
+ ));
+
+ return expression_tree_walker(node, rowfilter_walker, (void *)relation);
+}
+
/*
* Returns if relation represented by oid and Form_pg_class entry
* is publishable.
@@ -241,10 +374,6 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
PG_RETURN_BOOL(result);
}
-/*
- * Gets the relations based on the publication partition option for a specified
- * relation.
- */
List *
GetPubPartitionOptionRelations(List *result, PublicationPartOpt pub_partopt,
Oid relid)
@@ -298,7 +427,7 @@ GetTransformedWhereClause(ParseState *pstate, PublicationRelInfo *pri,
addNSItemToQuery(pstate, nsitem, false, true, true);
whereclause = transformWhereClause(pstate, copyObject(pri->whereClause),
- EXPR_KIND_PUBLICATION_WHERE,
+ EXPR_KIND_WHERE,
"PUBLICATION WHERE");
/* Fix up collation information */
@@ -308,6 +437,37 @@ GetTransformedWhereClause(ParseState *pstate, PublicationRelInfo *pri,
return whereclause;
}
+/*
+ * Check if any of the ancestors are published in the publication. If so,
+ * return the relid of the topmost ancestor that is published via this
+ * publication, otherwise 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.
*/
@@ -362,6 +522,13 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
* collation information.
*/
whereclause = GetTransformedWhereClause(pstate, pri, true);
+
+ /*
+ * Walk the parse-tree of this publication row filter expression and
+ * throw an error if anything not permitted or unexpected is
+ * encountered.
+ */
+ rowfilter_walker(whereclause, targetrel);
}
/* Form a tuple. */
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 574d7d27fd..42c5dbe2b9 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -568,14 +568,46 @@ void
CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
{
PublicationActions *pubactions;
+ AttrNumber bad_rfcolnum;
/* 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;
+
+ bad_rfcolnum = GetRelationPublicationInfo(rel, true);
+
+ /*
+ * 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,
+ * which means all referenced columns are part of REPLICA IDENTITY, or the
+ * table do not publish UPDATES or DELETES.
+ */
+ if (AttributeNumberIsValid(bad_rfcolnum))
+ {
+ const char *colname = get_attname(RelationGetRelid(rel),
+ bad_rfcolnum, 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/parser/parse_agg.c b/src/backend/parser/parse_agg.c
index 193c87d8b7..7d829a05a9 100644
--- a/src/backend/parser/parse_agg.c
+++ b/src/backend/parser/parse_agg.c
@@ -551,13 +551,6 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr)
err = _("grouping operations are not allowed in COPY FROM WHERE conditions");
break;
- case EXPR_KIND_PUBLICATION_WHERE:
- if (isAgg)
- err = _("aggregate functions are not allowed in publication WHERE expressions");
- else
- err = _("grouping operations are not allowed in publication WHERE expressions");
-
- break;
case EXPR_KIND_CYCLE_MARK:
errkind = true;
@@ -950,9 +943,6 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc,
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
- case EXPR_KIND_PUBLICATION_WHERE:
- err = _("window functions are not allowed in publication WHERE expressions");
- break;
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c
index 3d43839b35..2d1a477154 100644
--- a/src/backend/parser/parse_expr.c
+++ b/src/backend/parser/parse_expr.c
@@ -200,19 +200,8 @@ transformExprRecurse(ParseState *pstate, Node *expr)
break;
case T_FuncCall:
- {
- /*
- * Forbid functions in publication WHERE condition
- */
- if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("functions are not allowed in publication WHERE expressions"),
- parser_errposition(pstate, exprLocation(expr))));
-
- result = transformFuncCall(pstate, (FuncCall *) expr);
- break;
- }
+ result = transformFuncCall(pstate, (FuncCall *) expr);
+ break;
case T_MultiAssignRef:
result = transformMultiAssignRef(pstate, (MultiAssignRef *) expr);
@@ -515,7 +504,6 @@ transformColumnRef(ParseState *pstate, ColumnRef *cref)
case EXPR_KIND_COPY_WHERE:
case EXPR_KIND_GENERATED_COLUMN:
case EXPR_KIND_CYCLE_MARK:
- case EXPR_KIND_PUBLICATION_WHERE:
/* okay */
break;
@@ -1776,9 +1764,6 @@ transformSubLink(ParseState *pstate, SubLink *sublink)
case EXPR_KIND_GENERATED_COLUMN:
err = _("cannot use subquery in column generation expression");
break;
- case EXPR_KIND_PUBLICATION_WHERE:
- err = _("cannot use subquery in publication WHERE expression");
- break;
/*
* There is intentionally no default: case here, so that the
@@ -3099,8 +3084,6 @@ ParseExprKindName(ParseExprKind exprKind)
return "GENERATED AS";
case EXPR_KIND_CYCLE_MARK:
return "CYCLE";
- case EXPR_KIND_PUBLICATION_WHERE:
- return "publication expression";
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c
index 29bebb73eb..542f9167aa 100644
--- a/src/backend/parser/parse_func.c
+++ b/src/backend/parser/parse_func.c
@@ -2655,9 +2655,6 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location)
case EXPR_KIND_CYCLE_MARK:
errkind = true;
break;
- case EXPR_KIND_PUBLICATION_WHERE:
- err = _("set-returning functions are not allowed in publication WHERE expressions");
- break;
/*
* There is intentionally no default: case here, so that the
diff --git a/src/backend/parser/parse_oper.c b/src/backend/parser/parse_oper.c
index 29f8835ce1..bc34a23afc 100644
--- a/src/backend/parser/parse_oper.c
+++ b/src/backend/parser/parse_oper.c
@@ -718,13 +718,6 @@ make_op(ParseState *pstate, List *opname, Node *ltree, Node *rtree,
opform->oprright)),
parser_errposition(pstate, location)));
- /* Check it's not a custom operator for publication WHERE expressions */
- if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE && opform->oid >= FirstNormalObjectId)
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("user-defined operators are not allowed in publication WHERE expressions"),
- parser_errposition(pstate, location)));
-
/* Do typecasting and build the expression tree */
if (ltree == NULL)
{
diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index 036d9c6d26..c5c3f26ecf 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -3538,20 +3538,11 @@ errorMissingRTE(ParseState *pstate, RangeVar *relation)
rte->eref->aliasname)),
parser_errposition(pstate, relation->location)));
else
- {
- if (pstate->p_expr_kind == EXPR_KIND_PUBLICATION_WHERE)
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_TABLE),
- errmsg("publication WHERE expression invalid reference to table \"%s\"",
- relation->relname),
- parser_errposition(pstate, relation->location)));
-
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_TABLE),
errmsg("missing FROM-clause entry for table \"%s\"",
relation->relname),
parser_errposition(pstate, relation->location)));
- }
}
/*
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index be5ec1d28f..7192948399 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1572,26 +1572,17 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
*/
if (am_partition)
{
- List *ancestors = get_partition_ancestors(relid);
- ListCell *lc2;
+ Oid ancestor;
+ List *ancestors = get_partition_ancestors(relid);
- /*
- * Find the "topmost" ancestor that is in this
- * publication.
- */
- foreach(lc2, ancestors)
+ ancestor = GetTopMostAncestorInPublication(pub->oid,
+ ancestors);
+
+ if (ancestor != InvalidOid)
{
- Oid ancestor = lfirst_oid(lc2);
-
- 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;
- }
+ ancestor_published = true;
+ if (pub->pubviaroot)
+ publish_as_relid = ancestor;
}
}
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 105d8d4601..14184e1337 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -56,6 +56,7 @@
#include "catalog/pg_opclass.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
#include "catalog/pg_rewrite.h"
#include "catalog/pg_shseclabel.h"
#include "catalog/pg_statistic_ext.h"
@@ -71,6 +72,8 @@
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "optimizer/optimizer.h"
+#include "parser/parse_clause.h"
+#include "parser/parse_relation.h"
#include "rewrite/rewriteDefine.h"
#include "rewrite/rowsecurity.h"
#include "storage/lmgr.h"
@@ -84,6 +87,7 @@
#include "utils/memutils.h"
#include "utils/relmapper.h"
#include "utils/resowner_private.h"
+#include "utils/ruleutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
@@ -267,6 +271,19 @@ typedef struct opclasscacheent
static HTAB *OpClassCache = NULL;
+/*
+ * Information used to validate the columns in the row filter expression. see
+ * rowfilter_column_walker for details.
+ */
+typedef struct rf_context
+{
+ AttrNumber invalid_rfcolnum; /* invalid column number */
+ Bitmapset *bms_replident; /* bitset of replica identity col indexes */
+ 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;
/* non-export function prototypes */
@@ -5521,39 +5538,91 @@ RelationGetExclusionInfo(Relation indexRelation,
MemoryContextSwitchTo(oldcxt);
}
+
+
/*
- * Get publication actions for the given relation.
+ * Check if any columns used in the row-filter WHERE clause are not part of
+ * REPLICA IDENTITY and save the invalid column number in
+ * rf_context::invalid_rfcolnum.
*/
-struct PublicationActions *
-GetRelationPublicationActions(Relation relation)
+static bool
+rowfilter_column_walker(Node *node, rf_context *context)
{
- List *puboids;
- ListCell *lc;
- MemoryContext oldcxt;
- Oid schemaid;
- PublicationActions *pubactions = palloc0(sizeof(PublicationActions));
+ 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 we can only use the bitset of replica identity col
+ * indexes in the child table to check. So, we need to convert the
+ * column number in the row filter expression to the child table's in
+ * case the column order of the parent table is different from the
+ * child table's.
+ */
+ 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, rowfilter_column_walker,
+ (void *) context);
+}
+
+/*
+ * Get the publication information for the given relation.
+ *
+ * Traverse all the publications which the relation is in to get the
+ * publication actions. If the publication actions include UPDATE or DELETE and
+ * validate_rowfilter is true, then validate that if all columns referenced in
+ * the row filter expression are part of REPLICA IDENTITY.
+ *
+ * If not all the row filter columns are part of REPLICA IDENTITY, return the
+ * invalid column number, otherwise InvalidAttrNumber.
+ */
+AttrNumber
+GetRelationPublicationInfo(Relation relation, bool validate_rowfilter)
+{
+ List *puboids;
+ ListCell *lc;
+ MemoryContext oldcxt;
+ Oid schemaid;
+ List *ancestors = NIL;
+ Oid relid = RelationGetRelid(relation);
+ rf_context context = { 0 };
+ PublicationActions pubactions = { 0 };
+ bool rfcol_valid = true;
+ AttrNumber invalid_rfcolnum = InvalidAttrNumber;
/*
* If not publishable, it publishes no actions. (pgoutput_change() will
* ignore it.)
*/
- if (!is_publishable_relation(relation))
- return pubactions;
-
- if (relation->rd_pubactions)
- return memcpy(pubactions, relation->rd_pubactions,
- sizeof(PublicationActions));
+ if (!is_publishable_relation(relation) || relation->rd_rfcol_valid)
+ 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)
{
@@ -5568,6 +5637,20 @@ GetRelationPublicationActions(Relation relation)
}
puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+ /*
+ * Find what are the cols that are part of the REPLICA IDENTITY.
+ * Note that REPLICA IDENTITY DEFAULT means primary key or nothing.
+ */
+ if (validate_rowfilter)
+ {
+ if (relation->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT)
+ context.bms_replident = RelationGetIndexAttrBitmap(relation,
+ INDEX_ATTR_BITMAP_PRIMARY_KEY);
+ else if (relation->rd_rel->relreplident == REPLICA_IDENTITY_INDEX)
+ context.bms_replident = RelationGetIndexAttrBitmap(relation,
+ INDEX_ATTR_BITMAP_IDENTITY_KEY);
+ }
+
foreach(lc, puboids)
{
Oid pubid = lfirst_oid(lc);
@@ -5581,34 +5664,140 @@ 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.
+ * If the publication action include UPDATE and DELETE and
+ * validate_rowfilter flag is true, validates that any columns
+ * referenced in the filter expression are part of REPLICA IDENTITY
+ * index.
+ *
+ * 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 we already found the column in row filter which is not part of
+ * REPLICA IDENTITY index, skip the validation too.
*/
- if (pubactions->pubinsert && pubactions->pubupdate &&
- pubactions->pubdelete && pubactions->pubtruncate)
+ if (validate_rowfilter &&
+ (pubform->pubupdate || pubform->pubdelete) &&
+ relation->rd_rel->relreplident != REPLICA_IDENTITY_FULL &&
+ rfcol_valid)
+ {
+ HeapTuple rftuple;
+ Oid publish_as_relid = InvalidOid;
+
+ /*
+ * For a partition, if pubviaroot is true, check if any of the
+ * ancestors are published. If so, note down the topmost ancestor
+ * that is published via this publication, the row filter
+ * expression on which will be used to filter the partition's
+ * changes. We could have got the topmost ancestor when collecting
+ * the publication oids, but that will make the code more
+ * complicated.
+ */
+ if (pubform->pubviaroot && relation->rd_rel->relispartition)
+ {
+ if (pubform->puballtables)
+ publish_as_relid = llast_oid(ancestors);
+ else
+ publish_as_relid = GetTopMostAncestorInPublication(pubform->oid,
+ ancestors);
+ }
+
+ if (publish_as_relid == InvalidOid)
+ publish_as_relid = relid;
+
+ rftuple = SearchSysCache2(PUBLICATIONRELMAP,
+ ObjectIdGetDatum(publish_as_relid),
+ ObjectIdGetDatum(pubid));
+
+ if (HeapTupleIsValid(rftuple))
+ {
+ Datum rfdatum;
+ bool rfisnull;
+ Node *rfnode;
+
+ context.pubviaroot = pubform->pubviaroot;
+ context.parentid = publish_as_relid;
+ context.relid = relid;
+
+ rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
+ Anum_pg_publication_rel_prqual,
+ &rfisnull);
+
+ if (!rfisnull)
+ {
+ rfnode = stringToNode(TextDatumGetCString(rfdatum));
+ rfcol_valid = !rowfilter_column_walker(rfnode, &context);
+ invalid_rfcolnum = context.invalid_rfcolnum;
+ pfree(rfnode);
+ }
+
+ ReleaseSysCache(rftuple);
+ }
+ }
+
+ /*
+ * If we know everything is replicated and some columns are not part of
+ * replica identity, there is no point to check for other publications.
+ */
+ if (pubactions.pubinsert && pubactions.pubupdate &&
+ pubactions.pubdelete && pubactions.pubtruncate &&
+ (!validate_rowfilter || !rfcol_valid))
break;
}
+ bms_free(context.bms_replident);
+
if (relation->rd_pubactions)
{
pfree(relation->rd_pubactions);
relation->rd_pubactions = NULL;
}
+ if (validate_rowfilter)
+ relation->rd_rfcol_valid = rfcol_valid;
+
/* 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 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)
+ {
+ memcpy(pubactions, relation->rd_pubactions,
+ sizeof(PublicationActions));
+ return pubactions;
+ }
+
+ (void) GetRelationPublicationInfo(relation, false);
+ memcpy(pubactions, relation->rd_pubactions,
+ sizeof(PublicationActions));
+
return pubactions;
}
@@ -6163,6 +6352,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/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 96c55f627d..9e197defc0 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -135,6 +135,6 @@ extern char *get_publication_name(Oid pubid, bool missing_ok);
extern Node *GetTransformedWhereClause(ParseState *pstate,
PublicationRelInfo *pri,
bool bfixupcollation);
-
+extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors);
#endif /* PG_PUBLICATION_H */
diff --git a/src/include/parser/parse_node.h b/src/include/parser/parse_node.h
index d58ae6a63f..ee179082ce 100644
--- a/src/include/parser/parse_node.h
+++ b/src/include/parser/parse_node.h
@@ -80,7 +80,6 @@ typedef enum ParseExprKind
EXPR_KIND_COPY_WHERE, /* WHERE condition in COPY FROM */
EXPR_KIND_GENERATED_COLUMN, /* generation expression for a column */
EXPR_KIND_CYCLE_MARK, /* cycle mark value */
- EXPR_KIND_PUBLICATION_WHERE /* WHERE condition for a table in PUBLICATION */
} ParseExprKind;
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 31281279cf..27cec813c0 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -163,6 +163,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 replica identity, or the publication
+ * actions do not include UPDATE and 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
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 82316bba54..9cc4a380f8 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 02491c4b6b..e6651bdaae 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -243,18 +243,21 @@ 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 SCHEMA testpub_rf_myschema;
-CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
-CREATE SCHEMA testpub_rf_myschema1;
-CREATE TABLE testpub_rf_myschema1.testpub_rf_tbl6(i integer);
+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';
-CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5);
+-- 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 | t | t | t | f
+ 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))
@@ -264,7 +267,7 @@ ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 200
Publication testpub5
Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
--------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ 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))
@@ -275,7 +278,7 @@ ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2;
Publication testpub5
Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
--------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ regress_publication_user | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl3" WHERE ((e > 1000) AND (e < 2000))
@@ -286,7 +289,7 @@ ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500)
Publication testpub5
Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
--------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ regress_publication_user | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl3" WHERE ((e > 300) AND (e < 500))
@@ -308,40 +311,40 @@ Publications:
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);
+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 | t | t | t | f
+ 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_myschema.testpub_rf_tbl5 WHERE (h < 999);
+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 | t | t | t | f
+ regress_publication_user | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
- "testpub_rf_myschema.testpub_rf_tbl5" WHERE (h < 999)
+ "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_myschema WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a = 123);
ERROR: syntax error at or near "WHERE"
-LINE 1: ...tax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema WHERE (a =...
+LINE 1: ...ntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1 WHERE (a =...
^
-CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf_myschema WHERE (a = 123);
+CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_schema1, testpub_rf_schema1 WHERE (a = 123);
ERROR: WHERE clause for schema not allowed
-LINE 1: ...ax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf...
+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
@@ -353,43 +356,185 @@ ERROR: conflicting or redundant WHERE clauses for table "testpub_rf_tbl1"
RESET client_min_messages;
-- fail - aggregate functions not allowed in WHERE clause
ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
-ERROR: functions are not allowed in publication WHERE expressions
+ERROR: aggregate functions are not allowed in WHERE
LINE 1: ...TION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
^
--- fail - functions disallowed
-ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
-ERROR: functions are not allowed in publication WHERE expressions
-LINE 1: ...ICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) ...
- ^
--- fail - user-defined operators disallowed
-CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
-CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+-- 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: user-defined operators are not allowed in publication WHERE expressions
-LINE 1: ...ICATION 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 - builtin 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 builtin 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 and some built-in functions and operators.
+-- 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_tbl3 WHERE (e < 27);
+ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl1 WHERE (e < 27);
ERROR: cannot use a WHERE clause when removing a table from publication
-- fail - cannot ALTER SET table which is a member of a pre-existing schema
SET client_min_messages = 'ERROR';
-CREATE PUBLICATION testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_myschema1;
-ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_myschema1, TABLE testpub_rf_myschema1.testpub_rf_tbl6 WHERE (i < 99);
-ERROR: cannot add relation "testpub_rf_myschema1.testpub_rf_tbl6" to publication
-DETAIL: Table's schema "testpub_rf_myschema1" is already part of the publication or part of the specified schema list.
+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_myschema.testpub_rf_tbl5;
-DROP TABLE testpub_rf_myschema1.testpub_rf_tbl6;
-DROP SCHEMA testpub_rf_myschema;
-DROP SCHEMA testpub_rf_myschema1;
+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 testpub7;
+DROP PUBLICATION testpub6;
DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(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 9185d5a1d2..a95c71b1cb 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -138,12 +138,15 @@ 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 SCHEMA testpub_rf_myschema;
-CREATE TABLE testpub_rf_myschema.testpub_rf_tbl5(h integer);
-CREATE SCHEMA testpub_rf_myschema1;
-CREATE TABLE testpub_rf_myschema1.testpub_rf_tbl6(i integer);
+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';
-CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5);
+-- 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);
@@ -162,19 +165,19 @@ RESET client_min_messages;
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);
+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_myschema.testpub_rf_tbl5 WHERE (h < 999);
+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_myschema WHERE (a = 123);
-CREATE PUBLICATION testpub_syntax3 FOR ALL TABLES IN SCHEMA testpub_rf_myschema, testpub_rf_myschema WHERE (a = 123);
+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';
@@ -183,32 +186,157 @@ CREATE PUBLICATION testpub_dups FOR TABLE testpub_rf_tbl1, testpub_rf_tbl1 WHERE
RESET client_min_messages;
-- fail - aggregate functions not allowed in WHERE clause
ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e < AVG(e));
--- fail - functions disallowed
-ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl4 WHERE (length(g) < 6);
--- fail - user-defined operators disallowed
-CREATE FUNCTION testpub_rf_func(integer, integer) RETURNS boolean AS $$ SELECT hashint4($1) > $2 $$ LANGUAGE SQL;
-CREATE OPERATOR =#> (PROCEDURE = testpub_rf_func, LEFTARG = integer, RIGHTARG = integer);
+-- 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 - builtin 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 builtin 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_tbl3 WHERE (e < 27);
+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 testpub7 FOR ALL TABLES IN SCHEMA testpub_rf_myschema1;
-ALTER PUBLICATION testpub7 SET ALL TABLES IN SCHEMA testpub_rf_myschema1, TABLE testpub_rf_myschema1.testpub_rf_tbl6 WHERE (i < 99);
+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_myschema.testpub_rf_tbl5;
-DROP TABLE testpub_rf_myschema1.testpub_rf_tbl6;
-DROP SCHEMA testpub_rf_myschema;
-DROP SCHEMA testpub_rf_myschema1;
+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 testpub7;
+DROP PUBLICATION testpub6;
DROP OPERATOR =#>(integer, integer);
-DROP FUNCTION testpub_rf_func(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';
diff --git a/src/test/subscription/t/027_row_filter.pl b/src/test/subscription/t/027_row_filter.pl
index 64e71d0adb..de6b73d26a 100644
--- a/src/test/subscription/t/027_row_filter.pl
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -18,6 +18,8 @@ $node_subscriber->start;
# setup structure on publisher
$node_publisher->safe_psql('postgres',
"CREATE TABLE tab_rowfilter_1 (a int primary key, b text)");
+$node_publisher->safe_psql('postgres',
+ "ALTER TABLE tab_rowfilter_1 REPLICA IDENTITY FULL;");
$node_publisher->safe_psql('postgres',
"CREATE TABLE tab_rowfilter_2 (c int primary key)");
$node_publisher->safe_psql('postgres',
@@ -280,9 +282,7 @@ is($result, qq(13|0|12), 'check replicated rows to tab_rowfilter_4');
# - INSERT (1700, 'test 1700') YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
# - UPDATE (1600, NULL) NO, row filter evaluates to false because NULL is not <> 'filtered'
# - UPDATE (1601, 'test 1601 updated') YES, because 1601 > 1000 and 'test 1601 updated' <> 'filtered'
-# - DELETE (1700) NO, row filter contains column b that is not part of
-# the PK or REPLICA IDENTITY and old tuple contains b = NULL, hence, row filter
-# evaluates to false
+# - DELETE (1700) YES, because 1700 > 1000 and 'test 1700' <> 'filtered'
#
$result =
$node_subscriber->safe_psql('postgres',
@@ -291,7 +291,6 @@ is($result, qq(1001|test 1001
1002|test 1002
1600|test 1600
1601|test 1601 updated
-1700|test 1700
1980|not filtered), 'check replicated rows to table tab_rowfilter_1');
# Publish using root partitioned table
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 0c61ccbdd0..89f3917352 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3503,6 +3503,7 @@ replace_rte_variables_context
ret_type
rewind_source
rewrite_event
+rf_context
rijndael_ctx
rm_detail_t
role_auth_extra
--
2.20.1
[text/x-patch] v50-0004-fixes-0002.patch (8.6K, ../[email protected]/6-v50-0004-fixes-0002.patch)
download | inline diff:
From 89c3319e2a7bb5b4ec91c70c3a334e05a900cdbf Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email protected]>
Date: Sat, 18 Dec 2021 19:08:24 -0300
Subject: [PATCH v50 4/8] fixes 0002
---
src/backend/catalog/pg_publication.c | 48 +++++++++++------------
src/include/parser/parse_node.h | 2 +-
src/test/regress/expected/publication.out | 10 ++---
src/test/regress/sql/publication.sql | 6 +--
4 files changed, 32 insertions(+), 34 deletions(-)
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index e3c154e31f..44ec69e410 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -29,11 +29,11 @@
#include "catalog/objectaddress.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"
#include "catalog/pg_type.h"
-#include "catalog/pg_proc.h"
#include "commands/publicationcmds.h"
#include "funcapi.h"
#include "miscadmin.h"
@@ -145,31 +145,29 @@ IsRowFilterSimpleExpr(Node *node)
* expression".
*
* It allows only simple or compound expressions such as:
- * - "(Var Op Const)" or
- * - "(Var Op Var)" or
- * - "(Var Op Const) Bool (Var Op Const)"
+ * - (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)
*
- * Specifically,
- * - User-defined operators are not allowed.
- * - User-defined functions are not allowed.
- * - User-defined types are not allowed.
- * - Non-immutable builtin functions are not allowed.
+ * 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:
+ * NOTES
*
- * We don't allow user-defined functions/operators/types because (a) if the user
- * drops such a user-definition or if there is any other error via its function,
- * the walsender won't be able to recover from such an error even if we fix the
- * function's problem because a historic snapshot is used to access the
- * row-filter; (b) any other table could be accessed via a function, which won't
- * work because of historic snapshots in logical decoding environment.
- *
- * We don't allow anything other than immutable built-in functions because
- * non-immutable functions can access the database and would lead to the problem
- * (b) mentioned in the previous paragraph.
+ * 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 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
+ * non-immutable built-in functions are allowed in row filter expressions.
*/
static bool
rowfilter_walker(Node *node, Relation relation)
@@ -188,11 +186,11 @@ rowfilter_walker(Node *node, Relation relation)
{
Var *var = (Var *) node;
- /* User-defined types not allowed. */
+ /* User-defined types are not allowed. */
if (var->vartype >= FirstNormalObjectId)
- errdetail_msg = _("User-defined types are not allowed");
+ errdetail_msg = _("User-defined types are not allowed.");
- /* System columns not allowed. */
+ /* System columns are not allowed. */
else if (var->varattno < InvalidAttrNumber)
{
Oid relid = RelationGetRelid(relation);
@@ -225,12 +223,12 @@ rowfilter_walker(Node *node, Relation relation)
}
else
{
- elog(DEBUG1, "the row filter contained something unexpected: %s", nodeToString(node));
+ 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 and some built-in functions and operators.")
+ errdetail("Expressions only allow columns, constants, built-in operators, built-in data types and non-immutable built-in functions.")
));
}
diff --git a/src/include/parser/parse_node.h b/src/include/parser/parse_node.h
index ee179082ce..1d4f3a6eab 100644
--- a/src/include/parser/parse_node.h
+++ b/src/include/parser/parse_node.h
@@ -79,7 +79,7 @@ typedef enum ParseExprKind
EXPR_KIND_CALL_ARGUMENT, /* procedure argument in CALL */
EXPR_KIND_COPY_WHERE, /* WHERE condition in COPY FROM */
EXPR_KIND_GENERATED_COLUMN, /* generation expression for a column */
- EXPR_KIND_CYCLE_MARK, /* cycle mark value */
+ EXPR_KIND_CYCLE_MARK /* cycle mark value */
} ParseExprKind;
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index e6651bdaae..4a27ae86e2 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -376,23 +376,23 @@ 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 - builtin operators are allowed
+-- 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 builtin functions are allowed
+-- 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
+DETAIL: User-defined types are not allowed.
DROP TABLE rf_bug;
DROP TYPE rf_bug_status;
--- fail - row-filter expression is not simple
+-- 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 and some built-in functions and operators.
+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"
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index a95c71b1cb..73fc103b8c 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -197,10 +197,10 @@ ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl1 WHERE (a >= testpub_rf_func
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 - builtin operators are allowed
+-- 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 builtin functions are allowed
+-- 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');
@@ -208,7 +208,7 @@ 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
+-- 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);
--
2.20.1
[text/x-patch] v50-0005-UPDATEs-might-require-transformation.patch (16.4K, ../[email protected]/7-v50-0005-UPDATEs-might-require-transformation.patch)
download | inline diff:
From 97cfd82888e74f261e7cf446444dd75c20196abd Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email protected]>
Date: Wed, 8 Dec 2021 17:57:14 -0300
Subject: [PATCH v50 5/8] UPDATEs might require transformation
UPDATE should evaluate 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. Fix this issue
requires a transformation.
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 then, from the 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 satisfies the row filter then, from the data
consistency perspective, that row should inserted on the subscriber. The
UPDATE should be transformed into a INSERT statement and be sent to the
subscriber. Subsequent UPDATE or DELETE statements have no effect.
However, this might surprise someone who expects the data set to satisfy
the row filter expression on the provider.
---
src/backend/replication/pgoutput/pgoutput.c | 279 ++++++++++++++++++--
1 file changed, 253 insertions(+), 26 deletions(-)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 7192948399..7e5b93a135 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -94,6 +94,14 @@ static void send_repl_origin(LogicalDecodingContext *ctx,
RepOriginId origin_id, XLogRecPtr origin_lsn,
bool send_origin);
+/* index for exprstate array in RelationSyncEntry */
+enum PublishAction
+{
+ PUBLISH_ACTION_INSERT,
+ PUBLISH_ACTION_UPDATE,
+ PUBLISH_ACTION_DELETE
+};
+
/*
* Entry in the map used to remember which relation schemas we sent.
*
@@ -176,9 +184,11 @@ static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
static EState *create_estate_for_relation(Relation rel);
static ExprState *pgoutput_row_filter_init_expr(Node *rfnode);
static bool pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext);
-static bool pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
- Relation relation, HeapTuple oldtuple,
- HeapTuple newtuple, RelationSyncEntry *entry);
+static void pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry);
+static bool pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple,
+ HeapTuple newtuple, RelationSyncEntry *entry, int action);
+static bool pgoutput_row_filter_for_update(PGOutputData *data, Relation relation, HeapTuple oldtuple,
+ HeapTuple newtuple, RelationSyncEntry *entry, enum PublishAction *action, HeapTuple transformedtuple);
/*
* Specify output plugin callbacks
@@ -742,26 +752,15 @@ pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
}
/*
- * Change is checked against the row filter, if any.
- *
- * If it returns true, the change is replicated, otherwise, it is not.
+ * Initialize the row filter, the first time.
*/
-static bool
-pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
- Relation relation, HeapTuple oldtuple, HeapTuple newtuple,
- RelationSyncEntry *entry)
+static void
+pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntry *entry)
{
- EState *estate;
- ExprContext *ecxt;
ListCell *lc;
- bool result = true;
- Oid relid = RelationGetRelid(relation);
List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */
bool no_filter[] = {false, false, false}; /* One per pubaction */
- Assert(changetype == REORDER_BUFFER_CHANGE_INSERT ||
- changetype == REORDER_BUFFER_CHANGE_UPDATE ||
- changetype == REORDER_BUFFER_CHANGE_DELETE);
/*
* 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.
@@ -926,9 +925,31 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
entry->exprstate_valid = true;
}
+}
- /* Bail out if there is no row filter */
- if (!entry->exprstate[changetype])
+/*
+ * Change is checked against the row filter, if any.
+ *
+ * If it returns true, the change is replicated, otherwise, it is not.
+ */
+static bool
+pgoutput_row_filter(PGOutputData *data, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry, int action)
+{
+ EState *estate;
+ ExprContext *ecxt;
+ bool result = false;
+ Oid relid = RelationGetRelid(relation);
+
+ Assert(changetype == REORDER_BUFFER_CHANGE_INSERT ||
+ changetype == REORDER_BUFFER_CHANGE_UPDATE ||
+ changetype == REORDER_BUFFER_CHANGE_DELETE);
+
+ /*
+ * Bail out if for a certain operation there is no row filter to process.
+ * This is a fast path optimization. Read the explanation above about
+ * rf_in_all_pubs.
+ */
+ if (entry->exprstate[action] == NULL)
return true;
if (message_level_is_interesting(DEBUG3))
@@ -944,16 +965,22 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
ecxt = GetPerTupleExprContext(estate);
ecxt->ecxt_scantuple = entry->scantuple;
+ /*
+ * The default behavior for UPDATEs is to use the new tuple for row
+ * filtering.
+ * If the UPDATE requires a transformation, the new tuple will be replaced
+ * by the transformed tuple before calling this routine.
+ */
ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, ecxt->ecxt_scantuple, false);
/*
* NOTE: Multiple publication row filters have already been combined to a
* single exprstate (for this pubaction).
*/
- if (entry->exprstate[changetype])
+ if (entry->exprstate[action])
{
/* Evaluates row filter */
- result = pgoutput_row_filter_exec_expr(entry->exprstate[changetype], ecxt);
+ result = pgoutput_row_filter_exec_expr(entry->exprstate[action], ecxt);
}
/* Cleanup allocated resources */
@@ -964,6 +991,182 @@ pgoutput_row_filter(enum ReorderBufferChangeType changetype, PGOutputData *data,
return result;
}
+/*
+ * 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. Fix this issue requires a transformation.
+ *
+ * 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 then, from the 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
+ * satisfies the row filter then, from the data consistency perspective, that
+ * row should inserted on the subscriber. The UPDATE should be transformed into
+ * a INSERT statement and be sent to the subscriber. Subsequent UPDATE or
+ * DELETE statements have no effect (it matches no row -- see
+ * apply_handle_update_internal()). However, this might surprise someone who
+ * expects the data set to satisfy the row filter expression on the provider.
+ */
+static bool
+pgoutput_row_filter_for_update(PGOutputData *data, Relation relation, HeapTuple oldtuple, HeapTuple newtuple, RelationSyncEntry *entry, enum PublishAction *action, HeapTuple transformedtuple)
+{
+ MemoryContext oldctx;
+ Oid relid = RelationGetRelid(relation);
+ bool oldtuple_match, newtuple_match;
+ Datum *oldvalues, *newvalues;
+ bool *oldnulls, *newnulls, *replaces;
+ TupleDesc tupdesc;
+ int nratts;
+ int i;
+
+ /*
+ * Bail out if for a certain operation there is no row filter to process.
+ * This is a fast path optimization. Read the explanation in
+ * pgoutput_row_filter_init().
+ */
+ if (entry->exprstate[*action] == NULL)
+ return true;
+
+ /* Update requires a new tuple. */
+ Assert(newtuple);
+
+ elog(DEBUG3, "table \"%s.%s\" has row filter",
+ get_namespace_name(get_rel_namespace(relid)),
+ get_rel_name(relid));
+
+ /*
+ * If there is no oldtuple available then none of the replica identity
+ * columns were modified. No transformation is required in this case.
+ */
+ if (!oldtuple)
+ return pgoutput_row_filter(data, relation, NULL, newtuple, entry, PUBLISH_ACTION_UPDATE);
+
+ /* Evaluates row filter in the old tuple. */
+ oldtuple_match = pgoutput_row_filter(data, relation, oldtuple, NULL, entry, PUBLISH_ACTION_UPDATE);
+
+ /*
+ * Create a tuple table slot for row filter. TupleDesc must live as
+ * long as the cache remains. Release the tuple table slot if it
+ * already exists.
+ */
+ if (entry->scantuple != NULL)
+ {
+ ExecDropSingleTupleTableSlot(entry->scantuple);
+ entry->scantuple = NULL;
+ }
+ tupdesc = RelationGetDescr(relation);
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ entry->scantuple = MakeSingleTupleTableSlot(tupdesc, &TTSOpsHeapTuple);
+ MemoryContextSwitchTo(oldctx);
+
+ /*
+ * Obtain a transformed tuple based on the new tuple. This step is
+ * necessary because the new tuple might not have unchanged TOAST values
+ * from the replica identity. Copy them from the old tuple where it is
+ * available.
+ */
+ nratts = tupdesc->natts;
+
+ oldvalues = (Datum *) palloc(nratts * sizeof(Datum));
+ oldnulls = (bool *) palloc(nratts * sizeof(bool));
+ newvalues = (Datum *) palloc(nratts * sizeof(Datum));
+ newnulls = (bool *) palloc(nratts * sizeof(bool));
+ replaces = (bool *) palloc(nratts * sizeof(bool));
+
+ /* Indicates the modified values. */
+ memset(replaces, false, nratts * sizeof(bool));
+
+ /* Break down the tuples into fields. */
+ heap_deform_tuple(oldtuple, tupdesc, oldvalues, oldnulls);
+ heap_deform_tuple(newtuple, tupdesc, newvalues, newnulls);
+
+ /* Modify the transformed tuple if it does not have the value yet. */
+ for (i = 0; i < nratts; i++)
+ {
+ Form_pg_attribute att;
+
+ /* column in new tuple is null, nothing to do */
+ if (newnulls[i])
+ continue;
+
+ att = TupleDescAttr(tupdesc, i);
+
+ /*
+ * 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(newvalues[i]) &&
+ !oldnulls[i] && !VARATT_IS_EXTERNAL_ONDISK(oldvalues[i]))
+ {
+ newvalues[i] = oldvalues[i];
+ replaces[i] = true;
+ }
+ }
+
+ /* Transformed tuple is the new tuple with some possible modifications. */
+ transformedtuple = heap_modify_tuple(newtuple, tupdesc, newvalues, newnulls, replaces);
+
+ /*
+ * Evaluates row filter in the transformed tuple. The new tuple isn't used
+ * because it might not contain all values used in the row filter
+ * expression.
+ */
+ newtuple_match = pgoutput_row_filter(data, relation, NULL, transformedtuple, entry, PUBLISH_ACTION_UPDATE);
+
+ /*
+ * Case 1: if both tuples matches the row filter, transformation isn't
+ * required. Bail out. Send the UPDATE.
+ */
+ if (oldtuple_match && newtuple_match)
+ return true;
+
+ /*
+ * Case 2: if both tuples don't match the row filter, bail out. Send
+ * nothing.
+ */
+ if (!oldtuple_match && !newtuple_match)
+ return false;
+
+ /*
+ * 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.
+ */
+ if (oldtuple_match && !newtuple_match)
+ {
+ *action = PUBLISH_ACTION_DELETE;
+ return true;
+ }
+
+ /*
+ * Case 4: 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 (!oldtuple_match && newtuple_match)
+ *action = PUBLISH_ACTION_INSERT;
+
+ return true;
+}
+
/*
* Sends the decoded DML over wire.
*
@@ -1015,6 +1218,9 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
/* Avoid leaking memory by using and resetting our own context */
old = MemoryContextSwitchTo(data->context);
+ /* Initialize the row_filter */
+ pgoutput_row_filter_init(data, relation, relentry);
+
/* Send the data */
switch (change->action)
{
@@ -1023,7 +1229,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
HeapTuple tuple = &change->data.tp.newtuple->tuple;
/* Check row filter. */
- if (!pgoutput_row_filter(change->action, data, relation, NULL, tuple, relentry))
+ if (!pgoutput_row_filter(data, relation, NULL, tuple, relentry, PUBLISH_ACTION_INSERT))
break;
/*
@@ -1054,9 +1260,11 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
HeapTuple oldtuple = change->data.tp.oldtuple ?
&change->data.tp.oldtuple->tuple : NULL;
HeapTuple newtuple = &change->data.tp.newtuple->tuple;
+ HeapTuple transformedtuple = NULL;
+ enum PublishAction transformedaction = PUBLISH_ACTION_UPDATE;
/* Check row filter. */
- if (!pgoutput_row_filter(change->action, data, relation, oldtuple, newtuple, relentry))
+ if (!pgoutput_row_filter_for_update(data, relation, oldtuple, newtuple, relentry, &transformedaction, transformedtuple))
break;
maybe_send_schema(ctx, change, relation, relentry);
@@ -1075,12 +1283,31 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
relentry->map);
newtuple = execute_attr_map_tuple(newtuple,
relentry->map);
+ /* INSERT uses transformed tuple. */
+ if (transformedaction == PUBLISH_ACTION_INSERT)
+ transformedtuple = execute_attr_map_tuple(transformedtuple, relentry->map);
}
}
OutputPluginPrepareWrite(ctx, true);
- logicalrep_write_update(ctx->out, xid, relation, oldtuple,
- newtuple, data->binary);
+ switch (transformedaction)
+ {
+ case PUBLISH_ACTION_INSERT:
+ logicalrep_write_insert(ctx->out, xid, relation, transformedtuple,
+ data->binary);
+ break;
+ case PUBLISH_ACTION_UPDATE:
+ logicalrep_write_update(ctx->out, xid, relation,
+ oldtuple, newtuple,
+ data->binary);
+ break;
+ case PUBLISH_ACTION_DELETE:
+ logicalrep_write_delete(ctx->out, xid, relation, oldtuple,
+ data->binary);
+ break;
+ default:
+ Assert(false);
+ }
OutputPluginWrite(ctx, true);
break;
}
@@ -1090,7 +1317,7 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
HeapTuple oldtuple = &change->data.tp.oldtuple->tuple;
/* Check row filter. */
- if (!pgoutput_row_filter(change->action, data, relation, oldtuple, NULL, relentry))
+ if (!pgoutput_row_filter(data, relation, oldtuple, NULL, relentry, PUBLISH_ACTION_DELETE))
break;
maybe_send_schema(ctx, change, relation, relentry);
--
2.20.1
[text/x-patch] v50-0006-Row-filter-tab-auto-complete-and-pgdump.patch (5.6K, ../[email protected]/8-v50-0006-Row-filter-tab-auto-complete-and-pgdump.patch)
download | inline diff:
From 282d26be7f3a873c51a0457f010b23188350b88d Mon Sep 17 00:00:00 2001
From: Peter Smith <[email protected]>
Date: Fri, 17 Dec 2021 20:40:15 +1100
Subject: [PATCH v50 6/8] Row-filter tab auto-complete and pgdump
tab-auto-complete
-----------------
e.g.
"CREATE PUBLICATION <name> FOR TABLE <name>" - complete with "WHERE (".
"ALTER PUBLICATION <name> ADD|SET TABLE <name>" - complete with "WHERE (".
Author: Peter Smith
pg_dump
-------
Author: Euler Taveira
---
src/bin/pg_dump/pg_dump.c | 24 ++++++++++++++++++++----
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/tab-complete.c | 24 ++++++++++++++++++++++--
3 files changed, 43 insertions(+), 6 deletions(-)
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index b52f3ccda2..ea17e6d118 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4034,6 +4034,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
int i_oid;
int i_prpubid;
int i_prrelid;
+ int i_prrelqual;
int i,
j,
ntups;
@@ -4044,9 +4045,16 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
query = createPQExpBuffer();
/* Collect all publication membership info. */
- appendPQExpBufferStr(query,
- "SELECT tableoid, oid, prpubid, prrelid "
- "FROM pg_catalog.pg_publication_rel");
+ if (fout->remoteVersion >= 150000)
+ appendPQExpBufferStr(query,
+ "SELECT tableoid, oid, prpubid, prrelid, "
+ "pg_catalog.pg_get_expr(prqual, prrelid) AS prrelqual "
+ "FROM pg_catalog.pg_publication_rel");
+ else
+ appendPQExpBufferStr(query,
+ "SELECT tableoid, oid, prpubid, prrelid, "
+ "NULL AS prrelqual "
+ "FROM pg_catalog.pg_publication_rel");
res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
ntups = PQntuples(res);
@@ -4055,6 +4063,7 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
i_oid = PQfnumber(res, "oid");
i_prpubid = PQfnumber(res, "prpubid");
i_prrelid = PQfnumber(res, "prrelid");
+ i_prrelqual = PQfnumber(res, "prrelqual");
/* this allocation may be more than we need */
pubrinfo = pg_malloc(ntups * sizeof(PublicationRelInfo));
@@ -4095,6 +4104,10 @@ getPublicationTables(Archive *fout, TableInfo tblinfo[], int numTables)
pubrinfo[j].dobj.name = tbinfo->dobj.name;
pubrinfo[j].publication = pubinfo;
pubrinfo[j].pubtable = tbinfo;
+ if (PQgetisnull(res, i, i_prrelqual))
+ pubrinfo[j].pubrelqual = NULL;
+ else
+ pubrinfo[j].pubrelqual = pg_strdup(PQgetvalue(res, i, i_prrelqual));
/* Decide whether we want to dump it */
selectDumpablePublicationObject(&(pubrinfo[j].dobj), fout);
@@ -4162,8 +4175,11 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD TABLE ONLY",
fmtId(pubinfo->dobj.name));
- appendPQExpBuffer(query, " %s;\n",
+ appendPQExpBuffer(query, " %s",
fmtQualifiedDumpable(tbinfo));
+ if (pubrinfo->pubrelqual)
+ appendPQExpBuffer(query, " WHERE (%s)", pubrinfo->pubrelqual);
+ appendPQExpBufferStr(query, ";\n");
/*
* There is no point in creating a drop query as the drop is done by table
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index f011ace8a8..0ebdce56da 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -630,6 +630,7 @@ typedef struct _PublicationRelInfo
DumpableObject dobj;
PublicationInfo *publication;
TableInfo *pubtable;
+ char *pubrelqual;
} PublicationRelInfo;
/*
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index b524dc87fc..1d476348a6 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1654,6 +1654,19 @@ psql_completion(const char *text, int start, int end)
/* ALTER PUBLICATION <name> SET */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET"))
COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE");
+ /* ALTER PUBLICATION <name> SET TABLE <name> */
+ /* ALTER PUBLICATION <name> ADD TABLE <name> */
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET|ADD", "TABLE", MatchAny) && !ends_with(prev_wd, ','))
+ COMPLETE_WITH("WHERE (");
+ /*
+ * "ALTER PUBLICATION <name> SET TABLE <name> WHERE (" - complete with
+ * table attributes
+ *
+ * "ALTER PUBLICATION <name> ADD TABLE <name> WHERE (" - complete with
+ * table attributes
+ */
+ else if (HeadMatches("ALTER", "PUBLICATION", MatchAny) && TailMatches("WHERE", "("))
+ COMPLETE_WITH_ATTR(prev3_wd, "");
else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES", "IN", "SCHEMA"))
COMPLETE_WITH_QUERY(Query_for_list_of_schemas
" AND nspname != 'pg_catalog' "
@@ -2777,12 +2790,19 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("TABLES", "TABLES IN SCHEMA");
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES"))
COMPLETE_WITH("IN SCHEMA", "WITH (");
- else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE", MatchAny))
- COMPLETE_WITH("WITH (");
+ else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE", MatchAny) && !ends_with(prev_wd, ','))
+ COMPLETE_WITH("WHERE (", "WITH (");
/* Complete "CREATE PUBLICATION <name> FOR TABLE" with "<table>, ..." */
else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "TABLE"))
COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables, NULL);
+ /*
+ * "CREATE PUBLICATION <name> FOR TABLE <name> WHERE (" - complete with
+ * table attributes
+ */
+ else if (HeadMatches("CREATE", "PUBLICATION", MatchAny) && TailMatches("WHERE", "("))
+ COMPLETE_WITH_ATTR(prev3_wd, "");
+
/*
* Complete "CREATE PUBLICATION <name> FOR ALL TABLES IN SCHEMA <schema>,
* ..."
--
2.20.1
[text/x-patch] v50-0007-Row-filter-handle-FOR-ALL-TABLES.patch (14.1K, ../[email protected]/9-v50-0007-Row-filter-handle-FOR-ALL-TABLES.patch)
download | inline diff:
From 6e36362f9979b365264c3ed935bf7e9e607068f0 Mon Sep 17 00:00:00 2001
From: Peter Smith <[email protected]>
Date: Fri, 17 Dec 2021 20:41:40 +1100
Subject: [PATCH v50 7/8] Row-filter handle FOR ALL TABLES
If one of the subscriber's publications was created using FOR ALL TABLES then
that implies NO row-filtering will be applied.
If one of the subscriber's publications was created using FOR ALL TABLES IN
SCHEMA and the table belong to that same schmea, then that also implies NO
row-filtering will be applied.
These rules overrides any other row-filters from other subscribed publications.
Note that the initial COPY does not take publication operations into account.
Author: Peter Smith
Reported By: Tang
Discussion: https://www.postgresql.org/message-id/OS0PR01MB6113D82113AA081ACF710D0CFB6E9%40OS0PR01MB6113.jpnprd01.prod.outlook.com
---
src/backend/replication/logical/tablesync.c | 48 +++++---
src/backend/replication/pgoutput/pgoutput.c | 63 ++++++++++-
src/test/subscription/t/027_row_filter.pl | 118 +++++++++++++++++++-
3 files changed, 202 insertions(+), 27 deletions(-)
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index c20c2219fc..469aadc1c4 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -802,21 +802,22 @@ fetch_remote_table_info(char *nspname, char *relname,
walrcv_clear_result(res);
/*
+ * If any publication has puballtables true then all row-filtering is
+ * ignored.
+ *
+ * If the relation is a member of a schema of a subscribed publication that
+ * said ALL TABLES IN SCHEMA then all row-filtering is ignored.
+ *
* Get relation qual. DISTINCT avoids the same expression of a table in
* multiple publications from being included multiple times in the final
* expression.
*/
if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
{
- resetStringInfo(&cmd);
- appendStringInfo(&cmd,
- "SELECT DISTINCT pg_get_expr(prqual, prrelid) "
- " FROM pg_publication p "
- " INNER JOIN pg_publication_rel pr "
- " ON (p.oid = pr.prpubid) "
- " WHERE pr.prrelid = %u "
- " AND p.pubname IN (", lrel->remoteid);
+ StringInfoData pub_names;
+ /* Build the pubname list. */
+ initStringInfo(&pub_names);
first = true;
foreach(lc, MySubscription->publications)
{
@@ -825,11 +826,28 @@ fetch_remote_table_info(char *nspname, char *relname,
if (first)
first = false;
else
- appendStringInfoString(&cmd, ", ");
+ appendStringInfoString(&pub_names, ", ");
- appendStringInfoString(&cmd, quote_literal_cstr(pubname));
+ appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
}
- appendStringInfoChar(&cmd, ')');
+
+ /* Check for row-filters */
+ resetStringInfo(&cmd);
+ appendStringInfo(&cmd,
+ "SELECT DISTINCT pg_get_expr(prqual, prrelid) "
+ " FROM pg_publication p "
+ " INNER JOIN pg_publication_rel pr ON (p.oid = pr.prpubid) "
+ " WHERE pr.prrelid = %u AND p.pubname IN ( %s ) "
+ " AND NOT (select bool_or(puballtables) "
+ " FROM pg_publication "
+ " WHERE pubname in ( %s )) "
+ " AND (SELECT count(1)=0 "
+ " FROM pg_publication_namespace pn, pg_class c "
+ " WHERE c.oid = %u AND c.relnamespace = pn.pnnspid)",
+ lrel->remoteid,
+ pub_names.data,
+ pub_names.data,
+ lrel->remoteid);
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
@@ -847,18 +865,14 @@ fetch_remote_table_info(char *nspname, char *relname,
slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
{
- Datum rf = slot_getattr(slot, 1, &isnull);
+ Datum rf = slot_getattr(slot, 1, &isnull);
if (!isnull)
*qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
ExecClearTuple(slot);
- /*
- * One entry without a row filter expression means clean up
- * previous expressions (if there are any) and return with no
- * expressions.
- */
+ /* Ignore filters and cleanup as necessary. */
if (isnull)
{
if (*qual)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 7e5b93a135..a926ffc566 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -804,13 +804,68 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
* relation. Since row filter usage depends on the DML operation,
* there are multiple lists (one for each operation) which row filters
* will be appended.
+ *
+ * NOTE: FOR ALL TABLES implies "use no filters" so it takes precedence
+ *
+ * NOTE: ALL TABLES IN SCHEMA also implies "use not filters" if the
+ * table is a member of the same schema.
*/
foreach(lc, data->publications)
{
- Publication *pub = lfirst(lc);
- HeapTuple rftuple;
- Datum rfdatum;
- bool rfisnull;
+ Publication *pub = lfirst(lc);
+ HeapTuple rftuple;
+ Datum rfdatum;
+ bool rfisnull;
+ List *schemarelids = NIL;
+
+ /*
+ * If the publication is FOR ALL TABLES then it is treated the same
+ * as if this table has no filters (even if for some other
+ * publication it does).
+ */
+ if (pub->alltables)
+ {
+ if (pub->pubactions.pubinsert)
+ no_filter[idx_ins] = true;
+ if (pub->pubactions.pubupdate)
+ no_filter[idx_upd] = true;
+ if (pub->pubactions.pubdelete)
+ no_filter[idx_del] = true;
+
+ /* Quick exit loop if all pubactions have no row-filter. */
+ if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
+ break;
+
+ continue;
+ }
+
+ /*
+ * 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 filters (even if for some other publication it does).
+ */
+ schemarelids = GetAllSchemaPublicationRelations(pub->oid,
+ pub->pubviaroot ?
+ PUBLICATION_PART_ROOT :
+ PUBLICATION_PART_LEAF);
+ if (list_member_oid(schemarelids, entry->relid))
+ {
+ if (pub->pubactions.pubinsert)
+ no_filter[idx_ins] = true;
+ if (pub->pubactions.pubupdate)
+ no_filter[idx_upd] = true;
+ if (pub->pubactions.pubdelete)
+ no_filter[idx_del] = true;
+
+ list_free(schemarelids);
+
+ /* Quick exit loop if all pubactions have no row-filter. */
+ if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
+ break;
+
+ continue;
+ }
+ list_free(schemarelids);
/*
* Lookup if there is a row filter, and if yes remember it in a list (per
diff --git a/src/test/subscription/t/027_row_filter.pl b/src/test/subscription/t/027_row_filter.pl
index de6b73d26a..abeaf760c5 100644
--- a/src/test/subscription/t/027_row_filter.pl
+++ b/src/test/subscription/t/027_row_filter.pl
@@ -3,7 +3,7 @@ use strict;
use warnings;
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
-use Test::More tests => 10;
+use Test::More tests => 14;
# create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
@@ -15,6 +15,116 @@ 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)");
@@ -127,8 +237,6 @@ $node_publisher->safe_psql('postgres',
$node_publisher->safe_psql('postgres',
"INSERT INTO tab_rowfilter_greater_10k (a, b) VALUES(16000, 103)");
-my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
-my $appname = 'tap_sub';
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_1, tap_pub_2, tap_pub_3, tap_pub_4a, tap_pub_4b"
);
@@ -136,8 +244,6 @@ $node_subscriber->safe_psql('postgres',
$node_publisher->wait_for_catchup($appname);
# wait for initial table synchronization to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
$node_subscriber->poll_query_until('postgres', $synced_query)
or die "Timed out while waiting for subscriber to synchronize data";
@@ -148,7 +254,7 @@ $node_subscriber->poll_query_until('postgres', $synced_query)
# - INSERT (1980, 'not filtered') YES
# - generate_series(990,1002) YES, only for 1001,1002 because a > 1000
#
-my $result =
+$result =
$node_subscriber->safe_psql('postgres',
"SELECT a, b FROM tab_rowfilter_1 ORDER BY 1, 2");
is( $result, qq(1001|test 1001
--
2.20.1
[text/x-patch] v50-0008-fixes-0005.patch (6.6K, ../[email protected]/10-v50-0008-fixes-0005.patch)
download | inline diff:
From 06578a09b57558a1fce0327c748ecd074415bd63 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email protected]>
Date: Mon, 20 Dec 2021 10:05:23 -0300
Subject: [PATCH v50 8/8] fixes 0005
---
src/backend/replication/logical/tablesync.c | 45 ++++++++++++---------
src/backend/replication/pgoutput/pgoutput.c | 24 ++++++-----
2 files changed, 39 insertions(+), 30 deletions(-)
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 469aadc1c4..51271a6a00 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -802,15 +802,20 @@ fetch_remote_table_info(char *nspname, char *relname,
walrcv_clear_result(res);
/*
- * If any publication has puballtables true then all row-filtering is
- * ignored.
+ * 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.
*
- * If the relation is a member of a schema of a subscribed publication that
- * said ALL TABLES IN SCHEMA then all row-filtering is ignored.
+ * For initial synchronization, row filtering can be ignored in 2 cases:
*
- * Get relation qual. DISTINCT avoids the same expression of a table in
- * multiple publications from being included multiple times in the final
- * expression.
+ * 1) one of the subscribed publications has puballtables set to true
+ * 2) one of the subscribed publications is declared as ALL TABLES IN
+ * SCHEMA that includes this relation
+ *
+ * These 2 cases don't allow row filter expressions, so a absence of
+ * relation row filter expressions is a sufficient condition to copy the
+ * entire table, although other publications contain a row filter for this
+ * relation.
*/
if (walrcv_server_version(LogRepWorkerWalRcvConn) >= 150000)
{
@@ -831,23 +836,23 @@ fetch_remote_table_info(char *nspname, char *relname,
appendStringInfoString(&pub_names, quote_literal_cstr(pubname));
}
- /* Check for row-filters */
+ /* Check for row filters. */
resetStringInfo(&cmd);
appendStringInfo(&cmd,
- "SELECT DISTINCT pg_get_expr(prqual, prrelid) "
- " FROM pg_publication p "
- " INNER JOIN pg_publication_rel pr ON (p.oid = pr.prpubid) "
+ "SELECT DISTINCT pg_catalog.pg_get_expr(pr.prqual, pr.prrelid) "
+ " FROM pg_catalog.pg_publication p "
+ " INNER JOIN pg_catalog.pg_publication_rel pr ON (p.oid = pr.prpubid) "
" WHERE pr.prrelid = %u AND p.pubname IN ( %s ) "
- " AND NOT (select bool_or(puballtables) "
- " FROM pg_publication "
- " WHERE pubname in ( %s )) "
- " AND (SELECT count(1)=0 "
- " FROM pg_publication_namespace pn, pg_class c "
- " WHERE c.oid = %u AND c.relnamespace = pn.pnnspid)",
+ " AND NOT (SELECT pg_catalog.bool_or(b.puballtables) "
+ " FROM pg_catalog.pg_publication b "
+ " WHERE b.pubname IN ( %s )) "
+ " AND NOT EXISTS( "
+ " SELECT 1 FROM pg_catalog.pg_publication_namespace pn "
+ " INNER JOIN pg_catalog.pg_class c ON (pn.pnnspid = c.relnamespace) "
+ " WHERE c.oid = pr.prrelid)",
lrel->remoteid,
pub_names.data,
- pub_names.data,
- lrel->remoteid);
+ pub_names.data);
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 1, qualRow);
@@ -865,7 +870,7 @@ fetch_remote_table_info(char *nspname, char *relname,
slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
{
- Datum rf = slot_getattr(slot, 1, &isnull);
+ Datum rf = slot_getattr(slot, 1, &isnull);
if (!isnull)
*qual = lappend(*qual, makeString(TextDatumGetCString(rf)));
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index a926ffc566..30a6adb2d2 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -805,10 +805,11 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
* there are multiple lists (one for each operation) which row filters
* will be appended.
*
- * NOTE: FOR ALL TABLES implies "use no filters" so it takes precedence
+ * NOTE: FOR ALL TABLES implies "don't use row filter expression" so it
+ * takes precedence.
*
- * NOTE: ALL TABLES IN SCHEMA also implies "use not filters" if the
- * table is a member of the same schema.
+ * NOTE: 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)
{
@@ -820,8 +821,8 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
/*
* If the publication is FOR ALL TABLES then it is treated the same
- * as if this table has no filters (even if for some other
- * publication it does).
+ * as if this table has no row filters (even if for other
+ * publications it does).
*/
if (pub->alltables)
{
@@ -832,17 +833,19 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
if (pub->pubactions.pubdelete)
no_filter[idx_del] = true;
- /* Quick exit loop if all pubactions have no row-filter. */
+ /* Quick exit loop if all pubactions have no row filter. */
if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
break;
+ /* No additional work for this publication. Next one. */
continue;
}
/*
- * 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 filters (even if for some other publication it does).
+ * 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).
*/
schemarelids = GetAllSchemaPublicationRelations(pub->oid,
pub->pubviaroot ?
@@ -859,10 +862,11 @@ pgoutput_row_filter_init(PGOutputData *data, Relation relation, RelationSyncEntr
list_free(schemarelids);
- /* Quick exit loop if all pubactions have no row-filter. */
+ /* Quick exit loop if all pubactions have no row filter. */
if (no_filter[idx_ins] && no_filter[idx_upd] && no_filter[idx_del])
break;
+ /* No additional work for this publication. Next one. */
continue;
}
list_free(schemarelids);
--
2.20.1
view thread (489+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: row filtering for logical replication
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox