heapam_xlog.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * heapam_xlog.c
 *    WAL replay logic for heap access method.
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *    src/backend/access/heap/heapam_xlog.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"
 
#include "access/bufmask.h"
#include "access/heapam.h"
#include "access/visibilitymap.h"
#include "access/xlog.h"
#include "access/xlogutils.h"
#include "storage/freespace.h"
#include "storage/standby.h"
 
/*
 * Clear visibility map bits for a single heap block during heap redo.
 *
 * Used by records that modify one heap block and, at most, its corresponding
 * VM block (insert, delete, multi_insert, lock).  Records that can touch
 * multiple heap or VM blocks (e.g. updates) replay the VM changes inline
 * instead.
 *
 * 'record' is the WAL record being replayed
 * 'target_locator' identifies the relation whose VM is being updated
 * 'heap_blkno' is the heap block whose VM bits should be cleared
 * 'wal_vm_block_id' is the WAL block reference id of the VM page
 * 'flags' specifies which visibility map bits to clear
 */
static void
heap_xlog_vm_clear(XLogReaderState *record,
                   RelFileLocator target_locator,
                   BlockNumber heap_blkno,
                   uint8 wal_vm_block_id, uint8 flags)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    Buffer      vmbuffer = InvalidBuffer;
 
    if (!XLogRecHasBlockRef(record, wal_vm_block_id))
        return;
 
    /*
     * If the vmbuffer was registered, use the recovery-specific routines to
     * read it. These will either apply an FPI or indicate that we should
     * clear the requested bits ourselves.
     */
    if (XLogReadBufferForRedo(record, wal_vm_block_id,
                              &vmbuffer) == BLK_NEEDS_REDO)
    {
        if (visibilitymap_clear(target_locator, heap_blkno, vmbuffer, flags))
            PageSetLSN(BufferGetPage(vmbuffer), lsn);
    }
    if (BufferIsValid(vmbuffer))
        UnlockReleaseBuffer(vmbuffer);
}
 
/*
 * Replay XLOG_HEAP2_PRUNE_* records.
 */
static void
heap_xlog_prune_freeze(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    char       *maindataptr = XLogRecGetData(record);
    xl_heap_prune xlrec;
    Buffer      buffer;
    RelFileLocator rlocator;
    BlockNumber blkno;
    Buffer      vmbuffer = InvalidBuffer;
    uint8       vmflags = 0;
    Size        freespace = 0;
    bool        do_update_fsm = false;
 
    XLogRecGetBlockTag(record, 0, &rlocator, NULL, &blkno);
    memcpy(&xlrec, maindataptr, SizeOfHeapPrune);
    maindataptr += SizeOfHeapPrune;
 
    /*
     * We will take an ordinary exclusive lock or a cleanup lock depending on
     * whether the XLHP_CLEANUP_LOCK flag is set.  With an ordinary exclusive
     * lock, we better not be doing anything that requires moving existing
     * tuple data.
     */
    Assert((xlrec.flags & XLHP_CLEANUP_LOCK) != 0 ||
           (xlrec.flags & (XLHP_HAS_REDIRECTIONS | XLHP_HAS_DEAD_ITEMS)) == 0);
 
    if (xlrec.flags & XLHP_VM_ALL_VISIBLE)
    {
        vmflags = VISIBILITYMAP_ALL_VISIBLE;
        if (xlrec.flags & XLHP_VM_ALL_FROZEN)
            vmflags |= VISIBILITYMAP_ALL_FROZEN;
    }
 
    /*
     * After xl_heap_prune is the optional snapshot conflict horizon.
     *
     * In Hot Standby mode, we must ensure that there are no running queries
     * which would conflict with the changes in this record. That means we
     * can't replay this record if it removes tuples that are still visible to
     * transactions on the standby, freeze tuples with xids that are still
     * considered running on the standby, or set a page as all-visible in the
     * VM if it isn't all-visible to all transactions on the standby.
     */
    if ((xlrec.flags & XLHP_HAS_CONFLICT_HORIZON) != 0)
    {
        TransactionId snapshot_conflict_horizon;
 
        /* memcpy() because snapshot_conflict_horizon is stored unaligned */
        memcpy(&snapshot_conflict_horizon, maindataptr, sizeof(TransactionId));
        maindataptr += sizeof(TransactionId);
 
        if (InHotStandby)
            ResolveRecoveryConflictWithSnapshot(snapshot_conflict_horizon,
                                                (xlrec.flags & XLHP_IS_CATALOG_REL) != 0,
                                                rlocator);
    }
 
    /*
     * If we have a full-page image of the heap block, restore it and we're
     * done with the heap block.
     */
    if (XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL,
                                      (xlrec.flags & XLHP_CLEANUP_LOCK) != 0,
                                      &buffer) == BLK_NEEDS_REDO)
    {
        Page        page = BufferGetPage(buffer);
        OffsetNumber *redirected;
        OffsetNumber *nowdead;
        OffsetNumber *nowunused;
        int         nredirected;
        int         ndead;
        int         nunused;
        int         nplans;
        Size        datalen;
        xlhp_freeze_plan *plans;
        OffsetNumber *frz_offsets;
        char       *dataptr = XLogRecGetBlockData(record, 0, &datalen);
        bool        do_prune;
 
        heap_xlog_deserialize_prune_and_freeze(dataptr, xlrec.flags,
                                               &nplans, &plans, &frz_offsets,
                                               &nredirected, &redirected,
                                               &ndead, &nowdead,
                                               &nunused, &nowunused);
 
        do_prune = nredirected > 0 || ndead > 0 || nunused > 0;
 
        /* Ensure the record does something */
        Assert(do_prune || nplans > 0 || vmflags & VISIBILITYMAP_VALID_BITS);
 
        /*
         * Update all line pointers per the record, and repair fragmentation
         * if needed.
         */
        if (do_prune)
            heap_page_prune_execute(buffer,
                                    (xlrec.flags & XLHP_CLEANUP_LOCK) == 0,
                                    redirected, nredirected,
                                    nowdead, ndead,
                                    nowunused, nunused);
 
        /* Freeze tuples */
        for (int p = 0; p < nplans; p++)
        {
            HeapTupleFreeze frz;
 
            /*
             * Convert freeze plan representation from WAL record into
             * per-tuple format used by heap_execute_freeze_tuple
             */
            frz.xmax = plans[p].xmax;
            frz.t_infomask2 = plans[p].t_infomask2;
            frz.t_infomask = plans[p].t_infomask;
            frz.frzflags = plans[p].frzflags;
            frz.offset = InvalidOffsetNumber;   /* unused, but be tidy */
 
            for (int i = 0; i < plans[p].ntuples; i++)
            {
                OffsetNumber offset = *(frz_offsets++);
                ItemId      lp;
                HeapTupleHeader tuple;
 
                lp = PageGetItemId(page, offset);
                tuple = (HeapTupleHeader) PageGetItem(page, lp);
                heap_execute_freeze_tuple(tuple, &frz);
            }
        }
 
        /* There should be no more data */
        Assert((char *) frz_offsets == dataptr + datalen);
 
        /*
         * The critical integrity requirement here is that we must never end
         * up with the visibility map bit set and the page-level
         * PD_ALL_VISIBLE bit unset.  If that were to occur, a subsequent page
         * modification would fail to clear the visibility map bit.
         */
        if (vmflags & VISIBILITYMAP_VALID_BITS)
        {
            PageSetAllVisible(page);
            PageClearPrunable(page);
        }
 
        MarkBufferDirty(buffer);
 
        /*
         * See log_heap_prune_and_freeze() for commentary on when we set the
         * heap page LSN.
         */
        if (do_prune || nplans > 0 ||
            ((vmflags & VISIBILITYMAP_VALID_BITS) && XLogHintBitIsNeeded()))
            PageSetLSN(page, lsn);
 
        /*
         * Note: we don't worry about updating the page's prunability hints.
         * At worst this will cause an extra prune cycle to occur soon.
         */
    }
 
    /*
     * If we 1) released any space or line pointers or 2) set PD_ALL_VISIBLE
     * or the VM, update the freespace map.
     *
     * Even when no actual space is freed (when only marking the page
     * all-visible or frozen), we still update the FSM. Because the FSM is
     * unlogged and maintained heuristically, it often becomes stale on
     * standbys. If such a standby is later promoted and runs VACUUM, it will
     * skip recalculating free space for pages that were marked
     * all-visible/all-frozen. FreeSpaceMapVacuum() can then propagate overly
     * optimistic free space values upward, causing future insertions to
     * select pages that turn out to be unusable. In bulk, this can lead to
     * long stalls.
     *
     * To prevent this, always update the FSM even when only marking a page
     * all-visible/all-frozen.
     *
     * Do this regardless of whether a full-page image is logged, since FSM
     * data is not part of the page itself.
     */
    if (BufferIsValid(buffer))
    {
        if ((xlrec.flags & (XLHP_HAS_REDIRECTIONS |
                            XLHP_HAS_DEAD_ITEMS |
                            XLHP_HAS_NOW_UNUSED_ITEMS)) ||
            (vmflags & VISIBILITYMAP_VALID_BITS))
        {
            freespace = PageGetHeapFreeSpace(BufferGetPage(buffer));
            do_update_fsm = true;
        }
 
        /*
         * We want to avoid holding an exclusive lock on the heap buffer while
         * doing IO (either of the FSM or the VM), so we'll release it now.
         */
        UnlockReleaseBuffer(buffer);
    }
 
    /*
     * Now read and update the VM block.
     *
     * We must redo changes to the VM even if the heap page was skipped due to
     * LSN interlock. See comment in heap_xlog_multi_insert() for more details
     * on replaying changes to the VM.
     */
    if ((vmflags & VISIBILITYMAP_VALID_BITS) &&
        XLogReadBufferForRedoExtended(record, 1,
                                      RBM_ZERO_ON_ERROR,
                                      false,
                                      &vmbuffer) == BLK_NEEDS_REDO)
    {
        Page        vmpage = BufferGetPage(vmbuffer);
 
        /* initialize the page if it was read as zeros */
        if (PageIsNew(vmpage))
            PageInit(vmpage, BLCKSZ, 0);
 
        visibilitymap_set(blkno, vmbuffer, vmflags, rlocator);
 
        Assert(BufferIsDirty(vmbuffer));
        PageSetLSN(vmpage, lsn);
    }
 
    if (BufferIsValid(vmbuffer))
        UnlockReleaseBuffer(vmbuffer);
 
    if (do_update_fsm)
        XLogRecordPageWithFreeSpace(rlocator, blkno, freespace);
}
 
