v7-0001_add_log_lock_failure.patch

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index e55700f35b8..ecd5e36ffbc 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -7678,6 +7678,25 @@ log_line_prefix = '%m [%p] %q%u@%d/%a '
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-log-lock-failure" xreflabel="log_lock_failure">
+      <term><varname>log_lock_failure</varname> (<type>boolean</type>)
+      <indexterm>
+       <primary><varname>log_lock_failure</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Controls whether a log message is produced when a lock acquisition
+        fails.  This is useful for analyzing the causes of lock failures.
+        Currently, only lock failures due to <literal>SELECT NOWAIT</literal>
+        is supported.  The default is <literal>off</literal>.  Only superusers
+        and users with the appropriate <literal>SET</literal> privilege can change
+        this setting.
+       </para>
+      </listitem>
+     </varlistentry>
+
+
      <varlistentry id="guc-log-recovery-conflict-waits" xreflabel="log_recovery_conflict_waits">
       <term><varname>log_recovery_conflict_waits</varname> (<type>boolean</type>)
       <indexterm>
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index fa7935a0ed3..e640e55a285 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -98,7 +98,8 @@ static void MultiXactIdWait(MultiXactId multi, MultiXactStatus status, uint16 in
 							Relation rel, ItemPointer ctid, XLTW_Oper oper,
 							int *remaining);
 static bool ConditionalMultiXactIdWait(MultiXactId multi, MultiXactStatus status,
-									   uint16 infomask, Relation rel, int *remaining);
+									   uint16 infomask, Relation rel, int *remaining,
+									   LockHoldersAndWaiters *lockHoldersAndWaiters);
 static void index_delete_sort(TM_IndexDeleteOp *delstate);
 static int	bottomup_sort_and_shrink(TM_IndexDeleteOp *delstate);
 static XLogRecPtr log_heap_new_cid(Relation relation, HeapTuple tup);
@@ -162,8 +163,8 @@ static const struct
 	LockTuple((rel), (tup), tupleLockExtraInfo[mode].hwlock)
 #define UnlockTupleTuplock(rel, tup, mode) \
 	UnlockTuple((rel), (tup), tupleLockExtraInfo[mode].hwlock)
-#define ConditionalLockTupleTuplock(rel, tup, mode) \
-	ConditionalLockTuple((rel), (tup), tupleLockExtraInfo[mode].hwlock)
+#define ConditionalLockTupleTuplock(rel, tup, mode, buf) \
+	ConditionalLockTuple((rel), (tup), tupleLockExtraInfo[mode].hwlock, (buf))
 
 #ifdef USE_PREFETCH
 /*
@@ -4894,7 +4895,7 @@ l3:
 					case LockWaitSkip:
 						if (!ConditionalMultiXactIdWait((MultiXactId) xwait,
 														status, infomask, relation,
-														NULL))
+														NULL, NULL))
 						{
 							result = TM_WouldBlock;
 							/* recovery code expects to have buffer lock held */
@@ -4903,13 +4904,25 @@ l3:
 						}
 						break;
 					case LockWaitError:
+						LockHoldersAndWaiters *lockHoldersAndWaiters = CreateLockHoldersAndWaiters();
 						if (!ConditionalMultiXactIdWait((MultiXactId) xwait,
 														status, infomask, relation,
-														NULL))
+														NULL, lockHoldersAndWaiters))
 							ereport(ERROR,
-									(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-									 errmsg("could not obtain lock on row in relation \"%s\"",
-											RelationGetRelationName(relation))));
+								(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
+								 errmsg("could not obtain lock on row in relation \"%s\"",
+										RelationGetRelationName(relation)),
+								 errdetail_log_plural(
+									"process %d could not obtain %s on %s, "
+									"Process holding the lock: %s, Wait queue: %s.",
+									"process %d could not obtain %s on %s, "
+									"Processes holding the lock: %s, Wait queue: %s.",
+									*lockHoldersAndWaiters->lockHoldersNum,
+									MyProcPid,
+									lockHoldersAndWaiters->modename,
+									lockHoldersAndWaiters->describe_locktag->data,
+									lockHoldersAndWaiters->lock_holders_sbuf->data,
+									lockHoldersAndWaiters->lock_waiters_sbuf->data)));
 
 						break;
 				}
@@ -4934,7 +4947,7 @@ l3:
 										  XLTW_Lock);
 						break;
 					case LockWaitSkip:
