FOR KEY LOCK foreign keys

*** a/src/backend/access/heap/heapam.c
--- b/src/backend/access/heap/heapam.c
***************
*** 2417,2422 **** heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
--- 2417,2423 ----
  	HTSU_Result result;
  	TransactionId xid = GetCurrentTransactionId();
  	Bitmapset  *hot_attrs;
+ 	Bitmapset  *keylck_attrs;
  	ItemId		lp;
  	HeapTupleData oldtup;
  	HeapTuple	heaptup;
***************
*** 2430,2435 **** heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
--- 2431,2437 ----
  	bool		have_tuple_lock = false;
  	bool		iscombo;
  	bool		use_hot_update = false;
+ 	bool		keylocked_update = false;
  	bool		all_visible_cleared = false;
  	bool		all_visible_cleared_new = false;
  
***************
*** 2447,2453 **** heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
  	 * Note that we get a copy here, so we need not worry about relcache flush
  	 * happening midway through.
  	 */
! 	hot_attrs = RelationGetIndexAttrBitmap(relation);
  
  	buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(otid));
  	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
--- 2449,2456 ----
  	 * Note that we get a copy here, so we need not worry about relcache flush
  	 * happening midway through.
  	 */
! 	hot_attrs = RelationGetIndexAttrBitmap(relation, false);
! 	keylck_attrs = RelationGetIndexAttrBitmap(relation, true);
  
  	buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(otid));
  	LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
***************
*** 2484,2567 **** l2:
  		xwait = HeapTupleHeaderGetXmax(oldtup.t_data);
  		infomask = oldtup.t_data->t_infomask;
  
- 		LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
- 
  		/*
! 		 * Acquire tuple lock to establish our priority for the tuple (see
! 		 * heap_lock_tuple).  LockTuple will release us when we are
! 		 * next-in-line for the tuple.
! 		 *
! 		 * If we are forced to "start over" below, we keep the tuple lock;
! 		 * this arranges that we stay at the head of the line while rechecking
! 		 * tuple state.
  		 */
! 		if (!have_tuple_lock)
  		{
! 			LockTuple(relation, &(oldtup.t_self), ExclusiveLock);
! 			have_tuple_lock = true;
  		}
  
! 		/*
! 		 * Sleep until concurrent transaction ends.  Note that we don't care
! 		 * if the locker has an exclusive or shared lock, because we need
! 		 * exclusive.
! 		 */
! 
! 		if (infomask & HEAP_XMAX_IS_MULTI)
  		{
! 			/* wait for multixact */
! 			MultiXactIdWait((MultiXactId) xwait);
! 			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
  
  			/*
! 			 * If xwait had just locked the tuple then some other xact could
! 			 * update this tuple before we get to this point.  Check for xmax
! 			 * change, and start over if so.
  			 */
! 			if (!(oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 				!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 									 xwait))
! 				goto l2;
  
  			/*
! 			 * You might think the multixact is necessarily done here, but not
! 			 * so: it could have surviving members, namely our own xact or
! 			 * other subxacts of this backend.	It is legal for us to update
! 			 * the tuple in either case, however (the latter case is
! 			 * essentially a situation of upgrading our former shared lock to
! 			 * exclusive).	We don't bother changing the on-disk hint bits
! 			 * since we are about to overwrite the xmax altogether.
  			 */
! 		}
! 		else
! 		{
! 			/* wait for regular transaction to end */
! 			XactLockTableWait(xwait);
! 			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
  
  			/*
! 			 * xwait is done, but if xwait had just locked the tuple then some
! 			 * other xact could update this tuple before we get to this point.
! 			 * Check for xmax change, and start over if so.
  			 */
! 			if ((oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 				!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 									 xwait))
! 				goto l2;
! 
! 			/* Otherwise check if it committed or aborted */
! 			UpdateXmaxHintBits(oldtup.t_data, buffer, xwait);
  		}
- 
- 		/*
- 		 * We may overwrite if previous xmax aborted, or if it committed but
- 		 * only locked the tuple without updating it.
- 		 */
- 		if (oldtup.t_data->t_infomask & (HEAP_XMAX_INVALID |
- 										 HEAP_IS_LOCKED))
- 			result = HeapTupleMayBeUpdated;
- 		else
- 			result = HeapTupleUpdated;
  	}
  
  	if (crosscheck != InvalidSnapshot && result == HeapTupleMayBeUpdated)
--- 2487,2587 ----
  		xwait = HeapTupleHeaderGetXmax(oldtup.t_data);
  		infomask = oldtup.t_data->t_infomask;
  
  		/*
! 		 * if it's only key-locked and we're not updating an indexed column,
! 		 * we can act though MayBeUpdated was returned, but the resulting tuple
! 		 * needs a bunch of fields copied from the original.
  		 */
! 		if ((infomask & HEAP_XMAX_KEY_LOCK) &&
! 			!(infomask & HEAP_XMAX_SHARED_LOCK) &&
! 			HeapSatisfiesHOTUpdate(relation, keylck_attrs,
! 								   &oldtup, newtup))
  		{
! 			result = HeapTupleMayBeUpdated;
! 			keylocked_update = true;
  		}
  
! 		if (!keylocked_update)
  		{
! 			LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
  
  			/*
! 			 * Acquire tuple lock to establish our priority for the tuple (see
! 			 * heap_lock_tuple).  LockTuple will release us when we are
! 			 * next-in-line for the tuple.
! 			 *
! 			 * If we are forced to "start over" below, we keep the tuple lock;
! 			 * this arranges that we stay at the head of the line while rechecking
! 			 * tuple state.
  			 */
! 			if (!have_tuple_lock)
! 			{
! 				LockTuple(relation, &(oldtup.t_self), ExclusiveLock);
! 				have_tuple_lock = true;
! 			}
  
  			/*
! 			 * Sleep until concurrent transaction ends.  Note that we don't care
! 			 * if the locker has an exclusive or shared lock, because we need
! 			 * exclusive.
  			 */
! 
! 			if (infomask & HEAP_XMAX_IS_MULTI)
! 			{
! 				/* wait for multixact */
! 				MultiXactIdWait((MultiXactId) xwait);
! 				LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
! 
! 				/*
! 				 * If xwait had just locked the tuple then some other xact could
! 				 * update this tuple before we get to this point.  Check for xmax
! 				 * change, and start over if so.
! 				 */
! 				if (!(oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 					!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 										 xwait))
! 					goto l2;
! 
! 				/*
! 				 * You might think the multixact is necessarily done here, but not
! 				 * so: it could have surviving members, namely our own xact or
! 				 * other subxacts of this backend.	It is legal for us to update
! 				 * the tuple in either case, however (the latter case is
! 				 * essentially a situation of upgrading our former shared lock to
! 				 * exclusive).	We don't bother changing the on-disk hint bits
! 				 * since we are about to overwrite the xmax altogether.
! 				 */
! 			}
! 			else
! 			{
! 				/* wait for regular transaction to end */
! 				XactLockTableWait(xwait);
! 				LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
! 
! 				/*
! 				 * xwait is done, but if xwait had just locked the tuple then some
! 				 * other xact could update this tuple before we get to this point.
! 				 * Check for xmax change, and start over if so.
! 				 */
! 				if ((oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 					!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 										 xwait))
! 					goto l2;
! 
! 				/* Otherwise check if it committed or aborted */
! 				UpdateXmaxHintBits(oldtup.t_data, buffer, xwait);
! 			}
  
  			/*
! 			 * We may overwrite if previous xmax aborted, or if it committed but
! 			 * only locked the tuple without updating it.
  			 */
! 			if (oldtup.t_data->t_infomask & (HEAP_XMAX_INVALID |
! 											 HEAP_IS_LOCKED))
! 				result = HeapTupleMayBeUpdated;
! 			else
! 				result = HeapTupleUpdated;
  		}
  	}
  
  	if (crosscheck != InvalidSnapshot && result == HeapTupleMayBeUpdated)
***************
*** 2609,2621 **** l2:
  
  	newtup->t_data->t_infomask &= ~(HEAP_XACT_MASK);
  	newtup->t_data->t_infomask2 &= ~(HEAP2_XACT_MASK);
! 	newtup->t_data->t_infomask |= (HEAP_XMAX_INVALID | HEAP_UPDATED);
  	HeapTupleHeaderSetXmin(newtup->t_data, xid);
- 	HeapTupleHeaderSetCmin(newtup->t_data, cid);
- 	HeapTupleHeaderSetXmax(newtup->t_data, 0);	/* for cleanliness */
  	newtup->t_tableOid = RelationGetRelid(relation);
  
  	/*
  	 * Replace cid with a combo cid if necessary.  Note that we already put
  	 * the plain cid into the new tuple.
  	 */
--- 2629,2671 ----
  
  	newtup->t_data->t_infomask &= ~(HEAP_XACT_MASK);
  	newtup->t_data->t_infomask2 &= ~(HEAP2_XACT_MASK);
! 	newtup->t_data->t_infomask |= HEAP_UPDATED;
  	HeapTupleHeaderSetXmin(newtup->t_data, xid);
  	newtup->t_tableOid = RelationGetRelid(relation);
  
  	/*
+ 	 * If this update is touching a tuple that was key-locked, we need to
+ 	 * carry forward some bits from the old tuple into the new copy.
+ 	 */
+ 	if (keylocked_update)
+ 	{
+ 		HeapTupleHeaderSetXmax(newtup->t_data,
+ 							   HeapTupleHeaderGetXmax(oldtup.t_data));
+ 		newtup->t_data->t_infomask |= (oldtup.t_data->t_infomask & 
+ 									   (HEAP_XMAX_IS_MULTI |
+ 										HEAP_XMAX_KEY_LOCK));
+ 		/*
+ 		 * we also need to copy the combo CID stuff, but only if the original
+ 		 * tuple was created by us; otherwise the combocid module complains
+ 		 * (Alternatively we could use HeapTupleHeaderGetRawCommandId)
+ 		 */
+ 		if (TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmin(oldtup.t_data)))
+ 		{
+ 			newtup->t_data->t_infomask |= (oldtup.t_data->t_infomask & 
+ 										   HEAP_COMBOCID);
+ 			HeapTupleHeaderSetCmin(newtup->t_data,
+ 								   HeapTupleHeaderGetCmin(oldtup.t_data));
+ 		}
+ 
+ 	}
+ 	else
+ 	{
+ 		newtup->t_data->t_infomask |= HEAP_XMAX_INVALID;
+ 		HeapTupleHeaderSetXmax(newtup->t_data, 0);	/* for cleanliness */
+ 		HeapTupleHeaderSetCmin(newtup->t_data, cid);
+ 	}
+ 
+ 	/*
  	 * Replace cid with a combo cid if necessary.  Note that we already put
  	 * the plain cid into the new tuple.
  	 */
***************
*** 3142,3148 **** heap_lock_tuple(Relation relation, HeapTuple tuple, Buffer *buffer,
  	LOCKMODE	tuple_lock_type;
  	bool		have_tuple_lock = false;
  
! 	tuple_lock_type = (mode == LockTupleShared) ? ShareLock : ExclusiveLock;
  
  	*buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(tid));
  	LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
--- 3192,3211 ----
  	LOCKMODE	tuple_lock_type;
  	bool		have_tuple_lock = false;
  
! 	/* in FOR KEY LOCK mode, we use a share lock temporarily */
! 	switch (mode)
! 	{
! 		case LockTupleShared:
! 		case LockTupleKeylock:
! 			tuple_lock_type = ShareLock;
! 			break;
! 		case LockTupleExclusive:
! 			tuple_lock_type = ExclusiveLock;
! 			break;
! 		default:
! 			elog(ERROR, "invalid tuple lock mode");
! 			tuple_lock_type = 0;	/* keep compiler quiet */
! 	}
  
  	*buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(tid));
  	LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
***************
*** 3175,3192 **** l3:
  		LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);
  
  		/*
! 		 * If we wish to acquire share lock, and the tuple is already
! 		 * share-locked by a multixact that includes any subtransaction of the
  		 * current top transaction, then we effectively hold the desired lock
  		 * already.  We *must* succeed without trying to take the tuple lock,
  		 * else we will deadlock against anyone waiting to acquire exclusive
  		 * lock.  We don't need to make any state changes in this case.
  		 */
! 		if (mode == LockTupleShared &&
  			(infomask & HEAP_XMAX_IS_MULTI) &&
  			MultiXactIdIsCurrent((MultiXactId) xwait))
  		{
! 			Assert(infomask & HEAP_XMAX_SHARED_LOCK);
  			/* Probably can't hold tuple lock here, but may as well check */
  			if (have_tuple_lock)
  				UnlockTuple(relation, tid, tuple_lock_type);
--- 3238,3255 ----
  		LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);
  
  		/*
! 		 * If we wish to acquire a key or share lock, and the tuple is already
! 		 * share- or key-locked by a multixact that includes any subtransaction of the
  		 * current top transaction, then we effectively hold the desired lock
  		 * already.  We *must* succeed without trying to take the tuple lock,
  		 * else we will deadlock against anyone waiting to acquire exclusive
  		 * lock.  We don't need to make any state changes in this case.
  		 */
! 		if ((mode == LockTupleShared || mode == LockTupleKeylock) &&
  			(infomask & HEAP_XMAX_IS_MULTI) &&
  			MultiXactIdIsCurrent((MultiXactId) xwait))
  		{
! 			Assert(infomask & HEAP_IS_SHARE_LOCKED);
  			/* Probably can't hold tuple lock here, but may as well check */
  			if (have_tuple_lock)
  				UnlockTuple(relation, tid, tuple_lock_type);
***************
*** 3217,3226 **** l3:
  			have_tuple_lock = true;
  		}
  
! 		if (mode == LockTupleShared && (infomask & HEAP_XMAX_SHARED_LOCK))
  		{
  			/*
! 			 * Acquiring sharelock when there's at least one sharelocker
  			 * already.  We need not wait for him/them to complete.
  			 */
  			LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
--- 3280,3290 ----
  			have_tuple_lock = true;
  		}
  
! 		if ((mode == LockTupleShared || mode == LockTupleKeylock) &&
! 			(infomask & HEAP_IS_SHARE_LOCKED))
  		{
  			/*
! 			 * Acquiring sharelock or keylock when there's at least one such locker
  			 * already.  We need not wait for him/them to complete.
  			 */
  			LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
***************
*** 3229,3235 **** l3:
  			 * Make sure it's still a shared lock, else start over.  (It's OK
  			 * if the ownership of the shared lock has changed, though.)
  			 */
! 			if (!(tuple->t_data->t_infomask & HEAP_XMAX_SHARED_LOCK))
  				goto l3;
  		}
  		else if (infomask & HEAP_XMAX_IS_MULTI)
--- 3293,3299 ----
  			 * Make sure it's still a shared lock, else start over.  (It's OK
  			 * if the ownership of the shared lock has changed, though.)
  			 */
! 			if (!(tuple->t_data->t_infomask & HEAP_IS_SHARE_LOCKED))
  				goto l3;
  		}
  		else if (infomask & HEAP_XMAX_IS_MULTI)
