MDEV-25506: Kill during DDL leaves orphan .ibd file

Before we create an InnoDB data file, we must have persistently
started a DDL transaction and written a record in SYS_INDEXES
as well as a FILE_CREATE record for creating the file.
In that way, if InnoDB is killed before the DDL transaction is
committed, the rollback will be able to delete the file in
dict_drop_index_tree().

dict_build_table_def_step(): Do not create the tablespace.
At this point, we have not written any log, not even for
inserting the SYS_TABLES record.

dict_create_sys_indexes_tuple(): Relax an assertion to tolerate
a missing tablespace before the first index has been created in
dict_create_index_step().

dict_build_index_def_step(): Relax the dict_table_open_on_name()
parameter, because no tablespace may be available yet.

tab_create_graph_create(), row_create_table_for_mysql(), tab_node_t:
Remove key_id, mode.

ind_create_graph_create(), row_create_index_for_mysql(), ind_node_t:
Add key_id, mode.

dict_create_index_space(): New function, to create the tablespace
during clustered index creation.

dict_create_index_step(): After the SYS_INDEXES record has been
written, invoke dict_create_index_space() to create the tablespace
if needed.

fil_ibd_create(): Before creating the file, persistently write a
FILE_CREATE record. This will also ensure that an incomplete DDL
transaction will be recovered. After creating the file, invoke
fsp_header_init().

13 files changed, 141 insertions, 158 deletions

diff --git a/mysql-test/suite/innodb/r/log_file_name.result b/mysql-test/suite/innodb/r/log_file_name.result
index f183cb44ebe..40c15fa07b5 100644
--- a/mysql-test/suite/innodb/r/log_file_name.result
+++ b/mysql-test/suite/innodb/r/log_file_name.result
@@ -72,7 +72,7 @@ t2
 t3
 DROP TABLE t2,t3;
 CREATE TABLE t0(a INT PRIMARY KEY) ENGINE=InnoDB;
-ERROR HY000: Tablespace for table '`test`.`t0`' exists. Please DISCARD the tablespace before IMPORT
+ERROR HY000: Can't create table `test`.`t0` (errno: 184 "Tablespace already exists")
 CREATE TABLE t0(a INT PRIMARY KEY) ENGINE=InnoDB;
 DROP TABLE t0;
 CREATE TABLE u1(a INT PRIMARY KEY) ENGINE=InnoDB;
diff --git a/mysql-test/suite/innodb/r/truncate.result b/mysql-test/suite/innodb/r/truncate.result
index 180414f47c7..e12a401017f 100644
--- a/mysql-test/suite/innodb/r/truncate.result
+++ b/mysql-test/suite/innodb/r/truncate.result
@@ -25,7 +25,7 @@ call mtr.add_suppression("InnoDB: (Operating system )?[Ee]rror number");
 call mtr.add_suppression("InnoDB: Cannot create file '.*t1\\.ibd");
 FLUSH TABLES;
 CREATE TABLE t1 (a INT) ENGINE=InnoDB;
-ERROR HY000: Tablespace for table '`test`.`t1`' exists. Please DISCARD the tablespace before IMPORT
+ERROR 42S01: Table '`test`.`t1`' already exists
 SELECT * FROM t1;
 a
 1
diff --git a/mysql-test/suite/innodb/t/log_file_name.test b/mysql-test/suite/innodb/t/log_file_name.test
index 2be3f8e7c50..cde278e34c7 100644
--- a/mysql-test/suite/innodb/t/log_file_name.test
+++ b/mysql-test/suite/innodb/t/log_file_name.test
@@ -137,12 +137,9 @@ SELECT * FROM t3;
 SHOW TABLES;
 DROP TABLE t2,t3;
 
---error ER_TABLESPACE_EXISTS
+--error ER_CANT_CREATE_TABLE
 CREATE TABLE t0(a INT PRIMARY KEY) ENGINE=InnoDB;
 
-# Remove the orphan file from fault 0.
---remove_file $MYSQLD_DATADIR/test/t0.ibd
-
 CREATE TABLE t0(a INT PRIMARY KEY) ENGINE=InnoDB;
 DROP TABLE t0;
 
diff --git a/mysql-test/suite/innodb/t/truncate.test b/mysql-test/suite/innodb/t/truncate.test
index 80b10cd8ca5..dc2b2a81484 100644
--- a/mysql-test/suite/innodb/t/truncate.test
+++ b/mysql-test/suite/innodb/t/truncate.test
@@ -43,7 +43,7 @@ call mtr.add_suppression("InnoDB: (Operating system )?[Ee]rror number");
 call mtr.add_suppression("InnoDB: Cannot create file '.*t1\\.ibd");
 FLUSH TABLES;
 --move_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/hidden.frm