-						if (!ConditionalXactLockTableWait(xwait))
+						if (!ConditionalXactLockTableWait(xwait, NULL))
 						{
 							result = TM_WouldBlock;
 							/* recovery code expects to have buffer lock held */
@@ -4943,11 +4956,22 @@ l3:
 						}
 						break;
 					case LockWaitError:
-						if (!ConditionalXactLockTableWait(xwait))
+						LockHoldersAndWaiters *lockHoldersAndWaiters = CreateLockHoldersAndWaiters();
+						if (!ConditionalXactLockTableWait(xwait, lockHoldersAndWaiters))
 							ereport(ERROR,
-									(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-									 errmsg("could not obtain lock on row in relation \"%s\"",
-											RelationGetRelationName(relation))));
+								(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
+									errmsg("could not obtain lock on row in relation \"%s\"",
+										RelationGetRelationName(relation)),
+									errdetail_log_plural("process %d could not obtain %s on %s, "
+											"Process holding the lock: %s, Wait queue: %s.",
+											"process %d could not obtain %s on %s, "
+											"Processes holding the lock: %s, Wait queue: %s.",
+											*lockHoldersAndWaiters->lockHoldersNum,
+											MyProcPid,
+											lockHoldersAndWaiters->modename,
+											lockHoldersAndWaiters->describe_locktag->data,
+											lockHoldersAndWaiters->lock_holders_sbuf->data,
+											lockHoldersAndWaiters->lock_waiters_sbuf->data)));
 						break;
 				}
 			}
@@ -5203,16 +5227,27 @@ heap_acquire_tuplock(Relation relation, ItemPointer tid, LockTupleMode mode,
 			break;
 
 		case LockWaitSkip:
-			if (!ConditionalLockTupleTuplock(relation, tid, mode))
+			if (!ConditionalLockTupleTuplock(relation, tid, mode, NULL))
 				return false;
 			break;
 
 		case LockWaitError:
-			if (!ConditionalLockTupleTuplock(relation, tid, mode))
+			LockHoldersAndWaiters *lockHoldersAndWaiters = CreateLockHoldersAndWaiters();
+			if (!ConditionalLockTupleTuplock(relation, tid, mode, lockHoldersAndWaiters))
 				ereport(ERROR,
-						(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-						 errmsg("could not obtain lock on row in relation \"%s\"",
-								RelationGetRelationName(relation))));
+					(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
+					 errmsg("could not obtain lock on row in relation \"%s\"",
+							RelationGetRelationName(relation)),
+					 errdetail_log_plural("process %d could not obtain %s on %s, "
+								"Process holding the lock: %s, Wait queue: %s.",
+								"process %d could not obtain %s on %s, "
+								"Processes holding the lock: %s, Wait queue: %s.",
+								*lockHoldersAndWaiters->lockHoldersNum,
+								MyProcPid,
+								lockHoldersAndWaiters->modename,
+								lockHoldersAndWaiters->describe_locktag->data,
+								lockHoldersAndWaiters->lock_holders_sbuf->data,
+								lockHoldersAndWaiters->lock_waiters_sbuf->data)));
 			break;
 	}
 	*have_tuple_lock = true;
@@ -7602,7 +7637,7 @@ static bool
 Do_MultiXactIdWait(MultiXactId multi, MultiXactStatus status,
 				   uint16 infomask, bool nowait,
 				   Relation rel, ItemPointer ctid, XLTW_Oper oper,
-				   int *remaining)
+				   int *remaining, LockHoldersAndWaiters *buf)
 {
 	bool		result = true;
 	MultiXactMember *members;
@@ -7649,7 +7684,7 @@ Do_MultiXactIdWait(MultiXactId multi, MultiXactStatus status,
 			 */
 			if (nowait)
 			{
-				result = ConditionalXactLockTableWait(memxid);
+				result = ConditionalXactLockTableWait(memxid, buf);
 				if (!result)
 					break;
 			}
@@ -7682,7 +7717,7 @@ MultiXactIdWait(MultiXactId multi, MultiXactStatus status, uint16 infomask,
 				int *remaining)
 {
 	(void) Do_MultiXactIdWait(multi, status, infomask, false,
-							  rel, ctid, oper, remaining);
+							  rel, ctid, oper, remaining, NULL);
 }
 
 /*
@@ -7700,10 +7735,11 @@ MultiXactIdWait(MultiXactId multi, MultiXactStatus status, uint16 infomask,
  */
 static bool
 ConditionalMultiXactIdWait(MultiXactId multi, MultiXactStatus status,
-						   uint16 infomask, Relation rel, int *remaining)
+						   uint16 infomask, Relation rel, int *remaining,
+						   LockHoldersAndWaiters *lockHoldersAndWaiters)
 {
 	return Do_MultiXactIdWait(multi, status, infomask, true,
-							  rel, NULL, XLTW_None, remaining);
+							  rel, NULL, XLTW_None, remaining, lockHoldersAndWaiters);
 }
 
 /*
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index e78682c3cef..2ad65931f50 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -455,16 +455,28 @@ tuple_lock_retry:
 												  XLTW_FetchUpdated);
 								break;
 							case LockWaitSkip:
-								if (!ConditionalXactLockTableWait(SnapshotDirty.xmax))
+								if (!ConditionalXactLockTableWait(SnapshotDirty.xmax, NULL))
 									/* skip instead of waiting */
 									return TM_WouldBlock;
 								break;
 							case LockWaitError:
