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

On 01/08/18 16:00, Nikhil Sontakke wrote:

I was wondering if anything else would be needed for user-defined
catalog tables..

We don't need to do anything else for user-defined catalog tables
since they will also get accessed via the systable_* scan APIs.

They can be, but currently they might not be. So this requires at least
big fat warning in docs and description on how to access user catalogs
from plugins correctly (ie to always use systable_* API on them). It
would be nice if we could check for it in Assert builds at least.

--
Petr Jelinek http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Training & Services

On 2018-08-01 21:55:18 +0200, Petr Jelinek wrote:

On 01/08/18 16:00, Nikhil Sontakke wrote:

I was wondering if anything else would be needed for user-defined
catalog tables..

We don't need to do anything else for user-defined catalog tables
since they will also get accessed via the systable_* scan APIs.

They can be, but currently they might not be. So this requires at least
big fat warning in docs and description on how to access user catalogs
from plugins correctly (ie to always use systable_* API on them). It
would be nice if we could check for it in Assert builds at least.

Yea, I agree. I think we should just consider putting similar checks in
the general scan APIs. With an unlikely() and the easy predictability of
these checks, I think we should be fine, overhead-wise.

Greetings,

Andres Freund

They can be, but currently they might not be. So this requires at least
big fat warning in docs and description on how to access user catalogs
from plugins correctly (ie to always use systable_* API on them). It
would be nice if we could check for it in Assert builds at least.

Ok, modified the sgml documentation for the above.

Yea, I agree. I think we should just consider putting similar checks in
the general scan APIs. With an unlikely() and the easy predictability of
these checks, I think we should be fine, overhead-wise.

Ok, added unlikely() checks in the heap_* scan APIs.

Revised patchset attached.

Regards,
Nikhils

Greetings,

Andres Freund

--
Nikhil Sontakke http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Training & Services

Hello,

I have looked through the patches. I will first describe relativaly
serious issues I see and then proceed with small nitpicking.

- On decoding of aborted xacts. The idea to throw an error once we
detect the abort is appealing, however I think you will have problems
with subxacts in the current implementation. What if subxact issues
DDL and then aborted, but main transaction successfully committed?

- Decoding transactions at PREPARE record changes rules of the "we ship
all commits after lsn 'x'" game. Namely, it will break initial
tablesync: what if consistent snapshot was formed *after* PREPARE, but
before COMMIT PREPARED, and the plugin decides to employ 2pc? Instead
of getting inital contents + continious stream of changes the receiver
will miss the prepared xact contents and raise 'prepared xact doesn't
exist' error. I think the starting point to address this is to forbid
two-phase decoding of xacts with lsn of PREPARE less than
snapbuilder's start_decoding_at.

- Currently we will call abort_prepared cb even if we failed to actually
prepare xact due to concurrent abort. I think it is confusing for
users. We should either handle this by remembering not to invoke
abort_prepared in these cases or at least document this behaviour,
leaving this problem to the receiver side.

- I find it suspicious that DecodePrepare completely ignores actions of
SnapBuildCommitTxn. For example, to execute invalidations, the latter
sets base snapshot if our xact (or subxacts) did DDL and the snapshot
not set yet. My fantasy doesn't hint me the concrete example
where this would burn at the moment, but it should be considered.

Now, the bikeshedding.

First patch:
- I am one of those people upthread who don't think that converting
flags to bitmask is beneficial -- especially given that many of them
are mutually exclusive, e.g. xact can't be committed and aborted at
the same time. Apparently you have left this to the committer though.

Second patch:
- Applying gives me
Applying: Support decoding of two-phase transactions at PREPARE
.git/rebase-apply/patch:871: trailing whitespace.

+      row. The <function>change_cb</function> callback may access system or
+      user catalog tables to aid in the process of outputting the row
+      modification details. In case of decoding a prepared (but yet
+      uncommitted) transaction or decoding of an uncommitted transaction, this
+      change callback is ensured sane access to catalog tables regardless of
+      simultaneous rollback by another backend of this very same transaction.

I don't think we should explain this, at least in such words. As
mentioned upthread, we should warn about allowed systable_* accesses
instead. Same for message_cb.

+	/*
+	 * 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
+	 */

While we do certainly need to associate subxacts here, the explanation
looks weird to me. I would leave just the 'Tell the reorderbuffer about
the surviving subtransactions' as in DecodeCommit.

}
-
/*
* There's a speculative insertion remaining, just clean in up, it
* can't have been successful, otherwise we'd gotten a confirmation

Spurious newline deletion.

- I would rename ReorderBufferCommitInternal to ReorderBufferReplay:
we replay the xact there, not commit.

- If xact is empty, we will not prepare it (and call cb),
even if the output plugin asked us. However, we will call
commit_prepared cb.

- ReorderBufferTxnIsPrepared and ReorderBufferPrepareNeedSkip do the
same and should be merged with comments explaining that the answer
must be stable.

- filter_prepare_cb callback existence is checked in both decode.c and
in filter_prepare_cb_wrapper.

+	/*
+	 * The transaction may or may not exist (during restarts for example).
+	 * Anyways, 2PC transactions do not contain any reorderbuffers. So allow
+	 * it to be created below.
+	 */

Code around looks sane, but I think that ReorderBufferTXN for our xact
must *not* exist at this moment: if we are going to COMMIT/ABORT
PREPARED it, it must have been replayed and RBTXN purged immediately
after. Also, instead of misty '2PC transactions do not contain any
reorderbuffers' I would say something like 'create dummy
ReorderBufferTXN to pass it to the callback'.

