From b3568a5475526fbd7768bd4d74c6ad681ede920f Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Thu, 2 Nov 2017 18:35:10 +0100
Subject: [PATCH v2] Fix summarization concurrent with relation extension

---
 src/backend/access/brin/brin.c | 78 ++++++++++++++++++++++++++----------------
 1 file changed, 49 insertions(+), 29 deletions(-)

diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
index e6909d7aea..ad512c2c5f 100644
--- a/src/backend/access/brin/brin.c
+++ b/src/backend/access/brin/brin.c
@@ -67,7 +67,7 @@ static BrinBuildState *initialize_brin_buildstate(Relation idxRel,
 						   BrinRevmap *revmap, BlockNumber pagesPerRange);
 static void terminate_brin_buildstate(BrinBuildState *state);
 static void brinsummarize(Relation index, Relation heapRel, BlockNumber pageRange,
-			  double *numSummarized, double *numExisting);
+			  bool include_partial, double *numSummarized, double *numExisting);
 static void form_and_insert_tuple(BrinBuildState *state);
 static void union_tuples(BrinDesc *bdesc, BrinMemTuple *a,
 			 BrinTuple *b);
@@ -791,7 +791,7 @@ brinvacuumcleanup(IndexVacuumInfo *info, IndexBulkDeleteResult *stats)
 
 	brin_vacuum_scan(info->index, info->strategy);
 
-	brinsummarize(info->index, heapRel, BRIN_ALL_BLOCKRANGES,
+	brinsummarize(info->index, heapRel, BRIN_ALL_BLOCKRANGES, false,
 				  &stats->num_index_tuples, &stats->num_index_tuples);
 
 	heap_close(heapRel, AccessShareLock);
@@ -909,7 +909,7 @@ brin_summarize_range(PG_FUNCTION_ARGS)
 						RelationGetRelationName(indexRel))));
 
 	/* OK, do it */
-	brinsummarize(indexRel, heapRel, heapBlk, &numSummarized, NULL);
+	brinsummarize(indexRel, heapRel, heapBlk, true, &numSummarized, NULL);
 
 	relation_close(indexRel, ShareUpdateExclusiveLock);
 	relation_close(heapRel, ShareUpdateExclusiveLock);
@@ -1129,7 +1129,8 @@ terminate_brin_buildstate(BrinBuildState *state)
 }
 
 /*
- * Summarize the given page range of the given index.
+ * On the given BRIN index, summarize the heap page range that corresponds
+ * to the heap block number given.
  *
  * This routine can run in parallel with insertions into the heap.  To avoid
  * missing those values from the summary tuple, we first insert a placeholder
@@ -1139,6 +1140,12 @@ terminate_brin_buildstate(BrinBuildState *state)
  * update of the index value happens in a loop, so that if somebody updates
  * the placeholder tuple after we read it, we detect the case and try again.
  * This ensures that the concurrently inserted tuples are not lost.
+ *
+ * A further corner case is this routine being asked to summarize the partial
+ * range at the end of the table.  heapNumBlocks is the (possibly outdated)
+ * table size; if we notice that the requested range lies beyond that size,
+ * we re-compute the table size after inserting the placeholder tuple, to
+ * avoid missing pages that were appended recently.
  */
 static void
 summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
@@ -1160,6 +1167,20 @@ summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
 						   heapBlk, phtup, phsz);
 
 	/*
+	 * Whenever we're scanning what we believe to be the final range on the
+	 * table (i.e. one that might be partial) we need to recompute our idea of
+	 * what the latest page is after inserting the placeholder tuple.  This
+	 * ensures that we don't miss pages just because they were extended by
+	 * concurrent backends that didn't have the chance to see our placeholder
+	 * tuple.  Fortunately, this should occur infrequently.
+	 */
+	if (heapBlk + state->bs_pagesPerRange > heapNumBlks)
+		scanNumBlks = RelationGetNumberOfBlocks(heapRel) -
+			(heapBlk / state->bs_pagesPerRange) * state->bs_pagesPerRange;
+	else
+		scanNumBlks = state->bs_pagesPerRange;
+
+	/*
 	 * Execute the partial heap scan covering the heap blocks in the specified
 	 * page range, summarizing the heap tuples in it.  This scan stops just
 	 * short of brinbuildCallback creating the new index entry.
@@ -1169,8 +1190,6 @@ summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
 	 * by transactions that are still in progress, among other corner cases.
 	 */
 	state->bs_currRangeStart = heapBlk;
