MDEV-28709 unexpected X lock on Supremum in READ COMMITTED

The lock is created during page splitting after moving records and
locks(lock_move_rec_list_(start|end)()) to the new page, and inheriting
the locks to the supremum of left page from the successor of the infimum
on right page.

There is no need in such inheritance for READ COMMITTED isolation level
and not-gap locks, so the fix is to add the corresponding condition in
gap lock inheritance function.

One more fix is to forbid gap lock inheritance if XA was prepared. Use the
most significant bit of trx_t::n_ref to indicate that gap lock inheritance
is forbidden. This fix is based on
mysql/mysql-server@b063e52a8367dc9d5ed418e7f6d96400867e9f43

3 files changed, 97 insertions, 57 deletions

diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 68a9812468f..ce3eca7593f 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -646,14 +646,19 @@ struct trx_rsegs_t {
 struct trx_t : ilist_node<> {
 private:
   /**
-    Count of references.
+    Least significant 31 bits is count of references.
 
     We can't release the locks nor commit the transaction until this reference
     is 0. We can change the state to TRX_STATE_COMMITTED_IN_MEMORY to signify
     that it is no longer "active".
-  */
 
-  Atomic_counter<int32_t> n_ref;
+    If the most significant bit is set this transaction should stop inheriting
+    (GAP)locks. Generally set to true during transaction prepare for RC or lower
+    isolation, if requested. Needed for replication replay where
+    we don't want to get blocked on GAP locks taken for protecting
+    concurrent unique insert or replace operation.
+  */
+  Atomic_relaxed<uint32_t> skip_lock_inheritance_and_n_ref;
 
 
 public:
@@ -983,27 +988,47 @@ public:
   /** Commit the transaction. */
   void commit();
 
+  bool is_referenced() const
+  {
+    return (skip_lock_inheritance_and_n_ref & ~(1U << 31)) > 0;
+  }
 
-  bool is_referenced() const { return n_ref > 0; }
+  void reference()
+  {
+    ut_d(auto old_n_ref =)
+    skip_lock_inheritance_and_n_ref.fetch_add(1);
+    ut_ad(int32_t(old_n_ref << 1) >= 0);
+  }
 
+  void release_reference()
+  {
+    ut_d(auto old_n_ref =)
+    skip_lock_inheritance_and_n_ref.fetch_sub(1);
+    ut_ad(int32_t(old_n_ref << 1) > 0);
+  }
 
-  void reference()
+  bool is_not_inheriting_locks() const
   {
-#ifdef UNIV_DEBUG
-    auto old_n_ref=
-#endif
-    n_ref++;
-    ut_ad(old_n_ref >= 0);
+    return skip_lock_inheritance_and_n_ref >> 31;
   }
 
+  void set_skip_lock_inheritance()
+  {
+    ut_d(auto old_n_ref=) skip_lock_inheritance_and_n_ref.fetch_add(1U << 31);
+    ut_ad(!(old_n_ref >> 31));
+  }
 
-  void release_reference()
+  void reset_skip_lock_inheritance()
   {
-#ifdef UNIV_DEBUG
-    auto old_n_ref=
+#if defined __GNUC__ && (defined __i386__ || defined __x86_64__)
+    __asm__("lock btrl $31, %0" : : "m"(skip_lock_inheritance_and_n_ref));
+#elif defined _MSC_VER && (defined _M_IX86 || defined _M_X64)
+    _interlockedbittestandreset(
+        reinterpret_cast<volatile long *>(&skip_lock_inheritance_and_n_ref),
+        31);
+#else
+    skip_lock_inheritance_and_n_ref.fetch_and(~1U << 31);
 #endif
-    n_ref--;
-    ut_ad(old_n_ref > 0);
   }
 
   /** @return whether the table has lock on
@@ -1031,6 +1056,7 @@ public:
     ut_ad(!autoinc_locks || ib_vector_is_empty(autoinc_locks));
     ut_ad(UT_LIST_GET_LEN(lock.evicted_tables) == 0);
     ut_ad(dict_operation == TRX_DICT_OP_NONE);
+    ut_ad(!is_not_inheriting_locks());
   }
 
   /** @return whether this is a non-locking autocommit transaction */
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 380f9201ef5..da009ada289 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -2291,50 +2291,54 @@ lock_rec_reset_and_release_wait(
 		&lock_sys.prdt_page_hash, block, PAGE_HEAP_NO_INFIMUM);
 }
 
-/*************************************************************//**
-Makes a record to inherit the locks (except LOCK_INSERT_INTENTION type)
+/** Makes a record to inherit the locks (except LOCK_INSERT_INTENTION type)
 of another record as gap type locks, but does not reset the lock bits of
 the other record. Also waiting lock requests on rec are inherited as
-GRANTED gap locks. */
-static
-void
-lock_rec_inherit_to_gap(
-/*====================*/
-	const buf_block_t*	heir_block,	/*!< in: block containing the
-						record which inherits */
-	const buf_block_t*	block,		/*!< in: block containing the
-						record from which inherited;
-						does NOT reset the locks on
-						this record */
-	ulint			heir_heap_no,	/*!< in: heap_no of the
-						inheriting record */
-	ulint			heap_no)	/*!< in: heap_no of the
-						donating record */
+GRANTED gap locks.
+@param heir_block   block containing the record which inherits
+@param block block  containing the record from which inherited; does NOT reset
+                    the locks on this record
+@param heir_heap_no heap_no of the inheriting record
+@param heap_no      heap_no of the donating record
+@tparam from_split  true if the function is invoked from
+                    lock_update_split_(left|right)(), in this case not-gap
+                    locks are not inherited to supremum if transaction
+                    isolation level less or equal to READ COMMITTED */
+template <bool from_split= false>
+static void lock_rec_inherit_to_gap(const buf_block_t *heir_block,
+                                    const buf_block_t *block,
+                                    ulint heir_heap_no, ulint heap_no)
 {
-	lock_t*	lock;
-
-	ut_ad(lock_mutex_own());
+  ut_ad(lock_mutex_own());
+  ut_ad(!from_split || heir_heap_no == PAGE_HEAP_NO_SUPREMUM);
 
-	/* At READ UNCOMMITTED or READ COMMITTED isolation level,
-	we do not want locks set
-	by an UPDATE or a DELETE to be inherited as gap type locks. But we
-	DO want S-locks/X-locks(taken for replace) set by a consistency
-	constraint to be inherited also then. */
+  /* At READ UNCOMMITTED or READ COMMITTED isolation level,
+  we do not want locks set
+  by an UPDATE or a DELETE to be inherited as gap type locks. But we
+  DO want S-locks/X-locks(taken for replace) set by a consistency
+  constraint to be inherited also then. */
 
-	for (lock = lock_rec_get_first(&lock_sys.rec_hash, block, heap_no);
-	     lock != NULL;
-	     lock = lock_rec_get_next(heap_no, lock)) {
+  for (lock_t *lock= lock_rec_get_first(&lock_sys.rec_hash, block, heap_no);
+       lock != NULL; lock= lock_rec_get_next(heap_no, lock))
+  {
 
-		if (!lock_rec_get_insert_intention(lock)
-		    && (lock->trx->isolation_level > TRX_ISO_READ_COMMITTED
-			|| lock_get_mode(lock) !=
-			(lock->trx->duplicates ? LOCK_S : LOCK_X))) {
-			lock_rec_add_to_queue(
-				LOCK_REC | LOCK_GAP | lock_get_mode(lock),
-				heir_block, heir_heap_no, lock->index,
-				lock->trx, FALSE);
-		}
-	}
+    if (!lock->trx->is_not_inheriting_locks() &&
+        !lock_rec_get_insert_intention(lock) &&
+        (lock->trx->isolation_level > TRX_ISO_READ_COMMITTED ||
+         /* When we are in a page split (not purge), then we don't set a lock
+         on supremum if the donor lock type is LOCK_REC_NOT_GAP. That is, do
+         not create bogus gap locks for non-gap locks for READ UNCOMMITTED and
+         READ COMMITTED isolation levels. LOCK_ORDINARY and
+         LOCK_GAP require a gap before the record to be locked, that is why
+         setting lock on supremmum is necessary. */
+         ((!from_split || !lock->is_record_not_gap()) &&
+          (lock_get_mode(lock) != (lock->trx->duplicates ? LOCK_S : LOCK_X)))))
+    {
+      lock_rec_add_to_queue(LOCK_REC | LOCK_GAP | lock_get_mode(lock),
+                            heir_block, heir_heap_no, lock->index, lock->trx,
+                            FALSE);
+    }
+  }
 }
 
 /*************************************************************//**
@@ -2361,7 +2365,8 @@ lock_rec_inherit_to_gap_if_gap_lock(
 	     lock != NULL;
 	     lock = lock_rec_get_next(heap_no, lock)) {
 
-		if (!lock_rec_get_insert_intention(lock)
+		if (!lock->trx->is_not_inheriting_locks()
+		    && !lock_rec_get_insert_intention(lock)
 		    && (heap_no == PAGE_HEAP_NO_SUPREMUM
 			|| !lock_rec_get_rec_not_gap(lock))) {
 
@@ -2943,7 +2948,7 @@ lock_update_split_right(
 	/* Inherit the locks to the supremum of left page from the successor
 	of the infimum on right page */
 
-	lock_rec_inherit_to_gap(left_block, right_block,
+	lock_rec_inherit_to_gap<true>(left_block, right_block,
 				PAGE_HEAP_NO_SUPREMUM, heap_no);
 
 	lock_mutex_exit();
@@ -3063,7 +3068,7 @@ lock_update_split_left(
 	/* Inherit the locks to the supremum of the left page from the
 	successor of the infimum on the right page */
 
-	lock_rec_inherit_to_gap(left_block, right_block,
+	lock_rec_inherit_to_gap<true>(left_block, right_block,
 				PAGE_HEAP_NO_SUPREMUM, heap_no);
 
 	lock_mutex_exit();
@@ -4251,6 +4256,11 @@ void lock_release_on_prepare(trx_t *trx)
       {
         ut_ad(trx->dict_operation ||
               lock->index->table->id >= DICT_HDR_FIRST_ID);
+        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_get_mode(lock) == LOCK_X ||
+              !lock_rec_get_nth_bit(lock, PAGE_HEAP_NO_SUPREMUM));
 retain_lock:
         lock= UT_LIST_GET_PREV(trx_locks, lock);
         continue;
@@ -4287,6 +4297,8 @@ retain_lock:
   }
 
   lock_mutex_exit();
+
+  trx->set_skip_lock_inheritance();
 }
 
 /* True if a lock mode is S or X */
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index d8c720da604..54783e2d1a0 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -412,7 +412,8 @@ void trx_t::free()
 
   mod_tables.clear();
 
-  MEM_NOACCESS(&n_ref, sizeof n_ref);
+  MEM_NOACCESS(&skip_lock_inheritance_and_n_ref,
+               sizeof skip_lock_inheritance_and_n_ref);
   /* do not poison mutex */
   MEM_NOACCESS(&id, sizeof id);
   MEM_NOACCESS(&state, sizeof state);
@@ -518,6 +519,7 @@ inline void trx_t::release_locks()
   }
 
   lock.table_locks.clear();
+  reset_skip_lock_inheritance();
 }
 
 /** At shutdown, frees a transaction object. */