Merge 10.6 into 10.7

21 files changed, 528 insertions, 719 deletions

diff --git a/storage/innobase/btr/btr0bulk.cc b/storage/innobase/btr/btr0bulk.cc
index 3d5a0c2fb00..223d903c803 100644
--- a/storage/innobase/btr/btr0bulk.cc
+++ b/storage/innobase/btr/btr0bulk.cc
@@ -563,8 +563,9 @@ inline void PageBulk::finish()
 void PageBulk::commit(bool success)
 {
   finish();
-  if (success && !dict_index_is_clust(m_index) && page_is_leaf(m_page))
-    ibuf_set_bitmap_for_bulk_load(m_block, innobase_fill_factor == 100);
+  if (success && !m_index->is_clust() && page_is_leaf(m_page))
+    ibuf_set_bitmap_for_bulk_load(m_block, &m_mtr,
+                                  innobase_fill_factor == 100);
   m_mtr.commit();
 }
 
@@ -634,7 +635,7 @@ PageBulk::getSplitRec()
 		 < total_used_size / 2);
 
 	/* Keep at least one record on left page */
-	if (page_rec_is_second(rec, m_page)) {
+	if (page_rec_is_first(rec, m_page)) {
 		rec = page_rec_get_next(rec);
 		ut_ad(page_rec_is_user_rec(rec));
 	}
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 4729b4c0408..1d98ce058ed 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -588,7 +588,7 @@ bool buf_page_is_corrupted(bool check_lsn, const byte *read_buf,
 		DBUG_EXECUTE_IF(
 			"page_intermittent_checksum_mismatch", {
 			static int page_counter;
-			if (page_counter++ == 2) {
+			if (page_counter++ == 3) {
 				crc32++;
 			}
 		});
@@ -723,7 +723,7 @@ bool buf_page_is_corrupted(bool check_lsn, const byte *read_buf,
 			DBUG_EXECUTE_IF(
 				"page_intermittent_checksum_mismatch", {
 				static int page_counter;
-				if (page_counter++ == 2) return true;
+				if (page_counter++ == 3) return true;
 			});
 
 			if ((checksum_field1 != crc32
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 10ad82015ca..34362a6792f 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -685,8 +685,7 @@ dict_acquire_mdl_shared(dict_table_t *table,
   }
   else
   {
-    ut_ad(dict_sys.frozen());
-    ut_ad(!dict_sys.locked());
+    ut_ad(dict_sys.frozen_not_locked());
     db_len= dict_get_db_name_len(table->name.m_name);
   }
 
@@ -1003,7 +1002,7 @@ void dict_sys_t::lock_wait(SRW_LOCK_ARGS(const char *file, unsigned line))
     latch_ex_wait_start.store(0, std::memory_order_relaxed);
     ut_ad(!latch_readers);
     ut_ad(!latch_ex);
-    ut_d(latch_ex= true);
+    ut_d(latch_ex= pthread_self());
     return;
   }
 
@@ -1021,15 +1020,15 @@ void dict_sys_t::lock_wait(SRW_LOCK_ARGS(const char *file, unsigned line))
   latch.wr_lock(SRW_LOCK_ARGS(file, line));
   ut_ad(!latch_readers);
   ut_ad(!latch_ex);
-  ut_d(latch_ex= true);
+  ut_d(latch_ex= pthread_self());
 }
 
 #ifdef UNIV_PFS_RWLOCK
 ATTRIBUTE_NOINLINE void dict_sys_t::unlock()
 {
-  ut_ad(latch_ex);
+  ut_ad(latch_ex == pthread_self());
   ut_ad(!latch_readers);
-  ut_d(latch_ex= false);
+  ut_d(latch_ex= 0);
   latch.wr_unlock();
 }
 
@@ -1498,6 +1497,7 @@ dict_table_t::rename_tablespace(span<const char> new_name, bool replace) const
     err= DB_TABLESPACE_EXISTS;
   else
   {
+    space->x_lock();
     err= space->rename(path, true, replace);
     if (data_dir)
     {
@@ -1505,6 +1505,7 @@ dict_table_t::rename_tablespace(span<const char> new_name, bool replace) const
         new_name= {name.m_name, strlen(name.m_name)};
       RemoteDatafile::delete_link_file(new_name);
     }
+    space->x_unlock();
   }
 
   ut_free(path);
@@ -2757,17 +2758,6 @@ dict_index_build_internal_fts(
 }
 /*====================== FOREIGN KEY PROCESSING ========================*/
 
-/*********************************************************************//**
-Checks if a table is referenced by foreign keys.
-@return TRUE if table is referenced by a foreign key */
-ibool
-dict_table_is_referenced_by_foreign_key(
-/*====================================*/
-	const dict_table_t*	table)	/*!< in: InnoDB table */
-{
-	return(!table->referenced_set.empty());
-}
-
 /**********************************************************************//**
 Removes a foreign constraint struct from the dictionary cache. */
 void
diff --git a/storage/innobase/dict/drop.cc b/storage/innobase/dict/drop.cc
index edb6add0787..9013841ba5e 100644
--- a/storage/innobase/dict/drop.cc
+++ b/storage/innobase/dict/drop.cc
@@ -267,8 +267,7 @@ void trx_t::commit(std::vector<pfs_os_file_t> &deleted)
         if (btr_defragment_active)
           btr_defragment_remove_table(table);
         const fil_space_t *space= table->space;
-        ut_ad(!strstr(table->name.m_name, "/FTS_") ||
-              purge_sys.must_wait_FTS());
+        ut_ad(!p.second.is_aux_table() || purge_sys.must_wait_FTS());
         dict_sys.remove(table);
         if (const auto id= space ? space->id : 0)
         {
diff --git a/storage/innobase/fts/fts0fts.cc b/storage/innobase/fts/fts0fts.cc
index 31c6523c700..efa15251bbe 100644
--- a/storage/innobase/fts/fts0fts.cc
+++ b/storage/innobase/fts/fts0fts.cc
@@ -1373,6 +1373,13 @@ static dberr_t fts_drop_table(trx_t *trx, const char *table_name, bool rename)
       return err;
     }
 
+#ifdef UNIV_DEBUG
+    for (auto &p : trx->mod_tables)
+    {
+      if (p.first == table)
+	p.second.set_aux_table();
+    }
+#endif /* UNIV_DEBUG */
     return DB_SUCCESS;
   }
 
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index dd4f54092b1..acc179f28b5 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -54,7 +54,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include <my_bitmap.h>
 #include <mysql/service_thd_alloc.h>
 #include <mysql/service_thd_wait.h>
-#include "field.h"
+#include "sql_type_geom.h"
 #include "scope.h"
 #include "srv0srv.h"
 
@@ -11454,6 +11454,8 @@ innobase_fts_load_stopword(
 	trx_t*		trx,	/*!< in: transaction */
 	THD*		thd)	/*!< in: current thread */
 {
+  ut_ad(dict_sys.locked());
+
   const char *stopword_table= THDVAR(thd, ft_user_stopword_table);
   if (!stopword_table)
   {
@@ -11463,9 +11465,11 @@ innobase_fts_load_stopword(
     mysql_mutex_unlock(&LOCK_global_system_variables);
   }
 
-  return !high_level_read_only &&
-    fts_load_stopword(table, trx, stopword_table,
-                      THDVAR(thd, ft_enable_stopword), false);
+  table->fts->dict_locked= true;
+  bool success= fts_load_stopword(table, trx, stopword_table,
+                                  THDVAR(thd, ft_enable_stopword), false);
+  table->fts->dict_locked= false;
+  return success;
 }
 
 /** Parse the table name into normal name and remote path if needed.
@@ -12801,15 +12805,18 @@ int create_table_info_t::create_table(bool create_fk)
 	dberr_t err = create_fk ? create_foreign_keys() : DB_SUCCESS;
 
 	if (err == DB_SUCCESS) {
+		const dict_err_ignore_t ignore_err = m_trx->check_foreigns
+			? DICT_ERR_IGNORE_NONE : DICT_ERR_IGNORE_FK_NOKEY;
+
 		/* Check that also referencing constraints are ok */
 		dict_names_t	fk_tables;
 		err = dict_load_foreigns(m_table_name, nullptr,
 					 m_trx->id, true,
-					 DICT_ERR_IGNORE_NONE, fk_tables);
+					 ignore_err, fk_tables);
 		while (err == DB_SUCCESS && !fk_tables.empty()) {
 			dict_sys.load_table(
 				{fk_tables.front(), strlen(fk_tables.front())},
-				DICT_ERR_IGNORE_NONE);
+				ignore_err);
 			fk_tables.pop_front();
 		}
 	}
@@ -13090,96 +13097,59 @@ bool create_table_info_t::row_size_is_acceptable(
   return true;
 }
 
-/** Update a new table in an InnoDB database.
-@return error number */
-int
-create_table_info_t::create_table_update_dict()
+void create_table_info_t::create_table_update_dict(dict_table_t *table,
+                                                   THD *thd,
+                                                   const HA_CREATE_INFO &info,
+                                                   const TABLE &t)
 {
-	dict_table_t*	innobase_table;
-
-	DBUG_ENTER("create_table_update_dict");
-
-	innobase_table = dict_table_open_on_name(
-		m_table_name, false, DICT_ERR_IGNORE_NONE);
-
-	DBUG_ASSERT(innobase_table != 0);
-	if (innobase_table->fts != NULL) {
-		if (innobase_table->fts_doc_id_index == NULL) {
-			innobase_table->fts_doc_id_index
-				= dict_table_get_index_on_name(
-					innobase_table, FTS_DOC_ID_INDEX_NAME);
-			DBUG_ASSERT(innobase_table->fts_doc_id_index != NULL);
-		} else {
-			DBUG_ASSERT(innobase_table->fts_doc_id_index
-				    == dict_table_get_index_on_name(
-						innobase_table,
-						FTS_DOC_ID_INDEX_NAME));
-		}
-	}
-
-	DBUG_ASSERT((innobase_table->fts == NULL)
-		    == (innobase_table->fts_doc_id_index == NULL));
-
-	innobase_copy_frm_flags_from_create_info(innobase_table, m_create_info);
+  ut_ad(dict_sys.locked());
 
-	dict_stats_update(innobase_table, DICT_STATS_EMPTY_TABLE);
-
-	/* Load server stopword into FTS cache */
-	if (m_flags2 & DICT_TF2_FTS) {
-		if (!innobase_fts_load_stopword(innobase_table, NULL, m_thd)) {
-			innobase_table->release();
-			DBUG_RETURN(-1);
-		}
-
-		dict_sys.lock(SRW_LOCK_CALL);
-		fts_optimize_add_table(innobase_table);
-		dict_sys.unlock();
-	}
+  DBUG_ASSERT(table->get_ref_count());
+  if (table->fts)
+  {
+    if (!table->fts_doc_id_index)
+      table->fts_doc_id_index=
+        dict_table_get_index_on_name(table, FTS_DOC_ID_INDEX_NAME);
+    else
+      DBUG_ASSERT(table->fts_doc_id_index ==
+                  dict_table_get_index_on_name(table, FTS_DOC_ID_INDEX_NAME));
+  }
 
-	if (const Field* ai = m_form->found_next_number_field) {
-		ut_ad(ai->stored_in_db());
+  DBUG_ASSERT(!table->fts == !table->fts_doc_id_index);
 
-		ib_uint64_t	autoinc = m_create_info->auto_increment_value;
+  innobase_copy_frm_flags_from_create_info(table, &info);
 
-		if (autoinc == 0) {
-			autoinc = 1;
-		}
+  /* Load server stopword into FTS cache */
+  if (table->flags2 & DICT_TF2_FTS &&
+      innobase_fts_load_stopword(table, nullptr, thd))
+    fts_optimize_add_table(table);
 
-		innobase_table->autoinc_mutex.wr_lock();
-		dict_table_autoinc_initialize(innobase_table, autoinc);
+  if (const Field *ai = t.found_next_number_field)
+  {
+    ut_ad(ai->stored_in_db());
+    ib_uint64_t autoinc= info.auto_increment_value;
+    if (autoinc == 0)
+      autoinc= 1;
 
-		if (innobase_table->is_temporary()) {
-			/* AUTO_INCREMENT is not persistent for
-			TEMPORARY TABLE. Temporary tables are never
-			evicted. Keep the counter in memory only. */
-		} else {
-			const unsigned	col_no = innodb_col_no(ai);
-
-			innobase_table->persistent_autoinc
-				= static_cast<uint16_t>(
-					dict_table_get_nth_col_pos(
-						innobase_table, col_no, NULL)
-					+ 1)
-				& dict_index_t::MAX_N_FIELDS;
-
-			/* Persist the "last used" value, which
-			typically is AUTO_INCREMENT - 1.
-			In btr_create(), the value 0 was already written. */
-			if (--autoinc) {
-				btr_write_autoinc(
-					dict_table_get_first_index(
-						innobase_table),
-					autoinc);
-			}
-		}
+    table->autoinc_mutex.wr_lock();
+    dict_table_autoinc_initialize(table, autoinc);
 
-		innobase_table->autoinc_mutex.wr_unlock();
-	}
+    if (!table->is_temporary())
+    {
+      const unsigned col_no= innodb_col_no(ai);
+      table->persistent_autoinc= static_cast<uint16_t>
+        (dict_table_get_nth_col_pos(table, col_no, nullptr) + 1) &
+        dict_index_t::MAX_N_FIELDS;
+      /* Persist the "last used" value, which typically is AUTO_INCREMENT - 1.
+      In btr_create(), the value 0 was already written. */
+      if (--autoinc)
+        btr_write_autoinc(dict_table_get_first_index(table), autoinc);
+    }
 
-	innobase_parse_hint_from_comment(m_thd, innobase_table, m_form->s);
+    table->autoinc_mutex.wr_unlock();
+  }
 
-	dict_table_close(innobase_table);
-	DBUG_RETURN(0);
+  innobase_parse_hint_from_comment(thd, table, t.s);
 }
 
 /** Allocate a new trx. */
@@ -13196,89 +13166,80 @@ create_table_info_t::allocate_trx()
 @param[in]	create_info	Create info (including create statement string).
 @param[in]	file_per_table	whether to create .ibd file
 @param[in,out]	trx		dictionary transaction, or NULL to create new
-@return	0 if success else error number. */
-inline int
-ha_innobase::create(
-	const char*	name,
-	TABLE*		form,
-	HA_CREATE_INFO*	create_info,
-	bool		file_per_table,
-	trx_t*		trx)
+@return error code
+@retval	0 on success */
+int
+ha_innobase::create(const char *name, TABLE *form, HA_CREATE_INFO *create_info,
+                    bool file_per_table, trx_t *trx= nullptr)
 {
-	char		norm_name[FN_REFLEN];	/* {database}/{tablename} */
-	char		remote_path[FN_REFLEN];	/* Absolute path of table */
+  char norm_name[FN_REFLEN];	/* {database}/{tablename} */
+  char remote_path[FN_REFLEN];	/* Absolute path of table */
 
-	DBUG_ENTER("ha_innobase::create");
+  DBUG_ENTER("ha_innobase::create");
+  DBUG_ASSERT(form->s == table_share);
+  DBUG_ASSERT(table_share->table_type == TABLE_TYPE_SEQUENCE ||
+              table_share->table_type == TABLE_TYPE_NORMAL);
 
-	DBUG_ASSERT(form->s == table_share);
-	DBUG_ASSERT(table_share->table_type == TABLE_TYPE_SEQUENCE
-		    || table_share->table_type == TABLE_TYPE_NORMAL);
+  create_table_info_t info(ha_thd(), form, create_info, norm_name,
+                           remote_path, file_per_table, trx);
 
-	create_table_info_t	info(ha_thd(),
-				     form,
-				     create_info,
-				     norm_name,
-				     remote_path,
-				     file_per_table, trx);
+  int error= info.initialize();
+  if (!error)
+    error= info.prepare_create_table(name, !trx);
+  if (error)
+    DBUG_RETURN(error);
 
-	{
-		int error = info.initialize();
-		if (!error) {
-			error = info.prepare_create_table(name, !trx);
-		}
-		if (error) {
-			if (trx) {
-				trx_rollback_for_mysql(trx);
-				row_mysql_unlock_data_dictionary(trx);
-			}
-			DBUG_RETURN(error);
-		}
-	}
+  const bool own_trx= !trx;
+  if (own_trx)
+  {
+    info.allocate_trx();
+    trx= info.trx();
+    DBUG_ASSERT(trx_state_eq(trx, TRX_STATE_NOT_STARTED));
 
-	const bool own_trx = !trx;
-	int error = 0;
+    if (!(info.flags2() & DICT_TF2_TEMPORARY))
+    {
+      trx_start_for_ddl(trx);
+      if (dberr_t err= lock_sys_tables(trx))
+        error= convert_error_code_to_mysql(err, 0, nullptr);
+    }
+    row_mysql_lock_data_dictionary(trx);
+  }
 
-	if (own_trx) {
-		info.allocate_trx();
-		trx = info.trx();
-		DBUG_ASSERT(trx_state_eq(trx, TRX_STATE_NOT_STARTED));
-	}
-	if (own_trx && !(info.flags2() & DICT_TF2_TEMPORARY)) {
-		trx_start_for_ddl(trx);
-		if (dberr_t err = lock_sys_tables(trx)) {
-			error = convert_error_code_to_mysql(err, 0, nullptr);
-		}
-	}
-	if (own_trx) {
-		row_mysql_lock_data_dictionary(trx);
-	}
+  if (!error)
+    error= info.create_table(own_trx);
 
-	if (!error) {
-		error = info.create_table(own_trx);
-	}
+  if (own_trx || (info.flags2() & DICT_TF2_TEMPORARY))
+  {
+    if (error)
+      trx_rollback_for_mysql(trx);
+    else
+    {
+      std::vector<pfs_os_file_t> deleted;
+      trx->commit(deleted);
+      ut_ad(deleted.empty());
+      info.table()->acquire();
+      info.create_table_update_dict(info.table(), info.thd(),
+                                    *create_info, *form);
+    }
 
-	if (error) {
-		/* Rollback will drop the being-created table. */
-		trx_rollback_for_mysql(trx);
-		row_mysql_unlock_data_dictionary(trx);
-	} else {
-		/* When this is invoked as part of ha_innobase::truncate(),
-		the old copy of the table will be deleted here. */
-		std::vector<pfs_os_file_t> deleted;
-		trx->commit(deleted);
-		row_mysql_unlock_data_dictionary(trx);
-		for (pfs_os_file_t d : deleted) os_file_close(d);
-		error = info.create_table_update_dict();
-		if (!(info.flags2() & DICT_TF2_TEMPORARY)) {
-			log_write_up_to(trx->commit_lsn, true);
-		}
-	}
+    if (own_trx)
+    {
+      row_mysql_unlock_data_dictionary(trx);
 
-	if (own_trx) {
-		trx->free();
-	}
+      if (!error)
+      {
+        dict_stats_update(info.table(), DICT_STATS_EMPTY_TABLE);
+        if (!info.table()->is_temporary())
+          log_write_up_to(trx->commit_lsn, true);
+        info.table()->release();
+      }
+      trx->free();
+    }
+  }
+  else if (!error && m_prebuilt)
+    m_prebuilt->table= info.table();
 
-	DBUG_RETURN(error);
+  DBUG_RETURN(error);
 }
 
 /** Create a new table to an InnoDB database.
@@ -13286,13 +13247,10 @@ ha_innobase::create(
 @param[in]	form		Table format; columns and index information.
 @param[in]	create_info	Create info (including create statement string).
 @return	0 if success else error number. */
-int
-ha_innobase::create(
-	const char*	name,
-	TABLE*		form,
-	HA_CREATE_INFO*	create_info)
+int ha_innobase::create(const char *name, TABLE *form,
+                        HA_CREATE_INFO *create_info)
 {
-	return create(name, form, create_info, srv_file_per_table);
+  return create(name, form, create_info, srv_file_per_table);
 }
 
 /*****************************************************************//**
@@ -13821,229 +13779,247 @@ static dberr_t innobase_rename_table(trx_t *trx, const char *from,
 @retval	0	on success */
 int ha_innobase::truncate()
 {
-	DBUG_ENTER("ha_innobase::truncate");
+  DBUG_ENTER("ha_innobase::truncate");
 
-	update_thd();
+  update_thd();
 
-	if (is_read_only()) {
-		DBUG_RETURN(HA_ERR_TABLE_READONLY);
-	}
+  if (is_read_only())
+    DBUG_RETURN(HA_ERR_TABLE_READONLY);
 
-	HA_CREATE_INFO	info;
-	dict_table_t*	ib_table = m_prebuilt->table;
-	info.init();
-	update_create_info_from_table(&info, table);
-	switch (dict_tf_get_rec_format(ib_table->flags)) {
-	case REC_FORMAT_REDUNDANT:
-		info.row_type = ROW_TYPE_REDUNDANT;
-		break;
-	case REC_FORMAT_COMPACT:
-		info.row_type = ROW_TYPE_COMPACT;
-		break;
-	case REC_FORMAT_COMPRESSED:
-		info.row_type = ROW_TYPE_COMPRESSED;
-		break;
-	case REC_FORMAT_DYNAMIC:
-		info.row_type = ROW_TYPE_DYNAMIC;
-		break;
-	}
+  HA_CREATE_INFO info;
+  dict_table_t *ib_table= m_prebuilt->table;
+  info.init();
+  update_create_info_from_table(&info, table);
+  switch (dict_tf_get_rec_format(ib_table->flags)) {
+  case REC_FORMAT_REDUNDANT:
+    info.row_type= ROW_TYPE_REDUNDANT;
+    break;
+  case REC_FORMAT_COMPACT:
+    info.row_type= ROW_TYPE_COMPACT;
+    break;
+  case REC_FORMAT_COMPRESSED:
+    info.row_type= ROW_TYPE_COMPRESSED;
+    break;
+  case REC_FORMAT_DYNAMIC:
+    info.row_type= ROW_TYPE_DYNAMIC;
+    break;
+  }
 
-	const auto stored_lock = m_prebuilt->stored_select_lock_type;
-	trx_t*	trx = innobase_trx_allocate(m_user_thd);
-	trx_start_for_ddl(trx);
+  const auto stored_lock= m_prebuilt->stored_select_lock_type;
+  trx_t *trx= innobase_trx_allocate(m_user_thd);
+  trx_start_for_ddl(trx);
 
-	if (ib_table->is_temporary()) {
-		info.options|= HA_LEX_CREATE_TMP_TABLE;
-		btr_drop_temporary_table(*ib_table);
-		m_prebuilt->table = nullptr;
-		row_prebuilt_free(m_prebuilt);
-		m_prebuilt = nullptr;
-		my_free(m_upd_buf);
-		m_upd_buf = nullptr;
-		m_upd_buf_size = 0;
+  if (ib_table->is_temporary())
+  {
+    info.options|= HA_LEX_CREATE_TMP_TABLE;
+    btr_drop_temporary_table(*ib_table);
+    m_prebuilt->table= nullptr;
+    row_prebuilt_free(m_prebuilt);
+    m_prebuilt= nullptr;
+    my_free(m_upd_buf);
+    m_upd_buf= nullptr;
+    m_upd_buf_size= 0;
 
-		row_mysql_lock_data_dictionary(trx);
-		ib_table->release();
-		dict_sys.remove(ib_table, false, true);
+    row_mysql_lock_data_dictionary(trx);
+    ib_table->release();
+    dict_sys.remove(ib_table, false, true);
+    int err= create(ib_table->name.m_name, table, &info, true, trx);
+    row_mysql_unlock_data_dictionary(trx);
 
-		int err = create(ib_table->name.m_name, table, &info, true,
-				 trx);
-		ut_ad(!err);
-		if (!err) {
-			err = open(ib_table->name.m_name, 0, 0);
-			m_prebuilt->stored_select_lock_type = stored_lock;
-		}
+    ut_ad(!err);
+    if (!err)
+    {
+      err= open(ib_table->name.m_name, 0, 0);
+      m_prebuilt->table->release();
+      m_prebuilt->stored_select_lock_type= stored_lock;
+    }
 
-		trx->free();
+    trx->free();
 
 #ifdef BTR_CUR_HASH_ADAPT
-		if (UT_LIST_GET_LEN(ib_table->freed_indexes)) {
-			ib_table->vc_templ = nullptr;
-			ib_table->id = 0;
-			DBUG_RETURN(err);
-		}
+    if (UT_LIST_GET_LEN(ib_table->freed_indexes))
+    {
+      ib_table->vc_templ= nullptr;
+      ib_table->id= 0;
+    }
+    else
 #endif /* BTR_CUR_HASH_ADAPT */
+    dict_mem_table_free(ib_table);
 
-		dict_mem_table_free(ib_table);
-		DBUG_RETURN(err);
-	}
+    DBUG_RETURN(err);
+  }
 
-	mem_heap_t*	heap = mem_heap_create(1000);
+  mem_heap_t *heap= mem_heap_create(1000);
 
-	dict_get_and_save_data_dir_path(ib_table);
-	info.data_file_name = ib_table->data_dir_path;
-	const char* temp_name = dict_mem_create_temporary_tablename(
-		heap, ib_table->name.m_name, ib_table->id);
-	const char* name = mem_heap_strdup(heap, ib_table->name.m_name);
+  dict_get_and_save_data_dir_path(ib_table);
+  info.data_file_name= ib_table->data_dir_path;
+  const char *temp_name=
+    dict_mem_create_temporary_tablename(heap,
+                                        ib_table->name.m_name, ib_table->id);
+  const char *name= mem_heap_strdup(heap, ib_table->name.m_name);
 
-	dict_table_t *table_stats = nullptr, *index_stats = nullptr;
-	MDL_ticket *mdl_table = nullptr, *mdl_index = nullptr;
+  dict_table_t *table_stats = nullptr, *index_stats = nullptr;
+  MDL_ticket *mdl_table = nullptr, *mdl_index = nullptr;
 
-	dberr_t error = DB_SUCCESS;
+  dberr_t error= DB_SUCCESS;
 
-	dict_sys.freeze(SRW_LOCK_CALL);
-	for (const dict_foreign_t* f : ib_table->referenced_set) {
-		if (dict_table_t* child = f->foreign_table) {
-			error = lock_table_for_trx(child, trx, LOCK_X);
-			if (error != DB_SUCCESS) {
-				break;
-			}
-		}
-	}
-	dict_sys.unfreeze();
+  dict_sys.freeze(SRW_LOCK_CALL);
+  for (const dict_foreign_t *f : ib_table->referenced_set)
+    if (dict_table_t *child= f->foreign_table)
+      if ((error= lock_table_for_trx(child, trx, LOCK_X)) != DB_SUCCESS)
+        break;
+  dict_sys.unfreeze();
 
-	if (error == DB_SUCCESS) {
-		error = lock_table_for_trx(ib_table, trx, LOCK_X);
-	}
+  if (error == DB_SUCCESS)
+    error= lock_table_for_trx(ib_table, trx, LOCK_X);
 
-	const bool fts = error == DB_SUCCESS
-		&& ib_table->flags2 & (DICT_TF2_FTS_HAS_DOC_ID | DICT_TF2_FTS);
+  const bool fts= error == DB_SUCCESS &&
+    ib_table->flags2 & (DICT_TF2_FTS_HAS_DOC_ID | DICT_TF2_FTS);
 
-	if (fts) {
-		fts_optimize_remove_table(ib_table);
-		purge_sys.stop_FTS(*ib_table);
-		error = fts_lock_tables(trx, *ib_table);
-	}
+  if (fts)
+  {
+    fts_optimize_remove_table(ib_table);
+    purge_sys.stop_FTS(*ib_table);
+    error= fts_lock_tables(trx, *ib_table);
+  }
 
-	/* Wait for purge threads to stop using the table. */
-	for (uint n = 15; ib_table->get_ref_count() > 1; ) {
-		if (!--n) {
-			error = DB_LOCK_WAIT_TIMEOUT;
-			break;
-		}
+  /* Wait for purge threads to stop using the table. */
+  for (uint n = 15; ib_table->get_ref_count() > 1; )
+  {
+    if (!--n)
+    {
+      error= DB_LOCK_WAIT_TIMEOUT;
+      break;
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  }
 
-		std::this_thread::sleep_for(std::chrono::milliseconds(50));
-	}
+  if (error == DB_SUCCESS && dict_stats_is_persistent_enabled(ib_table) &&
+      !ib_table->is_stats_table())
+  {
+    table_stats= dict_table_open_on_name(TABLE_STATS_NAME, false,
+                                         DICT_ERR_IGNORE_NONE);
+    if (table_stats)
+    {
+      dict_sys.freeze(SRW_LOCK_CALL);
+      table_stats= dict_acquire_mdl_shared<false>(table_stats, m_user_thd,
+                                                  &mdl_table);
+      dict_sys.unfreeze();
+    }
+    index_stats= dict_table_open_on_name(INDEX_STATS_NAME, false,
+                                         DICT_ERR_IGNORE_NONE);
+    if (index_stats)
+    {
+      dict_sys.freeze(SRW_LOCK_CALL);
+      index_stats= dict_acquire_mdl_shared<false>(index_stats, m_user_thd,
+                                                  &mdl_index);
+      dict_sys.unfreeze();
+    }
 
-	if (error == DB_SUCCESS && dict_stats_is_persistent_enabled(ib_table)
-	    && !ib_table->is_stats_table()) {
-		table_stats= dict_table_open_on_name(TABLE_STATS_NAME, false,
-						     DICT_ERR_IGNORE_NONE);
-		if (table_stats) {
-			dict_sys.freeze(SRW_LOCK_CALL);
-			table_stats = dict_acquire_mdl_shared<false>(
-				table_stats, m_user_thd, &mdl_table);
-			dict_sys.unfreeze();
-		}
-		index_stats = dict_table_open_on_name(INDEX_STATS_NAME, false,
-						      DICT_ERR_IGNORE_NONE);
-		if (index_stats) {
-			dict_sys.freeze(SRW_LOCK_CALL);
-			index_stats = dict_acquire_mdl_shared<false>(
-				index_stats, m_user_thd, &mdl_index);
-			dict_sys.unfreeze();
-		}
+    if (table_stats && index_stats &&
+        !strcmp(table_stats->name.m_name, TABLE_STATS_NAME) &&
+        !strcmp(index_stats->name.m_name, INDEX_STATS_NAME) &&
+        !(error= lock_table_for_trx(table_stats, trx, LOCK_X)))
+      error= lock_table_for_trx(index_stats, trx, LOCK_X);
+  }
 
-		if (table_stats && index_stats
-		    && !strcmp(table_stats->name.m_name, TABLE_STATS_NAME)
-		    && !strcmp(index_stats->name.m_name, INDEX_STATS_NAME) &&
-		    !(error = lock_table_for_trx(table_stats, trx, LOCK_X))) {
-			error = lock_table_for_trx(index_stats, trx, LOCK_X);
-		}
-	}
+  if (error == DB_SUCCESS)
+    error= lock_sys_tables(trx);
 
-	if (error == DB_SUCCESS) {
-		error = lock_sys_tables(trx);
-	}
+  std::vector<pfs_os_file_t> deleted;
 
-	row_mysql_lock_data_dictionary(trx);
+  row_mysql_lock_data_dictionary(trx);
 
-	if (error == DB_SUCCESS) {
-		error = innobase_rename_table(trx, ib_table->name.m_name,
-					      temp_name, false);
+  if (error == DB_SUCCESS)
+  {
+    error= innobase_rename_table(trx, ib_table->name.m_name, temp_name, false);
+    if (error == DB_SUCCESS)
+      error= trx->drop_table(*ib_table);
+  }
 
-		if (error == DB_SUCCESS) {
-			error = trx->drop_table(*ib_table);
-		}
-	}
+  int err = convert_error_code_to_mysql(error, ib_table->flags, m_user_thd);
+  const auto update_time = ib_table->update_time;
 
-	int err = convert_error_code_to_mysql(error, ib_table->flags,
-					      m_user_thd);
-	if (err) {
-		trx_rollback_for_mysql(trx);
-		if (fts) {
-			fts_optimize_add_table(ib_table);
-			purge_sys.resume_FTS();
-		}
-		row_mysql_unlock_data_dictionary(trx);
-	} else {
-		const auto update_time = ib_table->update_time;
-		const auto stored_lock = m_prebuilt->stored_select_lock_type;
-		const auto def_trx_id = ib_table->def_trx_id;
-		ib_table->release();
-		m_prebuilt->table = nullptr;
+  if (err)
+  {
+    trx_rollback_for_mysql(trx);
+    if (fts)
+      fts_optimize_add_table(ib_table);
+  }
+  else
+  {
+    const auto def_trx_id= ib_table->def_trx_id;
+    ib_table->release();
+    m_prebuilt->table= nullptr;
 
-		err = create(name, table, &info,
-			     dict_table_is_file_per_table(ib_table), trx);
-		/* On success, create() durably committed trx. */
-		if (fts) {
-			purge_sys.resume_FTS();
-		}
+    err= create(name, table, &info, dict_table_is_file_per_table(ib_table),
+                trx);
+    if (!err)
+    {
+      m_prebuilt->table->acquire();
+      create_table_info_t::create_table_update_dict(m_prebuilt->table,
+                                                    m_user_thd, info, *table);
+      trx->commit(deleted);
+    }
+    else
+    {
+      trx_rollback_for_mysql(trx);
+      m_prebuilt->table= dict_table_open_on_name(name, true,
+                                                 DICT_ERR_IGNORE_FK_NOKEY);
+      m_prebuilt->table->def_trx_id= def_trx_id;
+    }
+    dict_names_t fk_tables;
+    dict_load_foreigns(m_prebuilt->table->name.m_name, nullptr, 1, true,
+                       DICT_ERR_IGNORE_FK_NOKEY, fk_tables);
+    for (const char *f : fk_tables)
+      dict_sys.load_table({f, strlen(f)});
+  }
 
-		if (err) {
-reload:
-			m_prebuilt->table = dict_table_open_on_name(
-				name, false, DICT_ERR_IGNORE_NONE);
-			m_prebuilt->table->def_trx_id = def_trx_id;
-		} else {
-			row_prebuilt_t* prebuilt = m_prebuilt;
-			uchar* upd_buf = m_upd_buf;
-			ulint upd_buf_size = m_upd_buf_size;
-			/* Mimic ha_innobase::close(). */
-			m_prebuilt = nullptr;
-			m_upd_buf = nullptr;
-			m_upd_buf_size = 0;
+  if (fts)
+    purge_sys.resume_FTS();
 
-			err = open(name, 0, 0);
+  row_mysql_unlock_data_dictionary(trx);
+  for (pfs_os_file_t d : deleted) os_file_close(d);
 
-			if (!err) {
-				m_prebuilt->stored_select_lock_type
-					= stored_lock;
-				m_prebuilt->table->update_time = update_time;
-				row_prebuilt_free(prebuilt);
-				my_free(upd_buf);
-			} else {
-				/* Revert to the old table. */
-				m_prebuilt = prebuilt;
-				m_upd_buf = upd_buf;
-				m_upd_buf_size = upd_buf_size;
-				goto reload;
-			}
-		}
-	}
+  if (!err)
+  {
+    dict_stats_update(m_prebuilt->table, DICT_STATS_EMPTY_TABLE);
+    log_write_up_to(trx->commit_lsn, true);
+    row_prebuilt_t *prebuilt= m_prebuilt;
+    uchar *upd_buf= m_upd_buf;
+    ulint upd_buf_size= m_upd_buf_size;
+    /* Mimic ha_innobase::close(). */
+    m_prebuilt= nullptr;
+    m_upd_buf= nullptr;
+    m_upd_buf_size= 0;
+
+    err= open(name, 0, 0);
+    if (!err)
+    {
+      m_prebuilt->stored_select_lock_type= stored_lock;
+      m_prebuilt->table->update_time= update_time;
+      row_prebuilt_free(prebuilt);
+      my_free(upd_buf);
+    }
+    else
+    {
+      /* Revert to the old table. */
+      m_prebuilt= prebuilt;
+      m_upd_buf= upd_buf;
+      m_upd_buf_size= upd_buf_size;
+    }
+  }
 
-	trx->free();
+  trx->free();
 
-	mem_heap_free(heap);
+  mem_heap_free(heap);
 
-	if (table_stats) {
-		dict_table_close(table_stats, false, m_user_thd, mdl_table);
-	}
-	if (index_stats) {
-		dict_table_close(index_stats, false, m_user_thd, mdl_index);
-	}
+  if (table_stats)
+    dict_table_close(table_stats, false, m_user_thd, mdl_table);
+  if (index_stats)
+    dict_table_close(index_stats, false, m_user_thd, mdl_index);
 
-	DBUG_RETURN(err);
+  DBUG_RETURN(err);
 }
 
 /*********************************************************************//**
@@ -15637,30 +15613,12 @@ delete is then allowed internally to resolve a duplicate key conflict in
 REPLACE, not an update.
 @return > 0 if referenced by a FOREIGN KEY */
 
-uint
-ha_innobase::referenced_by_foreign_key(void)
-/*========================================*/
-{
-	if (dict_table_is_referenced_by_foreign_key(m_prebuilt->table)) {
-
-		return(1);
-	}
-
-	return(0);
-}
-
-/*******************************************************************//**
-Frees the foreign key create info for a table stored in InnoDB, if it is
-non-NULL. */
-
-void
-ha_innobase::free_foreign_key_create_info(
-/*======================================*/
-	char*	str)	/*!< in, own: create info string to free */
+uint ha_innobase::referenced_by_foreign_key()
 {
-	if (str != NULL) {
-		my_free(str);
-	}
+  dict_sys.freeze(SRW_LOCK_CALL);
+  const bool empty= m_prebuilt->table->referenced_set.empty();
+  dict_sys.unfreeze();
+  return !empty;
 }
 
 /*******************************************************************//**
@@ -20639,6 +20597,26 @@ bool ha_innobase::can_convert_blob(const Field_blob *field,
   return true;
 }
 
+
+bool ha_innobase::can_convert_nocopy(const Field &field,
+                                     const Column_definition &new_type) const
+{
+  if (const Field_string *tf= dynamic_cast<const Field_string *>(&field))
+    return can_convert_string(tf, new_type);
+
+  if (const Field_varstring *tf= dynamic_cast<const Field_varstring *>(&field))
+    return can_convert_varstring(tf, new_type);
+
+  if (dynamic_cast<const Field_geom *>(&field))
+    return false;
+
+  if (const Field_blob *tf= dynamic_cast<const Field_blob *>(&field))
+    return can_convert_blob(tf, new_type);
+
+  return false;
+}
+
+
 Compare_keys ha_innobase::compare_key_parts(
     const Field &old_field, const Column_definition &new_field,
     const KEY_PART_INFO &old_part, const KEY_PART_INFO &new_part) const
@@ -20649,7 +20627,7 @@ Compare_keys ha_innobase::compare_key_parts(
 
   if (!is_equal)
   {
-    if (!old_field.can_be_converted_by_engine(new_field))
+    if (!old_field.table->file->can_convert_nocopy(old_field, new_field))
       return Compare_keys::NotEqual;
 
     if (!Charset(old_cs).eq_collation_specific_names(new_cs))
diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h
index 01acde3d8e6..1f42bf180a8 100644
--- a/storage/innobase/handler/ha_innodb.h
+++ b/storage/innobase/handler/ha_innodb.h
@@ -190,12 +190,12 @@ public:
 
 	void update_create_info(HA_CREATE_INFO* create_info) override;
 
-	inline int create(
+	int create(
 		const char*		name,
 		TABLE*			form,
 		HA_CREATE_INFO*		create_info,
 		bool			file_per_table,
-		trx_t*			trx = NULL);
+		trx_t*			trx);
 
 	int create(
 		const char*		name,
@@ -225,7 +225,7 @@ public:
 
 	uint referenced_by_foreign_key() override;
 
-	void free_foreign_key_create_info(char* str) override;
+	void free_foreign_key_create_info(char* str) override { my_free(str); }
 
 	uint lock_count(void) const override;
 
@@ -422,15 +422,9 @@ public:
 	@retval	false if pushed (always) */
 	bool rowid_filter_push(Rowid_filter *rowid_filter) override;
 
-	bool
-	can_convert_string(const Field_string* field,
-			   const Column_definition& new_field) const override;
-	bool can_convert_varstring(
-	    const Field_varstring* field,
-	    const Column_definition& new_field) const override;
-	bool
-	can_convert_blob(const Field_blob* field,
-			 const Column_definition& new_field) const override;
+	bool can_convert_nocopy(const Field &field,
+				const Column_definition& new_field) const
+		override;
 
 	/** @return whether innodb_strict_mode is active */
 	static bool is_innodb_strict_mode(THD* thd);
@@ -445,6 +439,16 @@ public:
 			  const KEY_PART_INFO& new_part) const override;
 
 protected:
+	bool
+	can_convert_string(const Field_string* field,
+			   const Column_definition& new_field) const;
+	bool can_convert_varstring(
+	    const Field_varstring* field,
+	    const Column_definition& new_field) const;
+	bool
+	can_convert_blob(const Field_blob* field,
+			 const Column_definition& new_field) const;
+
 	dberr_t innobase_get_autoinc(ulonglong* value);
 	dberr_t innobase_lock_autoinc();
 	ulonglong innobase_peek_autoinc();
@@ -639,8 +643,9 @@ public:
 	@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();
+  static void create_table_update_dict(dict_table_t* table, THD* thd,
+                                       const HA_CREATE_INFO& info,
+                                       const TABLE& t);
 
 	/** Validates the create options. Checks that the options
 	KEY_BLOCK_SIZE, ROW_FORMAT, DATA DIRECTORY, TEMPORARY & TABLESPACE
@@ -700,12 +705,13 @@ public:
 	trx_t* trx() const
 	{ return(m_trx); }
 
-	/** Return table name. */
-	const char* table_name() const
-	{ return(m_table_name); }
+	/** @return table name */
+	const char* table_name() const { return(m_table_name); }
+
+	/** @return the created table */
+	dict_table_t *table() const { return m_table; }
 
-	THD* thd() const
-	{ return(m_thd); }
+	THD* thd() const { return(m_thd); }
 
 private:
 	/** Parses the table name into normal name and either temp path or
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 544c1eeadc1..1f46f66a1b7 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -7286,13 +7286,10 @@ error_handling_drop_uncached:
 				goto error_handling;
 			}
 
-			ctx->new_table->fts->dict_locked = true;
-
 			error = innobase_fts_load_stopword(
 				ctx->new_table, ctx->trx,
 				ctx->prebuilt->trx->mysql_thd)
 				? DB_SUCCESS : DB_ERROR;
-			ctx->new_table->fts->dict_locked = false;
 
 			if (error != DB_SUCCESS) {
 				goto error_handling;
@@ -9882,7 +9879,7 @@ innobase_update_foreign_cache(
 
 	err = dict_load_foreigns(user_table->name.m_name,
 				 ctx->col_names, 1, true,
-				 DICT_ERR_IGNORE_NONE,
+				 DICT_ERR_IGNORE_FK_NOKEY,
 				 fk_tables);
 
 	if (err == DB_CANNOT_ADD_CONSTRAINT) {
diff --git a/storage/innobase/handler/i_s.cc b/storage/innobase/handler/i_s.cc
index 82b8968876f..bb2f1b6beda 100644
--- a/storage/innobase/handler/i_s.cc
+++ b/storage/innobase/handler/i_s.cc
@@ -6491,7 +6491,9 @@ static int i_s_sys_tablespaces_fill_table(THD *thd, TABLE_LIST *tables, Item*)
     {
       space.reacquire();
       mysql_mutex_unlock(&fil_system.mutex);
+      space.s_lock();
       err= i_s_sys_tablespaces_fill(thd, space, tables->table);
+      space.s_unlock();
       mysql_mutex_lock(&fil_system.mutex);
       space.release();
       if (err)
@@ -6719,8 +6721,10 @@ i_s_tablespaces_encryption_fill_table(
 		    && !space.is_stopping()) {
 			space.reacquire();
 			mysql_mutex_unlock(&fil_system.mutex);
+			space.s_lock();
 			err = i_s_dict_fill_tablespaces_encryption(
 				thd, &space, tables->table);
+			space.s_unlock();
 			mysql_mutex_lock(&fil_system.mutex);
 			space.release();
 			if (err) {
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index 0a6e951dbaa..37b00a8ff21 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -2464,6 +2464,7 @@ ibuf_merge_space(
 
 	ut_ad(space < SRV_SPACE_ID_UPPER_BOUND);
 
+	log_free_check();
 	ibuf_mtr_start(&mtr);
 
 	/* Position the cursor on the first matching record. */
@@ -2566,6 +2567,7 @@ ulint ibuf_merge_all()
 	ulint	n_pages = srv_io_capacity;
 
 	for (ulint sum_pages = 0; sum_pages < n_pages; ) {
+		log_free_check();
 		ulint n_pag2;
 		ulint n_bytes = ibuf_merge(&n_pag2);
 
@@ -4461,7 +4463,7 @@ reset_bit:
 }
 
 /** Delete all change buffer entries for a tablespace,
-in DISCARD TABLESPACE, IMPORT TABLESPACE, or crash recovery.
+in DISCARD TABLESPACE, IMPORT TABLESPACE, or read-ahead.
 @param[in]	space		missing or to-be-discarded tablespace */
 void ibuf_delete_for_discarded_space(uint32_t space)
 {
@@ -4483,6 +4485,7 @@ void ibuf_delete_for_discarded_space(uint32_t space)
 
 	memset(dops, 0, sizeof(dops));
 loop:
+	log_free_check();
 	ibuf_mtr_start(&mtr);
 
 	/* Position pcur in the insert buffer at the first entry for the
@@ -4622,9 +4625,6 @@ dberr_t ibuf_check_bitmap_on_import(const trx_t* trx, fil_space_t* space)
 		}
 
 		mtr_start(&mtr);
-
-		mtr_set_log_mode(&mtr, MTR_LOG_NO_REDO);
-
 		ibuf_enter(&mtr);
 
 		buf_block_t* bitmap_page = ibuf_bitmap_get_map_page(
@@ -4712,29 +4712,18 @@ dberr_t ibuf_check_bitmap_on_import(const trx_t* trx, fil_space_t* space)
 	return(DB_SUCCESS);
 }
 
-/** Updates free bits and buffered bits for bulk loaded page.
-@param[in]	block	index page
-@param[in]	reset	flag if reset free val */
-void
-ibuf_set_bitmap_for_bulk_load(
-	buf_block_t*	block,
-	bool		reset)
+void ibuf_set_bitmap_for_bulk_load(buf_block_t *block, mtr_t *mtr, bool reset)
 {
-  mtr_t	mtr;
-
   ut_a(page_is_leaf(block->page.frame));
-  mtr.start();
-  fil_space_t *space= mtr.set_named_space_id(block->page.id().space());
 
   if (buf_block_t *bitmap_page=
-      ibuf_bitmap_get_map_page(block->page.id(), space->zip_size(), &mtr))
+      ibuf_bitmap_get_map_page(block->page.id(), block->zip_size(), mtr))
   {
     ulint free_val= reset ? 0 : ibuf_index_page_calc_free(block);
     /* FIXME: update the bitmap byte only once! */
     ibuf_bitmap_page_set_bits<IBUF_BITMAP_FREE>
-      (bitmap_page, block->page.id(), block->physical_size(), free_val, &mtr);
+      (bitmap_page, block->page.id(), block->physical_size(), free_val, mtr);
     ibuf_bitmap_page_set_bits<IBUF_BITMAP_BUFFERED>
-      (bitmap_page, block->page.id(), block->physical_size(), false, &mtr);
+      (bitmap_page, block->page.id(), block->physical_size(), false, mtr);
   }
-  mtr.commit();
 }
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index a76fec31416..f70e50f5d48 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -421,14 +421,6 @@ dict_foreign_add_to_cache(
 	dict_err_ignore_t	ignore_err)
 				/*!< in: error to be ignored */
 	MY_ATTRIBUTE((nonnull(1), warn_unused_result));
-/*********************************************************************//**
-Checks if a table is referenced by foreign keys.
-@return TRUE if table is referenced by a foreign key */
-ibool
-dict_table_is_referenced_by_foreign_key(
-/*====================================*/
-	const dict_table_t*	table)	/*!< in: InnoDB table */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
 /**********************************************************************//**
 Replace the index passed in with another equivalent index in the
 foreign key lists of the table.
@@ -1326,7 +1318,7 @@ class dict_sys_t
   alignas(CPU_LEVEL1_DCACHE_LINESIZE) srw_lock latch;
 #ifdef UNIV_DEBUG
   /** whether latch is being held in exclusive mode (by any thread) */
-  bool latch_ex;
+  Atomic_relaxed<pthread_t> latch_ex;
   /** number of S-latch holders */
   Atomic_counter<uint32_t> latch_readers;
 #endif
@@ -1500,11 +1492,12 @@ public:
   /** @return whether any thread (not necessarily the current thread)
   is holding the latch; that is, this check may return false
   positives */
-  bool frozen() const { return latch_readers || locked(); }
+  bool frozen() const { return latch_readers || latch_ex; }
   /** @return whether any thread (not necessarily the current thread)
-  is holding the exclusive latch; that is, this check may return false
-  positives */
-  bool locked() const { return latch_ex; }
+  is holding a shared latch */
+  bool frozen_not_locked() const { return latch_readers; }
+  /** @return whether the current thread holds the exclusive latch */
+  bool locked() const { return latch_ex == pthread_self(); }
 #endif
 private:
   /** Acquire the exclusive latch */
@@ -1523,7 +1516,7 @@ public:
     {
       ut_ad(!latch_readers);
       ut_ad(!latch_ex);
-      ut_d(latch_ex= true);
+      ut_d(latch_ex= pthread_self());
     }
     else
       lock_wait(SRW_LOCK_ARGS(file, line));
@@ -1540,9 +1533,9 @@ public:
   /** Unlock the data dictionary cache. */
   void unlock()
   {
-    ut_ad(latch_ex);
+    ut_ad(latch_ex == pthread_self());
     ut_ad(!latch_readers);
-    ut_d(latch_ex= false);
+    ut_d(latch_ex= 0);
     latch.wr_unlock();
   }
   /** Acquire a shared lock on the dictionary cache. */
diff --git a/storage/innobase/include/dict0load.h b/storage/innobase/include/dict0load.h
index a94823b4a86..f7d33d5b43b 100644
--- a/storage/innobase/include/dict0load.h
+++ b/storage/innobase/include/dict0load.h
@@ -100,7 +100,7 @@ dict_load_foreigns(
 						which must be loaded
 						subsequently to load all the
 						foreign key constraints. */
-	MY_ATTRIBUTE((nonnull(1), warn_unused_result));
+	MY_ATTRIBUTE((nonnull(1)));
 
 /********************************************************************//**
 This function opens a system table, and return the first record.
diff --git a/storage/innobase/include/ibuf0ibuf.h b/storage/innobase/include/ibuf0ibuf.h
index 23d5258cfbc..e6f72b05087 100644
--- a/storage/innobase/include/ibuf0ibuf.h
+++ b/storage/innobase/include/ibuf0ibuf.h
@@ -264,7 +264,6 @@ ibuf_page_low(
 	MY_ATTRIBUTE((warn_unused_result));
 
 #ifdef UNIV_DEBUG
-
 /** Checks if a page is a level 2 or 3 page in the ibuf hierarchy of pages.
 Must not be called when recv_no_ibuf_operations==true.
 @param[in]	page_id		tablespace/page identifier
@@ -274,7 +273,7 @@ Must not be called when recv_no_ibuf_operations==true.
 # define ibuf_page(page_id, zip_size, mtr)	\
 	ibuf_page_low(page_id, zip_size, true, mtr)
 
-#else /* UVIV_DEBUG */
+#else /* UNIV_DEBUG */
 
 /** Checks if a page is a level 2 or 3 page in the ibuf hierarchy of pages.
 Must not be called when recv_no_ibuf_operations==true.
@@ -285,7 +284,7 @@ Must not be called when recv_no_ibuf_operations==true.
 # define ibuf_page(page_id, zip_size, mtr)	\
 	ibuf_page_low(page_id, zip_size, mtr)
 
-#endif /* UVIV_DEBUG */
+#endif /* UNIV_DEBUG */
 /***********************************************************************//**
 Frees excess pages from the ibuf free list. This function is called when an OS
 thread calls fsp services to allocate a new file segment, or a new page to a
@@ -334,7 +333,7 @@ dberr_t ibuf_merge_or_delete_for_page(buf_block_t *block,
                                       ulint zip_size);
 
 /** Delete all change buffer entries for a tablespace,
-in DISCARD TABLESPACE, IMPORT TABLESPACE, or crash recovery.
+in DISCARD TABLESPACE, IMPORT TABLESPACE, or read-ahead.
 @param[in]	space		missing or to-be-discarded tablespace */
 void ibuf_delete_for_discarded_space(uint32_t space);
 
@@ -385,13 +384,11 @@ ibuf_close(void);
 dberr_t ibuf_check_bitmap_on_import(const trx_t* trx, fil_space_t* space)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
-/** Updates free bits and buffered bits for bulk loaded page.
-@param[in]      block   index page
-@param]in]      reset   flag if reset free val */
-void
-ibuf_set_bitmap_for_bulk_load(
-	buf_block_t*    block,
-	bool		reset);
+/** Update free bits and buffered bits for bulk loaded page.
+@param block   secondary index leaf page
+@param mtr     mini-transaction
+@param reset   whether the page is full */
+void ibuf_set_bitmap_for_bulk_load(buf_block_t *block, mtr_t *mtr, bool reset);
 
 #define IBUF_HEADER_PAGE_NO	FSP_IBUF_HEADER_PAGE_NO
 #define IBUF_TREE_ROOT_PAGE_NO	FSP_IBUF_TREE_ROOT_PAGE_NO
diff --git a/storage/innobase/include/page0page.h b/storage/innobase/include/page0page.h
index b0e2eb98d01..0ad42474f84 100644
--- a/storage/innobase/include/page0page.h
+++ b/storage/innobase/include/page0page.h
@@ -795,17 +795,6 @@ page_rec_is_first(
 	MY_ATTRIBUTE((warn_unused_result));
 
 /************************************************************//**
-true if the record is the second user record on a page.
-@return true if the second user record */
-UNIV_INLINE
-bool
-page_rec_is_second(
-/*===============*/
-	const rec_t*	rec,	/*!< in: record */
-	const page_t*	page)	/*!< in: page */
-	MY_ATTRIBUTE((warn_unused_result));
-
-/************************************************************//**
 true if the record is the last user record on a page.
 @return true if the last user record */
 UNIV_INLINE
@@ -817,17 +806,6 @@ page_rec_is_last(
 	MY_ATTRIBUTE((warn_unused_result));
 
 /************************************************************//**
-true if the record is the second last user record on a page.
-@return true if the second last user record */
-UNIV_INLINE
-bool
-page_rec_is_second_last(
-/*====================*/
-	const rec_t*	rec,	/*!< in: record */
-	const page_t*	page)	/*!< in: page */
-	MY_ATTRIBUTE((warn_unused_result));
-
-/************************************************************//**
 Returns the maximum combined size of records which can be inserted on top
 of record heap.
 @return maximum combined size for inserted records */
diff --git a/storage/innobase/include/page0page.inl b/storage/innobase/include/page0page.inl
index 61c1b96ff79..6c0167edcf9 100644
--- a/storage/innobase/include/page0page.inl
+++ b/storage/innobase/include/page0page.inl
@@ -193,22 +193,6 @@ page_rec_is_first(
 }
 
 /************************************************************//**
-true if the record is the second user record on a page.
-@return true if the second user record */
-UNIV_INLINE
-bool
-page_rec_is_second(
-/*===============*/
-	const rec_t*	rec,	/*!< in: record */
-	const page_t*	page)	/*!< in: page */
-{
-  ut_ad(page_get_n_recs(page) > 1);
-  if (const rec_t *first= page_rec_get_next_const(page_get_infimum_rec(page)))
-    return page_rec_get_next_const(first) == rec;
-  return false;
-}
-
-/************************************************************//**
 true if the record is the last user record on a page.
 @return true if the last user record */
 UNIV_INLINE
@@ -224,24 +208,6 @@ page_rec_is_last(
 }
 
 /************************************************************//**
-true if the record is the second last user record on a page.
-@return true if the second last user record */
-UNIV_INLINE
-bool
-page_rec_is_second_last(
-/*====================*/
-	const rec_t*	rec,	/*!< in: record */
-	const page_t*	page)	/*!< in: page */
-{
-  ut_ad(page_get_n_recs(page) > 1);
-  ut_ad(!page_rec_is_last(rec, page));
-
-  if (const rec_t *next= page_rec_get_next_const(rec))
-    return page_rec_is_supremum(page_rec_get_next_const(next));
-  return false;
-}
-
-/************************************************************//**
 Returns the middle record of the records on the page. If there is an
 even number of records in the list, returns the first record of the
 upper half-list.
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 7933e9bc883..de3f617f76d 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -430,6 +430,10 @@ class trx_mod_table_time_t
   /** First modification of a system versioned column
   (NONE= no versioning, BULK= the table was dropped) */
   undo_no_t first_versioned= NONE;
+#ifdef UNIV_DEBUG
+  /** Whether the modified table is a FTS auxiliary table */
+  bool fts_aux_table= false;
+#endif /* UNIV_DEBUG */
 
   /** Buffer to store insert opertion */
   row_merge_bulk_t *bulk_store= nullptr;
@@ -496,6 +500,12 @@ public:
     return false;
   }
 
+#ifdef UNIV_DEBUG
+  void set_aux_table() { fts_aux_table= true; }
+
+  bool is_aux_table() const { return fts_aux_table; }
+#endif /* UNIV_DEBUG */
+
   /** @return the first undo record that modified the table */
   undo_no_t get_first() const
   {
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 448021d840a..db4035157b0 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -3934,8 +3934,7 @@ void lock_release(trx_t *trx)
 #ifdef UNIV_DEBUG
   std::set<table_id_t> to_evict;
   if (innodb_evict_tables_on_commit_debug &&
-      !trx->is_recovered && !trx->dict_operation &&
-      !trx->dict_operation_lock_mode)
+      !trx->is_recovered && !dict_sys.locked())
     for (const auto& p : trx->mod_tables)
       if (!p.first->is_temporary())
         to_evict.emplace(p.first->id);
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index da0f0668d1d..183500e19c3 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -1828,8 +1828,8 @@ ATTRIBUTE_COLD static dberr_t recv_log_recover_pre_10_2()
   log_sys.set_flushed_lsn(lsn);
   const lsn_t source_offset= log_sys.log.calc_lsn_offset_old(lsn);
 
-  static const char NO_UPGRADE_RECOVERY_MSG[]=
-    "Upgrade after a crash is not supported."
+  static constexpr char NO_UPGRADE_RECOVERY_MSG[]=
+    "InnoDB: Upgrade after a crash is not supported."
     " This redo log was created before MariaDB 10.2.2";
 
   recv_sys.read(source_offset & ~511, {buf, 512});
@@ -1837,8 +1837,7 @@ ATTRIBUTE_COLD static dberr_t recv_log_recover_pre_10_2()
   if (log_block_calc_checksum_format_0(buf) != log_block_get_checksum(buf) &&
       !log_crypt_101_read_block(buf, lsn))
   {
-    sql_print_error("InnoDB: %s, and it appears corrupted.",
-                    NO_UPGRADE_RECOVERY_MSG);
+    sql_print_error("%s, and it appears corrupted.", NO_UPGRADE_RECOVERY_MSG);
     return DB_CORRUPTION;
   }
 
@@ -1858,7 +1857,10 @@ ATTRIBUTE_COLD static dberr_t recv_log_recover_pre_10_2()
     sql_print_error("InnoDB: Cannot decrypt log for upgrading."
                     " The encrypted log was created before MariaDB 10.2.2.");
   else
-    sql_print_error("InnoDB: %s.", NO_UPGRADE_RECOVERY_MSG);
+    sql_print_error("%s. You must start up and shut down"
+                    " MariaDB 10.1 or MySQL 5.6 or earlier"
+                    " on the data directory.",
+                    NO_UPGRADE_RECOVERY_MSG);
 
   return DB_ERROR;
 }
@@ -1964,7 +1966,7 @@ recv_find_max_checkpoint(ulint* max_field)
 	if (log_sys.log.format != log_t::FORMAT_3_23
 	    && !recv_check_log_header_checksum(buf)) {
 		sql_print_error("InnoDB: Invalid redo log header checksum.");
-		return(DB_CORRUPTION);
+		return DB_CORRUPTION;
 	}
 
 	char creator[LOG_HEADER_CREATOR_END - LOG_HEADER_CREATOR + 1];
@@ -1988,7 +1990,7 @@ recv_find_max_checkpoint(ulint* max_field)
 	default:
 		sql_print_error("InnoDB: Unsupported redo log format."
 				" The redo log was created with %s.", creator);
-		return(DB_ERROR);
+		return DB_ERROR;
 	}
 
 	for (field = LOG_CHECKPOINT_1; field <= LOG_CHECKPOINT_2;
@@ -2044,7 +2046,7 @@ recv_find_max_checkpoint(ulint* max_field)
 				" (corrupted redo log)."
 				" You can try --innodb-force-recovery=6"
 				" as a last resort.");
-		return(DB_ERROR);
+		return DB_ERROR;
 	}
 
 	switch (log_sys.log.format) {
@@ -2053,12 +2055,15 @@ recv_find_max_checkpoint(ulint* max_field)
 		break;
 	default:
 		if (dberr_t err = recv_log_recover_10_4()) {
-			sql_print_error("InnoDB: Upgrade after a crash"
-					" is not supported."
+			sql_print_error("InnoDB: Upgrade after a crash "
+					"is not supported."
 					" The redo log was created with %s%s.",
 					creator,
-					(err == DB_ERROR
-					 ? "" : ", and it appears corrupted"));
+					err == DB_ERROR
+					? ". You must start up and shut down"
+					" MariaDB 10.4 or earlier"
+					" on the data directory"
+					: ", and it appears corrupted");
 			return err;
 		}
 	}
