From 917e67b93e86cd7cc962d045f1bb38474912b90f Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilip.kumar@enterprisedb.com>
Date: Mon, 5 Aug 2019 09:39:45 +0530
Subject: [PATCH 07/22] Provide interfaces to store and fetch undo records.

The undo interface layer provides a mechanism for packing/unpacking
the undo records into the undo pages.  We are assuming that the
certain operations are happening in the transaction and those
operations need to insert the undo records for handling the
visibility or for rolling back the effect of the transaction.
This layer provides interfaces for inserting and fetching the
undo records.  The patch will also provide a mechanism for
compressing the undo records.  Basically, if an undo record on
the page have the same fields value as that in the first complete
record on the page then we compress the undo record by avoid
including those field in the record provided the record is from
the same transaction as the first complete record on the page.
And, during fetch we get those informations from the first
complete record of the page.

Author: Dilip Kumar with contributions from Robert Haas, Amit Kapila,
        Thomas Munro, Vignesh C and Rafia Sabih
Reviewed-by: Earlier version of this patch is reviewed by Amit Kapila
Tested-by: Neha Sharma
Discussion: https://www.postgresql.org/message-id/CAFiTN-uVxxopn0UZ64%3DF-sydbETBbGjWapnBikNo1%3DXv78UeFw%40mail.gmail.com
---
 src/backend/access/undo/Makefile             |    2 +-
 src/backend/access/undo/README.undointerface |   47 +
 src/backend/access/undo/undoaccess.c         | 1652 ++++++++++++++++++
 src/backend/access/undo/undorecord.c         | 1014 +++++++++++
 src/backend/storage/page/bufpage.c           |   35 +
 src/include/access/transam.h                 |    1 +
 src/include/access/undoaccess.h              |  118 ++
 src/include/access/undolog.h                 |    6 +-
 src/include/access/undorecord.h              |  274 +++
 src/include/storage/bufpage.h                |   36 +
 10 files changed, 3183 insertions(+), 2 deletions(-)
 create mode 100644 src/backend/access/undo/README.undointerface
 create mode 100644 src/backend/access/undo/undoaccess.c
 create mode 100755 src/backend/access/undo/undorecord.c
 create mode 100644 src/include/access/undoaccess.h
 create mode 100644 src/include/access/undorecord.h

