nbtsearch.c

关系型数据库 Postgresql 6.5.2

{
	BTPageOpaque opaque;
	OffsetNumber limit;

	opaque = (BTPageOpaque) PageGetSpecialPointer(page);

	/* skip the high key, if any */
	limit = P_RIGHTMOST(opaque) ? P_HIKEY : P_FIRSTKEY;

	/* walk backwards looking for the first key in the chain of duplicates */
	while (offnum > limit
		   && _bt_compare(rel, itupdesc, page,
						  keysz, scankey, OffsetNumberPrev(offnum)) == 0)
		offnum = OffsetNumberPrev(offnum);

	return offnum;
}

/*
 *	_bt_compare() -- Compare scankey to a particular tuple on the page.
 *
 *		This routine returns:
 *			-1 if scankey < tuple at offnum;
 *			 0 if scankey == tuple at offnum;
 *			+1 if scankey > tuple at offnum.
 *
 *		-- Old comments:
 *		In order to avoid having to propagate changes up the tree any time
 *		a new minimal key is inserted, the leftmost entry on the leftmost
 *		page is less than all possible keys, by definition.
 *
 *		-- New ones:
 *		New insertion code (fix against updating _in_place_ if new minimal
 *		key has bigger size than old one) may delete P_HIKEY entry on the
 *		root page in order to insert new minimal key - and so this definition
 *		does not work properly in this case and breaks key' order on root
 *		page. BTW, this propagation occures only while page' splitting,
 *		but not "any time a new min key is inserted" (see _bt_insertonpg).
 *				- vadim 12/05/96
 */
static int
_bt_compare(Relation rel,
			TupleDesc itupdesc,
			Page page,
			int keysz,
			ScanKey scankey,
			OffsetNumber offnum)
{
	Datum		datum;
	BTItem		btitem;
	ItemId		itemid;
	IndexTuple	itup;
	BTPageOpaque opaque;
	ScanKey		entry;
	AttrNumber	attno;
	int			result;
	int			i;
	bool		null;

	/*
	 * If this is a leftmost internal page, and if our comparison is with
	 * the first key on the page, then the item at that position is by
	 * definition less than the scan key.
	 *
	 * - see new comments above...
	 */

	opaque = (BTPageOpaque) PageGetSpecialPointer(page);
	if (!(opaque->btpo_flags & BTP_LEAF)
		&& P_LEFTMOST(opaque)
		&& offnum == P_HIKEY)
	{
		itemid = PageGetItemId(page, offnum);

		/*
		 * we just have to believe that this will only be called with
		 * offnum == P_HIKEY when P_HIKEY is the OffsetNumber of the first
		 * actual data key (i.e., this is also a rightmost page).  there
		 * doesn't seem to be any code that implies that the leftmost page
		 * is normally missing a high key as well as the rightmost page.
		 * but that implies that this code path only applies to the root
		 * -- which seems unlikely..
		 *
		 * - see new comments above...
		 */
		if (!P_RIGHTMOST(opaque))
			elog(ERROR, "_bt_compare: invalid comparison to high key");

#ifdef NOT_USED

		/*
		 * We just have to belive that right answer will not break
		 * anything. I've checked code and all seems to be ok. See new
		 * comments above...
		 *
		 * -- Old comments If the item on the page is equal to the scankey,
		 * that's okay to admit.  We just can't claim that the first key
		 * on the page is greater than anything.
		 */

		if (_bt_skeycmp(rel, keysz, scankey, page, itemid,
						BTEqualStrategyNumber))
			return 0;
		return 1;
#endif
	}

	btitem = (BTItem) PageGetItem(page, PageGetItemId(page, offnum));
	itup = &(btitem->bti_itup);

	/*
	 * The scan key is set up with the attribute number associated with
	 * each term in the key.  It is important that, if the index is
	 * multi-key, the scan contain the first k key attributes, and that
	 * they be in order.  If you think about how multi-key ordering works,
	 * you'll understand why this is.
	 *
	 * We don't test for violation of this condition here.
	 */

	for (i = 1; i <= keysz; i++)
	{
		long		tmpres;

		entry = &scankey[i - 1];
		attno = entry->sk_attno;
		datum = index_getattr(itup, attno, itupdesc, &null);

		/* see comments about NULLs handling in btbuild */
		if (entry->sk_flags & SK_ISNULL)		/* key is NULL */
		{
			Assert(entry->sk_procedure == F_NULLVALUE);
			if (null)
				tmpres = (long) 0;		/* NULL "=" NULL */
			else
				tmpres = (long) 1;		/* NULL ">" NOT_NULL */
		}
		else if (null)			/* key is NOT_NULL and item is NULL */
		{
			tmpres = (long) -1; /* NOT_NULL "<" NULL */
		}
		else
			tmpres = (long) FMGR_PTR2(&entry->sk_func, entry->sk_argument, datum);
		result = tmpres;

		/* if the keys are unequal, return the difference */
		if (result != 0)
			return result;
	}

	/* by here, the keys are equal */
	return 0;
}