-								if (!ConditionalXactLockTableWait(SnapshotDirty.xmax))
+								LockHoldersAndWaiters *lockHoldersAndWaiters = CreateLockHoldersAndWaiters();
+								if (!ConditionalXactLockTableWait(SnapshotDirty.xmax, lockHoldersAndWaiters))
 									ereport(ERROR,
-											(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-											 errmsg("could not obtain lock on row in relation \"%s\"",
-													RelationGetRelationName(relation))));
+										(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
+										 errmsg("could not obtain lock on row in relation \"%s\"",
+												RelationGetRelationName(relation)),
+										 errdetail_log_plural(
+											"process %d could not obtain %s on %s, "
+											"Process holding the lock: %s, Wait queue: %s.",
+											"process %d could not obtain %s on %s, "
+											"Processes holding the lock: %s, Wait queue: %s.",
+											*lockHoldersAndWaiters->lockHoldersNum,
+											MyProcPid,
+											lockHoldersAndWaiters->modename,
+											lockHoldersAndWaiters->describe_locktag->data,
+											lockHoldersAndWaiters->lock_holders_sbuf->data,
+											lockHoldersAndWaiters->lock_waiters_sbuf->data)));
 								break;
 						}
 						continue;	/* loop back to repeat heap_fetch */
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index 68271cb6408..46f039ad389 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -22,6 +22,7 @@
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "storage/lmgr.h"
+#include "storage/lock.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "utils/inval.h"
@@ -112,7 +113,8 @@ LockRelationOid(Oid relid, LOCKMODE lockmode)
 
 	SetLocktagRelationOid(&tag, relid);
 
-	res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
+	res = LockAcquireExtended(&tag, lockmode, false, false, true,
+							&locallock, NULL);
 
 	/*
 	 * Now that we have the lock, check for invalidation messages, so that we
@@ -155,7 +157,8 @@ ConditionalLockRelationOid(Oid relid, LOCKMODE lockmode)
 
 	SetLocktagRelationOid(&tag, relid);
 
-	res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
+	res = LockAcquireExtended(&tag, lockmode, false, true, true,
+							&locallock, NULL);
 
 	if (res == LOCKACQUIRE_NOT_AVAIL)
 		return false;
@@ -188,7 +191,8 @@ LockRelationId(LockRelId *relid, LOCKMODE lockmode)
 
 	SET_LOCKTAG_RELATION(tag, relid->dbId, relid->relId);
 
-	res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
+	res = LockAcquireExtended(&tag, lockmode, false, false, true,
+							&locallock, NULL);
 
 	/*
 	 * Now that we have the lock, check for invalidation messages; see notes
@@ -250,7 +254,8 @@ LockRelation(Relation relation, LOCKMODE lockmode)
 						 relation->rd_lockInfo.lockRelId.dbId,
 						 relation->rd_lockInfo.lockRelId.relId);
 
-	res = LockAcquireExtended(&tag, lockmode, false, false, true, &locallock);
+	res = LockAcquireExtended(&tag, lockmode, false, false, true,
+							&locallock, NULL);
 
 	/*
 	 * Now that we have the lock, check for invalidation messages; see notes
@@ -281,7 +286,8 @@ ConditionalLockRelation(Relation relation, LOCKMODE lockmode)
 						 relation->rd_lockInfo.lockRelId.dbId,
 						 relation->rd_lockInfo.lockRelId.relId);
 
-	res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
+	res = LockAcquireExtended(&tag, lockmode, false, true, true,
+							&locallock, NULL);
 
 	if (res == LOCKACQUIRE_NOT_AVAIL)
 		return false;
@@ -574,7 +580,8 @@ LockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
  * Returns true iff the lock was acquired.
  */
 bool
-ConditionalLockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
+ConditionalLockTuple(Relation relation, ItemPointer tid,
+					LOCKMODE lockmode, LockHoldersAndWaiters *lockHoldersAndWaiters)
 {
 	LOCKTAG		tag;
 
@@ -584,7 +591,8 @@ ConditionalLockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode)
 					  ItemPointerGetBlockNumber(tid),
 					  ItemPointerGetOffsetNumber(tid));
 
-	return (LockAcquire(&tag, lockmode, false, true) != LOCKACQUIRE_NOT_AVAIL);
+	return (LockAcquireExtended(&tag, lockmode, false, true, true,
+			NULL, lockHoldersAndWaiters) != LOCKACQUIRE_NOT_AVAIL);
 }
 
 /*
@@ -726,7 +734,8 @@ XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid,
  * Returns true if the lock was acquired.
  */
 bool
-ConditionalXactLockTableWait(TransactionId xid)
+ConditionalXactLockTableWait(TransactionId xid,
+							LockHoldersAndWaiters *lockHoldersAndWaiters)
 {
 	LOCKTAG		tag;
 	bool		first = true;
@@ -738,7 +747,9 @@ ConditionalXactLockTableWait(TransactionId xid)
 
 		SET_LOCKTAG_TRANSACTION(tag, xid);
 
-		if (LockAcquire(&tag, ShareLock, false, true) == LOCKACQUIRE_NOT_AVAIL)
+		if (LockAcquireExtended(&tag, ShareLock, false, true, true,
+								NULL, lockHoldersAndWaiters)
+			== LOCKACQUIRE_NOT_AVAIL)
 			return false;
 
 		LockRelease(&tag, ShareLock, false);
@@ -1027,7 +1038,8 @@ ConditionalLockDatabaseObject(Oid classid, Oid objid, uint16 objsubid,
 					   objid,
 					   objsubid);
 
-	res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
+	res = LockAcquireExtended(&tag, lockmode, false, true, true,
+							&locallock, NULL);
 
 	if (res == LOCKACQUIRE_NOT_AVAIL)
 		return false;
@@ -1106,7 +1118,8 @@ ConditionalLockSharedObject(Oid classid, Oid objid, uint16 objsubid,
 					   objid,
 					   objsubid);
 
-	res = LockAcquireExtended(&tag, lockmode, false, true, true, &locallock);
+	res = LockAcquireExtended(&tag, lockmode, false, true, true,
+							&locallock, NULL);
 
 	if (res == LOCKACQUIRE_NOT_AVAIL)
 		return false;
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 11b4d1085bb..666cbce5acf 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -39,6 +39,7 @@
 #include "access/xlogutils.h"
 #include "miscadmin.h"
 #include "pg_trace.h"
+#include "storage/lmgr.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "storage/spin.h"
@@ -48,8 +49,9 @@
 #include "utils/resowner.h"
 
 
-/* This configuration variable is used to set the lock table size */
-int			max_locks_per_xact; /* set by guc.c */
+/* GUC variables. */
+int			max_locks_per_xact; /* This configuration variable is used to set the lock table size */
+bool		log_lock_failure = false;
 
 #define NLOCKENTS() \
 	mul_size(max_locks_per_xact, add_size(MaxBackends, max_prepared_xacts))
@@ -806,7 +808,7 @@ LockAcquire(const LOCKTAG *locktag,
 			bool dontWait)
 {
 	return LockAcquireExtended(locktag, lockmode, sessionLock, dontWait,
-							   true, NULL);
+							   true, NULL, NULL);
 }
 
 /*
@@ -829,7 +831,8 @@ LockAcquireExtended(const LOCKTAG *locktag,
 					bool sessionLock,
 					bool dontWait,
 					bool reportMemoryError,
-					LOCALLOCK **locallockp)
+					LOCALLOCK **locallockp,
+					LockHoldersAndWaiters *lockHoldersAndWaiters)
 {
 	LOCKMETHODID lockmethodid = locktag->locktag_lockmethodid;
 	LockMethod	lockMethodTable;
@@ -1145,6 +1148,24 @@ LockAcquireExtended(const LOCKTAG *locktag,
 
 		if (dontWait)
 		{
+			if (log_lock_failure && lockHoldersAndWaiters)
+			{
+				DescribeLockTag(lockHoldersAndWaiters->describe_locktag,
+								&locallock->tag.lock);
+				lockHoldersAndWaiters->modename =
+					GetLockmodeName(locallock->tag.lock.locktag_lockmethodid,
+									lockmode);
+
+				/* Collect lock holders and waiters */
+				LWLockAcquire(partitionLock, LW_SHARED);
+
+				CollectLockHoldersAndWaiters(locallock,
+					lockHoldersAndWaiters->lock_holders_sbuf,
+					lockHoldersAndWaiters->lock_waiters_sbuf,
+					lockHoldersAndWaiters->lockHoldersNum);
+
+				LWLockRelease(partitionLock);
+			}
 			if (locallockp)
 				*locallockp = NULL;
 			return LOCKACQUIRE_NOT_AVAIL;
