serializable_share_lock.patch

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 8975191..bf6c77e 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -83,6 +83,7 @@ static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf,
 				bool all_visible_cleared, bool new_all_visible_cleared);
 static bool HeapSatisfiesHOTUpdate(Relation relation, Bitmapset *hot_attrs,
 					   HeapTuple oldtup, HeapTuple newtup);
+static bool HeapSatisfiedLockCheckSnapshot(HeapTupleHeader tuple, Snapshot snapshot);
 
 
 /* ----------------------------------------------------------------
@@ -2033,8 +2034,9 @@ simple_heap_insert(Relation relation, HeapTuple tup)
  *	update_xmax - output parameter, used only for failure case (see below)
  *	cid - delete command ID (used for visibility test, and stored into
  *		cmax if successful)
- *	crosscheck - if not InvalidSnapshot, also check tuple against this
  *	wait - true if should wait for any conflicting update to commit/abort
+ *	lockcheck_snapshot - if not NULL, report the tuple as updated if it
+ *		was locked by a transaction not visible under this snapshot
  *
  * Normal, successful return value is HeapTupleMayBeUpdated, which
  * actually means we did delete it.  Failure return codes are
@@ -2049,7 +2051,7 @@ simple_heap_insert(Relation relation, HeapTuple tup)
 HTSU_Result
 heap_delete(Relation relation, ItemPointer tid,
 			ItemPointer ctid, TransactionId *update_xmax,
-			CommandId cid, Snapshot crosscheck, bool wait)
+			CommandId cid, bool wait, Snapshot lockcheck_snapshot)
 {
 	HTSU_Result result;
 	TransactionId xid = GetCurrentTransactionId();
@@ -2171,11 +2173,10 @@ l1:
 			result = HeapTupleUpdated;
 	}
 
-	if (crosscheck != InvalidSnapshot && result == HeapTupleMayBeUpdated)
+	if ((result == HeapTupleMayBeUpdated) &&
+		!HeapSatisfiedLockCheckSnapshot(tp.t_data, lockcheck_snapshot))
 	{
-		/* Perform additional check for serializable RI updates */
-		if (!HeapTupleSatisfiesVisibility(&tp, crosscheck, buffer))
-			result = HeapTupleUpdated;
+		result = HeapTupleUpdated;
 	}
 
 	if (result != HeapTupleMayBeUpdated)
@@ -2183,7 +2184,8 @@ l1:
 		Assert(result == HeapTupleSelfUpdated ||
 			   result == HeapTupleUpdated ||
 			   result == HeapTupleBeingUpdated);
-		Assert(!(tp.t_data->t_infomask & HEAP_XMAX_INVALID));
+		Assert(!(tp.t_data->t_infomask & HEAP_XMAX_INVALID) ||
+			   (tp.t_data->t_infomask & HEAP_IS_LOCKED));
 		*ctid = tp.t_data->t_ctid;
 		*update_xmax = HeapTupleHeaderGetXmax(tp.t_data);
 		UnlockReleaseBuffer(buffer);
@@ -2313,8 +2315,9 @@ simple_heap_delete(Relation relation, ItemPointer tid)
 
 	result = heap_delete(relation, tid,
 						 &update_ctid, &update_xmax,
-						 GetCurrentCommandId(true), InvalidSnapshot,
-						 true /* wait for commit */ );
+						 GetCurrentCommandId(true),
+						 true /* wait for commit */ ,
+						 InvalidSnapshot);
 	switch (result)
 	{
 		case HeapTupleSelfUpdated:
@@ -2349,7 +2352,8 @@ simple_heap_delete(Relation relation, ItemPointer tid)
  *	update_xmax - output parameter, used only for failure case (see below)
  *	cid - update command ID (used for visibility test, and stored into
  *		cmax/cmin if successful)
- *	crosscheck - if not InvalidSnapshot, also check old tuple against this
+ *	lockcheck_snapshot - if not NULL, report the tuple as updated if it
+ *		was locked by a transaction not visible under this snapshot
  *	wait - true if should wait for any conflicting update to commit/abort
  *
  * Normal, successful return value is HeapTupleMayBeUpdated, which
@@ -2371,7 +2375,7 @@ simple_heap_delete(Relation relation, ItemPointer tid)
 HTSU_Result
 heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 			ItemPointer ctid, TransactionId *update_xmax,
-			CommandId cid, Snapshot crosscheck, bool wait)
+			CommandId cid, bool wait, Snapshot lockcheck_snapshot)
 {
 	HTSU_Result result;
 	TransactionId xid = GetCurrentTransactionId();
@@ -2523,19 +2527,20 @@ l2:
 			result = HeapTupleUpdated;
 	}
 