- In DecodeAbort:
+	/*
+	 * If it's ROLLBACK PREPARED then handle it via callbacks.
+	 */
+	if (TransactionIdIsValid(xid) &&
+		!SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
+

How xid can be invalid here?

- It might be worthwile to put the check
+		!SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
+		parsed->dbId == ctx->slot->data.database &&
+		!FilterByOrigin(ctx, origin_id) &&

which appears 3 times now into separate function.

+	 * two-phase transactions - we either have to have all of them or none.
+	 * The filter_prepare callback is optional, but can only be defined when

Kind of controversial (all of them or none, but optional), might be
formulated more accurately.

+	/*
+	 * Capabilities of the output plugin.
+	 */
+	bool        enable_twophase;

I would rename this to 'supports_twophase' since this is not an option
but a description of the plugin capabilities.

+	/* filter_prepare is optional, but requires two-phase decoding */
+	if ((ctx->callbacks.filter_prepare_cb != NULL) && (!ctx->enable_twophase))
+		ereport(ERROR,
+				(errmsg("Output plugin does not support two-phase decoding, but "
+						"registered
filter_prepared callback.")));

Don't think we need to check that...

+		 * Otherwise call either PREPARE (for twophase transactions) or COMMIT
+		 * (for regular ones).
+		 */
+		if (rbtxn_rollback(txn))
+			rb->abort(rb, txn, commit_lsn);

This is the dead code since we don't have decoding of in-progress xacts
yet.

Third patch:
+/*
+ * An xid value pointing to a possibly ongoing or a prepared transaction.
+ * Currently used in logical decoding.  It's possible that such transactions
+ * can get aborted while the decoding is ongoing.
+ */

I would explain here that this xid is checked for abort after each
catalog scan, and sent for the details to SetupHistoricSnapshot.

+	/*
+	 * If CheckXidAlive is valid, then we check if it aborted. If it did, we
+	 * error out
+	 */
+	if (TransactionIdIsValid(CheckXidAlive) &&
+			!TransactionIdIsInProgress(CheckXidAlive) &&
+			!TransactionIdDidCommit(CheckXidAlive))
+			ereport(ERROR,
+				(errcode(ERRCODE_TRANSACTION_ROLLBACK),
+				 errmsg("transaction aborted during system catalog scan")));

Probably centralize checks in one function? As well as 'We don't expect
direct calls to heap_fetch...' ones.

P.S. Looks like you have torn the thread chain: In-Reply-To header of
mail [1]/messages/by-id/CAMGcDxeqEpWj3fTXwqhSwBdXd2RS9jzwWscO-XbeCfso6ts3+Q@mail.gmail.com is missing. Please don't do that.

[1]: /messages/by-id/CAMGcDxeqEpWj3fTXwqhSwBdXd2RS9jzwWscO-XbeCfso6ts3+Q@mail.gmail.com

--
Arseny Sher
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company

On 2018-08-06 21:06:13 +0300, Arseny Sher wrote:

Hello,

I have looked through the patches. I will first describe relativaly
serious issues I see and then proceed with small nitpicking.

- On decoding of aborted xacts. The idea to throw an error once we
detect the abort is appealing, however I think you will have problems
with subxacts in the current implementation. What if subxact issues
DDL and then aborted, but main transaction successfully committed?

I don't see a fundamental issue here. I've not reviewed the current
patchset meaningfully, however. Do you see a fundamental issue here?

- Decoding transactions at PREPARE record changes rules of the "we ship
all commits after lsn 'x'" game. Namely, it will break initial
tablesync: what if consistent snapshot was formed *after* PREPARE, but
before COMMIT PREPARED, and the plugin decides to employ 2pc? Instead
of getting inital contents + continious stream of changes the receiver
will miss the prepared xact contents and raise 'prepared xact doesn't
exist' error. I think the starting point to address this is to forbid
two-phase decoding of xacts with lsn of PREPARE less than
snapbuilder's start_decoding_at.

Yea, that sounds like it need to be addressed.

- Currently we will call abort_prepared cb even if we failed to actually
prepare xact due to concurrent abort. I think it is confusing for
users. We should either handle this by remembering not to invoke
abort_prepared in these cases or at least document this behaviour,
leaving this problem to the receiver side.

What precisely do you mean by "concurrent abort"?

- I find it suspicious that DecodePrepare completely ignores actions of
SnapBuildCommitTxn. For example, to execute invalidations, the latter
sets base snapshot if our xact (or subxacts) did DDL and the snapshot
not set yet. My fantasy doesn't hint me the concrete example
where this would burn at the moment, but it should be considered.

Yea, I think this need to mirror the actions (and thus generalize the
code to avoid duplication)

Now, the bikeshedding.

First patch:
- I am one of those people upthread who don't think that converting
flags to bitmask is beneficial -- especially given that many of them
are mutually exclusive, e.g. xact can't be committed and aborted at
the same time. Apparently you have left this to the committer though.

Similar.

- Andres

Andres Freund <andres@anarazel.de> writes:

- On decoding of aborted xacts. The idea to throw an error once we
detect the abort is appealing, however I think you will have problems
with subxacts in the current implementation. What if subxact issues
DDL and then aborted, but main transaction successfully committed?

I don't see a fundamental issue here. I've not reviewed the current
patchset meaningfully, however. Do you see a fundamental issue here?

Hmm, yes, this is not an issue for this patch because after reading
PREPARE record we know all aborted subxacts and won't try to decode
their changes. However, this will be raised once we decide to decode
in-progress transactions. Checking for all subxids is expensive;
moreover, WAL doesn't provide all of them until commit... it might be
easier to prevent vacuuming of aborted stuff while decoding needs it.
Matter for another patch, anyway.

- Currently we will call abort_prepared cb even if we failed to actually
prepare xact due to concurrent abort. I think it is confusing for
users. We should either handle this by remembering not to invoke
abort_prepared in these cases or at least document this behaviour,
leaving this problem to the receiver side.

What precisely do you mean by "concurrent abort"?

With current patch, the following is possible:
* We start decoding of some prepared xact;
* Xact aborts (ABORT PREPARED) for any reason;
* Decoding processs notices this on catalog scan and calls abort()
callback;
* Later decoding process reads abort record and calls abort_prepared
callback.

--
Arseny Sher
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company

Hi Arseny,

- Decoding transactions at PREPARE record changes rules of the "we ship
all commits after lsn 'x'" game. Namely, it will break initial
tablesync: what if consistent snapshot was formed *after* PREPARE, but
before COMMIT PREPARED, and the plugin decides to employ 2pc? Instead
of getting inital contents + continious stream of changes the receiver
will miss the prepared xact contents and raise 'prepared xact doesn't
exist' error. I think the starting point to address this is to forbid
two-phase decoding of xacts with lsn of PREPARE less than
snapbuilder's start_decoding_at.

It will be the job of the plugin to return a consistent answer for
every GID that is encountered. In this case, the plugin will decode
the transaction at COMMIT PREPARED time and not at PREPARE time.

- Currently we will call abort_prepared cb even if we failed to actually
prepare xact due to concurrent abort. I think it is confusing for
users. We should either handle this by remembering not to invoke
abort_prepared in these cases or at least document this behaviour,
leaving this problem to the receiver side.

The point is, when we reach the "ROLLBACK PREPARED", we have no idea
if the "PREPARE" was aborted by this rollback happening concurrently.
So it's possible that the 2PC has been successfully decoded and we
would have to send the rollback to the other side which would need to
check if it needs to rollback locally.

- I find it suspicious that DecodePrepare completely ignores actions of
SnapBuildCommitTxn. For example, to execute invalidations, the latter
sets base snapshot if our xact (or subxacts) did DDL and the snapshot
not set yet. My fantasy doesn't hint me the concrete example
where this would burn at the moment, but it should be considered.

I had discussed this area with Petr and we didn't see any issues as well, then.

Now, the bikeshedding.

First patch:
- I am one of those people upthread who don't think that converting
flags to bitmask is beneficial -- especially given that many of them
are mutually exclusive, e.g. xact can't be committed and aborted at
the same time. Apparently you have left this to the committer though.

Hmm, there seems to be divided opinion on this. I am willing to go
back to using the booleans if there's opposition and if the committer
so wishes. Note that this patch will end up adding 4/5 more booleans
in that case (we add new ones for prepare, commit prepare, abort,
rollback prepare etc).

Second patch:
- Applying gives me
Applying: Support decoding of two-phase transactions at PREPARE
.git/rebase-apply/patch:871: trailing whitespace.

+      row. The <function>change_cb</function> callback may access system or
+      user catalog tables to aid in the process of outputting the row
+      modification details. In case of decoding a prepared (but yet
+      uncommitted) transaction or decoding of an uncommitted transaction, this
+      change callback is ensured sane access to catalog tables regardless of
+      simultaneous rollback by another backend of this very same transaction.

I don't think we should explain this, at least in such words. As
mentioned upthread, we should warn about allowed systable_* accesses
instead. Same for message_cb.

Looks like you are looking at an earlier patchset. The latest patchset
has removed the above.

+       /*
+        * 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
+        */

While we do certainly need to associate subxacts here, the explanation
looks weird to me. I would leave just the 'Tell the reorderbuffer about
the surviving subtransactions' as in DecodeCommit.

}
-
/*
* There's a speculative insertion remaining, just clean in up, it
* can't have been successful, otherwise we'd gotten a confirmation

Spurious newline deletion.

- I would rename ReorderBufferCommitInternal to ReorderBufferReplay:
we replay the xact there, not commit.

- If xact is empty, we will not prepare it (and call cb),
even if the output plugin asked us. However, we will call
commit_prepared cb.

- ReorderBufferTxnIsPrepared and ReorderBufferPrepareNeedSkip do the
same and should be merged with comments explaining that the answer
must be stable.

- filter_prepare_cb callback existence is checked in both decode.c and
in filter_prepare_cb_wrapper.

+       /*
+        * The transaction may or may not exist (during restarts for example).
+        * Anyways, 2PC transactions do not contain any reorderbuffers. So allow
+        * it to be created below.
+        */

Code around looks sane, but I think that ReorderBufferTXN for our xact
must *not* exist at this moment: if we are going to COMMIT/ABORT
PREPARED it, it must have been replayed and RBTXN purged immediately
after. Also, instead of misty '2PC transactions do not contain any
reorderbuffers' I would say something like 'create dummy
ReorderBufferTXN to pass it to the callback'.

- In DecodeAbort:
+       /*
+        * If it's ROLLBACK PREPARED then handle it via callbacks.
+        */
+       if (TransactionIdIsValid(xid) &&
+               !SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
+

How xid can be invalid here?

- It might be worthwile to put the check
+               !SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
+               parsed->dbId == ctx->slot->data.database &&
+               !FilterByOrigin(ctx, origin_id) &&

which appears 3 times now into separate function.

+        * two-phase transactions - we either have to have all of them or none.
+        * The filter_prepare callback is optional, but can only be defined when

Kind of controversial (all of them or none, but optional), might be
formulated more accurately.

+       /*
+        * Capabilities of the output plugin.
+        */
+       bool        enable_twophase;

I would rename this to 'supports_twophase' since this is not an option
but a description of the plugin capabilities.

+       /* filter_prepare is optional, but requires two-phase decoding */
+       if ((ctx->callbacks.filter_prepare_cb != NULL) && (!ctx->enable_twophase))
+               ereport(ERROR,
+                               (errmsg("Output plugin does not support two-phase decoding, but "
+                                               "registered
filter_prepared callback.")));

Don't think we need to check that...

+                * Otherwise call either PREPARE (for twophase transactions) or COMMIT
+                * (for regular ones).
+                */
+               if (rbtxn_rollback(txn))
+                       rb->abort(rb, txn, commit_lsn);

This is the dead code since we don't have decoding of in-progress xacts
yet.

Yes, the above check can be done away with it.

Third patch:
+/*
+ * An xid value pointing to a possibly ongoing or a prepared transaction.
+ * Currently used in logical decoding.  It's possible that such transactions
+ * can get aborted while the decoding is ongoing.
+ */

I would explain here that this xid is checked for abort after each
catalog scan, and sent for the details to SetupHistoricSnapshot.

+       /*
+        * If CheckXidAlive is valid, then we check if it aborted. If it did, we
+        * error out
+        */
+       if (TransactionIdIsValid(CheckXidAlive) &&
+                       !TransactionIdIsInProgress(CheckXidAlive) &&
+                       !TransactionIdDidCommit(CheckXidAlive))
+                       ereport(ERROR,
+                               (errcode(ERRCODE_TRANSACTION_ROLLBACK),
+                                errmsg("transaction aborted during system catalog scan")));

Probably centralize checks in one function? As well as 'We don't expect
direct calls to heap_fetch...' ones.

P.S. Looks like you have torn the thread chain: In-Reply-To header of
mail [1] is missing. Please don't do that.

That wasn't me. I was also annoyed and surprised to see a new email
thread separate from the earlier containing 100 or so messages.

Regards,
Nikhils
--
Nikhil Sontakke http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Training & Services

Nikhil Sontakke <nikhils@2ndquadrant.com> writes:

- Decoding transactions at PREPARE record changes rules of the "we ship
all commits after lsn 'x'" game. Namely, it will break initial
tablesync: what if consistent snapshot was formed *after* PREPARE, but
before COMMIT PREPARED, and the plugin decides to employ 2pc? Instead
of getting inital contents + continious stream of changes the receiver
will miss the prepared xact contents and raise 'prepared xact doesn't
exist' error. I think the starting point to address this is to forbid
two-phase decoding of xacts with lsn of PREPARE less than
snapbuilder's start_decoding_at.

It will be the job of the plugin to return a consistent answer for
every GID that is encountered. In this case, the plugin will decode
the transaction at COMMIT PREPARED time and not at PREPARE time.

I can't imagine a scenario in which plugin would want to send COMMIT
PREPARED instead of replaying xact fully on CP record given it had never
seen PREPARE record. On the other hand, tracking such situations on
plugins side would make plugins life unneccessary complicated: either it
has to dig into snapbuilder/slot internals to learn when the snapshot
became consistent (which currently is impossible as this lsn is not
saved anywhere btw), or it must fsync each its decision to do or not to
do 2PC.

Technically, my concern covers not only tablesync, but just plain
decoding start: we don't want to ship COMMIT PREPARED if the downstream
had never had chance to see PREPARE.

As for tablesync, looking at current implementation I contemplate that
we would need to do something along the following lines:
- Tablesync worker performs COPY.
- It then speaks with main apply worker to arrange (origin)
lsn of sync point, as it does now.
- Tablesync worker applies changes up to arranged lsn; it never uses
two-phase decoding, all xacts are replayed on COMMIT PREPARED.
Moreover, instead of going into SYNCDONE state immediately after
reaching needed lsn, it stops replaying usual commits but continues
to receive changes to finish all transactions which were prepared
before sync point (we would need some additional support from
reorderbuffer to learn when this happens). Only then it goes into
SYNCDONE.
- Behaviour of the main apply worker doesn't change: it
ignores changes of the table in question before sync point and
applies them after sync point. It also can use 2PC decoding of any
transaction or not, as it desires.
I believe this approach would implement tablesync correctly (all changes
are applied, but only once) with minimal fuss.

- Currently we will call abort_prepared cb even if we failed to actually
prepare xact due to concurrent abort. I think it is confusing for
users. We should either handle this by remembering not to invoke
abort_prepared in these cases or at least document this behaviour,
leaving this problem to the receiver side.

The point is, when we reach the "ROLLBACK PREPARED", we have no idea
if the "PREPARE" was aborted by this rollback happening concurrently.
So it's possible that the 2PC has been successfully decoded and we
would have to send the rollback to the other side which would need to
check if it needs to rollback locally.

I understand this. But I find this confusing for the users, so I propose
to
- Either document that "you might get abort_prepared cb called even
after abort cb was invoked for the same transaction";
- Or consider adding some infrastructure to reorderbuffer to
remember not to call abort_prepared in these cases. Due to possible
reboots, I think this means that we need not to
ReorderBufferCleanupTXN immediately after failed attempt to replay
xact on PREPARE, but mark it as 'aborted' and keep it until we see
ABORT PREPARED record. If we see that xact is marked as aborted, we
don't call abort_prepared_cb. That way even if the decoder restarts
in between, we will see PREPARE in WAL, inquire xact status (even
if we skip it as already replayed) and mark it as aborted again.

- I find it suspicious that DecodePrepare completely ignores actions of
SnapBuildCommitTxn. For example, to execute invalidations, the latter
sets base snapshot if our xact (or subxacts) did DDL and the snapshot
not set yet. My fantasy doesn't hint me the concrete example
where this would burn at the moment, but it should be considered.

I had discussed this area with Petr and we didn't see any issues as well, then.

Ok, simplifying, what SnapBuildCommitTxn practically does is
* Decide whether we are interested in tracking this xact effects, and
if we are, mark it as committed.
* Build and distribute snapshot to all RBTXNs, if it is important.
* Set base snap of our xact if it did DDL, to execute invalidations
during replay.

I see that we don't need to do first two bullets during DecodePrepare:
xact effects are still invisible for everyone but itself after
PREPARE. As for seeing xacts own own changes, it is implemented via
logging cmin/cmax and we don't need to mark xact as committed for that
(c.f ReorderBufferCopySnap).

Regarding the third point... I think in 2PC decoding we might need to
execute invalidations twice:
1) After replaying xact on PREPARE to forget about catalog changes
xact did -- it is not yet committed and must be invisible to
other xacts until CP. In the latest patchset invalidations are
executed only if there is at least one change in xact (it has base
snap). It looks fine: we can't spoil catalogs if there were nothing
to decode. Better to explain that somewhere.
2) After decoding COMMIT PREPARED to make changes visible. In current
patchset it is always done. Actually, *this* is the reason
RBTXN might already exist when we enter ReorderBufferFinishPrepared,
not "(during restarts for example)" as comment says there:
if there were inval messages, RBTXN will be created
in DecodeCommit during their addition.

BTW, "that we might need to execute invalidations, add snapshot" in
SnapBuildCommitTxn looks like a cludge to me: I suppose it is better to
do that at ReorderBufferXidSetCatalogChanges.

Now, another issue is registering xact as committed in
SnapBuildCommitTxn during COMMIT PREPARED processing. Since RBTXN is
always purged after xact replay on PREPARE, the only medium we have for
noticing catalog changes during COMMIT PREPARED is invalidation messages
attached to the CP record. This raises the following question.
* If there is a guarantee that whenever xact makes catalog changes it
generates invalidation messages, then this code is fine. However,
currently ReorderBufferXidSetCatalogChanges is also called on
XLOG_HEAP_INPLACE processing and in SnapBuildProcessNewCid, which
is useless if such guarantee exists.
* If, on the other hand, there is no such guarantee, this code is
broken.

- I am one of those people upthread who don't think that converting
flags to bitmask is beneficial -- especially given that many of them
are mutually exclusive, e.g. xact can't be committed and aborted at
the same time. Apparently you have left this to the committer though.

Hmm, there seems to be divided opinion on this. I am willing to go
back to using the booleans if there's opposition and if the committer
so wishes. Note that this patch will end up adding 4/5 more booleans
in that case (we add new ones for prepare, commit prepare, abort,
rollback prepare etc).

Well, you can unite mutually exclusive fields into one enum or char with
macros defining possible values. Transaction can't be committed and
aborted at the same time, etc.

+      row. The <function>change_cb</function> callback may access system or
+      user catalog tables to aid in the process of outputting the row
+      modification details. In case of decoding a prepared (but yet
+      uncommitted) transaction or decoding of an uncommitted transaction, this
+      change callback is ensured sane access to catalog tables regardless of
+      simultaneous rollback by another backend of this very same transaction.

I don't think we should explain this, at least in such words. As
mentioned upthread, we should warn about allowed systable_* accesses
instead. Same for message_cb.

Looks like you are looking at an earlier patchset. The latest patchset
has removed the above.

I see, sorry.

--
Arseny Sher
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company

Hello,

Trying to revive this patch which attempts to support logical decoding of
two phase transactions. I've rebased and polished Nikhil's patch on the
current HEAD. Some of the logic in the previous patchset has already been
committed as part of large-in-progress transaction commits, like the
handling of concurrent aborts, so all that logic has been left out. I think
some of the earlier comments have already been addressed or are no longer
relevant. Do have a look at the patch and let me know what you think.I will
try and address any pending issues going forward.

regards,
Ajin Cherian
Fujitsu Australia

On Mon, Sep 7, 2020 at 10:54 AM Ajin Cherian <itsajin@gmail.com> wrote:

Hello,

Trying to revive this patch which attempts to support logical decoding of two phase transactions. I've rebased and polished Nikhil's patch on the current HEAD. Some of the logic in the previous patchset has already been committed as part of large-in-progress transaction commits, like the handling of concurrent aborts, so all that logic has been left out.

I am not sure about your point related to concurrent aborts. I think
we need some changes related to this patch. Have you tried to test
this behavior? Basically, we have the below code in
ReorderBufferProcessTXN() which will be hit for concurrent aborts, and
currently, the Asserts shown below will fail.

if (errdata->sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK)
{
/*
* This error can only occur when we are sending the data in
* streaming mode and the streaming is not finished yet.
*/
Assert(streaming);
Assert(stream_started);

Nikhil has a test for the same
(0004-Teach-test_decoding-plugin-to-work-with-2PC.Jan4) in his last
email [1]. You might want to use it to test this behavior. I think you
can also keep the tests as a separate patch as Nikhil had.

One other comment:
===================
@@ -27,6 +27,7 @@ typedef struct OutputPluginOptions
{
OutputPluginOutputType output_type;
bool receive_rewrites;
+ bool enable_twophase;
} OutputPluginOptions;
..
..
@@ -684,6 +699,33 @@ startup_cb_wrapper(LogicalDecodingContext *ctx,
OutputPluginOptions *opt, bool i
/* do the actual work: call callback */
ctx->callbacks.startup_cb(ctx, opt, is_init);

+ /*
+ * If the plugin claims to support two-phase transactions, then
+ * check that the plugin implements all callbacks necessary to decode
+ * two-phase transactions - we either have to have all of them or none.
+ * The filter_prepare callback is optional, but can only be defined when
+ * two-phase decoding is enabled (i.e. the three other callbacks are
+ * defined).
+ */
+ if (opt->enable_twophase)
+ {
+ int twophase_callbacks = (ctx->callbacks.prepare_cb != NULL) +
+ (ctx->callbacks.commit_prepared_cb != NULL) +
+ (ctx->callbacks.abort_prepared_cb != NULL);
+
+ /* Plugins with incorrect number of two-phase callbacks are broken. */
+ if ((twophase_callbacks != 3) && (twophase_callbacks != 0))
+ ereport(ERROR,
+ (errmsg("Output plugin registered only %d twophase callbacks. ",
+ twophase_callbacks)));
+ }

I don't know why the patch has used this way to implement an option to
enable two-phase. Can't we use how we implement 'stream-changes'
option in commit 7259736a6e? Just refer how we set ctx->streaming and
you can use a similar way to set this parameter.

--
With Regards,
Amit Kapila.

On Mon, Sep 7, 2020 at 11:17 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

Nikhil has a test for the same
(0004-Teach-test_decoding-plugin-to-work-with-2PC.Jan4) in his last
email [1]. You might want to use it to test this behavior. I think you
can also keep the tests as a separate patch as Nikhil had.

Done. I've added the tests and also tweaked code to make sure that the

aborts during 2 phase commits are also handled.

I don't know why the patch has used this way to implement an option to
enable two-phase. Can't we use how we implement 'stream-changes'
option in commit 7259736a6e? Just refer how we set ctx->streaming and
you can use a similar way to set this parameter.

Done, I've moved the checks for callbacks to inside the corresponding
wrappers.

Regards,
Ajin Cherian
Fujitsu Australia

On Wed, Sep 9, 2020 at 3:33 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Mon, Sep 7, 2020 at 11:17 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

Nikhil has a test for the same
(0004-Teach-test_decoding-plugin-to-work-with-2PC.Jan4) in his last
email [1]. You might want to use it to test this behavior. I think you
can also keep the tests as a separate patch as Nikhil had.

Done. I've added the tests and also tweaked code to make sure that the aborts during 2 phase commits are also handled.

Okay, I'll look into your changes but before that today, I have gone
through this entire thread to check if there are any design problems
and found that there were two major issues in the original proposal,
(a) one was to handle concurrent aborts which I think we should be
able to deal in a way similar to what we have done for decoding of
in-progress transactions and (b) what if someone specifically locks
pg_class or pg_attribute in exclusive mode (say be Lock pg_attribute
...), it seems the deadlock can happen in that case [0]/messages/by-id/20170328012546.473psm6546bgsi2c@alap3.anarazel.de. AFAIU, people
seem to think if there is no realistic scenario where deadlock can
happen apart from user explicitly locking the system catalog then we
might be able to get away by just ignoring such xacts to be decoded at
prepare time or would block it in some other way as any way that will
block the entire system. I am not sure what is the right thing but
something has to be done to avoid any sort of deadlock for this.

Another thing, I noticed is that originally we have subscriber-side
support as well, see [1]/messages/by-id/CAMGcDxchx=0PeQBVLzrgYG2AQ49QSRxHj5DCp7yy0QrJR0S0nA@mail.gmail.com (see *pgoutput* patch) but later dropped it
due to some reasons [2]/messages/by-id/CAMGcDxc-kuO9uq0zRCRwbHWBj_rePY9=raR7M9pZGWoj9EOGdg@mail.gmail.com. I think we should have pgoutput support as
well, so see what is required to get that incorporated.

I would also like to summarize my thinking on the usefulness of this
feature. One of the authors of this patch Stats wants this for a
conflict-free logical replication, see more details [3]/messages/by-id/CAMsr+YHQzGxnR-peT4SbX2-xiG2uApJMTgZ4a3TiRBM6COyfqg@mail.gmail.com. Craig seems
to suggest [3]/messages/by-id/CAMsr+YHQzGxnR-peT4SbX2-xiG2uApJMTgZ4a3TiRBM6COyfqg@mail.gmail.com that this will allow us to avoid conflicting schema
changes at different nodes though it is not clear to me if that is
possible without some external code support because we don't send
schema changes in logical replication, maybe Craig can shed some light
on this. Another use-case, I am thinking is if this can be used for
scaling-out reads as well. Because of 2PC, we can ensure that on
subscribers we have all the data committed on the master. Now, we can
design a system where different nodes are owners of some set of tables
and we can always get the data of those tables reliably from those
nodes, and then one can have some external process that will route the
reads accordingly. I know that the last idea is a bit of a hand-waving
but it seems to be possible after this feature.

[0]: /messages/by-id/20170328012546.473psm6546bgsi2c@alap3.anarazel.de
[1]: /messages/by-id/CAMGcDxchx=0PeQBVLzrgYG2AQ49QSRxHj5DCp7yy0QrJR0S0nA@mail.gmail.com
[2]: /messages/by-id/CAMGcDxc-kuO9uq0zRCRwbHWBj_rePY9=raR7M9pZGWoj9EOGdg@mail.gmail.com
[3]: /messages/by-id/CAMsr+YHQzGxnR-peT4SbX2-xiG2uApJMTgZ4a3TiRBM6COyfqg@mail.gmail.com

--
With Regards,
Amit Kapila.

On Sat, Sep 12, 2020 at 9:40 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

Another thing, I noticed is that originally we have subscriber-side
support as well, see [1] (see *pgoutput* patch) but later dropped it
due to some reasons [2]. I think we should have pgoutput support as
well, so see what is required to get that incorporated.

I have added the rebased patch-set for pgoutput and subscriber side

changes as well. This also includes a test case in subscriber.

regards,
Ajin Cherian

On Wed, Sep 9, 2020 at 3:33 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Mon, Sep 7, 2020 at 11:17 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

Nikhil has a test for the same
(0004-Teach-test_decoding-plugin-to-work-with-2PC.Jan4) in his last
email [1]. You might want to use it to test this behavior. I think you
can also keep the tests as a separate patch as Nikhil had.

Done. I've added the tests and also tweaked code to make sure that the aborts during 2 phase commits are also handled.

I don't think it is complete yet.
*
* This error can only occur when we are sending the data in
  * streaming mode and the streaming is not finished yet.
  */
- Assert(streaming);
- Assert(stream_started);
+ Assert(streaming || rbtxn_prepared(txn));
+ Assert(stream_started  || rbtxn_prepared(txn));

Here, you have updated the code but comments are still not updated.

*
@@ -2370,10 +2391,19 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
errdata = NULL;
curtxn->concurrent_abort = true;

- /* Reset the TXN so that it is allowed to stream remaining data. */
- ReorderBufferResetTXN(rb, txn, snapshot_now,
-   command_id, prev_lsn,
-   specinsert);
+ /* 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);
+ rb->abort(rb, txn, commit_lsn);
+ }

I don't think we need to perform abort here. Later we will anyway
encounter the WAL for Rollback Prepared for which we will call
abort_prepared_cb. As we have set the 'concurrent_abort' flag, it will
allow us to skip all the intermediate records. Here, we need only
enough state in ReorderBufferTxn that it can be later used for
ReorderBufferFinishPrepared(). Basically, you need functionality
similar to ReorderBufferTruncateTXN where except for invalidations you
can free memory for everything else. You can either write a new
function ReorderBufferTruncatePreparedTxn or pass another bool
parameter in ReorderBufferTruncateTXN to indicate it is prepared_xact
and then clean up additional things that are not required for prepared
xact.

*
Similarly, I don't understand why we need below code:
ReorderBufferProcessTXN()
{
..
+ if (rbtxn_rollback(txn))
+ rb->abort(rb, txn, commit_lsn);
..
}

There is nowhere we are setting the RBTXN_ROLLBACK flag, so how will
this check be true? If we decide to remove this code then don't forget
to update the comments.

*
If my previous two comments are correct then I don't think we need the
below interface.
+    <sect3 id="logicaldecoding-output-plugin-abort">
+     <title>Transaction Abort Callback</title>
+
+     <para>
+      The required <function>abort_cb</function> callback is called whenever
+      a transaction abort has to be initiated. This can happen if we are
+      decoding a transaction that has been prepared for two-phase commit and
+      a concurrent rollback happens while we are decoding it.
+<programlisting>
+typedef void (*LogicalDecodeAbortCB) (struct LogicalDecodingContext *ctx,
+                                       ReorderBufferTXN *txn,
+                                       XLogRecPtr abort_lsn);

I don't know why the patch has used this way to implement an option to
enable two-phase. Can't we use how we implement 'stream-changes'
option in commit 7259736a6e? Just refer how we set ctx->streaming and
you can use a similar way to set this parameter.

Done, I've moved the checks for callbacks to inside the corresponding wrappers.

This is not what I suggested. Please study the commit 7259736a6e and
see how streaming option is implemented. I want later subscribers can
specify whether they want transactions to be decoded at prepare time
similar to what we have done for streaming. Also, search for
ctx->streaming in the code and see how it is set to get the idea.

Note: Please use version number while sending patches, you can use
something like git format-patch -N -v n to do that. It makes easier
for the reviewer to compare it with the previous version.

Few other comments:
===================
1.
ReorderBufferProcessTXN()
{
..
if (streaming)
{
ReorderBufferTruncateTXN(rb, txn);

/* Reset the CheckXidAlive */
CheckXidAlive = InvalidTransactionId;
}
else
ReorderBufferCleanupTXN(rb, txn);
..
}

I don't think we can perform ReorderBufferCleanupTXN for the prepared
transactions because if we have removed the ReorderBufferTxn before
commit, the later code might not consider such a transaction in the
system and compute the wrong value of restart_lsn for a slot.
Basically, in SnapBuildProcessRunningXacts() when we call
ReorderBufferGetOldestTXN(), it should show the ReorderBufferTxn of
the prepared transaction which is not yet committed but because we
have removed it after prepare, it won't get that TXN and then that
leads to wrong computation of restart_lsn. Once we start from a wrong
point in WAL, the snapshot built was incorrect which will lead to the
wrong result. This is the same reason why the patch is not doing
ReorderBufferForget in DecodePrepare when we decide to skip the
transaction. Also, here, we need to set CheckXidAlive =
InvalidTransactionId; for prepared xact as well.

2. Have you thought about the interaction of streaming with prepared
transactions? You can try writing some tests using pg_logical* APIs
and see the behaviour. For ex. there is no handling in
ReorderBufferStreamCommit for the same. I think you need to introduce
stream_prepare API similar to stream_commit and then use the same.

3.
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
  {
  curtxn = change->txn;
  SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2254,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
  break;
  }
  }
-
  /*

Spurious line removal.

--
With Regards,
Amit Kapila.

On Tue, Sep 15, 2020 at 5:27 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Sat, Sep 12, 2020 at 9:40 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

Another thing, I noticed is that originally we have subscriber-side
support as well, see [1] (see *pgoutput* patch) but later dropped it
due to some reasons [2]. I think we should have pgoutput support as
well, so see what is required to get that incorporated.

I have added the rebased patch-set for pgoutput and subscriber side changes as well. This also includes a test case in subscriber.

As mentioned in my email there were some reasons due to which the
support has been left for later, have you checked those and if so, can
you please explain how you have addressed those or why they are not
relevant now if that is the case?

--
With Regards,
Amit Kapila.

On Tue, Sep 15, 2020 at 10:43 PM Amit Kapila <amit.kapila16@gmail.com>
wrote:

Few other comments:
===================
1.
ReorderBufferProcessTXN()
{
..
if (streaming)
{
ReorderBufferTruncateTXN(rb, txn);

/* Reset the CheckXidAlive */
CheckXidAlive = InvalidTransactionId;
}
else
ReorderBufferCleanupTXN(rb, txn);
..
}

I don't think we can perform ReorderBufferCleanupTXN for the prepared
transactions because if we have removed the ReorderBufferTxn before
commit, the later code might not consider such a transaction in the
system and compute the wrong value of restart_lsn for a slot.
Basically, in SnapBuildProcessRunningXacts() when we call
ReorderBufferGetOldestTXN(), it should show the ReorderBufferTxn of
the prepared transaction which is not yet committed but because we
have removed it after prepare, it won't get that TXN and then that
leads to wrong computation of restart_lsn. Once we start from a wrong
point in WAL, the snapshot built was incorrect which will lead to the
wrong result. This is the same reason why the patch is not doing
ReorderBufferForget in DecodePrepare when we decide to skip the
transaction. Also, here, we need to set CheckXidAlive =
InvalidTransactionId; for prepared xact as well.

Just to confirm what you are expecting here. so after we send out the
prepare transaction to the plugin, you are suggesting to NOT do a
ReorderBufferCleanupTXN, but what to do instead?. Are you suggesting to do
what you suggested
as part of concurrent abort handling? Something equivalent
to ReorderBufferTruncateTXN()? remove all changes of the transaction but
keep the invalidations and tuplecids etc? Do you think we should have a new
flag in txn to indicate that this transaction has already been decoded?
(prepare_decoded?) Any other special handling you think is required?

regards,
Ajin Cherian
Fujitsu Australia

On Thu, Sep 17, 2020 at 2:02 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Tue, Sep 15, 2020 at 10:43 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

Few other comments:
===================
1.
ReorderBufferProcessTXN()
{
..
if (streaming)
{
ReorderBufferTruncateTXN(rb, txn);

/* Reset the CheckXidAlive */
CheckXidAlive = InvalidTransactionId;
}
else
ReorderBufferCleanupTXN(rb, txn);
..
}

I don't think we can perform ReorderBufferCleanupTXN for the prepared
transactions because if we have removed the ReorderBufferTxn before
commit, the later code might not consider such a transaction in the
system and compute the wrong value of restart_lsn for a slot.
Basically, in SnapBuildProcessRunningXacts() when we call
ReorderBufferGetOldestTXN(), it should show the ReorderBufferTxn of
the prepared transaction which is not yet committed but because we
have removed it after prepare, it won't get that TXN and then that
leads to wrong computation of restart_lsn. Once we start from a wrong
point in WAL, the snapshot built was incorrect which will lead to the
wrong result. This is the same reason why the patch is not doing
ReorderBufferForget in DecodePrepare when we decide to skip the
transaction. Also, here, we need to set CheckXidAlive =
InvalidTransactionId; for prepared xact as well.

Just to confirm what you are expecting here. so after we send out the prepare transaction to the plugin, you are suggesting to NOT do a ReorderBufferCleanupTXN, but what to do instead?. Are you suggesting to do what you suggested
as part of concurrent abort handling?

Yes.

Something equivalent to ReorderBufferTruncateTXN()? remove all changes of the transaction but keep the invalidations and tuplecids etc?

I don't think you don't need tuplecids. I have checked
ReorderBufferFinishPrepared() and that seems to require only
invalidations, check if anything else is required.

Do you think we should have a new flag in txn to indicate that this transaction has already been decoded? (prepare_decoded?)

Yeah, I think that would be better. How about if name the new variable
as cleanup_prepared?

--
With Regards,
Amit Kapila.

On Tue, Sep 15, 2020 at 10:43 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

I don't think it is complete yet.
*
* This error can only occur when we are sending the data in
* streaming mode and the streaming is not finished yet.
*/
- Assert(streaming);
- Assert(stream_started);
+ Assert(streaming || rbtxn_prepared(txn));
+ Assert(stream_started  || rbtxn_prepared(txn));

Here, you have updated the code but comments are still not updated.

Updated the comments.

I don't think we need to perform abort here. Later we will anyway
encounter the WAL for Rollback Prepared for which we will call
abort_prepared_cb. As we have set the 'concurrent_abort' flag, it will
allow us to skip all the intermediate records. Here, we need only
enough state in ReorderBufferTxn that it can be later used for
ReorderBufferFinishPrepared(). Basically, you need functionality
similar to ReorderBufferTruncateTXN where except for invalidations you
can free memory for everything else. You can either write a new
function ReorderBufferTruncatePreparedTxn or pass another bool
parameter in ReorderBufferTruncateTXN to indicate it is prepared_xact
and then clean up additional things that are not required for prepared
xact.

Added a new parameter to ReorderBufferTruncatePreparedTxn for
prepared transactions and did cleanup of tupulecids as well, I have
left snapshots and transactions.
As a result of this, I also had to create a new function
ReorderBufferCleanupPreparedTXN which will clean up the rest as part
of FinishPrepared handling as we can't call
ReorderBufferCleanupTXN again after this.

*
Similarly, I don't understand why we need below code:
ReorderBufferProcessTXN()
{
..
+ if (rbtxn_rollback(txn))
+ rb->abort(rb, txn, commit_lsn);
..
}

There is nowhere we are setting the RBTXN_ROLLBACK flag, so how will
this check be true? If we decide to remove this code then don't forget
to update the comments.

Removed.

*
If my previous two comments are correct then I don't think we need the
below interface.
+    <sect3 id="logicaldecoding-output-plugin-abort">
+     <title>Transaction Abort Callback</title>
+
+     <para>
+      The required <function>abort_cb</function> callback is called whenever
+      a transaction abort has to be initiated. This can happen if we are
+      decoding a transaction that has been prepared for two-phase commit and
+      a concurrent rollback happens while we are decoding it.
+<programlisting>
+typedef void (*LogicalDecodeAbortCB) (struct LogicalDecodingContext *ctx,
+                                       ReorderBufferTXN *txn,
+                                       XLogRecPtr abort_lsn);

Removed.

I don't know why the patch has used this way to implement an option to
enable two-phase. Can't we use how we implement 'stream-changes'
option in commit 7259736a6e? Just refer how we set ctx->streaming and
you can use a similar way to set this parameter.

Done, I've moved the checks for callbacks to inside the corresponding wrappers.

This is not what I suggested. Please study the commit 7259736a6e and
see how streaming option is implemented. I want later subscribers can
specify whether they want transactions to be decoded at prepare time
similar to what we have done for streaming. Also, search for
ctx->streaming in the code and see how it is set to get the idea.

Changed it similar to ctx->streaming logic.

Note: Please use version number while sending patches, you can use
something like git format-patch -N -v n to do that. It makes easier
for the reviewer to compare it with the previous version.

Done.

Few other comments:
===================
1.
ReorderBufferProcessTXN()
{
..
if (streaming)
{
ReorderBufferTruncateTXN(rb, txn);

/* Reset the CheckXidAlive */
CheckXidAlive = InvalidTransactionId;
}
else
ReorderBufferCleanupTXN(rb, txn);
..
}

I don't think we can perform ReorderBufferCleanupTXN for the prepared
transactions because if we have removed the ReorderBufferTxn before
commit, the later code might not consider such a transaction in the
system and compute the wrong value of restart_lsn for a slot.
Basically, in SnapBuildProcessRunningXacts() when we call
ReorderBufferGetOldestTXN(), it should show the ReorderBufferTxn of
the prepared transaction which is not yet committed but because we
have removed it after prepare, it won't get that TXN and then that
leads to wrong computation of restart_lsn. Once we start from a wrong
point in WAL, the snapshot built was incorrect which will lead to the
wrong result. This is the same reason why the patch is not doing
ReorderBufferForget in DecodePrepare when we decide to skip the
transaction. Also, here, we need to set CheckXidAlive =
InvalidTransactionId; for prepared xact as well.

Updated as suggested above.

2. Have you thought about the interaction of streaming with prepared
transactions? You can try writing some tests using pg_logical* APIs
and see the behaviour. For ex. there is no handling in
ReorderBufferStreamCommit for the same. I think you need to introduce
stream_prepare API similar to stream_commit and then use the same.

This is pending. I will look at it in the next iteration. Also pending
is the investigation as to why the pgoutput changes were not added
initially.

regards,
Ajin Cherian
Fujitsu Australia

On Thu, Sep 17, 2020 at 10:35 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

Yeah, I think that would be better. How about if name the new variable
as cleanup_prepared?

I haven't added a new flag to indicate that the prepare was cleaned
up, as that wasn' really necessary. Instead I used a new function to
do partial cleanup to do whatever was not done in the truncate. If you
think, using a flag and doing special handling in
ReorderBufferCleanupTXN was a better idea, let me know.

regards,
Ajin Cherian
Fujitsu Australia

On Fri, Sep 18, 2020 at 6:02 PM Ajin Cherian <itsajin@gmail.com> wrote:

I have reviewed v4-0001 patch and I have a few comments. I haven't
yet completely reviewed the patch.

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.

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.

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.

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.

--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

On Fri, Sep 18, 2020 at 6:02 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Tue, Sep 15, 2020 at 10:43 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

I don't think it is complete yet.
*
* This error can only occur when we are sending the data in
* streaming mode and the streaming is not finished yet.
*/
- Assert(streaming);
- Assert(stream_started);
+ Assert(streaming || rbtxn_prepared(txn));
+ Assert(stream_started  || rbtxn_prepared(txn));

Here, you have updated the code but comments are still not updated.

Updated the comments.

I don't think we need to perform abort here. Later we will anyway
encounter the WAL for Rollback Prepared for which we will call
abort_prepared_cb. As we have set the 'concurrent_abort' flag, it will
allow us to skip all the intermediate records. Here, we need only
enough state in ReorderBufferTxn that it can be later used for
ReorderBufferFinishPrepared(). Basically, you need functionality
similar to ReorderBufferTruncateTXN where except for invalidations you
can free memory for everything else. You can either write a new
function ReorderBufferTruncatePreparedTxn or pass another bool
parameter in ReorderBufferTruncateTXN to indicate it is prepared_xact
and then clean up additional things that are not required for prepared
xact.

Added a new parameter to ReorderBufferTruncatePreparedTxn for
prepared transactions and did cleanup of tupulecids as well, I have
left snapshots and transactions.
As a result of this, I also had to create a new function
ReorderBufferCleanupPreparedTXN which will clean up the rest as part
of FinishPrepared handling as we can't call
ReorderBufferCleanupTXN again after this.

Why can't we call ReorderBufferCleanupTXN() from
ReorderBufferFinishPrepared after your changes?

+ * If streaming, keep the remaining info - transactions, tuplecids,
invalidations and
+ * snapshots.If after a PREPARE, keep only the invalidations and snapshots.
  */
 static void
-ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
+ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
bool txn_prepared)

Why do we need even snapshot for Prepared transactions? Also, note
that in the comment there is no space before you start a new line.

I don't know why the patch has used this way to implement an option to
enable two-phase. Can't we use how we implement 'stream-changes'
option in commit 7259736a6e? Just refer how we set ctx->streaming and
you can use a similar way to set this parameter.

Done, I've moved the checks for callbacks to inside the corresponding wrappers.

This is not what I suggested. Please study the commit 7259736a6e and
see how streaming option is implemented. I want later subscribers can
specify whether they want transactions to be decoded at prepare time
similar to what we have done for streaming. Also, search for
ctx->streaming in the code and see how it is set to get the idea.

Changed it similar to ctx->streaming logic.

Hmm, I still don't see changes relevant changes in pg_decode_startup().

--
With Regards,
Amit Kapila.

On Sun, Sep 20, 2020 at 11:01 AM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Fri, Sep 18, 2020 at 6:02 PM Ajin Cherian <itsajin@gmail.com> wrote:

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.

Yeah, probably we should have Assert for below three conditions:
+ !SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
+ parsed->dbId == ctx->slot->data.database &&
+ !FilterByOrigin(ctx, origin_id) &&

Your other comments make sense to me.

--
With Regards,
Amit Kapila.

Why can't we call ReorderBufferCleanupTXN() from
ReorderBufferFinishPrepared after your changes?

Since the truncate already removed the changes, it would fail on the
below Assert in ReorderBufferCleanupTXN()
/* Check we're not mixing changes from different transactions. */
Assert(change->txn == txn);

regards.
Ajin Cherian
Fujitsu Australia

On Mon, Sep 21, 2020 at 12:36 PM Ajin Cherian <itsajin@gmail.com> wrote:

Why can't we call ReorderBufferCleanupTXN() from
ReorderBufferFinishPrepared after your changes?

Since the truncate already removed the changes, it would fail on the
below Assert in ReorderBufferCleanupTXN()
/* Check we're not mixing changes from different transactions. */
Assert(change->txn == txn);

The changes list should be empty by that time because we removing each
change from the list:, see code "dlist_delete(&change->node);" in
ReorderBufferTruncateTXN. If you are hitting the Assert as you
mentioned then I think the problem is something else.

--
With Regards,
Amit Kapila.

On Mon, Sep 21, 2020 at 10:20 AM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Sun, Sep 20, 2020 at 11:01 AM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Fri, Sep 18, 2020 at 6:02 PM Ajin Cherian <itsajin@gmail.com> wrote:

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.

Yeah, probably we should have Assert for below three conditions:
+ !SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
+ parsed->dbId == ctx->slot->data.database &&
+ !FilterByOrigin(ctx, origin_id) &&

+1

--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

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

+ /*
+ * 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.

We could enter DecodeAbort even without a prepare, as the code is
common for both XLOG_XACT_ABORT and XLOG_XACT_ABORT_PREPARED. So, the
conditions !SnapBuildXactNeedsSkip, parsed->dbId

== ctx->slot->data.database and !FilterByOrigin could be true but the transaction is not prepared, then we dont need to do a ReorderBufferFinishPrepared (with commit flag false) but called ReorderBufferAbort. But I think there is a problem, if those conditions are in fact false, then we should return without trying to Abort using ReorderBufferAbort, what do you think?

I agree with all your other comments.

regards,
Ajin
Fujitsu Australia

On Mon, Sep 21, 2020 at 3:45 PM Ajin Cherian <itsajin@gmail.com> wrote:

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

+ /*
+ * 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.

We could enter DecodeAbort even without a prepare, as the code is
common for both XLOG_XACT_ABORT and XLOG_XACT_ABORT_PREPARED. So, the
conditions !SnapBuildXactNeedsSkip, parsed->dbId

== ctx->slot->data.database and !FilterByOrigin could be true but the transaction is not prepared, then we dont need to do a ReorderBufferFinishPrepared (with commit flag false) but called ReorderBufferAbort. But I think there is a problem, if those conditions are in fact false, then we should return without trying to Abort using ReorderBufferAbort, what do you think?

I think we need to call ReorderBufferAbort at least to clean up the
TXN. Also, if what you are saying is correct then that should be true
without this patch as well, no? If so, we don't need to worry about it
as far as this patch is concerned.

--
With Regards,
Amit Kapila.

On Mon, Sep 21, 2020 at 9:24 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

I think we need to call ReorderBufferAbort at least to clean up the
TXN. Also, if what you are saying is correct then that should be true
without this patch as well, no? If so, we don't need to worry about it
as far as this patch is concerned.

Yes, that is true. So will change this check to:

if (TransactionIdIsValid(xid) &&
ReorderBufferTxnIsPrepared(ctx->reorder, xid, parsed->twophase_gid)

regards,
Ajin Cherian
Fujitsu Australia

On Mon, Sep 21, 2020 at 5:23 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Mon, Sep 21, 2020 at 9:24 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

I think we need to call ReorderBufferAbort at least to clean up the
TXN. Also, if what you are saying is correct then that should be true
without this patch as well, no? If so, we don't need to worry about it
as far as this patch is concerned.

Yes, that is true. So will change this check to:

if (TransactionIdIsValid(xid) &&
ReorderBufferTxnIsPrepared(ctx->reorder, xid, parsed->twophase_gid)

Yeah and add the Assert for skip conditions as asked above.

--
With Regards,
Amit Kapila.

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

On Tue, Sep 22, 2020 at 5:18 PM Ajin Cherian <itsajin@gmail.com> wrote:

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

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.

If the transaction is prepared which you can ensure via
ReorderBufferTxnIsPrepared() (considering you have a proper check in
that function), it should not require skipping the transaction in
Abort. One way it could happen is if you clean up the ReorderBufferTxn
in Prepare which you were doing in earlier version of patch which I
pointed out was wrong, if you have changed that then I don't know why
it could fail, may be someplace else during prepare the patch is
freeing it. Just check that.

I will need to relook
this logic again in a future patch.

No problem. I think you can handle the other comments and then we can
come back to this and you might want to share the exact details of the
test (may be a narrow down version of the original test) and I or
someone else might be able to help you with that.

--
With Regards,
Amit Kapila.

On Wed, Sep 23, 2020 at 2:39 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

No problem. I think you can handle the other comments and then we can
come back to this and you might want to share the exact details of the
test (may be a narrow down version of the original test) and I or
someone else might be able to help you with that.

--
With Regards,
Amit Kapila.

I have added a new patch for supporting 2 phase commit semantics in
the streaming APIs for the logical decoding plugins. I have added 3
APIs
1. stream_prepare
2. stream_commit_prepared
3. stream_abort_prepared

I have also added the support for the new APIs in test_decoding
plugin. I have not yet added it to pgoutpout.

I have also added a fix for the error I saw while calling
ReorderBufferCleanupTXN as part of FinishPrepared handling. As a
result I have removed the function I added earlier,
ReorderBufferCleanupPreparedTXN.
Please have a look at the new changes and let me know what you think.

I will continue to look at:

1. Remove snapshots on prepare truncate.
2. Bug seen while abort of prepared transaction, the prepared flag is
lost, and not able to make out that it was a previously prepared
transaction.

regards,
Ajin Cherian
Fujitsu Australia

On Mon, Sep 28, 2020 at 1:13 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Wed, Sep 23, 2020 at 2:39 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

I have added a new patch for supporting 2 phase commit semantics in
the streaming APIs for the logical decoding plugins. I have added 3
APIs
1. stream_prepare
2. stream_commit_prepared
3. stream_abort_prepared

I have also added the support for the new APIs in test_decoding
plugin. I have not yet added it to pgoutpout.

I have also added a fix for the error I saw while calling
ReorderBufferCleanupTXN as part of FinishPrepared handling. As a
result I have removed the function I added earlier,
ReorderBufferCleanupPreparedTXN.

Can you explain what was the problem and how you fixed it?

Please have a look at the new changes and let me know what you think.

I will continue to look at:

1. Remove snapshots on prepare truncate.
2. Bug seen while abort of prepared transaction, the prepared flag is
lost, and not able to make out that it was a previously prepared
transaction.

And the support of new APIs in pgoutput, right?

--
With Regards,
Amit Kapila.

On Mon, Sep 28, 2020 at 6:36 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Mon, Sep 28, 2020 at 1:13 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Wed, Sep 23, 2020 at 2:39 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

I have added a new patch for supporting 2 phase commit semantics in
the streaming APIs for the logical decoding plugins. I have added 3
APIs
1. stream_prepare
2. stream_commit_prepared
3. stream_abort_prepared

I have also added the support for the new APIs in test_decoding
plugin. I have not yet added it to pgoutpout.

I have also added a fix for the error I saw while calling
ReorderBufferCleanupTXN as part of FinishPrepared handling. As a
result I have removed the function I added earlier,
ReorderBufferCleanupPreparedTXN.

Can you explain what was the problem and how you fixed it?

When I added the changes for cleaning up tuplecids in
ReorderBufferTruncateTXN, I was not deleting it from the list
(dlist_delete), only calling ReorderBufferReturnChange to free
memory. This logic was copied from ReorderBufferCleanupTXN, there the
lists were all cleaned up in the end, so was not present in each list
cleanup logic.

Please have a look at the new changes and let me know what you think.

I will continue to look at:

1. Remove snapshots on prepare truncate.
2. Bug seen while abort of prepared transaction, the prepared flag is
lost, and not able to make out that it was a previously prepared
transaction.

And the support of new APIs in pgoutput, right?

Yes, that also.

regards,
Ajin Cherian
Fujitsu Australia

On Wed, Sep 23, 2020 at 2:39 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

If the transaction is prepared which you can ensure via
ReorderBufferTxnIsPrepared() (considering you have a proper check in
that function), it should not require skipping the transaction in
Abort. One way it could happen is if you clean up the ReorderBufferTxn
in Prepare which you were doing in earlier version of patch which I
pointed out was wrong, if you have changed that then I don't know why
it could fail, may be someplace else during prepare the patch is
freeing it. Just check that.

I had a look at this problem. The problem happens when decoding is
done after a prepare but before the corresponding rollback
prepared/commit prepared.
For eg:

Begin;
<change 1>
<change 2>
PREPARE TRANSACTION '<prepare#1>';
SELECT data FROM pg_logical_slot_get_changes(...);
:
:
ROLLBACK PREPARED '<prepare#1>';
SELECT data FROM pg_logical_slot_get_changes(...);

Since the prepare is consumed in the first call to
pg_logical_slot_get_changes, subsequently when it is encountered in
the second call, it is skipped (as already decoded) in DecodePrepare
and the txn->flags are not set to
reflect the fact that it was prepared. The same behaviour is seen when
it is commit prepared after the original prepare was consumed.
Initially I was thinking about the following approach to fix it in DecodePrepare
Approach 1:
1. Break the big Skip check in DecodePrepare into 2 parts.
Return if the following conditions are true:
If (parsed->dbId != InvalidOid && parsed->dbId !=
ctx->slot->data.database) ||
ctx->fast_forward || FilterByOrigin(ctx, origin_id))

2. Check If this condition is true:
SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr)

Then this means we are skipping because this has already
been decoded, then instead of returning, call a new function
ReorderBufferMarkPrepare() which will only update the flags in the txn
to indicate that the transaction is prepared
Then later in DecodeAbort or DecodeCommit, we can confirm
that the transaction has been Prepared by checking if the flag is set
and call ReorderBufferFinishPrepared appropriately.

But then, thinking about this some more, I thought of a second approach.
Approach 2:
If the only purpose of all this was to differentiate between
Abort vs Rollback Prepared and Commit vs Commit Prepared, then we dont
need this. We already know the exact operation
in DecodeXactOp and can differentiate there. We only
overloaded DecodeAbort and DecodeCommit for convenience, we can always
call these functions with an extra flag to denote that we are either
commit or aborting a
previously prepared transaction and call
ReorderBufferFinishPrepared accordingly.

Let me know your thoughts on these two approaches or any other
suggestions on this.

regards,
Ajin Cherian
Fujitsu Australia

On Tue, Sep 29, 2020 at 5:08 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Wed, Sep 23, 2020 at 2:39 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

If the transaction is prepared which you can ensure via
ReorderBufferTxnIsPrepared() (considering you have a proper check in
that function), it should not require skipping the transaction in
Abort. One way it could happen is if you clean up the ReorderBufferTxn
in Prepare which you were doing in earlier version of patch which I
pointed out was wrong, if you have changed that then I don't know why
it could fail, may be someplace else during prepare the patch is
freeing it. Just check that.

I had a look at this problem. The problem happens when decoding is
done after a prepare but before the corresponding rollback
prepared/commit prepared.
For eg:

Begin;
<change 1>
<change 2>
PREPARE TRANSACTION '<prepare#1>';
SELECT data FROM pg_logical_slot_get_changes(...);
:
:
ROLLBACK PREPARED '<prepare#1>';
SELECT data FROM pg_logical_slot_get_changes(...);

Since the prepare is consumed in the first call to
pg_logical_slot_get_changes, subsequently when it is encountered in
the second call, it is skipped (as already decoded) in DecodePrepare
and the txn->flags are not set to
reflect the fact that it was prepared. The same behaviour is seen when
it is commit prepared after the original prepare was consumed.
Initially I was thinking about the following approach to fix it in DecodePrepare
Approach 1:
1. Break the big Skip check in DecodePrepare into 2 parts.
Return if the following conditions are true:
If (parsed->dbId != InvalidOid && parsed->dbId !=
ctx->slot->data.database) ||
ctx->fast_forward || FilterByOrigin(ctx, origin_id))

2. Check If this condition is true:
SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr)

Then this means we are skipping because this has already
been decoded, then instead of returning, call a new function
ReorderBufferMarkPrepare() which will only update the flags in the txn
to indicate that the transaction is prepared
Then later in DecodeAbort or DecodeCommit, we can confirm
that the transaction has been Prepared by checking if the flag is set
and call ReorderBufferFinishPrepared appropriately.

But then, thinking about this some more, I thought of a second approach.
Approach 2:
If the only purpose of all this was to differentiate between
Abort vs Rollback Prepared and Commit vs Commit Prepared, then we dont
need this. We already know the exact operation
in DecodeXactOp and can differentiate there. We only
overloaded DecodeAbort and DecodeCommit for convenience, we can always
call these functions with an extra flag to denote that we are either
commit or aborting a
previously prepared transaction and call
ReorderBufferFinishPrepared accordingly.

The second approach sounds better but you can see if there is not much
you want to reuse from DecodeCommit/DecodeAbort then you can even
write new functions DecodeCommitPrepared/DecodeAbortPrepared. OTOH, if
there is a common code among them then passing the flag would be a
better way.

--
With Regards,
Amit Kapila.

On Mon, Sep 28, 2020 at 1:13 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Wed, Sep 23, 2020 at 2:39 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

No problem. I think you can handle the other comments and then we can
come back to this and you might want to share the exact details of the
test (may be a narrow down version of the original test) and I or
someone else might be able to help you with that.

--
With Regards,
Amit Kapila.

I have added a new patch for supporting 2 phase commit semantics in
the streaming APIs for the logical decoding plugins. I have added 3
APIs
1. stream_prepare
2. stream_commit_prepared
3. stream_abort_prepared

I have also added the support for the new APIs in test_decoding
plugin. I have not yet added it to pgoutpout.

I have also added a fix for the error I saw while calling
ReorderBufferCleanupTXN as part of FinishPrepared handling. As a
result I have removed the function I added earlier,
ReorderBufferCleanupPreparedTXN.
Please have a look at the new changes and let me know what you think.

I will continue to look at:

1. Remove snapshots on prepare truncate.
2. Bug seen while abort of prepared transaction, the prepared flag is
lost, and not able to make out that it was a previously prepared
transaction.

I have started looking into you latest patches, as of now I have a
few comments.

v6-0001

@@ -1987,7 +2072,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
prev_lsn = change->lsn;

  /* Set the current xid to detect concurrent aborts. */
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
  {
  curtxn = change->txn;
  SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
  break;
  }
  }