diff --git a/src/backend/access/undo/Makefile b/src/backend/access/undo/Makefile
index 219c6963cf8..049a416f071 100644
--- a/src/backend/access/undo/Makefile
+++ b/src/backend/access/undo/Makefile
@@ -12,6 +12,6 @@ subdir = src/backend/access/undo
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = undolog.o
+OBJS = undoaccess.o undolog.o undorecord.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/access/undo/README.undointerface b/src/backend/access/undo/README.undointerface
new file mode 100644
index 00000000000..daf40d099ea
--- /dev/null
+++ b/src/backend/access/undo/README.undointerface
@@ -0,0 +1,47 @@
+The undo interface layer is about packing undo records into undo pages.
+We are assuming that certain operations are happening in a transaction and
+those operation might need to insert the undo for the visibility or for
+rolling back the effect of the transaction.  So this layer will provide
+interfaces for inserting and fetching the undo records.
+
+We want those undo records to be as space efficient as possible.  In most cases
+we won't need to write them out to the disk but in some cases when the
+transaction is big we might.  So we don't want them to be big.  In order to
+make that happen we have chosen to do the record compression therein we avoid
+storing the common fields in all the undo record on the page provided the first
+record on the page is from the same transaction.  This layer don't worry about
+compressing the payload data but that is the AM specific information so the AM
+layer might need to worry about compressing that information.
+
+Inserting an undo record
+------------------------
+For inserting an undo record the caller must first prepare undo records then
+insert the prepared undo records.  The prepare will just make the undo space
+ready and lock the necessary buffers and the insert will actually write the
+prepared records into the undo buffers.  There is an option to set the prepare
+limit which allows to prepare the multiple undo records and insert all of them
+in one shot.  This will allow the caller to prepared all the undo records
+(which are required under single WAL logged operation) outside the critical
+section and then insert all of them at once under the critical section.
+
+Fetching an undo record
+------------------------
+To fetch an undo record, a caller must provide a valid undo record pointer.
+The api will read the undo records from the underlying pages and return the
+unpacked undo record.  There is also an interface to bulk fetch the undo
+records which takes the start and end undo record pointers as input and it
+will return the array of unpacked undo records between that range.
+
+Undo record compression:
+------------------------
+Under a single transaction, there are some fields which will be common across
+the undo records, for example, the full transaction id is always the same,
+apart from that, the reloid, cid and rmid can also be same.  So we can avoid
+storing the common information in all the records and we can fetch it from the
+record in which it is stored.  We don't want to do this compression for
+complete transaction otherwise we might need to fetch the undo record across
+pages so we have decided to do the page level compression that is if the first
+record on the page is from the same transaction and if any of these common
+field value is same as the first record of the page then we will not store that
+value in our record. But, while fetching the undo record we will read that
+value from the first complete record of the page and return it to the caller.
diff --git a/src/backend/access/undo/undoaccess.c b/src/backend/access/undo/undoaccess.c
new file mode 100644
index 00000000000..dd9d8a48261
--- /dev/null
+++ b/src/backend/access/undo/undoaccess.c
@@ -0,0 +1,1652 @@
+/*-------------------------------------------------------------------------
+ *
+ * undoaccess.c
+ *	  entry points for inserting/fetching undo records
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/access/undo/undoaccess.c
+ *
+ *	INTERFACE ROUTINES
+ *		BeginUndoRecordInsert	- begin inserting one or multiple undo records
+ *		PrepareUndoInsert		- prepare undo record
+ *		InsertPreparedUndo		- insert prepared undo records
+ *		FinishUndoRecordInsert	- cleanup the insertion
+ *		BeginUndoFetch			- begin undo record fetch
+ *		UndoFetchRecord			- fetch the actual record
+ *		FinishUndoFetch			- cleanup after undo fetch record
+ *		UndoRecordRelease		- Release memory for unpacked undo record
+ *		UndoBulkFetchRecord		- Fetch undo record in bulk
+ *
+ * NOTES:
+ * Undo record layout:
+ *
+ * In each undo log undo records are stored in sequential order.  Each undo
+ * record consists of a undo record header, some optional headers and optional
+ * payload information.  The first undo record of each transaction in each undo
+ * log contains a group header that points to the next transaction's first undo
+ * record in the same undo log. This allows us to discard the entire
+ * transaction's undo log in one-shot.  The callers are not aware of the group
+ * header, it is entirely maintained by the undo interface layer for discarding
+ * the undo logs in groups.
+ *
+ * See undorecord.h for detailed information about the undo record header.
+ *
+ * Multiple logs:
+ *
+ * It is possible that the undo records for a transaction spans multiple undo
+ * logs.  We need some special handling while inserting them to ensure that
+ * discard and rollbacks can work sanely.
+ *
+ * We add a group header for the first record of the transaction for every log
+ * and link those group headers by uur_next_group pointer.  Additionally, In the
+ * first record of the transaction in new log after the log switch we will
+ * include an additional header called log switch header which will keep a
+ * pointer to the first undo record of the transaction in the previous log.
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/subtrans.h"
+#include "access/transam.h"
+#include "access/undoaccess.h"
+#include "access/undolog.h"
+#include "access/undolog_xlog.h"
+#include "access/undorecord.h"
+#include "access/xact.h"
+#include "access/xlog.h"
+#include "access/xlogutils.h"
+#include "catalog/pg_tablespace.h"
+#include "commands/tablecmds.h"
+#include "miscadmin.h"
+#include "storage/block.h"
+#include "storage/buf.h"
+#include "storage/buf_internals.h"
+#include "storage/bufmgr.h"
+
+/*
+ * Structure to hold the prepared undo information.  PreparedUndoInsert will
+ * fill this information for each prepared undo.  InsertPreparedUndo will use
+ * this information to insert all prepared undo records.
+ */
+struct PreparedUndoSpace
+{
+	UndoRecPtr	urp;			/* undo record pointer */
+	UnpackedUndoRecord *urec;	/* unpacked undo record */
+	uint16		size;			/* undo record size */
+	int			undo_buffer_idx[MAX_BUFFER_PER_UNDO];	/* undo_buffer array
+														 * index */
+};
+
+/*
+ * Holds the undo buffers informations.  During prepare time (which is called
+ * outside the critical section) we pin and lock all the buffer for inserting
+ * the undo record and store the information in this structure.  Later, during
+ * actual insert we use this information to insert the record into the buffers.
+ */
+struct PreparedUndoBuffer
+{
+	UndoLogNumber logno;		/* Undo log number */
+	BlockNumber blk;			/* block number */
+	Buffer		buf;			/* buffer allocated for the block */
+	bool		zero;			/* new block full of zeroes */
+};
+
+static UnpackedUndoRecord *UndoGetOneRecord(UnpackedUndoRecord *urec,
+											UndoRecPtr urp, RelFileNode rnode,
+											UndoLogCategory category,
+											Buffer *prevbuf);
+static int	UndoGetBufferSlot(UndoRecordInsertContext *context,
+							  RelFileNode rnode, BlockNumber blk,
+							  ReadBufferMode rbm);
+
+/*
+ * Prepare undo record update
+ *
+ * It's a helper function for UndoRecordPrepareUpdateNext
+ *
+ * urecptr - Undo record pointer of the record which needs to be updated.
+ * undo_offset - Offset of the undo record field which needs to be updated.
+ */
+static int
+PrepareUndoRecordUpdate(UndoRecordInsertContext *context, UndoRecPtr urecptr,
+						int size, int undo_offset)
+{
+	BlockNumber cur_blk;
+	RelFileNode rnode;
+	int			starting_byte;
+	int			bufidx;
+	int			index = 0;
+	int			remaining_bytes;
+	UndoRecordUpdateInfo *urec_update_info;
+
+	/*
+	 * Get a free slot to hold the prepared information for updating the undo
+	 * record.
+	 */
+	urec_update_info = &context->urec_update_info[context->nurec_update_info];
+
+	UndoRecPtrAssignRelFileNode(rnode, urecptr);
+	cur_blk = UndoRecPtrGetBlockNum(urecptr);
+	starting_byte = UndoRecPtrGetPageOffset(urecptr);
+
+	/* Remaining bytes on the current block. */
+	remaining_bytes = BLCKSZ - starting_byte;
+
+	/*
+	 * Compute the block number and the offset of the block where we need to
+	 * start updating the undo record.
+	 */
+	if (remaining_bytes <= undo_offset)
+	{
+		/*
+		 * If the given undo record offset is not in this block then go to the
+		 * next block.
+		 */
+		cur_blk++;
+		starting_byte = UndoLogBlockHeaderSize;
+		starting_byte += (undo_offset - remaining_bytes);
+	}
+	else
+		starting_byte += undo_offset;
+
+	/* Remember the offset where we need to start updating the undo record. */
+	urec_update_info->offset = starting_byte;
+
+	Assert(size <= sizeof(UndoRecPtr));
+
+	/*
+	 * Loop until we have locked all the buffers which we need to update. At
+	 * the max we will lock two buffers as we are just updating 8 bytes.
+	 * Blocks are locked in the increasing order so we need not to worry about
+	 * the deadlock.  We need to lock all the buffers in the prepare phase and
+	 * the actual update will be done in the update phase under the critical
+	 * section.
+	 */
+	while (1)
+	{
+		/* Read and lock the buffer if not already locked. */
+		bufidx = UndoGetBufferSlot(context, rnode, cur_blk, RBM_NORMAL);
+
+		/* Should never lock more than 2 buffers. */
+		Assert(index < 2);
+
+		/* Remember the prepared buffer index. */
+		urec_update_info->idx_undo_buffers[index++] = bufidx;
+		size -= (BLCKSZ - starting_byte);
+
+		/*
+		 * If the field we want to update is completely in the current block
+		 * then we are done.  Otherwise we need to go to the next block.
+		 */
+		if (size <= 0)
+			break;
+
+		starting_byte = UndoLogBlockHeaderSize;
+		cur_blk++;
+	}
+
+	/*
+	 * Remember the undo record pointer which we are updating and return the
+	 * index of the undo update info.  The caller will store the next/progress
+	 * field value in it based on which field it wants to update.
+	 */
+	urec_update_info->next = InvalidUndoRecPtr;
+	urec_update_info->progress = 0;
+	urec_update_info->urecptr = urecptr;
+
+	context->nurec_update_info++;
+	return (context->nurec_update_info - 1);
+}
+
+/*
+ * Prepare to update an undo record
+ *
+ * Prepare undo record update info for updating the the uur_next_group field
+ * in the undo record.  This function is called during the prepare phase it will
+ * pin and Lock all the necessary buffers required for updating the undo record.
+ * This function is called for
+ * a. Updating the next_group link in the group header.  This will make sure
+ * that the multiple transaction's undo record in the same log are linked and
+ * that will make it easy to traverse the undo during discard processing
+ * transaction by transaction.
+ * b. For multi-log transaction update the next_group link in the group header
+ * to connect the transaction first undo record across the undo logs.  This is
+ * required for collecting all the undo records of the transaction while
+ * applying the undo actions of the transaction.
+ *
+ * urecptr - undo record pointer of the next group which need to be set in the
+ *			 undo record pointed by prevurp.
+ * prevurp - undo record pointer to be updated.
+ */
+static void
+PrepareUndoRecordUpdateNext(UndoRecordInsertContext *context,
+							UndoRecPtr urecptr, UndoRecPtr prevurp)
+{
+	UndoLogSlot *slot;
+	int			index = 0;
+	int			offset;
+
+	/*
+	 * The absence of previous transaction's undo indicates that this backend
+	 * is preparing its first undo so there is nothing to be updated.
+	 */
+	if (!UndoRecPtrIsValid(prevurp))
+		return;
+
+	slot = UndoLogGetSlot(UndoRecPtrGetLogNo(prevurp), false);
+
+	/*
+	 * Acquire the discard update lock before reading the undo record so that
+	 * the undo record doesn't get discarded while we are reading undo
+	 * buffers.
+	 */
+	LWLockAcquire(&slot->discard_update_lock, LW_SHARED);
+
+	/* Check if it is already discarded. */
+	if (UndoRecPtrIsDiscarded(prevurp))
+	{
+		/* Release lock and return. */
+		LWLockRelease(&slot->discard_update_lock);
+		return;
+	}
+
+	/* Compute the offset of the uur_next_group in the undo record. */
+	offset = SizeOfUndoRecordHeader +
+		offsetof(UndoRecordGroup, urec_next_group);
+
+	index = PrepareUndoRecordUpdate(context, prevurp, sizeof(UndoRecPtr),
+									offset);
+
+	/* Store the next group's undo record pointer in urec_update_info. */
+	context->urec_update_info[index].next = urecptr;
+
+	/*
+	 * We can now release the discard lock as we have already acquired the
+	 * buffer locks.
+	 */
+	LWLockRelease(&slot->discard_update_lock);
+}
+
+/*
+ * Update the undo record
+ *
+ * This will overwrite uur_next_group or uur_progress fields in the undo record.
+ * Exact offset to be updated is already computed and necessary buffers are
+ * locked during the prepare phase.
+ */
+static void
+UndoRecordUpdate(UndoRecordInsertContext *context, int idx)
+{
+	Page		page = NULL;
+	int			i = 0;
+	int			write_bytes;
+	int			write_offset;
+	char	   *sourceptr;
+	UndoRecordUpdateInfo *urec_update_info = &context->urec_update_info[idx];
+
+	/* Whether to update the next or progress. */
+	if (UndoRecPtrIsValid(urec_update_info->next))
+	{
+		sourceptr = (char *) &urec_update_info->next;
+		write_bytes = sizeof(urec_update_info->next);
+	}
+	else
+	{
+		sourceptr = (char *) &urec_update_info->progress;
+		write_bytes = sizeof(urec_update_info->progress);
+	}
+
+	/* Where to start writing in the current block. */
+	write_offset = urec_update_info->offset;
+
+	/*
+	 * Start writing directly from the write offset calculated during prepare
+	 * phase.  And, loop until we write required bytes.
+	 */
+	while (1)
+	{
+		Buffer		buffer;
+		int			buf_idx;
+		int			can_write;
+		char	   *writeptr;
+
+		/* Should never write in more than 2 buffers. */
+		Assert(i < 2);
+
+		buf_idx = urec_update_info->idx_undo_buffers[i];
+		buffer = context->prepared_undo_buffers[buf_idx].buf;
+
+		/* How may bytes can be written in the current page. */
+		can_write = Min((BLCKSZ - write_offset), write_bytes);
+
+		/* If buffer is valid then write it otherwise just skip writing it. */
+		if (BufferIsValid(buffer))
+		{
+			page = BufferGetPage(buffer);
+
+			/* Compute the write pointer; write the buffer and mark it dirty. */
+			writeptr = (char *) page + write_offset;
+			memcpy(writeptr, sourceptr, can_write);
+			MarkBufferDirty(buffer);
+		}
+		else
+			Assert(InRecovery);
+
+		write_bytes -= can_write;
+
+		/*
+		 * If we have no more data to be written the break it otherwise go to
+		 * next block and continue writing there.
+		 */
+		if (write_bytes <= 0)
+			break;
+
+		sourceptr += can_write;
+		write_offset = UndoLogBlockHeaderSize;
+		i++;
+	}
+}
+
+/*
+ * Find or add the block info in the insert context's prepared buffer array.
+ *
+ * If the block is it is present in the prepared buffer array then just return
+ * its index otherwise read and lock the buffer and add an entry.
+ *
+ * Undo log insertions are append-only.  If the caller is writing new data that
+ * begins exactly at the beginning of a page, then there cannot be any useful
+ * data after that point.  In that case RBM_ZERO can be passed in as rbm so that
+ * we can skip a useless read of a disk block.  In all other cases, RBM_NORMAL
+ * should be passed in, to read the page in if it doesn't happen to be already
+ * in the buffer pool.
+ */
+static int
+UndoGetBufferSlot(UndoRecordInsertContext *context,
+				  RelFileNode rnode,
+				  BlockNumber blk,
+				  ReadBufferMode rbm)
+{
+	int			blkIndex;
+	Buffer		buffer;
+	XLogRedoAction action = BLK_NEEDS_REDO;
+	PreparedUndoBuffer *prepared_buffer;
+	UndoLogCategory category = context->alloc_context.category;
+
+	/*
+	 * Search the block in the prepared undo buffer array in our insert
+	 * context if we find it then simply return the index.
+	 */
+	for (blkIndex = 0; blkIndex < context->nprepared_undo_buffer; blkIndex++)
+	{
+		prepared_buffer = &context->prepared_undo_buffers[blkIndex];
+
+		/*
+		 * It's not enough to just compare the block number because this might
+		 * hold the blocks from different undo logs so compare the logno and
+		 * the blkno.
+		 */
+		if ((blk == prepared_buffer->blk) &&
+			(prepared_buffer->logno == rnode.relNode))
+		{
+			/* caller must hold exclusive lock on buffer */
+			Assert(BufferIsLocal(prepared_buffer->buf) ||
+				   LWLockHeldByMeInMode(BufferDescriptorGetContentLock(
+																	   GetBufferDescriptor(prepared_buffer->buf - 1)),
+										LW_EXCLUSIVE));
+			return blkIndex;
+		}
+	}
+
+	/*
+	 * We did not find the block the prepared buffer array so read the buffer
+	 * and add its entry.
+	 */
+	if (InRecovery)
+	{
+		/*
+		 * If block is found then this API will return the locked buffer so we
+		 * need not to lock it outside.
+		 */
+		action = XLogReadBufferForRedoBlock(context->alloc_context.xlog_record,
+											rnode,
+											UndoLogForkNum,
+											blk,
+											rbm,
+											false,
+											&buffer);
+	}
+	else
+	{
+		buffer = ReadBufferWithoutRelcache(rnode,
+										   UndoLogForkNum,
+										   blk,
+										   rbm,
+										   NULL,
+										   RelPersistenceForUndoLogCategory(category));
+
+		/* Lock the buffer */
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+	}
+
+	/*
+	 * Get the next empty slot in the prepared buffer array to store the
+	 * information of the new buffer we have locked.
+	 */
+	prepared_buffer =
+		&context->prepared_undo_buffers[context->nprepared_undo_buffer];
+
+	/*
+	 * During the recovery it's possible that the block is completely
+	 * discarded so just store it as InvalidBuffer in the prepared buffer
+	 * entry so that during actual insert we can skip inserting into these
+	 * blocks.
+	 */
+	if (action == BLK_NOTFOUND)
+	{
+		Assert(InRecovery);
+
+		prepared_buffer->buf = InvalidBuffer;
+		prepared_buffer->blk = blk;
+	}
+	else
+	{
+		/*
+		 * Remember the block and buffer information so that during multi
+		 * prepare we don't try to lock the same buffer again.
+		 */
+		prepared_buffer->buf = buffer;
+		prepared_buffer->blk = blk;
+		prepared_buffer->logno = rnode.relNode;
+		prepared_buffer->zero = rbm == RBM_ZERO;
+	}
+
+	context->nprepared_undo_buffer++;
+
+	return blkIndex;
+}
+
+/*
+ * Compress the undo record.
+ *
+ * Read the compression information from the first complete record of the page
+ * and try to compress the input undo record based on the compression
+ * information.
+ */
+static bool
+CompressUndoRecord(UndoRecordInsertContext *context, UndoRecPtr urecptr,
+				   UnpackedUndoRecord *urec, RelFileNode rnode)
+{
+	UndoPageHeader phdr;
+	Buffer		buffer;
+	Page		page;
+	int			bufidx;
+	int			offset;
+	UndoCompressionInfo compression_info;
+
+	/*
+	 * Read the buffer and compute the compression info.  We anyway need to
+	 * lock this buffer for inserting the undo record so locking here will not
+	 * cost us anything.  UndoGetBufferSlot will remember the reference to
+	 * this so that we will not try to read the buffer again.
+	 */
+	bufidx = UndoGetBufferSlot(context, rnode, UndoRecPtrGetBlockNum(urecptr),
+							   RBM_NORMAL);
+	buffer = context->prepared_undo_buffers[bufidx].buf;
+	page = BufferGetPage(buffer);
+	phdr = (UndoPageHeader) page;
+
+	/* Compute the offset of the next complete record on the page. */
+	offset = SizeOfUndoPageHeaderData + phdr->undo_len - phdr->record_offset;
+
+	/*
+	 * If we are inserting the first record after the partial record then
+	 * there is no complete record on the page based on which we can compress
+	 * our record.
+	 */
+	if (UndoRecPtrGetPageOffset(urecptr) == offset)
+		return false;
+
+	/* Read compression information from the record. */
+	UndoRecordGetCompressionInfo(page, offset, &compression_info);
+
+	/*
+	 * If the first record is not for the same transaction id for which we are
+	 * inserting the record then we can not compress this record.
+	 */
+	if (!FullTransactionIdEquals(urec->uur_fxid, compression_info.fxid))
+		return false;
+
+	/*
+	 * Exclude the information from the record which are same as the first
+	 * complete record of the page.
+	 */
+	urec->uur_info &= ~UREC_INFO_FXID;
+
+	if (urec->uur_rmid == compression_info.rmid)
+		urec->uur_info &= ~UREC_INFO_RMID;
+
+	if (urec->uur_reloid == compression_info.reloid)
+		urec->uur_info &= ~UREC_INFO_RELOID;
+
+	if (urec->uur_cid == compression_info.cid)
+		urec->uur_info &= ~UREC_INFO_CID;
+
+	return true;
+}
+
+/*
+ * Begin inserting undo records.
+ *
+ * This function must be called before all the undo records which are going to
+ * get inserted under a single WAL record.
+ * context - Undo insert context,  this will holds the memory and buffer
+ * information for preparing and inserting the undo records.  Caller must hold
+ * this memory until it insert undo record.
+ * category - Undo log category
+ * nprepared - max number of undo records that can be prepared before insert.
+ * xlog_record - In recovery for inserting the undo record the caller must pass
+ * its xlog reader state so that we can identify from which undo logno we need
+ * to allocate the space for the WAL.  To achieve this during DO time the caller
+ * must register all the undo buffer it has updated by calling
+ * RegisterUndoBuffers.
+ */
+void
+BeginUndoRecordInsert(UndoRecordInsertContext *context,
+					  UndoLogCategory category,
+					  int nprepared,
+					  XLogReaderState *xlog_record)
+{
+	uint32		nbuffers;
+
+	/* At least one prepared record should be there. */
+	if (nprepared <= 0)
+		elog(ERROR, "at least one undo record should be prepared");
+
+	/* Initialize undo log context. */
+	UndoLogBeginInsert(&context->alloc_context, category, xlog_record);
+
+	/* Initialize undo insert context. */
+	context->max_prepared_undo = nprepared;
+	context->nprepared_undo = 0;
+	context->nprepared_undo_buffer = 0;
+	context->nurec_update_info = 0;
+
+	/* Allocate memory for prepared undo record space. */
+	context->prepared_undo = (PreparedUndoSpace *) palloc(nprepared *
+														  sizeof(PreparedUndoSpace));
+
+	/* Compute number of buffers. */
+	nbuffers = (nprepared + MAX_UNDO_UPDATE_INFO) * MAX_BUFFER_PER_UNDO;
+
+	/* Allocate memory for the prepared buffers. */
+	context->prepared_undo_buffers =
+		palloc(nbuffers * sizeof(PreparedUndoBuffer));
+}
+
+/*
+ * Prepare to insert an undo record.
+ *
+ * Call PrepareUndoInsert to tell the undo subsystem about the undo record you
+ * intended to insert.  Upon return, the necessary undo buffers are pinned and
+ * locked.
+ *
+ * This should be called outside the  critical section.
+ */
+UndoRecPtr
+PrepareUndoInsert(UndoRecordInsertContext *context,
+				  UnpackedUndoRecord *urec,
+				  Oid dbid)
+{
+	UndoRecordSize size;
+	UndoRecPtr	urecptr = InvalidUndoRecPtr;
+	RelFileNode rnode;
+	UndoRecordSize cur_size = 0;
+	BlockNumber cur_blk;
+	FullTransactionId fxid;
+	int			starting_byte;
+	int			index = 0;
+	int			bufidx;
+	bool		resize = false;
+	ReadBufferMode rbm;
+	bool		need_xact_header;
+	UndoRecPtr	last_xact_start;
+	UndoRecPtr	prevlog_xact_start = InvalidUndoRecPtr;
+	PreparedUndoSpace *prepared_undo;
+
+	/* Already reached maximum prepared limit. */
+	if (context->nprepared_undo == context->max_prepared_undo)
+		elog(ERROR, "already reached the maximum prepared limit");
+
+	/* Extract the full transaction id from the input undo record. */
+	fxid = urec->uur_fxid;
+	Assert(FullTransactionIdIsValid(fxid));
+
+	/*
+	 * We don't yet know if this record needs a transaction header (ie is the
+	 * first undo record for a given transaction in a given undo log), because
+	 * you can only find out by allocating.  We'll resolve this circularity by
+	 * allocating enough space for a transaction header.  Similarly log switch
+	 * will only be detected after allocation so include the log switch header
+	 * and common information because in case of log switch we need to include
+	 * log switch header and also we need to include common header.  After
+	 * allocation, we'll only advance by as many bytes as we turn out to need.
+	 */
+	UndoRecordSetInfo(urec);
+	urec->uur_info |= UREC_INFO_GROUP;
+	urec->uur_info |= UREC_INFO_LOGSWITCH;
+	urec->uur_info |= UREC_INFO_PAGE_COMMON;
+
+	size = UndoRecordExpectedSize(urec);
+
+	/* Allocate space for the undo record. */
+	if (InRecovery)
+	{
+		/*
+		 * We'll figure out where the space needs to be allocated by
+		 * inspecting the xlog_record.  We don't expect to see temporary or
+		 * unlogged undo data here.
+		 */
+		Assert(context->alloc_context.category != UNDO_TEMP &&
+			   context->alloc_context.category != UNDO_UNLOGGED);
+		urecptr = UndoLogAllocateInRecovery(&context->alloc_context,
+											XidFromFullTransactionId(fxid),
+											size,
+											&need_xact_header,
+											&last_xact_start,
+											&prevlog_xact_start);
+	}
+	else
+		urecptr = UndoLogAllocate(&context->alloc_context,
+								  size,
+								  &need_xact_header, &last_xact_start,
+								  &prevlog_xact_start);
+
+	/*
+	 * If we need a group header then allocate memory for it and initialize
+	 * the same.
+	 */
+	if (need_xact_header)
+	{
+		urec->uur_group = palloc(sizeof(UndoRecordGroup));
+		urec->uur_group->urec_dbid = dbid;
+		urec->uur_group->urec_progress = InvalidBlockNumber;
+		urec->uur_group->urec_next_group = InvalidUndoRecPtr;
+	}
+	else
+	{
+		/* We don't need a group header after all. */
+		urec->uur_info &= ~UREC_INFO_GROUP;
+		resize = true;
+		urec->uur_group = NULL;
+	}
+
+	/*
+	 * If undo log got switched then allocate the memory for the log switch
+	 * header and initialize the same.
+	 */
+	if (UndoRecPtrIsValid(prevlog_xact_start))
+	{
+		urec->uur_logswitch = palloc(sizeof(UndoRecordLogSwitch));
+		urec->uur_logswitch->urec_prevlogstart = prevlog_xact_start;
+	}
+	else
+	{
+		/* We don't need a log switch header after all. */
+		urec->uur_info &= ~UREC_INFO_LOGSWITCH;
+		resize = true;
+		urec->uur_logswitch = NULL;
+	}
+
+	/* Populate the rnode for the undo record pointer. */
+	UndoRecPtrAssignRelFileNode(rnode, urecptr);
+
+	/*
+	 * If this is not the first record of the transaction for this log and we
+	 * are not inserting the first record on the page then we can compress
+	 * this record by avoid including the field which are same as the first
+	 * complete record on the page provided that is from the same transaction.
+	 */
+	if (!need_xact_header && !prevlog_xact_start &&
+		UndoRecPtrGetPageOffset(urecptr) != SizeOfUndoPageHeaderData &&
+		CompressUndoRecord(context, urecptr, urec, rnode))
+		resize = true;
+
+	/*
+	 * We might have excluded some of the header from the undo record so
+	 * recompute the expected record size.
+	 */
+	if (resize)
+		size = UndoRecordExpectedSize(urec);
+
+	/*
+	 * If the transaction's undo records are split across the undo logs then
+	 * link the transaction's group header in the previous log to the
+	 * transaction group header in the current log by updating the
+	 * uur_next_group field in the group header.
+	 */
+	if (UndoRecPtrIsValid(prevlog_xact_start))
+		PrepareUndoRecordUpdateNext(context, urecptr, prevlog_xact_start);
+
+	/*
+	 * If there is a physically preceding transaction in this undo log, and we
+	 * are writing the first record for this transaction that is in this undo
+	 * log (not necessarily the first ever for the transaction, because we
+	 * could have switched logs), then we need to set this transaction's group
+	 * header pointer in the previous transaction's group header.
+	 */
+	if (need_xact_header)
+		PrepareUndoRecordUpdateNext(context, urecptr, last_xact_start);
+
+	cur_blk = UndoRecPtrGetBlockNum(urecptr);
+	starting_byte = UndoRecPtrGetPageOffset(urecptr);
+
+	/*
+	 * If we happen to be writing the very first byte into this page, then
+	 * there is no need to read from disk.
+	 */
+	if (starting_byte == UndoLogBlockHeaderSize)
+		rbm = RBM_ZERO;
+	else
+		rbm = RBM_NORMAL;
+
+	/* Get a free slot in the prepare undo space for preparing our undo. */
+	prepared_undo = &context->prepared_undo[context->nprepared_undo];
+
+	/* Loop to lock the required buffers for inserting this undo record. */
+	do
+	{
+		bufidx = UndoGetBufferSlot(context, rnode, cur_blk, rbm);
+		if (cur_size == 0)
+			cur_size = BLCKSZ - starting_byte;
+		else
+			cur_size += BLCKSZ - UndoLogBlockHeaderSize;
+
+		/* undo record can't use buffers more than MAX_BUFFER_PER_UNDO. */
+		Assert(index < MAX_BUFFER_PER_UNDO);
+
+		/* Keep the track of the buffers we have pinned and locked. */
+		prepared_undo->undo_buffer_idx[index++] = bufidx;
+
+		/*
+		 * If we need more pages they'll be all new so we can definitely skip
+		 * reading from disk.
+		 */
+		rbm = RBM_ZERO;
+		cur_blk++;
+	} while (cur_size < size);
+
+	/*
+	 * Advance the local insert pointer in the context past this record so
+	 * that during multi-prepare we get the correct insert location for the
+	 * next record.
+	 */
+	UndoLogAdvance(&context->alloc_context, size);
+
+	/*
+	 * Save prepared undo record information into the context which will be
+	 * used by InsertPreparedUndo to insert the undo record.
+	 */
+	prepared_undo->urec = urec;
+	prepared_undo->urp = urecptr;
+	prepared_undo->size = size;
+	context->nprepared_undo++;
+
+	return urecptr;
+}
+
+/*
+ * Insert one undo record
+ *
+ * Helper function for InsertPreparedUndo.
+ */
+static void
+InsertUndoRecord(UndoRecordInsertContext *context,
+				 PreparedUndoSpace * prepared_undo)
+{
+	Buffer		buffer;
+	Page		page = NULL;
+	int			starting_byte;
+	int			bufidx = 0;
+	UndoPackContext ucontext = {{0}};
+	PreparedUndoBuffer *prepared_buffres = context->prepared_undo_buffers;
+	UnpackedUndoRecord *urec = prepared_undo->urec;
+
+	/* Compute starting offset of the page where to start inserting. */
+	starting_byte = UndoRecPtrGetPageOffset(prepared_undo->urp);
+
+	/* Initiate inserting the undo record. */
+	BeginInsertUndo(&ucontext, urec, prepared_undo->size);
+
+	/* Main loop for writing the undo record. */
+	while (1)
+	{
+		/* undo record can't use buffers more than MAX_BUFFER_PER_UNDO. */
+		Assert(bufidx < MAX_BUFFER_PER_UNDO);
+		buffer = prepared_buffres[prepared_undo->undo_buffer_idx[bufidx]].buf;
+
+		/*
+		 * During recovery, there might be some blocks which are already
+		 * deleted due to some discard command so we can just skip inserting
+		 * into those blocks.
+		 */
+		if (!BufferIsValid(buffer))
+		{
+			Assert(InRecovery);
+
+			/*
+			 * Instead of inserting the actual record this function will just
+			 * update the bookkeeping information in the context.
+			 */
+			SkipInsertingUndoData(&ucontext, BLCKSZ - starting_byte);
+		}
+		else
+		{
+			page = BufferGetPage(buffer);
+
+			/*
+			 * Initialize the page whenever we try to write the first record
+			 * in page.  We start writing immediately after the block header.
+			 */
+			if (starting_byte == UndoLogBlockHeaderSize)
+				UndoPageInit(page, BLCKSZ, urec->uur_info,
+							 ucontext.already_processed, prepared_undo->size);
+
+			/*
+			 * Write undo record data into the page and mark the buffer dirty.
+			 */
+			InsertUndoData(&ucontext, page, starting_byte);
+			MarkBufferDirty(buffer);
+		}
+
+		/* Record insertion is complete. */
+		if (ucontext.stage == UNDO_PACK_STAGE_DONE)
+			break;
+
+		/*
+		 * Record couldn't fit in the current block so insert the remaining
+		 * record in the next block.  In next block start inserting right
+		 * after the block header.
+		 */
+		starting_byte = UndoLogBlockHeaderSize;
+		bufidx++;
+	}
+}
+
+/*
+ * Insert previously-prepared undo records.
+ *
+ * This function will write the actual undo record into the buffers which are
+ * already pinned and locked in PreparedUndoInsert, and mark them dirty.  This
+ * step should be performed inside a critical section.
+ */
+void
+InsertPreparedUndo(UndoRecordInsertContext *context)
+{
+	PreparedUndoSpace *prepared_undo;
+	int			idx;
+	int			i;
+
+	/* There must be at least one prepared undo record. */
+	Assert(context->nprepared_undo > 0);
+
+	/*
+	 * This must be called under a critical section or we must be in recovery.
+	 */
+	Assert(InRecovery || CritSectionCount > 0);
+
+	for (idx = 0; idx < context->nprepared_undo; idx++)
+	{
+		prepared_undo = &context->prepared_undo[idx];
+
+		Assert(prepared_undo->size ==
+			   UndoRecordExpectedSize(prepared_undo->urec));
+
+		/* Insert the undo record. */
+		InsertUndoRecord(context, prepared_undo);
+
+		/* Advance the insert pointer past this record. */
+		UndoLogAdvanceFinal(prepared_undo->urp, prepared_undo->size);
+	}
+
+	/* Update the group header of the previous transaction. */
+	for (i = 0; i < context->nurec_update_info; i++)
+		UndoRecordUpdate(context, i);
+}
+
+/*
+ * Release all the memory and buffer pins hold for inserting the undo records.
+ */
+void
+FinishUndoRecordInsert(UndoRecordInsertContext *context)
+{
+	int			i;
+
+	/* Release buffer pins and locks. */
+	for (i = 0; i < context->nprepared_undo_buffer; i++)
+		if (BufferIsValid(context->prepared_undo_buffers[i].buf))
+			UnlockReleaseBuffer(context->prepared_undo_buffers[i].buf);
+
+	/*
+	 * Release memory for the transaction header and log switch header if we
+	 * have allocated it during the prepare time.
+	 */
+	for (i = 0; i < context->nprepared_undo; i++)
+	{
+		if (context->prepared_undo[i].urec->uur_group)
+			pfree(context->prepared_undo[i].urec->uur_group);
+
+		if (context->prepared_undo[i].urec->uur_logswitch)
+			pfree(context->prepared_undo[i].urec->uur_logswitch);
+	}
+
+	/*
+	 * Free the memory allocated for the prepared undo space and the prepared
+	 * undo buffers.
+	 */
+	pfree(context->prepared_undo_buffers);
+	pfree(context->prepared_undo);
+}
+
+/*
+ * Get undo page compression information.
+ *
+ * Read undo record compression information from the first complete record of
+ * the page and store them in the current unpack context.
+ */
+static void
+UndoPageGetCompressionInfo(UndoPackContext *ucontext, UndoRecPtr urp, Page page)
+{
+	int			offset;
+	UndoPageHeader phdr = (UndoPageHeader) page;
+	UndoCompressionInfo compresssion_info;
+
+	/* Compute the offset of the next complete record on the page. */
+	offset = SizeOfUndoPageHeaderData + phdr->undo_len - phdr->record_offset;
+
+	/* Fetch the undo page compression informations. */
+	UndoRecordGetCompressionInfo(page, offset, &compresssion_info);
+
+	/*
+	 * Get all missing common header information from the first undo record of
+	 * the page.
+	 */
+	if ((ucontext->urec_hd.urec_info & UREC_INFO_RMID) == 0)
+		ucontext->urec_rmid = compresssion_info.rmid;
+
+	if ((ucontext->urec_hd.urec_info & UREC_INFO_RELOID) == 0)
+		ucontext->urec_reloid = compresssion_info.reloid;
+
+	if ((ucontext->urec_hd.urec_info & UREC_INFO_FXID) == 0)
+		ucontext->urec_fxid = compresssion_info.fxid;
+
+	if ((ucontext->urec_hd.urec_info & UREC_INFO_CID) == 0)
+		ucontext->urec_cid = compresssion_info.cid;
+
+	ucontext->urec_hd.urec_info |= UREC_INFO_PAGE_COMMON;
+}
+
+/*
+ * Helper function for UndoFetchRecord and UndoBulkFetchRecord
+ *
+ * curbuf - If an input buffer is valid then this function will not release the
+ * pin on that buffer.  If the buffer is not valid then it will assign curbuf
+ * with the first buffer of the current undo record and also it will keep the
+ * pin on that buffer in a hope that while traversing the undo chain the caller
+ * might want to read the previous undo record from the same block.
+ */
+static UnpackedUndoRecord *
+UndoGetOneRecord(UnpackedUndoRecord *urec, UndoRecPtr urp, RelFileNode rnode,
+				 UndoLogCategory category, Buffer *curbuf)
+{
+	Page		page;
+	int			starting_byte = UndoRecPtrGetPageOffset(urp);
+	BlockNumber cur_blk;
+	UndoPackContext ucontext = {{0}};
+	Buffer		buffer = *curbuf;
+
+	cur_blk = UndoRecPtrGetBlockNum(urp);
+
+	/* Initiate unpacking one undo record. */
+	BeginUnpackUndo(&ucontext);
+
+	while (true)
+	{
+		/* If we already have a buffer then no need to allocate a new one. */
+		if (!BufferIsValid(buffer))
+		{
+			buffer = ReadBufferWithoutRelcache(rnode, UndoLogForkNum, cur_blk,
+											   RBM_NORMAL, NULL,
+											   RelPersistenceForUndoLogCategory(category));
+
+			/*
+			 * Remember the first buffer where this undo started as next undo
+			 * record what we fetch might fall on the same buffer.
+			 */
+			if (!BufferIsValid(*curbuf))
+				*curbuf = buffer;
+		}
+
+		/* Acquire shared lock on the buffer before reading undo from it. */
+		LockBuffer(buffer, BUFFER_LOCK_SHARE);
+
+		page = BufferGetPage(buffer);
+
+		UnpackUndoData(&ucontext, page, starting_byte);
+
+		/*
+		 * We are done if we have reached to the done stage otherwise move to
+		 * next block and continue reading from there.
+		 */
+		if (ucontext.stage == UNDO_PACK_STAGE_DONE)
+		{
+			/*
+			 * If the buffer is input buffer or the first buffer of the record
+			 * then just release the lock and keep the pin.  Otherwise, unlock
+			 * the buffer and also release the pin.
+			 */
+			if (buffer == *curbuf)
+				LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
+			else
+				UnlockReleaseBuffer(buffer);
+
+			/*
+			 * If any of the common header field is not available in the
+			 * current undo record then we must read it from the first
+			 * complete record of the page.  If any of these common field is
+			 * missing from the undo record then it's guarantee that the first
+			 * complete record on the page must be from the same transaction
+			 * otherwise we will always include all these common fields in the
+			 * undo record.
+			 */
+			if ((ucontext.urec_hd.urec_info & UREC_INFO_PAGE_COMMON) !=
+				UREC_INFO_PAGE_COMMON)
+				UndoPageGetCompressionInfo(&ucontext, urp,
+										   BufferGetPage(*curbuf));
+
+			break;
+		}
+
+		/*
+		 * If the buffer is input buffer or the first buffer of the record
+		 * then just release the lock and keep the pin.  Otherwise, unlock the
+		 * buffer and also release the pin.
+		 */
+		if (buffer == *curbuf)
+			LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
+		else
+			UnlockReleaseBuffer(buffer);
+
+		buffer = InvalidBuffer;
+
+		/* Go to next block. */
+		cur_blk++;
+		starting_byte = UndoLogBlockHeaderSize;
+	}
+
+	/* Final step of unpacking. */
+	FinishUnpackUndo(&ucontext, urec);
+
+	return urec;
+}
+
+/*
+ * BeginUndoFetch to fetch undo record.
+ */
+void
+BeginUndoFetch(UndoRecordFetchContext *context)
+{
+	context->buffer = InvalidBuffer;
+	context->urp = InvalidUndoRecPtr;
+}
+
+/*
+ * Fetch the undo record for given undo record pointer.
+ *
+ * This will internally allocate the memory for the unpacked undo record which
+ * will hold the pointers to the optional headers and the variable data.
+ * The undo record should be freed by the caller by calling ReleaseUndoRecord.
+ * This function will hold the pin on the buffer where we read the previous undo
+ * record so that when this function is called repeatedly with the same context
+ * then it can be benefitted by avoid reading the buffer again if the current
+ * undo record is in the same buffer.
+ */
+UnpackedUndoRecord *
+UndoFetchRecord(UndoRecordFetchContext *context, UndoRecPtr urp)
+{
+	RelFileNode rnode;
+	int			logno;
+	UndoLogSlot *slot;
+	UnpackedUndoRecord *uur = NULL;
+
+	logno = UndoRecPtrGetLogNo(urp);
+	slot = UndoLogGetSlot(logno, true);
+
+	/*
+	 * If slot is NULL that means undo log number is unknown.  Presumably it
+	 * has been entirely discarded.
+	 */
+	if (slot == NULL)
+		return NULL;
+
+	/*
+	 * Prevent UndoDiscardOneLog() from discarding data while we try to read
+	 * it.  Usually we would acquire log->mutex to read log->meta members, but
+	 * in this case we know that discard can't move without also holding
+	 * log->discard_lock.
+	 *
+	 * In Hot Standby mode log->oldest_data is never initialized because it's
+	 * get updated by undo discard worker whereas in HotStandby undo logs are
+	 * getting discarded using discard WAL.  So in HotStandby we can directly
+	 * check whether the undo record pointer is discarded or not.  But, we can
+	 * not do same for normal case because discard worker can concurrently
+	 * discard the undo logs.
+	 *
+	 * XXX We can avoid this check by always initializing log->oldest_data in
+	 * HotStandby mode as well whenever we apply discard WAL.  But, for doing
+	 * that we need to acquire discard lock just for setting this variable?
+	 */
+	if (InHotStandby)
+	{
+		if (UndoRecPtrIsDiscarded(urp))
+			return NULL;
+	}
+	else
+	{
+		LWLockAcquire(&slot->discard_lock, LW_SHARED);
+		if (slot->logno != logno || urp < slot->oldest_data)
+		{
+			/*
+			 * The slot has been recycled because the undo log was entirely
+			 * discarded, or the pointer is before the oldest data.
+			 */
+			LWLockRelease(&slot->discard_lock);
+			return NULL;
+		}
+	}
+
+	/*
+	 * Allocate memory for holding the undo record, caller should be
+	 * responsible for freeing this memory by calling UndoRecordRelease.
+	 */
+	uur = palloc0(sizeof(UnpackedUndoRecord));
+	UndoRecPtrAssignRelFileNode(rnode, urp);
+
+	/*
+	 * Before fetching the next record check whether we have a valid buffer in
+	 * the context.  If so and if we are reading the current record from the
+	 * same then pass that buffer to fetch the undo record, otherwise release
+	 * the buffer.
+	 */
+	if (BufferIsValid(context->buffer) &&
+		(UndoRecPtrGetLogNo(context->urp) != UndoRecPtrGetLogNo(urp) ||
+		 UndoRecPtrGetBlockNum(context->urp) != UndoRecPtrGetBlockNum(urp)))
+	{
+		ReleaseBuffer(context->buffer);
+		context->buffer = InvalidBuffer;
+	}
+
+	/* Fetch the current undo record. */
+	UndoGetOneRecord(uur, urp, rnode, slot->meta.category, &context->buffer);
+
+	/* Release the discard lock after fetching the record. */
+	if (!InHotStandby)
+		LWLockRelease(&slot->discard_lock);
+
+	context->urp = urp;
+
+	return uur;
+}
+
+/*
+ * Finish undo record fetch.
+ */
+void
+FinishUndoFetch(UndoRecordFetchContext *context)
+{
+	if (BufferIsValid(context->buffer))
+		ReleaseBuffer(context->buffer);
+}
+
+/*
+ * Release the memory of the undo record.
+ *
+ * Release memory of the undo record and other variable data of the undo record
+ * allocated by UndoFetchRecord and UndoBulkFetchRecord.
+ */
+void
+UndoRecordRelease(UnpackedUndoRecord *urec)
+{
+	if (urec->uur_payload.data)
+		pfree(urec->uur_payload.data);
+
+	if (urec->uur_tuple.data)
+		pfree(urec->uur_tuple.data);
+
+	if (urec->uur_group)
+		pfree(urec->uur_group);
+
+	if (urec->uur_logswitch)
+		pfree(urec->uur_logswitch);
+
+	/* Release the memory of the undo record. */
+	pfree(urec);
+}
+
+/*
+ * Prefetch undo pages, if the prefetch_pages are behind the prefetch_target
+ */
+static void
+PrefetchUndoPages(RelFileNode rnode, int prefetch_target, int *prefetch_pages,
+				  BlockNumber to_blkno, BlockNumber from_blkno,
+				  UndoLogCategory category)
+{
+	int			nprefetch;
+	BlockNumber startblock;
+	BlockNumber lastprefetched;
+
+	/* Calculate last prefetched page in the previous iteration. */
+	lastprefetched = from_blkno - *prefetch_pages;
+
+	/* We have already prefetched all the pages of the transaction's undo. */
+	if (lastprefetched <= to_blkno)
+		return;
+
+	/* Calculate number of blocks to be prefetched. */
+	nprefetch =
+		Min(prefetch_target - *prefetch_pages, lastprefetched - to_blkno);
+
+	/* Where to start prefetch. */
+	startblock = lastprefetched - nprefetch;
+
+	while (nprefetch--)
+	{
+		PrefetchBufferWithoutRelcache(rnode, MAIN_FORKNUM, startblock++,
+									  category == UNDO_TEMP);
+		(*prefetch_pages)++;
+	}
+}
+
+/*
+ * Read undo records of the transaction in bulk
+ *
+ * Read undo records between from_urecptr and to_urecptr until we exhaust the
+ * the memory size specified by max_result_size.  This will start reading undo
+ * records starting from from_urecptr.  It will read the transaction's undo
+ * record backwardly all the way upto to_urecptr.
+ *
+ * If we could not read all the records till to_urecptr then the caller should
+ * consume current set of records and call this function again.
+ *
+ * While traversing the undo record if the log switch is encountered then this
+ * API will stop processing further record and set the last record in the
+ * previous log as the from_urecptr so that the caller can call this API again
+ * to fetch the remaining records.
+ *
+ * XXX instead of this API handles this we can make caller to pass from and to
+ * urecptr from the same undo log?
+ *
+ * from_urecptr		- Where to start fetching the undo records.  If we can not
+ *					  read all the records because of memory limit then this
+ *					  will be set to the next from undo record pointer from
+ *					  where we need to start fetching on next call. Otherwise it
+ *					  will be set to InvalidUndoRecPtr.
+ * to_urecptr		- Last undo record pointer to be fetched.
+ * max_result_size	- Memory segment limit to collect undo records.
+ * nrecords			- Number of undo records read.
+ */
+UndoRecInfo *
+UndoBulkFetchRecord(UndoRecPtr *from_urecptr, UndoRecPtr to_urecptr,
+					int max_result_size, int *nrecords)
+{
+	RelFileNode rnode;
+	UndoRecPtr	urecptr,
+				prev_urec_ptr;
+	BlockNumber blkno;
+	BlockNumber to_blkno;
+	Buffer		buffer = InvalidBuffer;
+	UnpackedUndoRecord *uur = NULL;
+	UndoRecInfo *urp_array;
+	int			urp_array_size = 1024;
+	int			urp_index = 0;
+	int			prefetch_target = 0;
+	int			prefetch_pages = 0;
+	Size		total_size = 0;
+
+	/*
+	 * If we are fetching undo records from more than one logs, We can not
+	 * compute how many block to prefetch from the current log because we
+	 * don't know that in how many blocks of the log this transaction has
+	 * inserted the undo.  Hence, we can't use prefetching in this case.
+	 *
+	 * XXX should we always restrict caller to input from and to urecptr from
+	 * the same undo log.  But, it may not be possible for the caller to
+	 * always detect that?
+	 */
+	if (UndoRecPtrGetLogNo(*from_urecptr) == UndoRecPtrGetLogNo(to_urecptr))
+		prefetch_target = target_prefetch_pages;
+
+	/*
+	 * Allocate initial memory to hold the undo record info, we can expand it
+	 * if needed.
+	 */
+	urp_array = (UndoRecInfo *) palloc(sizeof(UndoRecInfo) * urp_array_size);
+	urecptr = *from_urecptr;
+
+	prev_urec_ptr = InvalidUndoRecPtr;
+	*from_urecptr = InvalidUndoRecPtr;
+
+	/* Read undo chain backward until we reach to the first undo record.  */
+	while (1)
+	{
+		BlockNumber from_blkno;
+		UndoLogSlot *slot;
+		UndoLogCategory category;
+		int			logno;
+
+		logno = UndoRecPtrGetLogNo(urecptr);
+		slot = UndoLogGetSlot(logno, true);
+		if (slot == NULL)
+		{
+			if (BufferIsValid(buffer))
+				ReleaseBuffer(buffer);
+			return NULL;
+		}
+		category = slot->meta.category;
+
+		UndoRecPtrAssignRelFileNode(rnode, urecptr);
+		to_blkno = UndoRecPtrGetBlockNum(to_urecptr);
+		from_blkno = UndoRecPtrGetBlockNum(urecptr);
+
+		/* Allocate memory for next undo record. */
+		uur = palloc0(sizeof(UnpackedUndoRecord));
+
+		/*
+		 * If next undo record pointer to be fetched is not on the same block
+		 * then release the old buffer and reduce the prefetch_pages count by
+		 * one as we have consumed one page. Otherwise, just pass the old
+		 * buffer into the UndoGetOneRecord so that it doesn't read the buffer
+		 * again.
+		 */
+		blkno = UndoRecPtrGetBlockNum(urecptr);
+		if (!UndoRecPtrIsValid(prev_urec_ptr) ||
+			UndoRecPtrGetLogNo(prev_urec_ptr) != logno ||
+			UndoRecPtrGetBlockNum(prev_urec_ptr) != blkno)
+		{
+			/* Release the previous buffer */
+			if (BufferIsValid(buffer))
+			{
+				ReleaseBuffer(buffer);
+				buffer = InvalidBuffer;
+			}
+
+			if (prefetch_pages > 0)
+				prefetch_pages--;
+		}
+
+		/*
+		 * If prefetch_pages are behind prefetch_target then it's time to
+		 * prefetch again.
+		 */
+		if (prefetch_pages < prefetch_target)
+			PrefetchUndoPages(rnode, prefetch_target, &prefetch_pages, to_blkno,
+							  from_blkno, category);
+
+		/* Get the undo record. */
+		UndoGetOneRecord(uur, urecptr, rnode, category, &buffer);
+
+		/* Remember the previous undo record pointer. */
+		prev_urec_ptr = urecptr;
+
+		/*
+		 * Calculate the previous undo record pointer of the transaction.  If
+		 * we have detected the log switch then compute the latest undo record
+		 * of the transaction in the previous log and set this as next
+		 * from_urecptr.  On log_switch don't process further record and
+		 * return handle to the caller with appropriate from_urecptr so that
+		 * caller can resume the fetching.
+		 */
+		if (uur->uur_logswitch)
+		{
+			urecptr = UndoGetPrevUrp(uur, prev_urec_ptr, buffer, category);
+			*from_urecptr = urecptr;
+		}
+		else if (prev_urec_ptr == to_urecptr)
+			urecptr = InvalidUndoRecPtr;
+		else
+			urecptr = UndoGetPrevUrp(uur, prev_urec_ptr, buffer, category);
+
+		/*
+		 * We have consumed all the elements of the urp_array so expand its
+		 * size.
+		 */
+		if (urp_index >= urp_array_size)
+		{
+			urp_array_size *= 2;
+			urp_array =
+				repalloc(urp_array, sizeof(UndoRecInfo) * urp_array_size);
+		}
+
+		/* Add entry in the urp_array */
+		urp_array[urp_index].index = urp_index;
+		urp_array[urp_index].urp = prev_urec_ptr;
+		urp_array[urp_index].uur = uur;
+		urp_index++;
+
+		/* We have fetched all the undo records for the transaction. */
+		if (!UndoRecPtrIsValid(urecptr) || (prev_urec_ptr == to_urecptr))
+			break;
+
+		/* Add the size required to store unpacked undo record in memory. */
+		total_size += sizeof(UnpackedUndoRecord) + UndoRecordPayloadSize(uur);
+
+		/*
+		 * Including current record, if we have crossed the memory limit or
+		 * undo log got switched then stop processing more records.  Remember
+		 * to set the from_urecptr so that on next call we can resume fetching
+		 * undo records where we left it.
+		 *
+		 * XXX we need this special handling for the log switch because in
+		 * some cases caller is enable to identify the log boundary but it
+		 * expect us to read the undo record only for one log at a time.
+		 */
+		if (total_size >= max_result_size || uur->uur_logswitch)
+		{
+			*from_urecptr = urecptr;
+			break;
+		}
+	}
+
+	/* Release the last buffer. */
+	if (BufferIsValid(buffer))
+		ReleaseBuffer(buffer);
+
+	*nrecords = urp_index;
+
+	return urp_array;
+}
+
+/*
+ * Register the undo buffers.
+ */
+void
+RegisterUndoLogBuffers(UndoRecordInsertContext *context, uint8 first_block_id)
+{
+	int			idx;
+	int			flags;
+
+	for (idx = 0; idx < context->nprepared_undo_buffer; idx++)
+	{
+		flags = context->prepared_undo_buffers[idx].zero
+			? REGBUF_KEEP_DATA_AFTER_CP | REGBUF_WILL_INIT
+			: REGBUF_KEEP_DATA_AFTER_CP;
+		XLogRegisterBuffer(first_block_id + idx,
+						   context->prepared_undo_buffers[idx].buf, flags);
+		UndoLogRegister(&context->alloc_context, first_block_id + idx,
+						context->prepared_undo_buffers[idx].logno);
+	}
+}
+
+/*
+ * Set LSN on undo page.
+*/
+void
+UndoLogBuffersSetLSN(UndoRecordInsertContext *context, XLogRecPtr recptr)
+{
+	int			idx;
+
+	for (idx = 0; idx < context->nprepared_undo_buffer; idx++)
+		PageSetLSN(BufferGetPage(context->prepared_undo_buffers[idx].buf),
+				   recptr);
+}
+
+/*
+ * Read length of the previous undo record.
+ *
+ * This function will take an undo record pointer as an input and read the
+ * length of the previous undo record which is stored at the end of the previous
+ * undo record.  If the undo record is split then this will add the undo block
+ * header size in the total length.
+ */
+static uint16
+UndoGetPrevRecordLen(UndoRecPtr urp, Buffer input_buffer,
+					 UndoLogCategory category)
+{
+	UndoLogOffset offset = UndoRecPtrGetPageOffset(urp) - 1;
+	BlockNumber cur_blk = UndoRecPtrGetBlockNum(urp);
+	Buffer		buffer = input_buffer;
+	Page		page = NULL;
+	char	   *pagedata = NULL;
+	union
+	{
+		char		len_bytes[2];
+		uint16		len;
+	}			prevlen;
+	RelFileNode rnode;
+	int			byte_to_read = 2;
+	char		persistence;
+	uint16		prev_rec_len = 0;
+
+	/* Get relfilenode. */
+	UndoRecPtrAssignRelFileNode(rnode, urp);
+	persistence = RelPersistenceForUndoLogCategory(category);
+
+	if (BufferIsValid(buffer))
+	{
+		page = BufferGetPage(buffer);
+		pagedata = (char *) page;
+	}
+
+	/*
+	 * Length of the previous undo record is store at the end of that record
+	 * so just read the last 2 bytes.  But, these 2 bytes can be split acorss
+	 * pages so we can not directly memcpy these 2 bytes, instead we need to
+	 * read them byte by byte.
+	 */
+	while (byte_to_read > 0)
+	{
+		/* Read buffer if the current buffer is not valid. */
+		if (!BufferIsValid(buffer))
+		{
+			buffer = ReadBufferWithoutRelcache(rnode, UndoLogForkNum,
+											   cur_blk, RBM_NORMAL, NULL,
+											   persistence);
+			LockBuffer(buffer, BUFFER_LOCK_SHARE);
+			page = BufferGetPage(buffer);
+			pagedata = (char *) page;
+		}
+
+		/*
+		 * Read current prevlen byte from current block if the read offset
+		 * hasn't reach to undo block header.  Otherwise, go to the previous
+		 * block and continue reading from there.
+		 */
+		if (offset >= UndoLogBlockHeaderSize)
+		{
+			prevlen.len_bytes[byte_to_read - 1] = pagedata[offset];
+			byte_to_read -= 1;
+			offset -= 1;
+
+			continue;
+		}
+
+		/*
+		 * Could not read complete prevlen from the current block so go to the
+		 * previous block and start reading from end of the block.
+		 */
+		cur_blk -= 1;
+		offset = BLCKSZ - 1;
+
+		/* Release the current buffer if it is not provide by the caller. */
+		if (input_buffer != buffer)
+			UnlockReleaseBuffer(buffer);
+
+		buffer = InvalidBuffer;
+	}
+
+	prev_rec_len = prevlen.len;
+
+	/*
+	 * If previous undo record is split across the pages then include the
+	 * block header size in its length for computing the start location of the
+	 * previous undo record.
+	 */
+	if (UndoRecPtrGetPageOffset(urp) - UndoLogBlockHeaderSize < prevlen.len)
+		prev_rec_len += UndoLogBlockHeaderSize;
+
+	/* Release the buffer if we have locally read it. */
+	if (input_buffer != buffer)
+		UnlockReleaseBuffer(buffer);
+
+	return prev_rec_len;
+}
+
+/*
+ * Calculate the previous undo record pointer for the transaction.
+ *
+ * This will take current undo record pointer of the transaction as an input
+ * and calculate the previous undo record pointer of the transaction.
+ */
+UndoRecPtr
+UndoGetPrevUrp(UnpackedUndoRecord *uur, UndoRecPtr urp, Buffer buffer,
+			   UndoLogCategory category)
+{
+	UndoLogOffset offset = UndoRecPtrGetOffset(urp);
+	uint16		prevlen;
+
+	/*
+	 * If this is the first record of the transaction for this log then we
+	 * need to get the previous undo record pointer from the previous undo
+	 * log. We can get the logno from the log switch header stored in this
+	 * record and then we can compute the undo record pointer of the last
+	 * record on that log by using the insert location and the length of the
+	 * last record on that log.
+	 */
+	if (uur && uur->uur_logswitch)
+	{
+		UndoLogNumber logno =
+		UndoRecPtrGetLogNo(uur->uur_logswitch->urec_prevlogstart);
+
+		urp = MakeUndoRecPtr(logno, UndoLogGetNextInsertPtr(logno));
+
+		/*
+		 * If we are reading the length from the another undo log then we can
+		 * not use the buffer of the current log so pass invalid buffer.
+		 */
+		prevlen = UndoGetPrevRecordLen(urp, InvalidBuffer, category);
+	}
+	else
+		prevlen = UndoGetPrevRecordLen(urp, buffer, category);
+
+	/*
+	 * We have got the length of the previous record.  Now using the offset of
+	 * the current record and the length of the previous record we can compute
+	 * the undo record pointer of the previous record.
+	 */
+	offset = UndoRecPtrGetOffset(urp);
+
+	/* calculate the previous undo record pointer */
+	return MakeUndoRecPtr(UndoRecPtrGetLogNo(urp), offset - prevlen);
+}
diff --git a/src/backend/access/undo/undorecord.c b/src/backend/access/undo/undorecord.c
new file mode 100755
index 00000000000..25d2973d771
--- /dev/null
+++ b/src/backend/access/undo/undorecord.c
@@ -0,0 +1,1014 @@
+/*-------------------------------------------------------------------------
+ *
+ * undorecord.c
+ *	  encode and decode undo records
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/backend/access/undo/undorecord.c
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/bufmask.h"
+#include "access/subtrans.h"
+#include "access/undorecord.h"
+#include "catalog/pg_tablespace.h"
+#include "storage/block.h"
+
+/* Prototypes for static functions. */
+static bool InsertUndoBytes(char *sourceptr, int sourcelen,
+							char **writeptr, char *endptr,
+							int *total_bytes_written, int *partial_write);
+static bool ReadUndoBytes(char *destptr, int readlen,
+						  char **readptr, char *endptr,
+						  int *total_bytes_read, int *partial_read);
+
+ /*
+  * Compute the header size of the undo record.
+  */
+static size_t
+UndoRecordHeaderSize(uint16 uur_info)
+{
+	size_t		size;
+
+	/* Add fixed header size. */
+	size = SizeOfUndoRecordHeader;
+
+	/* Add optional headers sizes. */
+	if ((uur_info & UREC_INFO_GROUP) != 0)
+		size += SizeOfUndoRecordGroup;
+
+	if ((uur_info & UREC_INFO_RMID) != 0)
+		size += sizeof(RmgrId);
+
+	if ((uur_info & UREC_INFO_RELOID) != 0)
+		size += sizeof(Oid);
+
+	if ((uur_info & UREC_INFO_FXID) != 0)
+		size += sizeof(FullTransactionId);
+
+	if ((uur_info & UREC_INFO_CID) != 0)
+		size += sizeof(CommandId);
+
+	if ((uur_info & UREC_INFO_FORK) != 0)
+		size += sizeof(ForkNumber);
+
+	if ((uur_info & UREC_INFO_PREVUNDO) != 0)
+		size += sizeof(UndoRecPtr);
+
+	if ((uur_info & UREC_INFO_BLOCK) != 0)
+		size += SizeOfUndoRecordBlock;
+
+	if ((uur_info & UREC_INFO_LOGSWITCH) != 0)
+		size += SizeOfUndoRecordLogSwitch;
+
+	if ((uur_info & UREC_INFO_PAYLOAD) != 0)
+		size += SizeOfUndoRecordPayload;
+
+	return size;
+}
+
+/*
+ * Compute the size of the payload data for the unpacked undo record.
+ */
+size_t
+UndoRecordPayloadSize(UnpackedUndoRecord *uur)
+{
+	size_t		size = 0;
+
+	/* Add payload size if record contains payload data. */
+	if ((uur->uur_info & UREC_INFO_PAYLOAD) != 0)
+	{
+		size += uur->uur_payload.len;
+		size += uur->uur_tuple.len;
+	}
+
+	return size;
+}
+
+/*
+ * Compute the expected size of the undo record.
+ *
+ * This function will take UnpackedUndoRecord as input and base on the uur_info
+ * and the variable headers it will compute the exact size required to store
+ * this record on the page.
+ */
+Size
+UndoRecordExpectedSize(UnpackedUndoRecord *uur)
+{
+	/* Header size + payload data + undo record length. */
+	return UndoRecordHeaderSize(uur->uur_info) +
+		UndoRecordPayloadSize(uur) +
+		sizeof(uint16);
+}
+
+/*
+ * Initiate inserting an undo record.
+ *
+ * This function will initialize the context for inserting and undo record
+ * which will be inserted by calling InsertUndoData.
+ */
+void
+BeginInsertUndo(UndoPackContext *ucontext, UnpackedUndoRecord *uur,
+				uint16 undo_len)
+{
+	ucontext->stage = UNDO_PACK_STAGE_HEADER;
+	ucontext->already_processed = 0;
+	ucontext->partial_bytes = 0;
+
+	/* Copy undo record header. */
+	ucontext->urec_hd.urec_type = uur->uur_type;
+	ucontext->urec_hd.urec_info = uur->uur_info;
+
+	/* Copy optional headers into the context. */
+	if ((uur->uur_info & UREC_INFO_GROUP) != 0)
+		ucontext->urec_group = *uur->uur_group;
+
+	if ((uur->uur_info & UREC_INFO_RMID) != 0)
+		ucontext->urec_rmid = uur->uur_rmid;
+
+	if ((uur->uur_info & UREC_INFO_RELOID) != 0)
+		ucontext->urec_reloid = uur->uur_reloid;
+
+	if ((uur->uur_info & UREC_INFO_FXID) != 0)
+		ucontext->urec_fxid = uur->uur_fxid;
+
+	if ((uur->uur_info & UREC_INFO_CID) != 0)
+		ucontext->urec_cid = uur->uur_cid;
+
+	if ((uur->uur_info & UREC_INFO_FORK) != 0)
+		ucontext->urec_fork = uur->uur_fork;
+
+	if ((uur->uur_info & UREC_INFO_PREVUNDO) != 0)
+		ucontext->urec_prevundo = uur->uur_prevundo;
+
+	if ((uur->uur_info & UREC_INFO_BLOCK) != 0)
+	{
+		ucontext->urec_blk.urec_block = uur->uur_block;
+		ucontext->urec_blk.urec_offset = uur->uur_offset;
+	}
+
+	if ((uur->uur_info & UREC_INFO_LOGSWITCH) != 0)
+		ucontext->urec_logswitch = *uur->uur_logswitch;
+
+	/* Copy undo record payload header and data. */
+	if ((uur->uur_info & UREC_INFO_PAYLOAD) != 0)
+	{
+		ucontext->urec_payload.urec_payload_len = uur->uur_payload.len;
+		ucontext->urec_payload.urec_tuple_len = uur->uur_tuple.len;
+		ucontext->urec_payloaddata = uur->uur_payload.data;
+		ucontext->urec_tupledata = uur->uur_tuple.data;
+	}
+	else
+	{
+		ucontext->urec_payload.urec_payload_len = 0;
+		ucontext->urec_payload.urec_tuple_len = 0;
+	}
+
+	ucontext->undo_len = undo_len;
+}
+
+/*
+ * Insert the undo record into the input page from the unpack undo context.
+ *
+ * Caller can  call this function multiple times until desired stage is reached.
+ * This will write the undo record into the page.
+ */
+void
+InsertUndoData(UndoPackContext *ucontext, Page page, int starting_byte)
+{
+	char	   *writeptr = (char *) page + starting_byte;
+	char	   *endptr = (char *) page + BLCKSZ;
+
+	switch (ucontext->stage)
+	{
+		case UNDO_PACK_STAGE_HEADER:
+			/* Insert undo record header. */
+			if (!InsertUndoBytes((char *) &ucontext->urec_hd,
+								 SizeOfUndoRecordHeader, &writeptr, endptr,
+								 &ucontext->already_processed,
+								 &ucontext->partial_bytes))
+				return;
+			ucontext->stage = UNDO_PACK_STAGE_GROUP;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_GROUP:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_GROUP) != 0)
+			{
+				/* Insert undo record group header. */
+				if (!InsertUndoBytes((char *) &ucontext->urec_group,
+									 SizeOfUndoRecordGroup,
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_RMID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_RMID:
+			/* Write rmid(if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_RMID) != 0)
+			{
+				if (!InsertUndoBytes((char *) &(ucontext->urec_rmid), sizeof(RmgrId),
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_RELOID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_RELOID:
+			/* Write reloid(if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_RELOID) != 0)
+			{
+				if (!InsertUndoBytes((char *) &(ucontext->urec_reloid), sizeof(Oid),
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_XID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_XID:
+			/* Write xid(if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_FXID) != 0)
+			{
+				if (!InsertUndoBytes((char *) &(ucontext->urec_fxid), sizeof(FullTransactionId),
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_CID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_CID:
+			/* Write cid(if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_CID) != 0)
+			{
+				if (!InsertUndoBytes((char *) &(ucontext->urec_cid), sizeof(CommandId),
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_FORKNUM;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_FORKNUM:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_FORK) != 0)
+			{
+				/* Insert undo record fork number. */
+				if (!InsertUndoBytes((char *) &ucontext->urec_fork,
+									 sizeof(ForkNumber),
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_PREVUNDO;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_PREVUNDO:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_PREVUNDO) != 0)
+			{
+				/* Insert undo record blkprev. */
+				if (!InsertUndoBytes((char *) &ucontext->urec_prevundo,
+									 sizeof(UndoRecPtr),
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_BLOCK;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_BLOCK:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_BLOCK) != 0)
+			{
+				/* Insert undo record block header. */
+				if (!InsertUndoBytes((char *) &ucontext->urec_blk,
+									 SizeOfUndoRecordBlock,
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_LOGSWITCH;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_LOGSWITCH:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_LOGSWITCH) != 0)
+			{
+				/* Insert undo record log switch header. */
+				if (!InsertUndoBytes((char *) &ucontext->urec_logswitch,
+									 SizeOfUndoRecordLogSwitch,
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_PAYLOAD;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_PAYLOAD:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_PAYLOAD) != 0)
+			{
+				/* Insert undo record payload header. */
+				if (!InsertUndoBytes((char *) &ucontext->urec_payload,
+									 SizeOfUndoRecordPayload,
+									 &writeptr, endptr,
+									 &ucontext->already_processed,
+									 &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_PAYLOAD_DATA;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_PAYLOAD_DATA:
+			{
+				int			len = ucontext->urec_payload.urec_payload_len;
+
+				if (len > 0)
+				{
+					/* Insert payload data. */
+					if (!InsertUndoBytes((char *) ucontext->urec_payloaddata,
+										 len, &writeptr, endptr,
+										 &ucontext->already_processed,
+										 &ucontext->partial_bytes))
+						return;
+				}
+				ucontext->stage = UNDO_PACK_STAGE_TUPLE_DATA;
+			}
+			/* fall through */
+
+		case UNDO_PACK_STAGE_TUPLE_DATA:
+			{
+				int			len = ucontext->urec_payload.urec_tuple_len;
+
+				if (len > 0)
+				{
+					/* Insert tuple data. */
+					if (!InsertUndoBytes((char *) ucontext->urec_tupledata,
+										 len, &writeptr, endptr,
+										 &ucontext->already_processed,
+										 &ucontext->partial_bytes))
+						return;
+				}
+				ucontext->stage = UNDO_PACK_STAGE_UNDO_LENGTH;
+			}
+			/* fall through */
+
+		case UNDO_PACK_STAGE_UNDO_LENGTH:
+			/* Insert undo length. */
+			if (!InsertUndoBytes((char *) &ucontext->undo_len,
+								 sizeof(uint16), &writeptr, endptr,
+								 &ucontext->already_processed,
+								 &ucontext->partial_bytes))
+				return;
+
+			ucontext->stage = UNDO_PACK_STAGE_DONE;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_DONE:
+			/* Nothing to be done. */
+			break;
+
+		default:
+			Assert(0);			/* Invalid stage */
+	}
+}
+
+/*
+ * Skip inserting undo record
+ *
+ * Don't insert the actual undo record instead just update the context data
+ * so that if we need to insert the remaining partial record to the next
+ * block then we have right context.
+ */
+void
+SkipInsertingUndoData(UndoPackContext *ucontext, int bytes_to_skip)
+{
+	switch (ucontext->stage)
+	{
+		case UNDO_PACK_STAGE_HEADER:
+			if (bytes_to_skip < SizeOfUndoRecordHeader)
+			{
+				ucontext->partial_bytes = bytes_to_skip;
+				return;
+			}
+			bytes_to_skip -= SizeOfUndoRecordHeader;
+			ucontext->stage = UNDO_PACK_STAGE_GROUP;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_GROUP:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_GROUP) != 0)
+			{
+				if (bytes_to_skip < SizeOfUndoRecordGroup)
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= SizeOfUndoRecordGroup;
+			}
+
+			ucontext->stage = UNDO_PACK_STAGE_RMID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_RMID:
+			/* Write rmid (if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UNDO_PACK_STAGE_RMID) != 0)
+			{
+				if (bytes_to_skip < (sizeof(RmgrId)))
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= sizeof(RmgrId);
+			}
+			ucontext->stage = UNDO_PACK_STAGE_RELOID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_RELOID:
+			/* Write reloid (if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UNDO_PACK_STAGE_RELOID) != 0)
+			{
+				if (bytes_to_skip < sizeof(Oid))
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= sizeof(Oid);
+			}
+			ucontext->stage = UNDO_PACK_STAGE_XID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_XID:
+			/* Write xid (if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UNDO_PACK_STAGE_XID) != 0)
+			{
+				if (bytes_to_skip < (sizeof(TransactionId)))
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= sizeof(TransactionId);
+			}
+			ucontext->stage = UNDO_PACK_STAGE_CID;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_CID:
+			/* Write cid (if needed and not already done). */
+			if ((ucontext->urec_hd.urec_info & UNDO_PACK_STAGE_CID) != 0)
+			{
+				if (bytes_to_skip < sizeof(CommandId))
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= sizeof(CommandId);
+			}
+			ucontext->stage = UNDO_PACK_STAGE_FORKNUM;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_FORKNUM:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_FORK) != 0)
+			{
+				if (bytes_to_skip < sizeof(ForkNumber))
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= sizeof(ForkNumber);
+			}
+
+			ucontext->stage = UNDO_PACK_STAGE_PREVUNDO;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_PREVUNDO:
+			if ((ucontext->urec_hd.urec_info & UNDO_PACK_STAGE_PREVUNDO) != 0)
+			{
+				if (bytes_to_skip < sizeof(UndoRecPtr))
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= sizeof(UndoRecPtr);
+			}
+			ucontext->stage = UNDO_PACK_STAGE_BLOCK;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_BLOCK:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_BLOCK) != 0)
+			{
+				if (bytes_to_skip < SizeOfUndoRecordBlock)
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= SizeOfUndoRecordBlock;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_LOGSWITCH;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_LOGSWITCH:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_LOGSWITCH) != 0)
+			{
+				if (bytes_to_skip < SizeOfUndoRecordLogSwitch)
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= SizeOfUndoRecordLogSwitch;
+			}
+
+			ucontext->stage = UNDO_PACK_STAGE_PAYLOAD;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_PAYLOAD:
+			/* Skip payload header. */
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_PAYLOAD) != 0)
+			{
+				if (bytes_to_skip < SizeOfUndoRecordPayload)
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= SizeOfUndoRecordPayload;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_PAYLOAD_DATA;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_PAYLOAD_DATA:
+			if (ucontext->urec_payload.urec_payload_len > 0)
+			{
+				if (bytes_to_skip < ucontext->urec_payload.urec_payload_len)
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= ucontext->urec_payload.urec_payload_len;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_TUPLE_DATA;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_TUPLE_DATA:
+			if (ucontext->urec_payload.urec_tuple_len > 0)
+			{
+				if (bytes_to_skip < ucontext->urec_payload.urec_tuple_len)
+				{
+					ucontext->partial_bytes = bytes_to_skip;
+					return;
+				}
+				bytes_to_skip -= ucontext->urec_payload.urec_tuple_len;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_UNDO_LENGTH;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_UNDO_LENGTH:
+			ucontext->stage = UNDO_PACK_STAGE_DONE;
+			 /* fall through */ ;
+
+		case UNDO_PACK_STAGE_DONE:
+			/* Nothing to be done. */
+			break;
+
+		default:
+			Assert(0);			/* Invalid stage */
+	}
+}
+
+/*
+ * Write undo bytes from a particular source, but only to the extent that
+ * they weren't written previously and will fit.
+ *
+ * 'sourceptr' points to the source data, and 'sourcelen' is the length of
+ * that data in bytes.
+ *
+ * 'writeptr' points to the insertion point for these bytes, and is updated
+ * for whatever we write.  The insertion point must not pass 'endptr', which
+ * represents the end of the buffer into which we are writing.
+ *
+ * 'my_bytes_written' is a pointer to the count of previous-written bytes
+ * from this and following structures in this undo record; that is, any
+ * bytes that are part of previous structures in the record have already
+ * been subtracted out.
+ *
+ * 'total_bytes_written' points to the count of all previously-written bytes,
+ * and must it must be updated for the bytes we write.
+ *
+ * The return value is false if we ran out of space before writing all
+ * the bytes, and otherwise true.
+ */
+static bool
+InsertUndoBytes(char *sourceptr, int sourcelen, char **writeptr, char *endptr,
+				int *total_bytes_written, int *partial_write)
+{
+	int			can_write;
+	int			remaining;
+
+	/* Compute number of bytes we can write. */
+	remaining = sourcelen - *partial_write;
+	can_write = Min(remaining, endptr - *writeptr);
+
+	/* Bail out if no bytes can be written. */
+	if (can_write == 0)
+		return false;
+
+	/* Copy the bytes we can write. */
+	memcpy(*writeptr, sourceptr + *partial_write, can_write);
+
+	/* Update bookkeeping information. */
+	*writeptr += can_write;
+	*total_bytes_written += can_write;
+
+	/* Could not read whole data so set the partial_read. */
+	if (can_write < remaining)
+	{
+		*partial_write += can_write;
+		return false;
+	}
+
+	/* Return true only if we wrote the whole thing. */
+	*partial_write = 0;
+	return true;
+}
+
+/*
+ * Initiate unpacking an undo record.
+ *
+ * This function will initialize the context for unpacking the undo record which
+ * will be unpacked by calling UnpackUndoData.
+ */
+void
+BeginUnpackUndo(UndoPackContext *ucontext)
+{
+	ucontext->stage = UNDO_PACK_STAGE_HEADER;
+	ucontext->already_processed = 0;
+	ucontext->partial_bytes = 0;
+}
+
+/*
+ * Read the undo record from the input page to the unpack undo context.
+ *
+ * Caller can  call this function multiple times until desired stage is reached.
+ * This will read the undo record from the page and store the data into unpack
+ * undo context, which can be later copied to unpacked undo record by calling
+ * FinishUnpackUndo.
+ */
+void
+UnpackUndoData(UndoPackContext *ucontext, Page page, int starting_byte)
+{
+	char	   *readptr = (char *) page + starting_byte;
+	char	   *endptr = (char *) page + BLCKSZ;
+
+	switch (ucontext->stage)
+	{
+		case UNDO_PACK_STAGE_HEADER:
+			if (!ReadUndoBytes((char *) &ucontext->urec_hd,
+							   SizeOfUndoRecordHeader, &readptr, endptr,
+							   &ucontext->already_processed,
+							   &ucontext->partial_bytes))
+				return;
+			ucontext->stage = UNDO_PACK_STAGE_GROUP;
+			/* fall through */
+		case UNDO_PACK_STAGE_GROUP:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_GROUP) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_group,
+								   SizeOfUndoRecordGroup,
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_RMID;
+			/* fall through */
+		case UNDO_PACK_STAGE_RMID:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_RMID) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_rmid,
+								   sizeof(RmgrId),
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_RELOID;
+			/* fall through */
+		case UNDO_PACK_STAGE_RELOID:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_RELOID) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_reloid,
+								   sizeof(Oid),
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_XID;
+			/* fall through */
+		case UNDO_PACK_STAGE_XID:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_FXID) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_fxid,
+								   sizeof(FullTransactionId),
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_CID;
+			/* fall through */
+		case UNDO_PACK_STAGE_CID:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_CID) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_cid,
+								   sizeof(CommandId),
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_FORKNUM;
+			/* fall through */
+		case UNDO_PACK_STAGE_FORKNUM:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_FORK) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_fork,
+								   sizeof(ForkNumber),
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_PREVUNDO;
+			/* fall through */
+		case UNDO_PACK_STAGE_PREVUNDO:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_PREVUNDO) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_prevundo,
+								   sizeof(UndoRecPtr),
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_BLOCK;
+			/* fall through */
+
+		case UNDO_PACK_STAGE_BLOCK:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_BLOCK) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_blk,
+								   SizeOfUndoRecordBlock,
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_LOGSWITCH;
+			/* fall through */
+		case UNDO_PACK_STAGE_LOGSWITCH:
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_LOGSWITCH) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_logswitch,
+								   SizeOfUndoRecordLogSwitch,
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_PAYLOAD;
+			/* fall through */
+		case UNDO_PACK_STAGE_PAYLOAD:
+			/* Read payload header. */
+			if ((ucontext->urec_hd.urec_info & UREC_INFO_PAYLOAD) != 0)
+			{
+				if (!ReadUndoBytes((char *) &ucontext->urec_payload,
+								   SizeOfUndoRecordPayload,
+								   &readptr, endptr, &ucontext->already_processed,
+								   &ucontext->partial_bytes))
+					return;
+			}
+			ucontext->stage = UNDO_PACK_STAGE_PAYLOAD_DATA;
+			/* fall through */
+		case UNDO_PACK_STAGE_PAYLOAD_DATA:
+			{
+				int			len = ucontext->urec_payload.urec_payload_len;
+
+				/* Allocate memory for the payload data if not already done. */
+				if (len > 0)
+				{
+					if (ucontext->urec_payloaddata == NULL)
+						ucontext->urec_payloaddata = (char *) palloc(len);
+
+					/* Read payload data. */
+					if (!ReadUndoBytes((char *) ucontext->urec_payloaddata, len,
+									   &readptr, endptr, &ucontext->already_processed,
+									   &ucontext->partial_bytes))
+						return;
+				}
+				ucontext->stage = UNDO_PACK_STAGE_TUPLE_DATA;
+				/* fall through */
+			}
+		case UNDO_PACK_STAGE_TUPLE_DATA:
+			{
+				int			len = ucontext->urec_payload.urec_tuple_len;
+
+				/* Allocate memory for the tuple data if not already done. */
+				if (len > 0)
+				{
+					if (ucontext->urec_tupledata == NULL)
+						ucontext->urec_tupledata = (char *) palloc(len);
+
+					/* Read tuple data. */
+					if (!ReadUndoBytes((char *) ucontext->urec_tupledata, len,
+									   &readptr, endptr, &ucontext->already_processed,
+									   &ucontext->partial_bytes))
+						return;
+				}
+
+				ucontext->stage = UNDO_PACK_STAGE_DONE;
+				/* fall through */
+			}
+		case UNDO_PACK_STAGE_DONE:
+			/* Nothing to be done. */
+			break;
+		default:
+			Assert(0);			/* Invalid stage */
+	}
+
+	return;
+}
+
+/*
+ * Final step of unpacking the undo record.
+ *
+ * Copy the undo record data from the unpack undo context to the input unpacked
+ * undo record.
+ */
+void
+FinishUnpackUndo(UndoPackContext *ucontext, UnpackedUndoRecord *uur)
+{
+	/* Copy undo record header. */
+	uur->uur_type = ucontext->urec_hd.urec_type;
+	uur->uur_info = ucontext->urec_hd.urec_info;
+
+	/* Copy undo record group header if it is present. */
+	if ((uur->uur_info & UREC_INFO_GROUP) != 0)
+	{
+		uur->uur_group = palloc(SizeOfUndoRecordGroup);
+		memcpy(uur->uur_group, &ucontext->urec_group, SizeOfUndoRecordGroup);
+	}
+
+	/*
+	 * Copy the common field.  All of these field must present in the final
+	 * unpacked undo record.
+	 */
+	Assert((uur->uur_info & UREC_INFO_PAGE_COMMON) == UREC_INFO_PAGE_COMMON);
+	uur->uur_rmid = ucontext->urec_rmid;
+	uur->uur_reloid = ucontext->urec_reloid;
+	uur->uur_fxid = ucontext->urec_fxid;
+	uur->uur_cid = ucontext->urec_cid;
+
+	/* Copy undo record optional headers. */
+	if ((uur->uur_info & UREC_INFO_FORK) != 0)
+		uur->uur_fork = ucontext->urec_fork;
+
+	if ((uur->uur_info & UREC_INFO_PREVUNDO) != 0)
+		uur->uur_prevundo = ucontext->urec_prevundo;
+
+	if ((uur->uur_info & UREC_INFO_BLOCK) != 0)
+	{
+		uur->uur_block = ucontext->urec_blk.urec_block;
+		uur->uur_offset = ucontext->urec_blk.urec_offset;
+	}
+
+	if ((uur->uur_info & UREC_INFO_LOGSWITCH) != 0)
+	{
+		uur->uur_logswitch = palloc(SizeOfUndoRecordLogSwitch);
+		memcpy(uur->uur_logswitch, &ucontext->urec_logswitch,
+			   SizeOfUndoRecordLogSwitch);
+	}
+
+	if ((uur->uur_info & UREC_INFO_PAYLOAD) != 0)
+	{
+		uur->uur_payload.len = ucontext->urec_payload.urec_payload_len;
+		uur->uur_tuple.len = ucontext->urec_payload.urec_tuple_len;
+
+		/* Read payload data if its length is not 0. */
+		if (uur->uur_payload.len != 0)
+			uur->uur_payload.data = ucontext->urec_payloaddata;
+
+		/* Read tuple data if its length is not 0. */
+		if (uur->uur_tuple.len != 0)
+			uur->uur_tuple.data = ucontext->urec_tupledata;
+	}
+}
+
+/*
+ * Directly read the undo compression info from the undo record starting at
+ * given offset.
+ */
+void
+UndoRecordGetCompressionInfo(Page page, int starting_byte,
+							 UndoCompressionInfo *compresssion_info)
+{
+	UndoRecordHeader urec_hd;
+	char	   *readptr = (char *) page + starting_byte;
+
+	/* Read the main header */
+	memcpy((char *) &urec_hd, readptr, SizeOfUndoRecordHeader);
+	readptr += SizeOfUndoRecordHeader;
+
+	/* If we have the group header then skip it. */
+	if (urec_hd.urec_info & UREC_INFO_GROUP)
+		readptr += SizeOfUndoRecordGroup;
+
+	/* Read compression info. */
+	memcpy((char *) compresssion_info, readptr, sizeof(UndoCompressionInfo));
+}
+
+/*
+ * Read undo bytes into a particular destination,
+ *
+ * 'destptr' points to the source data, and 'readlen' is the length of
+ * that data to be read in bytes.
+ *
+ * 'readptr' points to the read point for these bytes, and is updated
+ * for how much we read.  The read point must not pass 'endptr', which
+ * represents the end of the buffer from which we are reading.
+ *
+ * 'partial_read' is a pointer to the count of previous partial read bytes
+ *
+ * 'total_bytes_read' points to the count of all previously-read bytes,
+ * and must likewise be updated for the bytes we read.
+ *
+ * nocopy if this flag is set true then it will just skip the readlen
+ * size in undo but it will not copy into the buffer.
+ *
+ * The return value is false if we ran out of space before read all
+ * the bytes, and otherwise true.
+ */
+static bool
+ReadUndoBytes(char *destptr, int readlen, char **readptr, char *endptr,
+			  int *total_bytes_read, int *partial_read)
+{
+	int			can_read;
+	int			remaining;
+
+	/* Compute number of bytes we can read. */
+	remaining = readlen - *partial_read;
+	can_read = Min(remaining, endptr - *readptr);
+
+	/* Bail out if no bytes can be read. */
+	if (can_read == 0)
+		return false;
+
+	/* Copy the bytes we can read. */
+	memcpy(destptr + *partial_read, *readptr, can_read);
+
+	/* Update bookkeeping information. */
+	*readptr += can_read;
+	*total_bytes_read += can_read;
+
+	/* Could not read whole data so set the partial_read. */
+	if (can_read < remaining)
+	{
+		*partial_read += can_read;
+		return false;
+	}
+
+	/* Return true only if we wrote the whole thing. */
+	*partial_read = 0;
+
+	return true;
+}
+
+/*
+ * Set uur_info for an UnpackedUndoRecord appropriately based on which fields
+ * are set.
+ */
+void
+UndoRecordSetInfo(UnpackedUndoRecord *uur)
+{
+	/*
+	 * If fork number is not the main fork then we need to store it in the
+	 * undo record so set the flag.
+	 */
+	if (uur->uur_fork != MAIN_FORKNUM)
+		uur->uur_info |= UREC_INFO_FORK;
+
+	/* If prevundo is valid undo record pointer then set the flag. */
+	if (uur->uur_prevundo != InvalidUndoRecPtr)
+		uur->uur_info |= UREC_INFO_PREVUNDO;
+
+	/* If the block number is valid then set the flag for the block header. */
+	if (uur->uur_block != InvalidBlockNumber)
+		uur->uur_info |= UREC_INFO_BLOCK;
+
+	/*
+	 * Either of the payload or the tuple length is non-zero then we need the
+	 * payload header.
+	 */
+	if (uur->uur_payload.len || uur->uur_tuple.len)
+		uur->uur_info |= UREC_INFO_PAYLOAD;
+}
diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c
index 6b49810e378..548474a032c 100644
--- a/src/backend/storage/page/bufpage.c
+++ b/src/backend/storage/page/bufpage.c
@@ -59,6 +59,41 @@ PageInit(Page page, Size pageSize, Size specialSize)
 	/* p->pd_prune_xid = InvalidTransactionId;		done by above MemSet */
 }
 
