MDEV-27025 insert-intention lock conflicts with waiting ORDINARY lock

The code was backported from 10.5 be811386
commit. See that commit message for details.

mysql-test/suite/innodb/r/lock_wait_conflict.result

+#
+# MDEV-27025 insert-intention lock conflicts with waiting ORDINARY lock
+#
+CREATE TABLE t (a INT PRIMARY KEY, b INT NOT NULL UNIQUE) ENGINE=InnoDB;
+connect prevent_purge,localhost,root,,;
+start transaction with consistent snapshot;
+connection default;
+INSERT INTO t VALUES (20,20);
+DELETE FROM t WHERE b = 20;
+connect con_ins,localhost,root,,;
+SET DEBUG_SYNC = 'row_ins_sec_index_entry_dup_locks_created SIGNAL ins_set_locks WAIT_FOR ins_cont';
+INSERT INTO t VALUES(10, 20);
+connect con_del,localhost,root,,;
+SET DEBUG_SYNC = 'now WAIT_FOR ins_set_locks';
+SET DEBUG_SYNC = 'lock_wait_suspend_thread_enter SIGNAL del_locked';
+DELETE FROM t WHERE b = 20;
+connection default;
+SET DEBUG_SYNC = 'now WAIT_FOR del_locked';
+SET DEBUG_SYNC = 'now SIGNAL ins_cont';
+connection con_ins;
+disconnect con_ins;
+connection con_del;
+disconnect con_del;
+disconnect prevent_purge;
+connection default;
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t;

mysql-test/suite/innodb/t/lock_wait_conflict.test

+--source include/have_innodb.inc
+--source include/count_sessions.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+
+--echo #
+--echo # MDEV-27025 insert-intention lock conflicts with waiting ORDINARY lock
+--echo #
+
+# The test checks the ability to acquire exclusive record lock if the acquiring
+# transaction already holds a shared lock on the record and another transaction
+# is waiting for a lock.
+
+CREATE TABLE t (a INT PRIMARY KEY, b INT NOT NULL UNIQUE) ENGINE=InnoDB;
+
+--connect(prevent_purge,localhost,root,,)
+start transaction with consistent snapshot;
+
+--connection default
+INSERT INTO t VALUES (20,20);
+DELETE FROM t WHERE b = 20;
+
+--connect(con_ins,localhost,root,,)
+SET DEBUG_SYNC = 'row_ins_sec_index_entry_dup_locks_created SIGNAL ins_set_locks WAIT_FOR ins_cont';
+send
+INSERT INTO t VALUES(10, 20);
+
+--connect(con_del,localhost,root,,)
+SET DEBUG_SYNC = 'now WAIT_FOR ins_set_locks';
+SET DEBUG_SYNC = 'lock_wait_suspend_thread_enter SIGNAL del_locked';
+###############################################################################
+# This DELETE creates waiting ORDINARY X-lock for heap_no 2 as the record is
+# delete-marked, this lock conflicts with ORDINARY S-lock set by the the last
+# INSERT. After the last INSERT creates insert-intention lock on
+# heap_no 2, this lock will conflict with waiting ORDINARY X-lock of this
+# DELETE, what causes DEADLOCK error for this DELETE.
+###############################################################################
+send
+DELETE FROM t WHERE b = 20;
+
+--connection default
+SET DEBUG_SYNC = 'now WAIT_FOR del_locked';
+SET DEBUG_SYNC = 'now SIGNAL ins_cont';
+
+--connection con_ins
+--reap
+--disconnect con_ins
+
+--connection con_del
+# Without the fix, ER_LOCK_DEADLOCK would be reported here.
+--reap
+--disconnect con_del
+
+--disconnect prevent_purge
+
+--connection default
+
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t;
+--source include/wait_until_count_sessions.inc

mysql-test/suite/versioning/r/update.result

@@ -283,7 +283,6 @@ connection default;
 update t1 set b = 'foo';
 connection con1;
 update t1 set a = 'bar';
-ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
 disconnect con1;
 connection default;
 drop table t1;

mysql-test/suite/versioning/t/update.test

@@ -186,7 +186,9 @@ send update t1 set b = 'foo';
 connection con1;
 let $wait_condition= select count(*) from information_schema.innodb_lock_waits;
 source include/wait_condition.inc;
-error ER_LOCK_DEADLOCK;
+# There must no be DEADLOCK here as con1 transaction already holds locks, and
+# default's transaction lock is waiting, so the locks of the following "UPDATE"
+# must not conflict with waiting lock.
 update t1 set a = 'bar';
 disconnect con1;
 connection default;

