[CF 5542] v12 - Fix buffer pinning logic in [SP-]Gist

This branch was automatically generated by a robot using patches from an
email thread registered at:

https://commitfest.postgresql.org/patch/5542

The branch will be overwritten each time a new patch version is posted to
the thread, and also periodically to check for bitrot caused by changes
on the master branch.

Patch(es): https://www.postgresql.org/message-id/CAEze2WgnHJF66BtviDFNCPy26gLrb+tfns-Bi62DBfReH0-UCg@mail.gmail.com
Author(s): Peter Geoghegan, Michail Nikolaev, Matthias van de Meent, Mihail Nikalayeu

Author: Commitfest Bot

src/backend/access/gist/gistget.c

@@ -17,6 +17,7 @@
 #include "access/genam.h"
 #include "access/gist_private.h"
 #include "access/relscan.h"
+#include "access/tableam.h"
 #include "lib/pairingheap.h"
 #include "miscadmin.h"
 #include "pgstat.h"
@@ -394,10 +395,14 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem,
 		return;
 	}
 
-	so->nPageData = so->curPageData = 0;
+	if (scan->numberOfOrderBys)
+		so->os.nsortData = 0;
+	else
+		so->nos.nPageData = so->nos.curPageData = 0;
+
 	scan->xs_hitup = NULL;		/* might point into pageDataCxt */
-	if (so->pageDataCxt)
-		MemoryContextReset(so->pageDataCxt);
+	if (so->nos.pageDataCxt)
+		MemoryContextReset(so->nos.pageDataCxt);
 
 	/*
 	 * We save the LSN of the page as we read it, so that we know whether it
@@ -457,22 +462,22 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem,
 			/*
 			 * Non-ordered scan, so report tuples in so->pageData[]
 			 */
-			so->pageData[so->nPageData].heapPtr = it->t_tid;
-			so->pageData[so->nPageData].recheck = recheck;
-			so->pageData[so->nPageData].offnum = i;
+			so->nos.pageData[so->nos.nPageData].heapPtr = it->t_tid;
+			so->nos.pageData[so->nos.nPageData].recheck = recheck;
+			so->nos.pageData[so->nos.nPageData].offnum = i;
 
 			/*
 			 * In an index-only scan, also fetch the data from the tuple.  The
 			 * reconstructed tuples are stored in pageDataCxt.
 			 */
 			if (scan->xs_want_itup)
 			{
-				oldcxt = MemoryContextSwitchTo(so->pageDataCxt);
-				so->pageData[so->nPageData].recontup =
+				oldcxt = MemoryContextSwitchTo(so->nos.pageDataCxt);
+				so->nos.pageData[so->nos.nPageData].recontup =
 					gistFetchTuple(giststate, r, it);
 				MemoryContextSwitchTo(oldcxt);
 			}
-			so->nPageData++;
+			so->nos.nPageData++;
 		}
 		else
 		{
@@ -501,7 +506,11 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem,
 				 * In an index-only scan, also fetch the data from the tuple.
 				 */
 				if (scan->xs_want_itup)
+				{
 					item->data.heap.recontup = gistFetchTuple(giststate, r, it);
+					so->os.sortData[so->os.nsortData] = &item->data.heap;
+					so->os.nsortData += 1;
+				}
 			}
 			else
 			{
@@ -526,7 +535,101 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem,
 		}
 	}
 
