diff --git a/contrib/pg_visibility/pg_visibility.c b/contrib/pg_visibility/pg_visibility.c
index 2cc9575d9f..f6ee3b7734 100644
--- a/contrib/pg_visibility/pg_visibility.c
+++ b/contrib/pg_visibility/pg_visibility.c
@@ -12,6 +12,7 @@
 
 #include "access/htup_details.h"
 #include "access/visibilitymap.h"
+#include "catalog/catalog.h"
 #include "catalog/pg_type.h"
 #include "catalog/storage_xlog.h"
 #include "funcapi.h"
@@ -52,7 +53,7 @@ static corrupt_items *collect_corrupt_items(Oid relid, bool all_visible,
 					  bool all_frozen);
 static void record_corrupt_item(corrupt_items *items, ItemPointer tid);
 static bool tuple_all_visible(HeapTuple tup, TransactionId OldestXmin,
-				  Buffer buffer);
+				  Buffer buffer, bool is_catalog);
 static void check_relation_relkind(Relation rel);
 
 /*
@@ -553,6 +554,7 @@ collect_corrupt_items(Oid relid, bool all_visible, bool all_frozen)
 	Buffer		vmbuffer = InvalidBuffer;
 	BufferAccessStrategy bstrategy = GetAccessStrategy(BAS_BULKREAD);
 	TransactionId OldestXmin = InvalidTransactionId;
+	bool		is_catalog;
 
 	if (all_visible)
 	{
@@ -566,6 +568,7 @@ collect_corrupt_items(Oid relid, bool all_visible, bool all_frozen)
 	check_relation_relkind(rel);
 
 	nblocks = RelationGetNumberOfBlocks(rel);
+	is_catalog = RelationIsAccessibleInLogicalDecoding(rel);
 
 	/*
 	 * Guess an initial array size. We don't expect many corrupted tuples, so
@@ -656,7 +659,7 @@ collect_corrupt_items(Oid relid, bool all_visible, bool all_frozen)
 			 * the tuple to be all-visible.
 			 */
 			if (check_visible &&
-				!tuple_all_visible(&tuple, OldestXmin, buffer))
+				!tuple_all_visible(&tuple, OldestXmin, buffer, is_catalog))
 			{
 				TransactionId RecomputedOldestXmin;
 
@@ -681,7 +684,7 @@ collect_corrupt_items(Oid relid, bool all_visible, bool all_frozen)
 				else
 				{
 					OldestXmin = RecomputedOldestXmin;
-					if (!tuple_all_visible(&tuple, OldestXmin, buffer))
+					if (!tuple_all_visible(&tuple, OldestXmin, buffer, is_catalog))
 						record_corrupt_item(items, &tuple.t_self);
 				}
 			}
@@ -739,12 +742,13 @@ record_corrupt_item(corrupt_items *items, ItemPointer tid)
  * The buffer should contain the tuple and should be locked and pinned.
  */
 static bool
