From 89a9779ff68c77f150f701caffc750fae7410781 Mon Sep 17 00:00:00 2001
From: Peter Geoghegan <pg@bowt.ie>
Date: Sun, 12 Jun 2022 15:46:08 -0700
Subject: [PATCH v7 1/6] Add page-level freezing to VACUUM.

Teach VACUUM to decide on whether or not to trigger freezing at the
level of whole heap pages, not individual tuple fields.  OldestXmin is
now treated as the cutoff for freezing eligibility in all cases, while
FreezeLimit is used to trigger freezing at the level of each page (we
now freeze all eligible XIDs on a page when freezing is triggered for
the page).

This approach decouples the question of _how_ VACUUM could/will freeze a
given heap page (which of its XIDs are eligible to be frozen) from the
question of whether it actually makes sense to do so right now.

Just adding page-level freezing does not change all that much on its
own: VACUUM will still typically freeze very lazily, since we're only
forcing freezing of all of a page's eligible tuples when we decide to
freeze at least one (on the basis of XID age and FreezeLimit).  For now
VACUUM still freezes everything almost as lazily as it always has.
Later work will teach VACUUM to apply an alternative eager freezing
strategy that triggers page-level freezing earlier, based on additional
criteria.
---
 src/include/access/heapam.h          |  49 +++++-
 src/backend/access/heap/heapam.c     | 234 +++++++++++++++------------
 src/backend/access/heap/vacuumlazy.c |  95 +++++++----
 3 files changed, 233 insertions(+), 145 deletions(-)

diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index 810baaf9d..ddc81c0d1 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -112,6 +112,40 @@ typedef struct HeapTupleFreeze
 	OffsetNumber offset;
 } HeapTupleFreeze;
 
+/*
+ * State used by VACUUM to track the details of freezing all eligible tuples
+ * on a given heap page.
+ *
+ * VACUUM prepares freeze plans for each page via heap_prepare_freeze_tuple
+ * calls (every tuple with storage gets its own call).  This page-level freeze
+ * state is updated across each call, which ultimately determines whether or
+ * not freezing the page is required. (VACUUM freezes the page via a call to
+ * heap_freeze_execute_prepared, which freezes using prepared freeze plans.)
+ *
+ * Aside from the basic question of whether or not freezing will go ahead, the
+ * state also tracks the oldest extant XID/MXID in the table as a whole, for
+ * the purposes of advancing relfrozenxid/relminmxid values in pg_class later
+ * on.  Each heap_prepare_freeze_tuple call pushes NewRelfrozenXid and/or
+ * NewRelminMxid back as required to avoid unsafe final pg_class values.  Any
+ * and all unfrozen XIDs or MXIDs that remain after VACUUM finishes _must_
+ * have values >= the final relfrozenxid/relminmxid values in pg_class.  This
+ * includes XIDs that remain as MultiXact members from any tuple's xmax.
+ */
+typedef struct HeapPageFreeze
+{
+	/* Is heap_prepare_freeze_tuple caller required to freeze page? */
+	bool		FreezeRequired;
+
+	/* Values used when heap_freeze_execute_prepared is called for page */
+	TransactionId NewRelfrozenXid;
+	MultiXactId NewRelminMxid;
+
+	/* Used by callers that choose to not freeze the page */
+	TransactionId NoFreezeNewRelfrozenXid;
+	MultiXactId NoFreezeNewRelminMxid;
+
+} HeapPageFreeze;
+
 /* ----------------
  *		function prototypes for heap access method
  *
@@ -180,19 +214,20 @@ extern void heap_inplace_update(Relation relation, HeapTuple tuple);
 extern bool heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 									  TransactionId relfrozenxid, TransactionId relminmxid,
 									  TransactionId cutoff_xid, TransactionId cutoff_multi,
+									  TransactionId limit_xid, MultiXactId limit_multi,
 									  HeapTupleFreeze *frz, bool *totally_frozen,
-									  TransactionId *relfrozenxid_out,
-									  MultiXactId *relminmxid_out);
+									  HeapPageFreeze *xtrack);
 extern void heap_freeze_execute_prepared(Relation rel, Buffer buffer,
-										 TransactionId FreezeLimit,
+										 TransactionId OldestXmin,
 										 HeapTupleFreeze *tuples, int ntuples);
 extern bool heap_freeze_tuple(HeapTupleHeader tuple,
 							  TransactionId relfrozenxid, TransactionId relminmxid,
 							  TransactionId cutoff_xid, TransactionId cutoff_multi);
-extern bool heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
-									MultiXactId cutoff_multi,
-									TransactionId *relfrozenxid_out,
-									MultiXactId *relminmxid_out);
+extern bool heap_tuple_would_freeze(HeapTupleHeader tuple,
+									TransactionId MustFreezeLimit,
+									MultiXactId MustFreezeMultiLimit,
+									TransactionId *NewRelfrozenXid,
+									MultiXactId *NewRelminMxid);
 extern bool heap_tuple_needs_eventual_freeze(HeapTupleHeader tuple);
 
 extern void simple_heap_insert(Relation relation, HeapTuple tup);
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index d18c5ca6f..10de1adc3 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -6371,7 +6371,7 @@ FreezeMultiXactId(MultiXactId multi, uint16 t_infomask,
 				 * Running locker cannot possibly be older than the cutoff.
 				 *
 				 * The cutoff is <= VACUUM's OldestXmin, which is also the
-				 * initial value used for top-level relfrozenxid_out tracking
+				 * initial value used for top-level NewRelfrozenXid tracking
 				 * state.  A running locker cannot be older than VACUUM's
 				 * OldestXmin, either, so we don't need a temp_xid_out step.
 				 */
