Re: row filtering for logical replication

[Andres Freund]

Hi,

On 2018-11-23 13:15:08 -0300, Euler Taveira wrote:
> Besides the problem presented by Hironobu-san, I'm doing some cleanup
> and improving docs. I also forget to declare pg_publication_rel TOAST
> table.
> 
> Thanks for your review.

As far as I can tell, the patch has not been refreshed since. So I'm
marking this as returned with feedback for now. Please resubmit once
ready.

Greetings,

Andres Freund

Re: row filtering for logical replication

[David Fetter]

On Wed, Feb 28, 2018 at 08:03:02PM -0300, Euler Taveira wrote:
> Hi,
> 
> The attached patches add support for filtering rows in the publisher.
> The output plugin will do the work if a filter was defined in CREATE
> PUBLICATION command. An optional WHERE clause can be added after the
> table name in the CREATE PUBLICATION such as:
> 
> CREATE PUBLICATION foo FOR TABLE departments WHERE (id > 2000 AND id <= 3000);
> 
> Row that doesn't match the WHERE clause will not be sent to the subscribers.
> 
> Patches 0001 and 0002 are only refactors and can be applied
> independently. 0003 doesn't include row filtering on initial
> synchronization.
> 
> Comments?

Great feature!  I think a lot of people will like to have the option
of trading a little extra CPU on the pub side for a bunch of network
traffic and some work on the sub side.

I noticed that the WHERE clause applies to all tables in the
publication.  Is that actually the right thing?  I'm thinking of a
case where we have foo(id, ...) and bar(foo_id, ).  To slice that
correctly, we'd want to do the ids in the foo table and the foo_ids in
the bar table.  In the system as written, that would entail, at least
potentially, writing a lot of publications by hand.

Something like
WHERE (
(table_1,..., table_N) HAS (/* WHERE clause here */) AND
(table_N+1,..., table_M) HAS (/* WHERE clause here */) AND
...
)

could be one way to specify.

I also noticed that in psql, \dRp+ doesn't show the WHERE clause,
which it probably should.

Does it need regression tests?

Best,
David.
-- 
David Fetter  http://fetter.org/
Phone: +1 415 235 3778

Remember to vote!
Consider donating to Postgres: http://www.postgresql.org/about/donate

Re: row filtering for logical replication

[Craig Ringer]

On 1 March 2018 at 07:03, Euler Taveira  wrote:

> Hi,
>
> The attached patches add support for filtering rows in the publisher.
> The output plugin will do the work if a filter was defined in CREATE
> PUBLICATION command. An optional WHERE clause can be added after the
> table name in the CREATE PUBLICATION such as:
>
> CREATE PUBLICATION foo FOR TABLE departments WHERE (id > 2000 AND id <=
> 3000);
>
> Row that doesn't match the WHERE clause will not be sent to the
> subscribers.
>
> Patches 0001 and 0002 are only refactors and can be applied
> independently. 0003 doesn't include row filtering on initial
> synchronization.
>


Good idea. I haven't read this yet, but one thing to make sure you've
handled is limiting the clause to referencing only the current tuple and
the catalogs. user-catalog tables are OK, too, anything that is
RelationIsAccessibleInLogicalDecoding().

This means only immutable functions may be invoked, since a stable or
volatile function might attempt to access a table. And views must be
prohibited or recursively checked. (We have tree walkers that would help
with this).

It might be worth looking at the current logic for CHECK expressions, since
the requirements are similar. In my opinion you could safely not bother
with allowing access to user catalog tables in the filter expressions and
limit them strictly to immutable functions and the tuple its self.

-- 
 Craig Ringer   http://www.2ndQuadrant.com/
 PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Erik Rijkers]

On 2018-03-01 00:03, Euler Taveira wrote:

The attached patches add support for filtering rows in the publisher.



001-Refactor-function-create_estate_for_relation.patch
0002-Rename-a-WHERE-node.patch
0003-Row-filtering-for-logical-replication.patch



Comments?


Very, very useful.  I really do hope this patch survives the 
late-arrival-cull.


I built this functionality into a test program I have been using and in 
simple cascading replication tests it works well.


I did find what I think is a bug (a bug easy to avoid but also easy to 
run into):
The test I used was to cascade 3 instances (all on one machine) from 
A->B->C

I ran a pgbench session in instance A, and used:
  in A: alter publication pub0_6515 add table pgbench_accounts where 
(aid between 4 and 6-1);

  in B: alter publication pub1_6516 add table pgbench_accounts;

The above worked well, but when I did the same but used the filter in 
both publications:
  in A: alter publication pub0_6515 add table pgbench_accounts where 
(aid between 4 and 6-1);
  in B: alter publication pub1_6516 add table pgbench_accounts where 
(aid between 4 and 6-1);


then the replication only worked for (pgbench-)scale 1 (hence: very 
little data); with larger scales it became slow (taking many minutes 
where the above had taken less than 1 minute), and ended up using far 
too much memory (or blowing up/crashing altogether).  Something not 
quite right there.


Nevertheless, I am much in favour of acquiring this functionality as 
soon as possible.



Thanks,


Erik Rijkers

Re: row filtering for logical replication

[Euler Taveira]

2018-02-28 21:47 GMT-03:00 David Fetter :
> I noticed that the WHERE clause applies to all tables in the
> publication.  Is that actually the right thing?  I'm thinking of a
> case where we have foo(id, ...) and bar(foo_id, ).  To slice that
> correctly, we'd want to do the ids in the foo table and the foo_ids in
> the bar table.  In the system as written, that would entail, at least
> potentially, writing a lot of publications by hand.
>
I didn't make it clear in my previous email and I think you misread
the attached docs. Each table can have an optional WHERE clause. I'll
made it clear when I rewrite the tests. Something like:

CREATE PUBLICATION tap_pub FOR TABLE tab_rowfilter_1 WHERE (a > 1000
AND b <> 'filtered'), tab_rowfilter_2 WHERE (c % 2 = 0),
tab_rowfilter_3;

Such syntax will not block another future feature that will publish
only few columns of the table.

> I also noticed that in psql, \dRp+ doesn't show the WHERE clause,
> which it probably should.
>
Yea, it could be added be I'm afraid of such long WHERE clauses.

> Does it need regression tests?
>
I included some tests just to demonstrate the feature but I'm planning
to add a separate test file for it.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[David Fetter]

On Thu, Mar 01, 2018 at 12:41:04PM -0300, Euler Taveira wrote:
> 2018-02-28 21:47 GMT-03:00 David Fetter :
> > I noticed that the WHERE clause applies to all tables in the
> > publication.  Is that actually the right thing?  I'm thinking of a
> > case where we have foo(id, ...) and bar(foo_id, ).  To slice that
> > correctly, we'd want to do the ids in the foo table and the foo_ids in
> > the bar table.  In the system as written, that would entail, at least
> > potentially, writing a lot of publications by hand.
> >
> I didn't make it clear in my previous email and I think you misread
> the attached docs. Each table can have an optional WHERE clause. I'll
> made it clear when I rewrite the tests. Something like:

Sorry I misunderstood.

> CREATE PUBLICATION tap_pub FOR TABLE tab_rowfilter_1 WHERE (a > 1000
> AND b <> 'filtered'), tab_rowfilter_2 WHERE (c % 2 = 0),
> tab_rowfilter_3;

That's great!

> Such syntax will not block another future feature that will publish
> only few columns of the table.
> 
> > I also noticed that in psql, \dRp+ doesn't show the WHERE clause,
> > which it probably should.
> >
> Yea, it could be added be I'm afraid of such long WHERE clauses.

I think of + as signifying, "I am ready to get a LOT of output in
order to see more detail."  Perhaps that's just me.

> > Does it need regression tests?
> >
> I included some tests just to demonstrate the feature but I'm
> planning to add a separate test file for it.

Excellent. This feature looks like a nice big chunk of the user-space
infrastructure needed for sharding, among other things.

Best,
David.
-- 
David Fetter  http://fetter.org/
Phone: +1 415 235 3778

Remember to vote!
Consider donating to Postgres: http://www.postgresql.org/about/donate

Re: row filtering for logical replication

[Erik Rijkers]

On 2018-03-01 16:27, Erik Rijkers wrote:

On 2018-03-01 00:03, Euler Taveira wrote:

The attached patches add support for filtering rows in the publisher.



001-Refactor-function-create_estate_for_relation.patch
0002-Rename-a-WHERE-node.patch
0003-Row-filtering-for-logical-replication.patch



Comments?


Very, very useful.  I really do hope this patch survives the 
late-arrival-cull.


I built this functionality into a test program I have been using and
in simple cascading replication tests it works well.

I did find what I think is a bug (a bug easy to avoid but also easy to
run into):
The test I used was to cascade 3 instances (all on one machine) from 
A->B->C

I ran a pgbench session in instance A, and used:
  in A: alter publication pub0_6515 add table pgbench_accounts where
(aid between 4 and 6-1);
  in B: alter publication pub1_6516 add table pgbench_accounts;

The above worked well, but when I did the same but used the filter in
both publications:
  in A: alter publication pub0_6515 add table pgbench_accounts where
(aid between 4 and 6-1);
  in B: alter publication pub1_6516 add table pgbench_accounts where
(aid between 4 and 6-1);

then the replication only worked for (pgbench-)scale 1 (hence: very
little data); with larger scales it became slow (taking many minutes
where the above had taken less than 1 minute), and ended up using far
too much memory (or blowing up/crashing altogether).  Something not
quite right there.

Nevertheless, I am much in favour of acquiring this functionality as
soon as possible.



Attached is 'logrep_rowfilter.sh', a demonstration of above-described 
bug.


The program runs initdb for 3 instances in /tmp (using ports 6515, 6516, 
and 6517) and sets up logical replication from 1->2->3.


It can be made to work by removing de where-clause on the second 'create 
publication' ( i.e., outcomment the $where2 variable ).




Thanks,


Erik Rijkers
#!/bin/sh

# postges binary with 
#
#  0001-Refactor-function-create_estate_for_relation.patch
#  0002-Rename-a-WHERE-node.patch
#  0003-Row-filtering-for-logical-replication.patch
#

unset PGDATABASE PGPORT PGSERVICE
export PGDATABASE=postgres

scale=10

root_dir=/tmp/cascade

BIN=$HOME/pg_stuff/pg_installations/pgsql.logrep_rowfilter/bin.fast

export PATH=$BIN:$PATH

  initdb=$BIN/initdb
postgres=$BIN/postgres
  pg_ctl=$BIN/pg_ctl
baseport=6515

if [[ -d $root_dir/instance1 ]]; then rm -rf $root_dir/instance1; fi
if [[ -d $root_dir/instance2 ]]; then rm -rf $root_dir/instance2; fi
if [[ -d $root_dir/instance3 ]]; then rm -rf $root_dir/instance3; fi
if [[ -d $root_dir/instance1 ]]; then exit ; fi
if [[ -d $root_dir/instance2 ]]; then exit ; fi
if [[ -d $root_dir/instance3 ]]; then exit ; fi

devel_file=/tmp/bugs
echo filterbug>$devel_file

num_instances=3

for n in `seq 1 $num_instances`
do
  instance=instance$n
  server_dir=$root_dir/$instance
  data_dir=$server_dir/data
  port=$(( 6515 + $n -1 ))
  logfile=$server_dir/logfile.$port
  echo "-- $initdb --pgdata=$data_dir --encoding=UTF8 --pwfile=$devel_file "
   $initdb --pgdata=$data_dir --encoding=UTF8 --pwfile=$devel_file  &> /dev/null
  ( $postgres  -D $data_dir -p $port \
--wal_level=logical --logging_collector=on \
--client_min_messages=warning \
--log_directory=$server_dir --log_filename=logfile.${port} \
--log_replication_commands=on & ) &> /dev/null
done 

echo "sleep 3s"
sleep 3

echo "
  drop table if exists pgbench_accounts;
  drop table if exists pgbench_branches;
  drop table if exists pgbench_tellers;
  drop table if exists pgbench_history;" | psql -qXp 6515 \
&& echo "
  drop table if exists pgbench_accounts;
  drop table if exists pgbench_branches;
  drop table if exists pgbench_tellers;
  drop table if exists pgbench_history;" | psql -qXp 6516 \
&& pgbench -p 6515 -qis $scale \
&& echo "alter table pgbench_history add column hid serial primary key;" \
  | psql -q1Xp 6515 && pg_dump -F c -p 6515 \
 --exclude-table-data=pgbench_history  \
 --exclude-table-data=pgbench_accounts \
 --exclude-table-data=pgbench_branches \
 --exclude-table-data=pgbench_tellers  \
   -t pgbench_history -t pgbench_accounts \
   -t pgbench_branches -t pgbench_tellers \
  | pg_restore -1 -p 6516 -d postgres

appname=rowfilter

   where="where (aid between 4 and 6-1)"
  where2="where (aid between 4 and 6-1)"

echo "
create publication pub1;
alter publication pub1 add table pgbench_accounts $where ; --> where 1
alter publication pub1 add table pgbench_branches;
alter publication pub1 add table pgbench_tellers;
alter publication pub1 add table pgbench_history;
" | psql -p 6515 -aqtAX

if [[ $num_instances -eq 3 ]]; then

  pg_dump -F c -p 6515 \
 --exclude-table-data=pgbench_history  \
 --exclude-table-data=pgbench_accounts \
 --exclude-table-data=pgbench_branches \
 --exclude-table-data=pgbench_tellers  \
 -t pgbench_history -t pgbench_accounts \
 -t pgbench_branches -t pgbench_tellers \

Re: row filtering for logical replication

[Andres Freund]

Hi,

On 2018-03-01 16:27:11 +0100, Erik Rijkers wrote:
> Very, very useful.  I really do hope this patch survives the
> late-arrival-cull.

FWIW, I don't think it'd be fair or prudent. There's definitely some
issues (see e.g. Craig's reply), and I don't see why this patch'd
deserve an exemption from the "nontrivial patches shouldn't be submitted
to the last CF" policy?

- Andres

Re: row filtering for logical replication

[David Steele]

Hi,

On 3/1/18 4:27 PM, Andres Freund wrote:
> On 2018-03-01 16:27:11 +0100, Erik Rijkers wrote:
>> Very, very useful.  I really do hope this patch survives the
>> late-arrival-cull.
> 
> FWIW, I don't think it'd be fair or prudent. There's definitely some
> issues (see e.g. Craig's reply), and I don't see why this patch'd
> deserve an exemption from the "nontrivial patches shouldn't be submitted
> to the last CF" policy?

I'm unable to find this in the CF under the title or author name.  If it
didn't get entered then it is definitely out.

If it does have an entry, then I agree with Andres that it should be
pushed to the next CF.

-- 
-David
da...@pgmasters.net

Re: row filtering for logical replication

[Euler Taveira]

2018-03-01 18:27 GMT-03:00 Andres Freund :
> FWIW, I don't think it'd be fair or prudent. There's definitely some
> issues (see e.g. Craig's reply), and I don't see why this patch'd
> deserve an exemption from the "nontrivial patches shouldn't be submitted
> to the last CF" policy?
>
I forgot to mention but this feature is for v12. I know the rules and
that is why I didn't add it to the in progress CF.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Euler Taveira]

2018-03-01 18:25 GMT-03:00 Erik Rijkers :
> Attached is 'logrep_rowfilter.sh', a demonstration of above-described bug.
>
Thanks for testing. I will figure out what is happening. There are
some leaks around. I'll post another version when I fix some of those
bugs.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Euler Taveira]

2018-02-28 21:54 GMT-03:00 Craig Ringer :
> Good idea. I haven't read this yet, but one thing to make sure you've
> handled is limiting the clause to referencing only the current tuple and the
> catalogs. user-catalog tables are OK, too, anything that is
> RelationIsAccessibleInLogicalDecoding().
>
> This means only immutable functions may be invoked, since a stable or
> volatile function might attempt to access a table. And views must be
> prohibited or recursively checked. (We have tree walkers that would help
> with this).
>
> It might be worth looking at the current logic for CHECK expressions, since
> the requirements are similar. In my opinion you could safely not bother with
> allowing access to user catalog tables in the filter expressions and limit
> them strictly to immutable functions and the tuple its self.
>
IIRC implementation is similar to RLS expressions. I'll check all of
these rules.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[David Steele]

On 3/1/18 6:00 PM, Euler Taveira wrote:

2018-03-01 18:27 GMT-03:00 Andres Freund :

FWIW, I don't think it'd be fair or prudent. There's definitely some
issues (see e.g. Craig's reply), and I don't see why this patch'd
deserve an exemption from the "nontrivial patches shouldn't be submitted
to the last CF" policy?


I forgot to mention but this feature is for v12. I know the rules and
that is why I didn't add it to the in progress CF.


That was the right thing to do, thank you!

--
-David
da...@pgmasters.net

Re: row filtering for logical replication

[Petr Jelinek]

On 01/11/2018 01:29, Euler Taveira wrote:
> Em qua, 28 de fev de 2018 às 20:03, Euler Taveira
>  escreveu:
>> The attached patches add support for filtering rows in the publisher.
>>
> I rebased the patch. I added row filtering for initial
> synchronization, pg_dump support and psql support. 0001 removes unused
> code. 0002 reduces memory use. 0003 passes only structure member that
> is used in create_estate_for_relation. 0004 reuses a parser node for
> row filtering. 0005 is the feature. 0006 prints WHERE expression in
> psql. 0007 adds pg_dump support. 0008 is only for debug purposes (I'm
> not sure some of these messages will be part of the final patch).
> 0001, 0002, 0003 and 0008 are not mandatory for this feature.
> 
> Comments?
> 

Hi,

I think there are two main topics that still need to be discussed about
this patch.

Firstly, I am not sure if it's wise to allow UDFs in the filter clause
for the table. The reason for that is that we can't record all necessary
dependencies there because the functions are black box for parser. That
means if somebody drops object that an UDF used in replication filter
depends on, that function will start failing. But unlike for user
sessions it will start failing during decoding (well processing in
output plugin). And that's not recoverable by reading the missing
object, the only way to get out of that is either to move slot forward
which means losing part of replication stream and need for manual resync
or full rebuild of replication. Neither of which are good IMHO.

Secondly, do we want to at least notify user on filters (or maybe even
disallow them) with combination of action + column where column value
will not be logged? I mean for example we do this when processing the
filter against a row:

> + ExecStoreHeapTuple(new_tuple ? new_tuple : old_tuple, 
> ecxt->ecxt_scantuple, false);

But if user has expression on column which is not part of replica
identity that expression will always return NULL for DELETEs because
only replica identity is logged with actual values and everything else
in NULL in old_tuple. So if publication replicates deletes we should
check for this somehow.

Btw about code (you already fixed the wrong reloid in sync so skipping
that).

0002:
> + for (tupn = 0; tupn < walres->ntuples; tupn++)
>   {
> - char   *cstrs[MaxTupleAttributeNumber];
> + char**cstrs;
>  
>   CHECK_FOR_INTERRUPTS();
>  
>   /* Do the allocations in temporary context. */
>   oldcontext = MemoryContextSwitchTo(rowcontext);
>  
> + cstrs = palloc(nfields * sizeof(char *));

Not really sure that this is actually worth it given that we have to
allocate and free this in a loop now while before it was just sitting on
a stack.

0005:
> @@ -654,5 +740,10 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, 
> uint32 hashvalue)
>*/
>   hash_seq_init(&status, RelationSyncCache);
>   while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL)
> + {
>   entry->replicate_valid = false;
> + if (list_length(entry->row_filter) > 0)
> + list_free(entry->row_filter);
> + entry->row_filter = NIL;
> + }

Won't this leak memory? The list_free only frees the list cells, but not
the nodes you stored there before.

Also I think we should document here that the expression is run with the
session environment of the replication connection (so that it's more
obvious that things like CURRENT_USER will not return user which changed
tuple but the replication user).

It would be nice if 0006 had regression test and 0007 TAP test.

-- 
  Petr Jelinek  http://www.2ndQuadrant.com/
  PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Stephen Frost]

Greetings,

* Euler Taveira (eu...@timbira.com.br) wrote:
> 2018-02-28 21:54 GMT-03:00 Craig Ringer :
> > Good idea. I haven't read this yet, but one thing to make sure you've
> > handled is limiting the clause to referencing only the current tuple and the
> > catalogs. user-catalog tables are OK, too, anything that is
> > RelationIsAccessibleInLogicalDecoding().
> >
> > This means only immutable functions may be invoked, since a stable or
> > volatile function might attempt to access a table. And views must be
> > prohibited or recursively checked. (We have tree walkers that would help
> > with this).
> >
> > It might be worth looking at the current logic for CHECK expressions, since
> > the requirements are similar. In my opinion you could safely not bother with
> > allowing access to user catalog tables in the filter expressions and limit
> > them strictly to immutable functions and the tuple its self.
>
> IIRC implementation is similar to RLS expressions. I'll check all of
> these rules.

Given the similarity to RLS and the nearby discussion about allowing
non-superusers to create subscriptions, and probably publications later,
I wonder if we shouldn't be somehow associating this with RLS policies
instead of having the publication filtering be entirely independent..

Thanks!

Stephen


signature.asc
Description: PGP signature

Re: row filtering for logical replication

[Petr Jelinek]

On 23/11/2018 03:02, Stephen Frost wrote:
> Greetings,
> 
> * Euler Taveira (eu...@timbira.com.br) wrote:
>> 2018-02-28 21:54 GMT-03:00 Craig Ringer :
>>> Good idea. I haven't read this yet, but one thing to make sure you've
>>> handled is limiting the clause to referencing only the current tuple and the
>>> catalogs. user-catalog tables are OK, too, anything that is
>>> RelationIsAccessibleInLogicalDecoding().
>>>
>>> This means only immutable functions may be invoked, since a stable or
>>> volatile function might attempt to access a table. And views must be
>>> prohibited or recursively checked. (We have tree walkers that would help
>>> with this).
>>>
>>> It might be worth looking at the current logic for CHECK expressions, since
>>> the requirements are similar. In my opinion you could safely not bother with
>>> allowing access to user catalog tables in the filter expressions and limit
>>> them strictly to immutable functions and the tuple its self.
>>
>> IIRC implementation is similar to RLS expressions. I'll check all of
>> these rules.
> 
> Given the similarity to RLS and the nearby discussion about allowing
> non-superusers to create subscriptions, and probably publications later,
> I wonder if we shouldn't be somehow associating this with RLS policies
> instead of having the publication filtering be entirely independent..
> 
I do see the appeal here, if you consider logical replication to be a
streaming select it probably applies well.

But given that this is happening inside output plugin which does not
have full executor setup and has catalog-only snapshot I am not sure how
feasible it is to try to merge these two things. As per my previous
email it's possible that we'll have to be stricter about what we allow
in expressions here.

The other issue with merging this is that the use-case for filtering out
the data in logical replication is not necessarily about security, but
often about sending only relevant data. So it makes sense to have filter
on publication without RLS enabled on table and if we'd force that, we'd
limit usefulness of this feature.

We definitely want to eventually create subscriptions as non-superuser
but that has zero effect on this as everything here is happening on
different server than where subscription lives (we already allow
creation of publications with just CREATE privilege on database and
ownership of the table).

-- 
  Petr Jelinek  http://www.2ndQuadrant.com/
  PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Euler Taveira]

Em sex, 23 de nov de 2018 às 11:40, Petr Jelinek
 escreveu:
> But given that this is happening inside output plugin which does not
> have full executor setup and has catalog-only snapshot I am not sure how
> feasible it is to try to merge these two things. As per my previous
> email it's possible that we'll have to be stricter about what we allow
> in expressions here.
>
This feature should be as simple as possible. I don't want to
introduce a huge overhead just for filtering some data. Data sharding
generally uses simple expressions.

> The other issue with merging this is that the use-case for filtering out
> the data in logical replication is not necessarily about security, but
> often about sending only relevant data. So it makes sense to have filter
> on publication without RLS enabled on table and if we'd force that, we'd
> limit usefulness of this feature.
>
Use the same infrastructure as RLS could be a good idea but use RLS
for row filtering is not. RLS is complex.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Euler Taveira]

Em qui, 22 de nov de 2018 às 20:03, Petr Jelinek
 escreveu:
> Firstly, I am not sure if it's wise to allow UDFs in the filter clause
> for the table. The reason for that is that we can't record all necessary
> dependencies there because the functions are black box for parser. That
> means if somebody drops object that an UDF used in replication filter
> depends on, that function will start failing. But unlike for user
> sessions it will start failing during decoding (well processing in
> output plugin). And that's not recoverable by reading the missing
> object, the only way to get out of that is either to move slot forward
> which means losing part of replication stream and need for manual resync
> or full rebuild of replication. Neither of which are good IMHO.
>
It is a foot gun but there are several ways to do bad things in
postgres. CREATE PUBLICATION is restricted to superusers and role with
CREATE privilege in current database. AFAICS a role with CREATE
privilege cannot drop objects whose owner is not himself. I wouldn't
like to disallow UDFs in row filtering expressions just because
someone doesn't set permissions correctly. Do you have any other case
in mind?

> Secondly, do we want to at least notify user on filters (or maybe even
> disallow them) with combination of action + column where column value
> will not be logged? I mean for example we do this when processing the
> filter against a row:
>
> > + ExecStoreHeapTuple(new_tuple ? new_tuple : old_tuple, 
> > ecxt->ecxt_scantuple, false);
>
We could emit a LOG message. That could possibly be an option but it
could be too complex for the first version.

> But if user has expression on column which is not part of replica
> identity that expression will always return NULL for DELETEs because
> only replica identity is logged with actual values and everything else
> in NULL in old_tuple. So if publication replicates deletes we should
> check for this somehow.
>
In this case, we should document this behavior. That is a recurring
question in wal2json issues. Besides that we should explain that
UPDATE/DELETE tuples doesn't log all columns (people think the
behavior is equivalent to triggers; it is not unless you set REPLICA
IDENTITY FULL).

> Not really sure that this is actually worth it given that we have to
> allocate and free this in a loop now while before it was just sitting on
> a stack.
>
That is a experimentation code that should be in a separate patch.
Don't you think low memory use is a good goal? I also think that
MaxTupleAttributeNumber is an extreme value. I didn't some preliminary
tests and didn't notice overheads. I'll leave these modifications in a
separate patch.

> Won't this leak memory? The list_free only frees the list cells, but not
> the nodes you stored there before.
>
Good catch. It should be list_free_deep.

> Also I think we should document here that the expression is run with the
> session environment of the replication connection (so that it's more
> obvious that things like CURRENT_USER will not return user which changed
> tuple but the replication user).
>
Sure.

> It would be nice if 0006 had regression test and 0007 TAP test.
>
Sure.

Besides the problem presented by Hironobu-san, I'm doing some cleanup
and improving docs. I also forget to declare pg_publication_rel TOAST
table.

Thanks for your review.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Alvaro Herrera]

On 2018-Nov-23, Euler Taveira wrote:

> Em qui, 22 de nov de 2018 às 20:03, Petr Jelinek
>  escreveu:

> > Won't this leak memory? The list_free only frees the list cells, but not
> > the nodes you stored there before.
>
> Good catch. It should be list_free_deep.

Actually, if the nodes have more structure (say you palloc one list
item, but that list item also contains pointers to a Node) then a
list_free_deep won't be enough either.  I'd suggest to create a bespoke
memory context, which you can delete afterwards.

-- 
Álvaro Herrerahttps://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services

Re: row filtering for logical replication

[David Fetter]

