Introduce macros for WAL block reference IDs of some heap record types

When registering a buffer with the WAL machinery, the caller assigns it a
block reference ID, and replay must read each block back by that same ID.
Today these IDs are bare integers assigned by convention (0, 1, 2, ...),
which is easy to follow when a record registers a single block, or when the
blocks are handled during replay in their registration order.

An upcoming bug fix registers up to two visibility map blocks in addition
to the heap block(s) when clearing the VM, and these are not handled during
replay in a straightforward 1:1, in-registration-order fashion. Relying on
bare integers for the block IDs in that case is error-prone.

Introduce macros naming the block reference IDs for the heap record types
that the upcoming commit extends to register visibility map blocks, so the
registration and replay sites refer to the same block by a meaningful name.

Author: Melanie Plageman <melanieplageman@gmail.com>
Reviewed-by: Robert Haas <robertmhaas@gmail.com>
Discussion: https://postgr.es/m/66mqpfyti3qhfttcsv6r2lbvqqd32rrmpn6i47ovrsnvguts46%40gou54xc>
Backpatch through: 17

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index aff02cb003e8c8f021088d624c8759ac23e14b86..cf37956a2bf54be31f7cd734b49ec10614f1e6a3 100644 (file)
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2218,10 +2218,12 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
                 * write the whole page to the xlog, we don't need to store
                 * xl_heap_header in the xlog.
                 */
-               XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | bufflags);
-               XLogRegisterBufData(0, &xlhdr, SizeOfHeapHeader);
+               XLogRegisterBuffer(HEAP_INSERT_BLKREF_HEAP, buffer,
+                                                  REGBUF_STANDARD | bufflags);
+               XLogRegisterBufData(HEAP_INSERT_BLKREF_HEAP, &xlhdr,
+                                                       SizeOfHeapHeader);
                /* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
-               XLogRegisterBufData(0,
+               XLogRegisterBufData(HEAP_INSERT_BLKREF_HEAP,
                                                        (char *) heaptup->t_data + SizeofHeapTupleHeader,
                                                        heaptup->t_len - SizeofHeapTupleHeader);
 
@@ -2619,9 +2621,11 @@ heap_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
 
                        XLogBeginInsert();
                        XLogRegisterData(xlrec, tupledata - scratch.data);
-                       XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | bufflags);
+                       XLogRegisterBuffer(HEAP_MULTI_INSERT_BLKREF_HEAP, buffer,
+                                                          REGBUF_STANDARD | bufflags);
 
-                       XLogRegisterBufData(0, tupledata, totaldatalen);
+                       XLogRegisterBufData(HEAP_MULTI_INSERT_BLKREF_HEAP, tupledata,
+                                                               totaldatalen);
 
                        /* filtering by origin on a row level is much more efficient */
                        XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
@@ -3112,7 +3116,7 @@ l1:
                XLogBeginInsert();
                XLogRegisterData(&xlrec, SizeOfHeapDelete);
 
