MDEV-29603 btr_cur_open_at_index_side() is missing some consistency checks

btr_cur_t: Zero-initialize all fields in the default constructor.

btr_cur_t::index: Remove; it duplicated page_cur.index.

Many functions: Remove arguments that were duplicating
page_cur_t::index and page_cur_t::block.

page_cur_open_level(), btr_pcur_open_level(): Replaces
btr_cur_open_at_index_side() for dict_stats_analyze_index().
At the end, release all latches except the dict_index_t::lock
and the buf_page_t::lock on the requested page.

dict_stats_analyze_index(): Rely on mtr_t::rollback_to_savepoint()
to release all uninteresting page latches.

btr_search_guess_on_hash(): Simplify the logic, and invoke
mtr_t::rollback_to_savepoint().

We will use plain C++ std::vector<mtr_memo_slot_t> for mtr_t::m_memo.
In this way, we can avoid setting mtr_memo_slot_t::object to nullptr
and instead just remove garbage from m_memo.

mtr_t::rollback_to_savepoint(): Shrink the vector. We will be needing this
in dict_stats_analyze_index(), where we will release page latches and
only retain the index->lock in mtr_t::m_memo.

mtr_t::release_last_page(): Release the last acquired page latch.
Replaces btr_leaf_page_release().

mtr_t::release(const buf_block_t&): Release a single page latch.
Used in btr_pcur_move_backward_from_page().

mtr_t::memo_release(): Replaced with mtr_t::release().

mtr_t::upgrade_buffer_fix(): Acquire a latch for a buffer-fixed page.
This replaces the double bookkeeping in btr_cur_t::open_leaf().

Reviewed by: Vladislav Lesin

11 files changed, 141 insertions, 135 deletions

diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index b25864f91e8..9e336f267a3 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -257,9 +257,10 @@ public:
 	}
 
 	/** Position the cursor on the first user record. */
-	rec_t* open(buf_block_t* block) noexcept
+	rec_t* open(buf_block_t* block, const dict_index_t* index) noexcept
 		MY_ATTRIBUTE((warn_unused_result))
 	{
+		m_cur.index = const_cast<dict_index_t*>(index);
 		page_cur_set_before_first(block, &m_cur);
 		return next();
 	}
@@ -289,10 +290,9 @@ public:
 
 	/** Remove the current record
 	@return true on success */
-	bool remove(
-		const dict_index_t*	index,
-		rec_offs*		offsets) UNIV_NOTHROW
+	bool remove(rec_offs* offsets) UNIV_NOTHROW
 	{
+		const dict_index_t* const index = m_cur.index;
 		ut_ad(page_is_leaf(m_cur.block->page.frame));
 		/* We can't end up with an empty page unless it is root. */
 		if (page_get_n_recs(m_cur.block->page.frame) <= 1) {
@@ -315,7 +315,7 @@ public:
 			     page_zip, m_cur.block->page.frame, index));
 #endif /* UNIV_ZIP_DEBUG */
 
