Rename RBTXN_PREPARE to RBTXN_IS_PREPARE for better clarification.

RBTXN_PREPARE flag and rbtxn_prepared macro could be misinterpreted as
either indicating the transaction type (e.g. a prepared transaction or
a normal transaction) or its currentstate (e.g. skipped or its prepare
message is sent), especially after commit 072ee847ad4 introduced the
RBTXN_SENT_PREPARE flag and the rbtxn_sent_prepare macro.

The RBTXN_PREPARE flag (and its corresponding macro) have been renamed
to RBTXN_IS_PREPARE to explicitly indicate the transaction
type. Therefore, this commit also adds the RBTXN_IS_PREPARE flag to
the transaction that is a prepared transaction and has been skipped,
which previously had only the RBTXN_SKIPPED_PREPARE flag.

Reviewed-by: Amit Kapila, Peter Smith
Discussion: https://postgr.es/m/CAA4eK1KgNmBsG%3D155E7QQ6TX9RoWnM4z5Z20SvsbwxSe_QXYsg%40mail.gmail.com

diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index dc72b7c8f77a9c85eed0051ff215cf3b7177c4cb..1a352b542dc5670d2838149c5e51e1b042b9254c 100644 (file)
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -164,7 +164,7 @@ logicalrep_write_prepare_common(StringInfo out, LogicalRepMsgType type,
         * which case we expect to have a valid GID.
         */
        Assert(txn->gid != NULL);
-       Assert(rbtxn_prepared(txn));
+       Assert(rbtxn_is_prepared(txn));
        Assert(TransactionIdIsValid(txn->xid));
 
        /* send the flags field */

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index ed5a2946dc1ae8c5995c325724653bd25846643e..b42f4002ba84589bc4efd6d58fe06d6dcba0735f 100644 (file)
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -1793,7 +1793,7 @@ ReorderBufferCheckAndTruncateAbortedTXN(ReorderBuffer *rb, ReorderBufferTXN *txn
         * and the toast reconstruction data. The full cleanup will happen as part
         * of decoding ABORT record of this transaction.
         */
-       ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn));
+       ReorderBufferTruncateTXN(rb, txn, rbtxn_is_prepared(txn));
        ReorderBufferToastReset(rb, txn);
 
        /* All changes should be discarded */
@@ -1968,7 +1968,7 @@ ReorderBufferStreamCommit(ReorderBuffer *rb, ReorderBufferTXN *txn)
 
        ReorderBufferStreamTXN(rb, txn);
 
-       if (rbtxn_prepared(txn))
+       if (rbtxn_is_prepared(txn))
        {
                /*
                 * Note, we send stream prepare even if a concurrent abort is
@@ -2150,7 +2150,7 @@ ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                                          ReorderBufferChange *specinsert)
 {
        /* Discard the changes that we just streamed */
-       ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn));
+       ReorderBufferTruncateTXN(rb, txn, rbtxn_is_prepared(txn));
 
        /* Free all resources allocated for toast reconstruction */
        ReorderBufferToastReset(rb, txn);
@@ -2238,7 +2238,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                 */
                if (!streaming)
                {
-                       if (rbtxn_prepared(txn))
+                       if (rbtxn_is_prepared(txn))
                                rb->begin_prepare(rb, txn);
                        else
                                rb->begin(rb, txn);
@@ -2280,7 +2280,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                         * required for the cases when we decode the changes before the
                         * COMMIT record is processed.
                         */
-                       if (streaming || rbtxn_prepared(change->txn))
+                       if (streaming || rbtxn_is_prepared(change->txn))
                        {
                                curtxn = change->txn;
                                SetupCheckXidLive(curtxn->xid);
@@ -2625,7 +2625,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                         * Call either PREPARE (for two-phase transactions) or COMMIT (for
                         * regular ones).
                         */
-                       if (rbtxn_prepared(txn))
+                       if (rbtxn_is_prepared(txn))
                        {
                                Assert(!rbtxn_sent_prepare(txn));
                                rb->prepare(rb, txn, commit_lsn);
@@ -2680,12 +2680,12 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                 * For 4, as the entire txn has been decoded, we can fully clean up
                 * the TXN reorder buffer.
                 */
-               if (streaming || rbtxn_prepared(txn))
+               if (streaming || rbtxn_is_prepared(txn))
                {
                        if (streaming)
                                ReorderBufferMaybeMarkTXNStreamed(rb, txn);
 
-                       ReorderBufferTruncateTXN(rb, txn, rbtxn_prepared(txn));
+                       ReorderBufferTruncateTXN(rb, txn, rbtxn_is_prepared(txn));
                        /* Reset the CheckXidAlive */
                        CheckXidAlive = InvalidTransactionId;
                }
@@ -2729,7 +2729,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
                 * during a two-phase commit.
                 */
                if (errdata->sqlerrcode == ERRCODE_TRANSACTION_ROLLBACK &&
-                       (stream_started || rbtxn_prepared(txn)))
+                       (stream_started || rbtxn_is_prepared(txn)))
                {
                        /* curtxn must be set for streaming or prepared transactions */
                        Assert(curtxn);
@@ -2816,7 +2816,7 @@ ReorderBufferReplay(ReorderBufferTXN *txn,
                 * Removing this txn before a commit might result in the computation
                 * of an incorrect restart_lsn. See SnapBuildProcessRunningXacts.
                 */
-               if (!rbtxn_prepared(txn))
+               if (!rbtxn_is_prepared(txn))
                        ReorderBufferCleanupTXN(rb, txn);
                return;
        }
@@ -2853,7 +2853,8 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
 }
 
 /*
- * Record the prepare information for a transaction.
+ * Record the prepare information for a transaction. Also, mark the transaction
+ * as a prepared transaction.
  */
 bool
 ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid,
@@ -2879,6 +2880,10 @@ ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid,
        txn->origin_id = origin_id;
        txn->origin_lsn = origin_lsn;
 
+       /* Mark this transaction as a prepared transaction */
+       Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) == 0);
+       txn->txn_flags |= RBTXN_IS_PREPARED;
+
        return true;
 }
 
