MDEV-18518 Multi-table CREATE and DROP transactions for InnoDB

InnoDB used to support at most one CREATE TABLE or DROP TABLE
per transaction. This caused complications for DDL operations on
partitioned tables (where each partition is treated as a separate
table by InnoDB) and FULLTEXT INDEX (where each index is maintained
in a number of internal InnoDB tables).

dict_drop_index_tree(): Extend the MDEV-24589 logic and treat
the purge or rollback of SYS_INDEXES records of clustered indexes
specially: by dropping the tablespace if it exists. This is the only
form of recovery that we will need.

trx_undo_ddl_type: Document the DDL undo log record types better.

trx_t::dict_operation: Change the type to bool.

trx_t::ddl: Remove.

trx_t::table_id, trx_undo_t::table_id: Remove.

dict_build_table_def_step(): Remove trx_t::table_id logging.

dict_table_close_and_drop(), row_merge_drop_table(): Remove.

row_merge_lock_table(): Merged to the only callers, which can
call lock_table_for_trx() directly.

fts_aux_table_t, fts_aux_id, fts_space_set_t: Remove.

fts_drop_orphaned_tables(): Remove.

row_merge_rename_index_to_drop(): Remove. Thanks to MDEV-24589,
we can simply delete the to-be-dropped indexes from SYS_INDEXES,
while still being able to roll back the operation.

ha_innobase_inplace_ctx: Make a few data members const.
Preallocate trx.

prepare_inplace_alter_table_dict(): Simplify the logic. Let the
normal rollback take care of some cleanup.

row_undo_ins_remove_clust_rec(): Simplify the parsing of SYS_COLUMNS.

trx_rollback_active(): Remove the special DROP TABLE logic.

trx_undo_mem_create_at_db_start(), trx_undo_reuse_cached():
Always write TRX_UNDO_TABLE_ID as 0.

27 files changed, 359 insertions, 1029 deletions

diff --git a/storage/innobase/dict/dict0boot.cc b/storage/innobase/dict/dict0boot.cc
index e9c900a9f1c..afe8b5ec165 100644
--- a/storage/innobase/dict/dict0boot.cc
+++ b/storage/innobase/dict/dict0boot.cc
@@ -349,8 +349,7 @@ dict_boot(void)
 	dict_mem_table_add_col(table, heap, "NAME", DATA_BINARY, 0, 0);
 	dict_mem_table_add_col(table, heap, "N_FIELDS", DATA_INT, 0, 4);
 	dict_mem_table_add_col(table, heap, "TYPE", DATA_INT, 0, 4);
-	/* SYS_INDEXES.SPACE is redundant and not being read;
-	SYS_TABLES.SPACE is being used instead. */
+	/* SYS_INDEXES.SPACE is only read by in dict_drop_index_tree() */
 	dict_mem_table_add_col(table, heap, "SPACE", DATA_INT, 0, 4);
 	dict_mem_table_add_col(table, heap, "PAGE_NO", DATA_INT, 0, 4);
 	dict_mem_table_add_col(table, heap, "MERGE_THRESHOLD", DATA_INT, 0, 4);
