自学教程：C++ BufferIsValid函数代码示例

/* *	visibilitymap_set - set a bit on a previously pinned page * * recptr is the LSN of the XLOG record we're replaying, if we're in recovery, * or InvalidXLogRecPtr in normal running.	The page LSN is advanced to the * one provided; in normal running, we generate a new XLOG record and set the * page LSN to that value.	cutoff_xid is the largest xmin on the page being * marked all-visible; it is needed for Hot Standby, and can be * InvalidTransactionId if the page contains no tuples. * * Caller is expected to set the heap page's PD_ALL_VISIBLE bit before calling * this function. Except in recovery, caller should also pass the heap * buffer. When checksums are enabled and we're not in recovery, we must add * the heap buffer to the WAL chain to protect it from being torn. * * You must pass a buffer containing the correct map page to this function. * Call visibilitymap_pin first to pin the right one. This function doesn't do * any I/O. */voidvisibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,				  XLogRecPtr recptr, Buffer vmBuf, TransactionId cutoff_xid){	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	Assert(InRecovery || XLogRecPtrIsInvalid(recptr));	Assert(InRecovery || BufferIsValid(heapBuf));	/* Check that we have the right heap page pinned, if present */	if (BufferIsValid(heapBuf) && BufferGetBlockNumber(heapBuf) != heapBlk)		elog(ERROR, "wrong heap buffer passed to visibilitymap_set");	/* Check that we have the right VM page pinned */	if (!BufferIsValid(vmBuf) || BufferGetBlockNumber(vmBuf) != mapBlock)		elog(ERROR, "wrong VM buffer passed to visibilitymap_set");	page = BufferGetPage(vmBuf);	map = PageGetContents(page);	LockBuffer(vmBuf, BUFFER_LOCK_EXCLUSIVE);	if (!(map[mapByte] & (1 << mapBit)))	{		START_CRIT_SECTION();		map[mapByte] |= (1 << mapBit);		MarkBufferDirty(vmBuf);		if (RelationNeedsWAL(rel))		{			if (XLogRecPtrIsInvalid(recptr))			{				Assert(!InRecovery);				recptr = log_heap_visible(rel->rd_node, heapBuf, vmBuf,										  cutoff_xid);				/*				 * If data checksums are enabled, we need to protect the heap				 * page from being torn.				 */				if (DataChecksumsEnabled())				{					Page heapPage = BufferGetPage(heapBuf);					/* caller is expected to set PD_ALL_VISIBLE first */					Assert(PageIsAllVisible(heapPage));					PageSetLSN(heapPage, recptr);				}			}			PageSetLSN(page, recptr);		}		END_CRIT_SECTION();	}	LockBuffer(vmBuf, BUFFER_LOCK_UNLOCK);}