On Fri, Nov 23, 2018 at 12:03:27AM +0100, Petr Jelinek wrote:
> On 01/11/2018 01:29, Euler Taveira wrote:
> > Em qua, 28 de fev de 2018 às 20:03, Euler Taveira
> >  escreveu:
> >> The attached patches add support for filtering rows in the publisher.
> >>
> > I rebased the patch. I added row filtering for initial
> > synchronization, pg_dump support and psql support. 0001 removes unused
> > code. 0002 reduces memory use. 0003 passes only structure member that
> > is used in create_estate_for_relation. 0004 reuses a parser node for
> > row filtering. 0005 is the feature. 0006 prints WHERE expression in
> > psql. 0007 adds pg_dump support. 0008 is only for debug purposes (I'm
> > not sure some of these messages will be part of the final patch).
> > 0001, 0002, 0003 and 0008 are not mandatory for this feature.
> 
> Hi,
> 
> I think there are two main topics that still need to be discussed about
> this patch.
> 
> Firstly, I am not sure if it's wise to allow UDFs in the filter clause
> for the table. The reason for that is that we can't record all necessary
> dependencies there because the functions are black box for parser.

Some UDFs are not a black box for the parser, namely ones written in
SQL. Would it make sense at least not to foreclose the non-(black box)
option?

Best,
David.
-- 
David Fetter  http://fetter.org/
Phone: +1 415 235 3778

Remember to vote!
Consider donating to Postgres: http://www.postgresql.org/about/donate

Re: row filtering for logical replication

[Petr Jelinek]

On 23/11/2018 17:39, David Fetter wrote:
> On Fri, Nov 23, 2018 at 12:03:27AM +0100, Petr Jelinek wrote:
>> On 01/11/2018 01:29, Euler Taveira wrote:
>>> Em qua, 28 de fev de 2018 às 20:03, Euler Taveira
>>>  escreveu:
 The attached patches add support for filtering rows in the publisher.

>>> I rebased the patch. I added row filtering for initial
>>> synchronization, pg_dump support and psql support. 0001 removes unused
>>> code. 0002 reduces memory use. 0003 passes only structure member that
>>> is used in create_estate_for_relation. 0004 reuses a parser node for
>>> row filtering. 0005 is the feature. 0006 prints WHERE expression in
>>> psql. 0007 adds pg_dump support. 0008 is only for debug purposes (I'm
>>> not sure some of these messages will be part of the final patch).
>>> 0001, 0002, 0003 and 0008 are not mandatory for this feature.
>>
>> Hi,
>>
>> I think there are two main topics that still need to be discussed about
>> this patch.
>>
>> Firstly, I am not sure if it's wise to allow UDFs in the filter clause
>> for the table. The reason for that is that we can't record all necessary
>> dependencies there because the functions are black box for parser.
> 
> Some UDFs are not a black box for the parser, namely ones written in
> SQL. Would it make sense at least not to foreclose the non-(black box)
> option?
> 

Yeah inlinable SQL functions should be fine, we just need the ability to
extract dependencies.

-- 
  Petr Jelinek  http://www.2ndQuadrant.com/
  PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Petr Jelinek]

On 23/11/2018 17:15, Euler Taveira wrote:
> Em qui, 22 de nov de 2018 às 20:03, Petr Jelinek
>  escreveu:
>> Firstly, I am not sure if it's wise to allow UDFs in the filter clause
>> for the table. The reason for that is that we can't record all necessary
>> dependencies there because the functions are black box for parser. That
>> means if somebody drops object that an UDF used in replication filter
>> depends on, that function will start failing. But unlike for user
>> sessions it will start failing during decoding (well processing in
>> output plugin). And that's not recoverable by reading the missing
>> object, the only way to get out of that is either to move slot forward
>> which means losing part of replication stream and need for manual resync
>> or full rebuild of replication. Neither of which are good IMHO.
>>
> It is a foot gun but there are several ways to do bad things in
> postgres. CREATE PUBLICATION is restricted to superusers and role with
> CREATE privilege in current database. AFAICS a role with CREATE
> privilege cannot drop objects whose owner is not himself. I wouldn't
> like to disallow UDFs in row filtering expressions just because
> someone doesn't set permissions correctly. Do you have any other case
> in mind?

I don't think this has anything to do with security. Stupid example:

user1: CREATE EXTENSION citext;

user2: CREATE FUNCTION myfilter(col1 text, col2 text) returns boolean
language plpgsql as
$$BEGIN
RETURN col1::citext = col2::citext;
END;$$

user2: ALTER PUBLICATION mypub ADD TABLE mytab WHERE (myfilter(a,b));

[... replication happening ...]

user1: DROP EXTENSION citext;

And now replication is broken and unrecoverable without data loss.
Recreating extension will not help because the changes happening in
meantime will not see it in the historical snapshot.

I don't think it's okay to do completely nothing about this.

> 
>> Secondly, do we want to at least notify user on filters (or maybe even
>> disallow them) with combination of action + column where column value
>> will not be logged? I mean for example we do this when processing the
>> filter against a row:
>>
>>> + ExecStoreHeapTuple(new_tuple ? new_tuple : old_tuple, 
>>> ecxt->ecxt_scantuple, false);
>>
> We could emit a LOG message. That could possibly be an option but it
> could be too complex for the first version.
>

Well, it needs walker which extracts Vars from the expression and checks
them against replica identity columns. We already have a way to fetch
replica identity columns and the walker could be something like
simplified version of the find_expr_references_walker used by the
recordDependencyOnSingleRelExpr (I don't think there is anything ready
made already).

>> But if user has expression on column which is not part of replica
>> identity that expression will always return NULL for DELETEs because
>> only replica identity is logged with actual values and everything else
>> in NULL in old_tuple. So if publication replicates deletes we should
>> check for this somehow.
>>
> In this case, we should document this behavior. That is a recurring
> question in wal2json issues. Besides that we should explain that
> UPDATE/DELETE tuples doesn't log all columns (people think the
> behavior is equivalent to triggers; it is not unless you set REPLICA
> IDENTITY FULL).
> 
>> Not really sure that this is actually worth it given that we have to
>> allocate and free this in a loop now while before it was just sitting on
>> a stack.
>>
> That is a experimentation code that should be in a separate patch.
> Don't you think low memory use is a good goal? I also think that
> MaxTupleAttributeNumber is an extreme value. I didn't some preliminary
> tests and didn't notice overheads. I'll leave these modifications in a
> separate patch.
> 

It's static memory and it's a few KB of it (it's just single array of
pointers, not array of data, and does not depend on the number of rows).
Palloc will definitely need more CPU cycles.

-- 
  Petr Jelinek  http://www.2ndQuadrant.com/
  PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Fabrízio de Royes Mello]

On Fri, Nov 23, 2018 at 3:55 PM Petr Jelinek 
wrote:
>
> On 23/11/2018 17:15, Euler Taveira wrote:
> > Em qui, 22 de nov de 2018 às 20:03, Petr Jelinek
> >  escreveu:
> >> Firstly, I am not sure if it's wise to allow UDFs in the filter clause
> >> for the table. The reason for that is that we can't record all
necessary
> >> dependencies there because the functions are black box for parser. That
> >> means if somebody drops object that an UDF used in replication filter
> >> depends on, that function will start failing. But unlike for user
> >> sessions it will start failing during decoding (well processing in
> >> output plugin). And that's not recoverable by reading the missing
> >> object, the only way to get out of that is either to move slot forward
> >> which means losing part of replication stream and need for manual
resync
> >> or full rebuild of replication. Neither of which are good IMHO.
> >>
> > It is a foot gun but there are several ways to do bad things in
> > postgres. CREATE PUBLICATION is restricted to superusers and role with
> > CREATE privilege in current database. AFAICS a role with CREATE
> > privilege cannot drop objects whose owner is not himself. I wouldn't
> > like to disallow UDFs in row filtering expressions just because
> > someone doesn't set permissions correctly. Do you have any other case
> > in mind?
>
> I don't think this has anything to do with security. Stupid example:
>
> user1: CREATE EXTENSION citext;
>
> user2: CREATE FUNCTION myfilter(col1 text, col2 text) returns boolean
> language plpgsql as
> $$BEGIN
> RETURN col1::citext = col2::citext;
> END;$$
>
> user2: ALTER PUBLICATION mypub ADD TABLE mytab WHERE (myfilter(a,b));
>
> [... replication happening ...]
>
> user1: DROP EXTENSION citext;
>
> And now replication is broken and unrecoverable without data loss.
> Recreating extension will not help because the changes happening in
> meantime will not see it in the historical snapshot.
>
> I don't think it's okay to do completely nothing about this.
>

If carefully documented I see no problem with it... we already have an
analogous problem with functional indexes.

Regards,

--
   Fabrízio de Royes Mello Timbira - http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Petr Jelinek]

On 23/11/2018 19:05, Fabrízio de Royes Mello wrote:
> On Fri, Nov 23, 2018 at 3:55 PM Petr Jelinek
> mailto:petr.jeli...@2ndquadrant.com>> wrote:
>>
>> On 23/11/2018 17:15, Euler Taveira wrote:
>> > Em qui, 22 de nov de 2018 às 20:03, Petr Jelinek
>> > mailto:petr.jeli...@2ndquadrant.com>>
> escreveu:
>> >> Firstly, I am not sure if it's wise to allow UDFs in the filter clause
>> >> for the table. The reason for that is that we can't record all
> necessary
>> >> dependencies there because the functions are black box for parser. That
>> >> means if somebody drops object that an UDF used in replication filter
>> >> depends on, that function will start failing. But unlike for user
>> >> sessions it will start failing during decoding (well processing in
>> >> output plugin). And that's not recoverable by reading the missing
>> >> object, the only way to get out of that is either to move slot forward
>> >> which means losing part of replication stream and need for manual
> resync
>> >> or full rebuild of replication. Neither of which are good IMHO.
>> >>
>> > It is a foot gun but there are several ways to do bad things in
>> > postgres. CREATE PUBLICATION is restricted to superusers and role with
>> > CREATE privilege in current database. AFAICS a role with CREATE
>> > privilege cannot drop objects whose owner is not himself. I wouldn't
>> > like to disallow UDFs in row filtering expressions just because
>> > someone doesn't set permissions correctly. Do you have any other case
>> > in mind?
>>
>> I don't think this has anything to do with security. Stupid example:
>>
>> user1: CREATE EXTENSION citext;
>>
>> user2: CREATE FUNCTION myfilter(col1 text, col2 text) returns boolean
>> language plpgsql as
>> $$BEGIN
>> RETURN col1::citext = col2::citext;
>> END;$$
>>
>> user2: ALTER PUBLICATION mypub ADD TABLE mytab WHERE (myfilter(a,b));
>>
>> [... replication happening ...]
>>
>> user1: DROP EXTENSION citext;
>>
>> And now replication is broken and unrecoverable without data loss.
>> Recreating extension will not help because the changes happening in
>> meantime will not see it in the historical snapshot.
>>
>> I don't think it's okay to do completely nothing about this.
>>
> 
> If carefully documented I see no problem with it... we already have an
> analogous problem with functional indexes.

The difference is that with functional indexes you can recreate the
missing object and everything is okay again. With logical replication
recreating the object will not help.

-- 
  Petr Jelinek  http://www.2ndQuadrant.com/
  PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Fabrízio de Royes Mello]

On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek 
wrote:
>
> >
> > If carefully documented I see no problem with it... we already have an
> > analogous problem with functional indexes.
>
> The difference is that with functional indexes you can recreate the
> missing object and everything is okay again. With logical replication
> recreating the object will not help.
>

In this case with logical replication you should rsync the object. That is
the price of misunderstanding / bad use of the new feature.

As usual, there are no free beer ;-)

Regards,

--
   Fabrízio de Royes Mello Timbira - http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Stephen Frost]

Greetings,

* Fabrízio de Royes Mello (fabriziome...@gmail.com) wrote:
> On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek 
> wrote:
> > > If carefully documented I see no problem with it... we already have an
> > > analogous problem with functional indexes.
> >
> > The difference is that with functional indexes you can recreate the
> > missing object and everything is okay again. With logical replication
> > recreating the object will not help.
> >
> 
> In this case with logical replication you should rsync the object. That is
> the price of misunderstanding / bad use of the new feature.
> 
> As usual, there are no free beer ;-)

There's also certainly no shortage of other ways to break logical
replication, including ways that would also be hard to recover from
today other than doing a full resync.

What that seems to indicate, to me at least, is that it'd be awful nice
to have a way to resync the data which doesn't necessairly involve
transferring all of it over again.

Of course, it'd be nice if we could track those dependencies too,
but that's yet another thing.

In short, I'm not sure that I agree with the idea that we shouldn't
allow this and instead I'd rather we realize it and put the logical
replication into some kind of an error state that requires a resync.

Thanks!

Stephen


signature.asc
Description: PGP signature

Re: row filtering for logical replication

[Petr Jelinek]

On 23/11/2018 19:29, Fabrízio de Royes Mello wrote:
> 
> On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek
> mailto:petr.jeli...@2ndquadrant.com>> wrote:
>>
>> >
>> > If carefully documented I see no problem with it... we already have an
>> > analogous problem with functional indexes.
>>
>> The difference is that with functional indexes you can recreate the
>> missing object and everything is okay again. With logical replication
>> recreating the object will not help.
>>
> 
> In this case with logical replication you should rsync the object. That
> is the price of misunderstanding / bad use of the new feature.
> 
> As usual, there are no free beer ;-)
> 

Yeah but you have to resync whole subscription, not just single table
(removing table from the publication will also not help), that's pretty
severe punishment. What if you have triggers downstream that do
calculations or logging which you can't recover by simply rebuilding
replica? I think it's better to err on the side of no data loss.

We could also try to figure out a way to recover from this that does not
require resync, ie perhaps we could somehow temporarily force evaluation
of the expression to have current snapshot.

-- 
  Petr Jelinek  http://www.2ndQuadrant.com/
  PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Tomas Vondra]

On 11/23/18 8:03 PM, Stephen Frost wrote:
> Greetings,
> 
> * Fabrízio de Royes Mello (fabriziome...@gmail.com) wrote:
>> On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek 
>> wrote:
 If carefully documented I see no problem with it... we already have an
 analogous problem with functional indexes.
>>>
>>> The difference is that with functional indexes you can recreate the
>>> missing object and everything is okay again. With logical replication
>>> recreating the object will not help.
>>>
>>
>> In this case with logical replication you should rsync the object. That is
>> the price of misunderstanding / bad use of the new feature.
>>
>> As usual, there are no free beer ;-)
> 
> There's also certainly no shortage of other ways to break logical
> replication, including ways that would also be hard to recover from
> today other than doing a full resync.
> 

Sure, but that seems more like an argument against creating additional
ones (and for preventing those that already exist). I'm not sure this
particular feature is where we should draw the line, though.

> What that seems to indicate, to me at least, is that it'd be awful
> nice to have a way to resync the data which doesn't necessairly
> involve transferring all of it over again.
> 
> Of course, it'd be nice if we could track those dependencies too,
> but that's yet another thing.

Yep, that seems like a good idea in general. Both here and for
functional indexes (although I suppose sure is a technical reason why it
wasn't implemented right away for them).

> 
> In short, I'm not sure that I agree with the idea that we shouldn't
> allow this and instead I'd rather we realize it and put the logical
> replication into some kind of an error state that requires a resync.
> 

That would still mean a need to resync the data to recover, so I'm not
sure it's really an improvement. And I suppose it'd require tracking the
dependencies, because how else would you mark the subscription as
requiring a resync? At which point we could decline the DROP without a
CASCADE, just like we do elsewhere, no?

regards

-- 
Tomas Vondra  http://www.2ndQuadrant.com
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services

Re: row filtering for logical replication

[Tomas Vondra]

On 11/23/18 8:14 PM, Petr Jelinek wrote:
> On 23/11/2018 19:29, Fabrízio de Royes Mello wrote:
>>
>> On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek
>> mailto:petr.jeli...@2ndquadrant.com>> wrote:
>>>

 If carefully documented I see no problem with it... we already have an
 analogous problem with functional indexes.
>>>
>>> The difference is that with functional indexes you can recreate the
>>> missing object and everything is okay again. With logical replication
>>> recreating the object will not help.
>>>
>>
>> In this case with logical replication you should rsync the object. That
>> is the price of misunderstanding / bad use of the new feature.
>>
>> As usual, there are no free beer ;-)
>>
> 
> Yeah but you have to resync whole subscription, not just single table
> (removing table from the publication will also not help), that's pretty
> severe punishment. What if you have triggers downstream that do
> calculations or logging which you can't recover by simply rebuilding
> replica? I think it's better to err on the side of no data loss.
> 

Yeah, having to resync everything because you accidentally dropped a
function is quite annoying. Of course, you should notice that while
testing the upgrade in a testing environment, but still ...

> We could also try to figure out a way to recover from this that does not
> require resync, ie perhaps we could somehow temporarily force evaluation
> of the expression to have current snapshot.
> 

That seems like huge a can of worms ...


cheers

-- 
Tomas Vondra  http://www.2ndQuadrant.com
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services

Re: row filtering for logical replication

[Stephen Frost]

Greetings,

* Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
> On 23/11/2018 03:02, Stephen Frost wrote:
> > * Euler Taveira (eu...@timbira.com.br) wrote:
> >> 2018-02-28 21:54 GMT-03:00 Craig Ringer :
> >>> Good idea. I haven't read this yet, but one thing to make sure you've
> >>> handled is limiting the clause to referencing only the current tuple and 
> >>> the
> >>> catalogs. user-catalog tables are OK, too, anything that is
> >>> RelationIsAccessibleInLogicalDecoding().
> >>>
> >>> This means only immutable functions may be invoked, since a stable or
> >>> volatile function might attempt to access a table. And views must be
> >>> prohibited or recursively checked. (We have tree walkers that would help
> >>> with this).
> >>>
> >>> It might be worth looking at the current logic for CHECK expressions, 
> >>> since
> >>> the requirements are similar. In my opinion you could safely not bother 
> >>> with
> >>> allowing access to user catalog tables in the filter expressions and limit
> >>> them strictly to immutable functions and the tuple its self.
> >>
> >> IIRC implementation is similar to RLS expressions. I'll check all of
> >> these rules.
> > 
> > Given the similarity to RLS and the nearby discussion about allowing
> > non-superusers to create subscriptions, and probably publications later,
> > I wonder if we shouldn't be somehow associating this with RLS policies
> > instead of having the publication filtering be entirely independent..
>
> I do see the appeal here, if you consider logical replication to be a
> streaming select it probably applies well.
> 
> But given that this is happening inside output plugin which does not
> have full executor setup and has catalog-only snapshot I am not sure how
> feasible it is to try to merge these two things. As per my previous
> email it's possible that we'll have to be stricter about what we allow
> in expressions here.

I can certainly understand the concern about trying to combine the
implementation of this with that of RLS; perhaps that isn't a good fit
due to the additional constraints put on logical decoding.

That said, I still think it might make sense to consider these filters
for logical decoding to be policies and, ideally, to allow users to use
the same policy for both.

In the end, the idea of having to build a single large and complex
'create publication' command which has a bunch of tables, each with
their own filter clauses, just strikes me as pretty painful.

> The other issue with merging this is that the use-case for filtering out
> the data in logical replication is not necessarily about security, but
> often about sending only relevant data. So it makes sense to have filter
> on publication without RLS enabled on table and if we'd force that, we'd
> limit usefulness of this feature.

I definitely have a serious problem if we are going to say that you
can't use this filtering for security-sensitive cases.

> We definitely want to eventually create subscriptions as non-superuser
> but that has zero effect on this as everything here is happening on
> different server than where subscription lives (we already allow
> creation of publications with just CREATE privilege on database and
> ownership of the table).

What I wasn't clear about above was the idea that we might allow a user
other than the table owner to publish a given table, but that such a
publication should certanily only be allowed to include the rows which
that user has access to- as regulated by RLS.  If the RLS policy is too
complex to allow that then I would think we'd simply throw an error at
the create publication time and the would-be publisher would need to
figure that out with the table owner.

I'll admit that this might seem like a stretch, but what happens today?
Today, people write cronjobs to try to sync between tables with FDWs and
you don't need to own a table to use it as the target of a foreign
table.

I do think that we'll need to have some additional privileges around who
is allowed to create publications, I'm not entirely thrilled with that
being combined with the ability to create schemas; the two seem quite
different to me.

* Euler Taveira (eu...@timbira.com.br) wrote:
> Em sex, 23 de nov de 2018 às 11:40, Petr Jelinek
>  escreveu:
> > But given that this is happening inside output plugin which does not
> > have full executor setup and has catalog-only snapshot I am not sure how
> > feasible it is to try to merge these two things. As per my previous
> > email it's possible that we'll have to be stricter about what we allow
> > in expressions here.
>
> This feature should be as simple as possible. I don't want to
> introduce a huge overhead just for filtering some data. Data sharding
> generally uses simple expressions.

RLS often uses simple filters too.

> > The other issue with merging this is that the use-case for filtering out
> > the data in logical replication is not necessarily about security, but
> > often about sending only relevant data. So i

Re: row filtering for logical replication

[Stephen Frost]

Greetings,

* Tomas Vondra (tomas.von...@2ndquadrant.com) wrote:
> On 11/23/18 8:03 PM, Stephen Frost wrote:
> > * Fabrízio de Royes Mello (fabriziome...@gmail.com) wrote:
> >> On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek 
> >> wrote:
>  If carefully documented I see no problem with it... we already have an
>  analogous problem with functional indexes.
> >>>
> >>> The difference is that with functional indexes you can recreate the
> >>> missing object and everything is okay again. With logical replication
> >>> recreating the object will not help.
> >>>
> >>
> >> In this case with logical replication you should rsync the object. That is
> >> the price of misunderstanding / bad use of the new feature.
> >>
> >> As usual, there are no free beer ;-)
> > 
> > There's also certainly no shortage of other ways to break logical
> > replication, including ways that would also be hard to recover from
> > today other than doing a full resync.
> 
> Sure, but that seems more like an argument against creating additional
> ones (and for preventing those that already exist). I'm not sure this
> particular feature is where we should draw the line, though.

I was actually going in the other direction- we should allow it because
advanced users may know what they're doing better than we do and we
shouldn't prevent things just because they might be misused or
misunderstood by a user.

