src/backend/access/gin/gininsert.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * gininsert.c
   4  *        insert routines for the postgres inverted index access method.
   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/genam.h"
  18 #include "access/gin.h"
  19 #include "catalog/index.h"
  20 #include "miscadmin.h"
  21 #include "storage/bufmgr.h"
  22 #include "storage/indexfsm.h"
  23 #include "utils/memutils.h"
  24
  25
  26 typedef struct
  27 {
  28         GinState        ginstate;
  29         double          indtuples;
  30         MemoryContext tmpCtx;
  31         MemoryContext funcCtx;
  32         BuildAccumulator accum;
  33 } GinBuildState;
  34
  35 /*
  36  * Creates posting tree with one page. Function
  37  * suppose that items[] fits to page
  38  */
  39 static BlockNumber
  40 createPostingTree(Relation index, ItemPointerData *items, uint32 nitems)
  41 {
  42         BlockNumber blkno;
  43         Buffer          buffer = GinNewBuffer(index);
  44         Page            page;
  45
  46         START_CRIT_SECTION();
  47
  48         GinInitBuffer(buffer, GIN_DATA | GIN_LEAF);
  49         page = BufferGetPage(buffer);
  50         blkno = BufferGetBlockNumber(buffer);
  51
  52         memcpy(GinDataPageGetData(page), items, sizeof(ItemPointerData) * nitems);
  53         GinPageGetOpaque(page)->maxoff = nitems;
  54
  55         MarkBufferDirty(buffer);
  56
  57         if (!index->rd_istemp)
  58         {
  59                 XLogRecPtr      recptr;
  60                 XLogRecData rdata[2];
  61                 ginxlogCreatePostingTree data;
  62
  63                 data.node = index->rd_node;
  64                 data.blkno = blkno;
  65                 data.nitem = nitems;
  66
  67                 rdata[0].buffer = InvalidBuffer;
  68                 rdata[0].data = (char *) &data;
  69                 rdata[0].len = sizeof(ginxlogCreatePostingTree);
  70                 rdata[0].next = &rdata[1];
  71
  72                 rdata[1].buffer = InvalidBuffer;
  73                 rdata[1].data = (char *) items;
  74                 rdata[1].len = sizeof(ItemPointerData) * nitems;
  75                 rdata[1].next = NULL;
  76
  77
  78
  79                 recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_PTREE, rdata);
  80                 PageSetLSN(page, recptr);
  81                 PageSetTLI(page, ThisTimeLineID);
  82
  83         }
  84
  85         UnlockReleaseBuffer(buffer);
  86
  87         END_CRIT_SECTION();
  88
  89         return blkno;
  90 }
  91
  92
  93 /*
  94  * Adds array of item pointers to tuple's posting list or
  95  * creates posting tree and tuple pointed to tree in a case
  96  * of not enough space.  Max size of tuple is defined in
  97  * GinFormTuple().
  98  */
  99 static IndexTuple
 100 addItemPointersToTuple(Relation index, GinState *ginstate, GinBtreeStack *stack,
 101                   IndexTuple old, ItemPointerData *items, uint32 nitem, bool isBuild)
 102 {
 103         Datum           key = gin_index_getattr(ginstate, old);
 104         OffsetNumber attnum = gintuple_get_attrnum(ginstate, old);
 105         IndexTuple      res = GinFormTuple(ginstate, attnum, key,
 106                                                                    NULL, nitem + GinGetNPosting(old));
 107
 108         if (res)
 109         {
 110                 /* good, small enough */
 111                 uint32          newnitem;
 112
 113                 newnitem = MergeItemPointers(GinGetPosting(res),
 114                                                                          GinGetPosting(old), GinGetNPosting(old),
 115                                                                          items, nitem);
 116                 /* merge might have eliminated some duplicate items */
 117                 GinShortenTuple(res, newnitem);
 118         }
 119         else
 120         {
 121                 BlockNumber postingRoot;
 122                 GinPostingTreeScan *gdi;
 123
 124                 /* posting list becomes big, so we need to make posting's tree */
 125                 res = GinFormTuple(ginstate, attnum, key, NULL, 0);
 126                 postingRoot = createPostingTree(index, GinGetPosting(old), GinGetNPosting(old));
 127                 GinSetPostingTree(res, postingRoot);
 128
 129                 gdi = prepareScanPostingTree(index, postingRoot, FALSE);
 130                 gdi->btree.isBuild = isBuild;
 131
 132                 insertItemPointer(gdi, items, nitem);
 133
 134                 pfree(gdi);
 135         }
 136
 137         return res;
 138 }
 139
 140 /*
 141  * Inserts only one entry to the index, but it can add more than 1 ItemPointer.
 142  */
 143 void
 144 ginEntryInsert(Relation index, GinState *ginstate,
 145                            OffsetNumber attnum, Datum value,
 146                            ItemPointerData *items, uint32 nitem,
 147                            bool isBuild)
 148 {
 149         GinBtreeData btree;
 150         GinBtreeStack *stack;
 151         IndexTuple      itup;
 152         Page            page;
 153
 154         prepareEntryScan(&btree, index, attnum, value, ginstate);
 155
 156         stack = ginFindLeafPage(&btree, NULL);
 157         page = BufferGetPage(stack->buffer);
 158
 159         if (btree.findItem(&btree, stack))
 160         {
 161                 /* found entry */
 162                 itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, stack->off));
 163
 164                 if (GinIsPostingTree(itup))
 165                 {
 166                         /* lock root of posting tree */
 167                         GinPostingTreeScan *gdi;
 168                         BlockNumber rootPostingTree = GinGetPostingTree(itup);
 169
 170                         /* release all stack */
 171                         LockBuffer(stack->buffer, GIN_UNLOCK);
 172                         freeGinBtreeStack(stack);
 173
 174                         /* insert into posting tree */
 175                         gdi = prepareScanPostingTree(index, rootPostingTree, FALSE);
 176                         gdi->btree.isBuild = isBuild;
 177                         insertItemPointer(gdi, items, nitem);
 178
 179                         return;
 180                 }
 181
 182                 itup = addItemPointersToTuple(index, ginstate, stack, itup, items, nitem, isBuild);
 183
 184                 btree.isDelete = TRUE;
 185         }
 186         else
 187         {
 188                 /* We suppose, that tuple can store at list one itempointer */
 189                 itup = GinFormTuple(ginstate, attnum, value, items, 1);
 190                 if (itup == NULL || IndexTupleSize(itup) >= GinMaxItemSize)
 191                         elog(ERROR, "huge tuple");
 192
 193                 if (nitem > 1)
 194                 {
 195                         IndexTuple      previtup = itup;
 196
 197                         itup = addItemPointersToTuple(index, ginstate, stack, previtup, items + 1, nitem - 1, isBuild);
 198                         pfree(previtup);
 199                 }
 200         }
 201
 202         btree.entry = itup;
 203         ginInsertValue(&btree, stack);
 204         pfree(itup);
 205 }
 206
 207 /*
 208  * Saves indexed value in memory accumulator during index creation
 209  * Function isn't used during normal insert
 210  */
 211 static uint32
 212 ginHeapTupleBulkInsert(GinBuildState *buildstate, OffsetNumber attnum, Datum value, ItemPointer heapptr)
 213 {
 214         Datum      *entries;
 215         int32           nentries;
 216         MemoryContext oldCtx;
 217
 218         oldCtx = MemoryContextSwitchTo(buildstate->funcCtx);
 219         entries = extractEntriesSU(buildstate->accum.ginstate, attnum, value, &nentries);
 220         MemoryContextSwitchTo(oldCtx);
 221
 222         if (nentries == 0)
 223                 /* nothing to insert */
 224                 return 0;
 225
 226         ginInsertRecordBA(&buildstate->accum, heapptr, attnum, entries, nentries);
 227
 228         MemoryContextReset(buildstate->funcCtx);
 229
 230         return nentries;
 231 }
 232
 233 static void
 234 ginBuildCallback(Relation index, HeapTuple htup, Datum *values,
 235                                  bool *isnull, bool tupleIsAlive, void *state)
 236 {
 237         GinBuildState *buildstate = (GinBuildState *) state;
 238         MemoryContext oldCtx;
 239         int                     i;
 240
 241         oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
 242
 243         for (i = 0; i < buildstate->ginstate.origTupdesc->natts; i++)
 244                 if (!isnull[i])
 245                         buildstate->indtuples += ginHeapTupleBulkInsert(buildstate,
 246                                                                                    (OffsetNumber) (i + 1), values[i],
 247                                                                                                                         &htup->t_self);
 248
 249         /* If we've maxed out our available memory, dump everything to the index */
 250         /* Also dump if the tree seems to be getting too unbalanced */
 251         if (buildstate->accum.allocatedMemory >= maintenance_work_mem * 1024L ||
 252                 buildstate->accum.maxdepth > GIN_MAX_TREE_DEPTH)
 253         {
 254                 ItemPointerData *list;
 255                 Datum           entry;
 256                 uint32          nlist;
 257                 OffsetNumber attnum;
 258
 259                 while ((list = ginGetEntry(&buildstate->accum, &attnum, &entry, &nlist)) != NULL)
 260                 {
 261                         /* there could be many entries, so be willing to abort here */
 262                         CHECK_FOR_INTERRUPTS();
 263                         ginEntryInsert(index, &buildstate->ginstate, attnum, entry, list, nlist, TRUE);
 264                 }
 265
 266                 MemoryContextReset(buildstate->tmpCtx);
 267                 ginInitBA(&buildstate->accum);
 268         }
 269
 270         MemoryContextSwitchTo(oldCtx);
 271 }
 272
 273 Datum
 274 ginbuild(PG_FUNCTION_ARGS)
 275 {
 276         Relation        heap = (Relation) PG_GETARG_POINTER(0);
 277         Relation        index = (Relation) PG_GETARG_POINTER(1);
 278         IndexInfo  *indexInfo = (IndexInfo *) PG_GETARG_POINTER(2);
 279         IndexBuildResult *result;
 280         double          reltuples;
 281         GinBuildState buildstate;
 282         Buffer          RootBuffer,
 283                                 MetaBuffer;
 284         ItemPointerData *list;
 285         Datum           entry;
 286         uint32          nlist;
 287         MemoryContext oldCtx;
 288         OffsetNumber attnum;
 289
 290         if (RelationGetNumberOfBlocks(index) != 0)
 291                 elog(ERROR, "index \"%s\" already contains data",
 292                          RelationGetRelationName(index));
 293
 294         initGinState(&buildstate.ginstate, index);
 295
 296         /* initialize the meta page */
 297         MetaBuffer = GinNewBuffer(index);
 298
 299         /* initialize the root page */
 300         RootBuffer = GinNewBuffer(index);
 301
 302         START_CRIT_SECTION();
 303         GinInitMetabuffer(MetaBuffer);
 304         MarkBufferDirty(MetaBuffer);
 305         GinInitBuffer(RootBuffer, GIN_LEAF);
 306         MarkBufferDirty(RootBuffer);
 307
 308         if (!index->rd_istemp)
 309         {
 310                 XLogRecPtr      recptr;
 311                 XLogRecData rdata;
 312                 Page            page;
 313
 314                 rdata.buffer = InvalidBuffer;
 315                 rdata.data = (char *) &(index->rd_node);
 316                 rdata.len = sizeof(RelFileNode);
 317                 rdata.next = NULL;
 318
 319                 recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_INDEX, &rdata);
 320
 321                 page = BufferGetPage(RootBuffer);
 322                 PageSetLSN(page, recptr);
 323                 PageSetTLI(page, ThisTimeLineID);
 324
 325                 page = BufferGetPage(MetaBuffer);
 326                 PageSetLSN(page, recptr);
 327                 PageSetTLI(page, ThisTimeLineID);
 328         }
 329
 330         UnlockReleaseBuffer(MetaBuffer);
 331         UnlockReleaseBuffer(RootBuffer);
 332         END_CRIT_SECTION();
 333
 334         /* build the index */
 335         buildstate.indtuples = 0;
 336
 337         /*
 338          * create a temporary memory context that is reset once for each tuple
 339          * inserted into the index
 340          */
 341         buildstate.tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
 342                                                                                           "Gin build temporary context",
 343                                                                                           ALLOCSET_DEFAULT_MINSIZE,
 344                                                                                           ALLOCSET_DEFAULT_INITSIZE,
 345                                                                                           ALLOCSET_DEFAULT_MAXSIZE);
 346
 347         buildstate.funcCtx = AllocSetContextCreate(buildstate.tmpCtx,
 348                                          "Gin build temporary context for user-defined function",
 349                                                                                            ALLOCSET_DEFAULT_MINSIZE,
 350                                                                                            ALLOCSET_DEFAULT_INITSIZE,
 351                                                                                            ALLOCSET_DEFAULT_MAXSIZE);
 352
 353         buildstate.accum.ginstate = &buildstate.ginstate;
 354         ginInitBA(&buildstate.accum);
 355
 356         /*
 357          * Do the heap scan.  We disallow sync scan here because dataPlaceToPage
 358          * prefers to receive tuples in TID order.
 359          */
 360         reltuples = IndexBuildHeapScan(heap, index, indexInfo, false,
 361                                                                    ginBuildCallback, (void *) &buildstate);
 362
 363         /* dump remaining entries to the index */
 364         oldCtx = MemoryContextSwitchTo(buildstate.tmpCtx);
 365         while ((list = ginGetEntry(&buildstate.accum, &attnum, &entry, &nlist)) != NULL)
 366         {
 367                 /* there could be many entries, so be willing to abort here */
 368                 CHECK_FOR_INTERRUPTS();
 369                 ginEntryInsert(index, &buildstate.ginstate, attnum, entry, list, nlist, TRUE);
 370         }
 371         MemoryContextSwitchTo(oldCtx);
 372
 373         MemoryContextDelete(buildstate.tmpCtx);
 374
 375         /*
 376          * Return statistics
 377          */
 378         result = (IndexBuildResult *) palloc(sizeof(IndexBuildResult));
 379
 380         result->heap_tuples = reltuples;
 381         result->index_tuples = buildstate.indtuples;
 382
 383         PG_RETURN_POINTER(result);
 384 }
 385
 386 /*
 387  * Inserts value during normal insertion
 388  */
 389 static uint32
 390 ginHeapTupleInsert(Relation index, GinState *ginstate, OffsetNumber attnum, Datum value, ItemPointer item)
 391 {
 392         Datum      *entries;
 393         int32           i,
 394                                 nentries;
 395
 396         entries = extractEntriesSU(ginstate, attnum, value, &nentries);
 397
 398         if (nentries == 0)
 399                 /* nothing to insert */
 400                 return 0;
 401
 402         for (i = 0; i < nentries; i++)
 403                 ginEntryInsert(index, ginstate, attnum, entries[i], item, 1, FALSE);
 404
 405         return nentries;
 406 }
 407
 408 Datum
 409 gininsert(PG_FUNCTION_ARGS)
 410 {
 411         Relation        index = (Relation) PG_GETARG_POINTER(0);
 412         Datum      *values = (Datum *) PG_GETARG_POINTER(1);
 413         bool       *isnull = (bool *) PG_GETARG_POINTER(2);
 414         ItemPointer ht_ctid = (ItemPointer) PG_GETARG_POINTER(3);
 415
 416 #ifdef NOT_USED
 417         Relation        heapRel = (Relation) PG_GETARG_POINTER(4);
 418         bool            checkUnique = PG_GETARG_BOOL(5);
 419 #endif
 420         GinState        ginstate;
 421         MemoryContext oldCtx;
 422         MemoryContext insertCtx;
 423         uint32          res = 0;
 424         int                     i;
 425
 426         insertCtx = AllocSetContextCreate(CurrentMemoryContext,
 427                                                                           "Gin insert temporary context",
 428                                                                           ALLOCSET_DEFAULT_MINSIZE,
 429                                                                           ALLOCSET_DEFAULT_INITSIZE,
 430                                                                           ALLOCSET_DEFAULT_MAXSIZE);
 431
 432         oldCtx = MemoryContextSwitchTo(insertCtx);
 433
 434         initGinState(&ginstate, index);
 435
 436         if (GinGetUseFastUpdate(index))
 437         {
 438                 GinTupleCollector collector;
 439
 440                 memset(&collector, 0, sizeof(GinTupleCollector));
 441                 for (i = 0; i < ginstate.origTupdesc->natts; i++)
 442                         if (!isnull[i])
 443                                 res += ginHeapTupleFastCollect(index, &ginstate, &collector,
 444                                                                  (OffsetNumber) (i + 1), values[i], ht_ctid);
 445
 446                 ginHeapTupleFastInsert(index, &ginstate, &collector);
 447         }
 448         else
 449         {
 450                 for (i = 0; i < ginstate.origTupdesc->natts; i++)
 451                         if (!isnull[i])
 452                                 res += ginHeapTupleInsert(index, &ginstate,
 453                                                                  (OffsetNumber) (i + 1), values[i], ht_ctid);
 454
 455         }
 456
 457         MemoryContextSwitchTo(oldCtx);
 458         MemoryContextDelete(insertCtx);
 459
 460         PG_RETURN_BOOL(res > 0);
 461 }