Re: Reducing memory consumption for pending inval messages

I wrote:
> It turns out that the existing implementation in inval.c is quite
> inefficient when a lot of individual commands each register just
> a few invalidations --- but a few invalidations per command is
> pretty typical.

Per the cfbot, here's a rebase over 3788c6678 (actually just
undoing its effects on inval.c, since that code is removed here).

            regards, tom lane

diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index 9c79775725..9352c68090 100644
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -71,11 +71,6 @@
  *    manipulating the init file is in relcache.c, but we keep track of the
  *    need for it here.
  *
- *    The request lists proper are kept in CurTransactionContext of their
- *    creating (sub)transaction, since they can be forgotten on abort of that
- *    transaction but must be kept till top-level commit otherwise.  For
- *    simplicity we keep the controlling list-of-lists in TopTransactionContext.
- *
  *    Currently, inval messages are sent without regard for the possibility
  *    that the object described by the catalog tuple might be a session-local
  *    object such as a temporary table.  This is because (1) this code has
@@ -106,7 +101,6 @@
 #include "catalog/catalog.h"
 #include "catalog/pg_constraint.h"
 #include "miscadmin.h"
-#include "port/pg_bitutils.h"
 #include "storage/sinval.h"
 #include "storage/smgr.h"
 #include "utils/catcache.h"
@@ -121,35 +115,86 @@


 /*
- * To minimize palloc traffic, we keep pending requests in successively-
- * larger chunks (a slightly more sophisticated version of an expansible
- * array).  All request types can be stored as SharedInvalidationMessage
- * records.  The ordering of requests within a list is never significant.
+ * Pending requests are stored as ready-to-send SharedInvalidationMessages.
+ * We keep the messages themselves in arrays in TopTransactionContext
+ * (there are separate arrays for catcache and relcache messages).  Control
+ * information is kept in a chain of TransInvalidationInfo structs, also
+ * allocated in TopTransactionContext.  (We could keep a subtransaction's
+ * TransInvalidationInfo in its CurTransactionContext; but that's more
+ * wasteful not less so, since in very many scenarios it'd be the only
+ * allocation in the subtransaction's CurTransactionContext.)
+ *
+ * We can store the message arrays densely, and yet avoid moving data around
+ * within an array, because within any one subtransaction we need only
+ * distinguish between messages emitted by prior commands and those emitted
+ * by the current command.  Once a command completes and we've done local
+ * processing on its messages, we can fold those into the prior-commands
+ * messages just by changing array indexes in the TransInvalidationInfo
+ * struct.  Similarly, we need distinguish messages of prior subtransactions
+ * from those of the current subtransaction only until the subtransaction
+ * completes, after which we adjust the array indexes in the parent's
+ * TransInvalidationInfo to include the subtransaction's messages.
+ *
+ * The ordering of the individual messages within a command's or
+ * subtransaction's output is not considered significant, although this
+ * implementation happens to preserve the order in which they were queued.
+ * (Previous versions of this code did not preserve it.)
+ *
+ * For notational convenience, control information is kept in two-element
+ * arrays, the first for catcache messages and the second for relcache
+ * messages.
  */
-typedef struct InvalidationChunk
+#define CatCacheMsgs 0
+#define RelCacheMsgs 1
+
+/* Pointers to main arrays in TopTransactionContext */
+typedef struct InvalMessageArray
 {
-    struct InvalidationChunk *next; /* list link */
-    int            nitems;            /* # items currently stored in chunk */
-    int            maxitems;        /* size of allocated array in this chunk */
-    SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER];
-} InvalidationChunk;
+    SharedInvalidationMessage *msgs;    /* palloc'd array (can be expanded) */
+    int            maxmsgs;        /* current allocated size of array */
+} InvalMessageArray;