-tuple_all_visible(HeapTuple tup, TransactionId OldestXmin, Buffer buffer)
+tuple_all_visible(HeapTuple tup, TransactionId OldestXmin, Buffer buffer,
+				  bool is_catalog)
 {
 	HTSV_Result state;
 	TransactionId xmin;
 
-	state = HeapTupleSatisfiesVacuum(tup, OldestXmin, buffer);
+	state = HeapTupleSatisfiesVacuum(tup, OldestXmin, buffer, is_catalog);
 	if (state != HEAPTUPLE_LIVE)
 		return false;			/* all-visible implies live */
 
diff --git a/contrib/pgstattuple/pgstatapprox.c b/contrib/pgstattuple/pgstatapprox.c
index 5bf06138a5..e2aaf40ab6 100644
--- a/contrib/pgstattuple/pgstatapprox.c
+++ b/contrib/pgstattuple/pgstatapprox.c
@@ -69,6 +69,8 @@ statapprox_heap(Relation rel, output_type *stat)
 	BufferAccessStrategy bstrategy;
 	TransactionId OldestXmin;
 	uint64		misc_count = 0;
+	bool		is_catalog =
+		RelationIsAccessibleInLogicalDecoding(rel);
 
 	OldestXmin = GetOldestXmin(rel, PROCARRAY_FLAGS_VACUUM);
 	bstrategy = GetAccessStrategy(BAS_BULKREAD);
@@ -156,7 +158,8 @@ statapprox_heap(Relation rel, output_type *stat)
 			 * We count live and dead tuples, but we also need to add up
 			 * others in order to feed vac_estimate_reltuples.
 			 */
-			switch (HeapTupleSatisfiesVacuum(&tuple, OldestXmin, buf))
+			switch (HeapTupleSatisfiesVacuum(&tuple, OldestXmin, buf,
+											 is_catalog))
 			{
 				case HEAPTUPLE_RECENTLY_DEAD:
 					misc_count++;
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index 9f33e0ce07..7dffaf5ec5 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -366,6 +366,8 @@ heap_prune_chain(Relation relation, Buffer buffer, OffsetNumber rootoffnum,
 	int			nchain = 0,
 				i;
 	HeapTupleData tup;
+	bool		is_catalog =
+		RelationIsAccessibleInLogicalDecoding(relation);
 
 	tup.t_tableOid = RelationGetRelid(relation);
 
@@ -402,7 +404,7 @@ heap_prune_chain(Relation relation, Buffer buffer, OffsetNumber rootoffnum,
 			 * either here or while following a chain below.  Whichever path
 			 * gets there first will mark the tuple unused.
 			 */
-			if (HeapTupleSatisfiesVacuum(&tup, OldestXmin, buffer)
+			if (HeapTupleSatisfiesVacuum(&tup, OldestXmin, buffer, is_catalog)
 				== HEAPTUPLE_DEAD && !HeapTupleHeaderIsHotUpdated(htup))
 			{
 				heap_prune_record_unused(prstate, rootoffnum);
@@ -486,21 +488,28 @@ heap_prune_chain(Relation relation, Buffer buffer, OffsetNumber rootoffnum,
 		 */
 		tupdead = recent_dead = false;
 
-		switch (HeapTupleSatisfiesVacuum(&tup, OldestXmin, buffer))
+		switch (HeapTupleSatisfiesVacuum(&tup, OldestXmin, buffer, is_catalog))
 		{
 			case HEAPTUPLE_DEAD:
 				tupdead = true;
 				break;
 
 			case HEAPTUPLE_RECENTLY_DEAD:
-				recent_dead = true;
 
 				/*
 				 * This tuple may soon become DEAD.  Update the hint field so
 				 * that the page is reconsidered for pruning in future.
+				 *
+				 * Do this only if the xmax is valid though. HeapTupleSatisfiesVacuum
+				 * returns recently dead in some cases even for rows that were
+				 * inserted in aborted transactions
 				 */
-				heap_prune_record_prunable(prstate,
+				if (TransactionIdIsValid(HeapTupleHeaderGetUpdateXid(htup)))
+				{
+					recent_dead = true;
+					heap_prune_record_prunable(prstate,
 										   HeapTupleHeaderGetUpdateXid(htup));
+				}
 				break;
 
 			case HEAPTUPLE_DELETE_IN_PROGRESS:
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 0125c18bc1..d06fa5299e 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -2225,6 +2225,8 @@ IndexBuildHeapRangeScan(Relation heapRelation,
 	TransactionId OldestXmin;
 	BlockNumber root_blkno = InvalidBlockNumber;
 	OffsetNumber root_offsets[MaxHeapTuplesPerPage];
+	bool		is_catalog =
+		RelationIsAccessibleInLogicalDecoding(heapRelation);
 
 	/*
 	 * sanity checks
@@ -2361,7 +2363,7 @@ IndexBuildHeapRangeScan(Relation heapRelation,
 			LockBuffer(scan->rs_cbuf, BUFFER_LOCK_SHARE);
 
 			switch (HeapTupleSatisfiesVacuum(heapTuple, OldestXmin,
-											 scan->rs_cbuf))
+									scan->rs_cbuf, is_catalog))
 			{
 				case HEAPTUPLE_DEAD:
 					/* Definitely dead, we can ignore it */
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index f952b3c732..105d7f4fb0 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -1052,6 +1052,8 @@ acquire_sample_rows(Relation onerel, int elevel,
 	TransactionId OldestXmin;
 	BlockSamplerData bs;
 	ReservoirStateData rstate;
+	bool		is_catalog =
+		RelationIsAccessibleInLogicalDecoding(onerel);
 
 	Assert(targrows > 0);
 
@@ -1121,7 +1123,8 @@ acquire_sample_rows(Relation onerel, int elevel,
 
 			switch (HeapTupleSatisfiesVacuum(&targtuple,
 											 OldestXmin,
-											 targbuffer))
+											 targbuffer,
+											 is_catalog))
 			{
 				case HEAPTUPLE_LIVE:
 					sample_it = true;
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 48f1e6e2ad..0b387caf79 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -758,6 +758,7 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 				tups_recently_dead = 0;
 	int			elevel = verbose ? INFO : DEBUG2;
 	PGRUsage	ru0;
+	bool		is_catalog;
 
 	pg_rusage_init(&ru0);
 
@@ -771,6 +772,7 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 	else
 		OldIndex = NULL;
 
+	is_catalog = RelationIsAccessibleInLogicalDecoding(NewHeap);
 	/*
 	 * Their tuple descriptors should be exactly alike, but here we only need
 	 * assume that they have the same number of columns.
@@ -967,7 +969,7 @@ copy_heap_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 
 		LockBuffer(buf, BUFFER_LOCK_SHARE);
 
-		switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf))
+		switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf, is_catalog))
 		{
 			case HEAPTUPLE_DEAD:
 				/* Definitely dead */
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
index f95346acdb..38b956ebd9 100644
--- a/src/backend/commands/vacuumlazy.c
+++ b/src/backend/commands/vacuumlazy.c
@@ -489,6 +489,8 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 		PROGRESS_VACUUM_MAX_DEAD_TUPLES
 	};
 	int64		initprog_val[3];
+	bool		is_catalog =
+		RelationIsAccessibleInLogicalDecoding(onerel);
 
 	pg_rusage_init(&ru0);
 
@@ -988,7 +990,8 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
 
 			tupgone = false;
 
-			switch (HeapTupleSatisfiesVacuum(&tuple, OldestXmin, buf))
+			switch (HeapTupleSatisfiesVacuum(&tuple, OldestXmin,
+											 buf, is_catalog))
 			{
 				case HEAPTUPLE_DEAD:
 
@@ -2121,6 +2124,8 @@ heap_page_is_all_visible(Relation rel, Buffer buf,
 	OffsetNumber offnum,
 				maxoff;
 	bool		all_visible = true;
+	bool		is_catalog =
+		RelationIsAccessibleInLogicalDecoding(rel);
 
 	*visibility_cutoff_xid = InvalidTransactionId;
 	*all_frozen = true;
@@ -2162,7 +2167,8 @@ heap_page_is_all_visible(Relation rel, Buffer buf,
 		tuple.t_len = ItemIdGetLength(itemid);
 		tuple.t_tableOid = RelationGetRelid(rel);
 
-		switch (HeapTupleSatisfiesVacuum(&tuple, OldestXmin, buf))
+		switch (HeapTupleSatisfiesVacuum(&tuple, OldestXmin,
+										 buf, is_catalog))
 		{
 			case HEAPTUPLE_LIVE:
 				{
diff --git a/src/backend/storage/lmgr/predicate.c b/src/backend/storage/lmgr/predicate.c
index 251a359bff..82abf84e43 100644
--- a/src/backend/storage/lmgr/predicate.c
+++ b/src/backend/storage/lmgr/predicate.c
@@ -195,6 +195,7 @@
 #include "access/xlog.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "catalog/catalog.h"
 #include "storage/bufmgr.h"
 #include "storage/predicate.h"
 #include "storage/predicate_internals.h"
@@ -3951,6 +3952,8 @@ CheckForSerializableConflictOut(bool visible, Relation relation,
 	SERIALIZABLEXID *sxid;
 	SERIALIZABLEXACT *sxact;
 	HTSV_Result htsvResult;
+	bool		is_catalog =
+		RelationIsAccessibleInLogicalDecoding(relation);
 
 	if (!SerializationNeededForRead(relation, snapshot))
 		return;
@@ -3972,7 +3975,8 @@ CheckForSerializableConflictOut(bool visible, Relation relation,
 	 * tuple is visible to us, while HeapTupleSatisfiesVacuum checks what else
 	 * is going on with it.
 	 */
-	htsvResult = HeapTupleSatisfiesVacuum(tuple, TransactionXmin, buffer);
+	htsvResult = HeapTupleSatisfiesVacuum(tuple, TransactionXmin,
+										  buffer, is_catalog);
 	switch (htsvResult)
 	{
 		case HEAPTUPLE_LIVE:
diff --git a/src/backend/utils/adt/selfuncs.c b/src/backend/utils/adt/selfuncs.c
index ea95b8068d..f00396e42d 100644
--- a/src/backend/utils/adt/selfuncs.c
+++ b/src/backend/utils/adt/selfuncs.c
@@ -105,6 +105,7 @@
 #include "access/gin.h"
 #include "access/htup_details.h"
 #include "access/sysattr.h"
+#include "catalog/catalog.h"
 #include "catalog/index.h"
 #include "catalog/pg_am.h"
 #include "catalog/pg_collation.h"
@@ -5485,6 +5486,7 @@ get_actual_variable_range(PlannerInfo *root, VariableStatData *vardata,
 			Datum		values[INDEX_MAX_KEYS];
 			bool		isnull[INDEX_MAX_KEYS];
 			SnapshotData SnapshotNonVacuumable;
+			bool		is_catalog;
 
 			estate = CreateExecutorState();
 			econtext = GetPerTupleExprContext(estate);
@@ -5507,7 +5509,9 @@ get_actual_variable_range(PlannerInfo *root, VariableStatData *vardata,
 			slot = MakeSingleTupleTableSlot(RelationGetDescr(heapRel));
 			econtext->ecxt_scantuple = slot;
 			get_typlenbyval(vardata->atttype, &typLen, &typByVal);
-			InitNonVacuumableSnapshot(SnapshotNonVacuumable, RecentGlobalXmin);
+			is_catalog = RelationIsAccessibleInLogicalDecoding(heapRel);
+			InitNonVacuumableSnapshot(SnapshotNonVacuumable,
+									  RecentGlobalXmin, is_catalog);
 
 			/* set up an IS NOT NULL scan key so that we ignore nulls */
 			ScanKeyEntryInitialize(&scankeys[0],
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c
index 2b218e07e6..0c959b44b3 100644
--- a/src/backend/utils/time/tqual.c
+++ b/src/backend/utils/time/tqual.c
@@ -1159,10 +1159,15 @@ HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot,
  * deleted by XIDs >= OldestXmin are deemed "recently dead"; they might
  * still be visible to some open transaction, so we can't remove them,
  * even if we see that the deleting transaction has committed.
+ *
+ * In some cases, if the tuple was part of an aborted transaction,
+ * then if it belongs to a catalog table (either system or user),
+ * it could be visible to some transaction if the XID >= OldestXmin
+ * and is deemed "recently dead"
  */
 HTSV_Result
 HeapTupleSatisfiesVacuum(HeapTuple htup, TransactionId OldestXmin,
-						 Buffer buffer)
+						 Buffer buffer, bool is_catalog)
 {
 	HeapTupleHeader tuple = htup->t_data;
 
@@ -1252,6 +1257,19 @@ HeapTupleSatisfiesVacuum(HeapTuple htup, TransactionId OldestXmin,
 			 */
 			SetHintBits(tuple, buffer, HEAP_XMIN_INVALID,
 						InvalidTransactionId);
+
+			/*
+			 * If it's a catalog table, then perhaps it was recent enough
+			 * that some transactions (logical decoding, for example)
+			 * could still see the tuple.
+			 */
+			if (is_catalog && !TransactionIdPrecedes(
+						HeapTupleHeaderGetRawXmin(tuple), OldestXmin))
+			{
+				elog(DEBUG3, "returning recently DEAD");
+				return HEAPTUPLE_RECENTLY_DEAD;
+			}
+
 			return HEAPTUPLE_DEAD;
 		}
 
@@ -1398,7 +1416,8 @@ bool
 HeapTupleSatisfiesNonVacuumable(HeapTuple htup, Snapshot snapshot,
 								Buffer buffer)
 {
-	return HeapTupleSatisfiesVacuum(htup, snapshot->xmin, buffer)
+	return HeapTupleSatisfiesVacuum(htup, snapshot->xmin,
+									buffer, snapshot->is_catalog)
 		!= HEAPTUPLE_DEAD;
 }
 
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h
index bf519778df..7130d82ff8 100644
--- a/src/include/utils/snapshot.h
+++ b/src/include/utils/snapshot.h
@@ -94,6 +94,7 @@ typedef struct SnapshotData
 
 	bool		takenDuringRecovery;	/* recovery-shaped snapshot? */
 	bool		copied;			/* false if it's a static snapshot */
+	bool		is_catalog;		/* is it for a catalog table? */
 
 	CommandId	curcid;			/* in my xact, CID < curcid are visible */
 
diff --git a/src/include/utils/tqual.h b/src/include/utils/tqual.h
index 96eaf01ca0..bf6cf45e4e 100644
--- a/src/include/utils/tqual.h
+++ b/src/include/utils/tqual.h
@@ -75,7 +75,8 @@ extern bool HeapTupleSatisfiesHistoricMVCC(HeapTuple htup,
 extern HTSU_Result HeapTupleSatisfiesUpdate(HeapTuple htup,
 						 CommandId curcid, Buffer buffer);
 extern HTSV_Result HeapTupleSatisfiesVacuum(HeapTuple htup,
-						 TransactionId OldestXmin, Buffer buffer);
+						 TransactionId OldestXmin, Buffer buffer,
+						 bool is_catalog);
 extern bool HeapTupleIsSurelyDead(HeapTuple htup,
 					  TransactionId OldestXmin);
 extern bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
@@ -107,9 +108,10 @@ extern bool ResolveCminCmaxDuringDecoding(struct HTAB *tuplecid_data,
  * Similarly, some initialization is required for a NonVacuumable snapshot.
  * The caller must supply the xmin horizon to use (e.g., RecentGlobalXmin).
  */
-#define InitNonVacuumableSnapshot(snapshotdata, xmin_horizon)  \
+#define InitNonVacuumableSnapshot(snapshotdata, xmin_horizon, is_catalog)  \
 	((snapshotdata).satisfies = HeapTupleSatisfiesNonVacuumable, \
-	 (snapshotdata).xmin = (xmin_horizon))
+	 (snapshotdata).xmin = (xmin_horizon), \
+	 (snapshotdata).is_catalog = (is_catalog))
 
 /*
  * Similarly, some initialization is required for SnapshotToast.  We need