/*
 * Replay the clearing of F_FOLLOW_RIGHT flag on a child page.
 *
 * Even if the WAL record includes a full-page image, we have to update the
 * follow-right flag, because that change is not included in the full-page
 * image.  To be sure that the intermediate state with the wrong flag value is
 * not visible to concurrent Hot Standby queries, this function handles
 * restoring the full-page image as well as updating the flag.	(Note that
 * we never need to do anything else to the child page in the current WAL
 * action.)
 */
static void
gistRedoClearFollowRight(XLogRecPtr lsn, XLogRecord *record, int block_index,
						 RelFileNode node, BlockNumber childblkno)
{
	Buffer		buffer;
	Page		page;

	if (record->xl_info & XLR_BKP_BLOCK(block_index))
		buffer = RestoreBackupBlock(lsn, record, block_index, false, true);
	else
	{
		buffer = XLogReadBuffer(node, childblkno, false);
		if (!BufferIsValid(buffer))
			return;				/* page was deleted, nothing to do */
	}
	page = (Page) BufferGetPage(buffer);

	/*
	 * Note that we still update the page even if page LSN is equal to the LSN
	 * of this record, because the updated NSN is not included in the full
	 * page image.
	 */
	if (lsn >= PageGetLSN(page))
	{
		GistPageSetNSN(page, lsn);
		GistClearFollowRight(page);

		PageSetLSN(page, lsn);
		MarkBufferDirty(buffer);
	}
	UnlockReleaseBuffer(buffer);
}

static void
spgRedoVacuumRoot(XLogRecPtr lsn, XLogRecord *record)
{
	char	   *ptr = XLogRecGetData(record);
	spgxlogVacuumRoot *xldata = (spgxlogVacuumRoot *) ptr;
	OffsetNumber *toDelete;
	Buffer		buffer;
	Page		page;

	ptr += sizeof(spgxlogVacuumRoot);
	toDelete = (OffsetNumber *) ptr;

	if (!(record->xl_info & XLR_BKP_BLOCK_1))
	{
		buffer = XLogReadBuffer(xldata->node, SPGIST_HEAD_BLKNO, false);
		if (BufferIsValid(buffer))
		{
			page = BufferGetPage(buffer);
			if (!XLByteLE(lsn, PageGetLSN(page)))
			{
				/* The tuple numbers are in order */
				PageIndexMultiDelete(page, toDelete, xldata->nDelete);

				PageSetLSN(page, lsn);
				PageSetTLI(page, ThisTimeLineID);
				MarkBufferDirty(buffer);
			}
			UnlockReleaseBuffer(buffer);
		}
	}
}

static void
ginRedoVacuumDataLeafPage(XLogRecPtr lsn, XLogRecord *record)
{
    ginxlogVacuumDataLeafPage *xlrec = (ginxlogVacuumDataLeafPage *) XLogRecGetData(record);
    Buffer		buffer;
    Page		page;

    /* If we have a full-page image, restore it and we're done */
    if (record->xl_info & XLR_BKP_BLOCK(0))
    {
        (void) RestoreBackupBlock(lsn, record, 0, false, false);
        return;
    }

    buffer = XLogReadBuffer(xlrec->node, xlrec->blkno, false);
    if (!BufferIsValid(buffer))
        return;
    page = (Page) BufferGetPage(buffer);

    Assert(GinPageIsLeaf(page));
    Assert(GinPageIsData(page));

    if (lsn > PageGetLSN(page))
    {
        ginRedoRecompress(page, &xlrec->data);
        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }

    UnlockReleaseBuffer(buffer);
}

/*
 * XLogReadBufferForRedoExtended
 *		Like XLogReadBufferForRedo, but with extra options.
 *
 * If mode is RBM_ZERO or RBM_ZERO_ON_ERROR, if the page doesn't exist, the
 * relation is extended with all-zeroes pages up to the referenced block
 * number.  In RBM_ZERO mode, the return value is always BLK_NEEDS_REDO.
 *
 * If 'get_cleanup_lock' is true, a "cleanup lock" is acquired on the buffer
 * using LockBufferForCleanup(), instead of a regular exclusive lock.
 */
XLogRedoAction
XLogReadBufferForRedoExtended(XLogRecPtr lsn, XLogRecord *record,
							  int block_index, RelFileNode rnode,
							  ForkNumber forkno, BlockNumber blkno,
							  ReadBufferMode mode, bool get_cleanup_lock,
							  Buffer *buf)
{
	if (record->xl_info & XLR_BKP_BLOCK(block_index))
	{
		*buf = RestoreBackupBlock(lsn, record, block_index,
								  get_cleanup_lock, true);
		return BLK_RESTORED;
	}
	else
	{
		*buf = XLogReadBufferExtended(rnode, forkno, blkno, mode);
		if (BufferIsValid(*buf))
		{
			LockBuffer(*buf, BUFFER_LOCK_EXCLUSIVE);
			if (lsn <= PageGetLSN(BufferGetPage(*buf)))
				return BLK_DONE;
			else
				return BLK_NEEDS_REDO;
		}
		else
			return BLK_NOTFOUND;
	}
}

