Merge 10.5 into 10.6.

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

+CREATE TABLE t (
+`a` INT NOT NULL,
+PRIMARY KEY (`a`)
+) ENGINE=InnoDB;
+INSERT INTO t VALUES(10);
+INSERT INTO t VALUES(20);
+SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+XA START '1';
+SELECT * FROM t WHERE a > 20 FOR UPDATE;
+a
+INSERT INTO t VALUES(40);
+XA END '1';
+XA PREPARE '1';
+connect  con1,localhost,root;
+SET innodb_lock_wait_timeout=1;
+INSERT INTO t VALUES(50);
+disconnect con1;
+connection default;
+XA COMMIT '1';
+DROP TABLE t;

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

+--source include/have_innodb.inc
+--source include/count_sessions.inc
+
+CREATE TABLE t (
+  `a` INT NOT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=InnoDB;
+
+
+INSERT INTO t VALUES(10);
+INSERT INTO t VALUES(20);
+
+SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+XA START '1';
+SELECT * FROM t WHERE a > 20 FOR UPDATE;
+INSERT INTO t VALUES(40);
+XA END '1';
+XA PREPARE '1';
+
+connect (con1,localhost,root);
+SET innodb_lock_wait_timeout=1;
+# This will be finished with lock wait timeout error if XA PREPARE did not
+# reset lock on supremum
+INSERT INTO t VALUES(50);
+--disconnect con1
+
+--connection default
+XA COMMIT '1';
+DROP TABLE t;
+
+--source include/wait_until_count_sessions.inc

storage/innobase/include/lock0types.h

@@ -232,6 +232,11 @@ struct ib_lock_t
 		return(static_cast<enum lock_mode>(type_mode & LOCK_MODE_MASK));
 	}
 
+        bool is_rec_granted_exclusive_not_gap() const
+        {
+          return (type_mode & (LOCK_MODE_MASK | LOCK_GAP)) == LOCK_X;
+        }
+
 	/** Print the lock object into the given output stream.
 	@param[in,out]	out	the output stream
 	@return the given output stream. */

storage/innobase/lock/lock0lock.cc

@@ -3952,6 +3952,42 @@ dberr_t lock_sys_tables(trx_t *trx)
   return err;
 }
 
+/** Rebuild waiting queue after first_lock for heap_no. The queue is rebuilt
+close to the way lock_rec_dequeue_from_page() does it.
+@param trx        transaction that has set a lock, which caused the queue
+                  rebuild
+@param cell       rec hash cell of first_lock
+@param first_lock the lock after which waiting queue will be rebuilt
+@param heap_no    heap no of the record for which waiting queue to rebuild */
+static void lock_rec_rebuild_waiting_queue(
+#if defined(UNIV_DEBUG) || !defined(DBUG_OFF)
+    trx_t *trx,
+#endif /* defined(UNIV_DEBUG) || !defined(DBUG_OFF) */
+    hash_cell_t &cell, lock_t *first_lock, ulint heap_no)
+{
+  lock_sys.assert_locked(cell);
+
+  for (lock_t *lock= first_lock; lock != NULL;
+       lock= lock_rec_get_next(heap_no, lock))
+  {
+    if (!lock->is_waiting())
+      continue;
+    mysql_mutex_lock(&lock_sys.wait_mutex);
+    ut_ad(lock->trx->lock.wait_trx);
+    ut_ad(lock->trx->lock.wait_lock);
+
+    if (const lock_t *c= lock_rec_has_to_wait_in_queue(cell, lock))
+      lock->trx->lock.wait_trx= c->trx;
+    else
+    {
+      /* Grant the lock */
+      ut_ad(trx != lock->trx);
+      lock_grant(lock);
+    }
+    mysql_mutex_unlock(&lock_sys.wait_mutex);
+  }
+}
+
 /*=========================== LOCK RELEASE ==============================*/
 
 /*************************************************************//**
@@ -4015,26 +4051,11 @@ lock_rec_unlock(
 	}
 
 	/* Check if we can now grant waiting lock requests */
