postgresql/src/backend/access/gist/gistget.c

/*-------------------------------------------------------------------------
 *
 * gistget.c--
 *    fetch tuples from a GiST scan.
 *
 *
 *
 * IDENTIFICATION
 *    /usr/local/devel/pglite/cvs/src/backend/access/gisr/gistget.c,v 1.9 1995/08/01 20:16:02 jolly Exp
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "catalog/pg_attribute.h"
#include "access/attnum.h"
#include "nodes/pg_list.h"
#include "access/tupdesc.h"
#include "storage/fd.h"
#include "catalog/pg_am.h"  
#include "catalog/pg_class.h"
#include "nodes/nodes.h"
#include "rewrite/prs2lock.h"
#include "access/skey.h"
#include "access/strat.h"
#include "utils/rel.h"
 
#include "storage/block.h"
#include "storage/off.h"
#include "storage/itemptr.h"
#include <time.h>
#include "utils/nabstime.h"
#include "access/htup.h"
 
#include "utils/tqual.h"
#include "storage/buf.h"
#include "access/relscan.h"
 
#include "storage/itemid.h"
#include "storage/item.h" 
#include "storage/bufpage.h"

#include "access/sdir.h"
 
#include "access/itup.h"
 
#include "access/gist.h"

static OffsetNumber gistfindnext(IndexScanDesc s, Page p, OffsetNumber n, 
				 ScanDirection dir);
static RetrieveIndexResult gistscancache(IndexScanDesc s, ScanDirection dir);
static RetrieveIndexResult gistfirst(IndexScanDesc s, ScanDirection dir);
static RetrieveIndexResult gistnext(IndexScanDesc s, ScanDirection dir);
static ItemPointer gistheapptr(Relation r, ItemPointer itemp);


RetrieveIndexResult
gistgettuple(IndexScanDesc s, ScanDirection dir)
{
    RetrieveIndexResult res;
    
    /* if we have it cached in the scan desc, just return the value */
    if ((res = gistscancache(s, dir)) != (RetrieveIndexResult) NULL)
	return (res);
    
    /* not cached, so we'll have to do some work */
    if (ItemPointerIsValid(&(s->currentItemData))) {
	res = gistnext(s, dir);
    } else {
	res = gistfirst(s, dir);
    }
    return (res);
}

static RetrieveIndexResult
gistfirst(IndexScanDesc s, ScanDirection dir)
{
    Buffer b;
    Page p;
    OffsetNumber n;
    OffsetNumber maxoff;
    RetrieveIndexResult res;
    GISTPageOpaque po;
    GISTScanOpaque so;
    GISTSTACK *stk;
    BlockNumber blk;
    IndexTuple it;
    ItemPointer ip;

    b = ReadBuffer(s->relation, GISTP_ROOT);
    p = BufferGetPage(b);
    po = (GISTPageOpaque) PageGetSpecialPointer(p);
    so = (GISTScanOpaque) s->opaque;

    for (;;) {
	maxoff = PageGetMaxOffsetNumber(p);
	if (ScanDirectionIsBackward(dir))
	    n = gistfindnext(s, p, maxoff, dir);
	else
	    n = gistfindnext(s, p, FirstOffsetNumber, dir);
	
	while (n < FirstOffsetNumber || n > maxoff) {
	    
	    ReleaseBuffer(b);
	    if (so->s_stack == (GISTSTACK *) NULL)
		return ((RetrieveIndexResult) NULL);
	    
	    stk = so->s_stack;
	    b = ReadBuffer(s->relation, stk->gs_blk);
	    p = BufferGetPage(b);
	    po = (GISTPageOpaque) PageGetSpecialPointer(p);
	    maxoff = PageGetMaxOffsetNumber(p);
	    
	    if (ScanDirectionIsBackward(dir)) {
		n = OffsetNumberPrev(stk->gs_child);
	    } else {
		n = OffsetNumberNext(stk->gs_child);
	    }
	    so->s_stack = stk->gs_parent;
	    pfree(stk);
	    
	    n = gistfindnext(s, p, n, dir);
	}
	if (po->flags & F_LEAF) {
	    ItemPointerSet(&(s->currentItemData), BufferGetBlockNumber(b), n);
	    
	    it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
	    ip = (ItemPointer) palloc(sizeof(ItemPointerData));
	    memmove((char *) ip, (char *) &(it->t_tid),
		    sizeof(ItemPointerData));
	    ReleaseBuffer(b);
	    
	    res = FormRetrieveIndexResult(&(s->currentItemData), ip);
	    
	    return (res);
	} else {
	    stk = (GISTSTACK *) palloc(sizeof(GISTSTACK));
	    stk->gs_child = n;
	    stk->gs_blk = BufferGetBlockNumber(b);
	    stk->gs_parent = so->s_stack;
	    so->s_stack = stk;
	    
	    it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
	    blk = ItemPointerGetBlockNumber(&(it->t_tid));
	    
	    ReleaseBuffer(b);
	    b = ReadBuffer(s->relation, blk);
	    p = BufferGetPage(b);
	    po = (GISTPageOpaque) PageGetSpecialPointer(p);
	}
    }
}