-

For streaming transaction we need to check the xid everytime because
there could concurrent a subtransaction abort, but
for two-phase we don't need to call SetupCheckXidLive everytime,
because we are sure that transaction is going to be
the same throughout the processing.

Apart from this I have also noticed a couple of cosmetic changes

+ {
+ xl_xact_parsed_prepare parsed;
+ xl_xact_prepare *xlrec;
+ /* check that output plugin is capable of twophase decoding */
+ if (!ctx->enable_twophase)
+ {
+ ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
+ break;
+ }

One blank line after variable declations

- /* remove potential on-disk data, and deallocate */
+    /*
+     * remove potential on-disk data, and deallocate.
+     *
+     * We remove it even for prepared transactions (GID is enough to
+     * commit/abort those later).
+     */
+
  ReorderBufferCleanupTXN(rb, txn);

Comment not aligned properly

v6-0003

+LookupGXact(const char *gid)
+{
+ int i;
+
+ LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+
+ for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+ {
+ GlobalTransaction gxact = TwoPhaseState->prepXacts[i];

I think we should take LW_SHARED lowck here no?

--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

On Tue, Sep 29, 2020 at 8:04 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

I have started looking into you latest patches, as of now I have a
few comments.

v6-0001

@@ -1987,7 +2072,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
prev_lsn = change->lsn;

/* Set the current xid to detect concurrent aborts. */
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
{
curtxn = change->txn;
SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-

For streaming transaction we need to check the xid everytime because
there could concurrent a subtransaction abort, but
for two-phase we don't need to call SetupCheckXidLive everytime,
because we are sure that transaction is going to be
the same throughout the processing.

While decoding transactions at 'prepare' time there could be multiple
sub-transactions like in the case below. Won't that be impacted if we
follow your suggestion here?

postgres=# Begin;
BEGIN
postgres=*# insert into t1 values(1,'aaa');
INSERT 0 1
postgres=*# savepoint s1;
SAVEPOINT
postgres=*# insert into t1 values(2,'aaa');
INSERT 0 1
postgres=*# savepoint s2;
SAVEPOINT
postgres=*# insert into t1 values(3,'aaa');
INSERT 0 1
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION

--
With Regards,
Amit Kapila.

On Wed, Sep 30, 2020 at 2:36 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Tue, Sep 29, 2020 at 8:04 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

I have started looking into you latest patches, as of now I have a
few comments.

v6-0001

@@ -1987,7 +2072,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
prev_lsn = change->lsn;

/* Set the current xid to detect concurrent aborts. */
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
{
curtxn = change->txn;
SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-

For streaming transaction we need to check the xid everytime because
there could concurrent a subtransaction abort, but
for two-phase we don't need to call SetupCheckXidLive everytime,
because we are sure that transaction is going to be
the same throughout the processing.

While decoding transactions at 'prepare' time there could be multiple
sub-transactions like in the case below. Won't that be impacted if we
follow your suggestion here?

postgres=# Begin;
BEGIN
postgres=*# insert into t1 values(1,'aaa');
INSERT 0 1
postgres=*# savepoint s1;
SAVEPOINT
postgres=*# insert into t1 values(2,'aaa');
INSERT 0 1
postgres=*# savepoint s2;
SAVEPOINT
postgres=*# insert into t1 values(3,'aaa');
INSERT 0 1
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION

But once we prepare the transaction, we can not rollback individual
subtransaction. We can only rollback the main transaction so instead
of setting individual subxact as CheckXidLive, we can just set the
main XID so no need to check on every command. Just set it before
start processing.

--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

On Wed, Sep 30, 2020 at 2:46 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Wed, Sep 30, 2020 at 2:36 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Tue, Sep 29, 2020 at 8:04 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

I have started looking into you latest patches, as of now I have a
few comments.

v6-0001

@@ -1987,7 +2072,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
prev_lsn = change->lsn;

/* Set the current xid to detect concurrent aborts. */
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
{
curtxn = change->txn;
SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-

For streaming transaction we need to check the xid everytime because
there could concurrent a subtransaction abort, but
for two-phase we don't need to call SetupCheckXidLive everytime,
because we are sure that transaction is going to be
the same throughout the processing.

While decoding transactions at 'prepare' time there could be multiple
sub-transactions like in the case below. Won't that be impacted if we
follow your suggestion here?

postgres=# Begin;
BEGIN
postgres=*# insert into t1 values(1,'aaa');
INSERT 0 1
postgres=*# savepoint s1;
SAVEPOINT
postgres=*# insert into t1 values(2,'aaa');
INSERT 0 1
postgres=*# savepoint s2;
SAVEPOINT
postgres=*# insert into t1 values(3,'aaa');
INSERT 0 1
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION

But once we prepare the transaction, we can not rollback individual
subtransaction.

Sure but Rollback can come before prepare like in the case below which
will appear as concurrent abort (assume there is some DDL which
changes the table before the Rollback statement) because it has
already been done by the backend and that need to be caught by this
mechanism only.

Begin;
insert into t1 values(1,'aaa');
savepoint s1;
insert into t1 values(2,'aaa');
savepoint s2;
insert into t1 values(3,'aaa');
Rollback to savepoint s2;
insert into t1 values(4,'aaa');
Prepare Transaction 'foo';

--
With Regards,
Amit Kapila.

On Wed, Sep 30, 2020 at 3:08 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Wed, Sep 30, 2020 at 2:46 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Wed, Sep 30, 2020 at 2:36 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Tue, Sep 29, 2020 at 8:04 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

I have started looking into you latest patches, as of now I have a
few comments.

v6-0001

@@ -1987,7 +2072,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
prev_lsn = change->lsn;

/* Set the current xid to detect concurrent aborts. */
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
{
curtxn = change->txn;
SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-

For streaming transaction we need to check the xid everytime because
there could concurrent a subtransaction abort, but
for two-phase we don't need to call SetupCheckXidLive everytime,
because we are sure that transaction is going to be
the same throughout the processing.

While decoding transactions at 'prepare' time there could be multiple
sub-transactions like in the case below. Won't that be impacted if we
follow your suggestion here?

postgres=# Begin;
BEGIN
postgres=*# insert into t1 values(1,'aaa');
INSERT 0 1
postgres=*# savepoint s1;
SAVEPOINT
postgres=*# insert into t1 values(2,'aaa');
INSERT 0 1
postgres=*# savepoint s2;
SAVEPOINT
postgres=*# insert into t1 values(3,'aaa');
INSERT 0 1
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION

But once we prepare the transaction, we can not rollback individual
subtransaction.

Sure but Rollback can come before prepare like in the case below which
will appear as concurrent abort (assume there is some DDL which
changes the table before the Rollback statement) because it has
already been done by the backend and that need to be caught by this
mechanism only.

Begin;
insert into t1 values(1,'aaa');
savepoint s1;
insert into t1 values(2,'aaa');
savepoint s2;
insert into t1 values(3,'aaa');
Rollback to savepoint s2;
insert into t1 values(4,'aaa');
Prepare Transaction 'foo';

If we are streaming on the prepare that means we must have decoded
that rollback WAL which means we should have removed the
ReorderBufferTXN for those subxact.

--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

On Wed, Sep 30, 2020 at 3:12 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Wed, Sep 30, 2020 at 3:08 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Wed, Sep 30, 2020 at 2:46 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Wed, Sep 30, 2020 at 2:36 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Tue, Sep 29, 2020 at 8:04 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

I have started looking into you latest patches, as of now I have a
few comments.

v6-0001

@@ -1987,7 +2072,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
prev_lsn = change->lsn;

/* Set the current xid to detect concurrent aborts. */
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
{
curtxn = change->txn;
SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-

For streaming transaction we need to check the xid everytime because
there could concurrent a subtransaction abort, but
for two-phase we don't need to call SetupCheckXidLive everytime,
because we are sure that transaction is going to be
the same throughout the processing.

While decoding transactions at 'prepare' time there could be multiple
sub-transactions like in the case below. Won't that be impacted if we
follow your suggestion here?

postgres=# Begin;
BEGIN
postgres=*# insert into t1 values(1,'aaa');
INSERT 0 1
postgres=*# savepoint s1;
SAVEPOINT
postgres=*# insert into t1 values(2,'aaa');
INSERT 0 1
postgres=*# savepoint s2;
SAVEPOINT
postgres=*# insert into t1 values(3,'aaa');
INSERT 0 1
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION

But once we prepare the transaction, we can not rollback individual
subtransaction.

Sure but Rollback can come before prepare like in the case below which
will appear as concurrent abort (assume there is some DDL which
changes the table before the Rollback statement) because it has
already been done by the backend and that need to be caught by this
mechanism only.

Begin;
insert into t1 values(1,'aaa');
savepoint s1;
insert into t1 values(2,'aaa');
savepoint s2;
insert into t1 values(3,'aaa');
Rollback to savepoint s2;
insert into t1 values(4,'aaa');
Prepare Transaction 'foo';

If we are streaming on the prepare that means we must have decoded
that rollback WAL which means we should have removed the
ReorderBufferTXN for those subxact.

Okay, valid point. We can avoid setting it for each sub-transaction in
that case but OTOH even if we allow to set it there shouldn't be any
bug.

--
With Regards,
Amit Kapila.

On Wed, Sep 30, 2020 at 3:27 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Wed, Sep 30, 2020 at 3:12 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Wed, Sep 30, 2020 at 3:08 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Wed, Sep 30, 2020 at 2:46 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

On Wed, Sep 30, 2020 at 2:36 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Tue, Sep 29, 2020 at 8:04 PM Dilip Kumar <dilipbalaut@gmail.com> wrote:

I have started looking into you latest patches, as of now I have a
few comments.

v6-0001

@@ -1987,7 +2072,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
prev_lsn = change->lsn;

/* Set the current xid to detect concurrent aborts. */
- if (streaming)
+ if (streaming || rbtxn_prepared(change->txn))
{
curtxn = change->txn;
SetupCheckXidLive(curtxn->xid);
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-

For streaming transaction we need to check the xid everytime because
there could concurrent a subtransaction abort, but
for two-phase we don't need to call SetupCheckXidLive everytime,
because we are sure that transaction is going to be
the same throughout the processing.

While decoding transactions at 'prepare' time there could be multiple
sub-transactions like in the case below. Won't that be impacted if we
follow your suggestion here?

postgres=# Begin;
BEGIN
postgres=*# insert into t1 values(1,'aaa');
INSERT 0 1
postgres=*# savepoint s1;
SAVEPOINT
postgres=*# insert into t1 values(2,'aaa');
INSERT 0 1
postgres=*# savepoint s2;
SAVEPOINT
postgres=*# insert into t1 values(3,'aaa');
INSERT 0 1
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION

But once we prepare the transaction, we can not rollback individual
subtransaction.

Sure but Rollback can come before prepare like in the case below which
will appear as concurrent abort (assume there is some DDL which
changes the table before the Rollback statement) because it has
already been done by the backend and that need to be caught by this
mechanism only.

Begin;
insert into t1 values(1,'aaa');
savepoint s1;
insert into t1 values(2,'aaa');
savepoint s2;
insert into t1 values(3,'aaa');
Rollback to savepoint s2;
insert into t1 values(4,'aaa');
Prepare Transaction 'foo';

If we are streaming on the prepare that means we must have decoded
that rollback WAL which means we should have removed the
ReorderBufferTXN for those subxact.

Okay, valid point. We can avoid setting it for each sub-transaction in
that case but OTOH even if we allow to set it there shouldn't be any
bug.

Right, there will not be any bug, just an optimization.

--
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

Hello Ajin.

I have done some review of the v6 patches.

I had some difficulty replying my review comments to the OSS list, so
I am putting them as an attachment here.

Kind Regards,
Peter Smith
Fujitsu Australia

Hello Ajin.

I have gone through the v6 patch changes and have a list of review
comments below.

Apologies for the length of this email - I know that many of the
following comments are trivial, but I figured I should either just
ignore everything cosmetic, or list everything regardless. I chose the
latter.

There may be some duplication where the same review comment is written
for multiple files and/or where the same file is in your multiple
patches.

Kind Regards.
Peter Smith
Fujitsu Australia

[BEGIN]

==========
Patch V6-0001, File: contrib/test_decoding/expected/prepared.out (so
prepared.sql also)
==========

COMMENT
Line 30 - The INSERT INTO test_prepared1 VALUES (2); is kind of
strange because it is not really part of the prior test nor the
following test. Maybe it would be better to have a comment describing
the purpose of this isolated INSERT and to also consume the data from
the slot so it does not get jumbled with the data of the following
(abort) test.

;

COMMENT
Line 53 - Same comment for this test INSERT INTO test_prepared1 VALUES
(4); It kind of has nothing really to do with either the prior (abort)
test nor the following (ddl) test.

;

COMMENT
Line 60 - Seems to check which locks are held for the test_prepared_1
table while the transaction is in progress. Maybe it would be better
to have more comments describing what is expected here and why.

;

COMMENT
Line 88 - There is a comment in the test saying "-- We should see '7'
before '5' in our results since it commits first." but I did not see
any test code that actually verifies that happens.

;

QUESTION
Line 120 - I did not really understand the SQL checking the pg_class.
I expected this would be checking table 'test_prepared1' instead. Can
you explain it?
SELECT 'pg_class' AS relation, locktype, mode
FROM pg_locks
WHERE locktype = 'relation'
AND relation = 'pg_class'::regclass;
relation | locktype | mode
----------+----------+------
(0 rows)

;

QUESTION
Line 139 - SET statement_timeout = '1s'; is 1 seconds short enough
here for this test, or might it be that these statements would be
completed in less than one seconds anyhow?

;

QUESTION
Line 163 - How is this testing a SAVEPOINT? Or is it only to check
that the SAVEPOINT command is not part of the replicated changes?

;

COMMENT
Line 175 - Missing underscore in comment. Code requires also underscore:
"nodecode" --> "_nodecode"

==========
Patch V6-0001, File: contrib/test_decoding/test_decoding.c
==========

COMMENT
Line 43
@@ -36,6 +40,7 @@ typedef struct
bool skip_empty_xacts;
bool xact_wrote_changes;
bool only_local;
+ TransactionId check_xid; /* track abort of this txid */
} TestDecodingData;

The "check_xid" seems a meaningless name. Check what?
IIUC maybe should be something like "check_xid_aborted"

;

COMMENT
Line 105
@ -88,6 +93,19 @@ static void
pg_decode_stream_truncate(LogicalDecodingContext *ctx,
 ReorderBufferTXN *txn,
 int nrelations, Relation relations[],
 ReorderBufferChange *change);
+static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,

Remove extra blank line after these functions

;

COMMENT
Line 149
@@ -116,6 +134,11 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
 cb->stream_change_cb = pg_decode_stream_change;
 cb->stream_message_cb = pg_decode_stream_message;
 cb->stream_truncate_cb = pg_decode_stream_truncate;
+ cb->filter_prepare_cb = pg_decode_filter_prepare;
+ cb->prepare_cb = pg_decode_prepare_txn;
+ cb->commit_prepared_cb = pg_decode_commit_prepared_txn;
+ cb->abort_prepared_cb = pg_decode_abort_prepared_txn;
+
 }

There is a confusing mix of terminology where sometimes things are
referred as ROLLBACK/rollback and other times apparently the same
operation is referred as ABORT/abort. I do not know the root cause of
this mixture. IIUC maybe the internal functions and protocol generally
use the term "abort", whereas the SQL syntax is "ROLLBACK"... but
where those two terms collide in the middle it gets quite confusing.

At least I thought the names of the "callbacks" which get exposed to
the user (e.g. in the help) might be better if they would match the
SQL.
"abort_prepared_cb" --> "rollback_prepared_db"

There are similar review comments like this below where the
alternating terms caused me some confusion.

~

Also, Remove the extra blank line before the end of the function.

;

COMMENT
Line 267
@ -227,6 +252,42 @@ pg_decode_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt,
 errmsg("could not parse value \"%s\" for parameter \"%s\"",
 strVal(elem->arg), elem->defname)));
 }
+ else if (strcmp(elem->defname, "two-phase-commit") == 0)
+ {
+ if (elem->arg == NULL)
+ continue;

IMO the "check-xid" code might be better rearranged so the NULL check
is first instead of if/else.
e.g.
if (elem->arg == NULL)
ereport(FATAL,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("check-xid needs an input value")));
~

Also, is it really supposed to be FATAL instead or ERROR. That is not
the same as the other surrounding code.

;

COMMENT
Line 296
if (data->check_xid <= 0)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("Specify positive value for parameter \"%s\","
" you specified \"%s\"",
elem->defname, strVal(elem->arg))));

The code checking for <= 0 seems over-complicated. Because conversion
was using strtoul() I fail to see how this can ever be < 0. Wouldn't
it be easier to simply test the result of the strtoul() function?

BEFORE: if (errno == EINVAL || errno == ERANGE)
AFTER: if (data->check_xid == 0)

~

Also, should this be FATAL? Everything else similar is ERROR.

;

COMMENT
(general)
I don't recall seeing any of these decoding options (e.g.
"two-phase-commit", "check-xid") documented anywhere.
So how can a user even know these options exist so they can use them?
Perhaps options should be described on this page?
https://www.postgresql.org/docs/13/functions-admin.html#FUNCTIONS-REPLICATION

;

COMMENT
(general)
"check-xid" is a meaningless option name. Maybe something like
"checked-xid-aborted" is more useful?
Suggest changing the member, the option, and the error messages to
match some better name.

;

COMMENT
Line 314
@@ -238,6 +299,7 @@ pg_decode_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt,
}

ctx->streaming &= enable_streaming;
+ ctx->enable_twophase &= enable_2pc;
}

The "ctx->enable_twophase" is inconsistent naming with the
"ctx->streaming" member.
"enable_twophase" --> "twophase"

;

COMMENT
Line 374
@@ -297,6 +359,94 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
OutputPluginWrite(ctx, true);
}

+
+/*
+ * Filter out two-phase transactions.
+ *
+ * Each plugin can implement its own filtering logic. Here
+ * we demonstrate a simple logic by checking the GID. If the
+ * GID contains the "_nodecode" substring, then we filter
+ * it out.
+ */
+static bool
+pg_decode_filter_prepare(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Remove the extra preceding blank line.

~

I did not find anything in the help about "_nodecode". Should it be
there or is this deliberately not documented feature?

;

QUESTION
Line 440
+pg_decode_abort_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,

Is this a wrong comment
"ABORT PREPARED" --> "ROLLBACK PREPARED" ??

;

COMMENT
Line 620
@@ -455,6 +605,22 @@ pg_decode_change(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
}
data->xact_wrote_changes = true;

+ /* if check_xid is specified */
+ if (TransactionIdIsValid(data->check_xid))
+ {
+ elog(LOG, "waiting for %u to abort", data->check_xid);
+ while (TransactionIdIsInProgress(dat

The check_xid seems a meaningless name, and the comment "/* if
check_xid is specified */" was not helpful either.
IIUC purpose of this is to check that the nominated xid always is rolled back.
So the appropriate name may be more like "check-xid-aborted".

;

==========
Patch V6-0001, File: doc/src/sgml/logicaldecoding.sgml
==========

COMMENT/QUESTION
Section 48.6.1
@ -387,6 +387,10 @@ typedef struct OutputPluginCallbacks
 LogicalDecodeTruncateCB truncate_cb;
 LogicalDecodeCommitCB commit_cb;
 LogicalDecodeMessageCB message_cb;
+ LogicalDecodeFilterPrepareCB filter_prepare_cb;

Confused by the mixing of terminologies "abort" and "rollback".
Why is it LogicalDecodeAbortPreparedCB instead of
LogicalDecodeRollbackPreparedCB?
Why is it abort_prepared_cb instead of rollback_prepared_cb;?

I thought everything the user sees should be ROLLBACK/rollback (like
the SQL) regardless of what the internal functions might be called.

;

COMMENT
Section 48.6.1
The begin_cb, change_cb and commit_cb callbacks are required, while
startup_cb, filter_by_origin_cb, truncate_cb, and shutdown_cb are
optional. If truncate_cb is not set but a TRUNCATE is to be decoded,
the action will be ignored.

The 1st paragraph beneath the typedef does not mention the newly added
callbacks to say if they are required or optional.

;

COMMENT
Section 48.6.4.5
Section 48.6.4.6
Section 48.6.4.7
@@ -578,6 +588,55 @@ typedef void (*LogicalDecodeCommitCB) (struct
LogicalDecodingContext *ctx,
</para>
</sect3>

+ <sect3 id="logicaldecoding-output-plugin-prepare">
+    <sect3 id="logicaldecoding-output-plugin-commit-prepared">
+    <sect3 id="logicaldecoding-output-plugin-abort-prepared">
+<programlisting>

The wording and titles are a bit backwards compared to the others.
e.g. previously was "Transaction Begin" (not "Begin Transaction") and
"Transaction End" (not "End Transaction").

So for consistently following the existing IMO should change these new
titles (and wording) to:
- "Commit Prepared Transaction Callback" --> "Transaction Commit
Prepared Callback"
- "Rollback Prepared Transaction Callback" --> "Transaction Rollback
Prepared Callback"
- "whenever a commit prepared transaction has been decoded" -->
"whenever a transaction commit prepared has been decoded"
- "whenever a rollback prepared transaction has been decoded." -->
"whenever a transaction rollback prepared has been decoded."

;

==========
Patch V6-0001, File: src/backend/replication/logical/decode.c
==========

COMMENT
Line 74
@@ -70,6 +70,9 @@ static void DecodeCommit(LogicalDecodingContext
*ctx, XLogRecordBuffer *buf,
 xl_xact_parsed_commit *parsed, TransactionId xid);
 static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 xl_xact_parsed_abort *parsed, TransactionId xid);
+static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
+ xl_xact_parsed_prepare * parsed);

The 2nd line of DecodePrepare is misaligned by one space.

;

COMMENT
Line 321
@@ -312,17 +315,34 @@ DecodeXactOp(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf)
 }
 break;
 case XLOG_XACT_PREPARE:
+ {
+ xl_xact_parsed_prepare parsed;
+ xl_xact_prepare *xlrec;
+ /* check that output plugin is capable of twophase decoding */

"twophase" --> "two-phase"

~

Also, add a blank line after the declarations.

;

==========
Patch V6-0001, File: src/backend/replication/logical/logical.c
==========

COMMENT
Line 249
@@ -225,6 +237,19 @@ StartupDecodingContext(List *output_plugin_options,
(ctx->callbacks.stream_message_cb != NULL) ||
(ctx->callbacks.stream_truncate_cb != NULL);

+ /*
+ * To support two phase logical decoding, we require
prepare/commit-prepare/abort-prepare
+ * callbacks. The filter-prepare callback is optional. We however
enable two phase logical
+ * decoding when at least one of the methods is enabled so that we
can easily identify
+ * missing methods.

The terminology is generally well known as "two-phase" (with the
hyphen) https://en.wikipedia.org/wiki/Two-phase_commit_protocol so
let's be consistent for all the patch code comments. Please search the
code and correct this in all places, even where I might have missed to
identify it.

"two phase" --> "two-phase"

;

COMMENT
Line 822
@@ -782,6 +807,111 @@ commit_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
}

 static void
+prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr prepare_lsn)

"support 2 phase" --> "supports two-phase" in the comment

;

COMMENT
Line 844
Code condition seems strange and/or broken.
if (ctx->enable_twophase && ctx->callbacks.prepare_cb == NULL)
Because if the flag is null then this condition is skipped.
But then if the callback was also NULL then attempting to call it to
"do the actual work" will give NPE.

~

Also, I wonder should this check be the first thing in this function?
Because if it fails does it even make sense that all the errcallback
code was set up?
E.g errcallback.arg potentially is left pointing to a stack variable
on a stack that no longer exists.

;

COMMENT
Line 857
+commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

"support 2 phase" --> "supports two-phase" in the comment

~

Also, Same potential trouble with the condition:
if (ctx->enable_twophase && ctx->callbacks.commit_prepared_cb == NULL)
Same as previously asked. Should this check be first thing in this function?

;

COMMENT
Line 892
+abort_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

"support 2 phase" --> "supports two-phase" in the comment

~

Same potential trouble with the condition:
if (ctx->enable_twophase && ctx->callbacks.abort_prepared_cb == NULL)
Same as previously asked. Should this check be the first thing in this function?

;

COMMENT
Line 1013
@@ -858,6 +988,51 @@ truncate_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
error_context_stack = errcallback.previous;
}

+static bool
+filter_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ TransactionId xid, const char *gid)

Fix wording in comment:
"twophase" --> "two-phase transactions"
"twophase transactions" --> "two-phase transactions"

==========
Patch V6-0001, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
Line 255
@@ -251,7 +251,8 @@ static Size
ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn
 static void ReorderBufferRestoreChange(ReorderBuffer *rb,
ReorderBufferTXN *txn,
 char *change);
 static void ReorderBufferRestoreCleanup(ReorderBuffer *rb,
ReorderBufferTXN *txn);
-static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
+static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
+ bool txn_prepared);

The alignment is inconsistent. One more space needed before "bool txn_prepared"

;

COMMENT
Line 417
@@ -413,6 +414,11 @@ ReorderBufferReturnTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
}

 /* free data that's contained */
+ if (txn->gid != NULL)
+ {
+ pfree(txn->gid);
+ txn->gid = NULL;
+ }

Should add the blank link before this new code, as it was before.

;

COMMENT
Line 1564
@ -1502,12 +1561,14 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
}

 /*
- * Discard changes from a transaction (and subtransactions), after streaming
- * them. Keep the remaining info - transactions, tuplecids, invalidations and
- * snapshots.
+ * Discard changes from a transaction (and subtransactions), either
after streaming or
+ * after a PREPARE.

typo "snapshots.If" -> "snapshots. If"

;

COMMENT/QUESTION
Line 1590
@@ -1526,7 +1587,7 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
Assert(rbtxn_is_known_subxact(subtxn));
Assert(subtxn->nsubtxns == 0);

- ReorderBufferTruncateTXN(rb, subtxn);
+ ReorderBufferTruncateTXN(rb, subtxn, txn_prepared);
 }

There are some code paths here I did not understand how they match the comments.
Because this function is recursive it seems that it may be called
where the 2nd parameter txn is a sub-transaction.

But then this seems at odds with some of the other code comments of
this function which are processing the txn without ever testing is it
really toplevel or not:

e.g. Line 1593 "/* cleanup changes in the toplevel txn */"
e.g. Line 1632 "They are always stored in the toplevel transaction."

;

COMMENT
Line 1644
@@ -1560,9 +1621,33 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
 * about the toplevel xact (we send the XID in all messages), but we never
 * stream XIDs of empty subxacts.
 */
- if ((!txn->toptxn) || (txn->nentries_mem != 0))
+ if ((!txn_prepared) && ((!txn->toptxn) || (txn->nentries_mem != 0)))
 txn->txn_flags |= RBTXN_IS_STREAMED;

+ if (txn_prepared)

/* remove the change from it's containing list */
typo "it's" --> "its"

;

QUESTION
Line 1977
@@ -1880,7 +1965,7 @@ ReorderBufferResetTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
 ReorderBufferChange *specinsert)
 {
 /* Discard the changes that we just streamed */
- ReorderBufferTruncateTXN(rb, txn);
+ ReorderBufferTruncateTXN(rb, txn, false);

How do you know the 3rd parameter - i.e. txn_prepared - should be
hardwired false here?
e.g. I thought that maybe rbtxn_prepared(txn) can be true here.

;

COMMENT
Line 2345
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-
/*

Looks like accidental blank line deletion. This should be put back how it was

;

COMMENT/QUESTION
Line 2374
@@ -2278,7 +2362,16 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
 }
 }
 else
- rb->commit(rb, txn, commit_lsn);
+ {
+ /*
+ * Call either PREPARE (for twophase transactions) or COMMIT
+ * (for regular ones).

"twophase" --> "two-phase"

~

Also, I was confused by the apparent assumption of exclusiveness of
streaming and 2PC...
e.g. what if streaming AND 2PC then it won't do rb->prepare()

;

QUESTION
Line 2424
@@ -2319,11 +2412,17 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
 */
 if (streaming)
 {
- ReorderBufferTruncateTXN(rb, txn);
+ ReorderBufferTruncateTXN(rb, txn, false);

/* Reset the CheckXidAlive */
CheckXidAlive = InvalidTransactionId;
}
+ else if (rbtxn_prepared(txn))

I was confused by the exclusiveness of streaming/2PC.
e.g. what if streaming AND 2PC at same time - how can you pass false
as 3rd param to ReorderBufferTruncateTXN?

;

COMMENT
Line 2463
@@ -2352,17 +2451,18 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,

 /*
 * The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
- * abort of the (sub)transaction we are streaming. We need to do the
+ * abort of the (sub)transaction we are streaming or preparing. We
need to do the
 * cleanup and return gracefully on this error, see SetupCheckXidLive.
 */

"twoi phase" --> "two-phase"

;

QUESTIONS
Line 2482
@@ -2370,10 +2470,19 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
errdata = NULL;
curtxn->concurrent_abort = true;

- /* Reset the TXN so that it is allowed to stream remaining data. */
- ReorderBufferResetTXN(rb, txn, snapshot_now,
- command_id, prev_lsn,
- specinsert);
+ /* If streaming, reset the TXN so that it is allowed to stream
remaining data. */
+ if (streaming)

Re: /* If streaming, reset the TXN so that it is allowed to stream
remaining data. */
I was confused by the exclusiveness of streaming/2PC.
Is it not possible for streaming flags and rbtxn_prepared(txn) true at
the same time?

~

elog(LOG, "stopping decoding of %s (%u)",
txn->gid[0] != '\0'? txn->gid:"", txn->xid);

Is this a safe operation, or do you also need to test txn->gid is not NULL?

;

COMMENT
Line 2606
+ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,

"twophase" --> "two-phase"

;

QUESTION
Line 2655
+ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,

"This is used to handle COMMIT/ABORT PREPARED"
Should that say "COMMIT/ROLLBACK PREPARED"?

;

COMMENT
Line 2668

"Anyways, 2PC transactions" --> "Anyway, two-phase transactions"

;

COMMENT
Line 2765
@@ -2495,7 +2731,13 @@ ReorderBufferAbort(ReorderBuffer *rb,
TransactionId xid, XLogRecPtr lsn)
/* cosmetic... */
txn->final_lsn = lsn;

- /* remove potential on-disk data, and deallocate */
+ /*
+ * remove potential on-disk data, and deallocate.
+ *

Remove the blank between the comment and code.

==========
Patch V6-0001, File: src/include/replication/logical.h
==========

COMMENT
Line 89

"two phase" -> "two-phase"

;

COMMENT
Line 89

For consistency with the previous member naming really the new member
should just be called "twophase" rather than "enable_twophase"

;

==========
Patch V6-0001, File: src/include/replication/output_plugin.h
==========

QUESTION
Line 106

As previously asked, why is the callback function/typedef referred as
AbortPrepared instead of RollbackPrepared?
It does not match the SQL and the function comment, and seems only to
add some unnecessary confusion.

;

==========
Patch V6-0001, File: src/include/replication/reorderbuffer.h
==========

QUESTION
Line 116
@@ -162,9 +163,13 @@ typedef struct ReorderBufferChange
 #define RBTXN_HAS_CATALOG_CHANGES 0x0001
 #define RBTXN_IS_SUBXACT 0x0002
 #define RBTXN_IS_SERIALIZED 0x0004
-#define RBTXN_IS_STREAMED 0x0008
-#define RBTXN_HAS_TOAST_INSERT 0x0010
-#define RBTXN_HAS_SPEC_INSERT 0x0020
+#define RBTXN_PREPARE 0x0008
+#define RBTXN_COMMIT_PREPARED 0x0010
+#define RBTXN_ROLLBACK_PREPARED 0x0020
+#define RBTXN_COMMIT 0x0040
+#define RBTXN_IS_STREAMED 0x0080
+#define RBTXN_HAS_TOAST_INSERT 0x0100
+#define RBTXN_HAS_SPEC_INSERT 0x0200

I was wondering why when adding new flags, some of the existing flag
masks were also altered.
I am assuming this is ok because they are never persisted but are only
used in the protocol (??)

;

COMMENT
Line 226
@@ -218,6 +223,15 @@ typedef struct ReorderBufferChange
((txn)->txn_flags & RBTXN_IS_STREAMED) != 0 \
)

+/* is this txn prepared? */
+#define rbtxn_prepared(txn) (txn->txn_flags & RBTXN_PREPARE)
+/* was this prepared txn committed in the meanwhile? */
+#define rbtxn_commit_prepared(txn) (txn->txn_flags & RBTXN_COMMIT_PREPARED)
+/* was this prepared txn aborted in the meanwhile? */
+#define rbtxn_rollback_prepared(txn) (txn->txn_flags & RBTXN_ROLLBACK_PREPARED)
+/* was this txn committed in the meanwhile? */
+#define rbtxn_commit(txn) (txn->txn_flags & RBTXN_COMMIT)
+

Probably all the "txn->txn_flags" here might be more safely written
with parentheses in the macro like "(txn)->txn_flags".

~

Also, Start all comments with capital. And what is the meaning "in the
meanwhile?"

;

COMMENT
Line 410
@@ -390,6 +407,39 @@ typedef void (*ReorderBufferCommitCB) (ReorderBuffer *rb,
ReorderBufferTXN *txn,
XLogRecPtr commit_lsn);

The format is inconsistent with all other callback signatures here,
where the 1st arg was on the same line as the typedef.

;

COMMENT
Line 440-442

Excessive blank lines following this change?

;

COMMENT
Line 638
@@ -571,6 +631,15 @@ void
ReorderBufferXidSetCatalogChanges(ReorderBuffer *, TransactionId xid,
XLog
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *, TransactionId xid);
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *, TransactionId xid);

+bool ReorderBufferPrepareNeedSkip(ReorderBuffer *rb, TransactionId xid,
+ const char *gid);
+bool ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid,
+ const char *gid);
+void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
+ XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
+ TimestampTz commit_time,
+ RepOriginId origin_id, XLogRecPtr origin_lsn,
+ char *gid);

