Add macros for ReorderBufferTXN toptxn.

Currently, there are quite a few places in reorderbuffer.c that tries to
access top-transaction for a subtransaction. This makes the code to access
top-transaction consistent and easier to follow.

Author: Peter Smith
Reviewed-by: Vignesh C, Sawada Masahiko
Discussion: https://postgr.es/m/CAHut+PuCznOyTqBQwjRUu-ibG-=KHyCv-0FTcWQtZUdR88umfg@mail.gmail.com

diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
index b7e60486474a064e7c02b962f214621678b20fa6..628c6a2595786f91cc09627e10e2930b46106f3d 100644 (file)
--- a/contrib/test_decoding/test_decoding.c
+++ b/contrib/test_decoding/test_decoding.c
@@ -815,11 +815,11 @@ pg_decode_stream_abort(LogicalDecodingContext *ctx,
         * maintain the output_plugin_private only under the toptxn so if this is
         * not the toptxn then fetch the toptxn.
         */
-       ReorderBufferTXN *toptxn = txn->toptxn ? txn->toptxn : txn;
+       ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn);
        TestDecodingTxnData *txndata = toptxn->output_plugin_private;
        bool            xact_wrote_changes = txndata->xact_wrote_changes;
 
-       if (txn->toptxn == NULL)
+       if (rbtxn_is_toptxn(txn))
        {
                Assert(txn->output_plugin_private != NULL);
                pfree(txndata);

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 2d17c551a80f8190df869f03ee8801c4dcf6226f..9f4497447302d4f10663c2c056e69ae270a0b5a4 100644 (file)
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -717,10 +717,7 @@ ReorderBufferProcessPartialChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
                return;
 
        /* Get the top transaction. */
-       if (txn->toptxn != NULL)
-               toptxn = txn->toptxn;
-       else
-               toptxn = txn;
+       toptxn = rbtxn_get_toptxn(txn);
 
        /*
         * Indicate a partial change for toast inserts.  The change will be
@@ -809,13 +806,7 @@ ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
                change->action == REORDER_BUFFER_CHANGE_TRUNCATE ||
                change->action == REORDER_BUFFER_CHANGE_MESSAGE)
        {
-               ReorderBufferTXN *toptxn;
-
-               /* get the top transaction */
-               if (txn->toptxn != NULL)
-                       toptxn = txn->toptxn;
-               else
-                       toptxn = txn;
+               ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn);
 
                toptxn->txn_flags |= RBTXN_HAS_STREAMABLE_CHANGE;
        }
@@ -1655,9 +1646,9 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, bool txn_prep
        /*
         * Mark the transaction as streamed.
         *
-        * The toplevel transaction, identified by (toptxn==NULL), is marked as
-        * streamed always, even if it does not contain any changes (that is, when
-        * all the changes are in subtransactions).
+        * The top-level transaction, is marked as streamed always, even if it
+        * does not contain any changes (that is, when all the changes are in
+        * subtransactions).
         *
         * For subtransactions, we only mark them as streamed when there are
         * changes in them.
@@ -1667,7 +1658,7 @@ ReorderBufferTruncateTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, bool txn_prep
         * about the toplevel xact (we send the XID in all messages), but we never
         * stream XIDs of empty subxacts.
         */
-       if ((!txn_prepared) && ((!txn->toptxn) || (txn->nentries_mem != 0)))
+       if ((!txn_prepared) && (rbtxn_is_toptxn(txn) || (txn->nentries_mem != 0)))
                txn->txn_flags |= RBTXN_IS_STREAMED;
 
        if (txn_prepared)
@@ -3207,10 +3198,7 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
         * Update the total size in top level as well. This is later used to
         * compute the decoding stats.
         */
