change_v11_masahiko.patch

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;