-
-	for (lock = first_lock; lock != NULL;
-	     lock = lock_rec_get_next(heap_no, lock)) {
-		if (!lock->is_waiting()) {
-			continue;
-		}
-		mysql_mutex_lock(&lock_sys.wait_mutex);
-		ut_ad(lock->trx->lock.wait_trx);
-		ut_ad(lock->trx->lock.wait_lock);
-
-		if (const lock_t* c = lock_rec_has_to_wait_in_queue(g.cell(),
-								    lock)) {
-			lock->trx->lock.wait_trx = c->trx;
-		} else {
-			/* Grant the lock */
-			ut_ad(trx != lock->trx);
-			lock_grant(lock);
-		}
-		mysql_mutex_unlock(&lock_sys.wait_mutex);
-	}
+	lock_rec_rebuild_waiting_queue(
+#if defined(UNIV_DEBUG) || !defined(DBUG_OFF)
+					trx,
+#endif /* defined(UNIV_DEBUG) || !defined(DBUG_OFF) */
+					g.cell(), first_lock, heap_no);
 }
 
 /** Release the explicit locks of a committing transaction,
@@ -4228,6 +4249,30 @@ void lock_release_on_drop(trx_t *trx)
   }
 }
 
+/** Reset lock bit for supremum and rebuild waiting queue.
+@param cell rec hash cell of in_lock
+@param lock the lock with supemum bit set */
+static void lock_rec_unlock_supremum(hash_cell_t &cell, lock_t *lock)
+{
+  ut_ad(lock_rec_get_nth_bit(lock, PAGE_HEAP_NO_SUPREMUM));
+#ifdef SAFE_MUTEX
+  ut_ad(!mysql_mutex_is_owner(&lock_sys.wait_mutex));
+#endif /* SAFE_MUTEX */
+  ut_ad(!lock->is_table());
+  ut_ad(lock_sys.is_writer() || lock->trx->mutex_is_owner());
+
+  lock_rec_reset_nth_bit(lock, PAGE_HEAP_NO_SUPREMUM);
+
+  lock_t *first_lock= lock_sys_t::get_first(
+      cell, lock->un_member.rec_lock.page_id, PAGE_HEAP_NO_SUPREMUM);
+
+  lock_rec_rebuild_waiting_queue(
+#if defined(UNIV_DEBUG) || !defined(DBUG_OFF)
+      lock->trx,
+#endif /* defined(UNIV_DEBUG) || !defined(DBUG_OFF) */
+      cell, first_lock, PAGE_HEAP_NO_SUPREMUM);
+}
+
 /** Release non-exclusive locks on XA PREPARE,
 and wake up possible other transactions waiting because of these locks.
 @param trx   transaction in XA PREPARE state
@@ -4259,20 +4304,18 @@ static bool lock_release_on_prepare_try(trx_t *trx)
     if (!lock->is_table())
     {
       ut_ad(!lock->index->table->is_temporary());
-      if (lock->mode() == LOCK_X && !lock->is_gap()) {
-        ut_ad(lock->trx->isolation_level > TRX_ISO_READ_COMMITTED ||
-              /* Insert-intention lock is valid for supremum for isolation
-              level > TRX_ISO_READ_COMMITTED */
-              lock->mode() == LOCK_X ||
-              !lock_rec_get_nth_bit(lock, PAGE_HEAP_NO_SUPREMUM));
+      bool supremum_bit = lock_rec_get_nth_bit(lock, PAGE_HEAP_NO_SUPREMUM);
+      if (!supremum_bit && lock->is_rec_granted_exclusive_not_gap())
         continue;
-      }
       auto &lock_hash= lock_sys.hash_get(lock->type_mode);
       auto cell= lock_hash.cell_get(lock->un_member.rec_lock.page_id.fold());
       auto latch= lock_sys_t::hash_table::latch(cell);
       if (latch->try_acquire())
       {
-        lock_rec_dequeue_from_page(lock, false);
+        if (supremum_bit)
+          lock_rec_unlock_supremum(*cell, lock);
+        else
+          lock_rec_dequeue_from_page(lock, false);
         latch->release();
       }
       else
@@ -4312,7 +4355,7 @@ static bool lock_release_on_prepare_try(trx_t *trx)
 and release possible other transactions waiting because of these locks. */
 void lock_release_on_prepare(trx_t *trx)
 {
-  auto _ = make_scope_exit([trx]() { trx->set_skip_lock_inheritance(); });
+  trx->set_skip_lock_inheritance();
 
   for (ulint count= 5; count--; )
     if (lock_release_on_prepare_try(trx))
@@ -4329,8 +4372,14 @@ void lock_release_on_prepare(trx_t *trx)
     if (!lock->is_table())
     {
       ut_ad(!lock->index->table->is_temporary());
-      if (lock->mode() != LOCK_X || lock->is_gap())
+      if (!lock->is_rec_granted_exclusive_not_gap())
         lock_rec_dequeue_from_page(lock, false);
+      else if (lock_rec_get_nth_bit(lock, PAGE_HEAP_NO_SUPREMUM))
+      {
+        auto &lock_hash= lock_sys.hash_get(lock->type_mode);
+        auto cell= lock_hash.cell_get(lock->un_member.rec_lock.page_id.fold());
+        lock_rec_unlock_supremum(*cell, lock);
+      }
       else
         ut_ad(lock->trx->isolation_level > TRX_ISO_READ_COMMITTED ||
               /* Insert-intention lock is valid for supremum for isolation