More GIN refactoring.

Separate the insertion payload from the more static portions of GinBtree.
GinBtree now only contains information related to searching the tree, and
the information of what to insert is passed separately.

Add root block number to GinBtree, instead of passing it around all the
functions as argument.

Split off ginFinishSplit() from ginInsertValue(). ginFinishSplit is
responsible for finding the parent and inserting the downlink to it.

1 files changed, 113 insertions, 69 deletions

diff --git a/src/backend/access/gin/ginbtree.c b/src/backend/access/gin/ginbtree.c
index 7248b06326f..f3c0dfd128f 100644
--- a/src/backend/access/gin/ginbtree.c
+++ b/src/backend/access/gin/ginbtree.c
@@ -19,7 +19,7 @@
 #include "utils/rel.h"
 
 /*
- * Locks buffer by needed method for search.
+ * Lock buffer by needed method for search.
  */
 static int
 ginTraverseLock(Buffer buffer, bool searchMode)
@@ -53,9 +53,9 @@ ginTraverseLock(Buffer buffer, bool searchMode)
 }
 
 /*
- * Descends the tree to the leaf page that contains or would contain the
- * key we're searching for. The key should already be filled in 'btree',
- * in tree-type specific manner. If btree->fullScan is true, descends to the
+ * Descend the tree to the leaf page that contains or would contain the key
+ * we're searching for. The key should already be filled in 'btree', in
+ * tree-type specific manner. If btree->fullScan is true, descends to the
  * leftmost leaf page.
  *
  * If 'searchmode' is false, on return stack->buffer is exclusively locked,
@@ -63,13 +63,13 @@ ginTraverseLock(Buffer buffer, bool searchMode)
  * is share-locked, and stack->parent is NULL.
  */
 GinBtreeStack *
