postgresql/src/backend/access/nbtree/nbtinsert.c

/*-------------------------------------------------------------------------
 *
 * btinsert.c--
 *    Item insertion in Lehman and Yao btrees for Postgres.
 *
 * Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *    $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtinsert.c,v 1.12 1997/04/16 01:48:11 vadim Exp $
 *
 *-------------------------------------------------------------------------
 */

#include <postgres.h>

#include <utils/memutils.h>
#include <storage/bufpage.h>
#include <access/nbtree.h>
#include <access/heapam.h>
#include <storage/bufmgr.h>
#include <fmgr.h>

#ifndef HAVE_MEMMOVE
# include <regex/utils.h>
#else
# include <string.h>
#endif

static InsertIndexResult _bt_insertonpg(Relation rel, Buffer buf, BTStack stack, int keysz, ScanKey scankey, BTItem btitem, BTItem afteritem);
static Buffer _bt_split(Relation rel, Buffer buf);
static OffsetNumber _bt_findsplitloc(Relation rel, Page page, OffsetNumber start, OffsetNumber maxoff, Size llimit);
static void _bt_newroot(Relation rel, Buffer lbuf, Buffer rbuf);
static OffsetNumber _bt_pgaddtup(Relation rel, Buffer buf, int keysz, ScanKey itup_scankey, Size itemsize, BTItem btitem, BTItem afteritem);
static bool _bt_goesonpg(Relation rel, Buffer buf, Size keysz, ScanKey scankey, BTItem afteritem);
static void _bt_updateitem(Relation rel, Size keysz, Buffer buf, BTItem oldItem, BTItem newItem);
static bool _bt_isequal (TupleDesc itupdesc, Page page, OffsetNumber offnum, int keysz, ScanKey scankey);

/*
 *  _bt_doinsert() -- Handle insertion of a single btitem in the tree.
 *
 *	This routine is called by the public interface routines, btbuild
 *	and btinsert.  By here, btitem is filled in, and has a unique
 *	(xid, seqno) pair.
 */
InsertIndexResult
_bt_doinsert(Relation rel, BTItem btitem, bool index_is_unique, Relation heapRel)
{
    ScanKey itup_scankey;
    IndexTuple itup;
    BTStack stack;
    Buffer buf;
    BlockNumber blkno;
    int natts;
    InsertIndexResult res;
    
    itup = &(btitem->bti_itup);
    
    /* we need a scan key to do our search, so build one */
    itup_scankey = _bt_mkscankey(rel, itup);
    natts = rel->rd_rel->relnatts;
    
    /* find the page containing this key */
    stack = _bt_search(rel, natts, itup_scankey, &buf);

    blkno = BufferGetBlockNumber(buf);
    
    /* trade in our read lock for a write lock */
    _bt_relbuf(rel, buf, BT_READ);
    buf = _bt_getbuf(rel, blkno, BT_WRITE);
    
    /*
     *  If the page was split between the time that we surrendered our
     *  read lock and acquired our write lock, then this page may no
     *  longer be the right place for the key we want to insert.  In this
     *  case, we need to move right in the tree.  See Lehman and Yao for
     *  an excruciatingly precise description.
     */
    
    buf = _bt_moveright(rel, buf, natts, itup_scankey, BT_WRITE);

    /* if we're not allowing duplicates, make sure the key isn't */
    /* already in the node */
    if ( index_is_unique )
    {
	OffsetNumber offset, maxoff;
	Page page;

	page = BufferGetPage(buf);
	maxoff = PageGetMaxOffsetNumber (page);

	offset = _bt_binsrch(rel, buf, natts, itup_scankey, BT_DESCENT);

	/* make sure the offset we're given points to an actual */
	/* key on the page before trying to compare it */
	if ( !PageIsEmpty (page) && offset <= maxoff )
	{
	    TupleDesc itupdesc;
	    BTItem btitem;
	    IndexTuple itup;
	    HeapTuple htup;
	    BTPageOpaque opaque;
	    Buffer nbuf;
	    BlockNumber blkno;
	    
	    itupdesc = RelationGetTupleDescriptor(rel);
	    nbuf = InvalidBuffer;
	    opaque = (BTPageOpaque) PageGetSpecialPointer(page);
	    /*
	     * _bt_compare returns 0 for (1,NULL) and (1,NULL) -
	     * this's how we handling NULLs - and so we must not use
	     * _bt_compare in real comparison, but only for
	     * ordering/finding items on pages.	- vadim 03/24/97

	    while ( !_bt_compare (rel, itupdesc, page, 
			    	natts, itup_scankey, offset) )
	     */
	    while ( _bt_isequal (itupdesc, page, offset, natts, itup_scankey) )
	    {	/* they're equal */
		btitem = (BTItem) PageGetItem(page, PageGetItemId(page, offset));
		itup = &(btitem->bti_itup);
		htup = heap_fetch (heapRel, SelfTimeQual, &(itup->t_tid), NULL);
		if ( htup != (HeapTuple) NULL )
		{	/* it is a duplicate */
		    elog(WARN, "Cannot insert a duplicate key into a unique index.");
		}
	    	/* get next offnum */
	    	if ( offset < maxoff )
	    	{
	    	    offset = OffsetNumberNext(offset);
	    	}
	    	else
	    	{	/* move right ? */
		    if ( P_RIGHTMOST (opaque) )
		    	break;
	    	    if ( !_bt_isequal (itupdesc, page, P_HIKEY, 
	    	    				natts, itup_scankey) )
			break;
		    /* 
		     * min key of the right page is the same,
		     * ooh - so many dead duplicates...
		     */
		    blkno = opaque->btpo_next;
		    if ( nbuf != InvalidBuffer )
		    	_bt_relbuf (rel, nbuf, BT_READ);
		    for (nbuf = InvalidBuffer; ; )
		    {
		    	nbuf = _bt_getbuf (rel, blkno, BT_READ);
		    	page = BufferGetPage (nbuf);
		    	maxoff = PageGetMaxOffsetNumber(page);
		    	opaque = (BTPageOpaque) PageGetSpecialPointer(page);
		    	offset = P_RIGHTMOST(opaque) ? P_HIKEY : P_FIRSTKEY;
		    	if ( ! PageIsEmpty (page) && offset <= maxoff )
		    	{	/* Found some key */
			    break;
			}
			else
			{	/* Empty or "pseudo"-empty page - get next */
			    blkno = opaque->btpo_next;
		    	    _bt_relbuf (rel, nbuf, BT_READ);
		    	    nbuf = InvalidBuffer;
		    	    if ( blkno == P_NONE )
		    	    	break;
			}
		    }
		    if ( nbuf == InvalidBuffer )
		    	break;
	    	}
	    }
	    if ( nbuf != InvalidBuffer )
		_bt_relbuf(rel, nbuf, BT_READ);
	}
    }
    
    /* do the insertion */
    res = _bt_insertonpg(rel, buf, stack, natts, itup_scankey,
			 btitem, (BTItem) NULL);
    
    /* be tidy */
    _bt_freestack(stack);
    _bt_freeskey(itup_scankey);
    
    return (res);
}

