MDEV-27234: Data dictionary recovery was not READ COMMITTED

This also fixes MDEV-20198: Instant ALTER TABLE is not crash safe

InnoDB dictionary recovery wrongly used the READ UNCOMMITTED isolation
level, causing some mismatch. For example, if a table was renamed or
replaced in a transaction, according to READ UNCOMMITTED the table might
not exist at all.

We implement READ COMMITTED isolation level for accessing the dictionary
tables SYS_TABLES, SYS_COLUMNS, SYS_INDEXES, SYS_FIELDS, SYS_VIRTUAL,
SYS_FOREIGN, SYS_FOREIGN_COLS. For most of these tables, no secondary
index exists. For the secondary indexes (on SYS_TABLES.ID,
SYS_FOREIGN.FOR_NAME, SYS_FOREIGN.REF_NAME), we will always look up
the primary key in the clustered index and check if the record actually
is a committed version.

dict_check_sys_tables(): Recover tablespaces also from delete-marked
committed records, so that if a matching .ibd file exists, it will
be removed by fil_delete_tablespace() when the committed delete-marked
SYS_INDEXES record of the clustered index is purged
in row_purge_remove_clust_if_poss_low().

fil_ibd_open(): Change the Boolean parameter "validate" to a ternary
one, to suppress error messages when the file might not exist.
It is possible that a .ibd file was deleted and the server shut down
before the SYS_INDEXES and SYS_TABLES records were purged. Hence, if
dict_check_sys_tables() finds a committed delete-marked record,
we must not complain if the tablespace file is not found.
On Windows, we msut treat ERROR_PATH_NOT_FOUND (directory not found)
in the same way as ERROR_FILE_NOT_FOUND. This fixes a few failures where
a previous test successfully executed DROP DATABASE (and deleted all
files and the directory), but a committed delete-marked SYS_TABLES
record had not been purged before server restart.

dict_getnext_system_low(): Do not filter out delete-marked records.

dict_startscan_system(), dict_getnext_system(): Do filter out
delete-marked records, for accessing the INFORMATION_SCHEMA tables.

dict_sys_tables_rec_read(): Return the DB_TRX_ID of the committed
version of the record. This is needed in dict_load_table_low().

dict_load_foreign_cols(), dict_load_foreign(): Add a parameter for
the current transaction identifier. In some DDL operations, the
FOREIGN KEY constraints are being loaded from the data dictionary
before the DDL transaction has been committed. For SYS_FOREIGN
and SYS_FOREIGN_COLS, we must implement the special case of
READ COMMITTED that the changes of the uncommitted current transaction
are visible.

dict_load_foreign(): Validate the table name. We could find a
SYS_FOREIGN.ID via a committed delete-marked secondary index record
that does not match the REF_NAME or FOR_NAME of the secondary index record.

dict_load_index_low(): Optionally take the table as a parameter,
so that table->def_trx_id can be updated in case of a
committed delete-marked SYS_INDEXES record corresponding
to DROP INDEX, but not corresponding to an index stub of ADD INDEX.

dict_load_indexes(): Do not update table->def_trx_id
in case of delete-marked records.

rec_is_metadata(), rec_offs_make_valid(), rec_get_offsets_func(),
row_build_low(): Relax some assertions. We may now have
!index->is_instant() even if a metadata record is present in the index.
Previously, the recovery of instant ADD/DROP COLUMN assumed
that READ UNCOMMITTED of the data dictionary will be performed.
Now, we will have a READ COMMITTED copy of the data dictionary
cache, and a READ UNCOMMITTED copy of the metadata record.

btr_page_reorganize_low(): Correctly update the FIL_PAGE_TYPE
when rolling back an instant ADD/DROP COLUMN operation.

row_rec_to_index_entry_impl(): Relax some assertions,
and disallow accessing "extra" fields. This fixes the recovery
of a crash during an instant ADD COLUMN after a successful
instant DROP COLUMN, in the test innodb.instant_alter_crash.

Tested by: Matthias Leich

mysql-test/suite/innodb/r/row_format_redundant.result