@@ -4788,3 +4809,24 @@ LockWaiterCount(const LOCKTAG *locktag)
 
 	return waiters;
 }
+/*
+ * CreateLockHoldersAndWaiters
+ *
+ * initializing the LockHoldersAndWaiters
+ */
+LockHoldersAndWaiters * CreateLockHoldersAndWaiters(void)
+{
+	LockHoldersAndWaiters *buf =
+		(LockHoldersAndWaiters *) palloc(sizeof(LockHoldersAndWaiters));
+	buf->describe_locktag = palloc(sizeof(StringInfoData));
+	buf->lock_waiters_sbuf = palloc(sizeof(StringInfoData));
+	buf->lock_holders_sbuf = palloc(sizeof(StringInfoData));
+	buf->lockHoldersNum = palloc(sizeof(int));
+
+	initStringInfo(buf->describe_locktag);
+	initStringInfo(buf->lock_holders_sbuf);
+	initStringInfo(buf->lock_waiters_sbuf);
+	*(buf->lockHoldersNum) = 0;
+
+	return buf;
+}
\ No newline at end of file
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 49204f91a20..34d6cbbb896 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -1514,10 +1514,6 @@ ProcSleep(LOCALLOCK *locallock)
 			long		secs;
 			int			usecs;
 			long		msecs;
-			dlist_iter	proc_iter;
-			PROCLOCK   *curproclock;
-			bool		first_holder = true,
-						first_waiter = true;
 			int			lockHoldersNum = 0;
 
 			initStringInfo(&buf);
@@ -1533,53 +1529,11 @@ ProcSleep(LOCALLOCK *locallock)
 			msecs = secs * 1000 + usecs / 1000;
 			usecs = usecs % 1000;
 
-			/*
-			 * we loop over the lock's procLocks to gather a list of all
-			 * holders and waiters. Thus we will be able to provide more
-			 * detailed information for lock debugging purposes.
-			 *
-			 * lock->procLocks contains all processes which hold or wait for
-			 * this lock.
-			 */
-
 			LWLockAcquire(partitionLock, LW_SHARED);
 
-			dlist_foreach(proc_iter, &lock->procLocks)
-			{
-				curproclock =
-					dlist_container(PROCLOCK, lockLink, proc_iter.cur);
-
-				/*
-				 * we are a waiter if myProc->waitProcLock == curproclock; we
-				 * are a holder if it is NULL or something different
-				 */
-				if (curproclock->tag.myProc->waitProcLock == curproclock)
-				{
-					if (first_waiter)
-					{
-						appendStringInfo(&lock_waiters_sbuf, "%d",
-										 curproclock->tag.myProc->pid);
-						first_waiter = false;
-					}
-					else
-						appendStringInfo(&lock_waiters_sbuf, ", %d",
-										 curproclock->tag.myProc->pid);
-				}
-				else
-				{
-					if (first_holder)
-					{
-						appendStringInfo(&lock_holders_sbuf, "%d",
-										 curproclock->tag.myProc->pid);
-						first_holder = false;
-					}
-					else
-						appendStringInfo(&lock_holders_sbuf, ", %d",
-										 curproclock->tag.myProc->pid);
-
-					lockHoldersNum++;
-				}
-			}
+			/* Collect lock holders and waiters */
+			CollectLockHoldersAndWaiters(locallock, &lock_holders_sbuf,
+								&lock_waiters_sbuf, &lockHoldersNum);
 
 			LWLockRelease(partitionLock);
 
