Horton

Re: [HACKERS] logical decoding of two-phase transactions

Ajin Cherian <itsajin@gmail.com>

View thread View on postgr.es
From: Ajin Cherian <itsajin@gmail.com>
To: Dilip Kumar <dilipbalaut@gmail.com>
Cc: Amit Kapila <amit.kapila16@gmail.com>, Tom Lane <tgl@sss.pgh.pa.us>, Masahiko Sawada <sawada.mshk@gmail.com>, PostgreSQL Hackers <pgsql-hackers@postgresql.org>
Date: 2020-09-22T11:47:36Z
Lists: pgsql-hackers

Attachments

On Sun, Sep 20, 2020 at 3:31 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

> 1.
> + /*
> + * Process invalidation messages, even if we're not interested in the
> + * transaction's contents, since the various caches need to always be
> + * consistent.
> + */
> + if (parsed->nmsgs > 0)
> + {
> + if (!ctx->fast_forward)
> + ReorderBufferAddInvalidations(ctx->reorder, xid, buf->origptr,
> +   parsed->nmsgs, parsed->msgs);
> + ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);
> + }
> +
>
> I think we don't need to add prepare time invalidation messages as we now we
> are already logging the invalidations at the command level and adding them to
> reorder buffer.
>

Removed.

> 2.
>
> + /*
> + * Tell the reorderbuffer about the surviving subtransactions. We need to
> + * do this because the main transaction itself has not committed since we
> + * are in the prepare phase right now. So we need to be sure the snapshot
> + * is setup correctly for the main transaction in case all changes
> + * happened in subtransanctions
> + */
> + for (i = 0; i < parsed->nsubxacts; i++)
> + {
> + ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
> + buf->origptr, buf->endptr);
> + }
> +
> + if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
> + (parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
> + ctx->fast_forward || FilterByOrigin(ctx, origin_id))
> + return;
>
> Do we need to call ReorderBufferCommitChild if we are skiping this transaction?
> I think the below check should be before calling ReorderBufferCommitChild.
>

Done.

> 3.
>
> + /*
> + * If it's ROLLBACK PREPARED then handle it via callbacks.
> + */
> + if (TransactionIdIsValid(xid) &&
> + !SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
> + parsed->dbId == ctx->slot->data.database &&
> + !FilterByOrigin(ctx, origin_id) &&
> + ReorderBufferTxnIsPrepared(ctx->reorder, xid, parsed->twophase_gid))
> + {
> + ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
> + commit_time, origin_id, origin_lsn,
> + parsed->twophase_gid, false);
> + return;
> + }
>
>
> I think we have already checked !SnapBuildXactNeedsSkip, parsed->dbId
> == ctx->slot->data.database and !FilterByOrigin in DecodePrepare
> so if those are not true then we wouldn't have prepared this
> transaction i.e. ReorderBufferTxnIsPrepared will be false so why do we
> need
> to recheck this conditions.

I didnt change this, as I am seeing cases where the Abort is getting
called for transactions that needs to be skipped. I also see that the
same check is there both in DecodePrepare and DecodeCommit.
So, while the same transactions were not getting prepared or
committed, it tries to get ROLLBACK PREPARED (as part of finish
prepared handling). The check in if ReorderBufferTxnIsPrepared() is
also not proper. I will need to relook
this logic again in a future patch.

>
>
> 4.
>
> + /* If streaming, reset the TXN so that it is allowed to stream
> remaining data. */
> + if (streaming && stream_started)
> + {
> + ReorderBufferResetTXN(rb, txn, snapshot_now,
> +   command_id, prev_lsn,
> +   specinsert);
> + }
> + else
> + {
> + elog(LOG, "stopping decoding of %s (%u)",
> + txn->gid[0] != '\0'? txn->gid:"", txn->xid);
> + ReorderBufferTruncateTXN(rb, txn, true);
> + }
>
> Why only if (streaming) is not enough?  I agree if we are coming here
> and it is streaming mode then streaming started must be true
> but we already have an assert for that.
>
Changed.

Amit,

I have also changed  test_decode startup to support two_phase commits
only if specified similar to how it was done for streaming. I have
also changed the test cases accordingly. However, I have not added it
to the pgoutput startup as that would require create subscription
changes.  I will do that in a future patch. Some other pending changes
are:

1. Remove snapshots on prepare truncate.
2. Look at why ReorderBufferCleanupTXN is failing after a
ReorderBufferTruncateTXN
3. Add prepare support to streaming

regards,
Ajin Cherian
Fujitsu Australia

Commits

  1. Add prepare API support for streaming transactions in logical replication.

  2. Doc: minor improvements for logical replication protocol documentation.

  3. Fix test failure in 021_twophase.pl.

  4. Refactor to make common functions in proto.c and worker.c.

  5. Unify PostgresNode's new() and get_new_node() methods

  6. Fix potential buffer overruns in proto.c.

  7. Add support for prepared transactions to built-in logical replication.

  8. Refactor function parse_subscription_options.

  9. Allow enabling two-phase option via replication protocol.

  10. Don't use Asserts to check for violations of replication protocol.

  11. Improve psql tab completion for options of subcriptions and publications

  12. Rearrange logrep worker's snapshot handling some more.

  13. doc: Update information of new messages for logical replication.

  14. ALTER SUBSCRIPTION ... ADD/DROP PUBLICATION

  15. Allow pgoutput to send logical decoding messages.

  16. Refactor function parse_output_parameters.

  17. Avoid repeated decoding of prepared transactions after a restart.

  18. Fix an oversight in ReorderBufferFinishPrepared.

  19. Allow multiple xacts during table sync in logical replication.

  20. Fix replication of in-progress transactions in tablesync worker.

  21. Fix 'skip-empty-xacts' option in test_decoding for streaming mode.

  22. Use Enum for top level logical replication message types.

  23. Add support for streaming to built-in logical replication.

  24. Fix the logical streaming test.

  25. Implement streaming mode in ReorderBuffer.

  26. Extend the logical decoding output plugin API with stream methods.

  27. Store 2PC GID in commit/abort WAL recs for logical decoding