root / hw / hw2 / MyStuff / gistget.c

/*-------------------------------------------------------------------------
 *
 * gistget.c
 *    fetch tuples from a GIST scan.
 *
 *
 * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *    $PostgreSQL: pgsql/src/backend/access/gist/gistget.c,v 1.62 2006/11/12 06:55:53 neilc Exp $
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/gist_private.h"
#include "executor/execdebug.h"
#include "pgstat.h"
#include "utils/memutils.h"
#include "utils/builtins.h"
#include <stdio.h>


static OffsetNumber gistfindnext(IndexScanDesc scan, OffsetNumber n,
       ScanDirection dir);
static int  gistnext(IndexScanDesc scan, ScanDirection dir, ItemPointer tids, int maxtids, bool ignore_killed_tuples);
static double gistindex_keydistance(IndexTuple tuple, IndexScanDesc scan, 
          OffsetNumber offset);
static bool gistindex_keytest(IndexTuple tuple, IndexScanDesc scan,
          OffsetNumber offset);

static void
killtuple(Relation r, GISTScanOpaque so, ItemPointer iptr)
{
  Buffer    buffer = so->curbuf;

  for (;;)
  {
    Page    p;
    BlockNumber blkno;
    OffsetNumber offset,
          maxoff;

    LockBuffer(buffer, GIST_SHARE);
    gistcheckpage(r, buffer);
    p = (Page) BufferGetPage(buffer);

    if (buffer == so->curbuf && XLByteEQ(so->stack->lsn, PageGetLSN(p)))
    {
      /* page unchanged, so all is simple */
      offset = ItemPointerGetOffsetNumber(iptr);
      PageGetItemId(p, offset)->lp_flags |= LP_DELETE;
      SetBufferCommitInfoNeedsSave(buffer);
      LockBuffer(buffer, GIST_UNLOCK);
      break;
    }

    maxoff = PageGetMaxOffsetNumber(p);

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

      if (ItemPointerEquals(&(ituple->t_tid), iptr))
      {
        /* found */
        PageGetItemId(p, offset)->lp_flags |= LP_DELETE;
        SetBufferCommitInfoNeedsSave(buffer);
        LockBuffer(buffer, GIST_UNLOCK);
        if (buffer != so->curbuf)
          ReleaseBuffer(buffer);
        return;
      }
    }

    /* follow right link */

    /*
     * ??? is it good? if tuple dropped by concurrent vacuum, we will read
     * all leaf pages...
     */
    blkno = GistPageGetOpaque(p)->rightlink;
    LockBuffer(buffer, GIST_UNLOCK);
    if (buffer != so->curbuf)
      ReleaseBuffer(buffer);

    if (blkno == InvalidBlockNumber)
      /* can't found, dropped by somebody else */
      return;
    buffer = ReadBuffer(r, blkno);
  }
}

/*
 * gistgettuple() -- Get the next tuple in the scan
 */
Datum
gistgettuple(PG_FUNCTION_ARGS)
{
  IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
  ScanDirection dir = (ScanDirection) PG_GETARG_INT32(1);
  GISTScanOpaque so;
  ItemPointerData tid;
  bool    res;

  so = (GISTScanOpaque) scan->opaque;

  /*
   * If we have produced an index tuple in the past and the executor has
   * informed us we need to mark it as "killed", do so now.
   */
  if (scan->kill_prior_tuple && ItemPointerIsValid(&(scan->currentItemData)))
    killtuple(scan->indexRelation, so, &(scan->currentItemData));

  /*
   * Get the next tuple that matches the search key. If asked to skip killed
   * tuples, continue looping until we find a non-killed tuple that matches
   * the search key.
   */
  res = (gistnext(scan, dir, &tid, 1, scan->ignore_killed_tuples)) ? true : false;

  PG_RETURN_BOOL(res);
}