@@ -6444,26 +6444,14 @@ FreezeMultiXactId(MultiXactId multi, uint16 t_infomask,
  *
  * VACUUM caller must assemble HeapTupleFreeze entries for every tuple that we
  * returned true for when called.  A later heap_freeze_execute_prepared call
- * will execute freezing for caller's page as a whole.
+ * will execute freezing for caller's page as a whole.  Caller must initialize
+ * xtrack fields for page as a whole before calling here with first tuple for
+ * the page.
  *
  * It is assumed that the caller has checked the tuple with
  * HeapTupleSatisfiesVacuum() and determined that it is not HEAPTUPLE_DEAD
  * (else we should be removing the tuple, not freezing it).
  *
- * The *relfrozenxid_out and *relminmxid_out arguments are the current target
- * relfrozenxid and relminmxid for VACUUM caller's heap rel.  Any and all
- * unfrozen XIDs or MXIDs that remain in caller's rel after VACUUM finishes
- * _must_ have values >= the final relfrozenxid/relminmxid values in pg_class.
- * This includes XIDs that remain as MultiXact members from any tuple's xmax.
- * Each call here pushes back *relfrozenxid_out and/or *relminmxid_out as
- * needed to avoid unsafe final values in rel's authoritative pg_class tuple.
- *
- * NB: cutoff_xid *must* be <= VACUUM's OldestXmin, to ensure that any
- * XID older than it could neither be running nor seen as running by any
- * open transaction.  This ensures that the replacement will not change
- * anyone's idea of the tuple state.
- * Similarly, cutoff_multi must be <= VACUUM's OldestMxact.
- *
  * NB: This function has side effects: it might allocate a new MultiXactId.
  * It will be set as tuple's new xmax when our *frz output is processed within
  * heap_execute_freeze_tuple later on.  If the tuple is in a shared buffer
@@ -6473,14 +6461,15 @@ bool
 heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 						  TransactionId relfrozenxid, TransactionId relminmxid,
 						  TransactionId cutoff_xid, TransactionId cutoff_multi,
+						  TransactionId limit_xid, MultiXactId limit_multi,
 						  HeapTupleFreeze *frz, bool *totally_frozen,
-						  TransactionId *relfrozenxid_out,
-						  MultiXactId *relminmxid_out)
+						  HeapPageFreeze *xtrack)
 {
 	bool		changed = false;
-	bool		xmax_already_frozen = false;
-	bool		xmin_frozen;
-	bool		freeze_xmax;
+	bool		xmin_already_frozen = false,
+				xmax_already_frozen = false;
+	bool		freeze_xmin,
+				freeze_xmax;
 	TransactionId xid;
 
 	frz->frzflags = 0;
@@ -6489,18 +6478,17 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 	frz->xmax = HeapTupleHeaderGetRawXmax(tuple);
 
 	/*
-	 * Process xmin.  xmin_frozen has two slightly different meanings: in the
-	 * !XidIsNormal case, it means "the xmin doesn't need any freezing" (it's
-	 * already a permanent value), while in the block below it is set true to
-	 * mean "xmin won't need freezing after what we do to it here" (false
-	 * otherwise).  In both cases we're allowed to set totally_frozen, as far
-	 * as xmin is concerned.  Both cases also don't require relfrozenxid_out
-	 * handling, since either way the tuple's xmin will be a permanent value
-	 * once we're done with it.
+	 * Process xmin, while keeping track of whether it's already frozen, or
+	 * will become frozen iff our freeze plan is executed by caller (could be
+	 * neither).
 	 */
 	xid = HeapTupleHeaderGetXmin(tuple);
 	if (!TransactionIdIsNormal(xid))
-		xmin_frozen = true;
+	{
+		freeze_xmin = false;
+		xmin_already_frozen = true;
+		/* No need for NewRelfrozenXid handling for already-frozen xmin */
+	}
 	else
 	{
 		if (TransactionIdPrecedes(xid, relfrozenxid))
@@ -6509,8 +6497,8 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 					 errmsg_internal("found xmin %u from before relfrozenxid %u",
 									 xid, relfrozenxid)));
 
