4 files changed, 263 insertions, 238 deletions

diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 705ff035..6b3a3733 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -264,6 +264,8 @@ btr_root_block_get(
 	mtr_t*			mtr,	/*!< in: mtr */
 	dberr_t*		err)	/*!< out: error code */
 {
+  ut_ad(mode != RW_NO_LATCH);
+
   if (!index->table || !index->table->space)
   {
     *err= DB_TABLESPACE_NOT_FOUND;
@@ -285,13 +287,12 @@ btr_root_block_get(
 
   if (UNIV_LIKELY(block != nullptr))
   {
-    if (UNIV_UNLIKELY(mode == RW_NO_LATCH));
-    else if (!!page_is_comp(block->page.frame) !=
-             index->table->not_redundant() ||
-             btr_page_get_index_id(block->page.frame) != index->id ||
-             !fil_page_index_page_check(block->page.frame) ||
-             index->is_spatial() !=
-             (fil_page_get_type(block->page.frame) == FIL_PAGE_RTREE))
+    if (!!page_is_comp(block->page.frame) !=
+        index->table->not_redundant() ||
+        btr_page_get_index_id(block->page.frame) != index->id ||
+        !fil_page_index_page_check(block->page.frame) ||
+        index->is_spatial() !=
+        (fil_page_get_type(block->page.frame) == FIL_PAGE_RTREE))
     {
       *err= DB_PAGE_CORRUPTED;
       block= nullptr;
@@ -561,13 +562,39 @@ btr_page_alloc_for_ibuf(
   {
     buf_page_make_young_if_needed(&new_block->page);
     *err= flst_remove(root, PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST, new_block,
-                PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST_NODE, mtr);
+                      PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST_NODE,
+                      fil_system.sys_space->free_limit, mtr);
   }
   ut_d(if (*err == DB_SUCCESS)
          flst_validate(root, PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST, mtr));
   return new_block;
 }
 
+static MY_ATTRIBUTE((nonnull, warn_unused_result))
+/** Acquire a latch on the index root page for allocating or freeing pages.
+@param index   index tree
+@param mtr     mini-transaction
+@param err     error code
+@return root page
+@retval nullptr if an error occurred */
+buf_block_t *btr_root_block_sx(dict_index_t *index, mtr_t *mtr, dberr_t *err)
+{
+  buf_block_t *root=
+    mtr->get_already_latched(page_id_t{index->table->space_id, index->page},
+                             MTR_MEMO_PAGE_SX_FIX);
+  if (!root)
+  {
+    root= btr_root_block_get(index, RW_SX_LATCH, mtr, err);
+    if (UNIV_UNLIKELY(!root))
+      return root;
+  }
+#ifdef BTR_CUR_HASH_ADAPT
+  else
+    ut_ad(!root->index || !root->index->freed());
+#endif
+  return root;
+}
+
 /**************************************************************//**
 Allocates a new file page to be used in an index tree. NOTE: we assume
 that the caller has made the reservation for free extents!
@@ -589,21 +616,9 @@ btr_page_alloc_low(
 					page should be initialized. */
 	dberr_t*	err)		/*!< out: error code */
 {
-  const auto savepoint= mtr->get_savepoint();
-  buf_block_t *root= btr_root_block_get(index, RW_NO_LATCH, mtr, err);
+  buf_block_t *root= btr_root_block_sx(index, mtr, err);
   if (UNIV_UNLIKELY(!root))
     return root;
-
-  const bool have_latch= mtr->have_u_or_x_latch(*root);
-#ifdef BTR_CUR_HASH_ADAPT
-  ut_ad(!have_latch || !root->index || !root->index->freed());
-#endif
-  mtr->rollback_to_savepoint(savepoint);
-
-  if (!have_latch &&
-      UNIV_UNLIKELY(!(root= btr_root_block_get(index, RW_SX_LATCH, mtr, err))))
-    return root;
-
   fseg_header_t *seg_header= root->page.frame +
     (level ? PAGE_HEADER + PAGE_BTR_SEG_TOP : PAGE_HEADER + PAGE_BTR_SEG_LEAF);
   return fseg_alloc_free_page_general(seg_header, hint_page_no, file_direction,
@@ -652,7 +667,8 @@ btr_page_free_for_ibuf(
   buf_block_t *root= btr_get_latched_root(*index, mtr);
   dberr_t err=
     flst_add_first(root, PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST,
-                   block, PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST_NODE, mtr);
+                   block, PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST_NODE,
+                   fil_system.sys_space->free_limit, mtr);
   ut_d(if (err == DB_SUCCESS)
          flst_validate(root, PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST, mtr));
   return err;
@@ -696,24 +712,16 @@ dberr_t btr_page_free(dict_index_t* index, buf_block_t* block, mtr_t* mtr,
   fil_space_t *space= index->table->space;
   dberr_t err;
 
-  const auto savepoint= mtr->get_savepoint();
-  if (buf_block_t *root= btr_root_block_get(index, RW_NO_LATCH, mtr, &err))
+  if (buf_block_t *root= btr_root_block_sx(index, mtr, &err))
   {
-    const bool have_latch= mtr->have_u_or_x_latch(*root);
-#ifdef BTR_CUR_HASH_ADAPT
-    ut_ad(!have_latch || !root->index || !root->index->freed());
-#endif
-    mtr->rollback_to_savepoint(savepoint);
-    if (have_latch ||
-        (root= btr_root_block_get(index, RW_SX_LATCH, mtr, &err)))
-      err= fseg_free_page(&root->page.frame[blob ||
-                                            page_is_leaf(block->page.frame)
-                                            ? PAGE_HEADER + PAGE_BTR_SEG_LEAF
-                                            : PAGE_HEADER + PAGE_BTR_SEG_TOP],
-                          space, page, mtr, space_latched);
+    err= fseg_free_page(&root->page.frame[blob ||
+                                          page_is_leaf(block->page.frame)
+                                          ? PAGE_HEADER + PAGE_BTR_SEG_LEAF
+                                          : PAGE_HEADER + PAGE_BTR_SEG_TOP],
+                        space, page, mtr, space_latched);
+    if (err == DB_SUCCESS)
+      buf_page_free(space, page, mtr);
   }
-  if (err == DB_SUCCESS)
-    buf_page_free(space, page, mtr);
 
   /* The page was marked free in the allocation bitmap, but it
   should remain exclusively latched until mtr_t::commit() or until it
@@ -1291,54 +1299,71 @@ btr_read_autoinc(dict_index_t* index)
 	return autoinc;
 }
 
-/** Read the last used AUTO_INCREMENT value from PAGE_ROOT_AUTO_INC,
-or fall back to MAX(auto_increment_column).
-@param[in]	table	table containing an AUTO_INCREMENT column
-@param[in]	col_no	index of the AUTO_INCREMENT column
-@return	the AUTO_INCREMENT value
-@retval	0 on error or if no AUTO_INCREMENT value was used yet */
-ib_uint64_t
-btr_read_autoinc_with_fallback(const dict_table_t* table, unsigned col_no)
+dict_index_t *dict_table_t::get_index(const dict_col_t &col) const
 {
-	ut_ad(table->persistent_autoinc);
-	ut_ad(!table->is_temporary());
+  dict_index_t *index= dict_table_get_first_index(this);
 
-	dict_index_t*	index = dict_table_get_first_index(table);
+  while (index && (index->fields[0].col != &col || index->is_corrupted()))
+    index= dict_table_get_next_index(index);
 
-	if (index == NULL) {
-		return 0;
-	}
+  return index;
+}
 
-	mtr_t		mtr;
-	mtr.start();
-	buf_block_t*	block = buf_page_get(
-		page_id_t(index->table->space_id, index->page),
-		index->table->space->zip_size(),
-		RW_S_LATCH, &mtr);
-
-	ib_uint64_t	autoinc	= block
-		? page_get_autoinc(block->page.frame) : 0;
-	const bool	retry	= block && autoinc == 0
-		&& !page_is_empty(block->page.frame);
-	mtr.commit();
+/** Read the last used AUTO_INCREMENT value from PAGE_ROOT_AUTO_INC,
+or fall back to MAX(auto_increment_column).
+@param table          table containing an AUTO_INCREMENT column
+@param col_no         index of the AUTO_INCREMENT column
+@param mysql_version  TABLE_SHARE::mysql_version
+@param max            the maximum value of the AUTO_INCREMENT column
+@return the AUTO_INCREMENT value
+@retval 0 on error or if no AUTO_INCREMENT value was used yet */
+uint64_t btr_read_autoinc_with_fallback(const dict_table_t *table,
+                                        unsigned col_no, ulong mysql_version,
+                                        uint64_t max)
+{
+  ut_ad(table->persistent_autoinc);
+  ut_ad(!table->is_temporary());
 
-	if (retry) {
-		/* This should be an old data file where
-		PAGE_ROOT_AUTO_INC was initialized to 0.
-		Fall back to reading MAX(autoinc_col).
-		There should be an index on it. */
-		const dict_col_t*	autoinc_col
-			= dict_table_get_nth_col(table, col_no);
-		while (index && index->fields[0].col != autoinc_col) {
-			index = dict_table_get_next_index(index);
-		}
+  uint64_t autoinc= 0;
+  mtr_t mtr;
+  mtr.start();
 
-		if (index) {
-			autoinc = row_search_max_autoinc(index);
-		}
-	}
+  if (buf_block_t *block=
+      buf_page_get(page_id_t(table->space_id,
+                             dict_table_get_first_index(table)->page),
+                   table->space->zip_size(), RW_SX_LATCH, &mtr))
+  {
+    autoinc= page_get_autoinc(block->page.frame);
 
-	return autoinc;
+    if (autoinc > 0 && autoinc <= max && mysql_version >= 100210);
+    else if (dict_index_t *index=
+             table->get_index(*dict_table_get_nth_col(table, col_no)))
+    {
+      /* Read MAX(autoinc_col), in case this table had originally been
+      created before MariaDB 10.2.4 introduced persistent AUTO_INCREMENT
+      and MariaDB 10.2.10 fixed MDEV-12123, and there could be a garbage
+      value in the PAGE_ROOT_AUTO_INC field. */
+      const uint64_t max_autoinc= row_search_max_autoinc(index);
+      const bool need_adjust{autoinc > max || autoinc < max_autoinc};
+      ut_ad(max_autoinc <= max);
+
+      if (UNIV_UNLIKELY(need_adjust) && !high_level_read_only && !opt_readonly)
+      {
+        sql_print_information("InnoDB: Resetting PAGE_ROOT_AUTO_INC from "
+                              UINT64PF " to " UINT64PF
+                              " on table %`.*s.%`s (created with version %lu)",
+                              autoinc, max_autoinc,
+                              int(table->name.dblen()), table->name.m_name,
+                              table->name.basename(), mysql_version);
+        autoinc= max_autoinc;
+        index->set_modified(mtr);
+        page_set_autoinc(block, max_autoinc, &mtr, true);
+      }
+    }
+  }
+
+  mtr.commit();
+  return autoinc;
 }
 
 /** Write the next available AUTO_INCREMENT value to PAGE_ROOT_AUTO_INC.
diff --git a/storage/innobase/btr/btr0bulk.cc b/storage/innobase/btr/btr0bulk.cc
index 5bf68c58..e2513ad6 100644
--- a/storage/innobase/btr/btr0bulk.cc
+++ b/storage/innobase/btr/btr0bulk.cc
@@ -837,7 +837,7 @@ PageBulk::release()
 	m_block->page.fix();
 
 	/* No other threads can modify this block. */
-	m_modify_clock = buf_block_get_modify_clock(m_block);
+	m_modify_clock = m_block->modify_clock;
 
 	m_mtr.commit();
 }
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 46afb73b..2fc05b06 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -817,7 +817,7 @@ static ulint btr_node_ptr_max_size(const dict_index_t* index)
 		/* Determine the maximum length of the index field. */
 
 		field_max_size = dict_col_get_fixed_size(col, comp);
-		if (field_max_size) {
+		if (field_max_size && field->fixed_len) {
 			/* dict_index_add_col() should guarantee this */
 			ut_ad(!field->prefix_len
 			      || field->fixed_len == field->prefix_len);
@@ -935,7 +935,7 @@ static inline page_cur_mode_t btr_cur_nonleaf_mode(page_cur_mode_t mode)
   return PAGE_CUR_LE;
 }
 
-static MY_ATTRIBUTE((nonnull))
+MY_ATTRIBUTE((nonnull,warn_unused_result))
 /** Acquire a latch on the previous page without violating the latching order.
 @param block    index page
 @param page_id  page identifier with valid space identifier
@@ -946,8 +946,9 @@ static MY_ATTRIBUTE((nonnull))
 @retval 0  if an error occurred
 @retval 1  if the page could be latched in the wrong order
 @retval -1 if the latch on block was temporarily released */
-int btr_latch_prev(buf_block_t *block, page_id_t page_id, ulint zip_size,
-                   rw_lock_type_t rw_latch, mtr_t *mtr, dberr_t *err)
+static int btr_latch_prev(buf_block_t *block, page_id_t page_id,
+                          ulint zip_size,
+                          rw_lock_type_t rw_latch, mtr_t *mtr, dberr_t *err)
 {
   ut_ad(rw_latch == RW_S_LATCH || rw_latch == RW_X_LATCH);
   ut_ad(page_id.space() == block->page.id().space());
@@ -955,47 +956,80 @@ int btr_latch_prev(buf_block_t *block, page_id_t page_id, ulint zip_size,
   const auto prev_savepoint= mtr->get_savepoint();
   ut_ad(block == mtr->at_savepoint(prev_savepoint - 1));
 
-  page_id.set_page_no(btr_page_get_prev(block->page.frame));
+  const page_t *const page= block->page.frame;
+  page_id.set_page_no(btr_page_get_prev(page));
+  /* We are holding a latch on the current page.
+
+  We will start by buffer-fixing the left sibling. Waiting for a latch
+  on it while holding a latch on the current page could lead to a
+  deadlock, because another thread could hold that latch and wait for
+  a right sibling page latch (the current page).
+
+  If there is a conflict, we will temporarily release our latch on the
+  current block while waiting for a latch on the left sibling.  The
+  buffer-fixes on both blocks will prevent eviction. */
+
+ retry:
   buf_block_t *prev= buf_page_get_gen(page_id, zip_size, RW_NO_LATCH, nullptr,
                                       BUF_GET, mtr, err, false);
   if (UNIV_UNLIKELY(!prev))
     return 0;
 
   int ret= 1;
-  if (UNIV_UNLIKELY(rw_latch == RW_S_LATCH))
+  static_assert(MTR_MEMO_PAGE_S_FIX == mtr_memo_type_t(BTR_SEARCH_LEAF), "");
+  static_assert(MTR_MEMO_PAGE_X_FIX == mtr_memo_type_t(BTR_MODIFY_LEAF), "");
+
+  if (rw_latch == RW_S_LATCH
+      ? prev->page.lock.s_lock_try() : prev->page.lock.x_lock_try())
   {
-    if (UNIV_LIKELY(prev->page.lock.s_lock_try()))
+    mtr->lock_register(prev_savepoint, mtr_memo_type_t(rw_latch));
+    if (UNIV_UNLIKELY(prev->page.id() != page_id))
     {
-      mtr->lock_register(prev_savepoint, MTR_MEMO_PAGE_S_FIX);
-      goto prev_latched;
+    fail:
+      /* the page was just read and found to be corrupted */
+      mtr->rollback_to_savepoint(prev_savepoint);
+      return 0;
     }
-    block->page.lock.s_unlock();
   }
   else
   {
-    if (UNIV_LIKELY(prev->page.lock.x_lock_try()))
+    ut_ad(mtr->at_savepoint(mtr->get_savepoint() - 1)->page.id() == page_id);
+    mtr->release_last_page();
+    if (rw_latch == RW_S_LATCH)
+      block->page.lock.s_unlock();
+    else
+      block->page.lock.x_unlock();
+
+    prev= buf_page_get_gen(page_id, zip_size, rw_latch, prev,
+                           BUF_GET, mtr, err);
+    if (rw_latch == RW_S_LATCH)
+      block->page.lock.s_lock();
+    else
+      block->page.lock.x_lock();
+
+    const page_id_t prev_page_id= page_id;
+    page_id.set_page_no(btr_page_get_prev(page));
+
+    if (UNIV_UNLIKELY(page_id != prev_page_id))
     {
-      mtr->lock_register(prev_savepoint, MTR_MEMO_PAGE_X_FIX);
-      goto prev_latched;
+      mtr->release_last_page();
+      if (page_id.page_no() == FIL_NULL)
+        return -1;
+      goto retry;
     }
-    block->page.lock.x_unlock();
+
+    if (UNIV_UNLIKELY(!prev))
+      goto fail;
+
+    ret= -1;
   }
 
-  ret= -1;
-  mtr->lock_register(prev_savepoint - 1, MTR_MEMO_BUF_FIX);
-  mtr->rollback_to_savepoint(prev_savepoint);
-  prev= buf_page_get_gen(page_id, zip_size, rw_latch, prev,
-                         BUF_GET, mtr, err, false);
-  if (UNIV_UNLIKELY(!prev))
-    return 0;
-  mtr->upgrade_buffer_fix(prev_savepoint - 1, rw_latch);
-
- prev_latched:
-  if (memcmp_aligned<2>(FIL_PAGE_TYPE + prev->page.frame,
-                        FIL_PAGE_TYPE + block->page.frame, 2) ||
-      memcmp_aligned<2>(PAGE_HEADER + PAGE_INDEX_ID + prev->page.frame,
-                        PAGE_HEADER + PAGE_INDEX_ID + block->page.frame, 8) ||
-      page_is_comp(prev->page.frame) != page_is_comp(block->page.frame))
+  const page_t *const p= prev->page.frame;
+  if (memcmp_aligned<4>(FIL_PAGE_NEXT + p, FIL_PAGE_OFFSET + page, 4) ||
+      memcmp_aligned<2>(FIL_PAGE_TYPE + p, FIL_PAGE_TYPE + page, 2) ||
+      memcmp_aligned<2>(PAGE_HEADER + PAGE_INDEX_ID + p,
+                        PAGE_HEADER + PAGE_INDEX_ID + page, 8) ||
+      page_is_comp(p) != page_is_comp(page))
   {
     ut_ad("corrupted" == 0); // FIXME: remove this
     *err= DB_CORRUPTION;
@@ -6092,7 +6126,6 @@ btr_store_big_rec_extern_fields(
 		for (ulint blob_npages = 0;; ++blob_npages) {
 			buf_block_t*	block;
 			const ulint	commit_freq = 4;
-			uint32_t	r_extents;
 
 			ut_ad(page_align(field_ref) == page_align(rec));
 
@@ -6127,22 +6160,14 @@ btr_store_big_rec_extern_fields(
 				hint_prev = rec_block->page.id().page_no();
 			}
 
-			error = fsp_reserve_free_extents(
-				&r_extents, index->table->space, 1,
-				FSP_BLOB, &mtr, 1);
-			if (UNIV_UNLIKELY(error != DB_SUCCESS)) {
-alloc_fail:
-				mtr.commit();
-				goto func_exit;
-			}
-
 			block = btr_page_alloc(index, hint_prev + 1,
 					       FSP_NO_DIR, 0, &mtr, &mtr,
 					       &error);
 
-			index->table->space->release_free_extents(r_extents);
 			if (!block) {
-				goto alloc_fail;
+alloc_fail:
+                                mtr.commit();
+				goto func_exit;
 			}
 
 			const uint32_t space_id = block->page.id().space();
diff --git a/storage/innobase/btr/btr0pcur.cc b/storage/innobase/btr/btr0pcur.cc
index 2131fb94..de0f9e93 100644
--- a/storage/innobase/btr/btr0pcur.cc
+++ b/storage/innobase/btr/btr0pcur.cc
@@ -179,10 +179,8 @@ before_first:
 						 cursor->old_n_fields,
 						 &cursor->old_rec_buf,
 						 &cursor->buf_size);
-	cursor->block_when_stored.store(block);
-
-	/* Function try to check if block is S/X latch. */
-	cursor->modify_clock = buf_block_get_modify_clock(block);
+	cursor->old_page_id = block->page.id();
+	cursor->modify_clock = block->modify_clock;
 }
 
 /**************************************************************//**
@@ -214,101 +212,80 @@ btr_pcur_copy_stored_position(
 }
 
 /** Optimistically latches the leaf page or pages requested.
-@param[in]	block		guessed buffer block
-@param[in,out]	pcur		cursor
-@param[in,out]	latch_mode	BTR_SEARCH_LEAF, ...
-@param[in,out]	mtr		mini-transaction
-@return true if success */
+@param pcur        persistent cursor
+@param latch_mode  BTR_SEARCH_LEAF, ...
+@param mtr         mini-transaction
+@return true on success */
 TRANSACTIONAL_TARGET
-static bool btr_pcur_optimistic_latch_leaves(buf_block_t *block,
-                                             btr_pcur_t *pcur,
+static bool btr_pcur_optimistic_latch_leaves(btr_pcur_t *pcur,
                                              btr_latch_mode *latch_mode,
                                              mtr_t *mtr)
 {
-  ut_ad(block->page.buf_fix_count());
-  ut_ad(block->page.in_file());
-  ut_ad(block->page.frame);
-
   static_assert(BTR_SEARCH_PREV & BTR_SEARCH_LEAF, "");
   static_assert(BTR_MODIFY_PREV & BTR_MODIFY_LEAF, "");
   static_assert((BTR_SEARCH_PREV ^ BTR_MODIFY_PREV) ==
                 (RW_S_LATCH ^ RW_X_LATCH), "");
 
+  buf_block_t *const block=
+    buf_page_optimistic_fix(pcur->btr_cur.page_cur.block, pcur->old_page_id);
+
+  if (!block)
+    return false;
+
+  if (*latch_mode == BTR_SEARCH_LEAF || *latch_mode == BTR_MODIFY_LEAF)
+    return buf_page_optimistic_get(block, rw_lock_type_t(*latch_mode),
+                                   pcur->modify_clock, mtr);
+
+  ut_ad(*latch_mode == BTR_SEARCH_PREV || *latch_mode == BTR_MODIFY_PREV);
   const rw_lock_type_t mode=
     rw_lock_type_t(*latch_mode & (RW_X_LATCH | RW_S_LATCH));
 
-  switch (*latch_mode) {
-  default:
-    ut_ad(*latch_mode == BTR_SEARCH_LEAF || *latch_mode == BTR_MODIFY_LEAF);
-    return buf_page_optimistic_get(mode, block, pcur->modify_clock, mtr);
-  case BTR_SEARCH_PREV:
-  case BTR_MODIFY_PREV:
-    page_id_t id{0};
-    uint32_t left_page_no;
-    ulint zip_size;
-    buf_block_t *left_block= nullptr;
-    {
-      transactional_shared_lock_guard<block_lock> g{block->page.lock};
-      if (block->modify_clock != pcur->modify_clock)
-        return false;
-      id= block->page.id();
-      zip_size= block->zip_size();
-      left_page_no= btr_page_get_prev(block->page.frame);
-    }
-
-    if (left_page_no != FIL_NULL)
-    {
-      left_block=
-        buf_page_get_gen(page_id_t(id.space(), left_page_no), zip_size,
-                         mode, nullptr, BUF_GET_POSSIBLY_FREED, mtr);
-
-      if (!left_block);
-      else if (btr_page_get_next(left_block->page.frame) != id.page_no())
-      {
-release_left_block:
-        mtr->release_last_page();
-        return false;
-      }
-      else
-        buf_page_make_young_if_needed(&left_block->page);
-    }
-
-    if (buf_page_optimistic_get(mode, block, pcur->modify_clock, mtr))
-    {
-      if (btr_page_get_prev(block->page.frame) == left_page_no)
-      {
-        /* block was already buffer-fixed while entering the function and
-        buf_page_optimistic_get() buffer-fixes it again. */
-        ut_ad(2 <= block->page.buf_fix_count());
-        *latch_mode= btr_latch_mode(mode);
-        return true;
-      }
-
-      mtr->release_last_page();
-    }
-
-    ut_ad(block->page.buf_fix_count());
-    if (left_block)
-      goto release_left_block;
-    return false;
+  uint64_t modify_clock;
+  uint32_t left_page_no;
+  const page_t *const page= block->page.frame;
+  {
+    transactional_shared_lock_guard<block_lock> g{block->page.lock};
+    modify_clock= block->modify_clock;
+    left_page_no= btr_page_get_prev(page);
   }
-}
 
-/** Structure acts as functor to do the latching of leaf pages.
-It returns true if latching of leaf pages succeeded and false
-otherwise. */
-struct optimistic_latch_leaves
-{
-  btr_pcur_t *const cursor;
-  btr_latch_mode *const latch_mode;
-  mtr_t *const mtr;
+  const auto savepoint= mtr->get_savepoint();
+  mtr->memo_push(block, MTR_MEMO_BUF_FIX);
 
-  bool operator()(buf_block_t *hint) const
+  if (UNIV_UNLIKELY(modify_clock != pcur->modify_clock))
   {
-    return hint &&
-      btr_pcur_optimistic_latch_leaves(hint, cursor, latch_mode, mtr);
+  fail:
+    mtr->rollback_to_savepoint(savepoint);
+    return false;
+  }
+
+  buf_block_t *prev;
+  if (left_page_no != FIL_NULL)
+  {
+    prev= buf_page_get_gen(page_id_t(pcur->old_page_id.space(),
+                                     left_page_no), block->zip_size(),
+                           mode, nullptr, BUF_GET_POSSIBLY_FREED, mtr);
+    if (!prev ||
+        page_is_comp(prev->page.frame) != page_is_comp(block->page.frame) ||
+        memcmp_aligned<2>(block->page.frame, prev->page.frame, 2) ||
+        memcmp_aligned<2>(block->page.frame + PAGE_HEADER + PAGE_INDEX_ID,
+                          prev->page.frame + PAGE_HEADER + PAGE_INDEX_ID, 8))
+      goto fail;
   }
-};
+  else
+    prev= nullptr;
+
+  mtr->upgrade_buffer_fix(savepoint, mode);
+
+  if (UNIV_UNLIKELY(block->modify_clock != modify_clock) ||
+      UNIV_UNLIKELY(block->page.is_freed()) ||
+      (prev &&
+       memcmp_aligned<4>(FIL_PAGE_NEXT + prev->page.frame,
+                         FIL_PAGE_OFFSET + page, 4)))
+    goto fail;
+
+  return true;
+}
 
 /** Restores the stored position of a persistent cursor bufferfixing
 the page and obtaining the specified latches. If the cursor position
@@ -331,6 +308,7 @@ btr_pcur_t::SAME_UNIQ cursor position is on user rec and points on the
 record with the same unique field values as in the stored record,
 btr_pcur_t::NOT_SAME cursor position is not on user rec or points on
 the record with not the samebuniq field values as in the stored */
+TRANSACTIONAL_TARGET
 btr_pcur_t::restore_status
 btr_pcur_t::restore_position(btr_latch_mode restore_latch_mode, mtr_t *mtr)
 {
@@ -361,7 +339,6 @@ btr_pcur_t::restore_position(btr_latch_mode restore_latch_mode, mtr_t *mtr)
 		latch_mode =
 			BTR_LATCH_MODE_WITHOUT_INTENTION(restore_latch_mode);
 		pos_state = BTR_PCUR_IS_POSITIONED;
-		block_when_stored.clear();
 
 		return restore_status::NOT_SAME;
 	}
@@ -378,9 +355,8 @@ btr_pcur_t::restore_position(btr_latch_mode restore_latch_mode, mtr_t *mtr)
 	case BTR_SEARCH_PREV:
 	case BTR_MODIFY_PREV:
 		/* Try optimistic restoration. */
-		if (block_when_stored.run_with_hint(
-			optimistic_latch_leaves{this, &restore_latch_mode,
-						mtr})) {
+		if (btr_pcur_optimistic_latch_leaves(this, &restore_latch_mode,
+						     mtr)) {
 			pos_state = BTR_PCUR_IS_POSITIONED;
 			latch_mode = restore_latch_mode;
 
@@ -485,16 +461,22 @@ btr_pcur_t::restore_position(btr_latch_mode restore_latch_mode, mtr_t *mtr)
 			since the cursor can now be on a different page!
 			But we can retain the value of old_rec */
 
-			block_when_stored.store(btr_pcur_get_block(this));
-			modify_clock= buf_block_get_modify_clock(
-			    block_when_stored.block());
+			old_page_id = btr_cur.page_cur.block->page.id();
+			modify_clock = btr_cur.page_cur.block->modify_clock;
 
 			mem_heap_free(heap);
 
 			return restore_status::SAME_ALL;
 		}
-		if (n_matched_fields >= index->n_uniq)
-			ret_val= restore_status::SAME_UNIQ;
+		if (n_matched_fields >= index->n_uniq
+		    /* Unique indexes can contain "NULL" keys, and if all
+		    unique fields are NULL and not all tuple
+		    fields match to record fields, then treat it as if
+		    restored cursor position points to the record with
+		    not the same unique key. */
+		    && !(index->n_nullable
+			    && dtuple_contains_null(tuple, index->n_uniq)))
+			  ret_val= restore_status::SAME_UNIQ;
 	}
 
 	mem_heap_free(heap);
@@ -612,40 +594,33 @@ btr_pcur_move_backward_from_page(
 		return true;
 	}
 
-	buf_block_t* block = btr_pcur_get_block(cursor);
-
-	if (page_has_prev(block->page.frame)) {
-		buf_block_t* left_block
-			= mtr->at_savepoint(mtr->get_savepoint() - 1);
-		const page_t* const left = left_block->page.frame;
-		if (memcmp_aligned<4>(left + FIL_PAGE_NEXT,
-				      block->page.frame
-				      + FIL_PAGE_OFFSET, 4)) {
-			/* This should be the right sibling page, or
-			if there is none, the current block. */
-			ut_ad(left_block == block
-			      || !memcmp_aligned<4>(left + FIL_PAGE_PREV,
-						    block->page.frame
-						    + FIL_PAGE_OFFSET, 4));
-			/* The previous one must be the left sibling. */
-			left_block
-				= mtr->at_savepoint(mtr->get_savepoint() - 2);
-			ut_ad(!memcmp_aligned<4>(left_block->page.frame
-						 + FIL_PAGE_NEXT,
-						 block->page.frame
-						 + FIL_PAGE_OFFSET, 4));
-		}
+	buf_block_t* block = mtr->at_savepoint(0);
+	ut_ad(block == btr_pcur_get_block(cursor));
+	const page_t* const page = block->page.frame;
+	/* btr_pcur_optimistic_latch_leaves() will acquire a latch on
+	the preceding page if one exists;
+	if that fails, btr_cur_t::search_leaf() invoked by
+	btr_pcur_open_with_no_init() will also acquire a latch on the
+	succeeding page. Our caller only needs one page latch. */
+	ut_ad(mtr->get_savepoint() <= 3);
+
+	if (page_has_prev(page)) {
+		buf_block_t* const left_block = mtr->at_savepoint(1);
+		ut_ad(!memcmp_aligned<4>(page + FIL_PAGE_OFFSET,
+					 left_block->page.frame
+					 + FIL_PAGE_NEXT, 4));
 		if (btr_pcur_is_before_first_on_page(cursor)) {
+			/* Reposition on the previous page. */
 			page_cur_set_after_last(left_block,
 						&cursor->btr_cur.page_cur);
 			/* Release the right sibling. */
-		} else {
-			/* Release the left sibling. */
+			mtr->rollback_to_savepoint(0, 1);
 			block = left_block;
 		}
-		mtr->release(*block);
 	}
 
+	mtr->rollback_to_savepoint(1);
+	ut_ad(block == mtr->at_savepoint(0));
 	cursor->latch_mode = latch_mode;
 	cursor->old_rec = nullptr;
 	return false;