/*
 * Given an "infobits" field from an XLog record, set the correct bits in the
 * given infomask and infomask2 for the tuple touched by the record.
 *
 * (This is the reverse of compute_infobits).
 */
static void
fix_infomask_from_infobits(uint8 infobits, uint16 *infomask, uint16 *infomask2)
{
    *infomask &= ~(HEAP_XMAX_IS_MULTI | HEAP_XMAX_LOCK_ONLY |
                   HEAP_XMAX_KEYSHR_LOCK | HEAP_XMAX_EXCL_LOCK);
    *infomask2 &= ~HEAP_KEYS_UPDATED;
 
    if (infobits & XLHL_XMAX_IS_MULTI)
        *infomask |= HEAP_XMAX_IS_MULTI;
    if (infobits & XLHL_XMAX_LOCK_ONLY)
        *infomask |= HEAP_XMAX_LOCK_ONLY;
    if (infobits & XLHL_XMAX_EXCL_LOCK)
        *infomask |= HEAP_XMAX_EXCL_LOCK;
    /* note HEAP_XMAX_SHR_LOCK isn't considered here */
    if (infobits & XLHL_XMAX_KEYSHR_LOCK)
        *infomask |= HEAP_XMAX_KEYSHR_LOCK;
 
    if (infobits & XLHL_KEYS_UPDATED)
        *infomask2 |= HEAP_KEYS_UPDATED;
}
 
/*
 * Replay XLOG_HEAP_DELETE records.
 */
