src/backend/access/gin/ginxlog.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * ginxlog.c
   4  *        WAL replay logic for inverted index.
   5  *
   6  *
   7  * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
   8  * Portions Copyright (c) 1994, Regents of the University of California
   9  *
  10  * IDENTIFICATION
  11  *                       $PostgreSQL$
  12  *-------------------------------------------------------------------------
  13  */
  14 #include "postgres.h"
  15
  16 #include "access/gin.h"
  17 #include "access/xlogutils.h"
  18 #include "storage/bufmgr.h"
  19 #include "utils/memutils.h"
  20
  21 static MemoryContext opCtx;             /* working memory for operations */
  22 static MemoryContext topCtx;
  23
  24 typedef struct ginIncompleteSplit
  25 {
  26         RelFileNode node;
  27         BlockNumber leftBlkno;
  28         BlockNumber rightBlkno;
  29         BlockNumber rootBlkno;
  30 } ginIncompleteSplit;
  31
  32 static List *incomplete_splits;
  33
  34 static void
  35 pushIncompleteSplit(RelFileNode node, BlockNumber leftBlkno, BlockNumber rightBlkno, BlockNumber rootBlkno)
  36 {
  37         ginIncompleteSplit *split;
  38
  39         MemoryContextSwitchTo(topCtx);
  40
  41         split = palloc(sizeof(ginIncompleteSplit));
  42
  43         split->node = node;
  44         split->leftBlkno = leftBlkno;
  45         split->rightBlkno = rightBlkno;
  46         split->rootBlkno = rootBlkno;
  47
  48         incomplete_splits = lappend(incomplete_splits, split);
  49
  50         MemoryContextSwitchTo(opCtx);
  51 }
  52
  53 static void
  54 forgetIncompleteSplit(RelFileNode node, BlockNumber leftBlkno, BlockNumber updateBlkno)
  55 {
  56         ListCell   *l;
  57
  58         foreach(l, incomplete_splits)
  59         {
  60                 ginIncompleteSplit *split = (ginIncompleteSplit *) lfirst(l);
  61
  62                 if (RelFileNodeEquals(node, split->node) && leftBlkno == split->leftBlkno && updateBlkno == split->rightBlkno)
  63                 {
  64                         incomplete_splits = list_delete_ptr(incomplete_splits, split);
  65                         break;
  66                 }
  67         }
  68 }
  69
  70 static void
  71 ginRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
  72 {
  73         RelFileNode *node = (RelFileNode *) XLogRecGetData(record);
  74         Buffer          buffer;
  75         Page            page;
  76
  77         buffer = XLogReadBuffer(*node, GIN_ROOT_BLKNO, true);
  78         Assert(BufferIsValid(buffer));
  79         page = (Page) BufferGetPage(buffer);
  80
  81         GinInitBuffer(buffer, GIN_LEAF);
  82
  83         PageSetLSN(page, lsn);
  84         PageSetTLI(page, ThisTimeLineID);
  85
  86         MarkBufferDirty(buffer);
  87         UnlockReleaseBuffer(buffer);
  88 }
  89
  90 static void
  91 ginRedoCreatePTree(XLogRecPtr lsn, XLogRecord *record)
  92 {
  93         ginxlogCreatePostingTree *data = (ginxlogCreatePostingTree *) XLogRecGetData(record);
  94         ItemPointerData *items = (ItemPointerData *) (XLogRecGetData(record) + sizeof(ginxlogCreatePostingTree));
  95         Buffer          buffer;
  96         Page            page;
  97
  98         buffer = XLogReadBuffer(data->node, data->blkno, true);
  99         Assert(BufferIsValid(buffer));
 100         page = (Page) BufferGetPage(buffer);
 101
 102         GinInitBuffer(buffer, GIN_DATA | GIN_LEAF);
 103         memcpy(GinDataPageGetData(page), items, sizeof(ItemPointerData) * data->nitem);
 104         GinPageGetOpaque(page)->maxoff = data->nitem;
 105
 106         PageSetLSN(page, lsn);
 107         PageSetTLI(page, ThisTimeLineID);
 108
 109         MarkBufferDirty(buffer);
 110         UnlockReleaseBuffer(buffer);
 111 }
 112
 113 static void
 114 ginRedoInsert(XLogRecPtr lsn, XLogRecord *record)
 115 {
 116         ginxlogInsert *data = (ginxlogInsert *) XLogRecGetData(record);
 117         Buffer          buffer;
 118         Page            page;
 119
 120         /* nothing else to do if page was backed up */
 121         if (record->xl_info & XLR_BKP_BLOCK_1)
 122                 return;
 123
 124         buffer = XLogReadBuffer(data->node, data->blkno, false);
 125         Assert(BufferIsValid(buffer));
 126         page = (Page) BufferGetPage(buffer);
 127
 128         if (data->isData)
 129         {
 130                 Assert(data->isDelete == FALSE);
 131                 Assert(GinPageIsData(page));
 132
 133                 if (!XLByteLE(lsn, PageGetLSN(page)))
 134                 {
 135                         if (data->isLeaf)
 136                         {
 137                                 OffsetNumber i;
 138                                 ItemPointerData *items = (ItemPointerData *) (XLogRecGetData(record) + sizeof(ginxlogInsert));
 139
 140                                 Assert(GinPageIsLeaf(page));
 141                                 Assert(data->updateBlkno == InvalidBlockNumber);
 142
 143                                 for (i = 0; i < data->nitem; i++)
 144                                         GinDataPageAddItem(page, items + i, data->offset + i);
 145                         }
 146                         else
 147                         {
 148                                 PostingItem *pitem;
 149
 150                                 Assert(!GinPageIsLeaf(page));
 151
 152                                 if (data->updateBlkno != InvalidBlockNumber)
 153                                 {
 154                                         /* update link to right page after split */
 155                                         pitem = (PostingItem *) GinDataPageGetItem(page, data->offset);
 156                                         PostingItemSetBlockNumber(pitem, data->updateBlkno);
 157                                 }
 158
 159                                 pitem = (PostingItem *) (XLogRecGetData(record) + sizeof(ginxlogInsert));
 160
 161                                 GinDataPageAddItem(page, pitem, data->offset);
 162                         }
 163                 }
 164
 165                 if (!data->isLeaf && data->updateBlkno != InvalidBlockNumber)
 166                 {
 167                         PostingItem *pitem = (PostingItem *) (XLogRecGetData(record) + sizeof(ginxlogInsert));
 168
 169                         forgetIncompleteSplit(data->node, PostingItemGetBlockNumber(pitem), data->updateBlkno);
 170                 }
 171
 172         }
 173         else
 174         {
 175                 IndexTuple      itup;
 176
 177                 Assert(!GinPageIsData(page));
 178
 179                 if (!XLByteLE(lsn, PageGetLSN(page)))
 180                 {
 181                         if (data->updateBlkno != InvalidBlockNumber)
 182                         {
 183                                 /* update link to right page after split */
 184                                 Assert(!GinPageIsLeaf(page));
 185                                 Assert(data->offset >= FirstOffsetNumber && data->offset <= PageGetMaxOffsetNumber(page));
 186                                 itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, data->offset));
 187                                 ItemPointerSet(&itup->t_tid, data->updateBlkno, InvalidOffsetNumber);
 188                         }
 189
 190                         if (data->isDelete)
 191                         {
 192                                 Assert(GinPageIsLeaf(page));
 193                                 Assert(data->offset >= FirstOffsetNumber && data->offset <= PageGetMaxOffsetNumber(page));
 194                                 PageIndexTupleDelete(page, data->offset);
 195                         }
 196
 197                         itup = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogInsert));
 198
 199                         if (PageAddItem(page, (Item) itup, IndexTupleSize(itup), data->offset, false, false) == InvalidOffsetNumber)
 200                                 elog(ERROR, "failed to add item to index page in %u/%u/%u",
 201                                   data->node.spcNode, data->node.dbNode, data->node.relNode);
 202                 }
 203
 204                 if (!data->isLeaf && data->updateBlkno != InvalidBlockNumber)
 205                 {
 206                         itup = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogInsert));
 207                         forgetIncompleteSplit(data->node, GinItemPointerGetBlockNumber(&itup->t_tid), data->updateBlkno);
 208                 }
 209         }
 210
 211         if (!XLByteLE(lsn, PageGetLSN(page)))
 212         {
 213                 PageSetLSN(page, lsn);
 214                 PageSetTLI(page, ThisTimeLineID);
 215
 216                 MarkBufferDirty(buffer);
 217         }
 218         UnlockReleaseBuffer(buffer);
 219 }
 220
 221 static void
 222 ginRedoSplit(XLogRecPtr lsn, XLogRecord *record)
 223 {
 224         ginxlogSplit *data = (ginxlogSplit *) XLogRecGetData(record);
 225         Buffer          lbuffer,
 226                                 rbuffer;
 227         Page            lpage,
 228                                 rpage;
 229         uint32          flags = 0;
 230
 231         if (data->isLeaf)
 232                 flags |= GIN_LEAF;
 233         if (data->isData)
 234                 flags |= GIN_DATA;
 235
 236         lbuffer = XLogReadBuffer(data->node, data->lblkno, data->isRootSplit);
 237         Assert(BufferIsValid(lbuffer));
 238         lpage = (Page) BufferGetPage(lbuffer);
 239         GinInitBuffer(lbuffer, flags);
 240
 241         rbuffer = XLogReadBuffer(data->node, data->rblkno, true);
 242         Assert(BufferIsValid(rbuffer));
 243         rpage = (Page) BufferGetPage(rbuffer);
 244         GinInitBuffer(rbuffer, flags);
 245
 246         GinPageGetOpaque(lpage)->rightlink = BufferGetBlockNumber(rbuffer);
 247         GinPageGetOpaque(rpage)->rightlink = data->rrlink;
 248
 249         if (data->isData)
 250         {
 251                 char       *ptr = XLogRecGetData(record) + sizeof(ginxlogSplit);
 252                 Size            sizeofitem = GinSizeOfItem(lpage);
 253                 OffsetNumber i;
 254                 ItemPointer bound;
 255
 256                 for (i = 0; i < data->separator; i++)
 257                 {
 258                         GinDataPageAddItem(lpage, ptr, InvalidOffsetNumber);
 259                         ptr += sizeofitem;
 260                 }
 261
 262                 for (i = data->separator; i < data->nitem; i++)
 263                 {
 264                         GinDataPageAddItem(rpage, ptr, InvalidOffsetNumber);
 265                         ptr += sizeofitem;
 266                 }
 267
 268                 /* set up right key */
 269                 bound = GinDataPageGetRightBound(lpage);
 270                 if (data->isLeaf)
 271                         *bound = *(ItemPointerData *) GinDataPageGetItem(lpage, GinPageGetOpaque(lpage)->maxoff);
 272                 else
 273                         *bound = ((PostingItem *) GinDataPageGetItem(lpage, GinPageGetOpaque(lpage)->maxoff))->key;
 274
 275                 bound = GinDataPageGetRightBound(rpage);
 276                 *bound = data->rightbound;
 277         }
 278         else
 279         {
 280                 IndexTuple      itup = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogSplit));
 281                 OffsetNumber i;
 282
 283                 for (i = 0; i < data->separator; i++)
 284                 {
 285                         if (PageAddItem(lpage, (Item) itup, IndexTupleSize(itup), InvalidOffsetNumber, false, false) == InvalidOffsetNumber)
 286                                 elog(ERROR, "failed to add item to index page in %u/%u/%u",
 287                                   data->node.spcNode, data->node.dbNode, data->node.relNode);
 288                         itup = (IndexTuple) (((char *) itup) + MAXALIGN(IndexTupleSize(itup)));
 289                 }
 290
 291                 for (i = data->separator; i < data->nitem; i++)
 292                 {
 293                         if (PageAddItem(rpage, (Item) itup, IndexTupleSize(itup), InvalidOffsetNumber, false, false) == InvalidOffsetNumber)
 294                                 elog(ERROR, "failed to add item to index page in %u/%u/%u",
 295                                   data->node.spcNode, data->node.dbNode, data->node.relNode);
 296                         itup = (IndexTuple) (((char *) itup) + MAXALIGN(IndexTupleSize(itup)));
 297                 }
 298         }
 299
 300         PageSetLSN(rpage, lsn);
 301         PageSetTLI(rpage, ThisTimeLineID);
 302         MarkBufferDirty(rbuffer);
 303
 304         PageSetLSN(lpage, lsn);
 305         PageSetTLI(lpage, ThisTimeLineID);
 306         MarkBufferDirty(lbuffer);
 307
 308         if (!data->isLeaf && data->updateBlkno != InvalidBlockNumber)
 309                 forgetIncompleteSplit(data->node, data->leftChildBlkno, data->updateBlkno);
 310
 311         if (data->isRootSplit)
 312         {
 313                 Buffer          rootBuf = XLogReadBuffer(data->node, data->rootBlkno, false);
 314                 Page            rootPage = BufferGetPage(rootBuf);
 315
 316                 GinInitBuffer(rootBuf, flags & ~GIN_LEAF);
 317
 318                 if (data->isData)
 319                 {
 320                         Assert(data->rootBlkno != GIN_ROOT_BLKNO);
 321                         dataFillRoot(NULL, rootBuf, lbuffer, rbuffer);
 322                 }
 323                 else
 324                 {
 325                         Assert(data->rootBlkno == GIN_ROOT_BLKNO);
 326                         entryFillRoot(NULL, rootBuf, lbuffer, rbuffer);
 327                 }
 328
 329                 PageSetLSN(rootPage, lsn);
 330                 PageSetTLI(rootPage, ThisTimeLineID);
 331
 332                 MarkBufferDirty(rootBuf);
 333                 UnlockReleaseBuffer(rootBuf);
 334         }
 335         else
 336                 pushIncompleteSplit(data->node, data->lblkno, data->rblkno, data->rootBlkno);
 337
 338         UnlockReleaseBuffer(rbuffer);
 339         UnlockReleaseBuffer(lbuffer);
 340 }
 341
 342 static void
 343 ginRedoVacuumPage(XLogRecPtr lsn, XLogRecord *record)
 344 {
 345         ginxlogVacuumPage *data = (ginxlogVacuumPage *) XLogRecGetData(record);
 346         Buffer          buffer;
 347         Page            page;
 348
 349         /* nothing else to do if page was backed up (and no info to do it with) */
 350         if (record->xl_info & XLR_BKP_BLOCK_1)
 351                 return;
 352
 353         buffer = XLogReadBuffer(data->node, data->blkno, false);
 354         Assert(BufferIsValid(buffer));
 355         page = (Page) BufferGetPage(buffer);
 356
 357         if (GinPageIsData(page))
 358         {
 359                 memcpy(GinDataPageGetData(page), XLogRecGetData(record) + sizeof(ginxlogVacuumPage),
 360                            GinSizeOfItem(page) *data->nitem);
 361                 GinPageGetOpaque(page)->maxoff = data->nitem;
 362         }
 363         else
 364         {
 365                 OffsetNumber i,
 366                                    *tod;
 367                 IndexTuple      itup = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogVacuumPage));
 368
 369                 tod = (OffsetNumber *) palloc(sizeof(OffsetNumber) * PageGetMaxOffsetNumber(page));
 370                 for (i = FirstOffsetNumber; i <= PageGetMaxOffsetNumber(page); i++)
 371                         tod[i - 1] = i;
 372
 373                 PageIndexMultiDelete(page, tod, PageGetMaxOffsetNumber(page));
 374
 375                 for (i = 0; i < data->nitem; i++)
 376                 {
 377                         if (PageAddItem(page, (Item) itup, IndexTupleSize(itup), InvalidOffsetNumber, false, false) == InvalidOffsetNumber)
 378                                 elog(ERROR, "failed to add item to index page in %u/%u/%u",
 379                                   data->node.spcNode, data->node.dbNode, data->node.relNode);
 380                         itup = (IndexTuple) (((char *) itup) + MAXALIGN(IndexTupleSize(itup)));
 381                 }
 382         }
 383
 384         PageSetLSN(page, lsn);
 385         PageSetTLI(page, ThisTimeLineID);
 386
 387         MarkBufferDirty(buffer);
 388         UnlockReleaseBuffer(buffer);
 389 }
 390
 391 static void
 392 ginRedoDeletePage(XLogRecPtr lsn, XLogRecord *record)
 393 {
 394         ginxlogDeletePage *data = (ginxlogDeletePage *) XLogRecGetData(record);
 395         Buffer          buffer;
 396         Page            page;
 397
 398         if (!(record->xl_info & XLR_BKP_BLOCK_1))
 399         {
 400                 buffer = XLogReadBuffer(data->node, data->blkno, false);
 401                 page = BufferGetPage(buffer);
 402                 Assert(GinPageIsData(page));
 403                 GinPageGetOpaque(page)->flags = GIN_DELETED;
 404                 PageSetLSN(page, lsn);
 405                 PageSetTLI(page, ThisTimeLineID);
 406                 MarkBufferDirty(buffer);
 407                 UnlockReleaseBuffer(buffer);
 408         }
 409
 410         if (!(record->xl_info & XLR_BKP_BLOCK_2))
 411         {
 412                 buffer = XLogReadBuffer(data->node, data->parentBlkno, false);
 413                 page = BufferGetPage(buffer);
 414                 Assert(GinPageIsData(page));
 415                 Assert(!GinPageIsLeaf(page));
 416                 PageDeletePostingItem(page, data->parentOffset);
 417                 PageSetLSN(page, lsn);
 418                 PageSetTLI(page, ThisTimeLineID);
 419                 MarkBufferDirty(buffer);
 420                 UnlockReleaseBuffer(buffer);
 421         }
 422
 423         if (!(record->xl_info & XLR_BKP_BLOCK_3) && data->leftBlkno != InvalidBlockNumber)
 424         {
 425                 buffer = XLogReadBuffer(data->node, data->leftBlkno, false);
 426                 page = BufferGetPage(buffer);
 427                 Assert(GinPageIsData(page));
 428                 GinPageGetOpaque(page)->rightlink = data->rightLink;
 429                 PageSetLSN(page, lsn);
 430                 PageSetTLI(page, ThisTimeLineID);
 431                 MarkBufferDirty(buffer);
 432                 UnlockReleaseBuffer(buffer);
 433         }
 434 }
 435
 436 void
 437 gin_redo(XLogRecPtr lsn, XLogRecord *record)
 438 {
 439         uint8           info = record->xl_info & ~XLR_INFO_MASK;
 440
 441         topCtx = MemoryContextSwitchTo(opCtx);
 442         switch (info)
 443         {
 444                 case XLOG_GIN_CREATE_INDEX:
 445                         ginRedoCreateIndex(lsn, record);
 446                         break;
 447                 case XLOG_GIN_CREATE_PTREE:
 448                         ginRedoCreatePTree(lsn, record);
 449                         break;
 450                 case XLOG_GIN_INSERT:
 451                         ginRedoInsert(lsn, record);
 452                         break;
 453                 case XLOG_GIN_SPLIT:
 454                         ginRedoSplit(lsn, record);
 455                         break;
 456                 case XLOG_GIN_VACUUM_PAGE:
 457                         ginRedoVacuumPage(lsn, record);
 458                         break;
 459                 case XLOG_GIN_DELETE_PAGE:
 460                         ginRedoDeletePage(lsn, record);
 461                         break;
 462                 default:
 463                         elog(PANIC, "gin_redo: unknown op code %u", info);
 464         }
 465         MemoryContextSwitchTo(topCtx);
 466         MemoryContextReset(opCtx);
 467 }
 468
 469 static void
 470 desc_node(StringInfo buf, RelFileNode node, BlockNumber blkno)
 471 {
 472         appendStringInfo(buf, "node: %u/%u/%u blkno: %u",
 473                                          node.spcNode, node.dbNode, node.relNode, blkno);
 474 }
 475
 476 void
 477 gin_desc(StringInfo buf, uint8 xl_info, char *rec)
 478 {
 479         uint8           info = xl_info & ~XLR_INFO_MASK;
 480
 481         switch (info)
 482         {
 483                 case XLOG_GIN_CREATE_INDEX:
 484                         appendStringInfo(buf, "Create index, ");
 485                         desc_node(buf, *(RelFileNode *) rec, GIN_ROOT_BLKNO);
 486                         break;
 487                 case XLOG_GIN_CREATE_PTREE:
 488                         appendStringInfo(buf, "Create posting tree, ");
 489                         desc_node(buf, ((ginxlogCreatePostingTree *) rec)->node, ((ginxlogCreatePostingTree *) rec)->blkno);
 490                         break;
 491                 case XLOG_GIN_INSERT:
 492                         appendStringInfo(buf, "Insert item, ");
 493                         desc_node(buf, ((ginxlogInsert *) rec)->node, ((ginxlogInsert *) rec)->blkno);
 494                         appendStringInfo(buf, " offset: %u nitem: %u isdata: %c isleaf %c isdelete %c updateBlkno:%u",
 495                                                          ((ginxlogInsert *) rec)->offset,
 496                                                          ((ginxlogInsert *) rec)->nitem,
 497                                                          (((ginxlogInsert *) rec)->isData) ? 'T' : 'F',
 498                                                          (((ginxlogInsert *) rec)->isLeaf) ? 'T' : 'F',
 499                                                          (((ginxlogInsert *) rec)->isDelete) ? 'T' : 'F',
 500                                                          ((ginxlogInsert *) rec)->updateBlkno
 501                                 );
 502
 503                         break;
 504                 case XLOG_GIN_SPLIT:
 505                         appendStringInfo(buf, "Page split, ");
 506                         desc_node(buf, ((ginxlogSplit *) rec)->node, ((ginxlogSplit *) rec)->lblkno);
 507                         appendStringInfo(buf, " isrootsplit: %c", (((ginxlogSplit *) rec)->isRootSplit) ? 'T' : 'F');
 508                         break;
 509                 case XLOG_GIN_VACUUM_PAGE:
 510                         appendStringInfo(buf, "Vacuum page, ");
 511                         desc_node(buf, ((ginxlogVacuumPage *) rec)->node, ((ginxlogVacuumPage *) rec)->blkno);
 512                         break;
 513                 case XLOG_GIN_DELETE_PAGE:
 514                         appendStringInfo(buf, "Delete page, ");
 515                         desc_node(buf, ((ginxlogDeletePage *) rec)->node, ((ginxlogDeletePage *) rec)->blkno);
 516                         break;
 517                 default:
 518                         elog(PANIC, "gin_desc: unknown op code %u", info);
 519         }
 520 }
 521
 522 void
 523 gin_xlog_startup(void)
 524 {
 525         incomplete_splits = NIL;
 526
 527         opCtx = AllocSetContextCreate(CurrentMemoryContext,
 528                                                                   "GIN recovery temporary context",
 529                                                                   ALLOCSET_DEFAULT_MINSIZE,
 530                                                                   ALLOCSET_DEFAULT_INITSIZE,
 531                                                                   ALLOCSET_DEFAULT_MAXSIZE);
 532 }
 533
 534 static void
 535 ginContinueSplit(ginIncompleteSplit *split)
 536 {
 537         GinBtreeData btree;
 538         Relation        reln;
 539         Buffer          buffer;
 540         GinBtreeStack stack;
 541
 542         /*
 543          * elog(NOTICE,"ginContinueSplit root:%u l:%u r:%u",  split->rootBlkno,
 544          * split->leftBlkno, split->rightBlkno);
 545          */
 546         buffer = XLogReadBuffer(split->node, split->leftBlkno, false);
 547
 548         reln = CreateFakeRelcacheEntry(split->node);
 549
 550         if (split->rootBlkno == GIN_ROOT_BLKNO)
 551         {
 552                 prepareEntryScan(&btree, reln, InvalidOffsetNumber, (Datum) 0, NULL);
 553                 btree.entry = ginPageGetLinkItup(buffer);
 554         }
 555         else
 556         {
 557                 Page            page = BufferGetPage(buffer);
 558
 559                 prepareDataScan(&btree, reln);
 560
 561                 PostingItemSetBlockNumber(&(btree.pitem), split->leftBlkno);
 562                 if (GinPageIsLeaf(page))
 563                         btree.pitem.key = *(ItemPointerData *) GinDataPageGetItem(page,
 564                                                                                          GinPageGetOpaque(page)->maxoff);
 565                 else
 566                         btree.pitem.key = ((PostingItem *) GinDataPageGetItem(page,
 567                                                                            GinPageGetOpaque(page)->maxoff))->key;
 568         }
 569
 570         FreeFakeRelcacheEntry(reln);
 571
 572         btree.rightblkno = split->rightBlkno;
 573
 574         stack.blkno = split->leftBlkno;
 575         stack.buffer = buffer;
 576         stack.off = InvalidOffsetNumber;
 577         stack.parent = NULL;
 578
 579         findParents(&btree, &stack, split->rootBlkno);
 580         ginInsertValue(&btree, stack.parent);
 581
 582         UnlockReleaseBuffer(buffer);
 583 }
 584
 585 void
 586 gin_xlog_cleanup(void)
 587 {
 588         ListCell   *l;
 589         MemoryContext topCtx;
 590
 591         topCtx = MemoryContextSwitchTo(opCtx);
 592
 593         foreach(l, incomplete_splits)
 594         {
 595                 ginIncompleteSplit *split = (ginIncompleteSplit *) lfirst(l);
 596
 597                 ginContinueSplit(split);
 598                 MemoryContextReset(opCtx);
 599         }
 600
 601         MemoryContextSwitchTo(topCtx);
 602         MemoryContextDelete(opCtx);
 603         incomplete_splits = NIL;
 604 }
 605
 606 bool
 607 gin_safe_restartpoint(void)
 608 {
 609         if (incomplete_splits)
 610                 return false;
 611         return true;
 612 }