storage/innobase/include/hash0hash.h

 /*****************************************************************************
 
 Copyright (c) 1997, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, MariaDB Corporation.
+Copyright (c) 2018, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -31,7 +31,31 @@ Created 5/20/1997 Heikki Tuuri
 #include "sync0rw.h"
 
 struct hash_table_t;
-struct hash_cell_t;
+
+struct hash_cell_t
+{
+  /** singly-linked, nullptr terminated list of hash buckets */
+  void *node;
+
+  /** Insert an element after another.
+  @tparam T  type of the element
+  @param after   the element after which to insert
+  @param insert  the being-inserted element
+  @param next    the next-element pointer in T */
+  template<typename T>
+  void insert_after(T &after, T &insert, T *T::*next)
+  {
+#ifdef UNIV_DEBUG
+    for (const T *c= static_cast<const T*>(node); c; c= c->*next)
+      if (c == &after)
+        goto found;
+    ut_error;
+  found:
+#endif
+    insert.*next= after.*next;
+    after.*next= &insert;
+  }
+};
 
 typedef void*	hash_node_t;
 
@@ -477,10 +501,6 @@ hash_unlock_x_all_but(
 	hash_table_t*	table,		/*!< in: hash table */
 	rw_lock_t*	keep_lock);	/*!< in: lock to keep */
 