/*
 * Replay the clearing of F_FOLLOW_RIGHT flag.
 */
static void
gistRedoClearFollowRight(RelFileNode node, XLogRecPtr lsn,
						 BlockNumber leftblkno)
{
	Buffer		buffer;

	buffer = XLogReadBuffer(node, leftblkno, false);
	if (BufferIsValid(buffer))
	{
		Page		page = (Page) BufferGetPage(buffer);

		/*
		 * Note that we still update the page even if page LSN is equal to the
		 * LSN of this record, because the updated NSN is not included in the
		 * full page image.
		 */
		if (!XLByteLT(lsn, PageGetLSN(page)))
		{
			GistPageGetOpaque(page)->nsn = lsn;
			GistClearFollowRight(page);

			PageSetLSN(page, lsn);
			PageSetTLI(page, ThisTimeLineID);
			MarkBufferDirty(buffer);
		}
		UnlockReleaseBuffer(buffer);
	}
}

/*
 *	visibilitymap_set - set a bit on a previously pinned page
 *
 * recptr is the LSN of the heap page. The LSN of the visibility map page is
 * advanced to that, to make sure that the visibility map doesn't get flushed
 * to disk before the update to the heap page that made all tuples visible.
 *
 * This is an opportunistic function. It does nothing, unless *buf
 * contains the bit for heapBlk. Call visibilitymap_pin first to pin
 * the right map page. This function doesn't do any I/O.
 */
void
visibilitymap_set(Relation rel, BlockNumber heapBlk, XLogRecPtr recptr,
				  Buffer *buf)
{
	BlockNumber mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
	uint32		mapByte = HEAPBLK_TO_MAPBYTE(heapBlk);
	uint8		mapBit = HEAPBLK_TO_MAPBIT(heapBlk);
	Page		page;
	char	   *map;

#ifdef TRACE_VISIBILITYMAP
	elog(DEBUG1, "vm_set %s %d", RelationGetRelationName(rel), heapBlk);
#endif

	/* Check that we have the right page pinned */
	if (!BufferIsValid(*buf) || BufferGetBlockNumber(*buf) != mapBlock)
		return;

	page = BufferGetPage(*buf);
	map = PageGetContents(page);
	LockBuffer(*buf, BUFFER_LOCK_EXCLUSIVE);

	if (!(map[mapByte] & (1 << mapBit)))
	{
		map[mapByte] |= (1 << mapBit);

		if (XLByteLT(PageGetLSN(page), recptr))
			PageSetLSN(page, recptr);
		PageSetTLI(page, ThisTimeLineID);
		MarkBufferDirty(*buf);
	}

	LockBuffer(*buf, BUFFER_LOCK_UNLOCK);
}

static void
bitmap_xlog_insert_bitmap_lastwords(bool redo, XLogRecPtr lsn, XLogRecord* record)
{
	xl_bm_bitmap_lastwords	*xlrec = 
		(xl_bm_bitmap_lastwords*) XLogRecGetData(record);
	Relation reln;

	reln = XLogOpenRelation(xlrec->bm_node);
	if (!RelationIsValid(reln))
		return;

	if (redo)
	{
		Buffer		lovBuffer;
		Page		lovPage;
		BMLOVItem	lovItem;

#ifdef BM_DEBUG
		ereport(LOG, (errcode(LOG), 
			errmsg("call bitmap_xlog_insert_bitmap_lastwords: redo=%d\n", 
					redo)));
#endif

		lovBuffer = XLogReadBuffer(false, reln, xlrec->bm_lov_blkno);
		if (!BufferIsValid(lovBuffer))
			elog(PANIC, "bm_insert_redo: block unfound: %d",
				 xlrec->bm_lov_blkno);

		lovPage = BufferGetPage(lovBuffer);

		if (XLByteLT(PageGetLSN(lovPage), lsn))
		{
			lovItem = (BMLOVItem)
				PageGetItem(lovPage, PageGetItemId(lovPage, xlrec->bm_lov_offset));

			lovItem->bm_last_compword = xlrec->bm_last_compword;
			lovItem->bm_last_word = xlrec->bm_last_word;
			lovItem->bm_last_two_headerbits = xlrec->bm_last_two_headerbits;

			PageSetLSN(lovPage, lsn);
			PageSetTLI(lovPage, ThisTimeLineID);
			_bitmap_wrtbuf(lovBuffer);
		}

		else
			_bitmap_relbuf(lovBuffer);
	}

	else
		elog(PANIC, "bm_insert_undo: not implemented.");
}

