Re: Handle infinite recursion in logical replication setup

On Wed, Mar 30, 2022 at 7:22 PM Ashutosh Bapat
<ashutosh(dot)bapat(dot)oss(at)gmail(dot)com> wrote:
>
> Hi Vignesh,
> Some review comments on 0001
>
> On Wed, Mar 16, 2022 at 11:17 AM vignesh C <vignesh21(at)gmail(dot)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.

I have used filter_by_origin_cb for the implementation, removed
filter_remote_origin_cb

> - <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"?

Modified

>
> + <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".

Modified

> @@ -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?

Yes you are right, Modified

> 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.

Modified

> --- 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.

Modified

> --- 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.

This is set in ApplyWorkerMain before calling libpqrcv_startstreaming.
This will be used in building the START_REPLICATION command.
Thanks for the comments, the attached v6 patch has the changes for the same.

Regards,
Vignesh