> > What that seems to indicate, to me at least, is that it'd be awful
> > nice to have a way to resync the data which doesn't necessairly
> > involve transferring all of it over again.
> > 
> > Of course, it'd be nice if we could track those dependencies too,
> > but that's yet another thing.
> 
> Yep, that seems like a good idea in general. Both here and for
> functional indexes (although I suppose sure is a technical reason why it
> wasn't implemented right away for them).

We don't track function dependencies in general and I could certainly
see cases where you really wouldn't want to do so, at least not in the
same way that we track FKs or similar.  I do wonder if maybe we didn't
track function dependencies because we didn't (yet) have create or
replace function and that now we should.  We don't track dependencies
inside a function either though.

> > In short, I'm not sure that I agree with the idea that we shouldn't
> > allow this and instead I'd rather we realize it and put the logical
> > replication into some kind of an error state that requires a resync.
> 
> That would still mean a need to resync the data to recover, so I'm not
> sure it's really an improvement. And I suppose it'd require tracking the
> dependencies, because how else would you mark the subscription as
> requiring a resync? At which point we could decline the DROP without a
> CASCADE, just like we do elsewhere, no?

I was actually thinking more along the lines of just simply marking the
publication/subscription as being in a 'failed' state when a failure
actually happens, and maybe even at that point basically throwing away
everything except the shell of the publication/subscription (so the user
can see that it failed and come in and properly drop it); I'm thinking
about this as perhaps similar to a transaction being aborted.

Thanks!

Stephen


signature.asc
Description: PGP signature

Re: row filtering for logical replication

[Petr Jelinek]

On 14/12/2018 16:38, Stephen Frost wrote:
> Greetings,
> 
> * Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
>> On 23/11/2018 03:02, Stephen Frost wrote:
>>> * Euler Taveira (eu...@timbira.com.br) wrote:
 2018-02-28 21:54 GMT-03:00 Craig Ringer :
> Good idea. I haven't read this yet, but one thing to make sure you've
> handled is limiting the clause to referencing only the current tuple and 
> the
> catalogs. user-catalog tables are OK, too, anything that is
> RelationIsAccessibleInLogicalDecoding().
>
> This means only immutable functions may be invoked, since a stable or
> volatile function might attempt to access a table. And views must be
> prohibited or recursively checked. (We have tree walkers that would help
> with this).
>
> It might be worth looking at the current logic for CHECK expressions, 
> since
> the requirements are similar. In my opinion you could safely not bother 
> with
> allowing access to user catalog tables in the filter expressions and limit
> them strictly to immutable functions and the tuple its self.

 IIRC implementation is similar to RLS expressions. I'll check all of
 these rules.
>>>
>>> Given the similarity to RLS and the nearby discussion about allowing
>>> non-superusers to create subscriptions, and probably publications later,
>>> I wonder if we shouldn't be somehow associating this with RLS policies
>>> instead of having the publication filtering be entirely independent..
>>
>> I do see the appeal here, if you consider logical replication to be a
>> streaming select it probably applies well.
>>
>> But given that this is happening inside output plugin which does not
>> have full executor setup and has catalog-only snapshot I am not sure how
>> feasible it is to try to merge these two things. As per my previous
>> email it's possible that we'll have to be stricter about what we allow
>> in expressions here.
> 
> I can certainly understand the concern about trying to combine the
> implementation of this with that of RLS; perhaps that isn't a good fit
> due to the additional constraints put on logical decoding.
> 
> That said, I still think it might make sense to consider these filters
> for logical decoding to be policies and, ideally, to allow users to use
> the same policy for both.
> 

I am not against that as long as it's possible to have policy for
logical replication without having it for RLS and vice versa.

I also wonder if policies are flexible enough to allow for specifying
OLD and NEW - the replication filtering deals with DML, not with what's
visible, it might very well depend on differences between these (that's
something the current patch is missing as well BTW).

> In the end, the idea of having to build a single large and complex
> 'create publication' command which has a bunch of tables, each with
> their own filter clauses, just strikes me as pretty painful.
> 
>> The other issue with merging this is that the use-case for filtering out
>> the data in logical replication is not necessarily about security, but
>> often about sending only relevant data. So it makes sense to have filter
>> on publication without RLS enabled on table and if we'd force that, we'd
>> limit usefulness of this feature.
> 
> I definitely have a serious problem if we are going to say that you
> can't use this filtering for security-sensitive cases.

I am saying it should not be tied to only security sensitive cases,
because it has use cases that have nothing to do with security (ie, I
don't want this to depend on RLS being enabled for a table).

> 
>> We definitely want to eventually create subscriptions as non-superuser
>> but that has zero effect on this as everything here is happening on
>> different server than where subscription lives (we already allow
>> creation of publications with just CREATE privilege on database and
>> ownership of the table).
> 
> What I wasn't clear about above was the idea that we might allow a user
> other than the table owner to publish a given table, but that such a
> publication should certanily only be allowed to include the rows which
> that user has access to- as regulated by RLS.  If the RLS policy is too
> complex to allow that then I would think we'd simply throw an error at
> the create publication time and the would-be publisher would need to
> figure that out with the table owner.

My opinion is that this is useful, but not necessarily something v1
patch needs to solve. Having too many publications and subscriptions to
various places is not currently practical anyway due to decoding
duplicating all the work for every connection.

> 
> * Euler Taveira (eu...@timbira.com.br) wrote:
>> Em sex, 23 de nov de 2018 às 11:40, Petr Jelinek
>>  escreveu:
> 
>>> The other issue with merging this is that the use-case for filtering out
>>> the data in logical replication is not necessarily about security, but
>>> often about sending only relevant data. So it makes 

Re: row filtering for logical replication

[Petr Jelinek]

On 14/12/2018 16:56, Stephen Frost wrote:
> Greetings,
> 
> * Tomas Vondra (tomas.von...@2ndquadrant.com) wrote:
>> On 11/23/18 8:03 PM, Stephen Frost wrote:
>>> * Fabrízio de Royes Mello (fabriziome...@gmail.com) wrote:
 On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek 
 wrote:
>> If carefully documented I see no problem with it... we already have an
>> analogous problem with functional indexes.
>
> The difference is that with functional indexes you can recreate the
> missing object and everything is okay again. With logical replication
> recreating the object will not help.
>

 In this case with logical replication you should rsync the object. That is
 the price of misunderstanding / bad use of the new feature.

 As usual, there are no free beer ;-)
>>>
>>> There's also certainly no shortage of other ways to break logical
>>> replication, including ways that would also be hard to recover from
>>> today other than doing a full resync.
>>
>> Sure, but that seems more like an argument against creating additional
>> ones (and for preventing those that already exist). I'm not sure this
>> particular feature is where we should draw the line, though.
> 
> I was actually going in the other direction- we should allow it because
> advanced users may know what they're doing better than we do and we
> shouldn't prevent things just because they might be misused or
> misunderstood by a user.
> 

That's all good, but we need good escape hatch for when things go south
and we don't have it and IMHO it's not as easy to have one as you might
think.

That's why I would do the simple and safe way first before allowing
more, otherwise we'll be discussing this for next couple of PG versions.

>>> What that seems to indicate, to me at least, is that it'd be awful
>>> nice to have a way to resync the data which doesn't necessairly
>>> involve transferring all of it over again.
>>>
>>> Of course, it'd be nice if we could track those dependencies too,
>>> but that's yet another thing.
>>
>> Yep, that seems like a good idea in general. Both here and for
>> functional indexes (although I suppose sure is a technical reason why it
>> wasn't implemented right away for them).
> 
> We don't track function dependencies in general and I could certainly
> see cases where you really wouldn't want to do so, at least not in the
> same way that we track FKs or similar.  I do wonder if maybe we didn't
> track function dependencies because we didn't (yet) have create or
> replace function and that now we should.  We don't track dependencies
> inside a function either though.

Yeah we can't always have dependencies, it would break some perfectly
valid usage scenarios. Also it's not exactly clear to me how we'd track
dependencies of say plpython function...

> 
>>> In short, I'm not sure that I agree with the idea that we shouldn't
>>> allow this and instead I'd rather we realize it and put the logical
>>> replication into some kind of an error state that requires a resync.
>>
>> That would still mean a need to resync the data to recover, so I'm not
>> sure it's really an improvement. And I suppose it'd require tracking the
>> dependencies, because how else would you mark the subscription as
>> requiring a resync? At which point we could decline the DROP without a
>> CASCADE, just like we do elsewhere, no?
> 
> I was actually thinking more along the lines of just simply marking the
> publication/subscription as being in a 'failed' state when a failure
> actually happens, and maybe even at that point basically throwing away
> everything except the shell of the publication/subscription (so the user
> can see that it failed and come in and properly drop it); I'm thinking
> about this as perhaps similar to a transaction being aborted.

There are several problems with that. First this happens in historic
snapshot which can't write and on top of that we are in the middle of
error processing so we have our hands tied a bit, it's definitely going
to need bit of creative thinking to do this.

Second, and that's more soft issue (which is probably harder to solve)
what do we do with the slot and subscription. There is one failed
publication, but the subscription may be subscribed to 20 of them, do we
kill the whole subscription because of single failed publication? If we
don't do we continue replicating like nothing has happened but with data
in the failed publication missing (which can be considered data
loss/corruption from the view of user). If we stop replication, do we
clean the slot so that we don't keep back wal/catalog xmin forever
(which could lead to server stopping) or do we keep the slot so that
user can somehow fix the issue (reconfigure subscription to not care
about that publication for example) and continue replication without
further loss?

-- 
  Petr Jelinek  http://www.2ndQuadrant.com/
  PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Stephen Frost]

Greetings,

* Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
> On 14/12/2018 16:38, Stephen Frost wrote:
> > * Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
> >> I do see the appeal here, if you consider logical replication to be a
> >> streaming select it probably applies well.
> >>
> >> But given that this is happening inside output plugin which does not
> >> have full executor setup and has catalog-only snapshot I am not sure how
> >> feasible it is to try to merge these two things. As per my previous
> >> email it's possible that we'll have to be stricter about what we allow
> >> in expressions here.
> > 
> > I can certainly understand the concern about trying to combine the
> > implementation of this with that of RLS; perhaps that isn't a good fit
> > due to the additional constraints put on logical decoding.
> > 
> > That said, I still think it might make sense to consider these filters
> > for logical decoding to be policies and, ideally, to allow users to use
> > the same policy for both.
> 
> I am not against that as long as it's possible to have policy for
> logical replication without having it for RLS and vice versa.

RLS already is able to be enabled/disabled on a per-table basis.  I
could see how we might want to extend the existing policy system to have
a way to enable/disable individual policies for RLS but that should be
reasonably straight-forward to do, I would think.

> I also wonder if policies are flexible enough to allow for specifying
> OLD and NEW - the replication filtering deals with DML, not with what's
> visible, it might very well depend on differences between these (that's
> something the current patch is missing as well BTW).

The policy system already has the notion of a 'visible' check and a
'does the new row match this' check (USING vs. WITH CHECK policies).
Perhaps if you could outline the specific use-cases that you're thinking
about, we could discuss them and make sure that they fit within those
mechanisms- or, if not, discuss if such a use-case would make sense for
RLS as well and, if so, figure out a way to support that for both.

> > In the end, the idea of having to build a single large and complex
> > 'create publication' command which has a bunch of tables, each with
> > their own filter clauses, just strikes me as pretty painful.
> > 
> >> The other issue with merging this is that the use-case for filtering out
> >> the data in logical replication is not necessarily about security, but
> >> often about sending only relevant data. So it makes sense to have filter
> >> on publication without RLS enabled on table and if we'd force that, we'd
> >> limit usefulness of this feature.
> > 
> > I definitely have a serious problem if we are going to say that you
> > can't use this filtering for security-sensitive cases.
> 
> I am saying it should not be tied to only security sensitive cases,
> because it has use cases that have nothing to do with security (ie, I
> don't want this to depend on RLS being enabled for a table).

I'm fine with this being able to be independently enabled/disabled,
apart from RLS.

> >> We definitely want to eventually create subscriptions as non-superuser
> >> but that has zero effect on this as everything here is happening on
> >> different server than where subscription lives (we already allow
> >> creation of publications with just CREATE privilege on database and
> >> ownership of the table).
> > 
> > What I wasn't clear about above was the idea that we might allow a user
> > other than the table owner to publish a given table, but that such a
> > publication should certanily only be allowed to include the rows which
> > that user has access to- as regulated by RLS.  If the RLS policy is too
> > complex to allow that then I would think we'd simply throw an error at
> > the create publication time and the would-be publisher would need to
> > figure that out with the table owner.
> 
> My opinion is that this is useful, but not necessarily something v1
> patch needs to solve. Having too many publications and subscriptions to
> various places is not currently practical anyway due to decoding
> duplicating all the work for every connection.

I agree that supporting this could be done in a later patch, however, I
do feel that when we go to add support for non-owners to create
publications then RLS needs to be supported at that point (and by more
than just 'throw an error').  I can agree with incremental improvements
but I don't want to get to a point where we've got a bunch of
independent things only half of which work with other parts of the
system.

> > * Euler Taveira (eu...@timbira.com.br) wrote:
> >> Em sex, 23 de nov de 2018 às 11:40, Petr Jelinek
> >>  escreveu:
> > 
> >>> The other issue with merging this is that the use-case for filtering out
> >>> the data in logical replication is not necessarily about security, but
> >>> often about sending only relevant data. So it makes sense to have filter
> >>> on publication without RLS 

Re: row filtering for logical replication

[Stephen Frost]

Greetings,

* Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
> On 14/12/2018 16:56, Stephen Frost wrote:
> > * Tomas Vondra (tomas.von...@2ndquadrant.com) wrote:
> >> On 11/23/18 8:03 PM, Stephen Frost wrote:
> >>> * Fabrízio de Royes Mello (fabriziome...@gmail.com) wrote:
>  On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek 
>  
>  wrote:
> >> If carefully documented I see no problem with it... we already have an
> >> analogous problem with functional indexes.
> >
> > The difference is that with functional indexes you can recreate the
> > missing object and everything is okay again. With logical replication
> > recreating the object will not help.
> >
> 
>  In this case with logical replication you should rsync the object. That 
>  is
>  the price of misunderstanding / bad use of the new feature.
> 
>  As usual, there are no free beer ;-)
> >>>
> >>> There's also certainly no shortage of other ways to break logical
> >>> replication, including ways that would also be hard to recover from
> >>> today other than doing a full resync.
> >>
> >> Sure, but that seems more like an argument against creating additional
> >> ones (and for preventing those that already exist). I'm not sure this
> >> particular feature is where we should draw the line, though.
> > 
> > I was actually going in the other direction- we should allow it because
> > advanced users may know what they're doing better than we do and we
> > shouldn't prevent things just because they might be misused or
> > misunderstood by a user.
> 
> That's all good, but we need good escape hatch for when things go south
> and we don't have it and IMHO it's not as easy to have one as you might
> think.

We don't have a great solution but we should be able to at least drop
and recreate the publication or subscription, even today, can't we?
Sure, that means having to recopy everything, but that's what you get if
you break your publication/subscription.  If we allow the user to get to
a point where the system can't be fixed then I agree that's a serious
issue, but hopefully that isn't the case.

> >>> What that seems to indicate, to me at least, is that it'd be awful
> >>> nice to have a way to resync the data which doesn't necessairly
> >>> involve transferring all of it over again.
> >>>
> >>> Of course, it'd be nice if we could track those dependencies too,
> >>> but that's yet another thing.
> >>
> >> Yep, that seems like a good idea in general. Both here and for
> >> functional indexes (although I suppose sure is a technical reason why it
> >> wasn't implemented right away for them).
> > 
> > We don't track function dependencies in general and I could certainly
> > see cases where you really wouldn't want to do so, at least not in the
> > same way that we track FKs or similar.  I do wonder if maybe we didn't
> > track function dependencies because we didn't (yet) have create or
> > replace function and that now we should.  We don't track dependencies
> > inside a function either though.
> 
> Yeah we can't always have dependencies, it would break some perfectly
> valid usage scenarios. Also it's not exactly clear to me how we'd track
> dependencies of say plpython function...

Well, we could at leasts depend on the functions explicitly listed at
the top level and I don't believe we even do that today.  I can't think
of any downside off-hand to that, given that we have create-or-replace
function.

> >>> In short, I'm not sure that I agree with the idea that we shouldn't
> >>> allow this and instead I'd rather we realize it and put the logical
> >>> replication into some kind of an error state that requires a resync.
> >>
> >> That would still mean a need to resync the data to recover, so I'm not
> >> sure it's really an improvement. And I suppose it'd require tracking the
> >> dependencies, because how else would you mark the subscription as
> >> requiring a resync? At which point we could decline the DROP without a
> >> CASCADE, just like we do elsewhere, no?
> > 
> > I was actually thinking more along the lines of just simply marking the
> > publication/subscription as being in a 'failed' state when a failure
> > actually happens, and maybe even at that point basically throwing away
> > everything except the shell of the publication/subscription (so the user
> > can see that it failed and come in and properly drop it); I'm thinking
> > about this as perhaps similar to a transaction being aborted.
> 
> There are several problems with that. First this happens in historic
> snapshot which can't write and on top of that we are in the middle of
> error processing so we have our hands tied a bit, it's definitely going
> to need bit of creative thinking to do this.

We can't write to things inside the database in a historic snapshot and
we do have to deal with the fact that we're in error processing.  What
about writing somewhere that's outside of the regular database system?
Maybe a pg_logical/failed

Re: row filtering for logical replication

[Petr Jelinek]

Hi,

On 27/12/2018 20:05, Stephen Frost wrote:
> Greetings,
> 
> * Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
>> On 14/12/2018 16:38, Stephen Frost wrote:
>>> * Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
 I do see the appeal here, if you consider logical replication to be a
 streaming select it probably applies well.

 But given that this is happening inside output plugin which does not
 have full executor setup and has catalog-only snapshot I am not sure how
 feasible it is to try to merge these two things. As per my previous
 email it's possible that we'll have to be stricter about what we allow
 in expressions here.
>>>
>>> I can certainly understand the concern about trying to combine the
>>> implementation of this with that of RLS; perhaps that isn't a good fit
>>> due to the additional constraints put on logical decoding.
>>>
>>> That said, I still think it might make sense to consider these filters
>>> for logical decoding to be policies and, ideally, to allow users to use
>>> the same policy for both.
>>
>> I am not against that as long as it's possible to have policy for
>> logical replication without having it for RLS and vice versa.
> 
> RLS already is able to be enabled/disabled on a per-table basis.  I
> could see how we might want to extend the existing policy system to have
> a way to enable/disable individual policies for RLS but that should be
> reasonably straight-forward to do, I would think.

Sure, I was mostly referring to having ability of enable/disable this
independently of enabling/disabling RLS which you are okay with based on
bellow so no issue there from my side.

> 
>> I also wonder if policies are flexible enough to allow for specifying
>> OLD and NEW - the replication filtering deals with DML, not with what's
>> visible, it might very well depend on differences between these (that's
>> something the current patch is missing as well BTW).
> 
> The policy system already has the notion of a 'visible' check and a
> 'does the new row match this' check (USING vs. WITH CHECK policies).
> Perhaps if you could outline the specific use-cases that you're thinking
> about, we could discuss them and make sure that they fit within those
> mechanisms- or, if not, discuss if such a use-case would make sense for
> RLS as well and, if so, figure out a way to support that for both.

So we'd use USING for old row images (UPDATE/DELETE) and WITH CHECK for
new ones (UPDATE/INSERT)? I think OLD/NEW is somewhat more natural
naming of this as there is no "SELECT" part of operation here, but as
long as the functionality is there I don't mind syntax that much.

> 
>>> In the end, the idea of having to build a single large and complex
>>> 'create publication' command which has a bunch of tables, each with
>>> their own filter clauses, just strikes me as pretty painful.
>>>
 The other issue with merging this is that the use-case for filtering out
 the data in logical replication is not necessarily about security, but
 often about sending only relevant data. So it makes sense to have filter
 on publication without RLS enabled on table and if we'd force that, we'd
 limit usefulness of this feature.
>>>
>>> I definitely have a serious problem if we are going to say that you
>>> can't use this filtering for security-sensitive cases.
>>
>> I am saying it should not be tied to only security sensitive cases,
>> because it has use cases that have nothing to do with security (ie, I
>> don't want this to depend on RLS being enabled for a table).
> 
> I'm fine with this being able to be independently enabled/disabled,
> apart from RLS.
> 

Cool.

 We definitely want to eventually create subscriptions as non-superuser
 but that has zero effect on this as everything here is happening on
 different server than where subscription lives (we already allow
 creation of publications with just CREATE privilege on database and
 ownership of the table).
>>>
>>> What I wasn't clear about above was the idea that we might allow a user
>>> other than the table owner to publish a given table, but that such a
>>> publication should certanily only be allowed to include the rows which
>>> that user has access to- as regulated by RLS.  If the RLS policy is too
>>> complex to allow that then I would think we'd simply throw an error at
>>> the create publication time and the would-be publisher would need to
>>> figure that out with the table owner.
>>
>> My opinion is that this is useful, but not necessarily something v1
>> patch needs to solve. Having too many publications and subscriptions to
>> various places is not currently practical anyway due to decoding
>> duplicating all the work for every connection.
> 
> I agree that supporting this could be done in a later patch, however, I
> do feel that when we go to add support for non-owners to create
> publications then RLS needs to be supported at that point (and by more
> than just 'throw an error'

Re: row filtering for logical replication

[Petr Jelinek]

On 27/12/2018 20:19, Stephen Frost wrote:
> Greetings,
> 
> * Petr Jelinek (petr.jeli...@2ndquadrant.com) wrote:
>> On 14/12/2018 16:56, Stephen Frost wrote:
>>> * Tomas Vondra (tomas.von...@2ndquadrant.com) wrote:
 On 11/23/18 8:03 PM, Stephen Frost wrote:
> * Fabrízio de Royes Mello (fabriziome...@gmail.com) wrote:
>> On Fri, Nov 23, 2018 at 4:13 PM Petr Jelinek 
>> 
>> wrote:
 If carefully documented I see no problem with it... we already have an
 analogous problem with functional indexes.
>>>
>>> The difference is that with functional indexes you can recreate the
>>> missing object and everything is okay again. With logical replication
>>> recreating the object will not help.
>>>
>>
>> In this case with logical replication you should rsync the object. That 
>> is
>> the price of misunderstanding / bad use of the new feature.
>>
>> As usual, there are no free beer ;-)
>
> There's also certainly no shortage of other ways to break logical
> replication, including ways that would also be hard to recover from
> today other than doing a full resync.

 Sure, but that seems more like an argument against creating additional
 ones (and for preventing those that already exist). I'm not sure this
 particular feature is where we should draw the line, though.
>>>
>>> I was actually going in the other direction- we should allow it because
>>> advanced users may know what they're doing better than we do and we
>>> shouldn't prevent things just because they might be misused or
>>> misunderstood by a user.
>>
>> That's all good, but we need good escape hatch for when things go south
>> and we don't have it and IMHO it's not as easy to have one as you might
>> think.
> 
> We don't have a great solution but we should be able to at least drop
> and recreate the publication or subscription, even today, can't we?

Well we can drop thing always, yes, not having ability to drop things
when they break would be bad design. I am debating ability to recover
without rebuilding everything a there are cases where you simply can't
rebuild everything (ie we allow filtering out deletes). I don't like
disabling UDFs either as that means that user created types are unusable
in filters, I just wonder if saying "sorry your replica is gone" is any
better.

> Sure, that means having to recopy everything, but that's what you get if
> you break your publication/subscription.

This is but off-topic here, but I really wonder how are you currently
breaking your publications/subscriptions.

> What that seems to indicate, to me at least, is that it'd be awful
> nice to have a way to resync the data which doesn't necessairly
> involve transferring all of it over again.
>
> Of course, it'd be nice if we could track those dependencies too,
> but that's yet another thing.

 Yep, that seems like a good idea in general. Both here and for
 functional indexes (although I suppose sure is a technical reason why it
 wasn't implemented right away for them).
>>>
>>> We don't track function dependencies in general and I could certainly
>>> see cases where you really wouldn't want to do so, at least not in the
>>> same way that we track FKs or similar.  I do wonder if maybe we didn't
>>> track function dependencies because we didn't (yet) have create or
>>> replace function and that now we should.  We don't track dependencies
>>> inside a function either though.
>>
>> Yeah we can't always have dependencies, it would break some perfectly
>> valid usage scenarios. Also it's not exactly clear to me how we'd track
>> dependencies of say plpython function...
> 
> Well, we could at leasts depend on the functions explicitly listed at
> the top level and I don't believe we even do that today.  I can't think
> of any downside off-hand to that, given that we have create-or-replace
> function.
> 

I dunno how much is that worth it TBH, the situations where I've seen
this issue (pglogical has this feature for long time and suffers from
the same lack of dependency tracking) is that somebody drops table/type
used in a function that is used as filter.

> In short, I'm not sure that I agree with the idea that we shouldn't
> allow this and instead I'd rather we realize it and put the logical
> replication into some kind of an error state that requires a resync.

 That would still mean a need to resync the data to recover, so I'm not
 sure it's really an improvement. And I suppose it'd require tracking the
 dependencies, because how else would you mark the subscription as
 requiring a resync? At which point we could decline the DROP without a
 CASCADE, just like we do elsewhere, no?
>>>
>>> I was actually thinking more along the lines of just simply marking the
>>> publication/subscription as being in a 'failed' state when a failure
>>> actually happens, and maybe even at that point basically thro

Re: row filtering for logical replication

[Michael Paquier]

On Thu, Mar 01, 2018 at 06:16:17PM -0500, David Steele wrote:
> That was the right thing to do, thank you!

This patch has been waiting on author for a couple of months and does
not apply anymore, so I am marking as returned with feedback.  If you
can rebase, please feel free to resubmit.
--
Michael


signature.asc
Description: PGP signature

Re: row filtering for logical replication

[Euler Taveira]

Em qua, 28 de fev de 2018 às 20:03, Euler Taveira
 escreveu:
> The attached patches add support for filtering rows in the publisher.
>
I rebased the patch. I added row filtering for initial
synchronization, pg_dump support and psql support. 0001 removes unused
code. 0002 reduces memory use. 0003 passes only structure member that
is used in create_estate_for_relation. 0004 reuses a parser node for
row filtering. 0005 is the feature. 0006 prints WHERE expression in
psql. 0007 adds pg_dump support. 0008 is only for debug purposes (I'm
not sure some of these messages will be part of the final patch).
0001, 0002, 0003 and 0008 are not mandatory for this feature.

Comments?


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento
From b2e56eaa9e16246c8158ff2961a6a4b2acbd096b Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 18:39:22 +
Subject: [PATCH 1/8] Remove unused atttypmod column from initial table
 synchronization

 Since commit 7c4f52409a8c7d85ed169bbbc1f6092274d03920, atttypmod was
 added but not used. The removal is safe because COPY from publisher
 does not need such information.
---
 src/backend/replication/logical/tablesync.c | 7 +++
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 6e420d8..f285813 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -660,7 +660,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	StringInfoData cmd;
 	TupleTableSlot *slot;
 	Oid			tableRow[2] = {OIDOID, CHAROID};
-	Oid			attrRow[4] = {TEXTOID, OIDOID, INT4OID, BOOLOID};
+	Oid			attrRow[3] = {TEXTOID, OIDOID, BOOLOID};
 	bool		isnull;
 	int			natt;
 
@@ -704,7 +704,6 @@ fetch_remote_table_info(char *nspname, char *relname,
 	appendStringInfo(&cmd,
 	 "SELECT a.attname,"
 	 "   a.atttypid,"
-	 "   a.atttypmod,"
 	 "   a.attnum = ANY(i.indkey)"
 	 "  FROM pg_catalog.pg_attribute a"
 	 "  LEFT JOIN pg_catalog.pg_index i"
@@ -714,7 +713,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	 "   AND a.attrelid = %u"
 	 " ORDER BY a.attnum",
 	 lrel->remoteid, lrel->remoteid);
-	res = walrcv_exec(wrconn, cmd.data, 4, attrRow);
+	res = walrcv_exec(wrconn, cmd.data, 3, attrRow);
 
 	if (res->status != WALRCV_OK_TUPLES)
 		ereport(ERROR,
@@ -735,7 +734,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 		Assert(!isnull);
 		lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 2, &isnull));
 		Assert(!isnull);
-		if (DatumGetBool(slot_getattr(slot, 4, &isnull)))
+		if (DatumGetBool(slot_getattr(slot, 3, &isnull)))
 			lrel->attkeys = bms_add_member(lrel->attkeys, natt);
 
 		/* Should never happen. */
-- 
2.7.4

From 797a0e8d858b490df7a9e1526f76e49fe1e10215 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 17:37:36 +
Subject: [PATCH 2/8] Store number of tuples in WalRcvExecResult

It seems to be a useful information while allocating memory for queries
that returns more than one row. It reduces memory allocation
for initial table synchronization.

While in it, since we have the number of columns, allocate only nfields
for cstrs instead of MaxTupleAttributeNumber.
---
 src/backend/replication/libpqwalreceiver/libpqwalreceiver.c | 9 ++---
 src/backend/replication/logical/tablesync.c | 5 ++---
 src/include/replication/walreceiver.h   | 1 +
 3 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 1e1695e..2533e3a 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -865,6 +865,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
  errdetail("Expected %d fields, got %d fields.",
 		   nRetTypes, nfields)));
 
+	walres->ntuples = PQntuples(pgres);
 	walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
 
 	/* Create tuple descriptor corresponding to expected result. */
@@ -875,7 +876,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
 	attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
 
 	/* No point in doing more here if there were no tuples returned. */
-	if (PQntuples(pgres) == 0)
+	if (walres->ntuples == 0)
 		return;
 
 	/* Create temporary context for local allocations. */
@@ -884,15 +885,17 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
 	   ALLOCSET_DEFAULT_SIZES);
 
 	/* Process returned rows. */
-	for (tupn = 0; tupn < PQntuples(pgres); tupn++)
+	for (tupn = 0; tupn < walres->ntuples; tupn++)
 	{
-		char	   *cstrs[MaxTupleAttributeNumber];
+		char	**cstrs;
 
 		CHECK_FOR_INTE

Re: row filtering for logical replication

[Erik Rijkers]

On 2018-11-01 01:29, Euler Taveira wrote:

Em qua, 28 de fev de 2018 às 20:03, Euler Taveira
 escreveu:

The attached patches add support for filtering rows in the publisher.



I ran pgbench-over-logical-replication with a WHERE-clause and could not 
get this to do a correct replication.  Below is the output of the 
attached test program.



$ ./logrep_rowfilter.sh
--
/home/aardvark/pg_stuff/pg_installations/pgsql.logrep_rowfilter/bin.fast/initdb 
--pgdata=/tmp/cascade/instance1/data --encoding=UTF8 --pwfile=/tmp/bugs

--
/home/aardvark/pg_stuff/pg_installations/pgsql.logrep_rowfilter/bin.fast/initdb 
--pgdata=/tmp/cascade/instance2/data --encoding=UTF8 --pwfile=/tmp/bugs

--
/home/aardvark/pg_stuff/pg_installations/pgsql.logrep_rowfilter/bin.fast/initdb 
--pgdata=/tmp/cascade/instance3/data --encoding=UTF8 --pwfile=/tmp/bugs

sleep 3s
dropping old tables...
creating tables...
generating data...
10 of 10 tuples (100%) done (elapsed 0.09 s, remaining 0.00 s)
vacuuming...
creating primary keys...
done.
create publication pub_6515_to_6516;
alter publication pub_6515_to_6516 add table pgbench_accounts where (aid 
between 4 and 6-1) ; --> where 1

alter publication pub_6515_to_6516 add table pgbench_branches;
alter publication pub_6515_to_6516 add table pgbench_tellers;
alter publication pub_6515_to_6516 add table pgbench_history;
create publication pub_6516_to_6517;
alter publication pub_6516_to_6517 add table pgbench_accounts ; -- where 
(aid between 4 and 6-1) ; --> where 2

alter publication pub_6516_to_6517 add table pgbench_branches;
alter publication pub_6516_to_6517 add table pgbench_tellers;
alter publication pub_6516_to_6517 add table pgbench_history;

create subscription pub_6516_from_6515 connection 'port=6515 
application_name=rowfilter'

   publication pub_6515_to_6516 with(enabled=false);
alter subscription pub_6516_from_6515 enable;
create subscription pub_6517_from_6516 connection 'port=6516 
application_name=rowfilter'

   publication pub_6516_to_6517 with(enabled=false);
alter subscription pub_6517_from_6516 enable;
-- pgbench -p 6515 -c 16 -j 8 -T 5 -n postgres#  scale 1
transaction type: 
scaling factor: 1
query mode: simple
number of clients: 16
number of threads: 8
duration: 5 s
number of transactions actually processed: 80
latency average = 1178.106 ms
tps = 13.581120 (including connections establishing)
tps = 13.597443 (excluding connections establishing)

   accounts  branches   tellers   history
   - - - -
6515   6546b1f0f 2d328ed28 7406473b0 7c1351523e8c07347b
6516   6546b1f0f 2d328ed28 d41d8cd98 d41d8cd98e7235f541
6517   f7c0791c8 d9c63e471 d41d8cd98 d41d8cd9830892eea1   NOK

6515   6546b1f0f 2d328ed28 7406473b0 7c1351523e8c07347b
6516   6546b1f0f 2d328ed28 7406473b0 5a54cf7c5191ae1af3
6517   6546b1f0f 2d328ed28 7406473b0 5a54cf7c5191ae1af3   NOK

6515   6546b1f0f 2d328ed28 7406473b0 7c1351523e8c07347b
6516   6546b1f0f 2d328ed28 7406473b0 5a54cf7c5191ae1af3
6517   6546b1f0f 2d328ed28 7406473b0 5a54cf7c5191ae1af3   NOK

[...]

I let that run for 10 minutes or so but that pgbench_history table 
md5-values (of ports 6516 and 6517) do not change anymore, which shows 
that it is and remains different from the original pgbench_history table 
in 6515.



When there is a where-clause this goes *always* wrong.

Without a where-clause all logical replication tests were OK.  Perhaps 
the error is not in our patch but something in logical replication.


Attached is the test program (will need some tweaking of PATHs, 
PG-variables (PGPASSFILE) etc).  This is the same program I used in 
march when you first posted a version of this patch alhough the error is 
different.



thanks,


Erik Rijkers





#!/bin/bash

# postgres binary compiled with 
#
# 20181101
#   0001-Remove-unused-atttypmod-column-from-initial-table-sy.patch
#   0002-Store-number-of-tuples-in-WalRcvExecResult.patch
#   0003-Refactor-function-create_estate_for_relation.patch
#   0004-Rename-a-WHERE-node.patch
#   0005-Row-filtering-for-logical-replication.patch
#   0006-Print-publication-WHERE-condition-in-psql.patch
#   0007-Publication-where-condition-support-for-pg_dump.patch
#   0008-Debug-for-row-filtering.patch
#

unset PGDATABASE PGPORT PGSERVICE
export PGDATABASE=postgres

scale=1

root_dir=/tmp/cascade

BIN=$HOME/pg_stuff/pg_installations/pgsql.logrep_rowfilter/bin.fast

export PATH=$BIN:$PATH

  initdb=$BIN/initdb
postgres=$BIN/postgres
  pg_ctl=$BIN/pg_ctl
baseport=6515
 appname=rowfilter

if [[ -d $root_dir/instance1 ]]; then rm -rf $root_dir/instance1; fi
if [[ -d $root_dir/instance2 ]]; then rm -rf $root_dir/instance2; fi
if [[ -d $root_dir/instance3 ]]; then rm -rf $root_dir/instance3; fi
if [[ -d $root_dir/instance1 ]]; then exit ; fi
if [[ -d $root_dir/instance2 ]]; then exit ; fi
if [[ -d $root_dir/instance3 ]]; then exit ; fi

devel_file=/tmp/bugs
echo filterbug>$devel_file

num_instances=3

for n in

Re: row filtering for logical replication

[Erik Rijkers]

On 2018-11-01 08:56, Erik Rijkers wrote:

On 2018-11-01 01:29, Euler Taveira wrote:

Em qua, 28 de fev de 2018 às 20:03, Euler Taveira
 escreveu:

The attached patches add support for filtering rows in the publisher.



I ran pgbench-over-logical-replication with a WHERE-clause and could
not get this to do a correct replication.  Below is the output of the
attached test program.


$ ./logrep_rowfilter.sh


I have noticed that the failure to replicate correctly can be avoided by 
putting a wait state of (on my machine) at least 3 seconds between the 
setting up of the subscription and the start of pgbench.  See the bash 
program I attached in my previous mail.  The bug can be avoided by a 
'sleep 5' just before the start of the actual pgbench run.


So it seems this bug is due to some timing error in your patch (or 
possibly in logical replication itself).



Erik Rijkers

Re: row filtering for logical replication

[Euler Taveira]

Em qui, 1 de nov de 2018 às 05:30, Erik Rijkers  escreveu:
> > I ran pgbench-over-logical-replication with a WHERE-clause and could
> > not get this to do a correct replication.  Below is the output of the
> > attached test program.
> >
> >
> > $ ./logrep_rowfilter.sh
>
Erik, thanks for testing.

> So it seems this bug is due to some timing error in your patch (or
> possibly in logical replication itself).
>
It is a bug in the new synchronization code. I'm doing some code
cleanup/review and will post a new patchset after I finish it. If you
want to give it a try again, apply the following patch.

diff --git a/src/backend/replication/logical/tablesync.c
b/src/backend/replication/logical/tablesync.c
index e0eb73c..4797e0b 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -757,7 +757,7 @@ fetch_remote_table_info(char *nspname, char *relname,

/* Fetch row filtering info */
resetStringInfo(&cmd);
-   appendStringInfo(&cmd, "SELECT pg_get_expr(prrowfilter,
prrelid) FROM pg_publication p INNER JOIN pg_publication_rel pr ON
(p.oid = pr.prpubid) WHERE pr.prrelid = %u AND p.pubname IN (",
MyLogicalRepWorker->relid);
+   appendStringInfo(&cmd, "SELECT pg_get_expr(prrowfilter,
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);


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Erik Rijkers]

On 2018-11-02 02:59, Euler Taveira wrote:
Em qui, 1 de nov de 2018 às 05:30, Erik Rijkers  
escreveu:

> I ran pgbench-over-logical-replication with a WHERE-clause and could
> not get this to do a correct replication.  Below is the output of the
> attached test program.
>
>
> $ ./logrep_rowfilter.sh


Erik, thanks for testing.


So it seems this bug is due to some timing error in your patch (or
possibly in logical replication itself).


It is a bug in the new synchronization code. I'm doing some code
cleanup/review and will post a new patchset after I finish it. If you
want to give it a try again, apply the following patch.

diff --git a/src/backend/replication/logical/tablesync.c
b/src/backend/replication/logical/tablesync.c
index e0eb73c..4797e0b 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
[...]



That does indeed fix it.

Thank you,

Erik Rijkers

Re: row filtering for logical replication

[Hironobu SUZUKI]

On 2018/11/01 0:29, Euler Taveira wrote:

Em qua, 28 de fev de 2018 às 20:03, Euler Taveira
 escreveu:

The attached patches add support for filtering rows in the publisher.


I rebased the patch. I added row filtering for initial
synchronization, pg_dump support and psql support. 0001 removes unused
code. 0002 reduces memory use. 0003 passes only structure member that
is used in create_estate_for_relation. 0004 reuses a parser node for
row filtering. 0005 is the feature. 0006 prints WHERE expression in
psql. 0007 adds pg_dump support. 0008 is only for debug purposes (I'm
not sure some of these messages will be part of the final patch).
0001, 0002, 0003 and 0008 are not mandatory for this feature.

Comments?




Hi,

I reviewed your patches and I found a bug when I tested ALTER 
PUBLICATION statement.


In short, ALTER PUBLICATION SET with a WHERE clause does not applied new 
WHERE clause.


I describe the outline of the test I did and my conclusion.

[TEST]
I show the test case I tried in below.

(1)Publisher and Subscriber

I executed each statement on the publisher and the subscriber.

```
testdb=# CREATE PUBLICATION pub_testdb_t FOR TABLE t WHERE (id > 10);
CREATE PUBLICATION
```

```
testdb=# CREATE SUBSCRIPTION sub_testdb_t CONNECTION 'dbname=testdb 
port=5432 user=postgres' PUBLICATION pub_testdb_t;

NOTICE:  created replication slot "sub_testdb_t" on publisher
CREATE SUBSCRIPTION
```

(2)Publisher

I executed these statements shown below.

testdb=# INSERT INTO t VALUES (1,1);
INSERT 0 1
testdb=# INSERT INTO t VALUES (11,11);
INSERT 0 1

(3)Subscriber

I confirmed that the CREATE PUBLICATION statement worked well.

```
testdb=# SELECT * FROM t;
 id | data
+--
 11 |   11
(1 row)
```

(4)Publisher
After that, I executed ALTER PUBLICATION with a WHERE clause and 
inserted a new row.


```
testdb=# ALTER  PUBLICATION pub_testdb_t SET TABLE t WHERE (id > 5);
ALTER PUBLICATION

testdb=# INSERT INTO t VALUES (7,7);
INSERT 0 1

testdb=# SELECT * FROM t;
 id | data
+--
  1 |1
 11 |   11
  7 |7
(3 rows)
```

(5)Subscriber
I confirmed that the change of WHERE clause set by ALTER PUBLICATION 
statement was ignored.


```
testdb=# SELECT * FROM t;
 id | data
+--
 11 |   11
(1 row)
```

[Conclusion]
I think AlterPublicationTables()@publicationcmds.c has a bug.

In the foreach(oldlc, oldrelids) loop, oldrel must be appended to 
delrels if oldrel or newrel has a WHERE clause. However, the current 
implementation does not, therefore, old WHERE clause is not deleted and 
the new WHERE clause is ignored.


This is my speculation. It may not be correct, but , at least, it is a 
fact that ALTER PUBLICATION with a WHERE clause is not functioned in my 
environment and my operation described in above.



Best regards,

Re: row filtering for logical replication

[a . kondratov]

Hi Euler,

On 2019-02-03 13:14, Andres Freund wrote:


On 2018-11-23 13:15:08 -0300, Euler Taveira wrote:

Besides the problem presented by Hironobu-san, I'm doing some cleanup
and improving docs. I also forget to declare pg_publication_rel TOAST
table.

Thanks for your review.


As far as I can tell, the patch has not been refreshed since. So I'm
marking this as returned with feedback for now. Please resubmit once
ready.



Do you have any plans for continuing working on this patch and 
submitting it again on the closest September commitfest? There are only 
a few days left. Anyway, I will be glad to review the patch if you do 
submit it, though I didn't yet dig deeply into the code.


I've rebased recently the entire patch set (attached) and it works fine. 
Your tap test is passed. Also I've added a new test case (see 0009 
attached) with real life example of bidirectional replication (BDR) 
utilising this new WHERE clause. This naive BDR is implemented using 
is_cloud flag, which is set to TRUE/FALSE on cloud/remote nodes 
respectively.


Although almost all new tests are passed, there is a problem with DELETE 
replication, so 1 out of 10 tests is failed. It isn't replicated if the 
record was created with is_cloud=TRUE on cloud, replicated to remote; 
then updated with is_cloud=FALSE on remote, replicated to cloud; then 
deleted on remote.



Regards
--
Alexey Kondratov
Postgres Professional https://www.postgrespro.com
Russian Postgres CompanyFrom ae80e1616fb6374968a09e3c44f0abe59ebf3a87 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 18:39:22 +
Subject: [PATCH v2 1/9] Remove unused atttypmod column from initial table
 synchronization

 Since commit 7c4f52409a8c7d85ed169bbbc1f6092274d03920, atttypmod was
 added but not used. The removal is safe because COPY from publisher
 does not need such information.
---
 src/backend/replication/logical/tablesync.c | 7 +++
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 7881079e96..0a565dd837 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -647,7 +647,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	StringInfoData cmd;
 	TupleTableSlot *slot;
 	Oid			tableRow[2] = {OIDOID, CHAROID};
-	Oid			attrRow[4] = {TEXTOID, OIDOID, INT4OID, BOOLOID};
+	Oid			attrRow[3] = {TEXTOID, OIDOID, BOOLOID};
 	bool		isnull;
 	int			natt;
 
@@ -691,7 +691,6 @@ fetch_remote_table_info(char *nspname, char *relname,
 	appendStringInfo(&cmd,
 	 "SELECT a.attname,"
 	 "   a.atttypid,"
-	 "   a.atttypmod,"
 	 "   a.attnum = ANY(i.indkey)"
 	 "  FROM pg_catalog.pg_attribute a"
 	 "  LEFT JOIN pg_catalog.pg_index i"
@@ -703,7 +702,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	 lrel->remoteid,
 	 (walrcv_server_version(wrconn) >= 12 ? "AND a.attgenerated = ''" : ""),
 	 lrel->remoteid);
-	res = walrcv_exec(wrconn, cmd.data, 4, attrRow);
+	res = walrcv_exec(wrconn, cmd.data, 3, attrRow);
 
 	if (res->status != WALRCV_OK_TUPLES)
 		ereport(ERROR,
@@ -724,7 +723,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 		Assert(!isnull);
 		lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 2, &isnull));
 		Assert(!isnull);
-		if (DatumGetBool(slot_getattr(slot, 4, &isnull)))
+		if (DatumGetBool(slot_getattr(slot, 3, &isnull)))
 			lrel->attkeys = bms_add_member(lrel->attkeys, natt);
 
 		/* Should never happen. */

base-commit: 9acda731184c1ebdf99172cbb19d0404b7eebc37
-- 
2.19.1

From 362f2cc97745690ff4739b530f5ba95aea59be09 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 17:37:36 +
Subject: [PATCH v2 2/9] Store number of tuples in WalRcvExecResult

It seems to be a useful information while allocating memory for queries
that returns more than one row. It reduces memory allocation
for initial table synchronization.

While in it, since we have the number of columns, allocate only nfields
for cstrs instead of MaxTupleAttributeNumber.
---
 .../replication/libpqwalreceiver/libpqwalreceiver.c  | 9 ++---
 src/backend/replication/logical/tablesync.c  | 5 ++---
 src/include/replication/walreceiver.h| 1 +
 3 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 6eba08a920..846b6f89f1 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -878,6 +878,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
  errdetail("Expected %d fields, got %d fields.",
 		   nRetTypes, nfields)));
 
