gist-split-hotstandby-90.patch

diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c
index 82ba726..71c145d 100644
--- a/src/backend/access/gist/gist.c
+++ b/src/backend/access/gist/gist.c
@@ -377,9 +377,18 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 			state->ituplen++;
 		}
 
-		/* saves old rightlink */
+		/* save old rightlink and NSN */
 		if (state->stack->blkno != GIST_ROOT_BLKNO)
+		{
 			rrlink = GistPageGetOpaque(dist->page)->rightlink;
+			oldnsn = GistPageGetOpaque(dist->page)->nsn;
+		}
+		else
+		{
+			/* if root split we should put initial value */
+			rrlink = InvalidBlockNumber;
+			oldnsn = PageGetLSN(dist->page);
+		}
 
 		START_CRIT_SECTION();
 
@@ -407,7 +416,8 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 			XLogRecData *rdata;
 
 			rdata = formSplitRdata(state->r->rd_node, state->stack->blkno,
-								   is_leaf, &(state->key), dist);
+								   is_leaf, &(state->key), dist,
+								   rrlink, &oldnsn);
 
 			recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_SPLIT, rdata);
 
@@ -425,12 +435,6 @@ gistplacetopage(GISTInsertState *state, GISTSTATE *giststate)
 			}
 		}
 
-		/* set up NSN */
-		oldnsn = GistPageGetOpaque(dist->page)->nsn;
-		if (state->stack->blkno == GIST_ROOT_BLKNO)
-			/* if root split we should put initial value */
-			oldnsn = PageGetLSN(dist->page);
-
 		for (ptr = dist; ptr; ptr = ptr->next)
 		{
 			/* only for last set oldnsn */
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 7f5dd99..cdd8aaf 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -39,6 +39,8 @@ typedef struct
 {
 	gistxlogPageSplit *data;
 	NewPage    *page;
+	BlockNumber origrlink;
+	XLogRecPtr	orignsn;
 } PageSplitRecord;
 
 /* track for incomplete inserts, idea was taken from nbtxlog.c */
@@ -250,7 +252,6 @@ gistRedoPageUpdateRecord(XLogRecPtr lsn, XLogRecord *record, bool isnewroot)
 		 */
 		GistPageSetLeaf(page);
 
-	GistPageGetOpaque(page)->rightlink = InvalidBlockNumber;
 	PageSetLSN(page, lsn);
 	PageSetTLI(page, ThisTimeLineID);
 	MarkBufferDirty(buffer);
@@ -310,6 +311,26 @@ decodePageSplitRecord(PageSplitRecord *decoded, XLogRecord *record)
 			j++;
 		}
 	}
+
+	/*
+	 * Starting with 9.0.5, the original NSN and rightlink on the split page
+	 * are stored here. It would've been more logical to add them to the
+	 * gistxlogPageSplit struct, but that would've broken compatibility with
+	 * the pre-9.0.5 WAL format.
+	 */
+	if (ptr - begin < record->xl_len)
+	{
+		memcpy(&decoded->origrlink, ptr, sizeof(BlockNumber));
+		ptr += sizeof(BlockNumber);
+		memcpy(&decoded->orignsn, ptr, sizeof(XLogRecPtr));
+	}
+	else
+	{
+		/* pre-9.0.5 format, no rightlink/NSN information */
+		decoded->origrlink = InvalidBlockNumber;
+		decoded->orignsn.xlogid = 0;
+		decoded->orignsn.xrecoff = 0;
+	}
 }
 
 static void
@@ -320,17 +341,32 @@ gistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record)
 	Page		page;
 	int			i;
 	int			flags;
+	Buffer	   *buffers;
 
 	decodePageSplitRecord(&xlrec, record);
 	flags = xlrec.data->origleaf ? F_LEAF : 0;
 
-	/* loop around all pages */
+	/*
+	 * Lock all the pages involved in the split first, so that any concurrent
+	 * scans in hot standby mode will see the split as an atomic operation.
+	 */
+	buffers = palloc(xlrec.data->npage * sizeof(Buffer));
 	for (i = 0; i < xlrec.data->npage; i++)
 	{
 		NewPage    *newpage = xlrec.page + i;
 
-		buffer = XLogReadBuffer(xlrec.data->node, newpage->header->blkno, true);
-		Assert(BufferIsValid(buffer));
+		buffers[i] = XLogReadBuffer(xlrec.data->node,
+									newpage->header->blkno,
+									true);
+		page = (Page) BufferGetPage(buffers[i]);
+	}
+
+	/* Write out all the pages */
+	for (i = 0; i < xlrec.data->npage; i++)
+	{
+		NewPage    *newpage = xlrec.page + i;
+
+		buffer = buffers[i];
 		page = (Page) BufferGetPage(buffer);
 
 		/* ok, clear buffer */
@@ -339,6 +375,18 @@ gistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record)
 		/* and fill it */
 		gistfillbuffer(page, newpage->itup, newpage->header->num, FirstOffsetNumber);
 
