diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 606c94df7db..c4299c76fb1 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -115,10 +115,12 @@ #include "utils/relfilenumbermap.h" /* - * Each transaction is limited to a maximum of 8MB of inval messages distributed - * from other transaction. Once the number of distributed inval messages reach - * this threshold, the transaction is marked as RBTXN_INVAL_OVERFLOWED to - * invalidate the complete cache. + * Each transaction has an 8MB limit for invalidation messages distributed from + * other transactions. This limit is set considering scenarios with many + * concurrent logical decoding operations. When the distributed invalidation + * messages reach this threshold, the transaction is marked as + * RBTXN_DISTR_INVAL_OVERFLOWED to invalidate the complete cache as we have lost + * some inval messages and hence don't know what needs to be invalidated. */ #define MAX_DISTR_INVAL_MSG_PER_TXN \ ((8 * 1024 * 1024) / sizeof(SharedInvalidationMessage)) @@ -2677,7 +2679,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, AbortCurrentTransaction(); /* make sure there's no cache pollution */ - if (rbtxn_inval_overflowed(txn)) + if (rbtxn_distr_inval_overflowed(txn)) { Assert(txn->ninvalidations_distributed == 0); InvalidateSystemCaches(); @@ -2736,7 +2738,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, AbortCurrentTransaction(); /* make sure there's no cache pollution */ - if (rbtxn_inval_overflowed(txn)) + if (rbtxn_distr_inval_overflowed(txn)) { Assert(txn->ninvalidations_distributed == 0); InvalidateSystemCaches(); @@ -3095,7 +3097,8 @@ ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, * We might have decoded changes for this transaction that could load * the cache as per the current transaction's view (consider DDL's * happened in this transaction). We don't want the decoding of future - * transactions to use those cache entries so execute invalidations. + * transactions to use those cache entries so execute only the inval + * messages in this transaction. */ if (txn->ninvalidations > 0) ReorderBufferImmediateInvalidation(rb, txn->ninvalidations, @@ -3182,9 +3185,10 @@ ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn) txn->final_lsn = lsn; /* - * Process cache invalidation messages if there are any. Even if we're not - * interested in the transaction's contents, it could have manipulated the - * catalog and we need to update the caches according to that. + * Process only cache invalidation messages in this transaction if there + * are any. Even if we're not interested in the transaction's contents, it + * could have manipulated the catalog and we need to update the caches + * according to that. */ if (txn->base_snapshot != NULL && txn->ninvalidations > 0) ReorderBufferImmediateInvalidation(rb, txn->ninvalidations, @@ -3457,7 +3461,7 @@ ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, } /* - * Add a new invalidation message to the reorder buffer queue. + * Add new invalidation messages to the reorder buffer queue. */ static void ReorderBufferQueueInvalidations(ReorderBuffer *rb, TransactionId xid, @@ -3582,31 +3586,33 @@ ReorderBufferAddDistributedInvalidations(ReorderBuffer *rb, TransactionId xid, Assert(nmsgs > 0); - /* - * Check the transaction has enough space for storing distributed - * invalidation messages. - */ - if (txn->ninvalidations_distributed + nmsgs >= MAX_DISTR_INVAL_MSG_PER_TXN) + if (!rbtxn_distr_inval_overflowed(txn)) { /* - * Mark the invalidation message as overflowed and free up the - * messages accumulated so far. + * Check the transaction has enough space for storing distributed + * invalidation messages. */ - txn->txn_flags |= RBTXN_INVAL_OVERFLOWED; - - if (txn->invalidations_distributed) + if (txn->ninvalidations_distributed + nmsgs >= MAX_DISTR_INVAL_MSG_PER_TXN) { - pfree(txn->invalidations_distributed); - txn->invalidations_distributed = NULL; - txn->ninvalidations_distributed = 0; + /* + * Mark the invalidation message as overflowed and free up the + * messages accumulated so far. + */ + txn->txn_flags |= RBTXN_DISTR_INVAL_OVERFLOWED; + + if (txn->invalidations_distributed) + { + pfree(txn->invalidations_distributed); + txn->invalidations_distributed = NULL; + txn->ninvalidations_distributed = 0; + } } + else + ReorderBufferAccumulateInvalidations(&txn->invalidations_distributed, + &txn->ninvalidations_distributed, + msgs, nmsgs); } - if (!rbtxn_inval_overflowed(txn)) - ReorderBufferAccumulateInvalidations(&txn->invalidations_distributed, - &txn->ninvalidations_distributed, - msgs, nmsgs); - /* Queue the invalidation messages into the transaction */ ReorderBufferQueueInvalidations(rb, xid, lsn, nmsgs, msgs); diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 6579bee3f27..ed92830e75f 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -176,7 +176,7 @@ typedef struct ReorderBufferChange #define RBTXN_SENT_PREPARE 0x0200 #define RBTXN_IS_COMMITTED 0x0400 #define RBTXN_IS_ABORTED 0x0800 -#define RBTXN_INVAL_OVERFLOWED 0x1000 +#define RBTXN_DISTR_INVAL_OVERFLOWED 0x1000 #define RBTXN_PREPARE_STATUS_MASK (RBTXN_IS_PREPARED | RBTXN_SKIPPED_PREPARE | RBTXN_SENT_PREPARE) @@ -266,10 +266,10 @@ typedef struct ReorderBufferChange ((txn)->txn_flags & RBTXN_SKIPPED_PREPARE) != 0 \ ) -/* Is the array of inval message overflowed? */ -#define rbtxn_inval_overflowed(txn) \ +/* Is the array of distributed inval messages overflowed? */ +#define rbtxn_distr_inval_overflowed(txn) \ ( \ - ((txn)->txn_flags & RBTXN_INVAL_OVERFLOWED) != 0 \ + ((txn)->txn_flags & RBTXN_DISTR_INVAL_OVERFLOWED) != 0 \ ) /* Is this a top-level transaction? */ @@ -431,19 +431,6 @@ typedef struct ReorderBufferTXN /* * Stores cache invalidation messages distributed by other transactions. - * - * It is acceptable to skip invalidations received from concurrent - * transactions during ReorderBufferForget and ReorderBufferInvalidate, - * because the transaction being discarded wouldn't have loaded any shared - * catalog entries and invalidating in the source transaction is - * sufficient (see the comment at the caller of ReorderBufferForget). In - * the case of ReorderBufferAbort(), invalidations are processed only if - * the transaction has already streamed some changes. When streaming - * occurs, any concurrent invalidations that happened before the - * triggering statement would have already been processed. That should be - * sufficient to maintain consistency. Therefore, it's safe to skip - * processing invalidations_distributed for ReorderBufferForget, - * ReorderBufferInvalidate, and ReorderBufferAbort. */ uint32 ninvalidations_distributed; SharedInvalidationMessage *invalidations_distributed;