-typedef struct InvalidationListHeader
+static InvalMessageArray InvalMessageArrays[2];
+
+/* Control information for one logical group of messages */
+typedef struct InvalidationMsgsGroup
 {
-    InvalidationChunk *cclist;    /* list of chunks holding catcache msgs */
-    InvalidationChunk *rclist;    /* list of chunks holding relcache msgs */
-} InvalidationListHeader;
+    int            firstmsg[2];    /* first index in relevant array */
+    int            nextmsg[2];        /* last+1 index */
+} InvalidationMsgsGroup;
+
+/* Macros to help preserve InvalidationMsgsGroup abstraction */
+#define SetSubGroupToFollow(targetgroup, priorgroup, subgroup) \
+    do { \
+        (targetgroup)->firstmsg[subgroup] = \
+            (targetgroup)->nextmsg[subgroup] = \
+            (priorgroup)->nextmsg[subgroup]; \
+    } while (0)
+
+#define SetGroupToFollow(targetgroup, priorgroup) \
+    do { \
+        SetSubGroupToFollow(targetgroup, priorgroup, CatCacheMsgs); \
+        SetSubGroupToFollow(targetgroup, priorgroup, RelCacheMsgs); \
+    } while (0)
+
+#define NumMessagesInSubGroup(group, subgroup) \
+    ((group)->nextmsg[subgroup] - (group)->firstmsg[subgroup])
+
+#define NumMessagesInGroup(group) \
+    (NumMessagesInSubGroup(group, CatCacheMsgs) + \
+     NumMessagesInSubGroup(group, RelCacheMsgs))
+

 /*----------------
- * Invalidation info is divided into two lists:
+ * Invalidation messages are divided into two groups:
  *    1) events so far in current command, not yet reflected to caches.
  *    2) events in previous commands of current transaction; these have
  *       been reflected to local caches, and must be either broadcast to
  *       other backends or rolled back from local cache when we commit
  *       or abort the transaction.
- * Actually, we need two such lists for each level of nested transaction,
+ * Actually, we need such groups for each level of nested transaction,
  * so that we can discard events from an aborted subtransaction.  When
- * a subtransaction commits, we append its lists to the parent's lists.
+ * a subtransaction commits, we append its events to the parent's groups.
  *
  * The relcache-file-invalidated flag can just be a simple boolean,
  * since we only act on it at transaction commit; we don't care which
@@ -165,11 +210,11 @@ typedef struct TransInvalidationInfo
     /* Subtransaction nesting depth */
     int            my_level;

-    /* head of current-command event list */
-    InvalidationListHeader CurrentCmdInvalidMsgs;
+    /* Events emitted by current command */
+    InvalidationMsgsGroup CurrentCmdInvalidMsgs;

-    /* head of previous-commands event list */
-    InvalidationListHeader PriorCmdInvalidMsgs;
+    /* Events emitted by previous commands of this (sub)transaction */
+    InvalidationMsgsGroup PriorCmdInvalidMsgs;

     /* init file must be invalidated? */
     bool        RelcacheInitFileInval;
@@ -177,10 +222,6 @@ typedef struct TransInvalidationInfo

 static TransInvalidationInfo *transInvalInfo = NULL;

-static SharedInvalidationMessage *SharedInvalidMessagesArray;
-static int    numSharedInvalidMessagesArray;
-static int    maxSharedInvalidMessagesArray;
-
 /* GUC storage */
 int            debug_discard_caches = 0;

@@ -218,124 +259,118 @@ static struct RELCACHECALLBACK
 static int    relcache_callback_count = 0;

 /* ----------------------------------------------------------------
- *                Invalidation list support functions
- *
- * These three routines encapsulate processing of the "chunked"
- * representation of what is logically just a list of messages.
+ *                Invalidation subgroup support functions
  * ----------------------------------------------------------------
  */

 /*
  * AddInvalidationMessage
- *        Add an invalidation message to a list (of chunks).
+ *        Add an invalidation message to a (sub)group.
+ *
+ * The group must be the last active one, since we assume we can add to the
+ * end of the relevant InvalMessageArray.
  *
- * Note that we do not pay any great attention to maintaining the original
- * ordering of the messages.
+ * subgroup must be CatCacheMsgs or RelCacheMsgs.
  */
 static void
