patches/neon-pg17.11-consumers.patch
Rama main · réplica de lectura del repositorio original.
From 627583b85d7c0f624245e3ac6a240e775a5eed01 Mon Sep 17 00:00:00 2001
From: Hakopod-Acceptance <acceptance@invalid>
Date: Fri, 2 Oct 2026 19:34:34 +0000
Subject: [PATCH] Support PostgreSQL 17.11 visibility map WAL
---
libs/postgres_ffi/src/walrecord.rs | 4 +
libs/wal_decoder/src/decoder.rs | 104 +++-
pgxn/neon_rmgr/neon_rmgr.c | 839 ++++++++++++++++++++++++++++-
3 files changed, 920 insertions(+), 27 deletions(-)
diff --git a/libs/postgres_ffi/src/walrecord.rs b/libs/postgres_ffi/src/walrecord.rs
index 7ed07b0e7..3da169a17 100644
--- a/libs/postgres_ffi/src/walrecord.rs
+++ b/libs/postgres_ffi/src/walrecord.rs
@@ -145,6 +145,9 @@ pub fn decode(buf: &mut Bytes) -> MultiXactMember {
/// DecodedBkpBlock represents per-page data contained in a WAL record.
#[derive(Default)]
pub struct DecodedBkpBlock {
+ /* WAL block reference ID, which may be sparse. */
+ pub id: u8,
+
/* Is this block ref in use? */
//in_use: bool,
@@ -311,6 +314,7 @@ pub fn decode_wal_record(
0..=pg_constants::XLR_MAX_BLOCK_ID => {
/* XLogRecordBlockHeader */
let mut blk = DecodedBkpBlock::new();
+ blk.id = block_id;
if block_id <= max_block_id {
// TODO
diff --git a/libs/wal_decoder/src/decoder.rs b/libs/wal_decoder/src/decoder.rs
index 0843eb35b..4584538ba 100644
--- a/libs/wal_decoder/src/decoder.rs
+++ b/libs/wal_decoder/src/decoder.rs
@@ -69,6 +69,32 @@ pub fn from_bytes_filtered(
}
impl MetadataRecord {
+ fn heap_block_by_id(decoded: &DecodedWALRecord, id: u8) -> anyhow::Result<&DecodedBkpBlock> {
+ decoded
+ .blocks
+ .iter()
+ .find(|block| block.id == id)
+ .ok_or_else(|| anyhow::anyhow!("WAL record is missing heap block reference {id}"))
+ }
+
+ fn has_registered_vm_block(decoded: &DecodedWALRecord) -> bool {
+ decoded
+ .blocks
+ .iter()
+ .any(|block| block.forknum == VISIBILITYMAP_FORKNUM)
+ }
+
+ fn old_update_heap_block(decoded: &DecodedWALRecord) -> anyhow::Result<&DecodedBkpBlock> {
+ // Update block 0 is always the new heap page. Block 1 is the old heap
+ // page only when the update crosses pages. PG17.11 can append sparse VM
+ // references 2 and 3, so vector position and last() are not block IDs.
+ Ok(decoded
+ .blocks
+ .iter()
+ .find(|block| block.id == 1)
+ .unwrap_or(Self::heap_block_by_id(decoded, 0)?))
+ }
+
/// Populates the given `shard_records` with metadata records from this WAL record, if any,
/// discarding those belonging to other shards.
///
@@ -429,7 +455,7 @@ fn decode_heapam_record(
// we can't validate the remaining number of bytes without parsing
// the tuple data.
if (xlrec.flags & pg_constants::XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0 {
- old_heap_blkno = Some(decoded.blocks.last().unwrap().blkno);
+ old_heap_blkno = Some(Self::old_update_heap_block(decoded)?.blkno);
}
if (xlrec.flags & pg_constants::XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0 {
// PostgreSQL only uses XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED on a
@@ -475,6 +501,15 @@ fn decode_heapam_record(
}
}
+ // PostgreSQL 17.11 WAL includes modified VM pages as block references.
+ // Those pages are ingested in the regular block batch below. Emitting the
+ // legacy synthetic clear as well can apply a second same-LSN operation and,
+ // in the FPI case, race the authoritative page image. Keep the synthetic
+ // record only for legacy WAL that has no registered VM block.
+ if pg_version == PgMajorVersion::PG17 && Self::has_registered_vm_block(decoded) {
+ return Ok(None);
+ }
+
if new_heap_blkno.is_some() || old_heap_blkno.is_some() {
let vm_rel = RelTag {
forknum: VISIBILITYMAP_FORKNUM,
@@ -537,7 +572,7 @@ fn decode_neonmgr_record(
// we can't validate the remaining number of bytes without parsing
// the tuple data.
if (xlrec.flags & pg_constants::XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0 {
- old_heap_blkno = Some(decoded.blocks.last().unwrap().blkno);
+ old_heap_blkno = Some(Self::old_update_heap_block(decoded)?.blkno);
}
if (xlrec.flags & pg_constants::XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0 {
// PostgreSQL only uses XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED on a
@@ -579,6 +614,15 @@ fn decode_neonmgr_record(
),
}
+ // PostgreSQL 17.11 WAL includes modified VM pages as block references.
+ // Those pages are ingested in the regular block batch below. Emitting the
+ // legacy synthetic clear as well can apply a second same-LSN operation and,
+ // in the FPI case, race the authoritative page image. Keep the synthetic
+ // record only for legacy WAL that has no registered VM block.
+ if pg_version == PgMajorVersion::PG17 && Self::has_registered_vm_block(decoded) {
+ return Ok(None);
+ }
+
if new_heap_blkno.is_some() || old_heap_blkno.is_some() {
let vm_rel = RelTag {
forknum: VISIBILITYMAP_FORKNUM,
@@ -985,3 +1029,59 @@ fn decode_replorigin_record(
Ok(None)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn block(id: u8, blkno: u32) -> DecodedBkpBlock {
+ let mut block = DecodedBkpBlock::default();
+ block.id = id;
+ block.blkno = blkno;
+ block
+ }
+
+ #[test]
+ fn update_old_heap_block_uses_reference_one_before_vm_references() {
+ let decoded = DecodedWALRecord {
+ blocks: vec![block(0, 10), block(1, 20), block(2, 900), block(3, 901)],
+ ..Default::default()
+ };
+
+ assert_eq!(
+ MetadataRecord::old_update_heap_block(&decoded)
+ .unwrap()
+ .blkno,
+ 20
+ );
+ }
+
+ #[test]
+ fn update_old_heap_block_uses_reference_zero_for_same_page_update() {
+ let decoded = DecodedWALRecord {
+ // Reference 1 is absent for a same-page update. The VM reference is
+ // sparse and must not be mistaken for the old heap page.
+ blocks: vec![block(0, 10), block(2, 900)],
+ ..Default::default()
+ };
+
+ assert_eq!(
+ MetadataRecord::old_update_heap_block(&decoded)
+ .unwrap()
+ .blkno,
+ 10
+ );
+ }
+
+ #[test]
+ fn registered_vm_blocks_are_detected_by_fork_not_position() {
+ let mut vm = block(3, 901);
+ vm.forknum = VISIBILITYMAP_FORKNUM;
+ let decoded = DecodedWALRecord {
+ blocks: vec![block(0, 10), block(1, 20), vm],
+ ..Default::default()
+ };
+
+ assert!(MetadataRecord::has_registered_vm_block(&decoded));
+ }
+}
diff --git a/pgxn/neon_rmgr/neon_rmgr.c b/pgxn/neon_rmgr/neon_rmgr.c
index c3f726db8..48801dcc8 100644
--- a/pgxn/neon_rmgr/neon_rmgr.c
+++ b/pgxn/neon_rmgr/neon_rmgr.c
@@ -205,9 +205,170 @@ fix_infomask_from_infobits(uint8 infobits, uint16 *infomask, uint16 *infomask2)
*infomask2 |= HEAP_KEYS_UPDATED;
}
+#if PG_VERSION_NUM >= 170011
+static void
+neon_heap_xlog_vm_clear(XLogReaderState *record,
+ RelFileLocator target_locator,
+ BlockNumber heap_blkno,
+ uint8 wal_vm_block_id, uint8 flags)
+{
+ XLogRecPtr lsn = record->EndRecPtr;
+ Relation reln = CreateFakeRelcacheEntry(target_locator);
+ Buffer vmbuffer = InvalidBuffer;
+
+ /*
+ * 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.
+ *
+ * Originally, clearing the VM did not register the VM buffers, so since
+ * registering the VM buffer was a bug fix, we keep a fallback path to
+ * support replay of WAL generated from before the fix.
+ */
+ if (XLogRecHasBlockRef(record, wal_vm_block_id))
+ {
+ if (XLogReadBufferForRedo(record, wal_vm_block_id,
+ &vmbuffer) == BLK_NEEDS_REDO)
+ {
+ if (visibilitymap_clear_locked(reln,
+ heap_blkno, vmbuffer,
+ flags))
+ PageSetLSN(BufferGetPage(vmbuffer), lsn);
+ }
+ if (BufferIsValid(vmbuffer))
+ UnlockReleaseBuffer(vmbuffer);
+ }
+ else
+ {
+ /*
+ * This is the backwards compatibility path to clear VM bits for
+ * records predating VM buffer registration. It is also invoked if the
+ * heap page's PD_ALL_VISIBLE was cleared but the VM bits were already
+ * clear. The WAL record flags do not distinguish between these two
+ * situations. Though this is wasted effort, the behavior is
+ * historical and the situation should be rare.
+ */
+ visibilitymap_pin(reln, heap_blkno, &vmbuffer);
+ visibilitymap_clear(reln, heap_blkno, vmbuffer, flags);
+ ReleaseBuffer(vmbuffer);
+ }
+
+ FreeFakeRelcacheEntry(reln);
+}
+#endif
+
+
static void
redo_neon_heap_insert(XLogReaderState *record)
{
+#if PG_VERSION_NUM >= 170011
+ XLogRecPtr lsn = record->EndRecPtr;
+ xl_neon_heap_insert *xlrec = (xl_neon_heap_insert *) XLogRecGetData(record);
+ Buffer buffer;
+ Page page;
+ union
+ {
+ HeapTupleHeaderData hdr;
+ char data[MaxHeapTupleSize];
+ } tbuf;
+ HeapTupleHeader htup;
+ xl_neon_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)
+ neon_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, "neon_rm_redo: invalid max offset number");
+
+ data = XLogRecGetBlockData(record, HEAP_INSERT_BLKREF_HEAP, &datalen);
+
+ newlen = datalen - SizeOfNeonHeapHeader;
+ Assert(datalen > SizeOfNeonHeapHeader && newlen <= MaxHeapTupleSize);
+ memcpy((char *) &xlhdr, data, SizeOfNeonHeapHeader);
+ data += SizeOfNeonHeapHeader;
+
+ htup = &tbuf.hdr;
+ MemSet((char *) 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));
+ htup->t_choice.t_heap.t_field3.t_cid = xlhdr.t_cid;
+ htup->t_ctid = target_tid;
+
+ if (PageAddItem(page, (Item) htup, newlen, xlrec->offnum,
+ true, true) == InvalidOffsetNumber)
+ elog(PANIC, "neon_rm_redo: failed to add tuple");
+
+ freespace = PageGetHeapFreeSpace(page); /* needed to update FSM below */
+
+ 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);
+
+#else
XLogRecPtr lsn = record->EndRecPtr;
xl_neon_heap_insert *xlrec = (xl_neon_heap_insert *) XLogRecGetData(record);
Buffer buffer;
@@ -320,11 +481,79 @@ redo_neon_heap_insert(XLogReaderState *record)
*/
if (action == BLK_NEEDS_REDO && freespace < BLCKSZ / 5)
XLogRecordPageWithFreeSpace(target_locator, blkno, freespace);
+
+#endif
}
static void
redo_neon_heap_delete(XLogReaderState *record)
{
+#if PG_VERSION_NUM >= 170011
+ XLogRecPtr lsn = record->EndRecPtr;
+ xl_neon_heap_delete *xlrec = (xl_neon_heap_delete *) XLogRecGetData(record);
+ Buffer buffer;
+ Page page;
+ ItemId lp = NULL;
+ 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)
+ neon_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 (PageGetMaxOffsetNumber(page) >= xlrec->offnum)
+ lp = PageGetItemId(page, xlrec->offnum);
+
+ if (PageGetMaxOffsetNumber(page) < xlrec->offnum || !ItemIdIsNormal(lp))
+ elog(PANIC, "neon_rm_redo: 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);
+ htup->t_choice.t_heap.t_field3.t_cid = xlrec->t_cid;
+
+ /* 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);
+
+#else
XLogRecPtr lsn = record->EndRecPtr;
xl_neon_heap_delete *xlrec = (xl_neon_heap_delete *) XLogRecGetData(record);
Buffer buffer;
@@ -366,38 +595,385 @@ redo_neon_heap_delete(XLogReaderState *record)
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);
- htup->t_choice.t_heap.t_field3.t_cid = xlrec->t_cid;
+ 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);
+ htup->t_choice.t_heap.t_field3.t_cid = xlrec->t_cid;
+
+ /* 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);
+
+#endif
+}
+
+static void
+redo_neon_heap_update(XLogReaderState *record, bool hot_update)
+{
+#if PG_VERSION_NUM >= 170011
+ XLogRecPtr lsn = record->EndRecPtr;
+ xl_neon_heap_update *xlrec = (xl_neon_heap_update *) XLogRecGetData(record);
+ RelFileLocator rlocator;
+ BlockNumber oldblk;
+ BlockNumber newblk;
+ ItemPointerData newtid;
+ Buffer obuffer,
+ nbuffer;
+ Page page;
+ bool new_cleared,
+ old_cleared;
+ OffsetNumber offnum;
+ ItemId lp = NULL;
+ HeapTupleData oldtup;
+ HeapTupleHeader htup;
+ uint16 prefixlen = 0,
+ suffixlen = 0;
+ char *newp;
+ union
+ {
+ HeapTupleHeaderData hdr;
+ char data[MaxHeapTupleSize];
+ } tbuf;
+ xl_neon_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.
+ */
+ new_cleared = (xlrec->flags & XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0;
+ old_cleared = (xlrec->flags & XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0;
+ if (new_cleared || old_cleared)
+ {
+ Relation reln = CreateFakeRelcacheEntry(rlocator);
+ bool has_vm_old = XLogRecHasBlockRef(record, HEAP_UPDATE_BLKREF_VM_OLD);
+ bool 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_locked(reln, 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_locked(reln, 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_locked(reln, oldblk, vmbuffer_old,
+ VISIBILITYMAP_VALID_BITS))
+ PageSetLSN(BufferGetPage(vmbuffer_old), lsn);
+ }
+ if (BufferIsValid(vmbuffer_old))
+ UnlockReleaseBuffer(vmbuffer_old);
+ }
+ if (!has_vm_old && !has_vm_new)
+ {
+ /*
+ * Backwards compatibility path. Previously, the VM buffers were
+ * not registered in the WAL record. We need this path to replay
+ * WAL generated by a not-yet-patched primary during upgrade.
+ */
+ if (old_cleared)
+ {
+ Buffer vmbuffer = InvalidBuffer;
+
+ visibilitymap_pin(reln, oldblk, &vmbuffer);
+ visibilitymap_clear(reln, oldblk, vmbuffer,
+ VISIBILITYMAP_VALID_BITS);
+ ReleaseBuffer(vmbuffer);
+ }
+ if (new_cleared)
+ {
+ Buffer vmbuffer = InvalidBuffer;
+
+ visibilitymap_pin(reln, newblk, &vmbuffer);
+ visibilitymap_clear(reln, newblk, vmbuffer,
+ VISIBILITYMAP_VALID_BITS);
+ ReleaseBuffer(vmbuffer);
+ }
+ }
+
+ FreeFakeRelcacheEntry(reln);
+ }
+
+ /*
+ * 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)
+ {
+ page = BufferGetPage(obuffer);
+ offnum = xlrec->old_offnum;
+ if (PageGetMaxOffsetNumber(page) >= offnum)
+ lp = PageGetItemId(page, offnum);
+
+ if (PageGetMaxOffsetNumber(page) < offnum || !ItemIdIsNormal(lp))
+ elog(PANIC, "neon_rm_redo: invalid lp");
+
+ htup = (HeapTupleHeader) PageGetItem(page, 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);
+ htup->t_choice.t_heap.t_field3.t_cid = xlrec->t_cid;
+ /* Set forward chain link in t_ctid */
+ htup->t_ctid = newtid;
+
+ /* Mark the page as a candidate for pruning */
+ PageSetPrunable(page, XLogRecGetXid(record));
+
+ if (old_cleared)
+ PageClearAllVisible(page);
+
+ PageSetLSN(page, 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);
+ page = (Page) BufferGetPage(nbuffer);
+ PageInit(page, 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;
+
+ page = BufferGetPage(nbuffer);
+
+ offnum = xlrec->new_offnum;
+ if (PageGetMaxOffsetNumber(page) + 1 < offnum)
+ elog(PANIC, "neon_rm_redo: 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((char *) &xlhdr, recdata, SizeOfNeonHeapHeader);
+ recdata += SizeOfNeonHeapHeader;
+
+ tuplen = recdata_end - recdata;
+ Assert(tuplen <= MaxHeapTupleSize);
+
+ htup = &tbuf.hdr;
+ MemSet((char *) 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));
+ htup->t_choice.t_heap.t_field3.t_cid = xlhdr.t_cid;
+ HeapTupleHeaderSetXmax(htup, xlrec->new_xmax);
+ /* Make sure there is no forward chain link in t_ctid */
+ htup->t_ctid = newtid;
- /* Mark the page as a candidate for pruning */
- PageSetPrunable(page, XLogRecGetXid(record));
+ offnum = PageAddItem(page, (Item) htup, newlen, offnum, true, true);
+ if (offnum == InvalidOffsetNumber)
+ elog(PANIC, "neon_rm_redo: failed to add tuple");
- if (xlrec->flags & XLH_DELETE_ALL_VISIBLE_CLEARED)
+ if (new_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;
+ freespace = PageGetHeapFreeSpace(page); /* needed to update FSM below */
+
PageSetLSN(page, lsn);
- MarkBufferDirty(buffer);
+ MarkBufferDirty(nbuffer);
}
- if (BufferIsValid(buffer))
- UnlockReleaseBuffer(buffer);
-}
-static void
-redo_neon_heap_update(XLogReaderState *record, bool hot_update)
-{
+ 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);
+
+#else
XLogRecPtr lsn = record->EndRecPtr;
xl_neon_heap_update *xlrec = (xl_neon_heap_update *) XLogRecGetData(record);
RelFileLocator rlocator;
@@ -665,11 +1241,78 @@ redo_neon_heap_update(XLogReaderState *record, bool hot_update)
*/
if (newaction == BLK_NEEDS_REDO && !hot_update && freespace < BLCKSZ / 5)
XLogRecordPageWithFreeSpace(rlocator, newblk, freespace);
+
+#endif
}
static void
redo_neon_heap_lock(XLogReaderState *record)
{
+#if PG_VERSION_NUM >= 170011
+ XLogRecPtr lsn = record->EndRecPtr;
+ xl_neon_heap_lock *xlrec = (xl_neon_heap_lock *) XLogRecGetData(record);
+ Buffer buffer;
+ Page page;
+ OffsetNumber offnum;
+ ItemId lp = NULL;
+ 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)
+ {
+ BlockNumber block;
+ RelFileLocator rlocator;
+
+ XLogRecGetBlockTag(record, HEAP_LOCK_BLKREF_HEAP, &rlocator, NULL, &block);
+
+ neon_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 = (Page) BufferGetPage(buffer);
+
+ offnum = xlrec->offnum;
+ if (PageGetMaxOffsetNumber(page) >= offnum)
+ lp = PageGetItemId(page, offnum);
+
+ if (PageGetMaxOffsetNumber(page) < offnum || !ItemIdIsNormal(lp))
+ elog(PANIC, "neon_rm_redo: 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);
+ htup->t_choice.t_heap.t_field3.t_cid = xlrec->t_cid;
+ PageSetLSN(page, lsn);
+ MarkBufferDirty(buffer);
+ }
+ if (BufferIsValid(buffer))
+ UnlockReleaseBuffer(buffer);
+
+#else
XLogRecPtr lsn = record->EndRecPtr;
xl_neon_heap_lock *xlrec = (xl_neon_heap_lock *) XLogRecGetData(record);
Buffer buffer;
@@ -736,11 +1379,155 @@ redo_neon_heap_lock(XLogReaderState *record)
}
if (BufferIsValid(buffer))
UnlockReleaseBuffer(buffer);
+
+#endif
}
static void
redo_neon_heap_multi_insert(XLogReaderState *record)
{
+#if PG_VERSION_NUM >= 170011
+ XLogRecPtr lsn = record->EndRecPtr;
+ xl_neon_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;
+
+ /*
+ * Insertion doesn't overwrite MVCC data, so no conflict processing is
+ * required.
+ */
+ xlrec = (xl_neon_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)
+ neon_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 = (Page) BufferGetPage(buffer);
+
+ for (i = 0; i < xlrec->ntuples; i++)
+ {
+ OffsetNumber offnum;
+ xl_neon_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, "neon_rm_redo: invalid max offset number");
+
+ xlhdr = (xl_neon_multi_insert_tuple *) SHORTALIGN(tupdata);
+ tupdata = ((char *) xlhdr) + SizeOfNeonMultiInsertTuple;
+
+ newlen = xlhdr->datalen;
+ Assert(newlen <= MaxHeapTupleSize);
+ htup = &tbuf.hdr;
+ MemSet((char *) htup, 0, SizeofHeapTupleHeader);
+ /* PG73FORMAT: get bitmap [+ padding] [+ oid] + data */
+ memcpy((char *) htup + SizeofHeapTupleHeader,
+ (char *) 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));
+ htup->t_choice.t_heap.t_field3.t_cid = xlrec->t_cid;
+ ItemPointerSetBlockNumber(&htup->t_ctid, blkno);
+ ItemPointerSetOffsetNumber(&htup->t_ctid, offnum);
+
+ offnum = PageAddItem(page, (Item) htup, newlen, offnum, true, true);
+ if (offnum == InvalidOffsetNumber)
+ elog(PANIC, "neon_rm_redo: 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 (xlrec->flags & XLH_INSERT_ALL_FROZEN_SET)
+ PageSetAllVisible(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(rlocator, blkno, freespace);
+
+#else
XLogRecPtr lsn = record->EndRecPtr;
xl_neon_heap_multi_insert *xlrec;
RelFileLocator rlocator;
@@ -880,6 +1667,8 @@ redo_neon_heap_multi_insert(XLogReaderState *record)
*/
if (action == BLK_NEEDS_REDO && freespace < BLCKSZ / 5)
XLogRecordPageWithFreeSpace(rlocator, blkno, freespace);
+
+#endif
}
#else
--
2.43.0