diff --git a/src/backend/access/heap/README.tuplock b/src/backend/access/heap/README.tuplock
index b2f3a4ce90..d7a085366a 100644
--- a/src/backend/access/heap/README.tuplock
+++ b/src/backend/access/heap/README.tuplock
@@ -36,6 +36,17 @@ do LockTuple as well, if there is any conflict, to ensure that they don't
 starve out waiting exclusive-lockers.  However, if there is not any active
 conflict for a tuple, we don't incur any extra overhead.
 
+We make an exception from the rule above for those lockers that already hold
+some lock on a tuple and attempt to acquire a stronger one on it. We have to
+let them skip the LockTuple() call even when there are conflicts, provided
+that the target tuple is being locked, updated or deleted by multiple sessions
+concurrently. If we didn't make such an exception we could have a deadlock,
+e.g. between a session that has recorded its weaker lock in the tuple header
+and is waiting on a LockTuple() call to upgrade to the stronger lock level and
+another one that has already acquired the tuple-level lock from the
+LockTuple() but at the same time waiting for the first session transaction to
+finish due to a conflicting lock level.
+
 We provide four levels of tuple locking strength: SELECT FOR UPDATE obtains an
 exclusive lock which prevents any kind of modification of the tuple. This is
 the lock level that is implicitly taken by DELETE operations, and also by
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 8ac0f8a513..45100bab1d 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -95,7 +95,7 @@ static void GetMultiXactIdHintBits(MultiXactId multi, uint16 *new_infomask,
 static TransactionId MultiXactIdGetUpdateXid(TransactionId xmax,
 											 uint16 t_infomask);
 static bool DoesMultiXactIdConflict(MultiXactId multi, uint16 infomask,
-									LockTupleMode lockmode);
+									LockTupleMode lockmode, bool *has_current_xid);
 static void MultiXactIdWait(MultiXactId multi, MultiXactStatus status, uint16 infomask,
 							Relation rel, ItemPointer ctid, XLTW_Oper oper,
 							int *remaining);
