Reduce spurious Hot Standby conflicts from never-visible records.

Hot Standby conflicts only with tuples that were visible at
some point. So ignore tuples from aborted transactions or for
tuples updated/deleted during the inserting transaction when
generating the conflict transaction ids.

Following detailed analysis and test case by Noah Misch.
Original report covered btree delete records, correctly observed
by Heikki Linnakangas that this applies to other cases also.
Fix covers all sources of cleanup records via common code.

Author: simonat2ndQuadrant

src/backend/access/heap/heapam.c

@@ -3776,8 +3776,11 @@ heap_restrpos(HeapScanDesc scan)
 }
 
 /*
- * If 'tuple' contains any XID greater than latestRemovedXid, update
- * latestRemovedXid to the greatest one found.
+ * If 'tuple' contains any visible XID greater than latestRemovedXid,
+ * ratchet forwards latestRemovedXid to the greatest one found.
+ * This is used as the basis for generating Hot Standby conflicts, so
+ * if a tuple was never visible then removing it should not conflict
+ * with queries.
  */
 void
 HeapTupleHeaderAdvanceLatestRemovedXid(HeapTupleHeader tuple,
@@ -3793,13 +3796,27 @@ HeapTupleHeaderAdvanceLatestRemovedXid(HeapTupleHeader tuple,
 			*latestRemovedXid = xvac;
 	}
 
-	if (TransactionIdPrecedes(*latestRemovedXid, xmax))
-		*latestRemovedXid = xmax;
-
-	if (TransactionIdPrecedes(*latestRemovedXid, xmin))
-		*latestRemovedXid = xmin;
+	/*
+	 * Ignore tuples inserted by an aborted transaction or
+	 * if the tuple was updated/deleted by the inserting transaction.
+	 *
+	 * Look for a committed hint bit, or if no xmin bit is set, check clog.
+	 * This needs to work on both master and standby, where it is used
+	 * to assess btree delete records.
+	 */
+	if ((tuple->t_infomask & HEAP_XMIN_COMMITTED) ||
+		(!(tuple->t_infomask & HEAP_XMIN_COMMITTED) &&
+		 !(tuple->t_infomask & HEAP_XMIN_INVALID) &&
+		 TransactionIdDidCommit(xmin)))
+	{
+		if (TransactionIdFollows(xmax, xmin))
+		{
+			if (TransactionIdFollows(xmax, *latestRemovedXid))
+				*latestRemovedXid = xmax;
+		}
+	}
 
-	Assert(TransactionIdIsValid(*latestRemovedXid));
+	/* *latestRemovedXid may still be invalid at end */
 }
 
 /*

src/backend/access/heap/pruneheap.c

@@ -237,7 +237,6 @@ heap_page_prune(Relation relation, Buffer buffer, TransactionId OldestXmin,
 		{
 			XLogRecPtr	recptr;
 
-			Assert(TransactionIdIsValid(prstate.latestRemovedXid));
 			recptr = log_heap_clean(relation, buffer,
 									prstate.redirected, prstate.nredirected,
 									prstate.nowdead, prstate.ndead,

src/backend/access/nbtree/nbtxlog.c

@@ -580,7 +580,6 @@ btree_xlog_delete_get_latestRemovedXid(XLogRecord *record)
 	BlockNumber hblkno;
 	OffsetNumber hoffnum;
 	TransactionId latestRemovedXid = InvalidTransactionId;
-	TransactionId htupxid = InvalidTransactionId;
 	int			i;
 
 	/*
@@ -646,24 +645,16 @@ btree_xlog_delete_get_latestRemovedXid(XLogRecord *record)
 		}
 
 		/*
-		 * If the heap item has storage, then read the header. Some LP_DEAD
-		 * items may not be accessible, so we ignore them.
+		 * If the heap item has storage, then read the header and use that to
+		 * set latestRemovedXid.
+		 *
+		 * Some LP_DEAD items may not be accessible, so we ignore them.
 		 */
 		if (ItemIdHasStorage(hitemid))
 		{
 			htuphdr = (HeapTupleHeader) PageGetItem(hpage, hitemid);
 
-			/*
-			 * Get the heap tuple's xmin/xmax and ratchet up the
-			 * latestRemovedXid. No need to consider xvac values here.
-			 */
-			htupxid = HeapTupleHeaderGetXmin(htuphdr);
-			if (TransactionIdFollows(htupxid, latestRemovedXid))
-				latestRemovedXid = htupxid;
-
-			htupxid = HeapTupleHeaderGetXmax(htuphdr);
-			if (TransactionIdFollows(htupxid, latestRemovedXid))
-				latestRemovedXid = htupxid;
+			HeapTupleHeaderAdvanceLatestRemovedXid(htuphdr, &latestRemovedXid);
 		}
 		else if (ItemIdIsDead(hitemid))
 		{