static void
heap_xlog_delete(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_delete *xlrec = (xl_heap_delete *) XLogRecGetData(record);
    Buffer      buffer;
    Page        page;
    ItemId      lp;
    HeapTupleHeader htup;
    BlockNumber blkno;
    RelFileLocator target_locator;
    ItemPointerData target_tid;
 
    XLogRecGetBlockTag(record, HEAP_DELETE_BLKREF_HEAP, &target_locator, NULL,
                       &blkno);
    ItemPointerSetBlockNumber(&target_tid, blkno);
    ItemPointerSetOffsetNumber(&target_tid, xlrec->offnum);
 
    /*
     * The visibility map may need to be fixed even if the heap page is
     * already up-to-date.
     */
    if (xlrec->flags & XLH_DELETE_ALL_VISIBLE_CLEARED)
        heap_xlog_vm_clear(record, target_locator,
                           blkno, HEAP_DELETE_BLKREF_VM,
                           VISIBILITYMAP_VALID_BITS);
 
    if (XLogReadBufferForRedo(record, HEAP_DELETE_BLKREF_HEAP,
                              &buffer) == BLK_NEEDS_REDO)
    {
        page = BufferGetPage(buffer);
 
        if (xlrec->offnum < 1 || xlrec->offnum > PageGetMaxOffsetNumber(page))
            elog(PANIC, "offnum out of range");
        lp = PageGetItemId(page, xlrec->offnum);
        if (!ItemIdIsNormal(lp))
            elog(PANIC, "invalid lp");
 
        htup = (HeapTupleHeader) PageGetItem(page, lp);
 
        htup->t_infomask &= ~(HEAP_XMAX_BITS | HEAP_MOVED);
        htup->t_infomask2 &= ~HEAP_KEYS_UPDATED;
        HeapTupleHeaderClearHotUpdated(htup);
        fix_infomask_from_infobits(xlrec->infobits_set,
                                   &htup->t_infomask, &htup->t_infomask2);
        if (!(xlrec->flags & XLH_DELETE_IS_SUPER))
            HeapTupleHeaderSetXmax(htup, xlrec->xmax);
        else
            HeapTupleHeaderSetXmin(htup, InvalidTransactionId);
        HeapTupleHeaderSetCmax(htup, FirstCommandId, false);
 
        /* Mark the page as a candidate for pruning */
        PageSetPrunable(page, XLogRecGetXid(record));
 
        if (xlrec->flags & XLH_DELETE_ALL_VISIBLE_CLEARED)
            PageClearAllVisible(page);
 
        /* Make sure t_ctid is set correctly */
        if (xlrec->flags & XLH_DELETE_IS_PARTITION_MOVE)
            HeapTupleHeaderSetMovedPartitions(htup);
        else
            htup->t_ctid = target_tid;
        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))
        UnlockReleaseBuffer(buffer);
}
 
/*
 * Replay XLOG_HEAP_INSERT records.
 */
static void
heap_xlog_insert(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_insert *xlrec = (xl_heap_insert *) XLogRecGetData(record);
    Buffer      buffer;
    Page        page;
    union
    {
        HeapTupleHeaderData hdr;
        char        data[MaxHeapTupleSize];
    }           tbuf;
    HeapTupleHeader htup;
    xl_heap_header xlhdr;
    uint32      newlen;
    Size        freespace = 0;
    RelFileLocator target_locator;
    BlockNumber blkno;
    ItemPointerData target_tid;
    XLogRedoAction action;
 
    XLogRecGetBlockTag(record, HEAP_INSERT_BLKREF_HEAP, &target_locator, NULL,
                       &blkno);
    ItemPointerSetBlockNumber(&target_tid, blkno);
    ItemPointerSetOffsetNumber(&target_tid, xlrec->offnum);
 
    /* No freezing in the heap_insert() code path */
    Assert(!(xlrec->flags & XLH_INSERT_ALL_FROZEN_SET));
 
    /*
     * The visibility map may need to be fixed even if the heap page is
     * already up-to-date.
     */
    if (xlrec->flags & XLH_INSERT_ALL_VISIBLE_CLEARED)
        heap_xlog_vm_clear(record, target_locator,
                           blkno, HEAP_INSERT_BLKREF_VM,
                           VISIBILITYMAP_VALID_BITS);
 
    /*
     * If we inserted the first and only tuple on the page, re-initialize the
     * page from scratch.
     */
    if (XLogRecGetInfo(record) & XLOG_HEAP_INIT_PAGE)
    {
        buffer = XLogInitBufferForRedo(record, HEAP_INSERT_BLKREF_HEAP);
        page = BufferGetPage(buffer);
        PageInit(page, BufferGetPageSize(buffer), 0);
        action = BLK_NEEDS_REDO;
    }
    else
        action = XLogReadBufferForRedo(record, HEAP_INSERT_BLKREF_HEAP,
                                       &buffer);
    if (action == BLK_NEEDS_REDO)
    {
        Size        datalen;
        char       *data;
 
        page = BufferGetPage(buffer);
 
        if (PageGetMaxOffsetNumber(page) + 1 < xlrec->offnum)
            elog(PANIC, "invalid max offset number");
 
        data = XLogRecGetBlockData(record, HEAP_INSERT_BLKREF_HEAP, &datalen);
 
        newlen = datalen - SizeOfHeapHeader;
        Assert(datalen > SizeOfHeapHeader && newlen <= MaxHeapTupleSize);
        memcpy(&xlhdr, data, SizeOfHeapHeader);
        data += SizeOfHeapHeader;
 
        htup = &tbuf.hdr;
        MemSet(htup, 0, SizeofHeapTupleHeader);
        /* PG73FORMAT: get bitmap [+ padding] [+ oid] + data */
        memcpy((char *) htup + SizeofHeapTupleHeader,
               data,
               newlen);
        newlen += SizeofHeapTupleHeader;
        htup->t_infomask2 = xlhdr.t_infomask2;
        htup->t_infomask = xlhdr.t_infomask;
        htup->t_hoff = xlhdr.t_hoff;
        HeapTupleHeaderSetXmin(htup, XLogRecGetXid(record));
        HeapTupleHeaderSetCmin(htup, FirstCommandId);
        htup->t_ctid = target_tid;
 
        if (PageAddItem(page, htup, newlen, xlrec->offnum, true, true) == InvalidOffsetNumber)
            elog(PANIC, "failed to add tuple");
 
        freespace = PageGetHeapFreeSpace(page); /* needed to update FSM below */
 
        /*
         * Set the page prunable to trigger on-access pruning later, which may
         * set the page all-visible in the VM. See comments in heap_insert().
         */
        if (TransactionIdIsNormal(XLogRecGetXid(record)) &&
            !HeapTupleHeaderXminFrozen(htup))
            PageSetPrunable(page, XLogRecGetXid(record));
 
        PageSetLSN(page, lsn);
 
        if (xlrec->flags & XLH_INSERT_ALL_VISIBLE_CLEARED)
            PageClearAllVisible(page);
 
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))
        UnlockReleaseBuffer(buffer);
 
    /*
     * If the page is running low on free space, update the FSM as well.
     * Arbitrarily, our definition of "low" is less than 20%. We can't do much
     * better than that without knowing the fill-factor for the table.
     *
     * XXX: Don't do this if the page was restored from full page image. We
     * don't bother to update the FSM in that case, it doesn't need to be
     * totally accurate anyway.
     */
    if (action == BLK_NEEDS_REDO && freespace < BLCKSZ / 5)
        XLogRecordPageWithFreeSpace(target_locator, blkno, freespace);
}
 
