1 files changed, 95 insertions, 338 deletions

diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 066e3d43d27..fe88fce58a2 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2021, MariaDB Corporation.
+Copyright (c) 2015, 2023, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -127,10 +127,6 @@ row_upd_changes_first_fields_binary(
 Checks if index currently is mentioned as a referenced index in a foreign
 key constraint.
 
-NOTE that since we do not hold dict_sys.latch when leaving the
-function, it may be that the referencing table has been dropped when
-we leave this function: this function is only for heuristic use!
-
 @return true if referenced */
 static
 bool
@@ -139,64 +135,44 @@ row_upd_index_is_referenced(
 	dict_index_t*	index,	/*!< in: index */
 	trx_t*		trx)	/*!< in: transaction */
 {
-	dict_table_t*	table		= index->table;
-
-	if (table->referenced_set.empty()) {
-		return false;
-	}
-
-	const bool froze_data_dict = !trx->dict_operation_lock_mode;
-	if (froze_data_dict) {
-		row_mysql_freeze_data_dictionary(trx);
-	}
-
-	dict_foreign_set::iterator	it
-		= std::find_if(table->referenced_set.begin(),
-			       table->referenced_set.end(),
-			       dict_foreign_with_index(index));
-
-	const bool is_referenced = (it != table->referenced_set.end());
-
-	if (froze_data_dict) {
-		row_mysql_unfreeze_data_dictionary(trx);
-	}
-
-	return is_referenced;
+  dict_table_t *table= index->table;
+  /* The pointers in table->referenced_set are safe to dereference
+  thanks to the SQL layer having acquired MDL on all (grand)parent tables. */
+  dict_foreign_set::iterator end= table->referenced_set.end();
+  return end != std::find_if(table->referenced_set.begin(), end,
+                             dict_foreign_with_index(index));
 }
 
 #ifdef WITH_WSREP
 static
-ibool
+bool
 wsrep_row_upd_index_is_foreign(
 /*========================*/
 	dict_index_t*	index,	/*!< in: index */
 	trx_t*		trx)	/*!< in: transaction */
 {
-	dict_table_t*	table		= index->table;
-	ibool		froze_data_dict	= FALSE;
-	ibool		is_referenced	= FALSE;
+  if (!trx->is_wsrep())
+    return false;
 
-	if (table->foreign_set.empty()) {
-		return(FALSE);
-	}
-
-	if (trx->dict_operation_lock_mode == 0) {
-		row_mysql_freeze_data_dictionary(trx);
-		froze_data_dict = TRUE;
-	}
+  dict_table_t *table= index->table;
 
-	dict_foreign_set::iterator	it
-		= std::find_if(table->foreign_set.begin(),
-			       table->foreign_set.end(),
-			       dict_foreign_with_foreign_index(index));
+  if (table->foreign_set.empty())
+    return false;
 
-	is_referenced = (it != table->foreign_set.end());
+  /* No MDL protects dereferencing the members of table->foreign_set. */
+  const bool no_lock= !trx->dict_operation_lock_mode;
+  if (no_lock)
+    dict_sys.freeze(SRW_LOCK_CALL);
 
-	if (froze_data_dict) {
-		row_mysql_unfreeze_data_dictionary(trx);
-	}
+  auto end= table->foreign_set.end();
+  const bool is_referenced= end !=
+    std::find_if(table->foreign_set.begin(), end,
+                 [index](const dict_foreign_t* f)
+                 {return f->foreign_index == index;});
+  if (no_lock)
+    dict_sys.unfreeze();
 
-	return(is_referenced);
+  return is_referenced;
 }
 #endif /* WITH_WSREP */
 
@@ -224,10 +200,8 @@ row_upd_check_references_constraints(
 	dict_foreign_t*	foreign;
 	mem_heap_t*	heap;
 	dtuple_t*	entry;
-	trx_t*		trx;
 	const rec_t*	rec;
 	dberr_t		err;
-	ibool		got_s_lock	= FALSE;
 
 	DBUG_ENTER("row_upd_check_references_constraints");
 
@@ -235,8 +209,6 @@ row_upd_check_references_constraints(
 		DBUG_RETURN(DB_SUCCESS);
 	}
 
-	trx = thr_get_trx(thr);
-
 	rec = btr_pcur_get_rec(pcur);
 	ut_ad(rec_offs_validate(rec, index, offsets));
 
@@ -250,12 +222,6 @@ row_upd_check_references_constraints(
 
 	mtr->start();
 
-	if (trx->dict_operation_lock_mode == 0) {
-		got_s_lock = TRUE;
-
-		row_mysql_freeze_data_dictionary(trx);
-	}
-
 	DEBUG_SYNC_C_IF_THD(thr_get_trx(thr)->mysql_thd,
 			    "foreign_constraint_check_for_insert");
 
@@ -275,34 +241,19 @@ row_upd_check_references_constraints(
 			|| row_upd_changes_first_fields_binary(
 				entry, index, node->update,
 				foreign->n_fields))) {
-			dict_table_t*	foreign_table = foreign->foreign_table;
-
-			dict_table_t*	ref_table = NULL;
-
-			if (foreign_table == NULL) {
+			dict_table_t*	ref_table = nullptr;
 
+			if (!foreign->foreign_table) {
 				ref_table = dict_table_open_on_name(
 					foreign->foreign_table_name_lookup,
-					FALSE, FALSE, DICT_ERR_IGNORE_NONE);
+					false, DICT_ERR_IGNORE_NONE);
 			}
 
-			if (foreign_table) {
-				foreign_table->inc_fk_checks();
-			}
-
-			/* NOTE that if the thread ends up waiting for a lock
-			we will release dict_sys.latch temporarily!
-			But the inc_fk_checks() protects foreign_table from
-			being dropped while the check is running. */
-
 			err = row_ins_check_foreign_constraint(
 				FALSE, foreign, table, entry, thr);
 
-			if (foreign_table) {
-				foreign_table->dec_fk_checks();
-			}
-			if (ref_table != NULL) {
-				dict_table_close(ref_table, FALSE, FALSE);
+			if (ref_table) {
+				dict_table_close(ref_table);
 			}
 
 			if (err != DB_SUCCESS) {
@@ -314,10 +265,6 @@ row_upd_check_references_constraints(
 	err = DB_SUCCESS;
 
 func_exit:
-	if (got_s_lock) {
-		row_mysql_unfreeze_data_dictionary(trx);
-	}
-
 	mem_heap_free(heap);
 
 	DEBUG_SYNC_C("foreign_constraint_check_for_update_done");
@@ -341,18 +288,13 @@ wsrep_row_upd_check_foreign_constraints(
 	dict_foreign_t*	foreign;
 	mem_heap_t*	heap;
 	dtuple_t*	entry;
-	trx_t*		trx;
 	const rec_t*	rec;
 	dberr_t		err;
-	ibool		got_s_lock	= FALSE;
-	ibool		opened     	= FALSE;
 
 	if (table->foreign_set.empty()) {
 		return(DB_SUCCESS);
 	}
 
-	trx = thr_get_trx(thr);
-
 	/* TODO: make native slave thread bail out here */
 
 	rec = btr_pcur_get_rec(pcur);
@@ -366,12 +308,6 @@ wsrep_row_upd_check_foreign_constraints(
 
 	mtr_start(mtr);
 
-	if (trx->dict_operation_lock_mode == 0) {
-		got_s_lock = TRUE;
-
-		row_mysql_freeze_data_dictionary(trx);
-	}
-
 	for (dict_foreign_set::iterator it = table->foreign_set.begin();
 	     it != table->foreign_set.end();
 	     ++it) {
@@ -388,27 +324,21 @@ wsrep_row_upd_check_foreign_constraints(
 				entry, index, node->update,
 				foreign->n_fields))) {
 
-			if (foreign->referenced_table == NULL) {
+			dict_table_t *opened = nullptr;
+
+			if (!foreign->referenced_table) {
 				foreign->referenced_table =
 					dict_table_open_on_name(
 					  foreign->referenced_table_name_lookup,
-					  FALSE, FALSE, DICT_ERR_IGNORE_NONE);
-				opened = (foreign->referenced_table) ? TRUE : FALSE;
+					  false, DICT_ERR_IGNORE_NONE);
+				opened = foreign->referenced_table;
 			}
 
-			/* NOTE that if the thread ends up waiting for a lock
-			we will release dict_sys.latch temporarily!
-			But the counter on the table protects 'foreign' from
-			being dropped while the check is running. */
-
 			err = row_ins_check_foreign_constraint(
 				TRUE, foreign, table, entry, thr);
 
-			if (foreign->referenced_table) {
-				if (opened == TRUE) {
-					dict_table_close(foreign->referenced_table, FALSE, FALSE);
-					opened = FALSE;
-				}
+			if (opened) {
+				dict_table_close(opened);
 			}
 
 			if (err != DB_SUCCESS) {
@@ -419,10 +349,6 @@ wsrep_row_upd_check_foreign_constraints(
 
 	err = DB_SUCCESS;
 func_exit:
-	if (got_s_lock) {
-		row_mysql_unfreeze_data_dictionary(trx);
-	}
-
 	mem_heap_free(heap);
 
 	return(err);
@@ -543,46 +469,6 @@ row_upd_changes_field_size_or_external(
 	return(FALSE);
 }
 
-/***********************************************************//**
-Returns true if row update contains disowned external fields.
-@return true if the update contains disowned external fields. */
-bool
-row_upd_changes_disowned_external(
-/*==============================*/
-	const upd_t*	update)	/*!< in: update vector */
-{
-	const upd_field_t*	upd_field;
-	const dfield_t*		new_val;
-	ulint			new_len;
-	ulint                   n_fields;
-	ulint			i;
-
-	n_fields = upd_get_n_fields(update);
-
-	for (i = 0; i < n_fields; i++) {
-		const byte*	field_ref;
-
-		upd_field = upd_get_nth_field(update, i);
-		new_val = &(upd_field->new_val);
-		new_len = dfield_get_len(new_val);
-
-		if (!dfield_is_ext(new_val)) {
-			continue;
-		}
-
-		ut_ad(new_len >= BTR_EXTERN_FIELD_REF_SIZE);
-
-		field_ref = static_cast<const byte*>(dfield_get_data(new_val))
-			    + new_len - BTR_EXTERN_FIELD_REF_SIZE;
-
-		if (field_ref[BTR_EXTERN_LEN] & BTR_EXTERN_OWNER_FLAG) {
-			return(true);
-		}
-	}
-
-	return(false);
-}
-
 /***************************************************************//**
 Builds an update vector from those fields which in a secondary index entry
 differ from a record that has the equal ordering fields. NOTE: we compare
@@ -1146,16 +1032,7 @@ row_upd_replace_vcol(
 		/* If there is no index on the column, do not bother for
 		value update */
 		if (!col->m_col.ord_part) {
-			dict_index_t*	clust_index
-				= dict_table_get_first_index(table);
-
-			/* Skip the column if there is no online alter
-			table in progress or it is not being indexed
-			in new table */
-			if (!dict_index_is_online_ddl(clust_index)
-			    || !row_log_col_is_indexed(clust_index, col_no)) {
-				continue;
-			}
+			continue;
 		}
 
 		dfield = dtuple_get_nth_v_field(row, col_no);
@@ -1345,9 +1222,6 @@ row_upd_changes_ord_field_binary_func(
 	ulint			i;
 	const dict_index_t*	clust_index;
 
-	ut_ad(thr);
-	ut_ad(thr->graph);
-	ut_ad(thr->graph->trx);
 	ut_ad(!index->table->skip_alter_undo);
 
 	n_unique = dict_index_get_n_unique(index);
@@ -1547,9 +1421,11 @@ row_upd_changes_ord_field_binary_func(
 					trx_rollback_recovered()
 					when the server had crashed before
 					storing the field. */
-					ut_ad(thr->graph->trx->is_recovered);
-					ut_ad(thr->graph->trx
-					      == trx_roll_crash_recv_trx);
+					ut_ad(!thr
+					      || thr->graph->trx->is_recovered);
+					ut_ad(!thr
+					      || thr->graph->trx
+					         == trx_roll_crash_recv_trx);
 					return(TRUE);
 				}
 
@@ -1956,25 +1832,28 @@ row_upd_sec_index_entry(
 	que_thr_t*	thr)	/*!< in: query thread */
 {
 	mtr_t			mtr;
-	const rec_t*		rec;
 	btr_pcur_t		pcur;
 	mem_heap_t*		heap;
 	dtuple_t*		entry;
 	dict_index_t*		index;
-	btr_cur_t*		btr_cur;
 	dberr_t			err	= DB_SUCCESS;
 	trx_t*			trx	= thr_get_trx(thr);
-	ulint			mode;
+	btr_latch_mode		mode;
 	ulint			flags;
 	enum row_search_result	search_result;
 
 	ut_ad(trx->id != 0);
 
 	index = node->index;
+	ut_ad(index->is_committed());
+
+	/* For secondary indexes, index->online_status==ONLINE_INDEX_COMPLETE
+	if index->is_committed(). */
+	ut_ad(!dict_index_is_online_ddl(index));
 
 	const bool referenced = row_upd_index_is_referenced(index, trx);
 #ifdef WITH_WSREP
-	bool foreign = wsrep_row_upd_index_is_foreign(index, trx);
+	const bool foreign = wsrep_row_upd_index_is_foreign(index, trx);
 #endif /* WITH_WSREP */
 
 	heap = mem_heap_create(1024);
@@ -1989,6 +1868,7 @@ row_upd_sec_index_entry(
 			    "before_row_upd_sec_index_entry");
 
 	mtr.start();
+	mode = BTR_MODIFY_LEAF;
 
 	switch (index->table->space_id) {
 	case SRV_TMP_SPACE_ID:
@@ -2000,83 +1880,37 @@ row_upd_sec_index_entry(
 		/* fall through */
 	case IBUF_SPACE_ID:
 		flags = index->table->no_rollback() ? BTR_NO_ROLLBACK : 0;
+		/* We can only buffer delete-mark operations if there
+		are no foreign key constraints referring to the index. */
+		if (!referenced) {
+			mode = BTR_DELETE_MARK_LEAF;
+		}
 		break;
 	}
 
-	bool uncommitted = !index->is_committed();
-
-	if (uncommitted) {
-		/* The index->online_status may change if the index is
-		or was being created online, but not committed yet. It
-		is protected by index->lock. */
-
-		mtr_s_lock_index(index, &mtr);
+	/* Set the query thread, so that ibuf_insert_low() will be
+	able to invoke thd_get_trx(). */
+	pcur.btr_cur.thr = thr;
+	pcur.btr_cur.page_cur.index = index;
 
-		switch (dict_index_get_online_status(index)) {
-		case ONLINE_INDEX_COMPLETE:
-			/* This is a normal index. Do not log anything.
-			Perform the update on the index tree directly. */
-			break;
-		case ONLINE_INDEX_CREATION:
-			/* Log a DELETE and optionally INSERT. */
-			row_log_online_op(index, entry, 0);
-
-			if (!node->is_delete) {
-				mem_heap_empty(heap);
-				entry = row_build_index_entry(
-					node->upd_row, node->upd_ext,
-					index, heap);
-				ut_a(entry);
-				row_log_online_op(index, entry, trx->id);
-			}
-			/* fall through */
-		case ONLINE_INDEX_ABORTED:
-		case ONLINE_INDEX_ABORTED_DROPPED:
-			mtr_commit(&mtr);
-			goto func_exit;
+	if (index->is_spatial()) {
+		mode = btr_latch_mode(BTR_MODIFY_LEAF | BTR_RTREE_DELETE_MARK);
+		if (UNIV_LIKELY(!rtr_search(entry, mode, &pcur, &mtr))) {
+			goto found;
 		}
 
-		/* We can only buffer delete-mark operations if there
-		are no foreign key constraints referring to the index.
-		Change buffering is disabled for temporary tables and
-		spatial index. */
-		mode = (referenced || index->table->is_temporary()
-			|| dict_index_is_spatial(index))
-			? BTR_MODIFY_LEAF_ALREADY_S_LATCHED
-			: BTR_DELETE_MARK_LEAF_ALREADY_S_LATCHED;
-	} else {
-		/* For secondary indexes,
-		index->online_status==ONLINE_INDEX_COMPLETE if
-		index->is_committed(). */
-		ut_ad(!dict_index_is_online_ddl(index));
-
-		/* We can only buffer delete-mark operations if there
-		are no foreign key constraints referring to the index.
-		Change buffering is disabled for temporary tables and
-		spatial index. */
-		mode = (referenced || index->table->is_temporary()
-			|| dict_index_is_spatial(index))
-			? BTR_MODIFY_LEAF
-			: BTR_DELETE_MARK_LEAF;
-	}
+		if (pcur.btr_cur.rtr_info->fd_del) {
+			/* We found the record, but a delete marked */
+			goto close;
+		}
 
-	if (dict_index_is_spatial(index)) {
-		ut_ad(mode & BTR_MODIFY_LEAF);
-		mode |= BTR_RTREE_DELETE_MARK;
+		goto not_found;
 	}
 
-	/* 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;
-
-	search_result = row_search_index_entry(index, entry, mode,
-					       &pcur, &mtr);
-
-	btr_cur = btr_pcur_get_btr_cur(&pcur);
-
-	rec = btr_cur_get_rec(btr_cur);
+	search_result = row_search_index_entry(entry, mode, &pcur, &mtr);
 
 	switch (search_result) {
+	const rec_t* rec;
 	case ROW_NOT_DELETED_REF:	/* should only occur for BTR_DELETE */
 		ut_error;
 		break;
@@ -2085,24 +1919,8 @@ row_upd_sec_index_entry(
 		break;
 
 	case ROW_NOT_FOUND:
-		if (!index->is_committed()) {
-			/* When online CREATE INDEX copied the update
-			that we already made to the clustered index,
-			and completed the secondary index creation
-			before we got here, the old secondary index
-			record would not exist. The CREATE INDEX
-			should be waiting for a MySQL meta-data lock
-			upgrade at least until this UPDATE returns.
-			After that point, set_committed(true) would be
-			invoked by commit_inplace_alter_table(). */
-			break;
-		}
-
-		if (dict_index_is_spatial(index) && btr_cur->rtr_info->fd_del) {
-			/* We found the record, but a delete marked */
-			break;
-		}
-
+not_found:
+		rec = btr_pcur_get_rec(&pcur);
 		ib::error()
 			<< "Record in index " << index->name
 			<< " of table " << index->table->name
@@ -2116,7 +1934,9 @@ row_upd_sec_index_entry(
 #endif /* UNIV_DEBUG */
 		break;
 	case ROW_FOUND:
+found:
 		ut_ad(err == DB_SUCCESS);
+		rec = btr_pcur_get_rec(&pcur);
 
 		/* Delete mark the old index record; it can already be
 		delete marked if we return after a lock wait in
@@ -2125,14 +1945,14 @@ row_upd_sec_index_entry(
 			    rec, dict_table_is_comp(index->table))) {
 			err = lock_sec_rec_modify_check_and_lock(
 				flags,
-				btr_cur_get_block(btr_cur),
-				btr_cur_get_rec(btr_cur), index, thr, &mtr);
+				btr_pcur_get_block(&pcur),
+				btr_pcur_get_rec(&pcur), index, thr, &mtr);
 			if (err != DB_SUCCESS) {
 				break;
 			}
 
-			btr_rec_set_deleted<true>(btr_cur_get_block(btr_cur),
-						  btr_cur_get_rec(btr_cur),
+			btr_rec_set_deleted<true>(btr_pcur_get_block(&pcur),
+						  btr_pcur_get_rec(&pcur),
 						  &mtr);
 #ifdef WITH_WSREP
 			if (!referenced && foreign
@@ -2191,6 +2011,7 @@ row_upd_sec_index_entry(
 		}
 	}
 
+close:
 	btr_pcur_close(&pcur);
 	mtr_commit(&mtr);
 
@@ -2204,35 +2025,11 @@ row_upd_sec_index_entry(
 	DEBUG_SYNC_C_IF_THD(trx->mysql_thd,
 			    "before_row_upd_sec_new_index_entry");
 
-	uncommitted = !index->is_committed();
-	if (uncommitted) {
-		mtr.start();
-		/* The index->online_status may change if the index is
-		being rollbacked. It is protected by index->lock. */
-
-		mtr_s_lock_index(index, &mtr);
-
-		switch (dict_index_get_online_status(index)) {
-		case ONLINE_INDEX_COMPLETE:
-		case ONLINE_INDEX_CREATION:
-		       break;
-		case ONLINE_INDEX_ABORTED:
-		case ONLINE_INDEX_ABORTED_DROPPED:
-		       mtr_commit(&mtr);
-		       goto func_exit;
-		}
-
-	}
-
 	/* Build a new index entry */
 	entry = row_build_index_entry(node->upd_row, node->upd_ext,
 				      index, heap);
 	ut_a(entry);
 
-	if (uncommitted) {
-		mtr_commit(&mtr);
-	}
-
 	/* Insert new index entry */
 	err = row_ins_sec_index_entry(index, entry, thr, !node->is_delete);
 
@@ -2553,7 +2350,6 @@ row_upd_clust_rec(
 	btr_pcur_t*	pcur;
 	btr_cur_t*	btr_cur;
 	dberr_t		err;
-	const dtuple_t*	rebuilt_old_pk	= NULL;
 
 	ut_ad(dict_index_is_clust(index));
 	ut_ad(!thr_get_trx(thr)->in_rollback);
@@ -2567,11 +2363,6 @@ row_upd_clust_rec(
 				    dict_table_is_comp(index->table)));
 	ut_ad(rec_offs_validate(btr_cur_get_rec(btr_cur), index, offsets));
 
-	if (dict_index_is_online_ddl(index)) {
-		rebuilt_old_pk = row_log_table_get_pk(
-			btr_cur_get_rec(btr_cur), index, offsets, NULL, &heap);
-	}
-
 	/* Try optimistic updating of the record, keeping changes within
 	the page; we do not check locks because we assume the x-lock on the
 	record to update */
@@ -2589,7 +2380,7 @@ row_upd_clust_rec(
 	}
 
 	if (err == DB_SUCCESS) {
-		goto success;
+		goto func_exit;
 	}
 
 	if (buf_pool.running_out()) {
@@ -2618,7 +2409,7 @@ row_upd_clust_rec(
 	the same transaction do not modify the record in the meantime.
 	Therefore we can assert that the restoration of the cursor succeeds. */
 
-	ut_a(btr_pcur_restore_position(BTR_MODIFY_TREE, pcur, mtr) ==
+	ut_a(pcur->restore_position(BTR_MODIFY_TREE, mtr) ==
 	    btr_pcur_t::SAME_ALL);
 
 	ut_ad(!rec_get_deleted_flag(btr_pcur_get_rec(pcur),
@@ -2642,15 +2433,6 @@ row_upd_clust_rec(
 		DEBUG_SYNC_C("after_row_upd_extern");
 	}
 
-	if (err == DB_SUCCESS) {
-success:
-		if (dict_index_is_online_ddl(index)) {
-			row_log_table_update(
-				btr_cur_get_rec(btr_cur),
-				index, offsets, rebuilt_old_pk);
-		}
-	}
-
 func_exit:
 	if (heap) {
 		mem_heap_free(heap);
@@ -2776,6 +2558,10 @@ row_upd_clust_step(
 
 	index = dict_table_get_first_index(node->table);
 
+	if (index->is_corrupted()) {
+		return DB_TABLE_CORRUPT;
+	}
+
 	const bool referenced = row_upd_index_is_referenced(index, trx);
 #ifdef WITH_WSREP
 	const bool foreign = wsrep_row_upd_index_is_foreign(index, trx);
@@ -2810,57 +2596,30 @@ row_upd_clust_step(
 
 	ut_a(pcur->rel_pos == BTR_PCUR_ON);
 
-	ulint	mode;
+	btr_latch_mode mode;
 
 	DEBUG_SYNC_C_IF_THD(trx->mysql_thd, "innodb_row_upd_clust_step_enter");
 
 	if (dict_index_is_online_ddl(index)) {
 		ut_ad(node->table->id != DICT_INDEXES_ID);
-		mode = BTR_MODIFY_LEAF | BTR_ALREADY_S_LATCHED;
+		mode = BTR_MODIFY_LEAF_ALREADY_LATCHED;
 		mtr_s_lock_index(index, &mtr);
 	} else {
 		mode = BTR_MODIFY_LEAF;
 	}
 
-	if (btr_pcur_restore_position(mode, pcur, &mtr) !=
-	    btr_pcur_t::SAME_ALL) {
+	if (pcur->restore_position(mode, &mtr) != btr_pcur_t::SAME_ALL) {
 		err = DB_RECORD_NOT_FOUND;
 		goto exit_func;
 	}
 
-	/* If this is a row in SYS_INDEXES table of the data dictionary,
-	then we have to free the file segments of the index tree associated
-	with the index */
-
-	if (node->is_delete == PLAIN_DELETE
-	    && node->table->id == DICT_INDEXES_ID) {
-
-		ut_ad(!dict_index_is_online_ddl(index));
-
-		dict_drop_index_tree(pcur, trx, &mtr);
-
-		mtr.commit();
-
-		mtr.start();
-		index->set_modified(mtr);
-
-		if (btr_pcur_restore_position(BTR_MODIFY_LEAF, pcur, &mtr) !=
-		    btr_pcur_t::SAME_ALL) {
-			err = DB_ERROR;
-
-			mtr.commit();
-
-			return(err);
-		}
-	}
-
 	rec = btr_pcur_get_rec(pcur);
 	offsets = rec_get_offsets(rec, index, offsets_, index->n_core_fields,
 				  ULINT_UNDEFINED, &heap);
 
 	if (!flags && !node->has_clust_rec_x_lock) {
 		err = lock_clust_rec_modify_check_and_lock(
-			0, btr_pcur_get_block(pcur),
+			btr_pcur_get_block(pcur),
 			rec, index, offsets, thr);
 		if (err != DB_SUCCESS) {
 			goto exit_func;
@@ -2869,8 +2628,8 @@ row_upd_clust_step(
 
 	ut_ad(index->table->no_rollback() || index->table->is_temporary()
 	      || row_get_rec_trx_id(rec, index, offsets) == trx->id
-	      || lock_trx_has_expl_x_lock(trx, index->table,
-					  btr_pcur_get_block(pcur),
+	      || lock_trx_has_expl_x_lock(*trx, *index->table,
+					  btr_pcur_get_block(pcur)->page.id(),
 					  page_rec_get_heap_no(rec)));
 
 	if (node->is_delete == PLAIN_DELETE) {
@@ -3017,14 +2776,12 @@ row_upd(
 	DBUG_EXECUTE_IF("row_upd_skip_sec", node->index = NULL;);
 
 	do {
-		/* Skip corrupted index */
-		dict_table_skip_corrupt_index(node->index);
-
 		if (!node->index) {
 			break;
 		}
 
-		if (node->index->type != DICT_FTS) {
+		if (!(node->index->type & (DICT_FTS | DICT_CORRUPT))
+		    && node->index->is_committed()) {
 			err = row_upd_sec_step(node, thr);
 
 			if (err != DB_SUCCESS) {
@@ -3091,7 +2848,7 @@ row_upd_step(
 			/* It may be that the current session has not yet
 			started its transaction, or it has been committed: */
 
-			err = lock_table(0, node->table, LOCK_IX, thr);
+			err = lock_table(node->table, nullptr, LOCK_IX, thr);
 
 			if (err != DB_SUCCESS) {