/*
 *  _bt_insertonpg() -- Insert a tuple on a particular page in the index.
 *
 *	This recursive procedure does the following things:
 *
 *	    +  if necessary, splits the target page.
 *	    +  finds the right place to insert the tuple (taking into
 *	       account any changes induced by a split).
 *	    +  inserts the tuple.
 *	    +  if the page was split, pops the parent stack, and finds the
 *	       right place to insert the new child pointer (by walking
 *	       right using information stored in the parent stack).
 *	    +  invoking itself with the appropriate tuple for the right
 *	       child page on the parent.
 *
 *	On entry, we must have the right buffer on which to do the
 *	insertion, and the buffer must be pinned and locked.  On return,
 *	we will have dropped both the pin and the write lock on the buffer.
 *
 *	The locking interactions in this code are critical.  You should
 *	grok Lehman and Yao's paper before making any changes.  In addition,
 *	you need to understand how we disambiguate duplicate keys in this
 *	implementation, in order to be able to find our location using
 *	L&Y "move right" operations.  Since we may insert duplicate user
 *	keys, and since these dups may propogate up the tree, we use the
 *	'afteritem' parameter to position ourselves correctly for the
 *	insertion on internal pages.
 */
static InsertIndexResult
_bt_insertonpg(Relation rel,
	       Buffer buf,
	       BTStack stack,
	       int keysz,
	       ScanKey scankey,
	       BTItem btitem,
	       BTItem afteritem)
{
    InsertIndexResult res;
    Page page;
    Buffer rbuf;
    Buffer pbuf;
    Page rpage;
    ScanKey newskey;
    BTItem ritem;
    BTPageOpaque rpageop;
    BlockNumber rbknum, itup_blkno;
    OffsetNumber itup_off;
    int itemsz;
    InsertIndexResult newres;
    BTItem new_item = (BTItem) NULL;
    BTItem lowLeftItem;
    OffsetNumber leftmost_offset;
    Page ppage;
    BTPageOpaque ppageop;
    BlockNumber bknum;
    
    page = BufferGetPage(buf);
    itemsz = IndexTupleDSize(btitem->bti_itup)
	+ (sizeof(BTItemData) - sizeof(IndexTupleData));

    itemsz = DOUBLEALIGN(itemsz);	/* be safe, PageAddItem will do this
					   but we need to be consistent */
    
    if (PageGetFreeSpace(page) < itemsz) {
	
	/* split the buffer into left and right halves */
	rbuf = _bt_split(rel, buf);
	
	/* which new page (left half or right half) gets the tuple? */
	if (_bt_goesonpg(rel, buf, keysz, scankey, afteritem)) {
	    /* left page */
	    itup_off = _bt_pgaddtup(rel, buf, keysz, scankey,
				    itemsz, btitem, afteritem);
	    itup_blkno = BufferGetBlockNumber(buf);
	} else {
	    /* right page */
	    itup_off = _bt_pgaddtup(rel, rbuf, keysz, scankey,
				    itemsz, btitem, afteritem);
	    itup_blkno = BufferGetBlockNumber(rbuf);
	}
	
	/*
	 *  By here,
	 *
	 *	+  our target page has been split;
	 *	+  the original tuple has been inserted;
	 *	+  we have write locks on both the old (left half) and new
	 *	   (right half) buffers, after the split; and
	 *	+  we have the key we want to insert into the parent.
	 *
	 *  Do the parent insertion.  We need to hold onto the locks for
	 *  the child pages until we locate the parent, but we can release
	 *  them before doing the actual insertion (see Lehman and Yao for
	 *  the reasoning).
	 */
	
	if (stack == (BTStack) NULL) {
	    
	    /* create a new root node and release the split buffers */
	    _bt_newroot(rel, buf, rbuf);
	    _bt_relbuf(rel, buf, BT_WRITE);
	    _bt_relbuf(rel, rbuf, BT_WRITE);
	    
	} else {

	    /* form a index tuple that points at the new right page */
	    rbknum = BufferGetBlockNumber(rbuf);
	    rpage = BufferGetPage(rbuf);
	    rpageop = (BTPageOpaque) PageGetSpecialPointer(rpage);
	    
	    /*
	     *  By convention, the first entry (0) on every
	     *  non-rightmost page is the high key for that page.  In
	     *  order to get the lowest key on the new right page, we
	     *  actually look at its second (1) entry.
	     */
	    
	    if (! P_RIGHTMOST(rpageop)) {
		ritem = (BTItem) PageGetItem(rpage,
					     PageGetItemId(rpage, P_FIRSTKEY));
	    } else {
		ritem = (BTItem) PageGetItem(rpage,
					     PageGetItemId(rpage, P_HIKEY));
	    }
	    
	    /* get a unique btitem for this key */
	    new_item = _bt_formitem(&(ritem->bti_itup));
	    
	    ItemPointerSet(&(new_item->bti_itup.t_tid), rbknum, P_HIKEY);
	    
	    /* find the parent buffer */
	    pbuf = _bt_getstackbuf(rel, stack, BT_WRITE);
	    
	    /*
	     *  If the key of new_item is < than the key of the item
	     *  in the parent page pointing to the left page
	     *  (stack->bts_btitem), we have to update the latter key;
	     *  otherwise the keys on the parent page wouldn't be
	     *  monotonically increasing after we inserted the new
	     *  pointer to the right page (new_item). This only
	     *  happens if our left page is the leftmost page and a
	     *  new minimum key had been inserted before, which is not
	     *  reflected in the parent page but didn't matter so
	     *  far. If there are duplicate keys and this new minimum
	     *  key spills over to our new right page, we get an
	     *  inconsistency if we don't update the left key in the
	     *  parent page.
	     */
	    
	    if (_bt_itemcmp(rel, keysz, stack->bts_btitem, new_item,
	                    BTGreaterStrategyNumber)) {
		ppageop = (BTPageOpaque) PageGetSpecialPointer(page);
		Assert (P_LEFTMOST(ppageop));
		lowLeftItem =
		    (BTItem) PageGetItem(page,
					 PageGetItemId(page, P_FIRSTKEY));
		
		/* this method does not work--_bt_updateitem tries to     */
		/* overwrite an entry with another entry that might be    */
		/* bigger.  if lowLeftItem is bigger, it corrupts the     */
		/* parent page.  instead, we have to delete the original  */
		/* leftmost item from the parent, and insert the new one  */
		/* with a regular _bt_insertonpg (it could cause a split  */
		/* because it's bigger than what was there before).       */
                /*                                  --djm 8/21/96         */

		/* 
		 * but it works for items with the same size and so why don't
		 * use it for them ? - vadim 12/05/96
		 */
	    	if ( DOUBLEALIGN (IndexTupleDSize (lowLeftItem->bti_itup)) ==
       			DOUBLEALIGN (IndexTupleDSize (stack->bts_btitem->bti_itup)) ) 
       		{
		    _bt_updateitem(rel, keysz, pbuf, 
		    		stack->bts_btitem, lowLeftItem);
		    _bt_relbuf(rel, buf, BT_WRITE);
		    _bt_relbuf(rel, rbuf, BT_WRITE);
		}
		else
		{
		    /* get the parent page */
		    ppage = BufferGetPage(pbuf);
		    ppageop = (BTPageOpaque) PageGetSpecialPointer(ppage);

		    /* 
		     * figure out which key is leftmost (if the parent page
		     * is rightmost, too, it must be the root)
		     */
		    if(P_RIGHTMOST(ppageop)) {
		    	leftmost_offset = P_HIKEY;
		    } else {
		    	leftmost_offset = P_FIRSTKEY;
	    	    }
       		    PageIndexTupleDelete(ppage, leftmost_offset);
		
		   /* 
		    * don't write anything out yet--we still have the write
		    * lock, and now we call another _bt_insertonpg to
		    * insert the correct leftmost key
		    */

		    /* 
		     * make a new leftmost item, using the tuple data from
		     * lowLeftItem.  point it to the left child.
		     * update it on the stack at the same time.
		     */
		    bknum = BufferGetBlockNumber(buf);
		    pfree(stack->bts_btitem);
		    stack->bts_btitem = _bt_formitem(&(lowLeftItem->bti_itup));
		    ItemPointerSet(&(stack->bts_btitem->bti_itup.t_tid), 
			       bknum, P_HIKEY);
		
		    /* unlock the children before doing this */
		    _bt_relbuf(rel, buf, BT_WRITE);
		    _bt_relbuf(rel, rbuf, BT_WRITE);
		
		    /* 
		     * a regular _bt_binsrch should find the right place to
		     * put the new entry, since it should be lower than any
		     * other key on the page, therefore set afteritem to NULL
		     */
		    newskey = _bt_mkscankey(rel, &(stack->bts_btitem->bti_itup));
		    newres = _bt_insertonpg(rel, pbuf, stack->bts_parent,
					keysz, newskey, stack->bts_btitem,
					NULL);

		    pfree(newres);
		    pfree(newskey);
	    
		    /* 
		     * we have now lost our lock on the parent buffer, and
		     * need to get it back.
		     */
		    pbuf = _bt_getstackbuf(rel, stack, BT_WRITE);
		}
	    }
	    else
	    {
	    	_bt_relbuf(rel, buf, BT_WRITE);
	    	_bt_relbuf(rel, rbuf, BT_WRITE);
	    }
	    
	    newskey = _bt_mkscankey(rel, &(new_item->bti_itup));
	    newres = _bt_insertonpg(rel, pbuf, stack->bts_parent,
				    keysz, newskey, new_item,
				    stack->bts_btitem);
	    
	    /* be tidy */
	    pfree(newres);
	    pfree(newskey);
	    pfree(new_item);
	}
    } else {
	itup_off = _bt_pgaddtup(rel, buf, keysz, scankey,
				itemsz, btitem, afteritem);
	itup_blkno = BufferGetBlockNumber(buf);
	
	_bt_relbuf(rel, buf, BT_WRITE);
    }
    
    /* by here, the new tuple is inserted */
    res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
    ItemPointerSet(&(res->pointerData), itup_blkno, itup_off);
    
    return (res);
}

