*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 1925,1930 **** include_dir 'conf.d'
--- 1925,1966 ----
          </para>
         </listitem>
        </varlistentry>
+ 
+       <varlistentry id="guc-old-snapshot-threshold" xreflabel="old_snapshot_threshold">
+        <term><varname>old_snapshot_threshold</varname> (<type>integer</type>)
+        <indexterm>
+         <primary><varname>old_snapshot_threshold</> configuration parameter</primary>
+        </indexterm>
+        </term>
+        <listitem>
+         <para>
+          Sets the minimum time that a snapshot can be used without risk of a
+          <literal>snapshot too old</> error occurring when using the snapshot.
+          This parameter can only be set at server start.
+         </para>
+ 
+         <para>
+          Beyond the threshold, old data may be vacuumed away.  This can help
+          prevent bloat in the face of snapshots which remain in use for a
+          long time.  To prevent incorrect results due to cleanup of data which
+          would otherwise be visible to the snapshot, an error is generated
+          when the snapshot is older than this threshold and the snapshot is
+          used to read a page which has been modified since the snapshot was
+          built.
+         </para>
+ 
+         <para>
+          A value of <literal>-1</> disables this feature, and is the default.
+          Useful values for production work probably range from a small number
+          of hours to a few days.  The setting will be coerced to a granularity
+          of minutes, and small numbers (such as <literal>0</> or
+          <literal>1min</>) are only allowed because they may sometimes be
+          useful for testing.  While a setting as high as <literal>60d</> is
+          allowed, please note that in many workloads extreme bloat or
+          transaction ID wraparound may occur in much shorter time frames.
+         </para>
+        </listitem>
+       </varlistentry>
       </variablelist>
      </sect2>
     </sect1>
***************
*** 2850,2855 **** include_dir 'conf.d'
--- 2886,2895 ----
          You should also consider setting <varname>hot_standby_feedback</>
          on standby server(s) as an alternative to using this parameter.
         </para>
+        <para>
+         This does not prevent cleanup of dead rows which have reached the age
+         specified by <varname>old_snapshot_threshold</>.
+        </para>
        </listitem>
       </varlistentry>
  
***************
*** 2997,3002 **** include_dir 'conf.d'
--- 3037,3052 ----
          until it eventually reaches the primary.  Standbys make no other use
          of feedback they receive other than to pass upstream.
         </para>
+        <para>
+         This setting does not override the behavior of
+         <varname>old_snapshot_threshold</> on the primary; a snapshot on the
+         standby which exceeds the primary's age threshold can become invalid,
+         resulting in cancellation of transactions on the standby.  This is
+         because <varname>old_snapshot_threshold</> is intended to provide an
+         absolute limit on the time which dead rows can contribute to bloat,
+         which would otherwise be violated because of the configuration of a
+         standby.
+        </para>
        </listitem>
       </varlistentry>
  
*** a/src/backend/access/brin/brin.c
--- b/src/backend/access/brin/brin.c
***************
*** 96,102 **** brininsert(PG_FUNCTION_ARGS)
  	MemoryContext tupcxt = NULL;
  	MemoryContext oldcxt = NULL;
  
! 	revmap = brinRevmapInitialize(idxRel, &pagesPerRange);
  
  	for (;;)
  	{
--- 96,102 ----
  	MemoryContext tupcxt = NULL;
  	MemoryContext oldcxt = NULL;
  
! 	revmap = brinRevmapInitialize(idxRel, &pagesPerRange, NULL);
  
  	for (;;)
  	{
***************
*** 113,119 **** brininsert(PG_FUNCTION_ARGS)
  		/* normalize the block number to be the first block in the range */
  		heapBlk = (heapBlk / pagesPerRange) * pagesPerRange;
  		brtup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off, NULL,
! 										 BUFFER_LOCK_SHARE);
  
  		/* if range is unsummarized, there's nothing to do */
  		if (!brtup)
--- 113,119 ----
  		/* normalize the block number to be the first block in the range */
  		heapBlk = (heapBlk / pagesPerRange) * pagesPerRange;
  		brtup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off, NULL,
! 										 BUFFER_LOCK_SHARE, NULL);
  
  		/* if range is unsummarized, there's nothing to do */
  		if (!brtup)
***************
*** 248,254 **** brinbeginscan(PG_FUNCTION_ARGS)
  	scan = RelationGetIndexScan(r, nkeys, norderbys);
  
  	opaque = (BrinOpaque *) palloc(sizeof(BrinOpaque));
! 	opaque->bo_rmAccess = brinRevmapInitialize(r, &opaque->bo_pagesPerRange);
  	opaque->bo_bdesc = brin_build_desc(r);
  	scan->opaque = opaque;
  
--- 248,255 ----
  	scan = RelationGetIndexScan(r, nkeys, norderbys);
  
  	opaque = (BrinOpaque *) palloc(sizeof(BrinOpaque));
! 	opaque->bo_rmAccess = brinRevmapInitialize(r, &opaque->bo_pagesPerRange,
! 											   scan->xs_snapshot);
  	opaque->bo_bdesc = brin_build_desc(r);
  	scan->opaque = opaque;
  
