MDEV-17158 TRUNCATE is not atomic after MDEV-13564

16 files changed, 187 insertions, 103 deletions

diff --git a/mysql-test/suite/innodb/disabled.def b/mysql-test/suite/innodb/disabled.def
index a6259cdbfc6..35c941f8af7 100644
--- a/mysql-test/suite/innodb/disabled.def
+++ b/mysql-test/suite/innodb/disabled.def
@@ -11,4 +11,3 @@
 ##############################################################################
 
 create-index-debug : MDEV-13680 InnoDB may crash when btr_page_alloc() fails
-truncate_crash : MDEV-17158 log_write_up_to() sometimes fails
diff --git a/mysql-test/suite/innodb/r/truncate_crash.result b/mysql-test/suite/innodb/r/truncate_crash.result
index 6c20da46cf8..10ce8e92ab6 100644
--- a/mysql-test/suite/innodb/r/truncate_crash.result
+++ b/mysql-test/suite/innodb/r/truncate_crash.result
@@ -1,14 +1,14 @@
 FLUSH TABLES;
 CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
-INSERT INTO t1 SET a=1;
+INSERT INTO t1 VALUES (1),(2);
 connect  wait,localhost,root,,test;
-SET DEBUG_SYNC='after_trx_committed_in_memory SIGNAL c WAIT_FOR ever';
+SET DEBUG_SYNC='before_trx_state_committed_in_memory SIGNAL c WAIT_FOR ever';
 TRUNCATE TABLE t1;
 connection default;
 SET DEBUG_SYNC='now WAIT_FOR c';
 disconnect wait;
-SELECT * FROM t1;
-a
-1
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+0
 TRUNCATE TABLE t1;
 DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/t/truncate_crash.test b/mysql-test/suite/innodb/t/truncate_crash.test
index 15ba475e0e1..5cb39c745dc 100644
--- a/mysql-test/suite/innodb/t/truncate_crash.test
+++ b/mysql-test/suite/innodb/t/truncate_crash.test
@@ -5,10 +5,10 @@
 
 FLUSH TABLES;
 CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
-INSERT INTO t1 SET a=1;
+INSERT INTO t1 VALUES (1),(2);
 
 connect (wait,localhost,root,,test);
-SET DEBUG_SYNC='after_trx_committed_in_memory SIGNAL c WAIT_FOR ever';
+SET DEBUG_SYNC='before_trx_state_committed_in_memory SIGNAL c WAIT_FOR ever';
 send TRUNCATE TABLE t1;
 
 connection default;
@@ -17,6 +17,7 @@ SET DEBUG_SYNC='now WAIT_FOR c';
 --source include/restart_mysqld.inc
 disconnect wait;
 
-SELECT * FROM t1;
+--replace_result 2 0
+SELECT COUNT(*) FROM t1;
 TRUNCATE TABLE t1;
 DROP TABLE t1;
diff --git a/storage/innobase/dict/dict0crea.cc b/storage/innobase/dict/dict0crea.cc
index 1d7ee29d019..7d83c183fdc 100644
--- a/storage/innobase/dict/dict0crea.cc
+++ b/storage/innobase/dict/dict0crea.cc
@@ -38,6 +38,8 @@ Created 1/8/1996 Heikki Tuuri
 #include "row0mysql.h"
 #include "pars0pars.h"
 #include "trx0roll.h"
+#include "trx0rseg.h"
+#include "trx0undo.h"
 #include "ut0vec.h"
 #include "dict0priv.h"
 #include "fts0priv.h"