@@ -2547,15 +2547,22 @@ l1:
 		 */
 		if (infomask & HEAP_XMAX_IS_MULTI)
 		{
+			bool has_current_xid = false;
 			/* wait for multixact */
 			if (DoesMultiXactIdConflict((MultiXactId) xwait, infomask,
-										LockTupleExclusive))
+										LockTupleExclusive, &has_current_xid))
 			{
 				LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
 
-				/* acquire tuple lock, if necessary */
-				heap_acquire_tuplock(relation, &(tp.t_self), LockTupleExclusive,
-									 LockWaitBlock, &have_tuple_lock);
+				/*
+				 * Acquire the lock, if necessary. Avoid acquiring it when
+				 * getting the stronger lock while holding a weaker one to
+				 * prevent deadlocks with other lockers that already have
+				 * a tuple lock and waiting for our transaction to finish.
+				 */
+				if (!has_current_xid)
+					heap_acquire_tuplock(relation, &(tp.t_self), LockTupleExclusive,
+										 LockWaitBlock, &have_tuple_lock);
 
 				/* wait for multixact */
 				MultiXactIdWait((MultiXactId) xwait, MultiXactStatusUpdate, infomask,
@@ -3126,15 +3133,23 @@ l2:
 		{
 			TransactionId update_xact;
 			int			remain;
+			bool 		has_current_xid = false;
 
 			if (DoesMultiXactIdConflict((MultiXactId) xwait, infomask,
-										*lockmode))
+										*lockmode, &has_current_xid))
 			{
 				LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
 
-				/* acquire tuple lock, if necessary */
-				heap_acquire_tuplock(relation, &(oldtup.t_self), *lockmode,
-									 LockWaitBlock, &have_tuple_lock);
+				/*
+				 * Acquire the lock, if necessary. Avoid acquiring it when
+				 * getting the stronger lock while holding a weaker one to
+				 * prevent deadlocks with other lockers that already have
+				 * a tuple lock and waiting for our transaction to finish.
+				 */
+
+				if (!has_current_xid)
+					heap_acquire_tuplock(relation, &(oldtup.t_self), *lockmode,
+										 LockWaitBlock, &have_tuple_lock);
 
 				/* wait for multixact */
 				MultiXactIdWait((MultiXactId) xwait, mxact_status, infomask,
@@ -4028,6 +4043,7 @@ l3:
 		uint16		infomask;
 		uint16		infomask2;
 		bool		require_sleep;
+		bool		skip_tuple_lock;
 		ItemPointerData t_ctid;
 
 		/* must copy state data before unlocking buffer */
@@ -4053,6 +4069,7 @@ l3:
 		if (first_time)
 		{
 			first_time = false;
+			skip_tuple_lock = false;
 
 			if (infomask & HEAP_XMAX_IS_MULTI)
 			{
@@ -4081,6 +4098,21 @@ l3:
 						result = TM_Ok;
 						goto out_unlocked;
 					}
+					else
+					{
+						/*
+						 * Avoid acquiring the tuple level lock.
+						 *
+						 * Otherwise, when promoting a weaker lock, we may
+						 * deadlock with another potential locker that has
+						 * acquired a conflicting row lock and waits for our
+						 * transaction to finish.
+						 *
+						 * Note that we will wait for the multixact
+						 * if required to avoid acquiring conflicting locks.
+						 */
+						skip_tuple_lock = true;
+					}
 				}
 
 				if (members)
@@ -4235,7 +4267,7 @@ l3:
 			if (infomask & HEAP_XMAX_IS_MULTI)
 			{
 				if (!DoesMultiXactIdConflict((MultiXactId) xwait, infomask,
-											 mode))
+											 mode, NULL))
 				{
 					/*
 					 * No conflict, but if the xmax changed under us in the
@@ -4318,7 +4350,8 @@ l3:
 			 * this arranges that we stay at the head of the line while
 			 * rechecking tuple state.
 			 */
-			if (!heap_acquire_tuplock(relation, tid, mode, wait_policy,
+			if (!skip_tuple_lock &&
+				!heap_acquire_tuplock(relation, tid, mode, wait_policy,
 									  &have_tuple_lock))
 			{
 				/*
@@ -6516,10 +6549,12 @@ HeapTupleGetUpdateXid(HeapTupleHeader tuple)
  * tuple lock of the given strength?
  *
  * The passed infomask pairs up with the given multixact in the tuple header.
+ * has_current_xid is set when not-NULL if the given multixact contains current
+ * transaction.
  */
 static bool
 DoesMultiXactIdConflict(MultiXactId multi, uint16 infomask,
-						LockTupleMode lockmode)
+						LockTupleMode lockmode, bool *has_current_xid)
 {
 	int			nmembers;
 	MultiXactMember *members;
@@ -6540,15 +6575,24 @@ DoesMultiXactIdConflict(MultiXactId multi, uint16 infomask,
 			TransactionId memxid;
 			LOCKMODE	memlockmode;
 
-			memlockmode = LOCKMODE_from_mxstatus(members[i].status);
+			if (result && (has_current_xid == NULL || *has_current_xid))
+				break;
 
-			/* ignore members that don't conflict with the lock we want */
-			if (!DoLockModesConflict(memlockmode, wanted))
-				continue;
+			memlockmode = LOCKMODE_from_mxstatus(members[i].status);
 
 			/* ignore members from current xact */
 			memxid = members[i].xid;
 			if (TransactionIdIsCurrentTransactionId(memxid))
+			{
+				if (has_current_xid != NULL)
+					*has_current_xid = true;
+				continue;
+			}
+			else if (result)
+				continue;
+
+			/* ignore members that don't conflict with the lock we want */
+			if (!DoLockModesConflict(memlockmode, wanted))
 				continue;
 
 			if (ISUPDATE_from_mxstatus(members[i].status))
@@ -6567,10 +6611,11 @@ DoesMultiXactIdConflict(MultiXactId multi, uint16 infomask,
 			/*
 			 * Whatever remains are either live lockers that conflict with our
 			 * wanted lock, and updaters that are not aborted.  Those conflict
-			 * with what we want, so return true.
+			 * with what we want. Return true on the next loop iteration unless
+			 * we need to look for the current transaction among the multixact
+			 * members.
 			 */
 			result = true;
-			break;
 		}
 		pfree(members);
 	}
diff --git a/src/test/isolation/expected/rowlock-upgrade-deadlock.out b/src/test/isolation/expected/rowlock-upgrade-deadlock.out
new file mode 100644
index 0000000000..99d12b125b
--- /dev/null
+++ b/src/test/isolation/expected/rowlock-upgrade-deadlock.out
@@ -0,0 +1,110 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s1_share s2_for_update s3_share s3_for_update s1_rollback s3_rollback s2_rollback
+step s1_share: select id from tlu_job where id = 1 for share;
+id             
+
+1              
+step s2_for_update: select id from tlu_job where id = 1 for update; <waiting ...>
+step s3_share: select id from tlu_job where id = 1 for share;
+id             
+
+1              
+step s3_for_update: select id from tlu_job where id = 1 for update; <waiting ...>
+step s1_rollback: rollback;
+step s3_for_update: <... completed>
+id             
+
+1              
+step s3_rollback: rollback;
+step s2_for_update: <... completed>
+id             
+
+1              
+step s2_rollback: rollback;
+
+starting permutation: s1_keyshare s2_for_update s3_keyshare s1_update s3_update s1_rollback s3_rollback s2_rollback
+step s1_keyshare: select id from tlu_job where id = 1 for key share;
+id             
+
+1              
+step s2_for_update: select id from tlu_job where id = 1 for update; <waiting ...>
+step s3_keyshare: select id from tlu_job where id = 1 for key share;
+id             
+
+1              
+step s1_update: update tlu_job set name = 'b' where id = 1;
+step s3_update: update tlu_job set name = 'b' where id = 1; <waiting ...>
+step s1_rollback: rollback;
+step s3_update: <... completed>
+step s3_rollback: rollback;
+step s2_for_update: <... completed>
+id             
+
+1              
+step s2_rollback: rollback;
+
+starting permutation: s1_keyshare s2_for_update s3_keyshare s3_delete s1_rollback s3_rollback s2_rollback
+step s1_keyshare: select id from tlu_job where id = 1 for key share;
+id             
+
+1              
+step s2_for_update: select id from tlu_job where id = 1 for update; <waiting ...>
+step s3_keyshare: select id from tlu_job where id = 1 for key share;
+id             
+
+1              
+step s3_delete: delete from tlu_job where id = 1; <waiting ...>
+step s1_rollback: rollback;
+step s3_delete: <... completed>
+step s3_rollback: rollback;
+step s2_for_update: <... completed>
+id             
+
+1              
+step s2_rollback: rollback;
+
+starting permutation: s1_share s2_for_update s3_for_update s1_rollback s2_rollback s3_rollback
+step s1_share: select id from tlu_job where id = 1 for share;
+id             
+
+1              
+step s2_for_update: select id from tlu_job where id = 1 for update; <waiting ...>
+step s3_for_update: select id from tlu_job where id = 1 for update; <waiting ...>
+step s1_rollback: rollback;
+step s2_for_update: <... completed>
+id             
+
+1              
+step s2_rollback: rollback;
+step s3_for_update: <... completed>
+id             
+
+1              
+step s3_rollback: rollback;
+
+starting permutation: s1_share s2_update s3_update s1_rollback s2_rollback s3_rollback
+step s1_share: select id from tlu_job where id = 1 for share;
+id             
+
+1              
+step s2_update: update tlu_job set name = 'b' where id = 1; <waiting ...>
+step s3_update: update tlu_job set name = 'b' where id = 1; <waiting ...>
+step s1_rollback: rollback;
+step s2_update: <... completed>
+step s2_rollback: rollback;
+step s3_update: <... completed>
+step s3_rollback: rollback;
+
+starting permutation: s1_share s2_delete s3_delete s1_rollback s2_rollback s3_rollback
+step s1_share: select id from tlu_job where id = 1 for share;
+id             
+
+1              
+step s2_delete: delete from tlu_job where id = 1; <waiting ...>
+step s3_delete: delete from tlu_job where id = 1; <waiting ...>
+step s1_rollback: rollback;
+step s2_delete: <... completed>
+step s2_rollback: rollback;
+step s3_delete: <... completed>
+step s3_rollback: rollback;
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 889b4d827a..3b3bc10a65 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -42,6 +42,7 @@ test: delete-abort-savept-2
 test: aborted-keyrevoke
 test: multixact-no-deadlock
 test: multixact-no-forget
+test: rowlock-upgrade-deadlock
 test: lock-committed-update
 test: lock-committed-keyupdate
 test: update-locked-tuple
diff --git a/src/test/isolation/specs/rowlock-upgrade-deadlock.spec b/src/test/isolation/specs/rowlock-upgrade-deadlock.spec
new file mode 100644
index 0000000000..7e0230c7f7
--- /dev/null
+++ b/src/test/isolation/specs/rowlock-upgrade-deadlock.spec
@@ -0,0 +1,52 @@
+# This test checks that multiple sessions locking a single row in a table
+# does not deadlock each other, particularly when one of them upgrades its
+# lock while the others are waiting.
+
+setup
+{
+    drop table if exists tlu_job;
+    create table tlu_job (id integer primary key, name text);
+
+    insert into tlu_job values(1, 'a');
+}
+
+
+teardown
+{
+    drop table tlu_job;
+}
+
+session "s1"
+setup { begin; }
+step "s1_keyshare" { select id from tlu_job where id = 1 for key share;}
+step "s1_share" { select id from tlu_job where id = 1 for share; }
+step "s1_update" { update tlu_job set name = 'b' where id = 1;  }
+step "s1_delete" { delete from tlu_job where id = 1; }
+step "s1_rollback" { rollback; }
+
+session "s2"
+setup { begin; }
+step "s2_for_update" { select id from tlu_job where id = 1 for update; }
+step "s2_update" { update tlu_job set name = 'b' where id = 1; }
+step "s2_delete" { delete from tlu_job where id = 1; }
+step "s2_rollback" { rollback; }
+
+session "s3"
+setup { begin; }
+step "s3_keyshare" { select id from tlu_job where id = 1 for key share; }
+step "s3_share" { select id from tlu_job where id = 1 for share; }
+step "s3_for_update" { select id from tlu_job where id = 1 for update; }
+step "s3_update" { update tlu_job set name = 'b' where id = 1; }
+step "s3_delete" { delete from tlu_job where id = 1; }
+step "s3_rollback" { rollback; }
+
+# test that s2 will not deadlock with s3 when s1 is rolled back
+permutation "s1_share" "s2_for_update" "s3_share" "s3_for_update" "s1_rollback" "s3_rollback" "s2_rollback"
+# test that update does not cause deadlocks if it can proceed
+permutation  "s1_keyshare" "s2_for_update" "s3_keyshare" "s1_update" "s3_update" "s1_rollback" "s3_rollback" "s2_rollback"
+# test that delete does not cause deadlocks if it can proceed
+permutation "s1_keyshare" "s2_for_update" "s3_keyshare"  "s3_delete" "s1_rollback" "s3_rollback" "s2_rollback"
+# test that sessions that don't upgrade locks acquire them in order
+permutation "s1_share" "s2_for_update" "s3_for_update" "s1_rollback" "s2_rollback" "s3_rollback"
+permutation "s1_share" "s2_update" "s3_update" "s1_rollback" "s2_rollback" "s3_rollback"
+permutation "s1_share" "s2_delete" "s3_delete" "s1_rollback" "s2_rollback" "s3_rollback"