-		xmin_frozen = TransactionIdPrecedes(xid, cutoff_xid);
-		if (xmin_frozen)
+		freeze_xmin = TransactionIdPrecedes(xid, cutoff_xid);
+		if (freeze_xmin)
 		{
 			if (!TransactionIdDidCommit(xid))
 				ereport(ERROR,
@@ -6523,9 +6511,9 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 		}
 		else
 		{
-			/* xmin to remain unfrozen.  Could push back relfrozenxid_out. */
-			if (TransactionIdPrecedes(xid, *relfrozenxid_out))
-				*relfrozenxid_out = xid;
+			/* xmin to remain unfrozen.  Could push back NewRelfrozenXid. */
+			if (TransactionIdPrecedes(xid, xtrack->NewRelfrozenXid))
+				xtrack->NewRelfrozenXid = xid;
 		}
 	}
 
@@ -6536,7 +6524,8 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 	 * freezing, too.  Also, if a multi needs freezing, we cannot simply take
 	 * it out --- if there's a live updater Xid, it needs to be kept.
 	 *
-	 * Make sure to keep heap_tuple_would_freeze in sync with this.
+	 * Make sure to keep heap_tuple_would_freeze in sync with this.  It needs
+	 * to return true for any tuple that we would force to be frozen here.
 	 */
 	xid = HeapTupleHeaderGetRawXmax(tuple);
 
@@ -6544,7 +6533,7 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 	{
 		TransactionId newxmax;
 		uint16		flags;
-		TransactionId mxid_oldest_xid_out = *relfrozenxid_out;
+		TransactionId mxid_oldest_xid_out = xtrack->NewRelfrozenXid;
 
 		newxmax = FreezeMultiXactId(xid, tuple->t_infomask,
 									relfrozenxid, relminmxid,
@@ -6558,13 +6547,13 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 			/*
 			 * xmax will become an updater Xid (original MultiXact's updater
 			 * member Xid will be carried forward as a simple Xid in Xmax).
-			 * Might have to ratchet back relfrozenxid_out here, though never
-			 * relminmxid_out.
+			 * Might have to ratchet back NewRelfrozenXid here, though never
+			 * NewRelminMxid.
 			 */
 			Assert(!freeze_xmax);
 			Assert(TransactionIdIsValid(newxmax));
-			if (TransactionIdPrecedes(newxmax, *relfrozenxid_out))
-				*relfrozenxid_out = newxmax;
+			if (TransactionIdPrecedes(newxmax, xtrack->NewRelfrozenXid))
+				xtrack->NewRelfrozenXid = newxmax;
 
 			/*
 			 * NB -- some of these transformations are only valid because we
@@ -6587,15 +6576,15 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 			/*
 			 * xmax is an old MultiXactId that we have to replace with a new
 			 * MultiXactId, to carry forward two or more original member XIDs.
-			 * Might have to ratchet back relfrozenxid_out here, though never
-			 * relminmxid_out.
+			 * Might have to ratchet back NewRelfrozenXid here, though never
+			 * NewRelminMxid.
 			 */
 			Assert(!freeze_xmax);
 			Assert(MultiXactIdIsValid(newxmax));
-			Assert(!MultiXactIdPrecedes(newxmax, *relminmxid_out));
+			Assert(!MultiXactIdPrecedes(newxmax, xtrack->NewRelminMxid));
 			Assert(TransactionIdPrecedesOrEquals(mxid_oldest_xid_out,
-												 *relfrozenxid_out));
-			*relfrozenxid_out = mxid_oldest_xid_out;
+												 xtrack->NewRelfrozenXid));
+			xtrack->NewRelfrozenXid = mxid_oldest_xid_out;
 
 			/*
 			 * We can't use GetMultiXactIdHintBits directly on the new multi
@@ -6617,26 +6606,38 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 		{
 			/*
 			 * xmax is a MultiXactId, and nothing about it changes for now.
-			 * Might have to ratchet back relminmxid_out, relfrozenxid_out, or
+			 * Might have to ratchet back NewRelminMxid, NewRelfrozenXid, or
 			 * both together.
 			 */
 			Assert(!freeze_xmax);
 			Assert(MultiXactIdIsValid(newxmax) && xid == newxmax);
 			Assert(TransactionIdPrecedesOrEquals(mxid_oldest_xid_out,
-												 *relfrozenxid_out));
-			if (MultiXactIdPrecedes(xid, *relminmxid_out))
-				*relminmxid_out = xid;
-			*relfrozenxid_out = mxid_oldest_xid_out;
+												 xtrack->NewRelfrozenXid));
+			if (MultiXactIdPrecedes(xid, xtrack->NewRelminMxid))
+				xtrack->NewRelminMxid = xid;
+			xtrack->NewRelfrozenXid = mxid_oldest_xid_out;
 		}
 		else
 		{
 			/*
 			 * Keeping nothing (neither an Xid nor a MultiXactId) in xmax.
-			 * Won't have to ratchet back relminmxid_out or relfrozenxid_out.
+			 * Won't have to ratchet back NewRelminMxid or NewRelfrozenXid.
+			 *
+			 * Note: heap_tuple_would_freeze() might not insist that this xmax
+			 * be frozen now, but we always freeze Multis proactively.
 			 */
 			Assert(freeze_xmax);
 			Assert(!TransactionIdIsValid(newxmax));
 		}