/*
 *	_bt_next() -- Get the next item in a scan.
 *
 *		On entry, we have a valid currentItemData in the scan, and a
 *		read lock on the page that contains that item.	We do not have
 *		the page pinned.  We return the next item in the scan.	On
 *		exit, we have the page containing the next item locked but not
 *		pinned.
 */
RetrieveIndexResult
_bt_next(IndexScanDesc scan, ScanDirection dir)
{
	Relation	rel;
	Buffer		buf;
	Page		page;
	OffsetNumber offnum;
	RetrieveIndexResult res;
	ItemPointer current;
	BTItem		btitem;
	IndexTuple	itup;
	BTScanOpaque so;
	Size		keysok;

	rel = scan->relation;
	so = (BTScanOpaque) scan->opaque;
	current = &(scan->currentItemData);

	Assert(BufferIsValid(so->btso_curbuf));

	/* we still have the buffer pinned and locked */
	buf = so->btso_curbuf;

	do
	{
		/* step one tuple in the appropriate direction */
		if (!_bt_step(scan, &buf, dir))
			return (RetrieveIndexResult) NULL;

		/* by here, current is the tuple we want to return */
		offnum = ItemPointerGetOffsetNumber(current);
		page = BufferGetPage(buf);
		btitem = (BTItem) PageGetItem(page, PageGetItemId(page, offnum));
		itup = &btitem->bti_itup;

		if (_bt_checkkeys(scan, itup, &keysok))
		{
			Assert(keysok == so->numberOfKeys);
			res = FormRetrieveIndexResult(current, &(itup->t_tid));

			/* remember which buffer we have pinned and locked */
			so->btso_curbuf = buf;
			return res;
		}

	} while (keysok >= so->numberOfFirstKeys ||
			 (keysok == -1 && ScanDirectionIsBackward(dir)));

	ItemPointerSetInvalid(current);
	so->btso_curbuf = InvalidBuffer;
	_bt_relbuf(rel, buf, BT_READ);

	return (RetrieveIndexResult) NULL;
}

/*
 *	_bt_first() -- Find the first item in a scan.
 *
 *		We need to be clever about the type of scan, the operation it's
 *		performing, and the tree ordering.	We return the RetrieveIndexResult
 *		of the first item in the tree that satisfies the qualification
 *		associated with the scan descriptor.  On exit, the page containing
 *		the current index tuple is read locked and pinned, and the scan's
 *		opaque data entry is updated to include the buffer.
 */