Not aligned consistently with other function prototypes.

;

==========
Patch V6-0003, File: src/backend/access/transam/twophase.c
==========

COMMENT
Line 551
@@ -548,6 +548,37 @@ MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
}

 /*
+ * LookupGXact
+ * Check if the prepared transaction with the given GID is around
+ */
+bool
+LookupGXact(const char *gid)

There is potential to refactor/simplify this code:
e.g.

bool
LookupGXact(const char *gid)
{
int i;
bool found = false;

LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
{
GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
/* Ignore not-yet-valid GIDs */
if (gxact->valid && strcmp(gxact->gid, gid) == 0)
{
found = true;
break;
}
}
LWLockRelease(TwoPhaseStateLock);
return found;
}

;

==========
Patch V6-0003, File: src/backend/replication/logical/proto.c
==========

COMMENT
Line 86
@@ -72,12 +72,17 @@ logicalrep_read_begin(StringInfo in,
LogicalRepBeginData *begin_data)
*/
void
logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn,
- XLogRecPtr commit_lsn)

Since now the flags are used the code comment is wrong.
"/* send the flags field (unused for now) */"

;

COMMENT
Line 129
@ -106,6 +115,77 @@ logicalrep_read_commit(StringInfo in,
LogicalRepCommitData *commit_data)
}

 /*
+ * Write PREPARE to the output stream.
+ */
+void
+logicalrep_write_prepare(StringInfo out, ReorderBufferTXN *txn,

"2PC transactions" --> "two-phase commit transactions"

;

COMMENT
Line 133

Assert(strlen(txn->gid) > 0);
Shouldn't that assertion also check txn->gid is not NULL (to prevent
NPE in case gid was NULL)

;

COMMENT
Line 177
+logicalrep_read_prepare(StringInfo in, LogicalRepPrepareData * prepare_data)

prepare_data->prepare_type = flags;
This code may be OK but it does seem a bit of an abuse of the flags.

e.g. Are they flags or are the really enum values?
e.g. And if they are effectively enums (it appears they are) then
seemed inconsistent that |= was used when they were previously
assigned.

;

==========
Patch V6-0003, File: src/backend/replication/logical/worker.c
==========

COMMENT
Line 757
@@ -749,6 +753,141 @@ apply_handle_commit(StringInfo s)
pgstat_report_activity(STATE_IDLE, NULL);
}

+static void
+apply_handle_prepare_txn(LogicalRepPrepareData * prepare_data)
+{
+ Assert(prepare_data->prepare_lsn == remote_final_lsn);

Missing function comment to say this is called from apply_handle_prepare.

;

COMMENT
Line 798
+apply_handle_commit_prepared_txn(LogicalRepPrepareData * prepare_data)

Missing function comment to say this is called from apply_handle_prepare.

;

COMMENT
Line 824
+apply_handle_rollback_prepared_txn(LogicalRepPrepareData * prepare_data)

Missing function comment to say this is called from apply_handle_prepare.

==========
Patch V6-0003, File: src/backend/replication/pgoutput/pgoutput.c
==========

COMMENT
Line 50
@@ -47,6 +47,12 @@ static void pgoutput_truncate(LogicalDecodingContext *ctx,
 ReorderBufferChange *change);
 static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
 RepOriginId origin_id);
+static void pgoutput_prepare_txn(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn, XLogRecPtr prepare_lsn);

The parameter indentation (2nd lines) does not match everything else
in this context.

;

COMMENT
Line 152
@@ -143,6 +149,10 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
 cb->change_cb = pgoutput_change;
 cb->truncate_cb = pgoutput_truncate;
 cb->commit_cb = pgoutput_commit_txn;
+
+ cb->prepare_cb = pgoutput_prepare_txn;
+ cb->commit_prepared_cb = pgoutput_commit_prepared_txn;
+ cb->abort_prepared_cb = pgoutput_abort_prepared_txn;

Remove the unnecessary blank line.

;

QUESTION
Line 386
@@ -373,7 +383,49 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
OutputPluginUpdateProgress(ctx);

 OutputPluginPrepareWrite(ctx, true);
- logicalrep_write_commit(ctx->out, txn, commit_lsn);
+ logicalrep_write_commit(ctx->out, txn, commit_lsn, true);

Is the is_commit parameter of logicalrep_write_commit ever passed as false?
If yes, where?
If no, the what is the point of it?

;

COMMENT
Line 408
+pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,

Since all this function is identical to pg_output_prepare it might be
better to either
1. just leave this as a wrapper to delegate to that function
2. remove this one entirely and assign the callback to the common
pgoutput_prepare_txn

;

COMMENT
Line 419
+pgoutput_abort_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Since all this function is identical to pg_output_prepare if might be
better to either
1. just leave this as a wrapper to delegate to that function
2. remove this one entirely and assign the callback to the common
pgoutput_prepare_tx

;

COMMENT
Line 419
+pgoutput_abort_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Shouldn't this comment say be "ROLLBACK PREPARED"?

;

==========
Patch V6-0003, File: src/include/replication/logicalproto.h
==========

QUESTION
Line 101
@@ -87,20 +87,55 @@ typedef struct LogicalRepBeginData
TransactionId xid;
} LogicalRepBeginData;

+/* Commit (and abort) information */

#define LOGICALREP_IS_ABORT 0x02
Is there a good reason why this is not called:
#define LOGICALREP_IS_ROLLBACK 0x02

;

COMMENT
Line 105

((flags == LOGICALREP_IS_COMMIT) || (flags == LOGICALREP_IS_ABORT))

Macros would be safer if flags are in parentheses
(((flags) == LOGICALREP_IS_COMMIT) || ((flags) == LOGICALREP_IS_ABORT))

;

COMMENT
Line 115

Unexpected whitespace for the typedef
"} LogicalRepPrepareData;"

;

COMMENT
Line 122
/* prepare can be exactly one of PREPARE, [COMMIT|ABORT] PREPARED*/
#define PrepareFlagsAreValid(flags) \
((flags == LOGICALREP_IS_PREPARE) || \
(flags == LOGICALREP_IS_COMMIT_PREPARED) || \
(flags == LOGICALREP_IS_ROLLBACK_PREPARED))

There is confusing mixture in macros and comments of ABORT and ROLLBACK terms
"[COMMIT|ABORT] PREPARED" --> "[COMMIT|ROLLBACK] PREPARED"

~

Also, it would be safer if flags are in parentheses
(((flags) == LOGICALREP_IS_PREPARE) || \
((flags) == LOGICALREP_IS_COMMIT_PREPARED) || \
((flags) == LOGICALREP_IS_ROLLBACK_PREPARED))

;

==========
Patch V6-0003, File: src/test/subscription/t/020_twophase.pl
==========

COMMENT
Line 131 - # check inserts are visible

Isn't this supposed to be checking for rows 12 and 13, instead of 11 and 12?

;

==========
Patch V6-0004, File: contrib/test_decoding/test_decoding.c
==========

COMMENT
Line 81
@@ -78,6 +78,15 @@ static void
pg_decode_stream_stop(LogicalDecodingContext *ctx,
 static void pg_decode_stream_abort(LogicalDecodingContext *ctx,
 ReorderBufferTXN *txn,
 XLogRecPtr abort_lsn);
+static void pg_decode_stream_prepare(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+static

All these functions have a 3rd parameter called commit_lsn. Even
though the functions are not commit related. It seems like a cut/paste
error.

;

COMMENT
Line 142
@@ -130,6 +139,9 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
 cb->stream_start_cb = pg_decode_stream_start;
 cb->stream_stop_cb = pg_decode_stream_stop;
 cb->stream_abort_cb = pg_decode_stream_abort;
+ cb->stream_prepare_cb = pg_decode_stream_prepare;
+ cb->stream_commit_prepared_cb = pg_decode_stream_commit_prepared;
+ cb->stream_abort_prepared_cb = pg_decode_stream_abort_prepared;
 cb->stream_commit_cb = pg_decode_stream_commit;

Can the "cb->stream_abort_prepared_cb" be changed to
"cb->stream_rollback_prepared_cb"?

;

COMMENT
Line 827
@@ -812,6 +824,78 @@ pg_decode_stream_abort(LogicalDecodingContext *ctx,
}

 static void
+pg_decode_stream_prepare(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn)
+{
+ TestDecodingData *data = ctx->output_plugin_pr

The commit_lsn (3rd parameter) is unused and seems like a cut/paste name error.

;

COMMENT
Line 875
+pg_decode_stream_abort_prepared(LogicalDecodingContext *ctx,

The commit_lsn (3rd parameter) is unused and seems like a cut/paste name error.

;

==========
Patch V6-0004, File: doc/src/sgml/logicaldecoding.sgml
==========

COMMENT
48.6.1
@@ -396,6 +396,9 @@ typedef struct OutputPluginCallbacks
 LogicalDecodeStreamStartCB stream_start_cb;
 LogicalDecodeStreamStopCB stream_stop_cb;
 LogicalDecodeStreamAbortCB stream_abort_cb;
+ LogicalDecodeStreamPrepareCB stream_prepare_cb;
+ LogicalDecodeStreamCommitPreparedCB stream_commit_prepared_cb;
+ LogicalDecodeStreamAbortPreparedCB stream_abort_prepared_cb;

Same question from previous review comments - why using the
terminology "abort" instead of "rollback"

;

COMMENT
48.6.1
@@ -418,7 +421,9 @@ typedef void (*LogicalOutputPluginInit) (struct
OutputPluginCallbacks *cb);
 in-progress transactions. The <function>stream_start_cb</function>,
 <function>stream_stop_cb</function>, <function>stream_abort_cb</function>,
 <function>stream_commit_cb</function> and <function>stream_change_cb</function>
- are required, while <function>stream_message_cb</function> and
+ are required, while <function>stream_message_cb</function>,
+ <function>stream_prepare_cb</function>,
<function>stream_commit_prepared_cb</function>,
+ <function>stream_abort_prepared_cb</function>,

Missing "and".
... "stream_abort_prepared_cb, stream_truncate_cb are optional." -->
"stream_abort_prepared_cb, and stream_truncate_cb are optional."

;

COMMENT
Section 48.6.4.16
Section 48.6.4.17
Section 48.6.4.18
@@ -839,6 +844,45 @@ typedef void (*LogicalDecodeStreamAbortCB)
(struct LogicalDecodingContext *ctx,
</para>
</sect3>

+ <sect3 id="logicaldecoding-output-plugin-stream-prepare">
+ <title>Stream Prepare Callback</title>
+ <para>
+ The <function>stream_prepare_cb</function> callback is called to prepare
+ a previously streamed transaction as part of a two phase commit.
+<programlisting>
+typedef void (*LogicalDecodeStreamPrepareCB) (struct
LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn);
+</programlisting>
+ </para>
+ </sect3>
+
+ <sect3 id="logicaldecoding-output-plugin-stream-commit-prepared">
+ <title>Stream Commit Prepared Callback</title>
+ <para>
+ The <function>stream_commit_prepared_cb</function> callback is
called to commit prepared
+ a previously streamed transaction as part of a two phase commit.
+<programlisting>
+typedef void (*LogicalDecodeStreamCommitPreparedCB) (struct
LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn);
+</programlisting>
+ </para>
+ </sect3>
+
+ <sect3 id="logicaldecoding-output-plugin-stream-abort-prepared">
+ <title>Stream Abort Prepared Callback</title>
+ <para>
+ The <function>stream_abort_prepared_cb</function> callback is called
to abort prepared
+ a previously streamed transaction as part of a two phase commit.
+<programlisting>
+typedef void (*LogicalDecodeStreamAbortPreparedCB) (struct
LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn);
+</programlisting>
+ </para>
+ </sect3>

1. Everywhere it says "two phase" commit should be consistently
replaced to say "two-phase" commit (with the hyphen)

2. Search for "abort_lsn" parameter. It seems to be overused
(cut/paste error) even when the API is unrelated to abort

3. 48.6.4.17 and 48.6.4.18
Is this wording ok? Is the word "prepared" even necessary here?
- "... called to commit prepared a previously streamed transaction ..."
- "... called to abort prepared a previously streamed transaction ..."

;

COMMENT
Section 48.9
@@ -1017,9 +1061,13 @@ OutputPluginWrite(ctx, true);
 When streaming an in-progress transaction, the changes (and messages) are
 streamed in blocks demarcated by <function>stream_start_cb</function>
 and <function>stream_stop_cb</function> callbacks. Once all the decoded
- changes are transmitted, the transaction is committed using the
- <function>stream_commit_cb</function> callback (or possibly aborted using
- the <function>stream_abort_cb</function> callback).
+ changes are transmitted, the transaction can be committed using the
+ the <function>stream_commit_cb</function> callback

"two phase" --> "two-phase"

~

Also, Missing period on end of sentence.
"or aborted using the stream_abort_prepared_cb" --> "or aborted using
the stream_abort_prepared_cb."

;

==========
Patch V6-0004, File: src/backend/replication/logical/logical.c
==========

COMMENT
Line 84
@@ -81,6 +81,12 @@ static void stream_stop_cb_wrapper(ReorderBuffer
*cache, ReorderBufferTXN *txn,
 XLogRecPtr last_lsn);
 static void stream_abort_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
 XLogRecPtr abort_lsn);
+static void stream_prepare_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+static void stream_commit_prepared_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+static void stream_abort_prepared_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);

The 3rd parameter is always "commit_lsn" even for API unrelated to
commit, so seems like cut/paste error.

;

COMMENT
Line 1246
@@ -1231,6 +1243,105 @@ stream_abort_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
}

 static void
+stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn)
+{
+ LogicalDecodingContext *ctx = cache->private_data;
+ LogicalErrorCallbackState state;

Misnamed parameter "commit_lsn" ?

~

Also, Line 1272
There seem to be some missing integrity checking to make sure the
callback is not NULL.
A null callback will give NPE when wrapper attempts to call it

;

COMMENT
Line 1305
+static void
+stream_commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

There seem to be some missing integrity checking to make sure the
callback is not NULL.
A null callback will give NPE when wrapper attempts to call it.

;

COMMENT
Line 1312
+static void
+stream_abort_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

Misnamed parameter "commit_lsn" ?

~

Also, Line 1338
There seem to be some missing integrity checking to make sure the
callback is not NULL.
A null callback will give NPE when wrapper attempts to call it.

==========
Patch V6-0004, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
Line 2684
@@ -2672,15 +2681,31 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb,
TransactionId xid,
txn->gid = palloc(strlen(gid) + 1); /* trailing '\0' */
strcpy(txn->gid, gid);

- if (is_commit)
+ if (rbtxn_is_streamed(txn))
 {
- txn->txn_flags |= RBTXN_COMMIT_PREPARED;
- rb->commit_prepared(rb, txn, commit_lsn);
+ if (is_commit)
+ {
+ txn->txn_flags |= RBTXN_COMMIT_PREPARED;

The setting/checking of the flags could be refactored if you wanted to
write less code:
e.g.
if (is_commit)
txn->txn_flags |= RBTXN_COMMIT_PREPARED;
else
txn->txn_flags |= RBTXN_ROLLBACK_PREPARED;

if (rbtxn_is_streamed(txn) && rbtxn_commit_prepared(txn))
rb->stream_commit_prepared(rb, txn, commit_lsn);
else if (rbtxn_is_streamed(txn) && rbtxn_rollback_prepared(txn))
rb->stream_abort_prepared(rb, txn, commit_lsn);
else if (rbtxn_commit_prepared(txn))
rb->commit_prepared(rb, txn, commit_lsn);
else if (rbtxn_rollback_prepared(txn))
rb->abort_prepared(rb, txn, commit_lsn);

;

==========
Patch V6-0004, File: src/include/replication/output_plugin.h
==========

COMMENT
Line 171
@@ -157,6 +157,33 @@ typedef void (*LogicalDecodeStreamAbortCB)
(struct LogicalDecodingContext *ctx,
XLogRecPtr abort_lsn);

 /*
+ * Called to prepare changes streamed to remote node from in-progress
+ * transaction. This is called as part of a two-phase commit and only when
+ * two-phased commits are supported
+ */

1. Missing period all these comments.

2. Is the part that says "and only where two-phased commits are
supported" necessary to say? Is seems redundant since comments already
says called as part of a two-phase commit.

;

==========
Patch V6-0004, File: src/include/replication/reorderbuffer.h
==========

COMMENT
Line 467
@@ -466,6 +466,24 @@ typedef void (*ReorderBufferStreamAbortCB) (
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamPrepareCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamCommitPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamAbortPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);

Cut/paste error - repeated same comment 3 times?

[END]

On Tue, Oct 6, 2020 at 10:23 AM Peter.B.Smith@fujitsu.com
<Peter.B.Smith@fujitsu.com> wrote:

[BEGIN]

==========
Patch V6-0001, File: contrib/test_decoding/expected/prepared.out (so
prepared.sql also)
==========

COMMENT
Line 30 - The INSERT INTO test_prepared1 VALUES (2); is kind of
strange because it is not really part of the prior test nor the
following test. Maybe it would be better to have a comment describing
the purpose of this isolated INSERT and to also consume the data from
the slot so it does not get jumbled with the data of the following
(abort) test.

;

COMMENT
Line 53 - Same comment for this test INSERT INTO test_prepared1 VALUES
(4); It kind of has nothing really to do with either the prior (abort)
test nor the following (ddl) test.

;

COMMENT
Line 60 - Seems to check which locks are held for the test_prepared_1
table while the transaction is in progress. Maybe it would be better
to have more comments describing what is expected here and why.

;

COMMENT
Line 88 - There is a comment in the test saying "-- We should see '7'
before '5' in our results since it commits first." but I did not see
any test code that actually verifies that happens.

;

All the above comments are genuine and I think it is mostly because
the author has blindly modified the existing tests without completely
understanding the intent of the test. I suggest we write a completely
new regression file (decode_prepared.sql) for these and just copy
whatever is required from prepared.sql. Once we do that we might also
want to rename existing prepared.sql to decode_commit_prepared.sql or
something like that. I think modifying the existing test appears to be
quite ugly and also it is changing the intent of the existing tests.

QUESTION
Line 120 - I did not really understand the SQL checking the pg_class.
I expected this would be checking table 'test_prepared1' instead. Can
you explain it?
SELECT 'pg_class' AS relation, locktype, mode
FROM pg_locks
WHERE locktype = 'relation'
AND relation = 'pg_class'::regclass;
relation | locktype | mode
----------+----------+------
(0 rows)

;

Yes, I also think your expectation is correct and this should be on
'test_prepared_1'.

QUESTION
Line 139 - SET statement_timeout = '1s'; is 1 seconds short enough
here for this test, or might it be that these statements would be
completed in less than one seconds anyhow?

;

Good question. I think we have to mention the reason why logical
decoding is not blocked while it needs to acquire a shared lock on the
table and the previous commands already held an exclusive lock on the
table. I am not sure if I am missing something but like you, it is not
clear to me as well what this test intends to do, so surely more
commentary is required.

QUESTION
Line 163 - How is this testing a SAVEPOINT? Or is it only to check
that the SAVEPOINT command is not part of the replicated changes?

;

It is more of testing that subtransactions will not create a problem
while decoding.

COMMENT
Line 175 - Missing underscore in comment. Code requires also underscore:
"nodecode" --> "_nodecode"

makes sense.

==========
Patch V6-0001, File: contrib/test_decoding/test_decoding.c
==========

COMMENT
Line 43
@@ -36,6 +40,7 @@ typedef struct
bool skip_empty_xacts;
bool xact_wrote_changes;
bool only_local;
+ TransactionId check_xid; /* track abort of this txid */
} TestDecodingData;

The "check_xid" seems a meaningless name. Check what?
IIUC maybe should be something like "check_xid_aborted"

;

COMMENT
Line 105
@ -88,6 +93,19 @@ static void
pg_decode_stream_truncate(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
int nrelations, Relation relations[],
ReorderBufferChange *change);
+static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,

Remove extra blank line after these functions

;

The above two sounds reasonable suggestions.

COMMENT
Line 149
@@ -116,6 +134,11 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->stream_change_cb = pg_decode_stream_change;
cb->stream_message_cb = pg_decode_stream_message;
cb->stream_truncate_cb = pg_decode_stream_truncate;
+ cb->filter_prepare_cb = pg_decode_filter_prepare;
+ cb->prepare_cb = pg_decode_prepare_txn;
+ cb->commit_prepared_cb = pg_decode_commit_prepared_txn;
+ cb->abort_prepared_cb = pg_decode_abort_prepared_txn;
+
}

There is a confusing mix of terminology where sometimes things are
referred as ROLLBACK/rollback and other times apparently the same
operation is referred as ABORT/abort. I do not know the root cause of
this mixture. IIUC maybe the internal functions and protocol generally
use the term "abort", whereas the SQL syntax is "ROLLBACK"... but
where those two terms collide in the middle it gets quite confusing.

At least I thought the names of the "callbacks" which get exposed to
the user (e.g. in the help) might be better if they would match the
SQL.
"abort_prepared_cb" --> "rollback_prepared_db"

This suggestion sounds reasonable. I think it is to entertain the case
where due to error we need to rollback the transaction. I think it is
better if use 'rollback' terminology in the exposed functions. We
already have a function with the name stream_abort_cb in the code
which we also might want to rename but that is a separate thing and we
can deal it with a separate patch.

There are similar review comments like this below where the
alternating terms caused me some confusion.

~

Also, Remove the extra blank line before the end of the function.

;

COMMENT
Line 267
@ -227,6 +252,42 @@ pg_decode_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt,
errmsg("could not parse value \"%s\" for parameter \"%s\"",
strVal(elem->arg), elem->defname)));
}
+ else if (strcmp(elem->defname, "two-phase-commit") == 0)
+ {
+ if (elem->arg == NULL)
+ continue;

IMO the "check-xid" code might be better rearranged so the NULL check
is first instead of if/else.
e.g.
if (elem->arg == NULL)
ereport(FATAL,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("check-xid needs an input value")));
~

Also, is it really supposed to be FATAL instead or ERROR. That is not
the same as the other surrounding code.

;

+1.

COMMENT
Line 296
if (data->check_xid <= 0)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("Specify positive value for parameter \"%s\","
" you specified \"%s\"",
elem->defname, strVal(elem->arg))));

The code checking for <= 0 seems over-complicated. Because conversion
was using strtoul() I fail to see how this can ever be < 0. Wouldn't
it be easier to simply test the result of the strtoul() function?

BEFORE: if (errno == EINVAL || errno == ERANGE)
AFTER: if (data->check_xid == 0)

Better to use TransactionIdIsValid(data->check_xid) here.

~

Also, should this be FATAL? Everything else similar is ERROR.

;

It should be an error.

COMMENT
(general)
I don't recall seeing any of these decoding options (e.g.
"two-phase-commit", "check-xid") documented anywhere.
So how can a user even know these options exist so they can use them?
Perhaps options should be described on this page?
https://www.postgresql.org/docs/13/functions-admin.html#FUNCTIONS-REPLICATION

;

I think we should do what we are doing for other options, if they are
not documented then why to document this one separately. I guess we
can make a case to document all the existing options and write a
separate patch for that.

COMMENT
(general)
"check-xid" is a meaningless option name. Maybe something like
"checked-xid-aborted" is more useful?
Suggest changing the member, the option, and the error messages to
match some better name.

;

COMMENT
Line 314
@@ -238,6 +299,7 @@ pg_decode_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt,
}

ctx->streaming &= enable_streaming;
+ ctx->enable_twophase &= enable_2pc;
}

The "ctx->enable_twophase" is inconsistent naming with the
"ctx->streaming" member.
"enable_twophase" --> "twophase"

;

+1.

COMMENT
Line 374
@@ -297,6 +359,94 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
OutputPluginWrite(ctx, true);
}