+/*
+ * UndoPageInit
+ *		Initializes the contents of an undo page.
+ *		Note that we don't calculate an initial checksum here; that's not done
+ *		until it's time to write.
+ */
+void
+UndoPageInit(Page page, Size pageSize, uint16 uur_info, uint16 record_offset,
+			 uint16 len)
+{
+	UndoPageHeader	p = (UndoPageHeader) page;
+
+	Assert(pageSize == BLCKSZ);
+
+	/* Make sure all fields of page are zero, as well as unused space. */
+	MemSet(p, 0, pageSize);
+
+	p->pd_flags = 0;
+	p->pd_lower = SizeOfUndoPageHeaderData;
+	p->pd_upper = pageSize;
+	p->pd_special = pageSize;
+	p->uur_info = uur_info;
+	p->record_offset = record_offset;
+
+	/*
+	 * If we have partial record on the page then store complete length of the
+	 * undo record and the offset of the undo record which is starting in this
+	 * page.  This will be used to compute the offset of the first complete
+	 * record on the page.
+	 */
+	p->undo_len = (record_offset > 0) ? len : 0;
+
+	PageSetPageSizeAndVersion(page, pageSize, PG_PAGE_LAYOUT_VERSION);
+}
+
 
 /*
  * PageIsVerified
diff --git a/src/include/access/transam.h b/src/include/access/transam.h
index 33fd052156f..cc005096991 100644
--- a/src/include/access/transam.h
+++ b/src/include/access/transam.h
@@ -47,6 +47,7 @@
 #define EpochFromFullTransactionId(x)	((uint32) ((x).value >> 32))
 #define XidFromFullTransactionId(x)		((uint32) (x).value)
 #define U64FromFullTransactionId(x)		((x).value)
+#define FullTransactionIdEquals(a, b)	((a).value == (b).value)
 #define FullTransactionIdPrecedes(a, b)	((a).value < (b).value)
 #define FullTransactionIdIsValid(x)		TransactionIdIsValid(XidFromFullTransactionId(x))
 #define InvalidFullTransactionId		FullTransactionIdFromEpochAndXid(0, InvalidTransactionId)
diff --git a/src/include/access/undoaccess.h b/src/include/access/undoaccess.h
new file mode 100644
index 00000000000..815db4482f2
--- /dev/null
+++ b/src/include/access/undoaccess.h
@@ -0,0 +1,118 @@
+/*-------------------------------------------------------------------------
+ *
+ * undoaccess.h
+ *	  entry points for inserting/fetching undo records
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/undoaccess.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef UNDOACCESS_H
+#define UNDOACCESS_H
+
+#include "access/undolog.h"
+#include "access/undorecord.h"
+#include "access/xlogdefs.h"
+#include "catalog/pg_class.h"
+
+/*
+ * XXX Do we want to support undo tuple size which is more than the BLCKSZ
+ * if not than undo record can spread across 2 buffers at the max.
+ */
+#define MAX_BUFFER_PER_UNDO    2
+
+/*
+ * Maximum number of the UndoRecordUpdateInfo for updating the undo record.
+ * Usually it's 1 for updating next link of previous transaction's header
+ * if we are starting a new transaction.  But, in some cases where the same
+ * transaction is spilled to the next log, we update our own transaction's
+ * header in previous undo log as well as the header of the previous transaction
+ * in the new log.
+ */
+#define MAX_UNDO_UPDATE_INFO	2
+
+typedef struct PreparedUndoSpace PreparedUndoSpace;
+typedef struct PreparedUndoBuffer PreparedUndoBuffer;
+
+/*
+ * Undo record element.  Used for storing the group of undo record in a array
+ * using UndoBulkFetchRecord.
+ */
+typedef struct UndoRecInfo
+{
+	int			index;			/* Index of the element.  For stable qsort. */
+	UndoRecPtr	urp;			/* undo recptr (undo record location). */
+	UnpackedUndoRecord *uur;	/* actual undo record. */
+} UndoRecInfo;
+
+/*
+ * This structure holds the informations for updating the undo record.  We need
+ * to update the group header for various purposes
+ * a) Setting the "uur_next" which points to the next group's first undo record
+ * in the undo log.
+ * b) updating the undo apply progress while applying undo actions.  During
+ * prepare phase we will keep all the information handy in this structure and
+ * that will be used for updating the actual record inside the critical section.
+ */
+typedef struct UndoRecordUpdateInfo
+{
+	UndoRecPtr	urecptr;		/* Undo record pointer to be updated. */
+	uint32		offset;			/* offset in page where to start updating. */
+	UndoRecPtr	next;			/* first urp of the next group which is to be
+								 * set in the group header */
+	BlockNumber progress;		/* undo apply action progress. */
+	int			idx_undo_buffers[MAX_BUFFER_PER_UNDO];
+} UndoRecordUpdateInfo;
+
+/*
+ * Context for preparing and inserting undo records.
+ */
+typedef struct UndoRecordInsertContext
+{
+	UndoLogAllocContext alloc_context;
+	PreparedUndoSpace *prepared_undo;	/* prepared undo. */
+	PreparedUndoBuffer *prepared_undo_buffers;	/* Buffers for prepared undo. */
+	UndoRecordUpdateInfo urec_update_info[MAX_UNDO_UPDATE_INFO];	/* Information for undo
+																	 * update */
+	int			nprepared_undo; /* Number of prepared undo records. */
+	int			max_prepared_undo;	/* Max prepared undo for this operation. */
+	int			nprepared_undo_buffer;	/* Number of undo buffers. */
+	int			nurec_update_info;	/* Number of prepared undo update info. */
+} UndoRecordInsertContext;
+
+/*
+ * Context for fetching the required undo record.
+ */
+typedef struct UndoRecordFetchContext
+{
+	Buffer		buffer;			/* Previous undo record pinned buffer. */
+	UndoRecPtr	urp;			/* Previous undo record pointer. */
+} UndoRecordFetchContext;
+
+extern void BeginUndoRecordInsert(UndoRecordInsertContext *context,
+								  UndoLogCategory category,
+								  int nprepared,
+								  XLogReaderState *xlog_record);
+extern UndoRecPtr PrepareUndoInsert(UndoRecordInsertContext *context,
+									UnpackedUndoRecord *urec, Oid dbid);
+extern void InsertPreparedUndo(UndoRecordInsertContext *context);
+extern void FinishUndoRecordInsert(UndoRecordInsertContext *context);
+extern void BeginUndoFetch(UndoRecordFetchContext *context);
+extern UnpackedUndoRecord *UndoFetchRecord(UndoRecordFetchContext *context,
+										   UndoRecPtr urp);
+extern void FinishUndoFetch(UndoRecordFetchContext *context);
+extern void UndoRecordRelease(UnpackedUndoRecord *urec);
+extern UndoRecInfo *UndoBulkFetchRecord(UndoRecPtr *from_urecptr,
+										UndoRecPtr to_urecptr,
+										int undo_apply_size, int *nrecords);
+extern void RegisterUndoLogBuffers(UndoRecordInsertContext *context,
+								   uint8 first_block_id);
+extern void UndoLogBuffersSetLSN(UndoRecordInsertContext *context,
+								 XLogRecPtr recptr);
+extern UndoRecPtr UndoGetPrevUrp(UnpackedUndoRecord *uur, UndoRecPtr urp,
+								 Buffer buffer, UndoLogCategory category);
+
+#endif							/* UNDOINSERT_H */
diff --git a/src/include/access/undolog.h b/src/include/access/undolog.h
index 01c00a58dee..2d4486978f0 100644
--- a/src/include/access/undolog.h
+++ b/src/include/access/undolog.h
@@ -140,7 +140,7 @@ typedef int UndoLogNumber;
 	(((uint64) (logno) << UndoLogOffsetBits) | (offset))
 
 /* The number of unusable bytes in the header of each block. */