-	if (crosscheck != InvalidSnapshot && result == HeapTupleMayBeUpdated)
+	if ((result == HeapTupleMayBeUpdated) &&
+		!HeapSatisfiedLockCheckSnapshot(oldtup.t_data, lockcheck_snapshot))
 	{
-		/* Perform additional check for serializable RI updates */
-		if (!HeapTupleSatisfiesVisibility(&oldtup, crosscheck, buffer))
-			result = HeapTupleUpdated;
+		result = HeapTupleUpdated;
 	}
 
+
 	if (result != HeapTupleMayBeUpdated)
 	{
 		Assert(result == HeapTupleSelfUpdated ||
 			   result == HeapTupleUpdated ||
 			   result == HeapTupleBeingUpdated);
-		Assert(!(oldtup.t_data->t_infomask & HEAP_XMAX_INVALID));
+		Assert(!(oldtup.t_data->t_infomask & HEAP_XMAX_INVALID) ||
+			   (oldtup.t_data->t_infomask & HEAP_IS_LOCKED));
 		*ctid = oldtup.t_data->t_ctid;
 		*update_xmax = HeapTupleHeaderGetXmax(oldtup.t_data);
 		UnlockReleaseBuffer(buffer);
@@ -2961,6 +2966,56 @@ HeapSatisfiesHOTUpdate(Relation relation, Bitmapset *hot_attrs,
 	return true;
 }
 
+
+/*
+ * Check if the tuple was locked by a transaction invisible to
+ * lockcheck_snapshot. Returns false if so.
+ */
+static bool
+HeapSatisfiedLockCheckSnapshot(HeapTupleHeader tuple, Snapshot lockcheck_snapshot)
+{
+	if (tuple->t_infomask & HEAP_IS_LOCKED)
+	{
+		/* If the tuple was locked, we now check whether the locking
+		 * transaction(s) are visible under lockcheck_snapshot. If
+		 * they aren't, we pretent the tuple was updated.
+		 */
+
+		if (tuple->t_infomask & HEAP_XMAX_IS_MULTI)
+		{
+			/* XXXFGP: Do the same check as below, but for multi xacts. The
+			 *         hard part is to guarantee that the multixact will still
+			 *         be there when we need to examine it. Currently, no
+			 *         mechanism seems to exist that'd guarantee that all
+			 *         multi xacts containing members invisible to
+			 *         lockcheck_snapshot outlive lockcheck_snapshot. Especially
+			 *         if lockcheck_snapshot is a serializable snapshot, which
+			 *         unfortunately is it's designated use-case.
+			 */
+			return true;
+		}
+		else
+		{
+			if ((lockcheck_snapshot != InvalidSnapshot) &&
+			    XidInMVCCSnapshot(HeapTupleHeaderGetXmax(tuple), lockcheck_snapshot))
+			{
+				/* Locking transaction is visible to lockcheck_snapshot */
+				return false;
+			}
+			else
+			{
+				/* Locking transaction is visible to lockcheck_snapshot */
+				return true;
+			}
+		}
+	}
+	else
+	{
+		/* Tuple wasn't locked */
+		return true;
+	}
+}
+
 /*
  *	simple_heap_update - replace a tuple
  *
@@ -2978,8 +3033,8 @@ simple_heap_update(Relation relation, ItemPointer otid, HeapTuple tup)
 
 	result = heap_update(relation, otid, tup,
 						 &update_ctid, &update_xmax,
-						 GetCurrentCommandId(true), InvalidSnapshot,
-						 true /* wait for commit */ );
+						 GetCurrentCommandId(true),
+						 true /* wait for commit */, InvalidSnapshot);
 	switch (result)
 	{
 		case HeapTupleSelfUpdated:
@@ -3013,6 +3068,9 @@ simple_heap_update(Relation relation, ItemPointer otid, HeapTuple tup)
  *		tuple's cmax if lock is successful)
  *	mode: indicates if shared or exclusive tuple lock is desired
  *	nowait: if true, ereport rather than blocking if lock not available
+ *	lockcheck_snapshot: if not NULL, report the tuple as updated if it
+ *						was locked by a transaction not visible under
+ *						this snapshot
  *
  * Output parameters:
  *	*tuple: all fields filled in
@@ -3066,7 +3124,8 @@ simple_heap_update(Relation relation, ItemPointer otid, HeapTuple tup)
 HTSU_Result
 heap_lock_tuple(Relation relation, HeapTuple tuple, Buffer *buffer,
 				ItemPointer ctid, TransactionId *update_xmax,
-				CommandId cid, LockTupleMode mode, bool nowait)
+				CommandId cid, LockTupleMode mode, bool nowait,
+				Snapshot lockcheck_snapshot)
 {
 	HTSU_Result result;
 	ItemPointer tid = &(tuple->t_self);
@@ -3247,10 +3306,17 @@ l3:
 			result = HeapTupleUpdated;
 	}
 