+
+/*
+ * Filter out two-phase transactions.
+ *
+ * Each plugin can implement its own filtering logic. Here
+ * we demonstrate a simple logic by checking the GID. If the
+ * GID contains the "_nodecode" substring, then we filter
+ * it out.
+ */
+static bool
+pg_decode_filter_prepare(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Remove the extra preceding blank line.

~

I did not find anything in the help about "_nodecode". Should it be
there or is this deliberately not documented feature?

;

I guess we can document it along with filter_prepare API, if not
already documented.

QUESTION
Line 440
+pg_decode_abort_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,

Is this a wrong comment
"ABORT PREPARED" --> "ROLLBACK PREPARED" ??

;

COMMENT
Line 620
@@ -455,6 +605,22 @@ pg_decode_change(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
}
data->xact_wrote_changes = true;

+ /* if check_xid is specified */
+ if (TransactionIdIsValid(data->check_xid))
+ {
+ elog(LOG, "waiting for %u to abort", data->check_xid);
+ while (TransactionIdIsInProgress(dat

The check_xid seems a meaningless name, and the comment "/* if
check_xid is specified */" was not helpful either.
IIUC purpose of this is to check that the nominated xid always is rolled back.
So the appropriate name may be more like "check-xid-aborted".

;

Yeah, this part deserves better comments.

--
With Regards,
Amit Kapila.

On Tue, Oct 6, 2020 at 10:23 AM Peter.B.Smith@fujitsu.com
<Peter.B.Smith@fujitsu.com> wrote:

==========
Patch V6-0001, File: doc/src/sgml/logicaldecoding.sgml
==========

COMMENT/QUESTION
Section 48.6.1
@ -387,6 +387,10 @@ typedef struct OutputPluginCallbacks
LogicalDecodeTruncateCB truncate_cb;
LogicalDecodeCommitCB commit_cb;
LogicalDecodeMessageCB message_cb;
+ LogicalDecodeFilterPrepareCB filter_prepare_cb;

Confused by the mixing of terminologies "abort" and "rollback".
Why is it LogicalDecodeAbortPreparedCB instead of
LogicalDecodeRollbackPreparedCB?
Why is it abort_prepared_cb instead of rollback_prepared_cb;?

I thought everything the user sees should be ROLLBACK/rollback (like
the SQL) regardless of what the internal functions might be called.

;

Fair enough.

COMMENT
Section 48.6.1
The begin_cb, change_cb and commit_cb callbacks are required, while
startup_cb, filter_by_origin_cb, truncate_cb, and shutdown_cb are
optional. If truncate_cb is not set but a TRUNCATE is to be decoded,
the action will be ignored.

The 1st paragraph beneath the typedef does not mention the newly added
callbacks to say if they are required or optional.

;

Yeah, in code comments it was mentioned but is missed here, see the
comment "To support two phase logical decoding, we require
prepare/commit-prepare/abort-prepare callbacks. The filter-prepare
callback is optional.". I think instead of directly editing the above
paragraph we can write a new one similar to what we have done for
streaming of large in-progress transactions (Refer <para> An output
plugin may also define functions to support streaming of large,
in-progress transactions.).

COMMENT
Section 48.6.4.5
Section 48.6.4.6
Section 48.6.4.7
@@ -578,6 +588,55 @@ typedef void (*LogicalDecodeCommitCB) (struct
LogicalDecodingContext *ctx,
</para>
</sect3>

+ <sect3 id="logicaldecoding-output-plugin-prepare">
+    <sect3 id="logicaldecoding-output-plugin-commit-prepared">
+    <sect3 id="logicaldecoding-output-plugin-abort-prepared">
+<programlisting>

The wording and titles are a bit backwards compared to the others.
e.g. previously was "Transaction Begin" (not "Begin Transaction") and
"Transaction End" (not "End Transaction").

So for consistently following the existing IMO should change these new
titles (and wording) to:
- "Commit Prepared Transaction Callback" --> "Transaction Commit
Prepared Callback"
- "Rollback Prepared Transaction Callback" --> "Transaction Rollback
Prepared Callback"

makes sense.

- "whenever a commit prepared transaction has been decoded" -->
"whenever a transaction commit prepared has been decoded"
- "whenever a rollback prepared transaction has been decoded." -->
"whenever a transaction rollback prepared has been decoded."

;

I don't find above suggestions much better than current wording. How
about below instead?

"whenever we decode a transaction which is prepared for two-phase
commit is committed"
"whenever we decode a transaction which is prepared for two-phase
commit is rolled back"

Also, related to this:
+    <sect3 id="logicaldecoding-output-plugin-commit-prepared">
+     <title>Commit Prepared Transaction Callback</title>
+
+     <para>
+      The optional <function>commit_prepared_cb</function> callback
is called whenever
+      a commit prepared transaction has been decoded. The
<parameter>gid</parameter> field,
+      which is part of the <parameter>txn</parameter> parameter can
be used in this
+      callback.
+<programlisting>
+typedef void (*LogicalDecodeCommitPreparedCB) (struct
LogicalDecodingContext *ctx,
+                                               ReorderBufferTXN *txn,
+                                               XLogRecPtr commit_lsn);
+</programlisting>
+     </para>
+    </sect3>
+
+    <sect3 id="logicaldecoding-output-plugin-abort-prepared">
+     <title>Rollback Prepared Transaction Callback</title>
+
+     <para>
+      The optional <function>abort_prepared_cb</function> callback is
called whenever
+      a rollback prepared transaction has been decoded. The
<parameter>gid</parameter> field,
+      which is part of the <parameter>txn</parameter> parameter can
be used in this
+      callback.
+<programlisting>

Both the above are not optional as per code and I think code is
correct. I think the documentation is wrong here.

==========
Patch V6-0001, File: src/backend/replication/logical/decode.c
==========

COMMENT
Line 74
@@ -70,6 +70,9 @@ static void DecodeCommit(LogicalDecodingContext
*ctx, XLogRecordBuffer *buf,
xl_xact_parsed_commit *parsed, TransactionId xid);
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
xl_xact_parsed_abort *parsed, TransactionId xid);
+static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
+ xl_xact_parsed_prepare * parsed);

The 2nd line of DecodePrepare is misaligned by one space.

;

Yeah, probably pgindent is the answer. Ajin, can you please run
pgindent on all the patches?

COMMENT
Line 321
@@ -312,17 +315,34 @@ DecodeXactOp(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf)
}
break;
case XLOG_XACT_PREPARE:
+ {
+ xl_xact_parsed_prepare parsed;
+ xl_xact_prepare *xlrec;
+ /* check that output plugin is capable of twophase decoding */

"twophase" --> "two-phase"

~

Also, add a blank line after the declarations.

;

==========
Patch V6-0001, File: src/backend/replication/logical/logical.c
==========

COMMENT
Line 249
@@ -225,6 +237,19 @@ StartupDecodingContext(List *output_plugin_options,
(ctx->callbacks.stream_message_cb != NULL) ||
(ctx->callbacks.stream_truncate_cb != NULL);

+ /*
+ * To support two phase logical decoding, we require
prepare/commit-prepare/abort-prepare
+ * callbacks. The filter-prepare callback is optional. We however
enable two phase logical
+ * decoding when at least one of the methods is enabled so that we
can easily identify
+ * missing methods.

The terminology is generally well known as "two-phase" (with the
hyphen) https://en.wikipedia.org/wiki/Two-phase_commit_protocol so
let's be consistent for all the patch code comments. Please search the
code and correct this in all places, even where I might have missed to
identify it.

"two phase" --> "two-phase"

;

COMMENT
Line 822
@@ -782,6 +807,111 @@ commit_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
}

static void
+prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr prepare_lsn)

"support 2 phase" --> "supports two-phase" in the comment

;

COMMENT
Line 844
Code condition seems strange and/or broken.
if (ctx->enable_twophase && ctx->callbacks.prepare_cb == NULL)
Because if the flag is null then this condition is skipped.
But then if the callback was also NULL then attempting to call it to
"do the actual work" will give NPE.

~

Also, I wonder should this check be the first thing in this function?
Because if it fails does it even make sense that all the errcallback
code was set up?> E.g errcallback.arg potentially is left pointing to a stack variable
on a stack that no longer exists.

;

Right, I think we should have an Assert(ctx->enable_twophase) in the
beginning and then have the check (ctx->callbacks.prepare_cb == NULL)
t its current place. Refer any of the streaming APIs (for ex.
stream_stop_cb_wrapper).

COMMENT
Line 857
+commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

"support 2 phase" --> "supports two-phase" in the comment

~

Also, Same potential trouble with the condition:
if (ctx->enable_twophase && ctx->callbacks.commit_prepared_cb == NULL)
Same as previously asked. Should this check be first thing in this function?

;

Yeah, so the same solution as mentioned above can be used.

COMMENT
Line 892
+abort_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

"support 2 phase" --> "supports two-phase" in the comment

~

Same potential trouble with the condition:
if (ctx->enable_twophase && ctx->callbacks.abort_prepared_cb == NULL)
Same as previously asked. Should this check be the first thing in this function?

;

Again the same solution can be used.

COMMENT
Line 1013
@@ -858,6 +988,51 @@ truncate_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
error_context_stack = errcallback.previous;
}

+static bool
+filter_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ TransactionId xid, const char *gid)

Fix wording in comment:
"twophase" --> "two-phase transactions"
"twophase transactions" --> "two-phase transactions"

==========
Patch V6-0001, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
Line 255
@@ -251,7 +251,8 @@ static Size
ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn
static void ReorderBufferRestoreChange(ReorderBuffer *rb,
ReorderBufferTXN *txn,
char *change);
static void ReorderBufferRestoreCleanup(ReorderBuffer *rb,
ReorderBufferTXN *txn);
-static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
+static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
+ bool txn_prepared);

The alignment is inconsistent. One more space needed before "bool txn_prepared"

;

COMMENT
Line 417
@@ -413,6 +414,11 @@ ReorderBufferReturnTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
}

/* free data that's contained */
+ if (txn->gid != NULL)
+ {
+ pfree(txn->gid);
+ txn->gid = NULL;
+ }

Should add the blank link before this new code, as it was before.

;

COMMENT
Line 1564
@ -1502,12 +1561,14 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
}

/*
- * Discard changes from a transaction (and subtransactions), after streaming
- * them. Keep the remaining info - transactions, tuplecids, invalidations and
- * snapshots.
+ * Discard changes from a transaction (and subtransactions), either
after streaming or
+ * after a PREPARE.

typo "snapshots.If" -> "snapshots. If"

;

COMMENT/QUESTION
Line 1590
@@ -1526,7 +1587,7 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
Assert(rbtxn_is_known_subxact(subtxn));
Assert(subtxn->nsubtxns == 0);

- ReorderBufferTruncateTXN(rb, subtxn);
+ ReorderBufferTruncateTXN(rb, subtxn, txn_prepared);
}

There are some code paths here I did not understand how they match the comments.
Because this function is recursive it seems that it may be called
where the 2nd parameter txn is a sub-transaction.

But then this seems at odds with some of the other code comments of
this function which are processing the txn without ever testing is it
really toplevel or not:

e.g. Line 1593 "/* cleanup changes in the toplevel txn */"

I think this comment is wrong but this is not the fault of this patch.

e.g. Line 1632 "They are always stored in the toplevel transaction."

;

This seems to be correct and we probably need an Assert that the
transaction is a top-level transaction.

COMMENT
Line 1644
@@ -1560,9 +1621,33 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
* about the toplevel xact (we send the XID in all messages), but we never
* stream XIDs of empty subxacts.
*/
- if ((!txn->toptxn) || (txn->nentries_mem != 0))
+ if ((!txn_prepared) && ((!txn->toptxn) || (txn->nentries_mem != 0)))
txn->txn_flags |= RBTXN_IS_STREAMED;

+ if (txn_prepared)

/* remove the change from it's containing list */
typo "it's" --> "its"

;

QUESTION
Line 1977
@@ -1880,7 +1965,7 @@ ReorderBufferResetTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
ReorderBufferChange *specinsert)
{
/* Discard the changes that we just streamed */
- ReorderBufferTruncateTXN(rb, txn);
+ ReorderBufferTruncateTXN(rb, txn, false);

How do you know the 3rd parameter - i.e. txn_prepared - should be
hardwired false here?
e.g. I thought that maybe rbtxn_prepared(txn) can be true here.

;

COMMENT
Line 2345
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-
/*

Looks like accidental blank line deletion. This should be put back how it was

;

COMMENT/QUESTION
Line 2374
@@ -2278,7 +2362,16 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
}
}
else
- rb->commit(rb, txn, commit_lsn);
+ {
+ /*
+ * Call either PREPARE (for twophase transactions) or COMMIT
+ * (for regular ones).

"twophase" --> "two-phase"

~

Also, I was confused by the apparent assumption of exclusiveness of
streaming and 2PC...
e.g. what if streaming AND 2PC then it won't do rb->prepare()

;

QUESTION
Line 2424
@@ -2319,11 +2412,17 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
*/
if (streaming)
{
- ReorderBufferTruncateTXN(rb, txn);
+ ReorderBufferTruncateTXN(rb, txn, false);

/* Reset the CheckXidAlive */
CheckXidAlive = InvalidTransactionId;
}
+ else if (rbtxn_prepared(txn))

I was confused by the exclusiveness of streaming/2PC.
e.g. what if streaming AND 2PC at same time - how can you pass false
as 3rd param to ReorderBufferTruncateTXN?

;

Yeah, this and another handling wherever it is assumed that both can't
be true together is wrong.

COMMENT
Line 2463
@@ -2352,17 +2451,18 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,

/*
* The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
- * abort of the (sub)transaction we are streaming. We need to do the
+ * abort of the (sub)transaction we are streaming or preparing. We
need to do the
* cleanup and return gracefully on this error, see SetupCheckXidLive.
*/

"twoi phase" --> "two-phase"

;

QUESTIONS
Line 2482
@@ -2370,10 +2470,19 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
errdata = NULL;
curtxn->concurrent_abort = true;

- /* Reset the TXN so that it is allowed to stream remaining data. */
- ReorderBufferResetTXN(rb, txn, snapshot_now,
- command_id, prev_lsn,
- specinsert);
+ /* If streaming, reset the TXN so that it is allowed to stream
remaining data. */
+ if (streaming)

Re: /* If streaming, reset the TXN so that it is allowed to stream
remaining data. */
I was confused by the exclusiveness of streaming/2PC.
Is it not possible for streaming flags and rbtxn_prepared(txn) true at
the same time?

Yeah, I think it is not correct to assume that both can't be true at
the same time. But when prepared is true irrespective of whether
streaming is true or not we can use ReorderBufferTruncateTXN() API
instead of Reset API.

~

elog(LOG, "stopping decoding of %s (%u)",
txn->gid[0] != '\0'? txn->gid:"", txn->xid);

Is this a safe operation, or do you also need to test txn->gid is not NULL?

;

I think if 'prepared' is true then we can assume it to be non-NULL,
otherwise, not.

I am responding to your email in phases so that we can have a
discussion on specific points if required and I am slightly afraid
that the email might not bounce as it happened in your case when you
sent such a long email.

--
With Regards,
Amit Kapila.

On Wed, Oct 7, 2020 at 1:24 AM Amit Kapila <amit.kapila16@gmail.com> wrote:

There is a confusing mix of terminology where sometimes things are
referred as ROLLBACK/rollback and other times apparently the same
operation is referred as ABORT/abort. I do not know the root cause of
this mixture. IIUC maybe the internal functions and protocol generally
use the term "abort", whereas the SQL syntax is "ROLLBACK"... but
where those two terms collide in the middle it gets quite confusing.

At least I thought the names of the "callbacks" which get exposed to
the user (e.g. in the help) might be better if they would match the
SQL.
"abort_prepared_cb" --> "rollback_prepared_db"

This suggestion sounds reasonable. I think it is to entertain the case
where due to error we need to rollback the transaction. I think it is
better if use 'rollback' terminology in the exposed functions. We
already have a function with the name stream_abort_cb in the code
which we also might want to rename but that is a separate thing and we
can deal it with a separate patch.

So, for an ordinary transaction, rollback implies an explicit user
action, but an abort could either be an explicit user action (ABORT;
or ROLLBACK;) or an error. I agree that calling that case "abort"
rather than "rollback" is better. However, the situation is a bit
different for a prepared transaction: no error can prevent such a
transaction from being committed. That is the whole point of being
able to prepare transactions. So it is not unreasonable to think of
use "rollback" rather than "abort" for prepared transactions, but I
think it would be wrong in other cases. On the other hand, using
"abort" for all the cases also doesn't seem bad to me. It's true that
there is no ABORT PREPARED command at the SQL level, but I don't think
that is very important. I don't feel wrong saying that ROLLBACK
PREPARED causes a transaction abort.

--
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Thu, Oct 8, 2020 at 6:14 AM Robert Haas <robertmhaas@gmail.com> wrote:

So, for an ordinary transaction, rollback implies an explicit user
action, but an abort could either be an explicit user action (ABORT;
or ROLLBACK;) or an error. I agree that calling that case "abort"
rather than "rollback" is better. However, the situation is a bit
different for a prepared transaction: no error can prevent such a
transaction from being committed. That is the whole point of being
able to prepare transactions. So it is not unreasonable to think of
use "rollback" rather than "abort" for prepared transactions, but I
think it would be wrong in other cases. On the other hand, using
"abort" for all the cases also doesn't seem bad to me. It's true that
there is no ABORT PREPARED command at the SQL level, but I don't think
that is very important. I don't feel wrong saying that ROLLBACK
PREPARED causes a transaction abort.

So, as I understand you don't object to renaming the callback APIs for
ROLLBACK PREPARED transactions to "rollback_prepared_cb" but keeping
the "stream_abort" as such. This was what I was planning on doing.
I was just writing this up, so wanted to confirm.

regards,
Ajin Cherian
Fujitsu Australia

On Tue, Oct 6, 2020 at 10:23 AM Peter.B.Smith@fujitsu.com
<Peter.B.Smith@fujitsu.com> wrote:

==========
Patch V6-0001, File: src/include/replication/reorderbuffer.h
==========

QUESTION
Line 116
@@ -162,9 +163,13 @@ typedef struct ReorderBufferChange
#define RBTXN_HAS_CATALOG_CHANGES 0x0001
#define RBTXN_IS_SUBXACT 0x0002
#define RBTXN_IS_SERIALIZED 0x0004
-#define RBTXN_IS_STREAMED 0x0008
-#define RBTXN_HAS_TOAST_INSERT 0x0010
-#define RBTXN_HAS_SPEC_INSERT 0x0020
+#define RBTXN_PREPARE 0x0008
+#define RBTXN_COMMIT_PREPARED 0x0010
+#define RBTXN_ROLLBACK_PREPARED 0x0020
+#define RBTXN_COMMIT 0x0040
+#define RBTXN_IS_STREAMED 0x0080
+#define RBTXN_HAS_TOAST_INSERT 0x0100
+#define RBTXN_HAS_SPEC_INSERT 0x0200

I was wondering why when adding new flags, some of the existing flag
masks were also altered.
I am assuming this is ok because they are never persisted but are only
used in the protocol (??)

;

This is bad even though there is no direct problem. I don't think we
need to change the existing ones, we can add the new ones at the end
with the number starting where the last one ends.

COMMENT
Line 133

Assert(strlen(txn->gid) > 0);
Shouldn't that assertion also check txn->gid is not NULL (to prevent
NPE in case gid was NULL)

;

I think that would be better and a stronger Assertion than the current one.

COMMENT
Line 177
+logicalrep_read_prepare(StringInfo in, LogicalRepPrepareData * prepare_data)

prepare_data->prepare_type = flags;
This code may be OK but it does seem a bit of an abuse of the flags.

e.g. Are they flags or are the really enum values?
e.g. And if they are effectively enums (it appears they are) then
seemed inconsistent that |= was used when they were previously
assigned.

;

I don't understand this point. As far as I can see at the time of
write (logicalrep_write_prepare()), the patch has used |=, and at the
time of reading (logicalrep_read_prepare()) it has used assignment
which seems correct from the code perspective. Do you have a better
proposal?

COMMENT
Line 408
+pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,

Since all this function is identical to pg_output_prepare it might be
better to either
1. just leave this as a wrapper to delegate to that function
2. remove this one entirely and assign the callback to the common
pgoutput_prepare_txn

;

I think this is because as of now the patch uses the same function and
protocol message to send both Prepare and Commit/Rollback Prepare
which I am not sure is the right thing. I suggest keeping that code as
it is for now. Let's first try to figure out if it is a good idea to
overload the same protocol message and use flags to distinguish the
actual message. Also, I don't know whether prepare_lsn is required
during commit time?

COMMENT
Line 419
+pgoutput_abort_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Since all this function is identical to pg_output_prepare if might be
better to either
1. just leave this as a wrapper to delegate to that function
2. remove this one entirely and assign the callback to the common
pgoutput_prepare_tx

;

Due to reasons mentioned for the previous comment, let's keep this
also as it is for now.

--
With Regards,
Amit Kapila.

On Thu, Oct 8, 2020 at 5:25 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

COMMENT
Line 177
+logicalrep_read_prepare(StringInfo in, LogicalRepPrepareData * prepare_data)

prepare_data->prepare_type = flags;
This code may be OK but it does seem a bit of an abuse of the flags.

e.g. Are they flags or are the really enum values?
e.g. And if they are effectively enums (it appears they are) then
seemed inconsistent that |= was used when they were previously
assigned.

;

I don't understand this point. As far as I can see at the time of
write (logicalrep_write_prepare()), the patch has used |=, and at the
time of reading (logicalrep_read_prepare()) it has used assignment
which seems correct from the code perspective. Do you have a better
proposal?

OK. I will explain my thinking when I wrote that review comment.

I agree all is "correct" from a code perspective.

But IMO using bit arithmetic implies that different combinations are
also possible, whereas in current code they are not.
So code is kind of having a bet each-way - sometimes treating "flags"
as bit flags and sometimes as enums.

e.g. If these flags are not really bit flags at all then the
logicalrep_write_prepare() code might just as well be written as
below:

BEFORE
if (rbtxn_commit_prepared(txn))
flags |= LOGICALREP_IS_COMMIT_PREPARED;
else if (rbtxn_rollback_prepared(txn))
flags |= LOGICALREP_IS_ROLLBACK_PREPARED;
else
flags |= LOGICALREP_IS_PREPARE;

/* Make sure exactly one of the expected flags is set. */
if (!PrepareFlagsAreValid(flags))
elog(ERROR, "unrecognized flags %u in prepare message", flags);

AFTER
if (rbtxn_commit_prepared(txn))
flags = LOGICALREP_IS_COMMIT_PREPARED;
else if (rbtxn_rollback_prepared(txn))
flags = LOGICALREP_IS_ROLLBACK_PREPARED;
else
flags = LOGICALREP_IS_PREPARE;

~

OTOH, if you really do want to anticipate having future flag bit
combinations then maybe the PrepareFlagsAreValid() macro ought to to
be tweaked accordingly, and the logicalrep_read_prepare() code maybe
should look more like below:

BEFORE
/* set the action (reuse the constants used for the flags) */
prepare_data->prepare_type = flags;

AFTER
/* set the action (reuse the constants used for the flags) */
prepare_data->prepare_type =
flags & LOGICALREP_IS_COMMIT_PREPARED ? LOGICALREP_IS_COMMIT_PREPARED :
flags & LOGICALREP_IS_ROLLBACK_PREPARED ? LOGICALREP_IS_ROLLBACK_PREPARED :
LOGICALREP_IS_PREPARE;

Kind Regards.
Peter Smith
Fujitsu Australia

On Fri, Oct 9, 2020 at 5:45 AM Peter Smith <smithpb2250@gmail.com> wrote:

On Thu, Oct 8, 2020 at 5:25 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

COMMENT
Line 177
+logicalrep_read_prepare(StringInfo in, LogicalRepPrepareData * prepare_data)

prepare_data->prepare_type = flags;
This code may be OK but it does seem a bit of an abuse of the flags.

e.g. Are they flags or are the really enum values?
e.g. And if they are effectively enums (it appears they are) then
seemed inconsistent that |= was used when they were previously
assigned.

;

I don't understand this point. As far as I can see at the time of
write (logicalrep_write_prepare()), the patch has used |=, and at the
time of reading (logicalrep_read_prepare()) it has used assignment
which seems correct from the code perspective. Do you have a better
proposal?

OK. I will explain my thinking when I wrote that review comment.

I agree all is "correct" from a code perspective.

But IMO using bit arithmetic implies that different combinations are
also possible, whereas in current code they are not.
So code is kind of having a bet each-way - sometimes treating "flags"
as bit flags and sometimes as enums.

e.g. If these flags are not really bit flags at all then the
logicalrep_write_prepare() code might just as well be written as
below:

BEFORE
if (rbtxn_commit_prepared(txn))
flags |= LOGICALREP_IS_COMMIT_PREPARED;
else if (rbtxn_rollback_prepared(txn))
flags |= LOGICALREP_IS_ROLLBACK_PREPARED;
else
flags |= LOGICALREP_IS_PREPARE;

/* Make sure exactly one of the expected flags is set. */
if (!PrepareFlagsAreValid(flags))
elog(ERROR, "unrecognized flags %u in prepare message", flags);

AFTER
if (rbtxn_commit_prepared(txn))
flags = LOGICALREP_IS_COMMIT_PREPARED;
else if (rbtxn_rollback_prepared(txn))
flags = LOGICALREP_IS_ROLLBACK_PREPARED;
else
flags = LOGICALREP_IS_PREPARE;

~

OTOH, if you really do want to anticipate having future flag bit
combinations

I don't anticipate more combinations rather I am not yet sure whether
we want to distinguish these operations with flags or have separate
messages for each of these operations. I think for now we can go with
your proposal above.

--
With Regards,
Amit Kapila.

On Wed, Oct 7, 2020 at 4:24 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

All the above comments are genuine and I think it is mostly because
the author has blindly modified the existing tests without completely
understanding the intent of the test. I suggest we write a completely
new regression file (decode_prepared.sql) for these and just copy
whatever is required from prepared.sql. Once we do that we might also
want to rename existing prepared.sql to decode_commit_prepared.sql or
something like that. I think modifying the existing test appears to be
quite ugly and also it is changing the intent of the existing tests.

Updated this. Kept the original prepared.sql untouched and added a new
regression file called two_phase.sql
which is specific to test cases with the new flag two-phase-commit.

QUESTION
Line 120 - I did not really understand the SQL checking the pg_class.
I expected this would be checking table 'test_prepared1' instead. Can
you explain it?
SELECT 'pg_class' AS relation, locktype, mode
FROM pg_locks
WHERE locktype = 'relation'
AND relation = 'pg_class'::regclass;
relation | locktype | mode
----------+----------+------
(0 rows)

;

Yes, I also think your expectation is correct and this should be on
'test_prepared_1'.

Updated

QUESTION
Line 139 - SET statement_timeout = '1s'; is 1 seconds short enough
here for this test, or might it be that these statements would be
completed in less than one seconds anyhow?

;

Good question. I think we have to mention the reason why logical
decoding is not blocked while it needs to acquire a shared lock on the
table and the previous commands already held an exclusive lock on the
table. I am not sure if I am missing something but like you, it is not
clear to me as well what this test intends to do, so surely more
commentary is required.

Updated.

QUESTION
Line 163 - How is this testing a SAVEPOINT? Or is it only to check
that the SAVEPOINT command is not part of the replicated changes?

;

It is more of testing that subtransactions will not create a problem
while decoding.

Updated with a testcase that actually does a rollback to a savepoint

COMMENT
Line 175 - Missing underscore in comment. Code requires also underscore:
"nodecode" --> "_nodecode"

makes sense.

Updated.

==========
Patch V6-0001, File: contrib/test_decoding/test_decoding.c
==========

COMMENT
Line 43
@@ -36,6 +40,7 @@ typedef struct
bool skip_empty_xacts;
bool xact_wrote_changes;
bool only_local;
+ TransactionId check_xid; /* track abort of this txid */
} TestDecodingData;

The "check_xid" seems a meaningless name. Check what?
IIUC maybe should be something like "check_xid_aborted"

Updated.

;

COMMENT
Line 105
@ -88,6 +93,19 @@ static void
pg_decode_stream_truncate(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
int nrelations, Relation relations[],
ReorderBufferChange *change);
+static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,

Remove extra blank line after these functions

;

The above two sounds reasonable suggestions.

Updated.

COMMENT
Line 149
@@ -116,6 +134,11 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->stream_change_cb = pg_decode_stream_change;
cb->stream_message_cb = pg_decode_stream_message;
cb->stream_truncate_cb = pg_decode_stream_truncate;
+ cb->filter_prepare_cb = pg_decode_filter_prepare;
+ cb->prepare_cb = pg_decode_prepare_txn;
+ cb->commit_prepared_cb = pg_decode_commit_prepared_txn;
+ cb->abort_prepared_cb = pg_decode_abort_prepared_txn;
+
}

There is a confusing mix of terminology where sometimes things are
referred as ROLLBACK/rollback and other times apparently the same
operation is referred as ABORT/abort. I do not know the root cause of
this mixture. IIUC maybe the internal functions and protocol generally
use the term "abort", whereas the SQL syntax is "ROLLBACK"... but
where those two terms collide in the middle it gets quite confusing.

At least I thought the names of the "callbacks" which get exposed to
the user (e.g. in the help) might be better if they would match the
SQL.
"abort_prepared_cb" --> "rollback_prepared_db"

This suggestion sounds reasonable. I think it is to entertain the case
where due to error we need to rollback the transaction. I think it is
better if use 'rollback' terminology in the exposed functions. We
already have a function with the name stream_abort_cb in the code
which we also might want to rename but that is a separate thing and we
can deal it with a separate patch.

Changed the call back names from abort_prepared to rollback_prepapred
and stream_abort_prepared to stream_rollback_prepared.

There are similar review comments like this below where the
alternating terms caused me some confusion.

~

Also, Remove the extra blank line before the end of the function.

;

COMMENT
Line 267
@ -227,6 +252,42 @@ pg_decode_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt,
errmsg("could not parse value \"%s\" for parameter \"%s\"",
strVal(elem->arg), elem->defname)));
}
+ else if (strcmp(elem->defname, "two-phase-commit") == 0)
+ {
+ if (elem->arg == NULL)
+ continue;

IMO the "check-xid" code might be better rearranged so the NULL check
is first instead of if/else.
e.g.
if (elem->arg == NULL)
ereport(FATAL,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("check-xid needs an input value")));
~

Also, is it really supposed to be FATAL instead or ERROR. That is not
the same as the other surrounding code.

;

+1.

Updated.

COMMENT
Line 296
if (data->check_xid <= 0)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("Specify positive value for parameter \"%s\","
" you specified \"%s\"",
elem->defname, strVal(elem->arg))));

The code checking for <= 0 seems over-complicated. Because conversion
was using strtoul() I fail to see how this can ever be < 0. Wouldn't
it be easier to simply test the result of the strtoul() function?

BEFORE: if (errno == EINVAL || errno == ERANGE)
AFTER: if (data->check_xid == 0)

Better to use TransactionIdIsValid(data->check_xid) here.

Updated.

~

Also, should this be FATAL? Everything else similar is ERROR.

;

It should be an error.

Updated

COMMENT
(general)
I don't recall seeing any of these decoding options (e.g.
"two-phase-commit", "check-xid") documented anywhere.
So how can a user even know these options exist so they can use them?
Perhaps options should be described on this page?
https://www.postgresql.org/docs/13/functions-admin.html#FUNCTIONS-REPLICATION

;

I think we should do what we are doing for other options, if they are
not documented then why to document this one separately. I guess we
can make a case to document all the existing options and write a
separate patch for that.

I didnt see any of the test_decoding options updated in the
documentation as these seem specific for the test_decoder used in
testing.
https://www.postgresql.org/docs/13/test-decoding.html

COMMENT
(general)
"check-xid" is a meaningless option name. Maybe something like
"checked-xid-aborted" is more useful?
Suggest changing the member, the option, and the error messages to
match some better name.

Updated.

;

COMMENT
Line 314
@@ -238,6 +299,7 @@ pg_decode_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt,
}

ctx->streaming &= enable_streaming;
+ ctx->enable_twophase &= enable_2pc;
}

The "ctx->enable_twophase" is inconsistent naming with the
"ctx->streaming" member.
"enable_twophase" --> "twophase"

;

+1.

Updated

COMMENT
Line 374
@@ -297,6 +359,94 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
OutputPluginWrite(ctx, true);
}

