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 4f373b86028cde656b9f6c1a9821ddea0d01909a..4594e746c67fe954b94c0aed55e6bc20eebdaf5e 100644 (file)
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2149,10 +2149,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);
 
@@ -2573,11 +2575,13 @@ 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);
                        if (all_frozen_set)
-                               XLogRegisterBuffer(1, vmbuffer, 0);
+                               XLogRegisterBuffer(HEAP_MULTI_INSERT_BLKREF_VM, vmbuffer, 0);
 
-                       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);
@@ -3069,7 +3073,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
@@ -3844,7 +3848,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;
@@ -5183,7 +5187,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;
@@ -5935,7 +5939,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;
@@ -8901,9 +8905,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);
 
@@ -8916,15 +8920,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));
                }
        }
 
@@ -8938,10 +8945,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);
        }
@@ -8954,13 +8961,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 9ed7024e81474c1c06f80e203ad51f26a53eaf8c..963a886a1b366dd24d8cda0cb0d5f447a6295246 100644 (file)
--- a/src/backend/access/heap/heapam_xlog.c
+++ b/src/backend/access/heap/heapam_xlog.c
@@ -299,7 +299,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);
 
@@ -318,7 +319,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);
 
@@ -383,7 +385,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);
 
@@ -411,13 +414,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;
@@ -428,7 +432,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);
@@ -516,7 +520,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) &&
@@ -539,13 +544,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;
@@ -553,7 +559,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 = BufferGetPage(buffer);
@@ -728,8 +735,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);
@@ -765,7 +774,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)
        {
@@ -815,13 +825,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);
                npage = BufferGetPage(nbuffer);
                PageInit(npage, 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
@@ -846,7 +857,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;
 
                npage = BufferGetPage(nbuffer);
@@ -1033,7 +1045,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);
@@ -1043,7 +1056,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 = BufferGetPage(buffer);
 
@@ -1109,7 +1123,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);
@@ -1119,7 +1134,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 fdca7d821c87cdb415ebf023689fa929e61e7e39..fedee7f3501ddd90dcaa0c770246013d32467fa1 100644 (file)
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -112,6 +112,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 */
@@ -159,12 +161,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))
@@ -175,10 +179,14 @@ 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
+#define HEAP_MULTI_INSERT_BLKREF_VM            1
+
+/*
+ * 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
 {
@@ -203,7 +211,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.
@@ -215,8 +223,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 */
@@ -403,6 +416,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 */