Re: row filtering for logical replication

On Tue, Dec 21, 2021 at 2:29 PM Peter Smith <smithpb2250@gmail.com> wrote:
>
> Here is the v51* patch set:
>
> Main changes from Euler's v50* are
> 1. Most of Euler's "fixes" patches are now merged back in
> 2. Patches are then merged per Amit's suggestion [Amit 20/12]
> 3. Some other review comments are addressed
>
> ~~
>
> Phase 1 - Merge the Euler fixes
> ===============================
>
> v51-0001 (main) <== v50-0001 (main 0001) + v50-0002 (fixes 0001)
> - OK, accepted and merged all the "fixes" back into the 0001
> - (fixed typos)
> - There is a slight disagreement with what the PG docs say about
> NULLs, but I will raise a separate comment on -hackers later
> (meanwhile, the current PG docs text is from the Euler patch)
>
> v51-0002 (validation) <== v50-0003 (validation 0002) + v50-0004 (fixes 0002)
> - OK, accepted and merges all these "fixes" back into the 0002
> - (fixed typo)
>
> v51-0003 (new/old) <== v50-0005 (new/old 0003)
> - REVERTED to the v49 version of this patch!
> - Please see Ajin's explanation why the v49 code was required [Ajin 21/12]
>
> v51-0004 (tab-complete + dump) <== v50-0006 (tab-complete + dump 0004)
> - No changes
>
> v51-0005 (for all tables) <== v50-0007 (for all tables 0005) +
> v50-0008 (fixes 0005)
> - OK, accepted and merged most of these "fixes" back into the 0005
> - (fixed typo/grammar)
> - Amit requested we not use Euler's new tablesync SQL just yet [Amit 21/12]
>
>
> Phase 2 - Merge main (Phase 1) patches per Amit suggestion
> ================================================
>
> v51-0001 (main) <== v51-0001 (main) + v51-0002 (validation) + v51-0005
> (for all tables)
> - (also combined all the commit comments)
>
> v51-0002 (new/old) <== v51-0003 (new/old)
>
> v51-0005 (tab-complete + dump) <== v51-0004
>
>
> Review comments (details)
> =========================
>
> v51-0001 (main)
> - Addressed review comments from Amit. [Amit 20/12] #1,#2,#3,#4
>
> v51-0002 (new/old tuple)
> - Includes a patch from Greg (provided internally)
>
> v51-0003 (tab, dump)
> - No changes
>
> ------
> [Amit 20/12] https://www.postgresql.org/message-id/CAA4eK1JJgfEPKYvQAcwGa5jjuiUiQRQZ0Pgo-HF0KFHh-jyNQQ%40mail.gmail.com
> [Ajin 21/12] https://www.postgresql.org/message-id/CAFPTHDbfpPNh3GLGjySS%3DAuRfbQPQFNvfiyG1GDQW2kz1yT7Og%40mail.gmail.com
> [Amit 21/12] https://www.postgresql.org/message-id/CAA4eK1KwoA5k8v9z9e4ZPN_X%3D1GAmQmsWyauFwZpKiSHqy6eZA%40mail.gmail.com

Few comments:
1) list_free(schemarelids) is called inside if and and outside if in
break case. Can we move it above continue so that it does not gets
called in the break case:
+                       schemarelids =
GetAllSchemaPublicationRelations(pub->oid,
+
                                                 pub->pubviaroot ?
+
                                                 PUBLICATION_PART_ROOT
:
+

PUBLICATION_PART_LEAF);
+                       if (list_member_oid(schemarelids, entry->relid))
+                       {
+                               if (pub->pubactions.pubinsert)
+                                       no_filter[idx_ins] = true;
+                               if (pub->pubactions.pubupdate)
+                                       no_filter[idx_upd] = true;
+                               if (pub->pubactions.pubdelete)
+                                       no_filter[idx_del] = true;
+
+                               list_free(schemarelids);
+
+                               /* Quick exit loop if all pubactions
have no row-filter. */
+                               if (no_filter[idx_ins] &&
no_filter[idx_upd] && no_filter[idx_del])
+                                       break;
+
+                               continue;
+                       }
+                       list_free(schemarelids);

2) create_edata_for_relation also is using similar logic, can it also
call this function to reduce duplicate code
+static EState *
+create_estate_for_relation(Relation rel)
+{
+       EState                  *estate;
+       RangeTblEntry   *rte;
+
+       estate = CreateExecutorState();
+
+       rte = makeNode(RangeTblEntry);
+       rte->rtekind = RTE_RELATION;
+       rte->relid = RelationGetRelid(rel);
+       rte->relkind = rel->rd_rel->relkind;
+       rte->rellockmode = AccessShareLock;
+       ExecInitRangeTable(estate, list_make1(rte));
+
+       estate->es_output_cid = GetCurrentCommandId(false);
+
+       return estate;
+}