+
+		/*
+		 * Trigger page level freezing to ensure that we reliably process
+		 * MultiXacts as instructed by FreezeMultiXactId() in all cases.
+		 * There is no way to opt out of this, since FreezeMultiXactId()
+		 * doesn't provide for that.
+		 */
+		if ((flags & FRM_NOOP) == 0)
+			xtrack->FreezeRequired = true;
 	}
 	else if (TransactionIdIsNormal(xid))
 	{
@@ -6661,13 +6662,13 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 						 errmsg_internal("cannot freeze committed xmax %u",
 										 xid)));
 			freeze_xmax = true;
-			/* No need for relfrozenxid_out handling, since we'll freeze xmax */
+			/* No need for NewRelfrozenXid handling, since we'll freeze xmax */
 		}
 		else
 		{
 			freeze_xmax = false;
-			if (TransactionIdPrecedes(xid, *relfrozenxid_out))
-				*relfrozenxid_out = xid;
+			if (TransactionIdPrecedes(xid, xtrack->NewRelfrozenXid))
+				xtrack->NewRelfrozenXid = xid;
 		}
 	}
 	else if ((tuple->t_infomask & HEAP_XMAX_INVALID) ||
@@ -6675,7 +6676,7 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 	{
 		freeze_xmax = false;
 		xmax_already_frozen = true;
-		/* No need for relfrozenxid_out handling for already-frozen xmax */
+		/* No need for NewRelfrozenXid handling for already-frozen xmax */
 	}
 	else
 		ereport(ERROR,
@@ -6683,6 +6684,11 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 				 errmsg_internal("found xmax %u (infomask 0x%04x) not frozen, not multi, not normal",
 								 xid, tuple->t_infomask)));
 