@@ -370,7 +372,33 @@ dict_build_table_def_step(
 
 		ut_ad(DICT_TF_GET_ZIP_SSIZE(table->flags) == 0
 		      || dict_table_has_atomic_blobs(table));
-
+		trx_t* trx = thr_get_trx(thr);
+		ut_ad(trx->table_id);
+		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;
+			page_t* page = trx_undo_page_get(
+				page_id_t(trx->rsegs.m_redo.rseg->space->id,
+					  undo->hdr_page_no),
+				&mtr);
+			mlog_write_ulint(page + undo->hdr_offset
+					 + TRX_UNDO_DICT_TRANS,
+					 TRUE, MLOG_1BYTE, &mtr);
+			mlog_write_ull(page + undo->hdr_offset
+				       + TRX_UNDO_TABLE_ID,
+				       trx->table_id, &mtr);
+			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, table, false);
@@ -381,7 +409,7 @@ dict_build_table_def_step(
 		);
 
 		if (space_id == ULINT_UNDEFINED) {
-			return(DB_ERROR);
+			return DB_ERROR;
 		}
 
 		/* Determine the tablespace flags. */
@@ -416,7 +444,6 @@ 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);
@@ -426,8 +453,6 @@ dict_build_table_def_step(
 		      != REC_FORMAT_COMPRESSED);
 		table->space = fil_system.sys_space;
 		table->space_id = TRX_SYS_SPACE;
