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.

3 files changed, 176 insertions, 44 deletions

diff --git a/mysql-test/suite/innodb/r/lock_update_split_rc.result b/mysql-test/suite/innodb/r/lock_update_split_rc.result
new file mode 100644
index 00000000000..6bdce124c5b
--- /dev/null
+++ b/mysql-test/suite/innodb/r/lock_update_split_rc.result
@@ -0,0 +1,34 @@
+CREATE TABLE t (
+`a` INT NOT NULL,
+`b` INT NOT NULL,
+PRIMARY KEY (`a`)
+) ENGINE=InnoDB;
+SET GLOBAL innodb_limit_optimistic_insert_debug = 3;
+INSERT INTO t VALUES(10, 0);
+INSERT INTO t VALUES(20, 0);
+INSERT INTO t VALUES(30, 0);
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+XA START '1';
+REPLACE INTO t VALUES(10, 1);
+REPLACE INTO t VALUES(20, 1);
+SET DEBUG_SYNC= 'ib_after_row_insert SIGNAL inserted WAIT_FOR cont';
+REPLACE INTO t VALUES(30, 1);
+connect  con1,localhost,root;
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+XA START '2';
+SET DEBUG_SYNC= 'now WAIT_FOR inserted';
+INSERT INTO t VALUES(40, 2);
+SET DEBUG_SYNC= 'now SIGNAL cont';
+connection default;
+XA END '1';
+XA PREPARE '1';
+connection default;
+XA COMMIT '1';
+connection con1;
+XA END '2';
+XA PREPARE '2';
+XA COMMIT '2';
+disconnect con1;
+connection default;
+SET DEBUG_SYNC= "RESET";
+DROP TABLE t;
diff --git a/mysql-test/suite/innodb/t/lock_update_split_rc.test b/mysql-test/suite/innodb/t/lock_update_split_rc.test
new file mode 100644
index 00000000000..38910e53ef3
--- /dev/null
+++ b/mysql-test/suite/innodb/t/lock_update_split_rc.test
@@ -0,0 +1,76 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/count_sessions.inc
+
+CREATE TABLE t (
+  `a` INT NOT NULL,
+  `b` INT NOT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=InnoDB;
+
+--disable_query_log
+SET @old_innodb_limit_optimistic_insert_debug = @@innodb_limit_optimistic_insert_debug;
+--enable_query_log
+
+SET GLOBAL innodb_limit_optimistic_insert_debug = 3;
+
+INSERT INTO t VALUES(10, 0);
+INSERT INTO t VALUES(20, 0);
+INSERT INTO t VALUES(30, 0);
+
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+XA START '1';
+REPLACE INTO t VALUES(10, 1);
+REPLACE INTO t VALUES(20, 1);
+
+# We need the following sync point because mysql_insert() resets
+# trx->duplicates with the following condition:
+#
+#  if (duplic == DUP_REPLACE &&
+#      (!table->triggers || !table->triggers->has_delete_triggers()))
+#    table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
+#
+# and ha_innobase::extra() resets trx_t::duplicates, but we need
+# lock_update_split_right() to be invoked when trx->duplicates is set to
+# repeat the bug. So the transaction will hang just after
+# row_insert_for_mysql() call until another transaction inserts new row and
+# splits the page.
+SET DEBUG_SYNC= 'ib_after_row_insert SIGNAL inserted WAIT_FOR cont';
+--send REPLACE INTO t VALUES(30, 1)
+
+connect (con1,localhost,root);
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+XA START '2';
+SET DEBUG_SYNC= 'now WAIT_FOR inserted';
+# The following statement will cause page split and (20, ...) will be split
+# record. As the previous REPLACE set non-gap X-lock on it,
+# lock_update_split_right() and lock_rec_inherit_to_gap() will 'inherit' the
+# lock from the very first (20, ...) new right page record to the supremum of
+# the old left page, what should not be for READ COMMITTED isolation level
+INSERT INTO t VALUES(40, 2);
+SET DEBUG_SYNC= 'now SIGNAL cont';
+
+--connection default
+--reap
+XA END '1';
+# This will cause the assertion failure, because the supremum of the left page
+# has X-lock.
+XA PREPARE '1';
+--connection default
+XA COMMIT '1';
+
+--connection con1
+XA END '2';
+XA PREPARE '2';
+XA COMMIT '2';
+--disconnect con1
+
+--connection default
+SET DEBUG_SYNC= "RESET";
+DROP TABLE t;
+
+--disable_query_log
+SET GLOBAL innodb_limit_optimistic_insert_debug = @old_innodb_limit_optimistic_insert_debug;
+--enable_query_log
+
+--source include/wait_until_count_sessions.inc
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 1c11efafc7a..a2688a1b6cb 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -2118,49 +2118,57 @@ lock_rec_reset_and_release_wait(const hash_cell_t &cell, const page_id_t id,
   }
 }
 
