src/backend/access/gist/gistget.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * gistget.c
   4  *        fetch tuples from a GiST scan.
   5  *
   6  *
   7  * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
   8  * Portions Copyright (c) 1994, Regents of the University of California
   9  *
  10  * IDENTIFICATION
  11  *        $PostgreSQL$
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/gist_private.h"
  18 #include "access/relscan.h"
  19 #include "executor/execdebug.h"
  20 #include "miscadmin.h"
  21 #include "pgstat.h"
  22 #include "storage/bufmgr.h"
  23 #include "utils/memutils.h"
  24
  25
  26 static OffsetNumber gistfindnext(IndexScanDesc scan, OffsetNumber n);
  27 static int64 gistnext(IndexScanDesc scan, TIDBitmap *tbm);
  28 static bool gistindex_keytest(IndexTuple tuple, IndexScanDesc scan,
  29                                   OffsetNumber offset);
  30
  31 static void
  32 killtuple(Relation r, GISTScanOpaque so, ItemPointer iptr)
  33 {
  34         Page            p;
  35         OffsetNumber offset;
  36
  37         LockBuffer(so->curbuf, GIST_SHARE);
  38         gistcheckpage(r, so->curbuf);
  39         p = (Page) BufferGetPage(so->curbuf);
  40
  41         if (XLByteEQ(so->stack->lsn, PageGetLSN(p)))
  42         {
  43                 /* page unchanged, so all is simple */
  44                 offset = ItemPointerGetOffsetNumber(iptr);
  45                 ItemIdMarkDead(PageGetItemId(p, offset));
  46                 SetBufferCommitInfoNeedsSave(so->curbuf);
  47         }
  48         else
  49         {
  50                 OffsetNumber maxoff = PageGetMaxOffsetNumber(p);
  51
  52                 for (offset = FirstOffsetNumber; offset <= maxoff; offset = OffsetNumberNext(offset))
  53                 {
  54                         IndexTuple      ituple = (IndexTuple) PageGetItem(p, PageGetItemId(p, offset));
  55
  56                         if (ItemPointerEquals(&(ituple->t_tid), iptr))
  57                         {
  58                                 /* found */
  59                                 ItemIdMarkDead(PageGetItemId(p, offset));
  60                                 SetBufferCommitInfoNeedsSave(so->curbuf);
  61                                 break;
  62                         }
  63                 }
  64         }
  65
  66         LockBuffer(so->curbuf, GIST_UNLOCK);
  67 }
  68
  69 /*
  70  * gistgettuple() -- Get the next tuple in the scan
  71  */
  72 Datum
  73 gistgettuple(PG_FUNCTION_ARGS)
  74 {
  75         IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
  76         ScanDirection dir = (ScanDirection) PG_GETARG_INT32(1);
  77         GISTScanOpaque so;
  78         bool            res;
  79
  80         so = (GISTScanOpaque) scan->opaque;
  81
  82         if (dir != ForwardScanDirection)
  83                 elog(ERROR, "GiST doesn't support other scan directions than forward");
  84
  85         /*
  86          * If we have produced an index tuple in the past and the executor has
  87          * informed us we need to mark it as "killed", do so now.
  88          */
  89         if (scan->kill_prior_tuple && ItemPointerIsValid(&(so->curpos)))
  90                 killtuple(scan->indexRelation, so, &(so->curpos));
  91
  92         /*
  93          * Get the next tuple that matches the search key.
  94          */
  95         res = (gistnext(scan, NULL) > 0);
  96
  97         PG_RETURN_BOOL(res);
  98 }
  99
 100 Datum
 101 gistgetbitmap(PG_FUNCTION_ARGS)
 102 {
 103         IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
 104         TIDBitmap  *tbm = (TIDBitmap *) PG_GETARG_POINTER(1);
 105         int64           ntids;
 106
 107         ntids = gistnext(scan, tbm);
 108
 109         PG_RETURN_INT64(ntids);
 110 }
 111
 112 /*
 113  * Fetch tuple(s) that match the search key; this can be invoked
 114  * either to fetch the first such tuple or subsequent matching tuples.
 115  *
 116  * This function is used by both gistgettuple and gistgetbitmap. When
 117  * invoked from gistgettuple, tbm is null and the next matching tuple
 118  * is returned in scan->xs_ctup.t_self.  When invoked from getbitmap,
 119  * tbm is non-null and all matching tuples are added to tbm before
 120  * returning.  In both cases, the function result is the number of
 121  * returned tuples.
 122  *
 123  * If scan specifies to skip killed tuples, continue looping until we find a
 124  * non-killed tuple that matches the search key.
 125  */
 126 static int64
 127 gistnext(IndexScanDesc scan, TIDBitmap *tbm)
 128 {
 129         Page            p;
 130         OffsetNumber n;
 131         GISTScanOpaque so;
 132         GISTSearchStack *stk;
 133         IndexTuple      it;
 134         GISTPageOpaque opaque;
 135         int64           ntids = 0;
 136
 137         so = (GISTScanOpaque) scan->opaque;
 138
 139         if (so->qual_ok == false)
 140                 return 0;
 141
 142         if (so->curbuf == InvalidBuffer)
 143         {
 144                 if (ItemPointerIsValid(&so->curpos) == false)
 145                 {
 146                         /* Being asked to fetch the first entry, so start at the root */
 147                         Assert(so->curbuf == InvalidBuffer);
 148                         Assert(so->stack == NULL);
 149
 150                         so->curbuf = ReadBuffer(scan->indexRelation, GIST_ROOT_BLKNO);
 151
 152                         stk = so->stack = (GISTSearchStack *) palloc0(sizeof(GISTSearchStack));
 153
 154                         stk->next = NULL;
 155                         stk->block = GIST_ROOT_BLKNO;
 156
 157                         pgstat_count_index_scan(scan->indexRelation);
 158                 }
 159                 else
 160                 {
 161                         /* scan is finished */
 162                         return 0;
 163                 }
 164         }
 165
 166         /*
 167          * check stored pointers from last visit
 168          */
 169         if (so->nPageData > 0)
 170         {
 171                 /*
 172                  * gistgetmulti never should go here
 173                  */
 174                 Assert(tbm == NULL);
 175
 176                 if (so->curPageData < so->nPageData)
 177                 {
 178                         scan->xs_ctup.t_self = so->pageData[so->curPageData].heapPtr;
 179                         scan->xs_recheck = so->pageData[so->curPageData].recheck;
 180
 181                         ItemPointerSet(&so->curpos,
 182                                                    BufferGetBlockNumber(so->curbuf),
 183                                                    so->pageData[so->curPageData].pageOffset);
 184
 185                         so->curPageData++;
 186
 187                         return 1;
 188                 }
 189                 else
 190                 {
 191                         /*
 192                          * Go to the next page
 193                          */
 194                         stk = so->stack->next;
 195                         pfree(so->stack);
 196                         so->stack = stk;
 197
 198                         /* If we're out of stack entries, we're done */
 199                         if (so->stack == NULL)
 200                         {
 201                                 ReleaseBuffer(so->curbuf);
 202                                 so->curbuf = InvalidBuffer;
 203                                 return 0;
 204                         }
 205
 206                         so->curbuf = ReleaseAndReadBuffer(so->curbuf,
 207                                                                                           scan->indexRelation,
 208                                                                                           stk->block);
 209                 }
 210         }
 211
 212         for (;;)
 213         {
 214                 CHECK_FOR_INTERRUPTS();
 215
 216                 /* First of all, we need lock buffer */
 217                 Assert(so->curbuf != InvalidBuffer);
 218                 LockBuffer(so->curbuf, GIST_SHARE);
 219                 gistcheckpage(scan->indexRelation, so->curbuf);
 220                 p = BufferGetPage(so->curbuf);
 221                 opaque = GistPageGetOpaque(p);
 222
 223                 /* remember lsn to identify page changed for tuple's killing */
 224                 so->stack->lsn = PageGetLSN(p);
 225
 226                 /* check page split, occured since visit to parent */
 227                 if (!XLogRecPtrIsInvalid(so->stack->parentlsn) &&
 228                         XLByteLT(so->stack->parentlsn, opaque->nsn) &&
 229                         opaque->rightlink != InvalidBlockNumber /* sanity check */ &&
 230                         (so->stack->next == NULL || so->stack->next->block != opaque->rightlink)        /* check if already
 231                                 added */ )
 232                 {
 233                         /* detect page split, follow right link to add pages */
 234
 235                         stk = (GISTSearchStack *) palloc(sizeof(GISTSearchStack));
 236                         stk->next = so->stack->next;
 237                         stk->block = opaque->rightlink;
 238                         stk->parentlsn = so->stack->parentlsn;
 239                         memset(&(stk->lsn), 0, sizeof(GistNSN));
 240                         so->stack->next = stk;
 241                 }
 242
 243                 /* if page is empty, then just skip it */
 244                 if (PageIsEmpty(p))
 245                 {
 246                         LockBuffer(so->curbuf, GIST_UNLOCK);
 247                         stk = so->stack->next;
 248                         pfree(so->stack);
 249                         so->stack = stk;
 250
 251                         if (so->stack == NULL)
 252                         {
 253                                 ReleaseBuffer(so->curbuf);
 254                                 so->curbuf = InvalidBuffer;
 255                                 return ntids;
 256                         }
 257
 258                         so->curbuf = ReleaseAndReadBuffer(so->curbuf, scan->indexRelation,
 259                                                                                           stk->block);
 260                         continue;
 261                 }
 262
 263                 n = FirstOffsetNumber;
 264
 265                 /* wonderful, we can look at page */
 266                 so->nPageData = so->curPageData = 0;
 267
 268                 for (;;)
 269                 {
 270                         n = gistfindnext(scan, n);
 271
 272                         if (!OffsetNumberIsValid(n))
 273                         {
 274                                 /*
 275                                  * If we was called from gistgettuple and current buffer
 276                                  * contains something matched then make a recursive call - it
 277                                  * will return ItemPointer from so->pageData. But we save
 278                                  * buffer pinned to support tuple's killing
 279                                  */
 280                                 if (!tbm && so->nPageData > 0)
 281                                 {
 282                                         LockBuffer(so->curbuf, GIST_UNLOCK);
 283                                         return gistnext(scan, NULL);
 284                                 }
 285
 286                                 /*
 287                                  * We ran out of matching index entries on the current page,
 288                                  * so pop the top stack entry and use it to continue the
 289                                  * search.
 290                                  */
 291                                 LockBuffer(so->curbuf, GIST_UNLOCK);
 292                                 stk = so->stack->next;
 293                                 pfree(so->stack);
 294                                 so->stack = stk;
 295
 296                                 /* If we're out of stack entries, we're done */
 297
 298                                 if (so->stack == NULL)
 299                                 {
 300                                         ReleaseBuffer(so->curbuf);
 301                                         so->curbuf = InvalidBuffer;
 302                                         return ntids;
 303                                 }
 304
 305                                 so->curbuf = ReleaseAndReadBuffer(so->curbuf,
 306                                                                                                   scan->indexRelation,
 307                                                                                                   stk->block);
 308                                 /* XXX  go up */
 309                                 break;
 310                         }
 311
 312                         if (GistPageIsLeaf(p))
 313                         {
 314                                 /*
 315                                  * We've found a matching index entry in a leaf page, so
 316                                  * return success. Note that we keep "curbuf" pinned so that
 317                                  * we can efficiently resume the index scan later.
 318                                  */
 319
 320                                 if (!(scan->ignore_killed_tuples &&
 321                                           ItemIdIsDead(PageGetItemId(p, n))))
 322                                 {
 323                                         it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
 324                                         ntids++;
 325                                         if (tbm != NULL)
 326                                                 tbm_add_tuples(tbm, &it->t_tid, 1, scan->xs_recheck);
 327                                         else
 328                                         {
 329                                                 so->pageData[so->nPageData].heapPtr = it->t_tid;
 330                                                 so->pageData[so->nPageData].pageOffset = n;
 331                                                 so->pageData[so->nPageData].recheck = scan->xs_recheck;
 332                                                 so->nPageData++;
 333                                         }
 334                                 }
 335                         }
 336                         else
 337                         {
 338                                 /*
 339                                  * We've found an entry in an internal node whose key is
 340                                  * consistent with the search key, so push it to stack
 341                                  */
 342                                 stk = (GISTSearchStack *) palloc(sizeof(GISTSearchStack));
 343
 344                                 it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
 345                                 stk->block = ItemPointerGetBlockNumber(&(it->t_tid));
 346                                 memset(&(stk->lsn), 0, sizeof(GistNSN));
 347                                 stk->parentlsn = so->stack->lsn;
 348
 349                                 stk->next = so->stack->next;
 350                                 so->stack->next = stk;
 351                         }
 352
 353                         n = OffsetNumberNext(n);
 354                 }
 355         }
 356
 357         return ntids;
 358 }
 359
 360 /*
 361  * gistindex_keytest() -- does this index tuple satisfy the scan key(s)?
 362  *
 363  * On success return for a leaf tuple, scan->xs_recheck is set to indicate
 364  * whether recheck is needed.  We recheck if any of the consistent() functions
 365  * request it.
 366  *
 367  * We must decompress the key in the IndexTuple before passing it to the
 368  * sk_func (and we have previously overwritten the sk_func to use the
 369  * user-defined Consistent method, so we actually are invoking that).
 370  *
 371  * Note that this function is always invoked in a short-lived memory context,
 372  * so we don't need to worry about cleaning up allocated memory, either here
 373  * or in the implementation of any Consistent methods.
 374  */
 375 static bool
 376 gistindex_keytest(IndexTuple tuple,
 377                                   IndexScanDesc scan,
 378                                   OffsetNumber offset)
 379 {
 380         int                     keySize = scan->numberOfKeys;
 381         ScanKey         key = scan->keyData;
 382         Relation        r = scan->indexRelation;
 383         GISTScanOpaque so;
 384         Page            p;
 385         GISTSTATE  *giststate;
 386
 387         so = (GISTScanOpaque) scan->opaque;
 388         giststate = so->giststate;
 389         p = BufferGetPage(so->curbuf);
 390
 391         IncrIndexProcessed();
 392
 393         scan->xs_recheck = false;
 394
 395         /*
 396          * Tuple doesn't restore after crash recovery because of incomplete insert
 397          */
 398         if (!GistPageIsLeaf(p) && GistTupleIsInvalid(tuple))
 399                 return true;
 400
 401         while (keySize > 0)
 402         {
 403                 Datum           datum;
 404                 bool            isNull;
 405                 Datum           test;
 406                 bool            recheck;
 407                 GISTENTRY       de;
 408
 409                 datum = index_getattr(tuple,
 410                                                           key->sk_attno,
 411                                                           giststate->tupdesc,
 412                                                           &isNull);
 413
 414                 if (key->sk_flags & SK_ISNULL)
 415                 {
 416                         /*
 417                          * On non-leaf page we can't conclude that child hasn't NULL
 418                          * values because of assumption in GiST: uinon (VAL, NULL) is VAL
 419                          * But if on non-leaf page key IS  NULL then all childs has NULL.
 420                          */
 421
 422                         Assert(key->sk_flags & SK_SEARCHNULL);
 423
 424                         if (GistPageIsLeaf(p) && !isNull)
 425                                 return false;
 426                 }
 427                 else if (isNull)
 428                 {
 429                         return false;
 430                 }
 431                 else
 432                 {
 433                         gistdentryinit(giststate, key->sk_attno - 1, &de,
 434                                                    datum, r, p, offset,
 435                                                    FALSE, isNull);
 436
 437                         /*
 438                          * Call the Consistent function to evaluate the test.  The
 439                          * arguments are the index datum (as a GISTENTRY*), the comparison
 440                          * datum, the comparison operator's strategy number and subtype
 441                          * from pg_amop, and the recheck flag.
 442                          *
 443                          * (Presently there's no need to pass the subtype since it'll
 444                          * always be zero, but might as well pass it for possible future
 445                          * use.)
 446                          *
 447                          * We initialize the recheck flag to true (the safest assumption)
 448                          * in case the Consistent function forgets to set it.
 449                          */
 450                         recheck = true;
 451
 452                         test = FunctionCall5(&key->sk_func,
 453                                                                  PointerGetDatum(&de),
 454                                                                  key->sk_argument,
 455                                                                  Int32GetDatum(key->sk_strategy),
 456                                                                  ObjectIdGetDatum(key->sk_subtype),
 457                                                                  PointerGetDatum(&recheck));
 458
 459                         if (!DatumGetBool(test))
 460                                 return false;
 461                         scan->xs_recheck |= recheck;
 462                 }
 463
 464                 keySize--;
 465                 key++;
 466         }
 467
 468         return true;
 469 }
 470
 471 /*
 472  * Return the offset of the first index entry that is consistent with
 473  * the search key after offset 'n' in the current page. If there are
 474  * no more consistent entries, return InvalidOffsetNumber.
 475  * On success, scan->xs_recheck is set correctly, too.
 476  * Page should be locked....
 477  */
 478 static OffsetNumber
 479 gistfindnext(IndexScanDesc scan, OffsetNumber n)
 480 {
 481         OffsetNumber maxoff;
 482         IndexTuple      it;
 483         GISTScanOpaque so;
 484         MemoryContext oldcxt;
 485         Page            p;
 486
 487         so = (GISTScanOpaque) scan->opaque;
 488         p = BufferGetPage(so->curbuf);
 489         maxoff = PageGetMaxOffsetNumber(p);
 490
 491         /*
 492          * Make sure we're in a short-lived memory context when we invoke a
 493          * user-supplied GiST method in gistindex_keytest(), so we don't leak
 494          * memory
 495          */
 496         oldcxt = MemoryContextSwitchTo(so->tempCxt);
 497
 498         while (n >= FirstOffsetNumber && n <= maxoff)
 499         {
 500                 it = (IndexTuple) PageGetItem(p, PageGetItemId(p, n));
 501                 if (gistindex_keytest(it, scan, n))
 502                         break;
 503
 504                 n = OffsetNumberNext(n);
 505         }
 506
 507         MemoryContextSwitchTo(oldcxt);
 508         MemoryContextReset(so->tempCxt);
 509
 510         /*
 511          * If we found a matching entry, return its offset; otherwise return
 512          * InvalidOffsetNumber to inform the caller to go to the next page.
 513          */
 514         if (n >= FirstOffsetNumber && n <= maxoff)
 515                 return n;
 516         else
 517                 return InvalidOffsetNumber;
 518 }