/*
 * Replay XLOG_HEAP2_MULTI_INSERT records.
 */
static void
heap_xlog_multi_insert(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_multi_insert *xlrec;
    RelFileLocator rlocator;
    BlockNumber blkno;
    Buffer      buffer;
    Page        page;
    union
    {
        HeapTupleHeaderData hdr;
        char        data[MaxHeapTupleSize];
    }           tbuf;
    HeapTupleHeader htup;
    uint32      newlen;
    Size        freespace = 0;
    int         i;
    bool        isinit = (XLogRecGetInfo(record) & XLOG_HEAP_INIT_PAGE) != 0;
    XLogRedoAction action;
    Buffer      vmbuffer = InvalidBuffer;
 
    /*
     * Insertion doesn't overwrite MVCC data, so no conflict processing is
     * required.
     */
    xlrec = (xl_heap_multi_insert *) XLogRecGetData(record);
 
    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) &&
             (xlrec->flags & XLH_INSERT_ALL_FROZEN_SET)));
 
    /*
     * The visibility map may need to be fixed even if the heap page is
     * already up-to-date.
     *
     * Clear the VM (if needed) before clearing the heap page-level visibility
     * flag (PD_ALL_VISIBLE) to prevent the heap page from being marked
     * all-visible in the VM while its PD_ALL_VISIBLE is clear.
     */
    if (xlrec->flags & XLH_INSERT_ALL_VISIBLE_CLEARED)
        heap_xlog_vm_clear(record, rlocator,
                           blkno, HEAP_MULTI_INSERT_BLKREF_VM,
                           VISIBILITYMAP_VALID_BITS);
 
    if (isinit)
    {
        buffer = XLogInitBufferForRedo(record, HEAP_MULTI_INSERT_BLKREF_HEAP);
        page = BufferGetPage(buffer);
        PageInit(page, BufferGetPageSize(buffer), 0);
        action = BLK_NEEDS_REDO;
    }
    else
        action = XLogReadBufferForRedo(record, HEAP_MULTI_INSERT_BLKREF_HEAP,
                                       &buffer);
    if (action == BLK_NEEDS_REDO)
    {
        char       *tupdata;
        char       *endptr;
        Size        len;
 
        /* Tuples are stored as block data */
        tupdata = XLogRecGetBlockData(record, HEAP_MULTI_INSERT_BLKREF_HEAP,
                                      &len);
        endptr = tupdata + len;
 
        page = BufferGetPage(buffer);
 
        for (i = 0; i < xlrec->ntuples; i++)
        {
            OffsetNumber offnum;
            xl_multi_insert_tuple *xlhdr;
 
            /*
             * If we're reinitializing the page, the tuples are stored in
             * order from FirstOffsetNumber. Otherwise there's an array of
             * offsets in the WAL record, and the tuples come after that.
             */
            if (isinit)
                offnum = FirstOffsetNumber + i;
            else
                offnum = xlrec->offsets[i];
            if (PageGetMaxOffsetNumber(page) + 1 < offnum)
                elog(PANIC, "invalid max offset number");
 
            xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(tupdata);
            tupdata = ((char *) xlhdr) + SizeOfMultiInsertTuple;
 
            newlen = xlhdr->datalen;
            Assert(newlen <= MaxHeapTupleSize);
            htup = &tbuf.hdr;
            MemSet(htup, 0, SizeofHeapTupleHeader);
            /* PG73FORMAT: get bitmap [+ padding] [+ oid] + data */
            memcpy((char *) htup + SizeofHeapTupleHeader,
                   tupdata,
                   newlen);
            tupdata += newlen;
 
            newlen += SizeofHeapTupleHeader;
            htup->t_infomask2 = xlhdr->t_infomask2;
            htup->t_infomask = xlhdr->t_infomask;
            htup->t_hoff = xlhdr->t_hoff;
            HeapTupleHeaderSetXmin(htup, XLogRecGetXid(record));
            HeapTupleHeaderSetCmin(htup, FirstCommandId);
            ItemPointerSetBlockNumber(&htup->t_ctid, blkno);
            ItemPointerSetOffsetNumber(&htup->t_ctid, offnum);
 
            offnum = PageAddItem(page, htup, newlen, offnum, true, true);
            if (offnum == InvalidOffsetNumber)
                elog(PANIC, "failed to add tuple");
        }
        if (tupdata != endptr)
            elog(PANIC, "total tuple length mismatch");
 
        freespace = PageGetHeapFreeSpace(page); /* needed to update FSM below */
 
        PageSetLSN(page, lsn);
 
        if (xlrec->flags & XLH_INSERT_ALL_VISIBLE_CLEARED)
            PageClearAllVisible(page);
 
        /*
         * XLH_INSERT_ALL_FROZEN_SET implies that all tuples are visible. If
         * we are not setting the page frozen, then set the page's prunable
         * hint so that we trigger on-access pruning later which may set the
         * page all-visible in the VM.
         */
        if (xlrec->flags & XLH_INSERT_ALL_FROZEN_SET)
        {
            PageSetAllVisible(page);
            PageClearPrunable(page);
        }
        else
            PageSetPrunable(page, XLogRecGetXid(record));
 
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))
        UnlockReleaseBuffer(buffer);
 
    buffer = InvalidBuffer;
 
    /*
     * Read and update the visibility map (VM) block to set it frozen.
     *
     * We must always redo VM changes, even if the corresponding heap page
     * update was skipped due to the LSN interlock. Each VM block covers
     * multiple heap pages, so later WAL records may update other bits in the
     * same block. If this record includes an FPI (full-page image),
     * subsequent WAL records may depend on it to guard against torn pages.
     *
     * Heap page changes are replayed first to preserve the invariant:
     * PD_ALL_VISIBLE must be set on the heap page if the VM bit is set.
     *
     * Note that we released the heap page lock above. During normal
     * operation, this would be unsafe — a concurrent modification could
     * clear PD_ALL_VISIBLE while the VM bit remained set, violating the
     * invariant.
     *
     * During recovery, however, no concurrent writers exist. Therefore,
     * updating the VM without holding the heap page lock is safe enough. This
     * same approach is taken when replaying XLOG_HEAP2_PRUNE* records (see
     * heap_xlog_prune_freeze()).
     */
    if ((xlrec->flags & XLH_INSERT_ALL_FROZEN_SET) &&
        XLogReadBufferForRedoExtended(record, 1, RBM_ZERO_ON_ERROR, false,
                                      &vmbuffer) == BLK_NEEDS_REDO)
    {
        Page        vmpage = BufferGetPage(vmbuffer);
 
        /* initialize the page if it was read as zeros */
        if (PageIsNew(vmpage))
            PageInit(vmpage, BLCKSZ, 0);
 
        visibilitymap_set(blkno,
                          vmbuffer,
                          VISIBILITYMAP_ALL_VISIBLE |
                          VISIBILITYMAP_ALL_FROZEN,
                          rlocator);
 
        Assert(BufferIsDirty(vmbuffer));
        PageSetLSN(vmpage, lsn);
    }
 
    if (BufferIsValid(vmbuffer))
        UnlockReleaseBuffer(vmbuffer);
 
    /*
     * If the page is running low on free space, update the FSM as well.
     * Arbitrarily, our definition of "low" is less than 20%. We can't do much
     * better than that without knowing the fill-factor for the table.
     *
     * XXX: Don't do this if the page was restored from full page image. We
     * don't bother to update the FSM in that case, it doesn't need to be
     * totally accurate anyway.
     */
    if (action == BLK_NEEDS_REDO && freespace < BLCKSZ / 5)
        XLogRecordPageWithFreeSpace(rlocator, blkno, freespace);
}
 