-       if (txn->toptxn != NULL)
-               toptxn = txn->toptxn;
-       else
-               toptxn = txn;
+       toptxn = rbtxn_get_toptxn(txn);
 
        if (addition)
        {
@@ -3295,8 +3283,7 @@ ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid,
         * so that we can execute them all together.  See comments atop this
         * function.
         */
-       if (txn->toptxn)
-               txn = txn->toptxn;
+       txn = rbtxn_get_toptxn(txn);
 
        Assert(nmsgs > 0);
 
@@ -3354,7 +3341,6 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
                                                                  XLogRecPtr lsn)
 {
        ReorderBufferTXN *txn;
-       ReorderBufferTXN *toptxn;
 
        txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
@@ -3370,11 +3356,15 @@ ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid,
         * conveniently check just top-level transaction and decide whether to
         * build the hash table or not.
         */
-       toptxn = txn->toptxn;
-       if (toptxn != NULL && !rbtxn_has_catalog_changes(toptxn))
+       if (rbtxn_is_subtxn(txn))
        {
-               toptxn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
-               dclist_push_tail(&rb->catchange_txns, &toptxn->catchange_node);
+               ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn);
+
+               if (!rbtxn_has_catalog_changes(toptxn))
+               {
+                       toptxn->txn_flags |= RBTXN_HAS_CATALOG_CHANGES;
+                       dclist_push_tail(&rb->catchange_txns, &toptxn->catchange_node);
+               }
        }
 }
 
@@ -3619,7 +3609,7 @@ ReorderBufferCheckMemoryLimit(ReorderBuffer *rb)
                        (txn = ReorderBufferLargestStreamableTopTXN(rb)) != NULL)
                {
                        /* we know there has to be one, because the size is not zero */
-                       Assert(txn && !txn->toptxn);
+                       Assert(txn && rbtxn_is_toptxn(txn));
                        Assert(txn->total_size > 0);
                        Assert(rb->size >= txn->total_size);
 
@@ -4007,7 +3997,7 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
        bool            txn_is_streamed;
 
        /* We can never reach here for a subtransaction. */
-       Assert(txn->toptxn == NULL);
+       Assert(rbtxn_is_toptxn(txn));
 
        /*
         * We can't make any assumptions about base snapshot here, similar to what

diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 00a2d73dab034b8da85bd8e3d4fe17a36a98611d..3a2d2e357e031fed665ca012dd6f720133b2804d 100644 (file)
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -694,8 +694,8 @@ maybe_send_schema(LogicalDecodingContext *ctx,
        if (in_streaming)
                xid = change->txn->xid;
 
-       if (change->txn->toptxn)
-               topxid = change->txn->toptxn->xid;
+       if (rbtxn_is_subtxn(change->txn))
+               topxid = rbtxn_get_toptxn(change->txn)->xid;
        else
                topxid = xid;
 
@@ -1879,7 +1879,7 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
        Assert(!in_streaming);
 
        /* determine the toplevel transaction */
-       toptxn = (txn->toptxn) ? txn->toptxn : txn;
+       toptxn = rbtxn_get_toptxn(txn);
 
        Assert(rbtxn_is_streamed(toptxn));
 

diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 215d1494e9010b127cb99e05e46858be876b7198..e37f5120ebbf468c3e0777020b89eda3f3acb01a 100644 (file)
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -249,6 +249,24 @@ typedef struct ReorderBufferChange
        ((txn)->txn_flags & RBTXN_SKIPPED_PREPARE) != 0 \
 )
 
+/* Is this a top-level transaction? */
+#define rbtxn_is_toptxn(txn) \
+( \
+       (txn)->toptxn == NULL \
+)
+
+/* Is this a subtransaction? */
+#define rbtxn_is_subtxn(txn) \
+( \
+       (txn)->toptxn != NULL \
+)
+
+/* Get the top-level transaction of this (sub)transaction. */
+#define rbtxn_get_toptxn(txn) \
+( \
+       rbtxn_is_subtxn(txn) ? (txn)->toptxn : (txn) \
+)
+
 typedef struct ReorderBufferTXN
 {
        /* See above */