static RetrieveIndexResult
gistnext(IndexScanDesc s, ScanDirection dir)
{
    Buffer b;
    Page p;
    OffsetNumber n;
    OffsetNumber maxoff;
    RetrieveIndexResult res;
    GISTPageOpaque po;
    GISTScanOpaque so;
    GISTSTACK *stk;
    BlockNumber blk;
    IndexTuple it;
    ItemPointer ip;
    
    blk = ItemPointerGetBlockNumber(&(s->currentItemData));
    n = ItemPointerGetOffsetNumber(&(s->currentItemData));
    
    if (ScanDirectionIsForward(dir)) {
	n = OffsetNumberNext(n);
    } else {
	n = OffsetNumberPrev(n);
    }

    b = ReadBuffer(s->relation, blk);
    p = BufferGetPage(b);
    po = (GISTPageOpaque) PageGetSpecialPointer(p);
    so = (GISTScanOpaque) s->opaque;
    
    for (;;) {
	maxoff = PageGetMaxOffsetNumber(p);
	n = gistfindnext(s, p, n, dir);
	
	while (n < FirstOffsetNumber || n > maxoff) {
	    
	    ReleaseBuffer(b);
	    if (so->s_stack == (GISTSTACK *) NULL)
		return ((RetrieveIndexResult) NULL);
	    
	    stk = so->s_stack;
	    b = ReadBuffer(s->relation, stk->gs_blk);
	    p = BufferGetPage(b);
	    maxoff = PageGetMaxOffsetNumber(p);
	    po = (GISTPageOpaque) PageGetSpecialPointer(p);
	    
	    if (ScanDirectionIsBackward(dir)) {
		n = OffsetNumberPrev(stk->gs_child);
	    } else {
		n = OffsetNumberNext(stk->gs_child);
	    }
	    so->s_stack = stk->gs_parent;
	    pfree(stk);
	    
	    n = gistfindnext(s, p, n, dir);
	}
	if (po->flags & F_LEAF) {
	    ItemPointerSet(&(s->currentItemData), BufferGetBlockNumber(b), n);
	    
	    it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
	    ip = (ItemPointer) palloc(sizeof(ItemPointerData));
	    memmove((char *) ip, (char *) &(it->t_tid),
		    sizeof(ItemPointerData));
	    ReleaseBuffer(b);
	    
	    res = FormRetrieveIndexResult(&(s->currentItemData), ip);
	    
	    return (res);
	} else {
	    stk = (GISTSTACK *) palloc(sizeof(GISTSTACK));
	    stk->gs_child = n;
	    stk->gs_blk = BufferGetBlockNumber(b);
	    stk->gs_parent = so->s_stack;
	    so->s_stack = stk;
	    
	    it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
	    blk = ItemPointerGetBlockNumber(&(it->t_tid));
	    
	    ReleaseBuffer(b);
	    b = ReadBuffer(s->relation, blk);
	    p = BufferGetPage(b);
	    po = (GISTPageOpaque) PageGetSpecialPointer(p);
	    
	    if (ScanDirectionIsBackward(dir)) {
		n = PageGetMaxOffsetNumber(p);
	    } else {
		n = FirstOffsetNumber;
	    }
	}
    }
}

