5 files changed, 86 insertions, 18 deletions

diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 49df05653b3..f428729ea04 100644
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -882,9 +882,27 @@ gistNewBuffer(Relation r)
 bool
 gistPageRecyclable(Page page)
 {
-	return PageIsNew(page) ||
-		(GistPageIsDeleted(page) &&
-		 TransactionIdPrecedes(GistPageGetDeleteXid(page), RecentGlobalXmin));
+	if (PageIsNew(page))
+		return true;
+	if (GistPageIsDeleted(page))
+	{
+		/*
+		 * The page was deleted, but when? If it was just deleted, a scan
+		 * might have seen the downlink to it, and will read the page later.
+		 * As long as that can happen, we must keep the deleted page around as
+		 * a tombstone.
+		 *
+		 * Compare the deletion XID with RecentGlobalXmin. If deleteXid <
+		 * RecentGlobalXmin, then no scan that's still in progress could have
+		 * seen its downlink, and we can recycle it.
+		 */
+		FullTransactionId deletexid_full = GistPageGetDeleteXid(page);
+		FullTransactionId recentxmin_full = GetFullRecentGlobalXmin();
+
+		if (FullTransactionIdPrecedes(deletexid_full, recentxmin_full))
+			return true;
+	}
+	return false;
 }
 
 bytea *