@@ -1,3 +1,4 @@
+SET GLOBAL innodb_fast_shutdown=0;
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/row_format_redundant --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/row_format_redundant --innodb-data-file-path=ibdata1:1M:autoextend --innodb-undo-tablespaces=0 --innodb-stats-persistent=0
 SET GLOBAL innodb_file_per_table=1;
 #
@@ -8,25 +9,17 @@ SET GLOBAL innodb_file_per_table=ON;
 create table t1 (a int not null, d varchar(15) not null, b
 varchar(198) not null, c char(156)) engine=InnoDB
 row_format=redundant;
-insert into t1 values(123, 'abcdef', 'jghikl', 'mnop');
-insert into t1 values(456, 'abcdef', 'jghikl', 'mnop');
-insert into t1 values(789, 'abcdef', 'jghikl', 'mnop');
-insert into t1 values(134, 'kasdfsdsadf', 'adfjlasdkfjasd', 'adfsadflkasdasdfljasdf');
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
+create temporary table t like t1;
+insert into t values(123, 'abcdef', 'jghikl', 'mnop');
+insert into t values(456, 'abcdef', 'jghikl', 'mnop');
+insert into t values(789, 'abcdef', 'jghikl', 'mnop');
+insert into t values(134, 'kasdfsdsadf', 'adfjlasdkfjasd', 'adfsadflkasdasdfljasdf');
+insert into t1 select a,d,b,c from t, seq_1_to_1024;
 SET GLOBAL innodb_file_per_table=OFF;
 create table t2 (a int not null, d varchar(15) not null, b
 varchar(198) not null, c char(156), fulltext ftsic(c)) engine=InnoDB
 row_format=redundant;
-insert into t2 select * from t1;
+insert into t2 select a,d,b,c from t, seq_1_to_1024;
 create table t3 (a int not null, d varchar(15) not null, b varchar(198),
 c varchar(150), index k1(c(99), b(56)), index k2(b(5), c(10))) engine=InnoDB
 row_format=redundant;

mysql-test/suite/innodb/t/alter_crash_rebuild.test

@@ -0,0 +1,26 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+
+CREATE TABLE t1 (a INT NOT NULL) ENGINE=InnoDB STATS_PERSISTENT=0;
+
+connect ddl,localhost,root;
+SET DEBUG_SYNC='after_trx_committed_in_memory SIGNAL stuck WAIT_FOR ever EXECUTE 2';
+send ALTER TABLE t1 ADD PRIMARY KEY(a);
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR stuck';
+SET DEBUG_SYNC='now SIGNAL ever';
+SET DEBUG_SYNC='now WAIT_FOR stuck';
+
+SET GLOBAL innodb_log_checkpoint_now=ON;
+
+--let $shutdown_timeout=0
+--source include/restart_mysqld.inc
+
+disconnect ddl;
+
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+DROP TABLE t1;
+--source include/wait_all_purged.inc

mysql-test/suite/innodb/t/row_format_redundant.test

@@ -1,6 +1,7 @@
 --source include/have_innodb.inc
 # Embedded mode doesn't allow restarting
 --source include/not_embedded.inc
+--source include/have_sequence.inc
 
 --disable_query_log
 call mtr.add_suppression("InnoDB: Table `mysql`\\.`innodb_table_stats` not found");
@@ -21,6 +22,8 @@ let bugdir= $MYSQLTEST_VARDIR/tmp/row_format_redundant;
 --let $d=$d --innodb-data-file-path=ibdata1:1M:autoextend
 --let $d=$d --innodb-undo-tablespaces=0 --innodb-stats-persistent=0
 --let $restart_parameters= $d
+# Ensure that any DDL records from previous tests have been purged.
+SET GLOBAL innodb_fast_shutdown=0;
 --source include/restart_mysqld.inc
 
 SET GLOBAL innodb_file_per_table=1;
@@ -35,27 +38,21 @@ create table t1 (a int not null, d varchar(15) not null, b
 varchar(198) not null, c char(156)) engine=InnoDB
 row_format=redundant;
 