+	walres->ntuples = PQntuples(pgres);
 	walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
 
 	/* Create tuple descriptor 

Re: row filtering for logical replication

[Euler Taveira]

Em dom, 3 de fev de 2019 às 07:14, Andres Freund  escreveu:
>
> As far as I can tell, the patch has not been refreshed since. So I'm
> marking this as returned with feedback for now. Please resubmit once
> ready.
>
I fix all of the bugs pointed in this thread. I decide to disallow
UDFs in filters (it is safer for a first version). We can add this
functionality later. However, I'll check if allow "safe" functions
(aka builtin functions) are ok. I add more docs explaining that
expressions are executed with the role used for replication connection
and also that columns used in expressions must be part of PK or
REPLICA IDENTITY. I add regression tests.

Comments?



--
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento
From 87945236590e9fd37b203d325b74dc5baccee64d Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 18:39:22 +
Subject: [PATCH 1/8] Remove unused atttypmod column from initial table
 synchronization

 Since commit 7c4f52409a8c7d85ed169bbbc1f6092274d03920, atttypmod was
 added but not used. The removal is safe because COPY from publisher
 does not need such information.
---
 src/backend/replication/logical/tablesync.c | 7 +++
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 7881079e96..0a565dd837 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -647,7 +647,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	StringInfoData cmd;
 	TupleTableSlot *slot;
 	Oid			tableRow[2] = {OIDOID, CHAROID};
-	Oid			attrRow[4] = {TEXTOID, OIDOID, INT4OID, BOOLOID};
+	Oid			attrRow[3] = {TEXTOID, OIDOID, BOOLOID};
 	bool		isnull;
 	int			natt;
 
@@ -691,7 +691,6 @@ fetch_remote_table_info(char *nspname, char *relname,
 	appendStringInfo(&cmd,
 	 "SELECT a.attname,"
 	 "   a.atttypid,"
-	 "   a.atttypmod,"
 	 "   a.attnum = ANY(i.indkey)"
 	 "  FROM pg_catalog.pg_attribute a"
 	 "  LEFT JOIN pg_catalog.pg_index i"
@@ -703,7 +702,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	 lrel->remoteid,
 	 (walrcv_server_version(wrconn) >= 12 ? "AND a.attgenerated = ''" : ""),
 	 lrel->remoteid);
-	res = walrcv_exec(wrconn, cmd.data, 4, attrRow);
+	res = walrcv_exec(wrconn, cmd.data, 3, attrRow);
 
 	if (res->status != WALRCV_OK_TUPLES)
 		ereport(ERROR,
@@ -724,7 +723,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 		Assert(!isnull);
 		lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 2, &isnull));
 		Assert(!isnull);
-		if (DatumGetBool(slot_getattr(slot, 4, &isnull)))
+		if (DatumGetBool(slot_getattr(slot, 3, &isnull)))
 			lrel->attkeys = bms_add_member(lrel->attkeys, natt);
 
 		/* Should never happen. */
-- 
2.11.0

From 3a5b4c541982357c2231b9882ac01f1f0d0a8e29 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Tue, 27 Feb 2018 02:21:03 +
Subject: [PATCH 3/8] Refactor function create_estate_for_relation

Relation localrel is the only LogicalRepRelMapEntry structure member
that is useful for create_estate_for_relation.
---
 src/backend/replication/logical/worker.c | 14 +++---
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 43edfef089..31fc7c5048 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -173,7 +173,7 @@ ensure_transaction(void)
  * This is based on similar code in copy.c
  */
 static EState *
-create_estate_for_relation(LogicalRepRelMapEntry *rel)
+create_estate_for_relation(Relation rel)
 {
 	EState	   *estate;
 	ResultRelInfo *resultRelInfo;
@@ -183,13 +183,13 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel)
 
 	rte = makeNode(RangeTblEntry);
 	rte->rtekind = RTE_RELATION;
-	rte->relid = RelationGetRelid(rel->localrel);
-	rte->relkind = rel->localrel->rd_rel->relkind;
+	rte->relid = RelationGetRelid(rel);
+	rte->relkind = rel->rd_rel->relkind;
 	rte->rellockmode = AccessShareLock;
 	ExecInitRangeTable(estate, list_make1(rte));
 
 	resultRelInfo = makeNode(ResultRelInfo);
-	InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0);
+	InitResultRelInfo(resultRelInfo, rel, 1, NULL, 0);
 
 	estate->es_result_relations = resultRelInfo;
 	estate->es_num_result_relations = 1;
@@ -589,7 +589,7 @@ apply_handle_insert(StringInfo s)
 	}
 
 	/* Initialize the executor state. */
-	estate = create_estate_for_relation(rel);
+	estate = create_estate_for_relation(rel->localrel);
 	remoteslot = ExecInitExtraTupleSlot(estate,
 		RelationGetDescr(rel->localrel),
 		&TTSOpsVirtual);
@@ -696,7 +696,7 @@ apply_handle_update(StringInfo s)
 	check_relation_updatable(rel);
 
 	/* Initialize the executor state. */
-	estate = create_estate_for_relation(re

Re: row filtering for logical replication

[Euler Taveira]

Em ter, 27 de ago de 2019 às 18:10,  escreveu:
>
> Do you have any plans for continuing working on this patch and
> submitting it again on the closest September commitfest? There are only
> a few days left. Anyway, I will be glad to review the patch if you do
> submit it, though I didn't yet dig deeply into the code.
>
Sure. See my last email to this thread. I appreciate if you can review it.

> Although almost all new tests are passed, there is a problem with DELETE
> replication, so 1 out of 10 tests is failed. It isn't replicated if the
> record was created with is_cloud=TRUE on cloud, replicated to remote;
> then updated with is_cloud=FALSE on remote, replicated to cloud; then
> deleted on remote.
>
That's because you don't include is_cloud in PK or REPLICA IDENTITY. I
add a small note in docs.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Alexey Zagarin]

I think that I also have found one shortcoming when using the setup described 
by Alexey Kondratov. The problem that I face is that if both (cloud and remote) 
tables already have data the moment I add the subscription, then the whole 
table is copied in both directions initially. Which leads to duplicated data 
and broken replication because COPY doesn't take into account the filtering 
condition. In case there are filters in a publication, the COPY command that is 
executed when adding a subscription (or altering one to refresh a publication) 
should also filter the data based on the same condition, e.g. COPY (SELECT * 
FROM ... WHERE ...) TO ...

The current workaround is to always use WITH copy_data = false when subscribing 
or refreshing, and then manually copy data with the above statement.

Alexey Zagarin
On 1 Sep 2019 12:11 +0700, Euler Taveira , wrote:
> Em ter, 27 de ago de 2019 às 18:10,  escreveu:
> >
> > Do you have any plans for continuing working on this patch and
> > submitting it again on the closest September commitfest? There are only
> > a few days left. Anyway, I will be glad to review the patch if you do
> > submit it, though I didn't yet dig deeply into the code.
> >
> Sure. See my last email to this thread. I appreciate if you can review it.
>
> > Although almost all new tests are passed, there is a problem with DELETE
> > replication, so 1 out of 10 tests is failed. It isn't replicated if the
> > record was created with is_cloud=TRUE on cloud, replicated to remote;
> > then updated with is_cloud=FALSE on remote, replicated to cloud; then
> > deleted on remote.
> >
> That's because you don't include is_cloud in PK or REPLICA IDENTITY. I
> add a small note in docs.
>
>
> --
> Euler Taveira Timbira -
> http://www.timbira.com.br/
> PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento
>
>
>
>

Re: row filtering for logical replication

[Erik Rijkers]

On 2019-09-01 02:28, Euler Taveira wrote:
Em dom, 3 de fev de 2019 às 07:14, Andres Freund  
escreveu:


As far as I can tell, the patch has not been refreshed since. So I'm
marking this as returned with feedback for now. Please resubmit once
ready.


I fix all of the bugs pointed in this thread. I decide to disallow



0001-Remove-unused-atttypmod-column-from-initial-table-sy.patch
0002-Store-number-of-tuples-in-WalRcvExecResult.patch
0003-Refactor-function-create_estate_for_relation.patch
0004-Rename-a-WHERE-node.patch
0005-Row-filtering-for-logical-replication.patch
0006-Print-publication-WHERE-condition-in-psql.patch
0007-Publication-where-condition-support-for-pg_dump.patch
0008-Debug-for-row-filtering.patch


Hi,

The first 4 of these apply without error, but I can't get 0005 to apply. 
This is what I use:


patch --dry-run -b -l -F 5 -p 1 < 
/home/aardvark/download/pgpatches/0130/logrep_rowfilter/20190901/0005-Row-filtering-for-logical-replication.patch



checking file doc/src/sgml/catalogs.sgml
Hunk #1 succeeded at 5595 (offset 8 lines).
checking file doc/src/sgml/ref/alter_publication.sgml
checking file doc/src/sgml/ref/create_publication.sgml
checking file src/backend/catalog/pg_publication.c
checking file src/backend/commands/publicationcmds.c
Hunk #1 succeeded at 352 (offset 8 lines).
Hunk #2 succeeded at 381 (offset 8 lines).
Hunk #3 succeeded at 539 (offset 8 lines).
Hunk #4 succeeded at 570 (offset 8 lines).
Hunk #5 succeeded at 601 (offset 8 lines).
Hunk #6 succeeded at 626 (offset 8 lines).
Hunk #7 succeeded at 647 (offset 8 lines).
Hunk #8 succeeded at 679 (offset 8 lines).
Hunk #9 succeeded at 693 (offset 8 lines).
checking file src/backend/parser/gram.y
checking file src/backend/parser/parse_agg.c
checking file src/backend/parser/parse_expr.c
Hunk #4 succeeded at 3571 (offset -2 lines).
checking file src/backend/parser/parse_func.c
Hunk #1 succeeded at 2516 (offset -13 lines).
checking file src/backend/replication/logical/tablesync.c
checking file src/backend/replication/logical/worker.c
checking file src/backend/replication/pgoutput/pgoutput.c
Hunk #1 FAILED at 12.
Hunk #2 succeeded at 60 (offset 2 lines).
Hunk #3 succeeded at 336 (offset 2 lines).
Hunk #4 succeeded at 630 (offset 2 lines).
Hunk #5 succeeded at 647 (offset 2 lines).
Hunk #6 succeeded at 738 (offset 2 lines).
1 out of 6 hunks FAILED
checking file src/include/catalog/pg_publication.h
checking file src/include/catalog/pg_publication_rel.h
checking file src/include/catalog/toasting.h
checking file src/include/nodes/nodes.h
checking file src/include/nodes/parsenodes.h
Hunk #1 succeeded at 3461 (offset -1 lines).
Hunk #2 succeeded at 3486 (offset -1 lines).
checking file src/include/parser/parse_node.h
checking file src/include/replication/logicalrelation.h
checking file src/test/regress/expected/publication.out
Hunk #1 succeeded at 116 (offset 9 lines).
checking file src/test/regress/sql/publication.sql
Hunk #1 succeeded at 69 with fuzz 1 (offset 9 lines).
checking file src/test/subscription/t/013_row_filter.pl


perhaps that can be fixed?

thanks,

Erik Rijkers

Re: row filtering for logical replication

[Euler Taveira]

Em dom, 1 de set de 2019 às 06:09, Erik Rijkers  escreveu:
>
> The first 4 of these apply without error, but I can't get 0005 to apply.
> This is what I use:
>
Erik, I generate a new patch set with patience diff algorithm. It
seems it applies cleanly.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento
From c07af2f00b7a72ba9660e389bb1392fc9e5d2688 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Wed, 24 Jan 2018 17:01:31 -0200
Subject: [PATCH 4/8] Rename a WHERE node

A WHERE clause will be used for row filtering in logical replication. We
already have a similar node: 'WHERE (condition here)'. Let's rename the
node to a generic name and use it for row filtering too.
---
 src/backend/parser/gram.y | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index c97bb367f8..1de8f56794 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -476,7 +476,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type 	def_arg columnElem where_clause where_or_current_clause
 a_expr b_expr c_expr AexprConst indirection_el opt_slice_bound
 columnref in_expr having_clause func_table xmltable array_expr
-ExclusionWhereClause operator_def_arg
+OptWhereClause operator_def_arg
 %type 	rowsfrom_item rowsfrom_list opt_col_def_list
 %type  opt_ordinality
 %type 	ExclusionConstraintList ExclusionConstraintElem
@@ -3710,7 +3710,7 @@ ConstraintElem:
 	$$ = (Node *)n;
 }
 			| EXCLUDE access_method_clause '(' ExclusionConstraintList ')'
-opt_c_include opt_definition OptConsTableSpace  ExclusionWhereClause
+opt_c_include opt_definition OptConsTableSpace  OptWhereClause
 ConstraintAttributeSpec
 {
 	Constraint *n = makeNode(Constraint);
@@ -3812,7 +3812,7 @@ ExclusionConstraintElem: index_elem WITH any_operator
 			}
 		;
 
-ExclusionWhereClause:
+OptWhereClause:
 			WHERE '(' a_expr ')'	{ $$ = $3; }
 			| /*EMPTY*/{ $$ = NULL; }
 		;
-- 
2.11.0

From 367631ac4ba1e41170d59d39693e2eaf7c406621 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Tue, 27 Feb 2018 02:21:03 +
Subject: [PATCH 3/8] Refactor function create_estate_for_relation

Relation localrel is the only LogicalRepRelMapEntry structure member
that is useful for create_estate_for_relation.
---
 src/backend/replication/logical/worker.c | 14 +++---
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 11e6331f49..d9952c8b7e 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -173,7 +173,7 @@ ensure_transaction(void)
  * This is based on similar code in copy.c
  */
 static EState *
-create_estate_for_relation(LogicalRepRelMapEntry *rel)
+create_estate_for_relation(Relation rel)
 {
 	EState	   *estate;
 	ResultRelInfo *resultRelInfo;
@@ -183,13 +183,13 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel)
 
 	rte = makeNode(RangeTblEntry);
 	rte->rtekind = RTE_RELATION;
-	rte->relid = RelationGetRelid(rel->localrel);
-	rte->relkind = rel->localrel->rd_rel->relkind;
+	rte->relid = RelationGetRelid(rel);
+	rte->relkind = rel->rd_rel->relkind;
 	rte->rellockmode = AccessShareLock;
 	ExecInitRangeTable(estate, list_make1(rte));
 
 	resultRelInfo = makeNode(ResultRelInfo);
