More GIN refactoring.

Split off the portion of ginInsertValue that inserts the tuple to current
level into a separate function, ginPlaceToPage. ginInsertValue's charter
is now to recurse up the tree to insert the downlink, when a page split is
required.

This is in preparation for a patch to change the way incomplete splits are
handled, which will need to do these operations separately. And IMHO makes
the code more readable anyway.

diff --git a/src/backend/access/gin/ginbtree.c b/src/backend/access/gin/ginbtree.c
index c4801d5cd8937fbb3977733ab756427e5b5fe847..7248b06326fdb279d7cf7ce28369e472c73a17d7 100644 (file)
--- a/src/backend/access/gin/ginbtree.c
+++ b/src/backend/access/gin/ginbtree.c
@@ -280,80 +280,115 @@ ginFindParents(GinBtree btree, GinBtreeStack *stack,
 }
 
 /*
- * Insert value (stored in GinBtree) to tree described by stack
- *
- * During an index build, buildStats is non-null and the counters
- * it contains are incremented as needed.
+ * Returns true if the insertion is done, false if the page was split and
+ * downlink insertion is pending.
  *
- * NB: the passed-in stack is freed, as though by freeGinBtreeStack.
+ * stack->buffer is locked on entry, and is kept locked.
  */
-void
-ginInsertValue(GinBtree btree, GinBtreeStack *stack, GinStatsData *buildStats)
+static bool
+ginPlaceToPage(GinBtree btree, BlockNumber rootBlkno, GinBtreeStack *stack,
+              GinStatsData *buildStats)
 {
-   GinBtreeStack *parent;
-   BlockNumber rootBlkno;
-   Page        page,
-               rpage,
-               lpage;
+   Page        page = BufferGetPage(stack->buffer);
+   XLogRecData *rdata;
+   bool        fit;
 
-   /* extract root BlockNumber from stack */
-   Assert(stack != NULL);
-   parent = stack;
-   while (parent->parent)
-       parent = parent->parent;
-   rootBlkno = parent->blkno;
-   Assert(BlockNumberIsValid(rootBlkno));
+   START_CRIT_SECTION();
+   fit = btree->placeToPage(btree, stack->buffer, stack->off, &rdata);
+   if (fit)
+   {
+       MarkBufferDirty(stack->buffer);
 
-   /* this loop crawls up the stack until the insertion is complete */
-   for (;;)
+       if (RelationNeedsWAL(btree->index))
+       {
+           XLogRecPtr  recptr;
+
+           recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_INSERT, rdata);
+           PageSetLSN(page, recptr);
+       }
+
+       END_CRIT_SECTION();
+
+       return true;
+   }
+   else
    {
-       XLogRecData *rdata;
+       /* Didn't fit, have to split */
+       Buffer      rbuffer;
+       Page        newlpage;
        BlockNumber savedRightLink;
-       bool        fit;
+       GinBtreeStack *parent;
+       Page        lpage,
+                   rpage;
+
+       END_CRIT_SECTION();
+
+       rbuffer = GinNewBuffer(btree->index);
 
-       page = BufferGetPage(stack->buffer);
        savedRightLink = GinPageGetOpaque(page)->rightlink;
 
-       START_CRIT_SECTION();
-       fit = btree->placeToPage(btree, stack->buffer, stack->off, &rdata);
-       if (fit)
+       /*
+        * newlpage is a pointer to memory page, it is not associated with
+        * a buffer. stack->buffer is not touched yet.
+        */
+       newlpage = btree->splitPage(btree, stack->buffer, rbuffer, stack->off, &rdata);
+
+       ((ginxlogSplit *) (rdata->data))->rootBlkno = rootBlkno;
+
+       /* During index build, count the newly-split page */
+       if (buildStats)
        {
+           if (btree->isData)
+               buildStats->nDataPages++;
+           else
+               buildStats->nEntryPages++;
+       }
+
+       parent = stack->parent;
+
+       if (parent == NULL)
+       {
+           /*
+            * split root, so we need to allocate new left page and place
+            * pointer on root to left and right page
+            */
+           Buffer      lbuffer = GinNewBuffer(btree->index);
+
+           ((ginxlogSplit *) (rdata->data))->isRootSplit = TRUE;
+           ((ginxlogSplit *) (rdata->data))->rrlink = InvalidBlockNumber;
+
+           lpage = BufferGetPage(lbuffer);
+           rpage = BufferGetPage(rbuffer);
+
+           GinPageGetOpaque(rpage)->rightlink = InvalidBlockNumber;
+           GinPageGetOpaque(newlpage)->rightlink = BufferGetBlockNumber(rbuffer);
+           ((ginxlogSplit *) (rdata->data))->lblkno = BufferGetBlockNumber(lbuffer);
+
+           START_CRIT_SECTION();
+
+           GinInitBuffer(stack->buffer, GinPageGetOpaque(newlpage)->flags & ~GIN_LEAF);
+           PageRestoreTempPage(newlpage, lpage);
+           btree->fillRoot(btree, stack->buffer, lbuffer, rbuffer);
+
+           MarkBufferDirty(rbuffer);
+           MarkBufferDirty(lbuffer);
            MarkBufferDirty(stack->buffer);
 
            if (RelationNeedsWAL(btree->index))
            {
                XLogRecPtr  recptr;
 
-               recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_INSERT, rdata);
+               recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_SPLIT, rdata);
                PageSetLSN(page, recptr);
+               PageSetLSN(lpage, recptr);
+               PageSetLSN(rpage, recptr);
            }
 