-AddInvalidationMessage(InvalidationChunk **listHdr,
-                       SharedInvalidationMessage *msg)
+AddInvalidationMessage(InvalidationMsgsGroup *group, int subgroup,
+                       const SharedInvalidationMessage *msg)
 {
-    InvalidationChunk *chunk = *listHdr;
+    InvalMessageArray *ima = &InvalMessageArrays[subgroup];
+    int            nextindex = group->nextmsg[subgroup];

-    if (chunk == NULL)
-    {
-        /* First time through; create initial chunk */
-#define FIRSTCHUNKSIZE 32
-        chunk = (InvalidationChunk *)
-            MemoryContextAlloc(CurTransactionContext,
-                               offsetof(InvalidationChunk, msgs) +
-                               FIRSTCHUNKSIZE * sizeof(SharedInvalidationMessage));
-        chunk->nitems = 0;
-        chunk->maxitems = FIRSTCHUNKSIZE;
-        chunk->next = *listHdr;
-        *listHdr = chunk;
-    }
-    else if (chunk->nitems >= chunk->maxitems)
+    if (nextindex >= ima->maxmsgs)
     {
-        /* Need another chunk; double size of last chunk */
-        int            chunksize = 2 * chunk->maxitems;
-
-        chunk = (InvalidationChunk *)
-            MemoryContextAlloc(CurTransactionContext,
-                               offsetof(InvalidationChunk, msgs) +
-                               chunksize * sizeof(SharedInvalidationMessage));
-        chunk->nitems = 0;
-        chunk->maxitems = chunksize;
-        chunk->next = *listHdr;
-        *listHdr = chunk;
+        if (ima->msgs == NULL)
+        {
+            /* Create new storage array in TopTransactionContext */
+            int            reqsize = 32;    /* arbitrary */
+
+            ima->msgs = (SharedInvalidationMessage *)
+                MemoryContextAlloc(TopTransactionContext,
+                                   reqsize * sizeof(SharedInvalidationMessage));
+            ima->maxmsgs = reqsize;
+            Assert(nextindex == 0);
+        }
+        else
+        {
+            /* Enlarge storage array */
+            int            reqsize = 2 * ima->maxmsgs;
+
+            ima->msgs = (SharedInvalidationMessage *)
+                repalloc(ima->msgs,
+                         reqsize * sizeof(SharedInvalidationMessage));
+            ima->maxmsgs = reqsize;
+        }
     }
-    /* Okay, add message to current chunk */
-    chunk->msgs[chunk->nitems] = *msg;
-    chunk->nitems++;
+    /* Okay, add message to current group */
+    ima->msgs[nextindex] = *msg;
+    group->nextmsg[subgroup]++;
 }

 /*
- * Append one list of invalidation message chunks to another, resetting
- * the source chunk-list pointer to NULL.
+ * Append one subgroup of invalidation messages to another, resetting
+ * the source subgroup to empty.
  */
 static void
-AppendInvalidationMessageList(InvalidationChunk **destHdr,
-                              InvalidationChunk **srcHdr)
+AppendInvalidationMessageSubGroup(InvalidationMsgsGroup *dest,
+                                  InvalidationMsgsGroup *src,
+                                  int subgroup)
 {
-    InvalidationChunk *chunk = *srcHdr;
-
-    if (chunk == NULL)
-        return;                    /* nothing to do */
-
-    while (chunk->next != NULL)
-        chunk = chunk->next;
+    /* Messages must be adjacent in main array */
+    Assert(dest->nextmsg[subgroup] == src->firstmsg[subgroup]);

-    chunk->next = *destHdr;
+    /* ... which makes this easy: */
+    dest->nextmsg[subgroup] = src->nextmsg[subgroup];

-    *destHdr = *srcHdr;
-
-    *srcHdr = NULL;
+    /*
+     * This is handy for some callers and irrelevant for others.  But we do it
+     * always, reasoning that it's bad to leave different groups pointing at
+     * the same fragment of the message array.
+     */
+    SetSubGroupToFollow(src, dest, subgroup);
 }

 /*
- * Process a list of invalidation messages.
+ * Process a subgroup of invalidation messages.
  *
  * This is a macro that executes the given code fragment for each message in
- * a message chunk list.  The fragment should refer to the message as *msg.
+ * a message subgroup.  The fragment should refer to the message as *msg.
  */
