Implement forward scan with setting gap locks on delete-marked records during10.6-MDEV-20605-gap-locks

records delete-marking.

The following test fails with deadlock:

--source include/have_innodb.inc

CREATE TABLE t2 (a INT NOT NULL PRIMARY KEY) engine=innodb;
INSERT INTO t2 VALUES (134);
BEGIN;
SELECT * FROM t2 FOR UPDATE;

--connect(con2,localhost,root,,)
BEGIN;
--send SELECT * FROM t2 FOR UPDATE;
--connection default
select * from t2;
update t2 set a=135;
commit;

--connection con2
--reap
commit;

8 files changed, 111 insertions, 103 deletions

diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index ae8afa2be3f..045794a8657 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -3280,84 +3280,20 @@ btr_cur_ins_lock_and_undo(
 				trx->wsrep = 3;
 			}
 #endif /* WITH_WSREP */
-                        buf_block_t *old_block= btr_cur_get_block(cursor);
-                        ut_ad(old_block);
-                        rec_t *cur_rec= rec;
-                        buf_block_t *cur_block= old_block;
-                        bool comp= dict_table_is_comp(index->table);
-                        bool first_rec= true;
-                        ulint latch_mode;
                         dberr_t err;
-                        while (!(err= lock_rec_insert_check_and_lock(
-                                     cur_rec, cur_block, index, thr, mtr,
-                                     inherit, first_rec)))
-                        {
-                          first_rec= false;
-                          /*
-                          const rec_t *next_rec=
-                              page_rec_get_next_const(cur_rec);
-                          if (!page_rec_is_supremum(next_rec) &&
-                              !rec_get_deleted_flag(next_rec, comp))
-                            break;
-                          */
-                          get_next_rec:
-                          if (!page_cur_is_after_last(
-                                  btr_cur_get_page_cur(cursor)))
-                            page_cur_move_to_next(
-                                btr_cur_get_page_cur(cursor));
-                          else
-                          {
-                            uint32_t next_page_no=
-                                btr_page_get_next(btr_cur_get_page(cursor));
-                            if (next_page_no == FIL_NULL)
-                              break;
-                            buf_block_t *block= btr_cur_get_block(cursor);
-                            latch_mode=
-                                mtr->have_x_latch(page_id_t(
-                                    block->page.id().space(), next_page_no))
-                                    ? BTR_MODIFY_LEAF
-                                    : BTR_SEARCH_LEAF;
-                            block= btr_block_get(
-                                *index, next_page_no, latch_mode, false, mtr);
-                            if (cur_block != old_block &&
-                                latch_mode == BTR_SEARCH_LEAF)
-                              btr_leaf_page_release(btr_cur_get_block(cursor),
-                                                    latch_mode, mtr);
-                            page_cur_set_before_first(
-                                block, btr_cur_get_page_cur(cursor));
-                            page_cur_move_to_next(
-                                btr_cur_get_page_cur(cursor));
-
-                            ut_ad(!page_cur_is_after_last(
-                                btr_cur_get_page_cur(cursor)));
-                          }
-                          cur_rec= btr_cur_get_rec(cursor);
-                          cur_block= btr_cur_get_block(cursor);
-                          if (page_rec_is_supremum(cur_rec))
-                            goto get_next_rec;
-                          if (!page_rec_is_infimum(cur_rec) &&
-                              !rec_get_deleted_flag(cur_rec, comp))
-                            break;
-                        }
-                        /* Restore cursor position and release last latched
-                         block if necessary */
-                        page_cur_t *page_cur= btr_cur_get_page_cur(cursor);
-                        ut_ad(page_cur);
-                        if (cur_block != old_block)
-                        {
-                          ut_ad(latch_mode == BTR_MODIFY_LEAF ||
-                                latch_mode == BTR_SEARCH_LEAF);
-                          if (latch_mode == BTR_SEARCH_LEAF)
-                            btr_leaf_page_release(btr_cur_get_block(cursor),
-                                                  latch_mode, mtr);
-                          page_cur->block= old_block;
-                        }
-                        page_cur->rec= rec;
+                        for_each_delete_marked(
+                            btr_cur_get_page_cur(cursor), index, mtr,
+                            [index, thr, mtr, inherit,
+                             &err](rec_t *rec, buf_block_t *block,
+                                   bool first_rec) {
+                              return !(err= lock_rec_insert_check_and_lock(
+                                           rec, block, index, thr, mtr,
+                                           inherit, first_rec));
+                            });
                         if (err)