-           LockBuffer(stack->buffer, GIN_UNLOCK);
-           END_CRIT_SECTION();
-
-           freeGinBtreeStack(stack);
-
-           return;
-       }
-       else
-       {
-           /* Didn't fit, have to split */
-           Buffer      rbuffer;
-           Page        newlpage;
-
+           UnlockReleaseBuffer(rbuffer);
+           UnlockReleaseBuffer(lbuffer);
            END_CRIT_SECTION();
 
-           rbuffer = GinNewBuffer(btree->index);
-
-           /*
-            * newlpage is a pointer to memory page, it is not associated with
-            * a buffer. stack->buffer is not touched yet.
-            */
-           newlpage = btree->splitPage(btree, stack->buffer, rbuffer, stack->off, &rdata);
-
-           ((ginxlogSplit *) (rdata->data))->rootBlkno = rootBlkno;
-
-           /* During index build, count the newly-split page */
+           /* During index build, count the newly-added root page */
            if (buildStats)
            {
                if (btree->isData)
@@ -362,98 +397,83 @@ ginInsertValue(GinBtree btree, GinBtreeStack *stack, GinStatsData *buildStats)
                    buildStats->nEntryPages++;
            }
 
-           parent = stack->parent;
-
-           if (parent == NULL)
-           {
-               /*
-                * split root, so we need to allocate new left page and place
-                * pointer on root to left and right page
-                */
-               Buffer      lbuffer = GinNewBuffer(btree->index);
-
-               ((ginxlogSplit *) (rdata->data))->isRootSplit = TRUE;
-               ((ginxlogSplit *) (rdata->data))->rrlink = InvalidBlockNumber;
-
-               page = BufferGetPage(stack->buffer);
-               lpage = BufferGetPage(lbuffer);
-               rpage = BufferGetPage(rbuffer);
-
-               GinPageGetOpaque(rpage)->rightlink = InvalidBlockNumber;
-               GinPageGetOpaque(newlpage)->rightlink = BufferGetBlockNumber(rbuffer);
-               ((ginxlogSplit *) (rdata->data))->lblkno = BufferGetBlockNumber(lbuffer);
-
-               START_CRIT_SECTION();
-
-               GinInitBuffer(stack->buffer, GinPageGetOpaque(newlpage)->flags & ~GIN_LEAF);
-               PageRestoreTempPage(newlpage, lpage);
-               btree->fillRoot(btree, stack->buffer, lbuffer, rbuffer);
-
-               MarkBufferDirty(rbuffer);
-               MarkBufferDirty(lbuffer);
-               MarkBufferDirty(stack->buffer);
+           return true;
+       }
+       else
+       {
+           /* split non-root page */
+           ((ginxlogSplit *) (rdata->data))->isRootSplit = FALSE;
+           ((ginxlogSplit *) (rdata->data))->rrlink = savedRightLink;
 
-               if (RelationNeedsWAL(btree->index))
-               {
-                   XLogRecPtr  recptr;
+           lpage = BufferGetPage(stack->buffer);
+           rpage = BufferGetPage(rbuffer);
 
-                   recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_SPLIT, rdata);
-                   PageSetLSN(page, recptr);
-                   PageSetLSN(lpage, recptr);
-                   PageSetLSN(rpage, recptr);
-               }
+           GinPageGetOpaque(rpage)->rightlink = savedRightLink;
+           GinPageGetOpaque(newlpage)->rightlink = BufferGetBlockNumber(rbuffer);
 
-               UnlockReleaseBuffer(rbuffer);
-               UnlockReleaseBuffer(lbuffer);
-               LockBuffer(stack->buffer, GIN_UNLOCK);
-               END_CRIT_SECTION();
+           START_CRIT_SECTION();
+           PageRestoreTempPage(newlpage, lpage);
 
-               freeGinBtreeStack(stack);
+           MarkBufferDirty(rbuffer);
+           MarkBufferDirty(stack->buffer);
 
-               /* During index build, count the newly-added root page */
-               if (buildStats)
-               {
-                   if (btree->isData)
-                       buildStats->nDataPages++;
-                   else
-                       buildStats->nEntryPages++;
-               }
+           if (RelationNeedsWAL(btree->index))
+           {
+               XLogRecPtr  recptr;
 
-               return;
+               recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_SPLIT, rdata);
+               PageSetLSN(lpage, recptr);
+               PageSetLSN(rpage, recptr);
            }