/*
 *  _bt_split() -- split a page in the btree.
 *
 *	On entry, buf is the page to split, and is write-locked and pinned.
 *	Returns the new right sibling of buf, pinned and write-locked.  The
 *	pin and lock on buf are maintained.
 */
static Buffer
_bt_split(Relation rel, Buffer buf)
{
    Buffer rbuf;
    Page origpage;
    Page leftpage, rightpage;
    BTPageOpaque ropaque, lopaque, oopaque;
    Buffer sbuf;
    Page spage;
    BTPageOpaque sopaque;
    Size itemsz;
    ItemId itemid;
    BTItem item;
    OffsetNumber leftoff, rightoff;
    OffsetNumber start;
    OffsetNumber maxoff;
    OffsetNumber firstright;
    OffsetNumber i;
    Size llimit;
    
    rbuf = _bt_getbuf(rel, P_NEW, BT_WRITE);
    origpage = BufferGetPage(buf);
    leftpage = PageGetTempPage(origpage, sizeof(BTPageOpaqueData));
    rightpage = BufferGetPage(rbuf);
    
    _bt_pageinit(rightpage, BufferGetPageSize(rbuf));
    _bt_pageinit(leftpage, BufferGetPageSize(buf));
    
    /* init btree private data */
    oopaque = (BTPageOpaque) PageGetSpecialPointer(origpage);
    lopaque = (BTPageOpaque) PageGetSpecialPointer(leftpage);
    ropaque = (BTPageOpaque) PageGetSpecialPointer(rightpage);
    
    /* if we're splitting this page, it won't be the root when we're done */
    oopaque->btpo_flags &= ~BTP_ROOT;
    lopaque->btpo_flags = ropaque->btpo_flags = oopaque->btpo_flags;
    lopaque->btpo_prev = oopaque->btpo_prev;
    ropaque->btpo_prev = BufferGetBlockNumber(buf);
    lopaque->btpo_next = BufferGetBlockNumber(rbuf);
    ropaque->btpo_next = oopaque->btpo_next;
    
    /*
     *  If the page we're splitting is not the rightmost page at its
     *  level in the tree, then the first (0) entry on the page is the
     *  high key for the page.  We need to copy that to the right
     *  half.  Otherwise (meaning the rightmost page case), we should
     *  treat the line pointers beginning at zero as user data.
     *
     *  We leave a blank space at the start of the line table for the
     *  left page.  We'll come back later and fill it in with the high
     *  key item we get from the right key.
     */
    
    leftoff = P_FIRSTKEY;
    ropaque->btpo_next = oopaque->btpo_next;
    if (! P_RIGHTMOST(oopaque)) {
	/* splitting a non-rightmost page, start at the first data item */
	start = P_FIRSTKEY;

	/* copy the original high key to the new page */
	itemid = PageGetItemId(origpage, P_HIKEY);
	itemsz = ItemIdGetLength(itemid);
	item = (BTItem) PageGetItem(origpage, itemid);
	(void) PageAddItem(rightpage, (Item) item, itemsz, P_HIKEY, LP_USED);
	rightoff = P_FIRSTKEY;
    } else {
	/* splitting a rightmost page, "high key" is the first data item */
	start = P_HIKEY;

	/* the new rightmost page will not have a high key */
	rightoff = P_HIKEY;
    }
    maxoff = PageGetMaxOffsetNumber(origpage);
    llimit = PageGetFreeSpace(leftpage) / 2;
    firstright = _bt_findsplitloc(rel, origpage, start, maxoff, llimit);
    
    for (i = start; i <= maxoff; i = OffsetNumberNext(i)) {
	itemid = PageGetItemId(origpage, i);
	itemsz = ItemIdGetLength(itemid);
	item = (BTItem) PageGetItem(origpage, itemid);
	
	/* decide which page to put it on */
	if (i < firstright) {
	    (void) PageAddItem(leftpage, (Item) item, itemsz, leftoff,
			       LP_USED);
	    leftoff = OffsetNumberNext(leftoff);
	} else {
	    (void) PageAddItem(rightpage, (Item) item, itemsz, rightoff,
			       LP_USED);
	    rightoff = OffsetNumberNext(rightoff);
	}
    }
    
    /*
     *  Okay, page has been split, high key on right page is correct.  Now
     *  set the high key on the left page to be the min key on the right
     *  page.
     */
    
    if (P_RIGHTMOST(ropaque)) {
	itemid = PageGetItemId(rightpage, P_HIKEY);
    } else {
	itemid = PageGetItemId(rightpage, P_FIRSTKEY);
    }
    itemsz = ItemIdGetLength(itemid);
    item = (BTItem) PageGetItem(rightpage, itemid);
    
    /*
     *  We left a hole for the high key on the left page; fill it.  The
     *  modal crap is to tell the page manager to put the new item on the
     *  page and not screw around with anything else.  Whoever designed
     *  this interface has presumably crawled back into the dung heap they
     *  came from.  No one here will admit to it.
     */
    
    PageManagerModeSet(OverwritePageManagerMode);
    (void) PageAddItem(leftpage, (Item) item, itemsz, P_HIKEY, LP_USED);
    PageManagerModeSet(ShufflePageManagerMode);
    
    /*
     *  By here, the original data page has been split into two new halves,
     *  and these are correct.  The algorithm requires that the left page
     *  never move during a split, so we copy the new left page back on top
     *  of the original.  Note that this is not a waste of time, since we
     *  also require (in the page management code) that the center of a
     *  page always be clean, and the most efficient way to guarantee this
     *  is just to compact the data by reinserting it into a new left page.
     */
    
    PageRestoreTempPage(leftpage, origpage);
    
    /* write these guys out */
    _bt_wrtnorelbuf(rel, rbuf);
    _bt_wrtnorelbuf(rel, buf);
    
    /*
     *  Finally, we need to grab the right sibling (if any) and fix the
     *  prev pointer there.  We are guaranteed that this is deadlock-free
     *  since no other writer will be moving holding a lock on that page
     *  and trying to move left, and all readers release locks on a page
     *  before trying to fetch its neighbors.
     */
    
    if (! P_RIGHTMOST(ropaque)) {
	sbuf = _bt_getbuf(rel, ropaque->btpo_next, BT_WRITE);
	spage = BufferGetPage(sbuf);
	sopaque = (BTPageOpaque) PageGetSpecialPointer(spage);
	sopaque->btpo_prev = BufferGetBlockNumber(rbuf);
	
	/* write and release the old right sibling */
	_bt_wrtbuf(rel, sbuf);
    }
    
    /* split's done */
    return (rbuf);
}