+	if ((result == HeapTupleMayBeUpdated) &&
+		!HeapSatisfiedLockCheckSnapshot(tuple->t_data, lockcheck_snapshot))
+	{
+		result = HeapTupleUpdated;
+	}
+
 	if (result != HeapTupleMayBeUpdated)
 	{
 		Assert(result == HeapTupleSelfUpdated || result == HeapTupleUpdated);
-		Assert(!(tuple->t_data->t_infomask & HEAP_XMAX_INVALID));
+		Assert(!(tuple->t_data->t_infomask & HEAP_XMAX_INVALID) ||
+			   (tuple->t_data->t_infomask & HEAP_IS_LOCKED));
 		*ctid = tuple->t_data->t_ctid;
 		*update_xmax = HeapTupleHeaderGetXmax(tuple->t_data);
 		LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index 2cbc192..120ea60 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -2347,7 +2347,9 @@ ltrmark:;
 		test = heap_lock_tuple(relation, &tuple, &buffer,
 							   &update_ctid, &update_xmax,
 							   estate->es_output_cid,
-							   LockTupleExclusive, false);
+							   LockTupleExclusive, false,
+							   IsXactIsoLevelSerializable ? estate->es_snapshot :
+															InvalidSnapshot);
 		switch (test)
 		{
 			case HeapTupleSelfUpdated:
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index d299310..fd5ed3c 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1518,10 +1518,13 @@ EvalPlanQualFetch(EState *estate, Relation relation, int lockmode,
 			/*
 			 * This is a live tuple, so now try to lock it.
 			 */
+			Assert(!IsXactIsoLevelSerializable || (estate->es_snapshot != InvalidSnapshot));
 			test = heap_lock_tuple(relation, &tuple, &buffer,
 								   &update_ctid, &update_xmax,
 								   estate->es_output_cid,
-								   lockmode, false);
+								   lockmode, false,
+								   IsXactIsoLevelSerializable ? estate->es_snapshot :
+																InvalidSnapshot);
 			/* We now have two pins on the buffer, get rid of one */
 			ReleaseBuffer(buffer);
 
diff --git a/src/backend/executor/nodeLockRows.c b/src/backend/executor/nodeLockRows.c
index 00f0562..f8baa13 100644
--- a/src/backend/executor/nodeLockRows.c
+++ b/src/backend/executor/nodeLockRows.c
@@ -117,7 +117,9 @@ lnext:
 		test = heap_lock_tuple(erm->relation, &tuple, &buffer,
 							   &update_ctid, &update_xmax,
 							   estate->es_output_cid,
-							   lockmode, erm->noWait);
+							   lockmode, erm->noWait,
+							   IsXactIsoLevelSerializable ? estate->es_snapshot :
+															InvalidSnapshot);
 		ReleaseBuffer(buffer);
 		switch (test)
 		{
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 7856b66..ecc75f9 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -316,8 +316,9 @@ ldelete:;
 	result = heap_delete(resultRelationDesc, tupleid,
 						 &update_ctid, &update_xmax,
 						 estate->es_output_cid,
-						 estate->es_crosscheck_snapshot,
-						 true /* wait for commit */ );
+						 true, /* wait for commit */
+						 IsXactIsoLevelSerializable ? estate->es_snapshot :
+													  InvalidSnapshot);
 	switch (result)
 	{
 		case HeapTupleSelfUpdated:
@@ -504,8 +505,9 @@ lreplace:;
 	result = heap_update(resultRelationDesc, tupleid, tuple,
 						 &update_ctid, &update_xmax,
 						 estate->es_output_cid,
-						 estate->es_crosscheck_snapshot,
-						 true /* wait for commit */ );
+						 true, /* wait for commit */ 
+						 IsXactIsoLevelSerializable ? estate->es_snapshot :
+													  InvalidSnapshot);
 	switch (result)
 	{
 		case HeapTupleSelfUpdated:
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c
index 4f3630c..302a571 100644
--- a/src/backend/utils/time/tqual.c
+++ b/src/backend/utils/time/tqual.c
@@ -72,9 +72,6 @@ SnapshotData SnapshotSelfData = {HeapTupleSatisfiesSelf};
 SnapshotData SnapshotAnyData = {HeapTupleSatisfiesAny};
 SnapshotData SnapshotToastData = {HeapTupleSatisfiesToast};
 
-/* local functions */
-static bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
-
 
 /*
  * SetHintBits()
@@ -1253,7 +1250,7 @@ HeapTupleSatisfiesVacuum(HeapTupleHeader tuple, TransactionId OldestXmin,
  * by this function.  This is OK for current uses, because we actually only
  * apply this for known-committed XIDs.
  */
-static bool
+bool
 XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 {
 	uint32		i;
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index f963146..a17d62c 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -98,15 +98,16 @@ extern Oid heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 			int options, BulkInsertState bistate);
 extern HTSU_Result heap_delete(Relation relation, ItemPointer tid,
 			ItemPointer ctid, TransactionId *update_xmax,
-			CommandId cid, Snapshot crosscheck, bool wait);
+			CommandId cid, bool wait, Snapshot crosscheck);
 extern HTSU_Result heap_update(Relation relation, ItemPointer otid,
 			HeapTuple newtup,
 			ItemPointer ctid, TransactionId *update_xmax,
-			CommandId cid, Snapshot crosscheck, bool wait);
+			CommandId cid, bool wait, Snapshot crosscheck);
 extern HTSU_Result heap_lock_tuple(Relation relation, HeapTuple tuple,
 				Buffer *buffer, ItemPointer ctid,
 				TransactionId *update_xmax, CommandId cid,
-				LockTupleMode mode, bool nowait);
+				LockTupleMode mode, bool nowait,
+				Snapshot lockcheck_snapshot);
 extern void heap_inplace_update(Relation relation, HeapTuple tuple);
 extern bool heap_freeze_tuple(HeapTupleHeader tuple, TransactionId cutoff_xid,
 				  Buffer buf);
diff --git a/src/include/utils/tqual.h b/src/include/utils/tqual.h
index e85e820..fa386e7 100644
--- a/src/include/utils/tqual.h
+++ b/src/include/utils/tqual.h
@@ -41,6 +41,8 @@ extern PGDLLIMPORT SnapshotData SnapshotToastData;
 #define IsMVCCSnapshot(snapshot)  \
 	((snapshot)->satisfies == HeapTupleSatisfiesMVCC)
 
+bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
+
 /*
  * HeapTupleSatisfiesVisibility
  *		True iff heap tuple satisfies a time qual.