-insert into t1 values(123, 'abcdef', 'jghikl', 'mnop');
-insert into t1 values(456, 'abcdef', 'jghikl', 'mnop');
-insert into t1 values(789, 'abcdef', 'jghikl', 'mnop');
-insert into t1 values(134, 'kasdfsdsadf', 'adfjlasdkfjasd', 'adfsadflkasdasdfljasdf');
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
-insert into t1 select * from t1;
+create temporary table t like t1;
+
+insert into t values(123, 'abcdef', 'jghikl', 'mnop');
+insert into t values(456, 'abcdef', 'jghikl', 'mnop');
+insert into t values(789, 'abcdef', 'jghikl', 'mnop');
+insert into t values(134, 'kasdfsdsadf', 'adfjlasdkfjasd', 'adfsadflkasdasdfljasdf');
+
+insert into t1 select a,d,b,c from t, seq_1_to_1024;
 
 SET GLOBAL innodb_file_per_table=OFF;
 create table t2 (a int not null, d varchar(15) not null, b
 varchar(198) not null, c char(156), fulltext ftsic(c)) engine=InnoDB
 row_format=redundant;
 
-insert into t2 select * from t1;
+insert into t2 select a,d,b,c from t, seq_1_to_1024;
 
 create table t3 (a int not null, d varchar(15) not null, b varchar(198),
 c varchar(150), index k1(c(99), b(56)), index k2(b(5), c(10))) engine=InnoDB

storage/innobase/btr/btr0btr.cc

@@ -2,7 +2,7 @@
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2014, 2021, MariaDB Corporation.
+Copyright (c) 2014, 2022, 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
@@ -1334,11 +1334,23 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
   else
     ut_ad(cursor->rec == page_get_infimum_rec(block->page.frame));
 
-  if (block->page.id().page_no() == index->page &&
-      fil_page_get_type(old->page.frame) == FIL_PAGE_TYPE_INSTANT)
+  mtr->set_log_mode(log_mode);
+
+  if (block->page.id().page_no() != index->page ||
+      fil_page_get_type(old->page.frame) != FIL_PAGE_TYPE_INSTANT)
+    ut_ad(!memcmp(old->page.frame, block->page.frame, PAGE_HEADER));
+  else if (!index->is_instant())
+  {
+    ut_ad(!memcmp(old->page.frame, block->page.frame, FIL_PAGE_TYPE));
+    ut_ad(!memcmp(old->page.frame + FIL_PAGE_TYPE + 2,
+                  block->page.frame + FIL_PAGE_TYPE + 2,
+                  PAGE_HEADER - FIL_PAGE_TYPE - 2));
+    mtr->write<2,mtr_t::FORCED>(*block, FIL_PAGE_TYPE + block->page.frame,
+                                FIL_PAGE_INDEX);
+  }
+  else
   {
     /* Preserve the PAGE_INSTANT information. */
-    ut_ad(index->is_instant());
     memcpy_aligned<2>(FIL_PAGE_TYPE + block->page.frame,
                       FIL_PAGE_TYPE + old->page.frame, 2);
     memcpy_aligned<2>(PAGE_HEADER + PAGE_INSTANT + block->page.frame,
@@ -1358,9 +1370,10 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
       memcpy(PAGE_OLD_SUPREMUM + block->page.frame,
              PAGE_OLD_SUPREMUM + old->page.frame, 8);
     }
+
+    ut_ad(!memcmp(old->page.frame, block->page.frame, PAGE_HEADER));
   }
 
-  ut_ad(!memcmp(old->page.frame, block->page.frame, PAGE_HEADER));
   ut_ad(!memcmp(old->page.frame + PAGE_MAX_TRX_ID + PAGE_HEADER,
                 block->page.frame + PAGE_MAX_TRX_ID + PAGE_HEADER,
                 PAGE_DATA - (PAGE_MAX_TRX_ID + PAGE_HEADER)));
@@ -1369,7 +1382,6 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
     lock_move_reorganize_page(block, old);
 
   /* Write log for the changes, if needed. */