/*
 *  _bt_findsplitloc() -- find a safe place to split a page.
 *
 *	In order to guarantee the proper handling of searches for duplicate
 *	keys, the first duplicate in the chain must either be the first
 *	item on the page after the split, or the entire chain must be on
 *	one of the two pages.  That is,
 *		[1 2 2 2 3 4 5]
 *	must become
 *		[1] [2 2 2 3 4 5]
 *	or
 *		[1 2 2 2] [3 4 5]
 *	but not
 *		[1 2 2] [2 3 4 5].
 *	However,
 *		[2 2 2 2 2 3 4]
 *	may be split as
 *		[2 2 2 2] [2 3 4].
 */
static OffsetNumber
_bt_findsplitloc(Relation rel,
		 Page page,
		 OffsetNumber start,
		 OffsetNumber maxoff,
		 Size llimit)
{
    OffsetNumber i;
    OffsetNumber saferight;
    ItemId nxtitemid, safeitemid;
    BTItem safeitem, nxtitem;
    Size nbytes;
    int natts;
    
    natts = rel->rd_rel->relnatts;
    saferight = start;
    safeitemid = PageGetItemId(page, saferight);
    nbytes = ItemIdGetLength(safeitemid) + sizeof(ItemIdData);
    safeitem = (BTItem) PageGetItem(page, safeitemid);
    
    i = OffsetNumberNext(start);
    
    while (nbytes < llimit) {
	
	/* check the next item on the page */
	nxtitemid = PageGetItemId(page, i);
	nbytes += (ItemIdGetLength(nxtitemid) + sizeof(ItemIdData));
	nxtitem = (BTItem) PageGetItem(page, nxtitemid);

	/* 
	 * Test against last known safe item:
	 * if the tuple we're looking at isn't equal to the last safe 
	 * one we saw, then it's our new safe tuple.
	 */
	if ( !_bt_itemcmp (rel, natts, 
			safeitem, nxtitem, BTEqualStrategyNumber) )
	{
	    safeitem = nxtitem;
	    saferight = i;
	}
	i = OffsetNumberNext(i);
    }
    
    /*
     *  If the chain of dups starts at the beginning of the page and extends
     *  past the halfway mark, we can split it in the middle.
     */
    
    if (saferight == start)
	saferight = i;
    
    return (saferight);
}