/*
 * Replay XLOG_HEAP_UPDATE and XLOG_HEAP_HOT_UPDATE records.
 */
static void
heap_xlog_update(XLogReaderState *record, bool hot_update)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_update *xlrec = (xl_heap_update *) XLogRecGetData(record);
    RelFileLocator rlocator;
    BlockNumber oldblk;
    BlockNumber newblk;
    ItemPointerData newtid;
    Buffer      obuffer,
                nbuffer;
    Page        opage,
                npage;
    bool        has_vm_old,
                has_vm_new;
    OffsetNumber offnum;
    ItemId      lp;
    HeapTupleData oldtup;
    HeapTupleHeader htup;
    uint16      prefixlen = 0,
                suffixlen = 0;
    char       *newp;
    union
    {
        HeapTupleHeaderData hdr;
        char        data[MaxHeapTupleSize];
    }           tbuf;
    xl_heap_header xlhdr;
    uint32      newlen;
    Size        freespace = 0;
    XLogRedoAction oldaction;
    XLogRedoAction newaction;
 
    /* initialize to keep the compiler quiet */
    oldtup.t_data = NULL;
    oldtup.t_len = 0;
 
    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);
    }
    else
        oldblk = newblk;
 
    ItemPointerSet(&newtid, newblk, xlrec->new_offnum);
 
    /*
     * The visibility map may need to be fixed even if the heap page is
     * already up-to-date.
     */
    has_vm_old = XLogRecHasBlockRef(record, HEAP_UPDATE_BLKREF_VM_OLD);
    has_vm_new = XLogRecHasBlockRef(record, HEAP_UPDATE_BLKREF_VM_NEW);
 
    if (has_vm_new)
    {
        Buffer      vmbuffer_new = InvalidBuffer;
 
        Assert(xlrec->flags & XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED);
 
        if (XLogReadBufferForRedo(record, HEAP_UPDATE_BLKREF_VM_NEW,
                                  &vmbuffer_new) == BLK_NEEDS_REDO)
        {
            /*
             * If both the old and new heap pages were all-visible and their
             * VM bits are on the same VM page, that single VM page is
             * registered as HEAP_UPDATE_BLKREF_VM_NEW. Clear both heap
             * blocks' VM bits from the single provided VM buffer. It's
             * possible that one of the page's VM bits were already clear, but
             * visibilitymap_clear() is harmless as long as we provide it the
             * correct bits.
             *
             * We must verify that oldblk's VM bits really are on this VM
             * page, rather than relying on the absence of a separate VM_OLD
             * block reference: VM_OLD is also omitted when oldblk is on a
             * different VM page but its bit was already clear.
             */
            if (xlrec->flags & XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED &&
                visibilitymap_pin_ok(oldblk, vmbuffer_new))
            {
                if (visibilitymap_clear(rlocator, oldblk, vmbuffer_new,
                                        VISIBILITYMAP_VALID_BITS))
                    PageSetLSN(BufferGetPage(vmbuffer_new), lsn);
            }
            /* If VM_NEW is registered, we are sure newblk is on VM_NEW */
            if (visibilitymap_clear(rlocator, newblk, vmbuffer_new,
                                    VISIBILITYMAP_VALID_BITS))
                PageSetLSN(BufferGetPage(vmbuffer_new), lsn);
        }
        if (BufferIsValid(vmbuffer_new))
            UnlockReleaseBuffer(vmbuffer_new);
    }
    if (has_vm_old)
    {
        Buffer      vmbuffer_old = InvalidBuffer;
 
        Assert(xlrec->flags & XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED);
 
        if (XLogReadBufferForRedo(record, HEAP_UPDATE_BLKREF_VM_OLD,
                                  &vmbuffer_old) == BLK_NEEDS_REDO)
        {
            if (visibilitymap_clear(rlocator, oldblk, vmbuffer_old,
                                    VISIBILITYMAP_VALID_BITS))
                PageSetLSN(BufferGetPage(vmbuffer_old), lsn);
        }
        if (BufferIsValid(vmbuffer_old))
            UnlockReleaseBuffer(vmbuffer_old);
    }
 
    /*
     * In normal operation, it is important to lock the two pages in
     * page-number order, to avoid possible deadlocks against other update
     * operations going the other way.  However, during WAL replay there can
     * be no other update happening, so we don't need to worry about that.
     * Notice we also don't worry about this when locking VM buffers above.
     *
     * But we *do* need to worry that we don't expose an inconsistent state to
     * Hot Standby queries --- so the original page can't be unlocked before
     * we've added the new tuple to the new page.
     */
 
    /* Deal with old tuple version */
    oldaction = XLogReadBufferForRedo(record, (oldblk == newblk) ?
                                      HEAP_UPDATE_BLKREF_HEAP_NEW : HEAP_UPDATE_BLKREF_HEAP_OLD,
                                      &obuffer);
    if (oldaction == BLK_NEEDS_REDO)
    {
        opage = BufferGetPage(obuffer);
        offnum = xlrec->old_offnum;
        if (offnum < 1 || offnum > PageGetMaxOffsetNumber(opage))
            elog(PANIC, "offnum out of range");
        lp = PageGetItemId(opage, offnum);
        if (!ItemIdIsNormal(lp))
            elog(PANIC, "invalid lp");
 
        htup = (HeapTupleHeader) PageGetItem(opage, lp);
 
        oldtup.t_data = htup;
        oldtup.t_len = ItemIdGetLength(lp);
 
        htup->t_infomask &= ~(HEAP_XMAX_BITS | HEAP_MOVED);
        htup->t_infomask2 &= ~HEAP_KEYS_UPDATED;
        if (hot_update)
            HeapTupleHeaderSetHotUpdated(htup);
        else
            HeapTupleHeaderClearHotUpdated(htup);
        fix_infomask_from_infobits(xlrec->old_infobits_set, &htup->t_infomask,
                                   &htup->t_infomask2);
        HeapTupleHeaderSetXmax(htup, xlrec->old_xmax);
        HeapTupleHeaderSetCmax(htup, FirstCommandId, false);
        /* Set forward chain link in t_ctid */
        htup->t_ctid = newtid;
 
        /* Mark the page as a candidate for pruning */
        PageSetPrunable(opage, XLogRecGetXid(record));
 
        if (xlrec->flags & XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED)
            PageClearAllVisible(opage);
 
        PageSetLSN(opage, lsn);
        MarkBufferDirty(obuffer);
    }
 
    /*
     * Read the page the new tuple goes into, if different from old.
     */
    if (oldblk == newblk)
    {
        nbuffer = obuffer;
        newaction = oldaction;
    }
    else if (XLogRecGetInfo(record) & XLOG_HEAP_INIT_PAGE)
    {
        nbuffer = XLogInitBufferForRedo(record, HEAP_UPDATE_BLKREF_HEAP_NEW);
        npage = BufferGetPage(nbuffer);
        PageInit(npage, BufferGetPageSize(nbuffer), 0);
        newaction = BLK_NEEDS_REDO;
    }
    else
        newaction = XLogReadBufferForRedo(record, HEAP_UPDATE_BLKREF_HEAP_NEW,
                                          &nbuffer);
 
    /* Deal with new tuple */
    if (newaction == BLK_NEEDS_REDO)
    {
        char       *recdata;
        char       *recdata_end;
        Size        datalen;
        Size        tuplen;
 
        recdata = XLogRecGetBlockData(record, HEAP_UPDATE_BLKREF_HEAP_NEW,
                                      &datalen);
        recdata_end = recdata + datalen;
 
        npage = BufferGetPage(nbuffer);
 
        offnum = xlrec->new_offnum;
        if (PageGetMaxOffsetNumber(npage) + 1 < offnum)
            elog(PANIC, "invalid max offset number");
 
        if (xlrec->flags & XLH_UPDATE_PREFIX_FROM_OLD)
        {
            Assert(newblk == oldblk);
            memcpy(&prefixlen, recdata, sizeof(uint16));
            recdata += sizeof(uint16);
        }
        if (xlrec->flags & XLH_UPDATE_SUFFIX_FROM_OLD)
        {
            Assert(newblk == oldblk);
            memcpy(&suffixlen, recdata, sizeof(uint16));
            recdata += sizeof(uint16);
        }
 
        memcpy(&xlhdr, recdata, SizeOfHeapHeader);
        recdata += SizeOfHeapHeader;
 
        tuplen = recdata_end - recdata;
        Assert(tuplen <= MaxHeapTupleSize);
 
        htup = &tbuf.hdr;
        MemSet(htup, 0, SizeofHeapTupleHeader);
 
        /*
         * Reconstruct the new tuple using the prefix and/or suffix from the
         * old tuple, and the data stored in the WAL record.
         */
        newp = (char *) htup + SizeofHeapTupleHeader;
        if (prefixlen > 0)
        {
            int         len;
 
            /* copy bitmap [+ padding] [+ oid] from WAL record */
            len = xlhdr.t_hoff - SizeofHeapTupleHeader;
            memcpy(newp, recdata, len);
            recdata += len;
            newp += len;
 
            /* copy prefix from old tuple */
            memcpy(newp, (char *) oldtup.t_data + oldtup.t_data->t_hoff, prefixlen);
            newp += prefixlen;
 
            /* copy new tuple data from WAL record */
            len = tuplen - (xlhdr.t_hoff - SizeofHeapTupleHeader);
            memcpy(newp, recdata, len);
            recdata += len;
            newp += len;
        }
        else
        {
            /*
             * copy bitmap [+ padding] [+ oid] + data from record, all in one
             * go
             */
            memcpy(newp, recdata, tuplen);
            recdata += tuplen;
            newp += tuplen;
        }
        Assert(recdata == recdata_end);
 
        /* copy suffix from old tuple */
        if (suffixlen > 0)
            memcpy(newp, (char *) oldtup.t_data + oldtup.t_len - suffixlen, suffixlen);
 
        newlen = SizeofHeapTupleHeader + tuplen + prefixlen + suffixlen;
        htup->t_infomask2 = xlhdr.t_infomask2;
        htup->t_infomask = xlhdr.t_infomask;
        htup->t_hoff = xlhdr.t_hoff;
 
        HeapTupleHeaderSetXmin(htup, XLogRecGetXid(record));
        HeapTupleHeaderSetCmin(htup, FirstCommandId);
        HeapTupleHeaderSetXmax(htup, xlrec->new_xmax);
        /* Make sure there is no forward chain link in t_ctid */
        htup->t_ctid = newtid;
 
        offnum = PageAddItem(npage, htup, newlen, offnum, true, true);
        if (offnum == InvalidOffsetNumber)
            elog(PANIC, "failed to add tuple");
 
        if (xlrec->flags & XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED)
            PageClearAllVisible(npage);
 
        /* needed to update FSM below */
        freespace = PageGetHeapFreeSpace(npage);
 
        PageSetLSN(npage, lsn);
        /* See heap_insert() for why we set pd_prune_xid on insert */
        PageSetPrunable(npage, XLogRecGetXid(record));
        MarkBufferDirty(nbuffer);
    }
 
    if (BufferIsValid(nbuffer) && nbuffer != obuffer)
        UnlockReleaseBuffer(nbuffer);
    if (BufferIsValid(obuffer))
        UnlockReleaseBuffer(obuffer);
 
    /*
     * If the new page is running low on free space, update the FSM as well.
     * Arbitrarily, our definition of "low" is less than 20%. We can't do much
     * better than that without knowing the fill-factor for the table.
     *
     * However, don't update the FSM on HOT updates, because after crash
     * recovery, either the old or the new tuple will certainly be dead and
     * prunable. After pruning, the page will have roughly as much free space
     * as it did before the update, assuming the new tuple is about the same
     * size as the old one.
     *
     * XXX: Don't do this if the page was restored from full page image. We
     * don't bother to update the FSM in that case, it doesn't need to be
     * totally accurate anyway.
     */
    if (newaction == BLK_NEEDS_REDO && !hot_update && freespace < BLCKSZ / 5)
        XLogRecordPageWithFreeSpace(rlocator, newblk, freespace);
}
 
