src/backend/access/gist/gist.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * gist.c--
   4  *    interface routines for the postgres GiST index access method.
   5  *
   6  *
   7  *
   8  * IDENTIFICATION
   9  *
  10  *
  11  *-------------------------------------------------------------------------
  12  */
  13 #include "postgres.h"
  14
  15 #include "storage/bufmgr.h"
  16 #include "storage/bufpage.h"
  17
  18 #include "utils/elog.h"
  19 #include "utils/palloc.h"
  20 #include "utils/rel.h"
  21 #include "utils/excid.h"
  22
  23 #include "access/heapam.h"
  24 #include "access/genam.h"
  25 #include "access/gist.h"
  26 #include "access/gistscan.h"
  27 #include "access/funcindex.h"
  28 #include "access/tupdesc.h"
  29
  30 #include "nodes/execnodes.h"
  31 #include "nodes/plannodes.h"
  32
  33 #include "executor/executor.h"
  34 #include "executor/tuptable.h"
  35
  36 #include "catalog/index.h"
  37
  38 /* non-export function prototypes */
  39 static InsertIndexResult gistdoinsert(Relation r, IndexTuple itup,
  40                                     GISTSTATE *GISTstate);
  41 static InsertIndexResult gistentryinsert(Relation r, GISTSTACK *stk,
  42                                          IndexTuple tup,
  43                                          GISTSTATE *giststate);
  44 static void gistentryinserttwo(Relation r, GISTSTACK *stk, IndexTuple ltup,
  45                                IndexTuple rtup, GISTSTATE *giststate);
  46 static void gistAdjustKeys(Relation r, GISTSTACK *stk, BlockNumber blk,
  47                            char *datum, int att_size, GISTSTATE *giststate);
  48 static void gistintinsert(Relation r, GISTSTACK *stk, IndexTuple ltup,
  49                         IndexTuple rtup, GISTSTATE *giststate);
  50 static InsertIndexResult gistSplit(Relation r, Buffer buffer,
  51                                      GISTSTACK *stack, IndexTuple itup,
  52                                      GISTSTATE *giststate);
  53 static void gistnewroot(GISTSTATE *giststate, Relation r, IndexTuple lt,
  54                         IndexTuple rt);
  55 static void GISTInitBuffer(Buffer b, uint32 f);
  56 static BlockNumber gistChooseSubtree(Relation r, IndexTuple itup, int level,
  57                                      GISTSTATE *giststate,
  58                                      GISTSTACK **retstack, Buffer *leafbuf);
  59 static OffsetNumber gistchoose(Relation r, Page p, IndexTuple it,
  60                            GISTSTATE *giststate);
  61 static int gistnospace(Page p, IndexTuple it);
  62 void gistdelete(Relation r, ItemPointer tid);
  63 static IndexTuple gist_tuple_replacekey(Relation r, GISTENTRY entry, IndexTuple t);
  64
  65
  66 /*
  67 ** routine to build an index.  Basically calls insert over and over
  68 */
  69 void
  70 gistbuild(Relation heap,
  71           Relation index,
  72           int natts,
  73           AttrNumber *attnum,
  74           IndexStrategy istrat,
  75           uint16 pint,
  76           Datum *params,
  77           FuncIndexInfo *finfo,
  78           PredInfo *predInfo)
  79 {
  80     HeapScanDesc scan;
  81     Buffer buffer;
  82     AttrNumber i;
  83     HeapTuple htup;
  84     IndexTuple itup;
  85     TupleDesc hd, id;
  86     InsertIndexResult res;
  87     Datum *d;
  88     bool *nulls;
  89     int nb, nh, ni;
  90     ExprContext *econtext;
  91     TupleTable tupleTable;
  92     TupleTableSlot *slot;
  93     Oid hrelid, irelid;
  94     Node *pred, *oldPred;
  95     GISTSTATE giststate;
  96     GISTENTRY tmpcentry;
  97     bool *compvec;
  98
  99     /* GiSTs only know how to do stupid locking now */
 100     RelationSetLockForWrite(index);
 101
 102     setheapoverride(TRUE); /* so we can see the new pg_index tuple */
 103     initGISTstate(&giststate, index);
 104     setheapoverride(FALSE);
 105
 106     pred = predInfo->pred;
 107     oldPred = predInfo->oldPred;
 108
 109     /*
 110      *  We expect to be called exactly once for any index relation.
 111      *  If that's not the case, big trouble's what we have.
 112      */
 113
 114     if (oldPred == NULL && (nb = RelationGetNumberOfBlocks(index)) != 0)
 115         elog(WARN, "%.16s already contains data", &(index->rd_rel->relname.data[0]));
 116
 117     /* initialize the root page (if this is a new index) */
 118     if (oldPred == NULL) {
 119         buffer = ReadBuffer(index, P_NEW);
 120         GISTInitBuffer(buffer, F_LEAF);
 121         WriteBuffer(buffer);
 122     }
 123
 124     /* init the tuple descriptors and get set for a heap scan */
 125     hd = RelationGetTupleDescriptor(heap);
 126     id = RelationGetTupleDescriptor(index);
 127     d = (Datum *)palloc(natts * sizeof (*d));
 128     nulls = (bool *)palloc(natts * sizeof (*nulls));
 129
 130     /*
 131      * If this is a predicate (partial) index, we will need to evaluate the
 132      * predicate using ExecQual, which requires the current tuple to be in a
 133      * slot of a TupleTable.  In addition, ExecQual must have an ExprContext
 134      * referring to that slot.  Here, we initialize dummy TupleTable and
 135      * ExprContext objects for this purpose. --Nels, Feb '92
 136      */
 137 #ifndef OMIT_PARTIAL_INDEX
 138     if (pred != NULL || oldPred != NULL) {
 139         tupleTable = ExecCreateTupleTable(1);
 140         slot =  ExecAllocTableSlot(tupleTable);
 141         econtext = makeNode(ExprContext);
 142         FillDummyExprContext(econtext, slot, hd, buffer);
 143     }
 144 #endif /* OMIT_PARTIAL_INDEX */
 145     scan = heap_beginscan(heap, 0, NowTimeQual, 0, (ScanKey) NULL);
 146     htup = heap_getnext(scan, 0, &buffer);
 147
 148     /* int the tuples as we insert them */
 149     nh = ni = 0;
 150
 151     for (; HeapTupleIsValid(htup); htup = heap_getnext(scan, 0, &buffer)) {
 152
 153         nh++;
 154
 155         /*
 156          * If oldPred != NULL, this is an EXTEND INDEX command, so skip
 157          * this tuple if it was already in the existing partial index
 158          */
 159         if (oldPred != NULL) {
 160 #ifndef OMIT_PARTIAL_INDEX
 161             /*SetSlotContents(slot, htup); */
 162             slot->val = htup;
 163             if (ExecQual((List*)oldPred, econtext) == true) {
 164                 ni++;
 165                 continue;
 166             }
 167 #endif /* OMIT_PARTIAL_INDEX */
 168         }
 169
 170         /* Skip this tuple if it doesn't satisfy the partial-index predicate */
 171         if (pred != NULL) {
 172 #ifndef OMIT_PARTIAL_INDEX
 173             /*SetSlotContents(slot, htup); */
 174             slot->val = htup;
 175             if (ExecQual((List*)pred, econtext) == false)
 176                 continue;
 177 #endif /* OMIT_PARTIAL_INDEX */
 178         }
 179
 180         ni++;
 181
 182         /*
 183          *  For the current heap tuple, extract all the attributes
 184          *  we use in this index, and note which are null.
 185          */
 186
 187         for (i = 1; i <= natts; i++) {
 188             int  attoff;
 189             bool attnull;
 190
 191             /*
 192              *  Offsets are from the start of the tuple, and are
 193              *  zero-based; indices are one-based.  The next call
 194              *  returns i - 1.  That's data hiding for you.
 195              */
 196
 197             attoff = AttrNumberGetAttrOffset(i);
 198             /*
 199               d[attoff] = HeapTupleGetAttributeValue(htup, buffer,
 200               */
 201             d[attoff] = GetIndexValue(htup,
 202                                       hd,
 203                                       attoff,
 204                                       attnum,
 205                                       finfo,
 206                                       &attnull,
 207                                       buffer);
 208             nulls[attoff] = (attnull ? 'n' : ' ');
 209         }
 210
 211         /* immediately compress keys to normalize */
 212         compvec = (bool *)palloc(sizeof(bool) * natts);
 213         for (i = 0; i < natts; i++) {
 214             gistcentryinit(&giststate, &tmpcentry, (char *)d[i],
 215                            (Relation) NULL, (Page) NULL, (OffsetNumber) 0,
 216                            -1 /* size is currently bogus */, TRUE);
 217             if (d[i] != (Datum)tmpcentry.pred && !(giststate.keytypbyval))
 218               compvec[i] = TRUE;
 219             else compvec[i] = FALSE;
 220             d[i] = (Datum)tmpcentry.pred;
 221         }
 222
 223         /* form an index tuple and point it at the heap tuple */
 224         itup = index_formtuple(id, &d[0], nulls);
 225         itup->t_tid = htup->t_ctid;
 226
 227         /*
 228          *  Since we already have the index relation locked, we
 229          *  call gistdoinsert directly.  Normal access method calls
 230          *  dispatch through gistinsert, which locks the relation
 231          *  for write.  This is the right thing to do if you're
 232          *  inserting single tups, but not when you're initializing
 233          *  the whole index at once.
 234          */
 235
 236         res = gistdoinsert(index, itup, &giststate);
 237         for (i = 0; i < natts; i++)
 238           if (compvec[i] == TRUE) pfree((char *)d[i]);
 239         pfree(itup);
 240         pfree(res);
 241         pfree(compvec);
 242     }
 243
 244     /* okay, all heap tuples are indexed */
 245     heap_endscan(scan);
 246     RelationUnsetLockForWrite(index);
 247
 248     if (pred != NULL || oldPred != NULL) {
 249 #ifndef OMIT_PARTIAL_INDEX
 250         ExecDestroyTupleTable(tupleTable, true);
 251         pfree(econtext);
 252 #endif /* OMIT_PARTIAL_INDEX */
 253     }
 254
 255     /*
 256      *  Since we just inted the tuples in the heap, we update its
 257      *  stats in pg_relation to guarantee that the planner takes
 258      *  advantage of the index we just created.  UpdateStats() does a
 259      *  CommandinterIncrement(), which flushes changed entries from
 260      *  the system relcache.  The act of constructing an index changes
 261      *  these heap and index tuples in the system catalogs, so they
 262      *  need to be flushed.  We close them to guarantee that they
 263      *  will be.
 264      */
 265
 266     hrelid = heap->rd_id;
 267     irelid = index->rd_id;
 268     heap_close(heap);
 269     index_close(index);
 270
 271     UpdateStats(hrelid, nh, true);
 272     UpdateStats(irelid, ni, false);
 273
 274     if (oldPred != NULL) {
 275         if (ni == nh) pred = NULL;
 276         UpdateIndexPredicate(irelid, oldPred, pred);
 277     }
 278
 279     /* be tidy */
 280     pfree(nulls);
 281     pfree(d);
 282 }
 283
 284 /*
 285  *  gistinsert -- wrapper for GiST tuple insertion.
 286  *
 287  *    This is the public interface routine for tuple insertion in GiSTs.
 288  *    It doesn't do any work; just locks the relation and passes the buck.
 289  */
 290 InsertIndexResult
 291 gistinsert(Relation r, Datum *datum, char *nulls, ItemPointer ht_ctid)
 292 {
 293     InsertIndexResult res;
 294     IndexTuple itup;
 295     GISTSTATE giststate;
 296     GISTENTRY tmpentry;
 297     int i;
 298     bool *compvec;
 299
 300     initGISTstate(&giststate, r);
 301
 302     /* immediately compress keys to normalize */
 303     compvec = (bool *)palloc(sizeof(bool) * r->rd_att->natts);
 304     for (i = 0; i < r->rd_att->natts; i++) {
 305         gistcentryinit(&giststate, &tmpentry, (char *)datum[i],
 306                        (Relation) NULL, (Page) NULL, (OffsetNumber) 0,
 307                        -1 /* size is currently bogus */, TRUE);
 308         if (datum[i] != (Datum)tmpentry.pred && !(giststate.keytypbyval))
 309           compvec[i] = TRUE;
 310         else compvec[i] = FALSE;
 311         datum[i] = (Datum)tmpentry.pred;
 312     }
 313     itup = index_formtuple(RelationGetTupleDescriptor(r), datum, nulls);
 314     itup->t_tid = *ht_ctid;
 315
 316     RelationSetLockForWrite(r);
 317     res = gistdoinsert(r, itup, &giststate);
 318     for (i = 0; i < r->rd_att->natts; i++)
 319       if (compvec[i] == TRUE) pfree((char *)datum[i]);
 320     pfree(itup);
 321     pfree(compvec);
 322
 323     /* XXX two-phase locking -- don't unlock the relation until EOT */
 324     return (res);
 325 }
 326
 327 /*
 328 ** Take a compressed entry, and install it on a page.  Since we now know
 329 ** where the entry will live, we decompress it and recompress it using
 330 ** that knowledge (some compression routines may want to fish around
 331 ** on the page, for example, or do something special for leaf nodes.)
 332 */
 333 static OffsetNumber
 334 gistPageAddItem(GISTSTATE *giststate,
 335                 Relation r,
 336                 Page page,
 337                 Item item,
 338                 Size size,
 339                 OffsetNumber offsetNumber,
 340                 ItemIdFlags flags,
 341                 GISTENTRY *dentry,
 342                 IndexTuple *newtup)
 343 {
 344     GISTENTRY tmpcentry;
 345     IndexTuple itup = (IndexTuple)item;
 346
 347     /* recompress the item given that we now know the exact page and
 348        offset for insertion */
 349     gistdentryinit(giststate, dentry,
 350                    (((char *) itup) + sizeof(IndexTupleData)),
 351                    (Relation)0, (Page)0, (OffsetNumber)InvalidOffsetNumber,
 352                    IndexTupleSize(itup) - sizeof(IndexTupleData), FALSE);
 353     gistcentryinit(giststate, &tmpcentry, dentry->pred, r, page,
 354                    offsetNumber, dentry->bytes, FALSE);
 355     *newtup = gist_tuple_replacekey(r, *dentry, itup);
 356     /* be tidy */
 357     if (tmpcentry.pred != dentry->pred
 358         && tmpcentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
 359       pfree(tmpcentry.pred);
 360
 361     return(PageAddItem(page, (Item) *newtup, IndexTupleSize(*newtup),
 362                        offsetNumber, flags));
 363 }
 364
 365
 366 static InsertIndexResult
 367 gistdoinsert(Relation r,
 368              IndexTuple itup, /* itup contains compressed entry */
 369              GISTSTATE *giststate)
 370 {
 371     char *datum, *newdatum;
 372     GISTENTRY entry, tmpdentry;
 373     InsertIndexResult res;
 374     OffsetNumber l;
 375     GISTSTACK *stack, *tmpstk;
 376     Buffer buffer;
 377     BlockNumber blk;
 378     Page page;
 379     OffsetNumber off;
 380     IndexTuple newtup;
 381
 382     /* 3rd arg is ignored for now */
 383     blk = gistChooseSubtree(r, itup, 0, giststate, &stack, &buffer);
 384     page = (Page) BufferGetPage(buffer);
 385
 386     if (gistnospace(page, itup)) {
 387         /* need to do a split */
 388         res = gistSplit(r, buffer, stack, itup, giststate);
 389         gistfreestack(stack);
 390         WriteBuffer(buffer);  /* don't forget to release buffer! */
 391         return (res);
 392     }
 393
 394     if (PageIsEmpty(page))
 395       off = FirstOffsetNumber;
 396     else
 397       off = OffsetNumberNext(PageGetMaxOffsetNumber(page));
 398
 399     /* add the item and write the buffer */
 400     l = gistPageAddItem(giststate, r, page, (Item) itup, IndexTupleSize(itup),
 401                         off, LP_USED, &tmpdentry, &newtup);
 402     WriteBuffer(buffer);
 403
 404     /* now expand the page boundary in the parent to include the new child */
 405     gistAdjustKeys(r, stack, blk, tmpdentry.pred, tmpdentry.bytes, giststate);
 406     gistfreestack(stack);
 407
 408     /* be tidy */
 409     if (itup != newtup)
 410       pfree(newtup);
 411     if (tmpdentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
 412       pfree(tmpdentry.pred);
 413
 414     /* build and return an InsertIndexResult for this insertion */
 415     res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
 416     ItemPointerSet(&(res->pointerData), blk, l);
 417
 418     return (res);
 419 }
 420
 421
 422 static BlockNumber
 423 gistChooseSubtree(Relation r, IndexTuple itup, /* itup has compressed entry */
 424                   int level,
 425                   GISTSTATE *giststate,
 426                   GISTSTACK **retstack /*out*/,
 427                   Buffer *leafbuf /*out*/)
 428 {
 429   Buffer buffer;
 430   BlockNumber blk;
 431   GISTSTACK *stack;
 432   Page page;
 433   GISTPageOpaque opaque;
 434   IndexTuple which;
 435
 436   blk = GISTP_ROOT;
 437   buffer = InvalidBuffer;
 438   stack = (GISTSTACK *) NULL;
 439
 440   do {
 441     /* let go of current buffer before getting next */
 442     if (buffer != InvalidBuffer)
 443       ReleaseBuffer(buffer);
 444
 445     /* get next buffer */
 446     buffer = ReadBuffer(r, blk);
 447     page = (Page) BufferGetPage(buffer);
 448
 449     opaque = (GISTPageOpaque) PageGetSpecialPointer(page);
 450     if (!(opaque->flags & F_LEAF)) {
 451       GISTSTACK *n;
 452       ItemId iid;
 453
 454       n = (GISTSTACK *) palloc(sizeof(GISTSTACK));
 455       n->gs_parent = stack;
 456       n->gs_blk = blk;
 457       n->gs_child = gistchoose(r, page, itup, giststate);
 458       stack = n;
 459
 460       iid = PageGetItemId(page, n->gs_child);
 461       which = (IndexTuple) PageGetItem(page, iid);
 462       blk = ItemPointerGetBlockNumber(&(which->t_tid));
 463     }
 464   } while (!(opaque->flags & F_LEAF));
 465
 466   *retstack = stack;
 467   *leafbuf = buffer;
 468
 469   return(blk);
 470 }
 471
 472
 473 static void
 474 gistAdjustKeys(Relation r,
 475             GISTSTACK *stk,
 476             BlockNumber blk,
 477             char *datum,       /* datum is uncompressed */
 478             int att_size,
 479             GISTSTATE *giststate)
 480 {
 481     char *oldud;
 482     Page p;
 483     Buffer b;
 484     bool result;
 485     bytea *evec;
 486     GISTENTRY centry, *ev0p, *ev1p, *dentryp;
 487     int size, datumsize;
 488     IndexTuple tid;
 489
 490     if (stk == (GISTSTACK *) NULL)
 491         return;
 492
 493     b = ReadBuffer(r, stk->gs_blk);
 494     p = BufferGetPage(b);
 495
 496     oldud = (char *) PageGetItem(p, PageGetItemId(p, stk->gs_child));
 497     tid = (IndexTuple) oldud;
 498     size = IndexTupleSize((IndexTuple)oldud) - sizeof(IndexTupleData);
 499     oldud += sizeof(IndexTupleData);
 500
 501     evec = (bytea *) palloc(2*sizeof(GISTENTRY) + VARHDRSZ);
 502     VARSIZE(evec) = 2*sizeof(GISTENTRY) + VARHDRSZ;
 503
 504     /* insert decompressed oldud into entry vector */
 505     gistdentryinit(giststate, &((GISTENTRY *)VARDATA(evec))[0],
 506                    oldud, r, p, stk->gs_child,
 507                    size, FALSE);
 508     ev0p = &((GISTENTRY *)VARDATA(evec))[0];
 509
 510     /* insert datum entry into entry vector */
 511     gistentryinit(((GISTENTRY *)VARDATA(evec))[1], datum,
 512                   (Relation)NULL,(Page)NULL,(OffsetNumber)0, att_size, FALSE);
 513     ev1p = &((GISTENTRY *)VARDATA(evec))[1];
 514
 515     /* form union of decompressed entries */
 516     datum = (char *) (giststate->unionFn)(evec, &datumsize);
 517
 518     /* did union leave decompressed version of oldud unchanged? */
 519     (giststate->equalFn)(ev0p->pred, datum, &result);
 520     if (!result) {
 521         TupleDesc td = RelationGetTupleDescriptor(r);
 522
 523         /* compress datum for storage on page */
 524         gistcentryinit(giststate, &centry, datum, ev0p->rel, ev0p->page,
 525                        ev0p->offset, datumsize, FALSE);
 526         if (td->attrs[0]->attlen >= 0) {
 527           memmove(oldud, centry.pred, att_size);
 528           gistAdjustKeys(r, stk->gs_parent, stk->gs_blk, datum, att_size,
 529                          giststate);
 530         }
 531         else if (VARSIZE(centry.pred) == VARSIZE(oldud)) {
 532           memmove(oldud, centry.pred, VARSIZE(centry.pred));
 533           gistAdjustKeys(r, stk->gs_parent, stk->gs_blk, datum, att_size,
 534                          giststate);
 535         }
 536         else {
 537           /*
 538           ** new datum is not the same size as the old.
 539           ** We have to delete the old entry and insert the new
 540           ** one.  Note that this may cause a split here!
 541           */
 542           IndexTuple newtup;
 543           ItemPointerData oldtid;
 544           char *isnull;
 545           TupleDesc tupDesc;
 546           InsertIndexResult res;
 547
 548           /* delete old tuple */
 549           ItemPointerSet(&oldtid, stk->gs_blk, stk->gs_child);
 550           gistdelete(r, (ItemPointer)&oldtid);
 551
 552           /* generate and insert new tuple */
 553           tupDesc = r->rd_att;
 554           isnull = (char *) palloc(r->rd_rel->relnatts);
 555           memset(isnull, ' ', r->rd_rel->relnatts);
 556           newtup = (IndexTuple) index_formtuple(tupDesc,
 557                                                 (Datum *) &centry.pred, isnull);
 558           pfree(isnull);
 559           /* set pointer in new tuple to point to current child */
 560           ItemPointerSet(&oldtid, blk, 1);
 561           newtup->t_tid = oldtid;
 562
 563           /* inserting the new entry also adjust keys above */
 564           res = gistentryinsert(r, stk, newtup, giststate);
 565
 566           /* in stack, set info to point to new tuple */
 567           stk->gs_blk = ItemPointerGetBlockNumber(&(res->pointerData));
 568           stk->gs_child = ItemPointerGetOffsetNumber(&(res->pointerData));
 569
 570           pfree(res);
 571         }
 572         WriteBuffer(b);
 573
 574         if (centry.pred != datum)
 575           pfree(datum);
 576     }
 577     else {
 578         ReleaseBuffer(b);
 579     }
 580     pfree(evec);
 581 }
 582
 583 /*
 584  *  gistSplit -- split a page in the tree.
 585  *
 586  */
 587 static InsertIndexResult
 588 gistSplit(Relation r,
 589             Buffer buffer,
 590             GISTSTACK *stack,
 591             IndexTuple itup,  /* contains compressed entry */
 592             GISTSTATE *giststate)
 593 {
 594     Page p;
 595     Buffer leftbuf, rightbuf;
 596     Page left, right;
 597     ItemId itemid;
 598     IndexTuple item;
 599     IndexTuple ltup, rtup, newtup;
 600     OffsetNumber maxoff;
 601     OffsetNumber i;
 602     OffsetNumber leftoff, rightoff;
 603     BlockNumber lbknum, rbknum;
 604     BlockNumber bufblock;
 605     GISTPageOpaque opaque;
 606     int blank;
 607     InsertIndexResult res;
 608     char *isnull;
 609     GIST_SPLITVEC v;
 610     TupleDesc tupDesc;
 611     bytea *entryvec;
 612     bool *decompvec;
 613     IndexTuple item_1;
 614     GISTENTRY tmpdentry, tmpentry;
 615     char *datum;
 616
 617     isnull = (char *) palloc(r->rd_rel->relnatts);
 618     for (blank = 0; blank < r->rd_rel->relnatts; blank++)
 619         isnull[blank] = ' ';
 620     p = (Page) BufferGetPage(buffer);
 621     opaque = (GISTPageOpaque) PageGetSpecialPointer(p);
 622
 623
 624     /*
 625      *  The root of the tree is the first block in the relation.  If
 626      *  we're about to split the root, we need to do some hocus-pocus
 627      *  to enforce this guarantee.
 628      */
 629
 630     if (BufferGetBlockNumber(buffer) == GISTP_ROOT) {
 631         leftbuf = ReadBuffer(r, P_NEW);
 632         GISTInitBuffer(leftbuf, opaque->flags);
 633         lbknum = BufferGetBlockNumber(leftbuf);
 634         left = (Page) BufferGetPage(leftbuf);
 635     } else {
 636         leftbuf = buffer;
 637         IncrBufferRefCount(buffer);
 638         lbknum = BufferGetBlockNumber(buffer);
 639         left = (Page) PageGetTempPage(p, sizeof(GISTPageOpaqueData));
 640     }
 641
 642     rightbuf = ReadBuffer(r, P_NEW);
 643     GISTInitBuffer(rightbuf, opaque->flags);
 644     rbknum = BufferGetBlockNumber(rightbuf);
 645     right = (Page) BufferGetPage(rightbuf);
 646
 647     /* generate the item array */
 648     maxoff = PageGetMaxOffsetNumber(p);
 649     entryvec = (bytea *)palloc(VARHDRSZ + (maxoff + 2) * sizeof(GISTENTRY));
 650     decompvec = (bool *)palloc(VARHDRSZ + (maxoff + 2) * sizeof(bool));
 651     for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i)) {
 652         item_1 = (IndexTuple) PageGetItem(p, PageGetItemId(p, i));
 653         gistdentryinit(giststate, &((GISTENTRY *)VARDATA(entryvec))[i],
 654                        (((char *) item_1) + sizeof(IndexTupleData)),
 655                        r, p, i,
 656                        IndexTupleSize(item_1) - sizeof(IndexTupleData), FALSE);
 657         if ((char *)(((GISTENTRY *)VARDATA(entryvec))[i].pred)
 658             == (((char *) item_1) + sizeof(IndexTupleData)))
 659           decompvec[i] = FALSE;
 660         else decompvec[i] = TRUE;
 661     }
 662
 663     /* add the new datum as the last entry */
 664     gistdentryinit(giststate, &(((GISTENTRY *)VARDATA(entryvec))[maxoff+1]),
 665                    (((char *) itup) + sizeof(IndexTupleData)),
 666                    (Relation)NULL, (Page)NULL,
 667                    (OffsetNumber)0, tmpentry.bytes, FALSE);
 668     if ((char *)(((GISTENTRY *)VARDATA(entryvec))[maxoff+1]).pred !=
 669         (((char *) itup) + sizeof(IndexTupleData)))
 670       decompvec[maxoff+1] = TRUE;
 671     else decompvec[maxoff+1] = FALSE;
 672
 673     VARSIZE(entryvec) = (maxoff + 2) * sizeof(GISTENTRY) + VARHDRSZ;
 674
 675     /* now let the user-defined picksplit function set up the split vector */
 676     (giststate->picksplitFn)(entryvec, &v);
 677
 678     /* compress ldatum and rdatum */
 679     gistcentryinit(giststate, &tmpentry, v.spl_ldatum, (Relation)NULL,
 680                    (Page)NULL, (OffsetNumber)0,
 681                    ((GISTENTRY *)VARDATA(entryvec))[i].bytes, FALSE);
 682     if (v.spl_ldatum != tmpentry.pred)
 683       pfree(v.spl_ldatum);
 684     v.spl_ldatum = tmpentry.pred;
 685
 686     gistcentryinit(giststate, &tmpentry, v.spl_rdatum, (Relation)NULL,
 687                    (Page)NULL, (OffsetNumber)0,
 688                    ((GISTENTRY *)VARDATA(entryvec))[i].bytes, FALSE);
 689     if (v.spl_rdatum != tmpentry.pred)
 690       pfree(v.spl_rdatum);
 691     v.spl_rdatum = tmpentry.pred;
 692
 693     /* clean up the entry vector: its preds need to be deleted, too */
 694     for (i = FirstOffsetNumber; i <= maxoff+1; i = OffsetNumberNext(i))
 695       if (decompvec[i])
 696         pfree(((GISTENTRY *)VARDATA(entryvec))[i].pred);
 697     pfree(entryvec);
 698     pfree(decompvec);
 699
 700     leftoff = rightoff = FirstOffsetNumber;
 701     maxoff = PageGetMaxOffsetNumber(p);
 702     for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i)) {
 703         itemid = PageGetItemId(p, i);
 704         item = (IndexTuple) PageGetItem(p, itemid);
 705
 706         if (i == *(v.spl_left)) {
 707             (void) gistPageAddItem(giststate, r, left, (Item) item,
 708                                    IndexTupleSize(item),
 709                                    leftoff, LP_USED, &tmpdentry, &newtup);
 710             leftoff = OffsetNumberNext(leftoff);
 711             v.spl_left++;       /* advance in left split vector */
 712             /* be tidy */
 713             if (tmpdentry.pred != (((char *) item) + sizeof(IndexTupleData)))
 714               pfree(tmpdentry.pred);
 715             if ((IndexTuple)item != newtup)
 716               pfree(newtup);
 717         }
 718         else {
 719             (void) gistPageAddItem(giststate, r, right, (Item) item,
 720                                    IndexTupleSize(item),
 721                                    rightoff, LP_USED, &tmpdentry, &newtup);
 722             rightoff = OffsetNumberNext(rightoff);
 723             v.spl_right++;      /* advance in right split vector */
 724             /* be tidy */
 725             if (tmpdentry.pred != (((char *) item) + sizeof(IndexTupleData)))
 726               pfree(tmpdentry.pred);
 727             if (item != newtup)
 728               pfree(newtup);
 729         }
 730     }
 731
 732     /* build an InsertIndexResult for this insertion */
 733     res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
 734
 735     /* now insert the new index tuple */
 736     if (*(v.spl_left) != FirstOffsetNumber) {
 737         (void) gistPageAddItem(giststate, r, left, (Item) itup,
 738                                IndexTupleSize(itup),
 739                                leftoff, LP_USED, &tmpdentry, &newtup);
 740         leftoff = OffsetNumberNext(leftoff);
 741         ItemPointerSet(&(res->pointerData), lbknum, leftoff);
 742         /* be tidy */
 743         if (tmpdentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
 744           pfree(tmpdentry.pred);
 745         if (itup != newtup)
 746           pfree(newtup);
 747     } else {
 748         (void) gistPageAddItem(giststate, r, right, (Item) itup,
 749                                IndexTupleSize(itup),
 750                                rightoff, LP_USED, &tmpdentry, &newtup);
 751         rightoff = OffsetNumberNext(rightoff);
 752         ItemPointerSet(&(res->pointerData), rbknum, rightoff);
 753         /* be tidy */
 754         if (tmpdentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
 755           pfree(tmpdentry.pred);
 756         if (itup != newtup)
 757           pfree(newtup);
 758     }
 759
 760     if ((bufblock = BufferGetBlockNumber(buffer)) != GISTP_ROOT) {
 761         PageRestoreTempPage(left, p);
 762     }
 763     WriteBuffer(leftbuf);
 764     WriteBuffer(rightbuf);
 765
 766     /*
 767      *  Okay, the page is split.  We have three things left to do:
 768      *
 769      *    1)  Adjust any active scans on this index to cope with changes
 770      *        we introduced in its structure by splitting this page.
 771      *
 772      *    2)  "Tighten" the bounding box of the pointer to the left
 773      *        page in the parent node in the tree, if any.  Since we
 774      *        moved a bunch of stuff off the left page, we expect it
 775      *        to get smaller.  This happens in the internal insertion
 776      *        routine.
 777      *
 778      *    3)  Insert a pointer to the right page in the parent.  This
 779      *        may cause the parent to split.  If it does, we need to
 780      *        repeat steps one and two for each split node in the tree.
 781      */
 782
 783     /* adjust active scans */
 784     gistadjscans(r, GISTOP_SPLIT, bufblock, FirstOffsetNumber);
 785
 786     tupDesc = r->rd_att;
 787
 788     ltup = (IndexTuple) index_formtuple(tupDesc,
 789                                         (Datum *) &(v.spl_ldatum), isnull);
 790     rtup = (IndexTuple) index_formtuple(tupDesc,
 791                                         (Datum *) &(v.spl_rdatum), isnull);
 792     pfree(isnull);
 793
 794     /* set pointers to new child pages in the internal index tuples */
 795     ItemPointerSet(&(ltup->t_tid), lbknum, 1);
 796     ItemPointerSet(&(rtup->t_tid), rbknum, 1);
 797
 798     gistintinsert(r, stack, ltup, rtup, giststate);
 799
 800     pfree(ltup);
 801     pfree(rtup);
 802
 803     return (res);
 804 }
 805
 806 /*
 807 ** After a split, we need to overwrite the old entry's key in the parent,
 808 ** and install install an entry for the new key into the parent.
 809 */
 810 static void
 811 gistintinsert(Relation r,
 812               GISTSTACK *stk,
 813               IndexTuple ltup, /* new version of entry for old page */
 814               IndexTuple rtup, /* entry for new page */
 815               GISTSTATE *giststate)
 816 {
 817     IndexTuple old;
 818     Buffer b;
 819     Page p;
 820     ItemPointerData ltid;
 821
 822     if (stk == (GISTSTACK *) NULL) {
 823         gistnewroot(giststate, r, ltup, rtup);
 824         return;
 825     }
 826
 827     /* remove old left pointer, insert the 2 new entries */
 828     ItemPointerSet(&ltid, stk->gs_blk, stk->gs_child);
 829     gistdelete(r, (ItemPointer)&ltid);
 830     gistentryinserttwo(r, stk, ltup, rtup, giststate);
 831 }
 832
 833
 834 /*
 835 ** Insert two entries onto one page, handling a split for either one!
 836 */
 837 static void
 838 gistentryinserttwo(Relation r, GISTSTACK *stk, IndexTuple ltup,
 839                    IndexTuple rtup, GISTSTATE *giststate)
 840 {
 841     Buffer b;
 842     Page p;
 843     InsertIndexResult res;
 844     OffsetNumber off;
 845     bytea *evec;
 846     char *datum;
 847     int size;
 848     GISTENTRY tmpentry;
 849     IndexTuple newtup;
 850
 851     b = ReadBuffer(r, stk->gs_blk);
 852     p = BufferGetPage(b);
 853
 854     if (gistnospace(p, ltup)) {
 855         res = gistSplit(r, b, stk->gs_parent, ltup, giststate);
 856         WriteBuffer(b);  /* don't forget to release buffer!  - 01/31/94 */
 857         pfree(res);
 858         gistdoinsert(r, rtup, giststate);
 859     } else {
 860         (void) gistPageAddItem(giststate, r, p, (Item)ltup,
 861                                IndexTupleSize(ltup), InvalidOffsetNumber,
 862                                LP_USED, &tmpentry, &newtup);
 863         WriteBuffer(b);
 864         gistAdjustKeys(r, stk->gs_parent, stk->gs_blk, tmpentry.pred,
 865                        tmpentry.bytes, giststate);
 866         /* be tidy */
 867         if (tmpentry.pred != (((char *) ltup) + sizeof(IndexTupleData)))
 868           pfree(tmpentry.pred);
 869         if (ltup != newtup)
 870           pfree(newtup);
 871         (void)gistentryinsert(r, stk, rtup, giststate);
 872     }
 873 }
 874
 875
 876 /*
 877 ** Insert an entry onto a page
 878 */
 879 static InsertIndexResult
 880 gistentryinsert(Relation r, GISTSTACK *stk, IndexTuple tup,
 881                 GISTSTATE *giststate)
 882 {
 883     Buffer b;
 884     Page p;
 885     InsertIndexResult res;
 886     bytea *evec;
 887     char *datum;
 888     int size;
 889     OffsetNumber off;
 890     GISTENTRY tmpentry;
 891     IndexTuple newtup;
 892
 893     b = ReadBuffer(r, stk->gs_blk);
 894     p = BufferGetPage(b);
 895
 896     if (gistnospace(p, tup)) {
 897         res = gistSplit(r, b, stk->gs_parent, tup, giststate);
 898         WriteBuffer(b);  /* don't forget to release buffer!  - 01/31/94 */
 899         return(res);
 900     }
 901     else {
 902         res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
 903         off = gistPageAddItem(giststate, r, p, (Item) tup, IndexTupleSize(tup),
 904                               InvalidOffsetNumber, LP_USED, &tmpentry, &newtup);
 905         WriteBuffer(b);
 906         ItemPointerSet(&(res->pointerData), stk->gs_blk, off);
 907         gistAdjustKeys(r, stk->gs_parent, stk->gs_blk, tmpentry.pred,
 908                        tmpentry.bytes, giststate);
 909         /* be tidy */
 910         if (tmpentry.pred != (((char *) tup) + sizeof(IndexTupleData)))
 911           pfree(tmpentry.pred);
 912         if (tup != newtup)
 913           pfree(newtup);
 914         return(res);
 915     }
 916 }
 917
 918
 919 static void
 920 gistnewroot(GISTSTATE *giststate, Relation r, IndexTuple lt, IndexTuple rt)
 921 {
 922     Buffer b;
 923     Page p;
 924     GISTENTRY tmpentry;
 925     IndexTuple newtup;
 926
 927     b = ReadBuffer(r, GISTP_ROOT);
 928     GISTInitBuffer(b, 0);
 929     p = BufferGetPage(b);
 930     (void) gistPageAddItem(giststate, r, p, (Item) lt, IndexTupleSize(lt),
 931                            FirstOffsetNumber,
 932                            LP_USED, &tmpentry, &newtup);
 933     /* be tidy */
 934     if (tmpentry.pred != (((char *) lt) + sizeof(IndexTupleData)))
 935       pfree(tmpentry.pred);
 936     if (lt != newtup)
 937       pfree(newtup);
 938     (void) gistPageAddItem(giststate, r, p, (Item) rt, IndexTupleSize(rt),
 939                            OffsetNumberNext(FirstOffsetNumber), LP_USED,
 940                            &tmpentry, &newtup);
 941     /* be tidy */
 942     if (tmpentry.pred != (((char *) rt) + sizeof(IndexTupleData)))
 943       pfree(tmpentry.pred);
 944     if (rt != newtup)
 945       pfree(newtup);
 946     WriteBuffer(b);
 947 }
 948
 949 static void
 950 GISTInitBuffer(Buffer b, uint32 f)
 951 {
 952     GISTPageOpaque opaque;
 953     Page page;
 954     Size pageSize;
 955
 956     pageSize = BufferGetPageSize(b);
 957
 958     page = BufferGetPage(b);
 959     memset(page, 0, (int) pageSize);
 960     PageInit(page, pageSize, sizeof(GISTPageOpaqueData));
 961
 962     opaque = (GISTPageOpaque) PageGetSpecialPointer(page);
 963     opaque->flags = f;
 964 }
 965
 966
 967 /*
 968 ** find entry with lowest penalty
 969 */
 970 static OffsetNumber
 971 gistchoose(Relation r, Page p, IndexTuple it, /* it has compressed entry */
 972            GISTSTATE *giststate)
 973 {
 974     OffsetNumber maxoff;
 975     OffsetNumber i;
 976     char *ud, *id;
 977     char *datum;
 978     float usize, dsize;
 979     OffsetNumber which;
 980     float which_grow;
 981     GISTENTRY entry, identry;
 982     int size, idsize;
 983
 984     idsize = IndexTupleSize(it) - sizeof(IndexTupleData);
 985     id = ((char *) it) + sizeof(IndexTupleData);
 986     maxoff = PageGetMaxOffsetNumber(p);
 987     which_grow = -1.0;
 988     which = -1;
 989
 990     gistdentryinit(giststate,&identry,id,(Relation)NULL,(Page)NULL,
 991                    (OffsetNumber)0, idsize, FALSE);
 992
 993     for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i)) {
 994         datum = (char *) PageGetItem(p, PageGetItemId(p, i));
 995         size = IndexTupleSize(datum) - sizeof(IndexTupleData);
 996         datum += sizeof(IndexTupleData);
 997         gistdentryinit(giststate,&entry,datum,r,p,i,size,FALSE);
 998         (giststate->penaltyFn)(entry, identry, &usize);
 999         if (which_grow < 0 || usize < which_grow) {
1000             which = i;
1001             which_grow = usize;
1002             if (which_grow == 0)
1003                 break;
1004         }
1005         if (entry.pred != datum)
1006           pfree(entry.pred);
1007     }
1008     if (identry.pred != id)
1009       pfree(identry.pred);
1010
1011     return (which);
1012 }
1013
1014 static int
1015 gistnospace(Page p, IndexTuple it)
1016 {
1017     return (PageGetFreeSpace(p) < IndexTupleSize(it));
1018 }
1019
1020 void
1021 gistfreestack(GISTSTACK *s)
1022 {
1023     GISTSTACK *p;
1024
1025     while (s != (GISTSTACK *) NULL) {
1026         p = s->gs_parent;
1027         pfree(s);
1028         s = p;
1029     }
1030 }
1031
1032
1033 /*
1034 ** remove an entry from a page
1035 */
1036 void
1037 gistdelete(Relation r, ItemPointer tid)
1038 {
1039     BlockNumber blkno;
1040     OffsetNumber offnum;
1041     Buffer buf;
1042     Page page;
1043
1044     /* must write-lock on delete */
1045     RelationSetLockForWrite(r);
1046
1047     blkno = ItemPointerGetBlockNumber(tid);
1048     offnum = ItemPointerGetOffsetNumber(tid);
1049
1050     /* adjust any scans that will be affected by this deletion */
1051     gistadjscans(r, GISTOP_DEL, blkno, offnum);
1052
1053     /* delete the index tuple */
1054     buf = ReadBuffer(r, blkno);
1055     page = BufferGetPage(buf);
1056
1057     PageIndexTupleDelete(page, offnum);
1058
1059     WriteBuffer(buf);
1060
1061     /* XXX -- two-phase locking, don't release the write lock */
1062 }
1063
1064 void
1065 initGISTstate(GISTSTATE *giststate, Relation index)
1066 {
1067     RegProcedure consistent_proc, union_proc, compress_proc, decompress_proc;
1068     RegProcedure penalty_proc, picksplit_proc, equal_proc;
1069     func_ptr user_fn;
1070     int pronargs;
1071     HeapTuple htup;
1072     IndexTupleForm itupform;
1073
1074     consistent_proc = index_getprocid(index, 1, GIST_CONSISTENT_PROC);
1075     union_proc = index_getprocid(index, 1, GIST_UNION_PROC);
1076     compress_proc = index_getprocid(index, 1, GIST_COMPRESS_PROC);
1077     decompress_proc = index_getprocid(index, 1, GIST_DECOMPRESS_PROC);
1078     penalty_proc = index_getprocid(index, 1, GIST_PENALTY_PROC);
1079     picksplit_proc = index_getprocid(index, 1, GIST_PICKSPLIT_PROC);
1080     equal_proc = index_getprocid(index, 1, GIST_EQUAL_PROC);
1081     fmgr_info(consistent_proc, &user_fn, &pronargs);
1082     giststate->consistentFn = user_fn;
1083     fmgr_info(union_proc, &user_fn, &pronargs);
1084     giststate->unionFn = user_fn;
1085     fmgr_info(compress_proc, &user_fn, &pronargs);
1086     giststate->compressFn = user_fn;
1087     fmgr_info(decompress_proc, &user_fn, &pronargs);
1088     giststate->decompressFn = user_fn;
1089     fmgr_info(penalty_proc, &user_fn, &pronargs);
1090     giststate->penaltyFn = user_fn;
1091     fmgr_info(picksplit_proc, &user_fn, &pronargs);
1092     giststate->picksplitFn = user_fn;
1093     fmgr_info(equal_proc, &user_fn, &pronargs);
1094     giststate->equalFn = user_fn;
1095
1096     /* see if key type is different from type of attribute being indexed */
1097     htup = SearchSysCacheTuple(INDEXRELID, ObjectIdGetDatum(index->rd_id),
1098                                0,0,0);
1099     itupform = (IndexTupleForm)GETSTRUCT(htup);
1100     if (!HeapTupleIsValid(htup))
1101       elog(WARN, "initGISTstate: index %d not found", index->rd_id);
1102     giststate->haskeytype = itupform->indhaskeytype;
1103     if (giststate->haskeytype) {
1104         /* key type is different -- is it byval? */
1105         htup = SearchSysCacheTuple(ATTNUM,
1106                                    ObjectIdGetDatum(itupform->indexrelid),
1107                                    UInt16GetDatum(FirstOffsetNumber),
1108                                    0,0);
1109         if (!HeapTupleIsValid(htup)) {
1110           elog(WARN, "initGISTstate: no attribute tuple %d %d",
1111                itupform->indexrelid, FirstOffsetNumber);
1112           return;
1113         }
1114         giststate->keytypbyval = (((AttributeTupleForm)htup)->attbyval);
1115     }
1116     else
1117       giststate->keytypbyval = FALSE;
1118     return;
1119 }
1120
1121
1122 /*
1123 ** Given an IndexTuple to be inserted on a page, this routine replaces
1124 ** the key with another key, which may involve generating a new IndexTuple
1125 ** if the sizes don't match
1126 */
1127 static IndexTuple
1128 gist_tuple_replacekey(Relation r, GISTENTRY entry, IndexTuple t)
1129 {
1130   char * datum  = (((char *) t) + sizeof(IndexTupleData));
1131
1132   /* if new entry fits in index tuple, copy it in */
1133   if (entry.bytes < IndexTupleSize(t) - sizeof(IndexTupleData)) {
1134     memcpy(datum, entry.pred, entry.bytes);
1135     /* clear out old size */
1136     t->t_info &= 0xe000;
1137     /* or in new size */
1138     t->t_info |= MAXALIGN(entry.bytes + sizeof(IndexTupleData));
1139
1140     return(t);
1141   }
1142   else {
1143     /* generate a new index tuple for the compressed entry */
1144     TupleDesc tupDesc = r->rd_att;
1145     IndexTuple newtup;
1146     char *isnull;
1147     int blank;
1148
1149     isnull = (char *) palloc(r->rd_rel->relnatts);
1150     for (blank = 0; blank < r->rd_rel->relnatts; blank++)
1151       isnull[blank] = ' ';
1152     newtup = (IndexTuple) index_formtuple(tupDesc,
1153                                           (Datum *)&(entry.pred),
1154                                           isnull);
1155     newtup->t_tid = t->t_tid;
1156     pfree(isnull);
1157     return(newtup);
1158   }
1159 }
1160
1161
1162 /*
1163 ** initialize a GiST entry with a decompressed version of pred
1164 */
1165 void
1166 gistdentryinit(GISTSTATE *giststate, GISTENTRY *e, char *pr, Relation r,
1167                Page pg, OffsetNumber o, int b, bool l)
1168 {
1169     GISTENTRY *dep;
1170     gistentryinit(*e, pr, r, pg, o, b, l);
1171     if (giststate->haskeytype) {
1172       dep = (GISTENTRY *)((giststate->decompressFn)(e));
1173       gistentryinit(*e, dep->pred, dep->rel, dep->page, dep->offset, dep->bytes,
1174                     dep->leafkey);
1175       if (dep != e) pfree(dep);
1176     }
1177 }
1178
1179
1180 /*
1181 ** initialize a GiST entry with a compressed version of pred
1182 */
1183 void
1184 gistcentryinit(GISTSTATE *giststate, GISTENTRY *e, char *pr, Relation r,
1185                Page pg, OffsetNumber o, int b, bool l)
1186 {
1187     GISTENTRY *cep;
1188     gistentryinit(*e, pr, r, pg, o, b, l);
1189     if (giststate->haskeytype) {
1190       cep = (GISTENTRY *)((giststate->compressFn)(e));
1191       gistentryinit(*e, cep->pred, cep->rel, cep->page, cep->offset, cep->bytes,
1192                     cep->leafkey);
1193       if (cep != e) pfree(cep);
1194     }
1195 }
1196
1197
1198
1199 #ifdef GISTDEBUG
1200
1201 extern char *text_range_out();
1202 extern char *int_range_out();
1203
1204 /*
1205 ** sloppy debugging support routine, requires recompilation with appropriate
1206 ** "out" method for the index keys.  Could be fixed to find that info
1207 ** in the catalogs...
1208 */
1209 void
1210 _gistdump(Relation r)
1211 {
1212     Buffer buf;
1213     Page page;
1214     OffsetNumber offnum, maxoff;
1215     BlockNumber blkno;
1216     BlockNumber nblocks;
1217     GISTPageOpaque po;
1218     IndexTuple itup;
1219     BlockNumber itblkno;
1220     OffsetNumber itoffno;
1221     char *datum;
1222     char *itkey;
1223
1224     nblocks = RelationGetNumberOfBlocks(r);
1225     for (blkno = 0; blkno < nblocks; blkno++) {
1226         buf = ReadBuffer(r, blkno);
1227         page = BufferGetPage(buf);
1228         po = (GISTPageOpaque) PageGetSpecialPointer(page);
1229         maxoff = PageGetMaxOffsetNumber(page);
1230         printf("Page %d maxoff %d <%s>\n", blkno, maxoff,
1231                (po->flags & F_LEAF ? "LEAF" : "INTERNAL"));
1232
1233         if (PageIsEmpty(page)) {
1234             ReleaseBuffer(buf);
1235             continue;
1236         }
1237
1238         for (offnum = FirstOffsetNumber;
1239              offnum <= maxoff;
1240              offnum = OffsetNumberNext(offnum)) {
1241             itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, offnum));
1242             itblkno = ItemPointerGetBlockNumber(&(itup->t_tid));
1243             itoffno = ItemPointerGetOffsetNumber(&(itup->t_tid));
1244             datum = ((char *) itup);
1245             datum += sizeof(IndexTupleData);
1246             /* get out function for type of key, and out it! */
1247             itkey = (char *) int_range_out(datum);
1248             /* itkey = " unable to print"; */
1249             printf("\t[%d] size %d heap <%d,%d> key:%s\n",
1250                    offnum, IndexTupleSize(itup), itblkno, itoffno, itkey);
1251             pfree(itkey);
1252         }
1253
1254         ReleaseBuffer(buf);
1255     }
1256 }
1257
1258 #define TRLOWER(tr) (((tr)->bytes))
1259 #define TRUPPER(tr) (&((tr)->bytes[MAXALIGN(VARSIZE(TRLOWER(tr)))]))
1260 typedef struct txtrange {
1261     /* flag: NINF means that lower is negative infinity; PINF means that
1262     ** upper is positive infinity.  0 means that both are numbers.
1263     */
1264     int32 vl_len;
1265     int32 flag;
1266     char bytes[2];
1267 } TXTRANGE;
1268
1269 typedef struct intrange {
1270     int lower;
1271     int upper;
1272     /* flag: NINF means that lower is negative infinity; PINF means that
1273     ** upper is positive infinity.  0 means that both are numbers.
1274     */
1275     int flag;
1276 } INTRANGE;
1277
1278 char *text_range_out(TXTRANGE *r)
1279 {
1280     char        *result;
1281     char        *lower, *upper;
1282
1283     if (r == NULL)
1284         return(NULL);
1285     result = (char *)palloc(16 + VARSIZE(TRLOWER(r)) + VARSIZE(TRUPPER(r))
1286                             - 2*VARHDRSZ);
1287
1288     lower = (char *)palloc(VARSIZE(TRLOWER(r)) + 1 - VARHDRSZ);
1289     memcpy(lower, VARDATA(TRLOWER(r)), VARSIZE(TRLOWER(r)) - VARHDRSZ);
1290     lower[VARSIZE(TRLOWER(r)) - VARHDRSZ] = '\0';
1291     upper = (char *)palloc(VARSIZE(TRUPPER(r)) + 1 - VARHDRSZ);
1292     memcpy(upper, VARDATA(TRUPPER(r)), VARSIZE(TRUPPER(r)) - VARHDRSZ);
1293     upper[VARSIZE(TRUPPER(r)) - VARHDRSZ] = '\0';
1294
1295     (void) sprintf(result, "[%s,%s): %d", lower, upper, r->flag);
1296     pfree(lower);
1297     pfree(upper);
1298     return(result);
1299 }
1300
1301 char *
1302 int_range_out(INTRANGE *r)
1303 {
1304     char        *result;
1305
1306     if (r == NULL)
1307         return(NULL);
1308     result = (char *)palloc(80);
1309     (void) sprintf(result, "[%d,%d): %d",r->lower, r->upper, r->flag);
1310
1311     return(result);
1312 }
1313
1314 #endif /* defined GISTDEBUG */
1315