-		DBUG_EXECUTE_IF("ib_ddl_crash_during_tablespace_alloc",
-				DBUG_SUICIDE(););
 	}
 
 	ins_node_set_new_row(node->tab_def,
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 3fba1679cdf..3a4c8c19c00 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -1556,9 +1556,14 @@ dict_table_rename_in_cache(
 /*=======================*/
 	dict_table_t*	table,		/*!< in/out: table */
 	const char*	new_name,	/*!< in: new name */
-	ibool		rename_also_foreigns)/*!< in: in ALTER TABLE we want
+	bool		rename_also_foreigns,
+					/*!< in: in ALTER TABLE we want
 					to preserve the original table name
 					in constraints which reference it */
+	bool		replace_new_file)
+					/*!< in: whether to replace the
+					file with the new name
+					(as part of rolling back TRUNCATE) */
 {
 	dberr_t		err;
 	dict_foreign_t*	foreign;
@@ -1658,7 +1663,8 @@ dict_table_rename_in_cache(
 		}
 
 		/* New filepath must not exist. */
-		err = table->space->rename(new_name, new_path, true);
+		err = table->space->rename(new_name, new_path, true,
+					   replace_new_file);
 		ut_free(new_path);
 
 		/* If the tablespace is remote, a new .isl file was created
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index dc8d2cddf5b..e64c7040166 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -74,12 +74,14 @@ if that the old filepath exists and the new filepath does not exist.
 @param[in]	old_path	old filepath
 @param[in]	new_path	new filepath
 @param[in]	is_discarded	whether the tablespace is discarded
+@param[in]	replace_new	whether to ignore the existence of new_path
 @return innodb error code */
 static dberr_t
 fil_rename_tablespace_check(
 	const char*	old_path,
 	const char*	new_path,
-	bool		is_discarded);
+	bool		is_discarded,
+	bool		replace_new = false);
 /** Rename a single-table tablespace.
 The tablespace must exist in the memory cache.
 @param[in]	id		tablespace identifier
@@ -2866,12 +2868,14 @@ if that the old filepath exists and the new filepath does not exist.
 @param[in]	old_path	old filepath
 @param[in]	new_path	new filepath
 @param[in]	is_discarded	whether the tablespace is discarded
+@param[in]	replace_new	whether to ignore the existence of new_path
 @return innodb error code */
 static dberr_t
 fil_rename_tablespace_check(
 	const char*	old_path,
 	const char*	new_path,
-	bool		is_discarded)
+	bool		is_discarded,
+	bool		replace_new)
 {
 	bool	exists = false;
 	os_file_type_t	ftype;
@@ -2887,7 +2891,11 @@ fil_rename_tablespace_check(
 	}
 
 	exists = false;
-	if (!os_file_status(new_path, &exists, &ftype) || exists) {
+	if (os_file_status(new_path, &exists, &ftype) && !exists) {
+		return DB_SUCCESS;
+	}
+
+	if (!replace_new) {
 		ib::error() << "Cannot rename '" << old_path
 			<< "' to '" << new_path
 			<< "' because the target file exists."
@@ -2895,17 +2903,46 @@ fil_rename_tablespace_check(
 		return(DB_TABLESPACE_EXISTS);
 	}
 
+	/* This must be during the ROLLBACK of TRUNCATE TABLE.
+	Because InnoDB only allows at most one data dictionary
+	transaction at a time, and because this incomplete TRUNCATE
+	would have created a new tablespace file, we must remove
+	a possibly existing tablespace that is associated with the
+	new tablespace file. */
+retry:
+	mutex_enter(&fil_system.mutex);
+	for (fil_space_t* space = UT_LIST_GET_FIRST(fil_system.space_list);
+	     space; space = UT_LIST_GET_NEXT(space_list, space)) {
+		ulint id = space->id;
+		if (id && id < SRV_LOG_SPACE_FIRST_ID
+		    && space->purpose == FIL_TYPE_TABLESPACE
+		    && !strcmp(new_path,
+			       UT_LIST_GET_FIRST(space->chain)->name)) {
+			ib::info() << "TRUNCATE rollback: " << id
+				<< "," << new_path;
+			mutex_exit(&fil_system.mutex);
+			dberr_t err = fil_delete_tablespace(id);
+			if (err != DB_SUCCESS) {
+				return err;
+			}
+			goto retry;
+		}
+	}
+	mutex_exit(&fil_system.mutex);
+	fil_delete_file(new_path);
+
 	return(DB_SUCCESS);
 }
 
-dberr_t fil_space_t::rename(const char* name, const char* path, bool log)
+dberr_t fil_space_t::rename(const char* name, const char* path, bool log,
+			    bool replace)
 {
 	ut_ad(UT_LIST_GET_LEN(chain) == 1);
 	ut_ad(!is_system_tablespace(id));
 
 	if (log) {
 		dberr_t err = fil_rename_tablespace_check(
-			chain.start->name, path, false);
+			chain.start->name, path, false, replace);
 		if (err != DB_SUCCESS) {
 			return(err);
 		}
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index b65266fbf2e..2776ca18892 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -10831,8 +10831,6 @@ create_table_info_t::create_table_def()
 		}
 	}
 
-	ut_ad(trx_state_eq(m_trx, TRX_STATE_NOT_STARTED));
-
 	/* Check whether there already exists a FTS_DOC_ID column */
 	if (create_table_check_doc_id_col(m_trx, m_form, &doc_id_col)){
 
@@ -10931,9 +10929,6 @@ create_table_info_t::create_table_def()
 				mem_heap_free(heap);
 				dict_mem_table_free(table);
 
-				ut_ad(trx_state_eq(
-					m_trx, TRX_STATE_NOT_STARTED));
-
 				DBUG_RETURN(ER_CANT_CREATE_TABLE);
 			}
 		}
@@ -11050,8 +11045,6 @@ err_col:
 
 	dict_table_add_system_columns(table, heap);
 