Datum
gistgetmulti(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);

  *returned_tids = gistnext(scan, ForwardScanDirection, tids, max_tids, false);

  PG_RETURN_BOOL(*returned_tids == max_tids);
}

/*
 * CS186 HW2: RELEVANT CODE BEGINS HERE 
 */

/* insert a new priority queue entry into so->pq */
static void
gist_pq_insertpq(GISTScanOpaque so, /* State of a GiST scan */
                 GISTPQ *newpq /* a new GISTPQ entry */
                 )
{
  /* CS186 HW2: fill me in! */
}

/* 
 * given the current state of the index scan, generate a GISTPQ to insert into
 * the priority queue, and insert it. Note that the "stack" holds the blocks
 * from the root to the current item.
 * There are two cases to consider: 
 *   1) the item being inserted is an entry in a leaf node, identified by a
 *      block number (from the stack), and the offset.  It points to a data
 *      item in the database (isdata == true).
 *   2) the item being inserted is the block number of an index node to 
 *      be visited next (isdata == false).  The key corresponding to this 
 *      block number is in the block's *parent* in the tree (which can be
 *      found in the stack) at the given offset.
 */
static void
gist_pq_insert(IndexScanDesc scan, /* scan descriptor */
               OffsetNumber offset, /* "slot number" of item being inserted */
               GISTSearchStack *stack, /* search stack to this item */
               bool isData /* is the item being inserted a pointer to data? */
               )
{
  /* CS186 HW2: fill me in! Should call gist_pq_insertpq. */

}

/* Find the minimum item in the priority queue (without deleting it) */
static GISTPQ *
gist_pq_findmin(GISTScanOpaque so)
{
  /* CS186 HW2: implement me */
  return NULL;
}

/*
 * Find the min item in the priority queue, remove it from the queue, and 
 * return it.  Should NOT free the space allocated for the min; the caller 
 * does that when they're done with it.
 */
static GISTPQ *
gist_pq_deletemin(GISTScanOpaque so)
{
  /* CS186 HW2: implement me! */
  return NULL;
}

/* Optional debugging routine -- dump the state of the priority queue */
static void
gist_pq_dump(GISTScanOpaque so)
{
  /* CS186 HW2: fill me in if you like. */
}

/* Support routine for gistnext, to load in the root element on first call */
static void
gistnext_get_root(IndexScanDesc scan, GISTScanOpaque so)
{
  GISTSearchStack *stk;

  /* Being asked to fetch the first entry, so start at the root */
  Assert(so->curbuf == InvalidBuffer);
  Assert(so->stack == NULL);

  so->curbuf = ReadBuffer(scan->indexRelation, GIST_ROOT_BLKNO);

  stk = so->stack = (GISTSearchStack *) palloc0(sizeof(GISTSearchStack));

  stk->next = NULL;

  pgstat_count_index_scan(&scan->xs_pgstat_info);
}

/*
 * Support routine for gistnext.  Deals with concurrency control issues to
 * check if the top node on the search stack was split by a concurrent
 * inserter since the time it was put on the stack.  If so, the "rest"
 * of the split can be found by adding the right-hand
 * neighbor of the split node to the search stack.  See
 * Kornacker/Mohan/Hellerstein, SIGMOD 1997, for details.
 */