***************
*** 333,339 **** bringetbitmap(PG_FUNCTION_ARGS)
  		MemoryContextResetAndDeleteChildren(perRangeCxt);
  
  		tup = brinGetTupleForHeapBlock(opaque->bo_rmAccess, heapBlk, &buf,
! 									   &off, &size, BUFFER_LOCK_SHARE);
  		if (tup)
  		{
  			tup = brin_copy_tuple(tup, size);
--- 334,341 ----
  		MemoryContextResetAndDeleteChildren(perRangeCxt);
  
  		tup = brinGetTupleForHeapBlock(opaque->bo_rmAccess, heapBlk, &buf,
! 									   &off, &size, BUFFER_LOCK_SHARE,
! 									   scan->xs_snapshot);
  		if (tup)
  		{
  			tup = brin_copy_tuple(tup, size);
***************
*** 637,643 **** brinbuild(PG_FUNCTION_ARGS)
  	/*
  	 * Initialize our state, including the deformed tuple state.
  	 */
! 	revmap = brinRevmapInitialize(index, &pagesPerRange);
  	state = initialize_brin_buildstate(index, revmap, pagesPerRange);
  
  	/*
--- 639,645 ----
  	/*
  	 * Initialize our state, including the deformed tuple state.
  	 */
! 	revmap = brinRevmapInitialize(index, &pagesPerRange, NULL);
  	state = initialize_brin_buildstate(index, revmap, pagesPerRange);
  
  	/*
***************
*** 1004,1010 **** summarize_range(IndexInfo *indexInfo, BrinBuildState *state, Relation heapRel,
  		 * the same.)
  		 */
  		phtup = brinGetTupleForHeapBlock(state->bs_rmAccess, heapBlk, &phbuf,
! 										 &offset, &phsz, BUFFER_LOCK_SHARE);
  		/* the placeholder tuple must exist */
  		if (phtup == NULL)
  			elog(ERROR, "missing placeholder tuple");
--- 1006,1013 ----
  		 * the same.)
  		 */
  		phtup = brinGetTupleForHeapBlock(state->bs_rmAccess, heapBlk, &phbuf,
! 										 &offset, &phsz, BUFFER_LOCK_SHARE,
! 										 NULL);
  		/* the placeholder tuple must exist */
  		if (phtup == NULL)
  			elog(ERROR, "missing placeholder tuple");
***************
*** 1039,1045 **** brinsummarize(Relation index, Relation heapRel, double *numSummarized,
  	BlockNumber pagesPerRange;
  	Buffer		buf;
  
! 	revmap = brinRevmapInitialize(index, &pagesPerRange);
  
  	/*
  	 * Scan the revmap to find unsummarized items.
--- 1042,1048 ----
  	BlockNumber pagesPerRange;
  	Buffer		buf;
  
! 	revmap = brinRevmapInitialize(index, &pagesPerRange, NULL);
  
  	/*
  	 * Scan the revmap to find unsummarized items.
***************
*** 1054,1060 **** brinsummarize(Relation index, Relation heapRel, double *numSummarized,
  		CHECK_FOR_INTERRUPTS();
  
  		tup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off, NULL,
! 									   BUFFER_LOCK_SHARE);
  		if (tup == NULL)
  		{
  			/* no revmap entry for this heap range. Summarize it. */
--- 1057,1063 ----
  		CHECK_FOR_INTERRUPTS();
  
  		tup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off, NULL,
! 									   BUFFER_LOCK_SHARE, NULL);
  		if (tup == NULL)
  		{
  			/* no revmap entry for this heap range. Summarize it. */
*** a/src/backend/access/brin/brin_revmap.c
--- b/src/backend/access/brin/brin_revmap.c
***************
*** 31,36 ****
--- 31,37 ----
  #include "storage/bufmgr.h"
  #include "storage/lmgr.h"
  #include "utils/rel.h"
+ #include "utils/snapmgr.h"
  
  
  /*
***************
*** 68,82 **** static void revmap_physical_extend(BrinRevmap *revmap);
   * brinRevmapTerminate when caller is done with it.
   */
  BrinRevmap *
! brinRevmapInitialize(Relation idxrel, BlockNumber *pagesPerRange)
  {
  	BrinRevmap *revmap;
  	Buffer		meta;
  	BrinMetaPageData *metadata;
  
  	meta = ReadBuffer(idxrel, BRIN_METAPAGE_BLKNO);
  	LockBuffer(meta, BUFFER_LOCK_SHARE);
! 	metadata = (BrinMetaPageData *) PageGetContents(BufferGetPage(meta));
  
  	revmap = palloc(sizeof(BrinRevmap));
  	revmap->rm_irel = idxrel;
--- 69,87 ----
   * brinRevmapTerminate when caller is done with it.
   */
  BrinRevmap *
! brinRevmapInitialize(Relation idxrel, BlockNumber *pagesPerRange,
! 					 Snapshot snapshot)
  {
  	BrinRevmap *revmap;
  	Buffer		meta;
  	BrinMetaPageData *metadata;
+ 	Page		page;
  
  	meta = ReadBuffer(idxrel, BRIN_METAPAGE_BLKNO);
  	LockBuffer(meta, BUFFER_LOCK_SHARE);
! 	page = BufferGetPage(meta);
! 	TestForOldSnapshot(snapshot, idxrel, page);
! 	metadata = (BrinMetaPageData *) PageGetContents(page);
  
  	revmap = palloc(sizeof(BrinRevmap));
  	revmap->rm_irel = idxrel;
***************
*** 185,191 **** brinSetHeapBlockItemptr(Buffer buf, BlockNumber pagesPerRange,
   */
  BrinTuple *
  brinGetTupleForHeapBlock(BrinRevmap *revmap, BlockNumber heapBlk,
! 						 Buffer *buf, OffsetNumber *off, Size *size, int mode)
  {
  	Relation	idxRel = revmap->rm_irel;
  	BlockNumber mapBlk;
--- 190,197 ----
   */
  BrinTuple *
  brinGetTupleForHeapBlock(BrinRevmap *revmap, BlockNumber heapBlk,
! 						 Buffer *buf, OffsetNumber *off, Size *size, int mode,
! 						 Snapshot snapshot)
  {
  	Relation	idxRel = revmap->rm_irel;
  	BlockNumber mapBlk;
***************
*** 262,267 **** brinGetTupleForHeapBlock(BrinRevmap *revmap, BlockNumber heapBlk,
--- 268,274 ----
  		}
  		LockBuffer(*buf, mode);
  		page = BufferGetPage(*buf);
+ 		TestForOldSnapshot(snapshot, idxRel, page);
  
  		/* If we land on a revmap page, start over */
  		if (BRIN_IS_REGULAR_PAGE(page))
*** a/src/backend/access/gin/ginbtree.c
--- b/src/backend/access/gin/ginbtree.c
***************
*** 18,23 ****
--- 18,24 ----
  #include "access/xloginsert.h"
  #include "miscadmin.h"
  #include "utils/rel.h"
+ #include "utils/snapmgr.h"
  
  static void ginFindParents(GinBtree btree, GinBtreeStack *stack);
  static bool ginPlaceToPage(GinBtree btree, GinBtreeStack *stack,
***************
*** 71,77 **** ginTraverseLock(Buffer buffer, bool searchMode)
   * is share-locked, and stack->parent is NULL.
   */
  GinBtreeStack *
! ginFindLeafPage(GinBtree btree, bool searchMode)
  {
  	GinBtreeStack *stack;
  
--- 72,78 ----
   * is share-locked, and stack->parent is NULL.
   */
  GinBtreeStack *
! ginFindLeafPage(GinBtree btree, bool searchMode, Snapshot snapshot)
  {
  	GinBtreeStack *stack;
  
***************
*** 90,95 **** ginFindLeafPage(GinBtree btree, bool searchMode)
--- 91,97 ----
  		stack->off = InvalidOffsetNumber;
  
  		page = BufferGetPage(stack->buffer);
+ 		TestForOldSnapshot(snapshot, btree->index, page);
  
  		access = ginTraverseLock(stack->buffer, searchMode);
  
***************
*** 116,121 **** ginFindLeafPage(GinBtree btree, bool searchMode)
--- 118,124 ----
  			stack->buffer = ginStepRight(stack->buffer, btree->index, access);
  			stack->blkno = rightlink;
  			page = BufferGetPage(stack->buffer);
+ 			TestForOldSnapshot(snapshot, btree->index, page);
  
  			if (!searchMode && GinPageIsIncompleteSplit(page))
  				ginFinishSplit(btree, stack, false, NULL);
*** a/src/backend/access/gin/gindatapage.c
--- b/src/backend/access/gin/gindatapage.c
***************
*** 1820,1826 **** ginInsertItemPointers(Relation index, BlockNumber rootBlkno,
  	{
  		/* search for the leaf page where the first item should go to */
  		btree.itemptr = insertdata.items[insertdata.curitem];
! 		stack = ginFindLeafPage(&btree, false);
  
  		ginInsertValue(&btree, stack, &insertdata, buildStats);
  	}
--- 1820,1826 ----
  	{
  		/* search for the leaf page where the first item should go to */
  		btree.itemptr = insertdata.items[insertdata.curitem];
! 		stack = ginFindLeafPage(&btree, false, NULL);
  
  		ginInsertValue(&btree, stack, &insertdata, buildStats);
  	}
***************
*** 1830,1836 **** ginInsertItemPointers(Relation index, BlockNumber rootBlkno,
   * Starts a new scan on a posting tree.
   */
  GinBtreeStack *
! ginScanBeginPostingTree(GinBtree btree, Relation index, BlockNumber rootBlkno)
  {
  	GinBtreeStack *stack;
  
--- 1830,1837 ----
   * Starts a new scan on a posting tree.
   */
  GinBtreeStack *
! ginScanBeginPostingTree(GinBtree btree, Relation index, BlockNumber rootBlkno,
! 						Snapshot snapshot)
  {
  	GinBtreeStack *stack;
  
***************
*** 1838,1844 **** ginScanBeginPostingTree(GinBtree btree, Relation index, BlockNumber rootBlkno)
  
  	btree->fullScan = TRUE;
  
! 	stack = ginFindLeafPage(btree, TRUE);
  
  	return stack;
  }
--- 1839,1845 ----
  
  	btree->fullScan = TRUE;
  
! 	stack = ginFindLeafPage(btree, TRUE, snapshot);
  
  	return stack;
  }
*** a/src/backend/access/gin/ginget.c
--- b/src/backend/access/gin/ginget.c
***************
*** 19,24 ****
--- 19,25 ----
  #include "miscadmin.h"
  #include "utils/datum.h"
  #include "utils/memutils.h"
+ #include "utils/snapmgr.h"
  
  /* GUC parameter */
  int			GinFuzzySearchLimit = 0;
***************
*** 63,69 **** moveRightIfItNeeded(GinBtreeData *btree, GinBtreeStack *stack)
   */
  static void
  scanPostingTree(Relation index, GinScanEntry scanEntry,
! 				BlockNumber rootPostingTree)
  {
  	GinBtreeData btree;
  	GinBtreeStack *stack;
--- 64,70 ----
   */
  static void
  scanPostingTree(Relation index, GinScanEntry scanEntry,
! 				BlockNumber rootPostingTree, Snapshot snapshot)
  {
  	GinBtreeData btree;
  	GinBtreeStack *stack;
***************
*** 71,77 **** scanPostingTree(Relation index, GinScanEntry scanEntry,
  	Page		page;
  
  	/* Descend to the leftmost leaf page */
! 	stack = ginScanBeginPostingTree(&btree, index, rootPostingTree);
  	buffer = stack->buffer;
  	IncrBufferRefCount(buffer); /* prevent unpin in freeGinBtreeStack */
  
--- 72,78 ----
  	Page		page;
  
  	/* Descend to the leftmost leaf page */
! 	stack = ginScanBeginPostingTree(&btree, index, rootPostingTree, snapshot);
  	buffer = stack->buffer;
  	IncrBufferRefCount(buffer); /* prevent unpin in freeGinBtreeStack */
  
***************
*** 114,120 **** scanPostingTree(Relation index, GinScanEntry scanEntry,
   */
  static bool
  collectMatchBitmap(GinBtreeData *btree, GinBtreeStack *stack,
! 				   GinScanEntry scanEntry)
  {
  	OffsetNumber attnum;
  	Form_pg_attribute attr;
--- 115,121 ----
   */
  static bool
  collectMatchBitmap(GinBtreeData *btree, GinBtreeStack *stack,
! 				   GinScanEntry scanEntry, Snapshot snapshot)
  {
  	OffsetNumber attnum;
  	Form_pg_attribute attr;
***************
*** 145,150 **** collectMatchBitmap(GinBtreeData *btree, GinBtreeStack *stack,
--- 146,152 ----
  			return true;
  
  		page = BufferGetPage(stack->buffer);
+ 		TestForOldSnapshot(snapshot, btree->index, page);
  		itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, stack->off));
  
  		/*
***************
*** 224,230 **** collectMatchBitmap(GinBtreeData *btree, GinBtreeStack *stack,
  			LockBuffer(stack->buffer, GIN_UNLOCK);
  
  			/* Collect all the TIDs in this entry's posting tree */
! 			scanPostingTree(btree->index, scanEntry, rootPostingTree);
  
  			/*
  			 * We lock again the entry page and while it was unlocked insert
--- 226,232 ----
  			LockBuffer(stack->buffer, GIN_UNLOCK);
  
  			/* Collect all the TIDs in this entry's posting tree */
! 			scanPostingTree(btree->index, scanEntry, rootPostingTree, snapshot);
  
  			/*
  			 * We lock again the entry page and while it was unlocked insert
***************
*** 291,297 **** collectMatchBitmap(GinBtreeData *btree, GinBtreeStack *stack,
   * Start* functions setup beginning state of searches: finds correct buffer and pins it.
   */
  static void
! startScanEntry(GinState *ginstate, GinScanEntry entry)
  {
  	GinBtreeData btreeEntry;
  	GinBtreeStack *stackEntry;
--- 293,299 ----
   * Start* functions setup beginning state of searches: finds correct buffer and pins it.
   */
  static void
! startScanEntry(GinState *ginstate, GinScanEntry entry, Snapshot snapshot)
  {
  	GinBtreeData btreeEntry;
  	GinBtreeStack *stackEntry;
***************
*** 316,322 **** restartScanEntry:
  	ginPrepareEntryScan(&btreeEntry, entry->attnum,
  						entry->queryKey, entry->queryCategory,
  						ginstate);
! 	stackEntry = ginFindLeafPage(&btreeEntry, true);
  	page = BufferGetPage(stackEntry->buffer);
  	needUnlock = TRUE;
  
--- 318,324 ----
  	ginPrepareEntryScan(&btreeEntry, entry->attnum,
  						entry->queryKey, entry->queryCategory,
  						ginstate);
! 	stackEntry = ginFindLeafPage(&btreeEntry, true, snapshot);
  	page = BufferGetPage(stackEntry->buffer);
  	needUnlock = TRUE;
  
***************
*** 333,339 **** restartScanEntry:
  		 * for the entry type.
  		 */
  		btreeEntry.findItem(&btreeEntry, stackEntry);
! 		if (collectMatchBitmap(&btreeEntry, stackEntry, entry) == false)
  		{
  			/*
  			 * GIN tree was seriously restructured, so we will cleanup all
--- 335,341 ----
  		 * for the entry type.
  		 */
  		btreeEntry.findItem(&btreeEntry, stackEntry);
! 		if (!collectMatchBitmap(&btreeEntry, stackEntry, entry, snapshot))
  		{
  			/*
  			 * GIN tree was seriously restructured, so we will cleanup all
***************
*** 381,387 **** restartScanEntry:
  			needUnlock = FALSE;
  
  			stack = ginScanBeginPostingTree(&entry->btree, ginstate->index,
! 											rootPostingTree);
  			entry->buffer = stack->buffer;
  
  			/*
--- 383,389 ----
  			needUnlock = FALSE;
  
  			stack = ginScanBeginPostingTree(&entry->btree, ginstate->index,
! 											rootPostingTree, snapshot);
  			entry->buffer = stack->buffer;
  
  			/*
***************
*** 533,539 **** startScan(IndexScanDesc scan)
  	uint32		i;
  
  	for (i = 0; i < so->totalentries; i++)
! 		startScanEntry(ginstate, so->entries[i]);
  
  	if (GinFuzzySearchLimit > 0)
  	{
--- 535,541 ----
  	uint32		i;
  
  	for (i = 0; i < so->totalentries; i++)
! 		startScanEntry(ginstate, so->entries[i], scan->xs_snapshot);
  
  	if (GinFuzzySearchLimit > 0)
  	{
***************
*** 578,584 **** startScan(IndexScanDesc scan)
   * keep it pinned to prevent interference with vacuum.
   */
  static void
! entryLoadMoreItems(GinState *ginstate, GinScanEntry entry, ItemPointerData advancePast)
  {
  	Page		page;
  	int			i;
--- 580,587 ----
   * keep it pinned to prevent interference with vacuum.
   */
  static void
! entryLoadMoreItems(GinState *ginstate, GinScanEntry entry,
! 				   ItemPointerData advancePast, Snapshot snapshot)
  {
  	Page		page;
  	int			i;
***************
*** 622,628 **** entryLoadMoreItems(GinState *ginstate, GinScanEntry entry, ItemPointerData advan
  			entry->btree.itemptr.ip_posid++;
  		}
  		entry->btree.fullScan = false;
! 		stack = ginFindLeafPage(&entry->btree, true);
  
  		/* we don't need the stack, just the buffer. */
  		entry->buffer = stack->buffer;
--- 625,631 ----
  			entry->btree.itemptr.ip_posid++;
  		}
  		entry->btree.fullScan = false;
! 		stack = ginFindLeafPage(&entry->btree, true, snapshot);
  
  		/* we don't need the stack, just the buffer. */
  		entry->buffer = stack->buffer;
***************
*** 732,738 **** entryLoadMoreItems(GinState *ginstate, GinScanEntry entry, ItemPointerData advan
   */
  static void
  entryGetItem(GinState *ginstate, GinScanEntry entry,
! 			 ItemPointerData advancePast)
  {
  	Assert(!entry->isFinished);
  
--- 735,741 ----
   */
  static void
  entryGetItem(GinState *ginstate, GinScanEntry entry,
! 			 ItemPointerData advancePast, Snapshot snapshot)
  {
  	Assert(!entry->isFinished);
  
***************
*** 855,861 **** entryGetItem(GinState *ginstate, GinScanEntry entry,
  			/* If we've processed the current batch, load more items */
  			while (entry->offset >= entry->nlist)
  			{
! 				entryLoadMoreItems(ginstate, entry, advancePast);
  
  				if (entry->isFinished)
  				{
--- 858,864 ----
  			/* If we've processed the current batch, load more items */
  			while (entry->offset >= entry->nlist)
  			{
! 				entryLoadMoreItems(ginstate, entry, advancePast, snapshot);
  
  				if (entry->isFinished)
  				{
***************
*** 894,900 **** entryGetItem(GinState *ginstate, GinScanEntry entry,
   */
  static void
  keyGetItem(GinState *ginstate, MemoryContext tempCtx, GinScanKey key,
! 		   ItemPointerData advancePast)
  {
  	ItemPointerData minItem;
  	ItemPointerData curPageLossy;
--- 897,903 ----
   */
  static void
  keyGetItem(GinState *ginstate, MemoryContext tempCtx, GinScanKey key,
! 		   ItemPointerData advancePast, Snapshot snapshot)
  {
  	ItemPointerData minItem;
  	ItemPointerData curPageLossy;
***************
*** 941,947 **** keyGetItem(GinState *ginstate, MemoryContext tempCtx, GinScanKey key,
  		 */
  		if (ginCompareItemPointers(&entry->curItem, &advancePast) <= 0)
  		{
! 			entryGetItem(ginstate, entry, advancePast);
  			if (entry->isFinished)
  				continue;
  		}
--- 944,950 ----
  		 */
  		if (ginCompareItemPointers(&entry->curItem, &advancePast) <= 0)
  		{
! 			entryGetItem(ginstate, entry, advancePast, snapshot);
  			if (entry->isFinished)
  				continue;
  		}
***************
*** 999,1005 **** keyGetItem(GinState *ginstate, MemoryContext tempCtx, GinScanKey key,
  
  		if (ginCompareItemPointers(&entry->curItem, &advancePast) <= 0)
  		{
! 			entryGetItem(ginstate, entry, advancePast);
  			if (entry->isFinished)
  				continue;
  		}
--- 1002,1008 ----
  
  		if (ginCompareItemPointers(&entry->curItem, &advancePast) <= 0)
  		{
! 			entryGetItem(ginstate, entry, advancePast, snapshot);
  			if (entry->isFinished)
  				continue;
  		}
***************
*** 1208,1214 **** scanGetItem(IndexScanDesc scan, ItemPointerData advancePast,
  			GinScanKey	key = so->keys + i;
  
  			/* Fetch the next item for this key that is > advancePast. */
! 			keyGetItem(&so->ginstate, so->tempCtx, key, advancePast);
  
  			if (key->isFinished)
  				return false;
--- 1211,1218 ----
  			GinScanKey	key = so->keys + i;
  
  			/* Fetch the next item for this key that is > advancePast. */
! 			keyGetItem(&so->ginstate, so->tempCtx, key, advancePast,
! 					   scan->xs_snapshot);
  
  			if (key->isFinished)
  				return false;
***************
*** 1330,1335 **** scanGetCandidate(IndexScanDesc scan, pendingPosition *pos)
--- 1334,1340 ----
  	for (;;)
  	{
  		page = BufferGetPage(pos->pendingBuffer);
+ 		TestForOldSnapshot(scan->xs_snapshot, scan->indexRelation, page);
  
  		maxoff = PageGetMaxOffsetNumber(page);
  		if (pos->firstOffset > maxoff)
***************
*** 1510,1515 **** collectMatchesForHeapRow(IndexScanDesc scan, pendingPosition *pos)
--- 1515,1521 ----
  			   sizeof(bool) * (pos->lastOffset - pos->firstOffset));
  
  		page = BufferGetPage(pos->pendingBuffer);
+ 		TestForOldSnapshot(scan->xs_snapshot, scan->indexRelation, page);
  
  		for (i = 0; i < so->nkeys; i++)
  		{
***************
*** 1696,1707 **** scanPendingInsert(IndexScanDesc scan, TIDBitmap *tbm, int64 *ntids)
  	int			i;
  	pendingPosition pos;
  	Buffer		metabuffer = ReadBuffer(scan->indexRelation, GIN_METAPAGE_BLKNO);
  	BlockNumber blkno;
  
  	*ntids = 0;
  
  	LockBuffer(metabuffer, GIN_SHARE);
! 	blkno = GinPageGetMeta(BufferGetPage(metabuffer))->head;
  
  	/*
  	 * fetch head of list before unlocking metapage. head page must be pinned
--- 1702,1716 ----
  	int			i;
  	pendingPosition pos;
  	Buffer		metabuffer = ReadBuffer(scan->indexRelation, GIN_METAPAGE_BLKNO);
+ 	Page		page;
  	BlockNumber blkno;
  
  	*ntids = 0;
  
  	LockBuffer(metabuffer, GIN_SHARE);
! 	page = BufferGetPage(metabuffer);
! 	TestForOldSnapshot(scan->xs_snapshot, scan->indexRelation, page);
! 	blkno = GinPageGetMeta(page)->head;
  
  	/*
  	 * fetch head of list before unlocking metapage. head page must be pinned
*** a/src/backend/access/gin/gininsert.c
--- b/src/backend/access/gin/gininsert.c
***************
*** 192,198 **** ginEntryInsert(GinState *ginstate,
  
  	ginPrepareEntryScan(&btree, attnum, key, category, ginstate);
  
! 	stack = ginFindLeafPage(&btree, false);
  	page = BufferGetPage(stack->buffer);
  
  	if (btree.findItem(&btree, stack))
--- 192,198 ----
  
  	ginPrepareEntryScan(&btree, attnum, key, category, ginstate);
  
! 	stack = ginFindLeafPage(&btree, false, NULL);
  	page = BufferGetPage(stack->buffer);
  
  	if (btree.findItem(&btree, stack))
*** a/src/backend/access/gist/gistget.c
--- b/src/backend/access/gist/gistget.c
***************
*** 23,28 ****
--- 23,29 ----
  #include "utils/builtins.h"
  #include "utils/memutils.h"
  #include "utils/rel.h"
+ #include "utils/snapmgr.h"
  
  
  /*
***************
*** 266,271 **** gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem, double *myDistances,
--- 267,273 ----
  	LockBuffer(buffer, GIST_SHARE);
  	gistcheckpage(scan->indexRelation, buffer);
  	page = BufferGetPage(buffer);
+ 	TestForOldSnapshot(scan->xs_snapshot, r, page);
  	opaque = GistPageGetOpaque(page);
  
  	/*
*** a/src/backend/access/hash/hash.c
--- b/src/backend/access/hash/hash.c
***************
*** 26,31 ****
--- 26,32 ----
  #include "optimizer/plancat.h"
  #include "storage/bufmgr.h"
  #include "utils/rel.h"
+ #include "utils/snapmgr.h"
  
  
  /* Working state for hashbuild and its callback */
***************
*** 252,257 **** hashgettuple(PG_FUNCTION_ARGS)
--- 253,259 ----
  		buf = so->hashso_curbuf;
  		Assert(BufferIsValid(buf));
  		page = BufferGetPage(buf);
+ 		TestForOldSnapshot(scan->xs_snapshot, rel, page);
  		maxoffnum = PageGetMaxOffsetNumber(page);
  		for (offnum = ItemPointerGetOffsetNumber(current);
  			 offnum <= maxoffnum;
*** a/src/backend/access/hash/hashsearch.c
--- b/src/backend/access/hash/hashsearch.c
***************
*** 19,24 ****
--- 19,25 ----
  #include "miscadmin.h"
  #include "pgstat.h"
  #include "utils/rel.h"
+ #include "utils/snapmgr.h"
  
  
  /*
***************
*** 188,194 **** _hash_first(IndexScanDesc scan, ScanDirection dir)
  
  	/* Read the metapage */
  	metabuf = _hash_getbuf(rel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);
! 	metap = HashPageGetMeta(BufferGetPage(metabuf));
  
  	/*
  	 * Loop until we get a lock on the correct target bucket.
--- 189,197 ----
  
  	/* Read the metapage */
  	metabuf = _hash_getbuf(rel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);
! 	page = BufferGetPage(metabuf);
! 	TestForOldSnapshot(scan->xs_snapshot, rel, page);
! 	metap = HashPageGetMeta(page);
  
  	/*
  	 * Loop until we get a lock on the correct target bucket.
***************
*** 241,246 **** _hash_first(IndexScanDesc scan, ScanDirection dir)
--- 244,250 ----
  	/* Fetch the primary bucket page for the bucket */
  	buf = _hash_getbuf(rel, blkno, HASH_READ, LH_BUCKET_PAGE);
  	page = BufferGetPage(buf);
+ 	TestForOldSnapshot(scan->xs_snapshot, rel, page);
  	opaque = (HashPageOpaque) PageGetSpecialPointer(page);
  	Assert(opaque->hasho_bucket == bucket);
  
***************
*** 347,352 **** _hash_step(IndexScanDesc scan, Buffer *bufP, ScanDirection dir)
--- 351,357 ----
  					_hash_readnext(rel, &buf, &page, &opaque);
  					if (BufferIsValid(buf))
  					{
+ 						TestForOldSnapshot(scan->xs_snapshot, rel, page);
  						maxoff = PageGetMaxOffsetNumber(page);
  						offnum = _hash_binsearch(page, so->hashso_sk_hash);
  					}
***************
*** 388,393 **** _hash_step(IndexScanDesc scan, Buffer *bufP, ScanDirection dir)
--- 393,399 ----
  					_hash_readprev(rel, &buf, &page, &opaque);
  					if (BufferIsValid(buf))
  					{
+ 						TestForOldSnapshot(scan->xs_snapshot, rel, page);
  						maxoff = PageGetMaxOffsetNumber(page);
  						offnum = _hash_binsearch_last(page, so->hashso_sk_hash);
  					}
*** a/src/backend/access/heap/heapam.c
--- b/src/backend/access/heap/heapam.c
***************
*** 382,387 **** heapgetpage(HeapScanDesc scan, BlockNumber page)
--- 382,388 ----
  	LockBuffer(buffer, BUFFER_LOCK_SHARE);
  
  	dp = (Page) BufferGetPage(buffer);
+ 	TestForOldSnapshot(snapshot, scan->rs_rd, dp);
  	lines = PageGetMaxOffsetNumber(dp);
  	ntup = 0;
  
***************
*** 512,517 **** heapgettup(HeapScanDesc scan,
--- 513,519 ----
  		LockBuffer(scan->rs_cbuf, BUFFER_LOCK_SHARE);
  
  		dp = (Page) BufferGetPage(scan->rs_cbuf);
+ 		TestForOldSnapshot(snapshot, scan->rs_rd, dp);
  		lines = PageGetMaxOffsetNumber(dp);
  		/* page and lineoff now reference the physically next tid */
  
***************
*** 554,559 **** heapgettup(HeapScanDesc scan,
--- 556,562 ----
  		LockBuffer(scan->rs_cbuf, BUFFER_LOCK_SHARE);
  
  		dp = (Page) BufferGetPage(scan->rs_cbuf);
+ 		TestForOldSnapshot(snapshot, scan->rs_rd, dp);
  		lines = PageGetMaxOffsetNumber(dp);
  
  		if (!scan->rs_inited)
***************
*** 712,717 **** heapgettup(HeapScanDesc scan,
--- 715,721 ----
  		LockBuffer(scan->rs_cbuf, BUFFER_LOCK_SHARE);
  
  		dp = (Page) BufferGetPage(scan->rs_cbuf);
+ 		TestForOldSnapshot(snapshot, scan->rs_rd, dp);
  		lines = PageGetMaxOffsetNumber((Page) dp);
  		linesleft = lines;
  		if (backward)
***************
*** 1648,1653 **** heap_fetch(Relation relation,
--- 1652,1658 ----
  	 */
  	LockBuffer(buffer, BUFFER_LOCK_SHARE);
  	page = BufferGetPage(buffer);
+ 	TestForOldSnapshot(snapshot, relation, page);
  
  	/*
  	 * We'd better check for out-of-range offnum in case of VACUUM since the
***************
*** 1977,1982 **** heap_get_latest_tid(Relation relation,
--- 1982,1988 ----
  		buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(&ctid));
  		LockBuffer(buffer, BUFFER_LOCK_SHARE);
  		page = BufferGetPage(buffer);
+ 		TestForOldSnapshot(snapshot, relation, page);
  
  		/*
  		 * Check for bogus item number.  This is not treated as an error
*** a/src/backend/access/heap/pruneheap.c
--- b/src/backend/access/heap/pruneheap.c
***************
*** 92,103 **** heap_page_prune_opt(Relation relation, Buffer buffer)
  	 * need to use the horizon that includes slots, otherwise the data-only
  	 * horizon can be used. Note that the toast relation of user defined
  	 * relations are *not* considered catalog relations.
  	 */
  	if (IsCatalogRelation(relation) ||
  		RelationIsAccessibleInLogicalDecoding(relation))
  		OldestXmin = RecentGlobalXmin;
  	else
! 		OldestXmin = RecentGlobalDataXmin;
  
  	Assert(TransactionIdIsValid(OldestXmin));
  
--- 92,112 ----
  	 * need to use the horizon that includes slots, otherwise the data-only
  	 * horizon can be used. Note that the toast relation of user defined
  	 * relations are *not* considered catalog relations.
+ 	 *
+ 	 * It is OK to apply the old snapshot limit before acquiring the cleanup
+ 	 * lock because the worst that can happen is that we are not quite as
+ 	 * aggressive about the cleanup (by however many transaction IDs are
+ 	 * consumed between this point and acquiring the lock).  This allows us to
+ 	 * save significant overhead in the case where the page is found not to be
+ 	 * prunable.
  	 */
  	if (IsCatalogRelation(relation) ||
  		RelationIsAccessibleInLogicalDecoding(relation))
  		OldestXmin = RecentGlobalXmin;
  	else
! 		OldestXmin =
! 				TransactionIdLimitedForOldSnapshots(RecentGlobalDataXmin,
! 													relation);
  
  	Assert(TransactionIdIsValid(OldestXmin));
  
*** a/src/backend/access/nbtree/nbtinsert.c
--- b/src/backend/access/nbtree/nbtinsert.c
***************
*** 119,125 **** _bt_doinsert(Relation rel, IndexTuple itup,
  
  top:
  	/* find the first page containing this key */
! 	stack = _bt_search(rel, natts, itup_scankey, false, &buf, BT_WRITE);
  
  	offset = InvalidOffsetNumber;
  
--- 119,125 ----
  
  top:
  	/* find the first page containing this key */
! 	stack = _bt_search(rel, natts, itup_scankey, false, &buf, BT_WRITE, NULL);
  
  	offset = InvalidOffsetNumber;
  
***************
*** 135,141 **** top:
  	 * precise description.
  	 */
  	buf = _bt_moveright(rel, buf, natts, itup_scankey, false,
! 						true, stack, BT_WRITE);
  
  	/*
  	 * If we're not allowing duplicates, make sure the key isn't already in
--- 135,141 ----
  	 * precise description.
  	 */
  	buf = _bt_moveright(rel, buf, natts, itup_scankey, false,
! 						true, stack, BT_WRITE, NULL);
  
  	/*
  	 * If we're not allowing duplicates, make sure the key isn't already in
***************
*** 1671,1677 **** _bt_insert_parent(Relation rel,
  			elog(DEBUG2, "concurrent ROOT page split");
  			lpageop = (BTPageOpaque) PageGetSpecialPointer(page);
  			/* Find the leftmost page at the next level up */
! 			pbuf = _bt_get_endpoint(rel, lpageop->btpo.level + 1, false);
  			/* Set up a phony stack entry pointing there */
  			stack = &fakestack;
  			stack->bts_blkno = BufferGetBlockNumber(pbuf);
--- 1671,1678 ----
  			elog(DEBUG2, "concurrent ROOT page split");
  			lpageop = (BTPageOpaque) PageGetSpecialPointer(page);
  			/* Find the leftmost page at the next level up */
! 			pbuf = _bt_get_endpoint(rel, lpageop->btpo.level + 1, false,
! 									NULL);
  			/* Set up a phony stack entry pointing there */
  			stack = &fakestack;
  			stack->bts_blkno = BufferGetBlockNumber(pbuf);
*** a/src/backend/access/nbtree/nbtpage.c
--- b/src/backend/access/nbtree/nbtpage.c
***************
*** 1255,1261 **** _bt_pagedel(Relation rel, Buffer buf)
  				itup_scankey = _bt_mkscankey(rel, targetkey);
  				/* find the leftmost leaf page containing this key */
  				stack = _bt_search(rel, rel->rd_rel->relnatts, itup_scankey,
! 								   false, &lbuf, BT_READ);
  				/* don't need a pin on the page */
  				_bt_relbuf(rel, lbuf);
  
--- 1255,1261 ----
  				itup_scankey = _bt_mkscankey(rel, targetkey);
  				/* find the leftmost leaf page containing this key */
  				stack = _bt_search(rel, rel->rd_rel->relnatts, itup_scankey,
! 								   false, &lbuf, BT_READ, NULL);
  				/* don't need a pin on the page */
  				_bt_relbuf(rel, lbuf);
  
*** a/src/backend/access/nbtree/nbtsearch.c
--- b/src/backend/access/nbtree/nbtsearch.c
***************
*** 22,27 ****
--- 22,28 ----
  #include "storage/predicate.h"
  #include "utils/lsyscache.h"
  #include "utils/rel.h"
+ #include "utils/snapmgr.h"
  #include "utils/tqual.h"
  
  
***************
*** 30,36 **** static bool _bt_readpage(IndexScanDesc scan, ScanDirection dir,
  static void _bt_saveitem(BTScanOpaque so, int itemIndex,
  			 OffsetNumber offnum, IndexTuple itup);
  static bool _bt_steppage(IndexScanDesc scan, ScanDirection dir);
! static Buffer _bt_walk_left(Relation rel, Buffer buf);
  static bool _bt_endpoint(IndexScanDesc scan, ScanDirection dir);
  static void _bt_drop_lock_and_maybe_pin(IndexScanDesc scan, BTScanPos sp);
  
--- 31,37 ----
  static void _bt_saveitem(BTScanOpaque so, int itemIndex,
  			 OffsetNumber offnum, IndexTuple itup);
  static bool _bt_steppage(IndexScanDesc scan, ScanDirection dir);
! static Buffer _bt_walk_left(Relation rel, Buffer buf, Snapshot snapshot);
  static bool _bt_endpoint(IndexScanDesc scan, ScanDirection dir);
  static void _bt_drop_lock_and_maybe_pin(IndexScanDesc scan, BTScanPos sp);
  
***************
*** 79,84 **** _bt_drop_lock_and_maybe_pin(IndexScanDesc scan, BTScanPos sp)
--- 80,89 ----
   * address of the leaf-page buffer, which is read-locked and pinned.
   * No locks are held on the parent pages, however!
   *
+  * If the snapshot parameter is not NULL, "old snapshot" checking will take
+  * place during the descent through the tree.  This is not needed when
+  * positioning for an insert or delete, so NULL is used for those cases.
+  *
   * NOTE that the returned buffer is read-locked regardless of the access
   * parameter.  However, access = BT_WRITE will allow an empty root page
   * to be created and returned.  When access = BT_READ, an empty index
***************
*** 87,93 **** _bt_drop_lock_and_maybe_pin(IndexScanDesc scan, BTScanPos sp)
   */
  BTStack
  _bt_search(Relation rel, int keysz, ScanKey scankey, bool nextkey,
! 		   Buffer *bufP, int access)
  {
  	BTStack		stack_in = NULL;
  
--- 92,98 ----
   */
  BTStack
  _bt_search(Relation rel, int keysz, ScanKey scankey, bool nextkey,
! 		   Buffer *bufP, int access, Snapshot snapshot)
  {
  	BTStack		stack_in = NULL;
  
***************
*** 96,102 **** _bt_search(Relation rel, int keysz, ScanKey scankey, bool nextkey,
--- 101,109 ----
  
  	/* If index is empty and access = BT_READ, no root page is created. */
  	if (!BufferIsValid(*bufP))
+ 	{
  		return (BTStack) NULL;
+ 	}
  
  	/* Loop iterates once per level descended in the tree */
  	for (;;)
***************
*** 124,130 **** _bt_search(Relation rel, int keysz, ScanKey scankey, bool nextkey,
  		 */
  		*bufP = _bt_moveright(rel, *bufP, keysz, scankey, nextkey,
  							  (access == BT_WRITE), stack_in,
! 							  BT_READ);
  
  		/* if this is a leaf page, we're done */
  		page = BufferGetPage(*bufP);
--- 131,137 ----
  		 */
  		*bufP = _bt_moveright(rel, *bufP, keysz, scankey, nextkey,
  							  (access == BT_WRITE), stack_in,
! 							  BT_READ, snapshot);
  
  		/* if this is a leaf page, we're done */
  		page = BufferGetPage(*bufP);
***************
*** 197,202 **** _bt_search(Relation rel, int keysz, ScanKey scankey, bool nextkey,
--- 204,213 ----
   * On entry, we have the buffer pinned and a lock of the type specified by
   * 'access'.  If we move right, we release the buffer and lock and acquire
   * the same on the right sibling.  Return value is the buffer we stop at.
+  *
+  * If the snapshot parameter is not NULL, "old snapshot" checking will take
+  * place during the descent through the tree.  This is not needed when
+  * positioning for an insert or delete, so NULL is used for those cases.
   */
  Buffer
  _bt_moveright(Relation rel,
***************
*** 206,212 **** _bt_moveright(Relation rel,
  			  bool nextkey,
  			  bool forupdate,
  			  BTStack stack,
! 			  int access)
  {
  	Page		page;
  	BTPageOpaque opaque;
--- 217,224 ----
  			  bool nextkey,
  			  bool forupdate,
  			  BTStack stack,
! 			  int access,
! 			  Snapshot snapshot)
  {
  	Page		page;
  	BTPageOpaque opaque;
***************
*** 232,237 **** _bt_moveright(Relation rel,
--- 244,250 ----
  	for (;;)
  	{
  		page = BufferGetPage(buf);
+ 		TestForOldSnapshot(snapshot, rel, page);
  		opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  
  		if (P_RIGHTMOST(opaque))
***************
*** 970,976 **** _bt_first(IndexScanDesc scan, ScanDirection dir)
  	 * Use the manufactured insertion scan key to descend the tree and
  	 * position ourselves on the target leaf page.
  	 */
! 	stack = _bt_search(rel, keysCount, scankeys, nextkey, &buf, BT_READ);
  
  	/* don't need to keep the stack around... */
  	_bt_freestack(stack);
--- 983,990 ----
  	 * Use the manufactured insertion scan key to descend the tree and
  	 * position ourselves on the target leaf page.
  	 */
! 	stack = _bt_search(rel, keysCount, scankeys, nextkey, &buf, BT_READ,
! 					   scan->xs_snapshot);
  
  	/* don't need to keep the stack around... */
  	_bt_freestack(stack);
***************
*** 1363,1368 **** _bt_steppage(IndexScanDesc scan, ScanDirection dir)
--- 1377,1383 ----
  			so->currPos.buf = _bt_getbuf(rel, blkno, BT_READ);
  			/* check for deleted page */
  			page = BufferGetPage(so->currPos.buf);
+ 			TestForOldSnapshot(scan->xs_snapshot, rel, page);
  			opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  			if (!P_IGNORE(opaque))
  			{
***************
*** 1421,1427 **** _bt_steppage(IndexScanDesc scan, ScanDirection dir)
  			}
  
  			/* Step to next physical page */
! 			so->currPos.buf = _bt_walk_left(rel, so->currPos.buf);
  
  			/* if we're physically at end of index, return failure */
  			if (so->currPos.buf == InvalidBuffer)
--- 1436,1443 ----
  			}
  
  			/* Step to next physical page */
! 			so->currPos.buf = _bt_walk_left(rel, so->currPos.buf,
! 											scan->xs_snapshot);
  
  			/* if we're physically at end of index, return failure */
  			if (so->currPos.buf == InvalidBuffer)
***************
*** 1436,1441 **** _bt_steppage(IndexScanDesc scan, ScanDirection dir)
--- 1452,1458 ----
  			 * and do it all again.
  			 */
  			page = BufferGetPage(so->currPos.buf);
+ 			TestForOldSnapshot(scan->xs_snapshot, rel, page);
  			opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  			if (!P_IGNORE(opaque))
  			{
***************
*** 1469,1475 **** _bt_steppage(IndexScanDesc scan, ScanDirection dir)
   * again if it's important.
   */
  static Buffer
! _bt_walk_left(Relation rel, Buffer buf)
  {
  	Page		page;
  	BTPageOpaque opaque;
--- 1486,1492 ----
   * again if it's important.
   */
  static Buffer
! _bt_walk_left(Relation rel, Buffer buf, Snapshot snapshot)
  {
  	Page		page;
  	BTPageOpaque opaque;
***************
*** 1499,1504 **** _bt_walk_left(Relation rel, Buffer buf)
--- 1516,1522 ----
  		CHECK_FOR_INTERRUPTS();
  		buf = _bt_getbuf(rel, blkno, BT_READ);
  		page = BufferGetPage(buf);
+ 		TestForOldSnapshot(snapshot, rel, page);
  		opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  
  		/*
***************
*** 1525,1536 **** _bt_walk_left(Relation rel, Buffer buf)
--- 1543,1556 ----
  			blkno = opaque->btpo_next;
  			buf = _bt_relandgetbuf(rel, buf, blkno, BT_READ);
  			page = BufferGetPage(buf);
+ 			TestForOldSnapshot(snapshot, rel, page);
  			opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  		}
  
  		/* Return to the original page to see what's up */
  		buf = _bt_relandgetbuf(rel, buf, obknum, BT_READ);
  		page = BufferGetPage(buf);
+ 		TestForOldSnapshot(snapshot, rel, page);
  		opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  		if (P_ISDELETED(opaque))
  		{
***************
*** 1548,1553 **** _bt_walk_left(Relation rel, Buffer buf)
--- 1568,1574 ----
  				blkno = opaque->btpo_next;
  				buf = _bt_relandgetbuf(rel, buf, blkno, BT_READ);
  				page = BufferGetPage(buf);
+ 				TestForOldSnapshot(snapshot, rel, page);
  				opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  				if (!P_ISDELETED(opaque))
  					break;
***************
*** 1584,1590 **** _bt_walk_left(Relation rel, Buffer buf)
   * The returned buffer is pinned and read-locked.
   */
  Buffer
! _bt_get_endpoint(Relation rel, uint32 level, bool rightmost)
  {
  	Buffer		buf;
  	Page		page;
--- 1605,1612 ----
   * The returned buffer is pinned and read-locked.
   */
  Buffer
! _bt_get_endpoint(Relation rel, uint32 level, bool rightmost,
! 				 Snapshot snapshot)
  {
  	Buffer		buf;
  	Page		page;
***************
*** 1607,1612 **** _bt_get_endpoint(Relation rel, uint32 level, bool rightmost)
--- 1629,1635 ----
  		return InvalidBuffer;
  
  	page = BufferGetPage(buf);
+ 	TestForOldSnapshot(snapshot, rel, page);
  	opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  
  	for (;;)
***************
*** 1626,1631 **** _bt_get_endpoint(Relation rel, uint32 level, bool rightmost)
--- 1649,1655 ----
  					 RelationGetRelationName(rel));
  			buf = _bt_relandgetbuf(rel, buf, blkno, BT_READ);
  			page = BufferGetPage(buf);
+ 			TestForOldSnapshot(snapshot, rel, page);
  			opaque = (BTPageOpaque) PageGetSpecialPointer(page);
  		}
  
***************
*** 1678,1684 **** _bt_endpoint(IndexScanDesc scan, ScanDirection dir)
  	 * version of _bt_search().  We don't maintain a stack since we know we
  	 * won't need it.
  	 */
! 	buf = _bt_get_endpoint(rel, 0, ScanDirectionIsBackward(dir));
  
  	if (!BufferIsValid(buf))
  	{
--- 1702,1708 ----
  	 * version of _bt_search().  We don't maintain a stack since we know we
  	 * won't need it.
  	 */
! 	buf = _bt_get_endpoint(rel, 0, ScanDirectionIsBackward(dir), scan->xs_snapshot);
  
  	if (!BufferIsValid(buf))
  	{
*** a/src/backend/access/spgist/spgscan.c
--- b/src/backend/access/spgist/spgscan.c
***************
*** 22,27 ****
--- 22,28 ----
  #include "utils/datum.h"
  #include "utils/memutils.h"
  #include "utils/rel.h"
+ #include "utils/snapmgr.h"
  
  
  typedef void (*storeRes_func) (SpGistScanOpaque so, ItemPointer heapPtr,
***************
*** 319,325 **** spgLeafTest(Relation index, SpGistScanOpaque so,
   */
  static void
  spgWalk(Relation index, SpGistScanOpaque so, bool scanWholeIndex,
! 		storeRes_func storeRes)
  {
  	Buffer		buffer = InvalidBuffer;
  	bool		reportedSome = false;
--- 320,326 ----
   */
  static void
  spgWalk(Relation index, SpGistScanOpaque so, bool scanWholeIndex,
! 		storeRes_func storeRes, Snapshot snapshot)
  {
  	Buffer		buffer = InvalidBuffer;
  	bool		reportedSome = false;
***************
*** 360,365 **** redirect:
--- 361,367 ----
  		/* else new pointer points to the same page, no work needed */
  
  		page = BufferGetPage(buffer);
+ 		TestForOldSnapshot(snapshot, index, page);
  
  		isnull = SpGistPageStoresNulls(page) ? true : false;
  
***************
*** 584,590 **** spggetbitmap(PG_FUNCTION_ARGS)
  	so->tbm = tbm;
  	so->ntids = 0;
  
! 	spgWalk(scan->indexRelation, so, true, storeBitmap);
  
  	PG_RETURN_INT64(so->ntids);
  }
--- 586,592 ----
  	so->tbm = tbm;
  	so->ntids = 0;
  
! 	spgWalk(scan->indexRelation, so, true, storeBitmap, scan->xs_snapshot);
  
  	PG_RETURN_INT64(so->ntids);
  }
***************
*** 645,651 **** spggettuple(PG_FUNCTION_ARGS)
  		}
  		so->iPtr = so->nPtrs = 0;
  
! 		spgWalk(scan->indexRelation, so, false, storeGettuple);
  
  		if (so->nPtrs == 0)
  			break;				/* must have completed scan */
--- 647,654 ----
  		}
  		so->iPtr = so->nPtrs = 0;
  
! 		spgWalk(scan->indexRelation, so, false, storeGettuple,
! 				scan->xs_snapshot);
  
  		if (so->nPtrs == 0)
  			break;				/* must have completed scan */
*** a/src/backend/commands/vacuum.c
--- b/src/backend/commands/vacuum.c
***************
*** 489,495 **** vacuum_set_xid_limits(Relation rel,
  	 * working on a particular table at any time, and that each vacuum is
  	 * always an independent transaction.
  	 */
! 	*oldestXmin = GetOldestXmin(rel, true);
  
  	Assert(TransactionIdIsNormal(*oldestXmin));
  
--- 489,496 ----
  	 * working on a particular table at any time, and that each vacuum is
  	 * always an independent transaction.
  	 */
! 	*oldestXmin =
! 		TransactionIdLimitedForOldSnapshots(GetOldestXmin(rel, true), rel);
  
  	Assert(TransactionIdIsNormal(*oldestXmin));
  
*** a/src/backend/commands/vacuumlazy.c
--- b/src/backend/commands/vacuumlazy.c
***************
*** 59,64 ****
--- 59,65 ----
  #include "utils/lsyscache.h"
  #include "utils/memutils.h"
  #include "utils/pg_rusage.h"
+ #include "utils/snapmgr.h"
  #include "utils/timestamp.h"
  #include "utils/tqual.h"
  
***************
*** 270,276 **** lazy_vacuum_rel(Relation onerel, int options, VacuumParams *params,
  	possibly_freeable = vacrelstats->rel_pages - vacrelstats->nonempty_pages;
  	if (possibly_freeable > 0 &&
  		(possibly_freeable >= REL_TRUNCATE_MINIMUM ||
! 		 possibly_freeable >= vacrelstats->rel_pages / REL_TRUNCATE_FRACTION))
  		lazy_truncate_heap(onerel, vacrelstats);
  
  	/* Vacuum the Free Space Map */
--- 271,278 ----
  	possibly_freeable = vacrelstats->rel_pages - vacrelstats->nonempty_pages;
  	if (possibly_freeable > 0 &&
  		(possibly_freeable >= REL_TRUNCATE_MINIMUM ||
! 		 possibly_freeable >= vacrelstats->rel_pages / REL_TRUNCATE_FRACTION) &&
! 		old_snapshot_threshold < 0)
  		lazy_truncate_heap(onerel, vacrelstats);
  
  	/* Vacuum the Free Space Map */
*** a/src/backend/storage/ipc/ipci.c
--- b/src/backend/storage/ipc/ipci.c
***************
*** 43,48 ****
--- 43,49 ----
  #include "storage/procsignal.h"
  #include "storage/sinvaladt.h"
  #include "storage/spin.h"
+ #include "utils/snapmgr.h"
  
  
  shmem_startup_hook_type shmem_startup_hook = NULL;
***************
*** 136,141 **** CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
--- 137,143 ----
  		size = add_size(size, ReplicationOriginShmemSize());
  		size = add_size(size, WalSndShmemSize());
  		size = add_size(size, WalRcvShmemSize());
+ 		size = add_size(size, SnapMgrShmemSize());
  		size = add_size(size, BTreeShmemSize());
  		size = add_size(size, SyncScanShmemSize());
  		size = add_size(size, AsyncShmemSize());
***************
*** 247,252 **** CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
--- 249,255 ----
  	/*
  	 * Set up other modules that need some shared memory space
  	 */
+ 	SnapMgrInit();
  	BTreeShmemInit();
  	SyncScanShmemInit();
  	AsyncShmemInit();
*** a/src/backend/storage/ipc/procarray.c
--- b/src/backend/storage/ipc/procarray.c
***************
*** 1732,1737 **** GetSnapshotData(Snapshot snapshot)
--- 1732,1746 ----
  	snapshot->regd_count = 0;
  	snapshot->copied = false;
  
+ 	/*
+ 	 * Capture the current time and WAL stream location in case this snapshot
+ 	 * becomes old enough to need to fall back on the special "old snapshot"
+ 	 * logic.
+ 	 */
+ 	snapshot->lsn = GetXLogInsertRecPtr();
+ 	snapshot->whenTaken = GetSnapshotCurrentTimestamp();
+ 	MaintainOldSnapshotTimeMapping(snapshot->whenTaken, xmin);
+ 
  	return snapshot;
  }
  
*** a/src/backend/utils/errcodes.txt
--- b/src/backend/utils/errcodes.txt
***************
*** 413,418 **** Section: Class 58 - System Error (errors external to PostgreSQL itself)
--- 413,422 ----
  58P01    E    ERRCODE_UNDEFINED_FILE                                         undefined_file
  58P02    E    ERRCODE_DUPLICATE_FILE                                         duplicate_file
  
+ Section: Class 72 - Snapshot Failure
+ # (class borrowed from Oracle)
+ 72000    E    ERRCODE_SNAPSHOT_TOO_OLD                                       snapshot_too_old
+ 
  Section: Class F0 - Configuration File Error
  
  # (PostgreSQL-specific error class)
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 2556,2561 **** static struct config_int ConfigureNamesInt[] =
--- 2556,2572 ----
  	},
  
  	{
+ 		{"old_snapshot_threshold", PGC_POSTMASTER, RESOURCES_ASYNCHRONOUS,
+ 			gettext_noop("Time before a snapshot is too old to read pages changed after the snapshot was taken."),
+ 			gettext_noop("A value of -1 disables this feature."),
+ 			GUC_UNIT_MIN
+ 		},
+ 		&old_snapshot_threshold,
+ 		-1, -1, MAX_OLD_SNAPSHOT_THRESHOLD,
+ 		NULL, NULL, NULL
+ 	},
+ 
+ 	{
  		{"tcp_keepalives_idle", PGC_USERSET, CLIENT_CONN_OTHER,
  			gettext_noop("Time between issuing TCP keepalives."),
  			gettext_noop("A value of 0 uses the system default."),
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 164,169 ****
--- 164,171 ----
  
  #effective_io_concurrency = 1		# 1-1000; 0 disables prefetching
  #max_worker_processes = 8
+ #old_snapshot_threshold = -1		# 1min-60d; -1 disables; 0 is immediate
+ 									# (change requires restart)
  
  
  #------------------------------------------------------------------------------
*** a/src/backend/utils/time/snapmgr.c
--- b/src/backend/utils/time/snapmgr.c
***************
*** 46,59 ****
--- 46,62 ----
  
  #include "access/transam.h"
  #include "access/xact.h"
+ #include "access/xlog.h"
  #include "lib/pairingheap.h"
  #include "miscadmin.h"
  #include "storage/predicate.h"
  #include "storage/proc.h"
  #include "storage/procarray.h"
  #include "storage/sinval.h"
+ #include "storage/spin.h"
  #include "utils/builtins.h"
  #include "utils/memutils.h"
+ #include "utils/rel.h"
  #include "utils/resowner_private.h"
  #include "utils/snapmgr.h"
  #include "utils/syscache.h"
***************
*** 61,66 ****
--- 64,124 ----
  
  
  /*
+  * GUC parameters
+  */
+ int			old_snapshot_threshold;		/* number of minutes, -1 disables */
+ 
+ /*
+  * Structure for dealing with old_snapshot_threshold implementation.
+  */
+ typedef struct OldSnapshotControlData
+ {
+ 	/*
+ 	 * Variables for old snapshot handling are shared among processes and are
+ 	 * only allowed to move forward.
+ 	 */
+ 	slock_t		mutex_current;			/* protect current timestamp */
+ 	int64		current_timestamp;		/* latest snapshot timestamp */
+ 	slock_t		mutex_threshold;		/* protect threshold fields */
+ 	int64		threshold_timestamp;	/* earlier snapshot is old */
+ 	TransactionId threshold_xid;		/* earlier xid may be gone */
+ 
+ 	/*
+ 	 * Keep one xid per minute for old snapshot error handling.
+ 	 *
+ 	 * Use a circular buffer with a head offset, a length, and a timestamp
+ 	 * corresponding to the xid at the head offset.  A length of zero means
+ 	 * that there are no times stored; a length of XID_AGING_BUCKETS means
+ 	 * that the buffer is full and the head must be advanced to add new
+ 	 * entries.  Use timestamps aligned to minute boundaries, since that seems
+ 	 * less surprising than aligning based on the first usage timestamp.
+ 	 *
+ 	 * It is OK if the xid for a given time slot is from earlier than
+ 	 * calculated by adding the number of minutes corresponding to the
+ 	 * (possibly wrapped) distance from the head offset to the time of the
+ 	 * head entry, since that just results in the vacuuming of old tuples
+ 	 * being slightly less aggressive.  It would not be OK for it to be off in
+ 	 * the other direction, since it might result in vacuuming tuples that are
+ 	 * still expected to be there.
+ 	 *
+ 	 * Use of an SLRU was considered but not chosen because it is more
+ 	 * heavyweight than is needed for this, and would probably not be any less
+ 	 * code to implement.
+ 	 *
+ 	 * Persistence is not needed.
+ 	 */
+ 	int			head_offset;		/* subscript of oldest tracked time */
+ 	int64		head_timestamp;		/* time corresponding to head xid */
+ 	int			count_used;			/* how many slots are in use */
+ 	TransactionId xid_by_minute[XID_AGING_BUCKETS];
+ }	OldSnapshotControlData;
+ 
+ typedef struct OldSnapshotControlData *OldSnapshotControl;
+ 
+ static volatile OldSnapshotControl oldSnapshotControl;
+ 
+ 
+ /*
   * CurrentSnapshot points to the only snapshot taken in transaction-snapshot
   * mode, and to the latest one taken in a read-committed transaction.
   * SecondarySnapshot is a snapshot that's always up-to-date as of the current
***************
*** 153,158 **** static Snapshot FirstXactSnapshot = NULL;
--- 211,217 ----
  static List *exportedSnapshots = NIL;
  
  /* Prototypes for local functions */
+ static int64 AlignTimestampToMinuteBoundary(int64 ts);
  static Snapshot CopySnapshot(Snapshot snapshot);
  static void FreeSnapshot(Snapshot snapshot);
  static void SnapshotResetXmin(void);
***************
*** 174,179 **** typedef struct SerializedSnapshotData
--- 233,272 ----
  	CommandId	curcid;
  } SerializedSnapshotData;
  
+ Size
+ SnapMgrShmemSize(void)
+ {
+ 	return sizeof(OldSnapshotControlData);
+ }
+ 
+ /*
+  * Initialize for managing old snapshot detection.
+  */
+ void
+ SnapMgrInit(void)
+ {
+ 	bool		found;
+ 
+ 	/*
+ 	 * Create or attach to the OldSnapshotControl structure.
+ 	 */
+ 	oldSnapshotControl = (OldSnapshotControl)
+ 		ShmemInitStruct("OldSnapshotControlData",
+ 						sizeof(OldSnapshotControlData), &found);
+ 
+ 	if (!found)
+ 	{
+ 		SpinLockInit(&oldSnapshotControl->mutex_current);
+ 		oldSnapshotControl->current_timestamp = 0;
+ 		SpinLockInit(&oldSnapshotControl->mutex_threshold);
+ 		oldSnapshotControl->threshold_timestamp = 0;
+ 		oldSnapshotControl->threshold_xid = InvalidTransactionId;
+ 		oldSnapshotControl->head_offset = 0;
+ 		oldSnapshotControl->head_timestamp = 0;
+ 		oldSnapshotControl->count_used = 0;
+ 	}
+ }
+ 
  /*
   * GetTransactionSnapshot
   *		Get the appropriate snapshot for a new query in a transaction.
***************
*** 1405,1410 **** ThereAreNoPriorRegisteredSnapshots(void)
--- 1498,1743 ----
  	return false;
  }
  
+ 
+ static int64
+ AlignTimestampToMinuteBoundary(int64 ts)
+ {
+ 	int64		retval = ts + (USECS_PER_MINUTE - 1);
+ 
+ 	return retval - (retval % USECS_PER_MINUTE);
+ }
+ 
+ /*
+  * Get current timestamp for snapshots as int64 that never moves backward.
+  */
+ int64
+ GetSnapshotCurrentTimestamp(void)
+ {
+ 	int64		now = GetCurrentIntegerTimestamp();
+ 
+ 	/*
+ 	 * Don't let time move backward; if it hasn't advanced, use the old value.
+ 	 */
+ 	SpinLockAcquire(&oldSnapshotControl->mutex_current);
+ 	if (now <= oldSnapshotControl->current_timestamp)
+ 		now = oldSnapshotControl->current_timestamp;
+ 	else
+ 		oldSnapshotControl->current_timestamp = now;
+ 	SpinLockRelease(&oldSnapshotControl->mutex_current);
+ 
+ 	return now;
+ }
+ 
+ /*
+  * Get timestamp through which vacuum may have processed based on last stored
+  * value for threshold_timestamp.
+  *
+  * XXX: If we can trust a read of an int64 value to be atomic, we can skip the
+  * spinlock here.
+  */
+ int64
+ GetOldSnapshotThresholdTimestamp(void)
+ {
+ 	int64		threshold_timestamp;
+ 
+ 	SpinLockAcquire(&oldSnapshotControl->mutex_threshold);
+ 	threshold_timestamp = oldSnapshotControl->threshold_timestamp;
+ 	SpinLockRelease(&oldSnapshotControl->mutex_threshold);
+ 
+ 	return threshold_timestamp;
+ }
+ 
+ /*
+  * TransactionIdLimitedForOldSnapshots
+  *
+  * Apply old snapshot limit, if any.  This is intended to be called for page
+  * pruning and table vacuuming, to allow old_snapshot_threshold to override
+  * the normal global xmin value.  Actual testing for snapshot too old will be
+  * based on whether a snapshot timestamp is prior to the threshold timestamp
+  * set in this function.
+  */
+ TransactionId
+ TransactionIdLimitedForOldSnapshots(TransactionId recentXmin,
+ 									Relation relation)
+ {
+ 	if (TransactionIdIsNormal(recentXmin)
+ 		&& old_snapshot_threshold >= 0
+ 		&& RelationNeedsWAL(relation)
+ 		&& !IsCatalogRelation(relation)
+ 		&& !RelationIsAccessibleInLogicalDecoding(relation))
+ 	{
+ 		int64		ts;
+ 		TransactionId xlimit = recentXmin;
+ 		bool		same_ts_as_threshold = false;
+ 
+ 		ts = AlignTimestampToMinuteBoundary(GetSnapshotCurrentTimestamp())
+ 			 - (old_snapshot_threshold * USECS_PER_MINUTE);
+ 
+ 		/* Check for fast exit without LW locking. */
+ 		SpinLockAcquire(&oldSnapshotControl->mutex_threshold);
+ 		if (ts == oldSnapshotControl->threshold_timestamp)
+ 		{
+ 			xlimit = oldSnapshotControl->threshold_xid;
+ 			same_ts_as_threshold = true;
+ 		}
+ 		SpinLockRelease(&oldSnapshotControl->mutex_threshold);
+ 
+ 		if (!same_ts_as_threshold)
+ 		{
+ 			LWLockAcquire(OldSnapshotTimeMapLock, LW_SHARED);
+ 
+ 			if (oldSnapshotControl->count_used > 0
+ 				&& ts >= oldSnapshotControl->head_timestamp)
+ 			{
+ 				int		offset;
+ 
+ 				offset = ((ts - oldSnapshotControl->head_timestamp)
+ 						  / USECS_PER_MINUTE);
+ 				if (offset > oldSnapshotControl->count_used - 1)
+ 					offset = oldSnapshotControl->count_used - 1;
+ 				offset = (oldSnapshotControl->head_offset + offset)
+ 						% XID_AGING_BUCKETS;
+ 				xlimit = oldSnapshotControl->xid_by_minute[offset];
+ 
+ 				if (NormalTransactionIdFollows(xlimit, recentXmin))
+ 				{
+ 					SpinLockAcquire(&oldSnapshotControl->mutex_threshold);
+ 					oldSnapshotControl->threshold_timestamp = ts;
+ 					oldSnapshotControl->threshold_xid = xlimit;
+ 					SpinLockRelease(&oldSnapshotControl->mutex_threshold);
+ 				}
+ 			}
+ 
+ 			LWLockRelease(OldSnapshotTimeMapLock);
+ 		}
+ 
+ 		if (NormalTransactionIdFollows(xlimit, recentXmin))
+ 			return xlimit;
+ 	}
+ 
+ 	return recentXmin;
+ }
+ 
+ /*
+  * Take care of the circular buffer that maps time to xid.
+  */
+ void
+ MaintainOldSnapshotTimeMapping(int64 whenTaken, TransactionId xmin)
+ {
+ 	int64		ts;
+ 
+ 	/*
+ 	 * We don't want to do something stupid with unusual values, but we don't
+ 	 * want to litter the log with warnings or break otherwise normal
+ 	 * processing for this feature; so if something seems unreasonable, just
+ 	 * log at DEBUG level and return without doing anything.
+ 	 */
+ 	if (whenTaken < 0)
+ 	{
+ 		elog(DEBUG1,
+ 			 "MaintainOldSnapshotTimeMapping called with negative whenTaken = %ld",
+ 			 (long) whenTaken);
+ 		return;
+ 	}
+ 	if (!TransactionIdIsNormal(xmin))
+ 	{
+ 		elog(DEBUG1,
+ 			 "MaintainOldSnapshotTimeMapping called with xmin = %lu",
+ 			 (unsigned long) xmin);
+ 		return;
+ 	}
+ 
+ 	ts = AlignTimestampToMinuteBoundary(whenTaken);
+ 
+ 	LWLockAcquire(OldSnapshotTimeMapLock, LW_EXCLUSIVE);
+ 
+ 	Assert(oldSnapshotControl->head_offset >= 0);
+ 	Assert(oldSnapshotControl->head_offset < XID_AGING_BUCKETS);
+ 	Assert((oldSnapshotControl->head_timestamp % USECS_PER_MINUTE) == 0);
+ 	Assert(oldSnapshotControl->count_used >= 0);
+ 	Assert(oldSnapshotControl->count_used <= XID_AGING_BUCKETS);
+ 
+ 	if (oldSnapshotControl->count_used == 0)
+ 	{
+ 		/* set up first entry for empty mapping */
+ 		oldSnapshotControl->head_offset = 0;
+ 		oldSnapshotControl->head_timestamp = ts;
+ 		oldSnapshotControl->count_used = 1;
+ 		oldSnapshotControl->xid_by_minute[0] = xmin;
+ 	}
+ 	else if (ts < oldSnapshotControl->head_timestamp)
+ 	{
+ 		/* old ts; log it at DEBUG */
+ 		LWLockRelease(OldSnapshotTimeMapLock);
+ 		elog(DEBUG1,
+ 			 "MaintainOldSnapshotTimeMapping called with old whenTaken = %ld",
+ 			 (long) whenTaken);
+ 		return;
+ 	}
+ 	else if (ts <= (oldSnapshotControl->head_timestamp +
+ 					((oldSnapshotControl->count_used - 1)
+ 					 * USECS_PER_MINUTE)))
+ 	{
+ 		/* existing mapping; advance xid if possible */
+ 		int		bucket = (oldSnapshotControl->head_offset
+ 						  + ((ts - oldSnapshotControl->head_timestamp)
+ 							 / USECS_PER_MINUTE))
+ 						 % XID_AGING_BUCKETS;
+ 
+ 		if (TransactionIdPrecedes(oldSnapshotControl->xid_by_minute[bucket], xmin))
+ 			oldSnapshotControl->xid_by_minute[bucket] = xmin;
+ 	}
+ 	else
+ 	{
+ 		/* We need a new bucket, but it might not be the very next one. */
+ 		int		advance = ((ts - oldSnapshotControl->head_timestamp)
+ 						   / USECS_PER_MINUTE);
+ 
+ 		oldSnapshotControl->head_timestamp = ts;
+ 
+ 		if (advance >= XID_AGING_BUCKETS)
+ 		{
+ 			/* Advance is so far that all old data is junk; start over. */
+ 			oldSnapshotControl->head_offset = 0;
+ 			oldSnapshotControl->count_used = 1;
+ 			oldSnapshotControl->xid_by_minute[0] = xmin;
+ 		}
+ 		else
+ 		{
+ 			/* Store the new value in one or more buckets. */
+ 			int i;
+ 
+ 			for (i = 0; i < advance; i++)
+ 			{
+ 				if (oldSnapshotControl->count_used == XID_AGING_BUCKETS)
+ 				{
+ 					/* Map full and new value replaces old head. */
+ 					int		old_head = oldSnapshotControl->head_offset;
+ 
+ 					if (old_head == (XID_AGING_BUCKETS - 1))
+ 						oldSnapshotControl->head_offset = 0;
+ 					else
+ 						oldSnapshotControl->head_offset = old_head + 1;
+ 					oldSnapshotControl->xid_by_minute[old_head] = xmin;
+ 				}
+ 				else
+ 				{
+ 					/* Extend map to unused entry. */
+ 					int		new_tail = (oldSnapshotControl->head_offset
+ 										+ oldSnapshotControl->count_used)
+ 									   % XID_AGING_BUCKETS;
+ 
+ 					oldSnapshotControl->count_used++;
+ 					oldSnapshotControl->xid_by_minute[new_tail] = xmin;
+ 				}
+ 			}
+ 		}
+ 	}
+ 
+ 	LWLockRelease(OldSnapshotTimeMapLock);
+ }
+ 
+ 
  /*
   * Setup a snapshot that replaces normal catalog snapshots that allows catalog
   * access to behave just like it did at a certain point in the past.
*** a/src/include/access/brin_revmap.h
--- b/src/include/access/brin_revmap.h
***************
*** 18,29 ****
  #include "storage/itemptr.h"
  #include "storage/off.h"
  #include "utils/relcache.h"
  
  /* struct definition lives in brin_revmap.c */
  typedef struct BrinRevmap BrinRevmap;
  
  extern BrinRevmap *brinRevmapInitialize(Relation idxrel,
! 					 BlockNumber *pagesPerRange);
  extern void brinRevmapTerminate(BrinRevmap *revmap);
  
  extern void brinRevmapExtend(BrinRevmap *revmap,
--- 18,30 ----
  #include "storage/itemptr.h"
  #include "storage/off.h"
  #include "utils/relcache.h"
+ #include "utils/snapshot.h"
  
  /* struct definition lives in brin_revmap.c */
  typedef struct BrinRevmap BrinRevmap;
  
  extern BrinRevmap *brinRevmapInitialize(Relation idxrel,
! 					 BlockNumber *pagesPerRange, Snapshot snapshot);
  extern void brinRevmapTerminate(BrinRevmap *revmap);
  
  extern void brinRevmapExtend(BrinRevmap *revmap,
***************
*** 34,39 **** extern void brinSetHeapBlockItemptr(Buffer rmbuf, BlockNumber pagesPerRange,
  						BlockNumber heapBlk, ItemPointerData tid);
  extern BrinTuple *brinGetTupleForHeapBlock(BrinRevmap *revmap,
  						 BlockNumber heapBlk, Buffer *buf, OffsetNumber *off,
! 						 Size *size, int mode);
  
  #endif   /* BRIN_REVMAP_H */
--- 35,40 ----
  						BlockNumber heapBlk, ItemPointerData tid);
  extern BrinTuple *brinGetTupleForHeapBlock(BrinRevmap *revmap,
  						 BlockNumber heapBlk, Buffer *buf, OffsetNumber *off,
! 						 Size *size, int mode, Snapshot snapshot);
  
  #endif   /* BRIN_REVMAP_H */
*** a/src/include/access/gin_private.h
--- b/src/include/access/gin_private.h
***************
*** 699,705 **** typedef struct
   * PostingItem
   */
  
! extern GinBtreeStack *ginFindLeafPage(GinBtree btree, bool searchMode);
  extern Buffer ginStepRight(Buffer buffer, Relation index, int lockmode);
  extern void freeGinBtreeStack(GinBtreeStack *stack);
  extern void ginInsertValue(GinBtree btree, GinBtreeStack *stack,
--- 699,705 ----
   * PostingItem
   */
  
! extern GinBtreeStack *ginFindLeafPage(GinBtree btree, bool searchMode, Snapshot snapshot);
  extern Buffer ginStepRight(Buffer buffer, Relation index, int lockmode);
  extern void freeGinBtreeStack(GinBtreeStack *stack);
  extern void ginInsertValue(GinBtree btree, GinBtreeStack *stack,
***************
*** 727,733 **** extern void GinPageDeletePostingItem(Page page, OffsetNumber offset);
  extern void ginInsertItemPointers(Relation index, BlockNumber rootBlkno,
  					  ItemPointerData *items, uint32 nitem,
  					  GinStatsData *buildStats);
! extern GinBtreeStack *ginScanBeginPostingTree(GinBtree btree, Relation index, BlockNumber rootBlkno);
  extern void ginDataFillRoot(GinBtree btree, Page root, BlockNumber lblkno, Page lpage, BlockNumber rblkno, Page rpage);
  extern void ginPrepareDataScan(GinBtree btree, Relation index, BlockNumber rootBlkno);
  
--- 727,733 ----
  extern void ginInsertItemPointers(Relation index, BlockNumber rootBlkno,
  					  ItemPointerData *items, uint32 nitem,
  					  GinStatsData *buildStats);
! extern GinBtreeStack *ginScanBeginPostingTree(GinBtree btree, Relation index, BlockNumber rootBlkno, Snapshot snapshot);
  extern void ginDataFillRoot(GinBtree btree, Page root, BlockNumber lblkno, Page lpage, BlockNumber rblkno, Page rpage);
  extern void ginPrepareDataScan(GinBtree btree, Relation index, BlockNumber rootBlkno);
  
*** a/src/include/access/nbtree.h
--- b/src/include/access/nbtree.h
***************
*** 701,717 **** extern int	_bt_pagedel(Relation rel, Buffer buf);
   */
  extern BTStack _bt_search(Relation rel,
  		   int keysz, ScanKey scankey, bool nextkey,
! 		   Buffer *bufP, int access);
  extern Buffer _bt_moveright(Relation rel, Buffer buf, int keysz,
  			  ScanKey scankey, bool nextkey, bool forupdate, BTStack stack,
! 			  int access);
  extern OffsetNumber _bt_binsrch(Relation rel, Buffer buf, int keysz,
  			ScanKey scankey, bool nextkey);
  extern int32 _bt_compare(Relation rel, int keysz, ScanKey scankey,
  			Page page, OffsetNumber offnum);
  extern bool _bt_first(IndexScanDesc scan, ScanDirection dir);
  extern bool _bt_next(IndexScanDesc scan, ScanDirection dir);
! extern Buffer _bt_get_endpoint(Relation rel, uint32 level, bool rightmost);
  
  /*
   * prototypes for functions in nbtutils.c
--- 701,718 ----
   */
  extern BTStack _bt_search(Relation rel,
  		   int keysz, ScanKey scankey, bool nextkey,
! 		   Buffer *bufP, int access, Snapshot snapshot);
  extern Buffer _bt_moveright(Relation rel, Buffer buf, int keysz,
  			  ScanKey scankey, bool nextkey, bool forupdate, BTStack stack,
! 			  int access, Snapshot snapshot);
  extern OffsetNumber _bt_binsrch(Relation rel, Buffer buf, int keysz,
  			ScanKey scankey, bool nextkey);
  extern int32 _bt_compare(Relation rel, int keysz, ScanKey scankey,
  			Page page, OffsetNumber offnum);
  extern bool _bt_first(IndexScanDesc scan, ScanDirection dir);
  extern bool _bt_next(IndexScanDesc scan, ScanDirection dir);
! extern Buffer _bt_get_endpoint(Relation rel, uint32 level, bool rightmost,
! 							   Snapshot snapshot);
  
  /*
   * prototypes for functions in nbtutils.c
*** a/src/include/storage/lwlock.h
--- b/src/include/storage/lwlock.h
***************
*** 139,146 **** extern PGDLLIMPORT LWLockPadded *MainLWLockArray;
  #define CommitTsControlLock			(&MainLWLockArray[38].lock)
  #define CommitTsLock				(&MainLWLockArray[39].lock)
  #define ReplicationOriginLock		(&MainLWLockArray[40].lock)
  
! #define NUM_INDIVIDUAL_LWLOCKS		41
  
  /*
   * It's a bit odd to declare NUM_BUFFER_PARTITIONS and NUM_LOCK_PARTITIONS
--- 139,147 ----
  #define CommitTsControlLock			(&MainLWLockArray[38].lock)
  #define CommitTsLock				(&MainLWLockArray[39].lock)
  #define ReplicationOriginLock		(&MainLWLockArray[40].lock)
+ #define OldSnapshotTimeMapLock		(&MainLWLockArray[41].lock)
  
! #define NUM_INDIVIDUAL_LWLOCKS		42
  
  /*
   * It's a bit odd to declare NUM_BUFFER_PARTITIONS and NUM_LOCK_PARTITIONS
*** a/src/include/utils/rel.h
--- b/src/include/utils/rel.h
***************
*** 15,20 ****
--- 15,21 ----
  #define REL_H
  
  #include "access/tupdesc.h"
+ #include "access/xlog.h"
  #include "catalog/pg_am.h"
  #include "catalog/pg_class.h"
  #include "catalog/pg_index.h"
*** a/src/include/utils/snapmgr.h
--- b/src/include/utils/snapmgr.h
***************
*** 14,22 ****
--- 14,65 ----
  #define SNAPMGR_H
  
  #include "fmgr.h"
+ #include "catalog/catalog.h"
+ #include "utils/rel.h"
  #include "utils/resowner.h"
  #include "utils/snapshot.h"
+ #include "utils/tqual.h"
  
+ /*
+  * We keep a circular buffer of what transaction ID was next to assign by time
+  * with roughly one minute granularity, to support generation of "snapshot too
+  * old" errors.  Define the number of buckets.
+  */
+ #define XID_AGING_BUCKETS (MINS_PER_HOUR * HOURS_PER_DAY * 60)
+ 
+ /*
+  * At least for now, the maximum number of minutes for the GUC controlling the
+  * "snapshot too old" error is the same as the number of aging buckets; but
+  * let's use a different define in case the implementation changes.
+  */
+ #define MAX_OLD_SNAPSHOT_THRESHOLD XID_AGING_BUCKETS
+ 
+ /*
+  * This is a macro for speed; keep the tests that are fastest and/or most
+  * likely to exclude a page from old snapshot testing near the front.
+  */
+ #define TestForOldSnapshot(snapshot, relation, page) \
+ 	do { \
+ 		if ((snapshot) != NULL \
+ 		 && (snapshot)->satisfies == HeapTupleSatisfiesMVCC \
+ 		 && !XLogRecPtrIsInvalid((snapshot)->lsn) \
+ 		 && PageGetLSN(page) > (snapshot)->lsn \
+ 		 && !IsCatalogRelation(relation) \
+ 		 && !RelationIsAccessibleInLogicalDecoding(relation) \
+ 		 && (snapshot)->whenTaken < GetOldSnapshotThresholdTimestamp()) \
+ 			ereport(ERROR, \
+ 					(errcode(ERRCODE_SNAPSHOT_TOO_OLD), \
+ 					 errmsg("snapshot too old"))); \
+ 	} while (0)
+ 
+ 
+ /* GUC variables */
+ extern int	old_snapshot_threshold;
+ 
+ extern Size SnapMgrShmemSize(void);
+ extern void SnapMgrInit(void);
+ extern int64 GetSnapshotCurrentTimestamp(void);
+ extern int64 GetOldSnapshotThresholdTimestamp(void);
  
  extern bool FirstSnapshotSet;
  
***************
*** 54,59 **** extern void ImportSnapshot(const char *idstr);
--- 97,105 ----
  extern bool XactHasExportedSnapshots(void);
  extern void DeleteAllExportedSnapshotFiles(void);
  extern bool ThereAreNoPriorRegisteredSnapshots(void);
+ extern TransactionId TransactionIdLimitedForOldSnapshots(TransactionId recentXmin,
+ 														 Relation relation);
+ extern void MaintainOldSnapshotTimeMapping(int64 whenTaken, TransactionId xmin);
  
  extern char *ExportSnapshot(Snapshot snapshot);
  
*** a/src/include/utils/snapshot.h
--- b/src/include/utils/snapshot.h
***************
*** 14,19 ****
--- 14,20 ----
  #define SNAPSHOT_H
  
  #include "access/htup.h"
+ #include "access/xlogdefs.h"
  #include "lib/pairingheap.h"
  #include "storage/buf.h"
  
***************
*** 105,110 **** typedef struct SnapshotData
--- 106,114 ----
  	uint32		active_count;	/* refcount on ActiveSnapshot stack */
  	uint32		regd_count;		/* refcount on RegisteredSnapshots */
  	pairingheap_node ph_node;	/* link in the RegisteredSnapshots heap */
+ 
+ 	int64		whenTaken;		/* timestamp when snapshot was taken */
+ 	XLogRecPtr	lsn;			/* position in the WAL stream when taken */
  } SnapshotData;
  
  /*
*** a/src/test/Makefile
--- b/src/test/Makefile
***************
*** 12,18 **** subdir = src/test
  top_builddir = ../..
  include $(top_builddir)/src/Makefile.global
  
! SUBDIRS = regress isolation modules
  
  # We don't build or execute examples/, locale/, or thread/ by default,
  # but we do want "make clean" etc to recurse into them.  Likewise for ssl/,
--- 12,18 ----
  top_builddir = ../..
  include $(top_builddir)/src/Makefile.global
  
! SUBDIRS = regress isolation modules sto
  
  # We don't build or execute examples/, locale/, or thread/ by default,
  # but we do want "make clean" etc to recurse into them.  Likewise for ssl/,
*** a/src/test/perl/TestLib.pm
--- b/src/test/perl/TestLib.pm
***************
*** 135,141 **** sub tempdir_short
  # --config-auth).
  sub standard_initdb
  {
! 	my $pgdata = shift;
  	system_or_bail('initdb', '-D', "$pgdata", '-A' , 'trust', '-N');
  	system_or_bail($ENV{PG_REGRESS}, '--config-auth', $pgdata);
  
--- 135,141 ----
  # --config-auth).
  sub standard_initdb
  {
! 	my ($pgdata, $confappend) = @_;
  	system_or_bail('initdb', '-D', "$pgdata", '-A' , 'trust', '-N');
  	system_or_bail($ENV{PG_REGRESS}, '--config-auth', $pgdata);
  
***************
*** 152,157 **** sub standard_initdb
--- 152,162 ----
  		print CONF "unix_socket_directories = '$tempdir_short'\n";
  		print CONF "listen_addresses = ''\n";
  	}
+ 	if (defined $confappend and length $confappend)
+ 	{
+ 		print CONF "\n# Added by test\n";
+ 		print CONF "$confappend\n";
+ 	}
  	close CONF;
  
  	$ENV{PGHOST}         = $windows_os ? "127.0.0.1" : $tempdir_short;
***************
*** 182,192 **** my ($test_server_datadir, $test_server_logfile);
  # Initialize a new cluster for testing in given directory, and start it.
  sub start_test_server
  {
! 	my ($tempdir) = @_;
  	my $ret;
  
  	print("### Starting test server in $tempdir\n");
! 	standard_initdb "$tempdir/pgdata";
  
  	$ret = system_log('pg_ctl', '-D', "$tempdir/pgdata", '-w', '-l',
  	  "$log_path/postmaster.log", '-o', "--fsync=off --log-statement=all",
--- 187,197 ----
  # Initialize a new cluster for testing in given directory, and start it.
  sub start_test_server
  {
! 	my ($tempdir, $confappend) = @_;
  	my $ret;
  
  	print("### Starting test server in $tempdir\n");
! 	standard_initdb "$tempdir/pgdata", $confappend;
  
  	$ret = system_log('pg_ctl', '-D', "$tempdir/pgdata", '-w', '-l',
  	  "$log_path/postmaster.log", '-o', "--fsync=off --log-statement=all",
*** /dev/null
--- b/src/test/sto/.gitignore
***************
*** 0 ****
--- 1,2 ----
+ # Generated by regression tests
+ /tmp_check/
*** /dev/null
--- b/src/test/sto/Makefile
***************
*** 0 ****
--- 1,20 ----
+ #-------------------------------------------------------------------------
+ #
+ # Makefile for src/test/tso
+ #
+ # Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
+ # Portions Copyright (c) 1994, Regents of the University of California
+ #
+ # src/test/tso/Makefile
+ #
+ #-------------------------------------------------------------------------
+ 
+ subdir = src/test/tso
+ top_builddir = ../../..
+ include $(top_builddir)/src/Makefile.global
+ 
+ clean distclean maintainer-clean:
+ 	rm -rf tmp_check
+ 
+ check:
+ 	$(prove_check)
*** /dev/null
--- b/src/test/sto/t/001_snapshot_too_old.pl
***************
*** 0 ****
--- 1,121 ----
+ use strict;
+ use warnings;
+ 
+ use Config;
+ use TestLib;
+ use DBI;
+ use DBD::Pg ':async';
+ 
+ use Test::More tests => 9;
+ 
+ my $tempdir       = TestLib::tempdir;
+ 
+ my $driver   = "Pg";
+ my $database = "postgres";
+ 
+ sub log_sleep
+ {
+ 	my ($t) = @_;
+ 	print "sleeping $t seconds...";
+ 	sleep $t;
+ 	print "done!\n";
+ }
+ 
+ sub log_sleep_to_new_minute
+ {
+ 	my ($sec) = localtime;
+ 	log_sleep (61 - $sec);
+ }
+ 
+ sub connect_ok
+ {
+ 	my ($db) = @_;
+ 	my $dbh = DBI->connect("dbi:$driver:database=$db", '', '',
+ 						   {AutoCommit => 1, RaiseError => 0, PrintError => 1})
+ 		or print $DBI::errstr;
+ 	ok((defined $dbh), "get DBI connection to $database");
+ 	return $dbh;
+ }
+ 
+ sub setting_ok
+ {
+ 	my ($dbh, $setting, $expected_value) = @_;
+ 	my $actual_value = $dbh->selectrow_array("SHOW $setting");
+ 	is($actual_value, $expected_value, "check $setting");
+ }
+ 
+ sub select_scalar_ok
+ {
+ 	my ($dbh, $statement, $expected_value) = @_;
+ 	my $actual_value = $dbh->selectrow_array($statement);
+ 	is($actual_value, $expected_value, "SELECT scalar value");
+ }
+ 
+ sub fetch_first_ok
+ {
+ 	my ($dbh) = @_;
+ 	select_scalar_ok($dbh, 'FETCH FIRST FROM cursor1', '1');
+ }
+ 
+ sub snapshot_too_old_ok
+ {
+ 	my ($dbh) = @_;
+ 	my $actual_value = $dbh->selectrow_array('FETCH FIRST FROM cursor1');
+ 	is($dbh->state, '72000', 'expect "snapshot too old" error');
+ }
+ 
+ # Initialize cluster
+ start_test_server($tempdir, qq(
+ old_snapshot_threshold = '1min'
+ autovacuum = off
+ enable_indexonlyscan = off
+ max_connections = 10
+ ));
+ 
+ # Confirm that we can connect to the database.
+ my $conn1 = connect_ok($database);
+ my $conn2 = connect_ok($database);
+ 
+ # Confirm that the settings "took".
+ setting_ok($conn1, 'old_snapshot_threshold', '1min');
+ 
+ # Do some setup.
+ $conn1->do(qq(CREATE TABLE t (c int NOT NULL)));
+ $conn1->do(qq(INSERT INTO t SELECT generate_series(1, 1000)));
+ $conn1->do(qq(VACUUM ANALYZE t));
+ $conn1->do(qq(CREATE TABLE u (c int NOT NULL)));
+ 
+ # Open long-lived cursor
+ $conn1->begin_work;
+ $conn1->do(qq(DECLARE cursor1 CURSOR FOR SELECT c FROM t));
+ 
+ # Bump the last-used transaction number.
+ $conn2->do(qq(INSERT INTO u VALUES (1)));
+ 
+ # Confirm immediate fetch works.
+ fetch_first_ok($conn1);
+ 
+ # Confirm delayed fetch works with no modifications.
+ log_sleep_to_new_minute;
+ fetch_first_ok($conn1);
+ 
+ # Try again to confirm no pruning affect.
+ fetch_first_ok($conn1);
+ 
+ # Confirm that fetch immediately after update is OK.
+ # This assumes that they happen within one minute of each other.
+ # Still using same snapshot, so value shouldn't change.
+ $conn2->do(qq(UPDATE t SET c = 1001 WHERE c = 1));
+ fetch_first_ok($conn1);
+ 
+ # Try again to confirm no pruning affect.
+ fetch_first_ok($conn1);
+ 
+ # Passage of time should now make this get the "snapshot too old" error.
+ log_sleep_to_new_minute;
+ snapshot_too_old_ok($conn1);
+ $conn1->rollback;
+ 
+ # Clean up a bit, just to minimize noise in log file.
+ $conn2->disconnect;
+ $conn1->disconnect;