@@ -1985,3 +1939,76 @@ BecomeLockGroupMember(PGPROC *leader, int pid)
 
 	return ok;
 }
+
+/*
+ * CollectLockHoldersAndWaiters - collect lock holders and waiters for a lock
+ *
+ * The lock table's partition lock must be held on entry and remains held on exit.
+ * Fill lock_holders_sbuf and lock_waiters_sbuf with the PIDs of processes holding
+ * and waiting for the lock, and set lockHoldersNum to the number of lock holders.
+ */
+void
+CollectLockHoldersAndWaiters(LOCALLOCK *locallock, StringInfo lock_holders_sbuf,
+							StringInfo lock_waiters_sbuf, int *lockHoldersNum)
+{
+	bool first_holder = true;
+	bool first_waiter = true;
+	dlist_iter 	proc_iter;
+	PROCLOCK	*curproclock;
+	LOCK		*lock = locallock->lock;
+#ifdef USE_ASSERT_CHECKING
+	{
+		uint32		hashcode = locallock->hashcode;
+		LWLock	   *partitionLock = LockHashPartitionLock(hashcode);
+
+		Assert(LWLockHeldByMe(partitionLock));
+	}
+#endif
+
+	*lockHoldersNum = 0;
+
+	/*
+	* we loop over the lock's procLocks to gather a list of all
+	* holders and waiters. Thus we will be able to provide more
+	* detailed information for lock debugging purposes.
+	*
+	* lock->procLocks contains all processes which hold or wait for
+	* this lock.
+	*/
+	dlist_foreach(proc_iter, &lock->procLocks)
+	{
+		curproclock =
+			dlist_container(PROCLOCK, lockLink, proc_iter.cur);
+
+		/*
+		* we are a waiter if myProc->waitProcLock == curproclock; we
+		* are a holder if it is NULL or something different
+		*/
+		if (curproclock->tag.myProc->waitProcLock == curproclock)
+		{
+			if (first_waiter)
+			{
+				appendStringInfo(lock_waiters_sbuf, "%d",
+									curproclock->tag.myProc->pid);
+				first_waiter = false;
+			}
+			else
+				appendStringInfo(lock_waiters_sbuf, ", %d",
+									curproclock->tag.myProc->pid);
+		}
+		else
+		{
+			if (first_holder)
+			{
+				appendStringInfo(lock_holders_sbuf, "%d",
+									curproclock->tag.myProc->pid);
+				first_holder = false;
+			}
+			else
+				appendStringInfo(lock_holders_sbuf, ", %d",
+									curproclock->tag.myProc->pid);
+
+			(*lockHoldersNum)++;
+		}
+	}
+}
\ No newline at end of file
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index ad25cbb39c5..d7b3ed43920 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -75,6 +75,7 @@
 #include "storage/bufmgr.h"
 #include "storage/bufpage.h"
 #include "storage/large_object.h"
+#include "storage/lmgr.h"
 #include "storage/pg_shmem.h"
 #include "storage/predicate.h"
 #include "storage/procnumber.h"
@@ -1594,6 +1595,15 @@ struct config_bool ConfigureNamesBool[] =
 		false,
 		NULL, NULL, NULL
 	},
+	{
+		{"log_lock_failure", PGC_SUSET, LOGGING_WHAT,
+			gettext_noop("Logs lock failed."),
+			NULL
+		},
+		&log_lock_failure,
+		false,
+		NULL, NULL, NULL
+	},
 	{
 		{"log_recovery_conflict_waits", PGC_SIGHUP, LOGGING_WHAT,
 			gettext_noop("Logs standby recovery conflict waits."),
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 5362ff80519..0b570a81d76 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -608,6 +608,7 @@
 					#   %% = '%'
 					# e.g. '<%u%%%d> '
 #log_lock_waits = off			# log lock waits >= deadlock_timeout
+#log_lock_failure = off			# log lock failed
 #log_recovery_conflict_waits = off	# log standby recovery conflict waits
 					# >= deadlock_timeout
 #log_parameter_max_length = -1		# when logging statements, limit logged
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index 728260c1cc0..d0306229642 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -72,8 +72,8 @@ extern void UnlockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode);
 
 /* Lock a tuple (see heap_lock_tuple before assuming you understand this) */
 extern void LockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode);
