granicus.if.org Git - postgresql/blob - src/backend/access/gist/gistscan.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * gistscan.c
   4  *        routines to manage scans on index relations
   5  *
   6  *
   7  * IDENTIFICATION
   8  *        /usr/local/devel/pglite/cvs/src/backend/access/gist/gistscan.c,v 1.7 1995/06/14 00:10:05 jolly Exp
   9  *
  10  *-------------------------------------------------------------------------
  11  */
  12
  13 #include "postgres.h"
  14
  15 #include "access/genam.h"
  16 #include "access/gist.h"
  17 #include "access/gistscan.h"
  18
  19
  20 /* routines defined and used here */
  21 static void gistregscan(IndexScanDesc s);
  22 static void gistdropscan(IndexScanDesc s);
  23 static void gistadjone(IndexScanDesc s, int op, BlockNumber blkno,
  24                    OffsetNumber offnum);
  25 static void adjuststack(GISTSTACK *stk, BlockNumber blkno,
  26                         OffsetNumber offnum);
  27 static void adjustiptr(IndexScanDesc s, ItemPointer iptr,
  28                    int op, BlockNumber blkno, OffsetNumber offnum);
  29
  30 /*
  31  *      Whenever we start a GiST scan in a backend, we register it in private
  32  *      space.  Then if the GiST index gets updated, we check all registered
  33  *      scans and adjust them if the tuple they point at got moved by the
  34  *      update.  We only need to do this in private space, because when we update
  35  *      an GiST we have a write lock on the tree, so no other process can have
  36  *      any locks at all on it.  A single transaction can have write and read
  37  *      locks on the same object, so that's why we need to handle this case.
  38  */
  39
  40 typedef struct GISTScanListData
  41 {
  42         IndexScanDesc gsl_scan;
  43         struct GISTScanListData *gsl_next;
  44 } GISTScanListData;
  45
  46 typedef GISTScanListData *GISTScanList;
  47
  48 /* pointer to list of local scans on GiSTs */
  49 static GISTScanList GISTScans = (GISTScanList) NULL;
  50
  51 Datum
  52 gistbeginscan(PG_FUNCTION_ARGS)
  53 {
  54         Relation        r = (Relation) PG_GETARG_POINTER(0);
  55         bool            fromEnd = PG_GETARG_BOOL(1);
  56         uint16          nkeys = PG_GETARG_UINT16(2);
  57         ScanKey         key = (ScanKey) PG_GETARG_POINTER(3);
  58         IndexScanDesc s;
  59
  60         /*
  61          * Let index_beginscan does its work...
  62          *
  63          * RelationSetLockForRead(r);
  64          */
  65
  66         s = RelationGetIndexScan(r, fromEnd, nkeys, key);
  67         gistregscan(s);
  68
  69         PG_RETURN_POINTER(s);
  70 }
  71
  72 Datum
  73 gistrescan(PG_FUNCTION_ARGS)
  74 {
  75         IndexScanDesc   s = (IndexScanDesc) PG_GETARG_POINTER(0);
  76         bool                    fromEnd = PG_GETARG_BOOL(1);
  77         ScanKey                 key = (ScanKey) PG_GETARG_POINTER(2);
  78         GISTScanOpaque p;
  79         int                     i;
  80
  81         /*
  82          * Clear all the pointers.
  83          */
  84
  85         ItemPointerSetInvalid(&s->previousItemData);
  86         ItemPointerSetInvalid(&s->currentItemData);
  87         ItemPointerSetInvalid(&s->nextItemData);
  88         ItemPointerSetInvalid(&s->previousMarkData);
  89         ItemPointerSetInvalid(&s->currentMarkData);
  90         ItemPointerSetInvalid(&s->nextMarkData);
  91
  92         /*
  93          * Set flags.
  94          */
  95         if (RelationGetNumberOfBlocks(s->relation) == 0)
  96                 s->flags = ScanUnmarked;
  97         else if (fromEnd)
  98                 s->flags = ScanUnmarked | ScanUncheckedPrevious;
  99         else
 100                 s->flags = ScanUnmarked | ScanUncheckedNext;
 101
 102         s->scanFromEnd = fromEnd;
 103
 104         if (s->numberOfKeys > 0)
 105         {
 106                 memmove(s->keyData,
 107                                 key,
 108                                 s->numberOfKeys * sizeof(ScanKeyData));
 109         }
 110
 111         p = (GISTScanOpaque) s->opaque;
 112         if (p != (GISTScanOpaque) NULL)
 113         {
 114                 gistfreestack(p->s_stack);
 115                 gistfreestack(p->s_markstk);
 116                 p->s_stack = p->s_markstk = (GISTSTACK *) NULL;
 117                 p->s_flags = 0x0;
 118                 for (i = 0; i < s->numberOfKeys; i++)
 119                 {
 120                         s->keyData[i].sk_procedure
 121                                 = RelationGetGISTStrategy(s->relation, s->keyData[i].sk_attno,
 122                                                                                   s->keyData[i].sk_procedure);
 123                         s->keyData[i].sk_func = p->giststate->consistentFn;
 124                 }
 125         }
 126         else
 127         {
 128                 /* initialize opaque data */
 129                 p = (GISTScanOpaque) palloc(sizeof(GISTScanOpaqueData));
 130                 p->s_stack = p->s_markstk = (GISTSTACK *) NULL;
 131                 p->s_flags = 0x0;
 132                 s->opaque = p;
 133                 p->giststate = (GISTSTATE *) palloc(sizeof(GISTSTATE));
 134                 initGISTstate(p->giststate, s->relation);
 135                 if (s->numberOfKeys > 0)
 136
 137                         /*
 138                          * * Play games here with the scan key to use the Consistent *
 139                          * function for all comparisons: * 1) the sk_procedure field
 140                          * will now be used to hold the *        strategy number * 2) the
 141                          * sk_func field will point to the Consistent function
 142                          */
 143                         for (i = 0; i < s->numberOfKeys; i++)
 144                         {
 145
 146                                 /*
 147                                  * s->keyData[i].sk_procedure =
 148                                  * index_getprocid(s->relation, 1, GIST_CONSISTENT_PROC);
 149                                  */
 150                                 s->keyData[i].sk_procedure
 151                                         = RelationGetGISTStrategy(s->relation, s->keyData[i].sk_attno,
 152                                                                                           s->keyData[i].sk_procedure);
 153                                 s->keyData[i].sk_func = p->giststate->consistentFn;
 154                         }
 155         }
 156
 157         PG_RETURN_VOID();
 158 }
 159
 160 Datum
 161 gistmarkpos(PG_FUNCTION_ARGS)
 162 {
 163         IndexScanDesc   s = (IndexScanDesc) PG_GETARG_POINTER(0);
 164         GISTScanOpaque p;
 165         GISTSTACK  *o,
 166                            *n,
 167                            *tmp;
 168
 169         s->currentMarkData = s->currentItemData;
 170         p = (GISTScanOpaque) s->opaque;
 171         if (p->s_flags & GS_CURBEFORE)
 172                 p->s_flags |= GS_MRKBEFORE;
 173         else
 174                 p->s_flags &= ~GS_MRKBEFORE;
 175
 176         o = (GISTSTACK *) NULL;
 177         n = p->s_stack;
 178
 179         /* copy the parent stack from the current item data */
 180         while (n != (GISTSTACK *) NULL)
 181         {
 182                 tmp = (GISTSTACK *) palloc(sizeof(GISTSTACK));
 183                 tmp->gs_child = n->gs_child;
 184                 tmp->gs_blk = n->gs_blk;
 185                 tmp->gs_parent = o;
 186                 o = tmp;
 187                 n = n->gs_parent;
 188         }
 189
 190         gistfreestack(p->s_markstk);
 191         p->s_markstk = o;
 192
 193         PG_RETURN_VOID();
 194 }
 195
 196 Datum
 197 gistrestrpos(PG_FUNCTION_ARGS)
 198 {
 199         IndexScanDesc   s = (IndexScanDesc) PG_GETARG_POINTER(0);
 200         GISTScanOpaque p;
 201         GISTSTACK  *o,
 202                            *n,
 203                            *tmp;
 204
 205         s->currentItemData = s->currentMarkData;
 206         p = (GISTScanOpaque) s->opaque;
 207         if (p->s_flags & GS_MRKBEFORE)
 208                 p->s_flags |= GS_CURBEFORE;
 209         else
 210                 p->s_flags &= ~GS_CURBEFORE;
 211
 212         o = (GISTSTACK *) NULL;
 213         n = p->s_markstk;
 214
 215         /* copy the parent stack from the current item data */
 216         while (n != (GISTSTACK *) NULL)
 217         {
 218                 tmp = (GISTSTACK *) palloc(sizeof(GISTSTACK));
 219                 tmp->gs_child = n->gs_child;
 220                 tmp->gs_blk = n->gs_blk;
 221                 tmp->gs_parent = o;
 222                 o = tmp;
 223                 n = n->gs_parent;
 224         }
 225
 226         gistfreestack(p->s_stack);
 227         p->s_stack = o;
 228
 229         PG_RETURN_VOID();
 230 }
 231
 232 Datum
 233 gistendscan(PG_FUNCTION_ARGS)
 234 {
 235         IndexScanDesc   s = (IndexScanDesc) PG_GETARG_POINTER(0);
 236         GISTScanOpaque  p;
 237
 238         p = (GISTScanOpaque) s->opaque;
 239
 240         if (p != (GISTScanOpaque) NULL)
 241         {
 242                 gistfreestack(p->s_stack);
 243                 gistfreestack(p->s_markstk);
 244                 pfree(s->opaque);
 245         }
 246
 247         gistdropscan(s);
 248         /* XXX don't unset read lock -- two-phase locking */
 249
 250         PG_RETURN_VOID();
 251 }
 252
 253 static void
 254 gistregscan(IndexScanDesc s)
 255 {
 256         GISTScanList l;
 257
 258         l = (GISTScanList) palloc(sizeof(GISTScanListData));
 259         l->gsl_scan = s;
 260         l->gsl_next = GISTScans;
 261         GISTScans = l;
 262 }
 263
 264 static void
 265 gistdropscan(IndexScanDesc s)
 266 {
 267         GISTScanList l;
 268         GISTScanList prev;
 269
 270         prev = (GISTScanList) NULL;
 271
 272         for (l = GISTScans;
 273                  l != (GISTScanList) NULL && l->gsl_scan != s;
 274                  l = l->gsl_next)
 275                 prev = l;
 276
 277         if (l == (GISTScanList) NULL)
 278                 elog(ERROR, "GiST scan list corrupted -- cannot find 0x%p", (void *) s);
 279
 280         if (prev == (GISTScanList) NULL)
 281                 GISTScans = l->gsl_next;
 282         else
 283                 prev->gsl_next = l->gsl_next;
 284
 285         pfree(l);
 286 }
 287
 288 void
 289 gistadjscans(Relation rel, int op, BlockNumber blkno, OffsetNumber offnum)
 290 {
 291         GISTScanList l;
 292         Oid                     relid;
 293
 294         relid = RelationGetRelid(rel);
 295         for (l = GISTScans; l != (GISTScanList) NULL; l = l->gsl_next)
 296         {
 297                 if (l->gsl_scan->relation->rd_id == relid)
 298                         gistadjone(l->gsl_scan, op, blkno, offnum);
 299         }
 300 }
 301
 302 /*
 303  *      gistadjone() -- adjust one scan for update.
 304  *
 305  *              By here, the scan passed in is on a modified relation.  Op tells
 306  *              us what the modification is, and blkno and offind tell us what
 307  *              block and offset index were affected.  This routine checks the
 308  *              current and marked positions, and the current and marked stacks,
 309  *              to see if any stored location needs to be changed because of the
 310  *              update.  If so, we make the change here.
 311  */
 312 static void
 313 gistadjone(IndexScanDesc s,
 314                    int op,
 315                    BlockNumber blkno,
 316                    OffsetNumber offnum)
 317 {
 318         GISTScanOpaque so;
 319
 320         adjustiptr(s, &(s->currentItemData), op, blkno, offnum);
 321         adjustiptr(s, &(s->currentMarkData), op, blkno, offnum);
 322
 323         so = (GISTScanOpaque) s->opaque;
 324
 325         if (op == GISTOP_SPLIT)
 326         {
 327                 adjuststack(so->s_stack, blkno, offnum);
 328                 adjuststack(so->s_markstk, blkno, offnum);
 329         }
 330 }
 331
 332 /*
 333  *      adjustiptr() -- adjust current and marked item pointers in the scan
 334  *
 335  *              Depending on the type of update and the place it happened, we
 336  *              need to do nothing, to back up one record, or to start over on
 337  *              the same page.
 338  */
 339 static void
 340 adjustiptr(IndexScanDesc s,
 341                    ItemPointer iptr,
 342                    int op,
 343                    BlockNumber blkno,
 344                    OffsetNumber offnum)
 345 {
 346         OffsetNumber curoff;
 347         GISTScanOpaque so;
 348
 349         if (ItemPointerIsValid(iptr))
 350         {
 351                 if (ItemPointerGetBlockNumber(iptr) == blkno)
 352                 {
 353                         curoff = ItemPointerGetOffsetNumber(iptr);
 354                         so = (GISTScanOpaque) s->opaque;
 355
 356                         switch (op)
 357                         {
 358                                 case GISTOP_DEL:
 359                                         /* back up one if we need to */
 360                                         if (curoff >= offnum)
 361                                         {
 362
 363                                                 if (curoff > FirstOffsetNumber)
 364                                                 {
 365                                                         /* just adjust the item pointer */
 366                                                         ItemPointerSet(iptr, blkno, OffsetNumberPrev(curoff));
 367                                                 }
 368                                                 else
 369                                                 {
 370
 371                                                         /*
 372                                                          * remember that we're before the current
 373                                                          * tuple
 374                                                          */
 375                                                         ItemPointerSet(iptr, blkno, FirstOffsetNumber);
 376                                                         if (iptr == &(s->currentItemData))
 377                                                                 so->s_flags |= GS_CURBEFORE;
 378                                                         else
 379                                                                 so->s_flags |= GS_MRKBEFORE;
 380                                                 }
 381                                         }
 382                                         break;
 383
 384                                 case GISTOP_SPLIT:
 385                                         /* back to start of page on split */
 386                                         ItemPointerSet(iptr, blkno, FirstOffsetNumber);
 387                                         if (iptr == &(s->currentItemData))
 388                                                 so->s_flags &= ~GS_CURBEFORE;
 389                                         else
 390                                                 so->s_flags &= ~GS_MRKBEFORE;
 391                                         break;
 392
 393                                 default:
 394                                         elog(ERROR, "Bad operation in GiST scan adjust: %d", op);
 395                         }
 396                 }
 397         }
 398 }
 399
 400 /*
 401  *      adjuststack() -- adjust the supplied stack for a split on a page in
 402  *                                       the index we're scanning.
 403  *
 404  *              If a page on our parent stack has split, we need to back up to the
 405  *              beginning of the page and rescan it.  The reason for this is that
 406  *              the split algorithm for GiSTs doesn't order tuples in any useful
 407  *              way on a single page.  This means on that a split, we may wind up
 408  *              looking at some heap tuples more than once.  This is handled in the
 409  *              access method update code for heaps; if we've modified the tuple we
 410  *              are looking at already in this transaction, we ignore the update
 411  *              request.
 412  */
 413 /*ARGSUSED*/
 414 static void
 415 adjuststack(GISTSTACK *stk,
 416                         BlockNumber blkno,
 417                         OffsetNumber offnum)
 418 {
 419         while (stk != (GISTSTACK *) NULL)
 420         {
 421                 if (stk->gs_blk == blkno)
 422                         stk->gs_child = FirstOffsetNumber;
 423
 424                 stk = stk->gs_parent;
 425         }
 426 }