/*
 * Replay XLOG_HEAP_CONFIRM records.
 */
static void
heap_xlog_confirm(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_confirm *xlrec = (xl_heap_confirm *) XLogRecGetData(record);
    Buffer      buffer;
    Page        page;
    OffsetNumber offnum;
    ItemId      lp;
    HeapTupleHeader htup;
 
    if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
    {
        page = BufferGetPage(buffer);
 
        offnum = xlrec->offnum;
        if (offnum < 1 || offnum > PageGetMaxOffsetNumber(page))
            elog(PANIC, "offnum out of range");
        lp = PageGetItemId(page, offnum);
        if (!ItemIdIsNormal(lp))
            elog(PANIC, "invalid lp");
 
        htup = (HeapTupleHeader) PageGetItem(page, lp);
 
        /*
         * Confirm tuple as actually inserted
         */
        ItemPointerSet(&htup->t_ctid, BufferGetBlockNumber(buffer), offnum);
 
        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))
        UnlockReleaseBuffer(buffer);
}
 
/*
 * Replay XLOG_HEAP_LOCK records.
 */
static void
heap_xlog_lock(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_lock *xlrec = (xl_heap_lock *) XLogRecGetData(record);
    Buffer      buffer;
    Page        page;
    OffsetNumber offnum;
    ItemId      lp;
    HeapTupleHeader htup;
 
    /*
     * The visibility map may need to be fixed even if the heap page is
     * already up-to-date.
     */
    if (xlrec->flags & XLH_LOCK_ALL_FROZEN_CLEARED)
    {
        RelFileLocator rlocator;
        BlockNumber block;
 
        XLogRecGetBlockTag(record, HEAP_LOCK_BLKREF_HEAP, &rlocator, NULL,
                           &block);
 
        heap_xlog_vm_clear(record, rlocator,
                           block, HEAP_LOCK_BLKREF_VM,
                           VISIBILITYMAP_ALL_FROZEN);
    }
 
    if (XLogReadBufferForRedo(record, HEAP_LOCK_BLKREF_HEAP,
                              &buffer) == BLK_NEEDS_REDO)
    {
        page = BufferGetPage(buffer);
 
        offnum = xlrec->offnum;
        if (offnum < 1 || offnum > PageGetMaxOffsetNumber(page))
            elog(PANIC, "offnum out of range");
        lp = PageGetItemId(page, offnum);
        if (!ItemIdIsNormal(lp))
            elog(PANIC, "invalid lp");
 
        htup = (HeapTupleHeader) PageGetItem(page, lp);
 
        htup->t_infomask &= ~(HEAP_XMAX_BITS | HEAP_MOVED);
        htup->t_infomask2 &= ~HEAP_KEYS_UPDATED;
        fix_infomask_from_infobits(xlrec->infobits_set, &htup->t_infomask,
                                   &htup->t_infomask2);
 
        /*
         * Clear relevant update flags, but only if the modified infomask says
         * there's no update.
         */
        if (HEAP_XMAX_IS_LOCKED_ONLY(htup->t_infomask))
        {
            HeapTupleHeaderClearHotUpdated(htup);
            /* Make sure there is no forward chain link in t_ctid */
            ItemPointerSet(&htup->t_ctid,
                           BufferGetBlockNumber(buffer),
                           offnum);
        }
        HeapTupleHeaderSetXmax(htup, xlrec->xmax);
        HeapTupleHeaderSetCmax(htup, FirstCommandId, false);
        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))
        UnlockReleaseBuffer(buffer);
}
 
