MDEV-28422 Page split breaks a gap lockbb-10.6-MDEV-28422

btr_insert_into_right_sibling(): Inherit any gap lock from the
left sibling to the right sibling before inserting the record
to the right sibling and updating the node pointer(s).

lock_update_node_pointer(): Update locks in case a node pointer
will move.

Based on mysql/mysql-server@c7d93c274fdc5c56e36458fa4000fa3a483ffffd

5 files changed, 121 insertions, 4 deletions

diff --git a/mysql-test/suite/innodb/r/gap_lock_split.result b/mysql-test/suite/innodb/r/gap_lock_split.result
new file mode 100644
index 00000000000..25a3cf711f9
--- /dev/null
+++ b/mysql-test/suite/innodb/r/gap_lock_split.result
@@ -0,0 +1,27 @@
+SET @save_frequency=@@GLOBAL.innodb_purge_rseg_truncate_frequency;
+SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
+CREATE TABLE t1(id INT PRIMARY key, val VARCHAR(16000)) ENGINE=InnoDB;
+INSERT INTO t1 (id,val) SELECT 2*seq,'x' FROM seq_0_to_1023;
+connect con1,localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+DELETE FROM t1 WHERE id=1788;
+BEGIN;
+SELECT * FROM t1 WHERE id=1788 FOR UPDATE;
+id	val
+connection con1;
+COMMIT;
+InnoDB		0 transactions not purged
+connection default;
+INSERT INTO t1 (id,val) VALUES (1787, REPEAT('x',2000));
+connection con1;
+SET innodb_lock_wait_timeout=0;
+INSERT INTO t1 (id,val) VALUES (1788, 'x');
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+SELECT * FROM t1 WHERE id=1788 FOR UPDATE;
+id	val
+disconnect con1;
+connection default;
+COMMIT;
+DROP TABLE t1;
+SET GLOBAL innodb_purge_rseg_truncate_frequency=@save_frequency;
diff --git a/mysql-test/suite/innodb/t/gap_lock_split.test b/mysql-test/suite/innodb/t/gap_lock_split.test
new file mode 100644
index 00000000000..462f073ddce
--- /dev/null
+++ b/mysql-test/suite/innodb/t/gap_lock_split.test
@@ -0,0 +1,39 @@
+--source include/have_innodb.inc
+--source include/have_sequence.inc
+--source include/have_debug.inc
+
+SET @save_frequency=@@GLOBAL.innodb_purge_rseg_truncate_frequency;
+SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
+
+CREATE TABLE t1(id INT PRIMARY key, val VARCHAR(16000)) ENGINE=InnoDB;
+INSERT INTO t1 (id,val) SELECT 2*seq,'x' FROM seq_0_to_1023;
+
+connect(con1,localhost,root,,);
+# Prevent purge.
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+
+DELETE FROM t1 WHERE id=1788;
+
+BEGIN;
+# This will return no result, but should acquire a gap lock.
+SELECT * FROM t1 WHERE id=1788 FOR UPDATE;
+
+connection con1;
+COMMIT;
+source include/wait_all_purged.inc;
+connection default;
+
+INSERT INTO t1 (id,val) VALUES (1787, REPEAT('x',2000));
+
+connection con1;
+SET innodb_lock_wait_timeout=0;
+--error ER_LOCK_WAIT_TIMEOUT
+INSERT INTO t1 (id,val) VALUES (1788, 'x');
+SELECT * FROM t1 WHERE id=1788 FOR UPDATE;
+disconnect con1;
+
+connection default;
+COMMIT;
+DROP TABLE t1;
+SET GLOBAL innodb_purge_rseg_truncate_frequency=@save_frequency;
diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 7fdecf9e4ef..33e607b1637 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -2597,8 +2597,8 @@ btr_insert_into_right_sibling(
 	max_size = page_get_max_insert_size_after_reorganize(next_page, 1);
 
 	/* Extends gap lock for the next page */
-	if (cursor->index->has_locking()) {
-		lock_update_split_left(next_block, block);
+	if (is_leaf && cursor->index->has_locking()) {
+		lock_update_node_pointer(block, next_block);
 	}
 
 	rec = page_cur_tuple_insert(
diff --git a/storage/innobase/include/lock0lock.h b/storage/innobase/include/lock0lock.h
index a11bc60e7a0..b67a1011f6b 100644
--- a/storage/innobase/include/lock0lock.h
+++ b/storage/innobase/include/lock0lock.h
@@ -1,6 +1,6 @@
 /*****************************************************************************
 
-Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 1996, 2022, Oracle and/or its affiliates.
 Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
@@ -136,6 +136,41 @@ void lock_update_root_raise(const buf_block_t &block, const page_id_t root);
 @param new_block  the target page
 @param old        old page (not index root page) */
 void lock_update_copy_and_discard(const buf_block_t &new_block, page_id_t old);
+
+/** Update gap locks between the last record of the left_block and the
+first record of the right_block when a record is about to be inserted
+at the start of the right_block, even though it should "naturally" be
+inserted as the last record of the left_block according to the
+current node pointer in the parent page.
+
+That is, we assume that the lowest common ancestor of the left_block
+and right_block routes the key of the new record to the left_block,
+but a heuristic which tries to avoid overflowing left_block has chosen
+to insert the record into right_block instead. Said ancestor performs
+this routing by comparing the key of the record to a "split point" -
+all records greater or equal to than the split point (node pointer)
+are in right_block, and smaller ones in left_block.
+The split point may be smaller than the smallest key in right_block.
+
+The gap between the last record on the left_block and the first record
+on the right_block is represented as a gap lock attached to the supremum
+pseudo-record of left_block, and a gap lock attached to the new first
+record of right_block.
+
+Thus, inserting the new record, and subsequently adjusting the node
+pointers in parent pages to values smaller or equal to the new
+records' key, will mean that gap will be sliced at a different place
+("moved to the left"): fragment of the 1st gap will now become treated
+as 2nd. Therefore, we must copy any GRANTED locks from 1st gap to the
+2nd gap. Any WAITING locks must be of INSERT_INTENTION type (as no
+other GAP locks ever wait for anything) and can stay at 1st gap, as
+their only purpose is to notify the requester they can retry
+insertion, and there's no correctness requirement to avoid waking them
+up too soon.
+@param left_block   left page
+@param right_block  right page */
+void lock_update_node_pointer(const buf_block_t *left_block,
+                              const buf_block_t *right_block);
 /*************************************************************//**
 Updates the lock table when a page is split to the left. */
 void
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 2d27a674275..8287a03f1c6 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -1,6 +1,6 @@
 /*****************************************************************************
 
-Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 1996, 2022, Oracle and/or its affiliates.
 Copyright (c) 2014, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
@@ -2769,6 +2769,22 @@ lock_update_split_right(
                           PAGE_HEAP_NO_SUPREMUM, h);
 }
 
+void lock_update_node_pointer(const buf_block_t *left_block,
+                              const buf_block_t *right_block)
+{
+  const ulint h= lock_get_min_heap_no(right_block);
+  const page_id_t l{left_block->page.id()};
+  const page_id_t r{right_block->page.id()};
+
+  /* This would likely be too large for a memory transaction. */
+  LockMultiGuard g{lock_sys.rec_hash, l, r};
+
+  /* Inherit locks from the gap before supremum of the left page to the gap
+  before the successor of the infimum of the right page */
+  lock_rec_inherit_to_gap(g.cell2(), r, g.cell1(), l, right_block->page.frame,
+                          h, PAGE_HEAP_NO_SUPREMUM);
+}
+
 #ifdef UNIV_DEBUG
 static void lock_assert_no_spatial(const page_id_t id)
 {