+
+/*
+ * Filter out two-phase transactions.
+ *
+ * Each plugin can implement its own filtering logic. Here
+ * we demonstrate a simple logic by checking the GID. If the
+ * GID contains the "_nodecode" substring, then we filter
+ * it out.
+ */
+static bool
+pg_decode_filter_prepare(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Remove the extra preceding blank line.

Updated.

~

I did not find anything in the help about "_nodecode". Should it be
there or is this deliberately not documented feature?

;

I guess we can document it along with filter_prepare API, if not
already documented.

Again , this seems to be specific to test_decoder and an example of a
way to create a filter_prepare.

QUESTION
Line 440
+pg_decode_abort_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,

Is this a wrong comment
"ABORT PREPARED" --> "ROLLBACK PREPARED" ??

;

COMMENT
Line 620
@@ -455,6 +605,22 @@ pg_decode_change(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
}
data->xact_wrote_changes = true;

+ /* if check_xid is specified */
+ if (TransactionIdIsValid(data->check_xid))
+ {
+ elog(LOG, "waiting for %u to abort", data->check_xid);
+ while (TransactionIdIsInProgress(dat

The check_xid seems a meaningless name, and the comment "/* if
check_xid is specified */" was not helpful either.
IIUC purpose of this is to check that the nominated xid always is rolled back.
So the appropriate name may be more like "check-xid-aborted".

;

Yeah, this part deserves better comments.

Updated.

Other than these first batch of review comments from Peter Smith, I've
also updated new functions in decode.c for DecodeCommitPrepared
and DecodeAbortPrepared as agreed in a previous review comment by
Amit and Dilip.
I've also incorporated Dilip's comment on acquiring SHARED lock rather
than EXCLUSIVE lock while looking for transaction matching Gid.
Since Peter's comments are many, I'll be sending patch updates in
parts addressing his comments.

regards,
Ajin Cherian
Fujitsu Australia

On Wed, Oct 7, 2020 at 9:36 AM Peter Smith <smithpb2250@gmail.com> wrote:

==========
Patch V6-0001, File: doc/src/sgml/logicaldecoding.sgml
==========

COMMENT/QUESTION
Section 48.6.1
@ -387,6 +387,10 @@ typedef struct OutputPluginCallbacks
LogicalDecodeTruncateCB truncate_cb;
LogicalDecodeCommitCB commit_cb;
LogicalDecodeMessageCB message_cb;
+ LogicalDecodeFilterPrepareCB filter_prepare_cb;

Confused by the mixing of terminologies "abort" and "rollback".
Why is it LogicalDecodeAbortPreparedCB instead of
LogicalDecodeRollbackPreparedCB?
Why is it abort_prepared_cb instead of rollback_prepared_cb;?

I thought everything the user sees should be ROLLBACK/rollback (like
the SQL) regardless of what the internal functions might be called.

;

Modified.

COMMENT
Section 48.6.1
The begin_cb, change_cb and commit_cb callbacks are required, while
startup_cb, filter_by_origin_cb, truncate_cb, and shutdown_cb are
optional. If truncate_cb is not set but a TRUNCATE is to be decoded,
the action will be ignored.

The 1st paragraph beneath the typedef does not mention the newly added
callbacks to say if they are required or optional.

Added a new para for this.

;

COMMENT
Section 48.6.4.5
Section 48.6.4.6
Section 48.6.4.7
@@ -578,6 +588,55 @@ typedef void (*LogicalDecodeCommitCB) (struct
LogicalDecodingContext *ctx,
</para>
</sect3>

+ <sect3 id="logicaldecoding-output-plugin-prepare">
+    <sect3 id="logicaldecoding-output-plugin-commit-prepared">
+    <sect3 id="logicaldecoding-output-plugin-abort-prepared">
+<programlisting>

The wording and titles are a bit backwards compared to the others.
e.g. previously was "Transaction Begin" (not "Begin Transaction") and
"Transaction End" (not "End Transaction").

So for consistently following the existing IMO should change these new
titles (and wording) to:
- "Commit Prepared Transaction Callback" --> "Transaction Commit
Prepared Callback"
- "Rollback Prepared Transaction Callback" --> "Transaction Rollback
Prepared Callback"
- "whenever a commit prepared transaction has been decoded" -->
"whenever a transaction commit prepared has been decoded"
- "whenever a rollback prepared transaction has been decoded." -->
"whenever a transaction rollback prepared has been decoded."

;

Updated to this

==========
Patch V6-0001, File: src/backend/replication/logical/decode.c
==========

COMMENT
Line 74
@@ -70,6 +70,9 @@ static void DecodeCommit(LogicalDecodingContext
*ctx, XLogRecordBuffer *buf,
xl_xact_parsed_commit *parsed, TransactionId xid);
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
xl_xact_parsed_abort *parsed, TransactionId xid);
+static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
+ xl_xact_parsed_prepare * parsed);

The 2nd line of DecodePrepare is misaligned by one space.

;

COMMENT
Line 321
@@ -312,17 +315,34 @@ DecodeXactOp(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf)
}
break;
case XLOG_XACT_PREPARE:
+ {
+ xl_xact_parsed_prepare parsed;
+ xl_xact_prepare *xlrec;
+ /* check that output plugin is capable of twophase decoding */

"twophase" --> "two-phase"

~

Also, add a blank line after the declarations.

;

==========
Patch V6-0001, File: src/backend/replication/logical/logical.c
==========

COMMENT
Line 249
@@ -225,6 +237,19 @@ StartupDecodingContext(List *output_plugin_options,
(ctx->callbacks.stream_message_cb != NULL) ||
(ctx->callbacks.stream_truncate_cb != NULL);

+ /*
+ * To support two phase logical decoding, we require
prepare/commit-prepare/abort-prepare
+ * callbacks. The filter-prepare callback is optional. We however
enable two phase logical
+ * decoding when at least one of the methods is enabled so that we
can easily identify
+ * missing methods.

The terminology is generally well known as "two-phase" (with the
hyphen) https://en.wikipedia.org/wiki/Two-phase_commit_protocol so
let's be consistent for all the patch code comments. Please search the
code and correct this in all places, even where I might have missed to
identify it.

"two phase" --> "two-phase"

;

COMMENT
Line 822
@@ -782,6 +807,111 @@ commit_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
}

static void
+prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr prepare_lsn)

"support 2 phase" --> "supports two-phase" in the comment

;

COMMENT
Line 844
Code condition seems strange and/or broken.
if (ctx->enable_twophase && ctx->callbacks.prepare_cb == NULL)
Because if the flag is null then this condition is skipped.
But then if the callback was also NULL then attempting to call it to
"do the actual work" will give NPE.

~

Also, I wonder should this check be the first thing in this function?
Because if it fails does it even make sense that all the errcallback
code was set up?
E.g errcallback.arg potentially is left pointing to a stack variable
on a stack that no longer exists.

Updated accordingly.

;

COMMENT
Line 857
+commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

"support 2 phase" --> "supports two-phase" in the comment

~

Also, Same potential trouble with the condition:
if (ctx->enable_twophase && ctx->callbacks.commit_prepared_cb == NULL)
Same as previously asked. Should this check be first thing in this function?

;

COMMENT
Line 892
+abort_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

"support 2 phase" --> "supports two-phase" in the comment

~

Same potential trouble with the condition:
if (ctx->enable_twophase && ctx->callbacks.abort_prepared_cb == NULL)
Same as previously asked. Should this check be the first thing in this function?

;

COMMENT
Line 1013
@@ -858,6 +988,51 @@ truncate_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
error_context_stack = errcallback.previous;
}

+static bool
+filter_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ TransactionId xid, const char *gid)

Fix wording in comment:
"twophase" --> "two-phase transactions"
"twophase transactions" --> "two-phase transactions"

Updated accordingly.

==========
Patch V6-0001, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
Line 255
@@ -251,7 +251,8 @@ static Size
ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn
static void ReorderBufferRestoreChange(ReorderBuffer *rb,
ReorderBufferTXN *txn,
char *change);
static void ReorderBufferRestoreCleanup(ReorderBuffer *rb,
ReorderBufferTXN *txn);
-static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
+static void ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
+ bool txn_prepared);

The alignment is inconsistent. One more space needed before "bool txn_prepared"

;

COMMENT
Line 417
@@ -413,6 +414,11 @@ ReorderBufferReturnTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
}

/* free data that's contained */
+ if (txn->gid != NULL)
+ {
+ pfree(txn->gid);
+ txn->gid = NULL;
+ }

Should add the blank link before this new code, as it was before.

;

COMMENT
Line 1564
@ -1502,12 +1561,14 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
}

/*
- * Discard changes from a transaction (and subtransactions), after streaming
- * them. Keep the remaining info - transactions, tuplecids, invalidations and
- * snapshots.
+ * Discard changes from a transaction (and subtransactions), either
after streaming or
+ * after a PREPARE.

typo "snapshots.If" -> "snapshots. If"

;

Updated Accordingly.

COMMENT/QUESTION
Line 1590
@@ -1526,7 +1587,7 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
Assert(rbtxn_is_known_subxact(subtxn));
Assert(subtxn->nsubtxns == 0);

- ReorderBufferTruncateTXN(rb, subtxn);
+ ReorderBufferTruncateTXN(rb, subtxn, txn_prepared);
}

There are some code paths here I did not understand how they match the comments.
Because this function is recursive it seems that it may be called
where the 2nd parameter txn is a sub-transaction.

But then this seems at odds with some of the other code comments of
this function which are processing the txn without ever testing is it
really toplevel or not:

e.g. Line 1593 "/* cleanup changes in the toplevel txn */"
e.g. Line 1632 "They are always stored in the toplevel transaction."

;

I see that another commit in between has updated this now.

COMMENT
Line 1644
@@ -1560,9 +1621,33 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn)
* about the toplevel xact (we send the XID in all messages), but we never
* stream XIDs of empty subxacts.
*/
- if ((!txn->toptxn) || (txn->nentries_mem != 0))
+ if ((!txn_prepared) && ((!txn->toptxn) || (txn->nentries_mem != 0)))
txn->txn_flags |= RBTXN_IS_STREAMED;

+ if (txn_prepared)

/* remove the change from it's containing list */
typo "it's" --> "its"

Updated.

;

QUESTION
Line 1977
@@ -1880,7 +1965,7 @@ ReorderBufferResetTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
ReorderBufferChange *specinsert)
{
/* Discard the changes that we just streamed */
- ReorderBufferTruncateTXN(rb, txn);
+ ReorderBufferTruncateTXN(rb, txn, false);

How do you know the 3rd parameter - i.e. txn_prepared - should be
hardwired false here?
e.g. I thought that maybe rbtxn_prepared(txn) can be true here.

;

This particular function is only called when streaming and not when
handling a prepared transaction.

COMMENT
Line 2345
@@ -2249,7 +2334,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
break;
}
}
-
/*

Looks like accidental blank line deletion. This should be put back how it was

;

COMMENT/QUESTION
Line 2374
@@ -2278,7 +2362,16 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
}
}
else
- rb->commit(rb, txn, commit_lsn);
+ {
+ /*
+ * Call either PREPARE (for twophase transactions) or COMMIT
+ * (for regular ones).

"twophase" --> "two-phase"

~

Updated.

Also, I was confused by the apparent assumption of exclusiveness of
streaming and 2PC...
e.g. what if streaming AND 2PC then it won't do rb->prepare()

;

QUESTION
Line 2424
@@ -2319,11 +2412,17 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
*/
if (streaming)
{
- ReorderBufferTruncateTXN(rb, txn);
+ ReorderBufferTruncateTXN(rb, txn, false);

/* Reset the CheckXidAlive */
CheckXidAlive = InvalidTransactionId;
}
+ else if (rbtxn_prepared(txn))

I was confused by the exclusiveness of streaming/2PC.
e.g. what if streaming AND 2PC at same time - how can you pass false
as 3rd param to ReorderBufferTruncateTXN?

ReorderBufferProcessTXN can only be called when streaming individual
commands and not for streaming a prepare or a commit, Streaming of
prepare and commit would be handled as part of
ReorderBufferStreamCommit.

;

COMMENT
Line 2463
@@ -2352,17 +2451,18 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,

/*
* The error code ERRCODE_TRANSACTION_ROLLBACK indicates a concurrent
- * abort of the (sub)transaction we are streaming. We need to do the
+ * abort of the (sub)transaction we are streaming or preparing. We
need to do the
* cleanup and return gracefully on this error, see SetupCheckXidLive.
*/

"twoi phase" --> "two-phase"

;

QUESTIONS
Line 2482
@@ -2370,10 +2470,19 @@ ReorderBufferProcessTXN(ReorderBuffer *rb,
ReorderBufferTXN *txn,
errdata = NULL;
curtxn->concurrent_abort = true;

- /* Reset the TXN so that it is allowed to stream remaining data. */
- ReorderBufferResetTXN(rb, txn, snapshot_now,
- command_id, prev_lsn,
- specinsert);
+ /* If streaming, reset the TXN so that it is allowed to stream
remaining data. */
+ if (streaming)

Re: /* If streaming, reset the TXN so that it is allowed to stream
remaining data. */
I was confused by the exclusiveness of streaming/2PC.
Is it not possible for streaming flags and rbtxn_prepared(txn) true at
the same time?

Same as above.

~

elog(LOG, "stopping decoding of %s (%u)",
txn->gid[0] != '\0'? txn->gid:"", txn->xid);

Is this a safe operation, or do you also need to test txn->gid is not NULL?

Since this is in code where it is not streaming and therefore
rbtxn_prepared(txn), so gid has to be NOT NULL.

;

COMMENT
Line 2606
+ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,

"twophase" --> "two-phase"

;

QUESTION
Line 2655
+ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,

"This is used to handle COMMIT/ABORT PREPARED"
Should that say "COMMIT/ROLLBACK PREPARED"?

;

COMMENT
Line 2668

"Anyways, 2PC transactions" --> "Anyway, two-phase transactions"

;

COMMENT
Line 2765
@@ -2495,7 +2731,13 @@ ReorderBufferAbort(ReorderBuffer *rb,
TransactionId xid, XLogRecPtr lsn)
/* cosmetic... */
txn->final_lsn = lsn;

- /* remove potential on-disk data, and deallocate */
+ /*
+ * remove potential on-disk data, and deallocate.
+ *

Remove the blank between the comment and code.

==========
Patch V6-0001, File: src/include/replication/logical.h
==========

COMMENT
Line 89

"two phase" -> "two-phase"

;

COMMENT
Line 89

For consistency with the previous member naming really the new member
should just be called "twophase" rather than "enable_twophase"

;

Updated accordingly.

==========
Patch V6-0001, File: src/include/replication/output_plugin.h
==========

QUESTION
Line 106

As previously asked, why is the callback function/typedef referred as
AbortPrepared instead of RollbackPrepared?
It does not match the SQL and the function comment, and seems only to
add some unnecessary confusion.

;

==========
Patch V6-0001, File: src/include/replication/reorderbuffer.h
==========

QUESTION
Line 116
@@ -162,9 +163,13 @@ typedef struct ReorderBufferChange
#define RBTXN_HAS_CATALOG_CHANGES 0x0001
#define RBTXN_IS_SUBXACT 0x0002
#define RBTXN_IS_SERIALIZED 0x0004
-#define RBTXN_IS_STREAMED 0x0008
-#define RBTXN_HAS_TOAST_INSERT 0x0010
-#define RBTXN_HAS_SPEC_INSERT 0x0020
+#define RBTXN_PREPARE 0x0008
+#define RBTXN_COMMIT_PREPARED 0x0010
+#define RBTXN_ROLLBACK_PREPARED 0x0020
+#define RBTXN_COMMIT 0x0040
+#define RBTXN_IS_STREAMED 0x0080
+#define RBTXN_HAS_TOAST_INSERT 0x0100
+#define RBTXN_HAS_SPEC_INSERT 0x0200

I was wondering why when adding new flags, some of the existing flag
masks were also altered.
I am assuming this is ok because they are never persisted but are only
used in the protocol (??)

;

COMMENT
Line 226
@@ -218,6 +223,15 @@ typedef struct ReorderBufferChange
((txn)->txn_flags & RBTXN_IS_STREAMED) != 0 \
)

+/* is this txn prepared? */
+#define rbtxn_prepared(txn) (txn->txn_flags & RBTXN_PREPARE)
+/* was this prepared txn committed in the meanwhile? */
+#define rbtxn_commit_prepared(txn) (txn->txn_flags & RBTXN_COMMIT_PREPARED)
+/* was this prepared txn aborted in the meanwhile? */
+#define rbtxn_rollback_prepared(txn) (txn->txn_flags & RBTXN_ROLLBACK_PREPARED)
+/* was this txn committed in the meanwhile? */
+#define rbtxn_commit(txn) (txn->txn_flags & RBTXN_COMMIT)
+

Probably all the "txn->txn_flags" here might be more safely written
with parentheses in the macro like "(txn)->txn_flags".

~

Also, Start all comments with capital. And what is the meaning "in the
meanwhile?"

;

COMMENT
Line 410
@@ -390,6 +407,39 @@ typedef void (*ReorderBufferCommitCB) (ReorderBuffer *rb,
ReorderBufferTXN *txn,
XLogRecPtr commit_lsn);

The format is inconsistent with all other callback signatures here,
where the 1st arg was on the same line as the typedef.

;

COMMENT
Line 440-442

Excessive blank lines following this change?

;

COMMENT
Line 638
@@ -571,6 +631,15 @@ void
ReorderBufferXidSetCatalogChanges(ReorderBuffer *, TransactionId xid,
XLog
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *, TransactionId xid);
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *, TransactionId xid);

+bool ReorderBufferPrepareNeedSkip(ReorderBuffer *rb, TransactionId xid,
+ const char *gid);
+bool ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid,
+ const char *gid);
+void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
+ XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
+ TimestampTz commit_time,
+ RepOriginId origin_id, XLogRecPtr origin_lsn,
+ char *gid);

Not aligned consistently with other function prototypes.

;

Updated

==========
Patch V6-0003, File: src/backend/access/transam/twophase.c
==========

COMMENT
Line 551
@@ -548,6 +548,37 @@ MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
}

/*
+ * LookupGXact
+ * Check if the prepared transaction with the given GID is around
+ */
+bool
+LookupGXact(const char *gid)

There is potential to refactor/simplify this code:
e.g.

bool
LookupGXact(const char *gid)
{
int i;
bool found = false;

LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
{
GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
/* Ignore not-yet-valid GIDs */
if (gxact->valid && strcmp(gxact->gid, gid) == 0)
{
found = true;
break;
}
}
LWLockRelease(TwoPhaseStateLock);
return found;
}

;

Updated accordingly.

==========
Patch V6-0003, File: src/backend/replication/logical/proto.c
==========

COMMENT
Line 86
@@ -72,12 +72,17 @@ logicalrep_read_begin(StringInfo in,
LogicalRepBeginData *begin_data)
*/
void
logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn,
- XLogRecPtr commit_lsn)

Since now the flags are used the code comment is wrong.
"/* send the flags field (unused for now) */"

;

COMMENT
Line 129
@ -106,6 +115,77 @@ logicalrep_read_commit(StringInfo in,
LogicalRepCommitData *commit_data)
}

/*
+ * Write PREPARE to the output stream.
+ */
+void
+logicalrep_write_prepare(StringInfo out, ReorderBufferTXN *txn,

"2PC transactions" --> "two-phase commit transactions"

;

Updated

COMMENT
Line 133

Assert(strlen(txn->gid) > 0);
Shouldn't that assertion also check txn->gid is not NULL (to prevent
NPE in case gid was NULL)

In this case txn->gid has to be non NULL.

;

COMMENT
Line 177
+logicalrep_read_prepare(StringInfo in, LogicalRepPrepareData * prepare_data)

prepare_data->prepare_type = flags;
This code may be OK but it does seem a bit of an abuse of the flags.

e.g. Are they flags or are the really enum values?
e.g. And if they are effectively enums (it appears they are) then
seemed inconsistent that |= was used when they were previously
assigned.

;

I have not updated this as according to Amit this might require
refactoring again.

==========
Patch V6-0003, File: src/backend/replication/logical/worker.c
==========

COMMENT
Line 757
@@ -749,6 +753,141 @@ apply_handle_commit(StringInfo s)
pgstat_report_activity(STATE_IDLE, NULL);
}

+static void
+apply_handle_prepare_txn(LogicalRepPrepareData * prepare_data)
+{
+ Assert(prepare_data->prepare_lsn == remote_final_lsn);

Missing function comment to say this is called from apply_handle_prepare.

;

COMMENT
Line 798
+apply_handle_commit_prepared_txn(LogicalRepPrepareData * prepare_data)

Missing function comment to say this is called from apply_handle_prepare.

;

COMMENT
Line 824
+apply_handle_rollback_prepared_txn(LogicalRepPrepareData * prepare_data)

Missing function comment to say this is called from apply_handle_prepare.

Updated.

==========
Patch V6-0003, File: src/backend/replication/pgoutput/pgoutput.c
==========

COMMENT
Line 50
@@ -47,6 +47,12 @@ static void pgoutput_truncate(LogicalDecodingContext *ctx,
ReorderBufferChange *change);
static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
RepOriginId origin_id);
+static void pgoutput_prepare_txn(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn, XLogRecPtr prepare_lsn);

The parameter indentation (2nd lines) does not match everything else
in this context.

;

COMMENT
Line 152
@@ -143,6 +149,10 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->change_cb = pgoutput_change;
cb->truncate_cb = pgoutput_truncate;
cb->commit_cb = pgoutput_commit_txn;
+
+ cb->prepare_cb = pgoutput_prepare_txn;
+ cb->commit_prepared_cb = pgoutput_commit_prepared_txn;
+ cb->abort_prepared_cb = pgoutput_abort_prepared_txn;

Remove the unnecessary blank line.

;

QUESTION
Line 386
@@ -373,7 +383,49 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
OutputPluginUpdateProgress(ctx);

OutputPluginPrepareWrite(ctx, true);
- logicalrep_write_commit(ctx->out, txn, commit_lsn);
+ logicalrep_write_commit(ctx->out, txn, commit_lsn, true);

Is the is_commit parameter of logicalrep_write_commit ever passed as false?
If yes, where?
If no, the what is the point of it?

It was dead code from an earlier version. I have removed it, updated
accordingly.

;

COMMENT
Line 408
+pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,

Since all this function is identical to pg_output_prepare it might be
better to either
1. just leave this as a wrapper to delegate to that function
2. remove this one entirely and assign the callback to the common
pgoutput_prepare_txn

;

I have not changed this as this might require re-factoring according to Amit.

COMMENT
Line 419
+pgoutput_abort_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Since all this function is identical to pg_output_prepare if might be
better to either
1. just leave this as a wrapper to delegate to that function
2. remove this one entirely and assign the callback to the common
pgoutput_prepare_tx

;

Same as above.

COMMENT
Line 419
+pgoutput_abort_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,

Shouldn't this comment say be "ROLLBACK PREPARED"?

;

Updated.

==========
Patch V6-0003, File: src/include/replication/logicalproto.h
==========

QUESTION
Line 101
@@ -87,20 +87,55 @@ typedef struct LogicalRepBeginData
TransactionId xid;
} LogicalRepBeginData;

+/* Commit (and abort) information */

#define LOGICALREP_IS_ABORT 0x02
Is there a good reason why this is not called:
#define LOGICALREP_IS_ROLLBACK 0x02

;

Removed.

COMMENT
Line 105

((flags == LOGICALREP_IS_COMMIT) || (flags == LOGICALREP_IS_ABORT))

Macros would be safer if flags are in parentheses
(((flags) == LOGICALREP_IS_COMMIT) || ((flags) == LOGICALREP_IS_ABORT))

;

COMMENT
Line 115

Unexpected whitespace for the typedef
"} LogicalRepPrepareData;"

;

COMMENT
Line 122
/* prepare can be exactly one of PREPARE, [COMMIT|ABORT] PREPARED*/
#define PrepareFlagsAreValid(flags) \
((flags == LOGICALREP_IS_PREPARE) || \
(flags == LOGICALREP_IS_COMMIT_PREPARED) || \
(flags == LOGICALREP_IS_ROLLBACK_PREPARED))

There is confusing mixture in macros and comments of ABORT and ROLLBACK terms
"[COMMIT|ABORT] PREPARED" --> "[COMMIT|ROLLBACK] PREPARED"

~

Also, it would be safer if flags are in parentheses
(((flags) == LOGICALREP_IS_PREPARE) || \
((flags) == LOGICALREP_IS_COMMIT_PREPARED) || \
((flags) == LOGICALREP_IS_ROLLBACK_PREPARED))

;

updated.

==========
Patch V6-0003, File: src/test/subscription/t/020_twophase.pl
==========

COMMENT
Line 131 - # check inserts are visible

Isn't this supposed to be checking for rows 12 and 13, instead of 11 and 12?

;

Updated.

==========
Patch V6-0004, File: contrib/test_decoding/test_decoding.c
==========

COMMENT
Line 81
@@ -78,6 +78,15 @@ static void
pg_decode_stream_stop(LogicalDecodingContext *ctx,
static void pg_decode_stream_abort(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);
+static void pg_decode_stream_prepare(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+static

All these functions have a 3rd parameter called commit_lsn. Even
though the functions are not commit related. It seems like a cut/paste
error.

;

COMMENT
Line 142
@@ -130,6 +139,9 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->stream_start_cb = pg_decode_stream_start;
cb->stream_stop_cb = pg_decode_stream_stop;
cb->stream_abort_cb = pg_decode_stream_abort;
+ cb->stream_prepare_cb = pg_decode_stream_prepare;
+ cb->stream_commit_prepared_cb = pg_decode_stream_commit_prepared;
+ cb->stream_abort_prepared_cb = pg_decode_stream_abort_prepared;
cb->stream_commit_cb = pg_decode_stream_commit;

Can the "cb->stream_abort_prepared_cb" be changed to
"cb->stream_rollback_prepared_cb"?

;

COMMENT
Line 827
@@ -812,6 +824,78 @@ pg_decode_stream_abort(LogicalDecodingContext *ctx,
}

static void
+pg_decode_stream_prepare(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn)
+{
+ TestDecodingData *data = ctx->output_plugin_pr

The commit_lsn (3rd parameter) is unused and seems like a cut/paste name error.

;

COMMENT
Line 875
+pg_decode_stream_abort_prepared(LogicalDecodingContext *ctx,

The commit_lsn (3rd parameter) is unused and seems like a cut/paste name error.

;

Updated.

==========
Patch V6-0004, File: doc/src/sgml/logicaldecoding.sgml
==========

COMMENT
48.6.1
@@ -396,6 +396,9 @@ typedef struct OutputPluginCallbacks
LogicalDecodeStreamStartCB stream_start_cb;
LogicalDecodeStreamStopCB stream_stop_cb;
LogicalDecodeStreamAbortCB stream_abort_cb;
+ LogicalDecodeStreamPrepareCB stream_prepare_cb;
+ LogicalDecodeStreamCommitPreparedCB stream_commit_prepared_cb;
+ LogicalDecodeStreamAbortPreparedCB stream_abort_prepared_cb;

Same question from previous review comments - why using the
terminology "abort" instead of "rollback"

;

COMMENT
48.6.1
@@ -418,7 +421,9 @@ typedef void (*LogicalOutputPluginInit) (struct
OutputPluginCallbacks *cb);
in-progress transactions. The <function>stream_start_cb</function>,
<function>stream_stop_cb</function>, <function>stream_abort_cb</function>,
<function>stream_commit_cb</function> and <function>stream_change_cb</function>
- are required, while <function>stream_message_cb</function> and
+ are required, while <function>stream_message_cb</function>,
+ <function>stream_prepare_cb</function>,
<function>stream_commit_prepared_cb</function>,
+ <function>stream_abort_prepared_cb</function>,

Missing "and".
... "stream_abort_prepared_cb, stream_truncate_cb are optional." -->
"stream_abort_prepared_cb, and stream_truncate_cb are optional."

;

COMMENT
Section 48.6.4.16
Section 48.6.4.17
Section 48.6.4.18
@@ -839,6 +844,45 @@ typedef void (*LogicalDecodeStreamAbortCB)
(struct LogicalDecodingContext *ctx,
</para>
</sect3>

+ <sect3 id="logicaldecoding-output-plugin-stream-prepare">
+ <title>Stream Prepare Callback</title>
+ <para>
+ The <function>stream_prepare_cb</function> callback is called to prepare
+ a previously streamed transaction as part of a two phase commit.
+<programlisting>
+typedef void (*LogicalDecodeStreamPrepareCB) (struct
LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn);
+</programlisting>
+ </para>
+ </sect3>
+
+ <sect3 id="logicaldecoding-output-plugin-stream-commit-prepared">
+ <title>Stream Commit Prepared Callback</title>
+ <para>
+ The <function>stream_commit_prepared_cb</function> callback is
called to commit prepared
+ a previously streamed transaction as part of a two phase commit.
+<programlisting>
+typedef void (*LogicalDecodeStreamCommitPreparedCB) (struct
LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn);
+</programlisting>
+ </para>
+ </sect3>
+
+ <sect3 id="logicaldecoding-output-plugin-stream-abort-prepared">
+ <title>Stream Abort Prepared Callback</title>
+ <para>
+ The <function>stream_abort_prepared_cb</function> callback is called
to abort prepared
+ a previously streamed transaction as part of a two phase commit.
+<programlisting>
+typedef void (*LogicalDecodeStreamAbortPreparedCB) (struct
LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn);
+</programlisting>
+ </para>
+ </sect3>

1. Everywhere it says "two phase" commit should be consistently
replaced to say "two-phase" commit (with the hyphen)

2. Search for "abort_lsn" parameter. It seems to be overused
(cut/paste error) even when the API is unrelated to abort

3. 48.6.4.17 and 48.6.4.18
Is this wording ok? Is the word "prepared" even necessary here?
- "... called to commit prepared a previously streamed transaction ..."
- "... called to abort prepared a previously streamed transaction ..."

;

Updated accordingly.

COMMENT
Section 48.9
@@ -1017,9 +1061,13 @@ OutputPluginWrite(ctx, true);
When streaming an in-progress transaction, the changes (and messages) are
streamed in blocks demarcated by <function>stream_start_cb</function>
and <function>stream_stop_cb</function> callbacks. Once all the decoded
- changes are transmitted, the transaction is committed using the
- <function>stream_commit_cb</function> callback (or possibly aborted using
- the <function>stream_abort_cb</function> callback).
+ changes are transmitted, the transaction can be committed using the
+ the <function>stream_commit_cb</function> callback

"two phase" --> "two-phase"

~

Also, Missing period on end of sentence.
"or aborted using the stream_abort_prepared_cb" --> "or aborted using
the stream_abort_prepared_cb."

;

Updated accordingly.

==========
Patch V6-0004, File: src/backend/replication/logical/logical.c
==========

COMMENT
Line 84
@@ -81,6 +81,12 @@ static void stream_stop_cb_wrapper(ReorderBuffer
*cache, ReorderBufferTXN *txn,
XLogRecPtr last_lsn);
static void stream_abort_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);
+static void stream_prepare_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+static void stream_commit_prepared_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+static void stream_abort_prepared_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);

The 3rd parameter is always "commit_lsn" even for API unrelated to
commit, so seems like cut/paste error.

;

COMMENT
Line 1246
@@ -1231,6 +1243,105 @@ stream_abort_cb_wrapper(ReorderBuffer *cache,
ReorderBufferTXN *txn,
}

static void
+stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn)
+{
+ LogicalDecodingContext *ctx = cache->private_data;
+ LogicalErrorCallbackState state;

Misnamed parameter "commit_lsn" ?

~

Also, Line 1272
There seem to be some missing integrity checking to make sure the
callback is not NULL.
A null callback will give NPE when wrapper attempts to call it

;

COMMENT
Line 1305
+static void
+stream_commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

There seem to be some missing integrity checking to make sure the
callback is not NULL.
A null callback will give NPE when wrapper attempts to call it.

;

COMMENT
Line 1312
+static void
+stream_abort_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,

Misnamed parameter "commit_lsn" ?

~

Also, Line 1338
There seem to be some missing integrity checking to make sure the
callback is not NULL.
A null callback will give NPE when wrapper attempts to call it.

Updated accordingly.

==========
Patch V6-0004, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
Line 2684
@@ -2672,15 +2681,31 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb,
TransactionId xid,
txn->gid = palloc(strlen(gid) + 1); /* trailing '\0' */
strcpy(txn->gid, gid);

- if (is_commit)
+ if (rbtxn_is_streamed(txn))
{
- txn->txn_flags |= RBTXN_COMMIT_PREPARED;
- rb->commit_prepared(rb, txn, commit_lsn);
+ if (is_commit)
+ {
+ txn->txn_flags |= RBTXN_COMMIT_PREPARED;

The setting/checking of the flags could be refactored if you wanted to
write less code:
e.g.
if (is_commit)
txn->txn_flags |= RBTXN_COMMIT_PREPARED;
else
txn->txn_flags |= RBTXN_ROLLBACK_PREPARED;

if (rbtxn_is_streamed(txn) && rbtxn_commit_prepared(txn))
rb->stream_commit_prepared(rb, txn, commit_lsn);
else if (rbtxn_is_streamed(txn) && rbtxn_rollback_prepared(txn))
rb->stream_abort_prepared(rb, txn, commit_lsn);
else if (rbtxn_commit_prepared(txn))
rb->commit_prepared(rb, txn, commit_lsn);
else if (rbtxn_rollback_prepared(txn))
rb->abort_prepared(rb, txn, commit_lsn);

;

Updated accordingly.

==========
Patch V6-0004, File: src/include/replication/output_plugin.h
==========

COMMENT
Line 171
@@ -157,6 +157,33 @@ typedef void (*LogicalDecodeStreamAbortCB)
(struct LogicalDecodingContext *ctx,
XLogRecPtr abort_lsn);

/*
+ * Called to prepare changes streamed to remote node from in-progress
+ * transaction. This is called as part of a two-phase commit and only when
+ * two-phased commits are supported
+ */

1. Missing period all these comments.

2. Is the part that says "and only where two-phased commits are
supported" necessary to say? Is seems redundant since comments already
says called as part of a two-phase commit.

;

==========
Patch V6-0004, File: src/include/replication/reorderbuffer.h
==========

COMMENT
Line 467
@@ -466,6 +466,24 @@ typedef void (*ReorderBufferStreamAbortCB) (
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamPrepareCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamCommitPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamAbortPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);

Cut/paste error - repeated same comment 3 times?

Updated Accordingly.

[END]

I believe I have addressed all of Peter's comments. Peter, do have a
look and let me know if I missed anything or if you find anythinge
else. Thanks for your comments, much appreciated.

regards,
Ajin Cherian
Fujitsu Australia

On Wed, Oct 14, 2020 at 6:15 PM Ajin Cherian <itsajin@gmail.com> wrote:

I think it will be easier to review this work if we can split the
patches according to the changes made in different layers. The first
patch could be changes made in output plugin and the corresponding
changes in test_decoding, see the similar commit of in-progress
transactions [1]https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=45fdc9738b36d1068d3ad8fdb06436d6fd14436b. So you need to move corresponding changes from
v8-0001-Support-decoding-of-two-phase-transactions and
v8-0004-Support-two-phase-commits-in-streaming-mode-in-lo for this.
The second patch could be changes made in ReorderBuffer to support
this feature, see [2]https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=7259736a6e5b7c7588fff9578370736a6648acbb. The third patch could be changes made to
support pgoutput and subscriber-side stuff, see [3]https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=464824323e57dc4b397e8b05854d779908b55304. What do you
think?

[1]: https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=45fdc9738b36d1068d3ad8fdb06436d6fd14436b
[2]: https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=7259736a6e5b7c7588fff9578370736a6648acbb
[3]: https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=464824323e57dc4b397e8b05854d779908b55304

--
With Regards,
Amit Kapila.

On Thu, Oct 15, 2020 at 2:16 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Wed, Oct 14, 2020 at 6:15 PM Ajin Cherian <itsajin@gmail.com> wrote:

I think it will be easier to review this work if we can split the
patches according to the changes made in different layers. The first
patch could be changes made in output plugin and the corresponding
changes in test_decoding, see the similar commit of in-progress
transactions [1]. So you need to move corresponding changes from
v8-0001-Support-decoding-of-two-phase-transactions and
v8-0004-Support-two-phase-commits-in-streaming-mode-in-lo for this.
The second patch could be changes made in ReorderBuffer to support
this feature, see [2]. The third patch could be changes made to
support pgoutput and subscriber-side stuff, see [3]. What do you
think?

I agree. I have split the patches accordingly. Do have a look.
Pending work is:
1. Add pgoutput support for the new streaming two-phase commit APIs
2. Add test cases for two-phase commits with streaming for pub/sub and
test_decoding
3. Add CREATE SUBSCRIPTION command option to specify two-phase commits
rather than having it turned on by default.

regards,
Ajin Cherian
Fujitsu Australia

Hello Ajin,

The v9 patches provided support for two-phase transactions for NON-streaming.

Now I have added STREAM support for two-phase transactions, and bumped
all patches to version v10.

(The 0001 and 0002 patches are unchanged. Only 0003 is changed).

--

There are a few TODO/FIXME comments in the code highlighting parts
needing some attention.

There is a #define DEBUG_STREAM_2PC useful for debugging, which I can
remove later.

All the patches have some whitespaces issues when applied. We can
resolve them as we go.

Please let me know any comments/feedback.

Kind Regards
Peter Smith.
Fujitsu Australia.

Hello Ajin.

I have gone through the v10 patches to verify if and how my previous
v6 review comments got addressed.

Some issues remain, and there are a few newly introduced ones.

Mostly it is all very minor stuff.

Please find my revised review comments below.

Kind Regards.
Peter Smith
Fujitsu Australia

---

V10 REVIEW COMMENTS FOLLOW

==========
Patch v10-0001, File: contrib/test_decoding/test_decoding.c
==========

COMMENT
Line 285
+ {
+ errno = 0;
+ data->check_xid_aborted = (TransactionId)
+ strtoul(strVal(elem->arg), NULL, 0);
+
+ if (!TransactionIdIsValid(data->check_xid_aborted))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("check-xid-aborted is not a valid xid: \"%s\"",
+ strVal(elem->arg))));
+ }

I think it is risky to assign strtoul directly to the
check_xid_aborted member because it makes some internal assumption
that the invalid transaction is the same as the error return from
strtoul.

Maybe better to do in 2 steps like below:

BEFORE
errno = 0;
data->check_xid_aborted = (TransactionId)strtoul(strVal(elem->arg), NULL, 0);

AFTER
long xid;
errno = 0;
xid = strtoul(strVal(elem->arg), NULL, 0);
if (xid == 0 || errno != 0)
data->check_xid_aborted = InvalidTransactionId;
else
data->check_xid_aborted =(TransactionId)xid;

---

COMMENT
Line 430
+
+/* ABORT PREPARED callback */
+static void
+pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn)

Fix comment "ABORT PREPARED" --> "ROLLBACK PREPARED"

==========
Patch v10-0001, File: doc/src/sgml/logicaldecoding.sgml
==========

COMMENT
Section 48.6.1
Says:
An output plugin may also define functions to support streaming of
large, in-progress transactions. The stream_start_cb, stream_stop_cb,
stream_abort_cb, stream_commit_cb and stream_change_cb are required,
while stream_message_cb, stream_prepare_cb, stream_commit_prepared_cb,
stream_rollback_prepared_cb and stream_truncate_cb are optional.

An output plugin may also define functions to support two-phase
commits, which are decoded on PREPARE TRANSACTION. The prepare_cb,
commit_prepared_cb and rollback_prepared_cb callbacks are required,
while filter_prepare_cb is optional.

-

But is that correct? It seems strange/inconsistent to say that the 2PC
callbacks are mandatory for the non-streaming, but that they are
optional for streaming.

---

COMMENT
48.6.4.5 "Transaction Prepare Callback"
48.6.4.6 "Transaction Commit Prepared Callback"
48.6.4.7 "Transaction Rollback Prepared Callback"

There seems some confusion about what is optional and what is
mandatory. e.g. Why are the non-stream 2PC callbacks mandatory but the
stream 2PC callbacks are not? And also there is some inconsistency
with what is said in the paragraph at the top of the page versus what
each of the callback sections says wrt optional/mandatory.

The sub-sections 49.6.4.5, 49.6.4.6, 49.6.4.7 say those callbacks are
optional which IIUC Amit said is incorrect. This is similar to the
previous review comment

---

COMMENT
Section 48.6.4.7 "Transaction Rollback Prepared Callback"

parameter "abort_lsn" probably should be "rollback_lsn"

---

COMMENT
Section 49.6.4.18. "Stream Rollback Prepared Callback"
Says:
The stream_rollback_prepared_cb callback is called to abort a
previously streamed transaction as part of a two-phase commit.

maybe should say "is called to rollback"

==========
Patch v10-0001, File: src/backend/replication/logical/logical.c
==========

COMMENT
Line 252
Says: We however enable two phase logical...

"two phase" --> "two-phase"

--

COMMENT
Line 885
Line 923
Says: If the plugin support 2 phase commits...

"support 2 phase" --> "supports two-phase" in the comment. Same issue
occurs twice.

---

COMMENT
Line 830
Line 868
Line 906
Says:
/* We're only supposed to call this when two-phase commits are supported */

There is an extra space between the "are" and "supported" in the comment.
Same issue occurs 3 times.

---

COMMENT
Line 1023
+ /*
+ * Skip if decoding of two-phase at PREPARE time is not enabled. In that
+ * case all two-phase transactions are considered filtered out and will be
+ * applied as regular transactions at COMMIT PREPARED.
+ */

Comment still is missing the word "transactions"
"Skip if decoding of two-phase at PREPARE time is not enabled."
-> "Skip if decoding of two-phase transactions at PREPARE time is not enabled.

==========
Patch v10-0001, File: src/include/replication/reorderbuffer.h
==========

COMMENT
Line 459
/* abort prepared callback signature */
typedef void (*ReorderBufferRollbackPreparedCB) (
ReorderBuffer *rb,
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

There is no alignment consistency here for
ReorderBufferRollbackPreparedCB. Some function args are directly under
the "(" and some are on the same line. This function code is neither.

---

COMMENT
Line 638
@@ -431,6 +486,24 @@ typedef void (*ReorderBufferStreamAbortCB) (
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamPrepareCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr prepare_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamCommitPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamRollbackPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr rollback_lsn);

There is no inconsistent alignment with the arguments (compare how
other functions are aligned)

See:
- for ReorderBufferStreamCommitPreparedCB
- for ReorderBufferStreamRollbackPreparedCB
- for ReorderBufferPrepareNeedSkip
- for ReorderBufferTxnIsPrepared
- for ReorderBufferPrepare

---

COMMENT
Line 489
Line 495
Line 501
/* prepare streamed transaction callback signature */

Same comment cut/paste 3 times?
- for ReorderBufferStreamPrepareCB
- for ReorderBufferStreamCommitPreparedCB
- for ReorderBufferStreamRollbackPreparedCB

---

COMMENT
Line 457
/* abort prepared callback signature */
typedef void (*ReorderBufferRollbackPreparedCB) (
ReorderBuffer *rb,
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

"abort" --> "rollback" in the function comment.

---

COMMENT
Line 269
/* In case of 2PC we need to pass GID to output plugin */

"2PC" --> "two-phase commit"

==========
Patch v10-0002, File: contrib/test_decoding/expected/two_phase.out (and .sql)
==========

COMMENT
General

It is a bit hard to see what are the main tests here are what are just
sub-parts of some test case.

e.g. It seems like the main tests are.

1. Test that decoding happens at PREPARE time
2. Test decoding of an aborted tx
3. Test a prepared tx which contains some DDL
4. Test decoding works while an uncommitted prepared tx with DDL exists
5. Test operations holding exclusive locks won't block decoding
6. Test savepoints and sub-transactions
7. Test "_nodecode" will defer the decoding until the commit time

Can the comments be made more obvious so it is easy to distinguish the
main tests from the steps of those tests?

---

COMMENT
Line 1
-- Test two-phased transactions, when two-phase-commit is enabled,
transactions are
-- decoded at PREPARE time rather than at COMMIT PREPARED time.

Some commas to be removed and this comment to be split into several sentences.

---

COMMENT
Line 19
-- should show nothing

Comment could be more informative. E.g. "Should show nothing because
the PREPARE has not happened yet"

---

COMMENT
Line 77

Looks like there is a missing comment about here that should say
something like "Show that the DDL does not appear in the decoding"

---

COMMENT
Line 160
-- test savepoints and sub-xacts as a result

The subsequent test is testing savepoints. But is it testing sub
transactions like the comment says?

==========
Patch v10-0002, File: contrib/test_decoding/t/001_twophase.pl
==========

COMMENT
General

I think basically there are only 2 tests in this file.
1. to check that the concurrent abort works.
2. to check that the prepared tx can span a server shutdown/restart

But the tests comments do not make this clear at all.
e.g. All the "#" comments look equally important although most of them
are just steps of each test case.
Can the comments be better to distinguish the tests versus the steps
of each test?

==========
Patch v10-0002, File: src/backend/replication/logical/decode.c
==========

COMMENT
Line 71
static void DecodeCommitPrepared(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf,
xl_xact_parsed_commit *parsed, TransactionId xid);
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
xl_xact_parsed_abort *parsed, TransactionId xid);
static void DecodeAbortPrepared(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf,
xl_xact_parsed_abort *parsed, TransactionId xid);
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
xl_xact_parsed_prepare * parsed);

The 2nd line or args are not aligned properly.
- for DecodeCommitPrepared
- for DecodeAbortPrepared
- for DecodePrepare

==========
Patch v10-0002, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
There are some parts of the code where in my v6 review I had a doubt
about the mutually exclusive treatment of the "streaming" flag and the
"rbtxn_prepared(txn)" state.

Basically I did not see how some parts of the code are treating NOT
streaming as implying 2PC etc because it defies my understanding that
2PC can also work in streaming mode. Perhaps the "streaming" flag has
a different meaning to how I interpret it? Or perhaps some functions
are guarding higher up and can only be called under certain
conditions?

Anyway, this confusion manifests in several parts of the code, none of
which was changed after my v6 review.

Affected code includes the following:

CASE 1
Wherever the ReorderBufferTruncateTXN(...) "prepared" flag (third
parameter) is hardwired true/false, I think there must be some
preceding Assert to guarantee the prepared state condition holds true.
There can't be any room for doubts like "but what will it do for
streamed 2PC..."
Line 1805 - ReorderBufferTruncateTXN(rb, txn, true); // if rbtxn_prepared(txn)
Line 1941 - ReorderBufferTruncateTXN(rb, txn, false); // state ??
Line 2389 - ReorderBufferTruncateTXN(rb, txn, false); // if streaming
Line 2396 - ReorderBufferTruncateTXN(rb, txn, true); // if not
streaming and if rbtxm_prepared(txn)
Line 2459 - ReorderBufferTruncateTXN(rb, txn, true); // if not streaming

~

CASE 2
Wherever the "streaming" flag is tested I don't really understand how
NOT streaming can automatically imply 2PC.
Line 2330 - if (streaming) // what about if it is streaming AND 2PC at
the same time?
Line 2387 - if (streaming) // what about if it is streaming AND 2PC at
the same time?
Line 2449 - if (streaming) // what about if it is streaming AND 2PC at
the same time?

~

Case 1 and Case 2 above overlap a fair bit. I just listed them so they
all get checked again.

Even if the code is thought to be currently OK I do still think
something should be done like:
a) add some more substantial comments to explain WHY the combination
of streaming and 2PC is not valid in the context
b) the Asserts to be strengthened to 100% guarantee that the streaming
and prepared states really are exclusive (if indeed they are). For
this point I thought the following Assert condition could be better:
Assert(streaming || rbtxn_prepared(txn));
Assert(stream_started || rbtxn_prepared(txn));
because as it is you still are left wondering if both streaming AND
rbtxn_prepared(txn) can be possible at the same time...