-	scanNumBlks = heapBlk + state->bs_pagesPerRange <= heapNumBlks ?
-		state->bs_pagesPerRange : heapNumBlks - heapBlk;
 	IndexBuildHeapRangeScan(heapRel, state->bs_irel, indexInfo, false, true,
 							heapBlk, scanNumBlks,
 							brinbuildCallback, (void *) state);
@@ -1234,6 +1253,8 @@ summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
  * Summarize page ranges that are not already summarized.  If pageRange is
  * BRIN_ALL_BLOCKRANGES then the whole table is scanned; otherwise, only the
  * page range containing the given heap page number is scanned.
+ * If include_partial is true, then the partial range at the end of the table
+ * is summarized, otherwise not.
  *
  * For each new index tuple inserted, *numSummarized (if not NULL) is
  * incremented; for each existing tuple, *numExisting (if not NULL) is
@@ -1241,56 +1262,55 @@ summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
  */
 static void
 brinsummarize(Relation index, Relation heapRel, BlockNumber pageRange,
-			  double *numSummarized, double *numExisting)
+			  bool include_partial, double *numSummarized, double *numExisting)
 {
 	BrinRevmap *revmap;
 	BrinBuildState *state = NULL;
 	IndexInfo  *indexInfo = NULL;
 	BlockNumber heapNumBlocks;
-	BlockNumber heapBlk;
 	BlockNumber pagesPerRange;
 	Buffer		buf;
 	BlockNumber startBlk;
-	BlockNumber endBlk;
-
-	/* determine range of pages to process; nothing to do for an empty table */
-	heapNumBlocks = RelationGetNumberOfBlocks(heapRel);
-	if (heapNumBlocks == 0)
-		return;
 
 	revmap = brinRevmapInitialize(index, &pagesPerRange, NULL);
 
+	/* determine range of pages to process */
+	heapNumBlocks = RelationGetNumberOfBlocks(heapRel);
 	if (pageRange == BRIN_ALL_BLOCKRANGES)
-	{
 		startBlk = 0;
-		endBlk = heapNumBlocks;
-	}
 	else
-	{
 		startBlk = (pageRange / pagesPerRange) * pagesPerRange;
+	if (startBlk >= heapNumBlocks)
+	{
 		/* Nothing to do if start point is beyond end of table */
-		if (startBlk > heapNumBlocks)
-		{
-			brinRevmapTerminate(revmap);
-			return;
-		}
-		endBlk = startBlk + pagesPerRange;
-		if (endBlk > heapNumBlocks)
-			endBlk = heapNumBlocks;
+		brinRevmapTerminate(revmap);
+		return;
 	}
 
 	/*
 	 * Scan the revmap to find unsummarized items.
 	 */
 	buf = InvalidBuffer;
-	for (heapBlk = startBlk; heapBlk < endBlk; heapBlk += pagesPerRange)
+	for (; startBlk < heapNumBlocks; startBlk += pagesPerRange)
 	{
 		BrinTuple  *tup;
 		OffsetNumber off;
 
+		/*
+		 * Unless requested to summarize even a partial range, go away now if
+		 * we think the next range is partial.
+		 *
+		 * Maybe the table already grew to cover this range completely, but we
+		 * don't want to spend a whole RelationGetNumberOfBlocks to find out,
+		 * so just leave it for next time.
+		 */
+		if (!include_partial &&
+			(startBlk + pagesPerRange > heapNumBlocks))
+			break;
+
 		CHECK_FOR_INTERRUPTS();
 
-		tup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off, NULL,
+		tup = brinGetTupleForHeapBlock(revmap, startBlk, &buf, &off, NULL,
 									   BUFFER_LOCK_SHARE, NULL);
 		if (tup == NULL)
 		{
@@ -1303,7 +1323,7 @@ brinsummarize(Relation index, Relation heapRel, BlockNumber pageRange,
 												   pagesPerRange);
 				indexInfo = BuildIndexInfo(index);
 			}
-			summarize_range(indexInfo, state, heapRel, heapBlk, heapNumBlocks);
+			summarize_range(indexInfo, state, heapRel, startBlk, heapNumBlocks);
 
 			/* and re-initialize state for the next range */
 			brin_memtuple_initialize(state->bs_dtuple, state->bs_bdesc);
-- 
2.11.0

