1 files changed, 88 insertions, 52 deletions

diff --git a/src/backend/access/gin/ginentrypage.c b/src/backend/access/gin/ginentrypage.c
index 412f90da4db..bdf1f2e5889 100644
--- a/src/backend/access/gin/ginentrypage.c
+++ b/src/backend/access/gin/ginentrypage.c
@@ -20,9 +20,10 @@
 
 static void entrySplitPage(GinBtree btree, Buffer origbuf,
 			   GinBtreeStack *stack,
-			   void *insertPayload,
-			   BlockNumber updateblkno, XLogRecData **prdata,
-			   Page *newlpage, Page *newrpage);
+			   GinBtreeEntryInsertData *insertData,
+			   BlockNumber updateblkno,
+			   Page *newlpage, Page *newrpage,
+			   XLogRecData *rdata);
 
 /*
  * Form a tuple for entry tree.
@@ -507,40 +508,63 @@ entryPreparePage(GinBtree btree, Page page, OffsetNumber off,
 }
 
 /*
- * Place tuple on page and fills WAL record
+ * Prepare to insert data on an entry page.
  *
- * If the tuple doesn't fit, returns false without modifying the page.
+ * If it will fit, return GPTP_INSERT after doing whatever setup is needed
+ * before we enter the insertion critical section.  *ptp_workspace can be
+ * set to pass information along to the execPlaceToPage function.
  *
- * On insertion to an internal node, in addition to inserting the given item,
- * the downlink of the existing item at 'off' is updated to point to
- * 'updateblkno'.
+ * If it won't fit, perform a page split and return two temporary page
+ * images into *newlpage and *newrpage, with result GPTP_SPLIT.  Also,
+ * if WAL logging is needed, fill one or more entries of rdata[] with
+ * whatever data must be appended to the WAL record.
+ *
+ * In neither case should the given page buffer be modified here.
+ *
+ * Note: on insertion to an internal node, in addition to inserting the given
+ * item, the downlink of the existing item at stack->off will be updated to
+ * point to updateblkno.
  */
 static GinPlaceToPageRC
