Merging upstream version 1:11.4.2.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

12 files changed, 404 insertions, 493 deletions

diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 6194e9c3..5febd6df 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -293,6 +293,7 @@ struct fil_iterator_t {
 	byte*		io_buffer;		/*!< Buffer to use for IO */
 	fil_space_crypt_t *crypt_data;		/*!< Crypt data (if encrypted) */
 	byte*           crypt_io_buffer;        /*!< IO buffer when encrypted */
+	byte*           crypt_tmp_buffer;       /*!< Temporary buffer for crypt use */
 };
 
 /** Use the page cursor to iterate over records in a block. */
@@ -739,7 +740,8 @@ struct FetchIndexRootPages : public AbstractCallback {
 	/** Constructor
 	@param trx covering (user) transaction
 	@param table table definition in server .*/
-	FetchIndexRootPages(const dict_table_t* table, trx_t* trx)
+	FetchIndexRootPages(const dict_table_t* table = nullptr,
+			    trx_t* trx = nullptr)
 		:
 		AbstractCallback(trx, UINT32_MAX),
 		m_table(table), m_index(0, 0) UNIV_NOTHROW { }
@@ -754,18 +756,46 @@ struct FetchIndexRootPages : public AbstractCallback {
 	dberr_t run(const fil_iterator_t& iter,
 		    buf_block_t* block) UNIV_NOTHROW override;
 
-	/** Called for each block as it is read from the file.
+	/** Check that fsp flags and row formats match.
 	@param block block to convert, it is not from the buffer pool.
 	@retval DB_SUCCESS or error code. */
 	dberr_t operator()(buf_block_t* block) UNIV_NOTHROW override;
 
+	/** Get row format from the header and the root index page. */
+	enum row_type get_row_format(const buf_block_t &block)
+	{
+		if (!page_is_comp(block.page.frame))
+			return ROW_TYPE_REDUNDANT;
+		/* With full_crc32 we cannot tell between dynamic or
+		compact, and return not_used. We cannot simply return
+		dynamic or compact, as the client of this function
+		will not be able to tell whether it is dynamic because
+		of this or the other branch below. Returning default
+		would also work if it is immediately handled, but is
+		still more ambiguous than not_used, which is not a
+		row_format at all. */
+		if (fil_space_t::full_crc32(m_space_flags))
+			return ROW_TYPE_NOT_USED;
+		if (!(m_space_flags & FSP_FLAGS_MASK_ATOMIC_BLOBS))
+			return ROW_TYPE_COMPACT;
+		if (FSP_FLAGS_GET_ZIP_SSIZE(m_space_flags))
+			return ROW_TYPE_COMPRESSED;
+		return ROW_TYPE_DYNAMIC;
+	}
+
 	/** Update the import configuration that will be used to import
 	the tablespace. */
 	dberr_t build_row_import(row_import* cfg) const UNIV_NOTHROW;
 
-	/** Table definition in server. */
+	/** Table definition in server. When the table is being
+	created, there's no table yet so m_table is nullptr */
 	const dict_table_t*	m_table;
 
+	/** Table row format. Only used when a (stub) table is being
+	created in which case m_table is null, for obtaining row
+	format from the .ibd for the stub table. */
+	enum row_type m_row_format;
+
 	/** Index information */
 	Index			m_index;
 };
@@ -2151,9 +2181,9 @@ dberr_t PageConverter::operator()(buf_block_t* block) UNIV_NOTHROW
 	/* If we already had an old page with matching number
 	in the buffer pool, evict it now, because
 	we no longer evict the pages on DISCARD TABLESPACE. */
-	buf_page_get_low(block->page.id(), get_zip_size(), RW_NO_LATCH,
+	buf_page_get_gen(block->page.id(), get_zip_size(), RW_NO_LATCH,
 			 nullptr, BUF_PEEK_IF_IN_POOL,
-			 nullptr, nullptr, false);
+			 nullptr, nullptr);
 
 	uint16_t page_type;
 
@@ -2207,8 +2237,9 @@ row_import_cleanup(row_prebuilt_t* prebuilt,
                    dberr_t         err,
                    dict_table_t*   fts_table = nullptr)
 {
+	dict_table_t* table = prebuilt->table;
+
 	if (err != DB_SUCCESS) {
-		dict_table_t* table = prebuilt->table;
 		table->file_unreadable = true;
 		if (table->space) {
 			fil_close_tablespace(table->space_id);
@@ -2237,6 +2268,7 @@ row_import_cleanup(row_prebuilt_t* prebuilt,
 
 		if (err == DB_SUCCESS) {
 			reload_fts_table(prebuilt, fts_table);
+			table= prebuilt->table;
 			ib::warn() << "Added system generated FTS_DOC_ID "
 				   "and FTS_DOC_ID_INDEX while importing "
 				   "the tablespace " << prebuilt->table->name;
@@ -2272,7 +2304,25 @@ row_import_cleanup(row_prebuilt_t* prebuilt,
 
 	DBUG_EXECUTE_IF("ib_import_before_checkpoint_crash", DBUG_SUICIDE(););
 
-	return(err);
+	if (err != DB_SUCCESS
+	    || !dict_table_get_first_index(table)->is_gen_clust()) {
+		return err;
+	}
+
+	btr_cur_t cur;
+	mtr_t mtr;
+	mtr.start();
+	err = cur.open_leaf(false, dict_table_get_first_index(table),
+			    BTR_SEARCH_LEAF, &mtr);
+	if (err != DB_SUCCESS) {
+	} else if (const rec_t *rec =
+		   page_rec_get_prev(btr_cur_get_rec(&cur))) {
+		if (page_rec_is_user_rec(rec))
+			table->row_id= mach_read_from_6(rec);
+	}
+	mtr.commit();
+
+	return err;
 }
 
 /** Report error during tablespace import.
@@ -2411,55 +2461,6 @@ row_import_adjust_root_pages_of_secondary_indexes(
 }
 
 /*****************************************************************//**
-Ensure that dict_sys.row_id exceeds SELECT MAX(DB_ROW_ID). */
-MY_ATTRIBUTE((nonnull)) static
-void
-row_import_set_sys_max_row_id(
-/*==========================*/
-	row_prebuilt_t*		prebuilt,	/*!< in/out: prebuilt from
-						handler */
-	const dict_table_t*	table)		/*!< in: table to import */
-{
-	const rec_t*		rec;
-	mtr_t			mtr;
-	btr_pcur_t		pcur;
-	row_id_t		row_id	= 0;
-	dict_index_t*		index;
-
-	index = dict_table_get_first_index(table);
-	ut_ad(index->is_primary());
-	ut_ad(dict_index_is_auto_gen_clust(index));
-
-	mtr_start(&mtr);
-
-	mtr_set_log_mode(&mtr, MTR_LOG_NO_REDO);
-
-	if (pcur.open_leaf(false, index, BTR_SEARCH_LEAF, &mtr)
-	    == DB_SUCCESS) {
-		rec = btr_pcur_move_to_prev_on_page(&pcur);
-
-		if (!rec) {
-			/* The table is corrupted. */
-		} else if (page_rec_is_infimum(rec)) {
-			/* The table is empty. */
-		} else if (rec_is_metadata(rec, *index)) {
-			/* The clustered index contains the metadata
-			record only, that is, the table is empty. */
-		} else {
-			row_id = mach_read_from_6(rec);
-		}
-	}
-
-	mtr_commit(&mtr);
-
-	if (row_id) {
-		/* Update the system row id if the imported index row id is
-		greater than the max system row id. */
-		dict_sys.update_row_id(row_id);
-	}
-}
-
-/*****************************************************************//**
 Read the a string from the meta data file.
 @return DB_SUCCESS or error code. */
 static
@@ -3142,17 +3143,25 @@ row_import_read_meta_data(
 /* decrypt and decompress page if needed */
 static dberr_t decrypt_decompress(fil_space_crypt_t *space_crypt,
                                   uint32_t space_flags, span<byte> page,
-                                  uint32_t space_id, byte *page_compress_buf)
+                                  uint32_t space_id, byte *page_compress_buf,
+                                  byte *tmp_frame)
 {
   auto *data= page.data();
 
   if (space_crypt && space_crypt->should_encrypt())
   {
+    uint page_size= static_cast<uint>(page.size());
+
     if (!buf_page_verify_crypt_checksum(data, space_flags))
       return DB_CORRUPTION;
 
-    if (dberr_t err= fil_space_decrypt(space_id, space_flags, space_crypt,
-                                       data, page.size(), data))
+    dberr_t err=
+      fil_space_decrypt(space_id, space_flags, space_crypt,
+                        tmp_frame, page_size, data);
+
+    memcpy(data, tmp_frame, page_size);
+
+    if (err)
       return err;
   }
 
@@ -3404,11 +3413,16 @@ static dberr_t handle_instant_metadata(dict_table_t *table,
     return err;
 
   std::unique_ptr<byte[]> page_compress_buf(new byte[get_buf_size()]);
+  std::unique_ptr<byte[], decltype(&aligned_free)> crypt_tmp_frame(
+      static_cast<byte *>(
+          aligned_malloc(physical_size, CPU_LEVEL1_DCACHE_LINESIZE)),
+      &aligned_free);
 
   if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
                                       {page.get(), static_cast<size_t>
                                        (physical_size)},
-                                      space_id, page_compress_buf.get()))
+                                      space_id, page_compress_buf.get(),
+                                      crypt_tmp_frame.get()))
     return err;
 
   if (table->supports_instant())
@@ -3462,7 +3476,8 @@ static dberr_t handle_instant_metadata(dict_table_t *table,
       if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
                                           {page.get(), static_cast<size_t>
                                            (physical_size)}, space_id,
-                                          page_compress_buf.get()))
+                                          page_compress_buf.get(),
+                                          crypt_tmp_frame.get()))
         return err;
     }
 