static voidspgRedoAddNode(XLogRecPtr lsn, XLogRecord *record){	char	   *ptr = XLogRecGetData(record);	spgxlogAddNode *xldata = (spgxlogAddNode *) ptr;	SpGistInnerTuple innerTuple;	SpGistState state;	Buffer		buffer;	Page		page;	int			bbi;	/* we assume this is adequately aligned */	ptr += sizeof(spgxlogAddNode);	innerTuple = (SpGistInnerTuple) ptr;	fillFakeState(&state, xldata->stateSrc);	if (xldata->blknoNew == InvalidBlockNumber)	{		/* update in place */		Assert(xldata->blknoParent == InvalidBlockNumber);		if (!(record->xl_info & XLR_BKP_BLOCK_1))		{			buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);			if (BufferIsValid(buffer))			{				page = BufferGetPage(buffer);				if (!XLByteLE(lsn, PageGetLSN(page)))				{					PageIndexTupleDelete(page, xldata->offnum);					if (PageAddItem(page, (Item) innerTuple, innerTuple->size,									xldata->offnum,									false, false) != xldata->offnum)						elog(ERROR, "failed to add item of size %u to SPGiST index page",							 innerTuple->size);					PageSetLSN(page, lsn);					PageSetTLI(page, ThisTimeLineID);					MarkBufferDirty(buffer);				}				UnlockReleaseBuffer(buffer);			}		}	}	else	{		/* Install new tuple first so redirect is valid */		if (!(record->xl_info & XLR_BKP_BLOCK_2))		{			buffer = XLogReadBuffer(xldata->node, xldata->blknoNew,									xldata->newPage);			if (BufferIsValid(buffer))			{				page = BufferGetPage(buffer);				if (xldata->newPage)					SpGistInitBuffer(buffer, 0);				if (!XLByteLE(lsn, PageGetLSN(page)))				{					addOrReplaceTuple(page, (Item) innerTuple,									  innerTuple->size, xldata->offnumNew);					PageSetLSN(page, lsn);					PageSetTLI(page, ThisTimeLineID);					MarkBufferDirty(buffer);				}				UnlockReleaseBuffer(buffer);			}		}		/* Delete old tuple, replacing it with redirect or placeholder tuple */		if (!(record->xl_info & XLR_BKP_BLOCK_1))		{			buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);			if (BufferIsValid(buffer))			{				page = BufferGetPage(buffer);				if (!XLByteLE(lsn, PageGetLSN(page)))				{					SpGistDeadTuple dt;					if (state.isBuild)						dt = spgFormDeadTuple(&state, SPGIST_PLACEHOLDER,											  InvalidBlockNumber,											  InvalidOffsetNumber);					else						dt = spgFormDeadTuple(&state, SPGIST_REDIRECT,											  xldata->blknoNew,											  xldata->offnumNew);					PageIndexTupleDelete(page, xldata->offnum);					if (PageAddItem(page, (Item) dt, dt->size,									xldata->offnum,									false, false) != xldata->offnum)						elog(ERROR, "failed to add item of size %u to SPGiST index page",							 dt->size);					if (state.isBuild)						SpGistPageGetOpaque(page)->nPlaceholder++;//.........这里部分代码省略.........

static voidspgRedoPickSplit(XLogRecPtr lsn, XLogRecord *record){	char	   *ptr = XLogRecGetData(record);	spgxlogPickSplit *xldata = (spgxlogPickSplit *) ptr;	SpGistInnerTuple innerTuple;	SpGistState state;	OffsetNumber *toDelete;	OffsetNumber *toInsert;	uint8	   *leafPageSelect;	Buffer		srcBuffer;	Buffer		destBuffer;	Page		page;	int			bbi;	int			i;	fillFakeState(&state, xldata->stateSrc);	ptr += MAXALIGN(sizeof(spgxlogPickSplit));	innerTuple = (SpGistInnerTuple) ptr;	ptr += innerTuple->size;	toDelete = (OffsetNumber *) ptr;	ptr += MAXALIGN(sizeof(OffsetNumber) * xldata->nDelete);	toInsert = (OffsetNumber *) ptr;	ptr += MAXALIGN(sizeof(OffsetNumber) * xldata->nInsert);	leafPageSelect = (uint8 *) ptr;	ptr += MAXALIGN(sizeof(uint8) * xldata->nInsert);	/* now ptr points to the list of leaf tuples */	/*	 * It's a bit tricky to identify which pages have been handled as	 * full-page images, so we explicitly count each referenced buffer.	 */	bbi = 0;	if (xldata->blknoSrc == SPGIST_HEAD_BLKNO)	{		/* when splitting root, we touch it only in the guise of new inner */		srcBuffer = InvalidBuffer;	}	else if (xldata->initSrc)	{		/* just re-init the source page */		srcBuffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, true);		Assert(BufferIsValid(srcBuffer));		page = (Page) BufferGetPage(srcBuffer);		SpGistInitBuffer(srcBuffer, SPGIST_LEAF);		/* don't update LSN etc till we're done with it */	}	else	{		/* delete the specified tuples from source page */		if (!(record->xl_info & XLR_SET_BKP_BLOCK(bbi)))		{			srcBuffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, false);			if (BufferIsValid(srcBuffer))			{				page = BufferGetPage(srcBuffer);				if (!XLByteLE(lsn, PageGetLSN(page)))				{					/*					 * We have it a bit easier here than in doPickSplit(),					 * because we know the inner tuple's location already,					 * so we can inject the correct redirection tuple now.					 */					if (!state.isBuild)						spgPageIndexMultiDelete(&state, page,												toDelete, xldata->nDelete,												SPGIST_REDIRECT,												SPGIST_PLACEHOLDER,												xldata->blknoInner,												xldata->offnumInner);					else						spgPageIndexMultiDelete(&state, page,												toDelete, xldata->nDelete,												SPGIST_PLACEHOLDER,												SPGIST_PLACEHOLDER,												InvalidBlockNumber,												InvalidOffsetNumber);					/* don't update LSN etc till we're done with it */				}			}		}		else			srcBuffer = InvalidBuffer;		bbi++;	}	/* try to access dest page if any */	if (xldata->blknoDest == InvalidBlockNumber)	{		destBuffer = InvalidBuffer;	}	else if (xldata->initDest)	{		/* just re-init the dest page */		destBuffer = XLogReadBuffer(xldata->node, xldata->blknoDest, true);//.........这里部分代码省略.........

/* * redo any page update (except page split) */static voidgistRedoPageUpdateRecord(XLogRecPtr lsn, XLogRecord *record){	char	   *begin = XLogRecGetData(record);	gistxlogPageUpdate *xldata = (gistxlogPageUpdate *) begin;	Buffer		buffer;	Page		page;	char	   *data;	/*	 * We need to acquire and hold lock on target page while updating the left	 * child page.	If we have a full-page image of target page, getting the	 * lock is a side-effect of restoring that image.  Note that even if the	 * target page no longer exists, we'll still attempt to replay the change	 * on the child page.	 */	if (record->xl_info & XLR_BKP_BLOCK(0))		buffer = RestoreBackupBlock(lsn, record, 0, false, true);	else		buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);	/* Fix follow-right data on left child page */	if (BlockNumberIsValid(xldata->leftchild))		gistRedoClearFollowRight(lsn, record, 1,								 xldata->node, xldata->leftchild);	/* Done if target page no longer exists */	if (!BufferIsValid(buffer))		return;	/* nothing more to do if page was backed up (and no info to do it with) */	if (record->xl_info & XLR_BKP_BLOCK(0))	{		UnlockReleaseBuffer(buffer);		return;	}	page = (Page) BufferGetPage(buffer);	/* nothing more to do if change already applied */	if (lsn <= PageGetLSN(page))	{		UnlockReleaseBuffer(buffer);		return;	}	data = begin + sizeof(gistxlogPageUpdate);	/* Delete old tuples */	if (xldata->ntodelete > 0)	{		int			i;		OffsetNumber *todelete = (OffsetNumber *) data;		data += sizeof(OffsetNumber) * xldata->ntodelete;		for (i = 0; i < xldata->ntodelete; i++)			PageIndexTupleDelete(page, todelete[i]);		if (GistPageIsLeaf(page))			GistMarkTuplesDeleted(page);	}	/* add tuples */	if (data - begin < record->xl_len)	{		OffsetNumber off = (PageIsEmpty(page)) ? FirstOffsetNumber :		OffsetNumberNext(PageGetMaxOffsetNumber(page));		while (data - begin < record->xl_len)		{			IndexTuple	itup = (IndexTuple) data;			Size		sz = IndexTupleSize(itup);			OffsetNumber l;			data += sz;			l = PageAddItem(page, (Item) itup, sz, off, false, false);			if (l == InvalidOffsetNumber)				elog(ERROR, "failed to add item to GiST index page, size %d bytes",					 (int) sz);			off++;		}	}	else	{		/*		 * special case: leafpage, nothing to insert, nothing to delete, then		 * vacuum marks page		 */		if (GistPageIsLeaf(page) && xldata->ntodelete == 0)			GistClearTuplesDeleted(page);	}	if (!GistPageIsLeaf(page) &&		PageGetMaxOffsetNumber(page) == InvalidOffsetNumber &&		xldata->blkno == GIST_ROOT_BLKNO)	{//.........这里部分代码省略.........

//.........这里部分代码省略.........			oldcxt = MemoryContextSwitchTo(tupcxt);		}		dtup = brin_deform_tuple(bdesc, brtup);		/*		 * Compare the key values of the new tuple to the stored index values;		 * our deformed tuple will get updated if the new tuple doesn't fit		 * the original range (note this means we can't break out of the loop		 * early). Make a note of whether this happens, so that we know to		 * insert the modified tuple later.		 */		for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)		{			Datum		result;			BrinValues *bval;			FmgrInfo   *addValue;			bval = &dtup->bt_columns[keyno];			addValue = index_getprocinfo(idxRel, keyno + 1,										 BRIN_PROCNUM_ADDVALUE);			result = FunctionCall4Coll(addValue,									   idxRel->rd_indcollation[keyno],									   PointerGetDatum(bdesc),									   PointerGetDatum(bval),									   values[keyno],									   nulls[keyno]);			/* if that returned true, we need to insert the updated tuple */			need_insert |= DatumGetBool(result);		}		if (!need_insert)		{			/*			 * The tuple is consistent with the new values, so there's nothing			 * to do.			 */			LockBuffer(buf, BUFFER_LOCK_UNLOCK);		}		else		{			Page		page = BufferGetPage(buf);			ItemId		lp = PageGetItemId(page, off);			Size		origsz;			BrinTuple  *origtup;			Size		newsz;			BrinTuple  *newtup;			bool		samepage;			/*			 * Make a copy of the old tuple, so that we can compare it after			 * re-acquiring the lock.			 */			origsz = ItemIdGetLength(lp);			origtup = brin_copy_tuple(brtup, origsz);			/*			 * Before releasing the lock, check if we can attempt a same-page			 * update.  Another process could insert a tuple concurrently in			 * the same page though, so downstream we must be prepared to cope			 * if this turns out to not be possible after all.			 */			newtup = brin_form_tuple(bdesc, heapBlk, dtup, &newsz);			samepage = brin_can_do_samepage_update(buf, origsz, newsz);			LockBuffer(buf, BUFFER_LOCK_UNLOCK);			/*			 * Try to update the tuple.  If this doesn't work for whatever			 * reason, we need to restart from the top; the revmap might be			 * pointing at a different tuple for this block now, so we need to			 * recompute to ensure both our new heap tuple and the other			 * inserter's are covered by the combined tuple.  It might be that			 * we don't need to update at all.			 */			if (!brin_doupdate(idxRel, pagesPerRange, revmap, heapBlk,							   buf, off, origtup, origsz, newtup, newsz,							   samepage))			{				/* no luck; start over */				MemoryContextResetAndDeleteChildren(tupcxt);				continue;			}		}		/* success! */		break;	}	brinRevmapTerminate(revmap);	if (BufferIsValid(buf))		ReleaseBuffer(buf);	if (bdesc != NULL)	{		brin_free_desc(bdesc);		MemoryContextSwitchTo(oldcxt);		MemoryContextDelete(tupcxt);	}	return false;}

/* *	_bt_search() -- Search the tree for a particular scankey, *		or more precisely for the first leaf page it could be on. * * The passed scankey must be an insertion-type scankey (see nbtree/README), * but it can omit the rightmost column(s) of the index. * * When nextkey is false (the usual case), we are looking for the first * item >= scankey.  When nextkey is true, we are looking for the first * item strictly greater than scankey. * * Return value is a stack of parent-page pointers.  *bufP is set to the * address of the leaf-page buffer, which is read-locked and pinned. * No locks are held on the parent pages, however! * * NOTE that the returned buffer is read-locked regardless of the access * parameter.  However, access = BT_WRITE will allow an empty root page * to be created and returned.	When access = BT_READ, an empty index * will result in *bufP being set to InvalidBuffer. */BTStack_bt_search(Relation rel, int keysz, ScanKey scankey, bool nextkey,		   Buffer *bufP, int access){	BTStack		stack_in = NULL;	/* Get the root page to start with */	*bufP = _bt_getroot(rel, access);	/* If index is empty and access = BT_READ, no root page is created. */	if (!BufferIsValid(*bufP))		return (BTStack) NULL;	/* Loop iterates once per level descended in the tree */	for (;;)	{		Page		page;		BTPageOpaque opaque;		OffsetNumber offnum;		ItemId		itemid;		IndexTuple	itup;		BlockNumber blkno;		BlockNumber par_blkno;		BTStack		new_stack;		/*		 * Race -- the page we just grabbed may have split since we read its		 * pointer in the parent (or metapage).  If it has, we may need to		 * move right to its new sibling.  Do that.		 */		*bufP = _bt_moveright(rel, *bufP, keysz, scankey, nextkey, BT_READ);		/* if this is a leaf page, we're done */		page = BufferGetPage(*bufP);		opaque = (BTPageOpaque) PageGetSpecialPointer(page);		if (P_ISLEAF(opaque))			break;		/*		 * Find the appropriate item on the internal page, and get the child		 * page that it points to.		 */		offnum = _bt_binsrch(rel, *bufP, keysz, scankey, nextkey);		itemid = PageGetItemId(page, offnum);		itup = (IndexTuple) PageGetItem(page, itemid);		blkno = ItemPointerGetBlockNumber(&(itup->t_tid));		par_blkno = BufferGetBlockNumber(*bufP);		/*		 * We need to save the location of the index entry we chose in the		 * parent page on a stack. In case we split the tree, we'll use the		 * stack to work back up to the parent page.  We also save the actual		 * downlink (TID) to uniquely identify the index entry, in case it		 * moves right while we're working lower in the tree.  See the paper		 * by Lehman and Yao for how this is detected and handled. (We use the		 * child link to disambiguate duplicate keys in the index -- Lehman		 * and Yao disallow duplicate keys.)		 */		new_stack = (BTStack) palloc(sizeof(BTStackData));		new_stack->bts_blkno = par_blkno;		new_stack->bts_offset = offnum;		memcpy(&new_stack->bts_btentry, itup, sizeof(IndexTupleData));		new_stack->bts_parent = stack_in;		/* drop the read lock on the parent page, acquire one on the child */		*bufP = _bt_relandgetbuf(rel, *bufP, blkno, BT_READ);		/* okay, all set to move down a level */		stack_in = new_stack;	}	return stack_in;}

/* * Write WAL record of a page split. */XLogRecPtrgistXLogSplit(RelFileNode node, BlockNumber blkno, bool page_is_leaf,			  SplitedPageLayout *dist,			  BlockNumber origrlink, GistNSN orignsn,			  Buffer leftchildbuf, bool markfollowright){	XLogRecData *rdata;	gistxlogPageSplit xlrec;	SplitedPageLayout *ptr;	int			npage = 0,				cur;	XLogRecPtr	recptr;	for (ptr = dist; ptr; ptr = ptr->next)		npage++;	rdata = (XLogRecData *) palloc(sizeof(XLogRecData) * (npage * 2 + 2));	xlrec.node = node;	xlrec.origblkno = blkno;	xlrec.origrlink = origrlink;	xlrec.orignsn = orignsn;	xlrec.origleaf = page_is_leaf;	xlrec.npage = (uint16) npage;	xlrec.leftchild =		BufferIsValid(leftchildbuf) ? BufferGetBlockNumber(leftchildbuf) : InvalidBlockNumber;	xlrec.markfollowright = markfollowright;	rdata[0].data = (char *) &xlrec;	rdata[0].len = sizeof(gistxlogPageSplit);	rdata[0].buffer = InvalidBuffer;	cur = 1;	/*	 * Include a full page image of the child buf. (only necessary if a	 * checkpoint happened since the child page was split)	 */	if (BufferIsValid(leftchildbuf))	{		rdata[cur - 1].next = &(rdata[cur]);		rdata[cur].data = NULL;		rdata[cur].len = 0;		rdata[cur].buffer = leftchildbuf;		rdata[cur].buffer_std = true;		cur++;	}	for (ptr = dist; ptr; ptr = ptr->next)	{		rdata[cur - 1].next = &(rdata[cur]);		rdata[cur].buffer = InvalidBuffer;		rdata[cur].data = (char *) &(ptr->block);		rdata[cur].len = sizeof(gistxlogPage);		cur++;		rdata[cur - 1].next = &(rdata[cur]);		rdata[cur].buffer = InvalidBuffer;		rdata[cur].data = (char *) (ptr->list);		rdata[cur].len = ptr->lenlist;		cur++;	}	rdata[cur - 1].next = NULL;	recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_SPLIT, rdata);	pfree(rdata);	return recptr;}

/* * Scan a complete BRIN index, and summarize each page range that's not already * summarized.  The index and heap must have been locked by caller in at * least ShareUpdateExclusiveLock mode. * * For each new index tuple inserted, *numSummarized (if not NULL) is * incremented; for each existing tuple, numExisting (if not NULL) is * incremented. */static voidbrinsummarize(Relation index, Relation heapRel, double *numSummarized,			  double *numExisting){	BrinRevmap *revmap;	BrinBuildState *state = NULL;	IndexInfo  *indexInfo = NULL;	BlockNumber heapNumBlocks;	BlockNumber heapBlk;	BlockNumber pagesPerRange;	Buffer		buf;	revmap = brinRevmapInitialize(index, &pagesPerRange);	/*	 * Scan the revmap to find unsummarized items.	 */	buf = InvalidBuffer;	heapNumBlocks = RelationGetNumberOfBlocks(heapRel);	for (heapBlk = 0; heapBlk < heapNumBlocks; heapBlk += pagesPerRange)	{		BrinTuple  *tup;		OffsetNumber off;		CHECK_FOR_INTERRUPTS();		tup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off, NULL,									   BUFFER_LOCK_SHARE);		if (tup == NULL)		{			/* no revmap entry for this heap range. Summarize it. */			if (state == NULL)			{				/* first time through */				Assert(!indexInfo);				state = initialize_brin_buildstate(index, revmap,												   pagesPerRange);				indexInfo = BuildIndexInfo(index);				/*				 * We only have ShareUpdateExclusiveLock on the table, and				 * therefore other sessions may insert tuples into the range				 * we're going to scan.  This is okay, because we take				 * additional precautions to avoid losing the additional				 * tuples; see comments in summarize_range.  Set the				 * concurrent flag, which causes IndexBuildHeapRangeScan to				 * use a snapshot other than SnapshotAny, and silences				 * warnings emitted there.				 */				indexInfo->ii_Concurrent = true;				/*				 * If using transaction-snapshot mode, it would be possible				 * for another transaction to insert a tuple that's not				 * visible to our snapshot if we have already acquired one,				 * when in snapshot-isolation mode; therefore, disallow this				 * from running in such a transaction unless a snapshot hasn't				 * been acquired yet.				 *				 * This code is called by VACUUM and				 * brin_summarize_new_values. Have the error message mention				 * the latter because VACUUM cannot run in a transaction and				 * thus cannot cause this issue.				 */				if (IsolationUsesXactSnapshot() && FirstSnapshotSet)					ereport(ERROR,							(errcode(ERRCODE_INVALID_TRANSACTION_STATE),							 errmsg("brin_summarize_new_values() cannot run in a transaction that has already obtained a snapshot")));			}			summarize_range(indexInfo, state, heapRel, heapBlk);			/* and re-initialize state for the next range */			brin_memtuple_initialize(state->bs_dtuple, state->bs_bdesc);			if (numSummarized)				*numSummarized += 1.0;		}		else		{			if (numExisting)				*numExisting += 1.0;			LockBuffer(buf, BUFFER_LOCK_UNLOCK);		}	}	if (BufferIsValid(buf))		ReleaseBuffer(buf);	/* free resources */	brinRevmapTerminate(revmap);	if (state)//.........这里部分代码省略.........

//.........这里部分代码省略.........			Datum		result;			BrinValues *bval;			FmgrInfo   *addValue;			bval = &dtup->bt_columns[keyno];			addValue = index_getprocinfo(idxRel, keyno + 1,										 BRIN_PROCNUM_ADDVALUE);			result = FunctionCall4Coll(addValue,									   idxRel->rd_indcollation[keyno],									   PointerGetDatum(bdesc),									   PointerGetDatum(bval),									   values[keyno],									   nulls[keyno]);			/* if that returned true, we need to insert the updated tuple */			need_insert |= DatumGetBool(result);		}#ifdef USE_ASSERT_CHECKING		{			/*			 * Now we can compare the tuple produced by the union function			 * with the one from plain addValue.			 */			BrinTuple  *cmptup;			Size		cmpsz;			cmptup = brin_form_tuple(bdesc, heapBlk, dtup, &cmpsz);			Assert(brin_tuples_equal(tmptup, tmpsiz, cmptup, cmpsz));		}#endif		if (!need_insert)		{			/*			 * The tuple is consistent with the new values, so there's nothing			 * to do.			 */			LockBuffer(buf, BUFFER_LOCK_UNLOCK);		}		else		{			Page		page = BufferGetPage(buf);			ItemId		lp = PageGetItemId(page, off);			Size		origsz;			BrinTuple  *origtup;			Size		newsz;			BrinTuple  *newtup;			bool		samepage;			/*			 * Make a copy of the old tuple, so that we can compare it after			 * re-acquiring the lock.			 */			origsz = ItemIdGetLength(lp);			origtup = brin_copy_tuple(brtup, origsz);			/*			 * Before releasing the lock, check if we can attempt a same-page			 * update.  Another process could insert a tuple concurrently in			 * the same page though, so downstream we must be prepared to cope			 * if this turns out to not be possible after all.			 */			newtup = brin_form_tuple(bdesc, heapBlk, dtup, &newsz);			samepage = brin_can_do_samepage_update(buf, origsz, newsz);			LockBuffer(buf, BUFFER_LOCK_UNLOCK);			/*			 * Try to update the tuple.  If this doesn't work for whatever			 * reason, we need to restart from the top; the revmap might be			 * pointing at a different tuple for this block now, so we need to			 * recompute to ensure both our new heap tuple and the other			 * inserter's are covered by the combined tuple.  It might be that			 * we don't need to update at all.			 */			if (!brin_doupdate(idxRel, pagesPerRange, revmap, heapBlk,							   buf, off, origtup, origsz, newtup, newsz,							   samepage))			{				/* no luck; start over */				MemoryContextResetAndDeleteChildren(tupcxt);				continue;			}		}		/* success! */		break;	}	brinRevmapTerminate(revmap);	if (BufferIsValid(buf))		ReleaseBuffer(buf);	if (bdesc != NULL)	{		brin_free_desc(bdesc);		MemoryContextSwitchTo(oldcxt);		MemoryContextDelete(tupcxt);	}	return BoolGetDatum(false);}

/* *	_hash_step() -- step to the next valid item in a scan in the bucket. * *		If no valid record exists in the requested direction, return *		false.  Else, return true and set the hashso_curpos for the *		scan to the right thing. * *		'bufP' points to the current buffer, which is pinned and read-locked. *		On success exit, we have pin and read-lock on whichever page *		contains the right item; on failure, we have released all buffers. */bool_hash_step(IndexScanDesc scan, Buffer *bufP, ScanDirection dir){	Relation	rel = scan->indexRelation;	HashScanOpaque so = (HashScanOpaque) scan->opaque;	ItemPointer current;	Buffer		buf;	Page		page;	HashPageOpaque opaque;	OffsetNumber maxoff;	OffsetNumber offnum;	BlockNumber blkno;	IndexTuple	itup;	current = &(so->hashso_curpos);	buf = *bufP;	_hash_checkpage(rel, buf, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);	page = BufferGetPage(buf);	opaque = (HashPageOpaque) PageGetSpecialPointer(page);	/*	 * If _hash_step is called from _hash_first, current will not be valid, so	 * we can't dereference it.  However, in that case, we presumably want to	 * start at the beginning/end of the page...	 */	maxoff = PageGetMaxOffsetNumber(page);	if (ItemPointerIsValid(current))		offnum = ItemPointerGetOffsetNumber(current);	else		offnum = InvalidOffsetNumber;	/*	 * 'offnum' now points to the last tuple we examined (if any).	 *	 * continue to step through tuples until: 1) we get to the end of the	 * bucket chain or 2) we find a valid tuple.	 */	do	{		switch (dir)		{			case ForwardScanDirection:				if (offnum != InvalidOffsetNumber)					offnum = OffsetNumberNext(offnum);	/* move forward */				else				{					/* new page, locate starting position by binary search */					offnum = _hash_binsearch(page, so->hashso_sk_hash);				}				for (;;)				{					/*					 * check if we're still in the range of items with the					 * target hash key					 */					if (offnum <= maxoff)					{						Assert(offnum >= FirstOffsetNumber);						itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, offnum));						if (so->hashso_sk_hash == _hash_get_indextuple_hashkey(itup))							break;		/* yes, so exit for-loop */					}					/*					 * ran off the end of this page, try the next					 */					_hash_readnext(rel, &buf, &page, &opaque);					if (BufferIsValid(buf))					{						maxoff = PageGetMaxOffsetNumber(page);						offnum = _hash_binsearch(page, so->hashso_sk_hash);					}					else					{						/* end of bucket */						itup = NULL;						break;	/* exit for-loop */					}				}				break;			case BackwardScanDirection:				if (offnum != InvalidOffsetNumber)					offnum = OffsetNumberPrev(offnum);	/* move back */				else				{					/* new page, locate starting position by binary search *///.........这里部分代码省略.........

//.........这里部分代码省略.........                  #if REDCONF_READ_ONLY == 1                    CRITICAL_ERROR();                    ret = -RED_EFUBAR;                  #else                    ret = BufferWrite(bIdx);                  #endif                }            }            else            {                /*  All the buffers are used, which should have been caught by                    checking gBufCtx.uNumUsed.                */                CRITICAL_ERROR();                ret = -RED_EBUSY;            }            if(ret == 0)            {                if((uFlags & BFLAG_NEW) == 0U)                {                    /*  Invalidate the LRU buffer.  If the read fails, we do not                        want the buffer head to continue to refer to the old                        block number, since the read, even if it fails, may have                        partially overwritten the buffer data (consider the case                        where block size exceeds sector size, and some but not                        all of the sectors are read successfully), and if the                        buffer were to be used subsequently with its partially                        erroneous contents, bad things could happen.                    */                    pHead->ulBlock = BBLK_INVALID;                    ret = RedIoRead(gbRedVolNum, ulBlock, 1U, gBufCtx.b.aabBuffer[bIdx]);                    if((ret == 0) && ((uFlags & BFLAG_META) != 0U))                    {                        if(!BufferIsValid(gBufCtx.b.aabBuffer[bIdx], uFlags))                        {                            /*  A corrupt metadata node is usually a critical                                error.  The master block is an exception since                                it might be invalid because the volume is not                                mounted; that condition is expected and should                                not result in an assertion.                            */                            CRITICAL_ASSERT((uFlags & BFLAG_META_MASTER) == BFLAG_META_MASTER);                            ret = -RED_EIO;                        }                    }                  #ifdef REDCONF_ENDIAN_SWAP                    if(ret == 0)                    {                        BufferEndianSwap(gBufCtx.b.aabBuffer[bIdx], uFlags);                    }                  #endif                }                else                {                    RedMemSet(gBufCtx.b.aabBuffer[bIdx], 0U, REDCONF_BLOCK_SIZE);                }            }            if(ret == 0)            {                pHead->bVolNum = gbRedVolNum;                pHead->ulBlock = ulBlock;                pHead->uFlags = 0U;            }        }        /*  Reference the buffer, update its flags, and promote it to MRU.  This            happens both when BufferFind() found an existing buffer for the            block and when the LRU buffer was repurposed to create a buffer for            the block.        */        if(ret == 0)        {            BUFFERHEAD *pHead = &gBufCtx.aHead[bIdx];            pHead->bRefCount++;            if(pHead->bRefCount == 1U)            {                gBufCtx.uNumUsed++;            }            /*  BFLAG_NEW tells this function to zero the buffer instead of                reading it from disk; it has no meaning later on, and thus is                not saved.            */            pHead->uFlags |= (uFlags & (~BFLAG_NEW));            BufferMakeMRU(bIdx);            *ppBuffer = gBufCtx.b.aabBuffer[bIdx];        }    }    return ret;}

//.........这里部分代码省略.........					ptr->next->block.blkno;			else				GistPageGetOpaque(ptr->page)->rightlink = oldrlink;			/*			 * Mark the all but the right-most page with the follow-right			 * flag. It will be cleared as soon as the downlink is inserted			 * into the parent, but this ensures that if we error out before			 * that, the index is still consistent. (in buffering build mode,			 * any error will abort the index build anyway, so this is not			 * needed.)			 */			if (ptr->next && !is_rootsplit && markfollowright)				GistMarkFollowRight(ptr->page);			else				GistClearFollowRight(ptr->page);			/*			 * Copy the NSN of the original page to all pages. The			 * F_FOLLOW_RIGHT flags ensure that scans will follow the			 * rightlinks until the downlinks are inserted.			 */			GistPageSetNSN(ptr->page, oldnsn);		}		START_CRIT_SECTION();		/*		 * Must mark buffers dirty before XLogInsert, even though we'll still		 * be changing their opaque fields below.		 */		for (ptr = dist; ptr; ptr = ptr->next)			MarkBufferDirty(ptr->buffer);		if (BufferIsValid(leftchildbuf))			MarkBufferDirty(leftchildbuf);		/*		 * The first page in the chain was a temporary working copy meant to		 * replace the old page. Copy it over the old page.		 */		PageRestoreTempPage(dist->page, BufferGetPage(dist->buffer));		dist->page = BufferGetPage(dist->buffer);		/* Write the WAL record */		if (RelationNeedsWAL(rel))			recptr = gistXLogSplit(rel->rd_node, blkno, is_leaf,								   dist, oldrlink, oldnsn, leftchildbuf,								   markfollowright);		else			recptr = gistGetFakeLSN(rel);		for (ptr = dist; ptr; ptr = ptr->next)		{			PageSetLSN(ptr->page, recptr);		}		/*		 * Return the new child buffers to the caller.		 *		 * If this was a root split, we've already inserted the downlink		 * pointers, in the form of a new root page. Therefore we can release		 * all the new buffers, and keep just the root page locked.		 */		if (is_rootsplit)		{			for (ptr = dist->next; ptr; ptr = ptr->next)

static boolrtnext(IndexScanDesc s, ScanDirection dir){	Page		p;	OffsetNumber n;	RTreePageOpaque po;	RTreeScanOpaque so;	so = (RTreeScanOpaque) s->opaque;	if (!ItemPointerIsValid(&(s->currentItemData)))	{		/* first call: start at the root */		Assert(BufferIsValid(so->curbuf) == false);		so->curbuf = ReadBuffer(s->indexRelation, P_ROOT);		pgstat_count_index_scan(&s->xs_pgstat_info);	}	p = BufferGetPage(so->curbuf);	po = (RTreePageOpaque) PageGetSpecialPointer(p);	if (!ItemPointerIsValid(&(s->currentItemData)))	{		/* first call: start at first/last offset */		if (ScanDirectionIsForward(dir))			n = FirstOffsetNumber;		else			n = PageGetMaxOffsetNumber(p);	}	else	{		/* go on to the next offset */		n = ItemPointerGetOffsetNumber(&(s->currentItemData));		if (ScanDirectionIsForward(dir))			n = OffsetNumberNext(n);		else			n = OffsetNumberPrev(n);	}	for (;;)	{		IndexTuple	it;		RTSTACK    *stk;		n = findnext(s, n, dir);		/* no match on this page, so read in the next stack entry */		if (n == InvalidOffsetNumber)		{			/* if out of stack entries, we're done */			if (so->s_stack == NULL)			{				ReleaseBuffer(so->curbuf);				so->curbuf = InvalidBuffer;				return false;			}			stk = so->s_stack;			so->curbuf = ReleaseAndReadBuffer(so->curbuf, s->indexRelation,											  stk->rts_blk);			p = BufferGetPage(so->curbuf);			po = (RTreePageOpaque) PageGetSpecialPointer(p);			if (ScanDirectionIsBackward(dir))				n = OffsetNumberPrev(stk->rts_child);			else				n = OffsetNumberNext(stk->rts_child);			so->s_stack = stk->rts_parent;			pfree(stk);			continue;		}		if (po->flags & F_LEAF)		{			ItemPointerSet(&(s->currentItemData),						   BufferGetBlockNumber(so->curbuf),						   n);			it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));			s->xs_ctup.t_self = it->t_tid;			return true;		}		else		{			BlockNumber blk;			stk = (RTSTACK *) palloc(sizeof(RTSTACK));			stk->rts_child = n;			stk->rts_blk = BufferGetBlockNumber(so->curbuf);			stk->rts_parent = so->s_stack;			so->s_stack = stk;			it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));			blk = ItemPointerGetBlockNumber(&(it->t_tid));			/*			 * Note that we release the pin on the page as we descend down the			 * tree, even though there's a good chance we'll eventually need			 * to re-read the buffer later in this scan. This may or may not			 * be optimal, but it doesn't seem likely to make a huge//.........这里部分代码省略.........

/* *	visibilitymap_truncate - truncate the visibility map * * The caller must hold AccessExclusiveLock on the relation, to ensure that * other backends receive the smgr invalidation event that this function sends * before they access the VM again. * * nheapblocks is the new size of the heap. */voidvisibilitymap_truncate(Relation rel, BlockNumber nheapblocks){	BlockNumber newnblocks;	/* last remaining block, byte, and bit */	BlockNumber truncBlock = HEAPBLK_TO_MAPBLOCK(nheapblocks);	uint32		truncByte = HEAPBLK_TO_MAPBYTE(nheapblocks);	uint8		truncBit = HEAPBLK_TO_MAPBIT(nheapblocks);#ifdef TRACE_VISIBILITYMAP	elog(DEBUG1, "vm_truncate %s %d", RelationGetRelationName(rel), nheapblocks);#endif	RelationOpenSmgr(rel);	/*	 * If no visibility map has been created yet for this relation, there's	 * nothing to truncate.	 */	if (!smgrexists(rel->rd_smgr, VISIBILITYMAP_FORKNUM))		return;	/*	 * Unless the new size is exactly at a visibility map page boundary, the	 * tail bits in the last remaining map page, representing truncated heap	 * blocks, need to be cleared. This is not only tidy, but also necessary	 * because we don't get a chance to clear the bits if the heap is extended	 * again.	 */	if (truncByte != 0 || truncBit != 0)	{		Buffer		mapBuffer;		Page		page;		char	   *map;		newnblocks = truncBlock + 1;		mapBuffer = vm_readbuf(rel, truncBlock, false);		if (!BufferIsValid(mapBuffer))		{			/* nothing to do, the file was already smaller */			return;		}		page = BufferGetPage(mapBuffer);		map = PageGetContents(page);		LockBuffer(mapBuffer, BUFFER_LOCK_EXCLUSIVE);		/* Clear out the unwanted bytes. */		MemSet(&map[truncByte + 1], 0, MAPSIZE - (truncByte + 1));		/*		 * Mask out the unwanted bits of the last remaining byte.		 *		 * ((1 << 0) - 1) = 00000000 ((1 << 1) - 1) = 00000001 ... ((1 << 6) -		 * 1) = 00111111 ((1 << 7) - 1) = 01111111		 */		map[truncByte] &= (1 << truncBit) - 1;		MarkBufferDirty(mapBuffer);		UnlockReleaseBuffer(mapBuffer);	}	else		newnblocks = truncBlock;	if (smgrnblocks(rel->rd_smgr, VISIBILITYMAP_FORKNUM) <= newnblocks)	{		/* nothing to do, the file was already smaller than requested size */		return;	}	/* Truncate the unused VM pages, and send smgr inval message */	smgrtruncate(rel->rd_smgr, VISIBILITYMAP_FORKNUM, newnblocks);	/*	 * We might as well update the local smgr_vm_nblocks setting. smgrtruncate	 * sent an smgr cache inval message, which will cause other backends to	 * invalidate their copy of smgr_vm_nblocks, and this one too at the next	 * command boundary.  But this ensures it isn't outright wrong until then.	 */	if (rel->rd_smgr)		rel->rd_smgr->smgr_vm_nblocks = newnblocks;}

/* *	_bt_endpoint() -- Find the first or last page in the index, and scan * from there to the first key satisfying all the quals. * * This is used by _bt_first() to set up a scan when we've determined * that the scan must start at the beginning or end of the index (for * a forward or backward scan respectively).  Exit conditions are the * same as for _bt_first(). */static bool_bt_endpoint(IndexScanDesc scan, ScanDirection dir){	Relation	rel = scan->indexRelation;	BTScanOpaque so = (BTScanOpaque) scan->opaque;	Buffer		buf;	Page		page;	BTPageOpaque opaque;	OffsetNumber start;	BTScanPosItem *currItem;	/*	 * Scan down to the leftmost or rightmost leaf page.  This is a simplified	 * version of _bt_search().  We don't maintain a stack since we know we	 * won't need it.	 */	buf = _bt_get_endpoint(rel, 0, ScanDirectionIsBackward(dir));	if (!BufferIsValid(buf))	{		/*		 * Empty index. Lock the whole relation, as nothing finer to lock		 * exists.		 */		PredicateLockRelation(rel, scan->xs_snapshot);		so->currPos.buf = InvalidBuffer;		return false;	}	PredicateLockPage(rel, BufferGetBlockNumber(buf), scan->xs_snapshot);	page = BufferGetPage(buf);	opaque = (BTPageOpaque) PageGetSpecialPointer(page);	Assert(P_ISLEAF(opaque));	if (ScanDirectionIsForward(dir))	{		/* There could be dead pages to the left, so not this: */		/* Assert(P_LEFTMOST(opaque)); */		start = P_FIRSTDATAKEY(opaque);	}	else if (ScanDirectionIsBackward(dir))	{		Assert(P_RIGHTMOST(opaque));		start = PageGetMaxOffsetNumber(page);	}	else	{		elog(ERROR, "invalid scan direction: %d", (int) dir);		start = 0;				/* keep compiler quiet */	}	/* remember which buffer we have pinned */	so->currPos.buf = buf;	/* initialize moreLeft/moreRight appropriately for scan direction */	if (ScanDirectionIsForward(dir))	{		so->currPos.moreLeft = false;		so->currPos.moreRight = true;	}	else	{		so->currPos.moreLeft = true;		so->currPos.moreRight = false;	}	so->numKilled = 0;			/* just paranoia */	so->markItemIndex = -1;		/* ditto */	/*	 * Now load data from the first page of the scan.	 */	if (!_bt_readpage(scan, dir, start))	{		/*		 * There's no actually-matching data on this page.  Try to advance to		 * the next page.  Return false if there's no matching data at all.		 */		if (!_bt_steppage(scan, dir))			return false;	}	/* Drop the lock, but not pin, on the current page */	LockBuffer(so->currPos.buf, BUFFER_LOCK_UNLOCK);	/* OK, itemIndex says what to return */	currItem = &so->currPos.items[so->currPos.itemIndex];	scan->xs_ctup.t_self = currItem->heapTid;	if (scan->xs_want_itup)		scan->xs_itup = (IndexTuple) (so->currTuples + currItem->tupleOffset);//.........这里部分代码省略.........

/* *	hashgettuple() -- Get the next tuple in the scan. */Datumhashgettuple(PG_FUNCTION_ARGS){	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);	ScanDirection dir = (ScanDirection) PG_GETARG_INT32(1);	HashScanOpaque so = (HashScanOpaque) scan->opaque;	Relation	rel = scan->indexRelation;	Page		page;	OffsetNumber offnum;	bool		res;	/*	 * We hold pin but not lock on current buffer while outside the hash AM.	 * Reacquire the read lock here.	 */	if (BufferIsValid(so->hashso_curbuf))		_hash_chgbufaccess(rel, so->hashso_curbuf, HASH_NOLOCK, HASH_READ);	/*	 * If we've already initialized this scan, we can just advance it in the	 * appropriate direction.  If we haven't done so yet, we call a routine to	 * get the first item in the scan.	 */	if (ItemPointerIsValid(&(scan->currentItemData)))	{		/*		 * Check to see if we should kill the previously-fetched tuple.		 */		if (scan->kill_prior_tuple)		{			/*			 * Yes, so mark it by setting the LP_DELETE bit in the item flags.			 */			offnum = ItemPointerGetOffsetNumber(&(scan->currentItemData));			page = BufferGetPage(so->hashso_curbuf);			PageGetItemId(page, offnum)->lp_flags |= LP_DELETE;			/*			 * Since this can be redone later if needed, it's treated the same			 * as a commit-hint-bit status update for heap tuples: we mark the			 * buffer dirty but don't make a WAL log entry.			 */			SetBufferCommitInfoNeedsSave(so->hashso_curbuf);		}		/*		 * Now continue the scan.		 */		res = _hash_next(scan, dir);	}	else		res = _hash_first(scan, dir);	/*	 * Skip killed tuples if asked to.	 */	if (scan->ignore_killed_tuples)	{		while (res)		{			offnum = ItemPointerGetOffsetNumber(&(scan->currentItemData));			page = BufferGetPage(so->hashso_curbuf);			if (!ItemIdDeleted(PageGetItemId(page, offnum)))				break;			res = _hash_next(scan, dir);		}	}	/* Release read lock on current buffer, but keep it pinned */	if (BufferIsValid(so->hashso_curbuf))		_hash_chgbufaccess(rel, so->hashso_curbuf, HASH_READ, HASH_NOLOCK);	PG_RETURN_BOOL(res);}

//.........这里部分代码省略.........			 */			nextkey = false;			goback = false;			break;		case BTGreaterStrategyNumber:			/*			 * Find first item > scankey.  (This is only used for forward			 * scans.)			 */			nextkey = true;			goback = false;			break;		default:			/* can't get here, but keep compiler quiet */			elog(ERROR, "unrecognized strat_total: %d", (int) strat_total);			return false;	}	/*	 * Use the manufactured insertion scan key to descend the tree and	 * position ourselves on the target leaf page.	 */	stack = _bt_search(rel, keysCount, scankeys, nextkey, &buf, BT_READ);	/* don't need to keep the stack around... */	_bt_freestack(stack);	/* remember which buffer we have pinned, if any */	so->currPos.buf = buf;	if (!BufferIsValid(buf))	{		/*		 * We only get here if the index is completely empty. Lock relation		 * because nothing finer to lock exists.		 */		PredicateLockRelation(rel, scan->xs_snapshot);		return false;	}	else		PredicateLockPage(rel, BufferGetBlockNumber(buf),						  scan->xs_snapshot);	/* initialize moreLeft/moreRight appropriately for scan direction */	if (ScanDirectionIsForward(dir))	{		so->currPos.moreLeft = false;		so->currPos.moreRight = true;	}	else	{		so->currPos.moreLeft = true;		so->currPos.moreRight = false;	}	so->numKilled = 0;			/* just paranoia */	so->markItemIndex = -1;		/* ditto */	/* position to the precise item on the page */	offnum = _bt_binsrch(rel, buf, keysCount, scankeys, nextkey);	/*	 * If nextkey = false, we are positioned at the first item >= scan key, or	 * possibly at the end of a page on which all the existing items are less

/* *	hashgetmulti() -- get multiple tuples at once * * This is a somewhat generic implementation: it avoids lock reacquisition * overhead, but there's no smarts about picking especially good stopping * points such as index page boundaries. */Datumhashgetmulti(PG_FUNCTION_ARGS){	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);	ItemPointer tids = (ItemPointer) PG_GETARG_POINTER(1);	int32		max_tids = PG_GETARG_INT32(2);	int32	   *returned_tids = (int32 *) PG_GETARG_POINTER(3);	HashScanOpaque so = (HashScanOpaque) scan->opaque;	Relation	rel = scan->indexRelation;	bool		res = true;	int32		ntids = 0;	/*	 * We hold pin but not lock on current buffer while outside the hash AM.	 * Reacquire the read lock here.	 */	if (BufferIsValid(so->hashso_curbuf))		_hash_chgbufaccess(rel, so->hashso_curbuf, HASH_NOLOCK, HASH_READ);	while (ntids < max_tids)	{		/*		 * Start scan, or advance to next tuple.		 */		if (ItemPointerIsValid(&(scan->currentItemData)))			res = _hash_next(scan, ForwardScanDirection);		else			res = _hash_first(scan, ForwardScanDirection);		/*		 * Skip killed tuples if asked to.		 */		if (scan->ignore_killed_tuples)		{			while (res)			{				Page		page;				OffsetNumber offnum;				offnum = ItemPointerGetOffsetNumber(&(scan->currentItemData));				page = BufferGetPage(so->hashso_curbuf);				if (!ItemIdDeleted(PageGetItemId(page, offnum)))					break;				res = _hash_next(scan, ForwardScanDirection);			}		}		if (!res)			break;		/* Save tuple ID, and continue scanning */		tids[ntids] = scan->xs_ctup.t_self;		ntids++;	}	/* Release read lock on current buffer, but keep it pinned */	if (BufferIsValid(so->hashso_curbuf))		_hash_chgbufaccess(rel, so->hashso_curbuf, HASH_READ, HASH_NOLOCK);	*returned_tids = ntids;	PG_RETURN_BOOL(res);}

static voidgistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record){	gistxlogPageSplit *xldata = (gistxlogPageSplit *) XLogRecGetData(record);	PageSplitRecord xlrec;	Buffer		firstbuffer = InvalidBuffer;	Buffer		buffer;	Page		page;	int			i;	bool		isrootsplit = false;	decodePageSplitRecord(&xlrec, record);	/*	 * We must hold lock on the first-listed page throughout the action,	 * including while updating the left child page (if any).  We can unlock	 * remaining pages in the list as soon as they've been written, because	 * there is no path for concurrent queries to reach those pages without	 * first visiting the first-listed page.	 */	/* loop around all pages */	for (i = 0; i < xlrec.data->npage; i++)	{		NewPage    *newpage = xlrec.page + i;		int			flags;		if (newpage->header->blkno == GIST_ROOT_BLKNO)		{			Assert(i == 0);			isrootsplit = true;		}		buffer = XLogReadBuffer(xlrec.data->node, newpage->header->blkno, true);		Assert(BufferIsValid(buffer));		page = (Page) BufferGetPage(buffer);		/* ok, clear buffer */		if (xlrec.data->origleaf && newpage->header->blkno != GIST_ROOT_BLKNO)			flags = F_LEAF;		else			flags = 0;		GISTInitBuffer(buffer, flags);		/* and fill it */		gistfillbuffer(page, newpage->itup, newpage->header->num, FirstOffsetNumber);		if (newpage->header->blkno == GIST_ROOT_BLKNO)		{			GistPageGetOpaque(page)->rightlink = InvalidBlockNumber;			GistPageSetNSN(page, xldata->orignsn);			GistClearFollowRight(page);		}		else		{			if (i < xlrec.data->npage - 1)				GistPageGetOpaque(page)->rightlink = xlrec.page[i + 1].header->blkno;			else				GistPageGetOpaque(page)->rightlink = xldata->origrlink;			GistPageSetNSN(page, xldata->orignsn);			if (i < xlrec.data->npage - 1 && !isrootsplit &&				xldata->markfollowright)				GistMarkFollowRight(page);			else				GistClearFollowRight(page);		}		PageSetLSN(page, lsn);		MarkBufferDirty(buffer);		if (i == 0)			firstbuffer = buffer;		else			UnlockReleaseBuffer(buffer);	}	/* Fix follow-right data on left child page, if any */	if (BlockNumberIsValid(xldata->leftchild))		gistRedoClearFollowRight(lsn, record, 0,								 xldata->node, xldata->leftchild);	/* Finally, release lock on the first page */	UnlockReleaseBuffer(firstbuffer);}

//.........这里部分代码省略.........	 * No free pages --- have to extend the relation to add an overflow page.	 * First, check to see if we have to add a new bitmap page too.	 */	if (last_bit == (uint32) (BMPGSZ_BIT(metap) - 1))	{		/*		 * We create the new bitmap page with all pages marked "in use".		 * Actually two pages in the new bitmap's range will exist		 * immediately: the bitmap page itself, and the following page which		 * is the one we return to the caller.  Both of these are correctly		 * marked "in use".  Subsequent pages do not exist yet, but it is		 * convenient to pre-mark them as "in use" too.		 */		bit = metap->hashm_spares[splitnum];		/* metapage already has a write lock */		if (metap->hashm_nmaps >= HASH_MAX_BITMAPS)			ereport(ERROR,					(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),					 errmsg("out of overflow pages in hash index /"%s/"",							RelationGetRelationName(rel))));		newmapbuf = _hash_getnewbuf(rel, bitno_to_blkno(metap, bit), MAIN_FORKNUM);	}	else	{		/*		 * Nothing to do here; since the page will be past the last used page,		 * we know its bitmap bit was preinitialized to "in use".		 */	}	/* Calculate address of the new overflow page */	bit = BufferIsValid(newmapbuf) ?		metap->hashm_spares[splitnum] + 1 : metap->hashm_spares[splitnum];	blkno = bitno_to_blkno(metap, bit);	/*	 * Fetch the page with _hash_getnewbuf to ensure smgr's idea of the	 * relation length stays in sync with ours.  XXX It's annoying to do this	 * with metapage write lock held; would be better to use a lock that	 * doesn't block incoming searches.	 *	 * It is okay to hold two buffer locks here (one on tail page of bucket	 * and other on new overflow page) since there cannot be anyone else	 * contending for access to ovflbuf.	 */	ovflbuf = _hash_getnewbuf(rel, blkno, MAIN_FORKNUM);found:	/*	 * Do the update.  No ereport(ERROR) until changes are logged. We want to	 * log the changes for bitmap page and overflow page together to avoid	 * loss of pages in case the new page is added.	 */	START_CRIT_SECTION();	if (page_found)	{		Assert(BufferIsValid(mapbuf));		/* mark page "in use" in the bitmap */		SETBIT(freep, bitmap_page_bit);		MarkBufferDirty(mapbuf);	}

/* * Write XLOG record describing a page update. The update can include any * number of deletions and/or insertions of tuples on a single index page. * * If this update inserts a downlink for a split page, also record that * the F_FOLLOW_RIGHT flag on the child page is cleared and NSN set. * * Note that both the todelete array and the tuples are marked as belonging * to the target buffer; they need not be stored in XLOG if XLogInsert decides * to log the whole buffer contents instead.  Also, we take care that there's * at least one rdata item referencing the buffer, even when ntodelete and * ituplen are both zero; this ensures that XLogInsert knows about the buffer. */XLogRecPtrgistXLogUpdate(RelFileNode node, Buffer buffer,			   OffsetNumber *todelete, int ntodelete,			   IndexTuple *itup, int ituplen,			   Buffer leftchildbuf){	XLogRecData *rdata;	gistxlogPageUpdate xlrec;	int			cur,				i;	XLogRecPtr	recptr;	rdata = (XLogRecData *) palloc(sizeof(XLogRecData) * (3 + ituplen));	xlrec.node = node;	xlrec.blkno = BufferGetBlockNumber(buffer);	xlrec.ntodelete = ntodelete;	xlrec.leftchild =		BufferIsValid(leftchildbuf) ? BufferGetBlockNumber(leftchildbuf) : InvalidBlockNumber;	rdata[0].data = (char *) &xlrec;	rdata[0].len = sizeof(gistxlogPageUpdate);	rdata[0].buffer = InvalidBuffer;	rdata[0].next = &(rdata[1]);	rdata[1].data = (char *) todelete;	rdata[1].len = sizeof(OffsetNumber) * ntodelete;	rdata[1].buffer = buffer;	rdata[1].buffer_std = true;	cur = 2;	/* new tuples */	for (i = 0; i < ituplen; i++)	{		rdata[cur - 1].next = &(rdata[cur]);		rdata[cur].data = (char *) (itup[i]);		rdata[cur].len = IndexTupleSize(itup[i]);		rdata[cur].buffer = buffer;		rdata[cur].buffer_std = true;		cur++;	}	/*	 * Include a full page image of the child buf. (only necessary if a	 * checkpoint happened since the child page was split)	 */	if (BufferIsValid(leftchildbuf))	{		rdata[cur - 1].next = &(rdata[cur]);		rdata[cur].data = NULL;		rdata[cur].len = 0;		rdata[cur].buffer = leftchildbuf;		rdata[cur].buffer_std = true;		cur++;	}	rdata[cur - 1].next = NULL;	recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_UPDATE, rdata);	pfree(rdata);	return recptr;}

//.........这里部分代码省略.........	START_CRIT_SECTION();	/*	 * we have to insert tuples on the "write" page, being careful to preserve	 * hashkey ordering.  (If we insert many tuples into the same "write" page	 * it would be worth qsort'ing them).	 */	if (nitups > 0)	{		_hash_pgaddmultitup(rel, wbuf, itups, itup_offsets, nitups);		MarkBufferDirty(wbuf);	}	/*	 * Reinitialize the freed overflow page.  Just zeroing the page won't	 * work, because WAL replay routines expect pages to be initialized. See	 * explanation of RBM_NORMAL mode atop XLogReadBufferExtended.  We are	 * careful to make the special space valid here so that tools like	 * pageinspect won't get confused.	 */	_hash_pageinit(ovflpage, BufferGetPageSize(ovflbuf));	ovflopaque = (HashPageOpaque) PageGetSpecialPointer(ovflpage);	ovflopaque->hasho_prevblkno = InvalidBlockNumber;	ovflopaque->hasho_nextblkno = InvalidBlockNumber;	ovflopaque->hasho_bucket = -1;	ovflopaque->hasho_flag = LH_UNUSED_PAGE;	ovflopaque->hasho_page_id = HASHO_PAGE_ID;	MarkBufferDirty(ovflbuf);	if (BufferIsValid(prevbuf))	{		Page		prevpage = BufferGetPage(prevbuf);		HashPageOpaque prevopaque = (HashPageOpaque) PageGetSpecialPointer(prevpage);		Assert(prevopaque->hasho_bucket == bucket);		prevopaque->hasho_nextblkno = nextblkno;		MarkBufferDirty(prevbuf);	}	if (BufferIsValid(nextbuf))	{		Page		nextpage = BufferGetPage(nextbuf);		HashPageOpaque nextopaque = (HashPageOpaque) PageGetSpecialPointer(nextpage);		Assert(nextopaque->hasho_bucket == bucket);		nextopaque->hasho_prevblkno = prevblkno;		MarkBufferDirty(nextbuf);	}	/* Clear the bitmap bit to indicate that this overflow page is free */	CLRBIT(freep, bitmapbit);	MarkBufferDirty(mapbuf);	/* if this is now the first free page, update hashm_firstfree */	if (ovflbitno < metap->hashm_firstfree)	{		metap->hashm_firstfree = ovflbitno;		update_metap = true;		MarkBufferDirty(metabuf);	}	/* XLOG stuff */	if (RelationNeedsWAL(rel))

/* * Scan a complete BRIN index, and summarize each page range that's not already * summarized.  The index and heap must have been locked by caller in at * least ShareUpdateExclusiveLock mode. * * For each new index tuple inserted, *numSummarized (if not NULL) is * incremented; for each existing tuple, *numExisting (if not NULL) is * incremented. */static voidbrinsummarize(Relation index, Relation heapRel, double *numSummarized,			  double *numExisting){	BrinRevmap *revmap;	BrinBuildState *state = NULL;	IndexInfo  *indexInfo = NULL;	BlockNumber heapNumBlocks;	BlockNumber pagesPerRange;	Buffer		buf;	BlockNumber startBlk;	revmap = brinRevmapInitialize(index, &pagesPerRange, NULL);	/* determine range of pages to process: always start from the beginning */	heapNumBlocks = RelationGetNumberOfBlocks(heapRel);	startBlk = 0;	/*	 * Scan the revmap to find unsummarized items.	 */	buf = InvalidBuffer;	for (; startBlk < heapNumBlocks; startBlk += pagesPerRange)	{		BrinTuple  *tup;		OffsetNumber off;		/*		 * Go away now if we think the next range is partial.		 */		if (startBlk + pagesPerRange > heapNumBlocks)			break;		CHECK_FOR_INTERRUPTS();		tup = brinGetTupleForHeapBlock(revmap, startBlk, &buf, &off, NULL,									   BUFFER_LOCK_SHARE, NULL);		if (tup == NULL)		{			/* no revmap entry for this heap range. Summarize it. */			if (state == NULL)			{				/* first time through */				Assert(!indexInfo);				state = initialize_brin_buildstate(index, revmap,												   pagesPerRange);				indexInfo = BuildIndexInfo(index);			}			summarize_range(indexInfo, state, heapRel, startBlk, heapNumBlocks);			/* and re-initialize state for the next range */			brin_memtuple_initialize(state->bs_dtuple, state->bs_bdesc);			if (numSummarized)				*numSummarized += 1.0;		}		else		{			if (numExisting)				*numExisting += 1.0;			LockBuffer(buf, BUFFER_LOCK_UNLOCK);		}	}	if (BufferIsValid(buf))		ReleaseBuffer(buf);	/* free resources */	brinRevmapTerminate(revmap);	if (state)	{		terminate_brin_buildstate(state);		pfree(indexInfo);	}}

/* *	_bt_readpage() -- Load data from current index page into so->currPos * * Caller must have pinned and read-locked so->currPos.buf; the buffer's state * is not changed here.  Also, currPos.moreLeft and moreRight must be valid; * they are updated as appropriate.  All other fields of so->currPos are * initialized from scratch here. * * We scan the current page starting at offnum and moving in the indicated * direction.  All items matching the scan keys are loaded into currPos.items. * moreLeft or moreRight (as appropriate) is cleared if _bt_checkkeys reports * that there can be no more matching tuples in the current scan direction. * * Returns true if any matching items found on the page, false if none. */static bool_bt_readpage(IndexScanDesc scan, ScanDirection dir, OffsetNumber offnum){	BTScanOpaque so = (BTScanOpaque) scan->opaque;	Page		page;	BTPageOpaque opaque;	OffsetNumber minoff;	OffsetNumber maxoff;	int			itemIndex;	IndexTuple	itup;	bool		continuescan;	/* we must have the buffer pinned and locked */	Assert(BufferIsValid(so->currPos.buf));	page = BufferGetPage(so->currPos.buf);	opaque = (BTPageOpaque) PageGetSpecialPointer(page);	minoff = P_FIRSTDATAKEY(opaque);	maxoff = PageGetMaxOffsetNumber(page);	/*	 * we must save the page's right-link while scanning it; this tells us	 * where to step right to after we're done with these items.  There is no	 * corresponding need for the left-link, since splits always go right.	 */	so->currPos.nextPage = opaque->btpo_next;	/* initialize tuple workspace to empty */	so->currPos.nextTupleOffset = 0;	if (ScanDirectionIsForward(dir))	{		/* load items[] in ascending order */		itemIndex = 0;		offnum = Max(offnum, minoff);		while (offnum <= maxoff)		{			itup = _bt_checkkeys(scan, page, offnum, dir, &continuescan);			if (itup != NULL)			{				/* tuple passes all scan key conditions, so remember it */				_bt_saveitem(so, itemIndex, offnum, itup);				itemIndex++;			}			if (!continuescan)			{				/* there can't be any more matches, so stop */				so->currPos.moreRight = false;				break;			}			offnum = OffsetNumberNext(offnum);		}		Assert(itemIndex <= MaxIndexTuplesPerPage);		so->currPos.firstItem = 0;		so->currPos.lastItem = itemIndex - 1;		so->currPos.itemIndex = 0;	}	else	{		/* load items[] in descending order */		itemIndex = MaxIndexTuplesPerPage;		offnum = Min(offnum, maxoff);		while (offnum >= minoff)		{			itup = _bt_checkkeys(scan, page, offnum, dir, &continuescan);			if (itup != NULL)			{				/* tuple passes all scan key conditions, so remember it */				itemIndex--;				_bt_saveitem(so, itemIndex, offnum, itup);			}			if (!continuescan)			{				/* there can't be any more matches, so stop */				so->currPos.moreLeft = false;				break;			}			offnum = OffsetNumberPrev(offnum);//.........这里部分代码省略.........

static voidspgRedoMoveLeafs(XLogRecPtr lsn, XLogRecord *record){	char	   *ptr = XLogRecGetData(record);	spgxlogMoveLeafs *xldata = (spgxlogMoveLeafs *) ptr;	SpGistState state;	OffsetNumber *toDelete;	OffsetNumber *toInsert;	int			nInsert;	Buffer		buffer;	Page		page;	fillFakeState(&state, xldata->stateSrc);	nInsert = xldata->replaceDead ? 1 : xldata->nMoves + 1;	ptr += MAXALIGN(sizeof(spgxlogMoveLeafs));	toDelete = (OffsetNumber *) ptr;	ptr += MAXALIGN(sizeof(OffsetNumber) * xldata->nMoves);	toInsert = (OffsetNumber *) ptr;	ptr += MAXALIGN(sizeof(OffsetNumber) * nInsert);	/* now ptr points to the list of leaf tuples */	/* Insert tuples on the dest page (do first, so redirect is valid) */	if (!(record->xl_info & XLR_BKP_BLOCK_2))	{		buffer = XLogReadBuffer(xldata->node, xldata->blknoDst,								xldata->newPage);		if (BufferIsValid(buffer))		{			page = BufferGetPage(buffer);			if (xldata->newPage)				SpGistInitBuffer(buffer, SPGIST_LEAF);			if (!XLByteLE(lsn, PageGetLSN(page)))			{				int			i;				for (i = 0; i < nInsert; i++)				{					SpGistLeafTuple lt = (SpGistLeafTuple) ptr;					addOrReplaceTuple(page, (Item) lt, lt->size, toInsert[i]);					ptr += lt->size;				}				PageSetLSN(page, lsn);				PageSetTLI(page, ThisTimeLineID);				MarkBufferDirty(buffer);			}			UnlockReleaseBuffer(buffer);		}	}	/* Delete tuples from the source page, inserting a redirection pointer */	if (!(record->xl_info & XLR_BKP_BLOCK_1))	{		buffer = XLogReadBuffer(xldata->node, xldata->blknoSrc, false);		if (BufferIsValid(buffer))		{			page = BufferGetPage(buffer);			if (!XLByteLE(lsn, PageGetLSN(page)))			{				spgPageIndexMultiDelete(&state, page, toDelete, xldata->nMoves,										state.isBuild ? SPGIST_PLACEHOLDER : SPGIST_REDIRECT,										SPGIST_PLACEHOLDER,										xldata->blknoDst,										toInsert[nInsert - 1]);				PageSetLSN(page, lsn);				PageSetTLI(page, ThisTimeLineID);				MarkBufferDirty(buffer);			}			UnlockReleaseBuffer(buffer);		}	}	/* And update the parent downlink */	if (!(record->xl_info & XLR_BKP_BLOCK_3))	{		buffer = XLogReadBuffer(xldata->node, xldata->blknoParent, false);		if (BufferIsValid(buffer))		{			page = BufferGetPage(buffer);			if (!XLByteLE(lsn, PageGetLSN(page)))			{				SpGistInnerTuple tuple;				tuple = (SpGistInnerTuple) PageGetItem(page,													   PageGetItemId(page, xldata->offnumParent));				spgUpdateNodeLink(tuple, xldata->nodeI,								  xldata->blknoDst, toInsert[nInsert - 1]);				PageSetLSN(page, lsn);				PageSetTLI(page, ThisTimeLineID);				MarkBufferDirty(buffer);			}//.........这里部分代码省略.........

/* *	_bt_steppage() -- Step to next page containing valid data for scan * * On entry, so->currPos.buf must be pinned and read-locked.  We'll drop * the lock and pin before moving to next page. * * On success exit, we hold pin and read-lock on the next interesting page, * and so->currPos is updated to contain data from that page. * * If there are no more matching records in the given direction, we drop all * locks and pins, set so->currPos.buf to InvalidBuffer, and return FALSE. */static bool_bt_steppage(IndexScanDesc scan, ScanDirection dir){	BTScanOpaque so = (BTScanOpaque) scan->opaque;	Relation	rel;	Page		page;	BTPageOpaque opaque;	/* we must have the buffer pinned and locked */	Assert(BufferIsValid(so->currPos.buf));	/* Before leaving current page, deal with any killed items */	if (so->numKilled > 0)		_bt_killitems(scan, true);	/*	 * Before we modify currPos, make a copy of the page data if there was a	 * mark position that needs it.	 */	if (so->markItemIndex >= 0)	{		/* bump pin on current buffer for assignment to mark buffer */		IncrBufferRefCount(so->currPos.buf);		memcpy(&so->markPos, &so->currPos,			   offsetof(BTScanPosData, items[1]) +			   so->currPos.lastItem * sizeof(BTScanPosItem));		if (so->markTuples)			memcpy(so->markTuples, so->currTuples,				   so->currPos.nextTupleOffset);		so->markPos.itemIndex = so->markItemIndex;		so->markItemIndex = -1;	}	rel = scan->indexRelation;	if (ScanDirectionIsForward(dir))	{		/* Walk right to the next page with data */		/* We must rely on the previously saved nextPage link! */		BlockNumber blkno = so->currPos.nextPage;		/* Remember we left a page with data */		so->currPos.moreLeft = true;		for (;;)		{			/* release the previous buffer */			_bt_relbuf(rel, so->currPos.buf);			so->currPos.buf = InvalidBuffer;			/* if we're at end of scan, give up */			if (blkno == P_NONE || !so->currPos.moreRight)				return false;			/* check for interrupts while we're not holding any buffer lock */			CHECK_FOR_INTERRUPTS();			/* step right one page */			so->currPos.buf = _bt_getbuf(rel, blkno, BT_READ);			/* check for deleted page */			page = BufferGetPage(so->currPos.buf);			opaque = (BTPageOpaque) PageGetSpecialPointer(page);			if (!P_IGNORE(opaque))			{				PredicateLockPage(rel, blkno, scan->xs_snapshot);				/* see if there are any matches on this page */				/* note that this will clear moreRight if we can stop */				if (_bt_readpage(scan, dir, P_FIRSTDATAKEY(opaque)))					break;			}			/* nope, keep going */			blkno = opaque->btpo_next;		}	}	else	{		/* Remember we left a page with data */		so->currPos.moreRight = true;		/*		 * Walk left to the next page with data.  This is much more complex		 * than the walk-right case because of the possibility that the page		 * to our left splits while we are in flight to it, plus the		 * possibility that the page we were on gets deleted after we leave		 * it.	See nbtree/README for details.		 */		for (;;)		{			/* Done if we know there are no matching keys to the left */			if (!so->currPos.moreLeft)			{//.........这里部分代码省略.........

static voidspgRedoSplitTuple(XLogRecPtr lsn, XLogRecord *record){	char	   *ptr = XLogRecGetData(record);	spgxlogSplitTuple *xldata = (spgxlogSplitTuple *) ptr;	SpGistInnerTuple prefixTuple;	SpGistInnerTuple postfixTuple;	Buffer		buffer;	Page		page;	/* we assume this is adequately aligned */	ptr += sizeof(spgxlogSplitTuple);	prefixTuple = (SpGistInnerTuple) ptr;	ptr += prefixTuple->size;	postfixTuple = (SpGistInnerTuple) ptr;	/* insert postfix tuple first to avoid dangling link */	if (xldata->blknoPostfix != xldata->blknoPrefix &&		!(record->xl_info & XLR_BKP_BLOCK_2))	{		buffer = XLogReadBuffer(xldata->node, xldata->blknoPostfix,								xldata->newPage);		if (BufferIsValid(buffer))		{			page = BufferGetPage(buffer);			if (xldata->newPage)				SpGistInitBuffer(buffer, 0);			if (!XLByteLE(lsn, PageGetLSN(page)))			{				addOrReplaceTuple(page, (Item) postfixTuple,								  postfixTuple->size, xldata->offnumPostfix);				PageSetLSN(page, lsn);				PageSetTLI(page, ThisTimeLineID);				MarkBufferDirty(buffer);			}			UnlockReleaseBuffer(buffer);		}	}	/* now handle the original page */	if (!(record->xl_info & XLR_BKP_BLOCK_1))	{		buffer = XLogReadBuffer(xldata->node, xldata->blknoPrefix, false);		if (BufferIsValid(buffer))		{			page = BufferGetPage(buffer);			if (!XLByteLE(lsn, PageGetLSN(page)))			{				PageIndexTupleDelete(page, xldata->offnumPrefix);				if (PageAddItem(page, (Item) prefixTuple, prefixTuple->size,								xldata->offnumPrefix, false, false) != xldata->offnumPrefix)					elog(ERROR, "failed to add item of size %u to SPGiST index page",						 prefixTuple->size);				if (xldata->blknoPostfix == xldata->blknoPrefix)					addOrReplaceTuple(page, (Item) postfixTuple,									  postfixTuple->size,									  xldata->offnumPostfix);				PageSetLSN(page, lsn);				PageSetTLI(page, ThisTimeLineID);				MarkBufferDirty(buffer);			}			UnlockReleaseBuffer(buffer);		}	}}

/* * _bt_get_endpoint() -- Find the first or last page on a given tree level * * If the index is empty, we will return InvalidBuffer; any other failure * condition causes ereport().	We will not return a dead page. * * The returned buffer is pinned and read-locked. */Buffer_bt_get_endpoint(Relation rel, uint32 level, bool rightmost){	Buffer		buf;	Page		page;	BTPageOpaque opaque;	OffsetNumber offnum;	BlockNumber blkno;	IndexTuple	itup;	/*	 * If we are looking for a leaf page, okay to descend from fast root;	 * otherwise better descend from true root.  (There is no point in being	 * smarter about intermediate levels.)	 */	if (level == 0)		buf = _bt_getroot(rel, BT_READ);	else		buf = _bt_gettrueroot(rel);	if (!BufferIsValid(buf))		return InvalidBuffer;	page = BufferGetPage(buf);	opaque = (BTPageOpaque) PageGetSpecialPointer(page);	for (;;)	{		/*		 * If we landed on a deleted page, step right to find a live page		 * (there must be one).  Also, if we want the rightmost page, step		 * right if needed to get to it (this could happen if the page split		 * since we obtained a pointer to it).		 */		while (P_IGNORE(opaque) ||			   (rightmost && !P_RIGHTMOST(opaque)))		{			blkno = opaque->btpo_next;			if (blkno == P_NONE)				elog(ERROR, "fell off the end of index /"%s/"",					 RelationGetRelationName(rel));			buf = _bt_relandgetbuf(rel, buf, blkno, BT_READ);			page = BufferGetPage(buf);			opaque = (BTPageOpaque) PageGetSpecialPointer(page);		}		/* Done? */		if (opaque->btpo.level == level)			break;		if (opaque->btpo.level < level)			elog(ERROR, "btree level %u not found in index /"%s/"",				 level, RelationGetRelationName(rel));		/* Descend to leftmost or rightmost child page */		if (rightmost)			offnum = PageGetMaxOffsetNumber(page);		else			offnum = P_FIRSTDATAKEY(opaque);		itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, offnum));		blkno = ItemPointerGetBlockNumber(&(itup->t_tid));		buf = _bt_relandgetbuf(rel, buf, blkno, BT_READ);		page = BufferGetPage(buf);		opaque = (BTPageOpaque) PageGetSpecialPointer(page);	}	return buf;}

static voidspgRedoVacuumLeaf(XLogRecPtr lsn, XLogRecord *record){	char	   *ptr = XLogRecGetData(record);	spgxlogVacuumLeaf *xldata = (spgxlogVacuumLeaf *) ptr;	OffsetNumber *toDead;	OffsetNumber *toPlaceholder;	OffsetNumber *moveSrc;	OffsetNumber *moveDest;	OffsetNumber *chainSrc;	OffsetNumber *chainDest;	SpGistState state;	Buffer		buffer;	Page		page;	int			i;	fillFakeState(&state, xldata->stateSrc);	ptr += sizeof(spgxlogVacuumLeaf);	toDead = (OffsetNumber *) ptr;	ptr += sizeof(OffsetNumber) * xldata->nDead;	toPlaceholder = (OffsetNumber *) ptr;	ptr += sizeof(OffsetNumber) * xldata->nPlaceholder;	moveSrc = (OffsetNumber *) ptr;	ptr += sizeof(OffsetNumber) * xldata->nMove;	moveDest = (OffsetNumber *) ptr;	ptr += sizeof(OffsetNumber) * xldata->nMove;	chainSrc = (OffsetNumber *) ptr;	ptr += sizeof(OffsetNumber) * xldata->nChain;	chainDest = (OffsetNumber *) ptr;	if (!(record->xl_info & XLR_BKP_BLOCK_1))	{		buffer = XLogReadBuffer(xldata->node, xldata->blkno, false);		if (BufferIsValid(buffer))		{			page = BufferGetPage(buffer);			if (!XLByteLE(lsn, PageGetLSN(page)))			{				spgPageIndexMultiDelete(&state, page,										toDead, xldata->nDead,										SPGIST_DEAD, SPGIST_DEAD,										InvalidBlockNumber,										InvalidOffsetNumber);				spgPageIndexMultiDelete(&state, page,										toPlaceholder, xldata->nPlaceholder,										SPGIST_PLACEHOLDER, SPGIST_PLACEHOLDER,										InvalidBlockNumber,										InvalidOffsetNumber);				/* see comments in vacuumLeafPage() */				for (i = 0; i < xldata->nMove; i++)				{					ItemId		idSrc = PageGetItemId(page, moveSrc[i]);					ItemId		idDest = PageGetItemId(page, moveDest[i]);					ItemIdData	tmp;					tmp = *idSrc;					*idSrc = *idDest;					*idDest = tmp;				}				spgPageIndexMultiDelete(&state, page,										moveSrc, xldata->nMove,										SPGIST_PLACEHOLDER, SPGIST_PLACEHOLDER,										InvalidBlockNumber,										InvalidOffsetNumber);				for (i = 0; i < xldata->nChain; i++)				{					SpGistLeafTuple lt;					lt = (SpGistLeafTuple) PageGetItem(page,										   PageGetItemId(page, chainSrc[i]));					Assert(lt->tupstate == SPGIST_LIVE);					lt->nextOffset = chainDest[i];				}				PageSetLSN(page, lsn);				PageSetTLI(page, ThisTimeLineID);				MarkBufferDirty(buffer);			}			UnlockReleaseBuffer(buffer);		}	}}

/* *	_hash_step() -- step to the next valid item in a scan in the bucket. * *		If no valid record exists in the requested direction, return *		false.	Else, return true and set the hashso_curpos for the *		scan to the right thing. * *		'bufP' points to the current buffer, which is pinned and read-locked. *		On success exit, we have pin and read-lock on whichever page *		contains the right item; on failure, we have released all buffers. */bool_hash_step(IndexScanDesc scan, Buffer *bufP, ScanDirection dir){	Relation	rel = scan->indexRelation;	HashScanOpaque so = (HashScanOpaque) scan->opaque;	ItemPointer current;	Buffer		buf;	Page		page;	HashPageOpaque opaque;	OffsetNumber maxoff;	OffsetNumber offnum;	BlockNumber blkno;	IndexTuple	itup;	current = &(so->hashso_curpos);	buf = *bufP;	_hash_checkpage(rel, buf, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);	page = BufferGetPage(buf);	opaque = (HashPageOpaque) PageGetSpecialPointer(page);	/*	 * If _hash_step is called from _hash_first, current will not be valid, so	 * we can't dereference it.  However, in that case, we presumably want to	 * start at the beginning/end of the page...	 */	maxoff = PageGetMaxOffsetNumber(page);	if (ItemPointerIsValid(current))		offnum = ItemPointerGetOffsetNumber(current);	else		offnum = InvalidOffsetNumber;	/*	 * 'offnum' now points to the last tuple we have seen (if any).	 *	 * continue to step through tuples until: 1) we get to the end of the	 * bucket chain or 2) we find a valid tuple.	 */	do	{		switch (dir)		{			case ForwardScanDirection:				if (offnum != InvalidOffsetNumber)					offnum = OffsetNumberNext(offnum);	/* move forward */				else					offnum = FirstOffsetNumber; /* new page */				while (offnum > maxoff)				{					/*					 * either this page is empty (maxoff ==					 * InvalidOffsetNumber) or we ran off the end.					 */					_hash_readnext(rel, &buf, &page, &opaque);					if (BufferIsValid(buf))					{						maxoff = PageGetMaxOffsetNumber(page);						offnum = FirstOffsetNumber;					}					else					{						/* end of bucket */						maxoff = offnum = InvalidOffsetNumber;						break;	/* exit while */					}				}				break;			case BackwardScanDirection:				if (offnum != InvalidOffsetNumber)					offnum = OffsetNumberPrev(offnum);	/* move back */				else					offnum = maxoff;	/* new page */				while (offnum < FirstOffsetNumber)				{					/*					 * either this page is empty (offnum ==					 * InvalidOffsetNumber) or we ran off the end.					 */					_hash_readprev(rel, &buf, &page, &opaque);					if (BufferIsValid(buf))						maxoff = offnum = PageGetMaxOffsetNumber(page);					else					{						/* end of bucket */						maxoff = offnum = InvalidOffsetNumber;						break;	/* exit while *///.........这里部分代码省略.........