-	InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0);
+	InitResultRelInfo(resultRelInfo, rel, 1, NULL, 0);
 
 	estate->es_result_relations = resultRelInfo;
 	estate->es_num_result_relations = 1;
@@ -589,7 +589,7 @@ apply_handle_insert(StringInfo s)
 	}
 
 	/* Initialize the executor state. */
-	estate = create_estate_for_relation(rel);
+	estate = create_estate_for_relation(rel->localrel);
 	remoteslot = ExecInitExtraTupleSlot(estate,
 		RelationGetDescr(rel->localrel),
 		&TTSOpsVirtual);
@@ -696,7 +696,7 @@ apply_handle_update(StringInfo s)
 	check_relation_updatable(rel);
 
 	/* Initialize the executor state. */
-	estate = create_estate_for_relation(rel);
+	estate = create_estate_for_relation(rel->localrel);
 	remoteslot = ExecInitExtraTupleSlot(estate,
 		RelationGetDescr(rel->localrel),
 		&TTSOpsVirtual);
@@ -815,7 +815,7 @@ apply_handle_delete(StringInfo s)
 	check_relation_updatable(rel);
 
 	/* Initialize the executor state. */
-	estate = create_estate_for_relation(rel);
+	estate = create_estate_for_relation(rel->localrel);
 	remoteslot = ExecInitExtraTupleSlot(estate,
 		RelationGetDescr(rel->localrel),
 		&TTSOpsVirtual);
-- 
2.11.0

From e83de1cab559f4ca8f9a75356e220356814cd243 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 18:39:22 +
Subject: [PATCH 1/8] Remove unused atttypmod column from initial table
 synchronization

 Since commit 7c4f52409a8c7d85ed169bbbc1f6092274

Re: row filtering for logical replication

[Erik Rijkers]

On 2019-09-02 01:43, Euler Taveira wrote:
Em dom, 1 de set de 2019 às 06:09, Erik Rijkers  
escreveu:


The first 4 of these apply without error, but I can't get 0005 to 
apply.

This is what I use:


Erik, I generate a new patch set with patience diff algorithm. It
seems it applies cleanly.



It did apply cleanly, thanks.

But I can't get it to correctly do the partial replication in the 
attached pgbench-script (similar versions of which script I also used 
for earlier versions of the patch, last year).


There are complaints in the log (both pub and sub) like:
ERROR:  trying to store a heap tuple into wrong type of slot

I have no idea what causes that.

I attach a zip:

$ unzip -l logrep_rowfilter.zip
Archive:  logrep_rowfilter.zip
  Length  DateTimeName
-  -- -   
17942  2019-09-03 00:47   logfile.6525
10412  2019-09-03 00:47   logfile.6526
 6913  2019-09-03 00:47   logrep_rowfilter_2_nodes.sh
 3371  2019-09-03 00:47   output.txt
- ---
38638 4 files

That bash script runs 2 instances (as compiled on my local setup so it 
will not run as-is) and tries for one minute to get a slice of the 
pgbench_accounts table replicated.  One minute is short but I wanted 
short logfiles; I have tried the same up to 20 minutes without the 
replication completing.  I'll try even longer but in the meantime I hope 
you can figure out why these errors occur.



thanks,


Erik Rijkers


<>

Re: row filtering for logical replication

[Euler Taveira]

Em ter, 3 de set de 2019 às 00:16, Alexey Zagarin  escreveu:
>
> There are complaints in the log (both pub and sub) like:
> ERROR: trying to store a heap tuple into wrong type of slot
>
> I have no idea what causes that.
>
>
> Yeah, I've seen that too. It was fixed by Alexey Kondratov, in line 955 of 
> 0005-Row-filtering-for-logical-replication.patch it should be 
> &TTSOpsHeapTuple instead of &TTSOpsVirtual.
>
Ops... exact. That was an oversight while poking with different types of slots.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Alexey Zagarin]

> There are complaints in the log (both pub and sub) like:
> ERROR: trying to store a heap tuple into wrong type of slot
>
> I have no idea what causes that.

Yeah, I've seen that too. It was fixed by Alexey Kondratov, in line 955 of 
0005-Row-filtering-for-logical-replication.patch it should be &TTSOpsHeapTuple 
instead of &TTSOpsVirtual.

Re: row filtering for logical replication

[Erik Rijkers]

On 2019-09-03 05:32, Euler Taveira wrote:
Em ter, 3 de set de 2019 às 00:16, Alexey Zagarin  
escreveu:


There are complaints in the log (both pub and sub) like:
ERROR: trying to store a heap tuple into wrong type of slot

I have no idea what causes that.

Yeah, I've seen that too. It was fixed by Alexey Kondratov, in line 
955 of 0005-Row-filtering-for-logical-replication.patch it should be 
&TTSOpsHeapTuple instead of &TTSOpsVirtual.


Ops... exact. That was an oversight while poking with different types 
of slots.


OK, I'll consider Alexey Kondratov's set of patches as the current 
state-of-the-art then.  (They still apply.)


I found a problem where I'm not sure it's a bug:

The attached bash script does a test by setting up pgbench tables on 
both master and replica, and then sets up logical replication for a 
slice of pgbench_accounts. Then it does a short pgbench run, and loops 
until the results become identical(ok) (or breaks out after a certain 
time (NOK=not ok)).


It turns out this did not work until I added a wait state after the 
CREATE SUBSCRIPTION.  It always fails without the wait state, and always 
works with the wait state.


Do you agree this is a bug?


thanks (also to both Alexeys :))


Erik Rijkers


PS
by the way, this script won't run as-is on other machines; it has stuff 
particular to my local setup.



#!/bin/bash

# postgres binary compiled with 
#
# pgpatches/0130/logrep_rowfilter/20190902/v2-0001-Remove-unused-atttypmod-column-from-initial-table.patch
# pgpatches/0130/logrep_rowfilter/20190902/v2-0002-Store-number-of-tuples-in-WalRcvExecResult.patch   
# pgpatches/0130/logrep_rowfilter/20190902/v2-0003-Refactor-function-create_estate_for_relation.patch 
# pgpatches/0130/logrep_rowfilter/20190902/v2-0004-Rename-a-WHERE-node.patch  
# pgpatches/0130/logrep_rowfilter/20190902/v2-0005-Row-filtering-for-logical-replication.patch
# pgpatches/0130/logrep_rowfilter/20190902/v2-0006-Print-publication-WHERE-condition-in-psql.patch
# pgpatches/0130/logrep_rowfilter/20190902/v2-0007-Publication-where-condition-support-for-pg_dump.patch  
# pgpatches/0130/logrep_rowfilter/20190902/v2-0008-Debug-for-row-filtering.patch  
# pgpatches/0130/logrep_rowfilter/20190902/v2-0009-Add-simple-BDR-test-for-row-filtering.patch
 

unset PGDATABASE PGPORT PGSERVICE
export PGDATABASE=postgres

root_dir=/tmp/cascade/logrep_rowfilter

mkdir -p $root_dir

BIN=$HOME/pg_stuff/pg_installations/pgsql.logrep_rowfilter/bin

export PATH=$BIN:$PATH

  initdb=$BIN/initdb
postgres=$BIN/postgres
  pg_ctl=$BIN/pg_ctl
baseport=6525
   port1=$(( $baseport + 0 )) 
   port2=$(( $baseport + 1 ))
 appname=rowfilter

num_instances=2
scale=1  where="where (aid between 4 and 5-1)"
 #  scale=10 where="where (aid between 40 and 40+5-1)"
  clients=64
 duration=20
 wait=10  BASTA_COUNT=40  #   7200   #  wait seconds in total  

if [[ -d $root_dir/instance1 ]]; then rm -rf $root_dir/instance1; fi
if [[ -d $root_dir/instance2 ]]; then rm -rf $root_dir/instance2; fi
if [[ -d $root_dir/instance1 ]]; then exit ; fi
if [[ -d $root_dir/instance2 ]]; then exit ; fi

devel_file=/tmp/bugs
echo filterbug>$devel_file

for n in `seq 1 $num_instances`
do
  instance=instance$n
  server_dir=$root_dir/$instance
  data_dir=$server_dir/data
  port=$(( $baseport + $n -1 ))
  logfile=$server_dir/logfile.$port
  echo "-- $initdb --pgdata=$data_dir --encoding=UTF8 --pwfile=$devel_file "
   $initdb --pgdata=$data_dir --encoding=UTF8 --pwfile=$devel_file  &> /dev/null 
  ( $postgres  -D $data_dir -p $port \
--wal_level=logical --logging_collector=on \
--client_min_messages=warning \
--log_directory=$server_dir --log_filename=logfile.${port} \
--log_replication_commands=on & ) &> /dev/null
done 

echo "sleep 3s"
sleep 3

echo "
  drop table if exists pgbench_accounts;
  drop table if exists pgbench_branches;
  drop table if exists pgbench_tellers;
  drop table if exists pgbench_history;" | psql -qXp $port1 \
&& echo "
  drop table if exists pgbench_accounts;
  drop table if exists pgbench_branches;
  drop table if exists pgbench_tellers;
  drop table if exists pgbench_history;" | psql -qXp $port2 \
&& pgbench -p $port1 -qis $scale \
&& echo "alter table pgbench_history add column hid serial primary key;" \
  | psql -q1Xp $port1 && pg_dump -F c -p $port1 \
 --exclude-table-data=pgbench_history  \
 --exclude-table-data=pgbench_accounts \
 --exclude-table-data=pgbench_branches \
 --exclude-table-data=pgbench_tellers  \
   -t pgbench_history -t pgbench_accounts \
   -t pgbench_branches -t pgbench_tellers \
  | pg_restore -1 -p $port2 -d postgres


pub1=pub_${port1}_to_${port2}
sub1=sub_${port2}_fr_${port1}
echo -ne "
create publication $pub1;
alter  publication $pub1 add table pgbench_accounts $where ; --> where
alter  publication $pub1 add table pgbench_branches;
alter  publication $p

Re: row filtering for logical replication

[Alexey Zagarin]

> OK, I'll consider Alexey Kondratov's set of patches as the current
> state-of-the-art then. (They still apply.)

Alexey's patch is the rebased version of previous Euler's patch set, with slot 
type mistake fixed, and adapted to current changes in the master branch. It 
also has testing improvements. On the other hand, the new patches from Euler 
include more fixes and the implementation of filtering in COPY (as far as I can 
tell from code) which addresses my particular pain point with BDR. Hope they'll 
be joined soon. :)

> It turns out this did not work until I added a wait state after the
> CREATE SUBSCRIPTION. It always fails without the wait state, and always
> works with the wait state.
>
> Do you agree this is a bug?

I'm not sure this is a bug as after the subscription is added (or a new table 
added to the publication and then the subscription is refreshed), the whole 
table is synchronized using COPY statement. Depending on size of the table it 
can take some time. You may want to check srsubstate in pg_subscription_rel 
instead of just sleep for more reliable implementation.

Alexey

Re: row filtering for logical replication

[Euler Taveira]

Em ter, 3 de set de 2019 às 00:32, Euler Taveira
 escreveu:
>
> Ops... exact. That was an oversight while poking with different types of 
> slots.
>
Here is a rebased version including this small fix.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento
From 92474dd8e15d58e253d5b5aa76348d8973bf6d04 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 17:37:36 +
Subject: [PATCH 2/8] Store number of tuples in WalRcvExecResult

It seems to be a useful information while allocating memory for queries
that returns more than one row. It reduces memory allocation
for initial table synchronization.
---
 src/backend/replication/libpqwalreceiver/libpqwalreceiver.c | 5 +++--
 src/backend/replication/logical/tablesync.c | 5 ++---
 src/include/replication/walreceiver.h   | 1 +
 3 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 6eba08a920..343550a335 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -878,6 +878,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
  errdetail("Expected %d fields, got %d fields.",
 		   nRetTypes, nfields)));
 
+	walres->ntuples = PQntuples(pgres);
 	walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
 
 	/* Create tuple descriptor corresponding to expected result. */
@@ -888,7 +889,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
 	attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
 
 	/* No point in doing more here if there were no tuples returned. */
-	if (PQntuples(pgres) == 0)
+	if (walres->ntuples == 0)
 		return;
 
 	/* Create temporary context for local allocations. */
@@ -897,7 +898,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
 	   ALLOCSET_DEFAULT_SIZES);
 
 	/* Process returned rows. */
-	for (tupn = 0; tupn < PQntuples(pgres); tupn++)
+	for (tupn = 0; tupn < walres->ntuples; tupn++)
 	{
 		char	   *cstrs[MaxTupleAttributeNumber];
 
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 0a565dd837..42db4ada9e 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -709,9 +709,8 @@ fetch_remote_table_info(char *nspname, char *relname,
 (errmsg("could not fetch table info for table \"%s.%s\": %s",
 		nspname, relname, res->err)));
 
-	/* We don't know the number of rows coming, so allocate enough space. */
-	lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
-	lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
+	lrel->attnames = palloc0(res->ntuples * sizeof(char *));
+	lrel->atttyps = palloc0(res->ntuples * sizeof(Oid));
 	lrel->attkeys = NULL;
 
 	natt = 0;
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index e12a934966..0d32d598d8 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -196,6 +196,7 @@ typedef struct WalRcvExecResult
 	char	   *err;
 	Tuplestorestate *tuplestore;
 	TupleDesc	tupledesc;
+	int			ntuples;
 } WalRcvExecResult;
 
 /* libpqwalreceiver hooks */
-- 
2.11.0

From b8a8d98368ba032670788ac4f4b4ef666a4dedd0 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Tue, 27 Feb 2018 02:21:03 +
Subject: [PATCH 3/8] Refactor function create_estate_for_relation

Relation localrel is the only LogicalRepRelMapEntry structure member
that is useful for create_estate_for_relation.
---
 src/backend/replication/logical/worker.c | 14 +++---
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 11e6331f49..d9952c8b7e 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -173,7 +173,7 @@ ensure_transaction(void)
  * This is based on similar code in copy.c
  */
 static EState *
-create_estate_for_relation(LogicalRepRelMapEntry *rel)
+create_estate_for_relation(Relation rel)
 {
 	EState	   *estate;
 	ResultRelInfo *resultRelInfo;
@@ -183,13 +183,13 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel)
 
 	rte = makeNode(RangeTblEntry);
 	rte->rtekind = RTE_RELATION;
-	rte->relid = RelationGetRelid(rel->localrel);
-	rte->relkind = rel->localrel->rd_rel->relkind;
+	rte->relid = RelationGetRelid(rel);
+	rte->relkind = rel->rd_rel->relkind;
 	rte->rellockmode = AccessShareLock;
 	ExecInitRangeTable(estate, list_make1(rte));
 
 	resultRelInfo = makeNode(ResultRelInfo);
-	InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0);
+	InitResultRelInfo(resultRelInfo, rel, 1, NULL, 0);
 
 	estate->es_re

Re: row filtering for logical replication

[David Fetter]

On Wed, Feb 28, 2018 at 08:03:02PM -0300, Euler Taveira wrote:
> Hi,
> 
> The attached patches add support for filtering rows in the publisher.
> The output plugin will do the work if a filter was defined in CREATE
> PUBLICATION command. An optional WHERE clause can be added after the
> table name in the CREATE PUBLICATION such as:
> 
> CREATE PUBLICATION foo FOR TABLE departments WHERE (id > 2000 AND id <= 3000);
> 
> Row that doesn't match the WHERE clause will not be sent to the subscribers.
> 
> Patches 0001 and 0002 are only refactors and can be applied
> independently. 0003 doesn't include row filtering on initial
> synchronization.
> 
> Comments?

Great feature!  I think a lot of people will like to have the option
of trading a little extra CPU on the pub side for a bunch of network
traffic and some work on the sub side.

I noticed that the WHERE clause applies to all tables in the
publication.  Is that actually the right thing?  I'm thinking of a
case where we have foo(id, ...) and bar(foo_id, ).  To slice that
correctly, we'd want to do the ids in the foo table and the foo_ids in
the bar table.  In the system as written, that would entail, at least
potentially, writing a lot of publications by hand.

Something like
WHERE (
(table_1,..., table_N) HAS (/* WHERE clause here */) AND
(table_N+1,..., table_M) HAS (/* WHERE clause here */) AND
...
)

could be one way to specify.

I also noticed that in psql, \dRp+ doesn't show the WHERE clause,
which it probably should.

Does it need regression tests?

Best,
David.
-- 
David Fetter  http://fetter.org/
Phone: +1 415 235 3778

Remember to vote!
Consider donating to Postgres: http://www.postgresql.org/about/donate

Re: row filtering for logical replication

[Craig Ringer]

On 1 March 2018 at 07:03, Euler Taveira  wrote:

> Hi,
>
> The attached patches add support for filtering rows in the publisher.
> The output plugin will do the work if a filter was defined in CREATE
> PUBLICATION command. An optional WHERE clause can be added after the
> table name in the CREATE PUBLICATION such as:
>
> CREATE PUBLICATION foo FOR TABLE departments WHERE (id > 2000 AND id <=
> 3000);
>
> Row that doesn't match the WHERE clause will not be sent to the
> subscribers.
>
> Patches 0001 and 0002 are only refactors and can be applied
> independently. 0003 doesn't include row filtering on initial
> synchronization.
>


Good idea. I haven't read this yet, but one thing to make sure you've
handled is limiting the clause to referencing only the current tuple and
the catalogs. user-catalog tables are OK, too, anything that is
RelationIsAccessibleInLogicalDecoding().

This means only immutable functions may be invoked, since a stable or
volatile function might attempt to access a table. And views must be
prohibited or recursively checked. (We have tree walkers that would help
with this).

It might be worth looking at the current logic for CHECK expressions, since
the requirements are similar. In my opinion you could safely not bother
with allowing access to user catalog tables in the filter expressions and
limit them strictly to immutable functions and the tuple its self.

-- 
 Craig Ringer   http://www.2ndQuadrant.com/
 PostgreSQL Development, 24x7 Support, Training & Services

Re: row filtering for logical replication

[Erik Rijkers]

On 2018-03-01 00:03, Euler Taveira wrote:

The attached patches add support for filtering rows in the publisher.



001-Refactor-function-create_estate_for_relation.patch
0002-Rename-a-WHERE-node.patch
0003-Row-filtering-for-logical-replication.patch



Comments?


Very, very useful.  I really do hope this patch survives the 
late-arrival-cull.


I built this functionality into a test program I have been using and in 
simple cascading replication tests it works well.


I did find what I think is a bug (a bug easy to avoid but also easy to 
run into):
The test I used was to cascade 3 instances (all on one machine) from 
A->B->C

I ran a pgbench session in instance A, and used:
  in A: alter publication pub0_6515 add table pgbench_accounts where 
(aid between 4 and 6-1);

  in B: alter publication pub1_6516 add table pgbench_accounts;

The above worked well, but when I did the same but used the filter in 
both publications:
  in A: alter publication pub0_6515 add table pgbench_accounts where 
(aid between 4 and 6-1);
  in B: alter publication pub1_6516 add table pgbench_accounts where 
(aid between 4 and 6-1);


then the replication only worked for (pgbench-)scale 1 (hence: very 
little data); with larger scales it became slow (taking many minutes 
where the above had taken less than 1 minute), and ended up using far 
too much memory (or blowing up/crashing altogether).  Something not 
quite right there.


Nevertheless, I am much in favour of acquiring this functionality as 
soon as possible.



Thanks,


Erik Rijkers

Re: row filtering for logical replication

[Euler Taveira]

2018-02-28 21:47 GMT-03:00 David Fetter :
> I noticed that the WHERE clause applies to all tables in the
> publication.  Is that actually the right thing?  I'm thinking of a
> case where we have foo(id, ...) and bar(foo_id, ).  To slice that
> correctly, we'd want to do the ids in the foo table and the foo_ids in
> the bar table.  In the system as written, that would entail, at least
> potentially, writing a lot of publications by hand.
>
I didn't make it clear in my previous email and I think you misread
the attached docs. Each table can have an optional WHERE clause. I'll
made it clear when I rewrite the tests. Something like:

CREATE PUBLICATION tap_pub FOR TABLE tab_rowfilter_1 WHERE (a > 1000
AND b <> 'filtered'), tab_rowfilter_2 WHERE (c % 2 = 0),
tab_rowfilter_3;

Such syntax will not block another future feature that will publish
only few columns of the table.

> I also noticed that in psql, \dRp+ doesn't show the WHERE clause,
> which it probably should.
>
Yea, it could be added be I'm afraid of such long WHERE clauses.

> Does it need regression tests?
>
I included some tests just to demonstrate the feature but I'm planning
to add a separate test file for it.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[David Fetter]

On Thu, Mar 01, 2018 at 12:41:04PM -0300, Euler Taveira wrote:
> 2018-02-28 21:47 GMT-03:00 David Fetter :
> > I noticed that the WHERE clause applies to all tables in the
> > publication.  Is that actually the right thing?  I'm thinking of a
> > case where we have foo(id, ...) and bar(foo_id, ).  To slice that
> > correctly, we'd want to do the ids in the foo table and the foo_ids in
> > the bar table.  In the system as written, that would entail, at least
> > potentially, writing a lot of publications by hand.
> >
> I didn't make it clear in my previous email and I think you misread
> the attached docs. Each table can have an optional WHERE clause. I'll
> made it clear when I rewrite the tests. Something like:

Sorry I misunderstood.

> CREATE PUBLICATION tap_pub FOR TABLE tab_rowfilter_1 WHERE (a > 1000
> AND b <> 'filtered'), tab_rowfilter_2 WHERE (c % 2 = 0),
> tab_rowfilter_3;

That's great!

> Such syntax will not block another future feature that will publish
> only few columns of the table.
> 
> > I also noticed that in psql, \dRp+ doesn't show the WHERE clause,
> > which it probably should.
> >
> Yea, it could be added be I'm afraid of such long WHERE clauses.

I think of + as signifying, "I am ready to get a LOT of output in
order to see more detail."  Perhaps that's just me.

> > Does it need regression tests?
> >
> I included some tests just to demonstrate the feature but I'm
> planning to add a separate test file for it.

Excellent. This feature looks like a nice big chunk of the user-space
infrastructure needed for sharding, among other things.

Best,
David.
-- 
David Fetter  http://fetter.org/
Phone: +1 415 235 3778

Remember to vote!
Consider donating to Postgres: http://www.postgresql.org/about/donate

Re: row filtering for logical replication

[Erik Rijkers]

On 2018-03-01 16:27, Erik Rijkers wrote:

On 2018-03-01 00:03, Euler Taveira wrote:

The attached patches add support for filtering rows in the publisher.



001-Refactor-function-create_estate_for_relation.patch
0002-Rename-a-WHERE-node.patch
0003-Row-filtering-for-logical-replication.patch



Comments?


Very, very useful.  I really do hope this patch survives the 
late-arrival-cull.


I built this functionality into a test program I have been using and
in simple cascading replication tests it works well.

I did find what I think is a bug (a bug easy to avoid but also easy to
run into):
The test I used was to cascade 3 instances (all on one machine) from 
A->B->C

I ran a pgbench session in instance A, and used:
  in A: alter publication pub0_6515 add table pgbench_accounts where
(aid between 4 and 6-1);
  in B: alter publication pub1_6516 add table pgbench_accounts;

The above worked well, but when I did the same but used the filter in
both publications:
  in A: alter publication pub0_6515 add table pgbench_accounts where
(aid between 4 and 6-1);
  in B: alter publication pub1_6516 add table pgbench_accounts where
(aid between 4 and 6-1);

then the replication only worked for (pgbench-)scale 1 (hence: very
little data); with larger scales it became slow (taking many minutes
where the above had taken less than 1 minute), and ended up using far
too much memory (or blowing up/crashing altogether).  Something not
quite right there.

Nevertheless, I am much in favour of acquiring this functionality as
soon as possible.



Attached is 'logrep_rowfilter.sh', a demonstration of above-described 
bug.


The program runs initdb for 3 instances in /tmp (using ports 6515, 6516, 
and 6517) and sets up logical replication from 1->2->3.


It can be made to work by removing de where-clause on the second 'create 
publication' ( i.e., outcomment the $where2 variable ).




Thanks,


Erik Rijkers
#!/bin/sh

# postges binary with 
#
#  0001-Refactor-function-create_estate_for_relation.patch
#  0002-Rename-a-WHERE-node.patch
#  0003-Row-filtering-for-logical-replication.patch
#

unset PGDATABASE PGPORT PGSERVICE
export PGDATABASE=postgres

scale=10

root_dir=/tmp/cascade

BIN=$HOME/pg_stuff/pg_installations/pgsql.logrep_rowfilter/bin.fast

export PATH=$BIN:$PATH

  initdb=$BIN/initdb
postgres=$BIN/postgres
  pg_ctl=$BIN/pg_ctl
baseport=6515

if [[ -d $root_dir/instance1 ]]; then rm -rf $root_dir/instance1; fi
if [[ -d $root_dir/instance2 ]]; then rm -rf $root_dir/instance2; fi
if [[ -d $root_dir/instance3 ]]; then rm -rf $root_dir/instance3; fi
if [[ -d $root_dir/instance1 ]]; then exit ; fi
if [[ -d $root_dir/instance2 ]]; then exit ; fi
if [[ -d $root_dir/instance3 ]]; then exit ; fi

devel_file=/tmp/bugs
echo filterbug>$devel_file

num_instances=3

for n in `seq 1 $num_instances`
do
  instance=instance$n
  server_dir=$root_dir/$instance
  data_dir=$server_dir/data
  port=$(( 6515 + $n -1 ))
  logfile=$server_dir/logfile.$port
  echo "-- $initdb --pgdata=$data_dir --encoding=UTF8 --pwfile=$devel_file "
   $initdb --pgdata=$data_dir --encoding=UTF8 --pwfile=$devel_file  &> /dev/null
  ( $postgres  -D $data_dir -p $port \
--wal_level=logical --logging_collector=on \
--client_min_messages=warning \
--log_directory=$server_dir --log_filename=logfile.${port} \
--log_replication_commands=on & ) &> /dev/null
done 

echo "sleep 3s"
sleep 3

echo "
  drop table if exists pgbench_accounts;
  drop table if exists pgbench_branches;
  drop table if exists pgbench_tellers;
  drop table if exists pgbench_history;" | psql -qXp 6515 \
&& echo "
  drop table if exists pgbench_accounts;
  drop table if exists pgbench_branches;
  drop table if exists pgbench_tellers;
  drop table if exists pgbench_history;" | psql -qXp 6516 \
&& pgbench -p 6515 -qis $scale \
&& echo "alter table pgbench_history add column hid serial primary key;" \
  | psql -q1Xp 6515 && pg_dump -F c -p 6515 \
 --exclude-table-data=pgbench_history  \
 --exclude-table-data=pgbench_accounts \
 --exclude-table-data=pgbench_branches \
 --exclude-table-data=pgbench_tellers  \
   -t pgbench_history -t pgbench_accounts \
   -t pgbench_branches -t pgbench_tellers \
  | pg_restore -1 -p 6516 -d postgres

appname=rowfilter

   where="where (aid between 4 and 6-1)"
  where2="where (aid between 4 and 6-1)"

echo "
create publication pub1;
alter publication pub1 add table pgbench_accounts $where ; --> where 1
alter publication pub1 add table pgbench_branches;
alter publication pub1 add table pgbench_tellers;
alter publication pub1 add table pgbench_history;
" | psql -p 6515 -aqtAX

if [[ $num_instances -eq 3 ]]; then

  pg_dump -F c -p 6515 \
 --exclude-table-data=pgbench_history  \
 --exclude-table-data=pgbench_accounts \
 --exclude-table-data=pgbench_branches \
 --exclude-table-data=pgbench_tellers  \
 -t pgbench_history -t pgbench_accounts \
 -t pgbench_branches -t pgbench_tellers \