-	ut_ad(trx_state_eq(m_trx, TRX_STATE_NOT_STARTED));
-
 	if (table->is_temporary()) {
 		/* Get a new table ID. FIXME: Make this a private
 		sequence, not shared with persistent tables! */
@@ -12272,10 +12265,9 @@ create_table_info_t::prepare_create_table(
 	DBUG_RETURN(parse_table_name(name));
 }
 
-/** Create a new table to an InnoDB database.
-@return error number */
-int
-create_table_info_t::create_table()
+/** Create the internal innodb table.
+@param create_fk	whether to add FOREIGN KEY constraints */
+int create_table_info_t::create_table(bool create_fk)
 {
 	int		error;
 	int		primary_key_no;
@@ -12403,7 +12395,7 @@ create_table_info_t::create_table()
 
 	if (stmt) {
 		dberr_t	err = row_table_add_foreign_constraints(
-			m_trx, stmt, stmt_len, m_table_name,
+			create_fk ? m_trx : NULL, stmt, stmt_len, m_table_name,
 			m_create_info->options & HA_LEX_CREATE_TMP_TABLE);
 
 		switch (err) {
@@ -12579,56 +12571,56 @@ ha_innobase::create(
 				     remote_path,
 				     file_per_table, trx);
 
-	/* Initialize the object. */
-	if ((error = info.initialize())) {
+	if ((error = info.initialize())
+	    || (error = info.prepare_create_table(name))) {
+		if (trx) {
+			trx_rollback_for_mysql(trx);
+			row_mysql_unlock_data_dictionary(trx);
+		}
 		DBUG_RETURN(error);
 	}
 
-	/* Prepare for create and validate options. */
-	if ((error = info.prepare_create_table(name))) {
-		DBUG_RETURN(error);
-	}
+	const bool own_trx = !trx;
 
-	bool own_trx = !trx;
 	if (own_trx) {
 		info.allocate_trx();
 		trx = info.trx();
+		/* Latch the InnoDB data dictionary exclusively so that no deadlocks
+		or lock waits can happen in it during a table create operation.
+		Drop table etc. do this latching in row0mysql.cc. */
+		row_mysql_lock_data_dictionary(trx);
+		DBUG_ASSERT(trx_state_eq(trx, TRX_STATE_NOT_STARTED));
 	}
 
-	/* Latch the InnoDB data dictionary exclusively so that no deadlocks
-	or lock waits can happen in it during a table create operation.
-	Drop table etc. do this latching in row0mysql.cc. */
-	row_mysql_lock_data_dictionary(trx);
-
-	if ((error = info.create_table())) {
+	if ((error = info.create_table(own_trx))) {
+		trx_rollback_for_mysql(trx);
+		row_mysql_unlock_data_dictionary(trx);
 		if (own_trx) {
-			trx_rollback_for_mysql(trx);
+			trx_free(trx);
+			DBUG_RETURN(error);
 		}
-		row_mysql_unlock_data_dictionary(trx);
-		goto func_exit;
 	}
 
+	innobase_commit_low(trx);
+	row_mysql_unlock_data_dictionary(trx);
+
 	if (own_trx) {
-		innobase_commit_low(trx);
+		trx_free(trx);
 	}
 
-	ut_ad(!srv_read_only_mode);
-	row_mysql_unlock_data_dictionary(trx);
 	/* 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();
 
+	ut_ad(!srv_read_only_mode);
+
 	error = info.create_table_update_dict();
 
 	/* Tell the InnoDB server that there might be work for
 	utility threads: */
 
 	srv_active_wake_master_thread();
-func_exit:
-	if (own_trx) {
-		trx_free(trx);
-	}
 
 	DBUG_RETURN(error);
 }
@@ -13034,16 +13026,19 @@ innobase_drop_database(
 	trx_free(trx);
 }
 
-/*********************************************************************//**
-Renames an InnoDB table.
+/** Rename an InnoDB table.
+@param[in,out]	trx	InnoDB data dictionary transaction
+@param[in]	from	old table name
+@param[in]	to	new table name
+@param[in]	commit	whether to commit trx
 @return DB_SUCCESS or error code */
 inline
 dberr_t
 innobase_rename_table(
-/*==================*/
-	trx_t*		trx,	/*!< in: transaction */
-	const char*	from,	/*!< in: old name of the table */
-	const char*	to)	/*!< in: new name of the table */
+	trx_t*		trx,
+	const char*	from,
+	const char*	to,
+	bool		commit = true)
 {
 	dberr_t	error;
 	char	norm_to[FN_REFLEN];
@@ -13063,10 +13058,11 @@ innobase_rename_table(
 	trx_start_if_not_started(trx, true);
 	ut_ad(trx->will_lock > 0);
 
-	/* Serialize data dictionary operations with dictionary mutex:
-	no deadlocks can occur then in these operations. */
-
-	row_mysql_lock_data_dictionary(trx);
+	if (commit) {
+		/* Serialize data dictionary operations with dictionary mutex:
+		no deadlocks can occur then in these operations. */
+		row_mysql_lock_data_dictionary(trx);
+	}
 
 	dict_table_t*   table = dict_table_open_on_name(norm_from, TRUE, FALSE,
 							DICT_ERR_IGNORE_NONE);
@@ -13094,11 +13090,10 @@ innobase_rename_table(
 	/* FTS sync is in progress. We shall timeout this operation */
 	if (lock_wait_timeout < 0) {
 		error = DB_LOCK_WAIT_TIMEOUT;
-		row_mysql_unlock_data_dictionary(trx);
-		DBUG_RETURN(error);
+		goto func_exit;
 	}
 
-	error = row_rename_table_for_mysql(norm_from, norm_to, trx, TRUE);
+	error = row_rename_table_for_mysql(norm_from, norm_to, trx, commit);
 
 	if (error != DB_SUCCESS) {
 		if (error == DB_TABLE_NOT_FOUND
@@ -13147,7 +13142,10 @@ innobase_rename_table(
 		}
 	}
 
-	row_mysql_unlock_data_dictionary(trx);
+func_exit:
+	if (commit) {
+		row_mysql_unlock_data_dictionary(trx);
+	}
 
 	/* Flush the log to reduce probability that the .frm
 	files and the InnoDB data dictionary get out-of-sync
@@ -13199,20 +13197,18 @@ int ha_innobase::truncate()
 
 	++trx->will_lock;
 	trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
+	row_mysql_lock_data_dictionary(trx);
 	int err = convert_error_code_to_mysql(
-		innobase_rename_table(trx, ib_table->name.m_name, temp_name),
+		innobase_rename_table(trx, ib_table->name.m_name, temp_name, false),
 		ib_table->flags, m_user_thd);
-	if (!err) {
+	if (err) {
+		trx_rollback_for_mysql(trx);
+		row_mysql_unlock_data_dictionary(trx);
+	} else {
 		err = create(name, table, &info,
 			     dict_table_is_file_per_table(ib_table), trx);
 	}
 
-	if (err) {
-		innobase_rename_table(trx, temp_name, name);
-		trx_rollback_to_savepoint(trx, NULL);
-	}
-
-	innobase_commit_low(trx);
 	trx_free(trx);
 
 	if (!err) {
diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h
index fd8ea696faa..8b50129c946 100644
--- a/storage/innobase/handler/ha_innodb.h
+++ b/storage/innobase/handler/ha_innodb.h
@@ -675,8 +675,9 @@ public:
 	/** Set m_tablespace_type. */
 	void set_tablespace_type(bool table_being_altered_is_file_per_table);
 
-	/** Create the internal innodb table. */
-	int create_table();
+	/** Create the internal innodb table.
+	@param create_fk	whether to add FOREIGN KEY constraints */
+	int create_table(bool create_fk = true);
 
 	/** Update the internal data dictionary. */
 	int create_table_update_dict();
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index d0afac185f3..b050a045409 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -8767,11 +8767,11 @@ commit_cache_rebuild(
 	/* We already committed and redo logged the renames,
 	so this must succeed. */
 	error = dict_table_rename_in_cache(
-		ctx->old_table, ctx->tmp_name, FALSE);
+		ctx->old_table, ctx->tmp_name, false);
 	ut_a(error == DB_SUCCESS);
 
 	error = dict_table_rename_in_cache(
-		ctx->new_table, old_name, FALSE);
+		ctx->new_table, old_name, false);
 	ut_a(error == DB_SUCCESS);
 
 	DBUG_VOID_RETURN;
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index 3dcf290a276..0da785a4e5c 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -398,10 +398,14 @@ dict_table_rename_in_cache(
 /*=======================*/
 	dict_table_t*	table,		/*!< in/out: table */
 	const char*	new_name,	/*!< in: new name */
-	ibool		rename_also_foreigns)
+	bool		rename_also_foreigns,
 					/*!< in: in ALTER TABLE we want
 					to preserve the original table name
 					in constraints which reference it */
+	bool		replace_new_file = false)
+					/*!< in: whether to replace the
+					file with the new name
+					(as part of rolling back TRUNCATE) */
 	MY_ATTRIBUTE((nonnull));
 
 /** Removes an index from the dictionary cache.
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 890684af67e..d0dc02e6208 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -218,9 +218,11 @@ struct fil_space_t {
 	@param[in]	name	table name after renaming
 	@param[in]	path	tablespace file name after renaming
 	@param[in]	log	whether to write redo log
+	@param[in]	replace	whether to ignore the existence of path
 	@return	error code
 	@retval	DB_SUCCESS	on success */
-	dberr_t rename(const char* name, const char* path, bool log);
+	dberr_t rename(const char* name, const char* path, bool log,
+		       bool replace = false);
 
 	/** Note that the tablespace has been imported.
 	Initially, purpose=FIL_TYPE_IMPORT so that no redo log is
diff --git a/storage/innobase/include/row0mysql.h b/storage/innobase/include/row0mysql.h
index c59248d88c4..96454995e74 100644
--- a/storage/innobase/include/row0mysql.h
+++ b/storage/innobase/include/row0mysql.h
@@ -399,7 +399,7 @@ Each foreign key constraint must be accompanied with indexes in
 bot participating tables. The indexes are allowed to contain more
 fields than mentioned in the constraint.
 
-@param[in]	trx		transaction
+@param[in]	trx		transaction (NULL if not adding to dictionary)
 @param[in]	sql_string	table create statement where
 				foreign keys are declared like:
 				FOREIGN KEY (a, b) REFERENCES table2(c, d),
@@ -408,9 +408,8 @@ fields than mentioned in the constraint.
 				database id the database of parameter name
 @param[in]	sql_length	length of sql_string
 @param[in]	name		table full name in normalized form
-@param[in]	reject_fks	if TRUE, fail with error code
-				DB_CANNOT_ADD_CONSTRAINT if any
-				foreign keys are found.
+@param[in]	reject_fks	whether to fail with DB_CANNOT_ADD_CONSTRAINT
+				if any foreign keys are found
 @return error code or DB_SUCCESS */
 dberr_t
 row_table_add_foreign_constraints(
@@ -418,7 +417,7 @@ row_table_add_foreign_constraints(
 	const char*		sql_string,
 	size_t			sql_length,
 	const char*		name,
-	ibool			reject_fks)
+	bool			reject_fks)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /*********************************************************************//**
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index 16e2a384424..ada78615f15 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -203,6 +203,11 @@ trx_undo_truncate_start(
 	ulint		hdr_page_no,
 	ulint		hdr_offset,
 	undo_no_t	limit);
+/** Mark that an undo log header belongs to a data dictionary transaction.
+@param[in]	trx	dictionary transaction
+@param[in,out]	undo	undo log
+@param[in,out]	mtr	mini-transaction */
+void trx_undo_mark_as_dict(const trx_t* trx, trx_undo_t* undo, mtr_t* mtr);
 /** Assign an undo log for a persistent transaction.
 A new undo log is created or a cached undo log reused.
 @param[in,out]	trx	transaction
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index fa2316eee31..1d6c6a957dd 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -2632,7 +2632,7 @@ Each foreign key constraint must be accompanied with indexes in
 bot participating tables. The indexes are allowed to contain more
 fields than mentioned in the constraint.
 
-@param[in]	trx		transaction
+@param[in]	trx		transaction (NULL if not adding to dictionary)
 @param[in]	sql_string	table create statement where
 				foreign keys are declared like:
 				FOREIGN KEY (a, b) REFERENCES table2(c, d),
@@ -2641,9 +2641,8 @@ fields than mentioned in the constraint.
 				database id the database of parameter name
 @param[in]	sql_length	length of sql_string
 @param[in]	name		table full name in normalized form
-@param[in]	reject_fks	if TRUE, fail with error code
-				DB_CANNOT_ADD_CONSTRAINT if any
-				foreign keys are found.
+@param[in]	reject_fks	whether to fail with DB_CANNOT_ADD_CONSTRAINT
+				if any foreign keys are found
 @return error code or DB_SUCCESS */
 dberr_t
 row_table_add_foreign_constraints(
@@ -2651,7 +2650,7 @@ row_table_add_foreign_constraints(
 	const char*		sql_string,
 	size_t			sql_length,
 	const char*		name,
-	ibool			reject_fks)
+	bool			reject_fks)
 {
 	dberr_t	err;
 
@@ -2661,13 +2660,17 @@ row_table_add_foreign_constraints(
 	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_X));
 	ut_a(sql_string);
 
-	err = dict_create_foreign_constraints(
-		trx, sql_string, sql_length, name, reject_fks);
+	if (trx) {
+		err = dict_create_foreign_constraints(
+			trx, sql_string, sql_length, name, reject_fks);
 
-	DBUG_EXECUTE_IF("ib_table_add_foreign_fail",
-			err = DB_DUPLICATE_KEY;);
+		DBUG_EXECUTE_IF("ib_table_add_foreign_fail",
+				err = DB_DUPLICATE_KEY;);
 
-	DEBUG_SYNC_C("table_add_foreign_constraints");
+		DEBUG_SYNC_C("table_add_foreign_constraints");
+	} else {
+		err = DB_SUCCESS;
+	}
 
 	if (err == DB_SUCCESS) {
 		/* Check that also referencing constraints are ok */
@@ -2682,7 +2685,7 @@ row_table_add_foreign_constraints(
 		}
 	}
 
-	if (err != DB_SUCCESS) {
+	if (err != DB_SUCCESS && trx) {
 		/* We have special error handling here */
 
 		trx->error_state = DB_SUCCESS;
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 23867b7395a..67a11711fe2 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -435,7 +435,8 @@ row_undo_ins_parse_undo_rec(
 		ptr[len] = 0;
 		const char* name = reinterpret_cast<char*>(ptr);
 		if (strcmp(table->name.m_name, name)) {
-			dict_table_rename_in_cache(table, name, false);
+			dict_table_rename_in_cache(table, name, false,
+						   table_id != 0);
 		}
 		goto close_table;
 	}
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index 30c19669cee..d51e26aa771 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -675,7 +675,8 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
   if (undo->dict_operation)
   {
     trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
-    trx->table_id= undo->table_id;
+    if (!trx->table_id)
+      trx->table_id= undo->table_id;
   }
 
   trx_sys.rw_trx_hash.insert(trx);
@@ -1458,6 +1459,10 @@ void trx_commit_low(trx_t* trx, mtr_t* mtr)
 		}
 	}
 
+#ifndef DBUG_OFF
+	const bool debug_sync = trx->mysql_thd && trx->has_logged_persistent();
+#endif
+
 	if (mtr != NULL) {
 
 		mtr->set_sync();
@@ -1500,7 +1505,7 @@ void trx_commit_low(trx_t* trx, mtr_t* mtr)
            thd->debug_sync_control defined any longer. However the stack
            is possible only with a prepared trx not updating any data.
         */
-	if (trx->mysql_thd != NULL && trx->has_logged_persistent()) {
+	if (debug_sync) {
 		DEBUG_SYNC_C("before_trx_state_committed_in_memory");
 	}
 #endif