-ginFindLeafPage(GinBtree btree, BlockNumber rootBlkno, bool searchMode)
+ginFindLeafPage(GinBtree btree, bool searchMode)
 {
 	GinBtreeStack *stack;
 
 	stack = (GinBtreeStack *) palloc(sizeof(GinBtreeStack));
-	stack->blkno = rootBlkno;
-	stack->buffer = ReadBuffer(btree->index, rootBlkno);
+	stack->blkno = btree->rootBlkno;
+	stack->buffer = ReadBuffer(btree->index, btree->rootBlkno);
 	stack->parent = NULL;
 	stack->predictNumber = 1;
 
@@ -89,7 +89,7 @@ ginFindLeafPage(GinBtree btree, BlockNumber rootBlkno, bool searchMode)
 		 * ok, page is correctly locked, we should check to move right ..,
 		 * root never has a right link, so small optimization
 		 */
-		while (btree->fullScan == FALSE && stack->blkno != rootBlkno &&
+		while (btree->fullScan == FALSE && stack->blkno != btree->rootBlkno &&
 			   btree->isMoveRight(btree, page))
 		{
 			BlockNumber rightlink = GinPageGetOpaque(page)->rightlink;
@@ -146,7 +146,7 @@ ginStepRight(Buffer buffer, Relation index, int lockmode)
 	Page		page = BufferGetPage(buffer);
 	bool		isLeaf = GinPageIsLeaf(page);
 	bool		isData = GinPageIsData(page);
-	BlockNumber	blkno = GinPageGetOpaque(page)->rightlink;
+	BlockNumber blkno = GinPageGetOpaque(page)->rightlink;
 
 	nextbuffer = ReadBuffer(index, blkno);
 	LockBuffer(nextbuffer, lockmode);
@@ -158,10 +158,10 @@ ginStepRight(Buffer buffer, Relation index, int lockmode)
 		elog(ERROR, "right sibling of GIN page is of different type");
 
 	/*
-	 * Given the proper lock sequence above, we should never land on a
-	 * deleted page.
+	 * Given the proper lock sequence above, we should never land on a deleted
+	 * page.
 	 */
-	if  (GinPageIsDeleted(page))
+	if (GinPageIsDeleted(page))
 		elog(ERROR, "right sibling of GIN page was deleted");
 
 	return nextbuffer;
@@ -183,14 +183,12 @@ freeGinBtreeStack(GinBtreeStack *stack)
 }
 
 /*
- * Try to find parent for current stack position, returns correct
- * parent and child's offset in  stack->parent.
- * Function should never release root page to prevent conflicts
- * with vacuum process
+ * Try to find parent for current stack position. Returns correct parent and
+ * child's offset in stack->parent. The root page is never released, to
+ * to prevent conflict with vacuum process.
  */
 void
-ginFindParents(GinBtree btree, GinBtreeStack *stack,
-			   BlockNumber rootBlkno)
+ginFindParents(GinBtree btree, GinBtreeStack *stack)
 {
 	Page		page;
 	Buffer		buffer;
@@ -204,8 +202,8 @@ ginFindParents(GinBtree btree, GinBtreeStack *stack,
 	{
 		/* XLog mode... */
 		root = (GinBtreeStack *) palloc(sizeof(GinBtreeStack));
-		root->blkno = rootBlkno;
-		root->buffer = ReadBuffer(btree->index, rootBlkno);
+		root->blkno = btree->rootBlkno;
+		root->buffer = ReadBuffer(btree->index, btree->rootBlkno);
 		LockBuffer(root->buffer, GIN_EXCLUSIVE);
 		root->parent = NULL;
 	}
@@ -221,8 +219,8 @@ ginFindParents(GinBtree btree, GinBtreeStack *stack,
 			root = root->parent;
 		}
 
-		Assert(root->blkno == rootBlkno);
-		Assert(BufferGetBlockNumber(root->buffer) == rootBlkno);
+		Assert(root->blkno == btree->rootBlkno);
+		Assert(BufferGetBlockNumber(root->buffer) == btree->rootBlkno);
 		LockBuffer(root->buffer, GIN_EXCLUSIVE);
 	}
 	root->off = InvalidOffsetNumber;
@@ -268,7 +266,7 @@ ginFindParents(GinBtree btree, GinBtreeStack *stack,
 			ptr = (GinBtreeStack *) palloc(sizeof(GinBtreeStack));
 			ptr->blkno = blkno;
 			ptr->buffer = buffer;
-			ptr->parent = root; /* it's may be wrong, but in next call we will
+			ptr->parent = root; /* it may be wrong, but in next call we will
 								 * correct */
 			ptr->off = offset;
 			stack->parent = ptr;
@@ -280,21 +278,35 @@ ginFindParents(GinBtree btree, GinBtreeStack *stack,
 }
 
 /*
- * Returns true if the insertion is done, false if the page was split and
- * downlink insertion is pending.
+ * Insert a new item to a page.
+ *
+ * Returns true if the insertion was finished. On false, the page was split and
+ * the parent needs to be updated. (a root split returns true as it doesn't
+ * need any further action by the caller to complete)
+ *
+ * When inserting a downlink to a internal page, the existing item at the
+ * given location is updated to point to 'updateblkno'.
  *
  * stack->buffer is locked on entry, and is kept locked.
  */
 static bool
-ginPlaceToPage(GinBtree btree, BlockNumber rootBlkno, GinBtreeStack *stack,
+ginPlaceToPage(GinBtree btree, GinBtreeStack *stack,
+			   void *insertdata, BlockNumber updateblkno,
 			   GinStatsData *buildStats)
 {
 	Page		page = BufferGetPage(stack->buffer);
 	XLogRecData *rdata;
 	bool		fit;
 
+	/*
+	 * Try to put the incoming tuple on the page. If it doesn't fit,
+	 * placeToPage method will return false and leave the page unmodified, and
+	 * we'll have to split the page.
+	 */
 	START_CRIT_SECTION();
-	fit = btree->placeToPage(btree, stack->buffer, stack->off, &rdata);
+	fit = btree->placeToPage(btree, stack->buffer, stack->off,
+							 insertdata, updateblkno,
+							 &rdata);
 	if (fit)
 	{
 		MarkBufferDirty(stack->buffer);
@@ -324,18 +336,7 @@ ginPlaceToPage(GinBtree btree, BlockNumber rootBlkno, GinBtreeStack *stack,
 		END_CRIT_SECTION();
 
 		rbuffer = GinNewBuffer(btree->index);
-
-		savedRightLink = GinPageGetOpaque(page)->rightlink;
-
-		/*
-		 * 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 new page */
 		if (buildStats)
 		{
 			if (btree->isData)
@@ -344,6 +345,18 @@ ginPlaceToPage(GinBtree btree, BlockNumber rootBlkno, GinBtreeStack *stack,
 				buildStats->nEntryPages++;
 		}
 
+		savedRightLink = GinPageGetOpaque(page)->rightlink;
+
+		/*
+		 * 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,
+									insertdata, updateblkno,
+									&rdata);
+
+		((ginxlogSplit *) (rdata->data))->rootBlkno = btree->rootBlkno;
+
 		parent = stack->parent;
 
 		if (parent == NULL)
@@ -354,6 +367,15 @@ ginPlaceToPage(GinBtree btree, BlockNumber rootBlkno, GinBtreeStack *stack,
 			 */
 			Buffer		lbuffer = GinNewBuffer(btree->index);
 
+			/* During index build, count the new page */
+			if (buildStats)
+			{
+				if (btree->isData)
+					buildStats->nDataPages++;
+				else
+					buildStats->nEntryPages++;
+			}
+
 			((ginxlogSplit *) (rdata->data))->isRootSplit = TRUE;
 			((ginxlogSplit *) (rdata->data))->rrlink = InvalidBlockNumber;
 
@@ -434,46 +456,27 @@ ginPlaceToPage(GinBtree btree, BlockNumber rootBlkno, GinBtreeStack *stack,
 }
 
 /*
- * Insert value (stored in GinBtree) to tree described by stack
+ * Finish a split by inserting the downlink for the new page to parent.
  *
- * During an index build, buildStats is non-null and the counters
- * it contains are incremented as needed.
+ * On entry, stack->buffer is exclusively locked.
  *
  * NB: the passed-in stack is freed, as though by freeGinBtreeStack.
  */
 void
-ginInsertValue(GinBtree btree, GinBtreeStack *stack, GinStatsData *buildStats)
+ginFinishSplit(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));
+	bool		done;
 
 	/* this loop crawls up the stack until the insertion is complete */
-	for (;;)
+	do
 	{
-		bool done;
-
-		done = ginPlaceToPage(btree, rootBlkno, stack, buildStats);
-
-		/* just to be extra sure we don't delete anything by accident... */
-		btree->isDelete = FALSE;
-
-		if (done)
-		{
-			LockBuffer(stack->buffer, GIN_UNLOCK);
-			freeGinBtreeStack(stack);
-			break;
-		}
+		GinBtreeStack *parent = stack->parent;
+		void	   *insertdata;
+		BlockNumber updateblkno;
 
-		btree->prepareDownlink(btree, stack->buffer);
+		insertdata = btree->prepareDownlink(btree, stack->buffer);
+		updateblkno = GinPageGetOpaque(BufferGetPage(stack->buffer))->rightlink;
 
 		/* search parent to lock */
 		LockBuffer(parent->buffer, GIN_EXCLUSIVE);
@@ -491,7 +494,7 @@ ginInsertValue(GinBtree btree, GinBtreeStack *stack, GinStatsData *buildStats)
 				 * plain search...
 				 */
 				LockBuffer(parent->buffer, GIN_UNLOCK);
-				ginFindParents(btree, stack, rootBlkno);
+				ginFindParents(btree, stack);
 				parent = stack->parent;
 				Assert(parent != NULL);
 				break;
@@ -502,8 +505,49 @@ ginInsertValue(GinBtree btree, GinBtreeStack *stack, GinStatsData *buildStats)
 			page = BufferGetPage(parent->buffer);
 		}
 
+		/* release the child */
 		UnlockReleaseBuffer(stack->buffer);
 		pfree(stack);
 		stack = parent;
+
+		/* insert the downlink to parent */
+		done = ginPlaceToPage(btree, stack,
+							  insertdata, updateblkno,
+							  buildStats);
+		pfree(insertdata);
+	} while (!done);
+	LockBuffer(stack->buffer, GIN_UNLOCK);
+
+	/* free the rest of the stack */
+	freeGinBtreeStack(stack);
+}
+
+/*
+ * Insert a value to tree described by stack.
+ *
+ * The value to be inserted is given in 'insertdata'. Its format depends
+ * on whether this is an entry or data tree, ginInsertValue just passes it
+ * through to the tree-specific callback function.
+ *
+ * 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, void *insertdata,
+			   GinStatsData *buildStats)
+{
+	bool		done;
+
+	done = ginPlaceToPage(btree, stack,
+						  insertdata, InvalidBlockNumber,
+						  buildStats);
+	if (done)
+	{
+		LockBuffer(stack->buffer, GIN_UNLOCK);
+		freeGinBtreeStack(stack);
 	}
+	else
+		ginFinishSplit(btree, stack, buildStats);
 }