static bool
gistnext_checksplit(IndexScanDesc scan, GISTScanOpaque so, Page p,
                    GISTPageOpaque opaque)
{
  bool            resetoffset = false;
  GISTSearchStack *stk;

  if (XLogRecPtrIsInvalid(so->stack->lsn) || !XLByteEQ(so->stack->lsn, PageGetLSN(p)))
  {
    /* page changed from last visit or visit first time , reset offset */
    so->stack->lsn = PageGetLSN(p);
    resetoffset = true;

    /* check page split, occured from last visit or visit to parent */
    if (!XLogRecPtrIsInvalid(so->stack->parentlsn) &&
      XLByteLT(so->stack->parentlsn, opaque->nsn) &&
      opaque->rightlink != InvalidBlockNumber /* sanity check */ &&
      (so->stack->next == NULL || so->stack->next->block != opaque->rightlink)    /* check if already
        added */ )
    {
      /* detect page split, follow right link to add pages */

      /* 
       * CS186 HW2: Fix me!
       * Modify this next code to put the rightlinked block on
       * the priority queue.
       * Note: since this page was split since it was inserted into
       * the priority queue, it has fewer items on it, and a different
       * bounding box, so its priority may have changed.
       */
      stk = (GISTSearchStack *) palloc0(sizeof(GISTSearchStack));
      stk->next = so->stack->next;
      stk->block = opaque->rightlink;
      stk->parentlsn = so->stack->parentlsn;
      memset(&(stk->lsn), 0, sizeof(GistNSN));
      so->stack->next = stk;
    }
  }
  return resetoffset;
}

/*
 * Fetch tuples that matchs the search key; this can be invoked
 * either to fetch the first such tuple or subsequent matching
 * tuples. Returns true iff a matching tuple was found.
 */
static int
gistnext(IndexScanDesc scan, ScanDirection dir, ItemPointer tids, int maxtids, bool ignore_killed_tuples)
{
  Page    p;
  OffsetNumber n;
  GISTScanOpaque so;
  GISTSearchStack *stk;
  IndexTuple  it;
  GISTPageOpaque opaque;
  bool    resetoffset = false;
  int      ntids = 0;
  
  /* CS186 HW2: This code currently returns tuples that match the query in
   * an arbitrary order (depth-first left-to-right).  Modify it to return
   * matching tuples ordered by distance to search key 
   */

  so = (GISTScanOpaque) scan->opaque;

  if (ItemPointerIsValid(&scan->currentItemData) == false)
  {
    gistnext_get_root(scan, so);
  }
  else if (so->curbuf == InvalidBuffer)
  {
    return 0;
  }

  for (;;)
  {
    /* First of all, we need to get a "lightweight" lock on the buffer */
    Assert(so->curbuf != InvalidBuffer);
    LockBuffer(so->curbuf, GIST_SHARE);
    
    gistcheckpage(scan->indexRelation, so->curbuf);
    p = BufferGetPage(so->curbuf);
    opaque = GistPageGetOpaque(p);

    /* deal with case where this page split since last we saw it */
    resetoffset = gistnext_checksplit(scan, so, p, opaque);

    /* if page is empty, then just skip it */
    if (PageIsEmpty(p))
    {
      LockBuffer(so->curbuf, GIST_UNLOCK);
      stk = so->stack->next;
      pfree(so->stack);
      so->stack = stk;

      if (so->stack == NULL)
      {
        ReleaseBuffer(so->curbuf);
        so->curbuf = InvalidBuffer;
        return ntids;
      }

      so->curbuf = ReleaseAndReadBuffer(so->curbuf, scan->indexRelation,
                        stk->block);
      continue;
    }

    if (!GistPageIsLeaf(p) || resetoffset ||
      !ItemPointerIsValid(&scan->currentItemData))
    {
      if (ScanDirectionIsBackward(dir))
        n = PageGetMaxOffsetNumber(p);
      else
        n = FirstOffsetNumber;
    }
    else
    {
      n = ItemPointerGetOffsetNumber(&(scan->currentItemData));

      if (ScanDirectionIsBackward(dir))
        n = OffsetNumberPrev(n);
      else
        n = OffsetNumberNext(n);
    }

    /* wonderful, we can look at page */

    for (;;)
    {
      n = gistfindnext(scan, n, dir);

      if (!OffsetNumberIsValid(n))
      {
        /*
         * We ran out of matching index entries on the current page,
         * so pop the top stack entry and use it to continue the
         * search.
         */
        LockBuffer(so->curbuf, GIST_UNLOCK);
        stk = so->stack->next;
        pfree(so->stack);
        so->stack = stk;

        /* If we're out of stack entries, we're done */

        if (so->stack == NULL)
        {
          ReleaseBuffer(so->curbuf);
          so->curbuf = InvalidBuffer;
          return ntids;
        }

        so->curbuf = ReleaseAndReadBuffer(so->curbuf,
                          scan->indexRelation,
                          stk->block);
        /* XXX  go up */
        break;
      }

      if (GistPageIsLeaf(p))
      {
        /*
         * We've found a matching index entry in a leaf page, so
         * return success. Note that we keep "curbuf" pinned so that
         * we can efficiently resume the index scan later.
         */

        ItemPointerSet(&(scan->currentItemData),
                 BufferGetBlockNumber(so->curbuf), n);

        if (!(ignore_killed_tuples && ItemIdDeleted(PageGetItemId(p, n))))
        {
          it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
          tids[ntids] = scan->xs_ctup.t_self = it->t_tid;
          ntids++;

          if (ntids == maxtids)
          {
            LockBuffer(so->curbuf, GIST_UNLOCK);
            return ntids;
          }
        }
      }
      else
      {
        /*
         * We've found an entry in an internal node whose key is
         * consistent with the search key, so push it to stack
         */

        stk = (GISTSearchStack *) palloc0(sizeof(GISTSearchStack));

        it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
        stk->block = ItemPointerGetBlockNumber(&(it->t_tid));
        memset(&(stk->lsn), 0, sizeof(GistNSN));
        stk->parentlsn = so->stack->lsn;

        stk->next = so->stack->next;
        so->stack->next = stk;

      }

      if (ScanDirectionIsBackward(dir))
        n = OffsetNumberPrev(n);
      else
        n = OffsetNumberNext(n);
    }
  }

  return ntids;
}