-#define ProcessMessageList(listHdr, codeFragment) \
+#define ProcessMessageSubGroup(group, subgroup, codeFragment) \
     do { \
-        InvalidationChunk *_chunk; \
-        for (_chunk = (listHdr); _chunk != NULL; _chunk = _chunk->next) \
+        int        _msgindex = (group)->firstmsg[subgroup]; \
+        int        _endmsg = (group)->nextmsg[subgroup]; \
+        for (; _msgindex < _endmsg; _msgindex++) \
         { \
-            int        _cindex; \
-            for (_cindex = 0; _cindex < _chunk->nitems; _cindex++) \
-            { \
-                SharedInvalidationMessage *msg = &_chunk->msgs[_cindex]; \
-                codeFragment; \
-            } \
+            SharedInvalidationMessage *msg = \
+                &InvalMessageArrays[subgroup].msgs[_msgindex]; \
+            codeFragment; \
         } \
     } while (0)

 /*
- * Process a list of invalidation messages group-wise.
+ * Process a subgroup of invalidation messages as an array.
  *
  * As above, but the code fragment can handle an array of messages.
  * The fragment should refer to the messages as msgs[], with n entries.
  */
-#define ProcessMessageListMulti(listHdr, codeFragment) \
+#define ProcessMessageSubGroupMulti(group, subgroup, codeFragment) \
     do { \
-        InvalidationChunk *_chunk; \
-        for (_chunk = (listHdr); _chunk != NULL; _chunk = _chunk->next) \
-        { \
-            SharedInvalidationMessage *msgs = _chunk->msgs; \
-            int        n = _chunk->nitems; \
+        int        n = NumMessagesInSubGroup(group, subgroup); \
+        if (n > 0) { \
+            SharedInvalidationMessage *msgs = \
+                &InvalMessageArrays[subgroup].msgs[(group)->firstmsg[subgroup]]; \
             codeFragment; \
         } \
     } while (0)


 /* ----------------------------------------------------------------
- *                Invalidation set support functions
+ *                Invalidation group support functions
  *
  * These routines understand about the division of a logical invalidation
- * list into separate physical lists for catcache and relcache entries.
+ * group into separate physical arrays for catcache and relcache entries.
  * ----------------------------------------------------------------
  */

@@ -343,7 +378,7 @@ AppendInvalidationMessageList(InvalidationChunk **destHdr,
  * Add a catcache inval entry
  */
 static void
-AddCatcacheInvalidationMessage(InvalidationListHeader *hdr,
+AddCatcacheInvalidationMessage(InvalidationMsgsGroup *group,
                                int id, uint32 hashValue, Oid dbId)
 {
     SharedInvalidationMessage msg;
@@ -364,14 +399,14 @@ AddCatcacheInvalidationMessage(InvalidationListHeader *hdr,
      */
     VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg));

-    AddInvalidationMessage(&hdr->cclist, &msg);
+    AddInvalidationMessage(group, CatCacheMsgs, &msg);
 }

 /*
  * Add a whole-catalog inval entry
  */
 static void
-AddCatalogInvalidationMessage(InvalidationListHeader *hdr,
+AddCatalogInvalidationMessage(InvalidationMsgsGroup *group,
                               Oid dbId, Oid catId)
 {
     SharedInvalidationMessage msg;
@@ -382,14 +417,14 @@ AddCatalogInvalidationMessage(InvalidationListHeader *hdr,
     /* check AddCatcacheInvalidationMessage() for an explanation */
     VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg));

-    AddInvalidationMessage(&hdr->cclist, &msg);
+    AddInvalidationMessage(group, CatCacheMsgs, &msg);
 }

 /*
  * Add a relcache inval entry
  */
 static void
-AddRelcacheInvalidationMessage(InvalidationListHeader *hdr,
+AddRelcacheInvalidationMessage(InvalidationMsgsGroup *group,
                                Oid dbId, Oid relId)
 {
     SharedInvalidationMessage msg;
@@ -399,11 +434,11 @@ AddRelcacheInvalidationMessage(InvalidationListHeader *hdr,
      * it will never change. InvalidOid for relId means all relations so we
      * don't need to add individual ones when it is present.
      */
-    ProcessMessageList(hdr->rclist,
-                       if (msg->rc.id == SHAREDINVALRELCACHE_ID &&
-                           (msg->rc.relId == relId ||
-                            msg->rc.relId == InvalidOid))
-                       return);
+    ProcessMessageSubGroup(group, RelCacheMsgs,
+                           if (msg->rc.id == SHAREDINVALRELCACHE_ID &&
+                               (msg->rc.relId == relId ||
+                                msg->rc.relId == InvalidOid))
+                           return);

     /* OK, add the item */
     msg.rc.id = SHAREDINVALRELCACHE_ID;
@@ -412,24 +447,26 @@ AddRelcacheInvalidationMessage(InvalidationListHeader *hdr,
     /* check AddCatcacheInvalidationMessage() for an explanation */
     VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg));

-    AddInvalidationMessage(&hdr->rclist, &msg);
+    AddInvalidationMessage(group, RelCacheMsgs, &msg);
 }

 /*
  * Add a snapshot inval entry
+ *
+ * We put these into the relcache subgroup for simplicity.
  */
 static void
-AddSnapshotInvalidationMessage(InvalidationListHeader *hdr,
+AddSnapshotInvalidationMessage(InvalidationMsgsGroup *group,
                                Oid dbId, Oid relId)
 {
     SharedInvalidationMessage msg;

     /* Don't add a duplicate item */
     /* We assume dbId need not be checked because it will never change */
-    ProcessMessageList(hdr->rclist,
-                       if (msg->sn.id == SHAREDINVALSNAPSHOT_ID &&
-                           msg->sn.relId == relId)
-                       return);
+    ProcessMessageSubGroup(group, RelCacheMsgs,
+                           if (msg->sn.id == SHAREDINVALSNAPSHOT_ID &&
+                               msg->sn.relId == relId)
+                           return);

     /* OK, add the item */
     msg.sn.id = SHAREDINVALSNAPSHOT_ID;
@@ -438,33 +475,33 @@ AddSnapshotInvalidationMessage(InvalidationListHeader *hdr,
     /* check AddCatcacheInvalidationMessage() for an explanation */
     VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg));

-    AddInvalidationMessage(&hdr->rclist, &msg);
+    AddInvalidationMessage(group, RelCacheMsgs, &msg);
 }

 /*
- * Append one list of invalidation messages to another, resetting
- * the source list to empty.
+ * Append one group of invalidation messages to another, resetting
+ * the source group to empty.
  */
 static void
-AppendInvalidationMessages(InvalidationListHeader *dest,
-                           InvalidationListHeader *src)
+AppendInvalidationMessages(InvalidationMsgsGroup *dest,
+                           InvalidationMsgsGroup *src)
 {
-    AppendInvalidationMessageList(&dest->cclist, &src->cclist);
-    AppendInvalidationMessageList(&dest->rclist, &src->rclist);
+    AppendInvalidationMessageSubGroup(dest, src, CatCacheMsgs);
+    AppendInvalidationMessageSubGroup(dest, src, RelCacheMsgs);
 }

 /*
- * Execute the given function for all the messages in an invalidation list.
- * The list is not altered.
+ * Execute the given function for all the messages in an invalidation group.
+ * The group is not altered.
  *
  * catcache entries are processed first, for reasons mentioned above.
  */
 static void
-ProcessInvalidationMessages(InvalidationListHeader *hdr,
+ProcessInvalidationMessages(InvalidationMsgsGroup *group,
                             void (*func) (SharedInvalidationMessage *msg))
 {
-    ProcessMessageList(hdr->cclist, func(msg));
-    ProcessMessageList(hdr->rclist, func(msg));
+    ProcessMessageSubGroup(group, CatCacheMsgs, func(msg));
+    ProcessMessageSubGroup(group, RelCacheMsgs, func(msg));
 }

 /*
@@ -472,11 +509,11 @@ ProcessInvalidationMessages(InvalidationListHeader *hdr,
  * rather than just one at a time.
  */
 static void
-ProcessInvalidationMessagesMulti(InvalidationListHeader *hdr,
+ProcessInvalidationMessagesMulti(InvalidationMsgsGroup *group,
                                  void (*func) (const SharedInvalidationMessage *msgs, int n))
 {
-    ProcessMessageListMulti(hdr->cclist, func(msgs, n));
-    ProcessMessageListMulti(hdr->rclist, func(msgs, n));
+    ProcessMessageSubGroupMulti(group, CatCacheMsgs, func(msgs, n));
+    ProcessMessageSubGroupMulti(group, RelCacheMsgs, func(msgs, n));
 }

 /* ----------------------------------------------------------------
@@ -731,7 +768,7 @@ AcceptInvalidationMessages(void)

 /*
  * PrepareInvalidationState
- *        Initialize inval lists for the current (sub)transaction.
+ *        Initialize inval data for the current (sub)transaction.
  */
 static void
 PrepareInvalidationState(void)
@@ -748,12 +785,45 @@ PrepareInvalidationState(void)
     myInfo->parent = transInvalInfo;
     myInfo->my_level = GetCurrentTransactionNestLevel();

-    /*
-     * If there's any previous entry, this one should be for a deeper nesting
-     * level.
-     */
-    Assert(transInvalInfo == NULL ||
-           myInfo->my_level > transInvalInfo->my_level);
+    /* Now, do we have a previous stack entry? */
+    if (transInvalInfo != NULL)
+    {
+        /* Yes; this one should be for a deeper nesting level. */
+        Assert(myInfo->my_level > transInvalInfo->my_level);
+
+        /*
+         * The parent (sub)transaction must not have any current (i.e.,
+         * not-yet-locally-processed) messages.  If it did, we'd have a
+         * semantic problem: the new subtransaction presumably ought not be
+         * able to see those events yet, but since the CommandCounter is
+         * linear, that can't work once the subtransaction advances the
+         * counter.  This is a convenient place to check for that, as well as
+         * being important to keep management of the message arrays simple.
+         */
+        if (NumMessagesInGroup(&transInvalInfo->CurrentCmdInvalidMsgs) != 0)
+            elog(ERROR, "cannot start a subtransaction when there are unprocessed inval messages");
+
+        /*
+         * MemoryContextAllocZero set firstmsg = nextmsg = 0 in each group,
+         * which is fine for the first (sub)transaction, but otherwise we need
+         * to update them to follow whatever is already in the arrays.
+         */
+        SetGroupToFollow(&myInfo->PriorCmdInvalidMsgs,
+                         &transInvalInfo->CurrentCmdInvalidMsgs);
+        SetGroupToFollow(&myInfo->CurrentCmdInvalidMsgs,
+                         &myInfo->PriorCmdInvalidMsgs);
+    }
+    else
+    {
+        /*
+         * Here, we need only clear any array pointers left over from a prior
+         * transaction.
+         */
+        InvalMessageArrays[CatCacheMsgs].msgs = NULL;
+        InvalMessageArrays[CatCacheMsgs].maxmsgs = 0;
+        InvalMessageArrays[RelCacheMsgs].msgs = NULL;
+        InvalMessageArrays[RelCacheMsgs].maxmsgs = 0;
+    }

     transInvalInfo = myInfo;
 }
@@ -777,47 +847,8 @@ PostPrepare_Inval(void)
 }

 /*
- * Collect invalidation messages into SharedInvalidMessagesArray array.
- */
-static void
-MakeSharedInvalidMessagesArray(const SharedInvalidationMessage *msgs, int n)
-{
-    /*
-     * Initialise array first time through in each commit
-     */
-    if (SharedInvalidMessagesArray == NULL)
-    {
-        maxSharedInvalidMessagesArray = FIRSTCHUNKSIZE;
-        numSharedInvalidMessagesArray = 0;
-
-        /*
-         * Although this is being palloc'd we don't actually free it directly.
-         * We're so close to EOXact that we now we're going to lose it anyhow.
-         */
-        SharedInvalidMessagesArray = palloc(maxSharedInvalidMessagesArray
-                                            * sizeof(SharedInvalidationMessage));
-    }
-
-    if ((numSharedInvalidMessagesArray + n) > maxSharedInvalidMessagesArray)
-    {
-        maxSharedInvalidMessagesArray = pg_nextpower2_32(numSharedInvalidMessagesArray + n);
-
-        SharedInvalidMessagesArray = repalloc(SharedInvalidMessagesArray,
-                                              maxSharedInvalidMessagesArray
-                                              * sizeof(SharedInvalidationMessage));
-    }
-
-    /*
-     * Append the next chunk onto the array
-     */
-    memcpy(SharedInvalidMessagesArray + numSharedInvalidMessagesArray,
-           msgs, n * sizeof(SharedInvalidationMessage));
-    numSharedInvalidMessagesArray += n;
-}
-
-/*
- * xactGetCommittedInvalidationMessages() is executed by
- * RecordTransactionCommit() to add invalidation messages onto the
+ * xactGetCommittedInvalidationMessages() is called by
+ * RecordTransactionCommit() to collect invalidation messages to add to the
  * commit record. This applies only to commit message types, never to
  * abort records. Must always run before AtEOXact_Inval(), since that
  * removes the data we need to see.
@@ -832,7 +863,9 @@ int
 xactGetCommittedInvalidationMessages(SharedInvalidationMessage **msgs,
                                      bool *RelcacheInitFileInval)
 {
-    MemoryContext oldcontext;
+    SharedInvalidationMessage *msgarray;
+    int            nummsgs;
+    int            nmsgs;

     /* Quick exit if we haven't done anything with invalidation messages. */
     if (transInvalInfo == NULL)
@@ -853,27 +886,48 @@ xactGetCommittedInvalidationMessages(SharedInvalidationMessage **msgs,
     *RelcacheInitFileInval = transInvalInfo->RelcacheInitFileInval;

     /*
-     * Walk through TransInvalidationInfo to collect all the messages into a
-     * single contiguous array of invalidation messages. It must be contiguous
-     * so we can copy directly into WAL message. Maintain the order that they
-     * would be processed in by AtEOXact_Inval(), to ensure emulated behaviour
-     * in redo is as similar as possible to original. We want the same bugs,
-     * if any, not new ones.
+     * Collect all the pending messages into a single contiguous array of
+     * invalidation messages, to simplify what needs to happen while building
+     * the commit WAL message.  Maintain the order that they would be
+     * processed in by AtEOXact_Inval(), to ensure emulated behaviour in redo
+     * is as similar as possible to original.  We want the same bugs, if any,
+     * not new ones.
      */
-    oldcontext = MemoryContextSwitchTo(CurTransactionContext);
-
-    ProcessInvalidationMessagesMulti(&transInvalInfo->CurrentCmdInvalidMsgs,
-                                     MakeSharedInvalidMessagesArray);
-    ProcessInvalidationMessagesMulti(&transInvalInfo->PriorCmdInvalidMsgs,
-                                     MakeSharedInvalidMessagesArray);
-    MemoryContextSwitchTo(oldcontext);
-
-    Assert(!(numSharedInvalidMessagesArray > 0 &&
-             SharedInvalidMessagesArray == NULL));
-
-    *msgs = SharedInvalidMessagesArray;
-
-    return numSharedInvalidMessagesArray;
+    nummsgs = NumMessagesInGroup(&transInvalInfo->PriorCmdInvalidMsgs) +
+        NumMessagesInGroup(&transInvalInfo->CurrentCmdInvalidMsgs);
+
+    *msgs = msgarray = (SharedInvalidationMessage *)
+        MemoryContextAlloc(CurTransactionContext,
+                           nummsgs * sizeof(SharedInvalidationMessage));
+
+    nmsgs = 0;
+    ProcessMessageSubGroupMulti(&transInvalInfo->PriorCmdInvalidMsgs,
+                                CatCacheMsgs,
+                                (memcpy(msgarray + nmsgs,
+                                        msgs,
+                                        n * sizeof(SharedInvalidationMessage)),
+                                 nmsgs += n));
+    ProcessMessageSubGroupMulti(&transInvalInfo->CurrentCmdInvalidMsgs,
+                                CatCacheMsgs,
+                                (memcpy(msgarray + nmsgs,
+                                        msgs,
+                                        n * sizeof(SharedInvalidationMessage)),
+                                 nmsgs += n));
+    ProcessMessageSubGroupMulti(&transInvalInfo->PriorCmdInvalidMsgs,
+                                RelCacheMsgs,
+                                (memcpy(msgarray + nmsgs,
+                                        msgs,
+                                        n * sizeof(SharedInvalidationMessage)),
+                                 nmsgs += n));
+    ProcessMessageSubGroupMulti(&transInvalInfo->CurrentCmdInvalidMsgs,
+                                RelCacheMsgs,
+                                (memcpy(msgarray + nmsgs,
+                                        msgs,
+                                        n * sizeof(SharedInvalidationMessage)),
+                                 nmsgs += n));
+    Assert(nmsgs == nummsgs);
+
+    return nmsgs;
 }

 /*
@@ -942,7 +996,7 @@ ProcessCommittedInvalidationMessages(SharedInvalidationMessage *msgs,
  * about CurrentCmdInvalidMsgs too, since those changes haven't touched
  * the caches yet.
  *
- * In any case, reset the various lists to empty.  We need not physically
+ * In any case, reset our state to empty.  We need not physically
  * free memory here, since TopTransactionContext is about to be emptied
  * anyway.
  *
@@ -986,8 +1040,6 @@ AtEOXact_Inval(bool isCommit)

     /* Need not free anything explicitly */
     transInvalInfo = NULL;
-    SharedInvalidMessagesArray = NULL;
-    numSharedInvalidMessagesArray = 0;
 }

 /*
@@ -1043,10 +1095,21 @@ AtEOSubXact_Inval(bool isCommit)
             return;
         }

-        /* Pass up my inval messages to parent */
+        /*
+         * Pass up my inval messages to parent.  Notice that we stick them in
+         * PriorCmdInvalidMsgs, not CurrentCmdInvalidMsgs, since they've
+         * already been locally processed.  (This would trigger the Assert in
+         * AppendInvalidationMessageSubGroup if the parent's
+         * CurrentCmdInvalidMsgs isn't empty; but we already checked that in
+         * PrepareInvalidationState.)
+         */
         AppendInvalidationMessages(&myInfo->parent->PriorCmdInvalidMsgs,
                                    &myInfo->PriorCmdInvalidMsgs);

+        /* Must readjust parent's CurrentCmdInvalidMsgs indexes now */
+        SetGroupToFollow(&myInfo->parent->CurrentCmdInvalidMsgs,
+                         &myInfo->parent->PriorCmdInvalidMsgs);
+
         /* Pending relcache inval becomes parent's problem too */
         if (myInfo->RelcacheInitFileInval)
             myInfo->parent->RelcacheInitFileInval = true;
@@ -1514,31 +1577,24 @@ CallSyscacheCallbacks(int cacheid, uint32 hashvalue)
 /*
  * LogLogicalInvalidations
  *
- * Emit WAL for invalidations.  This is currently only used for logging
- * invalidations at the command end or at commit time if any invalidations
- * are pending.
+ * Emit WAL for invalidations caused by the current command.
+ *
+ * This is currently only used for logging invalidations at the command end
+ * or at commit time if any invalidations are pending.
  */
 void
-LogLogicalInvalidations()
+LogLogicalInvalidations(void)
 {
     xl_xact_invals xlrec;
-    SharedInvalidationMessage *invalMessages;
-    int            nmsgs = 0;
+    InvalidationMsgsGroup *group;
+    int            nmsgs;

     /* Quick exit if we haven't done anything with invalidation messages. */
     if (transInvalInfo == NULL)
         return;

-    ProcessInvalidationMessagesMulti(&transInvalInfo->CurrentCmdInvalidMsgs,
-                                     MakeSharedInvalidMessagesArray);
-
-    Assert(!(numSharedInvalidMessagesArray > 0 &&
-             SharedInvalidMessagesArray == NULL));
-
-    invalMessages = SharedInvalidMessagesArray;
-    nmsgs = numSharedInvalidMessagesArray;
-    SharedInvalidMessagesArray = NULL;
-    numSharedInvalidMessagesArray = 0;
+    group = &transInvalInfo->CurrentCmdInvalidMsgs;
+    nmsgs = NumMessagesInGroup(group);

     if (nmsgs > 0)
     {
@@ -1549,10 +1605,12 @@ LogLogicalInvalidations()
         /* perform insertion */
         XLogBeginInsert();
         XLogRegisterData((char *) (&xlrec), MinSizeOfXactInvals);
-        XLogRegisterData((char *) invalMessages,
-                         nmsgs * sizeof(SharedInvalidationMessage));
+        ProcessMessageSubGroupMulti(group, CatCacheMsgs,
+                                    XLogRegisterData((char *) msgs,
+                                                     n * sizeof(SharedInvalidationMessage)));
+        ProcessMessageSubGroupMulti(group, RelCacheMsgs,
+                                    XLogRegisterData((char *) msgs,
+                                                     n * sizeof(SharedInvalidationMessage)));
         XLogInsert(RM_XACT_ID, XLOG_XACT_INVALIDATIONS);
-
-        pfree(invalMessages);
     }
 }