Re: Skipping logical replication transactions on subscriber side

On Mon, Jan 10, 2022 at 8:50 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
>
> On Thu, Dec 16, 2021 at 11:12 AM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
> >
> > On Thu, Dec 16, 2021 at 2:21 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
> > >
> > > >
> > > > So if skip_xid is already changed, the apply worker would do
> > > > replorigin_advance() with WAL logging, instead of committing the
> > > > catalog change?
> > > >
> > >
> > > Right. BTW, how are you planning to advance the origin? Normally, a
> > > commit transaction would do it but when we are skipping all changes,
> > > the commit might not do it as there won't be any transaction id
> > > assigned.
> >
> > I've not tested it yet but replorigin_advance() with wal_log = true
> > seems to work for this case.
> >
>
> IIUC, the changes corresponding to above in the latest patch are as follows:
>
> --- a/src/backend/replication/logical/origin.c
> +++ b/src/backend/replication/logical/origin.c
> @@ -921,7 +921,8 @@ replorigin_advance(RepOriginId node,
> LWLockAcquire(&replication_state->lock, LW_EXCLUSIVE);
>
> /* Make sure it's not used by somebody else */
> - if (replication_state->acquired_by != 0)
> + if (replication_state->acquired_by != 0 &&
> + replication_state->acquired_by != MyProcPid)
> {
> ...
>
> clear_subscription_skip_xid()
> {
> ..
> + else if (!XLogRecPtrIsInvalid(origin_lsn))
> + {
> + /*
> + * User has already changed subskipxid before clearing the subskipxid, so
> + * don't change the catalog but just advance the replication origin.
> + */
> + replorigin_advance(replorigin_session_origin, origin_lsn,
> + GetXLogInsertRecPtr(),
> + false, /* go_backward */
> + true /* wal_log */);
> + }
> ..
> }
>
> I was thinking what if we don't advance origin explicitly in this
> case? Actually, that will be no different than the transactions where
> the apply worker doesn't apply any change because the initial sync is
> in progress (see should_apply_changes_for_rel()) or we have received
> an empty transaction. In those cases also, the origin lsn won't be
> advanced even though we acknowledge the advanced last_received
> location because of keep_alive messages. Now, it is possible after the
> restart we send the old start_lsn location because the replication
> origin was not updated before restart but we handle that case in the
> server by starting from the last confirmed location. See below code:
>
> CreateDecodingContext()
> {
> ..
> else if (start_lsn < slot->data.confirmed_flush)
> ..

Good point. Probably one minor thing that is different from the
transaction where the apply worker applied an empty transaction is a
case where the server restarts/crashes before sending an
acknowledgment of the flush location. That is, in the case of the
empty transaction, the publisher sends an empty transaction again. On
the other hand in the case of skipping the transaction, a non-empty
transaction will be sent again but skip_xid is already changed or
cleared, therefore the user will have to specify skip_xid again. If we
write replication origin WAL record to advance the origin lsn, it
reduces the possibility of that. But I think it’s a very minor case so
we won’t need to deal with that.

Anyway, according to your analysis, I think we don't necessarily need
to do replorigin_advance() in this case.

>
> Few other comments on the latest patch:
> =================================
> 1.
> A conflict will produce an error and will stop the replication; it must be
> resolved manually by the user. Details about the conflict can be found in
> - the subscriber's server log.
> + <xref linkend="monitoring-pg-stat-subscription-workers"/> as well as the
> + subscriber's server log.
>
> Can we slightly change the modified line to: "Details about the
> conflict can be found in <xref
> linkend="monitoring-pg-stat-subscription-workers"/> and the
> subscriber's server log."?

Will fix it.

> I think we can commit this change
> separately as this is true even without this patch.

Right. It seems an oversight of 8d74fc96db. I've attached the patch.

>
> 2.
> The resolution can be done either by changing data on the subscriber so
> - that it does not conflict with the incoming change or by skipping the
> - transaction that conflicts with the existing data. The transaction can be
> - skipped by calling the <link linkend="pg-replication-origin-advance">
> + that it does not conflict with the incoming changes or by skipping the whole
> + transaction. This option specifies the ID of the transaction whose
> + application is to be skipped by the logical replication worker. The logical
> + replication worker skips all data modification transaction conflicts with
> + the existing data. When a conflict produce an error, it is shown in
> + <structname>pg_stat_subscription_workers</structname> view as follows:
>
> I don't think most of the additional text added in the above paragraph
> is required. We can rephrase it as: "The resolution can be done either
> by changing data on the subscriber so that it does not conflict with
> the incoming change or by skipping the transaction that conflicts with
> the existing data. When a conflict produces an error, it is shown in
> <structname>pg_stat_subscription_workers</structname> view as
> follows:". After that keep the text, you have.

Agreed, will fix.

>
> 3.
> They skip the whole transaction, including changes that may not violate any
> + constraint. They may easily make the subscriber inconsistent, especially if
> + a user specifies the wrong transaction ID or the position of origin.
>
> Can we slightly reword the above text as: "Skipping the whole
> transaction includes skipping the changes that may not violate any
> constraint. This can easily make the subscriber inconsistent,
> especially if a user specifies the wrong transaction ID or the
> position of origin."?

Will fix.

>
> 4.
> The logical replication worker skips all data
> + modification changes within the specified transaction. Therefore, since
> + it skips the whole transaction including the changes that may not violate
> + the constraint, it should only be used as a last resort. This option has
> + no effect for the transaction that is already prepared with enabling
> + <literal>two_phase</literal> on susbscriber.
>
> Let's slightly reword the above text as: "The logical replication
> worker skips all data modification changes within the specified
> transaction including the changes that may not violate the constraint,
> so, it should only be used as a last resort. This option has no effect
> on the transaction that is already prepared by enabling
> <literal>two_phase</literal> on the subscriber."

Will fix.

>
> 5.
> + by the logical replication worker. Setting
> <literal>NONE</literal> means
> + to reset the transaction ID.
>
> Let's slightly reword the second part of the sentence as: "Setting
> <literal>NONE</literal> resets the transaction ID."

Will fix.

>
> 6.
> Once we start skipping
> + * changes, we don't stop it until the we skip all changes of the
> transaction even
> + * if the subscription invalidated and MySubscription->skipxid gets
> changed or reset.
>
> /subscription invalidated/subscription is invalidated

Will fix.

>
> What do you mean by subscription invalidated and how is it related to
> this feature? I think we should mention something on these lines in
> the docs as well.

I meant that MySubscription, a cache of pg_subscription entry, is
invalidated by the catalog change. IIUC while applying changes we
don't re-read pg_subscription (i.e., not calling
maybe_reread_subscription()). Similarly, while skipping changes, we
also don't do that. Therefore, even if skip_xid has been changed while
skipping changes, we don't stop skipping changes.

>
> 7.
> "Please refer to the comments in these functions for details.". We can
> slightly modify this part of the comment as: "Please refer to the
> comments in corresponding functions for details."

Will fix.

I'll submit an updated patch soon.

Regards,

--
Masahiko Sawada
EDB: https://www.enterprisedb.com/