/*
 * gistindex_keytest() -- does this index tuple satisfy the scan key(s)?
 *
 * We must decompress the key in the IndexTuple before passing it to the
 * sk_func (and we have previously overwritten the sk_func to use the
 * user-defined Consistent method, so we actually are invoking that).
 *
 * Note that this function is always invoked in a short-lived memory context,
 * so we don't need to worry about cleaning up allocated memory, either here
 * or in the implementation of any Consistent methods.
 */
static bool
gistindex_keytest(IndexTuple tuple,
          IndexScanDesc scan,
          OffsetNumber offset)
{
  int      keySize = scan->numberOfKeys;
  ScanKey    key = scan->keyData;
  Relation  r = scan->indexRelation;
  GISTScanOpaque so;
  Page    p;
  GISTSTATE  *giststate;

  so = (GISTScanOpaque) scan->opaque;
  giststate = so->giststate;
  p = BufferGetPage(so->curbuf);

  IncrIndexProcessed();

  /*
   * Tuple doesn't restore after crash recovery because of incomplete insert
   */
  if (!GistPageIsLeaf(p) && GistTupleIsInvalid(tuple))
    return true;

  while (keySize > 0)
  {
    Datum    datum;
    bool    isNull;
    Datum    test;
    GISTENTRY  de;

    datum = index_getattr(tuple,
                key->sk_attno,
                giststate->tupdesc,
                &isNull);

    if (key->sk_flags & SK_ISNULL)
    {
      /*
       * is the compared-to datum NULL? on non-leaf page it's possible
       * to have nulls in childs :(
       */

      if (isNull || !GistPageIsLeaf(p))
        return true;
      return false;
    }
    else if (isNull)
      return false;

    /* decompress the key! */
    gistdentryinit(giststate, key->sk_attno - 1, &de,
             datum, r, p, offset,
             FALSE, isNull);

    /*
     * Call the Consistent function to evaluate the test.  The arguments
     * are the index datum (as a GISTENTRY*), the comparison datum, and
     * the comparison operator's strategy number and subtype from pg_amop.
     *
     * (Presently there's no need to pass the subtype since it'll always
     * be zero, but might as well pass it for possible future use.)
     */
    test = FunctionCall4(&key->sk_func,
               PointerGetDatum(&de),
               key->sk_argument,
               Int32GetDatum(key->sk_strategy),
               ObjectIdGetDatum(key->sk_subtype));

    if (!DatumGetBool(test))
      return false;

    keySize--;
    key++;
  }

  return true;
}