-#define UndoLogBlockHeaderSize SizeOfPageHeaderData
+#define UndoLogBlockHeaderSize SizeOfUndoPageHeaderData
 
 /* The number of usable bytes we can store per block. */
 #define UndoLogUsableBytesPerPage (BLCKSZ - UndoLogBlockHeaderSize)
@@ -170,6 +170,10 @@ typedef int UndoLogNumber;
 #define UndoRecPtrGetPageOffset(urp)			\
 	(UndoRecPtrGetOffset(urp) % BLCKSZ)
 
+/* Compute the undo record pointer offset given the undo rec page offset and the block number. */
+#define UndoRecPageOffsetGetRecPtr(offset, blkno)             \
+	((blkno * BLCKSZ) + offset)
+
 /* Compare two undo checkpoint files to find the oldest file. */
 #define UndoCheckPointFilenamePrecedes(file1, file2)	\
 	(strcmp(file1, file2) < 0)
diff --git a/src/include/access/undorecord.h b/src/include/access/undorecord.h
new file mode 100644
index 00000000000..1c9afba4d6a
--- /dev/null
+++ b/src/include/access/undorecord.h
@@ -0,0 +1,274 @@
+/*-------------------------------------------------------------------------
+ *
+ * undorecord.h
+ *	  encode and decode undo records
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/undorecord.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef UNDORECORD_H
+#define UNDORECORD_H
+
+#include "access/undolog.h"
+#include "access/transam.h"
+#include "lib/stringinfo.h"
+#include "storage/block.h"
+#include "storage/bufpage.h"
+#include "storage/buf.h"
+#include "storage/off.h"
+
+/*
+ * The below common information will be stored in the first undo record of the
+ * page.  Every subsequent undo record will not store this information, if it's
+ * the same as the relevant fields in the first record.
+ */
+typedef struct UndoCompressionInfo
+{
+	RmgrId		rmid;			/* rmgr ID */
+	Oid			reloid;			/* relation OID */
+	FullTransactionId fxid;
+	CommandId	cid;			/* command id */
+} UndoCompressionInfo;
+
+/*
+ * If UREC_INFO_GROUP is set, an UndoRecordGroup structure follows.
+ * If UREC_INFO_RMID is set, rmgr id follows.
+ * if UREC_INFO_RELOID	is set, relation oid follows.
+ * If UREC_INFO_FXID	is set, full transaction id follows.
+ * If UREC_INFO_CID	is set, command id follows.
+ * If UREC_INFO_FORK is set, fork number follows.
+ * If UREC_INFO_PREVUNDO is set, previous undo record pointer follows.
+ * If UREC_INFO_BLOCK is set, an UndoRecordBlock structure follows.
+ * If UREC_INFO_LOGSWITCH is set, an UndoRecordLogSwitch structure follows.
+ * If UREC_INFO_PAYLOAD is set, an UndoRecordPayload structure follows.
+ *
+ * When (as will often be the case) multiple structures are present, they
+ * appear in the same order in which the constants are defined here.  That is,
+ * UndoRecordTransaction appears first.
+ */
+#define UREC_INFO_GROUP						0x001
+#define UREC_INFO_RMID						0x002
+#define UREC_INFO_RELOID					0x004
+#define UREC_INFO_FXID						0x008
+#define UREC_INFO_CID						0x010
+#define UREC_INFO_FORK						0x020
+#define UREC_INFO_PREVUNDO					0x040
+#define UREC_INFO_BLOCK						0x080
+#define UREC_INFO_LOGSWITCH					0x100
+#define UREC_INFO_PAYLOAD					0x200
+
+#define UREC_INFO_PAGE_COMMON  (UREC_INFO_RMID | UREC_INFO_RELOID | UREC_INFO_FXID | UREC_INFO_CID)
+
+/*
+ * Every undo record begins with an UndoRecordHeader structure, which is
+ * followed by the additional structures indicated by the contents of
+ * urec_info.  All structures are packed into undo pages without considering
+ * the alignment and without trailing padding bytes. So care must be taken when
+ * reading the header.
+ */
+typedef struct UndoRecordHeader
+{
+	uint8		urec_type;		/* record type code */
+	uint16		urec_info;		/* flag bits */
+} UndoRecordHeader;
+
+#define SizeOfUndoRecordHeader	\
+	(offsetof(UndoRecordHeader, urec_info) + sizeof(uint16))
+
+/*
+ * Undo group header keep the information per undo group which can be discarded
+ * in single shot.  This contains the undo record pointer to the next group.
+ * Additionally, it also contains the progress of undo apply and the dbid.
+ */
+typedef struct UndoRecordGroup
+{
+	/*
+	 * Undo block number where we need to start reading the undo for applying
+	 * the undo action.   InvalidBlockNumber means undo applying hasn't
+	 * started for the transaction and MaxBlockNumber mean undo completely
+	 * applied. And, any other block number means we have applied partial undo
+	 * so next we can start from this block.
+	 */
+	BlockNumber urec_progress;
+	Oid			urec_dbid;		/* database id */
+	UndoRecPtr	urec_next_group;	/* urec pointer of the next group */
+} UndoRecordGroup;
+
+#define SizeOfUndoRecordGroup \
+	(offsetof(UndoRecordGroup, urec_next_group) + sizeof(UndoRecPtr))
+
+/*
+ *  Information for a block to which this record pertains.
+ */
+typedef struct UndoRecordBlock
+{
+	BlockNumber urec_block;		/* block number */
+	OffsetNumber urec_offset;	/* offset number */
+} UndoRecordBlock;
+
+#define SizeOfUndoRecordBlock \
+	(offsetof(UndoRecordBlock, urec_offset) + sizeof(OffsetNumber))
+
+/*
+ * Information of the transaction's undo in the previous log.  If a transaction
+ * is split across the undo logs then this header will be included in the first
+ * undo record of the transaction in next log.
+ */
+typedef struct UndoRecordLogSwitch
+{
+	UndoRecPtr	urec_prevlogstart;	/* Transaction's first undo record pointer
+									 * in previous undo log. */
+} UndoRecordLogSwitch;
+
+#define SizeOfUndoRecordLogSwitch \
+	(offsetof(UndoRecordLogSwitch, urec_prevlogstart) + sizeof(UndoRecPtr))
+
+/*
+ * Information about the amount of payload data and tuple data present
+ * in this record.  The payload bytes immediately follow the structures
+ * specified by flag bits in urec_info, and the tuple bytes follow the
+ * payload bytes.
+ */
+typedef struct UndoRecordPayload
+{
+	uint16		urec_payload_len;	/* # of payload bytes */
+	uint16		urec_tuple_len; /* # of tuple bytes */
+} UndoRecordPayload;
+
+#define SizeOfUndoRecordPayload \
+	(offsetof(UndoRecordPayload, urec_tuple_len) + sizeof(uint16))
+
+typedef enum UndoPackStage
+{
+	UNDO_PACK_STAGE_HEADER,		/* We have not yet processed even the record
+								 * header; we need to do that next. */
+	UNDO_PACK_STAGE_GROUP,		/* The next thing to be processed is the group
+								 * header, if present. */
+	UNDO_PACK_STAGE_RMID,		/* The next thing to be processed is the rmid
+								 * if present */
+
+	UNDO_PACK_STAGE_RELOID,		/* The next thing to be processed is the
+								 * reloid if present */
+
+	UNDO_PACK_STAGE_XID,		/* The next thing to be processed is the xid
+								 * if present */
+
+	UNDO_PACK_STAGE_CID,		/* The next thing to be processed is the cid
+								 * if present */
+
+	UNDO_PACK_STAGE_FORKNUM,	/* The next thing to be processed is the
+								 * relation fork number, if present. */
+	UNDO_PACK_STAGE_PREVUNDO,	/* The next thing to be processed is the prev
+								 * undo info. */
+
+	UNDO_PACK_STAGE_BLOCK,		/* The next thing to be processed is the block
+								 * details, if present. */
+	UNDO_PACK_STAGE_LOGSWITCH,	/* The next thing to be processed is the log
+								 * switch details. */
+	UNDO_PACK_STAGE_PAYLOAD,	/* The next thing to be processed is the
+								 * payload details, if present */
+	UNDO_PACK_STAGE_PAYLOAD_DATA,	/* The next thing to be processed is the
+									 * payload data */
+	UNDO_PACK_STAGE_TUPLE_DATA, /* The next thing to be processed is the tuple
+								 * data */
+	UNDO_PACK_STAGE_UNDO_LENGTH,	/* Next thing to processed is undo length. */
+
+	UNDO_PACK_STAGE_DONE		/* complete */
+} UndoPackStage;
+
+/*
+ * Undo record context for inserting/unpacking undo record.  This will hold
+ * intermediate state of undo record processed so far.
+ */
+typedef struct UndoPackContext
+{
+	UndoRecordHeader urec_hd;	/* Main header */
+	UndoRecordGroup urec_group; /* Undo record group header */
+	RmgrId		urec_rmid;		/* rmgrid */
+	Oid			urec_reloid;	/* relation OID */
+
+	/*
+	 * Transaction id that has modified the tuple for which this undo record
+	 * is written.  We use this to skip the undo records.  See comments atop
+	 * function UndoFetchRecord.
+	 */
+	FullTransactionId urec_fxid;	/* Transaction id */
+	CommandId	urec_cid;		/* command id */
+
+	ForkNumber	urec_fork;		/* Relation fork number */
+	UndoRecPtr	urec_prevundo;	/* Block prev */
+	UndoRecordBlock urec_blk;	/* Block header */
+	UndoRecordLogSwitch urec_logswitch; /* Log switch header */
+	UndoRecordPayload urec_payload; /* Payload data */
+	char	   *urec_payloaddata;
+	char	   *urec_tupledata;
+	uint16		undo_len;		/* Length of the undo record. */
+	int			already_processed;	/* Number of bytes read/written so far */
+	int			partial_bytes;	/* Number of partial bytes read/written */
+	UndoPackStage stage;		/* Undo pack stage */
+} UndoPackContext;
+
+/*
+ * Information that can be used to create an undo record or that can be
+ * extracted from one previously created.  The raw undo record format is
+ * difficult to manage, so this structure provides a convenient intermediate
+ * form that is easier for callers to manage.
+ *
+ * When creating an undo record from an UnpackedUndoRecord, caller should
+ * set uur_info to 0.  It will be initialized by the first call to
+ * UndoRecordSetInfo or InsertUndoRecord.  We do set it in
+ * UndoRecordAllocate for transaction specific header information.
+ *
+ * When an undo record is decoded into an UnpackedUndoRecord, all fields
+ * will be initialized, but those for which no information is available
+ * will be set to invalid or default values, as appropriate.
+ */
+typedef struct UnpackedUndoRecord
+{
+	RmgrId		uur_rmid;		/* rmgr ID */
+	uint8		uur_type;		/* record type code */
+	uint16		uur_info;		/* flag bits */
+	Oid			uur_reloid;		/* relation OID */
+	CommandId	uur_cid;		/* command id */
+	ForkNumber	uur_fork;		/* fork number */
+	UndoRecPtr	uur_prevundo;	/* byte offset of previous undo for block */
+	BlockNumber uur_block;		/* block number */
+	OffsetNumber uur_offset;	/* offset number */
+	FullTransactionId uur_fxid; /* transaction id */
+	StringInfoData uur_payload; /* payload bytes */
+	StringInfoData uur_tuple;	/* tuple bytes */
+
+	/*
+	 * Below header will be internally set by the undo layer.  Above this all
+	 * information should be set by the caller.
+	 */
+	UndoRecordGroup *uur_group; /* Group header, included in the first record
+								 * of the group in a undo log. */
+	UndoRecordLogSwitch *uur_logswitch; /* Log switch header, included in the
+										 * first record of the transaction
+										 * only after undo log is switched
+										 * during a transaction. */
+} UnpackedUndoRecord;
+
+extern size_t UndoRecordExpectedSize(UnpackedUndoRecord *uur);
+extern size_t UndoRecordPayloadSize(UnpackedUndoRecord *uur);
+extern void BeginInsertUndo(UndoPackContext *ucontext, UnpackedUndoRecord *uur,
+							uint16 undo_len);
+extern void InsertUndoData(UndoPackContext *ucontext, Page page,
+						   int starting_byte);
+extern void SkipInsertingUndoData(UndoPackContext *ucontext,
+								  int bytes_to_skip);
+extern void BeginUnpackUndo(UndoPackContext *ucontext);
+extern void UnpackUndoData(UndoPackContext *ucontext, Page page,
+						   int starting_byte);
+extern void FinishUnpackUndo(UndoPackContext *ucontext,
+							 UnpackedUndoRecord *uur);
+extern void UndoRecordGetCompressionInfo(Page page, int starting_byte,
+										 UndoCompressionInfo *compresssion_info);
+extern void UndoRecordSetInfo(UnpackedUndoRecord *uur);
+
+#endif							/* UNDORECORD_H */
diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h
index 4ef6d8ddd4a..5c65b96a0cb 100644
--- a/src/include/storage/bufpage.h
+++ b/src/include/storage/bufpage.h
@@ -215,6 +215,40 @@ typedef PageHeaderData *PageHeader;
  */
 #define SizeOfPageHeaderData (offsetof(PageHeaderData, pd_linp))
 