-  mtr->set_log_mode(log_mode);
   if (log_mode == MTR_LOG_ALL)
   {
     /* Check and log the changes in the page header. */

storage/innobase/fil/fil0fil.cc

@@ -2136,7 +2136,7 @@ a remote tablespace is found it will be changed to true.
 If the fix_dict boolean is set, then it is safe to use an internal SQL
 statement to update the dictionary tables if they are incorrect.
 
-@param[in]	validate	true if we should validate the tablespace
+@param[in]	validate	0=maybe missing, 1=do not validate, 2=validate
 @param[in]	purpose		FIL_TYPE_TABLESPACE or FIL_TYPE_TEMPORARY
 @param[in]	id		tablespace ID
 @param[in]	flags		expected FSP_SPACE_FLAGS
@@ -2148,7 +2148,7 @@ If file-per-table, it is the table name in the databasename/tablename format
 @retval	NULL	if the tablespace could not be opened */
 fil_space_t*
 fil_ibd_open(
-	bool			validate,
+	unsigned		validate,
 	fil_type_t		purpose,
 	ulint			id,
 	ulint			flags,
@@ -2160,7 +2160,7 @@ fil_ibd_open(
 	fil_space_t* space = fil_space_get_by_id(id);
 	mysql_mutex_unlock(&fil_system.mutex);
 	if (space) {
-		if (validate && !srv_read_only_mode) {
+		if (validate > 1 && !srv_read_only_mode) {
 			fsp_flags_try_adjust(space,
 					     flags & ~FSP_FLAGS_MEM_MASK);
 		}
@@ -2197,8 +2197,9 @@ func_exit:
 
 	/* Look for a filepath embedded in an ISL where the default file
 	would be. */
-	if (df_remote.open_link_file(name)) {
-		validate = true;
+	bool must_validate = df_remote.open_link_file(name);
+
+	if (must_validate) {
 		if (df_remote.open_read_only(true) == DB_SUCCESS) {
 			ut_ad(df_remote.is_open());
 			++tablespaces_found;
@@ -2211,15 +2212,12 @@ func_exit:
 				    << df_remote.filepath()
 				    << "' could not be opened read-only.";
 		}
-	}
-
-	/* Attempt to open the tablespace at the dictionary filepath. */
-	if (path_in) {
-		if (!df_default.same_filepath_as(path_in)) {
-			/* Dict path is not the default path. Always validate
-			remote files. If default is opened, it was moved. */
-			validate = true;
-		}
+	} else if (path_in && !df_default.same_filepath_as(path_in)) {
+		/* Dict path is not the default path. Always validate
+		remote files. If default is opened, it was moved. */
+		must_validate = true;
+	} else if (validate > 1) {
+		must_validate = true;
 	}
 
 	/* Always look for a file at the default location. But don't log
@@ -2231,7 +2229,7 @@ func_exit:
 	the first server startup. The tables ought to be dropped by
 	drop_garbage_tables_after_restore() a little later. */
 
-	const bool strict = !tablespaces_found
+	const bool strict = validate && !tablespaces_found
 		&& !(srv_operation == SRV_OPERATION_NORMAL
 		     && srv_start_after_restore
 		     && srv_force_recovery < SRV_FORCE_NO_BACKGROUND
@@ -2257,7 +2255,7 @@ func_exit:
 	normal, we only found 1. */
 	/* For encrypted tablespace, we need to check the
 	encryption in header of first page. */
-	if (!validate && tablespaces_found == 1) {
+	if (!must_validate && tablespaces_found == 1) {
 		goto skip_validate;
 	}
 
@@ -2273,7 +2271,8 @@ func_exit:
 	First, bail out if no tablespace files were found. */
 	if (valid_tablespaces_found == 0) {
 		if (!strict
-		    && IF_WIN(GetLastError() == ERROR_FILE_NOT_FOUND,
+		    && IF_WIN(GetLastError() == ERROR_FILE_NOT_FOUND
+			      || GetLastError() == ERROR_PATH_NOT_FOUND,
 			      errno == ENOENT)) {
 			/* Suppress a message about a missing file. */
 			goto corrupted;
@@ -2286,7 +2285,7 @@ func_exit:
 				TROUBLESHOOT_DATADICT_MSG);
 		goto corrupted;
 	}
-	if (!validate) {
+	if (!must_validate) {
 		goto skip_validate;
 	}
 
@@ -2369,7 +2368,7 @@ skip_validate:
 		df_remote.is_open() ? df_remote.filepath() :
 		df_default.filepath(), OS_FILE_CLOSED, 0, false, true);
 
-	if (validate && !srv_read_only_mode) {
+	if (must_validate && !srv_read_only_mode) {
 		df_remote.close();
 		df_default.close();
 		if (space->acquire()) {

storage/innobase/handler/ha_innodb.cc

@@ -12793,7 +12793,8 @@ int create_table_info_t::create_table(bool create_fk)
 	if (err == DB_SUCCESS) {
 		/* Check that also referencing constraints are ok */
 		dict_names_t	fk_tables;
-		err = dict_load_foreigns(m_table_name, NULL, false, true,
+		err = dict_load_foreigns(m_table_name, nullptr,
+					 m_trx->id, true,
 					 DICT_ERR_IGNORE_NONE, fk_tables);
 		while (err == DB_SUCCESS && !fk_tables.empty()) {
 			dict_sys.load_table(
@@ -13245,9 +13246,7 @@ ha_innobase::create(
 	}
 
 	if (error) {
-		/* Drop the being-created table before rollback,
-		so that rollback can possibly rename back a table
-		that could have been renamed before the failed creation. */
+		/* Rollback will drop the being-created table. */
 		trx_rollback_for_mysql(trx);
 		row_mysql_unlock_data_dictionary(trx);
 	} else {

storage/innobase/handler/handler0alter.cc

@@ -9730,7 +9730,7 @@ innobase_update_foreign_cache(
 	dict_names_t	fk_tables;
 
 	err = dict_load_foreigns(user_table->name.m_name,
-				 ctx->col_names, false, true,
+				 ctx->col_names, 1, true,
 				 DICT_ERR_IGNORE_NONE,
 				 fk_tables);
 
@@ -9741,7 +9741,7 @@ innobase_update_foreign_cache(
 		loaded with "foreign_key checks" off,
 		so let's retry the loading with charset_check is off */
 		err = dict_load_foreigns(user_table->name.m_name,
-					 ctx->col_names, false, false,
+					 ctx->col_names, 1, false,
 					 DICT_ERR_IGNORE_NONE,
 					 fk_tables);
 

storage/innobase/handler/i_s.cc

@@ -4808,12 +4808,13 @@ i_s_dict_fill_sys_tables(
 
 /** Convert one SYS_TABLES record to dict_table_t.
 @param pcur      persistent cursor position on SYS_TABLES record
+@param mtr       mini-transaction (nullptr=use the dict_sys cache)
 @param rec       record to read from (nullptr=use the dict_sys cache)
 @param table     the converted dict_table_t
 @return error message
 @retval nullptr on success */
-static const char *i_s_sys_tables_rec(const btr_pcur_t &pcur, const rec_t *rec,
-                                      dict_table_t **table)
+static const char *i_s_sys_tables_rec(const btr_pcur_t &pcur, mtr_t *mtr,
+                                      const rec_t *rec, dict_table_t **table)
 {
   static_assert(DICT_FLD__SYS_TABLES__NAME == 0, "compatibility");
   size_t len;
@@ -4831,12 +4832,11 @@ static const char *i_s_sys_tables_rec(const btr_pcur_t &pcur, const rec_t *rec,
       return "corrupted SYS_TABLES.NAME";
   }
 
-  const span<const char>name{reinterpret_cast<const char*>(pcur.old_rec), len};
-
   if (rec)
-    return dict_load_table_low(name, rec, table);
+    return dict_load_table_low(mtr, rec, table);
 
-  *table= dict_sys.load_table(name);
+  *table= dict_sys.load_table
+    (span<const char>{reinterpret_cast<const char*>(pcur.old_rec), len});
   return *table ? nullptr : "Table not found in cache";
 }
 
@@ -4878,7 +4878,7 @@ i_s_sys_tables_fill_table(
 
 		/* Create and populate a dict_table_t structure with
 		information from SYS_TABLES row */
-		err_msg = i_s_sys_tables_rec(pcur, rec, &table_rec);
+		err_msg = i_s_sys_tables_rec(pcur, &mtr, rec, &table_rec);
 		mtr.commit();
 		dict_sys.unlock();
 
@@ -5116,7 +5116,8 @@ i_s_sys_tables_fill_table_stats(
 		mtr.commit();
 		/* Fetch the dict_table_t structure corresponding to
 		this SYS_TABLES record */
-		err_msg = i_s_sys_tables_rec(pcur, nullptr, &table_rec);
+		err_msg = i_s_sys_tables_rec(pcur, nullptr, nullptr,
+                                             &table_rec);
 
 		if (UNIV_LIKELY(!err_msg)) {
 			bool evictable = dict_sys.prevent_eviction(table_rec);

storage/innobase/include/dict0load.h

@@ -89,7 +89,8 @@ dict_load_foreigns(
 	const char*		table_name,	/*!< in: table name */
 	const char**		col_names,	/*!< in: column names, or NULL
 						to use table->col_names */
-	bool			check_recursive,/*!< in: Whether to check
+	trx_id_t		trx_id,		/*!< in: DDL transaction id,
+						or 0 to check
 						recursive load of tables
 						chained by FK */
 	bool			check_charsets,	/*!< in: whether to check
@@ -123,12 +124,12 @@ dict_getnext_system(
 
 /** Load a table definition from a SYS_TABLES record to dict_table_t.
 Do not load any columns or indexes.
-@param[in]	name		Table name
+@param[in,out]	mtr		mini-transaction
 @param[in]	rec		SYS_TABLES record
 @param[out,own]	table		table, or nullptr
 @return	error message
 @retval	nullptr on success */
-const char *dict_load_table_low(const span<const char> &name,
+const char *dict_load_table_low(mtr_t *mtr,
                                 const rec_t *rec, dict_table_t **table)
   MY_ATTRIBUTE((nonnull, warn_unused_result));
 

storage/innobase/include/fil0fil.h

@@ -1740,10 +1740,7 @@ file inode probably is much faster (the OS caches them) than accessing
 the first page of the file.  This boolean may be initially false, but if
 a remote tablespace is found it will be changed to true.
 
-If the fix_dict boolean is set, then it is safe to use an internal SQL
-statement to update the dictionary tables if they are incorrect.
-
-@param[in]	validate	true if we should validate the tablespace
+@param[in]	validate	0=maybe missing, 1=do not validate, 2=validate
 @param[in]	purpose		FIL_TYPE_TABLESPACE or FIL_TYPE_TEMPORARY
 @param[in]	id		tablespace ID
 @param[in]	flags		expected FSP_SPACE_FLAGS
@@ -1755,7 +1752,7 @@ If file-per-table, it is the table name in the databasename/tablename format
 @retval	NULL	if the tablespace could not be opened */
 fil_space_t*
 fil_ibd_open(
-	bool			validate,
+	unsigned		validate,
 	fil_type_t		purpose,
 	ulint			id,
 	ulint			flags,

storage/innobase/include/rem0rec.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -727,11 +727,9 @@ in the clustered index for instant ADD COLUMN or ALTER TABLE.
 @param[in]	rec	leaf page record
 @param[in]	index	index of the record
 @return	whether the record is the metadata pseudo-record */
-inline bool rec_is_metadata(const rec_t* rec, const dict_index_t& index)
+inline bool rec_is_metadata(const rec_t *rec, const dict_index_t &index)
 {
-	bool is = rec_is_metadata(rec, dict_table_is_comp(index.table));
-	ut_ad(!is || index.is_instant());
-	return is;
+  return rec_is_metadata(rec, index.table->not_redundant());
 }
 
 /** Determine if the record is the metadata pseudo-record

storage/innobase/rem/rem0rec.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -478,7 +478,7 @@ rec_offs_make_valid(
 {
 	const bool is_alter_metadata = leaf
 		&& rec_is_alter_metadata(rec, *index);
-	ut_ad(is_alter_metadata
+	ut_ad((leaf && rec_is_metadata(rec, *index))
 	      || index->is_dummy || index->is_ibuf()
 	      || (leaf
 		  ? rec_offs_n_fields(offsets)
@@ -572,7 +572,8 @@ rec_offs_validate(
 		}
 		/* index->n_def == 0 for dummy indexes if !comp */
 		ut_ad(!comp || index->n_def);
-		ut_ad(!index->n_def || i <= max_n_fields);
+		ut_ad(!index->n_def || i <= max_n_fields
+		      || rec_is_metadata(rec, *index));
 	}
 	while (i--) {
 		ulint curr = get_value(rec_offs_base(offsets)[1 + i]);
@@ -897,9 +898,7 @@ rec_get_offsets_func(
 		ut_ad(!is_user_rec || !n_core || index->is_dummy
 		      || dict_index_is_ibuf(index)
 		      || n == n_fields /* btr_pcur_restore_position() */
-		      || (n + (index->id == DICT_INDEXES_ID)
-			  >= n_core && n <= index->n_fields
-			  + unsigned(rec_is_alter_metadata(rec, false))));
+		      || (n + (index->id == DICT_INDEXES_ID) >= n_core));
 
 		if (is_user_rec && n_core && n < index->n_fields) {
 			ut_ad(!index->is_dummy);

storage/innobase/row/row0import.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2012, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2021, MariaDB Corporation.
+Copyright (c) 2015, 2022, 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
@@ -4517,7 +4517,7 @@ row_import_for_mysql(
 	ulint	fsp_flags = dict_tf_to_fsp_flags(table->flags);
 
 	table->space = fil_ibd_open(
-		true, FIL_TYPE_IMPORT, table->space_id,
+		2, FIL_TYPE_IMPORT, table->space_id,
 		fsp_flags, name, filepath, &err);
 
 	ut_ad((table->space == NULL) == (err != DB_SUCCESS));

storage/innobase/row/row0mysql.cc

@@ -2912,7 +2912,7 @@ row_rename_table_for_mysql(
 		dict_names_t	fk_tables;
 
 		err = dict_load_foreigns(
-			new_name, NULL, false,
+			new_name, nullptr, trx->id,
 			!old_is_tmp || trx->check_foreigns,
 			use_fk
 			? DICT_ERR_IGNORE_NONE

storage/innobase/row/row0row.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, 2021, MariaDB Corporation.
+Copyright (c) 2018, 2022, 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
@@ -531,7 +531,11 @@ row_build_low(
 			continue;
 		}
 
-		ut_ad(ind_field < &index->fields[index->n_fields]);
+		if (UNIV_UNLIKELY(ind_field
+				  >= &index->fields[index->n_fields])) {
+			ut_ad(rec_is_metadata(rec, *index));
+			continue;
+		}
 
 		const dict_col_t* col = dict_field_get_col(ind_field);
 
@@ -745,11 +749,15 @@ row_rec_to_index_entry_impl(
 	if (mblob == 2) {
 		ut_ad(info_bits == REC_INFO_METADATA_ALTER
 		      || info_bits == REC_INFO_METADATA_ADD);
-		ut_ad(rec_len <= ulint(index->n_fields + got));
 		if (pad) {
+			ut_ad(rec_len <= ulint(index->n_fields + got));
 			rec_len = ulint(index->n_fields)
 				+ (info_bits == REC_INFO_METADATA_ALTER);
-		} else if (!got && info_bits == REC_INFO_METADATA_ALTER) {
+		} else if (got) {
+			rec_len = std::min(rec_len,
+					   ulint(index->n_fields + got));
+		} else if (info_bits == REC_INFO_METADATA_ALTER) {
+			ut_ad(rec_len <= index->n_fields);
 			rec_len++;
 		}
 	} else {