---

COMMENT
Line 2634
* Anyways, two-phase transactions do not contain any reorderbuffers.

"Anyways" --> "Anyway"

==========
Patch v10-0003, File: src/backend/access/transam/twophase.c
==========

COMMENT
Line 557
@@ -548,6 +548,33 @@ MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
}

 /*
+ * LookupGXact
+ * Check if the prepared transaction with the given GID is around
+ */
+bool
+LookupGXact(const char *gid)
+{
+ int i;
+ bool found = false;

The variable declarations (i and found) are not aligned.

==========
Patch v10-0003, File: src/backend/replication/logical/proto.c
==========

COMMENT
Line 125
Line 205
Assert(strlen(txn->gid) > 0);

I suggested that the assertion should also check txn->gid is not NULL.
You replied "In this case txn->gid has to be non NULL".

But that is exactly what I said :-)
If it HAS to be non-NULL then why not just Assert that in code instead
of leaving the reader wondering?

"Assert(strlen(txn->gid) > 0);" --> "Assert(tdx->gid && strlen(txn->gid) > 0);"
Same occurs several times.

---

COMMENT
Line 133
Line 213
if (rbtxn_commit_prepared(txn))
flags |= LOGICALREP_IS_COMMIT_PREPARED;
else if (rbtxn_rollback_prepared(txn))
flags |= LOGICALREP_IS_ROLLBACK_PREPARED;
else
flags |= LOGICALREP_IS_PREPARE;

Previously I wrote that the use of the bit flags on assignment in the
logicalrep_write_prepare was inconsistent with the way they are
treated when they are read. Really it should be using a direct
assignment instead of bit flags.

You said this is skipped anticipating a possible refactor. But IMO
this leaves the code in a half/half state. I think it is better to fix
it properly and if refactoring happens then deal with that at the
time.

The last comment I saw from Amit said to use my 1st proposal of direct
assignment instead of bit flag assignment.

(applies to both non-stream and stream functions)
- see logicalrep_write_prepare
- see logicalrep_write_stream_prepare

==========
Patch v10-0003, File: src/backend/replication/pgoutput/pgoutput.c
==========

COMMENT
Line 429
/*
* PREPARE callback
*/
static void
pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn)
The function comment looks wrong.
Shouldn't this comment say be "ROLLBACK PREPARED callback"?

==========
Patch v10-0003, File: src/include/replication/logicalproto.h
==========

Line 115
#define PrepareFlagsAreValid(flags) \
((flags == LOGICALREP_IS_PREPARE) || \
(flags == LOGICALREP_IS_COMMIT_PREPARED) || \
(flags == LOGICALREP_IS_ROLLBACK_PREPARED))

Would be safer if all the references to flags are in parentheses
e.g. "flags" --> "(flags)"

[END]

On Fri, Oct 16, 2020 at 5:21 PM Peter Smith <smithpb2250@gmail.com> wrote:

Hello Ajin,

The v9 patches provided support for two-phase transactions for NON-streaming.

Now I have added STREAM support for two-phase transactions, and bumped
all patches to version v10.

(The 0001 and 0002 patches are unchanged. Only 0003 is changed).

--

There are a few TODO/FIXME comments in the code highlighting parts
needing some attention.

There is a #define DEBUG_STREAM_2PC useful for debugging, which I can
remove later.

All the patches have some whitespaces issues when applied. We can
resolve them as we go.

Please let me know any comments/feedback.

Hi Peter,

Thanks for your patch. Some comments for your patch:

Comments:

src/backend/replication/logical/worker.c
@@ -888,6 +888,319 @@ apply_handle_prepare(StringInfo s)
+ /*
+ * FIXME - Following condition was in apply_handle_prepare_txn except
I found  it was ALWAYS IsTransactionState() == false
+ * The synchronization worker runs in single transaction. *
+ if (IsTransactionState() && !am_tablesync_worker())
+ */
+ if (!am_tablesync_worker())

Comment: I dont think a tablesync worker will use streaming, none of
the other stream APIs check this, this might not be relevant for
stream_prepare either.

+ /*
+ * ==================================================================================================
+ * The following chunk of code is largely cut/paste from the existing
apply_handle_prepare_commit_txn

Comment: Here, I think you meant apply_handle_stream_commit. Also
rather than duplicating this chunk of code, you could put it in a new
function.

+ /* open the spool file for the committed transaction */
+ changes_filename(path, MyLogicalRepWorker->subid, xid);

Comment: Here the comment should read "committed/prepared" rather than
"committed"

+ else
+ {
+ /* Process any invalidation messages that might have accumulated. */
+ AcceptInvalidationMessages();
+ maybe_reread_subscription();
+ }

Comment: This else block might not be necessary as a tablesync worker
will not initiate the streaming APIs.

+ BeginTransactionBlock();
+ CommitTransactionCommand();
+ StartTransactionCommand();

Comment: Rereading the code and the transaction state description in
src/backend/access/transam/README. I am not entirely sure if the
BeginTransactionBlock followed by CommitTransactionBlock is really
needed here.
I understand this code was copied over from apply_handle_prepare_txn,
but now looking back I'm not so sure if it is correct. The transaction
would have already begin as part of applying the changes, why begin it
again?
Maybe Amit could confirm this.

END

regards,
Ajin Cherian
Fujitsu Australia

The PG docs for PREPARE TRANSACTION [1]https://www.postgresql.org/docs/current/sql-prepare-transaction.html don't say anything about an
empty (zero length) transaction-id.
e.g. PREPARE TRANSACTION '';
[1]: https://www.postgresql.org/docs/current/sql-prepare-transaction.html

~

Meanwhile, during testing I found the 2PC prepare hangs when an empty
id is used.

Now I am not sure does this represent some bug within the 2PC code, or
in fact should the PREPARE never have allowed an empty transaction-id
to be specified in the first place?

Thoughts?

Kind Regards
Peter Smith.
Fujitsu Australia.

On Tue, Oct 20, 2020 at 4:32 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Fri, Oct 16, 2020 at 5:21 PM Peter Smith <smithpb2250@gmail.com> wrote:

Comments:

src/backend/replication/logical/worker.c
@@ -888,6 +888,319 @@ apply_handle_prepare(StringInfo s)
+ /*
+ * FIXME - Following condition was in apply_handle_prepare_txn except
I found  it was ALWAYS IsTransactionState() == false
+ * The synchronization worker runs in single transaction. *
+ if (IsTransactionState() && !am_tablesync_worker())
+ */
+ if (!am_tablesync_worker())

Comment: I dont think a tablesync worker will use streaming, none of
the other stream APIs check this, this might not be relevant for
stream_prepare either.

Yes, I think this is right. See pgoutput_startup where we are
disabling the streaming for init phase. But it is always good to once
test this and ensure the same.

+ /*
+ * ==================================================================================================
+ * The following chunk of code is largely cut/paste from the existing
apply_handle_prepare_commit_txn

Comment: Here, I think you meant apply_handle_stream_commit. Also
rather than duplicating this chunk of code, you could put it in a new
function.

+ /* open the spool file for the committed transaction */
+ changes_filename(path, MyLogicalRepWorker->subid, xid);

Comment: Here the comment should read "committed/prepared" rather than
"committed"

+ else
+ {
+ /* Process any invalidation messages that might have accumulated. */
+ AcceptInvalidationMessages();
+ maybe_reread_subscription();
+ }

Comment: This else block might not be necessary as a tablesync worker
will not initiate the streaming APIs.

I think it is better to have an Assert here for streaming-mode?

+ BeginTransactionBlock();
+ CommitTransactionCommand();
+ StartTransactionCommand();

Comment: Rereading the code and the transaction state description in
src/backend/access/transam/README. I am not entirely sure if the
BeginTransactionBlock followed by CommitTransactionBlock is really
needed here.

Yeah, I also find this strange. I guess the patch is doing so because
it needs to call PrepareTransactionBlock later but I am not sure. How
can we call CommitTransactionCommand(), won't it commit the on-going
transaction and make it visible before even it is visible on the
publisher. I think you can verify by having a breakpoint after
CommitTransactionCommand() and see if the changes for which we are
doing prepare become visible.

I understand this code was copied over from apply_handle_prepare_txn,
but now looking back I'm not so sure if it is correct. The transaction
would have already begin as part of applying the changes, why begin it
again?
Maybe Amit could confirm this.

I hope the above suggestions will help to proceed here.

--
With Regards,
Amit Kapila.

On Wed, Oct 21, 2020 at 1:38 PM Peter Smith <smithpb2250@gmail.com> wrote:

The PG docs for PREPARE TRANSACTION [1] don't say anything about an
empty (zero length) transaction-id.
e.g. PREPARE TRANSACTION '';
[1] https://www.postgresql.org/docs/current/sql-prepare-transaction.html

~

Meanwhile, during testing I found the 2PC prepare hangs when an empty
id is used.

Can you please take an example to explain what you are trying to say?
I have tried below and doesn't face any problem:

postgres=# Begin;
BEGIN
postgres=*# select txid_current();
txid_current
--------------
534
(1 row)
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION
postgres=# Commit Prepared 'foo';
COMMIT PREPARED
postgres=# Begin;
BEGIN
postgres=*# Prepare Transaction 'foo';
PREPARE TRANSACTION
postgres=# Commit Prepared 'foo';
COMMIT PREPARED

--
With Regards,
Amit Kapila.

On Tue, Oct 20, 2020 at 9:46 AM Peter Smith <smithpb2250@gmail.com> wrote:

==========
Patch v10-0002, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
There are some parts of the code where in my v6 review I had a doubt
about the mutually exclusive treatment of the "streaming" flag and the
"rbtxn_prepared(txn)" state.

I am not sure about the exact specifics here but we can always prepare
a transaction that is streamed. I have to raise one more point in this
regard. Why do we need stream_commit_prepared_cb,
stream_rollback_prepared_cb callbacks? Do we need to do something
separate in pgoutput or otherwise for these APIs? If not, can't we use
a non-stream version of these APIs instead? There appears to be a
use-case for stream_prepare_cb which is to apply the existing changes
on subscriber and call prepare but I can't see usecase for the other
two APIs.

One minor comment:
v10-0001-Support-2PC-txn-base

1.
@@ -574,6 +655,11 @@ void ReorderBufferQueueMessage(ReorderBuffer *,
TransactionId, Snapshot snapsho
 void ReorderBufferCommit(ReorderBuffer *, TransactionId,
  XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
  TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn);
+void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
+                           XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
+                           TimestampTz commit_time,
+                           RepOriginId origin_id, XLogRecPtr origin_lsn,
+                           char *gid, bool is_commit);
 void ReorderBufferAssignChild(ReorderBuffer *, TransactionId,
TransactionId, XLogRecPtr commit_lsn);
 void ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId,
  XLogRecPtr commit_lsn, XLogRecPtr end_lsn);
@@ -597,6 +683,15 @@ void
ReorderBufferXidSetCatalogChanges(ReorderBuffer *, TransactionId xid,
XLog
 bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *, TransactionId xid);
 bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *, TransactionId xid);

+bool ReorderBufferPrepareNeedSkip(ReorderBuffer *rb, TransactionId xid,
+ const char *gid);
+bool ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid,
+    const char *gid);
+void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
+ XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
+ TimestampTz commit_time,
+ RepOriginId origin_id, XLogRecPtr origin_lsn,
+ char *gid);
 ReorderBufferTXN *ReorderBufferGetOldestTXN(ReorderBuf

I don't think these changes belong to this patch as the definition of
these functions is not part of this patch.

--
With Regards,
Amit Kapila.

On Wed, Oct 21, 2020 at 7:42 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Wed, Oct 21, 2020 at 1:38 PM Peter Smith <smithpb2250@gmail.com> wrote:

The PG docs for PREPARE TRANSACTION [1] don't say anything about an
empty (zero length) transaction-id.
e.g. PREPARE TRANSACTION '';
[1] https://www.postgresql.org/docs/current/sql-prepare-transaction.html

~

Meanwhile, during testing I found the 2PC prepare hangs when an empty
id is used.

Can you please take an example to explain what you are trying to say?

I was referring to an empty (zero length) transaction ID, not an empty
transaction.

The example was already given as PREPARE TRANSACTION '';

A longer example from my regress test is shown below. Using 2PC
pub/sub this will currently hang:

# --------------------
# Test using empty GID
# --------------------
# check that 2PC gets replicated to subscriber
$node_publisher->safe_psql('postgres',
"BEGIN;INSERT INTO tab_full VALUES (51);PREPARE TRANSACTION '';");
$node_publisher->poll_query_until('postgres', $caughtup_query)
or die "Timed out while waiting for subscriber to catch up";
# check that transaction is in prepared state on subscriber
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM
pg_prepared_xacts where gid = '';");
is($result, qq(1), 'transaction is prepared on subscriber');
# ROLLBACK
$node_publisher->safe_psql('postgres',
"ROLLBACK PREPARED '';");
# check that 2PC gets aborted on subscriber
$node_publisher->poll_query_until('postgres', $caughtup_query)
or die "Timed out while waiting for subscriber to catch up";
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM
pg_prepared_xacts where gid = '';");
is($result, qq(0), 'transaction is aborted on subscriber');

~

Is that something that should be made to work for 2PC pub/sub, or was
Postgres PREPARE TRANSACTION statement wrong to allow the user to
specify an empty transaction ID in the first place?

Kind Regards
Peter Smith.
Fujitsu Australia.

On Thu, Oct 22, 2020 at 4:58 AM Peter Smith <smithpb2250@gmail.com> wrote:

On Wed, Oct 21, 2020 at 7:42 PM Amit Kapila <amit.kapila16@gmail.com> wrote:

On Wed, Oct 21, 2020 at 1:38 PM Peter Smith <smithpb2250@gmail.com> wrote:

The PG docs for PREPARE TRANSACTION [1] don't say anything about an
empty (zero length) transaction-id.
e.g. PREPARE TRANSACTION '';
[1] https://www.postgresql.org/docs/current/sql-prepare-transaction.html

~

Meanwhile, during testing I found the 2PC prepare hangs when an empty
id is used.

Can you please take an example to explain what you are trying to say?

I was referring to an empty (zero length) transaction ID, not an empty
transaction.

oh, I got it confused with the system generated 32-bit TransactionId.
But now, I got what you were referring to.

The example was already given as PREPARE TRANSACTION '';

Is that something that should be made to work for 2PC pub/sub, or was
Postgres PREPARE TRANSACTION statement wrong to allow the user to
specify an empty transaction ID in the first place?

I don't see any problem with the empty transaction identifier used in
Prepare Transaction. This is just used as an identifier to uniquely
identify the transaction. If you try to use an empty string ('') more
than once for Prepare Transaction, it will give an error like below:
postgres=*# prepare transaction '';
ERROR: transaction identifier "" is already in use

So, I think this should work for pub/sub as well. Did you find out the
reason of hang?

--
With Regards,
Amit Kapila.

On Tue, Oct 20, 2020 at 3:15 PM Peter Smith <smithpb2250@gmail.com> wrote:

Hello Ajin.

I have gone through the v10 patches to verify if and how my previous
v6 review comments got addressed.

Some issues remain, and there are a few newly introduced ones.

Mostly it is all very minor stuff.

Please find my revised review comments below.

Kind Regards.
Peter Smith
Fujitsu Australia

---

V10 REVIEW COMMENTS FOLLOW

==========
Patch v10-0001, File: contrib/test_decoding/test_decoding.c
==========

COMMENT
Line 285
+ {
+ errno = 0;
+ data->check_xid_aborted = (TransactionId)
+ strtoul(strVal(elem->arg), NULL, 0);
+
+ if (!TransactionIdIsValid(data->check_xid_aborted))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("check-xid-aborted is not a valid xid: \"%s\"",
+ strVal(elem->arg))));
+ }

I think it is risky to assign strtoul directly to the
check_xid_aborted member because it makes some internal assumption
that the invalid transaction is the same as the error return from
strtoul.

Maybe better to do in 2 steps like below:

BEFORE
errno = 0;
data->check_xid_aborted = (TransactionId)strtoul(strVal(elem->arg), NULL, 0);

AFTER
long xid;
errno = 0;
xid = strtoul(strVal(elem->arg), NULL, 0);
if (xid == 0 || errno != 0)
data->check_xid_aborted = InvalidTransactionId;
else
data->check_xid_aborted =(TransactionId)xid;

---

Updated accordingly.

COMMENT
Line 430
+
+/* ABORT PREPARED callback */
+static void
+pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn)

Fix comment "ABORT PREPARED" --> "ROLLBACK PREPARED"

Updated accordingly.

==========
Patch v10-0001, File: doc/src/sgml/logicaldecoding.sgml
==========

COMMENT
Section 48.6.1
Says:
An output plugin may also define functions to support streaming of
large, in-progress transactions. The stream_start_cb, stream_stop_cb,
stream_abort_cb, stream_commit_cb and stream_change_cb are required,
while stream_message_cb, stream_prepare_cb, stream_commit_prepared_cb,
stream_rollback_prepared_cb and stream_truncate_cb are optional.

An output plugin may also define functions to support two-phase
commits, which are decoded on PREPARE TRANSACTION. The prepare_cb,
commit_prepared_cb and rollback_prepared_cb callbacks are required,
while filter_prepare_cb is optional.

-

But is that correct? It seems strange/inconsistent to say that the 2PC
callbacks are mandatory for the non-streaming, but that they are
optional for streaming.

Updated making all the 2PC callbacks mandatory.

---

COMMENT
48.6.4.5 "Transaction Prepare Callback"
48.6.4.6 "Transaction Commit Prepared Callback"
48.6.4.7 "Transaction Rollback Prepared Callback"

There seems some confusion about what is optional and what is
mandatory. e.g. Why are the non-stream 2PC callbacks mandatory but the
stream 2PC callbacks are not? And also there is some inconsistency
with what is said in the paragraph at the top of the page versus what
each of the callback sections says wrt optional/mandatory.

The sub-sections 49.6.4.5, 49.6.4.6, 49.6.4.7 say those callbacks are
optional which IIUC Amit said is incorrect. This is similar to the
previous review comment

---

Updated making all the 2PC callbacks mandatory.

COMMENT
Section 48.6.4.7 "Transaction Rollback Prepared Callback"

parameter "abort_lsn" probably should be "rollback_lsn"

---

COMMENT
Section 49.6.4.18. "Stream Rollback Prepared Callback"
Says:
The stream_rollback_prepared_cb callback is called to abort a
previously streamed transaction as part of a two-phase commit.

maybe should say "is called to rollback"

==========
Patch v10-0001, File: src/backend/replication/logical/logical.c
==========

COMMENT
Line 252
Says: We however enable two phase logical...

"two phase" --> "two-phase"

--

COMMENT
Line 885
Line 923
Says: If the plugin support 2 phase commits...

"support 2 phase" --> "supports two-phase" in the comment. Same issue
occurs twice.

---

COMMENT
Line 830
Line 868
Line 906
Says:
/* We're only supposed to call this when two-phase commits are supported */

There is an extra space between the "are" and "supported" in the comment.
Same issue occurs 3 times.

---

COMMENT
Line 1023
+ /*
+ * Skip if decoding of two-phase at PREPARE time is not enabled. In that
+ * case all two-phase transactions are considered filtered out and will be
+ * applied as regular transactions at COMMIT PREPARED.
+ */

Comment still is missing the word "transactions"
"Skip if decoding of two-phase at PREPARE time is not enabled."
-> "Skip if decoding of two-phase transactions at PREPARE time is not enabled.

Updated accordingly.

==========
Patch v10-0001, File: src/include/replication/reorderbuffer.h
==========

COMMENT
Line 459
/* abort prepared callback signature */
typedef void (*ReorderBufferRollbackPreparedCB) (
ReorderBuffer *rb,
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

There is no alignment consistency here for
ReorderBufferRollbackPreparedCB. Some function args are directly under
the "(" and some are on the same line. This function code is neither.

---

COMMENT
Line 638
@@ -431,6 +486,24 @@ typedef void (*ReorderBufferStreamAbortCB) (
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamPrepareCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr prepare_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamCommitPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr commit_lsn);
+
+/* prepare streamed transaction callback signature */
+typedef void (*ReorderBufferStreamRollbackPreparedCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr rollback_lsn);

There is no inconsistent alignment with the arguments (compare how
other functions are aligned)

See:
- for ReorderBufferStreamCommitPreparedCB
- for ReorderBufferStreamRollbackPreparedCB
- for ReorderBufferPrepareNeedSkip
- for ReorderBufferTxnIsPrepared
- for ReorderBufferPrepare

---

COMMENT
Line 489
Line 495
Line 501
/* prepare streamed transaction callback signature */

Same comment cut/paste 3 times?
- for ReorderBufferStreamPrepareCB
- for ReorderBufferStreamCommitPreparedCB
- for ReorderBufferStreamRollbackPreparedCB

---

COMMENT
Line 457
/* abort prepared callback signature */
typedef void (*ReorderBufferRollbackPreparedCB) (
ReorderBuffer *rb,
ReorderBufferTXN *txn,
XLogRecPtr abort_lsn);

"abort" --> "rollback" in the function comment.

---

COMMENT
Line 269
/* In case of 2PC we need to pass GID to output plugin */

"2PC" --> "two-phase commit"

Updated accordingly.

==========
Patch v10-0002, File: contrib/test_decoding/expected/two_phase.out (and .sql)
==========

COMMENT
General

It is a bit hard to see what are the main tests here are what are just
sub-parts of some test case.

e.g. It seems like the main tests are.

1. Test that decoding happens at PREPARE time
2. Test decoding of an aborted tx
3. Test a prepared tx which contains some DDL
4. Test decoding works while an uncommitted prepared tx with DDL exists
5. Test operations holding exclusive locks won't block decoding
6. Test savepoints and sub-transactions
7. Test "_nodecode" will defer the decoding until the commit time

Can the comments be made more obvious so it is easy to distinguish the
main tests from the steps of those tests?

---

COMMENT
Line 1
-- Test two-phased transactions, when two-phase-commit is enabled,
transactions are
-- decoded at PREPARE time rather than at COMMIT PREPARED time.

Some commas to be removed and this comment to be split into several sentences.

---

COMMENT
Line 19
-- should show nothing

Comment could be more informative. E.g. "Should show nothing because
the PREPARE has not happened yet"

---

COMMENT
Line 77

Looks like there is a missing comment about here that should say
something like "Show that the DDL does not appear in the decoding"

---

COMMENT
Line 160
-- test savepoints and sub-xacts as a result

The subsequent test is testing savepoints. But is it testing sub
transactions like the comment says?

Updated accordingly.

==========
Patch v10-0002, File: contrib/test_decoding/t/001_twophase.pl
==========

COMMENT
General

I think basically there are only 2 tests in this file.
1. to check that the concurrent abort works.
2. to check that the prepared tx can span a server shutdown/restart

But the tests comments do not make this clear at all.
e.g. All the "#" comments look equally important although most of them
are just steps of each test case.
Can the comments be better to distinguish the tests versus the steps
of each test?

Updated accordingly.

==========
Patch v10-0002, File: src/backend/replication/logical/decode.c
==========

COMMENT
Line 71
static void DecodeCommitPrepared(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf,
xl_xact_parsed_commit *parsed, TransactionId xid);
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
xl_xact_parsed_abort *parsed, TransactionId xid);
static void DecodeAbortPrepared(LogicalDecodingContext *ctx,
XLogRecordBuffer *buf,
xl_xact_parsed_abort *parsed, TransactionId xid);
static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
xl_xact_parsed_prepare * parsed);

