database/postgres
database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-07 11:02:12 +03:00
Code

Add macros for ReorderBufferTXN toptxn.

Browse Source 
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
This commit is contained in:
Amit Kapila 2023-03-17 08:29:41 +05:30
parent eb7d043c9b
commit e709596b25
4 changed files with 41 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
contrib/test_decoding/test_decoding.c
View File

@ -815,11 +815,11 @@ pg_decode_stream_abort(LogicalDecodingContext *ctx,
* maintain the output_plugin_private only under the toptxn so if this is * maintain the output_plugin_private only under the toptxn so if this is
* not the toptxn then fetch the toptxn. * 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; TestDecodingTxnData *txndata = toptxn->output_plugin_private;
bool xact_wrote_changes = txndata->xact_wrote_changes; bool xact_wrote_changes = txndata->xact_wrote_changes;
if (txn->toptxn == NULL) if (rbtxn_is_toptxn(txn))
{ {
Assert(txn->output_plugin_private != NULL); Assert(txn->output_plugin_private != NULL);
pfree(txndata); pfree(txndata);

46
src/backend/replication/logical/reorderbuffer.c
View File

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

6
src/backend/replication/pgoutput/pgoutput.c
View File

@ -694,8 +694,8 @@ maybe_send_schema(LogicalDecodingContext *ctx,
if (in_streaming) if (in_streaming)
xid = change->txn->xid; xid = change->txn->xid;
if (change->txn->toptxn) if (rbtxn_is_subtxn(change->txn))
topxid = change->txn->toptxn->xid; topxid = rbtxn_get_toptxn(change->txn)->xid;
else else
topxid = xid; topxid = xid;
@ -1879,7 +1879,7 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
Assert(!in_streaming); Assert(!in_streaming);
/* determine the toplevel transaction */ /* determine the toplevel transaction */
toptxn = (txn->toptxn) ? txn->toptxn : txn; toptxn = rbtxn_get_toptxn(txn);
Assert(rbtxn_is_streamed(toptxn)); Assert(rbtxn_is_streamed(toptxn));

18
src/include/replication/reorderbuffer.h
View File

@ -249,6 +249,24 @@ typedef struct ReorderBufferChange
((txn)->txn_flags & RBTXN_SKIPPED_PREPARE) != 0 \ ((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 typedef struct ReorderBufferTXN
{ {
/* See above */ /* See above */