RetrieveIndexResult
_bt_first(IndexScanDesc scan, ScanDirection dir)
{
	Relation	rel;
	TupleDesc	itupdesc;
	Buffer		buf;
	Page		page;
	BTPageOpaque pop;
	BTStack		stack;
	OffsetNumber offnum,
				maxoff;
	bool		offGmax = false;
	BTItem		btitem;
	IndexTuple	itup;
	ItemPointer current;
	BlockNumber blkno;
	StrategyNumber strat;
	RetrieveIndexResult res;
	RegProcedure proc;
	int			result;
	BTScanOpaque so;
	ScanKeyData skdata;
	Size		keysok;
	int			i;
	int			nKeyIndex = -1;

	rel = scan->relation;
	so = (BTScanOpaque) scan->opaque;

	/*
	 * Order the keys in the qualification and be sure that the scan
	 * exploits the tree order.
	 */
	so->numberOfFirstKeys = 0;	/* may be changed by _bt_orderkeys */
	so->qual_ok = 1;			/* may be changed by _bt_orderkeys */
	scan->scanFromEnd = false;
	if (so->numberOfKeys > 0)
	{
		_bt_orderkeys(rel, so);

		if (ScanDirectionIsBackward(dir))
		{
			for (i = 0; i < so->numberOfKeys; i++)
			{
				if (so->keyData[i].sk_attno != 1)
					break;
				strat = _bt_getstrat(rel, so->keyData[i].sk_attno,
									 so->keyData[i].sk_procedure);
				if (strat == BTLessStrategyNumber ||
					strat == BTLessEqualStrategyNumber ||
					strat == BTEqualStrategyNumber)
				{
					nKeyIndex = i;
					break;
				}
			}
		}
		else
		{
			strat = _bt_getstrat(rel, 1, so->keyData[0].sk_procedure);

			if (strat == BTLessStrategyNumber ||
				strat == BTLessEqualStrategyNumber)
				;
			else
				nKeyIndex = 0;
		}
		if (nKeyIndex < 0)
			scan->scanFromEnd = true;
	}
	else
		scan->scanFromEnd = true;

	if (so->qual_ok == 0)
		return (RetrieveIndexResult) NULL;

	/* if we just need to walk down one edge of the tree, do that */
	if (scan->scanFromEnd)
		return _bt_endpoint(scan, dir);

	itupdesc = RelationGetDescr(rel);
	current = &(scan->currentItemData);

	/*
	 * Okay, we want something more complicated.  What we'll do is use the
	 * first item in the scan key passed in (which has been correctly
	 * ordered to take advantage of index ordering) to position ourselves
	 * at the right place in the scan.
	 */
	/* _bt_orderkeys disallows it, but it's place to add some code latter */
	if (so->keyData[0].sk_flags & SK_ISNULL)
	{
		elog(ERROR, "_bt_first: btree doesn't support is(not)null, yet");
		return (RetrieveIndexResult) NULL;
	}
	proc = index_getprocid(rel, 1, BTORDER_PROC);
	ScanKeyEntryInitialize(&skdata, so->keyData[nKeyIndex].sk_flags,
						   1, proc, so->keyData[nKeyIndex].sk_argument);

	stack = _bt_search(rel, 1, &skdata, &buf);
	_bt_freestack(stack);

	blkno = BufferGetBlockNumber(buf);
	page = BufferGetPage(buf);

	/*
	 * This will happen if the tree we're searching is entirely empty, or
	 * if we're doing a search for a key that would appear on an entirely
	 * empty internal page.  In either case, there are no matching tuples
	 * in the index.
	 */

	if (PageIsEmpty(page))
	{
		ItemPointerSetInvalid(current);
		so->btso_curbuf = InvalidBuffer;
		_bt_relbuf(rel, buf, BT_READ);
		return (RetrieveIndexResult) NULL;
	}
	maxoff = PageGetMaxOffsetNumber(page);
	pop = (BTPageOpaque) PageGetSpecialPointer(page);

	/*
	 * Now _bt_moveright doesn't move from non-rightmost leaf page if
	 * scankey == hikey and there is only hikey there. It's good for
	 * insertion, but we need to do work for scan here. - vadim 05/27/97
	 */

	while (maxoff == P_HIKEY && !P_RIGHTMOST(pop) &&
		   _bt_skeycmp(rel, 1, &skdata, page,
					   PageGetItemId(page, P_HIKEY),
					   BTGreaterEqualStrategyNumber))
	{
		/* step right one page */
		blkno = pop->btpo_next;
		_bt_relbuf(rel, buf, BT_READ);
		buf = _bt_getbuf(rel, blkno, BT_READ);
		page = BufferGetPage(buf);
		if (PageIsEmpty(page))
		{
			ItemPointerSetInvalid(current);
			so->btso_curbuf = InvalidBuffer;
			_bt_relbuf(rel, buf, BT_READ);
			return (RetrieveIndexResult) NULL;
		}
		maxoff = PageGetMaxOffsetNumber(page);
		pop = (BTPageOpaque) PageGetSpecialPointer(page);
	}


	/* find the nearest match to the manufactured scan key on the page */
	offnum = _bt_binsrch(rel, buf, 1, &skdata, BT_DESCENT);

	if (offnum > maxoff)
	{
		offnum = maxoff;
		offGmax = true;
	}

	ItemPointerSet(current, blkno, offnum);

	/*
	 * Now find the right place to start the scan.	Result is the value
	 * we're looking for minus the value we're looking at in the index.
	 */

	result = _bt_compare(rel, itupdesc, page, 1, &skdata, offnum);

	/* it's yet other place to add some code latter for is(not)null */

	strat = _bt_getstrat(rel, 1, so->keyData[nKeyIndex].sk_procedure);

	switch (strat)
	{
		case BTLessStrategyNumber:
			if (result <= 0)
			{
				do
				{
					if (!_bt_twostep(scan, &buf, BackwardScanDirection))
						break;

					offnum = ItemPointerGetOffsetNumber(current);
					page = BufferGetPage(buf);
					result = _bt_compare(rel, itupdesc, page, 1, &skdata, offnum);
				} while (result <= 0);

			}
			break;

		case BTLessEqualStrategyNumber:
			if (result >= 0)
			{
				do
				{
					if (!_bt_twostep(scan, &buf, ForwardScanDirection))
						break;

					offnum = ItemPointerGetOffsetNumber(current);
					page = BufferGetPage(buf);
					result = _bt_compare(rel, itupdesc, page, 1, &skdata, offnum);
				} while (result >= 0);

				if (result < 0)
					_bt_twostep(scan, &buf, BackwardScanDirection);
			}
			break;

		case BTEqualStrategyNumber:
			if (result != 0)
			{
				_bt_relbuf(scan->relation, buf, BT_READ);
				so->btso_curbuf = InvalidBuffer;
				ItemPointerSetInvalid(&(scan->currentItemData));
				return (RetrieveIndexResult) NULL;
			}
			else if (ScanDirectionIsBackward(dir))
			{
				do
				{
					if (!_bt_twostep(scan, &buf, ForwardScanDirection))
						break;

					offnum = ItemPointerGetOffsetNumber(current);
					page = BufferGetPage(buf);
					result = _bt_compare(rel, itupdesc, page, 1, &skdata, offnum);
				} while (result == 0);

				if (result < 0)
					_bt_twostep(scan, &buf, BackwardScanDirection);
			}
			break;

		case BTGreaterEqualStrategyNumber:
			if (offGmax)
			{
				if (result < 0)
				{
					Assert(!P_RIGHTMOST(pop) && maxoff == P_HIKEY);
					if (!_bt_step(scan, &buf, ForwardScanDirection))
					{
						_bt_relbuf(scan->relation, buf, BT_READ);
						so->btso_curbuf = InvalidBuffer;
						ItemPointerSetInvalid(&(scan->currentItemData));
						return (RetrieveIndexResult) NULL;
					}
				}
				else if (result > 0)
				{				/* Just remember:  _bt_binsrch() returns
								 * the OffsetNumber of the first matching
								 * key on the page, or the OffsetNumber at
								 * which the matching key WOULD APPEAR IF
								 * IT WERE on this page. No key on this
								 * page, but offnum from _bt_binsrch()
								 * greater maxoff - have to move right. -
								 * vadim 12/06/96 */
					_bt_twostep(scan, &buf, ForwardScanDirection);
				}
			}
			else if (result < 0)
			{
				do
				{
					if (!_bt_twostep(scan, &buf, BackwardScanDirection))
						break;

					page = BufferGetPage(buf);
					offnum = ItemPointerGetOffsetNumber(current);
					result = _bt_compare(rel, itupdesc, page, 1, &skdata, offnum);
				} while (result < 0);

				if (result > 0)
					_bt_twostep(scan, &buf, ForwardScanDirection);
			}
			break;

		case BTGreaterStrategyNumber:
			/* offGmax helps as above */
			if (result >= 0 || offGmax)
			{
				do
				{
					if (!_bt_twostep(scan, &buf, ForwardScanDirection))
						break;

					offnum = ItemPointerGetOffsetNumber(current);
					page = BufferGetPage(buf);
					result = _bt_compare(rel, itupdesc, page, 1, &skdata, offnum);