/*
 *  _bt_newroot() -- Create a new root page for the index.
 *
 *	We've just split the old root page and need to create a new one.
 *	In order to do this, we add a new root page to the file, then lock
 *	the metadata page and update it.  This is guaranteed to be deadlock-
 *	free, because all readers release their locks on the metadata page
 *	before trying to lock the root, and all writers lock the root before
 *	trying to lock the metadata page.  We have a write lock on the old
 *	root page, so we have not introduced any cycles into the waits-for
 *	graph.
 *
 *	On entry, lbuf (the old root) and rbuf (its new peer) are write-
 *	locked.  We don't drop the locks in this routine; that's done by
 *	the caller.  On exit, a new root page exists with entries for the
 *	two new children.  The new root page is neither pinned nor locked.
 */
static void
_bt_newroot(Relation rel, Buffer lbuf, Buffer rbuf)
{
    Buffer rootbuf;
    Page lpage, rpage, rootpage;
    BlockNumber lbkno, rbkno;
    BlockNumber rootbknum;
    BTPageOpaque rootopaque;
    ItemId itemid;
    BTItem item;
    Size itemsz;
    BTItem new_item;
    
    /* get a new root page */
    rootbuf = _bt_getbuf(rel, P_NEW, BT_WRITE);
    rootpage = BufferGetPage(rootbuf);
    _bt_pageinit(rootpage, BufferGetPageSize(rootbuf));
    
    /* set btree special data */
    rootopaque = (BTPageOpaque) PageGetSpecialPointer(rootpage);
    rootopaque->btpo_prev = rootopaque->btpo_next = P_NONE;
    rootopaque->btpo_flags |= BTP_ROOT;
    
    /*
     *  Insert the internal tuple pointers.
     */
    
    lbkno = BufferGetBlockNumber(lbuf);
    rbkno = BufferGetBlockNumber(rbuf);
    lpage = BufferGetPage(lbuf);
    rpage = BufferGetPage(rbuf);

    /*
     * step over the high key on the left page while building the 
     * left page pointer.
     */
    itemid = PageGetItemId(lpage, P_FIRSTKEY);
    itemsz = ItemIdGetLength(itemid);
    item = (BTItem) PageGetItem(lpage, itemid);
    new_item = _bt_formitem(&(item->bti_itup));
    ItemPointerSet(&(new_item->bti_itup.t_tid), lbkno, P_FIRSTKEY);
    
    /*
     * insert the left page pointer into the new root page.  the root
     * page is the rightmost page on its level so the "high key" item
     * is the first data item.
     */
    (void) PageAddItem(rootpage, (Item) new_item, itemsz, P_HIKEY, LP_USED);
    pfree(new_item);
    
    /*
     * the right page is the rightmost page on the second level, so 
     * the "high key" item is the first data item on that page as well.
     */
    itemid = PageGetItemId(rpage, P_HIKEY);
    itemsz = ItemIdGetLength(itemid);
    item = (BTItem) PageGetItem(rpage, itemid);
    new_item = _bt_formitem(&(item->bti_itup));
    ItemPointerSet(&(new_item->bti_itup.t_tid), rbkno, P_HIKEY);
    
    /*
     * insert the right page pointer into the new root page.
     */
    (void) PageAddItem(rootpage, (Item) new_item, itemsz, P_FIRSTKEY, LP_USED);
    pfree(new_item);
    
    /* write and let go of the root buffer */
    rootbknum = BufferGetBlockNumber(rootbuf);
    _bt_wrtbuf(rel, rootbuf);
    
    /* update metadata page with new root block number */
    _bt_metaproot(rel, rootbknum, 0);
}