diff --git a/storage/innobase/plugin_exports b/storage/innobase/plugin_exports
deleted file mode 100644
index 235ae3d5e72..00000000000
--- a/storage/innobase/plugin_exports
+++ /dev/null
@@ -1,14 +0,0 @@
-{
- global:
-   _maria_plugin_interface_version_;
-   _maria_sizeof_struct_st_plugin_;
-   _maria_plugin_declarations_;
-   my_snprintf_service;
-   thd_alloc_service;
-   thd_autoinc_service;
-   thd_error_context_service;
-   thd_kill_statement_service;
-   thd_wait_service;
- local:
-   *;
-};
diff --git a/storage/innobase/que/que0que.cc b/storage/innobase/que/que0que.cc
index 80c34af2790..c5fe7c04a17 100644
--- a/storage/innobase/que/que0que.cc
+++ b/storage/innobase/que/que0que.cc
@@ -260,17 +260,10 @@ que_graph_free_recursive(
 	ind_node_t*	cre_ind;
 	purge_node_t*	purge;
 
-	DBUG_ENTER("que_graph_free_recursive");
-
 	if (node == NULL) {
-
-		DBUG_VOID_RETURN;
+		return;
 	}
 
-	DBUG_PRINT("que_graph_free_recursive",
-		   ("node: %p, type: " ULINTPF, node,
-		    que_node_get_type(node)));
-
 	switch (que_node_get_type(node)) {
 
 	case QUE_NODE_FORK:
@@ -410,8 +403,6 @@ que_graph_free_recursive(
 	default:
 		ut_error;
 	}
-
-	DBUG_VOID_RETURN;
 }
 
 /**********************************************************************//**
@@ -507,66 +498,6 @@ que_node_get_containing_loop_node(
 	return(node);
 }
 
-#ifdef DBUG_TRACE
-/** Gets information of an SQL query graph node.
-@return type description */
-static MY_ATTRIBUTE((warn_unused_result, nonnull))
-const char*
-que_node_type_string(
-/*=================*/
-	const que_node_t*	node)	/*!< in: query graph node */
-{
-	switch (que_node_get_type(node)) {
-	case QUE_NODE_SELECT:
-		return("SELECT");
-	case QUE_NODE_INSERT:
-		return("INSERT");
-	case QUE_NODE_UPDATE:
-		return("UPDATE");
-	case QUE_NODE_WHILE:
-		return("WHILE");
-	case QUE_NODE_ASSIGNMENT:
-		return("ASSIGNMENT");
-	case QUE_NODE_IF:
-		return("IF");
-	case QUE_NODE_FETCH:
-		return("FETCH");
-	case QUE_NODE_OPEN:
-		return("OPEN");
-	case QUE_NODE_PROC:
-		return("STORED PROCEDURE");
-	case QUE_NODE_FUNC:
-		return("FUNCTION");
-	case QUE_NODE_LOCK:
-		return("LOCK");
-	case QUE_NODE_THR:
-		return("QUERY THREAD");
-	case QUE_NODE_COMMIT:
-		return("COMMIT");
-	case QUE_NODE_UNDO:
-		return("UNDO ROW");
-	case QUE_NODE_PURGE:
-		return("PURGE ROW");
-	case QUE_NODE_ROLLBACK:
-		return("ROLLBACK");
-	case QUE_NODE_CREATE_TABLE:
-		return("CREATE TABLE");
-	case QUE_NODE_CREATE_INDEX:
-		return("CREATE INDEX");
-	case QUE_NODE_FOR:
-		return("FOR LOOP");
-	case QUE_NODE_RETURN:
-		return("RETURN");
-	case QUE_NODE_EXIT:
-		return("EXIT");
-	default:
-		ut_ad(0);
-		return("UNKNOWN NODE TYPE");
-	}
-}
-#endif /* DBUG_TRACE */
-
-
 /**********************************************************************//**
 Performs an execution step of an open or close cursor statement node.
 @param thr query thread */
@@ -614,10 +545,6 @@ que_thr_step(
 
 	old_thr = thr;
 
-	DBUG_PRINT("ib_que", ("Execute %u (%s) at %p",
-			      unsigned(type), que_node_type_string(node),
-			      (const void*) node));
-
 	if (type & QUE_NODE_CONTROL_STAT) {
 		if ((thr->prev_node != que_node_get_parent(node))
 		    && que_node_get_next(thr->prev_node)) {
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 81882f84c8a..c8e94991500 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -61,57 +61,49 @@ Completed by Sunny Bains and Marko Makela
 /* Whether to disable file system cache */
 char	srv_disable_sort_file_cache;
 
-/** Class that caches index row tuples made from a single cluster
+/** Class that caches spatial index row tuples made from a single cluster
 index page scan, and then insert into corresponding index tree */
-class index_tuple_info_t {
+class spatial_index_info {
 public:
-	/** constructor
-	@param[in]	heap	memory heap
-	@param[in]	index	index to be created */
-	index_tuple_info_t(mem_heap_t* heap, dict_index_t* index) :
-		m_dtuple_vec(UT_NEW_NOKEY(idx_tuple_vec())),
-		m_index(index), m_heap(heap)
-	{ ut_ad(index->is_spatial()); }
-
-	/** destructor */
-	~index_tuple_info_t()
-	{
-		UT_DELETE(m_dtuple_vec);
-	}
-
-	/** Get the index object
-	@return the index object */
-	dict_index_t*   get_index() UNIV_NOTHROW
-	{
-		return(m_index);
-	}
-
-	/** Caches an index row into index tuple vector
-	@param[in]	row	table row
-	@param[in]	ext	externally stored column
-	prefixes, or NULL */
-	void add(
-		const dtuple_t*		row,
-		const row_ext_t*	ext) UNIV_NOTHROW
-	{
-		dtuple_t*	dtuple;
-
-		dtuple = row_build_index_entry(row, ext, m_index, m_heap);
-
-		ut_ad(dtuple);
+  /** constructor
+  @param index	spatial index to be created */
+  spatial_index_info(dict_index_t *index) : index(index)
+  {
+    ut_ad(index->is_spatial());
+  }
 
-		m_dtuple_vec->push_back(dtuple);
-	}
+  /** Caches an index row into index tuple vector
+  @param[in]	row	table row
+  @param[in]	ext	externally stored column prefixes, or NULL */
+  void add(const dtuple_t *row, const row_ext_t *ext, mem_heap_t *heap)
+  {
+    dtuple_t *dtuple= row_build_index_entry(row, ext, index, heap);
+    ut_ad(dtuple);
+    ut_ad(dtuple->n_fields == index->n_fields);
+    if (ext)
+    {
+      /* Replace any references to ext, because ext will be allocated
+      from row_heap. */
+      for (ulint i= 1; i < dtuple->n_fields; i++)
+      {
+        dfield_t &dfield= dtuple->fields[i];
+        if (dfield.data >= ext->buf &&
+            dfield.data <= &ext->buf[ext->n_ext * ext->max_len])
+          dfield_dup(&dfield, heap);
+      }
+    }
+    m_dtuple_vec.push_back(dtuple);
+  }
 
 	/** Insert spatial index rows cached in vector into spatial index
 	@param[in]	trx_id		transaction id
-	@param[in,out]	row_heap	memory heap
 	@param[in]	pcur		cluster index scanning cursor
 	@param[in,out]	mtr_started	whether scan_mtr is active
+	@param[in,out]	heap		temporary memory heap
 	@param[in,out]	scan_mtr	mini-transaction for pcur
 	@return DB_SUCCESS if successful, else error number */
-	dberr_t insert(trx_id_t trx_id, mem_heap_t* row_heap, btr_pcur_t* pcur,
-		       bool& mtr_started, mtr_t* scan_mtr) const
+	dberr_t insert(trx_id_t trx_id, btr_pcur_t* pcur,
+		       bool& mtr_started, mem_heap_t* heap, mtr_t* scan_mtr)
 	{
 		big_rec_t*      big_rec;
 		rec_t*          rec;
@@ -130,8 +122,8 @@ public:
 		DBUG_EXECUTE_IF("row_merge_instrument_log_check_flush",
 				log_sys.set_check_flush_or_checkpoint(););
 
-		for (idx_tuple_vec::iterator it = m_dtuple_vec->begin();
-		     it != m_dtuple_vec->end();
+		for (idx_tuple_vec::iterator it = m_dtuple_vec.begin();
+		     it != m_dtuple_vec.end();
 		     ++it) {
 			dtuple = *it;
 			ut_ad(dtuple);
@@ -151,29 +143,29 @@ public:
 			}
 
 			mtr.start();
-			m_index->set_modified(mtr);
+			index->set_modified(mtr);
 
-			ins_cur.index = m_index;
-			rtr_init_rtr_info(&rtr_info, false, &ins_cur, m_index,
+			ins_cur.index = index;
+			rtr_init_rtr_info(&rtr_info, false, &ins_cur, index,
 					  false);
 			rtr_info_update_btr(&ins_cur, &rtr_info);
 
 			error = btr_cur_search_to_nth_level(
-				m_index, 0, dtuple, PAGE_CUR_RTREE_INSERT,
+				index, 0, dtuple, PAGE_CUR_RTREE_INSERT,
 				BTR_MODIFY_LEAF, &ins_cur, &mtr);
 
 			/* It need to update MBR in parent entry,
 			so change search mode to BTR_MODIFY_TREE */
 			if (error == DB_SUCCESS && rtr_info.mbr_adj) {
-				mtr_commit(&mtr);
+				mtr.commit();
 				rtr_clean_rtr_info(&rtr_info, true);
 				rtr_init_rtr_info(&rtr_info, false, &ins_cur,
-						  m_index, false);
+						  index, false);
 				rtr_info_update_btr(&ins_cur, &rtr_info);
-				mtr_start(&mtr);
-				m_index->set_modified(mtr);
+				mtr.start();
+				index->set_modified(mtr);
 				error = btr_cur_search_to_nth_level(
-					m_index, 0, dtuple,
+					index, 0, dtuple,
 					PAGE_CUR_RTREE_INSERT,
 					BTR_MODIFY_TREE, &ins_cur, &mtr);
 			}
@@ -181,7 +173,7 @@ public:
 			if (error == DB_SUCCESS) {
 				error = btr_cur_optimistic_insert(
 					flag, &ins_cur, &ins_offsets,
-					&row_heap, dtuple, &rec, &big_rec,
+					&heap, dtuple, &rec, &big_rec,
 					0, NULL, &mtr);
 			}
 
@@ -190,15 +182,15 @@ public:
 			if (error == DB_FAIL) {
 				mtr.commit();
 				mtr.start();
-				m_index->set_modified(mtr);
+				index->set_modified(mtr);
 
 				rtr_clean_rtr_info(&rtr_info, true);
 				rtr_init_rtr_info(&rtr_info, false,
-						  &ins_cur, m_index, false);
+						  &ins_cur, index, false);
 
 				rtr_info_update_btr(&ins_cur, &rtr_info);
 				error = btr_cur_search_to_nth_level(
-					m_index, 0, dtuple,
+					index, 0, dtuple,
 					PAGE_CUR_RTREE_INSERT,
 					BTR_MODIFY_TREE,
 					&ins_cur, &mtr);
@@ -206,7 +198,7 @@ public:
 				if (error == DB_SUCCESS) {
 					error = btr_cur_pessimistic_insert(
 						flag, &ins_cur, &ins_offsets,
-						&row_heap, dtuple, &rec,
+						&heap, dtuple, &rec,
 						&big_rec, 0, NULL, &mtr);
 				}
 			}
@@ -232,30 +224,26 @@ public:
 				}
 			}
 
-			mtr_commit(&mtr);
+			mtr.commit();
 
 			rtr_clean_rtr_info(&rtr_info, true);
 		}
 
-		m_dtuple_vec->clear();
+		m_dtuple_vec.clear();
 
 		return(error);
 	}
 
 private:
-	/** Cache index rows made from a cluster index scan. Usually
-	for rows on single cluster index page */
-	typedef std::vector<dtuple_t*, ut_allocator<dtuple_t*> >
-		idx_tuple_vec;
+  /** Cache index rows made from a cluster index scan. Usually
+  for rows on single cluster index page */
+  typedef std::vector<dtuple_t*, ut_allocator<dtuple_t*> > idx_tuple_vec;
 
-	/** vector used to cache index rows made from cluster index scan */
-	idx_tuple_vec* const	m_dtuple_vec;
-
-	/** the index being built */
-	dict_index_t* const	m_index;
-
-	/** memory heap for creating index tuples */
-	mem_heap_t* const	m_heap;
+  /** vector used to cache index rows made from cluster index scan */
+  idx_tuple_vec m_dtuple_vec;
+public:
+  /** the index being built */
+  dict_index_t*const	index;
 };
 
 /* Maximum pending doc memory limit in bytes for a fts tokenization thread */
@@ -1749,8 +1737,7 @@ row_mtuple_cmp(
 @param[in]	trx_id		transaction id
 @param[in]	sp_tuples	cached spatial rows
 @param[in]	num_spatial	number of spatial indexes
-@param[in,out]	heap		heap for insert
-@param[in,out]	sp_heap		heap for tuples
+@param[in,out]	heap		temporary memory heap
 @param[in,out]	pcur		cluster index cursor
 @param[in,out]	started		whether mtr is active
 @param[in,out]	mtr		mini-transaction
@@ -1759,10 +1746,9 @@ static
 dberr_t
 row_merge_spatial_rows(
 	trx_id_t		trx_id,
-	index_tuple_info_t**	sp_tuples,
+	spatial_index_info**	sp_tuples,
 	ulint			num_spatial,
 	mem_heap_t*		heap,
-	mem_heap_t*		sp_heap,
 	btr_pcur_t*		pcur,
 	bool&			started,
 	mtr_t*			mtr)
@@ -1771,10 +1757,10 @@ row_merge_spatial_rows(
     return DB_SUCCESS;
 
   for (ulint j= 0; j < num_spatial; j++)
-    if (dberr_t err= sp_tuples[j]->insert(trx_id, heap, pcur, started, mtr))
+    if (dberr_t err= sp_tuples[j]->insert(trx_id, pcur, started, heap, mtr))
       return err;
 
-  mem_heap_empty(sp_heap);
+  mem_heap_empty(heap);
   return DB_SUCCESS;
 }
 
@@ -1890,8 +1876,7 @@ row_merge_read_clustered_index(
 	doc_id_t		max_doc_id = 0;
 	ibool			add_doc_id = FALSE;
 	pthread_cond_t*		fts_parallel_sort_cond = nullptr;
-	index_tuple_info_t**	sp_tuples = NULL;
-	mem_heap_t*		sp_heap = NULL;
+	spatial_index_info**	sp_tuples = nullptr;
 	ulint			num_spatial = 0;
 	BtrBulk*		clust_btr_bulk = NULL;
 	bool			clust_temp_file = false;
@@ -1980,9 +1965,7 @@ row_merge_read_clustered_index(
 	if (num_spatial > 0) {
 		ulint	count = 0;
 
-		sp_heap = mem_heap_create(512);
-
-		sp_tuples = static_cast<index_tuple_info_t**>(
+		sp_tuples = static_cast<spatial_index_info**>(
 			ut_malloc_nokey(num_spatial
 					* sizeof(*sp_tuples)));
 
@@ -1990,9 +1973,7 @@ row_merge_read_clustered_index(
 			if (dict_index_is_spatial(index[i])) {
 				sp_tuples[count]
 					= UT_NEW_NOKEY(
-						index_tuple_info_t(
-							sp_heap,
-							index[i]));
+						spatial_index_info(index[i]));
 				count++;
 			}
 		}
@@ -2167,7 +2148,7 @@ corrupted_rec:
 			/* Insert the cached spatial index rows. */
 			err = row_merge_spatial_rows(
 				trx->id, sp_tuples, num_spatial,
-				row_heap, sp_heap, &pcur, mtr_started, &mtr);
+				row_heap, &pcur, mtr_started, &mtr);
 
 			if (err != DB_SUCCESS) {
 				goto func_exit;
@@ -2561,7 +2542,7 @@ write_buffers:
 					continue;
 				}
 
-				ut_ad(sp_tuples[s_idx_cnt]->get_index()
+				ut_ad(sp_tuples[s_idx_cnt]->index
 				      == buf->index);
 
 				/* If the geometry field is invalid, report
@@ -2571,7 +2552,7 @@ write_buffers:
 					break;
 				}
 
-				sp_tuples[s_idx_cnt]->add(row, ext);
+				sp_tuples[s_idx_cnt]->add(row, ext, buf->heap);
 				s_idx_cnt++;
 
 				continue;
@@ -2693,7 +2674,7 @@ write_buffers:
 						err = row_merge_spatial_rows(
 							trx->id, sp_tuples,
 							num_spatial,
-							row_heap, sp_heap,
+							row_heap,
 							&pcur, mtr_started,
 							&mtr);
 
@@ -3059,10 +3040,6 @@ wait_again:
 			UT_DELETE(sp_tuples[i]);
 		}
 		ut_free(sp_tuples);
-
-		if (sp_heap) {
-			mem_heap_free(sp_heap);
-		}
 	}
 
 	/* Update the next Doc ID we used. Table should be locked, so