-               XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
+               XLogRegisterBuffer(HEAP_DELETE_BLKREF_HEAP, buffer, REGBUF_STANDARD);
 
                /*
                 * Log replica identity of the deleted tuple if there is one
@@ -3883,7 +3887,7 @@ l2:
                        XLogRecPtr      recptr;
 
                        XLogBeginInsert();
-                       XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
+                       XLogRegisterBuffer(HEAP_LOCK_BLKREF_HEAP, buffer, REGBUF_STANDARD);
 
                        xlrec.offnum = ItemPointerGetOffsetNumber(&oldtup.t_self);
                        xlrec.xmax = xmax_lock_old_tuple;
@@ -5219,7 +5223,7 @@ failed:
                XLogRecPtr      recptr;
 
                XLogBeginInsert();
-               XLogRegisterBuffer(0, *buffer, REGBUF_STANDARD);
+               XLogRegisterBuffer(HEAP_LOCK_BLKREF_HEAP, *buffer, REGBUF_STANDARD);
 
                xlrec.offnum = ItemPointerGetOffsetNumber(&tuple->t_self);
                xlrec.xmax = xid;
@@ -5971,7 +5975,7 @@ l4:
                        Page            page = BufferGetPage(buf);
 
                        XLogBeginInsert();
-                       XLogRegisterBuffer(0, buf, REGBUF_STANDARD);
+                       XLogRegisterBuffer(HEAP_LOCK_BLKREF_HEAP, buf, REGBUF_STANDARD);
 
                        xlrec.offnum = ItemPointerGetOffsetNumber(&mytup.t_self);
                        xlrec.xmax = new_xmax;
@@ -8972,9 +8976,9 @@ log_heap_update(Relation reln, Buffer oldbuf,
        if (need_tuple_data)
                bufflags |= REGBUF_KEEP_DATA;
 
-       XLogRegisterBuffer(0, newbuf, bufflags);
+       XLogRegisterBuffer(HEAP_UPDATE_BLKREF_HEAP_NEW, newbuf, bufflags);
        if (oldbuf != newbuf)
-               XLogRegisterBuffer(1, oldbuf, REGBUF_STANDARD);
+               XLogRegisterBuffer(HEAP_UPDATE_BLKREF_HEAP_OLD, oldbuf, REGBUF_STANDARD);
 
        XLogRegisterData(&xlrec, SizeOfHeapUpdate);
 
@@ -8987,15 +8991,18 @@ log_heap_update(Relation reln, Buffer oldbuf,
                {
                        prefix_suffix[0] = prefixlen;
                        prefix_suffix[1] = suffixlen;
-                       XLogRegisterBufData(0, &prefix_suffix, sizeof(uint16) * 2);
+                       XLogRegisterBufData(HEAP_UPDATE_BLKREF_HEAP_NEW, &prefix_suffix,
+                                                               sizeof(uint16) * 2);
                }
                else if (prefixlen > 0)
                {
-                       XLogRegisterBufData(0, &prefixlen, sizeof(uint16));
+                       XLogRegisterBufData(HEAP_UPDATE_BLKREF_HEAP_NEW, &prefixlen,
+                                                               sizeof(uint16));
                }
                else
                {
-                       XLogRegisterBufData(0, &suffixlen, sizeof(uint16));
+                       XLogRegisterBufData(HEAP_UPDATE_BLKREF_HEAP_NEW, &suffixlen,
+                                                               sizeof(uint16));
                }
        }
 
@@ -9009,10 +9016,10 @@ log_heap_update(Relation reln, Buffer oldbuf,
         *
         * The 'data' doesn't include the common prefix or suffix.
         */
