Re: Handle infinite recursion in logical replication setup

Hi Vignesh,
Some review comments on 0001

On Wed, Mar 16, 2022 at 11:17 AM vignesh C <vignesh21@gmail.com> wrote:

>
> The changes for the same are available int the v5 patch available at [1].
> [1] - https://www.postgresql.org/message-id/CALDaNm3wCf0YcvVo%2BgHMGpupk9K6WKJxCyLUvhPC2GkPKRZUWA%40mail.gmail.com
>

     cb->truncate_cb = pg_decode_truncate;
     cb->commit_cb = pg_decode_commit_txn;
     cb->filter_by_origin_cb = pg_decode_filter;
+    cb->filter_remote_origin_cb = pg_decode_filter_remote_origin;

Why do we need a new hook? Can we use filter_by_origin_cb? Also it looks like
implementation of pg_decode_filter and pg_decode_filter_remote_origin is same,
unless my eyes are deceiving me.

-      <literal>binary</literal>, <literal>streaming</literal>, and
-      <literal>disable_on_error</literal>.
+      <literal>binary</literal>, <literal>streaming</literal>,
+      <literal>disable_on_error</literal> and
+      <literal>publish_local_only</literal>.

"publish_local_only" as a "subscription" option looks odd. Should it be
"subscribe_local_only"?


+       <varlistentry>
+        <term><literal>publish_local_only</literal>
(<type>boolean</type>)</term>
+        <listitem>
+         <para>
+          Specifies whether the subscription should subscribe only to the
+          locally generated changes or subscribe to both the locally generated
+          changes and the replicated changes that was generated from other
+          nodes in the publisher. The default is <literal>false</literal>.

This description to uses verb "subscribe" instead of "publish".

@@ -936,6 +951,13 @@ AlterSubscription(ParseState *pstate,
AlterSubscriptionStmt *stmt,
                         = true;
                 }

+                if (IsSet(opts.specified_opts, SUBOPT_PUBLISH_LOCAL_ONLY))
+                {
+                    values[Anum_pg_subscription_sublocal - 1] =
+                        BoolGetDatum(opts.streaming);

should this be opts.sublocal?

     cb->commit_prepared_cb = pgoutput_commit_prepared_txn;
     cb->rollback_prepared_cb = pgoutput_rollback_prepared_txn;
     cb->filter_by_origin_cb = pgoutput_origin_filter;
+    cb->filter_remote_origin_cb = pgoutput_remote_origin_filter;

pgoutput_origin_filter just returns false now. I think we should just enhance
that function and reuse the callback, instead of adding a new callback.

--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -99,6 +99,8 @@ typedef struct LogicalDecodingContext
      */
     bool        twophase_opt_given;

+    bool        only_local;        /* publish only locally generated data */
+

If we get rid of the new callback, we won't need this new member as well.
Anyway the filtering happens within the output plugin. There is nothing that
core is required to do here.

--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -183,6 +183,7 @@ typedef struct
             bool        streaming;    /* Streaming of large transactions */
             bool        twophase;    /* Streaming of two-phase transactions at
                                      * prepare time */
+            bool        only_local; /* publish only locally generated data */

Are we using this anywhere. I couldn't spot any usage of this member. I might
be missing something though.



-- 
Best Wishes,
Ashutosh Bapat