/*
 *  _bt_pgaddtup() -- add a tuple to a particular page in the index.
 *
 *	This routine adds the tuple to the page as requested, and keeps the
 *	write lock and reference associated with the page's buffer.  It is
 *	an error to call pgaddtup() without a write lock and reference.  If
 *	afteritem is non-null, it's the item that we expect our new item
 *	to follow.  Otherwise, we do a binary search for the correct place
 *	and insert the new item there.
 */
static OffsetNumber
_bt_pgaddtup(Relation rel,
	     Buffer buf,
	     int keysz,
	     ScanKey itup_scankey,
	     Size itemsize,
	     BTItem btitem,
	     BTItem afteritem)
{
    OffsetNumber itup_off;
    OffsetNumber first;
    Page page;
    BTPageOpaque opaque;
    BTItem chkitem;
    
    page = BufferGetPage(buf);
    opaque = (BTPageOpaque) PageGetSpecialPointer(page);
    first = P_RIGHTMOST(opaque) ? P_HIKEY : P_FIRSTKEY;
    
    if (afteritem == (BTItem) NULL) {
	itup_off = _bt_binsrch(rel, buf, keysz, itup_scankey, BT_INSERTION);
    } else {
	itup_off = first;
	
	do {
	    chkitem =
		(BTItem) PageGetItem(page, PageGetItemId(page, itup_off));
	    itup_off = OffsetNumberNext(itup_off);
	} while ( ! BTItemSame (chkitem, afteritem) );
    }

    (void) PageAddItem(page, (Item) btitem, itemsize, itup_off, LP_USED);
    
    /* write the buffer, but hold our lock */
    _bt_wrtnorelbuf(rel, buf);
    
    return (itup_off);
}