3) In one place select is in lower case, it can be changed to upper case
+               resetStringInfo(&cmd);
+               appendStringInfo(&cmd,
+                                                "SELECT DISTINCT
pg_get_expr(prqual, prrelid) "
+                                                "  FROM pg_publication p "
+                                                "  INNER JOIN
pg_publication_rel pr ON (p.oid = pr.prpubid) "
+                                                "    WHERE pr.prrelid
= %u AND p.pubname IN ( %s ) "
+                                                "    AND NOT (select
bool_or(puballtables) "
+                                                "      FROM pg_publication "
+                                                "      WHERE pubname
in ( %s )) "
+                                                "    AND (SELECT count(1)=0 "
+                                                "      FROM
pg_publication_namespace pn, pg_class c "
+                                                "      WHERE c.oid =
%u AND c.relnamespace = pn.pnnspid)",
+                                               lrel->remoteid,
+                                               pub_names.data,
+                                               pub_names.data,
+                                               lrel->remoteid);

4) we could run pgindent once to fix indentation issues
+       /* Cache ExprState using CacheMemoryContext. */
+       Assert(CurrentMemoryContext = CacheMemoryContext);
+
+       /* Prepare expression for execution */
+       exprtype = exprType(rfnode);
+       expr = (Expr *) coerce_to_target_type(NULL, rfnode, exprtype,
BOOLOID, -1, COERCION_ASSIGNMENT, COERCE_IMPLICIT_CAST, -1);
+
+       if (expr == NULL)

5) Should we mention user should take care of deletion of row filter
records after table sync is done.
+   ALL TABLES</literal> or <literal>FOR ALL TABLES IN
SCHEMA</literal> publication,
+   then all other <literal>WHERE</literal> clauses (for the same
publish operation)
+   become redundant.
+   If the subscriber is a <productname>PostgreSQL</productname>
version before 15
+   then any row filtering is ignored during the initial data
synchronization phase.

6) Should this be an Assert, since this will be taken care earlier by
GetTransformedWhereClause->transformWhereClause->coerce_to_boolean:
+       expr = (Expr *) coerce_to_target_type(NULL, rfnode, exprtype,
BOOLOID, -1, COERCION_ASSIGNMENT, COERCE_IMPLICIT_CAST, -1);
+
+       if (expr == NULL)
+               ereport(ERROR,
+                               (errcode(ERRCODE_CANNOT_COERCE),
+                                errmsg("row filter returns type %s
that cannot be cast to the expected type %s",
+                                               format_type_be(exprtype),
+                                               format_type_be(BOOLOID)),
+                                errhint("You will need to rewrite the
row filter.")));

7) This code is present in 3 places, can we make it a function or
macro and use it:
+                               if (pub->pubactions.pubinsert)
+                                       no_filter[idx_ins] = true;
+                               if (pub->pubactions.pubupdate)
+                                       no_filter[idx_upd] = true;
+                               if (pub->pubactions.pubdelete)
+                                       no_filter[idx_del] = true;
+
+                               /* Quick exit loop if all pubactions
have no row-filter. */
+                               if (no_filter[idx_ins] &&
no_filter[idx_upd] && no_filter[idx_del])
+                                       break;
+
+                               continue;

8) Can the below transformation be done just before the
values[Anum_pg_publication_rel_prqual - 1] is set to reduce one of the
checks:
+       if (pri->whereClause != NULL)
+       {
+               /* Set up a ParseState to parse with */
+               pstate = make_parsestate(NULL);
+
+               /*
+                * Get the transformed WHERE clause, of boolean type,
with necessary
+                * collation information.
+                */
+               whereclause = GetTransformedWhereClause(pstate, pri, true);
+       }

        /* Form a tuple. */
        memset(values, 0, sizeof(values));
@@ -328,6 +376,12 @@ publication_add_relation(Oid pubid,
PublicationRelInfo *targetrel,
        values[Anum_pg_publication_rel_prrelid - 1] =
                ObjectIdGetDatum(relid);

+       /* Add qualifications, if available */
+       if (whereclause)
+               values[Anum_pg_publication_rel_prqual - 1] =
CStringGetTextDatum(nodeToString(whereclause));
+       else
+               nulls[Anum_pg_publication_rel_prqual - 1] = true;
+

Like:
/* Add qualifications, if available */
if (pri->whereClause != NULL)
{
/* Set up a ParseState to parse with */
pstate = make_parsestate(NULL);

/*
* Get the transformed WHERE clause, of boolean type, with necessary
* collation information.
*/
whereclause = GetTransformedWhereClause(pstate, pri, true);

/*
* Walk the parse-tree of this publication row filter expression and
* throw an error if anything not permitted or unexpected is
* encountered.
*/
rowfilter_walker(whereclause, targetrel);
values[Anum_pg_publication_rel_prqual - 1] =
CStringGetTextDatum(nodeToString(whereclause));
}
else
nulls[Anum_pg_publication_rel_prqual - 1] = true;

Regards,
Vignesh