---error ER_TABLESPACE_EXISTS
+--error ER_TABLE_EXISTS_ERROR
 CREATE TABLE t1 (a INT) ENGINE=InnoDB;
 --move_file $MYSQLD_DATADIR/test/hidden.frm $MYSQLD_DATADIR/test/t1.frm
 SELECT * FROM t1;
diff --git a/storage/innobase/dict/dict0crea.cc b/storage/innobase/dict/dict0crea.cc
index 16270039d77..34d2a00083a 100644
--- a/storage/innobase/dict/dict0crea.cc
+++ b/storage/innobase/dict/dict0crea.cc
@@ -362,55 +362,16 @@ dict_build_table_def_step(
 		ut_ad(DICT_TF_GET_ZIP_SSIZE(table->flags) == 0
 		      || dict_table_has_atomic_blobs(table));
 		/* Get a new tablespace ID */
-		ulint space_id;
-		dict_hdr_get_new_id(NULL, NULL, &space_id);
+		dict_hdr_get_new_id(NULL, NULL, &table->space_id);
 
 		DBUG_EXECUTE_IF(
 			"ib_create_table_fail_out_of_space_ids",
-			space_id = ULINT_UNDEFINED;
+			table->space_id = ULINT_UNDEFINED;
 		);
 
-		if (space_id == ULINT_UNDEFINED) {
+		if (table->space_id == ULINT_UNDEFINED) {
 			return DB_ERROR;
 		}
-
-		/* Determine the tablespace flags. */
-		bool	has_data_dir = DICT_TF_HAS_DATA_DIR(table->flags);
-		ulint	fsp_flags = dict_tf_to_fsp_flags(table->flags);
-		ut_ad(!has_data_dir || table->data_dir_path);
-		char*	filepath = fil_make_filepath(has_data_dir
-						     ? table->data_dir_path
-						     : nullptr,
-						     table->name, IBD,
-						     has_data_dir);
-
-		/* We create a new single-table tablespace for the table.
-		We initially let it be 4 pages:
-		- page 0 is the fsp header and an extent descriptor page,
-		- page 1 is an ibuf bitmap page,
-		- page 2 is the first inode page,
-		- page 3 will contain the root of the clustered index of
-		the table we create here. */
-
-		dberr_t err;
-		table->space = fil_ibd_create(
-			space_id, table->name, filepath, fsp_flags,
-			FIL_IBD_FILE_INITIAL_SIZE,
-			node->mode, node->key_id, &err);
-
-		ut_free(filepath);
-
-		if (!table->space) {
-			ut_ad(err != DB_SUCCESS);
-			return err;
-		}
-
-		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);
-		mtr.commit();
 	} else {
 		ut_ad(dict_tf_get_rec_format(table->flags)
 		      != REC_FORMAT_COMPRESSED);
@@ -457,7 +418,8 @@ dict_create_sys_indexes_tuple(
 
 	dict_sys.assert_locked();
 	ut_ad(index);
-	ut_ad(index->table->space || index->table->file_unreadable);
+	ut_ad(index->table->space || !UT_LIST_GET_LEN(index->table->indexes)
+	      || index->table->file_unreadable);
 	ut_ad(!index->table->space
 	      || index->table->space->id == index->table->space_id);
 	ut_ad(heap);
@@ -690,21 +652,20 @@ dict_build_index_def_step(
 
 	index = node->index;
 
-	table = index->table = node->table = dict_table_open_on_name(
-		node->table_name, TRUE, FALSE, DICT_ERR_IGNORE_NONE);
+	table = dict_table_open_on_name(
+		node->table_name, TRUE, FALSE, DICT_ERR_IGNORE_DROP);
 
-	if (table == NULL) {
-		return(DB_TABLE_NOT_FOUND);
+	if (!table) {
+		return DB_TABLE_NOT_FOUND;
 	}
 
+	index->table = table;
+
 	ut_ad((UT_LIST_GET_LEN(table->indexes) > 0)
 	      || dict_index_is_clust(index));
 
 	dict_hdr_get_new_id(NULL, &index->id, NULL);
 
-	/* Inherit the space id from the table; we store all indexes of a
-	table in the same tablespace */
-
 	node->page_no = FIL_NULL;
 	row = dict_create_sys_indexes_tuple(index, node->heap);
 	node->ind_row = row;
@@ -715,7 +676,7 @@ dict_build_index_def_step(
 	index->trx_id = trx->id;
 	ut_ad(table->def_trx_id <= trx->id);
 	table->def_trx_id = trx->id;
-	dict_table_close(table, true, false);
+	table->release();
 
 	return(DB_SUCCESS);
 }
@@ -949,9 +910,7 @@ tab_create_graph_create(
 /*====================*/
 	dict_table_t*	table,	/*!< in: table to create, built as a memory data
 				structure */
-	mem_heap_t*	heap,	/*!< in: heap where created */
-	fil_encryption_t mode,	/*!< in: encryption mode */
-	uint32_t	key_id)	/*!< in: encryption key_id */
+	mem_heap_t*	heap)	/*!< in: heap where created */
 {
 	tab_node_t*	node;
 
@@ -964,8 +923,6 @@ tab_create_graph_create(
 
 	node->state = TABLE_BUILD_TABLE_DEF;
 	node->heap = mem_heap_create(256);
-	node->mode = mode;
-	node->key_id = key_id;
 
 	node->tab_def = ins_node_create(INS_DIRECT, dict_sys.sys_tables,
 					heap);
@@ -986,6 +943,8 @@ tab_create_graph_create(
 @param[in]	index	index to create, built as a memory data structure
 @param[in]	table	table name
 @param[in,out]	heap	heap where created
+@param[in]	mode	encryption mode (for creating a table)
+@param[in]	key_id	encryption key identifier (for creating a table)
 @param[in]	add_v	new virtual columns added in the same clause with
 			add index
 @return own: index create node */
@@ -994,6 +953,8 @@ ind_create_graph_create(
 	dict_index_t*		index,
 	const char*		table,
 	mem_heap_t*		heap,
+	fil_encryption_t	mode,
+	uint32_t		key_id,
 	const dict_add_v_col_t*	add_v)
 {
 	ind_node_t*	node;
@@ -1007,6 +968,8 @@ ind_create_graph_create(
 
 	node->table_name = table;
 
+	node->key_id = key_id;
+	node->mode = mode;
 	node->add_v = add_v;
 
 	node->state = INDEX_BUILD_INDEX_DEF;
@@ -1068,7 +1031,6 @@ dict_create_table_step(
 	}
 
 	if (node->state == TABLE_BUILD_COL_DEF) {
-
 		if (node->col_no + DATA_N_SYS_COLS
 		    < (static_cast<ulint>(node->table->n_def)
 		       + static_cast<ulint>(node->table->n_v_def))) {
@@ -1166,6 +1128,40 @@ function_exit:
 	return(thr);
 }
 
+static dberr_t dict_create_index_space(const ind_node_t &node)
+{
+  dict_table_t *table= node.index->table;
+  if (table->space || (table->flags2 & DICT_TF2_DISCARDED))
+    return DB_SUCCESS;
+  ut_ad(table->space_id);
+  ut_ad(table->space_id < SRV_TMP_SPACE_ID);
+  /* Determine the tablespace flags. */
+  const bool has_data_dir= DICT_TF_HAS_DATA_DIR(table->flags);
+  ut_ad(!has_data_dir || table->data_dir_path);
+  char* filepath= fil_make_filepath(has_data_dir
+                                    ? table->data_dir_path : nullptr,
+                                    table->name, IBD, has_data_dir);
+  if (!filepath)
+    return DB_OUT_OF_MEMORY;
+
+  /* We create a new single-table tablespace for the table.
+  We initially let it be 4 pages:
+  - page 0 is the fsp header and an extent descriptor page,
+  - page 1 is an ibuf bitmap page,
+  - page 2 is the first inode page,
+  - page 3 will contain the root of the clustered index of
+  the table we create here. */
+  dberr_t err;
+  table->space= fil_ibd_create(table->space_id, table->name, filepath,
+                               dict_tf_to_fsp_flags(table->flags),
+                               FIL_IBD_FILE_INITIAL_SIZE,
+                               node.mode, node.key_id, &err);
+  ut_ad((err != DB_SUCCESS) == !table->space);
+  ut_free(filepath);
+
+  return err;
+}
+
 /***********************************************************//**
 Creates an index. This is a high-level function used in SQL execution
 graphs.
@@ -1210,6 +1206,12 @@ dict_create_index_step(
 	}
 
 	if (node->state == INDEX_BUILD_FIELD_DEF) {
+		err = dict_create_index_space(*node);
+		if (err != DB_SUCCESS) {
+			dict_mem_index_free(node->index);
+			node->index = nullptr;
+			goto function_exit;
+		}
 
 		if (node->field_no < (node->index)->n_fields) {
 
@@ -1226,11 +1228,10 @@ dict_create_index_step(
 	}
 
 	if (node->state == INDEX_ADD_TO_CACHE) {
-		ut_ad(node->index->table == node->table);
 		err = dict_index_add_to_cache(node->index, FIL_NULL,
 					      node->add_v);
 
-		ut_ad((node->index == NULL) == (err != DB_SUCCESS));
+		ut_ad(!node->index == (err != DB_SUCCESS));
 
 		if (!node->index) {
 			goto function_exit;
@@ -1239,7 +1240,7 @@ dict_create_index_step(
 		ut_ad(!node->index->is_instant());
 		ut_ad(node->index->n_core_null_bytes
 		      == ((dict_index_is_clust(node->index)
-			   && node->table->supports_instant())
+			   && node->index->table->supports_instant())
 			  ? dict_index_t::NO_CORE_NULL_BYTES
 			  : UT_BITS_IN_BYTES(
 				  unsigned(node->index->n_nullable))));
@@ -1256,18 +1257,18 @@ dict_create_index_step(
 				err = DB_OUT_OF_MEMORY;);
 
 		if (err != DB_SUCCESS) {
+			dict_table_t* table = node->index->table;
 			/* If this is a FTS index, we will need to remove
 			it from fts->cache->indexes list as well */
-			if ((node->index->type & DICT_FTS)
-			    && node->table->fts) {
+			if (!(node->index->type & DICT_FTS)) {
+			} else if (auto fts = table->fts) {
 				fts_index_cache_t*	index_cache;
 
-				mysql_mutex_lock(
-					&node->table->fts->cache->init_lock);
+				mysql_mutex_lock(&fts->cache->init_lock);
 
 				index_cache = (fts_index_cache_t*)
 					 fts_find_index_cache(
-						node->table->fts->cache,
+						fts->cache,
 						node->index);
 
 				if (index_cache->words) {
@@ -1276,17 +1277,16 @@ dict_create_index_step(
 				}
 
 				ib_vector_remove(
-					node->table->fts->cache->indexes,
+					fts->cache->indexes,
 					*reinterpret_cast<void**>(index_cache));
 
-				mysql_mutex_unlock(
-					&node->table->fts->cache->init_lock);
+				mysql_mutex_unlock(&fts->cache->init_lock);
 			}
 
 #ifdef BTR_CUR_HASH_ADAPT
 			ut_ad(!node->index->search_info->ref_count);
 #endif /* BTR_CUR_HASH_ADAPT */
-			dict_index_remove_from_cache(node->table, node->index);
+			dict_index_remove_from_cache(table, node->index);
 			node->index = NULL;
 
 			goto function_exit;
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index ef23e036876..000b9c8d8ef 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -2015,6 +2015,7 @@ fil_ibd_create(
 	pfs_os_file_t	file;
 	byte*		page;
 	bool		success;
+	mtr_t		mtr;
 	bool		has_data_dir = FSP_FLAGS_HAS_DATA_DIR(flags) != 0;
 
 	ut_ad(!is_system_tablespace(space_id));
@@ -2030,6 +2031,11 @@ fil_ibd_create(
 		return NULL;
 	}
 
+	mtr.start();
+	mtr.log_file_op(FILE_CREATE, space_id, path);
+	mtr.commit();
+	log_write_up_to(mtr.commit_lsn(), true);
+
 	ulint type;
 	static_assert(((UNIV_ZIP_SIZE_MIN >> 1) << 3) == 4096,
 		      "compatibility");
@@ -2183,12 +2189,11 @@ err_exit:
 						     crypt_data, mode)) {
 		space->punch_hole = punch_hole;
 		fil_node_t* node = space->add(path, file, size, false, true);
-		mtr_t mtr;
+		node->find_metadata(file);
 		mtr.start();
-		mtr.log_file_op(FILE_CREATE, space_id, node->name);
+		mtr.set_named_space(space);
+		fsp_header_init(space, size, &mtr);
 		mtr.commit();
-
-		node->find_metadata(file);
 		*err = DB_SUCCESS;
 		return space;
 	}
diff --git a/storage/innobase/fts/fts0fts.cc b/storage/innobase/fts/fts0fts.cc
index 8ffbeedef2f..cce6abc340b 100644
--- a/storage/innobase/fts/fts0fts.cc
+++ b/storage/innobase/fts/fts0fts.cc
@@ -1750,8 +1750,7 @@ fts_create_one_common_table(
 	}
 
 	dict_table_add_system_columns(new_table, heap);
-	error = row_create_table_for_mysql(new_table, trx,
-		FIL_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+	error = row_create_table_for_mysql(new_table, trx);
 
 	if (error == DB_SUCCESS) {
 
@@ -1765,7 +1764,9 @@ fts_create_one_common_table(
 			dict_mem_index_add_field(index, "key", 0);
 		}
 
-		error =	row_create_index_for_mysql(index, trx, NULL);
+		error =	row_create_index_for_mysql(index, trx, NULL,
+						   FIL_ENCRYPTION_DEFAULT,
+						   FIL_DEFAULT_ENCRYPTION_KEY);
 	}
 
 	if (error != DB_SUCCESS) {
@@ -1878,7 +1879,9 @@ fts_create_common_tables(
 				      DICT_UNIQUE, 1);
 	dict_mem_index_add_field(index, FTS_DOC_ID_COL_NAME, 0);
 
-	error =	row_create_index_for_mysql(index, trx, NULL);
+	error =	row_create_index_for_mysql(index, trx, NULL,
+					   FIL_ENCRYPTION_DEFAULT,
+					   FIL_DEFAULT_ENCRYPTION_KEY);
 
 func_exit:
 	if (error != DB_SUCCESS) {
@@ -1957,8 +1960,7 @@ fts_create_one_index_table(
 		FTS_INDEX_ILIST_LEN);
 
 	dict_table_add_system_columns(new_table, heap);
-	error = row_create_table_for_mysql(new_table, trx,
-		FIL_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+	error = row_create_table_for_mysql(new_table, trx);
 
 	if (error == DB_SUCCESS) {
 		dict_index_t*	index = dict_mem_index_create(
@@ -1967,7 +1969,9 @@ fts_create_one_index_table(
 		dict_mem_index_add_field(index, "word", 0);
 		dict_mem_index_add_field(index, "first_doc_id", 0);
 
-		error =	row_create_index_for_mysql(index, trx, NULL);
+		error =	row_create_index_for_mysql(index, trx, NULL,
+						   FIL_ENCRYPTION_DEFAULT,
+						   FIL_DEFAULT_ENCRYPTION_KEY);
 	}
 
 	if (error != DB_SUCCESS) {
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 925674c35b6..6316cc71cc1 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -10400,10 +10400,7 @@ err_col:
 		table->add_to_cache();
 	} else {
 		if (err == DB_SUCCESS) {
-			err = row_create_table_for_mysql(
-				table, m_trx,
-				fil_encryption_t(options->encryption),
-				uint32_t(options->encryption_key_id));
+			err = row_create_table_for_mysql(table, m_trx);
 			m_drop_before_rollback = (err == DB_SUCCESS);
 		}
 
@@ -10411,9 +10408,6 @@ err_col:
 				DBUG_SUICIDE(););
 	}
 
-	DBUG_EXECUTE_IF("ib_create_err_tablespace_exist",
-			err = DB_TABLESPACE_EXISTS;);
-
 	switch (err) {
 	case DB_SUCCESS:
 		ut_ad(table);
@@ -10422,7 +10416,6 @@ err_col:
 	default:
 		break;
 	case DB_DUPLICATE_KEY:
-	case DB_TABLESPACE_EXISTS:
 		char display_name[FN_REFLEN];
 		char* buf_end = innobase_convert_identifier(
 			display_name, sizeof(display_name) - 1,
@@ -10431,9 +10424,7 @@ err_col:
 
 		*buf_end = '\0';
 
-		my_error(err == DB_DUPLICATE_KEY
-			 ? ER_TABLE_EXISTS_ERROR
-			 : ER_TABLESPACE_EXISTS, MYF(0), display_name);
+		my_error(ER_TABLE_EXISTS_ERROR, MYF(0), display_name);
 	}
 
 	DBUG_RETURN(convert_error_code_to_mysql(err, m_flags, m_thd));
@@ -10462,6 +10453,7 @@ create_index(
 
 	/* Assert that "GEN_CLUST_INDEX" cannot be used as non-primary index */
 	ut_a(innobase_strcasecmp(key->name.str, innobase_index_reserve_name) != 0);
+	const ha_table_option_struct& o = *form->s->option_struct;
 
 	if (key->flags & (HA_SPATIAL | HA_FULLTEXT)) {
 		/* Only one of these can be specified at a time. */
@@ -10488,7 +10480,9 @@ create_index(
 
 		DBUG_RETURN(convert_error_code_to_mysql(
 				    row_create_index_for_mysql(
-					    index, trx, NULL),
+					    index, trx, NULL,
+					    fil_encryption_t(o.encryption),
+					    uint32_t(o.encryption_key_id)),
 				    table->flags, NULL));
 	}
 
@@ -10584,7 +10578,9 @@ create_index(
 	ulint flags = table->flags;
 
 	error = convert_error_code_to_mysql(
-		row_create_index_for_mysql(index, trx, field_lengths),
+		row_create_index_for_mysql(index, trx, field_lengths,
+					   fil_encryption_t(o.encryption),
+					   uint32_t(o.encryption_key_id)),
 		flags, NULL);
 
 	my_free(field_lengths);
@@ -12239,8 +12235,12 @@ int create_table_info_t::create_table(bool create_fk)
 		dict_index_t* index = dict_mem_index_create(
 			m_table, innobase_index_reserve_name,
 			DICT_CLUSTERED, 0);
+		const ha_table_option_struct& o = *m_form->s->option_struct;
 		error = convert_error_code_to_mysql(
-			row_create_index_for_mysql(index, m_trx, NULL),
+			row_create_index_for_mysql(
+				index, m_trx, NULL,
+				fil_encryption_t(o.encryption),
+				uint32_t(o.encryption_key_id)),
 			flags, m_thd);
 		if (error) {
 			DBUG_RETURN(error);
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 0bcb6114ab4..ef98687fae0 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -6086,6 +6086,8 @@ innodb_v_adjust_idx_col(
 /** Create index metadata in the data dictionary.
 @param[in,out]	trx	dictionary transaction
 @param[in,out]	index	index being created
+@param[in]	mode	encryption mode (for creating a table)
+@param[in]	key_id	encryption key identifier (for creating a table)
 @param[in]	add_v	virtual columns that are being added, or NULL
 @return the created index */
 MY_ATTRIBUTE((nonnull(1,2), warn_unused_result))
@@ -6094,13 +6096,15 @@ dict_index_t*
 create_index_dict(
 	trx_t*			trx,
 	dict_index_t*		index,
+	fil_encryption_t	mode,
+	uint32_t		key_id,
 	const dict_add_v_col_t* add_v)
 {
 	DBUG_ENTER("create_index_dict");
 
 	mem_heap_t* heap = mem_heap_create(512);
 	ind_node_t* node = ind_create_graph_create(
-		index, index->table->name.m_name, heap, add_v);
+		index, index->table->name.m_name, heap, mode, key_id, add_v);
 	que_thr_t* thr = pars_complete_graph_for_exec(node, trx, heap, NULL);
 
 	que_fork_start_command(
@@ -6800,18 +6804,13 @@ wrong_column_name:
 		/* Create the table. */
 		ctx->trx->dict_operation = true;
 
-		error = row_create_table_for_mysql(
-			ctx->new_table, ctx->trx, mode, key_id);
+		error = row_create_table_for_mysql(ctx->new_table, ctx->trx);
 
 		switch (error) {
 		case DB_SUCCESS:
 			DBUG_ASSERT(ctx->new_table->get_ref_count() == 0);
 			DBUG_ASSERT(ctx->new_table->id != 0);
 			break;
-		case DB_TABLESPACE_EXISTS:
-			my_error(ER_TABLESPACE_EXISTS, MYF(0),
-				 altered_table->s->table_name.str);
-			goto new_table_failed;
 		case DB_DUPLICATE_KEY:
 			my_error(HA_ERR_TABLE_EXIST, MYF(0),
 				 altered_table->s->table_name.str);
@@ -6831,7 +6830,8 @@ new_table_failed:
 		for (ulint a = 0; a < ctx->num_to_add_index; a++) {
 			dict_index_t* index = ctx->add_index[a];
 			const ulint n_v_col = index->get_new_n_vcol();
-			index = create_index_dict(ctx->trx, index, add_v);
+			index = create_index_dict(ctx->trx, index,
+						  mode, key_id, add_v);
 			error = ctx->trx->error_state;
 			if (error != DB_SUCCESS) {
 				if (index) {
@@ -6940,7 +6940,10 @@ error_handling_drop_uncached_1:
 						DB_OUT_OF_FILE_SPACE;
 					goto index_created;
 				});
-			index = create_index_dict(ctx->trx, index, add_v);
+			index = create_index_dict(ctx->trx, index,
+						  FIL_ENCRYPTION_DEFAULT,
+						  FIL_DEFAULT_ENCRYPTION_KEY,
+						  add_v);
 #ifndef DBUG_OFF
 index_created:
 #endif
diff --git a/storage/innobase/include/dict0crea.h b/storage/innobase/include/dict0crea.h
index 5fe35d14cd4..8153faad446 100644
--- a/storage/innobase/include/dict0crea.h
+++ b/storage/innobase/include/dict0crea.h
@@ -41,14 +41,14 @@ tab_create_graph_create(
 /*====================*/
 	dict_table_t*	table,		/*!< in: table to create, built as
 					a memory data structure */
-	mem_heap_t*	heap,		/*!< in: heap where created */
-	fil_encryption_t mode,		/*!< in: encryption mode */
-	uint32_t	key_id);	/*!< in: encryption key_id */
+	mem_heap_t*	heap);		/*!< in: heap where created */
 
 /** Creates an index create graph.
 @param[in]	index	index to create, built as a memory data structure
 @param[in]	table	table name
 @param[in,out]	heap	heap where created
+@param[in]	mode	encryption mode (for creating a table)
+@param[in]	key_id	encryption key identifier (for creating a table)
 @param[in]	add_v	new virtual columns added in the same clause with
 			add index
 @return own: index create node */
@@ -57,6 +57,8 @@ ind_create_graph_create(
 	dict_index_t*		index,
 	const char*		table,
 	mem_heap_t*		heap,
+	fil_encryption_t	mode,
+	uint32_t		key_id,
 	const dict_add_v_col_t*	add_v = NULL);
 
 /***********************************************************//**
@@ -218,8 +220,6 @@ struct tab_node_t{
 	/* Local storage for this graph node */
 	ulint		state;		/*!< node execution state */
 	ulint		col_no;		/*!< next column definition to insert */
-	uint		key_id;	/*!< encryption key_id */
-	fil_encryption_t mode;	/*!< encryption mode */
 	ulint		base_col_no;	/*!< next base column to insert */
 	mem_heap_t*	heap;		/*!< memory heap used as auxiliary
 					storage */
@@ -251,11 +251,12 @@ struct ind_node_t{
 	/* Local storage for this graph node */
 	ulint		state;		/*!< node execution state */
 	uint32_t	page_no;	/* root page number of the index */
-	dict_table_t*	table;		/*!< table which owns the index */
 	dtuple_t*	ind_row;	/* index definition row built */
 	ulint		field_no;	/* next field definition to insert */
 	mem_heap_t*	heap;		/*!< memory heap used as auxiliary
 					storage */
+	uint		key_id;		/*!< encryption key_id */
+	fil_encryption_t mode;		/*!< encryption mode */
 	const dict_add_v_col_t*
 			add_v;		/*!< new virtual columns that being
 					added along with an add index call */
diff --git a/storage/innobase/include/row0mysql.h b/storage/innobase/include/row0mysql.h
index 51c588f4a2d..d9b4902884a 100644
--- a/storage/innobase/include/row0mysql.h
+++ b/storage/innobase/include/row0mysql.h
@@ -355,9 +355,7 @@ row_create_table_for_mysql(
 	dict_table_t*	table,	/*!< in, own: table definition
 				(will be freed, or on DB_SUCCESS
 				added to the data dictionary cache) */
-	trx_t*		trx,	/*!< in/out: transaction */
-	fil_encryption_t mode,	/*!< in: encryption mode */
-	uint32_t	key_id)	/*!< in: encryption key_id */
+	trx_t*		trx)	/*!< in/out: transaction */
 	MY_ATTRIBUTE((warn_unused_result));
 
 /*********************************************************************//**
@@ -370,12 +368,14 @@ row_create_index_for_mysql(
 	dict_index_t*	index,		/*!< in, own: index definition
 					(will be freed) */
 	trx_t*		trx,		/*!< in: transaction handle */
-	const ulint*	field_lengths)	/*!< in: if not NULL, must contain
+	const ulint*	field_lengths,	/*!< in: if not NULL, must contain
 					dict_index_get_n_fields(index)
 					actual field lengths for the
 					index columns, which are
 					then checked for not being too
 					large. */
+	fil_encryption_t mode,	/*!< in: encryption mode */
+	uint32_t	key_id)	/*!< in: encryption key_id */
 	MY_ATTRIBUTE((warn_unused_result));
 /*********************************************************************//**
 The master thread in srv0srv.cc calls this regularly to drop tables which
diff --git a/storage/innobase/pars/pars0pars.cc b/storage/innobase/pars/pars0pars.cc
index ec922e85e29..5ec408342a8 100644
--- a/storage/innobase/pars/pars0pars.cc
+++ b/storage/innobase/pars/pars0pars.cc
@@ -1802,9 +1802,7 @@ pars_create_table(
 	}
 
 	dict_table_add_system_columns(table, heap);
-	node = tab_create_graph_create(table, heap,
-				       FIL_ENCRYPTION_DEFAULT,
-				       FIL_DEFAULT_ENCRYPTION_KEY);
+	node = tab_create_graph_create(table, heap);
 
 	table_sym->resolved = TRUE;
 	table_sym->token_type = SYM_TABLE;
@@ -1858,7 +1856,9 @@ pars_create_index(
 	}
 
 	node = ind_create_graph_create(index, table_sym->name,
-				       pars_sym_tab_global->heap);
+				       pars_sym_tab_global->heap,
+				       FIL_ENCRYPTION_DEFAULT,
+				       FIL_DEFAULT_ENCRYPTION_KEY);
 
 	table_sym->resolved = TRUE;
 	table_sym->token_type = SYM_TABLE;
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 16f5fbdaced..2e9dbee57ed 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -2303,9 +2303,7 @@ row_create_table_for_mysql(
 	dict_table_t*	table,	/*!< in, own: table definition
 				(will be freed, or on DB_SUCCESS
 				added to the data dictionary cache) */
-	trx_t*		trx,	/*!< in/out: transaction */
-	fil_encryption_t mode,	/*!< in: encryption mode */
-	uint32_t	key_id)	/*!< in: encryption key_id */
+	trx_t*		trx)	/*!< in/out: transaction */
 {
 	tab_node_t*	node;
 	mem_heap_t*	heap;
@@ -2345,7 +2343,7 @@ err_exit:
 
 	trx->dict_operation = true;
 
-	node = tab_create_graph_create(table, heap, mode, key_id);
+	node = tab_create_graph_create(table, heap);
 
 	thr = pars_complete_graph_for_exec(node, trx, heap, NULL);
 
@@ -2356,37 +2354,10 @@ err_exit:
 
 	err = trx->error_state;
 
-	switch (err) {
-	case DB_SUCCESS:
-		break;
-	case DB_OUT_OF_FILE_SPACE:
-		trx->error_state = DB_SUCCESS;
-		trx->rollback();
-
-		ib::warn() << "Cannot create table "
-			<< table->name
-			<< " because tablespace full";
-		dict_mem_table_free(table);
-		break;
-
-	case DB_UNSUPPORTED:
-	case DB_TOO_MANY_CONCURRENT_TRXS:
-		/* We already have .ibd file here. it should be deleted. */
-
-		if (dict_table_is_file_per_table(table)
-		    && fil_delete_tablespace(table->space_id) != DB_SUCCESS) {
-			ib::error() << "Cannot delete the file of table "
-				<< table->name;
-		}
-		/* fall through */
-
-	case DB_DUPLICATE_KEY:
-	case DB_TABLESPACE_EXISTS:
-	default:
+	if (err != DB_SUCCESS) {
 		trx->error_state = DB_SUCCESS;
 		trx->rollback();
 		dict_mem_table_free(table);
-		break;
 	}
 
 	que_graph_free((que_t*) que_node_get_parent(thr));
@@ -2406,12 +2377,14 @@ row_create_index_for_mysql(
 	dict_index_t*	index,		/*!< in, own: index definition
 					(will be freed) */
 	trx_t*		trx,		/*!< in: transaction handle */
-	const ulint*	field_lengths)	/*!< in: if not NULL, must contain
+	const ulint*	field_lengths,	/*!< in: if not NULL, must contain
 					dict_index_get_n_fields(index)
 					actual field lengths for the
 					index columns, which are
 					then checked for not being too
 					large. */
+	fil_encryption_t mode,		/*!< in: encryption mode */
+	uint32_t	key_id)		/*!< in: encryption key_id */
 {
 	ind_node_t*	node;
 	mem_heap_t*	heap;
@@ -2459,7 +2432,7 @@ row_create_index_for_mysql(
 
 		heap = mem_heap_create(512);
 		node = ind_create_graph_create(index, table->name.m_name,
-					       heap);
+					       heap, mode, key_id);
 
 		thr = pars_complete_graph_for_exec(node, trx, heap, NULL);