@@ -3544,7 +3559,8 @@ static dberr_t handle_instant_metadata(dict_table_t *table,
       if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
                                           {second_page.get(),
                                            static_cast<size_t>(physical_size)},
-                                          space_id, page_compress_buf.get()))
+                                          space_id, page_compress_buf.get(),
+                                          crypt_tmp_frame.get()))
         return err;
 
       if (fil_page_get_type(second_page.get()) != FIL_PAGE_TYPE_BLOB ||
@@ -3627,6 +3643,35 @@ static dberr_t handle_instant_metadata(dict_table_t *table,
 }
 
 /**
+Read the contents of a .cfg file.
+@param[in]  filename  Path to the cfg file
+@param[in]  thd       Connection
+@param[out] cfg   Contents of the .cfg file.
+@return DB_SUCCESS or error code. */
+static dberr_t row_import_read_cfg_internal(const char *filename, THD *thd,
+                                            row_import &cfg)
+{
+  FILE *file= fopen(filename, "rb");
+
+  cfg.m_missing= !file;
+
+  if (!file)
+  {
+    char msg[BUFSIZ];
+    snprintf(msg, sizeof(msg),
+             "Error opening '%s', will attempt to import"
+             " without schema verification", filename);
+    ib_senderrf(thd, IB_LOG_LEVEL_WARN, ER_IO_READ_ERROR,
+                (ulong) errno, strerror(errno), msg);
+    return DB_FAIL;
+  }
+
+  dberr_t err= row_import_read_meta_data(file, thd, cfg);
+  fclose(file);
+  return err;
+}
+
+/**
 Read the contents of the <tablename>.cfg file.
 @return DB_SUCCESS or error code. */
 static	MY_ATTRIBUTE((nonnull, warn_unused_result))
@@ -3637,38 +3682,60 @@ row_import_read_cfg(
 	THD*		thd,	/*!< in: session */
 	row_import&	cfg)	/*!< out: contents of the .cfg file */
 {
-	dberr_t		err;
 	char		name[OS_FILE_MAX_PATH];
 
 	cfg.m_table = table;
 
 	srv_get_meta_data_filename(table, name, sizeof(name));
 
-	FILE*	file = fopen(name, "rb");
-
-	if (file == NULL) {
-		char	msg[BUFSIZ];
-
-		snprintf(msg, sizeof(msg),
-			 "Error opening '%s', will attempt to import"
-			 " without schema verification", name);
-
-		ib_senderrf(
-			thd, IB_LOG_LEVEL_WARN, ER_IO_READ_ERROR,
-			(ulong) errno, strerror(errno), msg);
-
-		cfg.m_missing = true;
+	return row_import_read_cfg_internal(name, thd, cfg);
+}
 
-		err = DB_FAIL;
-	} else {
 
-		cfg.m_missing = false;
+/** Convert the InnoDB ROW_FORMAT from rec_format_enum to row_type.
+@param[in]  from  ROW_FORMAT as a rec_format_enum
+@return the row_type representation of ROW_FORMAT. */
+static enum row_type from_rec_format(const rec_format_enum from)
+{
+  switch (from) {
+  case REC_FORMAT_COMPACT:
+    return ROW_TYPE_COMPACT;
+  case REC_FORMAT_DYNAMIC:
+    return ROW_TYPE_DYNAMIC;
+  case REC_FORMAT_REDUNDANT:
+    return ROW_TYPE_REDUNDANT;
+  case REC_FORMAT_COMPRESSED:
+    return ROW_TYPE_COMPRESSED;
+  }
 
-		err = row_import_read_meta_data(file, thd, cfg);
-		fclose(file);
-	}
+  ut_ad("invalid format" == 0);
+  return ROW_TYPE_NOT_USED;
+}
 
-	return(err);
+/**
+Read the row type from a .cfg file.
+@param  dir_path  Path to the data directory containing the .cfg file
+@param  name      Name of the table
+@param  thd       Connection
+@retval ROW_TYPE_COMPACT    for ROW_FORMAT=COMPACT
+@retval ROW_TYPE_DYNAMIC    for ROW_FORMAT=DYNAMIC
+@retval ROW_TYPE_REDUNDANT  for ROW_FORMAT=REDUNDANT
+@retval ROW_TYPE_COMPRESSED for ROW_FORMAT=COMPRESSED
+@retval ROW_TYPE_NOT_USED to signal error */
+static enum row_type get_row_type_from_cfg(const char* dir_path,
+                                           const char* name, THD* thd)
+{
+  char* filename= fil_make_filepath(dir_path,
+                                    table_name_t(const_cast<char*>(name)),
+                                    CFG, dir_path != nullptr);
+  if (!filename)
+    return ROW_TYPE_NOT_USED;
+  row_import cfg;
+  dberr_t err= row_import_read_cfg_internal(filename, thd, cfg);
+  ut_free(filename);
+  if (err == DB_SUCCESS)
+    return from_rec_format(dict_tf_get_rec_format(cfg.m_flags));
+  return ROW_TYPE_NOT_USED;
 }
 
 /** Update the root page numbers and tablespace ID of a table.
@@ -3801,11 +3868,20 @@ row_import_set_discarded(
 	ulint	flags2 = mach_read_from_4(
 		static_cast<byte*>(dfield_get_data(dfield)));
 
+#if defined __GNUC__ && !defined __clang__
+# pragma GCC diagnostic push
+# if __GNUC__ < 12 || defined WITH_UBSAN
+#  pragma GCC diagnostic ignored "-Wconversion"
+# endif
+#endif
 	if (discard->state) {
 		flags2 |= DICT_TF2_DISCARDED;
 	} else {
 		flags2 &= ~DICT_TF2_DISCARDED;
 	}
+#if defined __GNUC__ && !defined __clang__
+# pragma GCC diagnostic pop
+#endif
 
 	mach_write_to_4(reinterpret_cast<byte*>(&discard->flags2), flags2);
 
@@ -3986,8 +4062,14 @@ page_corrupted:
     if (!buf_page_verify_crypt_checksum(readptr, m_space_flags))
       goto page_corrupted;
 
-    if ((err= fil_space_decrypt(get_space_id(), m_space_flags, iter.crypt_data,
-                                readptr, size, readptr)))
+    dberr_t err= fil_space_decrypt(get_space_id(), m_space_flags,
+                                   iter.crypt_data, iter.crypt_tmp_buffer,
+                                   size, readptr);
+
+    memcpy_aligned<CPU_LEVEL1_DCACHE_LINESIZE>(readptr, iter.crypt_tmp_buffer,
+                                               size);
+
+    if (err)
       goto func_exit;
   }
 
@@ -4008,7 +4090,12 @@ page_corrupted:
            && buf_page_is_corrupted(false, readptr, m_space_flags))
     goto page_corrupted;
 
-  err= this->operator()(block);
+  /* m_table is null iff we are trying to create a (stub) table, in
+  which case we want to get row format for the table creation. */
+  if (m_table)
+    err= this->operator()(block);
+  else
+    m_row_format= get_row_format(*block);
 func_exit:
   free(page_compress_buf);
   return err;
@@ -4331,19 +4418,21 @@ func_exit:
 	return err;
 }
 
