MDEV-23484 Rollback unnecessarily acquires dict_operation_lock for every row

InnoDB transaction rollback includes an unnecessary work-around for
a data corruption bug that was fixed by me in MySQL 5.6.12
mysql/mysql-server@935ba09d52c1908bde273ad1940b5ab919d9763d
and ported to MariaDB 10.0.8 by
commit c291ddfdf774b50c34b9741c6e39c57bae8fd1dc
in 2013 and 2014, respectively.

By acquiring and releasing dict_operation_lock in shared mode,
row_undo() hopes to prevent the table from being dropped while
the undo log record is being rolled back. But, thanks to mentioned fix,
debug assertions (that we are adding) show that the rollback is
protected by transactional locks (table IX lock, in addition to
implicit or explicit exclusive locks on the records that had been modified).

Because row_drop_table_for_mysql() would invoke
row_add_table_to_background_drop_list() if any locks exist on the table,
the mere existence of locks (which is guaranteed during ROLLBACK) is
enough to protect the table from disappearing. Hence, acquiring and
releasing dict_operation_lock for every row that is being rolled back is
unnecessary.

row_undo(): Remove the unnecessary acquisition and release of
dict_operation_lock.

Note: row_add_table_to_background_drop_list() is mostly working around
bugs outside InnoDB:
MDEV-21175 (insufficient MDL protection of FOREIGN KEY operations)
MDEV-21602 (incorrect error handling of CREATE TABLE...SELECT).

3 files changed, 11 insertions, 58 deletions

diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 54fa244fca6..7e5aac1ba9f 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -81,6 +81,7 @@ row_undo_ins_remove_clust_rec(
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
 	} else {
 		mtr.set_named_space(index->space);
+		ut_ad(lock_table_has_locks(index->table));
 	}
 
 	/* This is similar to row_undo_mod_clust(). The DDL thread may
@@ -91,8 +92,7 @@ row_undo_ins_remove_clust_rec(
 
 	online = dict_index_is_online_ddl(index);
 	if (online) {
-		ut_ad(node->trx->dict_operation_lock_mode
-		      != RW_X_LATCH);
+		ut_ad(!node->trx->dict_operation_lock_mode);
 		ut_ad(node->table->id != DICT_INDEXES_ID);
 		mtr_s_lock(dict_index_get_lock(index), &mtr);
 	}
@@ -491,6 +491,9 @@ row_undo_ins(
 		return(DB_SUCCESS);
 	}
 
+	ut_ad(node->table->is_temporary()
+	      || lock_table_has_locks(node->table));
+
 	/* Iterate over all the indexes and undo the insert.*/
 
 	node->index = dict_table_get_first_index(node->table);
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 80d90f40379..b1cc994cdd8 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -264,10 +264,7 @@ row_undo_mod_clust(
 	bool		online;
 
 	ut_ad(thr_get_trx(thr) == node->trx);
-	ut_ad(node->trx->dict_operation_lock_mode);
 	ut_ad(node->trx->in_rollback);
-	ut_ad(rw_lock_own_flagged(&dict_operation_lock,
-				  RW_LOCK_FLAG_X | RW_LOCK_FLAG_S));
 
 	log_free_check();
 	pcur = &node->pcur;
@@ -278,11 +275,12 @@ row_undo_mod_clust(
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
 	} else {
 		mtr.set_named_space(index->space);
+		ut_ad(lock_table_has_locks(index->table));
 	}
 
 	online = dict_index_is_online_ddl(index);
 	if (online) {
-		ut_ad(node->trx->dict_operation_lock_mode != RW_X_LATCH);
+		ut_ad(!node->trx->dict_operation_lock_mode);
 		mtr_s_lock(dict_index_get_lock(index), &mtr);
 	}
 
@@ -807,37 +805,6 @@ func_exit_no_pcur:
 }
 
 /***********************************************************//**
-Flags a secondary index corrupted. */
-static MY_ATTRIBUTE((nonnull))
-void
-row_undo_mod_sec_flag_corrupted(
-/*============================*/
-	trx_t*		trx,	/*!< in/out: transaction */
-	dict_index_t*	index)	/*!< in: secondary index */
-{
-	ut_ad(!dict_index_is_clust(index));
-
-	switch (trx->dict_operation_lock_mode) {
-	case RW_S_LATCH:
-		/* Because row_undo() is holding an S-latch
-		on the data dictionary during normal rollback,
-		we can only mark the index corrupted in the
-		data dictionary cache. TODO: fix this somehow.*/
-		mutex_enter(&dict_sys->mutex);
-		dict_set_corrupted_index_cache_only(index);
-		mutex_exit(&dict_sys->mutex);
-		break;
-	default:
-		ut_ad(0);
-		/* fall through */
-	case RW_X_LATCH:
-		/* This should be the rollback of a data dictionary
-		transaction. */
-		dict_set_corrupted(index, trx, "rollback");
-	}
-}
-
-/***********************************************************//**
 Undoes a modify in secondary indexes when undo record type is UPD_DEL.
 @return DB_SUCCESS or DB_OUT_OF_FILE_SPACE */
 static MY_ATTRIBUTE((nonnull, warn_unused_result))
@@ -950,8 +917,7 @@ row_undo_mod_del_mark_sec(
 		}
 
 		if (err == DB_DUPLICATE_KEY) {
-			row_undo_mod_sec_flag_corrupted(
-				thr_get_trx(thr), index);
+			index->type |= DICT_CORRUPT;
 			err = DB_SUCCESS;
 			/* Do not return any error to the caller. The
 			duplicate will be reported by ALTER TABLE or
@@ -1097,8 +1063,7 @@ row_undo_mod_upd_exist_sec(
 		}
 
 		if (err == DB_DUPLICATE_KEY) {
-			row_undo_mod_sec_flag_corrupted(
-				thr_get_trx(thr), index);
+			index->type |= DICT_CORRUPT;
 			err = DB_SUCCESS;
 		} else if (err != DB_SUCCESS) {
 			break;
@@ -1250,6 +1215,8 @@ row_undo_mod(
 		return(DB_SUCCESS);
 	}
 
+	ut_ad(node->table->is_temporary()
+	      || lock_table_has_locks(node->table));
 	node->index = dict_table_get_first_index(node->table);
 	ut_ad(dict_index_is_clust(node->index));
 	/* Skip the clustered index (the first index) */
diff --git a/storage/innobase/row/row0undo.cc b/storage/innobase/row/row0undo.cc
index b65b173fedb..34b55f25527 100644
--- a/storage/innobase/row/row0undo.cc
+++ b/storage/innobase/row/row0undo.cc
@@ -279,18 +279,6 @@ row_undo(
 			? UNDO_NODE_INSERT : UNDO_NODE_MODIFY;
 	}
 
-	/* Prevent DROP TABLE etc. while we are rolling back this row.
-	If we are doing a TABLE CREATE or some other dictionary operation,
-	then we already have dict_operation_lock locked in x-mode. Do not
-	try to lock again, because that would cause a hang. */
-
-	const bool locked_data_dict = (trx->dict_operation_lock_mode == 0);
-
-	if (locked_data_dict) {
-
-		row_mysql_freeze_data_dictionary(trx);
-	}
-
 	dberr_t err;
 
 	if (node->state == UNDO_NODE_INSERT) {
@@ -303,11 +291,6 @@ row_undo(
 		err = row_undo_mod(node, thr);
 	}
 
-	if (locked_data_dict) {
-
-		row_mysql_unfreeze_data_dictionary(trx);
-	}
-
 	/* Do some cleanup */
 	btr_pcur_close(&(node->pcur));