static void
ginRedoDeleteListPages(XLogRecPtr lsn, XLogRecord *record)
{
    ginxlogDeleteListPages *data = (ginxlogDeleteListPages *) XLogRecGetData(record);
    Buffer		metabuffer;
    Page		metapage;
    int			i;

    /* Backup blocks are not used in delete_listpage records */
    Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));

    metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false);
    if (!BufferIsValid(metabuffer))
        return;					/* assume index was deleted, nothing to do */
    metapage = BufferGetPage(metabuffer);

    memcpy(GinPageGetMeta(metapage), &data->metadata, sizeof(GinMetaPageData));
    PageSetLSN(metapage, lsn);
    MarkBufferDirty(metabuffer);

    /*
     * In normal operation, shiftList() takes exclusive lock on all the
     * pages-to-be-deleted simultaneously.  During replay, however, it should
     * be all right to lock them one at a time.  This is dependent on the fact
     * that we are deleting pages from the head of the list, and that readers
     * share-lock the next page before releasing the one they are on. So we
     * cannot get past a reader that is on, or due to visit, any page we are
     * going to delete.  New incoming readers will block behind our metapage
     * lock and then see a fully updated page list.
     */
    for (i = 0; i < data->ndeleted; i++)
    {
        Buffer		buffer = XLogReadBuffer(data->node, data->toDelete[i], false);

        if (BufferIsValid(buffer))
        {
            Page		page = BufferGetPage(buffer);

            if (lsn > PageGetLSN(page))
            {
                GinPageGetOpaque(page)->flags = GIN_DELETED;

                PageSetLSN(page, lsn);
                MarkBufferDirty(buffer);
            }

            UnlockReleaseBuffer(buffer);
        }
    }
    UnlockReleaseBuffer(metabuffer);
}

/*
 * Determine whether the buffer referenced has to be backed up.
 *
 * Since we don't yet have the insert lock, fullPageWrites and forcePageWrites
 * could change later, so the result should be used for optimization purposes
 * only.
 */
bool
XLogCheckBufferNeedsBackup(Buffer buffer)
{
	XLogRecPtr	RedoRecPtr;
	bool		doPageWrites;
	Page		page;

	GetFullPageWriteInfo(&RedoRecPtr, &doPageWrites);

	page = BufferGetPage(buffer);

	if (doPageWrites && PageGetLSN(page) <= RedoRecPtr)
		return true;			/* buffer requires backup */

	return false;				/* buffer does not need to be backed up */
}

static void
killtuple(Relation r, GISTScanOpaque so, ItemPointer iptr)
{
	MIRROREDLOCK_BUFMGR_DECLARE;

	Page        p;
	OffsetNumber offset;

	// -------- MirroredLock ----------
	MIRROREDLOCK_BUFMGR_LOCK;

	LockBuffer(so->curbuf, GIST_SHARE);
	gistcheckpage(r, so->curbuf);
	p = (Page) BufferGetPage(so->curbuf);

	if (XLByteEQ(so->stack->lsn, PageGetLSN(p)))
	{
		/* page unchanged, so all is simple */
		offset = ItemPointerGetOffsetNumber(iptr);
		ItemIdMarkDead(PageGetItemId(p, offset));
		SetBufferCommitInfoNeedsSave(so->curbuf);
	}
	else
	{
		OffsetNumber maxoff = PageGetMaxOffsetNumber(p);

		for (offset = FirstOffsetNumber; offset <= maxoff; offset = OffsetNumberNext(offset))
		{
			IndexTuple  ituple = (IndexTuple) PageGetItem(p, PageGetItemId(p, offset));

			if (ItemPointerEquals(&(ituple->t_tid), iptr))
			{
				/* found */
				ItemIdMarkDead(PageGetItemId(p, offset));
				SetBufferCommitInfoNeedsSave(so->curbuf);
				break;
			}
		}
	}

	LockBuffer(so->curbuf, GIST_UNLOCK);

	MIRROREDLOCK_BUFMGR_UNLOCK;
	// -------- MirroredLock ----------
}

static void
bitmap_xlog_insert_meta(bool redo, XLogRecPtr lsn, XLogRecord* record)
{
	xl_bm_metapage	*xlrec = (xl_bm_metapage*) XLogRecGetData(record);
	Relation		reln;

	reln = XLogOpenRelation(xlrec->bm_node);
	
	if (!RelationIsValid(reln))
		return;

	if (redo)
	{
		Buffer			metabuf;
		BMMetaPage			metapage;

#ifdef BM_DEBUG
		ereport(LOG, (errcode(LOG), 
			errmsg("call bitmap_xlog_insert_meta: redo=%d\n", redo)));
#endif

		metabuf = XLogReadBuffer(false, reln, BM_METAPAGE);
		if (!BufferIsValid(metabuf))
			elog(PANIC, "bm_insert_redo: block unfound: %d", BM_METAPAGE);

		/* restore the page */
		metapage = (BMMetaPage)BufferGetPage(metabuf);

		if (XLByteLT(PageGetLSN(metapage), lsn))
		{
			PageSetLSN(metapage, lsn);
			PageSetTLI(metapage, ThisTimeLineID);
			_bitmap_wrtbuf(metabuf);
		}

		else
			_bitmap_relbuf(metabuf);
	}

	else
		elog(PANIC, "bm_insert_undo: not implemented.");
}