Re: row filtering for logical replication

[Andres Freund]

Hi,

On 2018-03-01 16:27:11 +0100, Erik Rijkers wrote:
> Very, very useful.  I really do hope this patch survives the
> late-arrival-cull.

FWIW, I don't think it'd be fair or prudent. There's definitely some
issues (see e.g. Craig's reply), and I don't see why this patch'd
deserve an exemption from the "nontrivial patches shouldn't be submitted
to the last CF" policy?

- Andres

Re: row filtering for logical replication

[David Steele]

Hi,

On 3/1/18 4:27 PM, Andres Freund wrote:
> On 2018-03-01 16:27:11 +0100, Erik Rijkers wrote:
>> Very, very useful.  I really do hope this patch survives the
>> late-arrival-cull.
> 
> FWIW, I don't think it'd be fair or prudent. There's definitely some
> issues (see e.g. Craig's reply), and I don't see why this patch'd
> deserve an exemption from the "nontrivial patches shouldn't be submitted
> to the last CF" policy?

I'm unable to find this in the CF under the title or author name.  If it
didn't get entered then it is definitely out.

If it does have an entry, then I agree with Andres that it should be
pushed to the next CF.

-- 
-David
da...@pgmasters.net

Re: row filtering for logical replication

[Euler Taveira]

2018-03-01 18:27 GMT-03:00 Andres Freund :
> FWIW, I don't think it'd be fair or prudent. There's definitely some
> issues (see e.g. Craig's reply), and I don't see why this patch'd
> deserve an exemption from the "nontrivial patches shouldn't be submitted
> to the last CF" policy?
>
I forgot to mention but this feature is for v12. I know the rules and
that is why I didn't add it to the in progress CF.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Euler Taveira]

2018-03-01 18:25 GMT-03:00 Erik Rijkers :
> Attached is 'logrep_rowfilter.sh', a demonstration of above-described bug.
>
Thanks for testing. I will figure out what is happening. There are
some leaks around. I'll post another version when I fix some of those
bugs.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Euler Taveira]

2018-02-28 21:54 GMT-03:00 Craig Ringer :
> Good idea. I haven't read this yet, but one thing to make sure you've
> handled is limiting the clause to referencing only the current tuple and the
> catalogs. user-catalog tables are OK, too, anything that is
> RelationIsAccessibleInLogicalDecoding().
>
> This means only immutable functions may be invoked, since a stable or
> volatile function might attempt to access a table. And views must be
> prohibited or recursively checked. (We have tree walkers that would help
> with this).
>
> It might be worth looking at the current logic for CHECK expressions, since
> the requirements are similar. In my opinion you could safely not bother with
> allowing access to user catalog tables in the filter expressions and limit
> them strictly to immutable functions and the tuple its self.
>
IIRC implementation is similar to RLS expressions. I'll check all of
these rules.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[David Steele]

On 3/1/18 6:00 PM, Euler Taveira wrote:

2018-03-01 18:27 GMT-03:00 Andres Freund :

FWIW, I don't think it'd be fair or prudent. There's definitely some
issues (see e.g. Craig's reply), and I don't see why this patch'd
deserve an exemption from the "nontrivial patches shouldn't be submitted
to the last CF" policy?


I forgot to mention but this feature is for v12. I know the rules and
that is why I didn't add it to the in progress CF.


That was the right thing to do, thank you!

--
-David
da...@pgmasters.net

Re: row filtering for logical replication

[Önder Kalacı]

Hi,

Thanks for working on this. I did some review and testing, please see my
comments below.

1)

   0008 is only for debug purposes (I'm
>not sure some of these messages will be part of the final patch).
>

I think if you are planning to keep the debug patch, there seems to be an
area of improvement in the following code:

/* If the tuple does not match one of the row filters, bail out 
*/
+   s = TextDatumGetCString(DirectFunctionCall2(pg_get_expr,
CStringGetTextDatum(nodeToString(rfnode)),
ObjectIdGetDatum(relation->rd_id)));
+   if (result)
+   elog(DEBUG2, "pgoutput_row_filter: row filter \"%s\" 
matched", s);
+   else
+   elog(DEBUG2, "pgoutput_row_filter: row filter \"%s\" 
not matched", s);
+   pfree(s);
+

We only need to calculate "s" if the debug level is DEBU2 or higher. So, we
could maybe do something like this:

if (log_min_messages <= DEBUG2 || client_min_messages <= DEBUG2)
{
   /* and the code block is moved to here */
}

2) I have done some tests with some different expressions that don't exist
on the regression tests, just to make sure that we don't have any edge
cases. All seems to work fine for the expressions
like: (column/1.0)::int::bool::text::bool, CASE WHEN column1> 4000 THEN
column2/ 100 > 1 ELSE false END, COALESCE((column/5)::bool, false),
NULLIF((column/5)::int::bool, false), column IS DISTINCT FROM 50040,
row(column, 2, 3) > row(2000, 2, 3), (column IS DISTINCT FROM), column IS
NULL, column IS NOT NULL, composite types

3) As another  edge case exploration, I tested with tables/types on
different schemas with escape chars on the schemas/custom types etc. All
looks good.

4) In terms of performance, I also separately verified that the overhead
seems pretty low with the final patch. I used the tests in
commands_to_perf_test.sql file which I shared earlier. The steps in the
file do not intend to measure the time precisely per tuple, but just to see
if there is any noticeable regression while moving lots of data. The
difference between (a) no filter (b) simple filter is between %1-%4, which
could even be considered in the noise level.

5) I guess we can by-pass the function limitation via operators. Do you see
anything problematic with that? I think that should be allowed as it helps
power users to create more complex replications if they need.

CREATE FUNCTION simple_f(int, int) RETURNS bool
> AS $$ SELECT hashint4($1) > $2  $$
> LANGUAGE SQL;
> CREATE OPERATOR =*>
> (
> PROCEDURE = simple_f,
> LEFTARG = int,
> RIGHTARG = int
> );
> CREATE PUBLICATION pub_1 FOR TABLE test WHERE (key =*> 1000);



5.1) It might be useful to have a regression test which has an user-defined
operator on the WHERE clause, and DROP without cascade is not allowed so
that we cover recordDependencyOnExpr() calls in the tests.

Thanks,
Onder KALACI
Software Engineer at Microsoft

Euler Taveira , 2 Şub 2021 Sal, 13:34 tarihinde şunu
yazdı:

> On Tue, Feb 2, 2021, at 8:38 AM, japin wrote:
>
> In 0003 patch, function GetPublicationRelationQuals() has been defined,
> but it
> never used.  So why should we define it?
>
> Thanks for taking a look again. It is an oversight. It was introduced in an
> attempt to refactor ALTER PUBLICATION SET TABLE. In
> AlterPublicationTables, we
> could possibly keep some publication-table mappings that does not change,
> however, since commit 3696a600e2, it is required to find the qual for all
> inheritors (see GetPublicationRelations). I explain this decision in the
> following comment:
>
> /*
>  * Remove all publication-table mappings.  We could possibly
>  * remove (i) tables that are not found in the new table list
> and
>  * (ii) tables that are being re-added with a different qual
>  * expression. For (ii), simply updating the existing tuple is
> not
>  * enough, because of qual expression dependencies.
>  */
>
> I will post a new patch set later.
>
>
> --
> Euler Taveira
> EDB   https://www.enterprisedb.com/
>
>

Re: row filtering for logical replication

[Euler Taveira]

On Mon, Feb 22, 2021, at 7:45 AM, Önder Kalacı wrote:
> Thanks for working on this. I did some review and testing, please see my 
> comments below.
I appreciate your review. I'm working on a new patch set and expect to post it 
soon.

> I think if you are planning to keep the debug patch, there seems to be an 
> area of improvement in the following code:
I was not planning to include the debug part, however, it would probably help 
to  
debug some use cases. In the "row [not] matched" message, it should be DEBUG3  
for a final version because it is too noisy. Since you mentioned I will inspect
and include in the main patch those DEBUG messages that could possibly be  
useful for debug purposes.

> 
> 4) In terms of performance, I also separately verified that the overhead 
> seems pretty low with the final patch. I used the tests in 
> commands_to_perf_test.sql file which I shared earlier. The steps in the file 
> do not intend to measure the time precisely per tuple, but just to see if 
> there is any noticeable regression while moving lots of data. The difference 
> between (a) no filter (b) simple filter is between %1-%4, which could even be 
> considered in the noise level.
I'm concerned about it too, I'm currently experimenting alternatives to reduce 
this overhead.

> 5) I guess we can by-pass the function limitation via operators. Do you see 
> anything problematic with that? I think that should be allowed as it helps 
> power users to create more complex replications if they need.
Yes, you can. I have to check if this user-defined operator could possibly  
   
break the replication. I will make sure to include a test covering this case
  
too.


--
Euler Taveira
EDB   https://www.enterprisedb.com/

Re: row filtering for logical replication

[Euler Taveira]

On Mon, Feb 22, 2021, at 9:28 AM, Euler Taveira wrote:
> On Mon, Feb 22, 2021, at 7:45 AM, Önder Kalacı wrote:
>> Thanks for working on this. I did some review and testing, please see my 
>> comments below.
> I appreciate your review. I'm working on a new patch set and expect to post 
> it soon.
I'm attaching a new patch set. This new version improves documentation and 
commit  
messages and incorporates a few debug messages. I did a couple of tests and
didn't find issues.

Here are some numbers from my i7 using a simple expression (aid > 0) on table
pgbench_accounts.

$ awk '/row filter time/ {print $9}' postgresql.log | /tmp/stat.pl 99
mean:   33.00 us
stddev: 17.65 us
median: 28.83 us
min-max:[3.48 .. 6404.84] us
percentile(99): 49.66 us
mode:   41.71 us

I don't expect 0005 and 0006 to be included. I attached them to help with some
tests.


--
Euler Taveira
EDB   https://www.enterprisedb.com/
From 596b278d4be77f67467aa1d61ce26e765b078197 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Mon, 18 Jan 2021 11:53:34 -0300
Subject: [PATCH 1/6] Rename a WHERE node

A WHERE clause will be used for row filtering in logical replication. We
already have a similar node: 'WHERE (condition here)'. Let's rename the
node to a generic name and use it for row filtering too.
---
 src/backend/parser/gram.y | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index dd72a9fc3c..ecfd98ba5b 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -485,7 +485,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type 	def_arg columnElem where_clause where_or_current_clause
 a_expr b_expr c_expr AexprConst indirection_el opt_slice_bound
 columnref in_expr having_clause func_table xmltable array_expr
-ExclusionWhereClause operator_def_arg
+OptWhereClause operator_def_arg
 %type 	rowsfrom_item rowsfrom_list opt_col_def_list
 %type  opt_ordinality
 %type 	ExclusionConstraintList ExclusionConstraintElem
@@ -3806,7 +3806,7 @@ ConstraintElem:
 	$$ = (Node *)n;
 }
 			| EXCLUDE access_method_clause '(' ExclusionConstraintList ')'
-opt_c_include opt_definition OptConsTableSpace  ExclusionWhereClause
+opt_c_include opt_definition OptConsTableSpace  OptWhereClause
 ConstraintAttributeSpec
 {
 	Constraint *n = makeNode(Constraint);
@@ -3908,7 +3908,7 @@ ExclusionConstraintElem: index_elem WITH any_operator
 			}
 		;
 
-ExclusionWhereClause:
+OptWhereClause:
 			WHERE '(' a_expr ')'	{ $$ = $3; }
 			| /*EMPTY*/{ $$ = NULL; }
 		;
-- 
2.20.1

From 9b1d46d5c146d591022c12c7acec8d61d4e0bfcc Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Mon, 18 Jan 2021 12:07:51 -0300
Subject: [PATCH 2/6] Row filter for logical replication

This feature adds row filter for publication tables. When you define or modify
a publication you can optionally filter rows that does not satisfy a WHERE
condition. It allows you to partially replicate a database or set of tables.
The row filter is per table which means that you can define different row
filters for different tables. A new row filter can be added simply by
informing the WHERE clause after the table name. The WHERE expression must be
enclosed by parentheses.

The WHERE clause should probably contain only columns that are part of the
primary key or that are covered by REPLICA IDENTITY. Otherwise, and DELETEs
won't be replicated. DELETE uses the old row version (that is limited to
primary key or REPLICA IDENTITY) to evaluate the row filter. INSERT and UPDATE
use the new row version to evaluate the row filter, hence, you can use any
column. If the row filter evaluates to NULL, it returns false. For simplicity,
functions are not allowed; it could possibly be addressed in a future patch.

If you choose to do the initial table synchronization, only data that satisfies
the row filters is sent. If the subscription has several publications in which
a table has been published with different WHERE clauses, rows must satisfy all
expressions to be copied.

If your publication contains partitioned table, the parameter
publish_via_partition_root determines if it uses the partition row filter (if
the parameter is false -- default) or the partitioned table row filter.
---
 doc/src/sgml/catalogs.sgml  |   8 +
 doc/src/sgml/ref/alter_publication.sgml |  11 +-
 doc/src/sgml/ref/create_publication.sgml|  38 +++-
 doc/src/sgml/ref/create_subscription.sgml   |   8 +-
 src/backend/catalog/pg_publication.c|  52 -
 src/backend/commands/publicationcmds.c  |  96 +
 src/backend/parser/gram.y   |  28 ++-
 src/backend/parser/parse_agg.c  |  10 +
 src/backend/parser/parse_expr.c |  13 ++
 src/backend/parser/parse_func.c |   3 +
 src/backend/replication/logical/tablesync.c |  93 -
 src/backend/replication

Re: row filtering for logical replication

[Rahila Syed]

Hi Euler,

Please find some comments below:

1. If the where clause contains non-replica identity columns, the delete
performed on a replicated row
 using DELETE from pub_tab where repl_ident_col = n;
is not being replicated, as logical replication does not have any info
whether the column has
to be filtered or not.
Shouldn't a warning be thrown in this case to notify the user that the
delete is not replicated.

2. Same for update, even if I update a row to match the quals on publisher,
it is still not being replicated to
the subscriber. (if the quals contain non-replica identity columns). I
think for UPDATE at least, the new value
of the non-replicate identity column is available which can be used to
filter and replicate the update.

3. 0001.patch,
Why is the name of the existing ExclusionWhereClause node being changed, if
the exact same definition is being used?

For 0002.patch,
4.   +
 +   memset(lrel, 0, sizeof(LogicalRepRelation));

Is this needed, apart from the above, patch does not use or update lrel at
all in that function.

5.  PublicationRelationQual and PublicationTable have similar fields, can
PublicationTable
be used in place of PublicationRelationQual instead of defining a new
struct?

Thank you,
Rahila Syed

Re: row filtering for logical replication

[Rahila Syed]

Hi Euler,

Please find below some review comments,

1.
   +
   + 
   +  prqual
   +  pg_node_tree
   +  
   +  Expression tree (in nodeToString()
   +  representation) for the relation's qualifying condition
   + 
I think the docs are being incorrectly updated to add a column to
pg_partitioned_table
instead of pg_publication_rel.

2.   +typedef struct PublicationRelationQual
 +{
+   Oid relid;
+   Relationrelation;
+   Node   *whereClause;
+} PublicationRelationQual;

Can this be given a more generic name like PublicationRelationInfo, so that
the same struct
can be used to store additional relation information in future, for ex.
column names, if column filtering is introduced.

3. Also, in the above structure, it seems that we can do with storing just
relid and derive relation information from it
using table_open when needed. Am I missing something?

4.  Currently in logical replication, I noticed that an UPDATE is being
applied on the subscriber even if the column values
 are unchanged. Can row-filtering feature be used to change it such that,
when all the OLD.columns = NEW.columns, filter out
the row from being sent to the subscriber. I understand this would need
REPLICA IDENTITY FULL to work, but would be an
improvement from the existing state.

On subscriber:

postgres=# select xmin, * from tab_rowfilter_1;
 xmin | a |  b
--+---+-
  555 | 1 | unfiltered
(1 row)

On publisher:
postgres=# ALTER TABLE tab_rowfilter_1 REPLICA IDENTITY FULL;
ALTER TABLE
postgres=# update tab_rowfilter_1 SET b = 'unfiltered' where a = 1;
UPDATE 1

On Subscriber:  The xmin has changed indicating the update from the
publisher was applied
even though nothing changed.

postgres=# select xmin, * from tab_rowfilter_1;
 xmin | a |  b
--+---+-
  556 | 1 | unfiltered
(1 row)

5. Currently, any existing rows that were not replicated, when updated to
match the publication quals
using UPDATE tab SET pub_qual_column = 'not_filtered' where a = 1; won't be
applied, as row
does not exist on the subscriber.  It would be good if ALTER SUBSCRIBER
REFRESH PUBLICATION
would help fetch such existing rows from publishers that match the qual
now(either because the row changed
or the qual changed)

Thank you,
Rahila Syed






On Tue, Mar 9, 2021 at 8:35 PM Rahila Syed  wrote:

> Hi Euler,
>
> Please find some comments below:
>
> 1. If the where clause contains non-replica identity columns, the delete
> performed on a replicated row
>  using DELETE from pub_tab where repl_ident_col = n;
> is not being replicated, as logical replication does not have any info
> whether the column has
> to be filtered or not.
> Shouldn't a warning be thrown in this case to notify the user that the
> delete is not replicated.
>
> 2. Same for update, even if I update a row to match the quals on
> publisher, it is still not being replicated to
> the subscriber. (if the quals contain non-replica identity columns). I
> think for UPDATE at least, the new value
> of the non-replicate identity column is available which can be used to
> filter and replicate the update.
>
> 3. 0001.patch,
> Why is the name of the existing ExclusionWhereClause node being changed,
> if the exact same definition is being used?
>
> For 0002.patch,
> 4.   +
>  +   memset(lrel, 0, sizeof(LogicalRepRelation));
>
> Is this needed, apart from the above, patch does not use or update lrel at
> all in that function.
>
> 5.  PublicationRelationQual and PublicationTable have similar fields, can
> PublicationTable
> be used in place of PublicationRelationQual instead of defining a new
> struct?
>
> Thank you,
> Rahila Syed
>
>

Re: row filtering for logical replication

[David Steele]

Hi Euler,

On 1/21/20 2:32 AM, Craig Ringer wrote:

On Fri, 17 Jan 2020 at 07:58, Euler Taveira  wrote:


Em qui., 16 de jan. de 2020 às 18:57, Tomas Vondra
 escreveu:


Euler, this patch is still in "waiting on author" since 11/25. Do you
plan to review changes made by Amit in the patches he submitted, or what
are your plans with this patch?


Yes, I'm working on Amit suggestions. I'll post a new patch as soon as possible.


Great. I think this'd be nice to see.


The last CF for PG13 has started. Do you have a new patch ready?

Regards,
--
-David
da...@pgmasters.net

Re: row filtering for logical replication

[David Steele]

On 3/3/20 12:39 PM, David Steele wrote:

Hi Euler,

On 1/21/20 2:32 AM, Craig Ringer wrote:

On Fri, 17 Jan 2020 at 07:58, Euler Taveira  wrote:


Em qui., 16 de jan. de 2020 às 18:57, Tomas Vondra
 escreveu:


Euler, this patch is still in "waiting on author" since 11/25. Do you
plan to review changes made by Amit in the patches he submitted, or 
what

are your plans with this patch?

Yes, I'm working on Amit suggestions. I'll post a new patch as soon 
as possible.


Great. I think this'd be nice to see.


The last CF for PG13 has started. Do you have a new patch ready?


I have marked this patch Returned with Feedback since no new patch has 
been posted.


Please submit to a future CF when a new patch is available.

Regards,
--
-David
da...@pgmasters.net

Re: row filtering for logical replication

[movead li]

Hello

I find several problems as below when I test the patches:

1. There be some regression problem after apply 0001.patch~0005.patch
   The regression problem is solved in 0006.patch
2. There be a data wrong after create subscription if the relation contains
 inherits table, for example:
   ##
   The Tables:
   CREATE TABLE cities (
   nametext,
   population  float,
   altitudeint
   );
   CREATE TABLE capitals (
   state   char(2)
   ) INHERITS (cities);
   
   Do on publication:
   insert into cities values('aaa',123, 134);
   insert into capitals values('bbb',123, 134);
   create publication pub_tc for table cities where (altitude > 100 and 
altitude < 200);
   postgres=# select * from cities ;
name | population | altitude 
   --++--
aaa  |123 |  134
bbb  |123 |  134
   (2 rows)
   
   Do on subscription:
   create subscription sub_tc connection 'host=localhost port=5432 
dbname=postgres' publication pub_tc;
   postgres=# select * from cities ;
name | population | altitude 
   --++--
aaa  |123 |  134
bbb  |123 |  134
bbb  |123 |  134
   (3 rows)
   ##
   An unexcept row appears.
   
3. I am puzzled when I test the update.
  Use the tables in problem 2 and test as below:
  #
  On publication:
  postgres=# insert into cities values('t1',123, 34);
  INSERT 0 1
  postgres=# update cities SET altitude = 134 where altitude = 34;
  UPDATE 1
  postgres=# select * from cities ;
   name | population | altitude 
  --++--
   t1   |123 |  134
  (1 row)
  On subscription:
  postgres=# select * from cities ;
   name | population | altitude 
  --++--
  (0 rows)
  
  On publication:
  insert into cities values('t1',1,'135');
  update cities set altitude=300 where altitude=135;
  postgres=# table cities ;
   name | population | altitude 
  --++--
   t1   |123 |  134
   t1   |  1 |  300
  (2 rows)
  
  On subscription:
  ostgres=# table cities ;
   name | population | altitude 
  --++--
   t1   |  1 |  135
  (1 row)
  #
  Result1:Update a row that is not suitable the publication condition to
  suitable, the subscription change nothing.
  Result2: Update a row that is suitable for the publication condition to
  not suitable, the subscription change nothing.
  If it is a bug? Or there should be an explanation about it?

4. SQL splicing code in fetch_remote_table_info() function is too long

---
Highgo Software (Canada/China/Pakistan) 
URL : www.highgo.ca 
EMAIL: mailto:movead...@highgo.ca

The new status of this patch is: Waiting on Author

Re: row filtering for logical replication

[Euler Taveira]

Em seg, 23 de set de 2019 às 01:59, movead li  escreveu:
>
> I find several problems as below when I test the patches:
>
First of all, thanks for your review.

> 1. There be some regression problem after apply 0001.patch~0005.patch
>The regression problem is solved in 0006.patch
>
Which regression?

> 2. There be a data wrong after create subscription if the relation contains
>  inherits table, for example:
>
Ouch. Good catch! Forgot about the ONLY in COPY with query. I will add
a test for it.

> 3. I am puzzled when I test the update.
>   Use the tables in problem 2 and test as below:
>   #
>   On publication:
>   postgres=# insert into cities values('t1',123, 34);
>   INSERT 0 1
>
INSERT isn't replicated.

>   postgres=# update cities SET altitude = 134 where altitude = 34;
>   UPDATE 1
>
There should be an error because you don't have a PK or REPLICA IDENTITY.

postgres=# update cities SET altitude = 134 where altitude = 34;
ERROR:  cannot update table "cities" because it does not have a
replica identity and publishes updates
HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.