@@ -2894,6 +2899,8 @@ ReorderBufferSkipPrepare(ReorderBuffer *rb, TransactionId xid)
        if (txn == NULL)
                return;
 
+       /* txn must have been marked as a prepared transaction */
+       Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) == RBTXN_IS_PREPARED);
        txn->txn_flags |= RBTXN_SKIPPED_PREPARE;
 }
 
@@ -2915,12 +2922,16 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
        if (txn == NULL)
                return;
 
-       txn->txn_flags |= RBTXN_PREPARE;
-       txn->gid = pstrdup(gid);
-
-       /* The prepare info must have been updated in txn by now. */
+       /*
+        * txn must have been marked as a prepared transaction and must have
+        * neither been skipped nor sent a prepare. Also, the prepare info must
+        * have been updated in it by now.
+        */
+       Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) == RBTXN_IS_PREPARED);
        Assert(txn->final_lsn != InvalidXLogRecPtr);
 
+       txn->gid = pstrdup(gid);
+
        ReorderBufferReplay(txn, rb, xid, txn->final_lsn, txn->end_lsn,
                                                txn->xact_time.prepare_time, txn->origin_id, txn->origin_lsn);
 
@@ -2976,12 +2987,13 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
         */
        if ((txn->final_lsn < two_phase_at) && is_commit)
        {
-               txn->txn_flags |= RBTXN_PREPARE;
-
                /*
-                * The prepare info must have been updated in txn even if we skip
-                * prepare.
+                * txn must have been marked as a prepared transaction and skipped but
+                * not sent a prepare. Also, the prepare info must have been updated
+                * in txn even if we skip prepare.
                 */
+               Assert((txn->txn_flags & RBTXN_PREPARE_STATUS_MASK) ==
+                          (RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE));
                Assert(txn->final_lsn != InvalidXLogRecPtr);
 
                /*

diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index bbedd3de31831e9590610c2f2cd7fb6b64b72825..05687fd75e5b4c1065c9baf9295b86b1143bb9cb 100644 (file)
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -761,7 +761,7 @@ SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
                 * We don't need to add snapshot to prepared transactions as they
                 * should not see the new catalog contents.
                 */
-               if (rbtxn_prepared(txn) || rbtxn_skip_prepared(txn))
+               if (rbtxn_is_prepared(txn))
                        continue;
 
                elog(DEBUG2, "adding a new snapshot to %u at %X/%X",

diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 9d9ac2f08305a0ba88ea6874489c4ac080fff5ff..517a8e3634fd1e147d96de844aed32dc2de29cb8 100644 (file)
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -170,13 +170,15 @@ typedef struct ReorderBufferChange
 #define RBTXN_IS_SERIALIZED_CLEAR      0x0008
 #define RBTXN_IS_STREAMED              0x0010
 #define RBTXN_HAS_PARTIAL_CHANGE       0x0020
-#define RBTXN_PREPARE                  0x0040
+#define RBTXN_IS_PREPARED                      0x0040
 #define RBTXN_SKIPPED_PREPARE          0x0080
 #define RBTXN_HAS_STREAMABLE_CHANGE    0x0100
 #define RBTXN_SENT_PREPARE                     0x0200
 #define RBTXN_IS_COMMITTED                     0x0400
 #define RBTXN_IS_ABORTED                       0x0800
 
+#define RBTXN_PREPARE_STATUS_MASK      (RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE | RBTXN_SENT_PREPARE)
+
 /* Does the transaction have catalog changes? */
 #define rbtxn_has_catalog_changes(txn) \
 ( \
@@ -234,9 +236,9 @@ typedef struct ReorderBufferChange
  * committed. To check whether a prepare or a stream_prepare has already
  * been sent for this transaction, we need to use rbtxn_sent_prepare().
  */
-#define rbtxn_prepared(txn) \
+#define rbtxn_is_prepared(txn) \
 ( \
-       ((txn)->txn_flags & RBTXN_PREPARE) != 0 \
+       ((txn)->txn_flags & RBTXN_IS_PREPARED) != 0 \
 )
 
 /* Has a prepare or stream_prepare already been sent? */