diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c index b756f6e..9a33597 100644 --- a/src/backend/access/gist/gist.c +++ b/src/backend/access/gist/gist.c @@ -59,6 +59,10 @@ static bool gistinserttuples(GISTInsertState *state, GISTInsertStack *stack, static void gistfinishsplit(GISTInsertState *state, GISTInsertStack *stack, GISTSTATE *giststate, List *splitinfo); +static void gistFindCorrectParent(Relation r, GISTInsertStack *child); +static GISTInsertStack *gistFindPath(Relation r, BlockNumber child, + OffsetNumber *newchildoffnum); + #define ROTATEDIST(d) do { \ SplitedPageLayout *tmp=(SplitedPageLayout*)palloc(sizeof(SplitedPageLayout)); \ @@ -626,6 +630,7 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate) firststack.blkno = GIST_ROOT_BLKNO; firststack.lsn.xrecoff = 0; firststack.parent = NULL; + firststack.childoffnum = InvalidOffsetNumber; state.stack = stack = &firststack; /* @@ -702,9 +707,10 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate) BlockNumber childblkno; IndexTuple newtup; GISTInsertStack *item; + OffsetNumber childoffnum; - stack->childoffnum = gistchoose(state.r, stack->page, itup, giststate); - iid = PageGetItemId(stack->page, stack->childoffnum); + childoffnum = gistchoose(state.r, stack->page, itup, giststate); + iid = PageGetItemId(stack->page, childoffnum); idxtuple = (IndexTuple) PageGetItem(stack->page, iid); childblkno = ItemPointerGetBlockNumber(&(idxtuple->t_tid)); @@ -754,7 +760,7 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate) * tuple. */ if (gistinserttuples(&state, stack, giststate, &newtup, 1, - stack->childoffnum, InvalidBuffer)) + childoffnum, InvalidBuffer)) { /* * If this was a root split, the root page continues to be @@ -778,6 +784,7 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate) item = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack)); item->blkno = childblkno; item->parent = stack; + item->childoffnum = childoffnum; state.stack = stack = item; } else @@ -852,14 +859,16 @@ gistdoinsert(Relation r, IndexTuple itup, Size freespace, GISTSTATE *giststate) } /* - * Traverse the tree to find path from root page to specified "child" block. + * Traverse the tree to find path from root page to specified block. * - * returns from the beginning of closest parent; + * returns a new insertion stack, starting from the parent of 'child', up + * to the root. *newchildoffnum is set to the offset number of the downlink + * in the direct parent of child that points to the child. * * To prevent deadlocks, this should lock only one page at a time. */ GISTInsertStack * -gistFindPath(Relation r, BlockNumber child) +gistFindPath(Relation r, BlockNumber child, OffsetNumber *newchildoffnum) { Page page; Buffer buffer; @@ -867,16 +876,22 @@ gistFindPath(Relation r, BlockNumber child) maxoff; ItemId iid; IndexTuple idxtuple; + List *fifo; GISTInsertStack *top, - *tail, *ptr; BlockNumber blkno; - top = tail = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack)); + top = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack)); top->blkno = GIST_ROOT_BLKNO; + top->childoffnum = InvalidOffsetNumber; - while (top && top->blkno != child) + fifo = list_make1(top); + while (fifo != NIL) { + /* Get next page from the fifo */ + top = linitial(fifo); + fifo = list_delete_first(fifo); + buffer = ReadBuffer(r, top->blkno); LockBuffer(buffer, GIST_SHARE); gistcheckpage(r, buffer); @@ -884,9 +899,12 @@ gistFindPath(Relation r, BlockNumber child) if (GistPageIsLeaf(page)) { - /* we can safety go away, follows only leaf pages */ + /* + * we can stop looking, there are only leaf pages left in the + * stack. + */ UnlockReleaseBuffer(buffer); - return NULL; + break; } top->lsn = PageGetLSN(page); @@ -901,14 +919,18 @@ gistFindPath(Relation r, BlockNumber child) if (top->parent && XLByteLT(top->parent->lsn, GistPageGetOpaque(page)->nsn) && GistPageGetOpaque(page)->rightlink != InvalidBlockNumber /* sanity check */ ) { - /* page splited while we thinking of... */ + /* + * Page was split while we were looking elsewhere. We didn't see + * the downlink to page to the right of this one, when we + * processed the parent level. Append it to the list of pages to + * visit later. + */ ptr = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack)); ptr->blkno = GistPageGetOpaque(page)->rightlink; ptr->childoffnum = InvalidOffsetNumber; ptr->parent = top; - ptr->next = NULL; - tail->next = ptr; - tail = ptr; + + fifo = lappend(fifo, ptr); } maxoff = PageGetMaxOffsetNumber(page); @@ -920,50 +942,27 @@ gistFindPath(Relation r, BlockNumber child) blkno = ItemPointerGetBlockNumber(&(idxtuple->t_tid)); if (blkno == child) { - OffsetNumber poff = InvalidOffsetNumber; - - /* make childs links */ - ptr = top; - while (ptr->parent) - { - /* move childoffnum.. */ - if (ptr == top) - { - /* first iteration */ - poff = ptr->parent->childoffnum; - ptr->parent->childoffnum = ptr->childoffnum; - } - else - { - OffsetNumber tmp = ptr->parent->childoffnum; - - ptr->parent->childoffnum = poff; - poff = tmp; - } - ptr = ptr->parent; - } - top->childoffnum = i; + /* Found it! */ UnlockReleaseBuffer(buffer); - return top; + *newchildoffnum = blkno; + return ptr; } else { - /* Install next inner page to the end of stack */ + /* Append this child to the list of page to visit later */ ptr = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack)); ptr->blkno = blkno; - ptr->childoffnum = i; /* set offsetnumber of child to child - * !!! */ + ptr->childoffnum = i; ptr->parent = top; - ptr->next = NULL; - tail->next = ptr; - tail = ptr; + + fifo = lappend(fifo, ptr); } } UnlockReleaseBuffer(buffer); - top = top->next; } + elog(ERROR, "failed to re-find parent of gist page %u", child); return NULL; } @@ -981,7 +980,8 @@ gistFindCorrectParent(Relation r, GISTInsertStack *child) parent->page = (Page) BufferGetPage(parent->buffer); /* here we don't need to distinguish between split and page update */ - if (parent->childoffnum == InvalidOffsetNumber || !XLByteEQ(parent->lsn, PageGetLSN(parent->page))) + if (child->childoffnum == InvalidOffsetNumber || + !XLByteEQ(parent->lsn, PageGetLSN(parent->page))) { /* parent is changed, look child in right links until found */ OffsetNumber i, @@ -1000,7 +1000,7 @@ gistFindCorrectParent(Relation r, GISTInsertStack *child) if (ItemPointerGetBlockNumber(&(idxtuple->t_tid)) == child->blkno) { /* yes!!, found */ - parent->childoffnum = i; + child->childoffnum = i; return; } } @@ -1034,7 +1034,7 @@ gistFindCorrectParent(Relation r, GISTInsertStack *child) } /* ok, find new path */ - ptr = parent = gistFindPath(r, child->blkno); + ptr = parent = gistFindPath(r, child->blkno, &child->childoffnum); Assert(ptr != NULL); /* read all buffers as expected by caller */ @@ -1104,7 +1104,7 @@ gistformdownlink(Relation rel, Buffer buf, GISTSTATE *giststate, LockBuffer(stack->parent->buffer, GIST_EXCLUSIVE); gistFindCorrectParent(rel, stack); - iid = PageGetItemId(stack->parent->page, stack->parent->childoffnum); + iid = PageGetItemId(stack->parent->page, stack->childoffnum); downlink = (IndexTuple) PageGetItem(stack->parent->page, iid); downlink = CopyIndexTuple(downlink); LockBuffer(stack->parent->buffer, GIST_UNLOCK); @@ -1132,7 +1132,7 @@ gistfixsplit(GISTInsertState *state, GISTSTATE *giststate) RelationGetRelationName(state->r), stack->blkno); Assert(GistFollowRight(stack->page)); - Assert(OffsetNumberIsValid(stack->parent->childoffnum)); + Assert(OffsetNumberIsValid(stack->childoffnum)); buf = stack->buffer; @@ -1269,7 +1269,7 @@ gistfinishsplit(GISTInsertState *state, GISTInsertStack *stack, tuples[1] = right->downlink; gistinserttuples(state, stack->parent, giststate, tuples, 2, - stack->parent->childoffnum, + stack->childoffnum, left->buf); LockBuffer(stack->parent->buffer, GIST_UNLOCK); UnlockReleaseBuffer(right->buf); diff --git a/src/include/access/gist_private.h b/src/include/access/gist_private.h index 77e3cb5..ca9fd25 100644 --- a/src/include/access/gist_private.h +++ b/src/include/access/gist_private.h @@ -223,9 +223,6 @@ typedef struct GISTInsertStack /* pointer to parent */ struct GISTInsertStack *parent; - - /* for gistFindPath */ - struct GISTInsertStack *next; } GISTInsertStack; typedef struct GistSplitVector @@ -293,8 +290,6 @@ extern void freeGISTstate(GISTSTATE *giststate); extern SplitedPageLayout *gistSplit(Relation r, Page page, IndexTuple *itup, int len, GISTSTATE *giststate); -extern GISTInsertStack *gistFindPath(Relation r, BlockNumber child); - /* gistxlog.c */ extern void gist_redo(XLogRecPtr lsn, XLogRecord *record); extern void gist_desc(StringInfo buf, uint8 xl_info, char *rec);