/*
 * Replay XLOG_HEAP2_LOCK_UPDATED records.
 */
static void
heap_xlog_lock_updated(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_lock_updated *xlrec;
    Buffer      buffer;
    Page        page;
    OffsetNumber offnum;
    ItemId      lp;
    HeapTupleHeader htup;
 
    xlrec = (xl_heap_lock_updated *) XLogRecGetData(record);
 
    /*
     * The visibility map may need to be fixed even if the heap page is
     * already up-to-date.
     */
    if (xlrec->flags & XLH_LOCK_ALL_FROZEN_CLEARED)
    {
        RelFileLocator rlocator;
        BlockNumber block;
 
        XLogRecGetBlockTag(record, HEAP_LOCK_BLKREF_HEAP, &rlocator, NULL,
                           &block);
 
        heap_xlog_vm_clear(record, rlocator,
                           block, HEAP_LOCK_BLKREF_VM,
                           VISIBILITYMAP_ALL_FROZEN);
    }
 
    if (XLogReadBufferForRedo(record, HEAP_LOCK_BLKREF_HEAP,
                              &buffer) == BLK_NEEDS_REDO)
    {
        page = BufferGetPage(buffer);
 
        offnum = xlrec->offnum;
        if (offnum < 1 || offnum > PageGetMaxOffsetNumber(page))
            elog(PANIC, "offnum out of range");
        lp = PageGetItemId(page, offnum);
        if (!ItemIdIsNormal(lp))
            elog(PANIC, "invalid lp");
 
        htup = (HeapTupleHeader) PageGetItem(page, lp);
 
        htup->t_infomask &= ~(HEAP_XMAX_BITS | HEAP_MOVED);
        htup->t_infomask2 &= ~HEAP_KEYS_UPDATED;
        fix_infomask_from_infobits(xlrec->infobits_set, &htup->t_infomask,
                                   &htup->t_infomask2);
        HeapTupleHeaderSetXmax(htup, xlrec->xmax);
 
        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))
        UnlockReleaseBuffer(buffer);
}
 
/*
 * Replay XLOG_HEAP_INPLACE records.
 */