diff --git a/storage/innobase/dict/dict0crea.cc b/storage/innobase/dict/dict0crea.cc
index 0b9d98fd2ca..16270039d77 100644
--- a/storage/innobase/dict/dict0crea.cc
+++ b/storage/innobase/dict/dict0crea.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, 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
@@ -345,12 +345,10 @@ dict_build_table_def_step(
 {
 	dict_sys.assert_locked();
 	dict_table_t*	table = node->table;
-	trx_t* trx = thr_get_trx(thr);
 	ut_ad(!table->is_temporary());
 	ut_ad(!table->space);
 	ut_ad(table->space_id == ULINT_UNDEFINED);
 	dict_hdr_get_new_id(&table->id, NULL, NULL);
-	trx->table_id = table->id;
 
 	/* Always set this bit for all new created tables */
 	DICT_TF2_FLAG_SET(table, DICT_TF2_FTS_AUX_HEX_NAME);
@@ -363,33 +361,6 @@ dict_build_table_def_step(
 
 		ut_ad(DICT_TF_GET_ZIP_SSIZE(table->flags) == 0
 		      || dict_table_has_atomic_blobs(table));
-		mtr_t mtr;
-		trx_undo_t* undo = trx->rsegs.m_redo.undo;
-		if (undo && !undo->table_id
-		    && trx_get_dict_operation(trx) == TRX_DICT_OP_TABLE) {
-			/* This must be a TRUNCATE operation where
-			the empty table is created after the old table
-			was renamed. Be sure to mark the transaction
-			associated with the new empty table, so that
-			we can remove it on recovery. */
-			mtr.start();
-			undo->table_id = trx->table_id;
-			undo->dict_operation = TRUE;
-			buf_block_t* block = trx_undo_page_get(
-				page_id_t(trx->rsegs.m_redo.rseg->space->id,
-					  undo->hdr_page_no),
-				&mtr);
-			mtr.write<1,mtr_t::MAYBE_NOP>(
-				*block,
-				block->frame + undo->hdr_offset
-				+ TRX_UNDO_DICT_TRANS, 1U);
-			mtr.write<8,mtr_t::MAYBE_NOP>(
-				*block,
-				block->frame + undo->hdr_offset
-				+ TRX_UNDO_TABLE_ID, trx->table_id);
-			mtr.commit();
-			log_write_up_to(mtr.commit_lsn(), true);
-		}
 		/* Get a new tablespace ID */
 		ulint space_id;
 		dict_hdr_get_new_id(NULL, NULL, &space_id);
@@ -435,6 +406,7 @@ dict_build_table_def_step(
 		}
 
 		table->space_id = space_id;
+		mtr_t mtr;
 		mtr.start();
 		mtr.set_named_space(table->space);
 		fsp_header_init(table->space, FIL_IBD_FILE_INITIAL_SIZE, &mtr);
@@ -725,11 +697,6 @@ dict_build_index_def_step(
 		return(DB_TABLE_NOT_FOUND);
 	}
 
-	if (!trx->table_id) {
-		/* Record only the first table id. */
-		trx->table_id = table->id;
-	}
-
 	ut_ad((UT_LIST_GET_LEN(table->indexes) > 0)
 	      || dict_index_is_clust(index));
 
@@ -765,11 +732,6 @@ dict_build_index_def(
 {
 	dict_sys.assert_locked();
 
-	if (trx->table_id == 0) {
-		/* Record only the first table id. */
-		trx->table_id = table->id;
-	}
-
 	ut_ad((UT_LIST_GET_LEN(table->indexes) > 0)
 	      || dict_index_is_clust(index));
 
@@ -903,73 +865,80 @@ dict_create_index_tree_in_mem(
 /** Drop the index tree associated with a row in SYS_INDEXES table.
 @param[in,out]	pcur	persistent cursor on rec
 @param[in,out]	trx	dictionary transaction
+@param[in,out]	table	table that the record belongs to
 @param[in,out]	mtr	mini-transaction */
-void dict_drop_index_tree(btr_pcur_t* pcur, trx_t* trx, mtr_t* mtr)
+void dict_drop_index_tree(btr_pcur_t *pcur, trx_t *trx, dict_table_t *table,
+                          mtr_t *mtr)
 {
-	rec_t*	rec = btr_pcur_get_rec(pcur);
-	byte*	ptr;
-	ulint	len;
+  rec_t *rec= btr_pcur_get_rec(pcur);
 
-	ut_d(if (trx) dict_sys.assert_locked());
-	ut_ad(!dict_table_is_comp(dict_sys.sys_indexes));
-	btr_pcur_store_position(pcur, mtr);
+  ut_d(if (trx) dict_sys.assert_locked());
+  ut_ad(!dict_table_is_comp(dict_sys.sys_indexes));
+  btr_pcur_store_position(pcur, mtr);
 
-	static_assert(DICT_FLD__SYS_INDEXES__TABLE_ID == 0, "compatibility");
-	static_assert(DICT_FLD__SYS_INDEXES__ID == 1, "compatibility");
+  static_assert(DICT_FLD__SYS_INDEXES__TABLE_ID == 0, "compatibility");
+  static_assert(DICT_FLD__SYS_INDEXES__ID == 1, "compatibility");
 
-	if (rec_get_1byte_offs_flag(rec)) {
-		if (rec_1_get_field_end_info(rec, 0) != 8
-		    || rec_1_get_field_end_info(rec, 1) != 8 + 8) {
+  ulint len= rec_get_n_fields_old(rec);
+  if (len < DICT_FLD__SYS_INDEXES__MERGE_THRESHOLD ||
+      len > DICT_NUM_FIELDS__SYS_INDEXES)
+  {
 rec_corrupted:
-			ib::error() << "Corrupted SYS_INDEXES record";
-			return;
-		}
-	} else if (rec_2_get_field_end_info(rec, 0) != 8
-		   || rec_2_get_field_end_info(rec, 1) != 8 + 8) {
-		goto rec_corrupted;
-	}
-
-	ptr = rec_get_nth_field_old(rec, DICT_FLD__SYS_INDEXES__PAGE_NO, &len);
-	if (len != 4) {
-		goto rec_corrupted;
-	}
-
-	const uint32_t root_page_no = mach_read_from_4(ptr);
-
-	if (root_page_no == FIL_NULL) {
-		/* The tree has already been freed */
-		return;
-	}
-
-	compile_time_assert(FIL_NULL == 0xffffffff);
-	mtr->memset(btr_pcur_get_block(pcur), page_offset(ptr), 4, 0xff);
-
-	ptr = rec_get_nth_field_old(
-		rec, DICT_FLD__SYS_INDEXES__SPACE, &len);
-
-	if (len != 4) {
-		goto rec_corrupted;
-	}
-
-	const uint32_t space_id = mach_read_from_4(ptr);
-	ut_ad(space_id < SRV_TMP_SPACE_ID);
-	if (space_id != TRX_SYS_SPACE && trx
-	    && trx_get_dict_operation(trx) == TRX_DICT_OP_TABLE) {
-		/* We are about to delete the entire .ibd file;
-		do not bother to free pages inside it. */
-		return;
-	}
-
-	if (fil_space_t* s = fil_space_t::get(space_id)) {
-		/* Ensure that the tablespace file exists
-		in order to avoid a crash in buf_page_get_gen(). */
-		if (root_page_no < s->get_size()) {
-			btr_free_if_exists(page_id_t(space_id, root_page_no),
-					   s->zip_size(),
-					   mach_read_from_8(rec + 8), mtr);
-		}
-		s->release();
-	}
+    ib::error() << "Corrupted SYS_INDEXES record";
+    return;
+  }
+
+  if (rec_get_1byte_offs_flag(rec))
+  {
+    if (rec_1_get_field_end_info(rec, 0) != 8 ||
+        rec_1_get_field_end_info(rec, 1) != 8 + 8)
+      goto rec_corrupted;
+  }
+  else if (rec_2_get_field_end_info(rec, 0) != 8 ||
+           rec_2_get_field_end_info(rec, 1) != 8 + 8)
+    goto rec_corrupted;
+
+  const byte *p= rec_get_nth_field_old(rec, DICT_FLD__SYS_INDEXES__TYPE, &len);
+  if (len != 4)
+    goto rec_corrupted;
+  const uint32_t type= mach_read_from_4(p);
+  p= rec_get_nth_field_old(rec, DICT_FLD__SYS_INDEXES__PAGE_NO, &len);
+  if (len != 4)
+    goto rec_corrupted;
+  const uint32_t root_page_no= mach_read_from_4(p);
+  p= rec_get_nth_field_old(rec, DICT_FLD__SYS_INDEXES__SPACE, &len);
+  if (len != 4)
+    goto rec_corrupted;
+
+  if (root_page_no == FIL_NULL)
+    /* The tree has already been freed */
+    return;
+
+  static_assert(FIL_NULL == 0xffffffff, "compatibility");
+  static_assert(DICT_FLD__SYS_INDEXES__PAGE_NO ==
+                DICT_FLD__SYS_INDEXES__SPACE + 1, "compatibility");
+  mtr->memset(btr_pcur_get_block(pcur), page_offset(p + 4), 4, 0xff);
+
+  const uint32_t space_id= mach_read_from_4(p);
+  ut_ad(space_id < SRV_TMP_SPACE_ID);
+
+  if (space_id && (type & DICT_CLUSTERED))
+  {
+    if (table && table->space_id == space_id)
+      table->space= nullptr;
+    else
+      ut_ad(!table);
+    fil_delete_tablespace(space_id, true);
+  }
+  else if (fil_space_t*s= fil_space_t::get(space_id))
+  {
+    /* Ensure that the tablespace file exists
+    in order to avoid a crash in buf_page_get_gen(). */
+    if (root_page_no < s->get_size())
+      btr_free_if_exists(page_id_t(space_id, root_page_no), s->zip_size(),
+                         mach_read_from_8(rec + 8), mtr);
+    s->release();
+  }
 }
 
 /*********************************************************************//**
@@ -1431,7 +1400,7 @@ dict_create_or_check_foreign_constraint_tables(void)
 
 	trx = trx_create();
 
-	trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
+	trx->dict_operation = true;
 
 	trx->op_info = "creating foreign key sys tables";
 
@@ -1570,7 +1539,7 @@ dict_create_or_check_sys_virtual()
 
 	trx = trx_create();
 
-	trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
+	trx->dict_operation = true;
 
 	trx->op_info = "creating sys_virtual tables";
 
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 3287041ccb2..8818cd085d8 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -247,7 +247,7 @@ dict_table_try_drop_aborted(
 	trx = trx_create();
 	trx->op_info = "try to drop any indexes after an aborted index creation";
 	row_mysql_lock_data_dictionary(trx);
-	trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);
+	trx->dict_operation = true;
 
 	if (table == NULL) {
 		table = dict_table_open_on_id_low(
@@ -367,42 +367,6 @@ dict_table_close(
 	}
 }
 
-/********************************************************************//**
-Closes the only open handle to a table and drops a table while assuring
-that dict_sys.mutex is held the whole time.  This assures that the table
-is not evicted after the close when the count of open handles goes to zero.
-Because dict_sys.mutex is held, we do not need to call
-dict_table_prevent_eviction().  */
-void
-dict_table_close_and_drop(
-/*======================*/
-	trx_t*		trx,		/*!< in: data dictionary transaction */
-	dict_table_t*	table)		/*!< in/out: table */
-{
-	dberr_t err = DB_SUCCESS;
-
-	ut_d(dict_sys.assert_locked());
-	ut_ad(trx->dict_operation != TRX_DICT_OP_NONE);
-	ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE));
-
-	dict_table_close(table, true, false);
-
-#if defined UNIV_DEBUG || defined UNIV_DDL_DEBUG
-	/* Nobody should have initialized the stats of the newly created
-	table when this is called. So we know that it has not been added
-	for background stats gathering. */
-	ut_a(!table->stat_initialized);
-#endif /* UNIV_DEBUG || UNIV_DDL_DEBUG */
-
-	err = row_merge_drop_table(trx, table);
-
-	if (err != DB_SUCCESS) {
-		ib::error() << "At " << __FILE__ << ":" << __LINE__
-			    << " row_merge_drop_table returned error: " << err
-			    << " table: " << table->name;
-	}
-}
-
 /** Check if the table has a given (non_virtual) column.
 @param[in]	table		table object
 @param[in]	col_name	column name
@@ -943,8 +907,6 @@ dict_table_open_on_id(table_id_t table_id, bool dict_locked,
                       dict_table_op_t table_op, THD *thd,
                       MDL_ticket **mdl)
 {
-	ut_ad(!dict_locked || !thd);
-
 	if (!dict_locked) {
 		dict_sys.mutex_lock();
 	}
@@ -2013,7 +1975,7 @@ void dict_sys_t::remove(dict_table_t* table, bool lru, bool keep)
 		/* Mimic row_mysql_lock_data_dictionary(). */
 		trx->dict_operation_lock_mode = RW_X_LATCH;
 
-		trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);
+		trx->dict_operation = true;
 		row_merge_drop_indexes_dict(trx, table->id);
 		trx_commit_for_mysql(trx);
 		trx->dict_operation_lock_mode = 0;
diff --git a/storage/innobase/fts/fts0fts.cc b/storage/innobase/fts/fts0fts.cc
index ae88d0bd3c0..8ffbeedef2f 100644
--- a/storage/innobase/fts/fts0fts.cc
+++ b/storage/innobase/fts/fts0fts.cc
@@ -134,14 +134,6 @@ const char *fts_default_stopword[] =
 	NULL
 };
 
-/** For storing table info when checking for orphaned tables. */
-struct fts_aux_table_t {
-	table_id_t	id;		/*!< Table id */
-	table_id_t	parent_id;	/*!< Parent table id */
-	table_id_t	index_id;	/*!< Table FT index id */
-	char*		name;		/*!< Name of the table */
-};
-
 /** FTS auxiliary table suffixes that are common to all FT indexes. */
 const char* fts_common_tables[] = {
 	"BEING_DELETED",
@@ -1555,14 +1547,8 @@ on the given table. row_mysql_lock_data_dictionary must have been called
 before this.
 @param[in]	trx		transaction to drop fts common table
 @param[in]	fts_table	table with an FTS index
-@param[in]	drop_orphan	True if the function is used to drop
-				orphaned table
 @return DB_SUCCESS or error code */
-static dberr_t
-fts_drop_common_tables(
-	trx_t*		trx,
-	fts_table_t*	fts_table,
-	bool		drop_orphan=false)
+static dberr_t fts_drop_common_tables(trx_t *trx, fts_table_t *fts_table)
 {
 	ulint		i;
 	dberr_t		error = DB_SUCCESS;
@@ -1580,16 +1566,6 @@ fts_drop_common_tables(
 		if (err != DB_SUCCESS && err != DB_FAIL) {
 			error = err;
 		}
-
-		if (drop_orphan && err == DB_FAIL) {
-			char* path = fil_make_filepath(
-				NULL, table_name_t{table_name}, IBD, false);
-			if (path != NULL) {
-				os_file_delete_if_exists(
-					innodb_data_file_key, path, NULL);
-				ut_free(path);
-			}
-		}
 	}
 
 	return(error);
@@ -1789,14 +1765,7 @@ fts_create_one_common_table(
 			dict_mem_index_add_field(index, "key", 0);
 		}
 
-		/* We save and restore trx->dict_operation because
-		row_create_index_for_mysql() changes the operation to
-		TRX_DICT_OP_TABLE. */
-		trx_dict_op_t op = trx_get_dict_operation(trx);
-
 		error =	row_create_index_for_mysql(index, trx, NULL);
-
-		trx->dict_operation = op;
 	}
 
 	if (error != DB_SUCCESS) {
@@ -1845,7 +1814,6 @@ fts_create_common_tables(
 				[MAX_FULL_NAME_LEN];
 
 	dict_index_t*					index = NULL;
-	trx_dict_op_t					op;
 	/* common_tables vector is used for dropping FTS common tables
 	on error condition. */
 	std::vector<dict_table_t*>			common_tables;
@@ -1910,12 +1878,8 @@ fts_create_common_tables(
 				      DICT_UNIQUE, 1);
 	dict_mem_index_add_field(index, FTS_DOC_ID_COL_NAME, 0);
 
-	op = trx_get_dict_operation(trx);
-
 	error =	row_create_index_for_mysql(index, trx, NULL);
 
-	trx->dict_operation = op;
-
 func_exit:
 	if (error != DB_SUCCESS) {
 		for (it = common_tables.begin(); it != common_tables.end();
@@ -2003,11 +1967,7 @@ fts_create_one_index_table(
 		dict_mem_index_add_field(index, "word", 0);
 		dict_mem_index_add_field(index, "first_doc_id", 0);
 
-		trx_dict_op_t op = trx_get_dict_operation(trx);
-
 		error =	row_create_index_for_mysql(index, trx, NULL);
-
-		trx->dict_operation = op;
 	}
 
 	if (error != DB_SUCCESS) {
@@ -5737,161 +5697,6 @@ bool fts_check_aux_table(const char *name,
   return false;
 }
 
-typedef std::pair<table_id_t,index_id_t> fts_aux_id;
-typedef std::set<fts_aux_id> fts_space_set_t;
-
-/** Iterate over all the spaces in the space list and fetch the
-fts parent table id and index id.
-@param[in,out]	fts_space_set	store the list of tablespace id and
-				index id */
-static void fil_get_fts_spaces(fts_space_set_t& fts_space_set)
-{
-  mysql_mutex_lock(&fil_system.mutex);
-
-  for (fil_space_t &space : fil_system.space_list)
-  {
-    index_id_t index_id= 0;
-    table_id_t table_id= 0;
-
-    if (space.purpose == FIL_TYPE_TABLESPACE && space.id &&
-        space.chain.start &&
-        fts_check_aux_table(space.chain.start->name, &table_id, &index_id))
-      fts_space_set.insert(std::make_pair(table_id, index_id));
-  }
-
-  mysql_mutex_unlock(&fil_system.mutex);
-}
-
-/** Check whether the parent table id and index id of fts auxilary
-tables with SYS_INDEXES. If it exists then we can safely ignore the
-fts table from orphaned tables.
-@param[in,out]	fts_space_set	fts space set contains set of auxiliary
-				table ids */
-static void fts_check_orphaned_tables(fts_space_set_t& fts_space_set)
-{
-  btr_pcur_t pcur;
-  mtr_t	     mtr;
-  trx_t*     trx = trx_create();
-  trx->op_info = "checking fts orphaned tables";
-
-  row_mysql_lock_data_dictionary(trx);
-
-  mtr.start();
-  btr_pcur_open_at_index_side(
-    true, dict_table_get_first_index(dict_sys.sys_indexes),
-    BTR_SEARCH_LEAF, &pcur, true, 0, &mtr);
-
-  do
-  {
-    const rec_t *rec;
-    const byte *tbl_field;
-    const byte *index_field;
-    ulint len;
-
-    btr_pcur_move_to_next_user_rec(&pcur, &mtr);
-    if (!btr_pcur_is_on_user_rec(&pcur))
-      break;
-
-    rec= btr_pcur_get_rec(&pcur);
-    if (rec_get_deleted_flag(rec, 0))
-      continue;
-
-    tbl_field= rec_get_nth_field_old(rec, 0, &len);
-    if (len != 8)
-      continue;
-
-    index_field= rec_get_nth_field_old(rec, 1, &len);
-    if (len != 8)
-      continue;
-
-    table_id_t table_id = mach_read_from_8(tbl_field);
-    index_id_t index_id = mach_read_from_8(index_field);
-
-    fts_space_set_t::iterator it = fts_space_set.find(
-	fts_aux_id(table_id, index_id));
-
-    if (it != fts_space_set.end())
-      fts_space_set.erase(*it);
-    else
-    {
-      it= fts_space_set.find(fts_aux_id(table_id, 0));
-      if (it != fts_space_set.end())
-        fts_space_set.erase(*it);
-    }
-  } while(!fts_space_set.empty());
-
-  btr_pcur_close(&pcur);
-  mtr.commit();
-  row_mysql_unlock_data_dictionary(trx);
-  trx->free();
-}
-
-/** Drop all fts auxilary table for the respective fts_id
-@param[in]	fts_id	fts auxilary table ids */
-static void fts_drop_all_aux_tables(trx_t *trx, fts_table_t *fts_table)
-{
-  char fts_table_name[MAX_FULL_NAME_LEN];
-  for (ulint i= 0;i < FTS_NUM_AUX_INDEX; i++)
-  {
-    fts_table->suffix= fts_get_suffix(i);
-    fts_get_table_name(fts_table, fts_table_name, true);
-
-    /* Drop all fts aux and common table */
-    if (fts_drop_table(trx, fts_table_name) != DB_FAIL)
-      continue;
-
-    if (char *path= fil_make_filepath(nullptr, table_name_t{fts_table_name},
-                                      IBD, false))
-    {
-      os_file_delete_if_exists(innodb_data_file_key, path, nullptr);
-      ut_free(path);
-    }
-  }
-}
-
-/** Drop all orphaned FTS auxiliary tables, those that don't have
-a parent table or FTS index defined on them. */
-void fts_drop_orphaned_tables()
-{
-  fts_space_set_t fts_space_set;
-  fil_get_fts_spaces(fts_space_set);
-
-  if (fts_space_set.empty())
-    return;
-
-  fts_check_orphaned_tables(fts_space_set);
-
-  if (fts_space_set.empty())
-    return;
-
-  trx_t* trx= trx_create();
-  trx->op_info= "Drop orphaned aux FTS tables";
-  row_mysql_lock_data_dictionary(trx);
-
-  for (fts_space_set_t::iterator it = fts_space_set.begin();
-       it != fts_space_set.end(); it++)
-  {
-    fts_table_t fts_table;
-    dict_table_t *table= dict_table_open_on_id(it->first, TRUE,
-                                               DICT_TABLE_OP_NORMAL);
-    if (!table)
-      continue;
-
-    FTS_INIT_FTS_TABLE(&fts_table, NULL, FTS_COMMON_TABLE, table);
-    fts_drop_common_tables(trx, &fts_table, true);
-
-    fts_table.type= FTS_INDEX_TABLE;
-    fts_table.index_id= it->second;
-    fts_drop_all_aux_tables(trx, &fts_table);
-
-    dict_table_close(table, true, false);
-  }
-  trx_commit_for_mysql(trx);
-  row_mysql_unlock_data_dictionary(trx);
-  trx->dict_operation_lock_mode= 0;
-  trx->free();
-}
-
 /**********************************************************************//**
 Check whether user supplied stopword table is of the right format.
 Caller is responsible to hold dictionary locks.
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 75c076eeb96..925674c35b6 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -2653,7 +2653,7 @@ ha_innobase::update_thd(
 	trx_t*	trx = check_trx_exists(thd);
 
 	ut_ad(trx->dict_operation_lock_mode == 0);
-	ut_ad(trx->dict_operation == TRX_DICT_OP_NONE);
+	ut_ad(!trx->dict_operation);
 
 	if (m_prebuilt->trx != trx) {
 
@@ -4057,7 +4057,7 @@ innobase_commit(
 	trx_t*	trx = check_trx_exists(thd);
 
 	ut_ad(trx->dict_operation_lock_mode == 0);
-	ut_ad(trx->dict_operation == TRX_DICT_OP_NONE);
+	ut_ad(!trx->dict_operation);
 
 	/* Transaction is deregistered only in a commit or a rollback. If
 	it is deregistered we know there cannot be resources to be freed
@@ -4146,7 +4146,7 @@ innobase_rollback(
 	trx_t*	trx = check_trx_exists(thd);
 
 	ut_ad(trx->dict_operation_lock_mode == 0);
-	ut_ad(trx->dict_operation == TRX_DICT_OP_NONE);
+	ut_ad(!trx->dict_operation);
 
 	/* Reset the number AUTO-INC rows required */
 
@@ -10392,8 +10392,7 @@ err_col:
 					    "temporary table creation.");
 		}
 
-		m_trx->table_id = table->id
-			= dict_sys.get_temporary_table_id();
+		table->id = dict_sys.get_temporary_table_id();
 		ut_ad(dict_tf_get_rec_format(table->flags)
 		      != REC_FORMAT_COMPRESSED);
 		table->space_id = SRV_TMP_SPACE_ID;
@@ -12185,7 +12184,7 @@ create_table_info_t::create_foreign_keys()
 
 	trx_start_if_not_started_xa(m_trx, true);
 
-	trx_set_dict_operation(m_trx, TRX_DICT_OP_TABLE);
+	m_trx->dict_operation = true;
 
 	error = dict_create_add_foreigns_to_dictionary(local_fk_set, table,
 						       m_trx);
@@ -12712,8 +12711,7 @@ create_table_info_t::allocate_trx()
 {
 	m_trx = innobase_trx_allocate(m_thd);
 
-	m_trx->will_lock++;
-	m_trx->ddl = true;
+	m_trx->will_lock = 1;
 }
 
 /** Create a new table to an InnoDB database.
@@ -13259,8 +13257,7 @@ inline dberr_t innobase_rename_table(trx_t *trx, const char *from,
 	char	norm_from[FN_REFLEN];
 
 	DBUG_ENTER("innobase_rename_table");
-	DBUG_ASSERT(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX
-		    || trx_get_dict_operation(trx) == TRX_DICT_OP_TABLE);
+	DBUG_ASSERT(trx->dict_operation);
 
 	ut_ad(!srv_read_only_mode);
 
@@ -13381,8 +13378,8 @@ int ha_innobase::truncate()
 	const char* name = mem_heap_strdup(heap, ib_table->name.m_name);
 	trx_t*	trx = innobase_trx_allocate(m_user_thd);
 
-	++trx->will_lock;
-	trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
+	trx->will_lock = 1;
+	trx->dict_operation = true;
 	row_mysql_lock_data_dictionary(trx);
 	dict_stats_wait_bg_to_stop_using_table(ib_table, trx);
 
@@ -13468,8 +13465,8 @@ ha_innobase::rename_table(
 	trx_t*	trx = innobase_trx_allocate(thd);
 
 	/* We are doing a DDL operation. */
-	++trx->will_lock;
-	trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);
+	trx->will_lock = 1;
+	trx->dict_operation = true;
 
 	dberr_t	error = innobase_rename_table(trx, from, to, true);
 
@@ -20755,10 +20752,7 @@ ib_push_frm_error(
 		sql_print_error("InnoDB: Table %s contains " ULINTPF " "
 			"indexes inside InnoDB, which "
 			"is different from the number of "
-			"indexes %u defined in the MariaDB "
-			" Have you mixed up "
-			".frm files from different "
-			"installations? See "
+			"indexes %u defined in the .frm file. See "
 			"https://mariadb.com/kb/en/innodb-troubleshooting/\n",
 			ib_table->name.m_name, n_keys,
 			table->s->keys);
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 70939618c96..0bcb6114ab4 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -839,7 +839,7 @@ inline void dict_table_t::rollback_instant(
 struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 {
 	/** Dummy query graph */
-	que_thr_t*	thr;
+	que_thr_t*const	thr;
 	/** The prebuilt struct of the creating instance */
 	row_prebuilt_t*&	prebuilt;
 	/** InnoDB indexes being created */
@@ -861,9 +861,9 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 	/** number of InnoDB foreign key constraints being dropped */
 	const ulint	num_to_add_fk;
 	/** whether to create the indexes online */
-	bool		online;
+	const bool	online;
 	/** memory heap */
-	mem_heap_t*	heap;
+	mem_heap_t* const heap;
 	/** dictionary transaction */
 	trx_t*		trx;
 	/** original table (if rebuilt, differs from indexed_table) */
@@ -948,12 +948,15 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 				bool page_compressed,
 				ulonglong page_compression_level_arg) :
 		inplace_alter_handler_ctx(),
+		thr (pars_complete_graph_for_exec(nullptr, prebuilt_arg->trx,
+						  heap_arg, prebuilt_arg)),
 		prebuilt (prebuilt_arg),
 		add_index (0), add_key_numbers (0), num_to_add_index (0),
 		drop_index (drop_arg), num_to_drop_index (num_to_drop_arg),
 		drop_fk (drop_fk_arg), num_to_drop_fk (num_to_drop_fk_arg),
 		add_fk (add_fk_arg), num_to_add_fk (num_to_add_fk_arg),
-		online (online_arg), heap (heap_arg), trx (0),
+		online (online_arg), heap (heap_arg),
+		trx (innobase_trx_allocate(prebuilt_arg->trx->mysql_thd)),
 		old_table (prebuilt_arg->table),
 		new_table (new_table_arg), instant_table (0),
 		col_map (0), col_names (col_names_arg),
@@ -1000,8 +1003,7 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 		}
 #endif /* UNIV_DEBUG */
 
-		thr = pars_complete_graph_for_exec(NULL, prebuilt->trx, heap,
-			prebuilt);
+		trx_start_for_ddl(trx);
 	}
 
 	~ha_innobase_inplace_ctx()
@@ -1896,7 +1898,7 @@ innobase_fts_check_doc_id_col(
 			col = dict_table_get_nth_col(table, i);
 
 			/* Because the FTS_DOC_ID does not exist in
-			the MySQL data dictionary, this must be the
+			the .frm file or TABLE_SHARE, this must be the
 			internally created FTS_DOC_ID column. */
 			ut_ad(col->mtype == DATA_INT);
 			ut_ad(col->len == 8);
@@ -4070,7 +4072,7 @@ online_retry_drop_indexes_low(
 {
 	dict_sys.assert_locked();
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_ad(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	ut_ad(trx->dict_operation);
 
 	/* We can have table->n_ref_count > 1, because other threads
 	may have prebuilt->table pointing to the table. However, these
@@ -4096,7 +4098,7 @@ online_retry_drop_indexes(
 	if (table->drop_aborted) {
 		trx_t*	trx = innobase_trx_allocate(user_thd);
 
-		trx_start_for_ddl(trx, TRX_DICT_OP_INDEX);
+		trx_start_for_ddl(trx);
 
 		row_mysql_lock_data_dictionary(trx);
 		online_retry_drop_indexes_low(table, trx);
@@ -4129,10 +4131,7 @@ online_retry_drop_indexes_with_trx(
 	drop any incompletely created indexes that may have been left
 	behind in rollback_inplace_alter_table() earlier. */
 	if (table->drop_aborted) {
-
-		trx->table_id = 0;
-
-		trx_start_for_ddl(trx, TRX_DICT_OP_INDEX);
+		trx_start_for_ddl(trx);
 
 		online_retry_drop_indexes_low(table, trx);
 		trx_commit_for_mysql(trx);
@@ -4811,7 +4810,7 @@ innobase_update_gis_column_type(
 
 	DBUG_ENTER("innobase_update_gis_column_type");
 
-	DBUG_ASSERT(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_d(dict_sys.assert_locked());
 
@@ -6274,25 +6273,21 @@ prepare_inplace_alter_table_dict(
 			       * sizeof *ctx->add_key_numbers));
 
 	/* Acquire a lock on the table before creating any indexes. */
+	bool table_lock_failed = false;
 
 	if (ctx->online) {
 		error = DB_SUCCESS;
 	} else {
-		error = row_merge_lock_table(
-			ctx->prebuilt->trx, ctx->new_table, LOCK_S);
+		ctx->prebuilt->trx->op_info = "acquiring table lock";
+		error = lock_table_for_trx(ctx->new_table,
+					   ctx->prebuilt->trx, LOCK_S);
 
 		if (error != DB_SUCCESS) {
-
+			table_lock_failed = true;
 			goto error_handling;
 		}
 	}
 
-	/* Create a background transaction for the operations on
-	the data dictionary tables. */
-	ctx->trx = innobase_trx_allocate(ctx->prebuilt->trx->mysql_thd);
-
-	trx_start_for_ddl(ctx->trx, TRX_DICT_OP_INDEX);
-
 	/* Latch the InnoDB data dictionary exclusively so that no deadlocks
 	or lock waits can happen in it during an index create operation. */
 
@@ -6803,30 +6798,15 @@ wrong_column_name:
 		}
 
 		/* Create the table. */
-		trx_set_dict_operation(ctx->trx, TRX_DICT_OP_TABLE);
+		ctx->trx->dict_operation = true;
 
 		error = row_create_table_for_mysql(
 			ctx->new_table, ctx->trx, mode, key_id);
 
 		switch (error) {
-			dict_table_t*	temp_table;
 		case DB_SUCCESS:
-			/* We need to bump up the table ref count and
-			before we can use it we need to open the
-			table. The new_table must be in the data
-			dictionary cache, because we are still holding
-			the dict_sys.mutex. */
-			dict_sys.assert_locked();
-			temp_table = dict_table_open_on_name(
-				ctx->new_table->name.m_name, TRUE, FALSE,
-				DICT_ERR_IGNORE_NONE);
-			ut_a(ctx->new_table == temp_table);
-			/* n_ref_count must be 1, because purge cannot
-			be executing on this very table as we are
-			holding dict_sys.latch X-latch. */
-			DBUG_ASSERT(ctx->new_table->get_ref_count() == 1);
+			DBUG_ASSERT(ctx->new_table->get_ref_count() == 0);
 			DBUG_ASSERT(ctx->new_table->id != 0);
-			DBUG_ASSERT(ctx->new_table->id == ctx->trx->table_id);
 			break;
 		case DB_TABLESPACE_EXISTS:
 			my_error(ER_TABLESPACE_EXISTS, MYF(0),
@@ -6949,7 +6929,6 @@ error_handling_drop_uncached_1:
 		}
 	} else if (ctx->num_to_add_index) {
 		ut_ad(!ctx->is_instant());
-		ctx->trx->table_id = user_table->id;
 
 		for (ulint a = 0; a < ctx->num_to_add_index; a++) {
 			dict_index_t* index = ctx->add_index[a];
@@ -7047,21 +7026,7 @@ error_handling_drop_uncached:
 	}
 
 	if (fts_index) {
-		/* Ensure that the dictionary operation mode will
-		not change while creating the auxiliary tables. */
-		trx_dict_op_t	op = trx_get_dict_operation(ctx->trx);
-
-#ifdef UNIV_DEBUG
-		switch (op) {
-		case TRX_DICT_OP_NONE:
-			break;
-		case TRX_DICT_OP_TABLE:
-		case TRX_DICT_OP_INDEX:
-			goto op_ok;
-		}
-		ut_error;
-op_ok:
-#endif /* UNIV_DEBUG */
+		ut_ad(ctx->trx->dict_operation);
 		ut_ad(ctx->trx->dict_operation_lock_mode == RW_X_LATCH);
 		ut_d(dict_sys.assert_locked());
 
@@ -7086,9 +7051,6 @@ op_ok:
 			goto error_handling;
 		}
 
-		ctx->trx->commit();
-		trx_start_for_ddl(ctx->trx, op);
-
 		if (!ctx->new_table->fts
 		    || ib_vector_size(ctx->new_table->fts->indexes) == 0) {
 			error = fts_create_common_tables(
@@ -7114,40 +7076,38 @@ op_ok:
 				goto error_handling;
 			}
 		}
-
-		ut_ad(trx_get_dict_operation(ctx->trx) == op);
 	}
 
 	DBUG_ASSERT(error == DB_SUCCESS);
 
-	/* Commit the data dictionary transaction in order to release
-	the table locks on the system tables.  This means that if
-	MySQL crashes while creating a new primary key inside
-	row_merge_build_indexes(), ctx->new_table will not be dropped
-	by trx_rollback_active().  It will have to be recovered or
-	dropped by the database administrator. */
-	trx_commit_for_mysql(ctx->trx);
+	if (UT_LIST_GET_LEN(ctx->trx->lock.trx_locks)) {
+		/* Commit the data dictionary transaction in order to release
+		the table locks on the system tables.  This means that if
+		MariaDB is killed while rebuilding the table inside
+		row_merge_build_indexes(), ctx->new_table will not be dropped
+		by trx_rollback_active(). */
+		trx_commit_for_mysql(ctx->trx);
+		trx_start_for_ddl(ctx->trx);
+
+		if (ctx->need_rebuild()) {
+			ctx->new_table->acquire();
+		}
+	}
 
+	ut_d(dict_table_check_for_dup_indexes(user_table, CHECK_PARTIAL_OK));
 	row_mysql_unlock_data_dictionary(ctx->trx);
-	dict_locked = false;
 
 	if (ctx->old_table->fts) {
 		fts_sync_during_ddl(ctx->old_table);
 	}
 
+	DBUG_RETURN(false);
+
 error_handling:
 	/* After an error, remove all those index definitions from the
 	dictionary which were defined. */
 
 	switch (error) {
-	case DB_SUCCESS:
-		ut_a(!dict_locked);
-
-		ut_d(dict_sys.mutex_lock());
-		ut_d(dict_table_check_for_dup_indexes(
-			     user_table, CHECK_PARTIAL_OK));
-		ut_d(dict_sys.mutex_unlock());
-		DBUG_RETURN(false);
 	case DB_TABLESPACE_EXISTS:
 		my_error(ER_TABLESPACE_EXISTS, MYF(0), "(unknown)");
 		break;
@@ -7161,59 +7121,48 @@ error_handling:
 		my_error_innodb(error, table_name, user_table->flags);
 	}
 
+	ctx->trx->rollback();
+
 error_handled:
 
 	ctx->prebuilt->trx->error_info = NULL;
-
-	if (!ctx->trx) {
-		goto err_exit;
-	}
-
 	ctx->trx->error_state = DB_SUCCESS;
 
 	if (!dict_locked) {
 		row_mysql_lock_data_dictionary(ctx->trx);
+		if (table_lock_failed) {
+			goto err_exit;
+		}
 	}
 
-	if (new_clustered) {
-		if (ctx->need_rebuild()) {
-
-			if (DICT_TF2_FLAG_IS_SET(
-				    ctx->new_table, DICT_TF2_FTS)) {
-				innobase_drop_fts_index_table(
-					ctx->new_table, ctx->trx);
-			}
-
-			dict_table_close_and_drop(ctx->trx, ctx->new_table);
-
-			/* Free the log for online table rebuild, if
-			one was allocated. */
-
-			dict_index_t* clust_index = dict_table_get_first_index(
-				user_table);
+	if (ctx->need_rebuild()) {
+		/* Free the log for online table rebuild, if
+		one was allocated. */
 
-			clust_index->lock.x_lock(SRW_LOCK_CALL);
+		dict_index_t* clust_index = dict_table_get_first_index(
+			user_table);
 
-			if (clust_index->online_log) {
-				ut_ad(ctx->online);
-				row_log_abort_sec(clust_index);
-				clust_index->online_status
-					= ONLINE_INDEX_COMPLETE;
-			}
+		clust_index->lock.x_lock(SRW_LOCK_CALL);
 
-			clust_index->lock.x_unlock();
+		if (clust_index->online_log) {
+			ut_ad(ctx->online);
+			row_log_abort_sec(clust_index);
+			clust_index->online_status
+				= ONLINE_INDEX_COMPLETE;
 		}
 
-		trx_commit_for_mysql(ctx->trx);
-		/* n_ref_count must be 1, because purge cannot
-		be executing on this very table as we are
-		holding dict_sys.latch X-latch. */
-		ut_ad(!stats_wait || ctx->online
-		      || user_table->get_ref_count() == 1);
+		clust_index->lock.x_unlock();
+	}
 
+	/* n_ref_count must be 1, because purge cannot
+	be executing on this very table as we are
+	holding dict_sys.latch X-latch. */
+	ut_ad(!stats_wait || ctx->online || user_table->get_ref_count() == 1);
+
+	if (new_clustered) {
 		online_retry_drop_indexes_with_trx(user_table, ctx->trx);
 	} else {
-		ut_ad(!ctx->need_rebuild());
+		trx_start_for_ddl(ctx->trx);
 		row_merge_drop_indexes(ctx->trx, user_table, true);
 		trx_commit_for_mysql(ctx->trx);
 	}
@@ -7231,7 +7180,7 @@ err_exit:
 
 	if (ctx->trx) {
 		row_mysql_unlock_data_dictionary(ctx->trx);
-
+		ctx->trx->rollback();
 		ctx->trx->free();
 	}
 	trx_commit_for_mysql(ctx->prebuilt->trx);
@@ -8737,25 +8686,20 @@ rollback_inplace_alter_table(
 
 	DBUG_ENTER("rollback_inplace_alter_table");
 
-	if (!ctx || !ctx->trx) {
+	if (!ctx) {
 		/* If we have not started a transaction yet,
 		(almost) nothing has been or needs to be done. */
 		goto func_exit;
 	}
 
-	trx_start_for_ddl(ctx->trx, ctx->need_rebuild()
-			  ? TRX_DICT_OP_TABLE : TRX_DICT_OP_INDEX);
 	row_mysql_lock_data_dictionary(ctx->trx);
+	ctx->trx->dict_operation = true;
 
-	if (ctx->need_rebuild()) {
+	if (!ctx->new_table) {
+	} else if (ctx->need_rebuild()) {
 		/* DML threads can access ctx->new_table via the
 		online rebuild log. Free it first. */
 		innobase_online_rebuild_log_free(prebuilt->table);
-	}
-
-	if (!ctx->new_table) {
-		ut_ad(ctx->need_rebuild());
-	} else if (ctx->need_rebuild()) {
 		dberr_t	err= DB_SUCCESS;
 		ulint	flags	= ctx->new_table->flags;
 
@@ -8775,12 +8719,12 @@ rollback_inplace_alter_table(
 			}
 		}
 
-		dict_table_close_and_drop(ctx->trx, ctx->new_table);
-
-		switch (err) {
-		case DB_SUCCESS:
-			break;
-		default:
+		ut_d(const bool last_handle=) ctx->new_table->release();
+		ut_ad(last_handle);
+		err = row_drop_table_for_mysql(ctx->new_table->name.m_name,
+					       ctx->trx, SQLCOM_DROP_TABLE,
+					       false, false);
+		if (err != DB_SUCCESS) {
 			my_error_innodb(err, table->s->table_name.str,
 					flags);
 			fail = true;
@@ -8899,7 +8843,7 @@ innobase_drop_foreign_try(
 {
 	DBUG_ENTER("innobase_drop_foreign_try");
 
-	DBUG_ASSERT(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_d(dict_sys.assert_locked());
 
@@ -8955,7 +8899,7 @@ innobase_rename_column_try(
 
 	DBUG_ENTER("innobase_rename_column_try");
 
-	DBUG_ASSERT(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_d(dict_sys.assert_locked());
 
@@ -9273,7 +9217,7 @@ innobase_rename_or_enlarge_column_try(
 	DBUG_ENTER("innobase_rename_or_enlarge_column_try");
 	DBUG_ASSERT(!ctx->need_rebuild());
 
-	DBUG_ASSERT(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_d(dict_sys.assert_locked());
 
@@ -10295,57 +10239,53 @@ commit_try_norebuild(
 	}
 
 	dberr_t	error;
+	dict_index_t* index;
+	const char *op = "rename index to add";
+	ulint i;
 
 	/* We altered the table in place. Mark the indexes as committed. */
-	for (ulint i = 0; i < ctx->num_to_add_index; i++) {
-		dict_index_t*	index = ctx->add_index[i];
+	for (i = 0; i < ctx->num_to_add_index; i++) {
+		index = ctx->add_index[i];
 		DBUG_ASSERT(dict_index_get_online_status(index)
 			    == ONLINE_INDEX_COMPLETE);
 		DBUG_ASSERT(!index->is_committed());
 		error = row_merge_rename_index_to_add(
 			trx, ctx->new_table->id, index->id);
+handle_error:
 		switch (error) {
 		case DB_SUCCESS:
 			break;
 		case DB_TOO_MANY_CONCURRENT_TRXS:
-			/* If we wrote some undo log here, then the
-			persistent data dictionary for this table may
-			probably be corrupted. This is because a
-			'trigger' on SYS_INDEXES could already have invoked
-			btr_free_if_exists(), which cannot be rolled back. */
-			DBUG_ASSERT(trx->undo_no == 0);
 			my_error(ER_TOO_MANY_CONCURRENT_TRXS, MYF(0));
 			DBUG_RETURN(true);
 		default:
-			sql_print_error(
-				"InnoDB: rename index to add: %lu\n",
-				(ulong) error);
+			sql_print_error("InnoDB: %s: %s\n", op,
+					ut_strerr(error));
 			DBUG_ASSERT(0);
-			my_error(ER_INTERNAL_ERROR, MYF(0),
-				 "rename index to add");
+			my_error(ER_INTERNAL_ERROR, MYF(0), op);
 			DBUG_RETURN(true);
 		}
 	}
 
-	/* Drop any indexes that were requested to be dropped.
-	Flag them in the data dictionary first. */
-
-	for (ulint i = 0; i < ctx->num_to_drop_index; i++) {
-		dict_index_t*	index = ctx->drop_index[i];
+	for (i = 0; i < ctx->num_to_drop_index; i++) {
+		index = ctx->drop_index[i];
 		DBUG_ASSERT(index->is_committed());
 		DBUG_ASSERT(index->table == ctx->new_table);
 		DBUG_ASSERT(index->to_be_dropped);
+		op = "DROP INDEX";
+
+		static const char drop_index[] =
+			"PROCEDURE DROP_INDEX_PROC () IS\n"
+			"BEGIN\n"
+			"DELETE FROM SYS_FIELDS WHERE INDEX_ID=:indexid;\n"
+			"DELETE FROM SYS_INDEXES WHERE ID=:indexid;\n"
+			"END;\n";
 
-		error = row_merge_rename_index_to_drop(
-			trx, index->table->id, index->id);
+		pars_info_t* info = pars_info_create();
+		pars_info_add_ull_literal(info, "indexid", index->id);
+		error = que_eval_sql(info, drop_index, FALSE, trx);
 		if (error != DB_SUCCESS) {
-			sql_print_error(
-				"InnoDB: rename index to drop: %lu\n",
-				(ulong) error);
-			DBUG_ASSERT(0);
-			my_error(ER_INTERNAL_ERROR, MYF(0),
-				 "rename index to drop");
-			DBUG_RETURN(true);
+			goto handle_error;
 		}
 	}
 
@@ -10520,15 +10460,6 @@ commit_cache_norebuild(
 	}
 
 	if (ctx->num_to_drop_index) {
-		/* Really drop the indexes that were dropped.
-		The transaction had to be committed first
-		(after renaming the indexes), so that in the
-		event of a crash, crash recovery will drop the
-		indexes, because it drops all indexes whose
-		names start with TEMP_INDEX_PREFIX_STR. Once we
-		have started dropping an index tree, there is
-		no way to roll it back. */
-
 		for (ulint i = 0; i < ctx->num_to_drop_index; i++) {
 			dict_index_t*	index = ctx->drop_index[i];
 			DBUG_ASSERT(index->is_committed());
@@ -10550,8 +10481,7 @@ commit_cache_norebuild(
 			index->lock.u_unlock();
 		}
 
-		trx_start_for_ddl(trx, TRX_DICT_OP_INDEX);
-		row_merge_drop_indexes_dict(trx, ctx->new_table->id);
+		trx_start_for_ddl(trx);
 
 		for (ulint i = 0; i < ctx->num_to_drop_index; i++) {
 			dict_index_t*	index = ctx->drop_index[i];
@@ -10979,11 +10909,13 @@ ha_innobase::commit_inplace_alter_table(
 	ha_alter_info->group_commit_ctx = NULL;
 
 	trx_start_if_not_started_xa(m_prebuilt->trx, true);
+	const bool new_clustered = ctx0->need_rebuild();
 
 	for (inplace_alter_handler_ctx** pctx = ctx_array; *pctx; pctx++) {
 		ha_innobase_inplace_ctx*	ctx
 			= static_cast<ha_innobase_inplace_ctx*>(*pctx);
 		DBUG_ASSERT(ctx->prebuilt->trx == m_prebuilt->trx);
+		DBUG_ASSERT(new_clustered == ctx->need_rebuild());
 
 		/* If decryption failed for old table or new table
 		fail here. */
@@ -10998,18 +10930,19 @@ ha_innobase::commit_inplace_alter_table(
 			DBUG_RETURN(true);
 		}
 
+		if (new_clustered) {
+			continue;
+		}
+
 		/* Exclusively lock the table, to ensure that no other
 		transaction is holding locks on the table while we
-		change the table definition. The MySQL meta-data lock
-		should normally guarantee that no conflicting locks
-		exist. However, FOREIGN KEY constraints checks and any
-		transactions collected during crash recovery could be
-		holding InnoDB locks only, not MySQL locks. */
-
-		dberr_t error = row_merge_lock_table(
-			m_prebuilt->trx, ctx->old_table, LOCK_X);
+		change the table definition. Any recovered incomplete
+		transactions would be holding InnoDB locks only, not MDL. */
+		ctx->prebuilt->trx->op_info = "acquiring table lock";
 
-		if (error != DB_SUCCESS) {
+		if (dberr_t error = lock_table_for_trx(ctx->new_table,
+						       ctx->prebuilt->trx,
+						       LOCK_X)) {
 			my_error_innodb(
 				error, table_share->table_name.str, 0);
 			DBUG_RETURN(true);
@@ -11018,7 +10951,6 @@ ha_innobase::commit_inplace_alter_table(
 
 	DEBUG_SYNC(m_user_thd, "innodb_alter_commit_after_lock_table");
 
-	const bool	new_clustered	= ctx0->need_rebuild();
 	trx_t*		trx		= ctx0->trx;
 	bool		fail		= false;
 
@@ -11057,15 +10989,12 @@ ha_innobase::commit_inplace_alter_table(
 		}
 	}
 
-	if (!trx) {
-		DBUG_ASSERT(!new_clustered);
-		trx = innobase_trx_allocate(m_user_thd);
-	}
-
-	trx_start_for_ddl(trx, TRX_DICT_OP_INDEX);
 	/* Latch the InnoDB data dictionary exclusively so that no deadlocks
 	or lock waits can happen in it during the data dictionary operation. */
 	row_mysql_lock_data_dictionary(trx);
+	if (trx->state != TRX_STATE_ACTIVE) {
+		trx_start_for_ddl(trx);
+	}
 
 	/* Prevent the background statistics collection from accessing
 	the tables. */
@@ -11159,8 +11088,6 @@ ha_innobase::commit_inplace_alter_table(
 				= static_cast<ha_innobase_inplace_ctx*>(*pctx);
 			ctx->rollback_instant();
 		}
-	} else if (!new_clustered) {
-		trx_commit_for_mysql(trx);
 	} else {
 		/* Test what happens on crash if the redo logs
 		are flushed to disk here. The log records
@@ -11172,29 +11099,15 @@ ha_innobase::commit_inplace_alter_table(
 				DBUG_SUICIDE(););
 		ut_ad(!trx->fts_trx);
 
-		if (fail) {
-			trx_rollback_for_mysql(trx);
-		} else {
-			ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE));
-			ut_ad(trx->has_logged());
-			trx->commit();
-		}
+		ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE));
+		ut_ad(!new_clustered || trx->has_logged());
 
-		/* If server crashes here, the dictionary in
-		InnoDB and MySQL will differ.  The .ibd files
-		and the .frm files must be swapped manually by
-		the administrator. No loss of data. */
+		trx->commit();
+		log_write_up_to(trx->commit_lsn, true);
 		DBUG_EXECUTE_IF("innodb_alter_commit_crash_after_commit",
-				log_buffer_flush_to_disk();
 				DBUG_SUICIDE(););
 	}
 
-	/* Flush the log to reduce probability that the .frm files and
-	the InnoDB data dictionary get out-of-sync if the user runs
-	with innodb_flush_log_at_trx_commit = 0 */
-
-	log_buffer_flush_to_disk();
-
 	/* At this point, the changes to the persistent storage have
 	been committed or rolled back. What remains to be done is to
 	update the in-memory structures, close some handles, release
@@ -11213,9 +11126,14 @@ ha_innobase::commit_inplace_alter_table(
 
 		if (fail) {
 			if (new_clustered) {
-				trx_start_for_ddl(trx, TRX_DICT_OP_TABLE);
+				trx_start_for_ddl(trx);
 
-				dict_table_close_and_drop(trx, ctx->new_table);
+				ut_d(const bool last_handle=)
+					ctx->new_table->release();
+				ut_ad(last_handle);
+				row_drop_table_for_mysql(
+					ctx->new_table->name.m_name,
+					trx, SQLCOM_DROP_TABLE, false, false);
 
 				trx_commit_for_mysql(trx);
 				ctx->new_table = NULL;
@@ -11224,7 +11142,7 @@ ha_innobase::commit_inplace_alter_table(
 				Roll back any ADD INDEX, or get rid of garbage
 				ADD INDEX that was left over from a previous
 				ALTER TABLE statement. */
-				trx_start_for_ddl(trx, TRX_DICT_OP_INDEX);
+				trx_start_for_ddl(trx);
 				innobase_rollback_sec_index(
 					ctx->new_table, table, TRUE, trx);
 				trx_commit_for_mysql(trx);
@@ -11329,6 +11247,7 @@ foreign_fail:
 			= static_cast<ha_innobase_inplace_ctx*>(*pctx);
 
 		if (ctx->trx) {
+			ctx->trx->rollback();
 			ctx->trx->free();
 			ctx->trx = NULL;
 		}
@@ -11439,10 +11358,10 @@ foreign_fail:
 			transaction commit.  If the system crashes
 			before this is completed, some orphan tables
 			with ctx->tmp_name may be recovered. */
-			trx_start_for_ddl(trx, TRX_DICT_OP_TABLE);
-			dberr_t error = row_merge_drop_table(trx, ctx->old_table);
-
-			if (UNIV_UNLIKELY(error != DB_SUCCESS)) {
+			trx_start_for_ddl(trx);
+			if (dberr_t error = row_drop_table_for_mysql(
+				ctx->old_table->name.m_name,
+				trx, SQLCOM_DROP_TABLE, false, false)) {
 				ib::error() << "Inplace alter table " << ctx->old_table->name
 					    << " dropping copy of the old table failed error "
 					    << error
diff --git a/storage/innobase/include/dict0crea.h b/storage/innobase/include/dict0crea.h
index 9f8e5442e52..5fe35d14cd4 100644
--- a/storage/innobase/include/dict0crea.h
+++ b/storage/innobase/include/dict0crea.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, 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
@@ -99,9 +99,11 @@ dict_create_index_tree(
 /** Drop the index tree associated with a row in SYS_INDEXES table.
 @param[in,out]	pcur	persistent cursor on rec
 @param[in,out]	trx	dictionary transaction
+@param[in,out]	table	table that the record belongs to
 @param[in,out]	mtr	mini-transaction */
-void dict_drop_index_tree(btr_pcur_t* pcur, trx_t* trx, mtr_t* mtr)
-	MY_ATTRIBUTE((nonnull(1,3)));
+void dict_drop_index_tree(btr_pcur_t *pcur, trx_t *trx, dict_table_t *table,
+                          mtr_t *mtr)
+	MY_ATTRIBUTE((nonnull(1,4)));
 
 /***************************************************************//**
 Creates an index tree for the index if it is not a member of a cluster.
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index 02ed956b039..ee6c94e9433 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -170,17 +170,6 @@ dict_table_close(
 	MDL_ticket*	mdl = NULL);
 
 /*********************************************************************//**
-Closes the only open handle to a table and drops a table while assuring
-that dict_sys.mutex is held the whole time.  This assures that the table
-is not evicted after the close when the count of open handles goes to zero.
-Because dict_sys.mutex is held, we do not need to call prevent_eviction(). */
-void
-dict_table_close_and_drop(
-/*======================*/
-	trx_t*		trx,		/*!< in: data dictionary transaction */
-	dict_table_t*	table);		/*!< in/out: table */
-
-/*********************************************************************//**
 Gets the minimum number of bytes per character.
 @return minimum multi-byte char size, in bytes */
 UNIV_INLINE
diff --git a/storage/innobase/include/fts0fts.h b/storage/innobase/include/fts0fts.h
index 6c97c9ed868..d13c1a6d3c1 100644
--- a/storage/innobase/include/fts0fts.h
+++ b/storage/innobase/include/fts0fts.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2011, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, 2020, MariaDB Corporation.
+Copyright (c) 2016, 2021, 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
@@ -725,10 +725,6 @@ fts_savepoint_rollback_last_stmt(
 /*=============================*/
 	trx_t*		trx);			/*!< in: transaction */
 
-/** Drop all orphaned FTS auxiliary tables, those that don't have a parent
-table or FTS index defined on them. */
-void fts_drop_orphaned_tables();
-
 /** Run SYNC on the table, i.e., write out data from the cache to the
 FTS auxiliary INDEX table and clear the cache at the end.
 @param[in,out]	table		fts table
diff --git a/storage/innobase/include/row0merge.h b/storage/innobase/include/row0merge.h
index 1d7f9bb145b..262df6ad1c0 100644
--- a/storage/innobase/include/row0merge.h
+++ b/storage/innobase/include/row0merge.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2005, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2020, MariaDB Corporation.
+Copyright (c) 2015, 2021, 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
@@ -146,17 +146,6 @@ row_merge_dup_report(
 	MY_ATTRIBUTE((nonnull));
 
 /*********************************************************************//**
-Sets an exclusive lock on a table, for the duration of creating indexes.
-@return error code or DB_SUCCESS */
-dberr_t
-row_merge_lock_table(
-/*=================*/
-	trx_t*		trx,		/*!< in/out: transaction */
-	dict_table_t*	table,		/*!< in: table to lock */
-	enum lock_mode	mode)		/*!< in: LOCK_X or LOCK_S */
-	MY_ATTRIBUTE((nonnull(1,2), warn_unused_result));
-
-/*********************************************************************//**
 Drop indexes that were created before an error occurred.
 The data dictionary must have been locked exclusively by the caller,
 because the transaction will not be committed. */
@@ -217,19 +206,6 @@ row_merge_rename_index_to_add(
 	index_id_t	index_id)	/*!< in: index identifier */
 	MY_ATTRIBUTE((nonnull(1), warn_unused_result));
 
-/*********************************************************************//**
-Rename an index in the dictionary that is to be dropped. The data
-dictionary must have been locked exclusively by the caller, because
-the transaction will not be committed.
-@return DB_SUCCESS if all OK */
-dberr_t
-row_merge_rename_index_to_drop(
-/*===========================*/
-	trx_t*		trx,		/*!< in/out: transaction */
-	table_id_t	table_id,	/*!< in: table identifier */
-	index_id_t	index_id)	/*!< in: index identifier */
-	MY_ATTRIBUTE((nonnull(1), warn_unused_result));
-
 /** Create the index and load in to the dictionary.
 @param[in,out]	table		the index is on this table
 @param[in]	index_def	the index definition
@@ -253,19 +229,6 @@ row_merge_is_index_usable(
 	const dict_index_t*	index)	/*!< in: index to check */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
-/*********************************************************************//**
-Drop a table. The caller must have ensured that the background stats
-thread is not processing the table. This can be done by calling
-dict_stats_wait_bg_to_stop_using_table() after locking the dictionary and
-before calling this function.
-@return DB_SUCCESS or error code */
-dberr_t
-row_merge_drop_table(
-/*=================*/
-	trx_t*		trx,		/*!< in: transaction */
-	dict_table_t*	table)		/*!< in: table instance to drop */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
-
 /** Build indexes on a table by reading a clustered index, creating a temporary
 file containing index entries, merge sorting these index entries and inserting
 sorted index entries to indexes.
diff --git a/storage/innobase/include/trx0rec.h b/storage/innobase/include/trx0rec.h
index b377847c73a..a7f517a086b 100644
--- a/storage/innobase/include/trx0rec.h
+++ b/storage/innobase/include/trx0rec.h
@@ -285,9 +285,22 @@ trx_undo_read_v_idx(
 compilation info multiplied by 16 is ORed to this value in an undo log
 record */
 
-#define	TRX_UNDO_RENAME_TABLE	9	/*!< RENAME TABLE */
-#define	TRX_UNDO_INSERT_METADATA 10	/*!< insert a metadata
-					pseudo-record for instant ALTER */
+/** Undo log records for DDL operations
+
+Note: special rollback and purge triggers exist for SYS_INDEXES records:
+@see dict_drop_index_tree() */
+enum trx_undo_ddl_type
+{
+  /** RENAME TABLE (logging the old table name).
+
+  Because SYS_TABLES has PRIMARY KEY(NAME), the row-level undo log records
+  for SYS_TABLES cannot be distinguished from DROP TABLE, CREATE TABLE. */
+  TRX_UNDO_RENAME_TABLE= 9,
+  /** insert a metadata pseudo-record for instant ALTER TABLE */
+  TRX_UNDO_INSERT_METADATA= 10
+};
+
+/* DML operations */
 #define	TRX_UNDO_INSERT_REC	11	/* fresh insert into clustered index */
 #define	TRX_UNDO_UPD_EXIST_REC	12	/* update of a non-delete-marked
 					record */
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 51dfe143248..29672a4eebd 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -151,25 +151,20 @@ trx_start_internal_read_only_low(
 	trx_start_if_not_started_xa_low((t), (rw))
 #endif /* UNIV_DEBUG */
 
-/*************************************************************//**
-Starts the transaction for a DDL operation. */
-void
-trx_start_for_ddl_low(
-/*==================*/
-	trx_t*		trx,	/*!< in/out: transaction */
-	trx_dict_op_t	op);	/*!< in: dictionary operation type */
+/** Start a transaction for a DDL operation.
+@param trx   transaction */
+void trx_start_for_ddl_low(trx_t *trx);
 
 #ifdef UNIV_DEBUG
-#define trx_start_for_ddl(t, o)					\
+# define trx_start_for_ddl(t)					\
 	do {							\
 	ut_ad((t)->start_file == 0);				\
 	(t)->start_line = __LINE__;				\
 	(t)->start_file = __FILE__;				\
-	trx_start_for_ddl_low((t), (o));			\
+	trx_start_for_ddl_low(t);				\
 	} while (0)
 #else
-#define trx_start_for_ddl(t, o)					\
-	trx_start_for_ddl_low((t), (o))
+# define trx_start_for_ddl(t) trx_start_for_ddl_low(t)
 #endif /* UNIV_DEBUG */
 
 /**********************************************************************//**
@@ -274,25 +269,6 @@ trx_print(
 					or 0 to use the default max length */
 
 /**********************************************************************//**
-Determine if a transaction is a dictionary operation.
-@return dictionary operation mode */
-UNIV_INLINE
-enum trx_dict_op_t
-trx_get_dict_operation(
-/*===================*/
-	const trx_t*	trx)	/*!< in: transaction */
-	MY_ATTRIBUTE((warn_unused_result));
-/**********************************************************************//**
-Flag a transaction a dictionary operation. */
-UNIV_INLINE
-void
-trx_set_dict_operation(
-/*===================*/
-	trx_t*			trx,	/*!< in/out: transaction */
-	enum trx_dict_op_t	op);	/*!< in: operation, not
-					TRX_DICT_OP_NONE */
-
-/**********************************************************************//**
 Determines if a transaction is in the given state.
 The caller must hold trx->mutex, or it must be the thread
 that is serving a running transaction.
@@ -818,8 +794,8 @@ public:
 					flush the log in
 					trx_commit_complete_for_mysql() */
 	ulint		duplicates;	/*!< TRX_DUP_IGNORE | TRX_DUP_REPLACE */
-	trx_dict_op_t	dict_operation;	/**< @see enum trx_dict_op_t */
-
+	bool		dict_operation;	/**< whether this modifies InnoDB
+					data dictionary */
 	ib_uint32_t	dict_operation_lock_mode;
 					/*!< 0, RW_S_LATCH, or RW_X_LATCH:
 					the latch mode trx currently holds
@@ -832,8 +808,6 @@ public:
 	/** microsecond_interval_timer() of transaction start */
 	ulonglong	start_time_micro;
 	lsn_t		commit_lsn;	/*!< lsn at the time of the commit */
-	table_id_t	table_id;	/*!< Table to drop iff dict_operation
-					== TRX_DICT_OP_TABLE, or 0. */
 	/*------------------------------*/
 	THD*		mysql_thd;	/*!< MySQL thread handle corresponding
 					to this trx, or NULL */
@@ -928,8 +902,6 @@ public:
 					count of tables being flushed. */
 
 	/*------------------------------*/
-	bool		ddl;		/*!< true if it is an internal
-					transaction for DDL */
 	bool		internal;	/*!< true if it is a system/internal
 					transaction background task. This
 					includes DDL transactions too.  Such
@@ -1070,7 +1042,7 @@ public:
     ut_ad(lock.table_locks.empty());
     ut_ad(!autoinc_locks || ib_vector_is_empty(autoinc_locks));
     ut_ad(UT_LIST_GET_LEN(lock.evicted_tables) == 0);
-    ut_ad(dict_operation == TRX_DICT_OP_NONE);
+    ut_ad(!dict_operation);
   }
 
   /** This has to be invoked on SAVEPOINT or at the end of a statement.
diff --git a/storage/innobase/include/trx0trx.ic b/storage/innobase/include/trx0trx.ic
index a7ab040a9d9..4b566998a90 100644
--- a/storage/innobase/include/trx0trx.ic
+++ b/storage/innobase/include/trx0trx.ic
@@ -80,62 +80,3 @@ trx_get_error_info(
 {
 	return(trx->error_info);
 }
-
-/**********************************************************************//**
-Determine if a transaction is a dictionary operation.
-@return dictionary operation mode */
-UNIV_INLINE
-enum trx_dict_op_t
-trx_get_dict_operation(
-/*===================*/
-	const trx_t*	trx)	/*!< in: transaction */
-{
-	trx_dict_op_t op = static_cast<trx_dict_op_t>(trx->dict_operation);
-
-#ifdef UNIV_DEBUG
-	switch (op) {
-	case TRX_DICT_OP_NONE:
-	case TRX_DICT_OP_TABLE:
-	case TRX_DICT_OP_INDEX:
-		return(op);
-	}
-	ut_error;
-#endif /* UNIV_DEBUG */
-	return(op);
-}
-/**********************************************************************//**
-Flag a transaction a dictionary operation. */
-UNIV_INLINE
-void
-trx_set_dict_operation(
-/*===================*/
-	trx_t*			trx,	/*!< in/out: transaction */
-	enum trx_dict_op_t	op)	/*!< in: operation, not
-					TRX_DICT_OP_NONE */
-{
-#ifdef UNIV_DEBUG
-	enum trx_dict_op_t	old_op = trx_get_dict_operation(trx);
-
-	switch (op) {
-	case TRX_DICT_OP_NONE:
-		ut_error;
-		break;
-	case TRX_DICT_OP_TABLE:
-		switch (old_op) {
-		case TRX_DICT_OP_NONE:
-		case TRX_DICT_OP_INDEX:
-		case TRX_DICT_OP_TABLE:
-			goto ok;
-		}
-		ut_error;
-		break;
-	case TRX_DICT_OP_INDEX:
-		ut_ad(old_op == TRX_DICT_OP_NONE);
-		break;
-	}
-ok:
-#endif /* UNIV_DEBUG */
-
-	trx->ddl = true;
-	trx->dict_operation = op;
-}
diff --git a/storage/innobase/include/trx0types.h b/storage/innobase/include/trx0types.h
index dbabf7754fb..edc6cf33e93 100644
--- a/storage/innobase/include/trx0types.h
+++ b/storage/innobase/include/trx0types.h
@@ -61,21 +61,6 @@ enum trx_state_t {
 	TRX_STATE_COMMITTED_IN_MEMORY
 };
 
-/** Type of data dictionary operation */
-enum trx_dict_op_t {
-	/** The transaction is not modifying the data dictionary. */
-	TRX_DICT_OP_NONE = 0,
-	/** The transaction is creating a table or an index, or
-	dropping a table.  The table must be dropped in crash
-	recovery.  This and TRX_DICT_OP_NONE are the only possible
-	operation modes in crash recovery. */
-	TRX_DICT_OP_TABLE = 1,
-	/** The transaction is creating or dropping an index in an
-	existing table.  In crash recovery, the data dictionary
-	must be locked, but the table must not be dropped. */
-	TRX_DICT_OP_INDEX = 2
-};
-
 /** Memory objects */
 /* @{ */
 /** Transaction */
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index e032c53263a..c89afffd983 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, 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
@@ -301,9 +301,7 @@ struct trx_undo_t {
 					log */
 	XID		xid;		/*!< X/Open XA transaction
 					identification */
-	ibool		dict_operation;	/*!< TRUE if a dict operation trx */
-	table_id_t	table_id;	/*!< if a dict operation, then the table
-					id */
+	bool		dict_operation;	/*!< TRUE if a dict operation trx */
 	trx_rseg_t*	rseg;		/*!< rseg where the undo log belongs */
 	/*-----------------------------*/
 	uint32_t	hdr_page_no;	/*!< page number of the header page in
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 6261154e32b..d46036b9243 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -1256,11 +1256,7 @@ lock_rec_enqueue_waiting(
 	trx_t* trx = thr_get_trx(thr);
 	ut_ad(trx->mutex_is_owner());
 
-	switch (trx_get_dict_operation(trx)) {
-	case TRX_DICT_OP_NONE:
-		break;
-	case TRX_DICT_OP_TABLE:
-	case TRX_DICT_OP_INDEX:
+	if (UNIV_UNLIKELY(trx->dict_operation)) {
 		ib::error() << "A record lock wait happens in a dictionary"
 			" operation. index "
 			<< index->name
@@ -3309,11 +3305,7 @@ lock_table_enqueue_waiting(
 	trx_t* trx = thr_get_trx(thr);
 	ut_ad(trx->mutex_is_owner());
 
-	switch (trx_get_dict_operation(trx)) {
-	case TRX_DICT_OP_NONE:
-		break;
-	case TRX_DICT_OP_TABLE:
-	case TRX_DICT_OP_INDEX:
+	if (UNIV_UNLIKELY(trx->dict_operation)) {
 		ib::error() << "A table lock wait happens in a dictionary"
 			" operation. Table " << table->name
 			<< ". " << BUG_REPORT_MSG;
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 3060ec05bc8..5ad7de3fb8c 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -3927,19 +3927,13 @@ row_import_for_mysql(
 
 	trx = trx_create();
 
-	/* So that the table is not DROPped during recovery. */
-	trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);
+	trx->dict_operation = true;
 
 	trx_start_if_not_started(trx, true);
 
 	/* So that we can send error messages to the user. */
 	trx->mysql_thd = prebuilt->trx->mysql_thd;
 
-	/* Ensure that the table will be dropped by trx_rollback_active()
-	in case of a crash. */
-
-	trx->table_id = table->id;
-
 	/* Assign an undo segment for the transaction, so that the
 	transaction will be recovered after a crash. */
 
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 453b260e894..6cb1ad64c88 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -2666,7 +2666,7 @@ commit_exit:
 	    && page_is_empty(block->frame)
 	    && !entry->is_metadata() && !trx->duplicates
 	    && !trx->check_unique_secondary && !trx->check_foreigns
-	    && !trx->ddl && !trx->internal
+	    && !trx->dict_operation && !trx->internal
 	    && block->page.id().page_no() == index->page
 	    && !index->table->skip_alter_undo
 	    && !index->table->n_rec_locks
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 44d857fb158..3726be84653 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -3658,25 +3658,6 @@ err_exit:
 }
 
 /*********************************************************************//**
-Sets an exclusive lock on a table, for the duration of creating indexes.
-@return error code or DB_SUCCESS */
-dberr_t
-row_merge_lock_table(
-/*=================*/
-	trx_t*		trx,		/*!< in/out: transaction */
-	dict_table_t*	table,		/*!< in: table to lock */
-	enum lock_mode	mode)		/*!< in: LOCK_X or LOCK_S */
-{
-	ut_ad(!srv_read_only_mode);
-	ut_ad(mode == LOCK_X || mode == LOCK_S);
-
-	trx->op_info = "setting table lock for creating or dropping index";
-	trx->ddl = true;
-
-	return(lock_table_for_trx(table, trx, mode));
-}
-
-/*********************************************************************//**
 Drop an index that was created before an error occurred.
 The data dictionary must have been locked exclusively by the caller,
 because the transaction will not be committed. */
@@ -3698,7 +3679,7 @@ row_merge_drop_index_dict(
 
 	ut_ad(!srv_read_only_mode);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_ad(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	ut_ad(trx->dict_operation);
 	ut_d(dict_sys.assert_locked());
 
 	info = pars_info_create();
@@ -3760,7 +3741,7 @@ row_merge_drop_indexes_dict(
 
 	ut_ad(!srv_read_only_mode);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_ad(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	ut_ad(trx->dict_operation);
 	ut_d(dict_sys.assert_locked());
 
 	/* It is possible that table->n_ref_count > 1 when
@@ -3812,7 +3793,7 @@ row_merge_drop_indexes(
 
 	ut_ad(!srv_read_only_mode);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_ad(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	ut_ad(trx->dict_operation);
 	ut_d(dict_sys.assert_locked());
 
 	index = dict_table_get_first_index(table);
@@ -3926,8 +3907,11 @@ row_merge_drop_indexes(
 	/* Invalidate all row_prebuilt_t::ins_graph that are referring
 	to this table. That is, force row_get_prebuilt_insert_row() to
 	rebuild prebuilt->ins_node->entry_list). */
-	ut_ad(table->def_trx_id <= trx->id);
-	table->def_trx_id = trx->id;
+	if (table->def_trx_id < trx->id) {
+		table->def_trx_id = trx->id;
+	} else {
+		ut_ad(table->def_trx_id == trx->id || table->name.part());
+	}
 
 	next_index = dict_table_get_next_index(index);
 
@@ -4017,7 +4001,7 @@ row_merge_drop_temp_indexes(void)
 	/* Ensure that this transaction will be rolled back and locks
 	will be released, if the server gets killed before the commit
 	gets written to the redo log. */
-	trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);
+	trx->dict_operation = true;
 
 	trx->op_info = "dropping indexes";
 	error = que_eval_sql(NULL, sql, FALSE, trx);
@@ -4164,7 +4148,7 @@ row_merge_rename_index_to_add(
 		"END;\n";
 
 	ut_a(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_ad(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
+	ut_ad(trx->dict_operation);
 
 	trx->op_info = "renaming index to add";
 
@@ -4188,59 +4172,6 @@ row_merge_rename_index_to_add(
 	return(err);
 }
 
-/*********************************************************************//**
-Rename an index in the dictionary that is to be dropped. The data
-dictionary must have been locked exclusively by the caller, because
-the transaction will not be committed.
-@return DB_SUCCESS if all OK */
-dberr_t
-row_merge_rename_index_to_drop(
-/*===========================*/
-	trx_t*		trx,		/*!< in/out: transaction */
-	table_id_t	table_id,	/*!< in: table identifier */
-	index_id_t	index_id)	/*!< in: index identifier */
-{
-	dberr_t		err;
-	pars_info_t*	info = pars_info_create();
-
-	ut_ad(!srv_read_only_mode);
-
-	/* We use the private SQL parser of Innobase to generate the
-	query graphs needed in renaming indexes. */
-
-	static const char rename_index[] =
-		"PROCEDURE RENAME_INDEX_PROC () IS\n"
-		"BEGIN\n"
-		"UPDATE SYS_INDEXES SET NAME=CONCAT('"
-		TEMP_INDEX_PREFIX_STR "',NAME)\n"
-		"WHERE TABLE_ID = :tableid AND ID = :indexid;\n"
-		"END;\n";
-
-	ut_a(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_ad(trx_get_dict_operation(trx) == TRX_DICT_OP_INDEX);
-
-	trx->op_info = "renaming index to drop";
-
-	pars_info_add_ull_literal(info, "tableid", table_id);
-	pars_info_add_ull_literal(info, "indexid", index_id);
-
-	err = que_eval_sql(info, rename_index, FALSE, trx);
-
-	if (err != DB_SUCCESS) {
-		/* Even though we ensure that DDL transactions are WAIT
-		and DEADLOCK free, we could encounter other errors e.g.,
-		DB_TOO_MANY_CONCURRENT_TRXS. */
-		trx->error_state = DB_SUCCESS;
-
-		ib::error() << "row_merge_rename_index_to_drop failed with"
-			" error " << err;
-	}
-
-	trx->op_info = "";
-
-	return(err);
-}
-
 /** Create the index and load in to the dictionary.
 @param[in,out]	table		the index is on this table
 @param[in]	index_def	the index definition
@@ -4323,27 +4254,6 @@ row_merge_is_index_usable(
 			   index->table->name)));
 }
 
-/*********************************************************************//**
-Drop a table. The caller must have ensured that the background stats
-thread is not processing the table. This can be done by calling
-dict_stats_wait_bg_to_stop_using_table() after locking the dictionary and
-before calling this function.
-@return DB_SUCCESS or error code */
-dberr_t
-row_merge_drop_table(
-/*=================*/
-	trx_t*		trx,		/*!< in: transaction */
-	dict_table_t*	table)		/*!< in: table to drop */
-{
-	ut_ad(!srv_read_only_mode);
-
-	/* There must be no open transactions on the table. */
-	ut_a(table->get_ref_count() == 0);
-
-	return(row_drop_table_for_mysql(table->name.m_name,
-			trx, SQLCOM_DROP_TABLE, false, false));
-}
-
 /** Build indexes on a table by reading a clustered index, creating a temporary
 file containing index entries, merge sorting these index entries and inserting
 sorted index entries to indexes.
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index e557c13e986..16f5fbdaced 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -2343,17 +2343,7 @@ err_exit:
 
 	heap = mem_heap_create(512);
 
-	switch (trx_get_dict_operation(trx)) {
-	case TRX_DICT_OP_NONE:
-		trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
-	case TRX_DICT_OP_TABLE:
-		break;
-	case TRX_DICT_OP_INDEX:
-		/* If the transaction was previously flagged as
-		TRX_DICT_OP_INDEX, we should be creating auxiliary
-		tables for full-text indexes. */
-		ut_ad(strstr(table->name.m_name, "/FTS_") != NULL);
-	}
+	trx->dict_operation = true;
 
 	node = tab_create_graph_create(table, heap, mode, key_id);
 
@@ -2376,15 +2366,7 @@ err_exit:
 		ib::warn() << "Cannot create table "
 			<< table->name
 			<< " because tablespace full";
-
-		if (dict_table_open_on_name(table->name.m_name, TRUE, FALSE,
-					    DICT_ERR_IGNORE_NONE)) {
-
-			dict_table_close_and_drop(trx, table);
-		} else {
-			dict_mem_table_free(table);
-		}
-
+		dict_mem_table_free(table);
 		break;
 
 	case DB_UNSUPPORTED:
@@ -2470,7 +2452,7 @@ row_create_index_for_mysql(
 	just updates dictonary cache. */
 	if (!table->is_temporary()) {
 		trx_start_if_not_started_xa(trx, true);
-		trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
+		trx->dict_operation = true;
 		/* Note that the space id where we store the index is
 		inherited from the table in dict_build_index_def_step()
 		in dict0crea.cc. */
@@ -2837,7 +2819,7 @@ row_discard_tablespace_begin(
 {
 	trx->op_info = "discarding tablespace";
 
-	trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
+	trx->dict_operation = true;
 
 	trx_start_if_not_started_xa(trx, true);
 
@@ -3335,7 +3317,7 @@ row_drop_table_for_mysql(
 
 	/* This function is called recursively via fts_drop_tables(). */
 	if (!trx_is_started(trx)) {
-		trx_start_for_ddl(trx, TRX_DICT_OP_TABLE);
+		trx_start_for_ddl(trx);
 	}
 
 	/* Turn on this drop bit before we could release the dictionary
@@ -3502,18 +3484,7 @@ defer:
 	and it is free to be dropped */
 	table->to_be_dropped = false;
 
-	switch (trx_get_dict_operation(trx)) {
-	case TRX_DICT_OP_NONE:
-		trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
-		trx->table_id = table->id;
-	case TRX_DICT_OP_TABLE:
-		break;
-	case TRX_DICT_OP_INDEX:
-		/* If the transaction was previously flagged as
-		TRX_DICT_OP_INDEX, we should be dropping auxiliary
-		tables for full-text indexes. */
-		ut_ad(strstr(table->name.m_name, "/FTS_"));
-	}
+	trx->dict_operation = true;
 
 	/* Mark all indexes unavailable in the data dictionary cache
 	before starting to drop the table. */
@@ -3872,7 +3843,7 @@ row_drop_database_for_mysql(
 
 	*found = 0;
 
-	trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
+	trx->dict_operation = true;
 
 	trx_start_if_not_started_xa(trx, true);
 
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index bea748591e1..58b9add6b3b 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -107,34 +107,36 @@ row_purge_remove_clust_if_poss_low(
 	dict_index_t* index = dict_table_get_first_index(node->table);
 	table_id_t table_id = 0;
 	index_id_t index_id = 0;
+	MDL_ticket* mdl_ticket = nullptr;
+	dict_table_t *table = nullptr;
 retry:
 	if (table_id) {
-		MDL_ticket* mdl_ticket = nullptr;
-		if (dict_table_t *table = dict_table_open_on_id(
-			    table_id, false,
-			    DICT_TABLE_OP_OPEN_ONLY_IF_CACHED,
-			    node->purge_thd, &mdl_ticket)) {
-			if (table->n_rec_locks) {
-				for (dict_index_t* ind = UT_LIST_GET_FIRST(
-					     table->indexes); ind;
-				     ind = UT_LIST_GET_NEXT(indexes, ind)) {
-					if (ind->id == index_id) {
-						lock_discard_for_index(*ind);
-					}
+		table = dict_table_open_on_id(
+			table_id, false, DICT_TABLE_OP_OPEN_ONLY_IF_CACHED,
+			node->purge_thd, &mdl_ticket);
+		if (table && table->n_rec_locks) {
+			for (dict_index_t* ind = UT_LIST_GET_FIRST(
+				     table->indexes); ind;
+			     ind = UT_LIST_GET_NEXT(indexes, ind)) {
+				if (ind->id == index_id) {
+					lock_discard_for_index(*ind);
 				}
 			}
-			dict_table_close(table, false, false,
-					 node->purge_thd, mdl_ticket);
 		}
 	}
-	log_free_check();
 	mtr_t mtr;
 	mtr.start();
 	index->set_modified(mtr);
+	log_free_check();
 
 	if (!row_purge_reposition_pcur(mode, node, &mtr)) {
 		/* The record was already removed. */
+removed:
 		mtr.commit();
+		if (table) {
+			dict_table_close(table, false, false,
+					 node->purge_thd, mdl_ticket);
+		}
 		return true;
 	}
 
@@ -153,14 +155,13 @@ retry:
 			}
 			ut_ad("corrupted SYS_INDEXES record" == 0);
 		}
-		dict_drop_index_tree(&node->pcur, nullptr, &mtr);
+		dict_drop_index_tree(&node->pcur, nullptr, table, &mtr);
 		mtr.commit();
 		mtr.start();
 		index->set_modified(mtr);
 
 		if (!row_purge_reposition_pcur(mode, node, &mtr)) {
-			mtr.commit();
-			return true;
+			goto removed;
 		}
 	}
 
@@ -216,6 +217,11 @@ func_exit:
 		mtr_commit(&mtr);
 	}
 
+	if (UNIV_LIKELY_NULL(table)) {
+		dict_table_close(table, false, false, node->purge_thd,
+				 mdl_ticket);
+	}
+
 	return(success);
 }
 
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 3c67fcee9e1..1b0c53f6280 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -71,6 +71,18 @@ row_undo_ins_remove_clust_rec(
 	mtr_t		mtr;
 	dict_index_t*	index	= node->pcur.btr_cur.index;
 	bool		online;
+	table_id_t table_id = 0;
+	const bool dict_locked = node->trx->dict_operation_lock_mode
+		== RW_X_LATCH;
+restart:
+	MDL_ticket* mdl_ticket = nullptr;
+	ut_ad(!table_id || dict_locked
+	      || node->trx->dict_operation_lock_mode == 0);
+	dict_table_t *table = table_id
+		? dict_table_open_on_id(table_id, dict_locked,
+					DICT_TABLE_OP_OPEN_ONLY_IF_CACHED,
+					node->trx->mysql_thd, &mdl_ticket)
+		: nullptr;
 
 	ut_ad(index->is_primary());
 	ut_ad(node->trx->in_rollback);
@@ -132,8 +144,16 @@ row_undo_ins_remove_clust_rec(
 			ut_ad(node->trx->dict_operation_lock_mode
 			      == RW_X_LATCH);
 			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
-
-			dict_drop_index_tree(&node->pcur, node->trx, &mtr);
+			if (!table_id) {
+				table_id = mach_read_from_8(rec);
+				if (table_id) {
+					mtr.commit();
+					goto restart;
+				}
+				ut_ad("corrupted SYS_INDEXES record" == 0);
+			}
+			dict_drop_index_tree(&node->pcur, node->trx,
+					     table, &mtr);
 			mtr.commit();
 
 			mtr.start();
@@ -153,16 +173,14 @@ row_undo_ins_remove_clust_rec(
 			      == RW_X_LATCH);
 			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
 			if (rec_get_n_fields_old(rec)
-			    != DICT_NUM_FIELDS__SYS_COLUMNS) {
-				break;
-			}
-			ulint len;
-			const byte* data = rec_get_nth_field_old(
-				rec, DICT_FLD__SYS_COLUMNS__TABLE_ID, &len);
-			if (len != 8) {
+			    != DICT_NUM_FIELDS__SYS_COLUMNS
+			    || (rec_get_1byte_offs_flag(rec)
+				? rec_1_get_field_end_info(rec, 0) != 8
+				: rec_2_get_field_end_info(rec, 0) != 8)) {
 				break;
 			}
-			node->trx->evict_table(mach_read_from_8(data));
+			static_assert(!DICT_FLD__SYS_COLUMNS__TABLE_ID, "");
+			node->trx->evict_table(mach_read_from_8(rec));
 		}
 	}
 
@@ -213,6 +231,12 @@ func_exit:
 	}
 
 	btr_pcur_commit_specify_mtr(&node->pcur, &mtr);
+
+	if (UNIV_LIKELY_NULL(table)) {
+		dict_table_close(table, dict_locked, false,
+				 node->trx->mysql_thd, mdl_ticket);
+	}
+
 	return(err);
 }
 
diff --git a/storage/innobase/row/row0undo.cc b/storage/innobase/row/row0undo.cc
index bfffcbdcdc6..0399c4e3c58 100644
--- a/storage/innobase/row/row0undo.cc
+++ b/storage/innobase/row/row0undo.cc
@@ -480,7 +480,7 @@ row_undo_step(
 
 	ut_ad(que_node_get_type(node) == QUE_NODE_UNDO);
 
-	if (UNIV_UNLIKELY(trx_get_dict_operation(trx) == TRX_DICT_OP_NONE
+	if (UNIV_UNLIKELY(!trx->dict_operation
 			  && !srv_undo_sources
 			  && srv_shutdown_state != SRV_SHUTDOWN_NONE)
 	    && (srv_fast_shutdown == 3 || trx == trx_roll_crash_recv_trx)) {
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index d1eaea3095e..5331373551c 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -1770,13 +1770,6 @@ file_checked:
 			/* Drop garbage tables. */
 			row_mysql_drop_garbage_tables();
 
-			/* Drop any auxiliary tables that were not
-			dropped when the parent table was
-			dropped. This can happen if the parent table
-			was dropped but the server crashed before the
-			auxiliary tables were dropped. */
-			fts_drop_orphaned_tables();
-
 			/* Rollback incomplete non-DDL transactions */
 			trx_rollback_is_active = true;
 			srv_thread_pool->submit_task(&rollback_all_recovered_task);
diff --git a/storage/innobase/trx/trx0roll.cc b/storage/innobase/trx/trx0roll.cc
index 44f54df6071..97881edc45a 100644
--- a/storage/innobase/trx/trx0roll.cc
+++ b/storage/innobase/trx/trx0roll.cc
@@ -591,8 +591,7 @@ trx_rollback_active(
 
 	trx_roll_crash_recv_trx	= trx;
 
-	const bool dictionary_locked = trx_get_dict_operation(trx)
-		!= TRX_DICT_OP_NONE;
+	const bool dictionary_locked = trx->dict_operation;
 
 	if (dictionary_locked) {
 		row_mysql_lock_data_dictionary(trx);
@@ -608,24 +607,10 @@ trx_rollback_active(
 
 	if (UNIV_UNLIKELY(!trx->rollback_finish())) {
 		ut_ad(!dictionary_locked);
-		goto func_exit;
-	}
-
-	if (!dictionary_locked || !trx->table_id) {
-	} else if (dict_table_t* table = dict_table_open_on_id(
-			   trx->table_id, TRUE, DICT_TABLE_OP_NORMAL)) {
-		ib::info() << "Dropping table " << table->name
-			   << ", with id " << trx->table_id
-			   << " in recovery";
-
-		dict_table_close_and_drop(trx, table);
-
-		trx_commit_for_mysql(trx);
+	} else {
+		ib::info() << "Rolled back recovered transaction " << trx_id;
 	}
 
-	ib::info() << "Rolled back recovered transaction " << trx_id;
-
-func_exit:
 	if (dictionary_locked) {
 		row_mysql_unlock_data_dictionary(trx);
 	}
@@ -751,7 +736,7 @@ void trx_rollback_recovered(bool all)
         srv_fast_shutdown)
       goto discard;
 
-    if (all || trx_get_dict_operation(trx) != TRX_DICT_OP_NONE)
+    if (all || trx->dict_operation)
     {
       trx_rollback_active(trx);
       if (trx->error_state != DB_SUCCESS)
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index ca868cab2e9..11cceda8529 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -113,9 +113,7 @@ trx_init(
 
 	trx->lock.n_rec_locks = 0;
 
-	trx->dict_operation = TRX_DICT_OP_NONE;
-
-	trx->table_id = 0;
+	trx->dict_operation = false;
 
 	trx->error_state = DB_SUCCESS;
 
@@ -133,8 +131,6 @@ trx_init(
 
 	trx->will_lock = 0;
 
-	trx->ddl = false;
-
 	trx->internal = false;
 
 	trx->bulk_insert = false;
@@ -383,7 +379,7 @@ void trx_t::free()
   ut_ad(!read_only);
   ut_ad(!lock.wait_lock);
 
-  dict_operation= TRX_DICT_OP_NONE;
+  dict_operation= false;
   trx_sys.deregister_trx(this);
   assert_freed();
   trx_sys.rw_trx_hash.put_pins(this);
@@ -423,7 +419,6 @@ void trx_t::free()
   MEM_NOACCESS(&start_time, sizeof start_time);
   MEM_NOACCESS(&start_time_micro, sizeof start_time_micro);
   MEM_NOACCESS(&commit_lsn, sizeof commit_lsn);
-  MEM_NOACCESS(&table_id, sizeof table_id);
   MEM_NOACCESS(&mysql_thd, sizeof mysql_thd);
   MEM_NOACCESS(&mysql_log_file_name, sizeof mysql_log_file_name);
   MEM_NOACCESS(&mysql_log_offset, sizeof mysql_log_offset);
@@ -448,7 +443,6 @@ void trx_t::free()
   MEM_NOACCESS(&fts_trx, sizeof fts_trx);
   MEM_NOACCESS(&fts_next_doc_id, sizeof fts_next_doc_id);
   MEM_NOACCESS(&flush_tables, sizeof flush_tables);
-  MEM_NOACCESS(&ddl, sizeof ddl);
   MEM_NOACCESS(&internal, sizeof internal);
 #ifdef UNIV_DEBUG
   MEM_NOACCESS(&start_line, sizeof start_line);
@@ -693,13 +687,7 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
   trx->is_recovered= true;
   trx->start_time= start_time;
   trx->start_time_micro= start_time_micro;
-
-  if (undo->dict_operation)
-  {
-    trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
-    if (!trx->table_id)
-      trx->table_id= undo->table_id;
-  }
+  trx->dict_operation= undo->dict_operation;
 
   trx_sys.rw_trx_hash.insert(trx);
   trx_sys.rw_trx_hash.put_pins(trx);
@@ -941,7 +929,7 @@ trx_start_low(
 	trx->auto_commit = thd_trx_is_auto_commit(trx->mysql_thd);
 
 	trx->read_only = srv_read_only_mode
-		|| (!trx->ddl && !trx->internal
+		|| (!trx->dict_operation && !trx->internal
 		    && thd_trx_is_read_only(trx->mysql_thd));
 
 	if (!trx->auto_commit) {
@@ -973,7 +961,7 @@ trx_start_low(
 	list too. */
 
 	if (!trx->read_only
-	    && (trx->mysql_thd == 0 || read_write || trx->ddl)) {
+	    && (!trx->mysql_thd || read_write || trx->dict_operation)) {
 
 		/* Temporary rseg is assigned only if the transaction
 		updates a temporary table */
@@ -1434,7 +1422,7 @@ inline void trx_t::commit_in_memory(const mtr_t *mtr)
   lock.was_chosen_as_deadlock_victim= false;
 #endif /* WITH_WSREP */
   mutex.wr_lock();
-  dict_operation= TRX_DICT_OP_NONE;
+  dict_operation= false;
 
   DBUG_LOG("trx", "Commit in memory: " << this);
   state= TRX_STATE_NOT_STARTED;
@@ -2192,38 +2180,17 @@ trx_start_internal_read_only_low(
 	trx_start_low(trx, false);
 }
 
-/*************************************************************//**
-Starts the transaction for a DDL operation. */
-void
-trx_start_for_ddl_low(
-/*==================*/
-	trx_t*		trx,	/*!< in/out: transaction */
-	trx_dict_op_t	op)	/*!< in: dictionary operation type */
+/** Start a transaction for a DDL operation.
+@param trx   transaction */
+void trx_start_for_ddl_low(trx_t *trx)
 {
-	switch (trx->state) {
-	case TRX_STATE_NOT_STARTED:
-		/* Flag this transaction as a dictionary operation, so that
-		the data dictionary will be locked in crash recovery. */
-
-		trx_set_dict_operation(trx, op);
-
-		/* Ensure it is not flagged as an auto-commit-non-locking
-		transation. */
-		trx->will_lock = 1;
-
-		trx->ddl= true;
-
-		trx_start_internal_low(trx);
-		return;
-
-	case TRX_STATE_ACTIVE:
-	case TRX_STATE_PREPARED:
-	case TRX_STATE_PREPARED_RECOVERED:
-	case TRX_STATE_COMMITTED_IN_MEMORY:
-		break;
-	}
-
-	ut_error;
+  ut_a(trx->state == TRX_STATE_NOT_STARTED);
+  /* Flag this transaction as a dictionary operation, so that
+  the data dictionary will be locked in crash recovery. */
+  trx->dict_operation= true;
+  /* Ensure it is not flagged as an auto-commit-non-locking transaction. */
+  trx->will_lock= 1;
+  trx_start_internal_low(trx);
 }
 
 /*************************************************************//**
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index 1724b18572c..b71656a198b 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -876,7 +876,6 @@ trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
 	mysql_mutex_unlock(&rseg->mutex);
 
 	undo->dict_operation = undo_header[TRX_UNDO_DICT_TRANS];
-	undo->table_id = mach_read_from_8(undo_header + TRX_UNDO_TABLE_ID);
 	undo->size = flst_get_len(TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST
 				  + block->frame);
 
@@ -1046,21 +1045,12 @@ trx_undo_create(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** undo,
 		return block;
 	}
 
-	switch (trx_get_dict_operation(trx)) {
-	case TRX_DICT_OP_NONE:
-		break;
-	case TRX_DICT_OP_INDEX:
-		/* Do not discard the table on recovery. */
-		trx->table_id = 0;
-		/* fall through */
-	case TRX_DICT_OP_TABLE:
-		(*undo)->table_id = trx->table_id;
-		(*undo)->dict_operation = TRUE;
+	if (trx->dict_operation) {
+		(*undo)->dict_operation = true;
 		mtr->write<1,mtr_t::MAYBE_NOP>(*block, block->frame + offset
 					       + TRX_UNDO_DICT_TRANS, 1U);
 		mtr->write<8,mtr_t::MAYBE_NOP>(*block, block->frame + offset
-					       + TRX_UNDO_TABLE_ID,
-					       trx->table_id);
+					       + TRX_UNDO_TABLE_ID, 0U);
 	}
 
 	*err = DB_SUCCESS;
@@ -1111,21 +1101,12 @@ trx_undo_reuse_cached(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** pundo,
 		return block;
 	}
 
-	switch (trx_get_dict_operation(trx)) {
-	case TRX_DICT_OP_NONE:
-		return block;
-	case TRX_DICT_OP_INDEX:
-		/* Do not discard the table on recovery. */
-		trx->table_id = 0;
-		/* fall through */
-	case TRX_DICT_OP_TABLE:
-		undo->table_id = trx->table_id;
+	if (trx->dict_operation) {
 		undo->dict_operation = TRUE;
 		mtr->write<1,mtr_t::MAYBE_NOP>(*block, block->frame + offset
 					       + TRX_UNDO_DICT_TRANS, 1U);
 		mtr->write<8,mtr_t::MAYBE_NOP>(*block, block->frame + offset
-					       + TRX_UNDO_TABLE_ID,
-					       trx->table_id);
+					       + TRX_UNDO_TABLE_ID, 0U);
 	}
 
 	return block;