Even if you create a PK or REPLICA IDENTITY, it won't turn this UPDATE
into a INSERT and send it to the other node (indeed UPDATE will be
sent however there isn't a tuple to update). Also, filter columns must
be in PK or REPLICA IDENTITY. I explain this in documentation.

> 4. SQL splicing code in fetch_remote_table_info() function is too long
>
I split it into small pieces. I also run pgindent to improve code style.

I'll send a patchset later today.


Regards,


--
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento

Re: row filtering for logical replication

[Euler Taveira]

Em qua, 25 de set de 2019 às 08:08, Euler Taveira
 escreveu:
>
> I'll send a patchset later today.
>
... and it is attached.


-- 
   Euler Taveira   Timbira -
http://www.timbira.com.br/
   PostgreSQL: Consultoria, Desenvolvimento, Suporte 24x7 e Treinamento
From b5d4d1369dbb4e7ec20182507dc5ae920dd8d2e9 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 18:39:22 +
Subject: [PATCH 1/8] Remove unused atttypmod column from initial table
 synchronization

 Since commit 7c4f52409a8c7d85ed169bbbc1f6092274d03920, atttypmod was
 added but not used. The removal is safe because COPY from publisher
 does not need such information.
---
 src/backend/replication/logical/tablesync.c | 7 +++
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 7881079..0a565dd 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -647,7 +647,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	StringInfoData cmd;
 	TupleTableSlot *slot;
 	Oid			tableRow[2] = {OIDOID, CHAROID};
-	Oid			attrRow[4] = {TEXTOID, OIDOID, INT4OID, BOOLOID};
+	Oid			attrRow[3] = {TEXTOID, OIDOID, BOOLOID};
 	bool		isnull;
 	int			natt;
 
@@ -691,7 +691,6 @@ fetch_remote_table_info(char *nspname, char *relname,
 	appendStringInfo(&cmd,
 	 "SELECT a.attname,"
 	 "   a.atttypid,"
-	 "   a.atttypmod,"
 	 "   a.attnum = ANY(i.indkey)"
 	 "  FROM pg_catalog.pg_attribute a"
 	 "  LEFT JOIN pg_catalog.pg_index i"
@@ -703,7 +702,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 	 lrel->remoteid,
 	 (walrcv_server_version(wrconn) >= 12 ? "AND a.attgenerated = ''" : ""),
 	 lrel->remoteid);
-	res = walrcv_exec(wrconn, cmd.data, 4, attrRow);
+	res = walrcv_exec(wrconn, cmd.data, 3, attrRow);
 
 	if (res->status != WALRCV_OK_TUPLES)
 		ereport(ERROR,
@@ -724,7 +723,7 @@ fetch_remote_table_info(char *nspname, char *relname,
 		Assert(!isnull);
 		lrel->atttyps[natt] = DatumGetObjectId(slot_getattr(slot, 2, &isnull));
 		Assert(!isnull);
-		if (DatumGetBool(slot_getattr(slot, 4, &isnull)))
+		if (DatumGetBool(slot_getattr(slot, 3, &isnull)))
 			lrel->attkeys = bms_add_member(lrel->attkeys, natt);
 
 		/* Should never happen. */
-- 
2.7.4

From 406b2dbe4df63a94364e548a67d085e255ea2644 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Fri, 9 Mar 2018 17:37:36 +
Subject: [PATCH 2/8] Store number of tuples in WalRcvExecResult

It seems to be a useful information while allocating memory for queries
that returns more than one row. It reduces memory allocation
for initial table synchronization.
---
 src/backend/replication/libpqwalreceiver/libpqwalreceiver.c | 5 +++--
 src/backend/replication/logical/tablesync.c | 5 ++---
 src/include/replication/walreceiver.h   | 1 +
 3 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 6eba08a..343550a 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -878,6 +878,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
  errdetail("Expected %d fields, got %d fields.",
 		   nRetTypes, nfields)));
 
+	walres->ntuples = PQntuples(pgres);
 	walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
 
 	/* Create tuple descriptor corresponding to expected result. */
@@ -888,7 +889,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
 	attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
 
 	/* No point in doing more here if there were no tuples returned. */
-	if (PQntuples(pgres) == 0)
+	if (walres->ntuples == 0)
 		return;
 
 	/* Create temporary context for local allocations. */
@@ -897,7 +898,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
 	   ALLOCSET_DEFAULT_SIZES);
 
 	/* Process returned rows. */
-	for (tupn = 0; tupn < PQntuples(pgres); tupn++)
+	for (tupn = 0; tupn < walres->ntuples; tupn++)
 	{
 		char	   *cstrs[MaxTupleAttributeNumber];
 
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 0a565dd..42db4ad 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -709,9 +709,8 @@ fetch_remote_table_info(char *nspname, char *relname,
 (errmsg("could not fetch table info for table \"%s.%s\": %s",
 		nspname, relname, res->err)));
 
-	/* We don't know the number of rows coming, so allocate enough space. */
-	lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
-	lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
+	lrel->attnames = palloc0(res->ntuples * sizeof(char *))

Re: row filtering for logical replication

[Euler Taveira]

On Tue, Mar 9, 2021, at 12:05 PM, Rahila Syed wrote:
> Please find some comments below:
Thanks for your review.

> 1. If the where clause contains non-replica identity columns, the delete 
> performed on a replicated row
>  using DELETE from pub_tab where repl_ident_col = n;
> is not being replicated, as logical replication does not have any info 
> whether the column has 
> to be filtered or not. 
> Shouldn't a warning be thrown in this case to notify the user that the delete 
> is not replicated.
Isn't documentation enough? If you add a WARNING, it should be printed per row,
hence, a huge DELETE will flood the client with WARNING messages by default. If
you are thinking about LOG messages, it is a different story. However, we
should limit those messages to one per transaction. Even if we add such an aid,
it would impose a performance penalty while checking the DELETE is not
replicating because the row filter contains a column that is not part of the PK
or REPLICA IDENTITY. If I were to add any message, it would be to warn at the
creation time (CREATE PUBLICATION or ALTER PUBLICATION ... [ADD|SET] TABLE).

> 2. Same for update, even if I update a row to match the quals on publisher, 
> it is still not being replicated to 
> the subscriber. (if the quals contain non-replica identity columns). I think 
> for UPDATE at least, the new value
> of the non-replicate identity column is available which can be used to filter 
> and replicate the update.
Indeed, the row filter for UPDATE uses the new tuple. Maybe your non-replica
identity column contains NULL that evaluates the expression to false.

> 3. 0001.patch, 
> Why is the name of the existing ExclusionWhereClause node being changed, if 
> the exact same definition is being used?
Because this node ExclusionWhereClause is used for exclusion constraint. This
patch renames the node to made it clear it is a generic node that could be used
for other filtering features in the future.

> For 0002.patch,
> 4.   +
>  +   memset(lrel, 0, sizeof(LogicalRepRelation));
> 
> Is this needed, apart from the above, patch does not use or update lrel at 
> all in that function.
Good catch. It is a leftover from a previous patch. It will be fixed in the
next patch set.

> 5.  PublicationRelationQual and PublicationTable have similar fields, can 
> PublicationTable
> be used in place of PublicationRelationQual instead of defining a new struct?
I don't think it is a good idea to have additional fields in a parse node. The
DDL commands use Relation (PublicationTableQual) and parse code uses RangeVar
(PublicationTable). publicationcmds.c uses Relation everywhere so I decided to
create a new struct to store Relation and qual as a list item. It also 
minimizes the places
you have to modify.


--
Euler Taveira
EDB   https://www.enterprisedb.com/

Re: row filtering for logical replication

[Euler Taveira]

On Thu, Mar 18, 2021, at 7:51 AM, Rahila Syed wrote:
> 1. 
> I think the docs are being incorrectly updated to add a column to 
> pg_partitioned_table
> instead of pg_publication_rel.
Good catch.

> 2.   +typedef struct PublicationRelationQual
>  +{
> +   Oid relid;
> +   Relationrelation;
> +   Node   *whereClause;
> +} PublicationRelationQual;
> 
> Can this be given a more generic name like PublicationRelationInfo, so that 
> the same struct 
> can be used to store additional relation information in future, for ex. 
> column names, if column filtering is introduced.
Good idea. I rename it and it'll be in this next patch set.

> 3. Also, in the above structure, it seems that we can do with storing just 
> relid and derive relation information from it
> using table_open when needed. Am I missing something?
We need the Relation. See OpenTableList(). The way this code is organized, it
opens all publication tables and append each Relation to a list. This list is
used in PublicationAddTables() to update the catalog. I tried to minimize the
number of refactors while introducing this feature. We could probably revise
this code in the future (someone said in a previous discussion that it is weird
to open relations in one source code file -- publicationcmds.c -- and use it
into another one -- pg_publication.c).

> 4.  Currently in logical replication, I noticed that an UPDATE is being 
> applied on the subscriber even if the column values
>  are unchanged. Can row-filtering feature be used to change it such that, 
> when all the OLD.columns = NEW.columns, filter out 
> the row from being sent to the subscriber. I understand this would need 
> REPLICA IDENTITY FULL to work, but would be an
> improvement from the existing state.
This is how Postgres works.

postgres=# create table foo (a integer, b integer);
CREATE TABLE
postgres=# insert into foo values(1, 100);
INSERT 0 1
postgres=# select ctid, xmin, xmax, a, b from foo;
ctid  |  xmin  | xmax | a |  b 
---++--+---+-
(0,1) | 488920 |0 | 1 | 100
(1 row)

postgres=# update foo set b = 101 where a = 1;
UPDATE 1
postgres=# select ctid, xmin, xmax, a, b from foo;
ctid  |  xmin  | xmax | a |  b 
---++--+---+-
(0,2) | 488921 |0 | 1 | 101
(1 row)

postgres=# update foo set b = 101 where a = 1;
UPDATE 1
postgres=# select ctid, xmin, xmax, a, b from foo;
ctid  |  xmin  | xmax | a |  b 
---++--+---+-
(0,3) | 488922 |0 | 1 | 101
(1 row)

You could probably abuse this feature and skip some UPDATEs when old tuple is
identical to new tuple. The question is: why would someone issue the same
command multiple times? A broken application? I would say: don't do it. Besides
that, this feature could impose an overhead into a code path that already
consume substantial CPU time. I've seen some tables with RIF and dozens of
columns that would certainly contribute to increase the replication lag.

> 5. Currently, any existing rows that were not replicated, when updated to 
> match the publication quals
> using UPDATE tab SET pub_qual_column = 'not_filtered' where a = 1; won't be 
> applied, as row 
> does not exist on the subscriber.  It would be good if ALTER SUBSCRIBER 
> REFRESH PUBLICATION
> would help fetch such existing rows from publishers that match the qual 
> now(either because the row changed
> or the qual changed)
I see. This should be addressed by a resynchronize feature. Such option is
useful when you have to change the row filter. It should certainly be implement
as an ALTER SUBSCRIPTION subcommand.

I attached a new patch set that addresses:

* fix documentation;
* rename PublicationRelationQual to PublicationRelationInfo;
* remove the memset that was leftover from a previous patch set;
* add new tests to improve coverage (INSERT/UPDATE/DELETE to exercise the row
  filter code).


--
Euler Taveira
EDB   https://www.enterprisedb.com/
From a6d893be0091bc8cd8569ac380e6f628263d31c0 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Mon, 18 Jan 2021 11:53:34 -0300
Subject: [PATCH v12 1/5] Rename a WHERE node

A WHERE clause will be used for row filtering in logical replication. We
already have a similar node: 'WHERE (condition here)'. Let's rename the
node to a generic name and use it for row filtering too.
---
 src/backend/parser/gram.y | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index bc43641ffe..22bbb27041 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -495,7 +495,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type 	def_arg columnElem where_clause where_or_current_clause
 a_expr b_expr c_expr AexprConst indirection_el opt_slice_bound
 columnref in_expr having_clause func_table xmltable array_expr
-ExclusionWhereClause operator_def_arg
+OptWhereClause operator_def_arg
 %type 	rowsfrom_item rowsfro

Re: row filtering for logical replication

[Peter Eisentraut]

On 22.03.21 03:15, Euler Taveira wrote:

I attached a new patch set that addresses:

* fix documentation;
* rename PublicationRelationQual to PublicationRelationInfo;
* remove the memset that was leftover from a previous patch set;
* add new tests to improve coverage (INSERT/UPDATE/DELETE to exercise 
the row

   filter code).


I have committed the 0001 patch.

Attached are a few fixup patches that I recommend you integrate into 
your patch set.  They address backward compatibility with PG13, and a 
few more stylistic issues.


I suggest you combine your 0002, 0003, and 0004 patches into one.  They 
can't be used separately, and for example the psql changes in patch 0003 
already appear as regression test output changes in 0002, so this 
arrangement isn't useful.  (0005 can be kept separately, since it's 
mostly for debugging right now.)
From c29459505db88c86dd7aa61019fa406202c30b0a Mon Sep 17 00:00:00 2001
From: Peter Eisentraut 
Date: Thu, 25 Mar 2021 11:57:48 +0100
Subject: [PATCH 1/6] fixup! Row filter for logical replication

Remove unused header files, clean up whitespace.
---
 src/backend/catalog/pg_publication.c| 2 --
 src/backend/replication/pgoutput/pgoutput.c | 4 
 2 files changed, 6 deletions(-)

diff --git a/src/backend/catalog/pg_publication.c 
b/src/backend/catalog/pg_publication.c
index f31ae28de2..78f5780fb7 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -33,11 +33,9 @@
 #include "catalog/pg_type.h"
 #include "funcapi.h"
 #include "miscadmin.h"
-
 #include "parser/parse_clause.h"
 #include "parser/parse_collate.h"
 #include "parser/parse_relation.h"
-
 #include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/catcache.h"
diff --git a/src/backend/replication/pgoutput/pgoutput.c 
b/src/backend/replication/pgoutput/pgoutput.c
index ce6da8de19..6151f34925 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -16,14 +16,10 @@
 #include "catalog/partition.h"
 #include "catalog/pg_publication.h"
 #include "catalog/pg_publication_rel.h"
-#include "catalog/pg_type.h"
 #include "commands/defrem.h"
 #include "executor/executor.h"
 #include "fmgr.h"
-#include "nodes/execnodes.h"
 #include "nodes/nodeFuncs.h"
-#include "optimizer/planner.h"
-#include "optimizer/optimizer.h"
 #include "parser/parse_coerce.h"
 #include "replication/logical.h"
 #include "replication/logicalproto.h"
-- 
2.30.2

From 91c20ecb45a8627a9252ed589fe6b85927be8b45 Mon Sep 17 00:00:00 2001
From: Peter Eisentraut 
Date: Thu, 25 Mar 2021 11:59:13 +0100
Subject: [PATCH 2/6] fixup! Row filter for logical replication

Allow replication from older PostgreSQL versions without prqual.
---
 src/backend/replication/logical/tablesync.c | 70 +++--
 1 file changed, 37 insertions(+), 33 deletions(-)

diff --git a/src/backend/replication/logical/tablesync.c 
b/src/backend/replication/logical/tablesync.c
index 40d84dadb5..246510c82e 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -796,49 +796,53 @@ fetch_remote_table_info(char *nspname, char *relname,
walrcv_clear_result(res);
 
/* Get relation qual */
-   resetStringInfo(&cmd);
-   appendStringInfo(&cmd,
-"SELECT pg_get_expr(prqual, prrelid) "
-"  FROM pg_publication p "
-"  INNER JOIN pg_publication_rel pr "
-"   ON (p.oid = pr.prpubid) "
-" WHERE pr.prrelid = %u "
-"   AND p.pubname IN (", 
lrel->remoteid);
-
-   first = true;
-   foreach(lc, MySubscription->publications)
+   if (walrcv_server_version(wrconn) >= 14)
{
-   char   *pubname = strVal(lfirst(lc));
+   resetStringInfo(&cmd);
+   appendStringInfo(&cmd,
+"SELECT pg_get_expr(prqual, 
prrelid) "
+"  FROM pg_publication p "
+"  INNER JOIN 
pg_publication_rel pr "
+"   ON (p.oid = 
pr.prpubid) "
+" WHERE pr.prrelid = %u "
+"   AND p.pubname IN (", 
lrel->remoteid);
+
+   first = true;
+   foreach(lc, MySubscription->publications)
+   {
+   char   *pubname = strVal(lfirst(lc));
 
-   if (first)
-   first = false;
-   else
-   appendStringInfoString(&cmd, ", ");
+   if (first)
+   first = false;
+   else
+  

Re: row filtering for logical replication

[Euler Taveira]

On Thu, Mar 25, 2021, at 8:15 AM, Peter Eisentraut wrote:
> I have committed the 0001 patch.
> 
> Attached are a few fixup patches that I recommend you integrate into 
> your patch set.  They address backward compatibility with PG13, and a 
> few more stylistic issues.
> 
> I suggest you combine your 0002, 0003, and 0004 patches into one.  They 
> can't be used separately, and for example the psql changes in patch 0003 
> already appear as regression test output changes in 0002, so this 
> arrangement isn't useful.  (0005 can be kept separately, since it's 
> mostly for debugging right now.)
I appreciate your work on it. I split into psql and pg_dump support just
because it was developed after the main patch. I expect them to be combined
into the main patch (0002) before committing it. This new patch set integrates
them into the main patch.

I totally forgot about the backward compatibility support. Good catch.  While
inspecting the code again, I did a small fix into the psql support. I added an
else as shown below so the query always returns the same number of columns and
we don't possibly have an issue while using a column number that is out of
range in PQgetisnull() a few lines later.

if (pset.sversion >= 14)
appendPQExpBuffer(&buf,
  ", pg_get_expr(pr.prqual, c.oid)");
else
appendPQExpBuffer(&buf,
  ", NULL");

While testing the replication between v14 -> v10, I realized that even if the
tables in the publication have row filters, the data synchronization code won't
evaluate the row filter expressions. That's because the subscriber (v10) is
responsible to assemble the COPY command (possibly adding row filters) for data
synchronization and there is no such code in released versions. I added a new
sentence into copy_data parameter saying that row filters won't be used if
version is prior than 14. I also include this info into the commit message.

At this time, I didn't include the patch that changes the log_min_messages in
the row filter regression test. It was part of this patch set for testing
purposes only.

I don't expect the patch that measures row filter performance to be included
but I'm including it again in case someone wants to inspect the performance
numbers.


--
Euler Taveira
EDB   https://www.enterprisedb.com/
From 692572d0daaf8905e750889aa2f374764d86636c Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Mon, 18 Jan 2021 12:07:51 -0300
Subject: [PATCH v13 1/2] Row filter for logical replication

This feature adds row filter for publication tables. When you define or modify
a publication you can optionally filter rows that does not satisfy a WHERE
condition. It allows you to partially replicate a database or set of tables.
The row filter is per table which means that you can define different row
filters for different tables. A new row filter can be added simply by
informing the WHERE clause after the table name. The WHERE expression must be
enclosed by parentheses.

The WHERE clause should probably contain only columns that are part of the
primary key or that are covered by REPLICA IDENTITY. Otherwise, and DELETEs
won't be replicated. DELETE uses the old row version (that is limited to
primary key or REPLICA IDENTITY) to evaluate the row filter. INSERT and UPDATE
use the new row version to evaluate the row filter, hence, you can use any
column. If the row filter evaluates to NULL, it returns false. For simplicity,
functions are not allowed; it could possibly be addressed in a future patch.

If you choose to do the initial table synchronization, only data that satisfies
the row filters is sent. If the subscription has several publications in which
a table has been published with different WHERE clauses, rows must satisfy all
expressions to be copied. If subscriber is a pre-14 version, data
synchronization won't use row filters if they are defined in the publisher.
Previous versions cannot handle row filters.

If your publication contains partitioned table, the parameter
publish_via_partition_root determines if it uses the partition row filter (if
the parameter is false -- default) or the partitioned table row filter.
---
 doc/src/sgml/catalogs.sgml  |   8 +
 doc/src/sgml/ref/alter_publication.sgml |  11 +-
 doc/src/sgml/ref/create_publication.sgml|  38 +++-
 doc/src/sgml/ref/create_subscription.sgml   |  11 +-
 src/backend/catalog/pg_publication.c|  50 -
 src/backend/commands/publicationcmds.c  |  98 +
 src/backend/parser/gram.y   |  26 ++-
 src/backend/parser/parse_agg.c  |  10 +
 src/backend/parser/parse_expr.c |  13 ++
 src/backend/parser/parse_func.c |   3 +
 src/backend/replication/logical/tablesync.c |  95 -
 src/backend/replication/logical/worker.c|  14 +-
 src/backend/replication/pgoutput/pgoutput.c | 202 --
 src/bin/pg_dump/pg_dump.c

Re: row filtering for logical replication

[Rahila Syed]

Hi Euler,

While running some tests on v13 patches, I noticed that, in case the
published table data
already exists on the subscriber database before creating the subscription,
at the time of
CREATE subscription/table synchronization, an error as seen as follows

With the patch:

2021-03-29 14:32:56.265 IST [78467] STATEMENT:  CREATE_REPLICATION_SLOT
"pg_16406_sync_16390_6944995860755251708" LOGICAL pgoutput USE_SNAPSHOT
2021-03-29 14:32:56.279 IST [78467] LOG:  could not send data to client:
Broken pipe
2021-03-29 14:32:56.279 IST [78467] STATEMENT:  COPY (SELECT aid, bid,
abalance, filler FROM public.pgbench_accounts WHERE (aid > 0)) TO STDOUT
2021-03-29 14:32:56.279 IST [78467] FATAL:  connection to client lost
2021-03-29 14:32:56.279 IST [78467] STATEMENT:  COPY (SELECT aid, bid,
abalance, filler FROM public.pgbench_accounts WHERE (aid > 0)) TO STDOUT
2021-03-29 14:33:01.302 IST [78470] LOG:  logical decoding found consistent
point at 0/4E2B8460
2021-03-29 14:33:01.302 IST [78470] DETAIL:  There are no running
transactions.

Without the patch:

2021-03-29 15:05:01.581 IST [79029] ERROR:  duplicate key value violates
unique constraint "pgbench_branches_pkey"
2021-03-29 15:05:01.581 IST [79029] DETAIL:  Key (bid)=(1) already exists.
2021-03-29 15:05:01.581 IST [79029] CONTEXT:  COPY pgbench_branches, line 1
2021-03-29 15:05:01.583 IST [78538] LOG:  background worker "logical
replication worker" (PID 79029) exited with exit code 1
2021-03-29 15:05:06.593 IST [79031] LOG:  logical replication table
synchronization worker for subscription "test_sub2", table
"pgbench_branches" has started

Without the patch the COPY command throws an ERROR, but with the patch, a
similar scenario results in client connection being lost.

I didn't investigate it more, but looks like we should maintain the
existing behaviour when table synchronization fails
due to duplicate data.

Thank you,
Rahila Syed

Re: row filtering for logical replication

[Euler Taveira]

On Mon, Mar 29, 2021, at 6:45 AM, Rahila Syed wrote:
> While running some tests on v13 patches, I noticed that, in case the 
> published table data 
> already exists on the subscriber database before creating the subscription, 
> at the time of
> CREATE subscription/table synchronization, an error as seen as follows 
> 
> With the patch:
> 
> 2021-03-29 14:32:56.265 IST [78467] STATEMENT:  CREATE_REPLICATION_SLOT 
> "pg_16406_sync_16390_6944995860755251708" LOGICAL pgoutput USE_SNAPSHOT
> 2021-03-29 14:32:56.279 IST [78467] LOG:  could not send data to client: 
> Broken pipe
> 2021-03-29 14:32:56.279 IST [78467] STATEMENT:  COPY (SELECT aid, bid, 
> abalance, filler FROM public.pgbench_accounts WHERE (aid > 0)) TO STDOUT
> 2021-03-29 14:32:56.279 IST [78467] FATAL:  connection to client lost
> 2021-03-29 14:32:56.279 IST [78467] STATEMENT:  COPY (SELECT aid, bid, 
> abalance, filler FROM public.pgbench_accounts WHERE (aid > 0)) TO STDOUT
> 2021-03-29 14:33:01.302 IST [78470] LOG:  logical decoding found consistent 
> point at 0/4E2B8460
> 2021-03-29 14:33:01.302 IST [78470] DETAIL:  There are no running 
> transactions.
Rahila, I tried to reproduce this issue with the attached script but no luck. I 
always get

> Without the patch:
> 
> 2021-03-29 15:05:01.581 IST [79029] ERROR:  duplicate key value violates 
> unique constraint "pgbench_branches_pkey"
> 2021-03-29 15:05:01.581 IST [79029] DETAIL:  Key (bid)=(1) already exists.
> 2021-03-29 15:05:01.581 IST [79029] CONTEXT:  COPY pgbench_branches, line 1
> 2021-03-29 15:05:01.583 IST [78538] LOG:  background worker "logical 
> replication worker" (PID 79029) exited with exit code 1
> 2021-03-29 15:05:06.593 IST [79031] LOG:  logical replication table 
> synchronization worker for subscription "test_sub2", table "pgbench_branches" 
> has started
... this message. The code that reports this error is from the COPY command.
Row filter modifications has no control over it. It seems somehow your
subscriber close the replication connection causing this issue. Can you
reproduce it consistently? If so, please share your steps.


--
Euler Taveira
EDB   https://www.enterprisedb.com/


test-row-filter-pgbench.sh
Description: application/shellscript

Re: row filtering for logical replication

[Rahila Syed]

Hi,

>
> While running some tests on v13 patches, I noticed that, in case the
> published table data
> already exists on the subscriber database before creating the
> subscription, at the time of
> CREATE subscription/table synchronization, an error as seen as follows
>
> With the patch:
>
> 2021-03-29 14:32:56.265 IST [78467] STATEMENT:  CREATE_REPLICATION_SLOT
> "pg_16406_sync_16390_6944995860755251708" LOGICAL pgoutput USE_SNAPSHOT
> 2021-03-29 14:32:56.279 IST [78467] LOG:  could not send data to client:
> Broken pipe
> 2021-03-29 14:32:56.279 IST [78467] STATEMENT:  COPY (SELECT aid, bid,
> abalance, filler FROM public.pgbench_accounts WHERE (aid > 0)) TO STDOUT
> 2021-03-29 14:32:56.279 IST [78467] FATAL:  connection to client lost
> 2021-03-29 14:32:56.279 IST [78467] STATEMENT:  COPY (SELECT aid, bid,
> abalance, filler FROM public.pgbench_accounts WHERE (aid > 0)) TO STDOUT
> 2021-03-29 14:33:01.302 IST [78470] LOG:  logical decoding found
> consistent point at 0/4E2B8460
> 2021-03-29 14:33:01.302 IST [78470] DETAIL:  There are no running
> transactions.
>
> Rahila, I tried to reproduce this issue with the attached script but no
> luck. I always get
>
> OK, Sorry for confusion. Actually both the errors are happening on
different servers. *Broken pipe*  error on publisher and
the following error on subscriber end. And the behaviour is consistent with
or without row filtering.

> Without the patch:
>
> 2021-03-29 15:05:01.581 IST [79029] ERROR:  duplicate key value violates
> unique constraint "pgbench_branches_pkey"
> 2021-03-29 15:05:01.581 IST [79029] DETAIL:  Key (bid)=(1) already exists.
> 2021-03-29 15:05:01.581 IST [79029] CONTEXT:  COPY pgbench_branches, line 1
> 2021-03-29 15:05:01.583 IST [78538] LOG:  background worker "logical
> replication worker" (PID 79029) exited with exit code 1
> 2021-03-29 15:05:06.593 IST [79031] LOG:  logical replication table
> synchronization worker for subscription "test_sub2", table
> "pgbench_branches" has started
>
> ... this message. The code that reports this error is from the COPY
> command.
> Row filter modifications has no control over it. It seems somehow your
> subscriber close the replication connection causing this issue. Can you
> reproduce it consistently? If so, please share your steps.
>
> Please ignore the report.

Thank you,
Rahila Syed

Re: row filtering for logical replication

[Amit Kapila]

On Mon, Mar 29, 2021 at 6:47 PM Euler Taveira  wrote:
>
Few comments:
==
1. How can we specify row filters for multiple tables for a
publication? Consider a case as below:
postgres=# CREATE TABLE tab_rowfilter_1 (a int primary key, b text);
CREATE TABLE
postgres=# CREATE TABLE tab_rowfilter_2 (c int primary key);
CREATE TABLE