static void
ginRedoClearIncompleteSplit(XLogRecPtr lsn, RelFileNode node, BlockNumber blkno)
{
    Buffer		buffer;
    Page		page;

    buffer = XLogReadBuffer(node, blkno, false);
    if (!BufferIsValid(buffer))
        return;					/* page was deleted, nothing to do */
    page = (Page) BufferGetPage(buffer);

    if (lsn > PageGetLSN(page))
    {
        GinPageGetOpaque(page)->flags &= ~GIN_INCOMPLETE_SPLIT;

        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }

    UnlockReleaseBuffer(buffer);
}

static void
gistfindleaf(GISTInsertState *state, GISTSTATE *giststate)
{
	ItemId		iid;
	IndexTuple	idxtuple;
	GISTPageOpaque opaque;

	MIRROREDLOCK_BUFMGR_MUST_ALREADY_BE_HELD;

	/*
	 * walk down, We don't lock page for a long time, but so we should be
	 * ready to recheck path in a bad case... We remember, that page->lsn
	 * should never be invalid.
	 */
	for (;;)
	{
		if (XLogRecPtrIsInvalid(state->stack->lsn))
			state->stack->buffer = ReadBuffer(state->r, state->stack->blkno);
		LockBuffer(state->stack->buffer, GIST_SHARE);
		gistcheckpage(state->r, state->stack->buffer);

		state->stack->page = (Page) BufferGetPage(state->stack->buffer);
		opaque = GistPageGetOpaque(state->stack->page);

		state->stack->lsn = PageGetLSN(state->stack->page);
		Assert(state->r->rd_istemp || !XLogRecPtrIsInvalid(state->stack->lsn));

		if (state->stack->blkno != GIST_ROOT_BLKNO &&
			XLByteLT(state->stack->parent->lsn, opaque->nsn))
		{
			/*
			 * caused split non-root page is detected, go up to parent to
			 * choose best child
			 */
			UnlockReleaseBuffer(state->stack->buffer);
			state->stack = state->stack->parent;
			continue;
		}

		if (!GistPageIsLeaf(state->stack->page))
		{
			/*
			 * This is an internal page, so continue to walk down the tree. We
			 * find the child node that has the minimum insertion penalty and
			 * recursively invoke ourselves to modify that node. Once the
			 * recursive call returns, we may need to adjust the parent node
			 * for two reasons: the child node split, or the key in this node
			 * needs to be adjusted for the newly inserted key below us.
			 */
			GISTInsertStack *item = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack));

			state->stack->childoffnum = gistchoose(state->r, state->stack->page, state->itup[0], giststate);

			iid = PageGetItemId(state->stack->page, state->stack->childoffnum);
			idxtuple = (IndexTuple) PageGetItem(state->stack->page, iid);
			item->blkno = ItemPointerGetBlockNumber(&(idxtuple->t_tid));
			LockBuffer(state->stack->buffer, GIST_UNLOCK);

			item->parent = state->stack;
			item->child = NULL;
			if (state->stack)
				state->stack->child = item;
			state->stack = item;
		}
		else
		{
			/* be carefull, during unlock/lock page may be changed... */
			LockBuffer(state->stack->buffer, GIST_UNLOCK);
			LockBuffer(state->stack->buffer, GIST_EXCLUSIVE);
			state->stack->page = (Page) BufferGetPage(state->stack->buffer);
			opaque = GistPageGetOpaque(state->stack->page);

			if (state->stack->blkno == GIST_ROOT_BLKNO)
			{
				/*
				 * the only page can become inner instead of leaf is a root
				 * page, so for root we should recheck it
				 */
				if (!GistPageIsLeaf(state->stack->page))
				{
					/*
					 * very rarely situation: during unlock/lock index with
					 * number of pages = 1 was increased
					 */
					LockBuffer(state->stack->buffer, GIST_UNLOCK);
					continue;
				}

				/*
				 * we don't need to check root split, because checking
				 * leaf/inner is enough to recognize split for root
				 */

			}
			else if (XLByteLT(state->stack->parent->lsn, opaque->nsn))
			{
				/*
				 * detecting split during unlock/lock, so we should find
				 * better child on parent
				 */
//.........这里部分代码省略.........

static void
ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
{
    ginxlogUpdateMeta *data = (ginxlogUpdateMeta *) XLogRecGetData(record);
    Buffer		metabuffer;
    Page		metapage;
    Buffer		buffer;

    /*
     * Restore the metapage. This is essentially the same as a full-page
     * image, so restore the metapage unconditionally without looking at the
     * LSN, to avoid torn page hazards.
     */
    metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false);
    if (!BufferIsValid(metabuffer))
        return;					/* assume index was deleted, nothing to do */
    metapage = BufferGetPage(metabuffer);

    memcpy(GinPageGetMeta(metapage), &data->metadata, sizeof(GinMetaPageData));
    PageSetLSN(metapage, lsn);
    MarkBufferDirty(metabuffer);

    if (data->ntuples > 0)
    {
        /*
         * insert into tail page
         */
        if (record->xl_info & XLR_BKP_BLOCK(0))
            (void) RestoreBackupBlock(lsn, record, 0, false, false);
        else
        {
            buffer = XLogReadBuffer(data->node, data->metadata.tail, false);
            if (BufferIsValid(buffer))
            {
                Page		page = BufferGetPage(buffer);

                if (lsn > PageGetLSN(page))
                {
                    OffsetNumber l,
                                 off = (PageIsEmpty(page)) ? FirstOffsetNumber :
                                       OffsetNumberNext(PageGetMaxOffsetNumber(page));
                    int			i,
                                tupsize;
                    IndexTuple	tuples = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogUpdateMeta));

                    for (i = 0; i < data->ntuples; i++)
                    {
                        tupsize = IndexTupleSize(tuples);

                        l = PageAddItem(page, (Item) tuples, tupsize, off, false, false);

                        if (l == InvalidOffsetNumber)
                            elog(ERROR, "failed to add item to index page");

                        tuples = (IndexTuple) (((char *) tuples) + tupsize);

                        off++;
                    }

                    /*
                     * Increase counter of heap tuples
                     */
                    GinPageGetOpaque(page)->maxoff++;

                    PageSetLSN(page, lsn);
                    MarkBufferDirty(buffer);
                }
                UnlockReleaseBuffer(buffer);
            }
        }
    }
    else if (data->prevTail != InvalidBlockNumber)
    {
        /*
         * New tail
         */
        if (record->xl_info & XLR_BKP_BLOCK(0))
            (void) RestoreBackupBlock(lsn, record, 0, false, false);
        else
        {
            buffer = XLogReadBuffer(data->node, data->prevTail, false);
            if (BufferIsValid(buffer))
            {
                Page		page = BufferGetPage(buffer);

                if (lsn > PageGetLSN(page))
                {
                    GinPageGetOpaque(page)->rightlink = data->newRightlink;

                    PageSetLSN(page, lsn);
                    MarkBufferDirty(buffer);
                }
                UnlockReleaseBuffer(buffer);
            }
        }
    }

    UnlockReleaseBuffer(metabuffer);
}