-entryPlaceToPage(GinBtree btree, Buffer buf, GinBtreeStack *stack,
-				 void *insertPayload, BlockNumber updateblkno,
-				 XLogRecData **prdata, Page *newlpage, Page *newrpage)
+entryBeginPlaceToPage(GinBtree btree, Buffer buf, GinBtreeStack *stack,
+					  void *insertPayload, BlockNumber updateblkno,
+					  void **ptp_workspace,
+					  Page *newlpage, Page *newrpage,
+					  XLogRecData *rdata)
 {
 	GinBtreeEntryInsertData *insertData = insertPayload;
-	Page		page = BufferGetPage(buf);
 	OffsetNumber off = stack->off;
-	OffsetNumber placed;
-	int			cnt = 0;
 
-	/* these must be static so they can be returned to caller */
-	static XLogRecData rdata[3];
-	static ginxlogInsertEntry data;
-
-	/* quick exit if it doesn't fit */
+	/* If it doesn't fit, deal with split case */
 	if (!entryIsEnoughSpace(btree, buf, off, insertData))
 	{
-		entrySplitPage(btree, buf, stack, insertPayload, updateblkno,
-					   prdata, newlpage, newrpage);
-		return SPLIT;
+		entrySplitPage(btree, buf, stack, insertData, updateblkno,
+					   newlpage, newrpage, rdata);
+		return GPTP_SPLIT;
 	}
 
-	START_CRIT_SECTION();
+	/* Else, we're ready to proceed with insertion */
+	return GPTP_INSERT;
+}
+
+/*
+ * Perform data insertion after beginPlaceToPage has decided it will fit.
+ *
+ * This is invoked within a critical section.  It must modify the target
+ * buffer and store one or more XLogRecData records describing the changes
+ * in rdata[].
+ */
+static void
+entryExecPlaceToPage(GinBtree btree, Buffer buf, GinBtreeStack *stack,
+					 void *insertPayload, BlockNumber updateblkno,
+					 void *ptp_workspace,
+					 XLogRecData *rdata)
+{
+	GinBtreeEntryInsertData *insertData = insertPayload;
+	Page		page = BufferGetPage(buf);
+	OffsetNumber off = stack->off;
+	OffsetNumber placed;
 
-	*prdata = rdata;
 	entryPreparePage(btree, page, off, insertData, updateblkno);
 
 	placed = PageAddItem(page,
@@ -551,39 +575,47 @@ entryPlaceToPage(GinBtree btree, Buffer buf, GinBtreeStack *stack,
 		elog(ERROR, "failed to add item to index page in \"%s\"",
 			 RelationGetRelationName(btree->index));
 
-	data.isDelete = insertData->isDelete;
-	data.offset = off;
-
-	rdata[cnt].buffer = buf;
-	rdata[cnt].buffer_std = true;
-	rdata[cnt].data = (char *) &data;
-	rdata[cnt].len = offsetof(ginxlogInsertEntry, tuple);
-	rdata[cnt].next = &rdata[cnt + 1];
-	cnt++;
-
-	rdata[cnt].buffer = buf;
-	rdata[cnt].buffer_std = true;
-	rdata[cnt].data = (char *) insertData->entry;
-	rdata[cnt].len = IndexTupleSize(insertData->entry);
-	rdata[cnt].next = NULL;
-
-	return INSERTED;
+	if (RelationNeedsWAL(btree->index))
+	{
+		/*
+		 * This must be static, because it has to survive until XLogInsert,
+		 * and we can't palloc here.  Ugly, but the XLogInsert infrastructure
+		 * isn't reentrant anyway.
+		 */
+		static ginxlogInsertEntry data;
+
+		data.isDelete = insertData->isDelete;
+		data.offset = off;
+
+		rdata[0].buffer = buf;
+		rdata[0].buffer_std = true;
+		rdata[0].data = (char *) &data;
+		rdata[0].len = offsetof(ginxlogInsertEntry, tuple);
+		rdata[0].next = &rdata[1];
+
+		rdata[1].buffer = buf;
+		rdata[1].buffer_std = true;
+		rdata[1].data = (char *) insertData->entry;
+		rdata[1].len = IndexTupleSize(insertData->entry);
+		rdata[1].next = NULL;
+	}
 }
 
 /*
- * Place tuple and split page, original buffer(lbuf) leaves untouched,
- * returns shadow pages filled with new data.
- * Tuples are distributed between pages by equal size on its, not
- * an equal number!
+ * Split entry page and insert new data.
+ *
+ * Returns new temp pages to *newlpage and *newrpage.
+ * The original buffer is left untouched.
+ * Also, set up rdata[] entries describing data to be appended to WAL record.
  */
 static void
 entrySplitPage(GinBtree btree, Buffer origbuf,
 			   GinBtreeStack *stack,
-			   void *insertPayload,
-			   BlockNumber updateblkno, XLogRecData **prdata,
-			   Page *newlpage, Page *newrpage)
+			   GinBtreeEntryInsertData *insertData,
+			   BlockNumber updateblkno,
+			   Page *newlpage, Page *newrpage,
+			   XLogRecData *rdata)
 {
-	GinBtreeEntryInsertData *insertData = insertPayload;
 	OffsetNumber off = stack->off;
 	OffsetNumber i,
 				maxoff,
@@ -600,11 +632,9 @@ entrySplitPage(GinBtree btree, Buffer origbuf,
 	Size		pageSize = PageGetPageSize(lpage);
 
 	/* these must be static so they can be returned to caller */
-	static XLogRecData rdata[2];
 	static ginxlogSplitEntry data;
 	static char tupstore[2 * BLCKSZ];
 
-	*prdata = rdata;
 	entryPreparePage(btree, lpage, off, insertData, updateblkno);
 
 	/*
@@ -655,6 +685,10 @@ entrySplitPage(GinBtree btree, Buffer origbuf,
 	{
 		itup = (IndexTuple) ptr;
 
+		/*
+		 * Decide where to split.  We try to equalize the pages' total data
+		 * size, not number of tuples.
+		 */
 		if (lsize > totalsize / 2)
 		{
 			if (separator == InvalidOffsetNumber)
@@ -685,6 +719,7 @@ entrySplitPage(GinBtree btree, Buffer origbuf,
 	rdata[1].len = tupstoresize;
 	rdata[1].next = NULL;
 
+	/* return temp pages to caller */
 	*newlpage = lpage;
 	*newrpage = rpage;
 }
@@ -753,7 +788,8 @@ ginPrepareEntryScan(GinBtree btree, OffsetNumber attnum,
 	btree->isMoveRight = entryIsMoveRight;
 	btree->findItem = entryLocateLeafEntry;
 	btree->findChildPtr = entryFindChildPtr;
-	btree->placeToPage = entryPlaceToPage;
+	btree->beginPlaceToPage = entryBeginPlaceToPage;
+	btree->execPlaceToPage = entryExecPlaceToPage;
 	btree->fillRoot = ginEntryFillRoot;
 	btree->prepareDownlink = entryPrepareDownlink;