/* Similar to index_keytest, but decompresses the key in the IndexTuple */
bool
gistindex_keytest(IndexTuple tuple,
		  TupleDesc tupdesc,
		  int scanKeySize,
		  ScanKey key,
		  GISTSTATE *giststate,
		  Relation r,
		  Page p,
		  OffsetNumber offset)
{
    bool	    isNull;
    Datum	    datum;
    int		    test;
    GISTENTRY       de;

    IncrIndexProcessed();
    

    while (scanKeySize > 0) {
	datum = index_getattr(tuple,
			      1,
			      tupdesc,
			      &isNull);
	gistdentryinit(giststate, &de, (char *)datum, r, p, offset,
		       IndexTupleSize(tuple) - sizeof(IndexTupleData),
		       FALSE);
	
	if (isNull) {
	    /* XXX eventually should check if SK_ISNULL */
	    return (false);
	}
	
	if (key[0].sk_flags & SK_COMMUTE) {
	    test = (int) (*(key[0].sk_func))
		(DatumGetPointer(key[0].sk_argument),
		 &de, key[0].sk_procedure);
	} else {
	    test = (int) (*(key[0].sk_func))
		(&de,
		 DatumGetPointer(key[0].sk_argument),
		 key[0].sk_procedure);
	}
	
	if (!test == !(key[0].sk_flags & SK_NEGATE)) {
	    return (false);
	}
	
	scanKeySize -= 1;
	key++;
    }
    
    return (true);
}


static OffsetNumber
gistfindnext(IndexScanDesc s, Page p, OffsetNumber n, ScanDirection dir)
{
    OffsetNumber maxoff;
    char *it;
    GISTPageOpaque po;
    GISTScanOpaque so;
    GISTENTRY de;
    GISTSTATE *giststate;

    maxoff = PageGetMaxOffsetNumber(p);
    po = (GISTPageOpaque) PageGetSpecialPointer(p);
    so = (GISTScanOpaque) s->opaque;
    giststate = so->giststate;

    /*
     *  If we modified the index during the scan, we may have a pointer to
     *  a ghost tuple, before the scan.  If this is the case, back up one.
     */
    
    if (so->s_flags & GS_CURBEFORE) {
	so->s_flags &= ~GS_CURBEFORE;
	n = OffsetNumberPrev(n);
    }
    
    while (n >= FirstOffsetNumber && n <= maxoff) {
	it = (char *) PageGetItem(p, PageGetItemId(p, n));
	if (gistindex_keytest((IndexTuple) it,
			      RelationGetTupleDescriptor(s->relation),
			      s->numberOfKeys, s->keyData, giststate,
			      s->relation, p, n))
	  break;
	
	if (ScanDirectionIsBackward(dir)) {
	    n = OffsetNumberPrev(n);
	} else {
	    n = OffsetNumberNext(n);
	}
    }
    
    return (n);
}

static RetrieveIndexResult
gistscancache(IndexScanDesc s, ScanDirection dir)
{
    RetrieveIndexResult res;
    ItemPointer ip;
    
    if (!(ScanDirectionIsNoMovement(dir)
	  && ItemPointerIsValid(&(s->currentItemData)))) {
	
	return ((RetrieveIndexResult) NULL);
    } 
    
    ip = gistheapptr(s->relation, &(s->currentItemData));
    
    if (ItemPointerIsValid(ip))
	res = FormRetrieveIndexResult(&(s->currentItemData), ip);
    else
	res = (RetrieveIndexResult) NULL;
    
    return (res);
}

/*
 *  gistheapptr returns the item pointer to the tuple in the heap relation
 *  for which itemp is the index relation item pointer.
 */
static ItemPointer
gistheapptr(Relation r, ItemPointer itemp)
{
    Buffer b;
    Page p;
    IndexTuple it;
    ItemPointer ip;
    OffsetNumber n;
    
    ip = (ItemPointer) palloc(sizeof(ItemPointerData));
    if (ItemPointerIsValid(itemp)) {
	b = ReadBuffer(r, ItemPointerGetBlockNumber(itemp));
	p = BufferGetPage(b);
	n = ItemPointerGetOffsetNumber(itemp);
	it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
	memmove((char *) ip, (char *) &(it->t_tid),
		sizeof(ItemPointerData));
	ReleaseBuffer(b);
    } else {
	ItemPointerSetInvalid(ip);
    }
    
    return (ip);
}