/* 
 * Use the mindistance function to compute the distance of the given index
 * entry from the constant in the scan.  See gistindex_keytest for an 
 * analogous task.
 */

static double
gistindex_keydistance(  IndexTuple tuple,
            IndexScanDesc scan,
            OffsetNumber offset)
{
  int      keySize = scan->numberOfKeys;
  ScanKey    key = scan->keyData;
  Relation  r = scan->indexRelation;
  GISTScanOpaque so;
  Page    p;
  GISTSTATE  *giststate;
  double retval = 0;
  
  so = (GISTScanOpaque) scan->opaque;
  giststate = so->giststate;
  p = BufferGetPage(so->curbuf);

  while (keySize > 0)
  {
    Datum    datum;
    bool    isNull;
    Datum    test;
    GISTENTRY  de;

    datum = index_getattr(tuple,
                key->sk_attno,
                giststate->tupdesc,
                &isNull);

    if (key->sk_flags & SK_ISNULL)
    {
      /*
       * is the compared-to datum NULL? on non-leaf page it's possible
       * to have nulls in childs :(
       */

      if (isNull || !GistPageIsLeaf(p))
        return get_float8_infinity();
      return 0;
    }
    else if (isNull)
      return 0;

    gistdentryinit(giststate, key->sk_attno - 1, &de,
             datum, r, p, offset,
             FALSE, isNull);

    /*
     * Call the Distance function to evaluate the test.  The arguments
     * are the index datum (as a GISTENTRY*), and the comparison datum.
     */
    test = FunctionCall2(&giststate->distanceFn[key->sk_attno - 1],
               PointerGetDatum(&de),
               key->sk_argument);

    retval += DatumGetFloat8(test);

    keySize--;
    key++;
  }

  return retval;
}

/*
 * Return the offset of the first index entry that is consistent with
 * the search key after offset 'n' in the current page. If there are
 * no more consistent entries, return InvalidOffsetNumber.
 * Page should be locked....
 */
static OffsetNumber
gistfindnext(IndexScanDesc scan, OffsetNumber n, ScanDirection dir)
{
  OffsetNumber maxoff;
  IndexTuple  it;
  GISTScanOpaque so;
  MemoryContext oldcxt;
  Page    p;

  so = (GISTScanOpaque) scan->opaque;
  p = BufferGetPage(so->curbuf);
  maxoff = PageGetMaxOffsetNumber(p);

  /*
   * Make sure we're in a short-lived memory context when we invoke a
   * user-supplied GIST method in gistindex_keytest(), so we don't leak
   * memory
   */
  oldcxt = MemoryContextSwitchTo(so->tempCxt);

  /*
   * 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->flags & GS_CURBEFORE)
  {
    so->flags &= ~GS_CURBEFORE;
    n = OffsetNumberPrev(n);
  }

  while (n >= FirstOffsetNumber && n <= maxoff)
  {
    it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
    if (gistindex_keytest(it, scan, n))
      break;

    if (ScanDirectionIsBackward(dir))
      n = OffsetNumberPrev(n);
    else
      n = OffsetNumberNext(n);
  }

  MemoryContextSwitchTo(oldcxt);
  MemoryContextReset(so->tempCxt);

  /*
   * If we found a matching entry, return its offset; otherwise return
   * InvalidOffsetNumber to inform the caller to go to the next page.
   */
  if (n >= FirstOffsetNumber && n <= maxoff)
    return n;
  else
    return InvalidOffsetNumber;
}