-	UnlockReleaseBuffer(buffer);
+	/* Allow writes to the buffer, but don't yet allow VACUUM */
+	LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
+
+	/*
+	 * If we're in an index-only scan, we need to do visibility checks before
+	 * we release the pin, so that VACUUM can't clean up dead tuples from this
+	 * index page and mark the page ALL_VISIBLE before the tuple was returned.
+	 *
+	 * See also docs section "Index Locking Considerations".
+	 */
+	if (scan->xs_want_itup)
+	{
+		TM_IndexVisibilityCheckOp	op;
+		op.vmbuf = &so->vmbuf;
+
+		if (scan->numberOfOrderBys > 0)
+		{
+			op.checkntids = so->os.nsortData;
+
+			if (op.checkntids > 0)
+			{
+				op.checktids = palloc(op.checkntids * sizeof(TM_VisCheck));
+
+				for (int off = 0; off < op.checkntids; off++)
+				{
+					Assert(ItemPointerIsValid(&so->os.sortData[off]->heapPtr));
+
+					PopulateTMVischeck(&op.checktids[off],
+									   &so->os.sortData[off]->heapPtr,
+									   off);
+				}
+			}
+		}
+		else
+		{
+			op.checkntids = so->nos.nPageData;
+
+			if (op.checkntids > 0)
+			{
+				op.checktids = palloc_array(TM_VisCheck, op.checkntids);
+
+				for (int off = 0; off < op.checkntids; off++)
+				{
+					Assert(ItemPointerIsValid(&so->nos.pageData[off].heapPtr));
+
+					PopulateTMVischeck(&op.checktids[off],
+									   &so->nos.pageData[off].heapPtr,
+									   off);
+				}
+			}
+		}
+
+		if (op.checkntids > 0)
+		{
+			table_index_vischeck_tuples(scan->heapRelation, &op);
+
+			if (scan->numberOfOrderBys > 0)
+			{
+				for (int off = 0; off < op.checkntids; off++)
+				{
+					TM_VisCheck *check = &op.checktids[off];
+					GISTSearchHeapItem *item = so->os.sortData[check->idxoffnum];
+
+					/* sanity checks */
+					Assert(check->idxoffnum < op.checkntids);
+					Assert(check->tidblkno == ItemPointerGetBlockNumberNoCheck(&item->heapPtr));
+					Assert(check->tidoffset == ItemPointerGetOffsetNumberNoCheck(&item->heapPtr));
+
+					item->visrecheck = check->vischeckresult;
+				}
+				/* reset state */
+				so->os.nsortData = 0;
+			}
+			else
+			{
+				for (int off = 0; off < op.checkntids; off++)
+				{
+					TM_VisCheck *check = &op.checktids[off];
+					GISTSearchHeapItem *item = &so->nos.pageData[check->idxoffnum];
+
+					Assert(check->idxoffnum < op.checkntids);
+					Assert(check->tidblkno == ItemPointerGetBlockNumberNoCheck(&item->heapPtr));
+					Assert(check->tidoffset == ItemPointerGetOffsetNumberNoCheck(&item->heapPtr));
+
+					item->visrecheck = check->vischeckresult;
+				}
+			}
+
+			/* clean up the used resources */
+			pfree(op.checktids);
+		}
+	}
+
+	/* Allow VACUUM to process the buffer again */
+	ReleaseBuffer(buffer);
 }
 
 /*
@@ -588,7 +691,10 @@ getNextNearest(IndexScanDesc scan)
 
 			/* in an index-only scan, also return the reconstructed tuple. */
 			if (scan->xs_want_itup)
+			{
 				scan->xs_hitup = item->data.heap.recontup;
+				scan->xs_visrecheck = item->data.heap.visrecheck;
+			}
 			res = true;
 		}
 		else
@@ -629,10 +735,10 @@ gistgettuple(IndexScanDesc scan, ScanDirection dir)
 			scan->instrument->nsearches++;
 
 		so->firstCall = false;
-		so->curPageData = so->nPageData = 0;
+		so->nos.curPageData = so->nos.nPageData = 0;
 		scan->xs_hitup = NULL;
-		if (so->pageDataCxt)
-			MemoryContextReset(so->pageDataCxt);
+		if (so->nos.pageDataCxt)
+			MemoryContextReset(so->nos.pageDataCxt);
 
 		fakeItem.blkno = GIST_ROOT_BLKNO;
 		memset(&fakeItem.data.parentlsn, 0, sizeof(GistNSN));
@@ -649,9 +755,9 @@ gistgettuple(IndexScanDesc scan, ScanDirection dir)
 		/* Fetch tuples index-page-at-a-time */
 		for (;;)
 		{
-			if (so->curPageData < so->nPageData)
+			if (so->nos.curPageData < so->nos.nPageData)
 			{
-				if (scan->kill_prior_tuple && so->curPageData > 0)
+				if (scan->kill_prior_tuple && so->nos.curPageData > 0)
 				{
 
 					if (so->killedItems == NULL)
@@ -667,17 +773,20 @@ gistgettuple(IndexScanDesc scan, ScanDirection dir)
 					}
 					if (so->numKilled < MaxIndexTuplesPerPage)
 						so->killedItems[so->numKilled++] =
-							so->pageData[so->curPageData - 1].offnum;
+							so->nos.pageData[so->nos.curPageData - 1].offnum;
 				}
 				/* continuing to return tuples from a leaf page */
-				scan->xs_heaptid = so->pageData[so->curPageData].heapPtr;
-				scan->xs_recheck = so->pageData[so->curPageData].recheck;
+				scan->xs_heaptid = so->nos.pageData[so->nos.curPageData].heapPtr;
+				scan->xs_recheck = so->nos.pageData[so->nos.curPageData].recheck;
 
 				/* in an index-only scan, also return the reconstructed tuple */
 				if (scan->xs_want_itup)
-					scan->xs_hitup = so->pageData[so->curPageData].recontup;
+				{
+					scan->xs_hitup = so->nos.pageData[so->nos.curPageData].recontup;
+					scan->xs_visrecheck = so->nos.pageData[so->nos.curPageData].visrecheck;
+				}
 
-				so->curPageData++;
+				so->nos.curPageData++;
 
 				return true;
 			}