static void
ginRedoDeletePage(XLogRecPtr lsn, XLogRecord *record)
{
    ginxlogDeletePage *data = (ginxlogDeletePage *) XLogRecGetData(record);
    Buffer		dbuffer;
    Buffer		pbuffer;
    Buffer		lbuffer;
    Page		page;

    if (record->xl_info & XLR_BKP_BLOCK(0))
        dbuffer = RestoreBackupBlock(lsn, record, 0, false, true);
    else
    {
        dbuffer = XLogReadBuffer(data->node, data->blkno, false);
        if (BufferIsValid(dbuffer))
        {
            page = BufferGetPage(dbuffer);
            if (lsn > PageGetLSN(page))
            {
                Assert(GinPageIsData(page));
                GinPageGetOpaque(page)->flags = GIN_DELETED;
                PageSetLSN(page, lsn);
                MarkBufferDirty(dbuffer);
            }
        }
    }

    if (record->xl_info & XLR_BKP_BLOCK(1))
        pbuffer = RestoreBackupBlock(lsn, record, 1, false, true);
    else
    {
        pbuffer = XLogReadBuffer(data->node, data->parentBlkno, false);
        if (BufferIsValid(pbuffer))
        {
            page = BufferGetPage(pbuffer);
            if (lsn > PageGetLSN(page))
            {
                Assert(GinPageIsData(page));
                Assert(!GinPageIsLeaf(page));
                GinPageDeletePostingItem(page, data->parentOffset);
                PageSetLSN(page, lsn);
                MarkBufferDirty(pbuffer);
            }
        }
    }

    if (record->xl_info & XLR_BKP_BLOCK(2))
        (void) RestoreBackupBlock(lsn, record, 2, false, false);
    else if (data->leftBlkno != InvalidBlockNumber)
    {
        lbuffer = XLogReadBuffer(data->node, data->leftBlkno, false);
        if (BufferIsValid(lbuffer))
        {
            page = BufferGetPage(lbuffer);
            if (lsn > PageGetLSN(page))
            {
                Assert(GinPageIsData(page));
                GinPageGetOpaque(page)->rightlink = data->rightLink;
                PageSetLSN(page, lsn);
                MarkBufferDirty(lbuffer);
            }
            UnlockReleaseBuffer(lbuffer);
        }
    }

    if (BufferIsValid(pbuffer))
        UnlockReleaseBuffer(pbuffer);
    if (BufferIsValid(dbuffer))
        UnlockReleaseBuffer(dbuffer);
}