+	if (freeze_xmin)
+	{
+		Assert(!xmin_already_frozen);
+		Assert(changed);
+	}
 	if (freeze_xmax)
 	{
 		Assert(!xmax_already_frozen);
@@ -6709,18 +6715,11 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 	{
 		xid = HeapTupleHeaderGetXvac(tuple);
 
-		/*
-		 * For Xvac, we ignore the cutoff_xid and just always perform the
-		 * freeze operation.  The oldest release in which such a value can
-		 * actually be set is PostgreSQL 8.4, because old-style VACUUM FULL
-		 * was removed in PostgreSQL 9.0.  Note that if we were to respect
-		 * cutoff_xid here, we'd need to make surely to clear totally_frozen
-		 * when we skipped freezing on that basis.
-		 *
-		 * No need for relfrozenxid_out handling, since we always freeze xvac.
-		 */
+		/* For Xvac, we always freeze proactively */
 		if (TransactionIdIsNormal(xid))
 		{
+			Assert(TransactionIdPrecedes(xid, cutoff_xid));
+
 			/*
 			 * If a MOVED_OFF tuple is not dead, the xvac transaction must
 			 * have failed; whereas a non-dead MOVED_IN tuple must mean the
@@ -6731,18 +6730,39 @@ heap_prepare_freeze_tuple(HeapTupleHeader tuple,
 			else
 				frz->frzflags |= XLH_FREEZE_XVAC;
 
-			/*
-			 * Might as well fix the hint bits too; usually XMIN_COMMITTED
-			 * will already be set here, but there's a small chance not.
-			 */
+			/* Set XMIN_COMMITTED defensively */
 			Assert(!(tuple->t_infomask & HEAP_XMIN_INVALID));
 			frz->t_infomask |= HEAP_XMIN_COMMITTED;
+
+			/*
+			 * Force freezing any page with an xvac to keep things simple.
+			 * This allows totally_frozen tracking to ignore xvac.
+			 */
 			changed = true;
+			xtrack->FreezeRequired = true;
 		}
 	}
 
-	*totally_frozen = (xmin_frozen &&
+	/*
+	 * Determine if this tuple is already totally frozen, or will become
+	 * totally frozen (provided caller executes freeze plan for the page)
+	 */
+	*totally_frozen = ((freeze_xmin || xmin_already_frozen) &&
 					   (freeze_xmax || xmax_already_frozen));
+
+	/*
+	 * Force vacuumlazy.c to freeze page when avoiding it would violate the
+	 * rule that XIDs < limit_xid (and MXIDs < limit_multi) must never remain
+	 */
+	if (!xtrack->FreezeRequired &&
+		!(xmin_already_frozen && xmax_already_frozen))
+	{
+		xtrack->FreezeRequired =
+			heap_tuple_would_freeze(tuple, limit_xid, limit_multi,
+									&xtrack->NoFreezeNewRelfrozenXid,
+									&xtrack->NoFreezeNewRelminMxid);
+	}
+
 	return changed;
 }
 
@@ -6786,13 +6806,13 @@ heap_execute_freeze_tuple(HeapTupleHeader tuple, HeapTupleFreeze *frz)
  */
 void
 heap_freeze_execute_prepared(Relation rel, Buffer buffer,
-							 TransactionId FreezeLimit,
+							 TransactionId OldestXmin,
 							 HeapTupleFreeze *tuples, int ntuples)
 {
 	Page		page = BufferGetPage(buffer);
 
 	Assert(ntuples > 0);
-	Assert(TransactionIdIsNormal(FreezeLimit));
+	Assert(TransactionIdIsNormal(OldestXmin));
 
 	START_CRIT_SECTION();
 
@@ -6821,11 +6841,10 @@ heap_freeze_execute_prepared(Relation rel, Buffer buffer,
 		nplans = heap_xlog_freeze_plan(tuples, ntuples, plans, offsets);
 
 		/*
-		 * FreezeLimit is (approximately) the first XID not frozen by VACUUM.
-		 * Back up caller's FreezeLimit to avoid false conflicts when
-		 * FreezeLimit is precisely equal to VACUUM's OldestXmin cutoff.
+		 * OldestXmin is the first XID not frozen by VACUUM.  Back up caller's
+		 * OldestXmin to avoid false conflicts.
 		 */
-		snapshotConflictHorizon = FreezeLimit;
+		snapshotConflictHorizon = OldestXmin;
 		TransactionIdRetreat(snapshotConflictHorizon);
 
 		xlrec.snapshotConflictHorizon = snapshotConflictHorizon;
@@ -6867,14 +6886,20 @@ heap_freeze_tuple(HeapTupleHeader tuple,
 	HeapTupleFreeze frz;
 	bool		do_freeze;
 	bool		tuple_totally_frozen;
-	TransactionId relfrozenxid_out = cutoff_xid;
-	MultiXactId relminmxid_out = cutoff_multi;
+	HeapPageFreeze dummy;
+
+	dummy.FreezeRequired = true;
+	dummy.NewRelfrozenXid = cutoff_xid;
+	dummy.NewRelminMxid = cutoff_multi;
+	dummy.NoFreezeNewRelfrozenXid = cutoff_xid;
+	dummy.NoFreezeNewRelminMxid = cutoff_multi;
 
 	do_freeze = heap_prepare_freeze_tuple(tuple,
 										  relfrozenxid, relminmxid,
 										  cutoff_xid, cutoff_multi,
+										  cutoff_xid, cutoff_multi,
 										  &frz, &tuple_totally_frozen,
-										  &relfrozenxid_out, &relminmxid_out);
+										  &dummy);
 
 	/*
 	 * Note that because this is not a WAL-logged operation, we don't need to
@@ -7304,15 +7329,16 @@ heap_tuple_needs_eventual_freeze(HeapTupleHeader tuple)
  * We must also deal with dead tuples here, since (xmin, xmax, xvac) fields
  * could be processed by pruning away the whole tuple instead of freezing.
  *
- * The *relfrozenxid_out and *relminmxid_out input/output arguments work just
- * like the heap_prepare_freeze_tuple arguments that they're based on.  We
- * never freeze here, which makes tracking the oldest extant XID/MXID simple.
+ * The *NewRelfrozenXid and *NewRelminMxid input/output arguments work just
+ * like the similar fields from the FreezeCutoffs struct.  We never freeze
+ * here, which makes tracking the oldest extant XID/MXID simple.
  */
 bool
-heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
-						MultiXactId cutoff_multi,
-						TransactionId *relfrozenxid_out,
-						MultiXactId *relminmxid_out)
+heap_tuple_would_freeze(HeapTupleHeader tuple,
+						TransactionId MustFreezeLimit,
+						MultiXactId MustFreezeMultiLimit,
+						TransactionId *NewRelfrozenXid,
+						MultiXactId *NewRelminMxid)
 {
 	TransactionId xid;
 	MultiXactId multi;
@@ -7322,9 +7348,9 @@ heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
 	xid = HeapTupleHeaderGetXmin(tuple);
 	if (TransactionIdIsNormal(xid))
 	{
-		if (TransactionIdPrecedes(xid, *relfrozenxid_out))
-			*relfrozenxid_out = xid;
-		if (TransactionIdPrecedes(xid, cutoff_xid))
+		if (TransactionIdPrecedes(xid, *NewRelfrozenXid))
+			*NewRelfrozenXid = xid;
+		if (TransactionIdPrecedes(xid, MustFreezeLimit))
 			would_freeze = true;
 	}
 
@@ -7339,9 +7365,9 @@ heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
 	if (TransactionIdIsNormal(xid))
 	{
 		/* xmax is a non-permanent XID */
-		if (TransactionIdPrecedes(xid, *relfrozenxid_out))
-			*relfrozenxid_out = xid;
-		if (TransactionIdPrecedes(xid, cutoff_xid))
+		if (TransactionIdPrecedes(xid, *NewRelfrozenXid))
+			*NewRelfrozenXid = xid;
+		if (TransactionIdPrecedes(xid, MustFreezeLimit))
 			would_freeze = true;
 	}
 	else if (!MultiXactIdIsValid(multi))
@@ -7351,8 +7377,8 @@ heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
 	else if (HEAP_LOCKED_UPGRADED(tuple->t_infomask))
 	{
 		/* xmax is a pg_upgrade'd MultiXact, which can't have updater XID */
-		if (MultiXactIdPrecedes(multi, *relminmxid_out))
-			*relminmxid_out = multi;
+		if (MultiXactIdPrecedes(multi, *NewRelminMxid))
+			*NewRelminMxid = multi;
 		/* heap_prepare_freeze_tuple always freezes pg_upgrade'd xmax */
 		would_freeze = true;
 	}
@@ -7362,9 +7388,9 @@ heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
 		MultiXactMember *members;
 		int			nmembers;
 
-		if (MultiXactIdPrecedes(multi, *relminmxid_out))
-			*relminmxid_out = multi;
-		if (MultiXactIdPrecedes(multi, cutoff_multi))
+		if (MultiXactIdPrecedes(multi, *NewRelminMxid))
+			*NewRelminMxid = multi;
+		if (MultiXactIdPrecedes(multi, MustFreezeMultiLimit))
 			would_freeze = true;
 
 		/* need to check whether any member of the mxact is old */
@@ -7375,9 +7401,9 @@ heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
 		{
 			xid = members[i].xid;
 			Assert(TransactionIdIsNormal(xid));
-			if (TransactionIdPrecedes(xid, *relfrozenxid_out))
-				*relfrozenxid_out = xid;
-			if (TransactionIdPrecedes(xid, cutoff_xid))
+			if (TransactionIdPrecedes(xid, *NewRelfrozenXid))
+				*NewRelfrozenXid = xid;
+			if (TransactionIdPrecedes(xid, MustFreezeLimit))
 				would_freeze = true;
 		}
 		if (nmembers > 0)
@@ -7389,9 +7415,9 @@ heap_tuple_would_freeze(HeapTupleHeader tuple, TransactionId cutoff_xid,
 		xid = HeapTupleHeaderGetXvac(tuple);
 		if (TransactionIdIsNormal(xid))
 		{
-			if (TransactionIdPrecedes(xid, *relfrozenxid_out))
-				*relfrozenxid_out = xid;
-			/* heap_prepare_freeze_tuple always freezes xvac */
+			if (TransactionIdPrecedes(xid, *NewRelfrozenXid))
+				*NewRelfrozenXid = xid;
+			/* heap_prepare_freeze_tuple forces xvac freezing */
 			would_freeze = true;
 		}
 	}
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 834ab83a0..9c84f8397 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -169,8 +169,9 @@ typedef struct LVRelState
 
 	/* VACUUM operation's cutoffs for freezing and pruning */
 	TransactionId OldestXmin;
+	MultiXactId OldestMxact;
 	GlobalVisState *vistest;
-	/* VACUUM operation's target cutoffs for freezing XIDs and MultiXactIds */
+	/* Limits on the age of the oldest unfrozen XID and MXID */
 	TransactionId FreezeLimit;
 	MultiXactId MultiXactCutoff;
 	/* Tracks oldest extant XID/MXID for setting relfrozenxid/relminmxid */
@@ -511,6 +512,7 @@ heap_vacuum_rel(Relation rel, VacuumParams *params,
 	 */
 	vacrel->rel_pages = orig_rel_pages = RelationGetNumberOfBlocks(rel);
 	vacrel->OldestXmin = OldestXmin;
+	vacrel->OldestMxact = OldestMxact;
 	vacrel->vistest = GlobalVisTestFor(rel);
 	/* FreezeLimit controls XID freezing (always <= OldestXmin) */
 	vacrel->FreezeLimit = FreezeLimit;
@@ -1563,8 +1565,8 @@ lazy_scan_prune(LVRelState *vacrel,
 				live_tuples,
 				recently_dead_tuples;
 	int			nnewlpdead;
-	TransactionId NewRelfrozenXid;
-	MultiXactId NewRelminMxid;
+	HeapPageFreeze xtrack;
+	bool		freeze_all_eligible PG_USED_FOR_ASSERTS_ONLY;
 	OffsetNumber deadoffsets[MaxHeapTuplesPerPage];
 	HeapTupleFreeze frozen[MaxHeapTuplesPerPage];
 
@@ -1580,8 +1582,11 @@ lazy_scan_prune(LVRelState *vacrel,
 retry:
 
 	/* Initialize (or reset) page-level state */
-	NewRelfrozenXid = vacrel->NewRelfrozenXid;
-	NewRelminMxid = vacrel->NewRelminMxid;
+	xtrack.FreezeRequired = false;
+	xtrack.NewRelfrozenXid = vacrel->NewRelfrozenXid;
+	xtrack.NewRelminMxid = vacrel->NewRelminMxid;
+	xtrack.NoFreezeNewRelfrozenXid = vacrel->NewRelfrozenXid;
+	xtrack.NoFreezeNewRelminMxid = vacrel->NewRelminMxid;
 	tuples_deleted = 0;
 	tuples_frozen = 0;
 	lpdead_items = 0;
@@ -1634,27 +1639,23 @@ retry:
 			continue;
 		}
 
-		/*
-		 * LP_DEAD items are processed outside of the loop.
-		 *
-		 * Note that we deliberately don't set hastup=true in the case of an
-		 * LP_DEAD item here, which is not how count_nondeletable_pages() does
-		 * it -- it only considers pages empty/truncatable when they have no
-		 * items at all (except LP_UNUSED items).
-		 *
-		 * Our assumption is that any LP_DEAD items we encounter here will
-		 * become LP_UNUSED inside lazy_vacuum_heap_page() before we actually
-		 * call count_nondeletable_pages().  In any case our opinion of
-		 * whether or not a page 'hastup' (which is how our caller sets its
-		 * vacrel->nonempty_pages value) is inherently race-prone.  It must be
-		 * treated as advisory/unreliable, so we might as well be slightly
-		 * optimistic.
-		 */
 		if (ItemIdIsDead(itemid))
 		{
+			/*
+			 * Delay unsetting all_visible until after we have decided on
+			 * whether this page should be frozen.  We need to test "is this
+			 * page all_visible, assuming any LP_DEAD items are set LP_UNUSED
+			 * in final heap pass?" to reach a decision.  all_visible will be
+			 * unset before we return, as required by lazy_scan_heap caller.
+			 *
+			 * Deliberately don't set hastup for LP_DEAD items.  We make the
+			 * soft assumption that any LP_DEAD items encountered here will
+			 * become LP_UNUSED later on, before count_nondeletable_pages is
+			 * reached.  Whether the page 'hastup' is inherently race-prone.
+			 * It must be treated as unreliable by caller anyway, so we might
+			 * as well be slightly optimistic about it.
+			 */
 			deadoffsets[lpdead_items++] = offnum;
-			prunestate->all_visible = false;
-			prunestate->has_lpdead_items = true;
 			continue;
 		}
 
@@ -1782,11 +1783,13 @@ retry:
 		if (heap_prepare_freeze_tuple(tuple.t_data,
 									  vacrel->relfrozenxid,
 									  vacrel->relminmxid,
+									  vacrel->OldestXmin,
+									  vacrel->OldestMxact,
 									  vacrel->FreezeLimit,
 									  vacrel->MultiXactCutoff,
 									  &frozen[tuples_frozen],
 									  &tuple_totally_frozen,
-									  &NewRelfrozenXid, &NewRelminMxid))
+									  &xtrack))
 		{
 			/* Save prepared freeze plan for later */
 			frozen[tuples_frozen++].offset = offnum;
@@ -1807,9 +1810,33 @@ retry:
 	 * that will need to be vacuumed in indexes later, or a LP_NORMAL tuple
 	 * that remains and needs to be considered for freezing now (LP_UNUSED and
 	 * LP_REDIRECT items also remain, but are of no further interest to us).
+	 *
+	 * Freeze the page when heap_prepare_freeze_tuple indicates that at least
+	 * one XID/MXID from before FreezeLimit/MultiXactCutoff is present.
 	 */
-	vacrel->NewRelfrozenXid = NewRelfrozenXid;
-	vacrel->NewRelminMxid = NewRelminMxid;
+	if (xtrack.FreezeRequired || tuples_frozen == 0)
+	{
+		/*
+		 * We're freezing the page.  Our final NewRelfrozenXid doesn't need to
+		 * be affected by the XIDs that are just about to be frozen anyway.
+		 *
+		 * Note: although we're freezing all eligible tuples on this page, we
+		 * might not need to freeze anything (might be zero eligible tuples).
+		 */
+		vacrel->NewRelfrozenXid = xtrack.NewRelfrozenXid;
+		vacrel->NewRelminMxid = xtrack.NewRelminMxid;
+		freeze_all_eligible = true;
+	}
+	else
+	{
+		/* Not freezing this page, so use alternative cutoffs */
+		vacrel->NewRelfrozenXid = xtrack.NoFreezeNewRelfrozenXid;
+		vacrel->NewRelminMxid = xtrack.NoFreezeNewRelminMxid;
+
+		/* Might still set page all-visible, but never all-frozen */
+		tuples_frozen = 0;
+		freeze_all_eligible = prunestate->all_frozen = false;
+	}
 
 	/*
 	 * Consider the need to freeze any items with tuple storage from the page
@@ -1817,12 +1844,12 @@ retry:
 	 */
 	if (tuples_frozen > 0)
 	{
-		Assert(prunestate->hastup);
+		Assert(prunestate->hastup && freeze_all_eligible);
 
 		vacrel->frozen_pages++;
 
 		/* Execute all freeze plans for page as a single atomic action */
-		heap_freeze_execute_prepared(vacrel->rel, buf, vacrel->FreezeLimit,
+		heap_freeze_execute_prepared(vacrel->rel, buf, vacrel->OldestXmin,
 									 frozen, tuples_frozen);
 	}
 
@@ -1841,7 +1868,7 @@ retry:
 	 */
 #ifdef USE_ASSERT_CHECKING
 	/* Note that all_frozen value does not matter when !all_visible */
-	if (prunestate->all_visible)
+	if (prunestate->all_visible && lpdead_items == 0)
 	{
 		TransactionId cutoff;
 		bool		all_frozen;
@@ -1849,8 +1876,7 @@ retry:
 		if (!heap_page_is_all_visible(vacrel, buf, &cutoff, &all_frozen))
 			Assert(false);
 
-		Assert(lpdead_items == 0);
-		Assert(prunestate->all_frozen == all_frozen);
+		Assert(prunestate->all_frozen == all_frozen || !freeze_all_eligible);
 
 		/*
 		 * It's possible that we froze tuples and made the page's XID cutoff
@@ -1871,9 +1897,6 @@ retry:
 		VacDeadItems *dead_items = vacrel->dead_items;
 		ItemPointerData tmp;
 
-		Assert(!prunestate->all_visible);
-		Assert(prunestate->has_lpdead_items);
-
 		vacrel->lpdead_item_pages++;
 
 		ItemPointerSetBlockNumber(&tmp, blkno);
@@ -1887,6 +1910,10 @@ retry:
 		Assert(dead_items->num_items <= dead_items->max_items);
 		pgstat_progress_update_param(PROGRESS_VACUUM_NUM_DEAD_TUPLES,
 									 dead_items->num_items);
+
+		/* lazy_scan_heap caller expects LP_DEAD item to unset all_visible */
+		prunestate->all_visible = false;
+		prunestate->has_lpdead_items = true;
 	}
 
 	/* Finally, add page-local counts to whole-VACUUM counts */
-- 
2.34.1