-                          return err;
-                }
+                            return err;
+		}
 	}
-
 	if (!index->is_primary() || !page_is_leaf(page_align(rec))) {
 		return DB_SUCCESS;
 	}
@@ -5429,8 +5365,7 @@ undo log record created.
 dberr_t
 btr_cur_del_mark_set_clust_rec(
 /*===========================*/
-	buf_block_t*	block,	/*!< in/out: buffer block of the record */
-	rec_t*		rec,	/*!< in/out: record */
+	btr_cur_t *btr_cur,
 	dict_index_t*	index,	/*!< in: clustered index of the record */
 	const rec_offs*	offsets,/*!< in: rec_get_offsets(rec) */
 	que_thr_t*	thr,	/*!< in: query thread */
@@ -5441,6 +5376,8 @@ btr_cur_del_mark_set_clust_rec(
 	roll_ptr_t	roll_ptr;
 	dberr_t		err;
 	trx_t*		trx;
+	buf_block_t *block = btr_cur_get_block(btr_cur);
+	rec_t *rec = btr_cur_get_rec(btr_cur);
 
 	ut_ad(dict_index_is_clust(index));
 	ut_ad(rec_offs_validate(rec, index, offsets));
@@ -5471,6 +5408,9 @@ btr_cur_del_mark_set_clust_rec(
 
 	btr_rec_set_deleted<true>(block, rec, mtr);
 
+	for_each_delete_marked(btr_cur_get_page_cur(btr_cur), index, mtr,
+	  gap_lock_inherit_functor(index));
+
 	trx = thr_get_trx(thr);
 
 	DBUG_LOG("ib_cur",
@@ -5608,8 +5548,8 @@ ibool btr_cur_optimistic_delete(btr_cur_t *cursor, ulint flags, mtr_t *mtr,
 			    && rec_is_add_metadata(first_rec, *index));
 		if (UNIV_LIKELY(empty_table)) {
 			if (UNIV_LIKELY(!is_metadata)) {
-				lock_update_delete(block, rec, offsets, index,
-				    from_purge, convert_lock_to_gap);
+				lock_update_delete(block, rec, index,
+				    from_purge, convert_lock_to_gap, true);
 			}
 			btr_page_empty(block, buf_block_get_page_zip(block),
 				       index, 0, mtr);
@@ -5647,8 +5587,8 @@ ibool btr_cur_optimistic_delete(btr_cur_t *cursor, ulint flags, mtr_t *mtr,
 					    cursor->index, mtr);
 			goto func_exit;
 		} else {
-			lock_update_delete(block, rec, offsets, cursor->index,
-			    from_purge, convert_lock_to_gap);
+			lock_update_delete(block, rec, cursor->index,
+			    from_purge, convert_lock_to_gap, true);
 
 			btr_search_update_hash_on_delete(cursor);
 		}
@@ -5802,8 +5742,8 @@ ibool btr_cur_pessimistic_delete(dberr_t *err, ibool has_reserved_extents,
 			ut_ad(index->table->supports_instant());
 			ut_ad(index->is_primary());
 		} else if (flags == 0) {
-			lock_update_delete(block, rec, offsets, index,
-			    from_purge, convert_lock_to_gap);
+			lock_update_delete(block, rec, index,
+			    from_purge, convert_lock_to_gap, true);
 		}
 
 		if (block->page.id().page_no() != index->page) {
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index 734796c6ce5..7dc315cae36 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -3958,7 +3958,8 @@ ibuf_delete(
 			return;
 		}
 
-		lock_update_delete(block, rec, offsets, index, false, true);
+		lock_update_delete(block, rec, index, false, true,
+		    true);
 
 		if (!page_zip) {
 			max_ins_size
diff --git a/storage/innobase/include/btr0cur.h b/storage/innobase/include/btr0cur.h
index 7998e2322da..f485344beb5 100644
--- a/storage/innobase/include/btr0cur.h
+++ b/storage/innobase/include/btr0cur.h
@@ -454,8 +454,7 @@ undo log record created.
 dberr_t
 btr_cur_del_mark_set_clust_rec(
 /*===========================*/
-	buf_block_t*	block,	/*!< in/out: buffer block of the record */
-	rec_t*		rec,	/*!< in/out: record */
+	btr_cur_t *btr_cur,
 	dict_index_t*	index,	/*!< in: clustered index of the record */
 	const rec_offs*	offsets,/*!< in: rec_get_offsets(rec) */
 	que_thr_t*	thr,	/*!< in: query thread */
diff --git a/storage/innobase/include/lock0lock.h b/storage/innobase/include/lock0lock.h
index 60947df76e4..c09d07b01d6 100644
--- a/storage/innobase/include/lock0lock.h
+++ b/storage/innobase/include/lock0lock.h
@@ -197,8 +197,9 @@ lock_update_insert(
 otherwise
 */
 void lock_update_delete(const buf_block_t *block, const rec_t *rec,
-                        rec_offs *offsets, dict_index_t *index,
-                        bool from_purge, bool convert_lock_to_gap);
+                        dict_index_t *index,
+                        bool from_purge, bool convert_lock_to_gap,
+                        bool release_lock);
 /*********************************************************************//**
 Stores on the page infimum record the explicit locks of another record.
 This function is used to store the lock state of a record when it is
diff --git a/storage/innobase/include/lock0lock.ic b/storage/innobase/include/lock0lock.ic
index e481dabbbf5..2920eee001b 100644
--- a/storage/innobase/include/lock0lock.ic
+++ b/storage/innobase/include/lock0lock.ic
@@ -77,3 +77,69 @@ lock_rec_create(
 		type_mode, block->page.id(), block->frame, heap_no,
 		index, trx, caller_owns_trx_mutex);
 }
+
+
+struct gap_lock_inherit_functor
+{
+  gap_lock_inherit_functor(dict_index_t *index) : index(index) {}
+
+  bool operator()(rec_t *rec, buf_block_t *block, bool first_rec) const
+  {
+    lock_update_delete(block, rec, index, false, false, false);
+    return true;
+  }
+private:
+  dict_index_t *index;
+};
+
+template <typename Functor>
+void for_each_delete_marked(page_cur_t *page_cur, const dict_index_t *index,
+                               mtr_t *mtr, const Functor &functor)
+{
+  rec_t *old_rec= page_cur_get_rec(page_cur);
+  buf_block_t *old_block= page_cur_get_block(page_cur);
+
+  rec_t *cur_rec= old_rec;
+  buf_block_t *cur_block= old_block;
+  bool comp= dict_table_is_comp(index->table);
+  bool first_rec= true;
+  ulint latch_mode;
+  while (functor(cur_rec, cur_block, first_rec))
+  {
+    first_rec= false;
+  get_next_rec:
+    if (!page_cur_is_after_last(page_cur))
+      page_cur_move_to_next(page_cur);
+    else
+    {
+      uint32_t next_page_no= btr_page_get_next(page_cur_get_page(page_cur));
+      if (next_page_no == FIL_NULL)
+        break;
+      buf_block_t *block= page_cur_get_block(page_cur);
+      latch_mode=
+          mtr->have_x_latch(page_id_t(block->page.id().space(), next_page_no))
+              ? BTR_MODIFY_LEAF
+              : BTR_SEARCH_LEAF;
+      block= btr_block_get(*index, next_page_no, latch_mode, false, mtr);
+      if (cur_block != old_block && latch_mode == BTR_SEARCH_LEAF)
+        btr_leaf_page_release(page_cur_get_block(page_cur), latch_mode, mtr);
+      page_cur_set_before_first(block, page_cur);
+      page_cur_move_to_next(page_cur);
+      ut_ad(!page_cur_is_after_last(page_cur));
+    }
+    cur_rec= page_cur_get_rec(page_cur);
+    cur_block= page_cur_get_block(page_cur);
+    if (page_rec_is_supremum(cur_rec))
+      goto get_next_rec;
+    if (!page_rec_is_infimum(cur_rec) && !rec_get_deleted_flag(cur_rec, comp))
+      break;
+  }
+  if (cur_block != old_block)
+  {
+    ut_ad(latch_mode == BTR_MODIFY_LEAF || latch_mode == BTR_SEARCH_LEAF);
+    if (latch_mode == BTR_SEARCH_LEAF)
+      btr_leaf_page_release(page_cur_get_block(page_cur), latch_mode, mtr);
+    page_cur->block= old_block;
+  }
+  page_cur->rec= old_rec;
+}
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 83654aa547f..414576f7d28 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -2999,7 +2999,6 @@ lock_update_insert(
 inline bool lock_rec_has_gap_or_ordinary(hash_cell_t &cell,
                                          const page_id_t page_id,
                                          ulint heap_no, const rec_t *rec,
-                                         const rec_offs *offsets,
                                          dict_index_t *index,
                                          bool check_locking_read)
 {
@@ -3033,8 +3032,9 @@ inline bool lock_rec_has_gap_or_ordinary(hash_cell_t &cell,
 otherwise
 */
 void lock_update_delete(const buf_block_t *block, const rec_t *rec,
-                        rec_offs *offsets, dict_index_t *index,
-                        bool from_purge, bool convert_lock_to_gap)
+                        dict_index_t *index,
+                        bool from_purge, bool convert_lock_to_gap,
+                        bool release_lock)
 {
 	const page_t*	page = block->frame;
 	ulint		heap_no;
@@ -3064,14 +3064,15 @@ void lock_update_delete(const buf_block_t *block, const rec_t *rec,
 	      next_heap_no, heap_no, convert_lock_to_gap);
 #ifdef UNIV_DEBUG
 	else if (lock_rec_has_gap_or_ordinary(g.cell(), id, heap_no, rec,
-	      offsets, index, true)) {
+	      index, true)) {
 			ut_a(rec_get_deleted_flag(rec, page_is_comp(page)));
 			ut_a(lock_rec_has_gap_or_ordinary(g.cell(), id,
-				next_heap_no, rec, offsets, index, false));
+				next_heap_no, rec, index, false));
 	}
 #endif // UNIV_DEBUG
 	/* Reset the lock bits on rec and release waiting transactions */
-	lock_rec_reset_and_release_wait(g.cell(), id, heap_no);
+	if (release_lock)
+	  lock_rec_reset_and_release_wait(g.cell(), id, heap_no);
 }
 
 /*********************************************************************//**
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 79c644b333d..f9276e2c38d 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -626,6 +626,8 @@ row_undo_mod_del_mark_or_remove_sec_low(
 	    || row_vers_old_has_index_entry(
 		    false, btr_pcur_get_rec(&node->pcur),
 		    &mtr_vers, index, entry, 0, 0)) {
+		/* Temporary table records should not be locked, so there
+		is no need to inherit gap locks here */
 		btr_rec_set_deleted<true>(btr_cur_get_block(btr_cur),
 					  btr_cur_get_rec(btr_cur), &mtr);
 	} else {
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index ee88ee7a4a6..d78c5cf289f 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -2070,14 +2070,15 @@ row_upd_sec_index_entry(
 			btr_rec_set_deleted<true>(btr_cur_get_block(btr_cur),
 						  btr_cur_get_rec(btr_cur),
 						  &mtr);
+//			for_each_delete_marked(btr_cur_get_page_cur(btr_cur),
+//			    index, &mtr,
+//			    gap_lock_inherit_functor(index));
 #ifdef WITH_WSREP
 			if (!referenced && foreign
 			    && wsrep_must_process_fk(node, trx)
 			    && !wsrep_thd_is_BF(trx->mysql_thd, FALSE)) {
-
-				rec_offs* offsets = rec_get_offsets(
-					rec, index, NULL, index->n_core_fields,
-					ULINT_UNDEFINED, &heap);
+			rec_offs *offsets = rec_get_offsets(rec, index, NULL,
+			    index->n_core_fields, ULINT_UNDEFINED, &heap);
 
 				err = wsrep_row_upd_check_foreign_constraints(
 					node, &pcur, index->table,
@@ -2387,7 +2388,7 @@ row_upd_clust_rec_by_insert(
 		}
 
 		err = btr_cur_del_mark_set_clust_rec(
-			btr_cur_get_block(btr_cur), rec, index, offsets,
+			btr_cur, index, offsets,
 			thr, node->row, mtr);
 		if (err != DB_SUCCESS) {
 			goto err_exit;
@@ -2621,7 +2622,6 @@ row_upd_del_mark_clust_rec(
 {
 	btr_pcur_t*	pcur;
 	btr_cur_t*	btr_cur;
-	rec_t*		rec;
 	trx_t*		trx = thr_get_trx(thr);
 
 	ut_ad(dict_index_is_clust(index));
@@ -2642,10 +2642,8 @@ row_upd_del_mark_clust_rec(
 	/* Mark the clustered index record deleted; we do not have to check
 	locks, because we assume that we have an x-lock on the record */
 
-	rec = btr_cur_get_rec(btr_cur);
-
 	dberr_t err = btr_cur_del_mark_set_clust_rec(
-		btr_cur_get_block(btr_cur), rec,
+		btr_cur,
 		index, offsets, thr, node->row, mtr);
 
 	if (err != DB_SUCCESS) {