static void
ginRedoInsert(XLogRecPtr lsn, XLogRecord *record)
{
    ginxlogInsert *data = (ginxlogInsert *) XLogRecGetData(record);
    Buffer		buffer;
    Page		page;
    char	   *payload;
    BlockNumber leftChildBlkno = InvalidBlockNumber;
    BlockNumber rightChildBlkno = InvalidBlockNumber;
    bool		isLeaf = (data->flags & GIN_INSERT_ISLEAF) != 0;

    payload = XLogRecGetData(record) + sizeof(ginxlogInsert);

    /*
     * First clear incomplete-split flag on child page if this finishes a
     * split.
     */
    if (!isLeaf)
    {
        leftChildBlkno = BlockIdGetBlockNumber((BlockId) payload);
        payload += sizeof(BlockIdData);
        rightChildBlkno = BlockIdGetBlockNumber((BlockId) payload);
        payload += sizeof(BlockIdData);

        if (record->xl_info & XLR_BKP_BLOCK(0))
            (void) RestoreBackupBlock(lsn, record, 0, false, false);
        else
            ginRedoClearIncompleteSplit(lsn, data->node, leftChildBlkno);
    }

    /* If we have a full-page image, restore it and we're done */
    if (record->xl_info & XLR_BKP_BLOCK(isLeaf ? 0 : 1))
    {
        (void) RestoreBackupBlock(lsn, record, isLeaf ? 0 : 1, false, false);
        return;
    }

    buffer = XLogReadBuffer(data->node, data->blkno, false);
    if (!BufferIsValid(buffer))
        return;					/* page was deleted, nothing to do */
    page = (Page) BufferGetPage(buffer);

    if (lsn > PageGetLSN(page))
    {
        /* How to insert the payload is tree-type specific */
        if (data->flags & GIN_INSERT_ISDATA)
        {
            Assert(GinPageIsData(page));
            ginRedoInsertData(buffer, isLeaf, rightChildBlkno, payload);
        }
        else
        {
            Assert(!GinPageIsData(page));
            ginRedoInsertEntry(buffer, isLeaf, rightChildBlkno, payload);
        }

        PageSetLSN(page, lsn);
        MarkBufferDirty(buffer);
    }

    UnlockReleaseBuffer(buffer);
}