-           else
-           {
-               /* split non-root page */
-               ((ginxlogSplit *) (rdata->data))->isRootSplit = FALSE;
-               ((ginxlogSplit *) (rdata->data))->rrlink = savedRightLink;
+           UnlockReleaseBuffer(rbuffer);
+           END_CRIT_SECTION();
 
-               lpage = BufferGetPage(stack->buffer);
-               rpage = BufferGetPage(rbuffer);
+           return false;
+       }
+   }
+}
 
-               GinPageGetOpaque(rpage)->rightlink = savedRightLink;
-               GinPageGetOpaque(newlpage)->rightlink = BufferGetBlockNumber(rbuffer);
+/*
+ * Insert value (stored in GinBtree) to tree described by stack
+ *
+ * During an index build, buildStats is non-null and the counters
+ * it contains are incremented as needed.
+ *
+ * NB: the passed-in stack is freed, as though by freeGinBtreeStack.
+ */
+void
+ginInsertValue(GinBtree btree, GinBtreeStack *stack, GinStatsData *buildStats)
+{
+   GinBtreeStack *parent;
+   BlockNumber rootBlkno;
+   Page        page;
+
+   /* extract root BlockNumber from stack */
+   Assert(stack != NULL);
+   parent = stack;
+   while (parent->parent)
+       parent = parent->parent;
+   rootBlkno = parent->blkno;
+   Assert(BlockNumberIsValid(rootBlkno));
 
-               START_CRIT_SECTION();
-               PageRestoreTempPage(newlpage, lpage);
+   /* this loop crawls up the stack until the insertion is complete */
+   for (;;)
+   {
+       bool done;
 
-               MarkBufferDirty(rbuffer);
-               MarkBufferDirty(stack->buffer);
+       done = ginPlaceToPage(btree, rootBlkno, stack, buildStats);
 
-               if (RelationNeedsWAL(btree->index))
-               {
-                   XLogRecPtr  recptr;
+       /* just to be extra sure we don't delete anything by accident... */
+       btree->isDelete = FALSE;
 
-                   recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_SPLIT, rdata);
-                   PageSetLSN(lpage, recptr);
-                   PageSetLSN(rpage, recptr);
-               }
-               UnlockReleaseBuffer(rbuffer);
-               END_CRIT_SECTION();
-           }
+       if (done)
+       {
+           LockBuffer(stack->buffer, GIN_UNLOCK);
+           freeGinBtreeStack(stack);
+           break;
        }
 
        btree->prepareDownlink(btree, stack->buffer);
-       btree->isDelete = FALSE;
 
        /* search parent to lock */
        LockBuffer(parent->buffer, GIN_EXCLUSIVE);