@@ -687,8 +796,8 @@ gistgettuple(IndexScanDesc scan, ScanDirection dir)
 			 * necessary
 			 */
 			if (scan->kill_prior_tuple
-				&& so->curPageData > 0
-				&& so->curPageData == so->nPageData)
+				&& so->nos.curPageData > 0
+				&& so->nos.curPageData == so->nos.nPageData)
 			{
 
 				if (so->killedItems == NULL)
@@ -704,7 +813,7 @@ gistgettuple(IndexScanDesc scan, ScanDirection dir)
 				}
 				if (so->numKilled < MaxIndexTuplesPerPage)
 					so->killedItems[so->numKilled++] =
-						so->pageData[so->curPageData - 1].offnum;
+						so->nos.pageData[so->nos.curPageData - 1].offnum;
 			}
 			/* find and process the next index page */
 			do
@@ -733,7 +842,7 @@ gistgettuple(IndexScanDesc scan, ScanDirection dir)
 				gistScanPage(scan, item, item->distances, NULL, NULL);
 
 				pfree(item);
-			} while (so->nPageData == 0);
+			} while (so->nos.nPageData == 0);
 		}
 	}
 }
@@ -756,10 +865,10 @@ gistgetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
 		scan->instrument->nsearches++;
 
 	/* Begin the scan by processing the root page */
-	so->curPageData = so->nPageData = 0;
+	so->nos.curPageData = so->nos.nPageData = 0;
 	scan->xs_hitup = NULL;
-	if (so->pageDataCxt)
-		MemoryContextReset(so->pageDataCxt);
+	if (so->nos.pageDataCxt)
+		MemoryContextReset(so->nos.pageDataCxt);
 
 	fakeItem.blkno = GIST_ROOT_BLKNO;
 	memset(&fakeItem.data.parentlsn, 0, sizeof(GistNSN));

src/backend/access/gist/gistscan.c

@@ -204,9 +204,9 @@ gistrescan(IndexScanDesc scan, ScanKey key, int nkeys,
 		scan->xs_hitupdesc = so->giststate->fetchTupdesc;
 
 		/* Also create a memory context that will hold the returned tuples */
-		so->pageDataCxt = AllocSetContextCreate(so->giststate->scanCxt,
-												"GiST page data context",
-												ALLOCSET_DEFAULT_SIZES);
+		so->nos.pageDataCxt = AllocSetContextCreate(so->giststate->scanCxt,
+													"GiST page data context",
+													ALLOCSET_DEFAULT_SIZES);
 	}
 
 	/* create new, empty pairing heap for search queue */
@@ -347,6 +347,11 @@ void
 gistendscan(IndexScanDesc scan)
 {
 	GISTScanOpaque so = (GISTScanOpaque) scan->opaque;
+	if (BufferIsValid(so->vmbuf))
+	{
+		ReleaseBuffer(so->vmbuf);
+		so->vmbuf = InvalidBuffer;
+	}
 
 	/*
 	 * freeGISTstate is enough to clean up everything made by gistbeginscan,

src/backend/access/gist/gistvacuum.c

@@ -326,10 +326,10 @@ gistvacuumpage(GistVacState *vstate, Buffer buffer)
 	recurse_to = InvalidBlockNumber;
 
 	/*
-	 * We are not going to stay here for a long time, aggressively grab an
-	 * exclusive lock.
+	 * We are not going to stay here for a long time, aggressively grab a
+	 * cleanup lock.
 	 */
-	LockBuffer(buffer, GIST_EXCLUSIVE);
+	LockBufferForCleanup(buffer);
 	page = (Page) BufferGetPage(buffer);
 
 	if (gistPageRecyclable(page))