static int
FileRepPrimary_ResyncBufferPoolIncrementalWrite(ChangeTrackingRequest *request)
{
	int				status = STATUS_OK;
	Page			page;
	Buffer			buf; 
	BlockNumber		numBlocks = 0;
	SMgrRelation	smgr_relation = NULL;
	char			relidstr[OIDCHARS + 1 + OIDCHARS + 1 + OIDCHARS + 1];
	int				ii;
	XLogRecPtr		loc;
	XLogRecPtr		loc1;
	int				count = 0;
	int				thresholdCount = 0;
	bool			mirrorDataLossOccurred = FALSE;
	int				NumberOfRelations = request->count;
	
	FileRepResyncHashEntry_s	entry;
	ChangeTrackingResult		*result = NULL;	

	while (1)
	{
		/* allow flushing buffers from buffer pool during scan */
		FileRepResync_SetReadBufferRequest();
		if ((result = ChangeTracking_GetChanges(request)) != NULL) 
		{
			FileRepResync_ResetReadBufferRequest();
					
			for (ii = 0; ii < result->count; ii++)
			{
				
				if (smgr_relation == NULL)
				{
					NumberOfRelations--;
					
					smgr_relation = smgropen(result->entries[ii].relFileNode);
					
					snprintf(relidstr, sizeof(relidstr), "%u/%u/%u",
							 smgr_relation->smgr_rnode.spcNode,
							 smgr_relation->smgr_rnode.dbNode,
							 smgr_relation->smgr_rnode.relNode);

					numBlocks = smgrnblocks(smgr_relation);
					
					if (Debug_filerep_print)
						elog(LOG, "resynchronize buffer pool relation '%u/%u/%u' "
							 "number of blocks:'%u' ",
							 smgr_relation->smgr_rnode.spcNode,
							 smgr_relation->smgr_rnode.dbNode,
							 smgr_relation->smgr_rnode.relNode,
							 numBlocks);
					
					thresholdCount = Min(numBlocks, 1024);
				}
				
				loc1 =  result->entries[ii].lsn_end;
				
				/*
				 * if relation was truncated then block_num from change tracking can be beyond numBlocks 
				 */
				if (result->entries[ii].block_num >=  numBlocks)
				{
					ereport(LOG,	
							(errmsg("could not resynchonize buffer pool relation '%s' block '%d' (maybe due to truncate), "
									"lsn change tracking '%s(%u/%u)' "
									"number of blocks '%d' ",
									relidstr,
									result->entries[ii].block_num,
									XLogLocationToString(&loc1),
									loc1.xlogid,
									loc1.xrecoff,
									numBlocks),						
							 FileRep_errcontext()));						
					
					goto flush_check;
				}
				
				/* allow flushing buffers from buffer pool during scan */
				FileRepResync_SetReadBufferRequest();
				buf = ReadBuffer_Resync(smgr_relation,
										result->entries[ii].block_num,
										relidstr);
				FileRepResync_ResetReadBufferRequest();
				
				Assert(result->entries[ii].block_num < numBlocks);
				
				LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
				page = BufferGetPage(buf);
				
				loc = PageGetLSN(page); 
				
				if(Debug_filerep_print)
				{
					elog(LOG,	
							"incremental resync buffer pool identifier '%s' num blocks '%d' blkno '%d' lsn page '%s(%u/%u)' "
							"lsn end change tracking '%s(%u/%u)' ",
							relidstr,
							numBlocks,
							result->entries[ii].block_num,
							XLogLocationToString(&loc),
//.........这里部分代码省略.........

/*
 * Updates the stack so that child->parent is the correct parent of the
 * child. child->parent must be exclusively locked on entry, and will
 * remain so at exit, but it might not be the same page anymore.
 */
static void
gistFindCorrectParent(Relation r, GISTInsertStack *child)
{
	GISTInsertStack *parent = child->parent;

	gistcheckpage(r, parent->buffer);
	parent->page = (Page) BufferGetPage(parent->buffer);

	/* here we don't need to distinguish between split and page update */
	if (child->downlinkoffnum == InvalidOffsetNumber ||
		parent->lsn != PageGetLSN(parent->page))
	{
		/* parent is changed, look child in right links until found */
		OffsetNumber i,
					maxoff;
		ItemId		iid;
		IndexTuple	idxtuple;
		GISTInsertStack *ptr;

		while (true)
		{
			maxoff = PageGetMaxOffsetNumber(parent->page);
			for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
			{
				iid = PageGetItemId(parent->page, i);
				idxtuple = (IndexTuple) PageGetItem(parent->page, iid);
				if (ItemPointerGetBlockNumber(&(idxtuple->t_tid)) == child->blkno)
				{
					/* yes!!, found */
					child->downlinkoffnum = i;
					return;
				}
			}

			parent->blkno = GistPageGetOpaque(parent->page)->rightlink;
			UnlockReleaseBuffer(parent->buffer);
			if (parent->blkno == InvalidBlockNumber)
			{
				/*
				 * End of chain and still didn't find parent. It's a very-very
				 * rare situation when root splited.
				 */
				break;
			}
			parent->buffer = ReadBuffer(r, parent->blkno);
			LockBuffer(parent->buffer, GIST_EXCLUSIVE);
			gistcheckpage(r, parent->buffer);
			parent->page = (Page) BufferGetPage(parent->buffer);
		}

		/*
		 * awful!!, we need search tree to find parent ... , but before we
		 * should release all old parent
		 */

		ptr = child->parent->parent;	/* child->parent already released
										 * above */
		while (ptr)
		{
			ReleaseBuffer(ptr->buffer);
			ptr = ptr->parent;
		}

		/* ok, find new path */
		ptr = parent = gistFindPath(r, child->blkno, &child->downlinkoffnum);

		/* read all buffers as expected by caller */
		/* note we don't lock them or gistcheckpage them here! */
		while (ptr)
		{
			ptr->buffer = ReadBuffer(r, ptr->blkno);
			ptr->page = (Page) BufferGetPage(ptr->buffer);
			ptr = ptr->parent;
		}

		/* install new chain of parents to stack */
		child->parent = parent;

		/* make recursive call to normal processing */
		LockBuffer(child->parent->buffer, GIST_EXCLUSIVE);
		gistFindCorrectParent(r, child);
	}

	return;
}

/*
 * Workhouse routine for doing insertion into a GiST index. Note that
 * this routine assumes it is invoked in a short-lived memory context,
 * so it does not bother releasing palloc'd allocations.
 */
void
gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate)
{
	ItemId		iid;
	IndexTuple	idxtuple;
	GISTInsertStack firststack;
	GISTInsertStack *stack;
	GISTInsertState state;
	bool		xlocked = false;

	memset(&state, 0, sizeof(GISTInsertState));
	state.freespace = freespace;
	state.r = r;

	/* Start from the root */
	firststack.blkno = GIST_ROOT_BLKNO;
	firststack.lsn = 0;
	firststack.parent = NULL;
	firststack.downlinkoffnum = InvalidOffsetNumber;
	state.stack = stack = &firststack;

	/*
	 * Walk down along the path of smallest penalty, updating the parent
	 * pointers with the key we're inserting as we go. If we crash in the
	 * middle, the tree is consistent, although the possible parent updates
	 * were a waste.
	 */
	for (;;)
	{
		if (XLogRecPtrIsInvalid(stack->lsn))
			stack->buffer = ReadBuffer(state.r, stack->blkno);

		/*
		 * Be optimistic and grab shared lock first. Swap it for an exclusive
		 * lock later if we need to update the page.
		 */
		if (!xlocked)
		{
			LockBuffer(stack->buffer, GIST_SHARE);
			gistcheckpage(state.r, stack->buffer);
		}

		stack->page = (Page) BufferGetPage(stack->buffer);
		stack->lsn = PageGetLSN(stack->page);
		Assert(!RelationNeedsWAL(state.r) || !XLogRecPtrIsInvalid(stack->lsn));

		/*
		 * If this page was split but the downlink was never inserted to the
		 * parent because the inserting backend crashed before doing that, fix
		 * that now.
		 */
		if (GistFollowRight(stack->page))
		{
			if (!xlocked)
			{
				LockBuffer(stack->buffer, GIST_UNLOCK);
				LockBuffer(stack->buffer, GIST_EXCLUSIVE);
				xlocked = true;
				/* someone might've completed the split when we unlocked */
				if (!GistFollowRight(stack->page))
					continue;
			}
			gistfixsplit(&state, giststate);

			UnlockReleaseBuffer(stack->buffer);
			xlocked = false;
			state.stack = stack = stack->parent;
			continue;
		}

		if (stack->blkno != GIST_ROOT_BLKNO &&
			stack->parent->lsn < GistPageGetNSN(stack->page))
		{
			/*
			 * Concurrent split detected. There's no guarantee that the
			 * downlink for this page is consistent with the tuple we're
			 * inserting anymore, so go back to parent and rechoose the best
			 * child.
			 */
			UnlockReleaseBuffer(stack->buffer);
			xlocked = false;
			state.stack = stack = stack->parent;
			continue;
		}

		if (!GistPageIsLeaf(stack->page))
		{
			/*
			 * This is an internal page so continue to walk down the tree.
			 * Find the child node that has the minimum insertion penalty.
			 */
			BlockNumber childblkno;
			IndexTuple	newtup;
			GISTInsertStack *item;
			OffsetNumber downlinkoffnum;
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		{
			PageRestoreTempPage(dist->page, BufferGetPage(dist->buffer));
			dist->page = BufferGetPage(dist->buffer);
		}

		if (!state->r->rd_istemp)
		{
			XLogRecPtr	recptr;
			XLogRecData *rdata;

			rdata = formSplitRdata(state->r, state->stack->blkno,
								   is_leaf, &(state->key), dist);

			recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_SPLIT, rdata);

			for (ptr = dist; ptr; ptr = ptr->next)
			{
				PageSetLSN(ptr->page, recptr);
				PageSetTLI(ptr->page, ThisTimeLineID);
			}
		}
		else
		{
			for (ptr = dist; ptr; ptr = ptr->next)
			{
				PageSetLSN(ptr->page, XLogRecPtrForTemp);
			}
		}

		/* set up NSN */
		oldnsn = GistPageGetOpaque(dist->page)->nsn;
		if (state->stack->blkno == GIST_ROOT_BLKNO)
			/* if root split we should put initial value */
			oldnsn = PageGetLSN(dist->page);

		for (ptr = dist; ptr; ptr = ptr->next)
		{
			/* only for last set oldnsn */
			GistPageGetOpaque(ptr->page)->nsn = (ptr->next) ?
				PageGetLSN(ptr->page) : oldnsn;
		}

		/*
		 * release buffers, if it was a root split then release all buffers
		 * because we create all buffers
		 */
		ptr = (state->stack->blkno == GIST_ROOT_BLKNO) ? dist : dist->next;
		for (; ptr; ptr = ptr->next)
			UnlockReleaseBuffer(ptr->buffer);

		if (state->stack->blkno == GIST_ROOT_BLKNO)
		{
			gistnewroot(state->r, state->stack->buffer, state->itup, state->ituplen, &(state->key));
			state->needInsertComplete = false;
		}

		END_CRIT_SECTION();
	}
	else
	{
		/* enough space */
		START_CRIT_SECTION();

		if (!is_leaf)
			PageIndexTupleDelete(state->stack->page, state->stack->childoffnum);
		gistfillbuffer(state->r, state->stack->page, state->itup, state->ituplen, InvalidOffsetNumber);