Vacuum fix for ERROR updated tuple is already HEAP_MOVED_OFF.

`repair_frag()` should consult distributed snapshot
(`localXidSatisfiesAnyDistributedSnapshot()`) while following and moving chains
of updated tuples. Vacuum consults distributed snapshot
(`localXidSatisfiesAnyDistributedSnapshot()`) to find which tuples can be
deleted and not. For RECENTLY_DEAD tuples it used to make decision just based on
comparison with OldestXmin which is not sufficient and even there distributed
snapshot must be checked.

Fixes #4298

(cherry picked from commit 313ab24f)

src/backend/cdb/cdbdistributedsnapshot.c

@@ -52,6 +52,15 @@ localXidSatisfiesAnyDistributedSnapshot(TransactionId localXid)
 	DistributedSnapshotCommitted distributedSnapshotCommitted;
 	Assert(TransactionIdIsNormal(localXid));
 
+	/*
+	 * In general expect this function to be called only for normal xid, as
+	 * more performant for caller to avoid the call based on
+	 * TransactionIdIsNormal() check but just in case was called can safely
+	 * return false.
+	 */
+	if (!TransactionIdIsNormal(localXid))
+		return false;
+
 	/*
 	 * For single user mode operation like initdb time, let the vacuum
 	 * cleanout and freeze tuples.

src/backend/commands/vacuum.c

@@ -3732,8 +3732,10 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 			 * separately movable chain, ignoring any intervening DEAD ones.
 			 */
 			if (((tuple.t_data->t_infomask & HEAP_UPDATED) &&
-				 !TransactionIdPrecedes(HeapTupleHeaderGetXmin(tuple.t_data),
-										OldestXmin)) ||
+				 (!TransactionIdPrecedes(HeapTupleHeaderGetXmin(tuple.t_data),
+										 OldestXmin) ||
+				  (!(tuple.t_data->t_infomask2 & HEAP_XMIN_DISTRIBUTED_SNAPSHOT_IGNORE) &&
+				   localXidSatisfiesAnyDistributedSnapshot(HeapTupleHeaderGetXmin(tuple.t_data))))) ||
 				(!(tuple.t_data->t_infomask & (HEAP_XMAX_INVALID |
 											   HEAP_IS_LOCKED)) &&
 				 !(ItemPointerEquals(&(tuple.t_self),
@@ -3921,8 +3923,10 @@ repair_frag(VRelStats *vacrelstats, Relation onerel,
 
 					/* Done if at beginning of chain */
 					if (!(tp.t_data->t_infomask & HEAP_UPDATED) ||
-					 TransactionIdPrecedes(HeapTupleHeaderGetXmin(tp.t_data),
-										   OldestXmin))
+						(TransactionIdPrecedes(HeapTupleHeaderGetXmin(tp.t_data),
+											   OldestXmin) &&
+						 ((tuple.t_data->t_infomask2 & HEAP_XMIN_DISTRIBUTED_SNAPSHOT_IGNORE) ||
+							 !localXidSatisfiesAnyDistributedSnapshot(HeapTupleHeaderGetXmin(tp.t_data)))))
 						break;	/* out of check-all-items loop */
 
 					/* Move to tuple with prior row version */

src/test/isolation2/expected/vacuum_full_recently_dead_tuple_due_to_distributed_snapshot.out

+-- Test to validate bug fix for vacuum full and distributed snapshot.
+--
+-- Scenarios is where locally on segment tuple's xmin and xmax is lower than
+-- OldestXmin and hence safely can be declared DEAD but based on distributed
+-- snapshot HeapTupleSatisfiesVacuum() returns HEAPTUPLE_RECENTLY_DEAD. Later
+-- though while moving the tuples around as part of vacuum, distributed snapshot
+-- was not consulted but instead xmin was only checked against OldestXmin
+-- raising the "ERROR: updated tuple is already HEAP_MOVED_OFF".
+create table test_recently_dead_utility(a int, b int, c text);
+CREATE
+-- Insert enough data that it doesn't all fit in one page.
+insert into test_recently_dead_utility select 1, g, 'foobar' from generate_series(1, 1000) g;
+INSERT 1000
+-- Perform updates to form update chains and deleted tuples.
+update test_recently_dead_utility set b = 1;
+UPDATE 1000
+update test_recently_dead_utility set b = 1;
+UPDATE 1000
+
+-- Run VACUUM FULL in utility mode. It sees some of the old, updated, tuple
+-- versions as HEAPTUPLE_RECENTLY_DEAD, even though they are safely dead because
+-- localXidSatisfiesAnyDistributedSnapshot() is conservative and returns 'true'
+-- in utility mode. Helps to validate doesn't surprise the chain-following logic
+-- in VACUUM FULL.
+0U: vacuum full verbose test_recently_dead_utility;
+VACUUM
+0U: select count(*) from test_recently_dead_utility;
+count
+-----
+1000 
+(1 row)
+0U: set gp_select_invisible=1;
+SET
+-- print to make sure deleted tuples were not cleaned due to distributed
+-- snapshot to make test is future proof, if logic in
+-- localXidSatisfiesAnyDistributedSnapshot() changes for utility mode.
+0U: select count(*) from test_recently_dead_utility;
+count
+-----
+2000 
+(1 row)
+0U: set gp_select_invisible=0;
+SET

src/test/isolation2/isolation2_schedule

@@ -11,6 +11,7 @@ test: drop_rename
 test: instr_in_shmem_setup
 test: instr_in_shmem_terminate
 test: instr_in_shmem_cleanup
+test: vacuum_full_recently_dead_tuple_due_to_distributed_snapshot
 
 test: setup
 # Tests on Append-Optimized tables (row-oriented).

src/test/isolation2/sql/vacuum_full_recently_dead_tuple_due_to_distributed_snapshot.sql

+-- Test to validate bug fix for vacuum full and distributed snapshot.
+--
+-- Scenarios is where locally on segment tuple's xmin and xmax is lower than
+-- OldestXmin and hence safely can be declared DEAD but based on distributed
+-- snapshot HeapTupleSatisfiesVacuum() returns HEAPTUPLE_RECENTLY_DEAD. Later
+-- though while moving the tuples around as part of vacuum, distributed snapshot
+-- was not consulted but instead xmin was only checked against OldestXmin
+-- raising the "ERROR: updated tuple is already HEAP_MOVED_OFF".
+create table test_recently_dead_utility(a int, b int, c text);
+-- Insert enough data that it doesn't all fit in one page.
+insert into test_recently_dead_utility select 1, g, 'foobar' from generate_series(1, 1000) g;
+-- Perform updates to form update chains and deleted tuples.
+update test_recently_dead_utility set b = 1;
+update test_recently_dead_utility set b = 1;
+
+-- Run VACUUM FULL in utility mode. It sees some of the old, updated, tuple
+-- versions as HEAPTUPLE_RECENTLY_DEAD, even though they are safely dead because
+-- localXidSatisfiesAnyDistributedSnapshot() is conservative and returns 'true'
+-- in utility mode. Helps to validate doesn't surprise the chain-following logic
+-- in VACUUM FULL.
+0U: vacuum full verbose test_recently_dead_utility;
+0U: select count(*) from test_recently_dead_utility;
+0U: set gp_select_invisible=1;
+-- print to make sure deleted tuples were not cleaned due to distributed
+-- snapshot to make test is future proof, if logic in
+-- localXidSatisfiesAnyDistributedSnapshot() changes for utility mode.
+0U: select count(*) from test_recently_dead_utility;
+0U: set gp_select_invisible=0;