-		page_cur_delete_rec(&m_cur, index, offsets, &m_mtr);
+		page_cur_delete_rec(&m_cur, offsets, &m_mtr);
 
 #ifdef UNIV_ZIP_DEBUG
 		ut_a(!page_zip || page_zip_validate(
@@ -1734,10 +1734,8 @@ re-organising the B+tree.
 @return true if purge succeeded */
 inline bool PageConverter::purge() UNIV_NOTHROW
 {
-	const dict_index_t*	index = m_index->m_srv_index;
-
 	/* We can't have a page that is empty and not root. */
-	if (m_rec_iter.remove(index, m_offsets)) {
+	if (m_rec_iter.remove(m_offsets)) {
 
 		++m_index->m_stats.m_n_purged;
 
@@ -1801,7 +1799,7 @@ PageConverter::update_records(
 
 	/* This will also position the cursor on the first user record. */
 
-	if (!m_rec_iter.open(block)) {
+	if (!m_rec_iter.open(block, m_index->m_srv_index)) {
 		return DB_CORRUPTION;
 	}
 
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 1d772b02704..9b1d9a8b57f 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -194,8 +194,8 @@ row_ins_sec_index_entry_by_modify(
 
 	rec = btr_cur_get_rec(cursor);
 
-	ut_ad(!dict_index_is_clust(cursor->index));
-	ut_ad(rec_offs_validate(rec, cursor->index, *offsets));
+	ut_ad(!cursor->index()->is_clust());
+	ut_ad(rec_offs_validate(rec, cursor->index(), *offsets));
 	ut_ad(!entry->info_bits);
 
 	/* We know that in the alphabetical ordering, entry and rec are
@@ -204,7 +204,7 @@ row_ins_sec_index_entry_by_modify(
 	difference. */
 
 	update = row_upd_build_sec_rec_difference_binary(
-		rec, cursor->index, *offsets, entry, heap);
+		rec, cursor->index(), *offsets, entry, heap);
 
 	if (!rec_get_deleted_flag(rec, rec_offs_comp(*offsets))) {
 		/* We should never insert in place of a record that
@@ -218,8 +218,8 @@ row_ins_sec_index_entry_by_modify(
 		returns. After that point, set_committed(true)
 		would be invoked in commit_inplace_alter_table(). */
 		ut_a(update->n_fields == 0);
-		ut_a(!cursor->index->is_committed());
-		ut_ad(!dict_index_is_online_ddl(cursor->index));
+		ut_a(!cursor->index()->is_committed());
+		ut_ad(!dict_index_is_online_ddl(cursor->index()));
 		return(DB_SUCCESS);
 	}
 
@@ -288,15 +288,15 @@ row_ins_clust_index_entry_by_modify(
 	dberr_t		err = DB_SUCCESS;
 	btr_cur_t*	cursor	= btr_pcur_get_btr_cur(pcur);
 	TABLE*		mysql_table = NULL;
-	ut_ad(dict_index_is_clust(cursor->index));
+	ut_ad(cursor->index()->is_clust());
 
 	rec = btr_cur_get_rec(cursor);
 
 	ut_ad(rec_get_deleted_flag(rec,
-				   dict_table_is_comp(cursor->index->table)));
+				   cursor->index()->table->not_redundant()));
 	/* In delete-marked records, DB_TRX_ID must
 	always refer to an existing undo log record. */
-	ut_ad(rec_get_trx_id(rec, cursor->index));
+	ut_ad(rec_get_trx_id(rec, cursor->index()));
 
 	/* Build an update vector containing all the fields to be modified;
 	NOTE that this vector may NOT contain system columns trx_id or
@@ -307,7 +307,7 @@ row_ins_clust_index_entry_by_modify(
 	}
 
 	update = row_upd_build_difference_binary(
-		cursor->index, entry, rec, NULL, true, true,
+		cursor->index(), entry, rec, NULL, true, true,
 		thr_get_trx(thr), heap, mysql_table, &err);
 	if (err != DB_SUCCESS) {
 		return(err);
@@ -1115,7 +1115,7 @@ row_ins_foreign_check_on_constraint(
 		goto nonstandard_exit_func;
 	}
 
-	index = btr_pcur_get_btr_cur(pcur)->index;
+	index = pcur->index();
 
 	ut_a(index == foreign->foreign_index);
 
@@ -1139,7 +1139,8 @@ row_ins_foreign_check_on_constraint(
 		ref = row_build_row_ref(ROW_COPY_POINTERS, index, rec,
 					tmp_heap);
 		cascade->pcur->old_rec = nullptr;
-		err = btr_pcur_open_with_no_init(clust_index, ref,
+		cascade->pcur->btr_cur.page_cur.index = clust_index;
+		err = btr_pcur_open_with_no_init(ref,
 						 PAGE_CUR_LE, BTR_SEARCH_LEAF,
 						 cascade->pcur, mtr);
 		if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
@@ -1623,9 +1624,9 @@ row_ins_check_foreign_constraint(
 	n_fields_cmp = dtuple_get_n_fields_cmp(entry);
 
 	dtuple_set_n_fields_cmp(entry, foreign->n_fields);
-
-	err = btr_pcur_open(check_index, entry, PAGE_CUR_GE,
-			    BTR_SEARCH_LEAF, &pcur, &mtr);
+	pcur.btr_cur.page_cur.index = check_index;
+	err = btr_pcur_open(entry, PAGE_CUR_GE, BTR_SEARCH_LEAF, &pcur, 0,
+			    &mtr);
 	if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
 		goto end_scan;
 	}
@@ -2065,9 +2066,9 @@ row_ins_scan_sec_index_for_duplicate(
 
 	dtuple_set_n_fields_cmp(entry, n_unique);
 	const auto allow_duplicates = thr_get_trx(thr)->duplicates;
-
-	dberr_t err = btr_pcur_open(index, entry, PAGE_CUR_GE, BTR_SEARCH_LEAF,
-				    &pcur, mtr);
+	pcur.btr_cur.page_cur.index = index;
+	dberr_t err = btr_pcur_open(entry, PAGE_CUR_GE, BTR_SEARCH_LEAF,
+				    &pcur, 0, mtr);
 	if (err != DB_SUCCESS) {
 		goto end_scan;
 	}
@@ -2219,11 +2220,11 @@ row_ins_duplicate_error_in_clust_online(
 	dberr_t		err	= DB_SUCCESS;
 	const rec_t*	rec	= btr_cur_get_rec(cursor);
 
-	ut_ad(!cursor->index->is_instant());
+	ut_ad(!cursor->index()->is_instant());
 
 	if (cursor->low_match >= n_uniq && !page_rec_is_infimum(rec)) {
-		*offsets = rec_get_offsets(rec, cursor->index, *offsets,
-					   cursor->index->n_fields,
+		*offsets = rec_get_offsets(rec, cursor->index(), *offsets,
+					   cursor->index()->n_fields,
 					   ULINT_UNDEFINED, heap);
 		err = row_ins_duplicate_online(n_uniq, entry, rec, *offsets);
 		if (err != DB_SUCCESS) {
@@ -2236,8 +2237,8 @@ row_ins_duplicate_error_in_clust_online(
 	}
 
 	if (cursor->up_match >= n_uniq && !page_rec_is_supremum(rec)) {
-		*offsets = rec_get_offsets(rec, cursor->index, *offsets,
-					   cursor->index->n_fields,
+		*offsets = rec_get_offsets(rec, cursor->index(), *offsets,
+					   cursor->index()->n_fields,
 					   ULINT_UNDEFINED, heap);
 		err = row_ins_duplicate_online(n_uniq, entry, rec, *offsets);
 	}
@@ -2270,7 +2271,7 @@ row_ins_duplicate_error_in_clust(
 	rec_offs* offsets		= offsets_;
 	rec_offs_init(offsets_);
 
-	ut_ad(dict_index_is_clust(cursor->index));
+	ut_ad(cursor->index()->is_clust());
 
 	/* NOTE: For unique non-clustered indexes there may be any number
 	of delete marked records with the same value for the non-clustered
@@ -2285,15 +2286,17 @@ row_ins_duplicate_error_in_clust(
 	user records on the leaf level. So, even if low_match would suggest
 	that a duplicate key violation may occur, this may not be the case. */
 
-	n_unique = dict_index_get_n_unique(cursor->index);
+	n_unique = dict_index_get_n_unique(cursor->index());
 
 	if (cursor->low_match >= n_unique) {
 
 		rec = btr_cur_get_rec(cursor);
 
 		if (!page_rec_is_infimum(rec)) {
-			offsets = rec_get_offsets(rec, cursor->index, offsets,
-						  cursor->index->n_core_fields,
+			offsets = rec_get_offsets(rec, cursor->index(),
+						  offsets,
+						  cursor->index()
+						  ->n_core_fields,
 						  ULINT_UNDEFINED, &heap);
 
 			/* We set a lock on the possible duplicate: this
@@ -2314,13 +2317,13 @@ row_ins_duplicate_error_in_clust(
 				err = row_ins_set_exclusive_rec_lock(
 					LOCK_REC_NOT_GAP,
 					btr_cur_get_block(cursor),
-					rec, cursor->index, offsets, thr);
+					rec, cursor->index(), offsets, thr);
 			} else {
 
 				err = row_ins_set_shared_rec_lock(
 					LOCK_REC_NOT_GAP,
 					btr_cur_get_block(cursor), rec,
-					cursor->index, offsets, thr);
+					cursor->index(), offsets, thr);
 			}
 
 			switch (err) {
@@ -2332,11 +2335,11 @@ row_ins_duplicate_error_in_clust(
 			}
 
 			if (row_ins_dupl_error_with_rec(
-				    rec, entry, cursor->index, offsets)) {
+				    rec, entry, cursor->index(), offsets)) {
 duplicate:
-				trx->error_info = cursor->index;
+				trx->error_info = cursor->index();
 				err = DB_DUPLICATE_KEY;
-				if (cursor->index->table->versioned()
+				if (cursor->index()->table->versioned()
 				    && entry->vers_history_row())
 				{
 					ulint trx_id_len;
@@ -2360,8 +2363,10 @@ duplicate:
 		rec = page_rec_get_next(btr_cur_get_rec(cursor));
 
 		if (rec && !page_rec_is_supremum(rec)) {
-			offsets = rec_get_offsets(rec, cursor->index, offsets,
-						  cursor->index->n_core_fields,
+			offsets = rec_get_offsets(rec, cursor->index(),
+						  offsets,
+						  cursor->index()
+						  ->n_core_fields,
 						  ULINT_UNDEFINED, &heap);
 
 			if (trx->duplicates) {
@@ -2374,13 +2379,13 @@ duplicate:
 				err = row_ins_set_exclusive_rec_lock(
 					LOCK_REC_NOT_GAP,
 					btr_cur_get_block(cursor),
-					rec, cursor->index, offsets, thr);
+					rec, cursor->index(), offsets, thr);
 			} else {
 
 				err = row_ins_set_shared_rec_lock(
 					LOCK_REC_NOT_GAP,
 					btr_cur_get_block(cursor),
-					rec, cursor->index, offsets, thr);
+					rec, cursor->index(), offsets, thr);
 			}
 
 			switch (err) {
@@ -2391,7 +2396,7 @@ duplicate:
 				/* fall through */
 			case DB_SUCCESS:
 				if (row_ins_dupl_error_with_rec(
-					    rec, entry, cursor->index,
+					    rec, entry, cursor->index(),
 					    offsets)) {
 					goto duplicate;
 				}
@@ -2433,7 +2438,7 @@ row_ins_must_modify_rec(
 	and a secondary index node pointer contains all index fields. */
 
 	return(cursor->low_match
-	       >= dict_index_get_n_unique_in_tree(cursor->index)
+	       >= dict_index_get_n_unique_in_tree(cursor->index())
 	       && !page_rec_is_infimum(btr_cur_get_rec(cursor)));
 }
 
@@ -2462,6 +2467,7 @@ row_ins_index_entry_big_rec(
 	btr_pcur_t	pcur;
 	rec_t*		rec;
 
+	pcur.btr_cur.page_cur.index = index;
 	ut_ad(index->is_primary());
 
 	DEBUG_SYNC_C_IF_THD(thd, "before_row_ins_extern_latch");
@@ -2473,8 +2479,8 @@ row_ins_index_entry_big_rec(
 		index->set_modified(mtr);
 	}
 
-	dberr_t error = btr_pcur_open(index, entry, PAGE_CUR_LE,
-				      BTR_MODIFY_TREE, &pcur, &mtr);
+	dberr_t error = btr_pcur_open(entry, PAGE_CUR_LE,
+				      BTR_MODIFY_TREE, &pcur, 0, &mtr);
 	if (error != DB_SUCCESS) {
 		return error;
 	}
@@ -2600,8 +2606,8 @@ row_ins_clust_index_entry_low(
 	/* Note that we use PAGE_CUR_LE as the search mode, because then
 	the function will return in both low_match and up_match of the
 	cursor sensible values */
- 	err = btr_pcur_open_low(index, 0, entry, PAGE_CUR_LE, mode, &pcur,
-				auto_inc, &mtr);
+	pcur.btr_cur.page_cur.index = index;
+	err = btr_pcur_open(entry, PAGE_CUR_LE, mode, &pcur, auto_inc, &mtr);
 	if (err != DB_SUCCESS) {
 		index->table->file_unreadable = true;
 commit_exit:
@@ -2712,7 +2718,7 @@ skip_bulk_insert:
 				/* fall through */
 			case DB_SUCCESS_LOCKED_REC:
 			case DB_DUPLICATE_KEY:
-				trx->error_info = cursor->index;
+				trx->error_info = cursor->index();
 			}
 		} else {
 			/* Note that the following may return also
@@ -2861,6 +2867,7 @@ row_ins_sec_index_entry_low(
 
 	cursor.thr = thr;
 	cursor.rtr_info = NULL;
+	cursor.page_cur.index = index;
 	ut_ad(thr_get_trx(thr)->id != 0);
 
 	mtr.start();
@@ -2878,15 +2885,13 @@ row_ins_sec_index_entry_low(
 	the function will return in both low_match and up_match of the
 	cursor sensible values */
 
-	if (dict_index_is_spatial(index)) {
-		cursor.index = index;
+	if (index->is_spatial()) {
 		rtr_init_rtr_info(&rtr_info, false, &cursor, index, false);
 		rtr_info_update_btr(&cursor, &rtr_info);
 
-		err = btr_cur_search_to_nth_level(
-			index, 0, entry, PAGE_CUR_RTREE_INSERT,
-			search_mode,
-			&cursor, &mtr);
+		err = btr_cur_search_to_nth_level(0, entry,
+						  PAGE_CUR_RTREE_INSERT,
+						  search_mode, &cursor, &mtr);
 
 		if (err == DB_SUCCESS && search_mode == BTR_MODIFY_LEAF
 		    && rtr_info.mbr_adj) {
@@ -2903,9 +2908,8 @@ row_ins_sec_index_entry_low(
 				index->set_modified(mtr);
 			}
 			err = btr_cur_search_to_nth_level(
-				index, 0, entry, PAGE_CUR_RTREE_INSERT,
-				search_mode,
-				&cursor, &mtr);
+				0, entry, PAGE_CUR_RTREE_INSERT,
+				search_mode, &cursor, &mtr);
 		}
 
 		DBUG_EXECUTE_IF(
@@ -2921,10 +2925,8 @@ row_ins_sec_index_entry_low(
 				   : BTR_INSERT));
 		}
 
-		err = btr_cur_search_to_nth_level(
-			index, 0, entry, PAGE_CUR_LE,
-			search_mode,
-			&cursor, &mtr);
+		err = btr_cur_search_to_nth_level(0, entry, PAGE_CUR_LE,
+						  search_mode, &cursor, &mtr);
 	}
 
 	if (err != DB_SUCCESS) {
@@ -3001,7 +3003,7 @@ row_ins_sec_index_entry_low(
 		transaction. Let us now reposition the cursor and
 		continue the insertion (bypassing the change buffer). */
 		err = btr_cur_search_to_nth_level(
-			index, 0, entry, PAGE_CUR_LE,
+			0, entry, PAGE_CUR_LE,
 			btr_latch_mode(search_mode
 				       & ~(BTR_INSERT
 					   | BTR_IGNORE_SEC_UNIQUE)),
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index 1b56d18cd3d..0689f9ca029 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -1658,7 +1658,7 @@ row_log_table_apply_delete_low(
 	dberr_t		error;
 	row_ext_t*	ext;
 	dtuple_t*	row;
-	dict_index_t*	index	= btr_pcur_get_btr_cur(pcur)->index;
+	dict_index_t*	index	= pcur->index();
 
 	ut_ad(dict_index_is_clust(index));
 
@@ -1695,8 +1695,9 @@ err_exit:
 			row, ext, index, heap);
 		mtr->start();
 		index->set_modified(*mtr);
-		error = btr_pcur_open(index, entry, PAGE_CUR_LE,
-				      BTR_PURGE_TREE, pcur, mtr);
+		pcur->btr_cur.page_cur.index = index;
+		error = btr_pcur_open(entry, PAGE_CUR_LE,
+				      BTR_PURGE_TREE, pcur, 0, mtr);
 		if (error) {
 			goto err_exit;
 		}
@@ -1760,6 +1761,7 @@ row_log_table_apply_delete(
 	btr_pcur_t	pcur;
 	rec_offs*	offsets;
 
+	pcur.btr_cur.page_cur.index = index;
 	ut_ad(rec_offs_n_fields(moffsets) == index->first_user_field());
 	ut_ad(!rec_offs_any_extern(moffsets));
 
@@ -1778,8 +1780,8 @@ row_log_table_apply_delete(
 
 	mtr_start(&mtr);
 	index->set_modified(mtr);
-	dberr_t err = btr_pcur_open(index, old_pk, PAGE_CUR_LE,
-				    BTR_PURGE_TREE, &pcur, &mtr);
+	dberr_t err = btr_pcur_open(old_pk, PAGE_CUR_LE,
+				    BTR_PURGE_TREE, &pcur, 0, &mtr);
 	if (err != DB_SUCCESS) {
 		goto all_done;
 	}
@@ -1891,6 +1893,8 @@ row_log_table_apply_update(
 	dberr_t		error;
 	ulint		n_index = 0;
 
+	pcur.btr_cur.page_cur.index = index;
+
 	ut_ad(dtuple_get_n_fields_cmp(old_pk)
 	      == dict_index_get_n_unique(index));
 	ut_ad(dtuple_get_n_fields(old_pk) - (log->same_pk ? 0 : 2)
@@ -1913,8 +1917,8 @@ row_log_table_apply_update(
 
 	mtr.start();
 	index->set_modified(mtr);
-	error = btr_pcur_open(index, old_pk, PAGE_CUR_LE,
-			      BTR_MODIFY_TREE, &pcur, &mtr);
+	error = btr_pcur_open(old_pk, PAGE_CUR_LE,
+			      BTR_MODIFY_TREE, &pcur, 0, &mtr);
 	if (error != DB_SUCCESS) {
 func_exit:
 		mtr.commit();
@@ -2059,7 +2063,7 @@ func_exit_committed:
 
 	for (n_index += index->type != DICT_CLUSTERED;
 	     (index = dict_table_get_next_index(index)); n_index++) {
-		if (index->type & DICT_FTS) {
+		if (!index->is_btree()) {
 			continue;
 		}
 
@@ -2087,9 +2091,10 @@ func_exit_committed:
 
 		mtr.start();
 		index->set_modified(mtr);
+		pcur.btr_cur.page_cur.index = index;
 
 		if (ROW_FOUND != row_search_index_entry(
-			    index, entry, BTR_MODIFY_TREE, &pcur, &mtr)) {
+			    entry, BTR_MODIFY_TREE, &pcur, &mtr)) {
 			ut_ad(0);
 			error = DB_CORRUPTION;
 			break;
@@ -3069,13 +3074,14 @@ row_log_apply_op_low(
 
 	mtr_start(&mtr);
 	index->set_modified(mtr);
+	cursor.page_cur.index = index;
 
 	/* We perform the pessimistic variant of the operations if we
 	already hold index->lock exclusively. First, search the
 	record. The operation may already have been performed,
 	depending on when the row in the clustered index was
 	scanned. */
-	*error = btr_cur_search_to_nth_level(index, 0, entry, PAGE_CUR_LE,
+	*error = btr_cur_search_to_nth_level(0, entry, PAGE_CUR_LE,
 					     has_index_lock
 					     ? BTR_MODIFY_TREE
 					     : BTR_MODIFY_LEAF,
@@ -3130,7 +3136,7 @@ row_log_apply_op_low(
 				mtr_start(&mtr);
 				index->set_modified(mtr);
 				*error = btr_cur_search_to_nth_level(
-					index, 0, entry, PAGE_CUR_LE,
+					0, entry, PAGE_CUR_LE,
 					BTR_MODIFY_TREE, &cursor, &mtr);
 				if (UNIV_UNLIKELY(*error != DB_SUCCESS)) {
 					goto func_exit;
@@ -3234,7 +3240,7 @@ insert_the_rec:
 				mtr_start(&mtr);
 				index->set_modified(mtr);
 				*error = btr_cur_search_to_nth_level(
-					index, 0, entry, PAGE_CUR_LE,
+					0, entry, PAGE_CUR_LE,
 					BTR_MODIFY_TREE, &cursor, &mtr);
 				if (*error != DB_SUCCESS) {
 					break;
@@ -3949,8 +3955,8 @@ void UndorecApplier::log_insert(const dtuple_t &tuple,
     }
 
     bool success= true;
-    dict_index_t *index= dict_table_get_next_index(clust_index);
-    while (index)
+    for (dict_index_t *index= clust_index;
+         (index= dict_table_get_next_index(index)) != nullptr; )
     {
       index->lock.s_lock(SRW_LOCK_CALL);
       if (index->online_log &&
@@ -3969,7 +3975,6 @@ void UndorecApplier::log_insert(const dtuple_t &tuple,
         row_log_mark_other_online_index_abort(index->table);
         return;
       }
-      index= dict_table_get_next_index(index);
     }
   }
 }
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 56b5ab2f883..dba3c0a144a 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -145,13 +145,13 @@ public:
 			mtr.start();
 			index->set_modified(mtr);
 
-			ins_cur.index = index;
+			ins_cur.page_cur.index = index;
 			rtr_init_rtr_info(&rtr_info, false, &ins_cur, index,
 					  false);
 			rtr_info_update_btr(&ins_cur, &rtr_info);
 
 			error = btr_cur_search_to_nth_level(
-				index, 0, dtuple, PAGE_CUR_RTREE_INSERT,
+				0, dtuple, PAGE_CUR_RTREE_INSERT,
 				BTR_MODIFY_LEAF, &ins_cur, &mtr);
 
 			/* It need to update MBR in parent entry,
@@ -165,7 +165,7 @@ public:
 				mtr.start();
 				index->set_modified(mtr);
 				error = btr_cur_search_to_nth_level(
-					index, 0, dtuple,
+					0, dtuple,
 					PAGE_CUR_RTREE_INSERT,
 					BTR_MODIFY_TREE, &ins_cur, &mtr);
 			}
@@ -190,7 +190,7 @@ public:
 
 				rtr_info_update_btr(&ins_cur, &rtr_info);
 				error = btr_cur_search_to_nth_level(
-					index, 0, dtuple,
+					0, dtuple,
 					PAGE_CUR_RTREE_INSERT,
 					BTR_MODIFY_TREE,
 					&ins_cur, &mtr);
@@ -2061,13 +2061,15 @@ end_of_index:
 				if (!block) {
 					goto err_exit;
 				}
-				btr_leaf_page_release(page_cur_get_block(cur),
-						      BTR_SEARCH_LEAF, &mtr);
+
 				page_cur_set_before_first(block, cur);
 				if (!page_cur_move_to_next(cur)
 				    || page_cur_is_after_last(cur)) {
 					goto corrupted_rec;
 				}
+
+				const auto s = mtr.get_savepoint();
+				mtr.rollback_to_savepoint(s - 2, s - 1);
 			}
 		} else {
 			mem_heap_empty(row_heap);
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index de469c5b088..6ac0d26bb2a 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1456,11 +1456,8 @@ row_create_update_node_for_mysql(
 
 	node->in_mysql_interface = true;
 	node->is_delete = NO_DELETE;
-	node->searched_update = FALSE;
-	node->select = NULL;
-	node->pcur = btr_pcur_create_for_mysql();
-
-	DBUG_PRINT("info", ("node: %p, pcur: %p", node, node->pcur));
+	node->pcur = new (mem_heap_alloc(heap, sizeof(btr_pcur_t)))
+		btr_pcur_t();
 
 	node->table = table;
 
@@ -1472,10 +1469,6 @@ row_create_update_node_for_mysql(
 	UT_LIST_INIT(node->columns, &sym_node_t::col_var_list);
 
 	node->has_clust_rec_x_lock = TRUE;
-	node->cmpl_info = 0;
-
-	node->table_sym = NULL;
-	node->col_assign_list = NULL;
 
 	DBUG_RETURN(node);
 }
@@ -1650,8 +1643,7 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 	clust_index = dict_table_get_first_index(table);
 
 	btr_pcur_copy_stored_position(node->pcur,
-				      prebuilt->pcur->btr_cur.index
-				      == clust_index
+				      prebuilt->pcur->index() == clust_index
 				      ? prebuilt->pcur
 				      : prebuilt->clust_pcur);
 
@@ -1804,7 +1796,7 @@ row_unlock_for_mysql(
 		}
 
 		rec = btr_pcur_get_rec(pcur);
-		index = btr_pcur_get_btr_cur(pcur)->index;
+		index = pcur->index();
 
 		/* If the record has been modified by this
 		transaction, do not unlock it. */
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 3c3beeb0eda..e216d9faa3b 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -347,8 +347,9 @@ row_purge_remove_sec_if_poss_tree(
 	log_free_check();
 	mtr.start();
 	index->set_modified(mtr);
+	pcur.btr_cur.page_cur.index = index;
 
-	search_result = row_search_index_entry(index, entry, BTR_PURGE_TREE,
+	search_result = row_search_index_entry(entry, BTR_PURGE_TREE,
 					       &pcur, &mtr);
 
 	switch (search_result) {
@@ -452,6 +453,7 @@ row_purge_remove_sec_if_poss_leaf(
 	virtual index. */
 	mode = (index->type & (DICT_SPATIAL | DICT_VIRTUAL))
 		? BTR_MODIFY_LEAF : BTR_PURGE_LEAF;
+	pcur.btr_cur.page_cur.index = index;
 
 	/* Set the purge node for the call to row_purge_poss_sec(). */
 	pcur.btr_cur.purge_node = node;
@@ -459,7 +461,7 @@ row_purge_remove_sec_if_poss_leaf(
 		pcur.btr_cur.thr = NULL;
 		index->lock.u_lock(SRW_LOCK_CALL);
 		search_result = row_search_index_entry(
-			index, entry, mode, &pcur, &mtr);
+			entry, mode, &pcur, &mtr);
 		index->lock.u_unlock();
 	} else {
 		/* Set the query thread, so that ibuf_insert_low() will be
@@ -467,7 +469,7 @@ row_purge_remove_sec_if_poss_leaf(
 		pcur.btr_cur.thr = static_cast<que_thr_t*>(
 			que_node_get_parent(node));
 		search_result = row_search_index_entry(
-			index, entry, mode, &pcur, &mtr);
+			entry, mode, &pcur, &mtr);
 	}
 
 	switch (search_result) {
@@ -1344,7 +1346,7 @@ purge_node_t::validate_pcur()
 		return(true);
 	}
 
-	dict_index_t*	clust_index = pcur.btr_cur.index;
+	dict_index_t* clust_index = pcur.index();
 
 	rec_offs* offsets = rec_get_offsets(
 		pcur.old_rec, clust_index, NULL, pcur.old_n_core_fields,
diff --git a/storage/innobase/row/row0row.cc b/storage/innobase/row/row0row.cc
index 62a41be115e..599033353c5 100644
--- a/storage/innobase/row/row0row.cc
+++ b/storage/innobase/row/row0row.cc
@@ -1196,11 +1196,12 @@ row_search_on_row_ref(
 	ut_ad(dtuple_check_typed(ref));
 
 	dict_index_t *index = dict_table_get_first_index(table);
+	btr_pcur_init(pcur);
+	pcur->btr_cur.page_cur.index = index;
 
 	if (UNIV_UNLIKELY(ref->info_bits != 0)) {
 		ut_ad(ref->is_metadata());
 		ut_ad(ref->n_fields <= index->n_uniq);
-		btr_pcur_init(pcur);
 		if (pcur->open_leaf(true, index, mode, mtr) != DB_SUCCESS
 		    || !btr_pcur_move_to_next_user_rec(pcur, mtr)) {
 			return false;
@@ -1215,7 +1216,7 @@ row_search_on_row_ref(
 			& REC_INFO_MIN_REC_FLAG;
 	} else {
 		ut_a(ref->n_fields == index->n_uniq);
-		if (btr_pcur_open(index, ref, PAGE_CUR_LE, mode, pcur, mtr)
+		if (btr_pcur_open(ref, PAGE_CUR_LE, mode, pcur, 0, mtr)
 		    != DB_SUCCESS) {
 			return false;
 		}
@@ -1265,7 +1266,6 @@ Searches an index record.
 enum row_search_result
 row_search_index_entry(
 /*===================*/
-	dict_index_t*	index,	/*!< in: index */
 	const dtuple_t*	entry,	/*!< in: index entry */
 	btr_latch_mode	mode,	/*!< in: BTR_MODIFY_LEAF, ... */
 	btr_pcur_t*	pcur,	/*!< in/out: persistent cursor, which must
@@ -1278,12 +1278,12 @@ row_search_index_entry(
 
 	ut_ad(dtuple_check_typed(entry));
 
-	if (index->is_spatial()) {
-		if (rtr_pcur_open(index, entry, mode, pcur, mtr)) {
+	if (pcur->index()->is_spatial()) {
+		if (rtr_pcur_open(pcur->index(), entry, mode, pcur, mtr)) {
 			return ROW_NOT_FOUND;
 		}
 	} else {
-		if (btr_pcur_open(index, entry, PAGE_CUR_LE, mode, pcur, mtr)
+		if (btr_pcur_open(entry, PAGE_CUR_LE, mode, pcur, 0, mtr)
 		    != DB_SUCCESS) {
 			return ROW_NOT_FOUND;
 		}
@@ -1292,7 +1292,7 @@ row_search_index_entry(
 	switch (btr_pcur_get_btr_cur(pcur)->flag) {
 	case BTR_CUR_DELETE_REF:
 		ut_ad(!(~mode & BTR_DELETE));
-		ut_ad(!index->is_spatial());
+		ut_ad(!pcur->index()->is_spatial());
 		return(ROW_NOT_DELETED_REF);
 
 	case BTR_CUR_DEL_MARK_IBUF:
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 615f767e7d2..41c9f4456e4 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -1009,17 +1009,16 @@ row_sel_get_clust_rec(
 
 	*out_rec = NULL;
 
-	offsets = rec_get_offsets(rec,
-				  btr_pcur_get_btr_cur(&plan->pcur)->index,
-				  offsets,
-				  btr_pcur_get_btr_cur(&plan->pcur)->index
-				  ->n_core_fields, ULINT_UNDEFINED, &heap);
+	offsets = rec_get_offsets(rec, plan->pcur.index(), offsets,
+				  plan->pcur.index()->n_core_fields,
+				  ULINT_UNDEFINED, &heap);
 
 	row_build_row_ref_fast(plan->clust_ref, plan->clust_map, rec, offsets);
 
 	index = dict_table_get_first_index(plan->table);
 	plan->clust_pcur.old_rec = nullptr;
-	dberr_t err = btr_pcur_open_with_no_init(index, plan->clust_ref,
+	plan->clust_pcur.btr_cur.page_cur.index = index;
+	dberr_t err = btr_pcur_open_with_no_init(plan->clust_ref,
 						 PAGE_CUR_LE, BTR_SEARCH_LEAF,
 						 &plan->clust_pcur, mtr);
 	if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
@@ -1412,6 +1411,7 @@ row_sel_open_pcur(
 	}
 
 	plan->pcur.old_rec = nullptr;
+	plan->pcur.btr_cur.page_cur.index = index;
 
 	dberr_t err;
 
@@ -1432,7 +1432,7 @@ row_sel_open_pcur(
 					 que_node_get_val(exp));
 		}
 
-		err = btr_pcur_open_with_no_init(index, plan->tuple,
+		err = btr_pcur_open_with_no_init(plan->tuple,
 						 plan->mode, BTR_SEARCH_LEAF,
 						 &plan->pcur, mtr);
 	} else {
@@ -3383,9 +3383,9 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 				   sec_index, *offsets);
 
 	clust_index = dict_table_get_first_index(sec_index->table);
+	prebuilt->clust_pcur->btr_cur.page_cur.index = clust_index;
 
-	dberr_t err = btr_pcur_open_with_no_init(clust_index,
-						 prebuilt->clust_ref,
+	dberr_t err = btr_pcur_open_with_no_init(prebuilt->clust_ref,
 						 PAGE_CUR_LE, BTR_SEARCH_LEAF,
 						 prebuilt->clust_pcur, mtr);
 	if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
@@ -3450,9 +3450,10 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 				rec, sec_index, true,
 				sec_index->n_fields, heap);
 			page_cur_t     page_cursor;
+			page_cursor.block = block;
+			page_cursor.index = sec_index;
 			ulint up_match = 0, low_match = 0;
-			ut_ad(!page_cur_search_with_match(block, sec_index,
-							  tuple, PAGE_CUR_LE,
+			ut_ad(!page_cur_search_with_match(tuple, PAGE_CUR_LE,
 							  &up_match,
 							  &low_match,
 							  &page_cursor,
@@ -3672,7 +3673,7 @@ static bool sel_restore_position_for_mysql(bool *same_user_rec,
 next:
 			if (btr_pcur_move_to_next(pcur, mtr)
 			    && rec_is_metadata(btr_pcur_get_rec(pcur),
-					       *pcur->btr_cur.index)) {
+					       *pcur->index())) {
 				btr_pcur_move_to_next(pcur, mtr);
 			}
 
@@ -3688,7 +3689,7 @@ next:
 prev:
 		if (btr_pcur_is_on_user_rec(pcur) && !moves_up
 		    && !rec_is_metadata(btr_pcur_get_rec(pcur),
-					*pcur->btr_cur.index)) {
+					*pcur->index())) {
 			if (!btr_pcur_move_to_prev(pcur, mtr)) {
 				return true;
 			}
@@ -3970,7 +3971,7 @@ row_sel_try_search_shortcut_for_mysql(
 	ut_ad(trx->read_view.is_open());
 	pcur->old_rec = nullptr;
 
-	if (btr_pcur_open_with_no_init(index, search_tuple, PAGE_CUR_GE,
+	if (btr_pcur_open_with_no_init(search_tuple, PAGE_CUR_GE,
 				       BTR_SEARCH_LEAF, pcur, mtr)
 	    != DB_SUCCESS) {
 		return SEL_RETRY;
@@ -4400,6 +4401,8 @@ row_search_mvcc(
 		DBUG_RETURN(DB_CORRUPTION);
 	}
 
+	pcur->btr_cur.page_cur.index = index;
+
 	/* We need to get the virtual column values stored in secondary
 	index key, if this is covered index scan or virtual key read is
 	requested. */
@@ -4791,7 +4794,7 @@ wait_table_again:
 			}
 		}
 
-		err = btr_pcur_open_with_no_init(index, search_tuple, mode,
+		err = btr_pcur_open_with_no_init(search_tuple, mode,
 						 BTR_SEARCH_LEAF, pcur, &mtr);
 
 		if (err != DB_SUCCESS) {
@@ -6212,6 +6215,7 @@ dberr_t row_check_index(row_prebuilt_t *prebuilt, ulint *n_rows)
   mtr.start();
 
   dict_index_t *clust_index= dict_table_get_first_index(prebuilt->table);
+  prebuilt->clust_pcur->btr_cur.page_cur.index = clust_index;
   dberr_t err= prebuilt->pcur->open_leaf(true, index, BTR_SEARCH_LEAF, &mtr);
   if (UNIV_UNLIKELY(err != DB_SUCCESS))
   {
@@ -6387,7 +6391,7 @@ rec_loop:
       const auto savepoint= mtr.get_savepoint();
 
       row_build_row_ref_in_tuple(prebuilt->clust_ref, rec, index, offsets);
-      err= btr_pcur_open_with_no_init(clust_index, prebuilt->clust_ref,
+      err= btr_pcur_open_with_no_init(prebuilt->clust_ref,
                                       PAGE_CUR_LE, BTR_SEARCH_LEAF,
                                       prebuilt->clust_pcur, &mtr);
       if (err != DB_SUCCESS)
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index eae89a36a6d..6567019a33d 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -68,7 +68,7 @@ row_undo_ins_remove_clust_rec(
 	dberr_t		err;
 	ulint		n_tries	= 0;
 	mtr_t		mtr;
-	dict_index_t*	index	= node->pcur.btr_cur.index;
+	dict_index_t*	index	= node->pcur.index();
 	table_id_t table_id = 0;
 	const bool dict_locked = node->trx->dict_operation_lock_mode;
 restart:
@@ -265,6 +265,7 @@ row_undo_ins_remove_sec_low(
 	mtr_t			mtr;
 	const bool		modify_leaf = mode == BTR_MODIFY_LEAF;
 
+	pcur.btr_cur.page_cur.index = index;
 	row_mtr_start(&mtr, index, !modify_leaf);
 
 	if (modify_leaf) {
@@ -284,7 +285,7 @@ row_undo_ins_remove_sec_low(
 		btr_pcur_get_btr_cur(&pcur)->thr = thr;
 	}
 
-	switch (row_search_index_entry(index, entry, mode, &pcur, &mtr)) {
+	switch (row_search_index_entry(entry, mode, &pcur, &mtr)) {
 	case ROW_BUFFERED:
 	case ROW_NOT_DELETED_REF:
 		/* These are invalid outcomes, because the mode passed
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 01b639db085..2d04dca4003 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -224,14 +224,14 @@ static bool row_undo_mod_must_purge(const undo_node_t &node)
   ut_ad(!node.table->is_temporary());
 
   const btr_cur_t &btr_cur= node.pcur.btr_cur;
-  ut_ad(btr_cur.index->is_primary());
+  ut_ad(btr_cur.index()->is_primary());
   DEBUG_SYNC_C("rollback_purge_clust");
 
   if (!purge_sys.is_purgeable(node.new_trx_id))
     return false;
 
   const rec_t *rec= btr_cur_get_rec(&btr_cur);
-  return trx_read_trx_id(rec + row_trx_id_offset(rec, btr_cur.index)) ==
+  return trx_read_trx_id(rec + row_trx_id_offset(rec, btr_cur.index())) ==
     node.new_trx_id;
 }
 
@@ -495,6 +495,7 @@ row_undo_mod_del_mark_or_remove_sec_low(
 
 	row_mtr_start(&mtr, index, !modify_leaf);
 
+	pcur.btr_cur.page_cur.index = index;
 	btr_cur = btr_pcur_get_btr_cur(&pcur);
 
 	if (index->is_spatial()) {
@@ -522,8 +523,7 @@ row_undo_mod_del_mark_or_remove_sec_low(
 		ut_ad(!dict_index_is_online_ddl(index));
 	}
 
-	search_result = row_search_index_entry(index, entry, mode,
-					       &pcur, &mtr);
+	search_result = row_search_index_entry(entry, mode, &pcur, &mtr);
 
 	switch (UNIV_EXPECT(search_result, ROW_FOUND)) {
 	case ROW_NOT_FOUND:
@@ -668,6 +668,7 @@ row_undo_mod_del_unmark_sec_and_undo_update(
 	row_search_result	search_result;
 	const auto		orig_mode = mode;
 
+	pcur.btr_cur.page_cur.index = index;
 	ut_ad(trx->id != 0);
 
 	if (dict_index_is_spatial(index)) {
@@ -685,8 +686,7 @@ try_again:
 
 	btr_cur->thr = thr;
 
-	search_result = row_search_index_entry(index, entry, mode,
-					       &pcur, &mtr);
+	search_result = row_search_index_entry(entry, mode, &pcur, &mtr);
 
 	switch (search_result) {
 		mem_heap_t*	heap;
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 36941798bc2..3b2fe849bcd 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -1899,9 +1899,9 @@ row_upd_sec_index_entry(
 	/* Set the query thread, so that ibuf_insert_low() will be
 	able to invoke thd_get_trx(). */
 	btr_pcur_get_btr_cur(&pcur)->thr = thr;
+	pcur.btr_cur.page_cur.index = index;
 
-	search_result = row_search_index_entry(index, entry, mode,
-					       &pcur, &mtr);
+	search_result = row_search_index_entry(entry, mode, &pcur, &mtr);
 
 	btr_cur = btr_pcur_get_btr_cur(&pcur);