-       XLogRegisterBufData(0, &xlhdr, SizeOfHeapHeader);
+       XLogRegisterBufData(HEAP_UPDATE_BLKREF_HEAP_NEW, &xlhdr, SizeOfHeapHeader);
        if (prefixlen == 0)
        {
-               XLogRegisterBufData(0,
+               XLogRegisterBufData(HEAP_UPDATE_BLKREF_HEAP_NEW,
                                                        (char *) newtup->t_data + SizeofHeapTupleHeader,
                                                        newtup->t_len - SizeofHeapTupleHeader - suffixlen);
        }
@@ -9025,13 +9032,13 @@ log_heap_update(Relation reln, Buffer oldbuf,
                /* bitmap [+ padding] [+ oid] */
                if (newtup->t_data->t_hoff - SizeofHeapTupleHeader > 0)
                {
-                       XLogRegisterBufData(0,
+                       XLogRegisterBufData(HEAP_UPDATE_BLKREF_HEAP_NEW,
                                                                (char *) newtup->t_data + SizeofHeapTupleHeader,
                                                                newtup->t_data->t_hoff - SizeofHeapTupleHeader);
                }
 
                /* data after common prefix */
-               XLogRegisterBufData(0,
+               XLogRegisterBufData(HEAP_UPDATE_BLKREF_HEAP_NEW,
                                                        (char *) newtup->t_data + newtup->t_data->t_hoff + prefixlen,
                                                        newtup->t_len - newtup->t_data->t_hoff - prefixlen - suffixlen);
        }

diff --git a/src/backend/access/heap/heapam_xlog.c b/src/backend/access/heap/heapam_xlog.c
index eb4bd3d6ae3a3cd4281a1e4c03326b4b2403c37e..87b5c44fd46eee2bf8f89962f3aa3b809f2619d6 100644 (file)
--- a/src/backend/access/heap/heapam_xlog.c
+++ b/src/backend/access/heap/heapam_xlog.c
@@ -350,7 +350,8 @@ heap_xlog_delete(XLogReaderState *record)
        RelFileLocator target_locator;
        ItemPointerData target_tid;
 
-       XLogRecGetBlockTag(record, 0, &target_locator, NULL, &blkno);
+       XLogRecGetBlockTag(record, HEAP_DELETE_BLKREF_HEAP, &target_locator, NULL,
+                                          &blkno);
        ItemPointerSetBlockNumber(&target_tid, blkno);
        ItemPointerSetOffsetNumber(&target_tid, xlrec->offnum);
 
@@ -369,7 +370,8 @@ heap_xlog_delete(XLogReaderState *record)
                FreeFakeRelcacheEntry(reln);
        }
 
-       if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
+       if (XLogReadBufferForRedo(record, HEAP_DELETE_BLKREF_HEAP,
+                                                         &buffer) == BLK_NEEDS_REDO)
        {
                page = BufferGetPage(buffer);
 
@@ -434,7 +436,8 @@ heap_xlog_insert(XLogReaderState *record)
        ItemPointerData target_tid;
        XLogRedoAction action;
 
-       XLogRecGetBlockTag(record, 0, &target_locator, NULL, &blkno);
+       XLogRecGetBlockTag(record, HEAP_INSERT_BLKREF_HEAP, &target_locator, NULL,
+                                          &blkno);
        ItemPointerSetBlockNumber(&target_tid, blkno);
        ItemPointerSetOffsetNumber(&target_tid, xlrec->offnum);
 
@@ -462,13 +465,14 @@ heap_xlog_insert(XLogReaderState *record)
         */
        if (XLogRecGetInfo(record) & XLOG_HEAP_INIT_PAGE)
        {
-               buffer = XLogInitBufferForRedo(record, 0);
+               buffer = XLogInitBufferForRedo(record, HEAP_INSERT_BLKREF_HEAP);
                page = BufferGetPage(buffer);
                PageInit(page, BufferGetPageSize(buffer), 0);
                action = BLK_NEEDS_REDO;
        }
        else
-               action = XLogReadBufferForRedo(record, 0, &buffer);
+               action = XLogReadBufferForRedo(record, HEAP_INSERT_BLKREF_HEAP,
+                                                                          &buffer);
        if (action == BLK_NEEDS_REDO)
        {
                Size            datalen;
@@ -479,7 +483,7 @@ heap_xlog_insert(XLogReaderState *record)
                if (PageGetMaxOffsetNumber(page) + 1 < xlrec->offnum)
                        elog(PANIC, "invalid max offset number");
 
-               data = XLogRecGetBlockData(record, 0, &datalen);
+               data = XLogRecGetBlockData(record, HEAP_INSERT_BLKREF_HEAP, &datalen);
 
                newlen = datalen - SizeOfHeapHeader;
                Assert(datalen > SizeOfHeapHeader && newlen <= MaxHeapTupleSize);
@@ -559,7 +563,8 @@ heap_xlog_multi_insert(XLogReaderState *record)
         */
        xlrec = (xl_heap_multi_insert *) XLogRecGetData(record);
 
-       XLogRecGetBlockTag(record, 0, &rlocator, NULL, &blkno);
+       XLogRecGetBlockTag(record, HEAP_MULTI_INSERT_BLKREF_HEAP, &rlocator, NULL,
+                                          &blkno);
 
        /* check that the mutually exclusive flags are not both set */
        Assert(!((xlrec->flags & XLH_INSERT_ALL_VISIBLE_CLEARED) &&
@@ -582,13 +587,14 @@ heap_xlog_multi_insert(XLogReaderState *record)
 
        if (isinit)
        {
-               buffer = XLogInitBufferForRedo(record, 0);
+               buffer = XLogInitBufferForRedo(record, HEAP_MULTI_INSERT_BLKREF_HEAP);
                page = BufferGetPage(buffer);
                PageInit(page, BufferGetPageSize(buffer), 0);
                action = BLK_NEEDS_REDO;
        }
        else
-               action = XLogReadBufferForRedo(record, 0, &buffer);
+               action = XLogReadBufferForRedo(record, HEAP_MULTI_INSERT_BLKREF_HEAP,
+                                                                          &buffer);
        if (action == BLK_NEEDS_REDO)
        {
                char       *tupdata;
@@ -596,7 +602,8 @@ heap_xlog_multi_insert(XLogReaderState *record)
                Size            len;
 
                /* Tuples are stored as block data */
-               tupdata = XLogRecGetBlockData(record, 0, &len);
+               tupdata = XLogRecGetBlockData(record, HEAP_MULTI_INSERT_BLKREF_HEAP,
+                                                                         &len);
                endptr = tupdata + len;
 
                page = (Page) BufferGetPage(buffer);
@@ -713,8 +720,10 @@ heap_xlog_update(XLogReaderState *record, bool hot_update)
        oldtup.t_data = NULL;
        oldtup.t_len = 0;
 
-       XLogRecGetBlockTag(record, 0, &rlocator, NULL, &newblk);
-       if (XLogRecGetBlockTagExtended(record, 1, NULL, NULL, &oldblk, NULL))
+       XLogRecGetBlockTag(record, HEAP_UPDATE_BLKREF_HEAP_NEW, &rlocator, NULL,
+                                          &newblk);
+       if (XLogRecGetBlockTagExtended(record, HEAP_UPDATE_BLKREF_HEAP_OLD, NULL, NULL,
+                                                                  &oldblk, NULL))
        {
                /* HOT updates are never done across pages */
                Assert(!hot_update);
@@ -750,7 +759,8 @@ heap_xlog_update(XLogReaderState *record, bool hot_update)
         */
 
        /* Deal with old tuple version */
-       oldaction = XLogReadBufferForRedo(record, (oldblk == newblk) ? 0 : 1,
+       oldaction = XLogReadBufferForRedo(record, (oldblk == newblk) ?
+                                                                         HEAP_UPDATE_BLKREF_HEAP_NEW : HEAP_UPDATE_BLKREF_HEAP_OLD,
                                                                          &obuffer);
        if (oldaction == BLK_NEEDS_REDO)
        {
@@ -800,13 +810,14 @@ heap_xlog_update(XLogReaderState *record, bool hot_update)
        }
        else if (XLogRecGetInfo(record) & XLOG_HEAP_INIT_PAGE)
        {
-               nbuffer = XLogInitBufferForRedo(record, 0);
+               nbuffer = XLogInitBufferForRedo(record, HEAP_UPDATE_BLKREF_HEAP_NEW);
                page = (Page) BufferGetPage(nbuffer);
                PageInit(page, BufferGetPageSize(nbuffer), 0);
                newaction = BLK_NEEDS_REDO;
        }
        else
-               newaction = XLogReadBufferForRedo(record, 0, &nbuffer);
+               newaction = XLogReadBufferForRedo(record, HEAP_UPDATE_BLKREF_HEAP_NEW,
+                                                                                 &nbuffer);
 
        /*
         * The visibility map may need to be fixed even if the heap page is
@@ -831,7 +842,8 @@ heap_xlog_update(XLogReaderState *record, bool hot_update)
                Size            datalen;
                Size            tuplen;
 
-               recdata = XLogRecGetBlockData(record, 0, &datalen);
+               recdata = XLogRecGetBlockData(record, HEAP_UPDATE_BLKREF_HEAP_NEW,
+                                                                         &datalen);
                recdata_end = recdata + datalen;
 
                page = BufferGetPage(nbuffer);
@@ -1015,7 +1027,8 @@ heap_xlog_lock(XLogReaderState *record)
                BlockNumber block;
                Relation        reln;
 
-               XLogRecGetBlockTag(record, 0, &rlocator, NULL, &block);
+               XLogRecGetBlockTag(record, HEAP_LOCK_BLKREF_HEAP, &rlocator, NULL,
+                                                  &block);
                reln = CreateFakeRelcacheEntry(rlocator);
 
                visibilitymap_pin(reln, block, &vmbuffer);
@@ -1025,7 +1038,8 @@ heap_xlog_lock(XLogReaderState *record)
                FreeFakeRelcacheEntry(reln);
        }
 
-       if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
+       if (XLogReadBufferForRedo(record, HEAP_LOCK_BLKREF_HEAP,
+                                                         &buffer) == BLK_NEEDS_REDO)
        {
                page = (Page) BufferGetPage(buffer);
 
@@ -1091,7 +1105,8 @@ heap_xlog_lock_updated(XLogReaderState *record)
                BlockNumber block;
                Relation        reln;
 
-               XLogRecGetBlockTag(record, 0, &rlocator, NULL, &block);
+               XLogRecGetBlockTag(record, HEAP_LOCK_BLKREF_HEAP, &rlocator, NULL,
+                                                  &block);
                reln = CreateFakeRelcacheEntry(rlocator);
 
                visibilitymap_pin(reln, block, &vmbuffer);
@@ -1101,7 +1116,8 @@ heap_xlog_lock_updated(XLogReaderState *record)
                FreeFakeRelcacheEntry(reln);
        }
 
-       if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
+       if (XLogReadBufferForRedo(record, HEAP_LOCK_BLKREF_HEAP,
+                                                         &buffer) == BLK_NEEDS_REDO)
        {
                page = BufferGetPage(buffer);
 

diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h
index 277df6b3cf0b352dc423292cfcd9fa354a0cd8bf..92744c53899bc2a043cb103a6a437eeecd627adf 100644 (file)
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -110,6 +110,8 @@
        (XLH_DELETE_CONTAINS_OLD_TUPLE | XLH_DELETE_CONTAINS_OLD_KEY)
 
 /* This is what we need to know about delete */
+#define HEAP_DELETE_BLKREF_HEAP                0
+
 typedef struct xl_heap_delete
 {
        TransactionId xmax;                     /* xmax of the deleted tuple */
@@ -157,12 +159,14 @@ typedef struct xl_heap_header
 #define SizeOfHeapHeader       (offsetof(xl_heap_header, t_hoff) + sizeof(uint8))
 
 /* This is what we need to know about insert */
+#define HEAP_INSERT_BLKREF_HEAP                0
+
 typedef struct xl_heap_insert
 {
        OffsetNumber offnum;            /* inserted tuple's offset */
        uint8           flags;
 
-       /* xl_heap_header & TUPLE DATA in backup block 0 */
+       /* xl_heap_header & TUPLE DATA in HEAP_INSERT_BLKREF_HEAP */
 } xl_heap_insert;
 
 #define SizeOfHeapInsert       (offsetof(xl_heap_insert, flags) + sizeof(uint8))
@@ -173,10 +177,13 @@ typedef struct xl_heap_insert
  * The main data of the record consists of this xl_heap_multi_insert header.
  * 'offsets' array is omitted if the whole page is reinitialized
  * (XLOG_HEAP_INIT_PAGE).
- *
- * In block 0's data portion, there is an xl_multi_insert_tuple struct,
- * followed by the tuple data for each tuple. There is padding to align
- * each xl_multi_insert_tuple struct.
+ */
+#define HEAP_MULTI_INSERT_BLKREF_HEAP  0
+
+/*
+ * In HEAP_MULTI_INSERT_BLKREF_HEAP's data portion, there is an
+ * xl_multi_insert_tuple struct, followed by the tuple data for each tuple.
+ * There is padding to align each xl_multi_insert_tuple struct.
  */
 typedef struct xl_heap_multi_insert
 {
@@ -201,7 +208,7 @@ typedef struct xl_multi_insert_tuple
 /*
  * This is what we need to know about update|hot_update
  *
- * Backup blk 0: new page
+ * HEAP_UPDATE_BLKREF_HEAP_NEW: new page
  *
  * If XLH_UPDATE_PREFIX_FROM_OLD or XLH_UPDATE_SUFFIX_FROM_OLD flags are set,
  * the prefix and/or suffix come first, as one or two uint16s.
@@ -213,8 +220,13 @@ typedef struct xl_multi_insert_tuple
  * If XLH_UPDATE_CONTAINS_NEW_TUPLE flag is given, the tuple data is
  * included even if a full-page image was taken.
  *
- * Backup blk 1: old page, if different. (no data, just a reference to the blk)
+ * HEAP_UPDATE_BLKREF_HEAP_OLD: old page, if different. (no data, just a reference
+ * to the block)
  */
+
+#define HEAP_UPDATE_BLKREF_HEAP_NEW    0
+#define HEAP_UPDATE_BLKREF_HEAP_OLD    1
+
 typedef struct xl_heap_update
 {
        TransactionId old_xmax;         /* xmax of the old tuple */
@@ -393,6 +405,8 @@ typedef struct xlhp_prune_items
 #define XLH_LOCK_ALL_FROZEN_CLEARED            0x01
 
 /* This is what we need to know about lock */
+#define HEAP_LOCK_BLKREF_HEAP          0
+
 typedef struct xl_heap_lock
 {
        TransactionId xmax;                     /* might be a MultiXactId */