-/*************************************************************//**
-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(
-/*====================*/
-	hash_cell_t&		heir_cell,	/*!< heir hash table cell */
-	const page_id_t		heir,		/*!< in: page containing the
-						record which inherits */
-	const hash_cell_t&	donor_cell,	/*!< donor hash table cell */
-	const page_id_t		donor,		/*!< in: page containing the
-						record from which inherited;
-						does NOT reset the locks on
-						this record */
-	const page_t*		heir_page,	/*!< in: heir page frame */
-	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_cell    heir hash table cell
+@param heir         page containing the record which inherits
+@param donor_cell   donor hash table cell
+@param donor        page containing the record from which inherited; does NOT
+                    reset the locks on this record
+@param heir_page    heir page frame
+@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(hash_cell_t &heir_cell, const page_id_t heir,
+                        const hash_cell_t &donor_cell, const page_id_t donor,
+                        const page_t *heir_page, ulint heir_heap_no,
+                        ulint heap_no)
 {
-	/* 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. */
+  ut_ad(!from_split || heir_heap_no == PAGE_HEAP_NO_SUPREMUM);
 
-	for (lock_t* lock= lock_sys_t::get_first(donor_cell, donor, heap_no);
-	     lock;
-	     lock = lock_rec_get_next(heap_no, lock)) {
-		trx_t* lock_trx = lock->trx;
-		if (!lock->is_insert_intention()
-		    && (lock_trx->isolation_level > TRX_ISO_READ_COMMITTED
-			|| lock->mode() !=
-			(lock_trx->duplicates ? LOCK_S : LOCK_X))) {
-			lock_rec_add_to_queue(LOCK_GAP | lock->mode(),
-					      heir_cell, heir, heir_page,
-					      heir_heap_no,
-					      lock->index, lock_trx, false);
-		}
-	}
+  /* 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_t *lock= lock_sys_t::get_first(donor_cell, donor, heap_no); lock;
+       lock= lock_rec_get_next(heap_no, lock))
+  {
+    trx_t *lock_trx= lock->trx;
+    if (!lock->is_insert_intention() &&
+        (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->mode() != (lock_trx->duplicates ? LOCK_S : LOCK_X))))
+    {
+      lock_rec_add_to_queue(LOCK_GAP | lock->mode(), heir_cell, heir,
+                            heir_page, heir_heap_no, lock->index, lock_trx,
+                            false);
+    }
+  }
 }
 
 /*************************************************************//**
@@ -2801,8 +2809,9 @@ 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(g.cell1(), l, g.cell2(), r, left_block->page.frame,
-                          PAGE_HEAP_NO_SUPREMUM, h);
+  lock_rec_inherit_to_gap<true>(g.cell1(), l, g.cell2(), r,
+                                left_block->page.frame, PAGE_HEAP_NO_SUPREMUM,
+                                h);
 }
 
 void lock_update_node_pointer(const buf_block_t *left_block,
@@ -2917,8 +2926,9 @@ lock_update_split_left(
   LockMultiGuard g{lock_sys.rec_hash, l, r};
   /* 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(g.cell1(), l, g.cell2(), r, left_block->page.frame,
-                          PAGE_HEAP_NO_SUPREMUM, h);
+  lock_rec_inherit_to_gap<true>(g.cell1(), l, g.cell2(), r,
+                                left_block->page.frame, PAGE_HEAP_NO_SUPREMUM,
+                                h);
 }
 
 /** Update the lock table when a page is merged to the left.
@@ -4056,8 +4066,14 @@ 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())
+      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));
         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);
@@ -4120,6 +4136,12 @@ void lock_release_on_prepare(trx_t *trx)
       ut_ad(!lock->index->table->is_temporary());
       if (lock->mode() != LOCK_X || lock->is_gap())
         lock_rec_dequeue_from_page(lock, false);
+      else
+        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));
     }
     else
     {