/*
 *  _bt_goesonpg() -- Does a new tuple belong on this page?
 *
 *	This is part of the complexity introduced by allowing duplicate
 *	keys into the index.  The tuple belongs on this page if:
 *
 *		+ there is no page to the right of this one; or
 *		+ it is less than the high key on the page; or
 *		+ the item it is to follow ("afteritem") appears on this
 *		  page.
 */
static bool
_bt_goesonpg(Relation rel,
	     Buffer buf,
	     Size keysz,
	     ScanKey scankey,
	     BTItem afteritem)
{
    Page page;
    ItemId hikey;
    BTPageOpaque opaque;
    BTItem chkitem;
    OffsetNumber offnum, maxoff;
    bool found;
    
    page = BufferGetPage(buf);
    
    /* no right neighbor? */
    opaque = (BTPageOpaque) PageGetSpecialPointer(page);
    if (P_RIGHTMOST(opaque))
	return (true);
    
    /*
     *  this is a non-rightmost page, so it must have a high key item.
     *
     *  If the scan key is < the high key (the min key on the next page),
     *  then it for sure belongs here.
     */
    hikey = PageGetItemId(page, P_HIKEY);
    if (_bt_skeycmp(rel, keysz, scankey, page, hikey, BTLessStrategyNumber))
	return (true);
    
    /*
     *  If the scan key is > the high key, then it for sure doesn't belong
     *  here.
     */
    
    if (_bt_skeycmp(rel, keysz, scankey, page, hikey, BTGreaterStrategyNumber))
	return (false);
    
    /*
     *  If we have no adjacency information, and the item is equal to the
     *  high key on the page (by here it is), then the item does not belong
     *  on this page.
     */
    
    if (afteritem == (BTItem) NULL)
	return (false);
    
    /* damn, have to work for it.  i hate that. */
    maxoff = PageGetMaxOffsetNumber(page);
    
    /*
     *  Search the entire page for the afteroid.  We need to do this, rather
     *  than doing a binary search and starting from there, because if the
     *  key we're searching for is the leftmost key in the tree at this
     *  level, then a binary search will do the wrong thing.  Splits are
     *  pretty infrequent, so the cost isn't as bad as it could be.
     */
    
    found = false;
    for (offnum = P_FIRSTKEY;
	 offnum <= maxoff;
	 offnum = OffsetNumberNext(offnum)) {
	chkitem = (BTItem) PageGetItem(page, PageGetItemId(page, offnum));

	if ( BTItemSame (chkitem, afteritem) ) {
	    found = true;
	    break;
	}
    }
    
    return (found);
}

/*
 *	_bt_itemcmp() -- compare item1 to item2 using a requested
 *		         strategy (<, <=, =, >=, >)
 *
 */
