RE: row filtering for logical replication

On Thu, Jan 4, 2022 at 00:54 PM Peter Smith <smithpb2250@gmail.com> wrote:
> Modified in v58 [1] as suggested
Thanks for updating the patches.
A few comments about v58-0001 and v58-0002.

v58-0001
1.
How about modifying the following loop in copy_table by using for_each_from
instead of foreach?
Like the invocation of for_each_from in function get_rule_expr.
from:
        if (qual != NIL)
        {
            ListCell   *lc;
            bool        first = true;

            appendStringInfoString(&cmd, " WHERE ");
            foreach(lc, qual)
            {
                char       *q = strVal(lfirst(lc));

                if (first)
                    first = false;
                else
                    appendStringInfoString(&cmd, " OR ");
                appendStringInfoString(&cmd, q);
            }
            list_free_deep(qual);
        }
change to:
        if (qual != NIL)
        {
            ListCell   *lc;
            char       *q = strVal(linitial(qual));

            appendStringInfo(&cmd, " WHERE %s", q);
            for_each_from(lc, qual, 1)
            {
                q = strVal(lfirst(lc));
                appendStringInfo(&cmd, " OR %s", q);
            }
            list_free_deep(qual);
        }

2.
I find the API of get_rel_sync_entry is modified. 
-get_rel_sync_entry(PGOutputData *data, Oid relid)
+get_rel_sync_entry(PGOutputData *data, Relation relation)
It looks like just moving the invocation of RelationGetRelid from outside into
function get_rel_sync_entry. I am not sure whether this modification is
necessary to this feature or not.

v58-0002
1.
In function pgoutput_row_filter_init, if no_filter is set, I think we do not
need to add row filter to list(rfnodes).
So how about changing three conditions when add row filter to rfnodes like this:
-                    if (pub->pubactions.pubinsert)
+                    if (pub->pubactions.pubinsert && !no_filter[idx_ins])
                    {
                        rfnode = stringToNode(TextDatumGetCString(rfdatum));
                        rfnodes[idx_ins] = lappend(rfnodes[idx_ins], rfnode);
                    }

Regards,
Wang wei