-/********************************************************************//**
-Iterate over all the pages in the tablespace.
-@param table - the table definiton in the server
-@param n_io_buffers - number of blocks to read and write together
-@param callback - functor that will do the page updates
+/**
+Iterate over all or some pages in the tablespace.
+@param dir_path      the path to data dir storing the tablespace
+@param name          the table name
+@param n_io_buffers  number of blocks to read and write together
+@param callback      functor that will do the page queries or updates
 @return	DB_SUCCESS or error code */
 static
 dberr_t
 fil_tablespace_iterate(
 /*===================*/
-	dict_table_t*		table,
-	ulint			n_io_buffers,
-	AbstractCallback&	callback)
+	const char *name,
+	ulint n_io_buffers,
+	AbstractCallback &callback,
+	const char *dir_path)
 {
 	dberr_t		err;
 	pfs_os_file_t	file;
@@ -4355,18 +4444,9 @@ fil_tablespace_iterate(
 	DBUG_EXECUTE_IF("ib_import_trigger_corruption_1",
 			return(DB_CORRUPTION););
 
-	/* Make sure the data_dir_path is set. */
-	dict_get_and_save_data_dir_path(table);
-
-	ut_ad(!DICT_TF_HAS_DATA_DIR(table->flags) || table->data_dir_path);
-
-	const char *data_dir_path = DICT_TF_HAS_DATA_DIR(table->flags)
-		? table->data_dir_path : nullptr;
-
-	filepath = fil_make_filepath(data_dir_path,
-				     {table->name.m_name,
-				      strlen(table->name.m_name)},
-				     IBD, data_dir_path != nullptr);
+	table_name_t table_name(const_cast<char*>(name));
+	filepath= fil_make_filepath(dir_path, table_name, IBD,
+				    dir_path != nullptr);
 	if (!filepath) {
 		return(DB_OUT_OF_MEMORY);
 	} else {
@@ -4379,9 +4459,9 @@ fil_tablespace_iterate(
 		if (!success) {
 			/* The following call prints an error message */
 			os_file_get_last_error(true);
-			ib::error() << "Trying to import a tablespace,"
-				" but could not open the tablespace file "
-				    << filepath;
+			sql_print_error("InnoDB: could not open the "
+					"tablespace file %s.\n",
+					filepath);
 			ut_free(filepath);
 			return DB_TABLESPACE_NOT_FOUND;
 		} else {
@@ -4444,17 +4524,21 @@ fil_tablespace_iterate(
 		iter.file_size = file_size;
 		iter.n_io_buffers = n_io_buffers;
 
+		size_t buf_size = (1 + iter.n_io_buffers) * srv_page_size;
+
 		/* Add an extra page for compressed page scratch area. */
 		iter.io_buffer = static_cast<byte*>(
-			aligned_malloc((1 + iter.n_io_buffers)
-				       << srv_page_size_shift, srv_page_size));
+			aligned_malloc(buf_size, srv_page_size));
 
-		iter.crypt_io_buffer = iter.crypt_data
-			? static_cast<byte*>(
-				aligned_malloc((1 + iter.n_io_buffers)
-					       << srv_page_size_shift,
-					       srv_page_size))
-			: NULL;
+		if (iter.crypt_data) {
+			iter.crypt_io_buffer = static_cast<byte *>(
+				aligned_malloc(buf_size, srv_page_size));
+			iter.crypt_tmp_buffer = static_cast<byte *>(
+				aligned_malloc(buf_size, CPU_LEVEL1_DCACHE_LINESIZE));
+		} else {
+			iter.crypt_io_buffer = NULL;
+			iter.crypt_tmp_buffer = NULL;
+		}
 
 		if (block->page.zip.ssize) {
 			ut_ad(iter.n_io_buffers == 1);
@@ -4469,6 +4553,7 @@ fil_tablespace_iterate(
 			fil_space_destroy_crypt_data(&iter.crypt_data);
 		}
 
+		aligned_free(iter.crypt_tmp_buffer);
 		aligned_free(iter.crypt_io_buffer);
 		aligned_free(iter.io_buffer);
 	}
@@ -4493,6 +4578,24 @@ fil_tablespace_iterate(
 	return(err);
 }
 
+/**
+Iterate over all or some pages in the tablespace.
+@param table         the table definiton in the server
+@param n_io_buffers  number of blocks to read and write together
+@param callback      functor that will do the page queries or updates
+@return DB_SUCCESS or error code */
+static dberr_t fil_tablespace_iterate(dict_table_t *table, ulint n_io_buffers,
+                                      AbstractCallback &callback)
+{
+  /* Make sure the data_dir_path is set. */
+  dict_get_and_save_data_dir_path(table);
+  ut_ad(!DICT_TF_HAS_DATA_DIR(table->flags) || table->data_dir_path);
+  const char *data_dir_path= DICT_TF_HAS_DATA_DIR(table->flags)
+    ? table->data_dir_path : nullptr;
+  return fil_tablespace_iterate(table->name.m_name, n_io_buffers, callback,
+                                data_dir_path);
+}
+
 static void row_import_autoinc(dict_table_t *table, row_prebuilt_t *prebuilt,
                                uint64_t autoinc)
 {
@@ -4621,8 +4724,6 @@ row_import_for_mysql(
 	ut_ad(!table->is_readable());
 	ut_ad(prebuilt->table == table);
 
-	ibuf_delete_for_discarded_space(table->space_id);
-
 #ifdef BTR_CUR_HASH_ADAPT
 	/* On DISCARD TABLESPACE, we did not drop any adaptive hash
 	index entries. If we replaced the discarded tablespace with a
@@ -4830,12 +4931,6 @@ import_error:
 
 	ut_free(filepath);
 
-	if (err == DB_SUCCESS) {
-		err = ibuf_check_bitmap_on_import(trx, table->space);
-	}
-
-	DBUG_EXECUTE_IF("ib_import_check_bitmap_failure", err = DB_CORRUPTION;);
-
 	if (err != DB_SUCCESS) {
 		return row_import_cleanup(prebuilt, err);
 	}
@@ -4893,13 +4988,6 @@ import_error:
 		goto import_error;
 	}
 
-	/* Ensure that the next available DB_ROW_ID is not smaller than
-	any DB_ROW_ID stored in the table. */
-
-	if (prebuilt->clust_index_was_generated) {
-		row_import_set_sys_max_row_id(prebuilt, table);
-	}
-
 	ib::info() << "Phase III - Flush changes to disk";
 
 	/* Ensure that all pages dirtied during the IMPORT make it to disk.
@@ -4963,3 +5051,58 @@ import_error:
 
 	return row_import_cleanup(prebuilt, err, table);
 }
+
+/** Prepare the create info to create a new stub table for import.
+@param thd          Connection
+@param name         Table name, format: "db/table_name".
+@param create_info  The create info for creating a stub.
+@return ER_ error code
+@retval 0 on success */
+int prepare_create_stub_for_import(THD *thd, const char *name,
+                                   HA_CREATE_INFO& create_info)
+{
+  DBUG_ENTER("prepare_create_stub_for_import");
+  FetchIndexRootPages fetchIndexRootPages;
+  if (fil_tablespace_iterate(name, IO_BUFFER_SIZE(srv_page_size),
+                             fetchIndexRootPages, fil_path_to_mysql_datadir)
+      != DB_SUCCESS)
+  {
+    const char *ibd_path= fil_make_filepath(
+      fil_path_to_mysql_datadir, table_name_t(const_cast<char*>(name)), IBD,
+      true);
+    if (!ibd_path)
+      return(ER_ENGINE_OUT_OF_MEMORY);
+    sql_print_error("InnoDB: failed to get row format from %s.\n",
+                    ibd_path);
+    DBUG_RETURN(ER_INNODB_IMPORT_ERROR);
+  }
+  create_info.init();
+  /* get the row format from ibd. */
+  create_info.row_type= fetchIndexRootPages.m_row_format;
+  /* if .cfg exists, get the row format from cfg, and compare with
+  ibd, report error if different, except when cfg reports
+  compact/dynamic and ibd reports not_used (indicating either compact
+  or dynamic but not sure) */
+  const enum row_type row_type_from_cfg=
+    get_row_type_from_cfg(fil_path_to_mysql_datadir, name, thd);
+  if (row_type_from_cfg != ROW_TYPE_NOT_USED)
+  {
+    /* if ibd reports not_used but cfg reports compact or dynamic, go
+    with cfg. */
+    if (create_info.row_type != row_type_from_cfg &&
+        !((row_type_from_cfg == ROW_TYPE_COMPACT ||
+           row_type_from_cfg == ROW_TYPE_DYNAMIC) &&
+          create_info.row_type == ROW_TYPE_NOT_USED))
+    {
+      sql_print_error(
+        "InnoDB: cfg and ibd disagree on row format for table %s.\n",
+        name);
+      DBUG_RETURN(ER_INNODB_IMPORT_ERROR);
+    }
+    else
+      create_info.row_type= row_type_from_cfg;
+  }
+  else if (create_info.row_type == ROW_TYPE_NOT_USED)
+    create_info.row_type= ROW_TYPE_DYNAMIC;
+  DBUG_RETURN(0);
+}
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 1f319aae..952ccee4 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -31,7 +31,6 @@ Created 4/20/1996 Heikki Tuuri
 #include "btr0btr.h"
 #include "btr0cur.h"
 #include "mach0data.h"
-#include "ibuf0ibuf.h"
 #include "que0que.h"
 #include "row0upd.h"
 #include "row0sel.h"
@@ -2686,8 +2685,6 @@ err_exit:
 		page_set_autoinc(root, auto_inc, &mtr, false);
 	}
 
-	btr_pcur_get_btr_cur(&pcur)->thr = thr;
-
 #ifdef UNIV_DEBUG
 	{
 		page_t*	page = btr_pcur_get_page(&pcur);
@@ -2972,7 +2969,6 @@ row_ins_sec_index_entry_low(
 	ut_ad(!dict_index_is_clust(index));
 	ut_ad(mode == BTR_MODIFY_LEAF || mode == BTR_INSERT_TREE);
 
-	cursor.thr = thr;
 	cursor.rtr_info = NULL;
 	cursor.page_cur.index = index;
 	ut_ad(thr_get_trx(thr)->id != 0);
@@ -2994,9 +2990,10 @@ row_ins_sec_index_entry_low(
 
 	if (index->is_spatial()) {
 		rtr_init_rtr_info(&rtr_info, false, &cursor, index, false);
+		rtr_info.thr = thr;
 		rtr_info_update_btr(&cursor, &rtr_info);
 
-		err = rtr_insert_leaf(&cursor, entry, search_mode, &mtr);
+		err = rtr_insert_leaf(&cursor, thr, entry, search_mode, &mtr);
 
 		if (err == DB_SUCCESS && search_mode == BTR_MODIFY_LEAF
 		    && rtr_info.mbr_adj) {
@@ -3005,6 +3002,7 @@ row_ins_sec_index_entry_low(
 			rtr_clean_rtr_info(&rtr_info, true);
 			rtr_init_rtr_info(&rtr_info, false, &cursor,
 					  index, false);
+			rtr_info.thr = thr;
 			rtr_info_update_btr(&cursor, &rtr_info);
 			mtr.start();
 			if (index->table->is_temporary()) {
@@ -3012,7 +3010,7 @@ row_ins_sec_index_entry_low(
 			} else {
 				index->set_modified(mtr);
 			}
-			err = rtr_insert_leaf(&cursor, entry,
+			err = rtr_insert_leaf(&cursor, thr, entry,
 					      search_mode, &mtr);
 		}
 
@@ -3021,14 +3019,6 @@ row_ins_sec_index_entry_low(
 			goto func_exit;});
 
 	} else {
-		if (!index->table->is_temporary()) {
-			search_mode = btr_latch_mode(
-				search_mode
-				| (thr_get_trx(thr)->check_unique_secondary
-				   ? BTR_INSERT
-				   : BTR_INSERT | BTR_IGNORE_SEC_UNIQUE));
-		}
-
 		err = cursor.search_leaf(entry, PAGE_CUR_LE, search_mode,
 					 &mtr);
 	}
@@ -3040,12 +3030,6 @@ row_ins_sec_index_entry_low(
 		goto func_exit;
 	}
 
-	if (cursor.flag == BTR_CUR_INSERT_TO_IBUF) {
-		ut_ad(!dict_index_is_spatial(index));
-		/* The insert was buffered during the search: we are done */
-		goto func_exit;
-	}
-
 #ifdef UNIV_DEBUG
 	{
 		page_t*	page = btr_cur_get_page(&cursor);
@@ -3105,13 +3089,9 @@ row_ins_sec_index_entry_low(
 		locked with s-locks the necessary records to
 		prevent any insertion of a duplicate by another
 		transaction. Let us now reposition the cursor and
-		continue the insertion (bypassing the change buffer). */
-		err = cursor.search_leaf(
-			entry, PAGE_CUR_LE,
-			btr_latch_mode(search_mode
-				       & ~(BTR_INSERT
-					   | BTR_IGNORE_SEC_UNIQUE)),
-			&mtr);
+		continue the insertion. */
+		err = cursor.search_leaf(entry, PAGE_CUR_LE, search_mode,
+					 &mtr);
 		if (err != DB_SUCCESS) {
 			goto func_exit;
 		}
@@ -3342,11 +3322,6 @@ row_ins_sec_index_entry(
 	if (err == DB_FAIL) {
 		mem_heap_empty(heap);
 
-		if (index->table->space == fil_system.sys_space
-		    && !(index->type & (DICT_UNIQUE | DICT_SPATIAL))) {
-			ibuf_free_excess_pages();
-		}
-
 		/* Try then pessimistic descent to the B-tree */
 		log_free_check();
 
@@ -3570,19 +3545,6 @@ row_ins_index_entry_step(
 }
 
 /***********************************************************//**
-Allocates a row id for row and inits the node->index field. */
-UNIV_INLINE
-void
-row_ins_alloc_row_id_step(
-/*======================*/
-	ins_node_t*	node)	/*!< in: row insert node */
-{
-  ut_ad(node->state == INS_NODE_ALLOC_ROW_ID);
-  if (dict_table_get_first_index(node->table)->is_gen_clust())
-    dict_sys_write_row_id(node->sys_buf, dict_sys.get_new_row_id());
-}
-
-/***********************************************************//**
 Gets a row to insert from the values list. */
 UNIV_INLINE
 void
@@ -3662,13 +3624,18 @@ row_ins(
 	DBUG_PRINT("row_ins", ("table: %s", node->table->name.m_name));
 
 	if (node->state == INS_NODE_ALLOC_ROW_ID) {
-
-		row_ins_alloc_row_id_step(node);
-
 		node->index = dict_table_get_first_index(node->table);
 		ut_ad(node->entry_list.empty() == false);
 		node->entry = node->entry_list.begin();
 
+		if (node->index->is_gen_clust()) {
+			const uint64_t db_row_id{++node->table->row_id};
+			if (db_row_id >> 48) {
+				DBUG_RETURN(DB_OUT_OF_FILE_SPACE);
+			}
+			mach_write_to_6(node->sys_buf, db_row_id);
+		}
+
 		if (node->ins_type == INS_SEARCHED) {
 
 			row_ins_get_row_from_select(node);
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index c4f46304..882e7c64 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -1701,22 +1701,6 @@ err_exit:
 		if (error) {
 			goto err_exit;
 		}
-#ifdef UNIV_DEBUG
-		switch (btr_pcur_get_btr_cur(pcur)->flag) {
-		case BTR_CUR_DELETE_REF:
-		case BTR_CUR_DEL_MARK_IBUF:
-		case BTR_CUR_DELETE_IBUF:
-		case BTR_CUR_INSERT_TO_IBUF:
-			/* We did not request buffering. */
-			break;
-		case BTR_CUR_HASH:
-		case BTR_CUR_HASH_FAIL:
-		case BTR_CUR_BINARY:
-			goto flag_ok;
-		}
-		ut_ad(0);
-flag_ok:
-#endif /* UNIV_DEBUG */
 
 		if (page_rec_is_infimum(btr_pcur_get_rec(pcur))
 		    || btr_pcur_get_low_match(pcur) < index->n_uniq) {
@@ -1724,8 +1708,8 @@ flag_ok:
 			found, because new_table is being modified by
 			this thread only, and all indexes should be
 			updated in sync. */
-			mtr->commit();
-			return(DB_INDEX_CORRUPT);
+			error = DB_INDEX_CORRUPT;
+			goto err_exit;
 		}
 
 		btr_cur_pessimistic_delete(&error, FALSE,
@@ -1785,22 +1769,6 @@ row_log_table_apply_delete(
 	if (err != DB_SUCCESS) {
 		goto all_done;
 	}
-#ifdef UNIV_DEBUG
-	switch (btr_pcur_get_btr_cur(&pcur)->flag) {
-	case BTR_CUR_DELETE_REF:
-	case BTR_CUR_DEL_MARK_IBUF:
-	case BTR_CUR_DELETE_IBUF:
-	case BTR_CUR_INSERT_TO_IBUF:
-		/* We did not request buffering. */
-		break;
-	case BTR_CUR_HASH:
-	case BTR_CUR_HASH_FAIL:
-	case BTR_CUR_BINARY:
-		goto flag_ok;
-	}
-	ut_ad(0);
-flag_ok:
-#endif /* UNIV_DEBUG */
 
 	if (page_rec_is_infimum(btr_pcur_get_rec(&pcur))
 	    || btr_pcur_get_low_match(&pcur) < index->n_uniq) {
@@ -1934,19 +1902,6 @@ func_exit_committed:
 
 		return error;
 	}
-#ifdef UNIV_DEBUG
-	switch (btr_pcur_get_btr_cur(&pcur)->flag) {
-	case BTR_CUR_DELETE_REF:
-	case BTR_CUR_DEL_MARK_IBUF:
-	case BTR_CUR_DELETE_IBUF:
-	case BTR_CUR_INSERT_TO_IBUF:
-		ut_ad(0);/* We did not request buffering. */
-	case BTR_CUR_HASH:
-	case BTR_CUR_HASH_FAIL:
-	case BTR_CUR_BINARY:
-		break;
-	}
-#endif /* UNIV_DEBUG */
 
 	ut_ad(!page_rec_is_infimum(btr_pcur_get_rec(&pcur))
 	      && btr_pcur_get_low_match(&pcur) >= index->n_uniq);
@@ -2096,8 +2051,17 @@ func_exit_committed:
 		ut_free(pcur.old_rec_buf);
 		pcur.old_rec_buf = nullptr;
 
-		if (ROW_FOUND != row_search_index_entry(
-			    entry, BTR_MODIFY_TREE, &pcur, &mtr)) {
+		error = btr_pcur_open(entry, PAGE_CUR_LE, BTR_MODIFY_TREE,
+				      &pcur, &mtr);
+
+		if (error != DB_SUCCESS) {
+			ut_ad(0);
+			break;
+		}
+
+		if (btr_pcur_is_before_first_on_page(&pcur)
+		    || btr_pcur_get_low_match(&pcur)
+		    != dtuple_get_n_fields(entry)) {
 			ut_ad(0);
 			error = DB_CORRUPTION;
 			break;
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 6fb530f0..08ccaea2 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -150,7 +150,7 @@ public:
 					  false);
 			rtr_info_update_btr(&ins_cur, &rtr_info);
 
-			error = rtr_insert_leaf(&ins_cur, dtuple,
+			error = rtr_insert_leaf(&ins_cur, nullptr, dtuple,
 						BTR_MODIFY_LEAF, &mtr);
 
 			/* It need to update MBR in parent entry,
@@ -163,7 +163,8 @@ public:
 				rtr_info_update_btr(&ins_cur, &rtr_info);
 				mtr.start();
 				index->set_modified(mtr);
-				error = rtr_insert_leaf(&ins_cur, dtuple,
+				error = rtr_insert_leaf(&ins_cur, nullptr,
+							dtuple,
 							BTR_MODIFY_TREE, &mtr);
 			}
 
@@ -186,7 +187,8 @@ public:
 						  &ins_cur, index, false);
 
 				rtr_info_update_btr(&ins_cur, &rtr_info);
-				error = rtr_insert_leaf(&ins_cur, dtuple,
+				error = rtr_insert_leaf(&ins_cur, nullptr,
+							dtuple,
 							BTR_MODIFY_TREE, &mtr);
 
 				if (error == DB_SUCCESS) {
@@ -2227,7 +2229,7 @@ end_of_index:
 						  next_page_no),
 					old_table->space->zip_size(),
 					RW_S_LATCH, nullptr, BUF_GET, &mtr,
-					&err, false);
+					&err);
 				if (!block) {
 					goto err_exit;
 				}
@@ -3687,8 +3689,6 @@ row_merge_mtuple_to_dtuple(
 	dtuple_t*	dtuple,
 	const mtuple_t* mtuple)
 {
-	ut_ad(!dict_index_is_ibuf(index));
-
 	memcpy(dtuple->fields, mtuple->fields,
 	       dtuple->n_fields * sizeof *mtuple->fields);
 }
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 6a71cf3a..4b570779 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2000, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2022, MariaDB Corporation.
+Copyright (c) 2015, 2023, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -44,7 +44,6 @@ Created 9/17/2000 Heikki Tuuri
 #include "fsp0file.h"
 #include "fts0fts.h"
 #include "fts0types.h"
-#include "ibuf0ibuf.h"
 #include "lock0lock.h"
 #include "log0log.h"
 #include "pars0pars.h"
@@ -2339,12 +2338,7 @@ row_discard_tablespace(
 	2) Purge and rollback: we assign a new table id for the
 	table. Since purge and rollback look for the table based on
 	the table id, they see the table as 'dropped' and discard
-	their operations.
-
-	3) Insert buffer: we remove all entries for the tablespace in
-	the insert buffer tree. */
-
-	ibuf_delete_for_discarded_space(table->space_id);
+	their operations. */
 
 	table_id_t	new_id;
 
@@ -2446,15 +2440,23 @@ rollback:
   /* Note: The following cannot be rolled back. Rollback would see the
   UPDATE of SYS_INDEXES.TABLE_ID as two operations: DELETE and INSERT.
   It would invoke btr_free_if_exists() when rolling back the INSERT,
-  effectively dropping all indexes of the table. Furthermore, calls like
-  ibuf_delete_for_discarded_space() are already discarding data
-  before the transaction is committed.
+  effectively dropping all indexes of the table. Furthermore, we are
+  already discarding data before the transaction is committed.
 
   It would be better to remove the integrity-breaking
   ALTER TABLE...DISCARD TABLESPACE operation altogether. */
   table->file_unreadable= true;
   table->space= nullptr;
+#if defined __GNUC__ && !defined __clang__
+# pragma GCC diagnostic push
+# if __GNUC__ < 12 || defined WITH_UBSAN
+#  pragma GCC diagnostic ignored "-Wconversion"
+# endif
+#endif
   table->flags2|= DICT_TF2_DISCARDED;
+#if defined __GNUC__ && !defined __clang__
+# pragma GCC diagnostic pop
+#endif
   err= row_discard_tablespace(trx, table);
   DBUG_EXECUTE_IF("ib_discard_before_commit_crash",
                   log_buffer_flush_to_disk(); DBUG_SUICIDE(););
@@ -2474,7 +2476,6 @@ rollback:
   if (fts_exist)
     purge_sys.resume_FTS();
 
-  ibuf_delete_for_discarded_space(space_id);
   buf_flush_remove_pages(space_id);
   trx->op_info= "";
   return err;
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index d83ab861..adac8ecf 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -185,10 +185,6 @@ close_and_exit:
 			table = nullptr;
 		}
 
-		if (space_id) {
-			ibuf_delete_for_discarded_space(space_id);
-		}
-
 		mtr.start();
 		index->set_modified(mtr);
 
@@ -277,10 +273,10 @@ not delete marked version of a clustered index record where DB_TRX_ID
 is newer than the purge view.
 
 NOTE: This function should only be called by the purge thread, only
-while holding a latch on the leaf page of the secondary index entry
-(or keeping the buffer pool watch on the page).  It is possible that
-this function first returns true and then false, if a user transaction
-inserts a record that the secondary index entry would refer to.
+while holding a latch on the leaf page of the secondary index entry.
+It is possible that this function first returns true and then false,
+if a user transaction inserts a record that the secondary index entry
+would refer to.
 However, in that case, the user transaction would also re-insert the
 secondary index entry after purge has removed it and released the leaf
 page latch.
@@ -296,6 +292,7 @@ page latch.
 @param[in]	is_tree		true=pessimistic purge,
 				false=optimistic (leaf-page only)
 @return true if the secondary index record can be purged */
+static
 bool
 row_purge_poss_sec(
 	purge_node_t*	node,
@@ -353,14 +350,11 @@ row_purge_remove_sec_if_poss_tree(
 	pcur.btr_cur.page_cur.index = index;
 
 	if (index->is_spatial()) {
-		if (!rtr_search(entry, BTR_PURGE_TREE, &pcur, &mtr)) {
-			goto found;
+		if (rtr_search(entry, BTR_PURGE_TREE, &pcur, nullptr, &mtr)) {
+			goto func_exit;
 		}
-		goto func_exit;
-	}
-
-	switch (row_search_index_entry(entry, BTR_PURGE_TREE, &pcur, &mtr)) {
-	case ROW_NOT_FOUND:
+	} else if (!row_search_index_entry(entry, BTR_PURGE_TREE,
+					   &pcur, &mtr)) {
 		/* Not found.  This is a legitimate condition.  In a
 		rollback, InnoDB will remove secondary recs that would
 		be purged anyway.  Then the actual purge will not find
@@ -370,25 +364,13 @@ row_purge_remove_sec_if_poss_tree(
 		index, it will remove it.  Then if/when the purge
 		comes to consider the secondary index record a second
 		time, it will not exist any more in the index. */
-
-		/* fputs("PURGE:........sec entry not found\n", stderr); */
-		/* dtuple_print(stderr, entry); */
 		goto func_exit;
-	case ROW_FOUND:
-		break;
-	case ROW_BUFFERED:
-	case ROW_NOT_DELETED_REF:
-		/* These are invalid outcomes, because the mode passed
-		to row_search_index_entry() did not include any of the
-		flags BTR_INSERT, BTR_DELETE, or BTR_DELETE_MARK. */
-		ut_error;
 	}
 
 	/* We should remove the index record if no later version of the row,
 	which cannot be purged yet, requires its existence. If some requires,
 	we should do nothing. */
 
-found:
 	if (row_purge_poss_sec(node, index, entry, &pcur, &mtr, true)) {
 
 		/* Remove the index record, which should have been
@@ -457,24 +439,17 @@ row_purge_remove_sec_if_poss_leaf(
 
 	pcur.btr_cur.page_cur.index = index;
 
-	/* Set the purge node for the call to row_purge_poss_sec(). */
-	pcur.btr_cur.purge_node = node;
 	if (index->is_spatial()) {
-		pcur.btr_cur.thr = NULL;
-		if (!rtr_search(entry, BTR_MODIFY_LEAF, &pcur, &mtr)) {
+		if (!rtr_search(entry, BTR_MODIFY_LEAF, &pcur, nullptr,
+				&mtr)) {
 			goto found;
 		}
-		goto func_exit;
-	}
-
-	/* Set the query thread, so that ibuf_insert_low() will be
-	able to invoke thd_get_trx(). */
-	pcur.btr_cur.thr = static_cast<que_thr_t*>(que_node_get_parent(node));
-
-	switch (row_search_index_entry(entry, index->has_virtual()
-				       ? BTR_MODIFY_LEAF : BTR_PURGE_LEAF,
-				       &pcur, &mtr)) {
-	case ROW_FOUND:
+	} else if (btr_pcur_open(entry, PAGE_CUR_LE, BTR_MODIFY_LEAF, &pcur,
+				 &mtr)
+		   == DB_SUCCESS
+		   && !btr_pcur_is_before_first_on_page(&pcur)
+		   && btr_pcur_get_low_match(&pcur)
+		   == dtuple_get_n_fields(entry)) {
 found:
 		/* Before attempting to purge a record, check
 		if it is safe to do so. */
@@ -503,25 +478,18 @@ found:
 			if (index->is_spatial()) {
 				const buf_block_t* block = btr_cur_get_block(
 					btr_cur);
+                                const page_id_t id{block->page.id()};
 
-				if (block->page.id().page_no()
-				    != index->page
+				if (id.page_no() != index->page
 				    && page_get_n_recs(block->page.frame) < 2
-				    && !lock_test_prdt_page_lock(
-					    btr_cur->rtr_info
-					    && btr_cur->rtr_info->thr
-					    ? thr_get_trx(
-						    btr_cur->rtr_info->thr)
-					    : nullptr,
-					    block->page.id())) {
+				    && !lock_test_prdt_page_lock(nullptr, id)){
 					/* this is the last record on page,
 					and it has a "page" lock on it,
 					which mean search is still depending
 					on it, so do not delete */
 					DBUG_LOG("purge",
 						 "skip purging last"
-						 " record on page "
-						 << block->page.id());
+						 " record on page " << id);
 					goto func_exit;
 				}
 			}
@@ -529,25 +497,13 @@ found:
 			success = btr_cur_optimistic_delete(btr_cur, 0, &mtr)
 				!= DB_FAIL;
 		}
+	}
 
-		/* (The index entry is still needed,
-		or the deletion succeeded) */
-		/* fall through */
-	case ROW_NOT_DELETED_REF:
-		/* The index entry is still needed. */
-	case ROW_BUFFERED:
-		/* The deletion was buffered. */
-	case ROW_NOT_FOUND:
-		/* The index entry does not exist, nothing to do. */
 func_exit:
-		mtr.commit();
+	mtr.commit();
 cleanup:
-		btr_pcur_close(&pcur); // FIXME: do we need these? when is btr_cur->rtr_info set?
-		return(success);
-	}
-
-	ut_error;
-	return(false);
+	btr_pcur_close(&pcur);
+	return success;
 }
 
 /***********************************************************//**
@@ -600,10 +556,7 @@ Purges a delete marking of a record.
 @retval false the purge needs to be suspended because of
 running out of file space */
 static MY_ATTRIBUTE((nonnull, warn_unused_result))
-bool
-row_purge_del_mark(
-/*===============*/
-	purge_node_t*	node)	/*!< in/out: row purge node */
+bool row_purge_del_mark(purge_node_t *node)
 {
   if (node->index)
   {
diff --git a/storage/innobase/row/row0quiesce.cc b/storage/innobase/row/row0quiesce.cc
index 057b20c7..f86dbf2f 100644
--- a/storage/innobase/row/row0quiesce.cc
+++ b/storage/innobase/row/row0quiesce.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2012, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2023, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -27,7 +27,6 @@ Created 2012-02-08 by Sunny Bains.
 #include "row0quiesce.h"
 #include "row0mysql.h"
 #include "buf0flu.h"
-#include "ibuf0ibuf.h"
 #include "srv0start.h"
 #include "trx0purge.h"
 
@@ -539,18 +538,6 @@ row_quiesce_table_start(
 		purge_sys.stop();
 	}
 
-	for (ulint count = 0;
-	     ibuf_merge_space(table->space_id);
-	     ++count) {
-		if (trx_is_interrupted(trx)) {
-			goto aborted;
-		}
-		if (!(count % 20)) {
-			ib::info() << "Merging change buffer entries for "
-				<< table->name;
-		}
-	}
-
 	while (buf_flush_list_space(table->space)) {
 		if (trx_is_interrupted(trx)) {
 			goto aborted;
diff --git a/storage/innobase/row/row0row.cc b/storage/innobase/row/row0row.cc
index 4a00b2a4..a7cddee0 100644
--- a/storage/innobase/row/row0row.cc
+++ b/storage/innobase/row/row0row.cc
@@ -215,28 +215,20 @@ row_build_index_entry_low(
 		entry = dtuple_create(heap, entry_len);
 	}
 
-	if (dict_index_is_ibuf(index)) {
-		dtuple_set_n_fields_cmp(entry, entry_len);
-		/* There may only be externally stored columns
-		in a clustered index B-tree of a user table. */
-		ut_a(!ext);
-	} else {
-		dtuple_set_n_fields_cmp(
-			entry, dict_index_get_n_unique_in_tree(index));
-		if (dict_index_is_spatial(index)) {
-			/* Set the MBR field */
-			if (!row_build_spatial_index_key(
-				    index, ext,
-				    dtuple_get_nth_field(entry, 0),
-				    dtuple_get_nth_field(
-					    row,
-					    dict_index_get_nth_field(index, i)
-					    ->col->ind), flag, heap)) {
-				return NULL;
-			}
-
-			i = 1;
+	dtuple_set_n_fields_cmp(entry, dict_index_get_n_unique_in_tree(index));
+	if (index->is_spatial()) {
+		/* Set the MBR field */
+		if (!row_build_spatial_index_key(
+			    index, ext,
+			    dtuple_get_nth_field(entry, 0),
+			    dtuple_get_nth_field(
+				    row,
+				    dict_index_get_nth_field(index, i)
+				    ->col->ind), flag, heap)) {
+			return NULL;
 		}
+
+		i = 1;
 	}
 
 	for (; i < entry_len; i++) {
@@ -1262,8 +1254,8 @@ row_get_clust_rec(
 
 /***************************************************************//**
 Searches an index record.
-@return whether the record was found or buffered */
-enum row_search_result
+@return whether the record was found */
+bool
 row_search_index_entry(
 /*===================*/
 	const dtuple_t*	entry,	/*!< in: index entry */
@@ -1272,47 +1264,14 @@ row_search_index_entry(
 				be closed by the caller */
 	mtr_t*		mtr)	/*!< in: mtr */
 {
-	ulint	n_fields;
-	ulint	low_match;
-	rec_t*	rec;
-
 	ut_ad(dtuple_check_typed(entry));
 
 	if (btr_pcur_open(entry, PAGE_CUR_LE, mode, pcur, mtr) != DB_SUCCESS) {
-		return ROW_NOT_FOUND;
-	}
-
-	switch (btr_pcur_get_btr_cur(pcur)->flag) {
-	case BTR_CUR_DELETE_REF:
-		ut_ad(!(~mode & BTR_DELETE));
-		return(ROW_NOT_DELETED_REF);
-
-	case BTR_CUR_DEL_MARK_IBUF:
-	case BTR_CUR_DELETE_IBUF:
-	case BTR_CUR_INSERT_TO_IBUF:
-		return(ROW_BUFFERED);
-
-	case BTR_CUR_HASH:
-	case BTR_CUR_HASH_FAIL:
-	case BTR_CUR_BINARY:
-		break;
-	}
-
-	low_match = btr_pcur_get_low_match(pcur);
-
-	rec = btr_pcur_get_rec(pcur);
-
-	n_fields = dtuple_get_n_fields(entry);
-
-	if (page_rec_is_infimum(rec)) {
-
-		return(ROW_NOT_FOUND);
-	} else if (low_match != n_fields) {
-
-		return(ROW_NOT_FOUND);
+		return false;
 	}
 
-	return(ROW_FOUND);
+	return !btr_pcur_is_before_first_on_page(pcur)
+		&& btr_pcur_get_low_match(pcur) == dtuple_get_n_fields(entry);
 }
 
 /*******************************************************************//**
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 944f7358..0a38e30b 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -4039,7 +4039,8 @@ row_search_idx_cond_check(
 	ut_ad(rec_offs_validate(rec, prebuilt->index, offsets));
 
 	if (!prebuilt->idx_cond) {
-		if (!handler_rowid_filter_is_active(prebuilt->pk_filter)) {
+		if (!prebuilt->pk_filter ||
+                    !handler_rowid_filter_is_active(prebuilt->pk_filter)) {
 			return(CHECK_POS);
 		}
 	} else {
@@ -4081,7 +4082,8 @@ row_search_idx_cond_check(
 
 	switch (result) {
 	case CHECK_POS:
-	        if (handler_rowid_filter_is_active(prebuilt->pk_filter)) {
+	        if (prebuilt->pk_filter &&
+                  handler_rowid_filter_is_active(prebuilt->pk_filter)) {
 		        ut_ad(!prebuilt->index->is_primary());
 		        if (prebuilt->clust_index_was_generated) {
                                ulint len;
@@ -4765,14 +4767,13 @@ wait_table_again:
 		}
 
 	} else if (dtuple_get_n_fields(search_tuple) > 0) {
-		pcur->btr_cur.thr = thr;
 		pcur->old_rec = nullptr;
 
 		if (index->is_spatial()) {
 			if (!prebuilt->rtr_info) {
 				prebuilt->rtr_info = rtr_create_rtr_info(
-					set_also_gap_locks, true,
-					btr_pcur_get_btr_cur(pcur), index);
+					set_also_gap_locks, true, thr,
+					btr_pcur_get_btr_cur(pcur));
 				prebuilt->rtr_info->search_tuple = search_tuple;
 				prebuilt->rtr_info->search_mode = mode;
 				rtr_info_update_btr(btr_pcur_get_btr_cur(pcur),
@@ -4785,7 +4786,8 @@ wait_table_again:
 				prebuilt->rtr_info->search_mode = mode;
 			}
 
-			err = rtr_search_leaf(pcur, search_tuple, mode, &mtr);
+			err = rtr_search_leaf(pcur, thr, search_tuple, mode,
+					      &mtr);
 		} else {
 			err = btr_pcur_open_with_no_init(search_tuple, mode,
 							 BTR_SEARCH_LEAF,
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 23255cc9..b78a2c41 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -40,7 +40,6 @@ Created 2/25/1997 Heikki Tuuri
 #include "row0row.h"
 #include "row0upd.h"
 #include "que0que.h"
-#include "ibuf0ibuf.h"
 #include "log0log.h"
 #include "fil0fil.h"
 #include <mysql/service_thd_mdl.h>
@@ -188,10 +187,6 @@ restart:
 			os_file_close(d);
 		}
 
-		if (space_id) {
-			ibuf_delete_for_discarded_space(space_id);
-		}
-
 		mtr.start();
 		ut_a(node->pcur.restore_position(
 			BTR_MODIFY_LEAF, &mtr) == btr_pcur_t::SAME_ALL);
@@ -271,7 +266,7 @@ row_undo_ins_remove_sec_low(
 	const bool		modify_leaf = mode == BTR_MODIFY_LEAF;
 
 	pcur.btr_cur.page_cur.index = index;
-	row_mtr_start(&mtr, index, !modify_leaf);
+	row_mtr_start(&mtr, index);
 
 	if (index->is_spatial()) {
 		mode = modify_leaf
@@ -279,8 +274,7 @@ row_undo_ins_remove_sec_low(
 					 | BTR_RTREE_DELETE_MARK
 					 | BTR_RTREE_UNDO_INS)
 			: btr_latch_mode(BTR_PURGE_TREE | BTR_RTREE_UNDO_INS);
-		btr_pcur_get_btr_cur(&pcur)->thr = thr;
-		if (rtr_search(entry, mode, &pcur, &mtr)) {
+		if (rtr_search(entry, mode, &pcur, thr, &mtr)) {
 			goto func_exit;
 		}
 
@@ -301,28 +295,17 @@ row_undo_ins_remove_sec_low(
 		mtr_x_lock_index(index, &mtr);
 	}
 
-	switch (row_search_index_entry(entry, mode, &pcur, &mtr)) {
-	case ROW_BUFFERED:
-	case ROW_NOT_DELETED_REF:
-		/* These are invalid outcomes, because the mode passed
-		to row_search_index_entry() did not include any of the
-		flags BTR_INSERT, BTR_DELETE, or BTR_DELETE_MARK. */
-		ut_error;
-	case ROW_NOT_FOUND:
-		break;
-	case ROW_FOUND:
-        found:
-		btr_cur_t* btr_cur = btr_pcur_get_btr_cur(&pcur);
-
+	if (row_search_index_entry(entry, mode, &pcur, &mtr)) {
+found:
 		if (modify_leaf) {
-			err = btr_cur_optimistic_delete(btr_cur, 0, &mtr);
+			err = btr_cur_optimistic_delete(&pcur.btr_cur, 0, &mtr);
 		} else {
 			/* Passing rollback=false here, because we are
 			deleting a secondary index record: the distinction
 			only matters when deleting a record that contains
 			externally stored columns. */
-			btr_cur_pessimistic_delete(&err, FALSE, btr_cur, 0,
-						   false, &mtr);
+			btr_cur_pessimistic_delete(&err, FALSE, &pcur.btr_cur,
+						   0, false, &mtr);
 		}
 	}
 
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 52f54443..38d19882 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -33,7 +33,6 @@ Created 2/27/1997 Heikki Tuuri
 #include "trx0purge.h"
 #include "btr0btr.h"
 #include "mach0data.h"
-#include "ibuf0ibuf.h"
 #include "row0undo.h"
 #include "row0vers.h"
 #include "trx0trx.h"
@@ -491,7 +490,7 @@ row_undo_mod_del_mark_or_remove_sec_low(
 	mtr_t			mtr_vers;
 	const bool		modify_leaf = mode == BTR_MODIFY_LEAF;
 
-	row_mtr_start(&mtr, index, !modify_leaf);
+	row_mtr_start(&mtr, index);
 
 	pcur.btr_cur.page_cur.index = index;
 	btr_cur = btr_pcur_get_btr_cur(&pcur);
@@ -502,8 +501,7 @@ row_undo_mod_del_mark_or_remove_sec_low(
 					 | BTR_RTREE_DELETE_MARK
 					 | BTR_RTREE_UNDO_INS)
 			: btr_latch_mode(BTR_PURGE_TREE | BTR_RTREE_UNDO_INS);
-		btr_cur->thr = thr;
-		if (UNIV_LIKELY(!rtr_search(entry, mode, &pcur, &mtr))) {
+		if (UNIV_LIKELY(!rtr_search(entry, mode, &pcur, thr, &mtr))) {
 			goto found;
 		} else {
 			goto func_exit;
@@ -527,9 +525,7 @@ row_undo_mod_del_mark_or_remove_sec_low(
 		ut_ad(!dict_index_is_online_ddl(index));
 	}
 
-	switch (UNIV_EXPECT(row_search_index_entry(entry, mode, &pcur, &mtr),
-			    ROW_FOUND)) {
-	case ROW_NOT_FOUND:
+	if (!row_search_index_entry(entry, mode, &pcur, &mtr)) {
 		/* In crash recovery, the secondary index record may
 		be missing if the UPDATE did not have time to insert
 		the secondary index records before the crash.  When we
@@ -540,14 +536,6 @@ row_undo_mod_del_mark_or_remove_sec_low(
 		before it has inserted all updated secondary index
 		records, then the undo will not find those records. */
 		goto func_exit;
-	case ROW_FOUND:
-		break;
-	case ROW_BUFFERED:
-	case ROW_NOT_DELETED_REF:
-		/* These are invalid outcomes, because the mode passed
-		to row_search_index_entry() did not include any of the
-		flags BTR_INSERT, BTR_DELETE, or BTR_DELETE_MARK. */
-		ut_error;
 	}
 
 found:
@@ -685,12 +673,13 @@ row_undo_mod_del_unmark_sec_and_undo_update(
 	}
 
 try_again:
-	row_mtr_start(&mtr, index, mode & 8);
+	row_mtr_start(&mtr, index);
 
-	btr_cur->thr = thr;
+	mem_heap_t* offsets_heap = nullptr;
+	rec_offs* offsets = nullptr;
 
 	if (index->is_spatial()) {
-		if (!rtr_search(entry, mode, &pcur, &mtr)) {
+		if (!rtr_search(entry, mode, &pcur, thr, &mtr)) {
 			goto found;
 		}
 
@@ -704,17 +693,7 @@ try_again:
 		goto not_found;
 	}
 
-	switch (row_search_index_entry(entry, mode, &pcur, &mtr)) {
-		mem_heap_t*	heap;
-		mem_heap_t*	offsets_heap;
-		rec_offs*	offsets;
-	case ROW_BUFFERED:
-	case ROW_NOT_DELETED_REF:
-		/* These are invalid outcomes, because the mode passed
-		to row_search_index_entry() did not include any of the
-		flags BTR_INSERT, BTR_DELETE, or BTR_DELETE_MARK. */
-		ut_error;
-	case ROW_NOT_FOUND:
+	if (!row_search_index_entry(entry, mode, &pcur, &mtr)) {
 not_found:
 		if (btr_cur->up_match >= dict_index_get_n_unique(index)
 		    || btr_cur->low_match >= dict_index_get_n_unique(index)) {
@@ -726,7 +705,7 @@ not_found:
 				<< " at: " << rec_index_print(
 					btr_cur_get_rec(btr_cur), index);
 			err = DB_DUPLICATE_KEY;
-			break;
+			goto func_exit;
 		}
 
 		ib::warn() << "Record in index " << index->name
@@ -740,8 +719,6 @@ not_found:
 		delete-unmark. */
 		big_rec_t*	big_rec;
 		rec_t*		insert_rec;
-		offsets = NULL;
-		offsets_heap = NULL;
 
 		err = btr_cur_optimistic_insert(
 			flags, btr_cur, &offsets, &offsets_heap,
@@ -770,16 +747,13 @@ not_found:
 		if (offsets_heap) {
 			mem_heap_free(offsets_heap);
 		}
-
-		break;
-	case ROW_FOUND:
+	} else {
 found:
 		btr_rec_set_deleted<false>(btr_cur_get_block(btr_cur),
 					   btr_cur_get_rec(btr_cur), &mtr);
-		heap = mem_heap_create(
+		mem_heap_t* heap = mem_heap_create(
 			sizeof(upd_t)
 			+ dtuple_get_n_fields(entry) * sizeof(upd_field_t));
-		offsets_heap = NULL;
 		offsets = rec_get_offsets(
 			btr_cur_get_rec(btr_cur),
 			index, nullptr, index->n_core_fields, ULINT_UNDEFINED,
@@ -818,6 +792,7 @@ found:
 		mem_heap_free(offsets_heap);
 	}
 
+func_exit:
 	btr_pcur_close(&pcur);
 	mtr_commit(&mtr);
 
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index a39574d2..76bd1eff 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -1836,9 +1836,7 @@ row_upd_sec_index_entry(
 	dict_index_t*		index;
 	dberr_t			err	= DB_SUCCESS;
 	trx_t*			trx	= thr_get_trx(thr);
-	btr_latch_mode		mode;
 	ulint			flags;
-	enum row_search_result	search_result;
 
 	ut_ad(trx->id != 0);
 
@@ -1866,7 +1864,6 @@ row_upd_sec_index_entry(
 			    "before_row_upd_sec_index_entry");
 
 	mtr.start();
-	mode = BTR_MODIFY_LEAF;
 
 	switch (index->table->space_id) {
 	case SRV_TMP_SPACE_ID:
@@ -1876,24 +1873,17 @@ row_upd_sec_index_entry(
 	default:
 		index->set_modified(mtr);
 		/* fall through */
-	case IBUF_SPACE_ID:
+	case 0:
 		flags = index->table->no_rollback() ? BTR_NO_ROLLBACK : 0;
-		/* We can only buffer delete-mark operations if there
-		are no foreign key constraints referring to the index. */
-		if (!referenced) {
-			mode = BTR_DELETE_MARK_LEAF;
-		}
-		break;
 	}
 
-	/* Set the query thread, so that ibuf_insert_low() will be
-	able to invoke thd_get_trx(). */
-	pcur.btr_cur.thr = thr;
 	pcur.btr_cur.page_cur.index = index;
+	const rec_t *rec;
 
 	if (index->is_spatial()) {
-		mode = btr_latch_mode(BTR_MODIFY_LEAF | BTR_RTREE_DELETE_MARK);
-		if (UNIV_LIKELY(!rtr_search(entry, mode, &pcur, &mtr))) {
+		constexpr btr_latch_mode mode = btr_latch_mode(
+			BTR_MODIFY_LEAF | BTR_RTREE_DELETE_MARK);
+		if (UNIV_LIKELY(!rtr_search(entry, mode, &pcur, thr, &mtr))) {
 			goto found;
 		}
 
@@ -1903,20 +1893,8 @@ row_upd_sec_index_entry(
 		}
 
 		goto not_found;
-	}
-
-	search_result = row_search_index_entry(entry, mode, &pcur, &mtr);
-
-	switch (search_result) {
-	const rec_t* rec;
-	case ROW_NOT_DELETED_REF:	/* should only occur for BTR_DELETE */
-		ut_error;
-		break;
-	case ROW_BUFFERED:
-		/* Entry was delete marked already. */
-		break;
-
-	case ROW_NOT_FOUND:
+	} else if (!row_search_index_entry(entry, BTR_MODIFY_LEAF,
+                                           &pcur, &mtr)) {
 not_found:
 		rec = btr_pcur_get_rec(&pcur);
 		ib::error()
@@ -1930,8 +1908,7 @@ not_found:
 		ut_ad(btr_validate_index(index, 0) == DB_SUCCESS);
 		ut_ad(0);
 #endif /* UNIV_DEBUG */
-		break;
-	case ROW_FOUND:
+	} else {
 found:
 		ut_ad(err == DB_SUCCESS);
 		rec = btr_pcur_get_rec(&pcur);
@@ -1946,7 +1923,7 @@ found:
 				btr_pcur_get_block(&pcur),
 				btr_pcur_get_rec(&pcur), index, thr, &mtr);
 			if (err != DB_SUCCESS) {
-				break;
+				goto close;
 			}
 
 			btr_rec_set_deleted<true>(btr_pcur_get_block(&pcur),