+		/* Set NSN and rightlink, needed for concurrent scans in hot standby */
+		if (i == xlrec.data->npage - 1)
+		{
+			GistPageGetOpaque(page)->nsn = xlrec.orignsn;
+			GistPageGetOpaque(page)->rightlink = xlrec.origrlink;
+		}
+		else
+		{
+			GistPageGetOpaque(page)->nsn = lsn;
+			GistPageGetOpaque(page)->rightlink = xlrec.page[i + 1].header->blkno;
+		}
+
 		PageSetLSN(page, lsn);
 		PageSetTLI(page, ThisTimeLineID);
 		MarkBufferDirty(buffer);
@@ -350,6 +398,8 @@ gistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record)
 	pushIncompleteInsert(xlrec.data->node, lsn, xlrec.data->key,
 						 NULL, 0,
 						 &xlrec);
+
+	pfree(buffers);
 }
 
 static void
@@ -655,6 +705,8 @@ gistContinueInsert(gistIncompleteInsert *insert)
 			XLogRecPtr	recptr;
 			Buffer		tempbuffer = InvalidBuffer;
 			int			ntodelete = 0;
+			BlockNumber	rrlink;
+			XLogRecPtr	oldnsn;
 
 			numbuffer = 1;
 			buffers[0] = ReadBuffer(index, insert->path[i]);
@@ -691,6 +743,10 @@ gistContinueInsert(gistIncompleteInsert *insert)
 			if (ntodelete == 0)
 				elog(PANIC, "gistContinueInsert: cannot find pointer to page(s)");
 
+			/* Remember old rightlink and NSN */
+			rrlink = GistPageGetOpaque(pages[0])->rightlink;
+			oldnsn = GistPageGetOpaque(pages[0])->nsn;
+
 			/*
 			 * we check space with subtraction only first tuple to delete,
 			 * hope, that wiil be enough space....
@@ -742,7 +798,8 @@ gistContinueInsert(gistIncompleteInsert *insert)
 				xlinfo = XLOG_GIST_PAGE_SPLIT;
 				rdata = formSplitRdata(index->rd_node, insert->path[i],
 									   false, &(insert->key),
-									 gistMakePageLayout(buffers, numbuffer));
+									   gistMakePageLayout(buffers, numbuffer),
+									   rrlink, &oldnsn);
 
 			}
 			else
@@ -849,7 +906,8 @@ gist_safe_restartpoint(void)
 
 XLogRecData *
 formSplitRdata(RelFileNode node, BlockNumber blkno, bool page_is_leaf,
-			   ItemPointer key, SplitedPageLayout *dist)
+			   ItemPointer key, SplitedPageLayout *dist,
+			   BlockNumber origrlink, XLogRecPtr *orignsn)
 {
 	XLogRecData *rdata;
 	gistxlogPageSplit *xlrec = (gistxlogPageSplit *) palloc(sizeof(gistxlogPageSplit));
@@ -864,7 +922,7 @@ formSplitRdata(RelFileNode node, BlockNumber blkno, bool page_is_leaf,
 		ptr = ptr->next;
 	}
 
-	rdata = (XLogRecData *) palloc(sizeof(XLogRecData) * (npage * 2 + 2));
+	rdata = (XLogRecData *) palloc(sizeof(XLogRecData) * (npage * 2 + 4));
 
 	xlrec->node = node;
 	xlrec->origblkno = blkno;
@@ -893,11 +951,24 @@ formSplitRdata(RelFileNode node, BlockNumber blkno, bool page_is_leaf,
 		rdata[cur].data = (char *) (ptr->list);
 		rdata[cur].len = ptr->lenlist;
 		rdata[cur - 1].next = &(rdata[cur]);
-		rdata[cur].next = NULL;
 		cur++;
 		ptr = ptr->next;
 	}
 
+	/* Append origin rightlink and NSN */
+	rdata[cur].buffer = InvalidBuffer;
+	rdata[cur].data = (char *) &origrlink;
+	rdata[cur].len = sizeof(BlockNumber);
+	rdata[cur - 1].next = &(rdata[cur]);
+	cur++;
+
+	rdata[cur].buffer = InvalidBuffer;
+	rdata[cur].data = (char *) orignsn;
+	rdata[cur].len = sizeof(XLogRecPtr);
+	rdata[cur - 1].next = &(rdata[cur]);
+
+	rdata[cur].next = NULL;
+
 	return rdata;
 }
 
diff --git a/src/include/access/gist_private.h b/src/include/access/gist_private.h
index 4df5fed..d4c8f04 100644
--- a/src/include/access/gist_private.h
+++ b/src/include/access/gist_private.h
@@ -260,7 +260,8 @@ extern XLogRecData *formUpdateRdata(RelFileNode node, Buffer buffer,
 
 extern XLogRecData *formSplitRdata(RelFileNode node,
 			   BlockNumber blkno, bool page_is_leaf,
-			   ItemPointer key, SplitedPageLayout *dist);
+			   ItemPointer key, SplitedPageLayout *dist,
+			   BlockNumber origrlink, XLogRecPtr *orignsn);
 
 extern XLogRecPtr gistxlogInsertCompletion(RelFileNode node, ItemPointerData *keys, int len);