-struct hash_cell_t{
-	void*	node;	/*!< hash chain node, NULL if none */
-};
-
 /* The hash table structure */
 struct hash_table_t {
 	enum hash_table_sync_t	type;	/*<! type of hash_table. */

storage/innobase/include/lock0lock.h

 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2020, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -888,26 +888,29 @@ class lock_sys_t
 /*********************************************************************//**
 Creates a new record lock and inserts it to the lock queue. Does NOT check
 for deadlocks or lock compatibility!
+@param[in] c_lock conflicting lock
+@param[in] thr thread owning trx
+@param[in] type_mode lock mode and wait flag, type is ignored and replaced by
+LOCK_REC
+@param[in] block buffer block containing the record
+@param[in] heap_no heap number of the record
+@param[in] index index of record
+@param[in,out] trx transaction
+@param[in] caller_owns_trx_mutex TRUE if caller owns trx mutex
+@param[in] insert_before_waiting if true, inserts new B-tree record lock
+just after the last non-waiting lock of the current transaction which is
+located before the first waiting for the current transaction lock, otherwise
+the lock is inserted at the end of the queue
 @return created lock */
 UNIV_INLINE
-lock_t*
-lock_rec_create(
-/*============*/
+lock_t *lock_rec_create(lock_t *c_lock,
 #ifdef WITH_WSREP
-	lock_t*			c_lock,	/*!< conflicting lock */
-	que_thr_t*		thr,	/*!< thread owning trx */
+                        que_thr_t *thr,
 #endif
-	ulint			type_mode,/*!< in: lock mode and wait
-					flag, type is ignored and
-					replaced by LOCK_REC */
-	const buf_block_t*	block,	/*!< in: buffer block containing
-					the record */
-	ulint			heap_no,/*!< in: heap number of the record */
-	dict_index_t*		index,	/*!< in: index of record */
-	trx_t*			trx,	/*!< in,out: transaction */
-	bool			caller_owns_trx_mutex);
-					/*!< in: true if caller owns
-					trx mutex */
+                        ulint type_mode, const buf_block_t *block,
+                        ulint heap_no, dict_index_t *index, trx_t *trx,
+                        bool caller_owns_trx_mutex,
+                        bool insert_before_waiting= false);
 
 /*************************************************************//**
 Removes a record lock request, waiting or granted, from the queue. */
@@ -920,6 +923,7 @@ lock_rec_discard(
 
 /** Create a new record lock and inserts it to the lock queue,
 without checking for deadlocks or conflicts.
+@param[in]	c_lock		conflicting lock
 @param[in]	type_mode	lock mode and wait flag; type will be replaced
 				with LOCK_REC
 @param[in]	space		tablespace id
@@ -929,11 +933,15 @@ without checking for deadlocks or conflicts.
 @param[in]	index		the index tree
 @param[in,out]	trx		transaction
 @param[in]	holds_trx_mutex	whether the caller holds trx->mutex
+@param[in]	insert_before_waiting if true, inserts new B-tree record lock
+just after the last non-waiting lock of the current transaction which is
+located before the first waiting for the current transaction lock, otherwise
+the lock is inserted at the end of the queue
 @return created lock */
 lock_t*
 lock_rec_create_low(
+	lock_t*		c_lock,
 #ifdef WITH_WSREP
-	lock_t*		c_lock,	/*!< conflicting lock */
 	que_thr_t*	thr,	/*!< thread owning trx */
 #endif
 	ulint		type_mode,
@@ -943,9 +951,12 @@ lock_rec_create_low(
 	ulint		heap_no,
 	dict_index_t*	index,
 	trx_t*		trx,
-	bool		holds_trx_mutex);
+	bool		holds_trx_mutex,
+	bool		insert_before_waiting = false);
+
 /** Enqueue a waiting request for a lock which cannot be granted immediately.
 Check for deadlocks.
+@param[in]	c_lock		conflicting lock
 @param[in]	type_mode	the requested lock mode (LOCK_S or LOCK_X)
 				possibly ORed with LOCK_GAP or
 				LOCK_REC_NOT_GAP, ORed with
@@ -964,9 +975,7 @@ Check for deadlocks.
 				(or it happened to commit) */
 dberr_t
 lock_rec_enqueue_waiting(
-#ifdef WITH_WSREP
-	lock_t*			c_lock,	/*!< conflicting lock */
-#endif
+	lock_t*			c_lock,
 	ulint			type_mode,
 	const buf_block_t*	block,
 	ulint			heap_no,

storage/innobase/include/lock0lock.inl

@@ -101,34 +101,37 @@ lock_hash_get(
 /*********************************************************************//**
 Creates a new record lock and inserts it to the lock queue. Does NOT check
 for deadlocks or lock compatibility!
+@param[in] c_lock conflicting lock
+@param[in] thr thread owning trx
+@param[in] type_mode lock mode and wait flag, type is ignored and replaced by
+LOCK_REC
+@param[in] block buffer block containing the record
+@param[in] heap_no heap number of the record
+@param[in] index index of record
+@param[in,out] trx transaction
+@param[in] caller_owns_trx_mutex TRUE if caller owns trx mutex
+@param[in] insert_before_waiting if true, inserts new B-tree record lock
+just after the last non-waiting lock of the current transaction which is
+located before the first waiting for the current transaction lock, otherwise
+the lock is inserted at the end of the queue
 @return created lock */
 UNIV_INLINE
-lock_t*
-lock_rec_create(
-/*============*/
+lock_t *lock_rec_create(lock_t *c_lock,
 #ifdef WITH_WSREP
-	lock_t*			c_lock,	/*!< conflicting lock */
-	que_thr_t*		thr,	/*!< thread owning trx */
+                        que_thr_t *thr,
 #endif
-	ulint			type_mode,/*!< in: lock mode and wait
-					flag, type is ignored and
-					replaced by LOCK_REC */
-	const buf_block_t*	block,	/*!< in: buffer block containing
-					the record */
-	ulint			heap_no,/*!< in: heap number of the record */
-	dict_index_t*		index,	/*!< in: index of record */
-	trx_t*			trx,	/*!< in,out: transaction */
-	bool			caller_owns_trx_mutex)
-					/*!< in: TRUE if caller owns
-					trx mutex */
+                        ulint type_mode, const buf_block_t *block,
+                        ulint heap_no, dict_index_t *index, trx_t *trx,
+                        bool caller_owns_trx_mutex,
+                        bool insert_before_waiting)
 {
 	btr_assert_not_corrupted(block, index);
-	return lock_rec_create_low(
+	return lock_rec_create_low(c_lock,
 #ifdef WITH_WSREP
-		c_lock, thr,
+		thr,
 #endif
 		type_mode,
 		block->page.id.space(), block->page.id.page_no(),
-		block->frame, heap_no,
-		index, trx, caller_owns_trx_mutex);
+		block->frame, heap_no, index, trx,
+		caller_owns_trx_mutex, insert_before_waiting);
 }

storage/innobase/include/lock0priv.h

 /*****************************************************************************
 
 Copyright (c) 2007, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2018, MariaDB Corporation.
+Copyright (c) 2015, 2018, 2022 MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -585,6 +585,9 @@ lock_rec_get_next_const(
 
 /*********************************************************************//**
 Gets the first explicit lock request on a record.
+@param[in] hash hash chain the lock on
+@param[in] page_id page id
+@param[in] heap_no heap number of the record
 @return first lock, NULL if none exists */
 UNIV_INLINE
 lock_t*
@@ -660,15 +663,26 @@ lock_table_has(
 
 /** Set the wait status of a lock.
 @param[in,out]	lock	lock that will be waited for
-@param[in,out]	trx	transaction that will wait for the lock */
-inline void lock_set_lock_and_trx_wait(lock_t* lock, trx_t* trx)
+@param[in,out]	trx	transaction that will wait for the lock
+@param[in]      c_lock   conflicting lock */
+inline void lock_set_lock_and_trx_wait(lock_t* lock, trx_t* trx,
+	const lock_t *c_lock)
 {
 	ut_ad(lock);
 	ut_ad(lock->trx == trx);
-	ut_ad(trx->lock.wait_lock == NULL);
 	ut_ad(lock_mutex_own());
 	ut_ad(trx_mutex_own(trx));
 
+	if (trx->lock.wait_trx) {
+		ut_ad(!c_lock || trx->lock.wait_trx == c_lock->trx);
+		ut_ad(trx->lock.wait_lock);
+		ut_ad((*trx->lock.wait_lock).trx == trx);
+	} else {
+		ut_ad(c_lock);
+		trx->lock.wait_trx = c_lock->trx;
+		ut_ad(!trx->lock.wait_lock);
+	}
+
 	trx->lock.wait_lock = lock;
 	lock->type_mode |= LOCK_WAIT;
 }
@@ -681,6 +695,7 @@ inline void lock_reset_lock_and_trx_wait(lock_t* lock)
 	ut_ad(lock_mutex_own());
 	ut_ad(lock->trx->lock.wait_lock == NULL
 	      || lock->trx->lock.wait_lock == lock);
+	lock->trx->lock.wait_trx= NULL;
 	lock->trx->lock.wait_lock = NULL;
 	lock->type_mode &= ~LOCK_WAIT;
 }

storage/innobase/include/lock0priv.inl

@@ -145,22 +145,19 @@ lock_rec_get_first_on_page_addr(
 	return(NULL);
 }
 
-/*********************************************************************//**
-Gets the first record lock on a page, where the page is identified by a
+/** Gets the first record lock on a page, where the page is identified by a
 pointer to it.
+@param[in] lock_hash lock hash table
+@param[in] space page's space id
+@param[in] page_no page number
+@param[in] hash page's hash value in records hash table
 @return	first lock, NULL if none exists */
 UNIV_INLINE
-lock_t*
-lock_rec_get_first_on_page(
-/*=======================*/
-	hash_table_t*		lock_hash,	/*!< in: lock hash table */
-	const buf_block_t*	block)		/*!< in: buffer block */
+lock_t *lock_rec_get_first_on_page(hash_table_t *lock_hash, ulint space,
+                                   ulint page_no, ulint hash)
 {
 	ut_ad(lock_mutex_own());
 
-	ulint	space	= block->page.id.space();
-	ulint	page_no	= block->page.id.page_no();
-	ulint	hash = buf_block_get_lock_hash_val(block);
 
 	for (lock_t* lock = static_cast<lock_t*>(
 			HASH_GET_FIRST(lock_hash, hash));
@@ -177,6 +174,20 @@ lock_rec_get_first_on_page(
 	return(NULL);
 }
 
+/** Gets the first record lock on a page, where the page is identified by a
+pointer to it.
+@param[in] lock_hash lock hash table
+@param[in] block buffer block
+@return	first lock, NULL if none exists */
+UNIV_INLINE
+lock_t *lock_rec_get_first_on_page(hash_table_t *lock_hash,
+                                   const buf_block_t *block)
+{
+  return lock_rec_get_first_on_page(lock_hash, block->page.id.space(),
+                                    block->page.id.page_no(),
+                                    buf_block_get_lock_hash_val(block));
+}
+
 /*********************************************************************//**
 Gets the next explicit lock request on a record.
 @return	next lock, NULL if none exists or if heap_no == ULINT_UNDEFINED */
@@ -210,21 +221,21 @@ lock_rec_get_next_const(
 	return(lock_rec_get_next(heap_no, (lock_t*) lock));
 }
 
-/*********************************************************************//**
-Gets the first explicit lock request on a record.
+/** Gets the first explicit lock request on a record.
+@param[in] hash hash chain the lock on
+@param[in] space record's space id
+@param[in] page_no record's page number
+@param[in] lock_hash_val page's hash value in records hash table
+@param[in] heap_no heap number of the record
 @return	first lock, NULL if none exists */
 UNIV_INLINE
-lock_t*
-lock_rec_get_first(
-/*===============*/
-	hash_table_t*		hash,	/*!< in: hash chain the lock on */
-	const buf_block_t*	block,	/*!< in: block containing the record */
-	ulint			heap_no)/*!< in: heap number of the record */
+lock_t *lock_rec_get_first(hash_table_t *hash, ulint space, ulint page_no,
+                           uint32_t lock_hash_val, ulint heap_no)
 {
 	ut_ad(lock_mutex_own());
 
-	for (lock_t* lock = lock_rec_get_first_on_page(hash, block); lock;
-	     lock = lock_rec_get_next_on_page(lock)) {
+	for (lock_t* lock = lock_rec_get_first_on_page(hash, space, page_no,
+	      lock_hash_val); lock; lock = lock_rec_get_next_on_page(lock)) {
 		if (lock_rec_get_nth_bit(lock, heap_no)) {
 			return(lock);
 		}
@@ -233,6 +244,20 @@ lock_rec_get_first(
 	return(NULL);
 }
 
+/** Gets the first explicit lock request on a record.
+@param[in] hash hash chain the lock on
+@param[in] block block containing the record
+@param[in] heap_no heap number of the record
+@return	first lock, NULL if none exists */
+UNIV_INLINE
+lock_t *lock_rec_get_first(hash_table_t *hash, const buf_block_t *block,
+                           ulint heap_no)
+{
+  return lock_rec_get_first(hash, block->page.id.space(),
+                            block->page.id.page_no(),
+                            buf_block_get_lock_hash_val(block), heap_no);
+}
+
 /*********************************************************************//**
 Gets the nth bit of a record lock.
 @return TRUE if bit set also if i == ULINT_UNDEFINED return FALSE*/

storage/innobase/include/trx0trx.h

@@ -436,7 +436,9 @@ struct trx_lock_t {
 	trx_que_t	que_state;	/*!< valid when trx->state
 					== TRX_STATE_ACTIVE: TRX_QUE_RUNNING,
 					TRX_QUE_LOCK_WAIT, ... */
-
+	/** Transaction being waited for; protected by the same mutexes as
+	wait_lock */
+	trx_t*		wait_trx;
 	lock_t*		wait_lock;	/*!< if trx execution state is
 					TRX_QUE_LOCK_WAIT, this points to
 					the lock request, otherwise this is

storage/innobase/lock/lock0prdt.cc

 /*****************************************************************************
 
 Copyright (c) 2014, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, MariaDB Corporation.
+Copyright (c) 2018, 2022 MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -487,9 +487,13 @@ lock_prdt_add_to_queue(
 		}
 	}
 
-	lock = lock_rec_create(
+	/* Note: We will not pass any conflicting lock to lock_rec_create(),
+	because we should be moving an existing waiting lock request. */
+	ut_ad(!(type_mode & LOCK_WAIT) || trx->lock.wait_trx);
+
+	lock = lock_rec_create(NULL,
 #ifdef WITH_WSREP
-		NULL, NULL, /* FIXME: replicate SPATIAL INDEX locks */
+		NULL, /* FIXME: replicate SPATIAL INDEX locks */
 #endif
 		type_mode, block, PRDT_HEAPNO, index, trx,
 		caller_owns_trx_mutex);
@@ -579,9 +583,7 @@ lock_prdt_insert_check_and_lock(
 		trx_mutex_enter(trx);
 
 		err = lock_rec_enqueue_waiting(
-#ifdef WITH_WSREP
 			NULL, /* FIXME: replicate SPATIAL INDEX locks */
-#endif
 			LOCK_X | LOCK_PREDICATE | LOCK_INSERT_INTENTION,
 			block, PRDT_HEAPNO, index, thr, prdt);
 
@@ -829,9 +831,9 @@ lock_prdt_lock(
 	lock_t*		lock = lock_rec_get_first_on_page(hash, block);
 
 	if (lock == NULL) {
-		lock = lock_rec_create(
+		lock = lock_rec_create(NULL,
 #ifdef WITH_WSREP
-			NULL, NULL, /* FIXME: replicate SPATIAL INDEX locks */
+			NULL, /* FIXME: replicate SPATIAL INDEX locks */
 #endif
 			ulint(mode) | type_mode, block, PRDT_HEAPNO,
 			index, trx, FALSE);
@@ -861,10 +863,8 @@ lock_prdt_lock(
 				if (wait_for != NULL) {
 
 					err = lock_rec_enqueue_waiting(
-#ifdef WITH_WSREP
 						NULL, /* FIXME: replicate
 						      SPATIAL INDEX locks */
-#endif
 						ulint(mode) | type_mode,
 						block, PRDT_HEAPNO,
 						index, thr, prdt);
@@ -948,9 +948,9 @@ lock_place_prdt_page_lock(
 	}
 
 	if (lock == NULL) {
-		lock = lock_rec_create_low(
+		lock = lock_rec_create_low(NULL,
 #ifdef WITH_WSREP
-			NULL, NULL, /* FIXME: replicate SPATIAL INDEX locks */
+			NULL, /* FIXME: replicate SPATIAL INDEX locks */
 #endif
 			mode, space, page_no, NULL, PRDT_HEAPNO,
 			index, trx, FALSE);