bool
_bt_itemcmp(Relation rel,
	    Size keysz,
	    BTItem item1,
	    BTItem item2,
	    StrategyNumber strat)
{
    TupleDesc tupDes;
    IndexTuple indexTuple1, indexTuple2;
    Datum attrDatum1, attrDatum2;
    int i;
    bool isFirstNull, isSecondNull;
    bool compare;
    bool useEqual = false;
    
    if ( strat == BTLessEqualStrategyNumber )
    {
	useEqual = true;
	strat = BTLessStrategyNumber;
    }
    else if ( strat == BTGreaterEqualStrategyNumber )
    {
	useEqual = true;
	strat = BTGreaterStrategyNumber;
    }
    
    tupDes = RelationGetTupleDescriptor(rel);
    indexTuple1 = &(item1->bti_itup);
    indexTuple2 = &(item2->bti_itup);
    
    for (i = 1; i <= keysz; i++) {
	attrDatum1 = index_getattr(indexTuple1, i, tupDes, &isFirstNull);
	attrDatum2 = index_getattr(indexTuple2, i, tupDes, &isSecondNull);
	
	/* see comments about NULLs handling in btbuild */
	if ( isFirstNull )	/* attr in item1 is NULL */
	{
    	    if ( isSecondNull )	/* attr in item2 is NULL too */
    	    	compare = ( strat == BTEqualStrategyNumber ) ? true : false;
    	    else
    	    	compare = ( strat == BTGreaterStrategyNumber ) ? true : false;
    	}
    	else if ( isSecondNull )	/* attr in item1 is NOT_NULL and */
    	{				/* and attr in item2 is NULL */
    	    compare = ( strat == BTLessStrategyNumber ) ? true : false;
    	}
    	else
    	{
	    compare = _bt_invokestrat(rel, i, strat, attrDatum1, attrDatum2);
	}
	
	if ( compare )	/* true for one of ">, <, =" */
	{
	    if ( strat != BTEqualStrategyNumber )
	    	return (true);
	}
	else		/* false for one of ">, <, =" */
	{
	    if ( strat == BTEqualStrategyNumber )
		return (false);
	    /*
	     * if original strat was "<=, >=" OR
	     * "<, >" but some attribute(s) left
	     * - need to test for Equality
	     */
	    if ( useEqual || i < keysz )
	    {
	    	if ( isFirstNull || isSecondNull )
	    	    compare = ( isFirstNull && isSecondNull ) ? true : false;
	    	else
		    compare = _bt_invokestrat(rel, i, BTEqualStrategyNumber, 
						attrDatum1, attrDatum2);
		if ( compare )	/* item1' and item2' attributes are equal */
		    continue;	/* - try to compare next attributes */
	    }
	    return (false);
	}
    }
    return (true);
}

/*
 *	_bt_updateitem() -- updates the key of the item identified by the
 *			    oid with the key of newItem (done in place if
 *			    possible)
 *
 */
static void
_bt_updateitem(Relation rel,
	       Size keysz,
	       Buffer buf,
	       BTItem oldItem,
	       BTItem newItem)
{
    Page page;
    OffsetNumber maxoff;
    OffsetNumber i;
    ItemPointerData itemPtrData;
    BTItem item;
    IndexTuple oldIndexTuple, newIndexTuple;
    int first;
    
    page = BufferGetPage(buf);
    maxoff = PageGetMaxOffsetNumber(page);
    
    /* locate item on the page */
    first = P_RIGHTMOST((BTPageOpaque) PageGetSpecialPointer(page))
        ? P_HIKEY : P_FIRSTKEY;
    i = first;
    do {
	item = (BTItem) PageGetItem(page, PageGetItemId(page, i));
	i = OffsetNumberNext(i);
    } while (i <= maxoff && ! BTItemSame (item, oldItem));
    
    /* this should never happen (in theory) */
    if ( ! BTItemSame (item, oldItem) ) {
	elog(FATAL, "_bt_getstackbuf was lying!!");
    }
    
    /*
     * It's  defined by caller (_bt_insertonpg)
     */
    /*
    if(IndexTupleDSize(newItem->bti_itup) >
       IndexTupleDSize(item->bti_itup)) {
	elog(NOTICE, "trying to overwrite a smaller value with a bigger one in _bt_updateitem");
	elog(WARN, "this is not good.");
    }
     */

    oldIndexTuple = &(item->bti_itup);
    newIndexTuple = &(newItem->bti_itup);

	/* keep the original item pointer */
    ItemPointerCopy(&(oldIndexTuple->t_tid), &itemPtrData);
    CopyIndexTuple(newIndexTuple, &oldIndexTuple);
    ItemPointerCopy(&itemPtrData, &(oldIndexTuple->t_tid));
    
}

/*
 * _bt_isequal - used in _bt_doinsert in check for duplicates.
 *
 * Rule is simple: NOT_NULL not equal NULL, NULL not_equal NULL too.
 */
static bool
_bt_isequal (TupleDesc itupdesc, Page page, OffsetNumber offnum,
					int keysz, ScanKey scankey)
{
    Datum datum;
    BTItem btitem;
    IndexTuple itup;
    ScanKey entry;
    AttrNumber attno;
    long result;
    int i;
    bool null;
    
    btitem = (BTItem) PageGetItem(page, PageGetItemId(page, offnum));
    itup = &(btitem->bti_itup);
    
    for (i = 1; i <= keysz; i++)
    {
	entry = &scankey[i - 1];
	attno = entry->sk_attno;
	Assert (attno == i);
	datum = index_getattr(itup, attno, itupdesc, &null);

	/* NULLs are not equal */
	if ( entry->sk_flags & SK_ISNULL || null )
	    return (false);
	
	result = (long) FMGR_PTR2(entry->sk_func, entry->sk_procedure,
					entry->sk_argument, datum);
	if (result != 0)
	    return (false);
    }
    
    /* by here, the keys are equal */
    return (true);
}