Re: Skipping logical replication transactions on subscriber side

On Mon, May 24, 2021 at 7:51 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
>
> On Mon, May 24, 2021 at 1:32 PM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com> wrote:
> >
> > If a logical replication worker cannot apply the change on the
> > subscriber for some reason (e.g., missing table or violating a
> > constraint, etc.), logical replication stops until the problem is
> > resolved. Ideally, we resolve the problem on the subscriber (e.g., by
> > creating the missing table or removing the conflicting data, etc.) but
> > occasionally a problem cannot be fixed and it may be necessary to skip
> > the entire transaction in question. Currently, we have two ways to
> > skip transactions: advancing the LSN of the replication origin on the
> > subscriber and advancing the LSN of the replication slot on the
> > publisher. But both ways might not be able to skip exactly one
> > transaction in question and end up skipping other transactions too.
> >
> > I’d like to propose a way to skip the particular transaction on the
> > subscriber side. As the first step, a transaction can be specified to
> > be skipped by specifying remote XID on the subscriber. This feature
> > would need two sub-features: (1) a sub-feature for users to identify
> > the problem subscription and the problem transaction’s XID, and (2) a
> > sub-feature to skip the particular transaction to apply.
> >
> > For (1), I think the simplest way would be to put the details of the
> > change being applied in errcontext. For example, the following
> > errcontext shows the remote XID as well as the action name, the
> > relation name, and commit timestamp:
> >
> > ERROR: duplicate key value violates unique constraint "test_pkey"
> > DETAIL: Key (c)=(1) already exists.
> > CONTEXT: during apply of "INSERT" for relation "public.test" in
> > transaction with xid 590 commit timestamp 2021-05-21
> > 14:32:02.134273+09
> >
>
> In the above, the subscription name/id is not mentioned. I think you
> need it for sub-feature-2.

Agreed.

>
> > The user can identify which remote XID has a problem during applying
> > the change (XID=590 in this case). As another idea, we can have a
> > statistics view for logical replication workers, showing information
> > of the last failure transaction.
> >
>
> It might be good to display at both places. Having subscriber-side
> information in the view might be helpful in other ways as well like we
> can use it to display the number of transactions processed by a
> particular subscriber.

Yes. I think we can report that information to the stats collector. It
needs to live even after the worker exiting.

>
> I think you need to consider few more things here:
> (a) Say the error occurs after applying some part of changes, then
> just skipping the remaining part won't be sufficient, we probably need
> to someway rollback the applied changes (by rolling back the
> transaction or in some other way).

After more thought, it might be better to that setting and resetting
the XID to skip requires disabling the subscription. This would not be
a restriction for users since logical replication is likely to already
stop (and possibly repeating restarting and stopping) due to an error.
Setting and resetting the XID modifies the system catalog so it's a
crash-safe change and survives beyond the server restarts. When a
logical replication worker starts, it checks the XID. If the worker
receives changes associated with the transaction with the specified
XID, it can ignore the entire transaction.

> (b) How do you handle streamed transactions? It is possible that some
> of the streams are successful and the error occurs after that, say
> when writing to the stream file. Now, would you skip writing to stream
> file or will you write it, and then during apply, you will skip the
> entire transaction and remove the corresponding stream file.

I think streamed transactions can be handled in the same way described in (a).

> (c) There is also a possibility that the error occurs while applying
> the changes of some subtransaction (this is only possible for
> streaming xacts), so, in such cases, do we allow users to rollback the
> subtransaction or user has to rollback the entire transaction. I am
> not sure but maybe for very large transactions users might just want
> to rollback the subtransaction.

If the user specifies XID of a subtransaction, it would be better to
skip only the subtransaction. If specifies top transaction XID, it
would be better to skip the entire transaction. What do you think?

> (d) How about prepared transactions? Do we need to rollback the
> prepared transaction if user decides to skip such a transaction? We
> already allow prepared transactions to be streamed to plugins and the
> work for subscriber-side apply is in progress [1], so I think we need
> to consider this case as well.

If a transaction replicated from the subscriber could be prepared on
the subscriber, it would be guaranteed to be able to be either
committed or rolled back. Given that this feature is to skip a problem
transaction, I think it should not do anything for transactions that
are already prepared on the subscriber.

> (e) Do we want to provide such a feature via output plugins as well,
> if not, why?

You mean to specify an XID to skip on the publisher side? Since I've
been considering this feature as a way to resume the logical
replication having a problem I've not thought of that idea but It
would be a good idea. Do you have any use cases? If we specified the
XID on the publisher, multiple subscribers would skip that
transaction.

>
> > For (2), what I'm thinking is to add a new action to ALTER
> > SUBSCRIPTION command like ALTER SUBSCRIPTION test_sub SET SKIP
> > TRANSACTION 590. Also, we can have actions to reset it; ALTER
> > SUBSCRIPTION test_sub RESET SKIP TRANSACTION. Those commands add the
> > XID to a new column of pg_subscription or a new catalog, having the
> > worker reread its subscription information. Once the worker skipped
> > the specified transaction, it resets the transaction to skip on the
> > catalog.
> >
>
> What if we fail while updating the reset information in the catalog?
> Will it be the responsibility of the user to reset such a transaction
> or we will retry it after restart of worker? Now, say, we give such a
> responsibility to the user and the user forgets to reset it then there
> is a possibility that after wraparound we will again skip the
> transaction which is not intended. And, if we want to retry it after
> restart of worker, how will the worker remember the previous failure?

As described above, setting and resetting XID to skip is implemented
as a normal system catalog change, so it's crash-safe and persisted. I
think that the worker can either removes the XID or mark it as done
once it skipped the specified transaction so that it won't skip the
same XID again after wraparound. Also, it might be better if we reset
the XID also when a subscription field such as subconninfo is changed
because it could imply the worker will connect to another publisher
having a different XID space.

We also need to handle the cases where the user specifies an old XID
or XID whose transaction is already prepared on the subscriber. I
think the worker can reset the XID with a warning when it finds out
that the XID seems no longer valid or it cannot skip the specified
XID. For example in the former case, it can do that when the first
received transaction’s XID is newer than the specified XID. In the
latter case, it can do that when it receives the commit/rollback
prepared message of the specified XID.

Regards,

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