***************
*** 3339,3346 **** l3:
  	if (!(old_infomask & (HEAP_XMAX_INVALID |
  						  HEAP_XMAX_COMMITTED |
  						  HEAP_XMAX_IS_MULTI)) &&
! 		(mode == LockTupleShared ?
  		 (old_infomask & HEAP_IS_LOCKED) :
  		 (old_infomask & HEAP_XMAX_EXCL_LOCK)) &&
  		TransactionIdIsCurrentTransactionId(xmax))
  	{
--- 3403,3412 ----
  	if (!(old_infomask & (HEAP_XMAX_INVALID |
  						  HEAP_XMAX_COMMITTED |
  						  HEAP_XMAX_IS_MULTI)) &&
! 		(mode == LockTupleKeylock ?
  		 (old_infomask & HEAP_IS_LOCKED) :
+ 		 mode == LockTupleShared ?
+ 		 (old_infomask & (HEAP_XMAX_SHARED_LOCK | HEAP_XMAX_EXCL_LOCK)) :
  		 (old_infomask & HEAP_XMAX_EXCL_LOCK)) &&
  		TransactionIdIsCurrentTransactionId(xmax))
  	{
***************
*** 3364,3373 **** l3:
  									HEAP_IS_LOCKED |
  									HEAP_MOVED);
  
! 	if (mode == LockTupleShared)
  	{
  		/*
! 		 * If this is the first acquisition of a shared lock in the current
  		 * transaction, set my per-backend OldestMemberMXactId setting. We can
  		 * be certain that the transaction will never become a member of any
  		 * older MultiXactIds than that.  (We have to do this even if we end
--- 3430,3439 ----
  									HEAP_IS_LOCKED |
  									HEAP_MOVED);
  
! 	if (mode == LockTupleShared || mode == LockTupleKeylock)
  	{
  		/*
! 		 * If this is the first acquisition of a keylock or shared lock in the current
  		 * transaction, set my per-backend OldestMemberMXactId setting. We can
  		 * be certain that the transaction will never become a member of any
  		 * older MultiXactIds than that.  (We have to do this even if we end
***************
*** 3376,3382 **** l3:
  		 */
  		MultiXactIdSetOldestMember();
  
! 		new_infomask |= HEAP_XMAX_SHARED_LOCK;
  
  		/*
  		 * Check to see if we need a MultiXactId because there are multiple
--- 3442,3449 ----
  		 */
  		MultiXactIdSetOldestMember();
  
! 		new_infomask |= mode == LockTupleShared ? HEAP_XMAX_SHARED_LOCK :
! 			HEAP_XMAX_KEY_LOCK;
  
  		/*
  		 * Check to see if we need a MultiXactId because there are multiple
***************
*** 3476,3482 **** l3:
  		xlrec.target.tid = tuple->t_self;
  		xlrec.locking_xid = xid;
  		xlrec.xid_is_mxact = ((new_infomask & HEAP_XMAX_IS_MULTI) != 0);
! 		xlrec.shared_lock = (mode == LockTupleShared);
  		rdata[0].data = (char *) &xlrec;
  		rdata[0].len = SizeOfHeapLock;
  		rdata[0].buffer = InvalidBuffer;
--- 3543,3549 ----
  		xlrec.target.tid = tuple->t_self;
  		xlrec.locking_xid = xid;
  		xlrec.xid_is_mxact = ((new_infomask & HEAP_XMAX_IS_MULTI) != 0);
! 		xlrec.lock_strength = mode == LockTupleShared ? 's' : mode == LockTupleKeylock ? 'k' : 'x';
  		rdata[0].data = (char *) &xlrec;
  		rdata[0].len = SizeOfHeapLock;
  		rdata[0].buffer = InvalidBuffer;
***************
*** 4795,4802 **** heap_xlog_lock(XLogRecPtr lsn, XLogRecord *record)
  						  HEAP_MOVED);
  	if (xlrec->xid_is_mxact)
  		htup->t_infomask |= HEAP_XMAX_IS_MULTI;
! 	if (xlrec->shared_lock)
  		htup->t_infomask |= HEAP_XMAX_SHARED_LOCK;
  	else
  		htup->t_infomask |= HEAP_XMAX_EXCL_LOCK;
  	HeapTupleHeaderClearHotUpdated(htup);
--- 4862,4871 ----
  						  HEAP_MOVED);
  	if (xlrec->xid_is_mxact)
  		htup->t_infomask |= HEAP_XMAX_IS_MULTI;
! 	if (xlrec->lock_strength == 's')
  		htup->t_infomask |= HEAP_XMAX_SHARED_LOCK;
+ 	else if (xlrec->lock_strength == 'k')
+ 		htup->t_infomask |= HEAP_XMAX_KEY_LOCK;
  	else
  		htup->t_infomask |= HEAP_XMAX_EXCL_LOCK;
  	HeapTupleHeaderClearHotUpdated(htup);
***************
*** 4999,5006 **** heap_desc(StringInfo buf, uint8 xl_info, char *rec)
  	{
  		xl_heap_lock *xlrec = (xl_heap_lock *) rec;
  
! 		if (xlrec->shared_lock)
  			appendStringInfo(buf, "shared_lock: ");
  		else
  			appendStringInfo(buf, "exclusive_lock: ");
  		if (xlrec->xid_is_mxact)
--- 5068,5077 ----
  	{
  		xl_heap_lock *xlrec = (xl_heap_lock *) rec;
  
! 		if (xlrec->lock_strength == 's')
  			appendStringInfo(buf, "shared_lock: ");
+ 		else if (xlrec->lock_strength == 'k')
+ 			appendStringInfo(buf, "key_lock: ");
  		else
  			appendStringInfo(buf, "exclusive_lock: ");
  		if (xlrec->xid_is_mxact)
*** a/src/backend/catalog/index.c
--- b/src/backend/catalog/index.c
***************
*** 2863,2869 **** reindex_relation(Oid relid, bool toast_too, int flags)
  
  	/* Ensure rd_indexattr is valid; see comments for RelationSetIndexList */
  	if (is_pg_class)
! 		(void) RelationGetIndexAttrBitmap(rel);
  
  	PG_TRY();
  	{
--- 2863,2869 ----
  
  	/* Ensure rd_indexattr is valid; see comments for RelationSetIndexList */
  	if (is_pg_class)
! 		(void) RelationGetIndexAttrBitmap(rel, false);
  
  	PG_TRY();
  	{
*** a/src/backend/executor/execMain.c
--- b/src/backend/executor/execMain.c
***************
*** 701,707 **** InitPlan(QueryDesc *queryDesc, int eflags)
  	}
  
  	/*
! 	 * Similarly, we have to lock relations selected FOR UPDATE/FOR SHARE
  	 * before we initialize the plan tree, else we'd be risking lock upgrades.
  	 * While we are at it, build the ExecRowMark list.
  	 */
--- 701,707 ----
  	}
  
  	/*
! 	 * Similarly, we have to lock relations selected FOR UPDATE/FOR SHARE/KEY LOCK
  	 * before we initialize the plan tree, else we'd be risking lock upgrades.
  	 * While we are at it, build the ExecRowMark list.
  	 */
***************
*** 721,726 **** InitPlan(QueryDesc *queryDesc, int eflags)
--- 721,727 ----
  		{
  			case ROW_MARK_EXCLUSIVE:
  			case ROW_MARK_SHARE:
+ 			case ROW_MARK_KEYLOCK:
  				relid = getrelid(rc->rti, rangeTable);
  				relation = heap_open(relid, RowShareLock);
  				break;
*** a/src/backend/executor/nodeLockRows.c
--- b/src/backend/executor/nodeLockRows.c
***************
*** 112,119 **** lnext:
  		/* okay, try to lock the tuple */
  		if (erm->markType == ROW_MARK_EXCLUSIVE)
  			lockmode = LockTupleExclusive;
! 		else
  			lockmode = LockTupleShared;
  
  		test = heap_lock_tuple(erm->relation, &tuple, &buffer,
  							   &update_ctid, &update_xmax,
--- 112,126 ----
  		/* okay, try to lock the tuple */
  		if (erm->markType == ROW_MARK_EXCLUSIVE)
  			lockmode = LockTupleExclusive;
! 		else if (erm->markType == ROW_MARK_SHARE)
  			lockmode = LockTupleShared;
+ 		else if (erm->markType == ROW_MARK_KEYLOCK)
+ 			lockmode = LockTupleKeylock;
+ 		else
+ 		{
+ 			elog(ERROR, "unsupported rowmark type");
+ 			lockmode = LockTupleExclusive;	/* keep compiler quiet */
+ 		}
  
  		test = heap_lock_tuple(erm->relation, &tuple, &buffer,
  							   &update_ctid, &update_xmax,
*** a/src/backend/nodes/copyfuncs.c
--- b/src/backend/nodes/copyfuncs.c
***************
*** 1953,1959 **** _copyRowMarkClause(RowMarkClause *from)
  	RowMarkClause *newnode = makeNode(RowMarkClause);
  
  	COPY_SCALAR_FIELD(rti);
! 	COPY_SCALAR_FIELD(forUpdate);
  	COPY_SCALAR_FIELD(noWait);
  	COPY_SCALAR_FIELD(pushedDown);
  
--- 1953,1959 ----
  	RowMarkClause *newnode = makeNode(RowMarkClause);
  
  	COPY_SCALAR_FIELD(rti);
! 	COPY_SCALAR_FIELD(strength);
  	COPY_SCALAR_FIELD(noWait);
  	COPY_SCALAR_FIELD(pushedDown);
  
***************
*** 2310,2316 **** _copyLockingClause(LockingClause *from)
  	LockingClause *newnode = makeNode(LockingClause);
  
  	COPY_NODE_FIELD(lockedRels);
! 	COPY_SCALAR_FIELD(forUpdate);
  	COPY_SCALAR_FIELD(noWait);
  
  	return newnode;
--- 2310,2316 ----
  	LockingClause *newnode = makeNode(LockingClause);
  
  	COPY_NODE_FIELD(lockedRels);
! 	COPY_SCALAR_FIELD(strength);
  	COPY_SCALAR_FIELD(noWait);
  
  	return newnode;
*** a/src/backend/nodes/equalfuncs.c
--- b/src/backend/nodes/equalfuncs.c
***************
*** 2266,2272 **** static bool
  _equalLockingClause(LockingClause *a, LockingClause *b)
  {
  	COMPARE_NODE_FIELD(lockedRels);
! 	COMPARE_SCALAR_FIELD(forUpdate);
  	COMPARE_SCALAR_FIELD(noWait);
  
  	return true;
--- 2266,2272 ----
  _equalLockingClause(LockingClause *a, LockingClause *b)
  {
  	COMPARE_NODE_FIELD(lockedRels);
! 	COMPARE_SCALAR_FIELD(strength);
  	COMPARE_SCALAR_FIELD(noWait);
  
  	return true;
***************
*** 2335,2341 **** static bool
  _equalRowMarkClause(RowMarkClause *a, RowMarkClause *b)
  {
  	COMPARE_SCALAR_FIELD(rti);
! 	COMPARE_SCALAR_FIELD(forUpdate);
  	COMPARE_SCALAR_FIELD(noWait);
  	COMPARE_SCALAR_FIELD(pushedDown);
  
--- 2335,2341 ----
  _equalRowMarkClause(RowMarkClause *a, RowMarkClause *b)
  {
  	COMPARE_SCALAR_FIELD(rti);
! 	COMPARE_SCALAR_FIELD(strength);
  	COMPARE_SCALAR_FIELD(noWait);
  	COMPARE_SCALAR_FIELD(pushedDown);
  
*** a/src/backend/nodes/outfuncs.c
--- b/src/backend/nodes/outfuncs.c
***************
*** 2005,2011 **** _outLockingClause(StringInfo str, LockingClause *node)
  	WRITE_NODE_TYPE("LOCKINGCLAUSE");
  
  	WRITE_NODE_FIELD(lockedRels);
! 	WRITE_BOOL_FIELD(forUpdate);
  	WRITE_BOOL_FIELD(noWait);
  }
  
--- 2005,2011 ----
  	WRITE_NODE_TYPE("LOCKINGCLAUSE");
  
  	WRITE_NODE_FIELD(lockedRels);
! 	WRITE_ENUM_FIELD(strength, LockClauseStrength);
  	WRITE_BOOL_FIELD(noWait);
  }
  
***************
*** 2181,2187 **** _outRowMarkClause(StringInfo str, RowMarkClause *node)
  	WRITE_NODE_TYPE("ROWMARKCLAUSE");
  
  	WRITE_UINT_FIELD(rti);
! 	WRITE_BOOL_FIELD(forUpdate);
  	WRITE_BOOL_FIELD(noWait);
  	WRITE_BOOL_FIELD(pushedDown);
  }
--- 2181,2187 ----
  	WRITE_NODE_TYPE("ROWMARKCLAUSE");
  
  	WRITE_UINT_FIELD(rti);
! 	WRITE_BOOL_FIELD(strength);
  	WRITE_BOOL_FIELD(noWait);
  	WRITE_BOOL_FIELD(pushedDown);
  }
*** a/src/backend/nodes/readfuncs.c
--- b/src/backend/nodes/readfuncs.c
***************
*** 299,305 **** _readRowMarkClause(void)
  	READ_LOCALS(RowMarkClause);
  
  	READ_UINT_FIELD(rti);
! 	READ_BOOL_FIELD(forUpdate);
  	READ_BOOL_FIELD(noWait);
  	READ_BOOL_FIELD(pushedDown);
  
--- 299,305 ----
  	READ_LOCALS(RowMarkClause);
  
  	READ_UINT_FIELD(rti);
! 	READ_BOOL_FIELD(strength);
  	READ_BOOL_FIELD(noWait);
  	READ_BOOL_FIELD(pushedDown);
  
*** a/src/backend/optimizer/plan/initsplan.c
--- b/src/backend/optimizer/plan/initsplan.c
***************
*** 561,571 **** make_outerjoininfo(PlannerInfo *root,
  	Assert(jointype != JOIN_RIGHT);
  
  	/*
! 	 * Presently the executor cannot support FOR UPDATE/SHARE marking of rels
  	 * appearing on the nullable side of an outer join. (It's somewhat unclear
  	 * what that would mean, anyway: what should we mark when a result row is
  	 * generated from no element of the nullable relation?)  So, complain if
! 	 * any nullable rel is FOR UPDATE/SHARE.
  	 *
  	 * You might be wondering why this test isn't made far upstream in the
  	 * parser.	It's because the parser hasn't got enough info --- consider
--- 561,571 ----
  	Assert(jointype != JOIN_RIGHT);
  
  	/*
! 	 * Presently the executor cannot support FOR UPDATE/SHARE/KEY LOCK marking of rels
  	 * appearing on the nullable side of an outer join. (It's somewhat unclear
  	 * what that would mean, anyway: what should we mark when a result row is
  	 * generated from no element of the nullable relation?)  So, complain if
! 	 * any nullable rel is FOR UPDATE/SHARE/KEY LOCK.
  	 *
  	 * You might be wondering why this test isn't made far upstream in the
  	 * parser.	It's because the parser hasn't got enough info --- consider
***************
*** 583,589 **** make_outerjoininfo(PlannerInfo *root,
  			(jointype == JOIN_FULL && bms_is_member(rc->rti, left_rels)))
  			ereport(ERROR,
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 					 errmsg("SELECT FOR UPDATE/SHARE cannot be applied to the nullable side of an outer join")));
  	}
  
  	sjinfo->syn_lefthand = left_rels;
--- 583,589 ----
  			(jointype == JOIN_FULL && bms_is_member(rc->rti, left_rels)))
  			ereport(ERROR,
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 					 errmsg("SELECT FOR UPDATE/SHARE/KEY LOCK cannot be applied to the nullable side of an outer join")));
  	}
  
  	sjinfo->syn_lefthand = left_rels;
*** a/src/backend/optimizer/plan/planner.c
--- b/src/backend/optimizer/plan/planner.c
***************
*** 1830,1836 **** preprocess_rowmarks(PlannerInfo *root)
  	if (parse->rowMarks)
  	{
  		/*
! 		 * We've got trouble if FOR UPDATE/SHARE appears inside grouping,
  		 * since grouping renders a reference to individual tuple CTIDs
  		 * invalid.  This is also checked at parse time, but that's
  		 * insufficient because of rule substitution, query pullup, etc.
--- 1830,1836 ----
  	if (parse->rowMarks)
  	{
  		/*
! 		 * We've got trouble if FOR UPDATE/SHARE/KEY LOCK appears inside grouping,
  		 * since grouping renders a reference to individual tuple CTIDs
  		 * invalid.  This is also checked at parse time, but that's
  		 * insufficient because of rule substitution, query pullup, etc.
***************
*** 1840,1846 **** preprocess_rowmarks(PlannerInfo *root)
  	else
  	{
  		/*
! 		 * We only need rowmarks for UPDATE, DELETE, or FOR UPDATE/SHARE.
  		 */
  		if (parse->commandType != CMD_UPDATE &&
  			parse->commandType != CMD_DELETE)
--- 1840,1846 ----
  	else
  	{
  		/*
! 		 * We only need rowmarks for UPDATE, DELETE, or FOR UPDATE/SHARE/KEY LOCK.
  		 */
  		if (parse->commandType != CMD_UPDATE &&
  			parse->commandType != CMD_DELETE)
***************
*** 1850,1856 **** preprocess_rowmarks(PlannerInfo *root)
  	/*
  	 * We need to have rowmarks for all base relations except the target. We
  	 * make a bitmapset of all base rels and then remove the items we don't
! 	 * need or have FOR UPDATE/SHARE marks for.
  	 */
  	rels = get_base_rel_indexes((Node *) parse->jointree);
  	if (parse->resultRelation)
--- 1850,1856 ----
  	/*
  	 * We need to have rowmarks for all base relations except the target. We
  	 * make a bitmapset of all base rels and then remove the items we don't
! 	 * need or have FOR UPDATE/SHARE/KEY LOCK marks for.
  	 */
  	rels = get_base_rel_indexes((Node *) parse->jointree);
  	if (parse->resultRelation)
***************
*** 1887,1896 **** preprocess_rowmarks(PlannerInfo *root)
  		newrc = makeNode(PlanRowMark);
  		newrc->rti = newrc->prti = rc->rti;
  		newrc->rowmarkId = ++(root->glob->lastRowMarkId);
! 		if (rc->forUpdate)
! 			newrc->markType = ROW_MARK_EXCLUSIVE;
! 		else
! 			newrc->markType = ROW_MARK_SHARE;
  		newrc->noWait = rc->noWait;
  		newrc->isParent = false;
  
--- 1887,1904 ----
  		newrc = makeNode(PlanRowMark);
  		newrc->rti = newrc->prti = rc->rti;
  		newrc->rowmarkId = ++(root->glob->lastRowMarkId);
!  		switch (rc->strength)
!  		{
!  			case LCS_FORUPDATE:
!  				newrc->markType = ROW_MARK_EXCLUSIVE;
!  				break;
!  			case LCS_FORSHARE:
!  				newrc->markType = ROW_MARK_SHARE;
!  				break;
!  			case LCS_FORKEYLOCK:
!  				newrc->markType = ROW_MARK_KEYLOCK;
!  				break;
!  		}
  		newrc->noWait = rc->noWait;
  		newrc->isParent = false;
  
*** a/src/backend/parser/analyze.c
--- b/src/backend/parser/analyze.c
***************
*** 2161,2167 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
  	/* make a clause we can pass down to subqueries to select all rels */
  	allrels = makeNode(LockingClause);
  	allrels->lockedRels = NIL;	/* indicates all rels */
! 	allrels->forUpdate = lc->forUpdate;
  	allrels->noWait = lc->noWait;
  
  	if (lockedRels == NIL)
--- 2161,2167 ----
  	/* make a clause we can pass down to subqueries to select all rels */
  	allrels = makeNode(LockingClause);
  	allrels->lockedRels = NIL;	/* indicates all rels */
! 	allrels->strength = lc->strength;
  	allrels->noWait = lc->noWait;
  
  	if (lockedRels == NIL)
***************
*** 2177,2188 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
  			{
  				case RTE_RELATION:
  					applyLockingClause(qry, i,
! 									   lc->forUpdate, lc->noWait, pushedDown);
  					rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  					break;
  				case RTE_SUBQUERY:
  					applyLockingClause(qry, i,
! 									   lc->forUpdate, lc->noWait, pushedDown);
  
  					/*
  					 * FOR UPDATE/SHARE of subquery is propagated to all of
--- 2177,2188 ----
  			{
  				case RTE_RELATION:
  					applyLockingClause(qry, i,
! 									   lc->strength, lc->noWait, pushedDown);
  					rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  					break;
  				case RTE_SUBQUERY:
  					applyLockingClause(qry, i,
! 									   lc->strength, lc->noWait, pushedDown);
  
  					/*
  					 * FOR UPDATE/SHARE of subquery is propagated to all of
***************
*** 2226,2238 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
  					{
  						case RTE_RELATION:
  							applyLockingClause(qry, i,
! 											   lc->forUpdate, lc->noWait,
  											   pushedDown);
  							rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  							break;
  						case RTE_SUBQUERY:
  							applyLockingClause(qry, i,
! 											   lc->forUpdate, lc->noWait,
  											   pushedDown);
  							/* see comment above */
  							transformLockingClause(pstate, rte->subquery,
--- 2226,2238 ----
  					{
  						case RTE_RELATION:
  							applyLockingClause(qry, i,
! 											   lc->strength, lc->noWait,
  											   pushedDown);
  							rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  							break;
  						case RTE_SUBQUERY:
  							applyLockingClause(qry, i,
! 											   lc->strength, lc->noWait,
  											   pushedDown);
  							/* see comment above */
  							transformLockingClause(pstate, rte->subquery,
***************
*** 2291,2297 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
   */
  void
  applyLockingClause(Query *qry, Index rtindex,
! 				   bool forUpdate, bool noWait, bool pushedDown)
  {
  	RowMarkClause *rc;
  
--- 2291,2297 ----
   */
  void
  applyLockingClause(Query *qry, Index rtindex,
! 				   LockClauseStrength strength, bool noWait, bool pushedDown)
  {
  	RowMarkClause *rc;
  
***************
*** 2303,2312 **** applyLockingClause(Query *qry, Index rtindex,
  	if ((rc = get_parse_rowmark(qry, rtindex)) != NULL)
  	{
  		/*
! 		 * If the same RTE is specified both FOR UPDATE and FOR SHARE, treat
! 		 * it as FOR UPDATE.  (Reasonable, since you can't take both a shared
! 		 * and exclusive lock at the same time; it'll end up being exclusive
! 		 * anyway.)
  		 *
  		 * We also consider that NOWAIT wins if it's specified both ways. This
  		 * is a bit more debatable but raising an error doesn't seem helpful.
--- 2303,2312 ----
  	if ((rc = get_parse_rowmark(qry, rtindex)) != NULL)
  	{
  		/*
! 		 * If the same RTE is specified for more than one locking strength,
! 		 * treat is as the strongest.  (Reasonable, since you can't take both a
! 		 * shared and exclusive lock at the same time; it'll end up being
! 		 * exclusive anyway.)
  		 *
  		 * We also consider that NOWAIT wins if it's specified both ways. This
  		 * is a bit more debatable but raising an error doesn't seem helpful.
***************
*** 2315,2321 **** applyLockingClause(Query *qry, Index rtindex,
  		 *
  		 * And of course pushedDown becomes false if any clause is explicit.
  		 */
! 		rc->forUpdate |= forUpdate;
  		rc->noWait |= noWait;
  		rc->pushedDown &= pushedDown;
  		return;
--- 2315,2321 ----
  		 *
  		 * And of course pushedDown becomes false if any clause is explicit.
  		 */
! 		rc->strength = Max(rc->strength, strength);
  		rc->noWait |= noWait;
  		rc->pushedDown &= pushedDown;
  		return;
***************
*** 2324,2330 **** applyLockingClause(Query *qry, Index rtindex,
  	/* Make a new RowMarkClause */
  	rc = makeNode(RowMarkClause);
  	rc->rti = rtindex;
! 	rc->forUpdate = forUpdate;
  	rc->noWait = noWait;
  	rc->pushedDown = pushedDown;
  	qry->rowMarks = lappend(qry->rowMarks, rc);
--- 2324,2330 ----
  	/* Make a new RowMarkClause */
  	rc = makeNode(RowMarkClause);
  	rc->rti = rtindex;
! 	rc->strength = strength;
  	rc->noWait = noWait;
  	rc->pushedDown = pushedDown;
  	qry->rowMarks = lappend(qry->rowMarks, rc);
*** a/src/backend/parser/gram.y
--- b/src/backend/parser/gram.y
***************
*** 8542,8548 **** for_locking_item:
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->forUpdate = TRUE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
--- 8542,8548 ----
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->strength = LCS_FORUPDATE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
***************
*** 8550,8559 **** for_locking_item:
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->forUpdate = FALSE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
  		;
  
  locked_rels_list:
--- 8550,8567 ----
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->strength = LCS_FORSHARE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
+ 			| FOR KEY LOCK_P locked_rels_list opt_nowait
+ 				{
+ 					LockingClause *n = makeNode(LockingClause);
+ 					n->lockedRels = $4;
+ 					n->strength = LCS_FORKEYLOCK;
+ 					n->noWait = $5;
+ 					$$ = (Node *) n;
+ 				}
  		;
  
  locked_rels_list:
*** a/src/backend/rewrite/rewriteHandler.c
--- b/src/backend/rewrite/rewriteHandler.c
***************
*** 55,61 **** static void rewriteValuesRTE(RangeTblEntry *rte, Relation target_relation,
  static void rewriteTargetListUD(Query *parsetree, RangeTblEntry *target_rte,
  								Relation target_relation);
  static void markQueryForLocking(Query *qry, Node *jtnode,
! 					bool forUpdate, bool noWait, bool pushedDown);
  static List *matchLocks(CmdType event, RuleLock *rulelocks,
  		   int varno, Query *parsetree);
  static Query *fireRIRrules(Query *parsetree, List *activeRIRs,
--- 55,61 ----
  static void rewriteTargetListUD(Query *parsetree, RangeTblEntry *target_rte,
  								Relation target_relation);
  static void markQueryForLocking(Query *qry, Node *jtnode,
! 					LockClauseStrength strength, bool noWait, bool pushedDown);
  static List *matchLocks(CmdType event, RuleLock *rulelocks,
  		   int varno, Query *parsetree);
  static Query *fireRIRrules(Query *parsetree, List *activeRIRs,
***************
*** 1354,1361 **** ApplyRetrieveRule(Query *parsetree,
  	rte->modifiedCols = NULL;
  
  	/*
! 	 * If FOR UPDATE/SHARE of view, mark all the contained tables as implicit
! 	 * FOR UPDATE/SHARE, the same as the parser would have done if the view's
  	 * subquery had been written out explicitly.
  	 *
  	 * Note: we don't consider forUpdatePushedDown here; such marks will be
--- 1354,1361 ----
  	rte->modifiedCols = NULL;
  
  	/*
! 	 * If FOR UPDATE/SHARE/KEY LOCK of view, mark all the contained tables as implicit
! 	 * FOR UPDATE/SHARE/KEY LOCK, the same as the parser would have done if the view's
  	 * subquery had been written out explicitly.
  	 *
  	 * Note: we don't consider forUpdatePushedDown here; such marks will be
***************
*** 1363,1375 **** ApplyRetrieveRule(Query *parsetree,
  	 */
  	if (rc != NULL)
  		markQueryForLocking(rule_action, (Node *) rule_action->jointree,
! 							rc->forUpdate, rc->noWait, true);
  
  	return parsetree;
  }
  
  /*
!  * Recursively mark all relations used by a view as FOR UPDATE/SHARE.
   *
   * This may generate an invalid query, eg if some sub-query uses an
   * aggregate.  We leave it to the planner to detect that.
--- 1363,1375 ----
  	 */
  	if (rc != NULL)
  		markQueryForLocking(rule_action, (Node *) rule_action->jointree,
! 							rc->strength, rc->noWait, true);
  
  	return parsetree;
  }
  
  /*
!  * Recursively mark all relations used by a view as FOR UPDATE/SHARE/KEY LOCK.
   *
   * This may generate an invalid query, eg if some sub-query uses an
   * aggregate.  We leave it to the planner to detect that.
***************
*** 1381,1387 **** ApplyRetrieveRule(Query *parsetree,
   */
  static void
  markQueryForLocking(Query *qry, Node *jtnode,
! 					bool forUpdate, bool noWait, bool pushedDown)
  {
  	if (jtnode == NULL)
  		return;
--- 1381,1387 ----
   */
  static void
  markQueryForLocking(Query *qry, Node *jtnode,
! 					LockClauseStrength strength, bool noWait, bool pushedDown)
  {
  	if (jtnode == NULL)
  		return;
***************
*** 1392,1406 **** markQueryForLocking(Query *qry, Node *jtnode,
  
  		if (rte->rtekind == RTE_RELATION)
  		{
! 			applyLockingClause(qry, rti, forUpdate, noWait, pushedDown);
  			rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  		}
  		else if (rte->rtekind == RTE_SUBQUERY)
  		{
! 			applyLockingClause(qry, rti, forUpdate, noWait, pushedDown);
! 			/* FOR UPDATE/SHARE of subquery is propagated to subquery's rels */
  			markQueryForLocking(rte->subquery, (Node *) rte->subquery->jointree,
! 								forUpdate, noWait, true);
  		}
  		/* other RTE types are unaffected by FOR UPDATE */
  	}
--- 1392,1406 ----
  
  		if (rte->rtekind == RTE_RELATION)
  		{
! 			applyLockingClause(qry, rti, strength, noWait, pushedDown);
  			rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  		}
  		else if (rte->rtekind == RTE_SUBQUERY)
  		{
! 			applyLockingClause(qry, rti, strength, noWait, pushedDown);
! 			/* FOR UPDATE/SHARE/KEY LOCK of subquery is propagated to subquery's rels */
  			markQueryForLocking(rte->subquery, (Node *) rte->subquery->jointree,
! 								strength, noWait, true);
  		}
  		/* other RTE types are unaffected by FOR UPDATE */
  	}
***************
*** 1410,1423 **** markQueryForLocking(Query *qry, Node *jtnode,
  		ListCell   *l;
  
  		foreach(l, f->fromlist)
! 			markQueryForLocking(qry, lfirst(l), forUpdate, noWait, pushedDown);
  	}
  	else if (IsA(jtnode, JoinExpr))
  	{
  		JoinExpr   *j = (JoinExpr *) jtnode;
  
! 		markQueryForLocking(qry, j->larg, forUpdate, noWait, pushedDown);
! 		markQueryForLocking(qry, j->rarg, forUpdate, noWait, pushedDown);
  	}
  	else
  		elog(ERROR, "unrecognized node type: %d",
--- 1410,1423 ----
  		ListCell   *l;
  
  		foreach(l, f->fromlist)
! 			markQueryForLocking(qry, lfirst(l), strength, noWait, pushedDown);
  	}
  	else if (IsA(jtnode, JoinExpr))
  	{
  		JoinExpr   *j = (JoinExpr *) jtnode;
  
! 		markQueryForLocking(qry, j->larg, strength, noWait, pushedDown);
! 		markQueryForLocking(qry, j->rarg, strength, noWait, pushedDown);
  	}
  	else
  		elog(ERROR, "unrecognized node type: %d",
*** a/src/backend/tcop/utility.c
--- b/src/backend/tcop/utility.c
***************
*** 2205,2214 **** CreateCommandTag(Node *parsetree)
  						else if (stmt->rowMarks != NIL)
  						{
  							/* not 100% but probably close enough */
! 							if (((RowMarkClause *) linitial(stmt->rowMarks))->forUpdate)
! 								tag = "SELECT FOR UPDATE";
! 							else
! 								tag = "SELECT FOR SHARE";
  						}
  						else
  							tag = "SELECT";
--- 2205,2225 ----
  						else if (stmt->rowMarks != NIL)
  						{
  							/* not 100% but probably close enough */
! 							switch (((RowMarkClause *) linitial(stmt->rowMarks))->strength)
! 							{
! 								case LCS_FORUPDATE:
! 									tag = "SELECT FOR UPDATE";
! 									break;
! 								case LCS_FORSHARE:
! 									tag = "SELECT FOR SHARE";
! 									break;
! 								case LCS_FORKEYLOCK:
! 									tag = "SELECT FOR KEY LOCK";
! 									break;
! 								default:
! 									tag =  "???";
! 									break;
! 							}
  						}
  						else
  							tag = "SELECT";
*** a/src/backend/utils/adt/ri_triggers.c
--- b/src/backend/utils/adt/ri_triggers.c
***************
*** 308,314 **** RI_FKey_check(PG_FUNCTION_ARGS)
  	 * Get the relation descriptors of the FK and PK tables.
  	 *
  	 * pk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = trigdata->tg_relation;
  	pk_rel = heap_open(riinfo.pk_relid, RowShareLock);
--- 308,314 ----
  	 * Get the relation descriptors of the FK and PK tables.
  	 *
  	 * pk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = trigdata->tg_relation;
  	pk_rel = heap_open(riinfo.pk_relid, RowShareLock);
***************
*** 338,349 **** RI_FKey_check(PG_FUNCTION_ARGS)
  
  			/* ---------
  			 * The query string built is
! 			 *	SELECT 1 FROM ONLY <pktable>
  			 * ----------
  			 */
  			quoteRelationName(pkrelname, pk_rel);
  			snprintf(querystr, sizeof(querystr),
! 					 "SELECT 1 FROM ONLY %s x FOR SHARE OF x",
  					 pkrelname);
  
  			/* Prepare and save the plan */
--- 338,349 ----
  
  			/* ---------
  			 * The query string built is
! 			 *	SELECT 1 FROM ONLY <pktable> x FOR KEY LOCK OF x
  			 * ----------
  			 */
  			quoteRelationName(pkrelname, pk_rel);
  			snprintf(querystr, sizeof(querystr),
! 					 "SELECT 1 FROM ONLY %s x FOR KEY LOCK OF x",
  					 pkrelname);
  
  			/* Prepare and save the plan */
***************
*** 463,469 **** RI_FKey_check(PG_FUNCTION_ARGS)
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> WHERE pkatt1 = $1 [AND ...] FOR SHARE
  		 * The type id's for the $ parameters are those of the
  		 * corresponding FK attributes.
  		 * ----------
--- 463,470 ----
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> x WHERE pkatt1 = $1 [AND ...]
! 		 *	       FOR KEY LOCK OF x
  		 * The type id's for the $ parameters are those of the
  		 * corresponding FK attributes.
  		 * ----------
***************
*** 487,493 **** RI_FKey_check(PG_FUNCTION_ARGS)
  			querysep = "AND";
  			queryoids[i] = fk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 488,494 ----
  			querysep = "AND";
  			queryoids[i] = fk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 625,631 **** ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> WHERE pkatt1 = $1 [AND ...] FOR SHARE
  		 * The type id's for the $ parameters are those of the
  		 * PK attributes themselves.
  		 * ----------
--- 626,633 ----
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> x WHERE pkatt1 = $1 [AND ...]
! 		 *	       FOR KEY LOCK OF x
  		 * The type id's for the $ parameters are those of the
  		 * PK attributes themselves.
  		 * ----------
***************
*** 648,654 **** ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
  			querysep = "AND";
  			queryoids[i] = pk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
--- 650,656 ----
  			querysep = "AND";
  			queryoids[i] = pk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
***************
*** 712,718 **** RI_FKey_noaction_del(PG_FUNCTION_ARGS)
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 714,720 ----
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 780,786 **** RI_FKey_noaction_del(PG_FUNCTION_ARGS)
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> WHERE $1 = fkatt1 [AND ...]
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
--- 782,789 ----
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> x WHERE $1 = fkatt1 [AND ...]
! 				 *	       FOR KEY LOCK OF x
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
***************
*** 805,811 **** RI_FKey_noaction_del(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 808,814 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 890,896 **** RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 893,899 ----
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 993,999 **** RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 996,1002 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 1431,1437 **** RI_FKey_restrict_del(PG_FUNCTION_ARGS)
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 1434,1440 ----
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 1489,1495 **** RI_FKey_restrict_del(PG_FUNCTION_ARGS)
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> WHERE $1 = fkatt1 [AND ...]
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
--- 1492,1499 ----
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> x WHERE $1 = fkatt1 [AND ...]
! 				 *	       FOR KEY LOCK OF x
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
***************
*** 1514,1520 **** RI_FKey_restrict_del(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 1518,1524 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 1604,1610 **** RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 1608,1614 ----
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 1672,1678 **** RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> WHERE $1 = fkatt1 [AND ...]
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
--- 1676,1683 ----
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> x WHERE $1 = fkatt1 [AND ...]
! 				 *	       FOR KEY LOCK OF x
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
***************
*** 1697,1703 **** RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 1702,1708 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
*** a/src/backend/utils/adt/ruleutils.c
--- b/src/backend/utils/adt/ruleutils.c
***************
*** 2837,2848 **** get_select_query_def(Query *query, deparse_context *context,
  			if (rc->pushedDown)
  				continue;
  
! 			if (rc->forUpdate)
! 				appendContextKeyword(context, " FOR UPDATE",
! 									 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 			else
! 				appendContextKeyword(context, " FOR SHARE",
! 									 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
  			appendStringInfo(buf, " OF %s",
  							 quote_identifier(rte->eref->aliasname));
  			if (rc->noWait)
--- 2837,2858 ----
  			if (rc->pushedDown)
  				continue;
  
! 			switch (rc->strength)
! 			{
! 				case LCS_FORKEYLOCK:
! 					appendContextKeyword(context, " FOR KEY LOCK",
! 										 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 					break;
! 				case LCS_FORSHARE:
! 					appendContextKeyword(context, " FOR SHARE",
! 										 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 					break;
! 				case LCS_FORUPDATE:
! 					appendContextKeyword(context, " FOR UPDATE",
! 										 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 					break;
! 			}
! 
  			appendStringInfo(buf, " OF %s",
  							 quote_identifier(rte->eref->aliasname));
  			if (rc->noWait)
*** a/src/backend/utils/cache/relcache.c
--- b/src/backend/utils/cache/relcache.c
***************
*** 3608,3613 **** RelationGetIndexPredicate(Relation relation)
--- 3608,3615 ----
   * simple index keys, but attributes used in expressions and partial-index
   * predicates.)
   *
+  * If "unique" is true, only attributes of unique indexes are considered.
+  *
   * Attribute numbers are offset by FirstLowInvalidHeapAttributeNumber so that
   * we can include system attributes (e.g., OID) in the bitmap representation.
   *
***************
*** 3615,3630 **** RelationGetIndexPredicate(Relation relation)
   * be bms_free'd when not needed anymore.
   */
  Bitmapset *
! RelationGetIndexAttrBitmap(Relation relation)
  {
  	Bitmapset  *indexattrs;
  	List	   *indexoidlist;
  	ListCell   *l;
  	MemoryContext oldcxt;
  
  	/* Quick exit if we already computed the result. */
  	if (relation->rd_indexattr != NULL)
! 		return bms_copy(relation->rd_indexattr);
  
  	/* Fast path if definitely no indexes */
  	if (!RelationGetForm(relation)->relhasindex)
--- 3617,3633 ----
   * be bms_free'd when not needed anymore.
   */
  Bitmapset *
! RelationGetIndexAttrBitmap(Relation relation, bool unique)
  {
  	Bitmapset  *indexattrs;
+ 	Bitmapset  *uindexattrs;
  	List	   *indexoidlist;
  	ListCell   *l;
  	MemoryContext oldcxt;
  
  	/* Quick exit if we already computed the result. */
  	if (relation->rd_indexattr != NULL)
! 		return bms_copy(unique ? relation->rd_uindexattr : relation->rd_indexattr);
  
  	/* Fast path if definitely no indexes */
  	if (!RelationGetForm(relation)->relhasindex)
***************
*** 3643,3648 **** RelationGetIndexAttrBitmap(Relation relation)
--- 3646,3652 ----
  	 * For each index, add referenced attributes to indexattrs.
  	 */
  	indexattrs = NULL;
+ 	uindexattrs = NULL;
  	foreach(l, indexoidlist)
  	{
  		Oid			indexOid = lfirst_oid(l);
***************
*** 3661,3675 **** RelationGetIndexAttrBitmap(Relation relation)
--- 3665,3688 ----
  			int			attrnum = indexInfo->ii_KeyAttrNumbers[i];
  
  			if (attrnum != 0)
+ 			{
  				indexattrs = bms_add_member(indexattrs,
  							   attrnum - FirstLowInvalidHeapAttributeNumber);
+ 				if (indexInfo->ii_Unique)
+ 					uindexattrs = bms_add_member(uindexattrs,
+ 											   	 attrnum - FirstLowInvalidHeapAttributeNumber);
+ 			}
  		}
  
  		/* Collect all attributes used in expressions, too */
  		pull_varattnos((Node *) indexInfo->ii_Expressions, &indexattrs);
+ 		if (indexInfo->ii_Unique)
+ 			pull_varattnos((Node *) indexInfo->ii_Expressions, &uindexattrs);
  
  		/* Collect all attributes in the index predicate, too */
  		pull_varattnos((Node *) indexInfo->ii_Predicate, &indexattrs);
+ 		if (indexInfo->ii_Unique)
+ 			pull_varattnos((Node *) indexInfo->ii_Predicate, &uindexattrs);
  
  		index_close(indexDesc, AccessShareLock);
  	}
***************
*** 3679,3688 **** RelationGetIndexAttrBitmap(Relation relation)
  	/* Now save a copy of the bitmap in the relcache entry. */
  	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
  	relation->rd_indexattr = bms_copy(indexattrs);
  	MemoryContextSwitchTo(oldcxt);
  
  	/* We return our original working copy for caller to play with */
! 	return indexattrs;
  }
  
  /*
--- 3692,3702 ----
  	/* Now save a copy of the bitmap in the relcache entry. */
  	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
  	relation->rd_indexattr = bms_copy(indexattrs);
+ 	relation->rd_uindexattr = bms_copy(uindexattrs);
  	MemoryContextSwitchTo(oldcxt);
  
  	/* We return our original working copy for caller to play with */
! 	return unique ? uindexattrs : indexattrs;
  }
  
  /*
*** a/src/include/access/heapam.h
--- b/src/include/access/heapam.h
***************
*** 33,38 **** typedef struct BulkInsertStateData *BulkInsertState;
--- 33,39 ----
  
  typedef enum
  {
+ 	LockTupleKeylock,
  	LockTupleShared,
  	LockTupleExclusive
  } LockTupleMode;
*** a/src/include/access/htup.h
--- b/src/include/access/htup.h
***************
*** 163,174 **** typedef HeapTupleHeaderData *HeapTupleHeader;
  #define HEAP_HASVARWIDTH		0x0002	/* has variable-width attribute(s) */
  #define HEAP_HASEXTERNAL		0x0004	/* has external stored attribute(s) */
  #define HEAP_HASOID				0x0008	/* has an object-id field */
! /* bit 0x0010 is available */
  #define HEAP_COMBOCID			0x0020	/* t_cid is a combo cid */
  #define HEAP_XMAX_EXCL_LOCK		0x0040	/* xmax is exclusive locker */
  #define HEAP_XMAX_SHARED_LOCK	0x0080	/* xmax is shared locker */
  /* if either LOCK bit is set, xmax hasn't deleted the tuple, only locked it */
! #define HEAP_IS_LOCKED	(HEAP_XMAX_EXCL_LOCK | HEAP_XMAX_SHARED_LOCK)
  #define HEAP_XMIN_COMMITTED		0x0100	/* t_xmin committed */
  #define HEAP_XMIN_INVALID		0x0200	/* t_xmin invalid/aborted */
  #define HEAP_XMAX_COMMITTED		0x0400	/* t_xmax committed */
--- 163,177 ----
  #define HEAP_HASVARWIDTH		0x0002	/* has variable-width attribute(s) */
  #define HEAP_HASEXTERNAL		0x0004	/* has external stored attribute(s) */
  #define HEAP_HASOID				0x0008	/* has an object-id field */
! #define HEAP_XMAX_KEY_LOCK		0x0010	/* xmax is a "key" locker */
  #define HEAP_COMBOCID			0x0020	/* t_cid is a combo cid */
  #define HEAP_XMAX_EXCL_LOCK		0x0040	/* xmax is exclusive locker */
  #define HEAP_XMAX_SHARED_LOCK	0x0080	/* xmax is shared locker */
+ /* if either SHARE or KEY lock bit is set, this is a "shared" lock */
+ #define HEAP_IS_SHARE_LOCKED (HEAP_XMAX_SHARED_LOCK | HEAP_XMAX_KEY_LOCK)
  /* if either LOCK bit is set, xmax hasn't deleted the tuple, only locked it */
! #define HEAP_IS_LOCKED	(HEAP_XMAX_EXCL_LOCK | HEAP_XMAX_SHARED_LOCK | \
! 						 HEAP_XMAX_KEY_LOCK)
  #define HEAP_XMIN_COMMITTED		0x0100	/* t_xmin committed */
  #define HEAP_XMIN_INVALID		0x0200	/* t_xmin invalid/aborted */
  #define HEAP_XMAX_COMMITTED		0x0400	/* t_xmax committed */
***************
*** 725,734 **** typedef struct xl_heap_lock
  	xl_heaptid	target;			/* locked tuple id */
  	TransactionId locking_xid;	/* might be a MultiXactId not xid */
  	bool		xid_is_mxact;	/* is it? */
! 	bool		shared_lock;	/* shared or exclusive row lock? */
  } xl_heap_lock;
  
! #define SizeOfHeapLock	(offsetof(xl_heap_lock, shared_lock) + sizeof(bool))
  
  /* This is what we need to know about in-place update */
  typedef struct xl_heap_inplace
--- 728,737 ----
  	xl_heaptid	target;			/* locked tuple id */
  	TransactionId locking_xid;	/* might be a MultiXactId not xid */
  	bool		xid_is_mxact;	/* is it? */
! 	char		lock_strength;	/* keylock, shared, exclusive lock? */
  } xl_heap_lock;
  
! #define SizeOfHeapLock	(offsetof(xl_heap_lock, lock_strength) + sizeof(char))
  
  /* This is what we need to know about in-place update */
  typedef struct xl_heap_inplace
*** a/src/include/nodes/execnodes.h
--- b/src/include/nodes/execnodes.h
***************
*** 404,410 **** typedef struct EState
   * ExecRowMark -
   *	   runtime representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE, we should have an
   * ExecRowMark for each non-target relation in the query (except inheritance
   * parent RTEs, which can be ignored at runtime).  See PlanRowMark for details
   * about most of the fields.  In addition to fields directly derived from
--- 404,410 ----
   * ExecRowMark -
   *	   runtime representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE/KEY LOCK, we should have an
   * ExecRowMark for each non-target relation in the query (except inheritance
   * parent RTEs, which can be ignored at runtime).  See PlanRowMark for details
   * about most of the fields.  In addition to fields directly derived from
*** a/src/include/nodes/parsenodes.h
--- b/src/include/nodes/parsenodes.h
***************
*** 554,571 **** typedef struct DefElem
  } DefElem;
  
  /*
!  * LockingClause - raw representation of FOR UPDATE/SHARE options
   *
   * Note: lockedRels == NIL means "all relations in query".	Otherwise it
   * is a list of RangeVar nodes.  (We use RangeVar mainly because it carries
   * a location field --- currently, parse analysis insists on unqualified
   * names in LockingClause.)
   */
  typedef struct LockingClause
  {
  	NodeTag		type;
  	List	   *lockedRels;		/* FOR UPDATE or FOR SHARE relations */
! 	bool		forUpdate;		/* true = FOR UPDATE, false = FOR SHARE */
  	bool		noWait;			/* NOWAIT option */
  } LockingClause;
  
--- 554,579 ----
  } DefElem;
  
  /*
!  * LockingClause - raw representation of FOR UPDATE/SHARE/KEY LOCK options
   *
   * Note: lockedRels == NIL means "all relations in query".	Otherwise it
   * is a list of RangeVar nodes.  (We use RangeVar mainly because it carries
   * a location field --- currently, parse analysis insists on unqualified
   * names in LockingClause.)
   */
+ typedef enum LockClauseStrength
+ {
+ 	/* order is important -- see applyLockingClause */
+ 	LCS_FORKEYLOCK,
+ 	LCS_FORSHARE,
+ 	LCS_FORUPDATE
+ } LockClauseStrength;
+ 
  typedef struct LockingClause
  {
  	NodeTag		type;
  	List	   *lockedRels;		/* FOR UPDATE or FOR SHARE relations */
! 	LockClauseStrength strength;
  	bool		noWait;			/* NOWAIT option */
  } LockingClause;
  
***************
*** 839,856 **** typedef struct WindowClause
   *	   parser output representation of FOR UPDATE/SHARE clauses
   *
   * Query.rowMarks contains a separate RowMarkClause node for each relation
!  * identified as a FOR UPDATE/SHARE target.  If FOR UPDATE/SHARE is applied
!  * to a subquery, we generate RowMarkClauses for all normal and subquery rels
!  * in the subquery, but they are marked pushedDown = true to distinguish them
!  * from clauses that were explicitly written at this query level.  Also,
!  * Query.hasForUpdate tells whether there were explicit FOR UPDATE/SHARE
!  * clauses in the current query level.
   */
  typedef struct RowMarkClause
  {
  	NodeTag		type;
  	Index		rti;			/* range table index of target relation */
! 	bool		forUpdate;		/* true = FOR UPDATE, false = FOR SHARE */
  	bool		noWait;			/* NOWAIT option */
  	bool		pushedDown;		/* pushed down from higher query level? */
  } RowMarkClause;
--- 847,864 ----
   *	   parser output representation of FOR UPDATE/SHARE clauses
   *
   * Query.rowMarks contains a separate RowMarkClause node for each relation
!  * identified as a FOR UPDATE/SHARE/KEY LOCK target.  If one of these clauses
!  * is applied to a subquery, we generate RowMarkClauses for all normal and
!  * subquery rels in the subquery, but they are marked pushedDown = true to
!  * distinguish them from clauses that were explicitly written at this query
!  * level.  Also, Query.hasForUpdate tells whether there were explicit FOR
!  * UPDATE/SHARE clauses in the current query level.
   */
  typedef struct RowMarkClause
  {
  	NodeTag		type;
  	Index		rti;			/* range table index of target relation */
! 	LockClauseStrength strength;
  	bool		noWait;			/* NOWAIT option */
  	bool		pushedDown;		/* pushed down from higher query level? */
  } RowMarkClause;
*** a/src/include/nodes/plannodes.h
--- b/src/include/nodes/plannodes.h
***************
*** 706,712 **** typedef struct Limit
   * RowMarkType -
   *	  enums for types of row-marking operations
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE, we have to uniquely
   * identify all the source rows, not only those from the target relations, so
   * that we can perform EvalPlanQual rechecking at need.  For plain tables we
   * can just fetch the TID, the same as for a target relation.  Otherwise (for
--- 706,712 ----
   * RowMarkType -
   *	  enums for types of row-marking operations
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE/KEY LOCK, we have to uniquely
   * identify all the source rows, not only those from the target relations, so
   * that we can perform EvalPlanQual rechecking at need.  For plain tables we
   * can just fetch the TID, the same as for a target relation.  Otherwise (for
***************
*** 718,736 **** typedef enum RowMarkType
  {
  	ROW_MARK_EXCLUSIVE,			/* obtain exclusive tuple lock */
  	ROW_MARK_SHARE,				/* obtain shared tuple lock */
  	ROW_MARK_REFERENCE,			/* just fetch the TID */
  	ROW_MARK_COPY				/* physically copy the row value */
  } RowMarkType;
  
! #define RowMarkRequiresRowShareLock(marktype)  ((marktype) <= ROW_MARK_SHARE)
  
  /*
   * PlanRowMark -
   *	   plan-time representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE, we create a separate
   * PlanRowMark node for each non-target relation in the query.	Relations that
!  * are not specified as FOR UPDATE/SHARE are marked ROW_MARK_REFERENCE (if
   * real tables) or ROW_MARK_COPY (if not).
   *
   * Initially all PlanRowMarks have rti == prti and isParent == false.
--- 718,737 ----
  {
  	ROW_MARK_EXCLUSIVE,			/* obtain exclusive tuple lock */
  	ROW_MARK_SHARE,				/* obtain shared tuple lock */
+ 	ROW_MARK_KEYLOCK,			/* obtain keylock tuple lock */
  	ROW_MARK_REFERENCE,			/* just fetch the TID */
  	ROW_MARK_COPY				/* physically copy the row value */
  } RowMarkType;
  
! #define RowMarkRequiresRowShareLock(marktype)  ((marktype) <= ROW_MARK_KEYLOCK)
  
  /*
   * PlanRowMark -
   *	   plan-time representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE/KEY LOCK, we create a separate
   * PlanRowMark node for each non-target relation in the query.	Relations that
!  * are not specified as FOR UPDATE/SHARE/KEY LOCK are marked ROW_MARK_REFERENCE (if
   * real tables) or ROW_MARK_COPY (if not).
   *
   * Initially all PlanRowMarks have rti == prti and isParent == false.
*** a/src/include/parser/analyze.h
--- b/src/include/parser/analyze.h
***************
*** 31,36 **** extern bool analyze_requires_snapshot(Node *parseTree);
  
  extern void CheckSelectLocking(Query *qry);
  extern void applyLockingClause(Query *qry, Index rtindex,
! 				   bool forUpdate, bool noWait, bool pushedDown);
  
  #endif   /* ANALYZE_H */
--- 31,36 ----
  
  extern void CheckSelectLocking(Query *qry);
  extern void applyLockingClause(Query *qry, Index rtindex,
! 				   LockClauseStrength strength, bool noWait, bool pushedDown);
  
  #endif   /* ANALYZE_H */
*** a/src/include/utils/rel.h
--- b/src/include/utils/rel.h
***************
*** 156,161 **** typedef struct RelationData
--- 156,162 ----
  	Oid			rd_id;			/* relation's object id */
  	List	   *rd_indexlist;	/* list of OIDs of indexes on relation */
  	Bitmapset  *rd_indexattr;	/* identifies columns used in indexes */
+ 	Bitmapset  *rd_uindexattr;	/* identifies columns used in unique indexes */
  	Oid			rd_oidindex;	/* OID of unique index on OID, if any */
  	LockInfoData rd_lockInfo;	/* lock mgr's info for locking relation */
  	RuleLock   *rd_rules;		/* rewrite rules */
*** a/src/include/utils/relcache.h
--- b/src/include/utils/relcache.h
***************
*** 42,48 **** extern List *RelationGetIndexList(Relation relation);
  extern Oid	RelationGetOidIndex(Relation relation);
  extern List *RelationGetIndexExpressions(Relation relation);
  extern List *RelationGetIndexPredicate(Relation relation);
! extern Bitmapset *RelationGetIndexAttrBitmap(Relation relation);
  extern void RelationGetExclusionInfo(Relation indexRelation,
  						 Oid **operators,
  						 Oid **procs,
--- 42,48 ----
  extern Oid	RelationGetOidIndex(Relation relation);
  extern List *RelationGetIndexExpressions(Relation relation);
  extern List *RelationGetIndexPredicate(Relation relation);
! extern Bitmapset *RelationGetIndexAttrBitmap(Relation relation, bool unique);
  extern void RelationGetExclusionInfo(Relation indexRelation,
  						 Oid **operators,
  						 Oid **procs,

From e069cef91c686aa87e220336198267e5a5a2aeac Mon Sep 17 00:00:00 2001
From: Marti Raudsepp <marti@juffo.org>
Date: Tue, 15 Feb 2011 00:33:35 +0200
Subject: [PATCH] Only acquire KEY LOCK for colums that can be referenced by foreign keys

Don't consider columns in unique indexes that have expressions or WHERE
predicates.
---
 src/backend/utils/cache/relcache.c |   23 +++++++++++++----------
 src/include/utils/rel.h            |    2 +-
 src/include/utils/relcache.h       |    2 +-
 3 files changed, 15 insertions(+), 12 deletions(-)

diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 4d37e8e..5119288 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -3608,7 +3608,8 @@ RelationGetIndexPredicate(Relation relation)
  * simple index keys, but attributes used in expressions and partial-index
  * predicates.)
  *
- * If "unique" is true, only attributes of unique indexes are considered.
+ * If "keyAttrs" is true, only attributes that can be referenced by foreign
+ * keys are considered.
  *
  * Attribute numbers are offset by FirstLowInvalidHeapAttributeNumber so that
  * we can include system attributes (e.g., OID) in the bitmap representation.
@@ -3617,7 +3618,7 @@ RelationGetIndexPredicate(Relation relation)
  * be bms_free'd when not needed anymore.
  */
 Bitmapset *
-RelationGetIndexAttrBitmap(Relation relation, bool unique)
+RelationGetIndexAttrBitmap(Relation relation, bool keyAttrs)
 {
 	Bitmapset  *indexattrs;
 	Bitmapset  *uindexattrs;
@@ -3627,7 +3628,7 @@ RelationGetIndexAttrBitmap(Relation relation, bool unique)
 
 	/* Quick exit if we already computed the result. */
 	if (relation->rd_indexattr != NULL)
-		return bms_copy(unique ? relation->rd_uindexattr : relation->rd_indexattr);
+		return bms_copy(keyAttrs ? relation->rd_keyattr : relation->rd_indexattr);
 
 	/* Fast path if definitely no indexes */
 	if (!RelationGetForm(relation)->relhasindex)
@@ -3653,12 +3654,18 @@ RelationGetIndexAttrBitmap(Relation relation, bool unique)
 		Relation	indexDesc;
 		IndexInfo  *indexInfo;
 		int			i;
+		bool		isKey;
 
 		indexDesc = index_open(indexOid, AccessShareLock);
 
 		/* Extract index key information from the index's pg_index row */
 		indexInfo = BuildIndexInfo(indexDesc);
 
+		/* Can this index be referenced by a foreign key? */
+		isKey = indexInfo->ii_Unique &&
+				indexInfo->ii_Expressions == NIL &&
+				indexInfo->ii_Predicate == NIL;
+
 		/* Collect simple attribute references */
 		for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
 		{
@@ -3668,7 +3675,7 @@ RelationGetIndexAttrBitmap(Relation relation, bool unique)
 			{
 				indexattrs = bms_add_member(indexattrs,
 							   attrnum - FirstLowInvalidHeapAttributeNumber);
-				if (indexInfo->ii_Unique)
+				if (isKey)
 					uindexattrs = bms_add_member(uindexattrs,
 											   	 attrnum - FirstLowInvalidHeapAttributeNumber);
 			}
@@ -3676,13 +3683,9 @@ RelationGetIndexAttrBitmap(Relation relation, bool unique)
 
 		/* Collect all attributes used in expressions, too */
 		pull_varattnos((Node *) indexInfo->ii_Expressions, &indexattrs);
-		if (indexInfo->ii_Unique)
-			pull_varattnos((Node *) indexInfo->ii_Expressions, &uindexattrs);
 
 		/* Collect all attributes in the index predicate, too */
 		pull_varattnos((Node *) indexInfo->ii_Predicate, &indexattrs);
-		if (indexInfo->ii_Unique)
-			pull_varattnos((Node *) indexInfo->ii_Predicate, &uindexattrs);
 
 		index_close(indexDesc, AccessShareLock);
 	}
@@ -3692,11 +3695,11 @@ RelationGetIndexAttrBitmap(Relation relation, bool unique)
 	/* Now save a copy of the bitmap in the relcache entry. */
 	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
 	relation->rd_indexattr = bms_copy(indexattrs);
-	relation->rd_uindexattr = bms_copy(uindexattrs);
+	relation->rd_keyattr = bms_copy(uindexattrs);
 	MemoryContextSwitchTo(oldcxt);
 
 	/* We return our original working copy for caller to play with */
-	return unique ? uindexattrs : indexattrs;
+	return keyAttrs ? uindexattrs : indexattrs;
 }
 
 /*
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 2251b25..9b70d81 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -156,7 +156,7 @@ typedef struct RelationData
 	Oid			rd_id;			/* relation's object id */
 	List	   *rd_indexlist;	/* list of OIDs of indexes on relation */
 	Bitmapset  *rd_indexattr;	/* identifies columns used in indexes */
-	Bitmapset  *rd_uindexattr;	/* identifies columns used in unique indexes */
+	Bitmapset  *rd_keyattr;		/* cols that can be ref'd by foreign keys */
 	Oid			rd_oidindex;	/* OID of unique index on OID, if any */
 	LockInfoData rd_lockInfo;	/* lock mgr's info for locking relation */
 	RuleLock   *rd_rules;		/* rewrite rules */
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 6d1e64f..d4a09e3 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -42,7 +42,7 @@ extern List *RelationGetIndexList(Relation relation);
 extern Oid	RelationGetOidIndex(Relation relation);
 extern List *RelationGetIndexExpressions(Relation relation);
 extern List *RelationGetIndexPredicate(Relation relation);
-extern Bitmapset *RelationGetIndexAttrBitmap(Relation relation, bool unique);
+extern Bitmapset *RelationGetIndexAttrBitmap(Relation relation, bool keyAttrs);
 extern void RelationGetExclusionInfo(Relation indexRelation,
 						 Oid **operators,
 						 Oid **procs,
-- 
1.7.4

From 04298459b514495a8f1ef269b7a43c2ff3a50710 Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Mon, 14 Feb 2011 14:48:40 -0300
Subject: [PATCH 1/2] Fix visibility bug and poorly worded comment

per Noah Misch
---
 src/backend/access/heap/heapam.c |    9 ++++++---
 1 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 7515dc8..5d5ccbf 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2644,10 +2644,11 @@ l2:
 		newtup->t_data->t_infomask |= (oldtup.t_data->t_infomask & 
 									   (HEAP_XMAX_IS_MULTI |
 										HEAP_XMAX_KEY_LOCK));
+
 		/*
-		 * we also need to copy the combo CID stuff, but only if the original
-		 * tuple was created by us; otherwise the combocid module complains
-		 * (Alternatively we could use HeapTupleHeaderGetRawCommandId)
+		 * If the tuple was created in this transaction, and we're going to
+		 * delete it, then it must have a combo-cid, which we need to preserve.
+		 * Otherwise, just use the passed cid.
 		 */
 		if (TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmin(oldtup.t_data)))
 		{
@@ -2656,6 +2657,8 @@ l2:
 			HeapTupleHeaderSetCmin(newtup->t_data,
 								   HeapTupleHeaderGetCmin(oldtup.t_data));
 		}
+		else
+			HeapTupleHeaderSetCmin(newtup->t_data, cid);
 
 	}
 	else
-- 
1.7.2.3

*** /dev/null
--- b/src/test/isolation/expected/fk-contention.out
***************
*** 0 ****
--- 1,16 ----
+ Parsed test spec with 2 sessions
+ 
+ starting permutation: ins com upd
+ step ins:  INSERT INTO bar VALUES (42); 
+ step com:  COMMIT; 
+ step upd:  UPDATE foo SET b = 'Hello World'; 
+ 
+ starting permutation: ins upd com
+ step ins:  INSERT INTO bar VALUES (42); 
+ step upd:  UPDATE foo SET b = 'Hello World'; 
+ step com:  COMMIT; 
+ 
+ starting permutation: upd ins com
+ step upd:  UPDATE foo SET b = 'Hello World'; 
+ step ins:  INSERT INTO bar VALUES (42); 
+ step com:  COMMIT; 
*** /dev/null
--- b/src/test/isolation/expected/fk-deadlock.out
***************
*** 0 ****
--- 1,63 ----
+ Parsed test spec with 2 sessions
+ 
+ starting permutation: s1i s1u s1c s2i s2u s2c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ step s1c:  COMMIT; 
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1i s1u s2i s1c s2u s2c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ step s2i:  INSERT INTO child VALUES (2, 1);  <waiting ...>
+ step s1c:  COMMIT; 
+ step s2i: <... completed>
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1i s2i s1u s2u s1c s2c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
+ step s1c:  COMMIT; 
+ step s2u: <... completed>
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1i s2i s2u s1u s2c s1c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
+ step s2c:  COMMIT; 
+ step s1u: <... completed>
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2i s1i s1u s2u s1c s2c
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
+ step s1c:  COMMIT; 
+ step s2u: <... completed>
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2i s1i s2u s1u s2c s1c
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
+ step s2c:  COMMIT; 
+ step s1u: <... completed>
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2i s2u s1i s2c s1u s1c
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s1i:  INSERT INTO child VALUES (1, 1);  <waiting ...>
+ step s2c:  COMMIT; 
+ step s1i: <... completed>
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ step s1c:  COMMIT; 
*** /dev/null
--- b/src/test/isolation/expected/fk-deadlock2.out
***************
*** 0 ****
--- 1,106 ----
+ Parsed test spec with 2 sessions
+ 
+ starting permutation: s1u1 s1u2 s1c s2u1 s2u2 s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1c:  COMMIT; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s1u2 s2u1 s1c s2u2 s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1c:  COMMIT; 
+ step s2u1: <... completed>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s1u2 s2u2 s1c s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s1u2 s2u2 s2c s1c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s2u2 s1u2 s1c s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s2u2 s1u2 s2c s1c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s1u2 s2u2 s1c s2c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s1u2 s2u2 s2c s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s2u2 s1u2 s1c s2c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s2u2 s1u2 s2c s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s2u2 s1u1 s2c s1u2 s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2c:  COMMIT; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1c:  COMMIT; 
*** a/src/test/isolation/isolation_schedule
--- b/src/test/isolation/isolation_schedule
***************
*** 9,11 **** test: ri-trigger
--- 9,14 ----
  test: partial-index
  test: two-ids
  test: multiple-row-versions
+ test: fk-contention
+ test: fk-deadlock
+ test: fk-deadlock2
*** a/src/test/isolation/isolationtester.c
--- b/src/test/isolation/isolationtester.c
***************
*** 9,23 ****
  #include <windows.h>
  #endif
  
  #include <stddef.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <string.h>
- #include "libpq-fe.h"
  
  #include "isolationtester.h"
  
  static PGconn  **conns = NULL;
  static int	   nconns = 0;
  
  static void run_all_permutations(TestSpec *testspec);
--- 9,36 ----
  #include <windows.h>
  #endif
  
+ #include <errno.h>
+ #include <unistd.h>
  #include <stddef.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <string.h>
  
+ #ifdef HAVE_SYS_SELECT_H
+ #include <sys/select.h>
+ #endif
+ 
+ #include "libpq-fe.h"
  #include "isolationtester.h"
  
+ #define PREP_WAITING "isolationtester_waiting"
+ 
+ /*
+  * conns[0] is the global setup, teardown, and watchdog connection.  Additional
+  * connections represent spec-defined sessions.
+  */
  static PGconn  **conns = NULL;
+ static const char **backend_ids = NULL;
  static int	   nconns = 0;
  
  static void run_all_permutations(TestSpec *testspec);
***************
*** 25,30 **** static void run_all_permutations_recurse(TestSpec *testspec, int nsteps, Step **
--- 38,47 ----
  static void run_named_permutations(TestSpec *testspec);
  static void run_permutation(TestSpec *testspec, int nsteps, Step **steps);
  
+ #define STEP_NONBLOCK	0x1	/* return 0 as soon as cmd waits for a lock */
+ #define STEP_RETRY		0x2	/* this is a retry of a previously-waiting cmd */
+ static int try_complete_step(Step *step, int flags);
+ 
  static int step_qsort_cmp(const void *a, const void *b);
  static int step_bsearch_cmp(const void *a, const void *b);
  
***************
*** 46,51 **** main(int argc, char **argv)
--- 63,69 ----
  	const char *conninfo;
  	TestSpec   *testspec;
  	int			i;
+ 	PGresult   *res;
  
  	/*
  	 * If the user supplies a parameter on the command line, use it as the
***************
*** 63,75 **** main(int argc, char **argv)
  	testspec = &parseresult;
  	printf("Parsed test spec with %d sessions\n", testspec->nsessions);
  
! 	/* Establish connections to the database, one for each session */
! 	nconns = testspec->nsessions;
  	conns = calloc(nconns, sizeof(PGconn *));
! 	for (i = 0; i < testspec->nsessions; i++)
  	{
- 		PGresult *res;
- 
  		conns[i] = PQconnectdb(conninfo);
  		if (PQstatus(conns[i]) != CONNECTION_OK)
  		{
--- 81,95 ----
  	testspec = &parseresult;
  	printf("Parsed test spec with %d sessions\n", testspec->nsessions);
  
! 	/*
! 	 * Establish connections to the database, one for each session and an extra
! 	 * for lock wait detection and global work.
! 	 */
! 	nconns = 1 + testspec->nsessions;
  	conns = calloc(nconns, sizeof(PGconn *));
! 	backend_ids = calloc(nconns, sizeof(*backend_ids));
! 	for (i = 0; i < nconns; i++)
  	{
  		conns[i] = PQconnectdb(conninfo);
  		if (PQstatus(conns[i]) != CONNECTION_OK)
  		{
***************
*** 89,94 **** main(int argc, char **argv)
--- 109,136 ----
  			exit_nicely();
  		}
  		PQclear(res);
+ 
+ 		/* Get the backend ID for lock wait checking. */
+ 		res = PQexec(conns[i], "SELECT i FROM pg_stat_get_backend_idset() t(i) "
+ 					 "WHERE pg_stat_get_backend_pid(i) = pg_backend_pid()");
+ 		if (PQresultStatus(res) == PGRES_TUPLES_OK)
+ 		{
+ 			if (PQntuples(res) == 1 && PQnfields(res) == 1)
+ 				backend_ids[i] = strdup(PQgetvalue(res, 0, 0));
+ 			else
+ 			{
+ 				fprintf(stderr, "backend id query returned %d rows and %d columns, expected 1 row and 1 column",
+ 						PQntuples(res), PQnfields(res));
+ 				exit_nicely();
+ 			}
+ 		}
+ 		else
+ 		{
+ 			fprintf(stderr, "backend id query failed: %s",
+ 					PQerrorMessage(conns[i]));
+ 			exit_nicely();
+ 		}
+ 		PQclear(res);
  	}
  
  	/* Set the session index fields in steps. */
***************
*** 100,105 **** main(int argc, char **argv)
--- 142,157 ----
  			session->steps[stepindex]->session = i;
  	}
  
+ 	res = PQprepare(conns[0], PREP_WAITING,
+ 					"SELECT 1 WHERE pg_stat_get_backend_waiting($1)", 0, NULL);
+ 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ 	{
+ 		fprintf(stderr, "prepare of lock wait query failed: %s",
+ 				PQerrorMessage(conns[0]));
+ 		exit_nicely();
+ 	}
+ 	PQclear(res);
+ 
  	/*
  	 * Run the permutations specified in the spec, or all if none were
  	 * explicitly specified.
***************
*** 254,259 **** run_permutation(TestSpec *testspec, int nsteps, Step **steps)
--- 306,312 ----
  {
  	PGresult *res;
  	int i;
+ 	Step	   *waiting = NULL;
  
  	printf("\nstarting permutation:");
  	for (i = 0; i < nsteps; i++)
***************
*** 277,288 **** run_permutation(TestSpec *testspec, int nsteps, Step **steps)
  	{
  		if (testspec->sessions[i]->setupsql)
  		{
! 			res = PQexec(conns[i], testspec->sessions[i]->setupsql);
  			if (PQresultStatus(res) != PGRES_COMMAND_OK)
  			{
  				fprintf(stderr, "setup of session %s failed: %s",
  						testspec->sessions[i]->name,
! 						PQerrorMessage(conns[0]));
  				exit_nicely();
  			}
  			PQclear(res);
--- 330,341 ----
  	{
  		if (testspec->sessions[i]->setupsql)
  		{
! 			res = PQexec(conns[i+1], testspec->sessions[i]->setupsql);
  			if (PQresultStatus(res) != PGRES_COMMAND_OK)
  			{
  				fprintf(stderr, "setup of session %s failed: %s",
  						testspec->sessions[i]->name,
! 						PQerrorMessage(conns[i+1]));
  				exit_nicely();
  			}
  			PQclear(res);
***************
*** 293,334 **** run_permutation(TestSpec *testspec, int nsteps, Step **steps)
  	for (i = 0; i < nsteps; i++)
  	{
  		Step *step = steps[i];
- 		printf("step %s: %s\n", step->name, step->sql);
- 		res = PQexec(conns[step->session], step->sql);
  
! 		switch(PQresultStatus(res))
  		{
! 			case PGRES_COMMAND_OK:
! 				break;
! 
! 			case PGRES_TUPLES_OK:
! 				printResultSet(res);
! 				break;
  
! 			case PGRES_FATAL_ERROR:
! 				/* Detail may contain xid values, so just show primary. */
! 				printf("%s:  %s\n", PQresultErrorField(res, PG_DIAG_SEVERITY),
! 					   PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY));
! 				break;
  
! 			default:
! 				printf("unexpected result status: %s\n",
! 					   PQresStatus(PQresultStatus(res)));
  		}
! 		PQclear(res);
  	}
  
  	/* Perform per-session teardown */
  	for (i = 0; i < testspec->nsessions; i++)
  	{
  		if (testspec->sessions[i]->teardownsql)
  		{
! 			res = PQexec(conns[i], testspec->sessions[i]->teardownsql);
  			if (PQresultStatus(res) != PGRES_COMMAND_OK)
  			{
  				fprintf(stderr, "teardown of session %s failed: %s",
  						testspec->sessions[i]->name,
! 						PQerrorMessage(conns[0]));
  				/* don't exit on teardown failure */
  			}
  			PQclear(res);
--- 346,387 ----
  	for (i = 0; i < nsteps; i++)
  	{
  		Step *step = steps[i];
  
! 		if (!PQsendQuery(conns[1 + step->session], step->sql))
  		{
! 			fprintf(stdout, "failed to send query: %s\n",
! 					PQerrorMessage(conns[1 + step->session]));
! 			exit_nicely();
! 		}
  
! 		if (waiting != NULL)
! 		{
! 			/* Some other step is already waiting: just block. */
! 			try_complete_step(step, 0);
  
! 			/* See if this step unblocked the waiting step. */
! 			if (try_complete_step(waiting, STEP_NONBLOCK | STEP_RETRY))
! 				waiting = NULL;
  		}
! 		else if (!try_complete_step(step, STEP_NONBLOCK))
! 			waiting = step;
  	}
  
+ 	/* Finish any waiting query. */
+ 	if (waiting != NULL)
+ 		try_complete_step(waiting, STEP_RETRY);
+ 
  	/* Perform per-session teardown */
  	for (i = 0; i < testspec->nsessions; i++)
  	{
  		if (testspec->sessions[i]->teardownsql)
  		{
! 			res = PQexec(conns[i+1], testspec->sessions[i]->teardownsql);
  			if (PQresultStatus(res) != PGRES_COMMAND_OK)
  			{
  				fprintf(stderr, "teardown of session %s failed: %s",
  						testspec->sessions[i]->name,
! 						PQerrorMessage(conns[i+1]));
  				/* don't exit on teardown failure */
  			}
  			PQclear(res);
***************
*** 350,355 **** run_permutation(TestSpec *testspec, int nsteps, Step **steps)
--- 403,507 ----
  	}
  }
  
+ /*
+  * Our caller already sent the query associated with this step.  Wait for it to
+  * either complete or (only when given the STEP_NONBLOCK flag) to block while
+  * waiting for a lock.  We assume that any lock wait will persist until we have
+  * executed additional steps in the permutation.  This is not fully robust -- a
+  * concurrent autovacuum could briefly take a lock with which we conflict.  The
+  * risk may be low enough to discount.
+  *
+  * When calling this function on behalf of a given step for a second or later
+  * time, pass the STEP_RETRY flag.  This only affects the messages printed.
+  *
+  * If the STEP_NONBLOCK flag was specified and the query is waiting to acquire a
+  * lock, returns 0.  Otherwise, returns 1.
+  */
+ static int
+ try_complete_step(Step *step, int flags)
+ {
+ 	PGconn	   *conn = conns[1 + step->session];
+ 	fd_set		read_set;
+ 	struct timeval timeout;
+ 	int			sock = PQsocket(conn);
+ 	int			ret;
+ 	PGresult   *res;
+ 
+ 	FD_ZERO(&read_set);
+ 
+ 	while (flags & STEP_NONBLOCK && PQisBusy(conn))
+ 	{
+ 		FD_SET(sock, &read_set);
+ 		timeout.tv_sec = 0;
+ 		timeout.tv_usec = 10000;	/* Check for lock waits every 10ms. */
+ 
+ 		ret = select(sock + 1, &read_set, NULL, NULL, &timeout);
+ 		if (ret < 0)	/* error in select() */
+ 		{
+ 			fprintf(stderr, "select failed: %s\n", strerror(errno));
+ 			exit_nicely();
+ 		}
+ 		else if (ret == 0)	/* select() timeout: check for lock wait */
+ 		{
+ 			int			ntuples;
+ 
+ 			res = PQexecPrepared(conns[0], PREP_WAITING, 1,
+ 								 &backend_ids[step->session + 1],
+ 								 NULL, NULL, 0);
+ 			if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ 			{
+ 				fprintf(stderr, "lock wait query failed: %s",
+ 						PQerrorMessage(conn));
+ 				exit_nicely();
+ 			}
+ 			ntuples = PQntuples(res);
+ 			PQclear(res);
+ 
+ 			if (ntuples >= 1)	/* waiting to acquire a lock */
+ 			{
+ 				if (!(flags & STEP_RETRY))
+ 					printf("step %s: %s <waiting ...>\n",
+ 						   step->name, step->sql);
+ 				return 0;
+ 			}
+ 			/* else, not waiting: give it more time */
+ 		}
+ 		else if (!PQconsumeInput(conn)) /* select(): data available */
+ 		{
+ 			fprintf(stderr, "PQconsumeInput failed: %s", PQerrorMessage(conn));
+ 			exit_nicely();
+ 		}
+ 	}
+ 
+ 	if (flags & STEP_RETRY)
+ 		printf("step %s: <... completed>\n", step->name);
+ 	else
+ 		printf("step %s: %s\n", step->name, step->sql);
+ 
+ 	while ((res = PQgetResult(conn)))
+ 	{
+ 		switch (PQresultStatus(res))
+ 		{
+ 			case PGRES_COMMAND_OK:
+ 				break;
+ 			case PGRES_TUPLES_OK:
+ 				printResultSet(res);
+ 				break;
+ 			case PGRES_FATAL_ERROR:
+ 				/* Detail may contain xid values, so just show primary. */
+ 				printf("%s:  %s\n", PQresultErrorField(res, PG_DIAG_SEVERITY),
+ 					   PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY));
+ 				break;
+ 			default:
+ 				printf("unexpected result status: %s\n",
+ 					   PQresStatus(PQresultStatus(res)));
+ 		}
+ 		PQclear(res);
+ 	}
+ 
+ 	return 1;
+ }
+ 
  static void
  printResultSet(PGresult *res)
  {
*** /dev/null
--- b/src/test/isolation/specs/fk-contention.spec
***************
*** 0 ****
--- 1,19 ----
+ setup
+ {
+   CREATE TABLE foo (a int PRIMARY KEY, b text);
+   CREATE TABLE bar (a int NOT NULL REFERENCES foo);
+   INSERT INTO foo VALUES (42);
+ }
+ 
+ teardown
+ {
+   DROP TABLE foo, bar;
+ }
+ 
+ session "s1"
+ setup		{ BEGIN; }
+ step "ins"	{ INSERT INTO bar VALUES (42); }
+ step "com"	{ COMMIT; }
+ 
+ session "s2"
+ step "upd"	{ UPDATE foo SET b = 'Hello World'; }
*** /dev/null
--- b/src/test/isolation/specs/fk-deadlock.spec
***************
*** 0 ****
--- 1,54 ----
+ setup
+ {
+   CREATE TABLE parent (
+ 	parent_key	int		PRIMARY KEY,
+ 	aux			text	NOT NULL
+   );
+ 
+   CREATE TABLE child (
+ 	child_key	int		PRIMARY KEY,
+ 	parent_key	int		NOT NULL REFERENCES parent
+   );
+ 
+   INSERT INTO parent VALUES (1, 'foo');
+ }
+ 
+ teardown
+ {
+   DROP TABLE parent, child;
+ }
+ 
+ session "s1"
+ setup		{ BEGIN; }
+ step "s1i"	{ INSERT INTO child VALUES (1, 1); }
+ step "s1u"	{ UPDATE parent SET aux = 'bar'; }
+ step "s1c"	{ COMMIT; }
+ 
+ session "s2"
+ setup		{ BEGIN; }
+ step "s2i"	{ INSERT INTO child VALUES (2, 1); }
+ step "s2u"	{ UPDATE parent SET aux = 'baz'; }
+ step "s2c"	{ COMMIT; }
+ 
+ ## Most theoretical permutations require that a blocked session execute a
+ ## command, making them impossible in practice.
+ permutation "s1i" "s1u" "s1c" "s2i" "s2u" "s2c"
+ permutation "s1i" "s1u" "s2i" "s1c" "s2u" "s2c"
+ #permutation "s1i" "s1u" "s2i" "s2u" "s1c" "s2c"
+ #permutation "s1i" "s1u" "s2i" "s2u" "s2c" "s1c"
+ #permutation "s1i" "s2i" "s1u" "s1c" "s2u" "s2c"
+ permutation "s1i" "s2i" "s1u" "s2u" "s1c" "s2c"
+ #permutation "s1i" "s2i" "s1u" "s2u" "s2c" "s1c"
+ #permutation "s1i" "s2i" "s2u" "s1u" "s1c" "s2c"
+ permutation "s1i" "s2i" "s2u" "s1u" "s2c" "s1c"
+ #permutation "s1i" "s2i" "s2u" "s2c" "s1u" "s1c"
+ #permutation "s2i" "s1i" "s1u" "s1c" "s2u" "s2c"
+ permutation "s2i" "s1i" "s1u" "s2u" "s1c" "s2c"
+ #permutation "s2i" "s1i" "s1u" "s2u" "s2c" "s1c"
+ #permutation "s2i" "s1i" "s2u" "s1u" "s1c" "s2c"
+ permutation "s2i" "s1i" "s2u" "s1u" "s2c" "s1c"
+ #permutation "s2i" "s1i" "s2u" "s2c" "s1u" "s1c"
+ #permutation "s2i" "s2u" "s1i" "s1u" "s1c" "s2c"
+ #permutation "s2i" "s2u" "s1i" "s1u" "s2c" "s1c"
+ permutation "s2i" "s2u" "s1i" "s2c" "s1u" "s1c"
+ #permutation "s2i" "s2u" "s2c" "s1i" "s1u" "s1c"
*** /dev/null
--- b/src/test/isolation/specs/fk-deadlock2.spec
***************
*** 0 ****
--- 1,59 ----
+ setup
+ {
+   CREATE TABLE A (
+ 	AID integer not null,
+ 	Col1 integer,
+ 	PRIMARY KEY (AID)
+   );
+ 
+   CREATE TABLE B (
+ 	BID integer not null,
+ 	AID integer not null,
+ 	Col2 integer,
+ 	PRIMARY KEY (BID),
+ 	FOREIGN KEY (AID) REFERENCES A(AID)
+   );
+ 
+   INSERT INTO A (AID) VALUES (1);
+   INSERT INTO B (BID,AID) VALUES (2,1);
+ }
+ 
+ teardown
+ {
+   DROP TABLE a, b;
+ }
+ 
+ session "s1"
+ setup		{ BEGIN; }
+ step "s1u1"	{ UPDATE A SET Col1 = 1 WHERE AID = 1; }
+ step "s1u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
+ step "s1c"	{ COMMIT; }
+ 
+ session "s2"
+ setup		{ BEGIN; }
+ step "s2u1"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
+ step "s2u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
+ step "s2c"	{ COMMIT; }
+ 
+ ## Many theoretical permutations require that a blocked session execute a
+ ## command, making them impossible in practice.
+ permutation "s1u1" "s1u2" "s1c" "s2u1" "s2u2" "s2c"
+ permutation "s1u1" "s1u2" "s2u1" "s1c" "s2u2" "s2c"
+ #permutation "s1u1" "s1u2" "s2u1" "s2u2" "s1c" "s2c"
+ #permutation "s1u1" "s1u2" "s2u1" "s2u2" "s2c" "s1c"
+ #permutation "s1u1" "s2u1" "s1u2" "s1c" "s2u2" "s2c"
+ permutation "s1u1" "s2u1" "s1u2" "s2u2" "s1c" "s2c"
+ permutation "s1u1" "s2u1" "s1u2" "s2u2" "s2c" "s1c"
+ permutation "s1u1" "s2u1" "s2u2" "s1u2" "s1c" "s2c"
+ permutation "s1u1" "s2u1" "s2u2" "s1u2" "s2c" "s1c"
+ #permutation "s1u1" "s2u1" "s2u2" "s2c" "s1u2" "s1c"
+ #permutation "s2u1" "s1u1" "s1u2" "s1c" "s2u2" "s2c"
+ permutation "s2u1" "s1u1" "s1u2" "s2u2" "s1c" "s2c"
+ permutation "s2u1" "s1u1" "s1u2" "s2u2" "s2c" "s1c"
+ permutation "s2u1" "s1u1" "s2u2" "s1u2" "s1c" "s2c"
+ permutation "s2u1" "s1u1" "s2u2" "s1u2" "s2c" "s1c"
+ #permutation "s2u1" "s1u1" "s2u2" "s2c" "s1u2" "s1c"
+ #permutation "s2u1" "s2u2" "s1u1" "s1u2" "s1c" "s2c"
+ #permutation "s2u1" "s2u2" "s1u1" "s1u2" "s2c" "s1c"
+ permutation "s2u1" "s2u2" "s1u1" "s2c" "s1u2" "s1c"
+ #permutation "s2u1" "s2u2" "s2c" "s1u1" "s1u2" "s1c"

diff --git a/src/test/isolation/expected/fk-deadlock.out b/src/test/isolation/expected/fk-deadlock.out
index 6b6ee16..0d86cda 100644
*** a/src/test/isolation/expected/fk-deadlock.out
--- b/src/test/isolation/expected/fk-deadlock.out
***************
*** 32,39 **** step s1i:  INSERT INTO child VALUES (1, 1);
  step s2i:  INSERT INTO child VALUES (2, 1); 
  step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
  step s1u:  UPDATE parent SET aux = 'bar'; 
- step s2u: <... completed>
  ERROR:  deadlock detected
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
--- 32,39 ----
  step s2i:  INSERT INTO child VALUES (2, 1); 
  step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
  step s1u:  UPDATE parent SET aux = 'bar'; 
  ERROR:  deadlock detected
+ step s2u: <... completed>
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
***************
*** 52,59 **** step s2i:  INSERT INTO child VALUES (2, 1);
  step s1i:  INSERT INTO child VALUES (1, 1); 
  step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
  step s1u:  UPDATE parent SET aux = 'bar'; 
- step s2u: <... completed>
  ERROR:  deadlock detected
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
--- 52,59 ----
  step s1i:  INSERT INTO child VALUES (1, 1); 
  step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
  step s1u:  UPDATE parent SET aux = 'bar'; 
  ERROR:  deadlock detected
+ step s2u: <... completed>
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
diff --git a/src/test/isolation/expected/fk-deadloindex af3ce8e..6e7f12d 100644
*** a/src/test/isolation/expected/fk-deadlock2.out
--- b/src/test/isolation/expected/fk-deadlock2.out
***************
*** 42,49 **** step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1;
  step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
- step s2u2: <... completed>
  ERROR:  deadlock detected
  step s1c:  COMMIT; 
  step s2c:  COMMIT; 
  
--- 42,49 ----
  step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  ERROR:  deadlock detected
+ step s2u2: <... completed>
  step s1c:  COMMIT; 
  step s2c:  COMMIT; 
  
***************
*** 52,59 **** step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1;
  step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
- step s2u2: <... completed>
  ERROR:  deadlock detected
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
--- 52,59 ----
  step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  ERROR:  deadlock detected
+ step s2u2: <... completed>
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
***************
*** 82,89 **** step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2;
  step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
- step s2u2: <... completed>
  ERROR:  deadlock detected
  step s1c:  COMMIT; 
  step s2c:  COMMIT; 
  
--- 82,89 ----
  step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  ERROR:  deadlock detected
+ step s2u2: <... completed>
  step s1c:  COMMIT; 
  step s2c:  COMMIT; 
  
***************
*** 92,99 **** step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2;
  step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
- step s2u2: <... completed>
  ERROR:  deadlock detected
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
--- 92,99 ----
  step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  ERROR:  deadlock detected
+ step s2u2: <... completed>
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
diff --git a/src/test/isolation/specs/fk-deadlock.sindex 530cf10..b533d77 100644
*** a/src/test/isolation/specs/fk-deadlock.spec
--- b/src/test/isolation/specs/fk-deadlock.spec
***************
*** 19,31 **** teardown
  }
  
  session "s1"
! setup		{ BEGIN; }
  step "s1i"	{ INSERT INTO child VALUES (1, 1); }
  step "s1u"	{ UPDATE parent SET aux = 'bar'; }
  step "s1c"	{ COMMIT; }
  
  session "s2"
! setup		{ BEGIN; }
  step "s2i"	{ INSERT INTO child VALUES (2, 1); }
  step "s2u"	{ UPDATE parent SET aux = 'baz'; }
  step "s2c"	{ COMMIT; }
--- 19,31 ----
  }
  
  session "s1"
! setup		{ BEGIN; SET deadlock_timeout = '20ms'; }
  step "s1i"	{ INSERT INTO child VALUES (1, 1); }
  step "s1u"	{ UPDATE parent SET aux = 'bar'; }
  step "s1c"	{ COMMIT; }
  
  session "s2"
! setup		{ BEGIN; SET deadlock_timeout = '10s'; }
  step "s2i"	{ INSERT INTO child VALUES (2, 1); }
  step "s2u"	{ UPDATE parent SET aux = 'baz'; }
  step "s2c"	{ COMMIT; }
diff --git a/src/test/isolation/specs/fk-deadlocindex 91a87d1..5653628 100644
*** a/src/test/isolation/specs/fk-deadlock2.spec
--- b/src/test/isolation/specs/fk-deadlock2.spec
***************
*** 24,36 **** teardown
  }
  
  session "s1"
! setup		{ BEGIN; }
  step "s1u1"	{ UPDATE A SET Col1 = 1 WHERE AID = 1; }
  step "s1u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s1c"	{ COMMIT; }
  
  session "s2"
! setup		{ BEGIN; }
  step "s2u1"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s2u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s2c"	{ COMMIT; }
--- 24,36 ----
  }
  
  session "s1"
! setup		{ BEGIN; SET deadlock_timeout = '20ms'; }
  step "s1u1"	{ UPDATE A SET Col1 = 1 WHERE AID = 1; }
  step "s1u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s1c"	{ COMMIT; }
  
  session "s2"
! setup		{ BEGIN; SET deadlock_timeout = '10s'; }
  step "s2u1"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s2u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s2c"	{ COMMIT; }

diff --git a/src/test/isolation/isolationtester.c b/src/test/isolation/isolationtester.c
index 126e185..96d7f17 100644
*** a/src/test/isolation/isolationtester.c
--- b/src/test/isolation/isolationtester.c
***************
*** 21,26 ****
--- 21,27 ----
  #endif
  
  #include "libpq-fe.h"
+ #include "pqexpbuffer.h"
  
  #include "isolationtester.h"
  
***************
*** 31,37 ****
   * connections represent spec-defined sessions.
   */
  static PGconn **conns = NULL;
! static const char **backend_ids = NULL;
  static int	nconns = 0;
  
  static void run_all_permutations(TestSpec * testspec);
--- 32,38 ----
   * connections represent spec-defined sessions.
   */
  static PGconn **conns = NULL;
! static const char **backend_pids = NULL;
  static int	nconns = 0;
  
  static void run_all_permutations(TestSpec * testspec);
***************
*** 67,72 **** main(int argc, char **argv)
--- 68,74 ----
  	TestSpec   *testspec;
  	int			i;
  	PGresult   *res;
+ 	PQExpBufferData wait_query;
  
  	/*
  	 * If the user supplies a parameter on the command line, use it as the
***************
*** 89,95 **** main(int argc, char **argv)
  	 */
  	nconns = 1 + testspec->nsessions;
  	conns = calloc(nconns, sizeof(PGconn *));
! 	backend_ids = calloc(nconns, sizeof(*backend_ids));
  	for (i = 0; i < nconns; i++)
  	{
  		conns[i] = PQconnectdb(conninfo);
--- 91,97 ----
  	 */
  	nconns = 1 + testspec->nsessions;
  	conns = calloc(nconns, sizeof(PGconn *));
! 	backend_pids = calloc(nconns, sizeof(*backend_pids));
  	for (i = 0; i < nconns; i++)
  	{
  		conns[i] = PQconnectdb(conninfo);
***************
*** 112,134 **** main(int argc, char **argv)
  		}
  		PQclear(res);
  
! 		/* Get the backend ID for lock wait checking. */
! 		res = PQexec(conns[i], "SELECT i FROM pg_stat_get_backend_idset() t(i) "
! 					 "WHERE pg_stat_get_backend_pid(i) = pg_backend_pid()");
  		if (PQresultStatus(res) == PGRES_TUPLES_OK)
  		{
  			if (PQntuples(res) == 1 && PQnfields(res) == 1)
! 				backend_ids[i] = strdup(PQgetvalue(res, 0, 0));
  			else
  			{
! 				fprintf(stderr, "backend id query returned %d rows and %d columns, expected 1 row and 1 column",
  						PQntuples(res), PQnfields(res));
  				exit_nicely();
  			}
  		}
  		else
  		{
! 			fprintf(stderr, "backend id query failed: %s",
  					PQerrorMessage(conns[i]));
  			exit_nicely();
  		}
--- 114,135 ----
  		}
  		PQclear(res);
  
! 		/* Get the backend pid for lock wait checking. */
! 		res = PQexec(conns[i], "SELECT pg_backend_pid()");
  		if (PQresultStatus(res) == PGRES_TUPLES_OK)
  		{
  			if (PQntuples(res) == 1 && PQnfields(res) == 1)
! 				backend_pids[i] = strdup(PQgetvalue(res, 0, 0));
  			else
  			{
! 				fprintf(stderr, "backend pid query returned %d rows and %d columns, expected 1 row and 1 column",
  						PQntuples(res), PQnfields(res));
  				exit_nicely();
  			}
  		}
  		else
  		{
! 			fprintf(stderr, "backend pid query failed: %s",
  					PQerrorMessage(conns[i]));
  			exit_nicely();
  		}
***************
*** 145,152 **** main(int argc, char **argv)
  			session->steps[stepindex]->session = i;
  	}
  
! 	res = PQprepare(conns[0], PREP_WAITING,
! 					"SELECT 1 WHERE pg_stat_get_backend_waiting($1)", 0, NULL);
  	if (PQresultStatus(res) != PGRES_COMMAND_OK)
  	{
  		fprintf(stderr, "prepare of lock wait query failed: %s",
--- 146,232 ----
  			session->steps[stepindex]->session = i;
  	}
  
! 	/*
! 	 * Build the query we'll use to detect lock contention among sessions in
! 	 * the test specification.  Most of the time, we could get away with
! 	 * simply checking whether a session is waiting for *any* lock: we don't
! 	 * exactly expect concurrent use of test tables.  However, autovacuum will
! 	 * occasionally take AccessExclusiveLock to truncate a table, and we must
! 	 * ignore that transient wait.
! 	 */
! 	initPQExpBuffer(&wait_query);
! 	appendPQExpBufferStr(&wait_query,
! 						 "SELECT 1 FROM pg_locks holder, pg_locks waiter "
! 						 "WHERE NOT waiter.granted AND waiter.pid = $1 "
! 						 "AND holder.granted "
! 						 "AND holder.pid <> $1 AND holder.pid IN (");
! 	/* The spec syntax requires at least one session; assume that here. */
! 	appendPQExpBuffer(&wait_query, "%s", backend_pids[1]);
! 	for (i = 2; i < nconns; i++)
! 		appendPQExpBuffer(&wait_query, ", %s", backend_pids[i]);
! 	appendPQExpBufferStr(&wait_query,
! 						 ") "
! 
! 						 "AND holder.mode = ANY (CASE waiter.mode "
! 						 "WHEN 'AccessShareLock' THEN ARRAY["
! 						 "'AccessExclusiveLock'] "
! 						 "WHEN 'RowShareLock' THEN ARRAY["
! 						 "'ExclusiveLock',"
! 						 "'AccessExclusiveLock'] "
! 						 "WHEN 'RowExclusiveLock' THEN ARRAY["
! 						 "'ShareLock',"
! 						 "'ShareRowExclusiveLock',"
! 						 "'ExclusiveLock',"
! 						 "'AccessExclusiveLock'] "
! 						 "WHEN 'ShareUpdateExclusiveLock' THEN ARRAY["
! 						 "'ShareUpdateExclusiveLock',"
! 						 "'ShareLock',"
! 						 "'ShareRowExclusiveLock',"
! 						 "'ExclusiveLock',"
! 						 "'AccessExclusiveLock'] "
! 						 "WHEN 'ShareLock' THEN ARRAY["
! 						 "'RowExclusiveLock',"
! 						 "'ShareUpdateExclusiveLock',"
! 						 "'ShareRowExclusiveLock',"
! 						 "'ExclusiveLock',"
! 						 "'AccessExclusiveLock'] "
! 						 "WHEN 'ShareRowExclusiveLock' THEN ARRAY["
! 						 "'RowExclusiveLock',"
! 						 "'ShareUpdateExclusiveLock',"
! 						 "'ShareLock',"
! 						 "'ShareRowExclusiveLock',"
! 						 "'ExclusiveLock',"
! 						 "'AccessExclusiveLock'] "
! 						 "WHEN 'ExclusiveLock' THEN ARRAY["
! 						 "'RowShareLock',"
! 						 "'RowExclusiveLock',"
! 						 "'ShareUpdateExclusiveLock',"
! 						 "'ShareLock',"
! 						 "'ShareRowExclusiveLock',"
! 						 "'ExclusiveLock',"
! 						 "'AccessExclusiveLock'] "
! 						 "WHEN 'AccessExclusiveLock' THEN ARRAY["
! 						 "'AccessShareLock',"
! 						 "'RowShareLock',"
! 						 "'RowExclusiveLock',"
! 						 "'ShareUpdateExclusiveLock',"
! 						 "'ShareLock',"
! 						 "'ShareRowExclusiveLock',"
! 						 "'ExclusiveLock',"
! 						 "'AccessExclusiveLock'] END) "
! 
! 						 "AND holder.locktype IS NOT DISTINCT FROM waiter.locktype "
! 						 "AND holder.database IS NOT DISTINCT FROM waiter.database "
! 						 "AND holder.relation IS NOT DISTINCT FROM waiter.relation "
! 						 "AND holder.page IS NOT DISTINCT FROM waiter.page "
! 						 "AND holder.tuple IS NOT DISTINCT FROM waiter.tuple "
! 						 "AND holder.virtualxid IS NOT DISTINCT FROM waiter.virtualxid "
! 						 "AND holder.transactionid IS NOT DISTINCT FROM waiter.transactionid "
! 						 "AND holder.classid IS NOT DISTINCT FROM waiter.classid "
! 						 "AND holder.objid IS NOT DISTINCT FROM waiter.objid "
! 						 "AND holder.objsubid IS NOT DISTINCT FROM waiter.objsubid ");
! 
! 	res = PQprepare(conns[0], PREP_WAITING, wait_query.data, 0, NULL);
  	if (PQresultStatus(res) != PGRES_COMMAND_OK)
  	{
  		fprintf(stderr, "prepare of lock wait query failed: %s",
***************
*** 154,159 **** main(int argc, char **argv)
--- 234,240 ----
  		exit_nicely();
  	}
  	PQclear(res);
+ 	termPQExpBuffer(&wait_query);
  
  	/*
  	 * Run the permutations specified in the spec, or all if none were
***************
*** 411,419 **** run_permutation(TestSpec * testspec, int nsteps, Step ** steps)
   * Our caller already sent the query associated with this step.  Wait for it
   * to either complete or (if given the STEP_NONBLOCK flag) to block while
   * waiting for a lock.  We assume that any lock wait will persist until we
!  * have executed additional steps in the permutation.  This is not fully
!  * robust -- a concurrent autovacuum could briefly take a lock with which we
!  * conflict.  The risk may be low enough to discount.
   *
   * When calling this function on behalf of a given step for a second or later
   * time, pass the STEP_RETRY flag.  This only affects the messages printed.
--- 492,498 ----
   * Our caller already sent the query associated with this step.  Wait for it
   * to either complete or (if given the STEP_NONBLOCK flag) to block while
   * waiting for a lock.  We assume that any lock wait will persist until we
!  * have executed additional steps in the permutation.
   *
   * When calling this function on behalf of a given step for a second or later
   * time, pass the STEP_RETRY flag.  This only affects the messages printed.
***************
*** 450,456 **** try_complete_step(Step *step, int flags)
  			int			ntuples;
  
  			res = PQexecPrepared(conns[0], PREP_WAITING, 1,
! 								 &backend_ids[step->session + 1],
  								 NULL, NULL, 0);
  			if (PQresultStatus(res) != PGRES_TUPLES_OK)
  			{
--- 529,535 ----
  			int			ntuples;
  
  			res = PQexecPrepared(conns[0], PREP_WAITING, 1,
! 								 &backend_pids[step->session + 1],
  								 NULL, NULL, 0);
  			if (PQresultStatus(res) != PGRES_TUPLES_OK)
  			{
diff --git a/src/test/isolation/specs/fk-deindex b533d77..9f46c6b 100644
*** a/src/test/isolation/specs/fk-deadlock.spec
--- b/src/test/isolation/specs/fk-deadlock.spec
***************
*** 19,25 **** teardown
  }
  
  session "s1"
! setup		{ BEGIN; SET deadlock_timeout = '20ms'; }
  step "s1i"	{ INSERT INTO child VALUES (1, 1); }
  step "s1u"	{ UPDATE parent SET aux = 'bar'; }
  step "s1c"	{ COMMIT; }
--- 19,25 ----
  }
  
  session "s1"
! setup		{ BEGIN; SET deadlock_timeout = '100ms'; }
  step "s1i"	{ INSERT INTO child VALUES (1, 1); }
  step "s1u"	{ UPDATE parent SET aux = 'bar'; }
  step "s1c"	{ COMMIT; }
diff --git a/src/test/isolation/specs/fk-deadlocindex 5653628..a8f1516 100644
*** a/src/test/isolation/specs/fk-deadlock2.spec
--- b/src/test/isolation/specs/fk-deadlock2.spec
***************
*** 24,30 **** teardown
  }
  
  session "s1"
! setup		{ BEGIN; SET deadlock_timeout = '20ms'; }
  step "s1u1"	{ UPDATE A SET Col1 = 1 WHERE AID = 1; }
  step "s1u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s1c"	{ COMMIT; }
--- 24,30 ----
  }
  
  session "s1"
! setup		{ BEGIN; SET deadlock_timeout = '100ms'; }
  step "s1u1"	{ UPDATE A SET Col1 = 1 WHERE AID = 1; }
  step "s1u2"	{ UPDATE B SET Col2 = 1 WHERE BID = 2; }
  step "s1c"	{ COMMIT; }

*** /dev/null
--- b/src/test/isolation/expected/fk-deadlock2_1.out
***************
*** 0 ****
--- 1,110 ----
+ Parsed test spec with 2 sessions
+ 
+ starting permutation: s1u1 s1u2 s1c s2u1 s2u2 s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1c:  COMMIT; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s1u2 s2u1 s1c s2u2 s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1c:  COMMIT; 
+ step s2u1: <... completed>
+ ERROR:  could not serialize access due to concurrent update
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  current transaction is aborted, commands ignored until end of transaction block
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s1u2 s2u2 s1c s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s1u2 s2u2 s2c s1c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s2u2 s1u2 s1c s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s2u2 s1u2 s2c s1c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s1u2 s2u2 s1c s2c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s1u2 s2u2 s2c s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s2u2 s1u2 s1c s2c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s2u2 s1u2 s2c s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s2u2 s1u1 s2c s1u2 s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1;  <waiting ...>
+ step s2c:  COMMIT; 
+ step s1u1: <... completed>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  could not serialize access due to read/write dependencies among transactions
+ step s1c:  COMMIT; 
*** /dev/null
--- b/src/test/isolation/expected/fk-deadlock2_2.out
***************
*** 0 ****
--- 1,110 ----
+ Parsed test spec with 2 sessions
+ 
+ starting permutation: s1u1 s1u2 s1c s2u1 s2u2 s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1c:  COMMIT; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s1u2 s2u1 s1c s2u2 s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1c:  COMMIT; 
+ step s2u1: <... completed>
+ ERROR:  could not serialize access due to concurrent update
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  current transaction is aborted, commands ignored until end of transaction block
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s1u2 s2u2 s1c s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s1u2 s2u2 s2c s1c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s2u2 s1u2 s1c s2c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1u1 s2u1 s2u2 s1u2 s2c s1c
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s1u2 s2u2 s1c s2c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s1u2 s2u2 s2c s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u2: <... completed>
+ ERROR:  deadlock detected
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s2u2 s1u2 s1c s2c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2u1 s1u1 s2u2 s1u2 s2c s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2;  <waiting ...>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  deadlock detected
+ step s2u2: <... completed>
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2u1 s2u2 s1u1 s2c s1u2 s1c
+ step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1;  <waiting ...>
+ step s2c:  COMMIT; 
+ step s1u1: <... completed>
+ step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
+ ERROR:  could not serialize access due to concurrent update
+ step s1c:  COMMIT; 
*** /dev/null
--- b/src/test/isolation/expected/fk-deadlock_1.out
***************
*** 0 ****
--- 1,71 ----
+ Parsed test spec with 2 sessions
+ 
+ starting permutation: s1i s1u s1c s2i s2u s2c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ step s1c:  COMMIT; 
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1i s1u s2i s1c s2u s2c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ step s2i:  INSERT INTO child VALUES (2, 1);  <waiting ...>
+ step s1c:  COMMIT; 
+ step s2i: <... completed>
+ ERROR:  could not serialize access due to concurrent update
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ ERROR:  current transaction is aborted, commands ignored until end of transaction block
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1i s2i s1u s2u s1c s2c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s1u: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s1i s2i s2u s1u s2c s1c
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ ERROR:  deadlock detected
+ step s2u: <... completed>
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2i s1i s1u s2u s1c s2c
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s1u: <... completed>
+ ERROR:  deadlock detected
+ step s1c:  COMMIT; 
+ step s2c:  COMMIT; 
+ 
+ starting permutation: s2i s1i s2u s1u s2c s1c
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s1i:  INSERT INTO child VALUES (1, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ ERROR:  deadlock detected
+ step s2u: <... completed>
+ step s2c:  COMMIT; 
+ step s1c:  COMMIT; 
+ 
+ starting permutation: s2i s2u s1i s2c s1u s1c
+ step s2i:  INSERT INTO child VALUES (2, 1); 
+ step s2u:  UPDATE parent SET aux = 'baz'; 
+ step s1i:  INSERT INTO child VALUES (1, 1);  <waiting ...>
+ step s2c:  COMMIT; 
+ step s1i: <... completed>
+ ERROR:  could not serialize access due to concurrent update
+ step s1u:  UPDATE parent SET aux = 'bar'; 
+ ERROR:  current transaction is aborted, commands ignored until end of transaction block
+ step s1c:  COMMIT; 

*** a/src/backend/access/heap/heapam.c
--- b/src/backend/access/heap/heapam.c
***************
*** 2450,2455 **** heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
--- 2450,2456 ----
  	HTSU_Result result;
  	TransactionId xid = GetCurrentTransactionId();
  	Bitmapset  *hot_attrs;
+ 	Bitmapset  *keylck_attrs;
  	ItemId		lp;
  	HeapTupleData oldtup;
  	HeapTuple	heaptup;
***************
*** 2466,2471 **** heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
--- 2467,2473 ----
  	bool		have_tuple_lock = false;
  	bool		iscombo;
  	bool		use_hot_update = false;
+ 	bool		keylocked_update = false;
  	bool		all_visible_cleared = false;
  	bool		all_visible_cleared_new = false;
  
***************
*** 2483,2489 **** heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
  	 * Note that we get a copy here, so we need not worry about relcache flush
  	 * happening midway through.
  	 */
! 	hot_attrs = RelationGetIndexAttrBitmap(relation);
  
  	block = ItemPointerGetBlockNumber(otid);
  	buffer = ReadBuffer(relation, block);
--- 2485,2492 ----
  	 * Note that we get a copy here, so we need not worry about relcache flush
  	 * happening midway through.
  	 */
! 	hot_attrs = RelationGetIndexAttrBitmap(relation, false);
! 	keylck_attrs = RelationGetIndexAttrBitmap(relation, true);
  
  	block = ItemPointerGetBlockNumber(otid);
  	buffer = ReadBuffer(relation, block);
***************
*** 2524,2614 **** l2:
  	}
  	else if (result == HeapTupleBeingUpdated && wait)
  	{
- 		TransactionId xwait;
  		uint16		infomask;
  
- 		/* must copy state data before unlocking buffer */
- 		xwait = HeapTupleHeaderGetXmax(oldtup.t_data);
  		infomask = oldtup.t_data->t_infomask;
  
- 		LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
- 
  		/*
! 		 * Acquire tuple lock to establish our priority for the tuple (see
! 		 * heap_lock_tuple).  LockTuple will release us when we are
! 		 * next-in-line for the tuple.
! 		 *
! 		 * If we are forced to "start over" below, we keep the tuple lock;
! 		 * this arranges that we stay at the head of the line while rechecking
! 		 * tuple state.
  		 */
! 		if (!have_tuple_lock)
  		{
! 			LockTuple(relation, &(oldtup.t_self), ExclusiveLock);
! 			have_tuple_lock = true;
  		}
  
! 		/*
! 		 * Sleep until concurrent transaction ends.  Note that we don't care
! 		 * if the locker has an exclusive or shared lock, because we need
! 		 * exclusive.
! 		 */
! 
! 		if (infomask & HEAP_XMAX_IS_MULTI)
  		{
! 			/* wait for multixact */
! 			MultiXactIdWait((MultiXactId) xwait);
! 			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
  
  			/*
! 			 * If xwait had just locked the tuple then some other xact could
! 			 * update this tuple before we get to this point.  Check for xmax
! 			 * change, and start over if so.
  			 */
! 			if (!(oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 				!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 									 xwait))
! 				goto l2;
  
  			/*
! 			 * You might think the multixact is necessarily done here, but not
! 			 * so: it could have surviving members, namely our own xact or
! 			 * other subxacts of this backend.	It is legal for us to update
! 			 * the tuple in either case, however (the latter case is
! 			 * essentially a situation of upgrading our former shared lock to
! 			 * exclusive).	We don't bother changing the on-disk hint bits
! 			 * since we are about to overwrite the xmax altogether.
  			 */
! 		}
! 		else
! 		{
! 			/* wait for regular transaction to end */
! 			XactLockTableWait(xwait);
! 			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
  
  			/*
! 			 * xwait is done, but if xwait had just locked the tuple then some
! 			 * other xact could update this tuple before we get to this point.
! 			 * Check for xmax change, and start over if so.
  			 */
! 			if ((oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 				!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 									 xwait))
! 				goto l2;
! 
! 			/* Otherwise check if it committed or aborted */
! 			UpdateXmaxHintBits(oldtup.t_data, buffer, xwait);
  		}
- 
- 		/*
- 		 * We may overwrite if previous xmax aborted, or if it committed but
- 		 * only locked the tuple without updating it.
- 		 */
- 		if (oldtup.t_data->t_infomask & (HEAP_XMAX_INVALID |
- 										 HEAP_IS_LOCKED))
- 			result = HeapTupleMayBeUpdated;
- 		else
- 			result = HeapTupleUpdated;
  	}
  
  	if (crosscheck != InvalidSnapshot && result == HeapTupleMayBeUpdated)
--- 2527,2636 ----
  	}
  	else if (result == HeapTupleBeingUpdated && wait)
  	{
  		uint16		infomask;
  
  		infomask = oldtup.t_data->t_infomask;
  
  		/*
! 		 * if it's only key-locked and we're not updating an indexed column,
! 		 * we can act though MayBeUpdated was returned, but the resulting tuple
! 		 * needs a bunch of fields copied from the original.
  		 */
! 		if ((infomask & HEAP_XMAX_KEY_LOCK) &&
! 			!(infomask & HEAP_XMAX_SHARED_LOCK) &&
! 			HeapSatisfiesHOTUpdate(relation, keylck_attrs,
! 								   &oldtup, newtup))
  		{
! 			result = HeapTupleMayBeUpdated;
! 			keylocked_update = true;
  		}
  
! 		if (!keylocked_update)
  		{
! 			TransactionId xwait;
! 
! 			/* must copy state data before unlocking buffer */
! 			xwait = HeapTupleHeaderGetXmax(oldtup.t_data);
! 
! 			LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
  
  			/*
! 			 * Acquire tuple lock to establish our priority for the tuple (see
! 			 * heap_lock_tuple).  LockTuple will release us when we are
! 			 * next-in-line for the tuple.
! 			 *
! 			 * If we are forced to "start over" below, we keep the tuple lock;
! 			 * this arranges that we stay at the head of the line while rechecking
! 			 * tuple state.
  			 */
! 			if (!have_tuple_lock)
! 			{
! 				LockTuple(relation, &(oldtup.t_self), ExclusiveLock);
! 				have_tuple_lock = true;
! 			}
  
  			/*
! 			 * Sleep until concurrent transaction ends.  Note that we don't care
! 			 * if the locker has an exclusive or shared lock, because we need
! 			 * exclusive.
  			 */
! 
! 			if (infomask & HEAP_XMAX_IS_MULTI)
! 			{
! 				/* wait for multixact */
! 				MultiXactIdWait((MultiXactId) xwait);
! 				LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
! 
! 				/*
! 				 * If xwait had just locked the tuple then some other xact could
! 				 * update this tuple before we get to this point.  Check for xmax
! 				 * change, and start over if so.
! 				 */
! 				if (!(oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 					!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 										 xwait))
! 					goto l2;
! 
! 				/*
! 				 * You might think the multixact is necessarily done here, but not
! 				 * so: it could have surviving members, namely our own xact or
! 				 * other subxacts of this backend.	It is legal for us to update
! 				 * the tuple in either case, however (the latter case is
! 				 * essentially a situation of upgrading our former shared lock to
! 				 * exclusive).	We don't bother changing the on-disk hint bits
! 				 * since we are about to overwrite the xmax altogether.
! 				 */
! 			}
! 			else
! 			{
! 				/* wait for regular transaction to end */
! 				XactLockTableWait(xwait);
! 				LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
! 
! 				/*
! 				 * xwait is done, but if xwait had just locked the tuple then some
! 				 * other xact could update this tuple before we get to this point.
! 				 * Check for xmax change, and start over if so.
! 				 */
! 				if ((oldtup.t_data->t_infomask & HEAP_XMAX_IS_MULTI) ||
! 					!TransactionIdEquals(HeapTupleHeaderGetXmax(oldtup.t_data),
! 										 xwait))
! 					goto l2;
! 
! 				/* Otherwise check if it committed or aborted */
! 				UpdateXmaxHintBits(oldtup.t_data, buffer, xwait);
! 			}
  
  			/*
! 			 * We may overwrite if previous xmax aborted, or if it committed but
! 			 * only locked the tuple without updating it.
  			 */
! 			if (oldtup.t_data->t_infomask & (HEAP_XMAX_INVALID |
! 											 HEAP_IS_LOCKED))
! 				result = HeapTupleMayBeUpdated;
! 			else
! 				result = HeapTupleUpdated;
  		}
  	}
  
  	if (crosscheck != InvalidSnapshot && result == HeapTupleMayBeUpdated)
***************
*** 2632,2637 **** l2:
--- 2654,2660 ----
  		if (vmbuffer != InvalidBuffer)
  			ReleaseBuffer(vmbuffer);
  		bms_free(hot_attrs);
+ 		bms_free(keylck_attrs);
  		return result;
  	}
  
***************
*** 2670,2682 **** l2:
  		Assert(!(newtup->t_data->t_infomask & HEAP_HASOID));
  	}
  
  	newtup->t_data->t_infomask &= ~(HEAP_XACT_MASK);
  	newtup->t_data->t_infomask2 &= ~(HEAP2_XACT_MASK);
! 	newtup->t_data->t_infomask |= (HEAP_XMAX_INVALID | HEAP_UPDATED);
  	HeapTupleHeaderSetXmin(newtup->t_data, xid);
  	HeapTupleHeaderSetCmin(newtup->t_data, cid);
- 	HeapTupleHeaderSetXmax(newtup->t_data, 0);	/* for cleanliness */
  	newtup->t_tableOid = RelationGetRelid(relation);
  
  	/*
  	 * Replace cid with a combo cid if necessary.  Note that we already put
--- 2693,2721 ----
  		Assert(!(newtup->t_data->t_infomask & HEAP_HASOID));
  	}
  
+ 	/*
+ 	 * Prepare the new tuple with the appropriate initial values of Xmin and
+ 	 * Xmax, as well as initial infomask bits.
+ 	 */
  	newtup->t_data->t_infomask &= ~(HEAP_XACT_MASK);
  	newtup->t_data->t_infomask2 &= ~(HEAP2_XACT_MASK);
! 	newtup->t_data->t_infomask |= HEAP_UPDATED;
  	HeapTupleHeaderSetXmin(newtup->t_data, xid);
  	HeapTupleHeaderSetCmin(newtup->t_data, cid);
  	newtup->t_tableOid = RelationGetRelid(relation);
+ 	if (keylocked_update)
+ 	{
+ 		HeapTupleHeaderSetXmax(newtup->t_data,
+ 							   HeapTupleHeaderGetXmax(oldtup.t_data));
+ 		newtup->t_data->t_infomask |= (oldtup.t_data->t_infomask & 
+ 									   (HEAP_XMAX_IS_MULTI |
+ 										HEAP_XMAX_KEY_LOCK));
+ 	}
+ 	else
+ 	{
+ 		newtup->t_data->t_infomask |= HEAP_XMAX_INVALID;
+ 		HeapTupleHeaderSetXmax(newtup->t_data, 0);	/* for cleanliness */
+ 	}
  
  	/*
  	 * Replace cid with a combo cid if necessary.  Note that we already put
***************
*** 2971,2976 **** l2:
--- 3010,3016 ----
  	}
  
  	bms_free(hot_attrs);
+ 	bms_free(keylck_attrs);
  
  	return HeapTupleMayBeUpdated;
  }
***************
*** 3203,3209 **** heap_lock_tuple(Relation relation, HeapTuple tuple, Buffer *buffer,
  	LOCKMODE	tuple_lock_type;
  	bool		have_tuple_lock = false;
  
! 	tuple_lock_type = (mode == LockTupleShared) ? ShareLock : ExclusiveLock;
  
  	*buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(tid));
  	LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
--- 3243,3261 ----
  	LOCKMODE	tuple_lock_type;
  	bool		have_tuple_lock = false;
  
! 	switch (mode)
! 	{
! 		case LockTupleShared:
! 		case LockTupleKeylock:
! 			tuple_lock_type = ShareLock;
! 			break;
! 		case LockTupleExclusive:
! 			tuple_lock_type = ExclusiveLock;
! 			break;
! 		default:
! 			elog(ERROR, "invalid tuple lock mode");
! 			tuple_lock_type = 0;	/* keep compiler quiet */
! 	}
  
  	*buffer = ReadBuffer(relation, ItemPointerGetBlockNumber(tid));
  	LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
***************
*** 3242,3253 **** l3:
  		 * already.  We *must* succeed without trying to take the tuple lock,
  		 * else we will deadlock against anyone waiting to acquire exclusive
  		 * lock.  We don't need to make any state changes in this case.
  		 */
! 		if (mode == LockTupleShared &&
! 			(infomask & HEAP_XMAX_IS_MULTI) &&
  			MultiXactIdIsCurrent((MultiXactId) xwait))
  		{
- 			Assert(infomask & HEAP_XMAX_SHARED_LOCK);
  			/* Probably can't hold tuple lock here, but may as well check */
  			if (have_tuple_lock)
  				UnlockTuple(relation, tid, tuple_lock_type);
--- 3294,3312 ----
  		 * already.  We *must* succeed without trying to take the tuple lock,
  		 * else we will deadlock against anyone waiting to acquire exclusive
  		 * lock.  We don't need to make any state changes in this case.
+ 		 *
+ 		 * Likewise, if we wish to acquire a key lock, and the tuple is already
+ 		 * share- or key-locked by us, we effectively hold the lock already.
+ 		 *
+ 		 * Note we cannot do this if we're asking for share lock and the tuple
+ 		 * is only key-locked.
  		 */
! 		if ((infomask & HEAP_XMAX_IS_MULTI) &&
! 			(((mode == LockTupleShared) && (infomask & HEAP_XMAX_SHARED_LOCK)) ||
! 			 ((mode == LockTupleKeylock) &&
! 			  (infomask & (HEAP_XMAX_SHARED_LOCK | HEAP_XMAX_KEY_LOCK)))) &&
  			MultiXactIdIsCurrent((MultiXactId) xwait))
  		{
  			/* Probably can't hold tuple lock here, but may as well check */
  			if (have_tuple_lock)
  				UnlockTuple(relation, tid, tuple_lock_type);
***************
*** 3293,3298 **** l3:
--- 3352,3372 ----
  			if (!(tuple->t_data->t_infomask & HEAP_XMAX_SHARED_LOCK))
  				goto l3;
  		}
+ 		else if (mode == LockTupleKeylock &&
+ 				 (infomask & (HEAP_XMAX_SHARED_LOCK | HEAP_XMAX_KEY_LOCK)))
+ 		{
+ 			/*
+ 			 * As above: acquiring keylock when there's at least one share- or
+ 			 * key-locker already.  We need not wait for him/them to complete.
+ 			 */
+ 			LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
+ 
+ 			/*
+ 			 * Make sure it's still an appropriate lock, else start over.
+ 			 */
+ 			if (!(tuple->t_data->t_infomask & (HEAP_XMAX_SHARED_LOCK | HEAP_XMAX_KEY_LOCK)))
+ 				goto l3;
+ 		}
  		else if (infomask & HEAP_XMAX_IS_MULTI)
  		{
  			/* wait for multixact to end */
***************
*** 3400,3407 **** l3:
  	if (!(old_infomask & (HEAP_XMAX_INVALID |
  						  HEAP_XMAX_COMMITTED |
  						  HEAP_XMAX_IS_MULTI)) &&
! 		(mode == LockTupleShared ?
  		 (old_infomask & HEAP_IS_LOCKED) :
  		 (old_infomask & HEAP_XMAX_EXCL_LOCK)) &&
  		TransactionIdIsCurrentTransactionId(xmax))
  	{
--- 3474,3483 ----
  	if (!(old_infomask & (HEAP_XMAX_INVALID |
  						  HEAP_XMAX_COMMITTED |
  						  HEAP_XMAX_IS_MULTI)) &&
! 		(mode == LockTupleKeylock ?
  		 (old_infomask & HEAP_IS_LOCKED) :
+ 		 mode == LockTupleShared ?
+ 		 (old_infomask & (HEAP_XMAX_SHARED_LOCK | HEAP_XMAX_EXCL_LOCK)) :
  		 (old_infomask & HEAP_XMAX_EXCL_LOCK)) &&
  		TransactionIdIsCurrentTransactionId(xmax))
  	{
***************
*** 3425,3434 **** l3:
  									HEAP_IS_LOCKED |
  									HEAP_MOVED);
  
! 	if (mode == LockTupleShared)
  	{
  		/*
! 		 * If this is the first acquisition of a shared lock in the current
  		 * transaction, set my per-backend OldestMemberMXactId setting. We can
  		 * be certain that the transaction will never become a member of any
  		 * older MultiXactIds than that.  (We have to do this even if we end
--- 3501,3510 ----
  									HEAP_IS_LOCKED |
  									HEAP_MOVED);
  
! 	if (mode == LockTupleShared || mode == LockTupleKeylock)
  	{
  		/*
! 		 * If this is the first acquisition of a keylock or shared lock in the current
  		 * transaction, set my per-backend OldestMemberMXactId setting. We can
  		 * be certain that the transaction will never become a member of any
  		 * older MultiXactIds than that.  (We have to do this even if we end
***************
*** 3437,3443 **** l3:
  		 */
  		MultiXactIdSetOldestMember();
  
! 		new_infomask |= HEAP_XMAX_SHARED_LOCK;
  
  		/*
  		 * Check to see if we need a MultiXactId because there are multiple
--- 3513,3520 ----
  		 */
  		MultiXactIdSetOldestMember();
  
! 		new_infomask |= mode == LockTupleShared ? HEAP_XMAX_SHARED_LOCK :
! 			HEAP_XMAX_KEY_LOCK;
  
  		/*
  		 * Check to see if we need a MultiXactId because there are multiple
***************
*** 3537,3543 **** l3:
  		xlrec.target.tid = tuple->t_self;
  		xlrec.locking_xid = xid;
  		xlrec.xid_is_mxact = ((new_infomask & HEAP_XMAX_IS_MULTI) != 0);
! 		xlrec.shared_lock = (mode == LockTupleShared);
  		rdata[0].data = (char *) &xlrec;
  		rdata[0].len = SizeOfHeapLock;
  		rdata[0].buffer = InvalidBuffer;
--- 3614,3620 ----
  		xlrec.target.tid = tuple->t_self;
  		xlrec.locking_xid = xid;
  		xlrec.xid_is_mxact = ((new_infomask & HEAP_XMAX_IS_MULTI) != 0);
! 		xlrec.lock_strength = mode;
  		rdata[0].data = (char *) &xlrec;
  		rdata[0].len = SizeOfHeapLock;
  		rdata[0].buffer = InvalidBuffer;
***************
*** 4987,4996 **** heap_xlog_lock(XLogRecPtr lsn, XLogRecord *record)
  						  HEAP_MOVED);
  	if (xlrec->xid_is_mxact)
  		htup->t_infomask |= HEAP_XMAX_IS_MULTI;
! 	if (xlrec->shared_lock)
  		htup->t_infomask |= HEAP_XMAX_SHARED_LOCK;
  	else
  		htup->t_infomask |= HEAP_XMAX_EXCL_LOCK;
  	HeapTupleHeaderClearHotUpdated(htup);
  	HeapTupleHeaderSetXmax(htup, xlrec->locking_xid);
  	HeapTupleHeaderSetCmax(htup, FirstCommandId, false);
--- 5064,5078 ----
  						  HEAP_MOVED);
  	if (xlrec->xid_is_mxact)
  		htup->t_infomask |= HEAP_XMAX_IS_MULTI;
! 	if (xlrec->lock_strength == LockTupleShared)
  		htup->t_infomask |= HEAP_XMAX_SHARED_LOCK;
+ 	else if (xlrec->lock_strength == LockTupleKeylock)
+ 		htup->t_infomask |= HEAP_XMAX_KEY_LOCK;
  	else
+ 	{
+ 		Assert(xlrec->lock_strength == LockTupleExclusive);
  		htup->t_infomask |= HEAP_XMAX_EXCL_LOCK;
+ 	}
  	HeapTupleHeaderClearHotUpdated(htup);
  	HeapTupleHeaderSetXmax(htup, xlrec->locking_xid);
  	HeapTupleHeaderSetCmax(htup, FirstCommandId, false);
***************
*** 5194,5203 **** heap_desc(StringInfo buf, uint8 xl_info, char *rec)
  	{
  		xl_heap_lock *xlrec = (xl_heap_lock *) rec;
  
! 		if (xlrec->shared_lock)
  			appendStringInfo(buf, "shared_lock: ");
! 		else
  			appendStringInfo(buf, "exclusive_lock: ");
  		if (xlrec->xid_is_mxact)
  			appendStringInfo(buf, "mxid ");
  		else
--- 5276,5289 ----
  	{
  		xl_heap_lock *xlrec = (xl_heap_lock *) rec;
  
! 		if (xlrec->lock_strength == LockTupleShared)
  			appendStringInfo(buf, "shared_lock: ");
! 		else if (xlrec->lock_strength == LockTupleKeylock)
! 			appendStringInfo(buf, "key_lock: ");
! 		else if (xlrec->lock_strength == LockTupleExclusive)
  			appendStringInfo(buf, "exclusive_lock: ");
+ 		else
+ 			appendStringInfo(buf, "unknown_type_lock: ");
  		if (xlrec->xid_is_mxact)
  			appendStringInfo(buf, "mxid ");
  		else
*** a/src/backend/catalog/index.c
--- b/src/backend/catalog/index.c
***************
*** 2986,2992 **** reindex_relation(Oid relid, int flags)
  
  	/* Ensure rd_indexattr is valid; see comments for RelationSetIndexList */
  	if (is_pg_class)
! 		(void) RelationGetIndexAttrBitmap(rel);
  
  	PG_TRY();
  	{
--- 2986,2992 ----
  
  	/* Ensure rd_indexattr is valid; see comments for RelationSetIndexList */
  	if (is_pg_class)
! 		(void) RelationGetIndexAttrBitmap(rel, false);
  
  	PG_TRY();
  	{
*** a/src/backend/executor/execMain.c
--- b/src/backend/executor/execMain.c
***************
*** 801,807 **** InitPlan(QueryDesc *queryDesc, int eflags)
  	}
  
  	/*
! 	 * Similarly, we have to lock relations selected FOR UPDATE/FOR SHARE
  	 * before we initialize the plan tree, else we'd be risking lock upgrades.
  	 * While we are at it, build the ExecRowMark list.
  	 */
--- 801,807 ----
  	}
  
  	/*
! 	 * Similarly, we have to lock relations selected FOR UPDATE/FOR SHARE/KEY LOCK
  	 * before we initialize the plan tree, else we'd be risking lock upgrades.
  	 * While we are at it, build the ExecRowMark list.
  	 */
***************
*** 821,826 **** InitPlan(QueryDesc *queryDesc, int eflags)
--- 821,827 ----
  		{
  			case ROW_MARK_EXCLUSIVE:
  			case ROW_MARK_SHARE:
+ 			case ROW_MARK_KEYLOCK:
  				relid = getrelid(rc->rti, rangeTable);
  				relation = heap_open(relid, RowShareLock);
  				break;
*** a/src/backend/executor/nodeLockRows.c
--- b/src/backend/executor/nodeLockRows.c
***************
*** 111,120 **** lnext:
  		tuple.t_self = *((ItemPointer) DatumGetPointer(datum));
  
  		/* okay, try to lock the tuple */
! 		if (erm->markType == ROW_MARK_EXCLUSIVE)
! 			lockmode = LockTupleExclusive;
! 		else
! 			lockmode = LockTupleShared;
  
  		test = heap_lock_tuple(erm->relation, &tuple, &buffer,
  							   &update_ctid, &update_xmax,
--- 111,132 ----
  		tuple.t_self = *((ItemPointer) DatumGetPointer(datum));
  
  		/* okay, try to lock the tuple */
! 		switch (erm->markType)
! 		{
! 			case ROW_MARK_EXCLUSIVE:
! 				lockmode = LockTupleExclusive;
! 				break;
! 			case ROW_MARK_SHARE:
! 				lockmode = LockTupleShared;
! 				break;
! 			case ROW_MARK_KEYLOCK:
! 				lockmode = LockTupleKeylock;
! 				break;
! 			default:
! 				elog(ERROR, "unsupported rowmark type");
! 				lockmode = LockTupleExclusive;	/* keep compiler quiet */
! 				break;
! 		}
  
  		test = heap_lock_tuple(erm->relation, &tuple, &buffer,
  							   &update_ctid, &update_xmax,
*** a/src/backend/nodes/copyfuncs.c
--- b/src/backend/nodes/copyfuncs.c
***************
*** 2008,2014 **** _copyRowMarkClause(RowMarkClause *from)
  	RowMarkClause *newnode = makeNode(RowMarkClause);
  
  	COPY_SCALAR_FIELD(rti);
! 	COPY_SCALAR_FIELD(forUpdate);
  	COPY_SCALAR_FIELD(noWait);
  	COPY_SCALAR_FIELD(pushedDown);
  
--- 2008,2014 ----
  	RowMarkClause *newnode = makeNode(RowMarkClause);
  
  	COPY_SCALAR_FIELD(rti);
! 	COPY_SCALAR_FIELD(strength);
  	COPY_SCALAR_FIELD(noWait);
  	COPY_SCALAR_FIELD(pushedDown);
  
***************
*** 2366,2372 **** _copyLockingClause(LockingClause *from)
  	LockingClause *newnode = makeNode(LockingClause);
  
  	COPY_NODE_FIELD(lockedRels);
! 	COPY_SCALAR_FIELD(forUpdate);
  	COPY_SCALAR_FIELD(noWait);
  
  	return newnode;
--- 2366,2372 ----
  	LockingClause *newnode = makeNode(LockingClause);
  
  	COPY_NODE_FIELD(lockedRels);
! 	COPY_SCALAR_FIELD(strength);
  	COPY_SCALAR_FIELD(noWait);
  
  	return newnode;
*** a/src/backend/nodes/equalfuncs.c
--- b/src/backend/nodes/equalfuncs.c
***************
*** 2291,2297 **** static bool
  _equalLockingClause(LockingClause *a, LockingClause *b)
  {
  	COMPARE_NODE_FIELD(lockedRels);
! 	COMPARE_SCALAR_FIELD(forUpdate);
  	COMPARE_SCALAR_FIELD(noWait);
  
  	return true;
--- 2291,2297 ----
  _equalLockingClause(LockingClause *a, LockingClause *b)
  {
  	COMPARE_NODE_FIELD(lockedRels);
! 	COMPARE_SCALAR_FIELD(strength);
  	COMPARE_SCALAR_FIELD(noWait);
  
  	return true;
***************
*** 2362,2368 **** static bool
  _equalRowMarkClause(RowMarkClause *a, RowMarkClause *b)
  {
  	COMPARE_SCALAR_FIELD(rti);
! 	COMPARE_SCALAR_FIELD(forUpdate);
  	COMPARE_SCALAR_FIELD(noWait);
  	COMPARE_SCALAR_FIELD(pushedDown);
  
--- 2362,2368 ----
  _equalRowMarkClause(RowMarkClause *a, RowMarkClause *b)
  {
  	COMPARE_SCALAR_FIELD(rti);
! 	COMPARE_SCALAR_FIELD(strength);
  	COMPARE_SCALAR_FIELD(noWait);
  	COMPARE_SCALAR_FIELD(pushedDown);
  
*** a/src/backend/nodes/outfuncs.c
--- b/src/backend/nodes/outfuncs.c
***************
*** 2070,2076 **** _outLockingClause(StringInfo str, LockingClause *node)
  	WRITE_NODE_TYPE("LOCKINGCLAUSE");
  
  	WRITE_NODE_FIELD(lockedRels);
! 	WRITE_BOOL_FIELD(forUpdate);
  	WRITE_BOOL_FIELD(noWait);
  }
  
--- 2070,2076 ----
  	WRITE_NODE_TYPE("LOCKINGCLAUSE");
  
  	WRITE_NODE_FIELD(lockedRels);
! 	WRITE_ENUM_FIELD(strength, LockClauseStrength);
  	WRITE_BOOL_FIELD(noWait);
  }
  
***************
*** 2247,2253 **** _outRowMarkClause(StringInfo str, RowMarkClause *node)
  	WRITE_NODE_TYPE("ROWMARKCLAUSE");
  
  	WRITE_UINT_FIELD(rti);
! 	WRITE_BOOL_FIELD(forUpdate);
  	WRITE_BOOL_FIELD(noWait);
  	WRITE_BOOL_FIELD(pushedDown);
  }
--- 2247,2253 ----
  	WRITE_NODE_TYPE("ROWMARKCLAUSE");
  
  	WRITE_UINT_FIELD(rti);
! 	WRITE_ENUM_FIELD(strength, LockClauseStrength);
  	WRITE_BOOL_FIELD(noWait);
  	WRITE_BOOL_FIELD(pushedDown);
  }
*** a/src/backend/nodes/readfuncs.c
--- b/src/backend/nodes/readfuncs.c
***************
*** 301,307 **** _readRowMarkClause(void)
  	READ_LOCALS(RowMarkClause);
  
  	READ_UINT_FIELD(rti);
! 	READ_BOOL_FIELD(forUpdate);
  	READ_BOOL_FIELD(noWait);
  	READ_BOOL_FIELD(pushedDown);
  
--- 301,307 ----
  	READ_LOCALS(RowMarkClause);
  
  	READ_UINT_FIELD(rti);
! 	READ_ENUM_FIELD(strength, LockClauseStrength);
  	READ_BOOL_FIELD(noWait);
  	READ_BOOL_FIELD(pushedDown);
  
*** a/src/backend/optimizer/plan/initsplan.c
--- b/src/backend/optimizer/plan/initsplan.c
***************
*** 563,573 **** make_outerjoininfo(PlannerInfo *root,
  	Assert(jointype != JOIN_RIGHT);
  
  	/*
! 	 * Presently the executor cannot support FOR UPDATE/SHARE marking of rels
  	 * appearing on the nullable side of an outer join. (It's somewhat unclear
  	 * what that would mean, anyway: what should we mark when a result row is
  	 * generated from no element of the nullable relation?)  So, complain if
! 	 * any nullable rel is FOR UPDATE/SHARE.
  	 *
  	 * You might be wondering why this test isn't made far upstream in the
  	 * parser.	It's because the parser hasn't got enough info --- consider
--- 563,573 ----
  	Assert(jointype != JOIN_RIGHT);
  
  	/*
! 	 * Presently the executor cannot support FOR UPDATE/SHARE/KEY LOCK marking of rels
  	 * appearing on the nullable side of an outer join. (It's somewhat unclear
  	 * what that would mean, anyway: what should we mark when a result row is
  	 * generated from no element of the nullable relation?)  So, complain if
! 	 * any nullable rel is FOR UPDATE/SHARE/KEY LOCK.
  	 *
  	 * You might be wondering why this test isn't made far upstream in the
  	 * parser.	It's because the parser hasn't got enough info --- consider
***************
*** 585,591 **** make_outerjoininfo(PlannerInfo *root,
  			(jointype == JOIN_FULL && bms_is_member(rc->rti, left_rels)))
  			ereport(ERROR,
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 					 errmsg("SELECT FOR UPDATE/SHARE cannot be applied to the nullable side of an outer join")));
  	}
  
  	sjinfo->syn_lefthand = left_rels;
--- 585,591 ----
  			(jointype == JOIN_FULL && bms_is_member(rc->rti, left_rels)))
  			ereport(ERROR,
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 					 errmsg("SELECT FOR UPDATE/SHARE/KEY LOCK cannot be applied to the nullable side of an outer join")));
  	}
  
  	sjinfo->syn_lefthand = left_rels;
*** a/src/backend/optimizer/plan/planner.c
--- b/src/backend/optimizer/plan/planner.c
***************
*** 1837,1843 **** preprocess_rowmarks(PlannerInfo *root)
  	if (parse->rowMarks)
  	{
  		/*
! 		 * We've got trouble if FOR UPDATE/SHARE appears inside grouping,
  		 * since grouping renders a reference to individual tuple CTIDs
  		 * invalid.  This is also checked at parse time, but that's
  		 * insufficient because of rule substitution, query pullup, etc.
--- 1837,1843 ----
  	if (parse->rowMarks)
  	{
  		/*
! 		 * We've got trouble if FOR UPDATE/SHARE/KEY LOCK appears inside grouping,
  		 * since grouping renders a reference to individual tuple CTIDs
  		 * invalid.  This is also checked at parse time, but that's
  		 * insufficient because of rule substitution, query pullup, etc.
***************
*** 1847,1853 **** preprocess_rowmarks(PlannerInfo *root)
  	else
  	{
  		/*
! 		 * We only need rowmarks for UPDATE, DELETE, or FOR UPDATE/SHARE.
  		 */
  		if (parse->commandType != CMD_UPDATE &&
  			parse->commandType != CMD_DELETE)
--- 1847,1853 ----
  	else
  	{
  		/*
! 		 * We only need rowmarks for UPDATE, DELETE, or FOR UPDATE/SHARE/KEY LOCK.
  		 */
  		if (parse->commandType != CMD_UPDATE &&
  			parse->commandType != CMD_DELETE)
***************
*** 1857,1863 **** preprocess_rowmarks(PlannerInfo *root)
  	/*
  	 * We need to have rowmarks for all base relations except the target. We
  	 * make a bitmapset of all base rels and then remove the items we don't
! 	 * need or have FOR UPDATE/SHARE marks for.
  	 */
  	rels = get_base_rel_indexes((Node *) parse->jointree);
  	if (parse->resultRelation)
--- 1857,1863 ----
  	/*
  	 * We need to have rowmarks for all base relations except the target. We
  	 * make a bitmapset of all base rels and then remove the items we don't
! 	 * need or have FOR UPDATE/SHARE/KEY LOCK marks for.
  	 */
  	rels = get_base_rel_indexes((Node *) parse->jointree);
  	if (parse->resultRelation)
***************
*** 1894,1903 **** preprocess_rowmarks(PlannerInfo *root)
  		newrc = makeNode(PlanRowMark);
  		newrc->rti = newrc->prti = rc->rti;
  		newrc->rowmarkId = ++(root->glob->lastRowMarkId);
! 		if (rc->forUpdate)
! 			newrc->markType = ROW_MARK_EXCLUSIVE;
! 		else
! 			newrc->markType = ROW_MARK_SHARE;
  		newrc->noWait = rc->noWait;
  		newrc->isParent = false;
  
--- 1894,1913 ----
  		newrc = makeNode(PlanRowMark);
  		newrc->rti = newrc->prti = rc->rti;
  		newrc->rowmarkId = ++(root->glob->lastRowMarkId);
! 		switch (rc->strength)
! 		{
! 			case LCS_FORUPDATE:
! 				newrc->markType = ROW_MARK_EXCLUSIVE;
! 				break;
! 			case LCS_FORSHARE:
! 				newrc->markType = ROW_MARK_SHARE;
! 				break;
! 			case LCS_FORKEYLOCK:
! 				newrc->markType = ROW_MARK_KEYLOCK;
! 				break;
! 			default:
! 				elog(ERROR, "unsupported rowmark type %d", rc->strength);
! 		}
  		newrc->noWait = rc->noWait;
  		newrc->isParent = false;
  
*** a/src/backend/parser/analyze.c
--- b/src/backend/parser/analyze.c
***************
*** 2310,2316 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
  	/* make a clause we can pass down to subqueries to select all rels */
  	allrels = makeNode(LockingClause);
  	allrels->lockedRels = NIL;	/* indicates all rels */
! 	allrels->forUpdate = lc->forUpdate;
  	allrels->noWait = lc->noWait;
  
  	if (lockedRels == NIL)
--- 2310,2316 ----
  	/* make a clause we can pass down to subqueries to select all rels */
  	allrels = makeNode(LockingClause);
  	allrels->lockedRels = NIL;	/* indicates all rels */
! 	allrels->strength = lc->strength;
  	allrels->noWait = lc->noWait;
  
  	if (lockedRels == NIL)
***************
*** 2329,2340 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
  					if (rte->relkind == RELKIND_FOREIGN_TABLE)
  						break;
  					applyLockingClause(qry, i,
! 									   lc->forUpdate, lc->noWait, pushedDown);
  					rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  					break;
  				case RTE_SUBQUERY:
  					applyLockingClause(qry, i,
! 									   lc->forUpdate, lc->noWait, pushedDown);
  
  					/*
  					 * FOR UPDATE/SHARE of subquery is propagated to all of
--- 2329,2340 ----
  					if (rte->relkind == RELKIND_FOREIGN_TABLE)
  						break;
  					applyLockingClause(qry, i,
! 									   lc->strength, lc->noWait, pushedDown);
  					rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  					break;
  				case RTE_SUBQUERY:
  					applyLockingClause(qry, i,
! 									   lc->strength, lc->noWait, pushedDown);
  
  					/*
  					 * FOR UPDATE/SHARE of subquery is propagated to all of
***************
*** 2384,2396 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
  											 rte->eref->aliasname),
  									  parser_errposition(pstate, thisrel->location)));
  							applyLockingClause(qry, i,
! 											   lc->forUpdate, lc->noWait,
  											   pushedDown);
  							rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  							break;
  						case RTE_SUBQUERY:
  							applyLockingClause(qry, i,
! 											   lc->forUpdate, lc->noWait,
  											   pushedDown);
  							/* see comment above */
  							transformLockingClause(pstate, rte->subquery,
--- 2384,2396 ----
  											 rte->eref->aliasname),
  									  parser_errposition(pstate, thisrel->location)));
  							applyLockingClause(qry, i,
! 											   lc->strength, lc->noWait,
  											   pushedDown);
  							rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  							break;
  						case RTE_SUBQUERY:
  							applyLockingClause(qry, i,
! 											   lc->strength, lc->noWait,
  											   pushedDown);
  							/* see comment above */
  							transformLockingClause(pstate, rte->subquery,
***************
*** 2443,2449 **** transformLockingClause(ParseState *pstate, Query *qry, LockingClause *lc,
   */
  void
  applyLockingClause(Query *qry, Index rtindex,
! 				   bool forUpdate, bool noWait, bool pushedDown)
  {
  	RowMarkClause *rc;
  
--- 2443,2449 ----
   */
  void
  applyLockingClause(Query *qry, Index rtindex,
! 				   LockClauseStrength strength, bool noWait, bool pushedDown)
  {
  	RowMarkClause *rc;
  
***************
*** 2455,2464 **** applyLockingClause(Query *qry, Index rtindex,
  	if ((rc = get_parse_rowmark(qry, rtindex)) != NULL)
  	{
  		/*
! 		 * If the same RTE is specified both FOR UPDATE and FOR SHARE, treat
! 		 * it as FOR UPDATE.  (Reasonable, since you can't take both a shared
! 		 * and exclusive lock at the same time; it'll end up being exclusive
! 		 * anyway.)
  		 *
  		 * We also consider that NOWAIT wins if it's specified both ways. This
  		 * is a bit more debatable but raising an error doesn't seem helpful.
--- 2455,2464 ----
  	if ((rc = get_parse_rowmark(qry, rtindex)) != NULL)
  	{
  		/*
! 		 * If the same RTE is specified for more than one locking strength,
! 		 * treat is as the strongest.  (Reasonable, since you can't take both a
! 		 * shared and exclusive lock at the same time; it'll end up being
! 		 * exclusive anyway.)
  		 *
  		 * We also consider that NOWAIT wins if it's specified both ways. This
  		 * is a bit more debatable but raising an error doesn't seem helpful.
***************
*** 2467,2473 **** applyLockingClause(Query *qry, Index rtindex,
  		 *
  		 * And of course pushedDown becomes false if any clause is explicit.
  		 */
! 		rc->forUpdate |= forUpdate;
  		rc->noWait |= noWait;
  		rc->pushedDown &= pushedDown;
  		return;
--- 2467,2473 ----
  		 *
  		 * And of course pushedDown becomes false if any clause is explicit.
  		 */
! 		rc->strength = Max(rc->strength, strength);
  		rc->noWait |= noWait;
  		rc->pushedDown &= pushedDown;
  		return;
***************
*** 2476,2482 **** applyLockingClause(Query *qry, Index rtindex,
  	/* Make a new RowMarkClause */
  	rc = makeNode(RowMarkClause);
  	rc->rti = rtindex;
! 	rc->forUpdate = forUpdate;
  	rc->noWait = noWait;
  	rc->pushedDown = pushedDown;
  	qry->rowMarks = lappend(qry->rowMarks, rc);
--- 2476,2482 ----
  	/* Make a new RowMarkClause */
  	rc = makeNode(RowMarkClause);
  	rc->rti = rtindex;
! 	rc->strength = strength;
  	rc->noWait = noWait;
  	rc->pushedDown = pushedDown;
  	qry->rowMarks = lappend(qry->rowMarks, rc);
*** a/src/backend/parser/gram.y
--- b/src/backend/parser/gram.y
***************
*** 8760,8766 **** for_locking_item:
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->forUpdate = TRUE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
--- 8760,8766 ----
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->strength = LCS_FORUPDATE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
***************
*** 8768,8777 **** for_locking_item:
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->forUpdate = FALSE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
  		;
  
  locked_rels_list:
--- 8768,8785 ----
  				{
  					LockingClause *n = makeNode(LockingClause);
  					n->lockedRels = $3;
! 					n->strength = LCS_FORSHARE;
  					n->noWait = $4;
  					$$ = (Node *) n;
  				}
+ 			| FOR KEY LOCK_P locked_rels_list opt_nowait
+ 				{
+ 					LockingClause *n = makeNode(LockingClause);
+ 					n->lockedRels = $4;
+ 					n->strength = LCS_FORKEYLOCK;
+ 					n->noWait = $5;
+ 					$$ = (Node *) n;
+ 				}
  		;
  
  locked_rels_list:
*** a/src/backend/rewrite/rewriteHandler.c
--- b/src/backend/rewrite/rewriteHandler.c
***************
*** 56,62 **** static void rewriteValuesRTE(RangeTblEntry *rte, Relation target_relation,
  static void rewriteTargetListUD(Query *parsetree, RangeTblEntry *target_rte,
  					Relation target_relation);
  static void markQueryForLocking(Query *qry, Node *jtnode,
! 					bool forUpdate, bool noWait, bool pushedDown);
  static List *matchLocks(CmdType event, RuleLock *rulelocks,
  		   int varno, Query *parsetree);
  static Query *fireRIRrules(Query *parsetree, List *activeRIRs,
--- 56,62 ----
  static void rewriteTargetListUD(Query *parsetree, RangeTblEntry *target_rte,
  					Relation target_relation);
  static void markQueryForLocking(Query *qry, Node *jtnode,
! 					LockClauseStrength strength, bool noWait, bool pushedDown);
  static List *matchLocks(CmdType event, RuleLock *rulelocks,
  		   int varno, Query *parsetree);
  static Query *fireRIRrules(Query *parsetree, List *activeRIRs,
***************
*** 1402,1409 **** ApplyRetrieveRule(Query *parsetree,
  	rte->modifiedCols = NULL;
  
  	/*
! 	 * If FOR UPDATE/SHARE of view, mark all the contained tables as implicit
! 	 * FOR UPDATE/SHARE, the same as the parser would have done if the view's
  	 * subquery had been written out explicitly.
  	 *
  	 * Note: we don't consider forUpdatePushedDown here; such marks will be
--- 1402,1409 ----
  	rte->modifiedCols = NULL;
  
  	/*
! 	 * If FOR UPDATE/SHARE/KEY LOCK of view, mark all the contained tables as implicit
! 	 * FOR UPDATE/SHARE/KEY LOCK, the same as the parser would have done if the view's
  	 * subquery had been written out explicitly.
  	 *
  	 * Note: we don't consider forUpdatePushedDown here; such marks will be
***************
*** 1411,1423 **** ApplyRetrieveRule(Query *parsetree,
  	 */
  	if (rc != NULL)
  		markQueryForLocking(rule_action, (Node *) rule_action->jointree,
! 							rc->forUpdate, rc->noWait, true);
  
  	return parsetree;
  }
  
  /*
!  * Recursively mark all relations used by a view as FOR UPDATE/SHARE.
   *
   * This may generate an invalid query, eg if some sub-query uses an
   * aggregate.  We leave it to the planner to detect that.
--- 1411,1423 ----
  	 */
  	if (rc != NULL)
  		markQueryForLocking(rule_action, (Node *) rule_action->jointree,
! 							rc->strength, rc->noWait, true);
  
  	return parsetree;
  }
  
  /*
!  * Recursively mark all relations used by a view as FOR UPDATE/SHARE/KEY LOCK.
   *
   * This may generate an invalid query, eg if some sub-query uses an
   * aggregate.  We leave it to the planner to detect that.
***************
*** 1429,1435 **** ApplyRetrieveRule(Query *parsetree,
   */
  static void
  markQueryForLocking(Query *qry, Node *jtnode,
! 					bool forUpdate, bool noWait, bool pushedDown)
  {
  	if (jtnode == NULL)
  		return;
--- 1429,1435 ----
   */
  static void
  markQueryForLocking(Query *qry, Node *jtnode,
! 					LockClauseStrength strength, bool noWait, bool pushedDown)
  {
  	if (jtnode == NULL)
  		return;
***************
*** 1443,1458 **** markQueryForLocking(Query *qry, Node *jtnode,
  			/* ignore foreign tables */
  			if (rte->relkind != RELKIND_FOREIGN_TABLE)
  			{
! 				applyLockingClause(qry, rti, forUpdate, noWait, pushedDown);
  				rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  			}
  		}
  		else if (rte->rtekind == RTE_SUBQUERY)
  		{
! 			applyLockingClause(qry, rti, forUpdate, noWait, pushedDown);
! 			/* FOR UPDATE/SHARE of subquery is propagated to subquery's rels */
  			markQueryForLocking(rte->subquery, (Node *) rte->subquery->jointree,
! 								forUpdate, noWait, true);
  		}
  		/* other RTE types are unaffected by FOR UPDATE */
  	}
--- 1443,1458 ----
  			/* ignore foreign tables */
  			if (rte->relkind != RELKIND_FOREIGN_TABLE)
  			{
! 				applyLockingClause(qry, rti, strength, noWait, pushedDown);
  				rte->requiredPerms |= ACL_SELECT_FOR_UPDATE;
  			}
  		}
  		else if (rte->rtekind == RTE_SUBQUERY)
  		{
! 			applyLockingClause(qry, rti, strength, noWait, pushedDown);
! 			/* FOR UPDATE/SHARE/KEY LOCK of subquery is propagated to subquery's rels */
  			markQueryForLocking(rte->subquery, (Node *) rte->subquery->jointree,
! 								strength, noWait, true);
  		}
  		/* other RTE types are unaffected by FOR UPDATE */
  	}
***************
*** 1462,1475 **** markQueryForLocking(Query *qry, Node *jtnode,
  		ListCell   *l;
  
  		foreach(l, f->fromlist)
! 			markQueryForLocking(qry, lfirst(l), forUpdate, noWait, pushedDown);
  	}
  	else if (IsA(jtnode, JoinExpr))
  	{
  		JoinExpr   *j = (JoinExpr *) jtnode;
  
! 		markQueryForLocking(qry, j->larg, forUpdate, noWait, pushedDown);
! 		markQueryForLocking(qry, j->rarg, forUpdate, noWait, pushedDown);
  	}
  	else
  		elog(ERROR, "unrecognized node type: %d",
--- 1462,1475 ----
  		ListCell   *l;
  
  		foreach(l, f->fromlist)
! 			markQueryForLocking(qry, lfirst(l), strength, noWait, pushedDown);
  	}
  	else if (IsA(jtnode, JoinExpr))
  	{
  		JoinExpr   *j = (JoinExpr *) jtnode;
  
! 		markQueryForLocking(qry, j->larg, strength, noWait, pushedDown);
! 		markQueryForLocking(qry, j->rarg, strength, noWait, pushedDown);
  	}
  	else
  		elog(ERROR, "unrecognized node type: %d",
*** a/src/backend/tcop/utility.c
--- b/src/backend/tcop/utility.c
***************
*** 130,136 **** CommandIsReadOnly(Node *parsetree)
  				if (stmt->intoClause != NULL)
  					return false;		/* SELECT INTO */
  				else if (stmt->rowMarks != NIL)
! 					return false;		/* SELECT FOR UPDATE/SHARE */
  				else if (stmt->hasModifyingCTE)
  					return false;		/* data-modifying CTE */
  				else
--- 130,136 ----
  				if (stmt->intoClause != NULL)
  					return false;		/* SELECT INTO */
  				else if (stmt->rowMarks != NIL)
! 					return false;		/* SELECT FOR UPDATE/SHARE/KEY LOCK */
  				else if (stmt->hasModifyingCTE)
  					return false;		/* data-modifying CTE */
  				else
***************
*** 2181,2190 **** CreateCommandTag(Node *parsetree)
  						else if (stmt->rowMarks != NIL)
  						{
  							/* not 100% but probably close enough */
! 							if (((PlanRowMark *) linitial(stmt->rowMarks))->markType == ROW_MARK_EXCLUSIVE)
! 								tag = "SELECT FOR UPDATE";
! 							else
! 								tag = "SELECT FOR SHARE";
  						}
  						else
  							tag = "SELECT";
--- 2181,2201 ----
  						else if (stmt->rowMarks != NIL)
  						{
  							/* not 100% but probably close enough */
! 							switch (((RowMarkClause *) linitial(stmt->rowMarks))->strength)
! 							{
! 								case LCS_FORUPDATE:
! 									tag = "SELECT FOR UPDATE";
! 									break;
! 								case LCS_FORSHARE:
! 									tag = "SELECT FOR SHARE";
! 									break;
! 								case LCS_FORKEYLOCK:
! 									tag = "SELECT FOR KEY LOCK";
! 									break;
! 								default:
! 									tag =  "???";
! 									break;
! 							}
  						}
  						else
  							tag = "SELECT";
***************
*** 2231,2240 **** CreateCommandTag(Node *parsetree)
  						else if (stmt->rowMarks != NIL)
  						{
  							/* not 100% but probably close enough */
! 							if (((RowMarkClause *) linitial(stmt->rowMarks))->forUpdate)
! 								tag = "SELECT FOR UPDATE";
! 							else
! 								tag = "SELECT FOR SHARE";
  						}
  						else
  							tag = "SELECT";
--- 2242,2262 ----
  						else if (stmt->rowMarks != NIL)
  						{
  							/* not 100% but probably close enough */
! 							switch (((RowMarkClause *) linitial(stmt->rowMarks))->strength)
! 							{
! 								case LCS_FORUPDATE:
! 									tag = "SELECT FOR UPDATE";
! 									break;
! 								case LCS_FORSHARE:
! 									tag = "SELECT FOR SHARE";
! 									break;
! 								case LCS_FORKEYLOCK:
! 									tag = "SELECT FOR KEY LOCK";
! 									break;
! 								default:
! 									tag =  "???";
! 									break;
! 							}
  						}
  						else
  							tag = "SELECT";
*** a/src/backend/utils/adt/ri_triggers.c
--- b/src/backend/utils/adt/ri_triggers.c
***************
*** 309,315 **** RI_FKey_check(PG_FUNCTION_ARGS)
  	 * Get the relation descriptors of the FK and PK tables.
  	 *
  	 * pk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = trigdata->tg_relation;
  	pk_rel = heap_open(riinfo.pk_relid, RowShareLock);
--- 309,315 ----
  	 * Get the relation descriptors of the FK and PK tables.
  	 *
  	 * pk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = trigdata->tg_relation;
  	pk_rel = heap_open(riinfo.pk_relid, RowShareLock);
***************
*** 339,350 **** RI_FKey_check(PG_FUNCTION_ARGS)
  
  			/* ---------
  			 * The query string built is
! 			 *	SELECT 1 FROM ONLY <pktable>
  			 * ----------
  			 */
  			quoteRelationName(pkrelname, pk_rel);
  			snprintf(querystr, sizeof(querystr),
! 					 "SELECT 1 FROM ONLY %s x FOR SHARE OF x",
  					 pkrelname);
  
  			/* Prepare and save the plan */
--- 339,350 ----
  
  			/* ---------
  			 * The query string built is
! 			 *	SELECT 1 FROM ONLY <pktable> x FOR KEY LOCK OF x
  			 * ----------
  			 */
  			quoteRelationName(pkrelname, pk_rel);
  			snprintf(querystr, sizeof(querystr),
! 					 "SELECT 1 FROM ONLY %s x FOR KEY LOCK OF x",
  					 pkrelname);
  
  			/* Prepare and save the plan */
***************
*** 464,470 **** RI_FKey_check(PG_FUNCTION_ARGS)
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> WHERE pkatt1 = $1 [AND ...] FOR SHARE
  		 * The type id's for the $ parameters are those of the
  		 * corresponding FK attributes.
  		 * ----------
--- 464,471 ----
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> x WHERE pkatt1 = $1 [AND ...]
! 		 *	       FOR KEY LOCK OF x
  		 * The type id's for the $ parameters are those of the
  		 * corresponding FK attributes.
  		 * ----------
***************
*** 488,494 **** RI_FKey_check(PG_FUNCTION_ARGS)
  			querysep = "AND";
  			queryoids[i] = fk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 489,495 ----
  			querysep = "AND";
  			queryoids[i] = fk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 626,632 **** ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> WHERE pkatt1 = $1 [AND ...] FOR SHARE
  		 * The type id's for the $ parameters are those of the
  		 * PK attributes themselves.
  		 * ----------
--- 627,634 ----
  
  		/* ----------
  		 * The query string built is
! 		 *	SELECT 1 FROM ONLY <pktable> x WHERE pkatt1 = $1 [AND ...]
! 		 *	       FOR KEY LOCK OF x
  		 * The type id's for the $ parameters are those of the
  		 * PK attributes themselves.
  		 * ----------
***************
*** 649,655 **** ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
  			querysep = "AND";
  			queryoids[i] = pk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
--- 651,657 ----
  			querysep = "AND";
  			queryoids[i] = pk_type;
  		}
! 		appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  		/* Prepare and save the plan */
  		qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
***************
*** 713,719 **** RI_FKey_noaction_del(PG_FUNCTION_ARGS)
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 715,721 ----
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 781,787 **** RI_FKey_noaction_del(PG_FUNCTION_ARGS)
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> WHERE $1 = fkatt1 [AND ...]
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
--- 783,790 ----
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> x WHERE $1 = fkatt1 [AND ...]
! 				 *	       FOR KEY LOCK OF x
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
***************
*** 806,812 **** RI_FKey_noaction_del(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 809,815 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 891,897 **** RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 894,900 ----
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 994,1000 **** RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 997,1003 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 1432,1438 **** RI_FKey_restrict_del(PG_FUNCTION_ARGS)
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 1435,1441 ----
  	 * Get the relation descriptors of the FK and PK tables and the old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 1490,1496 **** RI_FKey_restrict_del(PG_FUNCTION_ARGS)
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> WHERE $1 = fkatt1 [AND ...]
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
--- 1493,1500 ----
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> x WHERE $1 = fkatt1 [AND ...]
! 				 *	       FOR KEY LOCK OF x
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
***************
*** 1515,1521 **** RI_FKey_restrict_del(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 1519,1525 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
***************
*** 1605,1611 **** RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR SHARE will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
--- 1609,1615 ----
  	 * old tuple.
  	 *
  	 * fk_rel is opened in RowShareLock mode since that's what our eventual
! 	 * SELECT FOR KEY LOCK will get on it.
  	 */
  	fk_rel = heap_open(riinfo.fk_relid, RowShareLock);
  	pk_rel = trigdata->tg_relation;
***************
*** 1673,1679 **** RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> WHERE $1 = fkatt1 [AND ...]
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
--- 1677,1684 ----
  
  				/* ----------
  				 * The query string built is
! 				 *	SELECT 1 FROM ONLY <fktable> x WHERE $1 = fkatt1 [AND ...]
! 				 *	       FOR KEY LOCK OF x
  				 * The type id's for the $ parameters are those of the
  				 * corresponding PK attributes.
  				 * ----------
***************
*** 1698,1704 **** RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR SHARE OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
--- 1703,1709 ----
  					querysep = "AND";
  					queryoids[i] = pk_type;
  				}
! 				appendStringInfo(&querybuf, " FOR KEY LOCK OF x");
  
  				/* Prepare and save the plan */
  				qplan = ri_PlanCheck(querybuf.data, riinfo.nkeys, queryoids,
*** a/src/backend/utils/adt/ruleutils.c
--- b/src/backend/utils/adt/ruleutils.c
***************
*** 2857,2868 **** get_select_query_def(Query *query, deparse_context *context,
  			if (rc->pushedDown)
  				continue;
  
! 			if (rc->forUpdate)
! 				appendContextKeyword(context, " FOR UPDATE",
! 									 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 			else
! 				appendContextKeyword(context, " FOR SHARE",
! 									 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
  			appendStringInfo(buf, " OF %s",
  							 quote_identifier(rte->eref->aliasname));
  			if (rc->noWait)
--- 2857,2880 ----
  			if (rc->pushedDown)
  				continue;
  
! 			switch (rc->strength)
! 			{
! 				case LCS_FORKEYLOCK:
! 					appendContextKeyword(context, " FOR KEY LOCK",
! 										 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 					break;
! 				case LCS_FORSHARE:
! 					appendContextKeyword(context, " FOR SHARE",
! 										 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 					break;
! 				case LCS_FORUPDATE:
! 					appendContextKeyword(context, " FOR UPDATE",
! 										 -PRETTYINDENT_STD, PRETTYINDENT_STD, 0);
! 					break;
! 				default:
! 					elog(ERROR, "unrecognized row locking clause %d", rc->strength);
! 			}
! 
  			appendStringInfo(buf, " OF %s",
  							 quote_identifier(rte->eref->aliasname));
  			if (rc->noWait)
*** a/src/backend/utils/cache/relcache.c
--- b/src/backend/utils/cache/relcache.c
***************
*** 3614,3619 **** RelationGetIndexPredicate(Relation relation)
--- 3614,3622 ----
   * simple index keys, but attributes used in expressions and partial-index
   * predicates.)
   *
+  * If "keyAttrs" is true, only attributes that can be referenced by foreign
+  * keys are considered.
+  *
   * Attribute numbers are offset by FirstLowInvalidHeapAttributeNumber so that
   * we can include system attributes (e.g., OID) in the bitmap representation.
   *
***************
*** 3625,3640 **** RelationGetIndexPredicate(Relation relation)
   * be bms_free'd when not needed anymore.
   */
  Bitmapset *
! RelationGetIndexAttrBitmap(Relation relation)
  {
  	Bitmapset  *indexattrs;
  	List	   *indexoidlist;
  	ListCell   *l;
  	MemoryContext oldcxt;
  
  	/* Quick exit if we already computed the result. */
  	if (relation->rd_indexattr != NULL)
! 		return bms_copy(relation->rd_indexattr);
  
  	/* Fast path if definitely no indexes */
  	if (!RelationGetForm(relation)->relhasindex)
--- 3628,3644 ----
   * be bms_free'd when not needed anymore.
   */
  Bitmapset *
! RelationGetIndexAttrBitmap(Relation relation, bool keyAttrs)
  {
  	Bitmapset  *indexattrs;
+ 	Bitmapset  *uindexattrs;
  	List	   *indexoidlist;
  	ListCell   *l;
  	MemoryContext oldcxt;
  
  	/* Quick exit if we already computed the result. */
  	if (relation->rd_indexattr != NULL)
! 		return bms_copy(keyAttrs ? relation->rd_keyattr : relation->rd_indexattr);
  
  	/* Fast path if definitely no indexes */
  	if (!RelationGetForm(relation)->relhasindex)
***************
*** 3653,3678 **** RelationGetIndexAttrBitmap(Relation relation)
--- 3657,3694 ----
  	 * For each index, add referenced attributes to indexattrs.
  	 */
  	indexattrs = NULL;
+ 	uindexattrs = NULL;
  	foreach(l, indexoidlist)
  	{
  		Oid			indexOid = lfirst_oid(l);
  		Relation	indexDesc;
  		IndexInfo  *indexInfo;
  		int			i;
+ 		bool		isKey;
  
  		indexDesc = index_open(indexOid, AccessShareLock);
  
  		/* Extract index key information from the index's pg_index row */
  		indexInfo = BuildIndexInfo(indexDesc);
  
+ 		/* Can this index be referenced by a foreign key? */
+ 		isKey = indexInfo->ii_Unique &&
+ 				indexInfo->ii_Expressions == NIL &&
+ 				indexInfo->ii_Predicate == NIL;
+ 
  		/* Collect simple attribute references */
  		for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
  		{
  			int			attrnum = indexInfo->ii_KeyAttrNumbers[i];
  
  			if (attrnum != 0)
+ 			{
  				indexattrs = bms_add_member(indexattrs,
  							   attrnum - FirstLowInvalidHeapAttributeNumber);
+ 				if (isKey)
+ 					uindexattrs = bms_add_member(uindexattrs,
+ 											   	 attrnum - FirstLowInvalidHeapAttributeNumber);
+ 			}
  		}
  
  		/* Collect all attributes used in expressions, too */
***************
*** 3689,3698 **** RelationGetIndexAttrBitmap(Relation relation)
  	/* Now save a copy of the bitmap in the relcache entry. */
  	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
  	relation->rd_indexattr = bms_copy(indexattrs);
  	MemoryContextSwitchTo(oldcxt);
  
  	/* We return our original working copy for caller to play with */
! 	return indexattrs;
  }
  
  /*
--- 3705,3715 ----
  	/* Now save a copy of the bitmap in the relcache entry. */
  	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
  	relation->rd_indexattr = bms_copy(indexattrs);
+ 	relation->rd_keyattr = bms_copy(uindexattrs);
  	MemoryContextSwitchTo(oldcxt);
  
  	/* We return our original working copy for caller to play with */
! 	return keyAttrs ? uindexattrs : indexattrs;
  }
  
  /*
*** a/src/include/access/heapam.h
--- b/src/include/access/heapam.h
***************
*** 31,38 ****
--- 31,44 ----
  
  typedef struct BulkInsertStateData *BulkInsertState;
  
+ /*
+  * This enum mirrors LockClauseStrength precisely, but we define it separately
+  * to reduce having to share otherwise unrelated headers.  To go from one to
+  * the other, we wade through the planner using a third enum, RowMarkType.
+  */
  typedef enum
  {
+ 	LockTupleKeylock,
  	LockTupleShared,
  	LockTupleExclusive
  } LockTupleMode;
*** a/src/include/access/htup.h
--- b/src/include/access/htup.h
***************
*** 163,174 **** typedef HeapTupleHeaderData *HeapTupleHeader;
  #define HEAP_HASVARWIDTH		0x0002	/* has variable-width attribute(s) */
  #define HEAP_HASEXTERNAL		0x0004	/* has external stored attribute(s) */
  #define HEAP_HASOID				0x0008	/* has an object-id field */
! /* bit 0x0010 is available */
  #define HEAP_COMBOCID			0x0020	/* t_cid is a combo cid */
  #define HEAP_XMAX_EXCL_LOCK		0x0040	/* xmax is exclusive locker */
  #define HEAP_XMAX_SHARED_LOCK	0x0080	/* xmax is shared locker */
! /* if either LOCK bit is set, xmax hasn't deleted the tuple, only locked it */
! #define HEAP_IS_LOCKED	(HEAP_XMAX_EXCL_LOCK | HEAP_XMAX_SHARED_LOCK)
  #define HEAP_XMIN_COMMITTED		0x0100	/* t_xmin committed */
  #define HEAP_XMIN_INVALID		0x0200	/* t_xmin invalid/aborted */
  #define HEAP_XMAX_COMMITTED		0x0400	/* t_xmax committed */
--- 163,177 ----
  #define HEAP_HASVARWIDTH		0x0002	/* has variable-width attribute(s) */
  #define HEAP_HASEXTERNAL		0x0004	/* has external stored attribute(s) */
  #define HEAP_HASOID				0x0008	/* has an object-id field */
! #define HEAP_XMAX_KEY_LOCK		0x0010	/* xmax is a "key" locker */
  #define HEAP_COMBOCID			0x0020	/* t_cid is a combo cid */
  #define HEAP_XMAX_EXCL_LOCK		0x0040	/* xmax is exclusive locker */
  #define HEAP_XMAX_SHARED_LOCK	0x0080	/* xmax is shared locker */
! /* if either SHARE or KEY lock bit is set, this is a "shared" lock */
! #define HEAP_IS_SHARE_LOCKED (HEAP_XMAX_SHARED_LOCK | HEAP_XMAX_KEY_LOCK)
! /* if any LOCK bit is set, xmax hasn't deleted the tuple, only locked it */
! #define HEAP_IS_LOCKED	(HEAP_XMAX_EXCL_LOCK | HEAP_XMAX_SHARED_LOCK | \
! 						 HEAP_XMAX_KEY_LOCK)
  #define HEAP_XMIN_COMMITTED		0x0100	/* t_xmin committed */
  #define HEAP_XMIN_INVALID		0x0200	/* t_xmin invalid/aborted */
  #define HEAP_XMAX_COMMITTED		0x0400	/* t_xmax committed */
***************
*** 726,735 **** typedef struct xl_heap_lock
  	xl_heaptid	target;			/* locked tuple id */
  	TransactionId locking_xid;	/* might be a MultiXactId not xid */
  	bool		xid_is_mxact;	/* is it? */
! 	bool		shared_lock;	/* shared or exclusive row lock? */
  } xl_heap_lock;
  
! #define SizeOfHeapLock	(offsetof(xl_heap_lock, shared_lock) + sizeof(bool))
  
  /* This is what we need to know about in-place update */
  typedef struct xl_heap_inplace
--- 729,738 ----
  	xl_heaptid	target;			/* locked tuple id */
  	TransactionId locking_xid;	/* might be a MultiXactId not xid */
  	bool		xid_is_mxact;	/* is it? */
! 	int8		lock_strength;	/* keylock, shared, exclusive lock? */
  } xl_heap_lock;
  
! #define SizeOfHeapLock	(offsetof(xl_heap_lock, lock_strength) + sizeof(int8))
  
  /* This is what we need to know about in-place update */
  typedef struct xl_heap_inplace
***************
*** 767,774 **** extern void HeapTupleHeaderAdvanceLatestRemovedXid(HeapTupleHeader tuple,
  extern CommandId HeapTupleHeaderGetCmin(HeapTupleHeader tup);
  extern CommandId HeapTupleHeaderGetCmax(HeapTupleHeader tup);
  extern void HeapTupleHeaderAdjustCmax(HeapTupleHeader tup,
! 						  CommandId *cmax,
! 						  bool *iscombo);
  
  /* ----------------
   *		fastgetattr
--- 770,776 ----
  extern CommandId HeapTupleHeaderGetCmin(HeapTupleHeader tup);
  extern CommandId HeapTupleHeaderGetCmax(HeapTupleHeader tup);
  extern void HeapTupleHeaderAdjustCmax(HeapTupleHeader tup,
! 						  CommandId *cmax, bool *iscombo);
  
  /* ----------------
   *		fastgetattr
*** a/src/include/access/xlog_internal.h
--- b/src/include/access/xlog_internal.h
***************
*** 71,77 **** typedef struct XLogContRecord
  /*
   * Each page of XLOG file has a header like this:
   */
! #define XLOG_PAGE_MAGIC 0xD068	/* can be used as WAL version indicator */
  
  typedef struct XLogPageHeaderData
  {
--- 71,77 ----
  /*
   * Each page of XLOG file has a header like this:
   */
! #define XLOG_PAGE_MAGIC 0xD069	/* can be used as WAL version indicator */
  
  typedef struct XLogPageHeaderData
  {
*** a/src/include/nodes/execnodes.h
--- b/src/include/nodes/execnodes.h
***************
*** 408,414 **** typedef struct EState
   * ExecRowMark -
   *	   runtime representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE, we should have an
   * ExecRowMark for each non-target relation in the query (except inheritance
   * parent RTEs, which can be ignored at runtime).  See PlanRowMark for details
   * about most of the fields.  In addition to fields directly derived from
--- 408,414 ----
   * ExecRowMark -
   *	   runtime representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE/KEY LOCK, we should have an
   * ExecRowMark for each non-target relation in the query (except inheritance
   * parent RTEs, which can be ignored at runtime).  See PlanRowMark for details
   * about most of the fields.  In addition to fields directly derived from
*** a/src/include/nodes/parsenodes.h
--- b/src/include/nodes/parsenodes.h
***************
*** 119,125 **** typedef struct Query
  	bool		hasDistinctOn;	/* distinctClause is from DISTINCT ON */
  	bool		hasRecursive;	/* WITH RECURSIVE was specified */
  	bool		hasModifyingCTE;	/* has INSERT/UPDATE/DELETE in WITH */
! 	bool		hasForUpdate;	/* FOR UPDATE or FOR SHARE was specified */
  
  	List	   *cteList;		/* WITH list (of CommonTableExpr's) */
  
--- 119,125 ----
  	bool		hasDistinctOn;	/* distinctClause is from DISTINCT ON */
  	bool		hasRecursive;	/* WITH RECURSIVE was specified */
  	bool		hasModifyingCTE;	/* has INSERT/UPDATE/DELETE in WITH */
! 	bool		hasForUpdate;	/* FOR UPDATE/SHARE/KEY LOCK was specified */
  
  	List	   *cteList;		/* WITH list (of CommonTableExpr's) */
  
***************
*** 569,586 **** typedef struct DefElem
  } DefElem;
  
  /*
!  * LockingClause - raw representation of FOR UPDATE/SHARE options
   *
   * Note: lockedRels == NIL means "all relations in query".	Otherwise it
   * is a list of RangeVar nodes.  (We use RangeVar mainly because it carries
   * a location field --- currently, parse analysis insists on unqualified
   * names in LockingClause.)
   */
  typedef struct LockingClause
  {
  	NodeTag		type;
  	List	   *lockedRels;		/* FOR UPDATE or FOR SHARE relations */
! 	bool		forUpdate;		/* true = FOR UPDATE, false = FOR SHARE */
  	bool		noWait;			/* NOWAIT option */
  } LockingClause;
  
--- 569,594 ----
  } DefElem;
  
  /*
!  * LockingClause - raw representation of FOR UPDATE/SHARE/KEY LOCK options
   *
   * Note: lockedRels == NIL means "all relations in query".	Otherwise it
   * is a list of RangeVar nodes.  (We use RangeVar mainly because it carries
   * a location field --- currently, parse analysis insists on unqualified
   * names in LockingClause.)
   */
+ typedef enum LockClauseStrength
+ {
+ 	/* order is important -- see applyLockingClause */
+ 	LCS_FORKEYLOCK,
+ 	LCS_FORSHARE,
+ 	LCS_FORUPDATE
+ } LockClauseStrength;
+ 
  typedef struct LockingClause
  {
  	NodeTag		type;
  	List	   *lockedRels;		/* FOR UPDATE or FOR SHARE relations */
! 	LockClauseStrength strength;
  	bool		noWait;			/* NOWAIT option */
  } LockingClause;
  
***************
*** 863,880 **** typedef struct WindowClause
   *	   parser output representation of FOR UPDATE/SHARE clauses
   *
   * Query.rowMarks contains a separate RowMarkClause node for each relation
!  * identified as a FOR UPDATE/SHARE target.  If FOR UPDATE/SHARE is applied
!  * to a subquery, we generate RowMarkClauses for all normal and subquery rels
!  * in the subquery, but they are marked pushedDown = true to distinguish them
!  * from clauses that were explicitly written at this query level.  Also,
!  * Query.hasForUpdate tells whether there were explicit FOR UPDATE/SHARE
!  * clauses in the current query level.
   */
  typedef struct RowMarkClause
  {
  	NodeTag		type;
  	Index		rti;			/* range table index of target relation */
! 	bool		forUpdate;		/* true = FOR UPDATE, false = FOR SHARE */
  	bool		noWait;			/* NOWAIT option */
  	bool		pushedDown;		/* pushed down from higher query level? */
  } RowMarkClause;
--- 871,888 ----
   *	   parser output representation of FOR UPDATE/SHARE clauses
   *
   * Query.rowMarks contains a separate RowMarkClause node for each relation
!  * identified as a FOR UPDATE/SHARE/KEY LOCK target.  If one of these clauses
!  * is applied to a subquery, we generate RowMarkClauses for all normal and
!  * subquery rels in the subquery, but they are marked pushedDown = true to
!  * distinguish them from clauses that were explicitly written at this query
!  * level.  Also, Query.hasForUpdate tells whether there were explicit FOR
!  * UPDATE/SHARE/KEY LOCK clauses in the current query level.
   */
  typedef struct RowMarkClause
  {
  	NodeTag		type;
  	Index		rti;			/* range table index of target relation */
! 	LockClauseStrength strength;
  	bool		noWait;			/* NOWAIT option */
  	bool		pushedDown;		/* pushed down from higher query level? */
  } RowMarkClause;
*** a/src/include/nodes/plannodes.h
--- b/src/include/nodes/plannodes.h
***************
*** 722,728 **** typedef struct Limit
   * RowMarkType -
   *	  enums for types of row-marking operations
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE, we have to uniquely
   * identify all the source rows, not only those from the target relations, so
   * that we can perform EvalPlanQual rechecking at need.  For plain tables we
   * can just fetch the TID, the same as for a target relation.  Otherwise (for
--- 722,728 ----
   * RowMarkType -
   *	  enums for types of row-marking operations
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE/KEY LOCK, we have to uniquely
   * identify all the source rows, not only those from the target relations, so
   * that we can perform EvalPlanQual rechecking at need.  For plain tables we
   * can just fetch the TID, the same as for a target relation.  Otherwise (for
***************
*** 734,752 **** typedef enum RowMarkType
  {
  	ROW_MARK_EXCLUSIVE,			/* obtain exclusive tuple lock */
  	ROW_MARK_SHARE,				/* obtain shared tuple lock */
  	ROW_MARK_REFERENCE,			/* just fetch the TID */
  	ROW_MARK_COPY				/* physically copy the row value */
  } RowMarkType;
  
! #define RowMarkRequiresRowShareLock(marktype)  ((marktype) <= ROW_MARK_SHARE)
  
  /*
   * PlanRowMark -
   *	   plan-time representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE, we create a separate
   * PlanRowMark node for each non-target relation in the query.	Relations that
!  * are not specified as FOR UPDATE/SHARE are marked ROW_MARK_REFERENCE (if
   * real tables) or ROW_MARK_COPY (if not).
   *
   * Initially all PlanRowMarks have rti == prti and isParent == false.
--- 734,753 ----
  {
  	ROW_MARK_EXCLUSIVE,			/* obtain exclusive tuple lock */
  	ROW_MARK_SHARE,				/* obtain shared tuple lock */
+ 	ROW_MARK_KEYLOCK,			/* obtain keylock tuple lock */
  	ROW_MARK_REFERENCE,			/* just fetch the TID */
  	ROW_MARK_COPY				/* physically copy the row value */
  } RowMarkType;
  
! #define RowMarkRequiresRowShareLock(marktype)  ((marktype) <= ROW_MARK_KEYLOCK)
  
  /*
   * PlanRowMark -
   *	   plan-time representation of FOR UPDATE/SHARE clauses
   *
!  * When doing UPDATE, DELETE, or SELECT FOR UPDATE/SHARE/KEY LOCK, we create a separate
   * PlanRowMark node for each non-target relation in the query.	Relations that
!  * are not specified as FOR UPDATE/SHARE/KEY LOCK are marked ROW_MARK_REFERENCE (if
   * real tables) or ROW_MARK_COPY (if not).
   *
   * Initially all PlanRowMarks have rti == prti and isParent == false.
*** a/src/include/parser/analyze.h
--- b/src/include/parser/analyze.h
***************
*** 31,36 **** extern bool analyze_requires_snapshot(Node *parseTree);
  
  extern void CheckSelectLocking(Query *qry);
  extern void applyLockingClause(Query *qry, Index rtindex,
! 				   bool forUpdate, bool noWait, bool pushedDown);
  
  #endif   /* ANALYZE_H */
--- 31,36 ----
  
  extern void CheckSelectLocking(Query *qry);
  extern void applyLockingClause(Query *qry, Index rtindex,
! 				   LockClauseStrength strength, bool noWait, bool pushedDown);
  
  #endif   /* ANALYZE_H */
*** a/src/include/utils/rel.h
--- b/src/include/utils/rel.h
***************
*** 103,108 **** typedef struct RelationData
--- 103,109 ----
  	Oid			rd_id;			/* relation's object id */
  	List	   *rd_indexlist;	/* list of OIDs of indexes on relation */
  	Bitmapset  *rd_indexattr;	/* identifies columns used in indexes */
+ 	Bitmapset  *rd_keyattr;		/* cols that can be ref'd by foreign keys */
  	Oid			rd_oidindex;	/* OID of unique index on OID, if any */
  	LockInfoData rd_lockInfo;	/* lock mgr's info for locking relation */
  	RuleLock   *rd_rules;		/* rewrite rules */
*** a/src/include/utils/relcache.h
--- b/src/include/utils/relcache.h
***************
*** 42,48 **** extern List *RelationGetIndexList(Relation relation);
  extern Oid	RelationGetOidIndex(Relation relation);
  extern List *RelationGetIndexExpressions(Relation relation);
  extern List *RelationGetIndexPredicate(Relation relation);
! extern Bitmapset *RelationGetIndexAttrBitmap(Relation relation);
  extern void RelationGetExclusionInfo(Relation indexRelation,
  						 Oid **operators,
  						 Oid **procs,
--- 42,48 ----
  extern Oid	RelationGetOidIndex(Relation relation);
  extern List *RelationGetIndexExpressions(Relation relation);
  extern List *RelationGetIndexPredicate(Relation relation);
! extern Bitmapset *RelationGetIndexAttrBitmap(Relation relation, bool keyAttrs);
  extern void RelationGetExclusionInfo(Relation indexRelation,
  						 Oid **operators,
  						 Oid **procs,
*** a/src/test/isolation/expected/fk-contention.out
--- b/src/test/isolation/expected/fk-contention.out
***************
*** 7,15 **** step upd:  UPDATE foo SET b = 'Hello World';
  
  starting permutation: ins upd com
  step ins:  INSERT INTO bar VALUES (42); 
! step upd:  UPDATE foo SET b = 'Hello World';  <waiting ...>
  step com:  COMMIT; 
- step upd: <... completed>
  
  starting permutation: upd ins com
  step upd:  UPDATE foo SET b = 'Hello World'; 
--- 7,14 ----
  
  starting permutation: ins upd com
  step ins:  INSERT INTO bar VALUES (42); 
! step upd:  UPDATE foo SET b = 'Hello World'; 
  step com:  COMMIT; 
  
  starting permutation: upd ins com
  step upd:  UPDATE foo SET b = 'Hello World'; 
*** a/src/test/isolation/expected/fk-deadlock.out
--- b/src/test/isolation/expected/fk-deadlock.out
***************
*** 20,60 **** step s2c:  COMMIT;
  starting permutation: s1i s2i s1u s2u s1c s2c
  step s1i:  INSERT INTO child VALUES (1, 1); 
  step s2i:  INSERT INTO child VALUES (2, 1); 
! step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
! step s2u:  UPDATE parent SET aux = 'baz'; 
! step s1u: <... completed>
! ERROR:  deadlock detected
  step s1c:  COMMIT; 
  step s2c:  COMMIT; 
  
  starting permutation: s1i s2i s2u s1u s2c s1c
  step s1i:  INSERT INTO child VALUES (1, 1); 
  step s2i:  INSERT INTO child VALUES (2, 1); 
! step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
! step s1u:  UPDATE parent SET aux = 'bar'; 
! ERROR:  deadlock detected
! step s2u: <... completed>
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
  starting permutation: s2i s1i s1u s2u s1c s2c
  step s2i:  INSERT INTO child VALUES (2, 1); 
  step s1i:  INSERT INTO child VALUES (1, 1); 
! step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
! step s2u:  UPDATE parent SET aux = 'baz'; 
! step s1u: <... completed>
! ERROR:  deadlock detected
  step s1c:  COMMIT; 
  step s2c:  COMMIT; 
  
  starting permutation: s2i s1i s2u s1u s2c s1c
  step s2i:  INSERT INTO child VALUES (2, 1); 
  step s1i:  INSERT INTO child VALUES (1, 1); 
! step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
! step s1u:  UPDATE parent SET aux = 'bar'; 
! ERROR:  deadlock detected
! step s2u: <... completed>
  step s2c:  COMMIT; 
  step s1c:  COMMIT; 
  
  starting permutation: s2i s2u s1i s2c s1u s1c
--- 20,56 ----
  starting permutation: s1i s2i s1u s2u s1c s2c
  step s1i:  INSERT INTO child VALUES (1, 1); 
  step s2i:  INSERT INTO child VALUES (2, 1); 
! step s1u:  UPDATE parent SET aux = 'bar'; 
! step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
  step s1c:  COMMIT; 
+ step s2u: <... completed>
  step s2c:  COMMIT; 
  
  starting permutation: s1i s2i s2u s1u s2c s1c
  step s1i:  INSERT INTO child VALUES (1, 1); 
  step s2i:  INSERT INTO child VALUES (2, 1); 
! step s2u:  UPDATE parent SET aux = 'baz'; 
! step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
  step s2c:  COMMIT; 
+ step s1u: <... completed>
  step s1c:  COMMIT; 
  
  starting permutation: s2i s1i s1u s2u s1c s2c
  step s2i:  INSERT INTO child VALUES (2, 1); 
  step s1i:  INSERT INTO child VALUES (1, 1); 
! step s1u:  UPDATE parent SET aux = 'bar'; 
! step s2u:  UPDATE parent SET aux = 'baz';  <waiting ...>
  step s1c:  COMMIT; 
+ step s2u: <... completed>
  step s2c:  COMMIT; 
  
  starting permutation: s2i s1i s2u s1u s2c s1c
  step s2i:  INSERT INTO child VALUES (2, 1); 
  step s1i:  INSERT INTO child VALUES (1, 1); 
! step s2u:  UPDATE parent SET aux = 'baz'; 
! step s1u:  UPDATE parent SET aux = 'bar';  <waiting ...>
  step s2c:  COMMIT; 
+ step s1u: <... completed>
  step s1c:  COMMIT; 
  
  starting permutation: s2i s2u s1i s2c s1u s1c
*** a/src/test/isolation/expected/fk-deadlock2.out
--- b/src/test/isolation/expected/fk-deadlock2.out
***************
*** 100,107 **** step s1c:  COMMIT;
  starting permutation: s2u1 s2u2 s1u1 s2c s1u2 s1c
  step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
! step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1;  <waiting ...>
  step s2c:  COMMIT; 
- step s1u1: <... completed>
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s1c:  COMMIT; 
--- 100,106 ----
  starting permutation: s2u1 s2u2 s1u1 s2c s1u2 s1c
  step s2u1:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s2u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
! step s1u1:  UPDATE A SET Col1 = 1 WHERE AID = 1; 
  step s2c:  COMMIT; 
  step s1u2:  UPDATE B SET Col2 = 1 WHERE BID = 2; 
  step s1c:  COMMIT;