+/*
+ * Same as PageHeaderData + some additional information to detect partial
+ * undo record on a undo page.
+ *
+ * FIXME : for undo page do we need to keep all the information which is
+ * required for the PageHeaderData e.g. pd_lower, pd_upper, pd_special?
+ */
+typedef struct UndoPageHeaderData
+{
+	/* XXX LSN is member of *any* block, not only page-organized ones */
+	PageXLogRecPtr pd_lsn;		/* LSN: next byte after last byte of xlog
+								 * record for last change to this page */
+	uint16		pd_checksum;	/* checksum */
+	uint16		pd_flags;		/* flag bits, see below */
+	LocationIndex pd_lower;		/* offset to start of free space */
+	LocationIndex pd_upper;		/* offset to end of free space */
+	LocationIndex pd_special;	/* offset to start of special space */
+	uint16		pd_pagesize_version;
+
+	/*
+	 * Below fields required for computing the offset of the first complete
+	 * record on a undo page, which will be used for the undo record compression
+	 * and undo page consistency checking.
+	 */
+	uint16		uur_info;		/* uur_info field of the partial record. */
+	uint16		record_offset;	/* offset of the partial undo record. */
+	uint16		undo_len;		/* Length of the complete undo record. */
+} UndoPageHeaderData;
+
+typedef UndoPageHeaderData *UndoPageHeader;
+
+#define SizeOfUndoPageHeaderData (offsetof(UndoPageHeaderData, undo_len) + \
+								  sizeof(uint16))
+
 /*
  * PageIsEmpty
  *		returns true iff no itemid has been allocated on the page
@@ -419,6 +453,8 @@ do { \
 						((is_heap) ? PAI_IS_HEAP : 0))
 
 extern void PageInit(Page page, Size pageSize, Size specialSize);
+extern void UndoPageInit(Page page, Size pageSize, uint16 uur_info,
+						 uint16 record_offset, uint16 len);
 extern bool PageIsVerified(Page page, BlockNumber blkno);
 extern OffsetNumber PageAddItemExtended(Page page, Item item, Size size,
 										OffsetNumber offsetNumber, int flags);
-- 
2.22.0