The 2nd line or args are not aligned properly.
- for DecodeCommitPrepared
- for DecodeAbortPrepared
- for DecodePrepare

Updated accordingly.

==========
Patch v10-0002, File: src/backend/replication/logical/reorderbuffer.c
==========

COMMENT
There are some parts of the code where in my v6 review I had a doubt
about the mutually exclusive treatment of the "streaming" flag and the
"rbtxn_prepared(txn)" state.

Basically I did not see how some parts of the code are treating NOT
streaming as implying 2PC etc because it defies my understanding that
2PC can also work in streaming mode. Perhaps the "streaming" flag has
a different meaning to how I interpret it? Or perhaps some functions
are guarding higher up and can only be called under certain
conditions?

Anyway, this confusion manifests in several parts of the code, none of
which was changed after my v6 review.

Affected code includes the following:

CASE 1
Wherever the ReorderBufferTruncateTXN(...) "prepared" flag (third
parameter) is hardwired true/false, I think there must be some
preceding Assert to guarantee the prepared state condition holds true.
There can't be any room for doubts like "but what will it do for
streamed 2PC..."
Line 1805 - ReorderBufferTruncateTXN(rb, txn, true); // if rbtxn_prepared(txn)
Line 1941 - ReorderBufferTruncateTXN(rb, txn, false); // state ??
Line 2389 - ReorderBufferTruncateTXN(rb, txn, false); // if streaming
Line 2396 - ReorderBufferTruncateTXN(rb, txn, true); // if not
streaming and if rbtxm_prepared(txn)
Line 2459 - ReorderBufferTruncateTXN(rb, txn, true); // if not streaming

~

CASE 2
Wherever the "streaming" flag is tested I don't really understand how
NOT streaming can automatically imply 2PC.
Line 2330 - if (streaming) // what about if it is streaming AND 2PC at
the same time?
Line 2387 - if (streaming) // what about if it is streaming AND 2PC at
the same time?
Line 2449 - if (streaming) // what about if it is streaming AND 2PC at
the same time?

~

Case 1 and Case 2 above overlap a fair bit. I just listed them so they
all get checked again.

Even if the code is thought to be currently OK I do still think
something should be done like:
a) add some more substantial comments to explain WHY the combination
of streaming and 2PC is not valid in the context
b) the Asserts to be strengthened to 100% guarantee that the streaming
and prepared states really are exclusive (if indeed they are). For
this point I thought the following Assert condition could be better:
Assert(streaming || rbtxn_prepared(txn));
Assert(stream_started || rbtxn_prepared(txn));
because as it is you still are left wondering if both streaming AND
rbtxn_prepared(txn) can be possible at the same time...

---

Updated with more comments and a new Assert.

COMMENT
Line 2634
* Anyways, two-phase transactions do not contain any reorderbuffers.

"Anyways" --> "Anyway"

Updated.

==========
Patch v10-0003, File: src/backend/access/transam/twophase.c
==========

COMMENT
Line 557
@@ -548,6 +548,33 @@ MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
}

/*
+ * LookupGXact
+ * Check if the prepared transaction with the given GID is around
+ */
+bool
+LookupGXact(const char *gid)
+{
+ int i;
+ bool found = false;

The variable declarations (i and found) are not aligned.

Updated.

==========
Patch v10-0003, File: src/backend/replication/logical/proto.c
==========

COMMENT
Line 125
Line 205
Assert(strlen(txn->gid) > 0);

I suggested that the assertion should also check txn->gid is not NULL.
You replied "In this case txn->gid has to be non NULL".

But that is exactly what I said :-)
If it HAS to be non-NULL then why not just Assert that in code instead
of leaving the reader wondering?

"Assert(strlen(txn->gid) > 0);" --> "Assert(tdx->gid && strlen(txn->gid) > 0);"
Same occurs several times.

---

Updated checking that gid is non-NULL as zero strlen is actually a valid case.

COMMENT
Line 133
Line 213
if (rbtxn_commit_prepared(txn))
flags |= LOGICALREP_IS_COMMIT_PREPARED;
else if (rbtxn_rollback_prepared(txn))
flags |= LOGICALREP_IS_ROLLBACK_PREPARED;
else
flags |= LOGICALREP_IS_PREPARE;

Previously I wrote that the use of the bit flags on assignment in the
logicalrep_write_prepare was inconsistent with the way they are
treated when they are read. Really it should be using a direct
assignment instead of bit flags.

You said this is skipped anticipating a possible refactor. But IMO
this leaves the code in a half/half state. I think it is better to fix
it properly and if refactoring happens then deal with that at the
time.

The last comment I saw from Amit said to use my 1st proposal of direct
assignment instead of bit flag assignment.

(applies to both non-stream and stream functions)
- see logicalrep_write_prepare
- see logicalrep_write_stream_prepare

Updated accordingly.

==========
Patch v10-0003, File: src/backend/replication/pgoutput/pgoutput.c
==========

COMMENT
Line 429
/*
* PREPARE callback
*/
static void
pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn)
The function comment looks wrong.
Shouldn't this comment say be "ROLLBACK PREPARED callback"?

==========
Patch v10-0003, File: src/include/replication/logicalproto.h
==========

Line 115
#define PrepareFlagsAreValid(flags) \
((flags == LOGICALREP_IS_PREPARE) || \
(flags == LOGICALREP_IS_COMMIT_PREPARED) || \
(flags == LOGICALREP_IS_ROLLBACK_PREPARED))

Would be safer if all the references to flags are in parentheses
e.g. "flags" --> "(flags)"

Updated accordingly.

Amit,
I have also modified the stream callback APIs to not include
stream_commit_prpeared and stream_rollback_prepared, instead use the
non-stream APIs for the same functionality.
I have also updated the test_decoding and pgoutput plugins accordingly.

regards,
Ajin Cherian
Fujitsu Australia

On Fri, Oct 23, 2020 at 3:41 PM Ajin Cherian <itsajin@gmail.com> wrote:

Amit,
I have also modified the stream callback APIs to not include
stream_commit_prpeared and stream_rollback_prepared, instead use the
non-stream APIs for the same functionality.
I have also updated the test_decoding and pgoutput plugins accordingly.

Thanks, I think you forgot to address one of my comments in the
previous email[1]/messages/by-id/CAA4eK1JzRvUX2XLEKo2f74Vjecnt6wq-kkk1OiyMJ5XjJN+GvQ@mail.gmail.com (See "One minor comment .."). You have not even
responded to it.

[1]: /messages/by-id/CAA4eK1JzRvUX2XLEKo2f74Vjecnt6wq-kkk1OiyMJ5XjJN+GvQ@mail.gmail.com

--
With Regards,
Amit Kapila.

Hi Ajin.

I've addressed your review comments (details below) and bumped the
patch set to v12 attached.

I also added more test cases.

On Tue, Oct 20, 2020 at 10:02 PM Ajin Cherian <itsajin@gmail.com> wrote:

Thanks for your patch. Some comments for your patch:

Comments:

src/backend/replication/logical/worker.c
@@ -888,6 +888,319 @@ apply_handle_prepare(StringInfo s)
+ /*
+ * FIXME - Following condition was in apply_handle_prepare_txn except
I found  it was ALWAYS IsTransactionState() == false
+ * The synchronization worker runs in single transaction. *
+ if (IsTransactionState() && !am_tablesync_worker())
+ */
+ if (!am_tablesync_worker())

Comment: I dont think a tablesync worker will use streaming, none of
the other stream APIs check this, this might not be relevant for
stream_prepare either.

Updated

+ /*
+ * ==================================================================================================
+ * The following chunk of code is largely cut/paste from the existing
apply_handle_prepare_commit_txn

Comment: Here, I think you meant apply_handle_stream_commit.

Updated.

Also
rather than duplicating this chunk of code, you could put it in a new
function.

Code is refactored to share a common function for the spool file processing.

+ else
+ {
+ /* Process any invalidation messages that might have accumulated. */
+ AcceptInvalidationMessages();
+ maybe_reread_subscription();
+ }

Comment: This else block might not be necessary as a tablesync worker
will not initiate the streaming APIs.

Updated

~

Kind Regards,
Peter Smith
Fujitsu Australia

Hi Ajin.

I checked the to see how my previous review comments (of v10) were
addressed by the latest patches (currently v12)

There are a couple of remaining items.

---

====================
v12-0001. File: doc/src/sgml/logicaldecoding.sgml
====================

COMMENT
Section 49.6.1
Says:
An output plugin may also define functions to support streaming of
large, in-progress transactions. The stream_start_cb, stream_stop_cb,
stream_abort_cb, stream_commit_cb, stream_change_cb, and
stream_prepare_cb are required, while stream_message_cb and
stream_truncate_cb are optional.

An output plugin may also define functions to support two-phase
commits, which are decoded on PREPARE TRANSACTION. The prepare_cb,
commit_prepared_cb and rollback_prepared_cb callbacks are required,
while filter_prepare_cb is optional.
~
I was not sure how the paragraphs are organised. e.g. 1st seems to be
about streams and 2nd seems to be about two-phase commit. But they are
not mutually exclusive, so I guess I thought it was odd that
stream_prepare_cb was not mentioned in the 2nd paragraph.

Or maybe it is OK as-is?

====================
v12-0002. File: contrib/test_decoding/expected/two_phase.out
====================

COMMENT
Line 26
PREPARE TRANSACTION 'test_prepared#1';
--
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,
NULL, 'two-phase-commit', '1', 'include-xids', '0',
'skip-empty-xacts', '1');
~
Seems like a missing comment to explain the expectation of that select.

---

COMMENT
Line 80
-- The insert should show the newly altered column.
~
Do you also need to mention something about the DDL not being present
in the decoding?

====================
v12-0002. File: src/backend/replication/logical/reorderbuffer.c
====================

COMMENT
Line 1807
/* Here we are streaming and part of the PREPARE of a two-phase commit
* The full cleanup will happen as part of the COMMIT PREPAREDs, so now
* just truncate txn by removing changes and tuple_cids
*/
~
Something seems strange about the first sentence of that comment

---

COMMENT
Line 1944
/* Discard the changes that we just streamed.
* This can only be called if streaming and not part of a PREPARE in
* a two-phase commit, so set prepared flag as false.
*/
~
I thought since this comment that is asserting various things, that
should also actually be written as code Assert.

---

COMMENT
Line 2401
/*
* We are here due to one of the 3 scenarios:
* 1. As part of streaming in-progress transactions
* 2. Prepare of a two-phase commit
* 3. Commit of a transaction.
*
* If we are streaming the in-progress transaction then discard the
* changes that we just streamed, and mark the transactions as
* streamed (if they contained changes), set prepared flag as false.
* If part of a prepare of a two-phase commit set the prepared flag
* as true so that we can discard changes and cleanup tuplecids.
* Otherwise, remove all the
* changes and deallocate the ReorderBufferTXN.
*/
~
The above comment is beyond my understanding. Anything you could do to
simplify it would be good.

For example, when viewing this function in isolation I have never
understood why the streaming flag and rbtxn_prepared(txn) flag are not
possible to be set at the same time?

Perhaps the code is relying on just internal knowledge of how this
helper function gets called? And if it is just that, then IMO there
really should be some Asserts in the code to give more assurance about
that. (Or maybe use completely different flags to represent those 3
scenarios instead of bending the meanings of the existing flags)

====================
v12-0003. File: src/backend/access/transam/twophase.c
====================

COMMENT
Line 557
@@ -548,6 +548,33 @@ MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
}

 /*
+ * LookupGXact
+ * Check if the prepared transaction with the given GID is around
+ */
+bool
+LookupGXact(const char *gid)
+{
+ int i;
+ bool found = false;
~
Alignment of the variable declarations in LookupGXact function

---

Kind Regards,
Peter Smith.
Fujitsu Australia

On Mon, Oct 26, 2020 at 6:49 PM Peter Smith <smithpb2250@gmail.com> wrote:

Hi Ajin.

I checked the to see how my previous review comments (of v10) were
addressed by the latest patches (currently v12)

There are a couple of remaining items.

---

====================
v12-0001. File: doc/src/sgml/logicaldecoding.sgml
====================

COMMENT
Section 49.6.1
Says:
An output plugin may also define functions to support streaming of
large, in-progress transactions. The stream_start_cb, stream_stop_cb,
stream_abort_cb, stream_commit_cb, stream_change_cb, and
stream_prepare_cb are required, while stream_message_cb and
stream_truncate_cb are optional.

An output plugin may also define functions to support two-phase
commits, which are decoded on PREPARE TRANSACTION. The prepare_cb,
commit_prepared_cb and rollback_prepared_cb callbacks are required,
while filter_prepare_cb is optional.
~
I was not sure how the paragraphs are organised. e.g. 1st seems to be
about streams and 2nd seems to be about two-phase commit. But they are
not mutually exclusive, so I guess I thought it was odd that
stream_prepare_cb was not mentioned in the 2nd paragraph.

Or maybe it is OK as-is?

I've added stream_prepare_cb to the 2nd paragraph as well.

====================
v12-0002. File: contrib/test_decoding/expected/two_phase.out
====================

COMMENT
Line 26
PREPARE TRANSACTION 'test_prepared#1';
--
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,
NULL, 'two-phase-commit', '1', 'include-xids', '0',
'skip-empty-xacts', '1');
~
Seems like a missing comment to explain the expectation of that select.

---

Updated.

COMMENT
Line 80
-- The insert should show the newly altered column.
~
Do you also need to mention something about the DDL not being present
in the decoding?

Updated.

====================
v12-0002. File: src/backend/replication/logical/reorderbuffer.c
====================

COMMENT
Line 1807
/* Here we are streaming and part of the PREPARE of a two-phase commit
* The full cleanup will happen as part of the COMMIT PREPAREDs, so now
* just truncate txn by removing changes and tuple_cids
*/
~
Something seems strange about the first sentence of that comment

---

COMMENT
Line 1944
/* Discard the changes that we just streamed.
* This can only be called if streaming and not part of a PREPARE in
* a two-phase commit, so set prepared flag as false.
*/
~
I thought since this comment that is asserting various things, that
should also actually be written as code Assert.

---

Added an assert.

COMMENT
Line 2401
/*
* We are here due to one of the 3 scenarios:
* 1. As part of streaming in-progress transactions
* 2. Prepare of a two-phase commit
* 3. Commit of a transaction.
*
* If we are streaming the in-progress transaction then discard the
* changes that we just streamed, and mark the transactions as
* streamed (if they contained changes), set prepared flag as false.
* If part of a prepare of a two-phase commit set the prepared flag
* as true so that we can discard changes and cleanup tuplecids.
* Otherwise, remove all the
* changes and deallocate the ReorderBufferTXN.
*/
~
The above comment is beyond my understanding. Anything you could do to
simplify it would be good.

For example, when viewing this function in isolation I have never
understood why the streaming flag and rbtxn_prepared(txn) flag are not
possible to be set at the same time?

Perhaps the code is relying on just internal knowledge of how this
helper function gets called? And if it is just that, then IMO there
really should be some Asserts in the code to give more assurance about
that. (Or maybe use completely different flags to represent those 3
scenarios instead of bending the meanings of the existing flags)

Left this for now, probably re-look at this at a later review.
But just to explain; this function is what does the main decoding of
changes of a transaction.
At what point this decoding happens is what this feature and the
streaming in-progress feature is about. As of PG13, this decoding only
happens at commit time. With the streaming of in-progress txn feature,
this began to happen (if streaming enabled) at the time when the
memory limit for decoding transactions was crossed. This 2PC feature
is supporting decoding at the time of a PREPARE transaction.
Now, if streaming is enabled and streaming has started as a result of
crossing the memory threshold, then there is no need to
again begin streaming at a PREPARE transaction as the transaction that
is being prepared has already been streamed. Which is why this
function will not be called when a streaming transaction is prepared
as part of a two-phase commit.

====================
v12-0003. File: src/backend/access/transam/twophase.c
====================

COMMENT
Line 557
@@ -548,6 +548,33 @@ MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
}

/*
+ * LookupGXact
+ * Check if the prepared transaction with the given GID is around
+ */
+bool
+LookupGXact(const char *gid)
+{
+ int i;
+ bool found = false;
~
Alignment of the variable declarations in LookupGXact function

---

Updated.

Amit, I have also updated your comment about removing function
declaration from commit 1 and I've added it to commit 2. Also removed
whitespace errors.

regards,
Ajin Cherian
Fujitsu Australia

FYI - Please find attached code coverage reports which I generated
(based on the v12 patches) after running the following tests:

1. cd contrib/test_decoding; make check

2. cd src/test/subscriber; make check

Kind Regards,
Peter Smith.
Fujitsu Australia

Show quoted text

On Tue, Oct 27, 2020 at 8:55 PM Ajin Cherian <itsajin@gmail.com> wrote:

On Mon, Oct 26, 2020 at 6:49 PM Peter Smith <smithpb2250@gmail.com> wrote:

Hi Ajin.

I checked the to see how my previous review comments (of v10) were
addressed by the latest patches (currently v12)

There are a couple of remaining items.

---

====================
v12-0001. File: doc/src/sgml/logicaldecoding.sgml
====================

COMMENT
Section 49.6.1
Says:
An output plugin may also define functions to support streaming of
large, in-progress transactions. The stream_start_cb, stream_stop_cb,
stream_abort_cb, stream_commit_cb, stream_change_cb, and
stream_prepare_cb are required, while stream_message_cb and
stream_truncate_cb are optional.

An output plugin may also define functions to support two-phase
commits, which are decoded on PREPARE TRANSACTION. The prepare_cb,
commit_prepared_cb and rollback_prepared_cb callbacks are required,
while filter_prepare_cb is optional.
~
I was not sure how the paragraphs are organised. e.g. 1st seems to be
about streams and 2nd seems to be about two-phase commit. But they are
not mutually exclusive, so I guess I thought it was odd that
stream_prepare_cb was not mentioned in the 2nd paragraph.

Or maybe it is OK as-is?

I've added stream_prepare_cb to the 2nd paragraph as well.

====================
v12-0002. File: contrib/test_decoding/expected/two_phase.out
====================

COMMENT
Line 26
PREPARE TRANSACTION 'test_prepared#1';
--
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,
NULL, 'two-phase-commit', '1', 'include-xids', '0',
'skip-empty-xacts', '1');
~
Seems like a missing comment to explain the expectation of that select.

---

Updated.

COMMENT
Line 80
-- The insert should show the newly altered column.
~
Do you also need to mention something about the DDL not being present
in the decoding?

Updated.

====================
v12-0002. File: src/backend/replication/logical/reorderbuffer.c
====================

COMMENT
Line 1807
/* Here we are streaming and part of the PREPARE of a two-phase commit
* The full cleanup will happen as part of the COMMIT PREPAREDs, so now
* just truncate txn by removing changes and tuple_cids
*/
~
Something seems strange about the first sentence of that comment

---

COMMENT
Line 1944
/* Discard the changes that we just streamed.
* This can only be called if streaming and not part of a PREPARE in
* a two-phase commit, so set prepared flag as false.
*/
~
I thought since this comment that is asserting various things, that
should also actually be written as code Assert.

---

Added an assert.

COMMENT
Line 2401
/*
* We are here due to one of the 3 scenarios:
* 1. As part of streaming in-progress transactions
* 2. Prepare of a two-phase commit
* 3. Commit of a transaction.
*
* If we are streaming the in-progress transaction then discard the
* changes that we just streamed, and mark the transactions as
* streamed (if they contained changes), set prepared flag as false.
* If part of a prepare of a two-phase commit set the prepared flag
* as true so that we can discard changes and cleanup tuplecids.
* Otherwise, remove all the
* changes and deallocate the ReorderBufferTXN.
*/
~
The above comment is beyond my understanding. Anything you could do to
simplify it would be good.

For example, when viewing this function in isolation I have never
understood why the streaming flag and rbtxn_prepared(txn) flag are not
possible to be set at the same time?

Perhaps the code is relying on just internal knowledge of how this
helper function gets called? And if it is just that, then IMO there
really should be some Asserts in the code to give more assurance about
that. (Or maybe use completely different flags to represent those 3
scenarios instead of bending the meanings of the existing flags)

Left this for now, probably re-look at this at a later review.
But just to explain; this function is what does the main decoding of
changes of a transaction.
At what point this decoding happens is what this feature and the
streaming in-progress feature is about. As of PG13, this decoding only
happens at commit time. With the streaming of in-progress txn feature,
this began to happen (if streaming enabled) at the time when the
memory limit for decoding transactions was crossed. This 2PC feature
is supporting decoding at the time of a PREPARE transaction.
Now, if streaming is enabled and streaming has started as a result of
crossing the memory threshold, then there is no need to
again begin streaming at a PREPARE transaction as the transaction that
is being prepared has already been streamed. Which is why this
function will not be called when a streaming transaction is prepared
as part of a two-phase commit.

====================
v12-0003. File: src/backend/access/transam/twophase.c
====================

COMMENT
Line 557
@@ -548,6 +548,33 @@ MarkAsPrepared(GlobalTransaction gxact, bool lock_held)
}

/*
+ * LookupGXact
+ * Check if the prepared transaction with the given GID is around
+ */
+bool
+LookupGXact(const char *gid)
+{
+ int i;
+ bool found = false;
~
Alignment of the variable declarations in LookupGXact function

---

Updated.

Amit, I have also updated your comment about removing function
declaration from commit 1 and I've added it to commit 2. Also removed
whitespace errors.

regards,
Ajin Cherian
Fujitsu Australia

Hi Ajin.

I have re-checked the v13 patches for how my remaining review comments
have been addressed.

On Tue, Oct 27, 2020 at 8:55 PM Ajin Cherian <itsajin@gmail.com> wrote:

====================
v12-0002. File: src/backend/replication/logical/reorderbuffer.c
====================

COMMENT
Line 2401
/*
* We are here due to one of the 3 scenarios:
* 1. As part of streaming in-progress transactions
* 2. Prepare of a two-phase commit
* 3. Commit of a transaction.
*
* If we are streaming the in-progress transaction then discard the
* changes that we just streamed, and mark the transactions as
* streamed (if they contained changes), set prepared flag as false.
* If part of a prepare of a two-phase commit set the prepared flag
* as true so that we can discard changes and cleanup tuplecids.
* Otherwise, remove all the
* changes and deallocate the ReorderBufferTXN.
*/
~
The above comment is beyond my understanding. Anything you could do to
simplify it would be good.

For example, when viewing this function in isolation I have never
understood why the streaming flag and rbtxn_prepared(txn) flag are not
possible to be set at the same time?

Perhaps the code is relying on just internal knowledge of how this
helper function gets called? And if it is just that, then IMO there
really should be some Asserts in the code to give more assurance about
that. (Or maybe use completely different flags to represent those 3
scenarios instead of bending the meanings of the existing flags)

Left this for now, probably re-look at this at a later review.
But just to explain; this function is what does the main decoding of
changes of a transaction.
At what point this decoding happens is what this feature and the
streaming in-progress feature is about. As of PG13, this decoding only
happens at commit time. With the streaming of in-progress txn feature,
this began to happen (if streaming enabled) at the time when the
memory limit for decoding transactions was crossed. This 2PC feature
is supporting decoding at the time of a PREPARE transaction.
Now, if streaming is enabled and streaming has started as a result of
crossing the memory threshold, then there is no need to
again begin streaming at a PREPARE transaction as the transaction that
is being prepared has already been streamed. Which is why this
function will not be called when a streaming transaction is prepared
as part of a two-phase commit.

AFAIK the last remaining issue now is only about the complexity of the
aforementioned code/comment. If you want to defer changing that until
we can come up with something better, then that is OK by me.

Apart from that I have no other pending review comments at this time.

Kind Regards,
Peter Smith.
Fujitsu Australia

Hi Ajin.

Looking at v13 patches again I found a couple more review comments:

===

(1) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_write_prepare
+ if (rbtxn_commit_prepared(txn))
+ flags = LOGICALREP_IS_COMMIT_PREPARED;
+ else if (rbtxn_rollback_prepared(txn))
+ flags = LOGICALREP_IS_ROLLBACK_PREPARED;
+ else
+ flags = LOGICALREP_IS_PREPARE;
+
+ /* Make sure exactly one of the expected flags is set. */
+ if (!PrepareFlagsAreValid(flags))
+ elog(ERROR, "unrecognized flags %u in prepare message", flags);

Since those flags are directly assigned, I think the subsequent if
(!PrepareFlagsAreValid(flags)) check is redundant.

===

(2) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_write_stream_prepare
+/*
+ * Write STREAM PREPARE to the output stream.
+ * (For stream PREPARE, stream COMMIT PREPARED, stream ROLLBACK PREPARED)
+ */

I think the function comment is outdated because IIUC the stream
COMMIT PREPARED and stream ROLLBACK PREPARED are not being handled by
the function logicalrep_write_prepare. SInce this approach seems
counter-intuitive there needs to be an improved function comment to
explain what is going on.

===

(3) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_read_stream_prepare
+/*
+ * Read STREAM PREPARE from the output stream.
+ * (For stream PREPARE, stream COMMIT PREPARED, stream ROLLBACK PREPARED)
+ */

This is the same as the previous review comment. The function comment
needs to explain the new handling for stream COMMIT PREPARED and
stream ROLLBACK PREPARED.

===

(4) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_read_stream_prepare
+TransactionId
+logicalrep_read_stream_prepare(StringInfo in, LogicalRepPrepareData
*prepare_data)
+{
+ TransactionId xid;
+ uint8 flags;
+
+ xid = pq_getmsgint(in, 4);
+
+ /* read flags */
+ flags = pq_getmsgbyte(in);
+
+ if (!PrepareFlagsAreValid(flags))
+ elog(ERROR, "unrecognized flags %u in prepare message", flags);

I think the logicalrep_write_stream_prepare now can only assign the
flags = LOGICALREP_IS_PREPARE. So that means the check here for bad
flags should be changed to match.
BEFORE: if (!PrepareFlagsAreValid(flags))
AFTER: if (flags != LOGICALREP_IS_PREPARE)

===

(5) COMMENT
General
Since the COMMENTs (2), (3) and (4) are all caused by the refactoring
that was done for removal of the commit/rollback stream callbacks. I
do wonder if it might have worked out better just to leave the
logicalrep_read/write_stream_prepared as it was instead of mixing up
stream/no-stream handling. A check for stream/no-stream could possibly
have been made higher up.

For example:
static void
pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn)
{
OutputPluginUpdateProgress(ctx);

OutputPluginPrepareWrite(ctx, true);
if (ctx->streaming)
logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
else
logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
}

===

Kind Regards,
Peter Smith.
Fujitsu Australia

FYI - I have cross-checked all the v12 patch code changes against the
v12 code coverage resulting from running the patch tests

Those v12 code coverage results were posted in this thread previously [1]/messages/by-id/CAHut+Pt6zB-YffCrMo7+ZOKn7C2yXkNYnuQTdbStEJJJXZZXaw@mail.gmail.com.

The purpose of this study was to identify if / where there are any
gaps in the testing of this patch - e.g is there some code not
currently getting executed?

I found in general there seems quite high coverage of the normal (not
error) code path,but there are a couple of current gaps in the test
coverage.

For details please find attached the study results. (MS Excel file)

===

[1]: /messages/by-id/CAHut+Pt6zB-YffCrMo7+ZOKn7C2yXkNYnuQTdbStEJJJXZZXaw@mail.gmail.com

Kind Regards,
Peter Smith.
Fujitsu Australia

On Thu, Oct 29, 2020 at 11:48 AM Peter Smith <smithpb2250@gmail.com> wrote:

Hi Ajin.

Looking at v13 patches again I found a couple more review comments:

===

(1) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_write_prepare
+ if (rbtxn_commit_prepared(txn))
+ flags = LOGICALREP_IS_COMMIT_PREPARED;
+ else if (rbtxn_rollback_prepared(txn))
+ flags = LOGICALREP_IS_ROLLBACK_PREPARED;
+ else
+ flags = LOGICALREP_IS_PREPARE;
+
+ /* Make sure exactly one of the expected flags is set. */
+ if (!PrepareFlagsAreValid(flags))
+ elog(ERROR, "unrecognized flags %u in prepare message", flags);

Since those flags are directly assigned, I think the subsequent if
(!PrepareFlagsAreValid(flags)) check is redundant.

===

Updated this.

(2) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_write_stream_prepare
+/*
+ * Write STREAM PREPARE to the output stream.
+ * (For stream PREPARE, stream COMMIT PREPARED, stream ROLLBACK PREPARED)
+ */

I think the function comment is outdated because IIUC the stream
COMMIT PREPARED and stream ROLLBACK PREPARED are not being handled by
the function logicalrep_write_prepare. SInce this approach seems
counter-intuitive there needs to be an improved function comment to
explain what is going on.

===

(3) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_read_stream_prepare
+/*
+ * Read STREAM PREPARE from the output stream.
+ * (For stream PREPARE, stream COMMIT PREPARED, stream ROLLBACK PREPARED)
+ */

This is the same as the previous review comment. The function comment
needs to explain the new handling for stream COMMIT PREPARED and
stream ROLLBACK PREPARED.

===

I think that these functions only writing/reading STREAM PREPARE as
the name suggests is more intuitive. Maybe the usage of flags is more
confusing. More below.

(4) COMMENT
File: src/backend/replication/logical/proto.c
Function: logicalrep_read_stream_prepare
+TransactionId
+logicalrep_read_stream_prepare(StringInfo in, LogicalRepPrepareData
*prepare_data)
+{
+ TransactionId xid;
+ uint8 flags;
+
+ xid = pq_getmsgint(in, 4);
+
+ /* read flags */
+ flags = pq_getmsgbyte(in);
+
+ if (!PrepareFlagsAreValid(flags))
+ elog(ERROR, "unrecognized flags %u in prepare message", flags);

I think the logicalrep_write_stream_prepare now can only assign the
flags = LOGICALREP_IS_PREPARE. So that means the check here for bad
flags should be changed to match.
BEFORE: if (!PrepareFlagsAreValid(flags))
AFTER: if (flags != LOGICALREP_IS_PREPARE)

===

Updated.

(5) COMMENT
General
Since the COMMENTs (2), (3) and (4) are all caused by the refactoring
that was done for removal of the commit/rollback stream callbacks. I
do wonder if it might have worked out better just to leave the
logicalrep_read/write_stream_prepared as it was instead of mixing up
stream/no-stream handling. A check for stream/no-stream could possibly
have been made higher up.

For example:
static void
pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
XLogRecPtr prepare_lsn)
{
OutputPluginUpdateProgress(ctx);

OutputPluginPrepareWrite(ctx, true);
if (ctx->streaming)
logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
else
logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
}

===

I think I'll keep this as such for now. Amit was talking about
considering removal of flags to overload PREPARE with COMMIT PREPARED
and ROLLBACK PREPARED. Separate functions for each.
Will wait if Amit thinks that is the way to go.

I've also added a new test case for test_decoding for streaming 2PC.
Removed function ReorderBufferTxnIsPrepared as it was never called
thanks to Peter's coverage report. And added stream_prepare to the
list of callbacks that would
enable two-phase commits.

regards,
Ajin Cherian
Fujitsu Australia