static void
heap_xlog_inplace(XLogReaderState *record)
{
    XLogRecPtr  lsn = record->EndRecPtr;
    xl_heap_inplace *xlrec = (xl_heap_inplace *) XLogRecGetData(record);
    Buffer      buffer;
    Page        page;
    OffsetNumber offnum;
    ItemId      lp;
    HeapTupleHeader htup;
    uint32      oldlen;
    Size        newlen;
 
    if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)
    {
        char       *newtup = XLogRecGetBlockData(record, 0, &newlen);
 
        page = BufferGetPage(buffer);
 
        offnum = xlrec->offnum;
        if (offnum < 1 || offnum > PageGetMaxOffsetNumber(page))
            elog(PANIC, "offnum out of range");
        lp = PageGetItemId(page, offnum);
        if (!ItemIdIsNormal(lp))
            elog(PANIC, "invalid lp");
 
        htup = (HeapTupleHeader) PageGetItem(page, lp);
 
        oldlen = ItemIdGetLength(lp) - htup->t_hoff;
        if (oldlen != newlen)
            elog(PANIC, "wrong tuple length");
 
        memcpy((char *) htup + htup->t_hoff, newtup, newlen);
 
        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))
        UnlockReleaseBuffer(buffer);
 
    ProcessCommittedInvalidationMessages(xlrec->msgs,
                                         xlrec->nmsgs,
                                         xlrec->relcacheInitFileInval,
                                         xlrec->dbId,
                                         xlrec->tsId);
}
 
void
heap_redo(XLogReaderState *record)
{
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    /*
     * These operations don't overwrite MVCC data so no conflict processing is
     * required. The ones in heap2 rmgr do.
     */
 
    switch (info & XLOG_HEAP_OPMASK)
    {
        case XLOG_HEAP_INSERT:
            heap_xlog_insert(record);
            break;
        case XLOG_HEAP_DELETE:
            heap_xlog_delete(record);
            break;
        case XLOG_HEAP_UPDATE:
            heap_xlog_update(record, false);
            break;
        case XLOG_HEAP_TRUNCATE:
 
            /*
             * TRUNCATE is a no-op because the actions are already logged as
             * SMGR WAL records.  TRUNCATE WAL record only exists for logical
             * decoding.
             */
            break;
        case XLOG_HEAP_HOT_UPDATE:
            heap_xlog_update(record, true);
            break;
        case XLOG_HEAP_CONFIRM:
            heap_xlog_confirm(record);
            break;
        case XLOG_HEAP_LOCK:
            heap_xlog_lock(record);
            break;
        case XLOG_HEAP_INPLACE:
            heap_xlog_inplace(record);
            break;
        default:
            elog(PANIC, "heap_redo: unknown op code %u", info);
    }
}
 
void
heap2_redo(XLogReaderState *record)
{
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    switch (info & XLOG_HEAP_OPMASK)
    {
        case XLOG_HEAP2_PRUNE_ON_ACCESS:
        case XLOG_HEAP2_PRUNE_VACUUM_SCAN:
        case XLOG_HEAP2_PRUNE_VACUUM_CLEANUP:
            heap_xlog_prune_freeze(record);
            break;
        case XLOG_HEAP2_MULTI_INSERT:
            heap_xlog_multi_insert(record);
            break;
        case XLOG_HEAP2_LOCK_UPDATED:
            heap_xlog_lock_updated(record);
            break;
        case XLOG_HEAP2_NEW_CID:
 
            /*
             * Nothing to do on a real replay, only used during logical
             * decoding.
             */
            break;
        case XLOG_HEAP2_REWRITE:
            heap_xlog_logical_rewrite(record);
            break;
        default:
            elog(PANIC, "heap2_redo: unknown op code %u", info);
    }
}
 
/*
 * Mask a heap page before performing consistency checks on it.
 */
void
heap_mask(char *pagedata, BlockNumber blkno)
{
    Page        page = (Page) pagedata;
    OffsetNumber off;
 
    mask_page_lsn_and_checksum(page);
 
    mask_page_hint_bits(page);
    mask_unused_space(page);
 
    for (off = 1; off <= PageGetMaxOffsetNumber(page); off++)
    {
        ItemId      iid = PageGetItemId(page, off);
        char       *page_item;
 
        page_item = (char *) (page + ItemIdGetOffset(iid));
 
        if (ItemIdIsNormal(iid))
        {
            HeapTupleHeader page_htup = (HeapTupleHeader) page_item;
 
            /*
             * If xmin of a tuple is not yet frozen, we should ignore
             * differences in hint bits, since they can be set without
             * emitting WAL.
             */
            if (!HeapTupleHeaderXminFrozen(page_htup))
                page_htup->t_infomask &= ~HEAP_XACT_MASK;
            else
            {
                /* Still we need to mask xmax hint bits. */
                page_htup->t_infomask &= ~HEAP_XMAX_INVALID;
                page_htup->t_infomask &= ~HEAP_XMAX_COMMITTED;
            }
 
            /*
             * During replay, we set Command Id to FirstCommandId. Hence, mask
             * it. See heap_xlog_insert() for details.
             */
            page_htup->t_choice.t_heap.t_field3.t_cid = MASK_MARKER;
 
            /*
             * For a speculative tuple, heap_insert() does not set ctid in the
             * caller-passed heap tuple itself, leaving the ctid field to
             * contain a speculative token value - a per-backend monotonically
             * increasing identifier. Besides, it does not WAL-log ctid under
             * any circumstances.
             *
             * During redo, heap_xlog_insert() sets t_ctid to current block
             * number and self offset number. It doesn't care about any
             * speculative insertions on the primary. Hence, we set t_ctid to
             * current block number and self offset number to ignore any
             * inconsistency.
             */
            if (HeapTupleHeaderIsSpeculative(page_htup))
                ItemPointerSet(&page_htup->t_ctid, blkno, off);
 
            /*
             * NB: Not ignoring ctid changes due to the tuple having moved
             * (i.e. HeapTupleHeaderIndicatesMovedPartitions), because that's
             * important information that needs to be in-sync between primary
             * and standby, and thus is WAL logged.
             */
        }
 
        /*
         * Ignore any padding bytes after the tuple, when the length of the
         * item is not MAXALIGNed.
         */
        if (ItemIdHasStorage(iid))
        {
            int         len = ItemIdGetLength(iid);
            int         padlen = MAXALIGN(len) - len;
 
            if (padlen > 0)
                memset(page_item + len, MASK_MARKER, padlen);
        }
    }
}