postgres=# CREATE PUBLICATION tap_pub_1 FOR TABLE tab_rowfilter_1,
tab_rowfilter_2 WHERE (a > 1000 AND b <> 'filtered');
ERROR:  column "a" does not exist
LINE 1: ...FOR TABLE tab_rowfilter_1, tab_rowfilter_2 WHERE (a > 1000 A...

 ^

postgres=# CREATE PUBLICATION tap_pub_1 FOR TABLE tab_rowfilter_1,
tab_rowfilter_2  WHERE (c > 1000);
CREATE PUBLICATION

It gives an error when I tried to specify the columns corresponding to
the first relation but is fine for columns for the second relation.
Then, I tried few more combinations like below but that didn't work.
CREATE PUBLICATION tap_pub_1 FOR TABLE tab_rowfilter_1 As t1,
tab_rowfilter_2 As t2 WHERE (t1.a > 1000 AND t1.b <> 'filtered');

Will users be allowed to specify join conditions among columns from
multiple tables?

2.
+ /*
+ * Although ALTER PUBLICATION grammar allows WHERE clause to be specified
+ * for DROP TABLE action, it doesn't make sense to allow it. We implement
+ * this restriction here, instead of complicating the grammar to enforce
+ * it.
+ */
+ if (stmt->tableAction == DEFELEM_DROP)
+ {
+ ListCell   *lc;
+
+ foreach(lc, stmt->tables)
+ {
+ PublicationTable *t = lfirst(lc);
+
+ if (t->whereClause)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot use a WHERE clause when removing table from
publication \"%s\"",
+ NameStr(pubform->pubname;
+ }
+ }

Is there a reason to deal with this here separately rather than in the
ALTER PUBLICATION grammar?


-- 
With Regards,
Amit Kapila.

Re: row filtering for logical replication

[Euler Taveira]

On Tue, Mar 30, 2021, at 8:23 AM, Amit Kapila wrote:
> On Mon, Mar 29, 2021 at 6:47 PM Euler Taveira  > wrote:
> >
> Few comments:
> ==
> 1. How can we specify row filters for multiple tables for a
> publication? Consider a case as below:
It is not possible. Row filter is a per table option. Isn't it clear from the
synopsis? The current design allows different row filter for tables in the same
publication. It is more flexible than a single row filter for a set of tables
(even if we would support such variant, there are some cases where the
condition should be different because the column names are not the same). You
can easily build a CREATE PUBLICATION command that adds the same row filter
multiple times using a DO block or use a similar approach in your favorite
language.

> postgres=# CREATE TABLE tab_rowfilter_1 (a int primary key, b text);
> CREATE TABLE
> postgres=# CREATE TABLE tab_rowfilter_2 (c int primary key);
> CREATE TABLE
> 
> postgres=# CREATE PUBLICATION tap_pub_1 FOR TABLE tab_rowfilter_1,
> tab_rowfilter_2 WHERE (a > 1000 AND b <> 'filtered');
> ERROR:  column "a" does not exist
> LINE 1: ...FOR TABLE tab_rowfilter_1, tab_rowfilter_2 WHERE (a > 1000 A...
> 
>  ^
> 
> postgres=# CREATE PUBLICATION tap_pub_1 FOR TABLE tab_rowfilter_1,
> tab_rowfilter_2  WHERE (c > 1000);
> CREATE PUBLICATION
> 
> It gives an error when I tried to specify the columns corresponding to
> the first relation but is fine for columns for the second relation.
> Then, I tried few more combinations like below but that didn't work.
> CREATE PUBLICATION tap_pub_1 FOR TABLE tab_rowfilter_1 As t1,
> tab_rowfilter_2 As t2 WHERE (t1.a > 1000 AND t1.b <> 'filtered');
> 
> Will users be allowed to specify join conditions among columns from
> multiple tables?
It seems you are envisioning row filter as a publication property instead of a
publication-relation property. Due to the flexibility that the later approach
provides, I decided to use it because it covers more use cases. Regarding
allowing joins, it could possibly slow down a critical path, no? This code path
is executed by every change. If there are interest in the join support, we
might add it in a future patch.

> 2.
> + /*
> + * Although ALTER PUBLICATION grammar allows WHERE clause to be specified
> + * for DROP TABLE action, it doesn't make sense to allow it. We implement
> + * this restriction here, instead of complicating the grammar to enforce
> + * it.
> + */
> + if (stmt->tableAction == DEFELEM_DROP)
> + {
> + ListCell   *lc;
> +
> + foreach(lc, stmt->tables)
> + {
> + PublicationTable *t = lfirst(lc);
> +
> + if (t->whereClause)
> + ereport(ERROR,
> + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> + errmsg("cannot use a WHERE clause when removing table from
> publication \"%s\"",
> + NameStr(pubform->pubname;
> + }
> + }
> 
> Is there a reason to deal with this here separately rather than in the
> ALTER PUBLICATION grammar?
Good question. IIRC the issue is that AlterPublicationStmt->tables has a list
element that was a relation_expr_list and was converted to
publication_table_list. If we share 'tables' with relation_expr_list (for ALTER
PUBLICATION ... DROP TABLE) and publication_table_list (for the other ALTER
PUBLICATION ... ADD|SET TABLE), the OpenTableList() has to know what list
element it is dealing with. I think I came to the conclusion that it is less
uglier to avoid changing OpenTableList() and CloseTableList().

[Doing some experimentation...]

Here is a patch that remove the referred code. It uses 2 distinct list
elements: relation_expr_list for ALTER PUBLICATION ... DROP TABLE and
publication_table_list for for ALTER PUBLICATION ... ADD|SET TABLE. A new
parameter was introduced to deal with the different elements of the list
'tables'.


--
Euler Taveira
EDB   https://www.enterprisedb.com/
From 99d36d0f5cb5f706c73fcdbb05772580f6814fe6 Mon Sep 17 00:00:00 2001
From: Euler Taveira 
Date: Mon, 18 Jan 2021 12:07:51 -0300
Subject: [PATCH] Row filter for logical replication

This feature adds row filter for publication tables. When you define or modify
a publication you can optionally filter rows that does not satisfy a WHERE
condition. It allows you to partially replicate a database or set of tables.
The row filter is per table which means that you can define different row
filters for different tables. A new row filter can be added simply by
informing the WHERE clause after the table name. The WHERE expression must be
enclosed by parentheses.

The WHERE clause should probably contain only columns that are part of the
primary key or that are covered by REPLICA IDENTITY. Otherwise, and DELETEs
won't be replicated. DELETE uses the old row version (that is limited to
primary key or REPLICA IDENTITY) to evaluate the row filter. INSERT and UPDATE
use the new row version to evaluate the row filter, hence, you can use any
column. If the row filter eval

Re: row filtering for logical replication

[Amit Kapila]

On Wed, Mar 31, 2021 at 7:17 AM Euler Taveira  wrote:
>
> On Tue, Mar 30, 2021, at 8:23 AM, Amit Kapila wrote:
>
> On Mon, Mar 29, 2021 at 6:47 PM Euler Taveira  wrote:
> >
> Few comments:
> ==
> 1. How can we specify row filters for multiple tables for a
> publication? Consider a case as below:
>
> It is not possible. Row filter is a per table option. Isn't it clear from the
> synopsis?
>

Sorry, it seems I didn't read it properly earlier, now I got it.

>
> 2.
> + /*
> + * Although ALTER PUBLICATION grammar allows WHERE clause to be specified
> + * for DROP TABLE action, it doesn't make sense to allow it. We implement
> + * this restriction here, instead of complicating the grammar to enforce
> + * it.
> + */
> + if (stmt->tableAction == DEFELEM_DROP)
> + {
> + ListCell   *lc;
> +
> + foreach(lc, stmt->tables)
> + {
> + PublicationTable *t = lfirst(lc);
> +
> + if (t->whereClause)
> + ereport(ERROR,
> + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> + errmsg("cannot use a WHERE clause when removing table from
> publication \"%s\"",
> + NameStr(pubform->pubname;
> + }
> + }
>
> Is there a reason to deal with this here separately rather than in the
> ALTER PUBLICATION grammar?
>
> Good question. IIRC the issue is that AlterPublicationStmt->tables has a list
> element that was a relation_expr_list and was converted to
> publication_table_list. If we share 'tables' with relation_expr_list (for 
> ALTER
> PUBLICATION ... DROP TABLE) and publication_table_list (for the other ALTER
> PUBLICATION ... ADD|SET TABLE), the OpenTableList() has to know what list
> element it is dealing with. I think I came to the conclusion that it is less
> uglier to avoid changing OpenTableList() and CloseTableList().
>
> [Doing some experimentation...]
>
> Here is a patch that remove the referred code.
>

Thanks, few more comments:
1. In pgoutput_change, we are always sending schema even though we
don't send actual data because of row filters. It may not be a problem
in many cases but I guess for some odd cases we can avoid sending
extra information.

2. In get_rel_sync_entry(), we are caching the qual for rel_sync_entry
even though we won't publish it which seems unnecessary?

3.
@@ -1193,5 +1365,11 @@ rel_sync_cache_publication_cb(Datum arg, int
cacheid, uint32 hashvalue)
  entry->pubactions.pubupdate = false;
  entry->pubactions.pubdelete = false;
  entry->pubactions.pubtruncate = false;
+
+ if (entry->qual != NIL)
+ list_free_deep(entry->qual);

Seeing one previous comment in this thread [1], I am wondering if
list_free_deep is enough here?

4. Can we write explicitly in the docs that row filters won't apply
for Truncate operation?

5. Getting some whitespace errors:
git am 
/d/PostgreSQL/Patches/logical_replication/row_filter/v14-0001-Row-filter-for-logical-replication.patch
.git/rebase-apply/patch:487: trailing whitespace.

warning: 1 line adds whitespace errors.
Applying: Row filter for logical replication


[1] - 
https://www.postgresql.org/message-id/20181123161933.jpepibtyayflz2xg%40alvherre.pgsql

-- 
With Regards,
Amit Kapila.

Re: row filtering for logical replication

[Andres Freund]

Hi,

As far as I can tell you have not *AT ALL* addressed that it is *NOT
SAFE* to evaluate arbitrary expressions from within an output
plugin. Despite that having been brought up multiple times.


> +static ExprState *
> +pgoutput_row_filter_prepare_expr(Node *rfnode, EState *estate)
> +{
> + ExprState  *exprstate;
> + Oid exprtype;
> + Expr   *expr;
> +
> + /* Prepare expression for execution */
> + exprtype = exprType(rfnode);
> + expr = (Expr *) coerce_to_target_type(NULL, rfnode, exprtype, BOOLOID, 
> -1, COERCION_ASSIGNMENT, COERCE_IMPLICIT_CAST, -1);
> +
> + if (expr == NULL)
> + ereport(ERROR,
> + (errcode(ERRCODE_CANNOT_COERCE),
> +  errmsg("row filter returns type %s that cannot 
> be coerced to the expected type %s",
> + format_type_be(exprtype),
> + format_type_be(BOOLOID)),
> +  errhint("You will need to rewrite the row 
> filter.")));
> +
> + exprstate = ExecPrepareExpr(expr, estate);
> +
> + 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 inline bool
> +pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
> +{
> + Datum   ret;
> + boolisnull;
> +
> + 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(Relation relation, HeapTuple oldtuple, HeapTuple 
> newtuple, List *rowfilter)
> +{
> + TupleDesc   tupdesc;
> + EState *estate;
> + ExprContext *ecxt;
> + MemoryContext oldcxt;
> + ListCell   *lc;
> + boolresult = true;
> +
> + /* Bail out if there is no row filter */
> + if (rowfilter == NIL)
> + return true;
> +
> + elog(DEBUG3, "table \"%s.%s\" has row filter",
> +  
> get_namespace_name(get_rel_namespace(RelationGetRelid(relation))),
> +  get_rel_name(relation->rd_id));
> +
> + tupdesc = RelationGetDescr(relation);
> +
> + estate = create_estate_for_relation(relation);
> +
> + /* Prepare context per tuple */
> + ecxt = GetPerTupleExprContext(estate);
> + oldcxt = MemoryContextSwitchTo(estate->es_query_cxt);
> + ecxt->ecxt_scantuple = ExecInitExtraTupleSlot(estate, tupdesc, 
> &TTSOpsHeapTuple);
> + MemoryContextSwitchTo(oldcxt);
> +
> + ExecStoreHeapTuple(newtuple ? newtuple : oldtuple, 
> ecxt->ecxt_scantuple, false);
> + /*
> +  * If the subscription has multiple publications and the same table has 
> a
> +  * different row filter in these publications, all row filters must be
> +  * matched in order to replicate this change.
> +  */
> + foreach(lc, rowfilter)
> + {
> + Node   *rfnode = (Node *) lfirst(lc);
> + ExprState  *exprstate;
> +
> + /* Prepare for expression execution */
> + exprstate = pgoutput_row_filter_prepare_expr(rfnode, estate);
> +
> + /* Evaluates row filter */
> + result = pgoutput_row_filter_exec_expr(exprstate, ecxt);

Also, this still seems like an *extremely* expensive thing to do for
each tuple. It'll often be *vastly* faster to just send the data than to
the other side.

This just cannot be done once per tuple. It has to be cached.

I don't see how these issues can be addressed in the next 7 days,
therefore I think this unfortunately needs to be marked as returned with
feedback.

Greetings,

Andres Freund

RE: row filtering for logical replication

[tanghy.f...@fujitsu.com]

On Friday, November 5, 2021 1:14 PM, Peter Smith  wrote:
> 
> PSA new set of v37* patches.
> 

Thanks for your patch. I have a problem when using this patch.

The document about "create publication" in patch says:

   The WHERE clause should contain only columns that are
   part of the primary key or be covered  by REPLICA
   IDENTITY otherwise, DELETE operations will not
   be replicated.

But I tried this patch, the columns which could be contained in WHERE clause 
must be
covered by REPLICA IDENTITY, but it doesn't matter if they are part of the 
primary key. 
(We can see it in Case 4 of publication.sql, too.) So maybe we should modify 
the document.

Regards
Tang

RE: row filtering for logical replication

[houzj.f...@fujitsu.com]

On Fri, Nov 5, 2021 4:49 PM Amit Kapila  wrote:
> On Fri, Nov 5, 2021 at 10:44 AM Peter Smith  wrote:
> >
> > PSA new set of v37* patches.
> 3.
> - | ColId
> + | ColId OptWhereClause
>   {
>   $$ = makeNode(PublicationObjSpec);
>   $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
> - $$->name = $1;
> + if ($2)
> + {
> + $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation =
> + makeRangeVar(NULL, $1, @1); $$->pubtable->whereClause = $2; } else {
> + $$->name = $1; }
> 
> Again this doesn't appear to be the right way. I think this should be handled 
> at
> a later point.

I think the difficulty to handle this at a later point is that we need to make
sure we don't lose the whereclause. Currently, we can only save the whereclause
in PublicationTable structure and the PublicationTable is only used for TABLE,
but '| ColId' can be used for either a SCHEMA or TABLE. We cannot distinguish
the actual type at this stage, so we always need to save the whereclause if
it's NOT NULL.

I think the possible approaches to delay this check are:

(1) we can delete the PublicationTable structure and put all the vars(relation,
whereclause) in PublicationObjSpec. In this approach, we don't need check if
the whereclause is NULL in the '| ColId', we can check this at a later point.

Or

(2) Add a new pattern for whereclause in PublicationObjSpec:

The change could be:

PublicationObjSpec:
...
| ColId
... 
+ | ColId WHERE '(' a_expr ')'
+ {
+ $$ = makeNode(PublicationObjSpec);
+ $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
+ $$->pubtable = makeNode(PublicationTable);
+ $$->pubtable->relation = makeRangeVar(NULL, $1, @1);
+ $$->pubtable->whereClause = $2;
+ }

In this approach, we also don't need the "if ($2)" check.

What do you think ?

Best regards,
Hou zj

Re: row filtering for logical replication

[Peter Smith]

On Fri, Nov 5, 2021 at 7:49 PM Amit Kapila  wrote:
>
> 2.
> +preprocess_pubobj_list(List *pubobjspec_list, core_yyscan_t
> yyscanner, bool alter_drop)
>  {
>   ListCell   *cell;
>   PublicationObjSpec *pubobj;
> @@ -17341,7 +17359,15 @@ 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)
> +
> + /* cannot use WHERE w-filter for DROP TABLE from publications */
> + if (pubobj->pubtable && pubobj->pubtable->whereClause && alter_drop)
> + ereport(ERROR,
> + errcode(ERRCODE_SYNTAX_ERROR),
> + errmsg("invalid use of WHERE row-filter in ALTER PUBLICATION ... DROP 
> TABLE"),
> + parser_errposition(pubobj->location));
> +
>
> This change looks a bit ad-hoc to me. Can we handle this at a later
> point of time in publicationcmds.c?
>

Fixed in v38-0001 [1].

--
[1] 
https://www.postgresql.org/message-id/CAHut%2BPvWCS%2BW_OLV60AZJucY1RFpkXS%3DhfvYWwpwyMvifdJxiQ%40mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Australia

Re: row filtering for logical replication

[Peter Smith]

On Mon, Nov 8, 2021 at 5:53 PM houzj.f...@fujitsu.com
 wrote:
>
> On Fri, Nov 5, 2021 1:14 PM Peter Smith  wrote:
> > PSA new set of v37* patches.
>
> Thanks for updating the patches.
> Few comments:
>
> 1) v37-0001
>
> I think it might be better to also show the filter expression in '\d+
> tablename' command after publication description.
>

Fixed in v38-0001 [1]

--
[1] 
https://www.postgresql.org/message-id/CAHut%2BPvWCS%2BW_OLV60AZJucY1RFpkXS%3DhfvYWwpwyMvifdJxiQ%40mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Austrlalia

Re: row filtering for logical replication

[Peter Smith]

On Tue, Nov 9, 2021 at 2:03 PM tanghy.f...@fujitsu.com
 wrote:
>
> On Friday, November 5, 2021 1:14 PM, Peter Smith  
> wrote:
> >
> > PSA new set of v37* patches.
> >
>
> Thanks for your patch. I have a problem when using this patch.
>
> The document about "create publication" in patch says:
>
>The WHERE clause should contain only columns that are
>part of the primary key or be covered  by REPLICA
>IDENTITY otherwise, DELETE operations will not
>be replicated.
>
> But I tried this patch, the columns which could be contained in WHERE clause 
> must be
> covered by REPLICA IDENTITY, but it doesn't matter if they are part of the 
> primary key.
> (We can see it in Case 4 of publication.sql, too.) So maybe we should modify 
> the document.
>

PG Docs is changed in v38-0004 [1]. Please check if it is OK.

--
[1] 
https://www.postgresql.org/message-id/CAHut%2BPvWCS%2BW_OLV60AZJucY1RFpkXS%3DhfvYWwpwyMvifdJxiQ%40mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Australia

Re: row filtering for logical replication

[Peter Smith]

On Thu, Nov 4, 2021 at 2:21 PM houzj.f...@fujitsu.com
 wrote:
>
> 3)
>
> +   oldctx = 
> MemoryContextSwitchTo(CacheMemoryContext);
> +   rfnode = 
> stringToNode(TextDatumGetCString(rfdatum));
> +   exprstate = 
> pgoutput_row_filter_init_expr(rfnode);
> +   entry->exprstates = 
> lappend(entry->exprstates, exprstate);
> +   MemoryContextSwitchTo(oldctx);
> +   }
>
> Currently in the patch, it save and execute each expression separately. I was
> thinking it might be better if we can use "AND" to combine all the expressions
> into one expression, then we can initialize and optimize the final expression
> and execute it only once.
>

Fixed in v38-0003 [1].

--
[1] 
https://www.postgresql.org/message-id/CAHut%2BPvWCS%2BW_OLV60AZJucY1RFpkXS%3DhfvYWwpwyMvifdJxiQ%40mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Australia

RE: row filtering for logical replication

[houzj.f...@fujitsu.com]

On Thur, Nov 4, 2021 10:47 AM Peter Smith  wrote:
> PROPOSAL
> 
> I propose that we change the way duplicate tables are processed to make it so
> that it is always the *last* one that takes effect (instead of the *first* 
> one). AFAIK
> doing this won't affect any current PG behaviour, but doing this will let the 
> new
> row-filter feature work in a consistent/predictable/sane way.
> 
> Thoughts?

Last one take effect sounds reasonable to me.

OTOH, I think we should make the behavior here consistent with Column Filter
Patch in another thread. IIRC, in the current column filter patch, only the
first one's filter takes effect. So, maybe better to get Rahila and Alvaro's
thoughts on this.

Best regards,
Hou zj

Re: row filtering for logical replication

[Amit Kapila]

On Tue, Nov 9, 2021 at 2:22 PM houzj.f...@fujitsu.com
 wrote:
>
> On Fri, Nov 5, 2021 4:49 PM Amit Kapila  wrote:
> > On Fri, Nov 5, 2021 at 10:44 AM Peter Smith  wrote:
> > >
> > > PSA new set of v37* patches.
> > 3.
> > - | ColId
> > + | ColId OptWhereClause
> >   {
> >   $$ = makeNode(PublicationObjSpec);
> >   $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
> > - $$->name = $1;
> > + if ($2)
> > + {
> > + $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation =
> > + makeRangeVar(NULL, $1, @1); $$->pubtable->whereClause = $2; } else {
> > + $$->name = $1; }
> >
> > Again this doesn't appear to be the right way. I think this should be 
> > handled at
> > a later point.
>
> I think the difficulty to handle this at a later point is that we need to make
> sure we don't lose the whereclause. Currently, we can only save the 
> whereclause
> in PublicationTable structure and the PublicationTable is only used for TABLE,
> but '| ColId' can be used for either a SCHEMA or TABLE. We cannot distinguish
> the actual type at this stage, so we always need to save the whereclause if
> it's NOT NULL.
>

I see your point. But, I think we can add some comments here
indicating that the user might have mistakenly given where clause with
some schema which we will identify later and give an appropriate
error. Then, in preprocess_pubobj_list(), identify if the user has
given the where clause with schema name and give an appropriate error.

> I think the possible approaches to delay this check are:
>
> (1) we can delete the PublicationTable structure and put all the 
> vars(relation,
> whereclause) in PublicationObjSpec. In this approach, we don't need check if
> the whereclause is NULL in the '| ColId', we can check this at a later point.
>

Yeah, we can do this but I don't think it will reduce any checks later
to identify if the user has given where clause only for tables. So,
let's keep this structure around as that will at least keep all things
related to the table together in one structure.

> Or
>
> (2) Add a new pattern for whereclause in PublicationObjSpec:
>
> The change could be:
>
> PublicationObjSpec:
> ...
> | ColId
> ...
> + | ColId WHERE '(' a_expr ')'
> + {
> + $$ = makeNode(PublicationObjSpec);
> + $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
> + $$->pubtable = makeNode(PublicationTable);
> + $$->pubtable->relation = makeRangeVar(NULL, $1, @1);
> + $$->pubtable->whereClause = $2;
> + }
>
> In this approach, we also don't need the "if ($2)" check.
>

This seems redundant and we still need same checks later to see if the
where clause is given with the table object.

-- 
With Regards,
Amit Kapila.

RE: row filtering for logical replication

[houzj.f...@fujitsu.com]

On Wed, Nov 10, 2021 10:48 AM Amit Kapila  wrote:
> On Tue, Nov 9, 2021 at 2:22 PM houzj.f...@fujitsu.com wrote:
> >
> > On Fri, Nov 5, 2021 4:49 PM Amit Kapila  wrote:
> > > On Fri, Nov 5, 2021 at 10:44 AM Peter Smith  wrote:
> > > >
> > > > PSA new set of v37* patches.
> > > 3.
> > > - | ColId
> > > + | ColId OptWhereClause
> > >   {
> > >   $$ = makeNode(PublicationObjSpec);
> > >   $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
> > > - $$->name = $1;
> > > + if ($2)
> > > + {
> > > + $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation
> > > + = makeRangeVar(NULL, $1, @1); $$->pubtable->whereClause = $2; }
> > > + else { $$->name = $1; }
> > >
> > > Again this doesn't appear to be the right way. I think this should
> > > be handled at a later point.
> >
> > I think the difficulty to handle this at a later point is that we need
> > to make sure we don't lose the whereclause. Currently, we can only
> > save the whereclause in PublicationTable structure and the
> > PublicationTable is only used for TABLE, but '| ColId' can be used for
> > either a SCHEMA or TABLE. We cannot distinguish the actual type at
> > this stage, so we always need to save the whereclause if it's NOT NULL.
> >
> 
> I see your point. But, I think we can add some comments here indicating that
> the user might have mistakenly given where clause with some schema which we
> will identify later and give an appropriate error. Then, in
> preprocess_pubobj_list(), identify if the user has given the where clause with
> schema name and give an appropriate error.
> 

OK, IIRC, in this approach, we need to set both $$->name and $$->pubtable in
'| ColId OptWhereClause'. And In preprocess_pubobj_list, we can add some check
if both name and pubtable is NOT NULL.

the grammar code could be:

| ColId OptWhereClause
{
$$ = makeNode(PublicationObjSpec);
$$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;

$$->name = $1;
+   /* xxx */
+   $$->pubtable = makeNode(PublicationTable);
+   $$->pubtable->relation = makeRangeVar(NULL, $1, @1);
+   $$->pubtable->whereClause = $2;
$$->location = @1;
}

preprocess_pubobj_list
...
else if (pubobj->pubobjtype == PUBLICATIONOBJ_REL_IN_SCHEMA ||
pubobj->pubobjtype == PUBLICATIONOBJ_CURRSCHEMA)
{
...
+if (pubobj->name &&
+(!pubobj->pubtable || !pubobj->pubtable->whereClause))
pubobj->pubobjtype = PUBLICATIONOBJ_REL_IN_SCHEMA;
else if (!pubobj->name && !pubobj->pubtable)
pubobj->pubobjtype = PUBLICATIONOBJ_CURRSCHEMA;
else
ereport(ERROR,
errcode(ERRCODE_SYNTAX_ERROR),
errmsg("invalid schema name at or near"),
parser_errposition(pubobj->location));
}


Best regards,
Hou zj

Re: row filtering for logical replication

[Peter Smith]

On Wed, Nov 10, 2021 at 4:57 PM houzj.f...@fujitsu.com
 wrote:
>
> On Wed, Nov 10, 2021 10:48 AM Amit Kapila  wrote:
> > On Tue, Nov 9, 2021 at 2:22 PM houzj.f...@fujitsu.com wrote:
> > >
> > > On Fri, Nov 5, 2021 4:49 PM Amit Kapila  wrote:
> > > > On Fri, Nov 5, 2021 at 10:44 AM Peter Smith  
> > > > wrote:
> > > > >
> > > > > PSA new set of v37* patches.
> > > > 3.
> > > > - | ColId
> > > > + | ColId OptWhereClause
> > > >   {
> > > >   $$ = makeNode(PublicationObjSpec);
> > > >   $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
> > > > - $$->name = $1;
> > > > + if ($2)
> > > > + {
> > > > + $$->pubtable = makeNode(PublicationTable); $$->pubtable->relation
> > > > + = makeRangeVar(NULL, $1, @1); $$->pubtable->whereClause = $2; }
> > > > + else { $$->name = $1; }
> > > >
> > > > Again this doesn't appear to be the right way. I think this should
> > > > be handled at a later point.
> > >
> > > I think the difficulty to handle this at a later point is that we need
> > > to make sure we don't lose the whereclause. Currently, we can only
> > > save the whereclause in PublicationTable structure and the
> > > PublicationTable is only used for TABLE, but '| ColId' can be used for
> > > either a SCHEMA or TABLE. We cannot distinguish the actual type at
> > > this stage, so we always need to save the whereclause if it's NOT NULL.
> > >
> >
> > I see your point. But, I think we can add some comments here indicating that
> > the user might have mistakenly given where clause with some schema which we
> > will identify later and give an appropriate error. Then, in
> > preprocess_pubobj_list(), identify if the user has given the where clause 
> > with
> > schema name and give an appropriate error.
> >
>
> OK, IIRC, in this approach, we need to set both $$->name and $$->pubtable in
> '| ColId OptWhereClause'. And In preprocess_pubobj_list, we can add some check
> if both name and pubtable is NOT NULL.
>
> the grammar code could be:
>
> | ColId OptWhereClause
> {
> $$ = makeNode(PublicationObjSpec);
> $$->pubobjtype = PUBLICATIONOBJ_CONTINUATION;
>
> $$->name = $1;
> +   /* xxx */
> +   $$->pubtable = makeNode(PublicationTable);
> +   $$->pubtable->relation = makeRangeVar(NULL, $1, @1);
> +   $$->pubtable->whereClause = $2;
> $$->location = @1;
> }
>
> preprocess_pubobj_list
> ...
> else if (pubobj->pubobjtype == PUBLICATIONOBJ_REL_IN_SCHEMA ||
> pubobj->pubobjtype == PUBLICATIONOBJ_CURRSCHEMA)
> {
> ...
> +if (pubobj->name &&
> +(!pubobj->pubtable || !pubobj->pubtable->whereClause))
> pubobj->pubobjtype = PUBLICATIONOBJ_REL_IN_SCHEMA;
> else if (!pubobj->name && !pubobj->pubtable)
> pubobj->pubobjtype = PUBLICATIONOBJ_CURRSCHEMA;
> else
> ereport(ERROR,
> errcode(ERRCODE_SYNTAX_ERROR),
> errmsg("invalid schema name at or near"),
> parser_errposition(pubobj->location));
> }
>

Hi Hou-san. Actually, I have already implemented this part according
to my understanding of Amit's suggestion and it seems to be working
well.

Please wait for v39-0001, then feel free to post review comments about
it if you think there are still problems.

--
Kind Regards,
Peter Smith.
Fujitsu Australia

Re: row filtering for logical replication

[Peter Smith]

On Mon, Nov 8, 2021 at 5:53 PM houzj.f...@fujitsu.com
 wrote:
>
> 3) v37-0005
>
> - no parse nodes of any kind other than Var, OpExpr, Const, BoolExpr, FuncExpr
>
> I think there could be other node type which can also be considered as simple
> expression, for exmaple T_NullIfExpr.

The current walker restrictions are from a previously agreed decision
by Amit/Tomas [1] and from an earlier suggestion from Andres [2] to
keep everything very simple for a first version.

Yes, you are right, there might be some additional node types that
might be fine, but at this time I don't want to add anything different
without getting their approval to do so. Anyway, additions like this
are all candidates for a future version of this row-filter feature.

>
> Personally, I think it's natural to only check the IMMUTABLE and
> whether-user-defined in the new function rowfilter_walker. We can keep the
> other row-filter errors which were thrown for EXPR_KIND_PUBLICATION_WHERE in
> the 0001 patch.
>

YMMV. IMO it is much more convenient for all the filter validations to
be centralized just in one walker function instead of scattered all
over the place like they were in the 0001 patch.

-
[1] 
https://www.postgresql.org/message-id/CAA4eK1%2BXoD49bz5-2TtiD0ugq4PHSRX2D1sLPR_X4LNtdMc4OQ%40mail.gmail.com
[2] 
https://www.postgresql.org/message-id/20210128022032.eq2qqc6zxkqn5syt%40alap3.anarazel.de

Kind Regards,
Peter Smith.
Fujitsu Australia