-extern bool ConditionalLockTuple(Relation relation, ItemPointer tid,
-								 LOCKMODE lockmode);
+extern bool ConditionalLockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode, 
+								LockHoldersAndWaiters *lockHoldersAndWaiters);
 extern void UnlockTuple(Relation relation, ItemPointer tid, LOCKMODE lockmode);
 
 /* Lock an XID (used to wait for a transaction to finish) */
@@ -81,7 +81,8 @@ extern void XactLockTableInsert(TransactionId xid);
 extern void XactLockTableDelete(TransactionId xid);
 extern void XactLockTableWait(TransactionId xid, Relation rel,
 							  ItemPointer ctid, XLTW_Oper oper);
-extern bool ConditionalXactLockTableWait(TransactionId xid);
+extern bool ConditionalXactLockTableWait(TransactionId xid,
+										 LockHoldersAndWaiters *lockHoldersAndWaiters);
 
 /* Lock VXIDs, specified by conflicting locktags */
 extern void WaitForLockers(LOCKTAG heaplocktag, LOCKMODE lockmode, bool progress);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index 1076995518f..2ba8a0f736d 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -30,6 +30,7 @@ typedef struct PGPROC PGPROC;
 
 /* GUC variables */
 extern PGDLLIMPORT int max_locks_per_xact;
+extern PGDLLIMPORT bool log_lock_failure;
 
 #ifdef LOCK_DEBUG
 extern PGDLLIMPORT int Trace_lock_oidmin;
@@ -515,6 +516,15 @@ typedef enum
 								 * worker */
 } DeadLockState;
 
+typedef struct LockHoldersAndWaiters
+{
+	StringInfoData *describe_locktag,
+				*lock_waiters_sbuf,
+				*lock_holders_sbuf;
+	const char	*modename;
+	int			*lockHoldersNum;
+} LockHoldersAndWaiters;
+
 /*
  * The lockmgr's shared hash tables are partitioned to reduce contention.
  * To determine which partition a given locktag belongs to, compute the tag's
@@ -560,7 +570,8 @@ extern LockAcquireResult LockAcquireExtended(const LOCKTAG *locktag,
 											 bool sessionLock,
 											 bool dontWait,
 											 bool reportMemoryError,
-											 LOCALLOCK **locallockp);
+											 LOCALLOCK **locallockp,
+											 LockHoldersAndWaiters *lockHoldersAndWaiters);
 extern void AbortStrongLockAcquire(void);
 extern void MarkLockClear(LOCALLOCK *locallock);
 extern bool LockRelease(const LOCKTAG *locktag,
@@ -613,6 +624,7 @@ extern void RememberSimpleDeadLock(PGPROC *proc1,
 extern void InitDeadLockChecking(void);
 
 extern int	LockWaiterCount(const LOCKTAG *locktag);
+extern LockHoldersAndWaiters * CreateLockHoldersAndWaiters(void);
 
 #ifdef LOCK_DEBUG
 extern void DumpLocks(PGPROC *proc);
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 20777f7d5ae..05b5b0f3337 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -497,4 +497,7 @@ extern PGPROC *AuxiliaryPidGetProc(int pid);
 extern void BecomeLockGroupLeader(void);
 extern bool BecomeLockGroupMember(PGPROC *leader, int pid);
 
+extern void CollectLockHoldersAndWaiters(LOCALLOCK *locallock, StringInfo lock_holders_sbuf,
+								StringInfo lock_waiters_sbuf, int *lockHoldersNum);
+
 #endif							/* _PROC_H_ */
diff --git a/src/test/regress/pg_regress.c b/src/test/regress/pg_regress.c
index 5d85dcc62f0..fe568b52b5d 100644
--- a/src/test/regress/pg_regress.c
+++ b/src/test/regress/pg_regress.c
@@ -2404,6 +2404,7 @@ regression_main(int argc, char *argv[],
 		fputs("log_checkpoints = on\n", pg_conf);
 		fputs("log_line_prefix = '%m %b[%p] %q%a '\n", pg_conf);
 		fputs("log_lock_waits = on\n", pg_conf);
+		fputs("log_lock_failure = on\n", pg_conf);
 		fputs("log_temp_files = 128kB\n", pg_conf);
 		fputs("max_prepared_transactions = 2\n", pg_conf);