Re: Handle infinite recursion in logical replication setup

On Wed, Mar 30, 2022 at 7:22 PM Ashutosh Bapat
<ashutosh.bapat.oss@gmail.com> wrote:
>
> 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.

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