diff --git a/src/backend/access/gist/gistvacuum.c b/src/backend/access/gist/gistvacuum.c
index 4270226eee2..bf754ea6d0d 100644
--- a/src/backend/access/gist/gistvacuum.c
+++ b/src/backend/access/gist/gistvacuum.c
@@ -595,7 +595,7 @@ gistdeletepage(IndexVacuumInfo *info, GistBulkDeleteResult *stats,
 	ItemId		iid;
 	IndexTuple	idxtuple;
 	XLogRecPtr	recptr;
-	TransactionId txid;
+	FullTransactionId txid;
 
 	/*
 	 * Check that the leaf is still empty and deletable.
@@ -648,14 +648,13 @@ gistdeletepage(IndexVacuumInfo *info, GistBulkDeleteResult *stats,
 	 * currently in progress must have ended.  (That's much more conservative
 	 * than needed, but let's keep it safe and simple.)
 	 */
-	txid = ReadNewTransactionId();
+	txid = ReadNextFullTransactionId();
 
 	START_CRIT_SECTION();
 
 	/* mark the page as deleted */
 	MarkBufferDirty(leafBuffer);
-	GistPageSetDeleteXid(leafPage, txid);
-	GistPageSetDeleted(leafPage);
+	GistPageSetDeleted(leafPage, txid);
 	stats->stats.pages_deleted++;
 
 	/* remove the downlink from the parent */
diff --git a/src/backend/access/gist/gistxlog.c b/src/backend/access/gist/gistxlog.c
index 503db34d863..3b28f546465 100644
--- a/src/backend/access/gist/gistxlog.c
+++ b/src/backend/access/gist/gistxlog.c
@@ -356,8 +356,7 @@ gistRedoPageDelete(XLogReaderState *record)
 	{
 		Page		page = (Page) BufferGetPage(leafBuffer);
 
-		GistPageSetDeleteXid(page, xldata->deleteXid);
-		GistPageSetDeleted(page);
+		GistPageSetDeleted(page, xldata->deleteXid);
 
 		PageSetLSN(page, lsn);
 		MarkBufferDirty(leafBuffer);
@@ -396,8 +395,27 @@ gistRedoPageReuse(XLogReaderState *record)
 	 */
 	if (InHotStandby)
 	{
-		ResolveRecoveryConflictWithSnapshot(xlrec->latestRemovedXid,
-											xlrec->node);
+		FullTransactionId latestRemovedFullXid = xlrec->latestRemovedFullXid;
+		FullTransactionId nextFullXid = ReadNextFullTransactionId();
+		uint64		diff;
+
+		/*
+		 * ResolveRecoveryConflictWithSnapshot operates on 32-bit
+		 * TransactionIds, so truncate the logged FullTransactionId. If the
+		 * logged value is very old, so that XID wrap-around already happened
+		 * on it, there can't be any snapshots that still see it.
+		 */
+		nextFullXid = ReadNextFullTransactionId();
+		diff = U64FromFullTransactionId(nextFullXid) -
+			U64FromFullTransactionId(latestRemovedFullXid);
+		if (diff < MaxTransactionId / 2)
+		{
+			TransactionId latestRemovedXid;
+
+			latestRemovedXid = XidFromFullTransactionId(latestRemovedFullXid);
+			ResolveRecoveryConflictWithSnapshot(latestRemovedXid,
+												xlrec->node);
+		}
 	}
 }
 
@@ -554,7 +572,7 @@ gistXLogSplit(bool page_is_leaf,
  * downlink from the parent page.
  */
 XLogRecPtr
-gistXLogPageDelete(Buffer buffer, TransactionId xid,
+gistXLogPageDelete(Buffer buffer, FullTransactionId xid,
 				   Buffer parentBuffer, OffsetNumber downlinkOffset)
 {
 	gistxlogPageDelete xlrec;
@@ -578,7 +596,7 @@ gistXLogPageDelete(Buffer buffer, TransactionId xid,
  * Write XLOG record about reuse of a deleted page.
  */
 void
-gistXLogPageReuse(Relation rel, BlockNumber blkno, TransactionId latestRemovedXid)
+gistXLogPageReuse(Relation rel, BlockNumber blkno, FullTransactionId latestRemovedXid)
 {
 	gistxlogPageReuse xlrec_reuse;
 
@@ -591,7 +609,7 @@ gistXLogPageReuse(Relation rel, BlockNumber blkno, TransactionId latestRemovedXi
 	/* XLOG stuff */
 	xlrec_reuse.node = rel->rd_node;
 	xlrec_reuse.block = blkno;
-	xlrec_reuse.latestRemovedXid = latestRemovedXid;
+	xlrec_reuse.latestRemovedFullXid = latestRemovedXid;
 
 	XLogBeginInsert();
 	XLogRegisterData((char *) &xlrec_reuse, SizeOfGistxlogPageReuse);
diff --git a/src/backend/access/rmgrdesc/gistdesc.c b/src/backend/access/rmgrdesc/gistdesc.c
index 767864b58e6..eccb6fd9428 100644
--- a/src/backend/access/rmgrdesc/gistdesc.c
+++ b/src/backend/access/rmgrdesc/gistdesc.c
@@ -26,10 +26,11 @@ out_gistxlogPageUpdate(StringInfo buf, gistxlogPageUpdate *xlrec)
 static void
 out_gistxlogPageReuse(StringInfo buf, gistxlogPageReuse *xlrec)
 {
-	appendStringInfo(buf, "rel %u/%u/%u; blk %u; latestRemovedXid %u",
+	appendStringInfo(buf, "rel %u/%u/%u; blk %u; latestRemovedXid %u:%u",
 					 xlrec->node.spcNode, xlrec->node.dbNode,
 					 xlrec->node.relNode, xlrec->block,
-					 xlrec->latestRemovedXid);
+					 EpochFromFullTransactionId(xlrec->latestRemovedFullXid),
+					 XidFromFullTransactionId(xlrec->latestRemovedFullXid));
 }
 
 static void
@@ -50,8 +51,10 @@ out_gistxlogPageSplit(StringInfo buf, gistxlogPageSplit *xlrec)
 static void
 out_gistxlogPageDelete(StringInfo buf, gistxlogPageDelete *xlrec)
 {
-	appendStringInfo(buf, "deleteXid %u; downlink %u",
-					 xlrec->deleteXid, xlrec->downlinkOffset);
+	appendStringInfo(buf, "deleteXid %u:%u; downlink %u",
+					 EpochFromFullTransactionId(xlrec->deleteXid),
+					 XidFromFullTransactionId(xlrec->deleteXid),
+					 xlrec->downlinkOffset);
 }
 
 void
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
index ef9fc15ac36..d07ca1b0b24 100644
--- a/src/backend/utils/time/snapmgr.c
+++ b/src/backend/utils/time/snapmgr.c
@@ -957,6 +957,36 @@ xmin_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg)
 }
 
 /*
+ * Get current RecentGlobalXmin value, as a FullTransactionId.
+ */
+FullTransactionId
+GetFullRecentGlobalXmin(void)
+{
+	FullTransactionId nextxid_full;
+	uint32		nextxid_epoch;
+	TransactionId nextxid_xid;
+	uint32		epoch;
+
+	Assert(TransactionIdIsNormal(RecentGlobalXmin));
+
+	/*
+	 * Compute the epoch from the next XID's epoch. This relies on the fact
+	 * that RecentGlobalXmin must be within the 2 billion XID horizon from the
+	 * next XID.
+	 */
+	nextxid_full = ReadNextFullTransactionId();
+	nextxid_epoch = EpochFromFullTransactionId(nextxid_full);
+	nextxid_xid = XidFromFullTransactionId(nextxid_full);
+
+	if (RecentGlobalXmin > nextxid_xid)
+		epoch = nextxid_epoch - 1;
+	else
+		epoch = nextxid_epoch;
+
+	return FullTransactionIdFromEpochAndXid(epoch, RecentGlobalXmin);
+}
+
+/*
  * SnapshotResetXmin
  *
  * If there are no more snapshots, we can reset our PGXACT->xmin to InvalidXid.