MDEV-18543 IMPORT TABLESPACE fails after instant DROP COLUMNbb-10.5-MDEV-18543-instant-import-bugs

ALTER TABLE IMPORT doesn't properly handle instant alter metadata.
This patch makes IMPORT read, parse and apply instant alter metadata at the
very beginning of operation. So, cases when source table has some metadata
and destination table doesn't have it now works fine.

DISCARD already removes instant metadata so importing normal table into
instant table worked fine before this patch.

decrypt_decompress(): decrypts and decompresses page if needed

handle_instant_metadata(): this should be the first thing to read source
table. Basically, it applies instant metadata to a destination
dict_table_t object. This is the first thing to read FSP flags so
all possible checks of it were moved to this function.

PageConverter::update_index_page(): it doesn't now read instant metadata.
This logic were moved into handle_instant_metadata()

row_import::match_flags(): this is a first part row_import::match_schema().
As a separate function it's used by handle_instant_metadata().

ANONYMOUS_VAR: this is a handy thing to use along with make_scope_exit()

full_crc32_import.test shows different results, because no
dict_table_close() and dict_table_open_on_id() happens.
Thus, SHOW CREATE TABLE shows a little bit older table definition.

16 files changed, 757 insertions, 172 deletions

diff --git a/include/scope.h b/include/scope.h
index b9e2e96ae5c..e0e9fc62969 100644
--- a/include/scope.h
+++ b/include/scope.h
@@ -64,3 +64,11 @@ make_scope_exit(Callable &&f)
   return detail::scope_exit<typename std::decay<Callable>::type>(
       std::forward<Callable>(f));
 }
+
+#define CONCAT_IMPL(x, y) x##y
+
+#define CONCAT(x, y) CONCAT_IMPL(x, y)
+
+#define ANONYMOUS_VARIABLE CONCAT(_anonymous_variable, __LINE__)
+
+#define SCOPE_EXIT auto ANONYMOUS_VARIABLE= make_scope_exit
diff --git a/mysql-test/suite/encryption/r/instant_alter_import.result b/mysql-test/suite/encryption/r/instant_alter_import.result
new file mode 100644
index 00000000000..cb6e7ea7307
--- /dev/null
+++ b/mysql-test/suite/encryption/r/instant_alter_import.result
@@ -0,0 +1,24 @@
+SET GLOBAL innodb_encrypt_tables = ON;
+SET GLOBAL innodb_encryption_threads = 4;
+CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, i2 INT, i1 INT)
+ENGINE=INNODB ENCRYPTED=YES ENCRYPTION_KEY_ID=4;
+INSERT INTO t1 (i2) SELECT 4 FROM seq_1_to_1024;
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t2, t1;
+CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, i2 INT, i1 INT)
+ENGINE=INNODB ENCRYPTED=YES ENCRYPTION_KEY_ID=4 PAGE_COMPRESSED=1;
+INSERT INTO t1 (i2) SELECT 4 FROM seq_1_to_1024;
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t2, t1;
+SET GLOBAL innodb_encrypt_tables = OFF;
+SET GLOBAL innodb_encryption_threads = 0;
diff --git a/mysql-test/suite/encryption/t/instant_alter_import.test b/mysql-test/suite/encryption/t/instant_alter_import.test
new file mode 100644
index 00000000000..3a00c54665c
--- /dev/null
+++ b/mysql-test/suite/encryption/t/instant_alter_import.test
@@ -0,0 +1,52 @@
+--source include/have_innodb.inc
+--source include/have_sequence.inc
+--source include/have_file_key_management_plugin.inc
+
+SET GLOBAL innodb_encrypt_tables = ON;
+SET GLOBAL innodb_encryption_threads = 4;
+
+
+--let $MYSQLD_DATADIR= `SELECT @@datadir`
+
+CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, i2 INT, i1 INT)
+  ENGINE=INNODB ENCRYPTED=YES ENCRYPTION_KEY_ID=4;
+
+INSERT INTO t1 (i2) SELECT 4 FROM seq_1_to_1024;
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+
+FLUSH TABLE t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+DROP TABLE t2, t1;
+
+
+CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, i2 INT, i1 INT)
+  ENGINE=INNODB ENCRYPTED=YES ENCRYPTION_KEY_ID=4 PAGE_COMPRESSED=1;
+
+INSERT INTO t1 (i2) SELECT 4 FROM seq_1_to_1024;
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+
+FLUSH TABLE t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+DROP TABLE t2, t1;
+
+
+SET GLOBAL innodb_encrypt_tables = OFF;
+SET GLOBAL innodb_encryption_threads = 0;
diff --git a/mysql-test/suite/innodb/r/default_row_format_compatibility.result b/mysql-test/suite/innodb/r/default_row_format_compatibility.result
index 5070fa33a2e..f7aa984b279 100644
--- a/mysql-test/suite/innodb/r/default_row_format_compatibility.result
+++ b/mysql-test/suite/innodb/r/default_row_format_compatibility.result
@@ -1,3 +1,4 @@
+call mtr.add_suppression("Index for table 'tab' is corrupt; try to repair it");
 SET @row_format = @@GLOBAL.innodb_default_row_format;
 # ###########################################################
 # Check with Import/Export tablespace with Default_row_format
@@ -38,7 +39,7 @@ tab	InnoDB	#	Compact	#	#	#	#	#	#	NULL	#	NULL	NULL	latin1_swedish_ci	NULL			0	N
 ALTER TABLE tab DISCARD TABLESPACE;
 call mtr.add_suppression("InnoDB: Tried to read .* bytes at offset 0");
 ALTER TABLE tab IMPORT TABLESPACE;
-ERROR HY000: Internal error: Cannot reset LSNs in table `test`.`tab` : I/O error
+ERROR HY000: Index for table 'tab' is corrupt; try to repair it
 ALTER TABLE tab IMPORT TABLESPACE;
 SELECT * FROM tab;
 a
diff --git a/mysql-test/suite/innodb/r/full_crc32_import.result b/mysql-test/suite/innodb/r/full_crc32_import.result
index 6fec6107f2c..677bff22d53 100644
--- a/mysql-test/suite/innodb/r/full_crc32_import.result
+++ b/mysql-test/suite/innodb/r/full_crc32_import.result
@@ -46,7 +46,7 @@ t1	CREATE TABLE `t1` (
   `b` blob DEFAULT NULL,
   `c` blob DEFAULT NULL,
   PRIMARY KEY (`a`)
-) ENGINE=InnoDB AUTO_INCREMENT=46 DEFAULT CHARSET=latin1 ROW_FORMAT=DYNAMIC
+) ENGINE=InnoDB AUTO_INCREMENT=57 DEFAULT CHARSET=latin1 ROW_FORMAT=DYNAMIC
 UPDATE t1 set b = repeat("de", 100) where b = repeat("cd", 200);
 explain SELECT a FROM t1 where b = repeat("de", 100);
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
@@ -127,7 +127,7 @@ t1	CREATE TABLE `t1` (
   `c2` point NOT NULL,
   `c3` linestring NOT NULL,
   PRIMARY KEY (`c1`)
-) ENGINE=InnoDB AUTO_INCREMENT=14325 DEFAULT CHARSET=latin1 ROW_FORMAT=DYNAMIC
+) ENGINE=InnoDB AUTO_INCREMENT=16372 DEFAULT CHARSET=latin1 ROW_FORMAT=DYNAMIC
 UPDATE t1 SET C2 = ST_GeomFromText('POINT(0 0)');
 SELECT COUNT(*) FROM t1;
 COUNT(*)
diff --git a/mysql-test/suite/innodb/r/innodb-wl5522-debug.result b/mysql-test/suite/innodb/r/innodb-wl5522-debug.result
index 1030fbd0502..0b5d9ab55b6 100644
--- a/mysql-test/suite/innodb/r/innodb-wl5522-debug.result
+++ b/mysql-test/suite/innodb/r/innodb-wl5522-debug.result
@@ -935,7 +935,7 @@ ERROR HY000: Tablespace has been discarded for table `t1`
 restore: t1 .ibd and .cfg files
 SET SESSION debug_dbug="+d,fsp_flags_is_valid_failure";
 ALTER TABLE t1 IMPORT TABLESPACE;
-ERROR HY000: Internal error: Cannot reset LSNs in table `test`.`t1` : Data structure corruption
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
 SET SESSION debug_dbug=@saved_debug_dbug;
 DROP TABLE t1;
 unlink: t1.ibd
diff --git a/mysql-test/suite/innodb/r/instant_alter_import.result b/mysql-test/suite/innodb/r/instant_alter_import.result
index 008a26ea6af..c31e9c8e656 100644
--- a/mysql-test/suite/innodb/r/instant_alter_import.result
+++ b/mysql-test/suite/innodb/r/instant_alter_import.result
@@ -61,13 +61,10 @@ alter table t1 discard tablespace;
 flush tables t2 for export;
 unlock tables;
 alter table t1 import tablespace;
-ERROR HY000: Schema mismatch (Index field count 4 doesn't match tablespace metadata file value 5)
 select * from t1;
-ERROR HY000: Tablespace has been discarded for table `t1`
-alter table t1 import tablespace;
-ERROR HY000: Internal error: Cannot reset LSNs in table `test`.`t1` : Unsupported
-select * from t1;
-ERROR HY000: Tablespace has been discarded for table `t1`
+z
+42
+41
 drop table t2;
 drop table t1;
 CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, i1 INT) ENGINE=INNODB;
@@ -78,3 +75,46 @@ FLUSH TABLE t1 FOR EXPORT;
 UNLOCK TABLES;
 ALTER TABLE t2 IMPORT TABLESPACE;
 DROP TABLE t2, t1;
+CREATE TABLE t1 (id INT PRIMARY KEY, i2 INT, i1 INT) ENGINE=INNODB;
+INSERT INTO t1 VALUES (1, 1, 1);
+ALTER TABLE t1 MODIFY COLUMN i2 INT AFTER i1, ALGORITHM=INSTANT;
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SELECT * FROM t2;
+id	i1	i2
+1	1	1
+DROP TABLE t2, t1;
+CREATE TABLE t1 (id INT PRIMARY KEY, i2 INT, i1 INT) ENGINE=INNODB;
+INSERT INTO t1 VALUES (1, 1, 1);
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SELECT * FROM t2;
+id	i1
+1	1
+DROP TABLE t2, t1;
+CREATE TABLE t1 (id INT PRIMARY KEY, i2 INT, i1 INT)
+ENGINE=INNODB PAGE_COMPRESSED=1;
+INSERT INTO t1 VALUES (1, 1, 1);
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t2, t1;
+CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, i2 INT, i1 INT) ENGINE=INNODB;
+INSERT INTO t1 (i2) SELECT 4 FROM seq_1_to_1024;
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t2, t1;
diff --git a/mysql-test/suite/innodb/t/default_row_format_compatibility.test b/mysql-test/suite/innodb/t/default_row_format_compatibility.test
index 3dea78b2ff0..0868974c9df 100644
--- a/mysql-test/suite/innodb/t/default_row_format_compatibility.test
+++ b/mysql-test/suite/innodb/t/default_row_format_compatibility.test
@@ -1,5 +1,8 @@
 --source include/have_innodb.inc
 
+call mtr.add_suppression("Index for table 'tab' is corrupt; try to repair it");
+
+
 SET @row_format = @@GLOBAL.innodb_default_row_format;
 
 # set the variables
@@ -79,7 +82,7 @@ ALTER TABLE tab DISCARD TABLESPACE;
 
 call mtr.add_suppression("InnoDB: Tried to read .* bytes at offset 0");
 
---error ER_INTERNAL_ERROR
+--error ER_NOT_KEYFILE
 ALTER TABLE tab IMPORT TABLESPACE;
 --remove_file $MYSQLD_DATADIR/test/tab.ibd
 --move_file $MYSQLD_DATADIR/tab.ibd $MYSQLD_DATADIR/test/tab.ibd
diff --git a/mysql-test/suite/innodb/t/innodb-wl5522-debug.test b/mysql-test/suite/innodb/t/innodb-wl5522-debug.test
index a22afc041b0..7bc71d87a03 100644
--- a/mysql-test/suite/innodb/t/innodb-wl5522-debug.test
+++ b/mysql-test/suite/innodb/t/innodb-wl5522-debug.test
@@ -1351,7 +1351,7 @@ SET SESSION debug_dbug="+d,fsp_flags_is_valid_failure";
 
 --replace_regex /'.*t1.cfg'/'t1.cfg'/
 
---error ER_INTERNAL_ERROR
+--error ER_NOT_KEYFILE
 ALTER TABLE t1 IMPORT TABLESPACE;
 
 SET SESSION debug_dbug=@saved_debug_dbug;
diff --git a/mysql-test/suite/innodb/t/instant_alter_import.test b/mysql-test/suite/innodb/t/instant_alter_import.test
index d01f4325f1e..4bec3f8b7f5 100644
--- a/mysql-test/suite/innodb/t/instant_alter_import.test
+++ b/mysql-test/suite/innodb/t/instant_alter_import.test
@@ -1,4 +1,7 @@
 --source include/have_innodb.inc
+--source include/have_sequence.inc
+--source include/innodb_checksum_algorithm.inc
+
 set default_storage_engine=innodb;
 
 --echo #
@@ -70,14 +73,7 @@ flush tables t2 for export;
 --copy_file $MYSQLD_DATADIR/test/t2.ibd $MYSQLD_DATADIR/test/t1.ibd
 unlock tables;
 
---error ER_TABLE_SCHEMA_MISMATCH
-alter table t1 import tablespace;
---error ER_TABLESPACE_DISCARDED
-select * from t1;
---remove_file $MYSQLD_DATADIR/test/t1.cfg
---error ER_INTERNAL_ERROR
 alter table t1 import tablespace;
---error ER_TABLESPACE_DISCARDED
 select * from t1;
 
 drop table t2;
@@ -101,3 +97,86 @@ UNLOCK TABLES;
 ALTER TABLE t2 IMPORT TABLESPACE;
 
 DROP TABLE t2, t1;
+
+
+CREATE TABLE t1 (id INT PRIMARY KEY, i2 INT, i1 INT) ENGINE=INNODB;
+
+INSERT INTO t1 VALUES (1, 1, 1);
+ALTER TABLE t1 MODIFY COLUMN i2 INT AFTER i1, ALGORITHM=INSTANT;
+
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+
+
+FLUSH TABLE t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+SELECT * FROM t2;
+
+DROP TABLE t2, t1;
+
+
+CREATE TABLE t1 (id INT PRIMARY KEY, i2 INT, i1 INT) ENGINE=INNODB;
+
+INSERT INTO t1 VALUES (1, 1, 1);
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+
+
+FLUSH TABLE t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+SELECT * FROM t2;
+
+DROP TABLE t2, t1;
+
+
+CREATE TABLE t1 (id INT PRIMARY KEY, i2 INT, i1 INT)
+  ENGINE=INNODB PAGE_COMPRESSED=1;
+
+INSERT INTO t1 VALUES (1, 1, 1);
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+
+FLUSH TABLE t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+DROP TABLE t2, t1;
+
+
+CREATE TABLE t1 (id INT PRIMARY KEY AUTO_INCREMENT, i2 INT, i1 INT) ENGINE=INNODB;
+
+INSERT INTO t1 (i2) SELECT 4 FROM seq_1_to_1024;
+ALTER TABLE t1 DROP COLUMN i2, ALGORITHM=INSTANT;
+
+CREATE TABLE t2 LIKE t1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+
+FLUSH TABLE t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+DROP TABLE t2, t1;
diff --git a/mysql-test/suite/innodb_zip/r/wl5522_debug_zip.result b/mysql-test/suite/innodb_zip/r/wl5522_debug_zip.result
index fccee685caf..3feb51bac98 100644
--- a/mysql-test/suite/innodb_zip/r/wl5522_debug_zip.result
+++ b/mysql-test/suite/innodb_zip/r/wl5522_debug_zip.result
@@ -402,7 +402,7 @@ ERROR HY000: Tablespace has been discarded for table `t1`
 restore: t1 .ibd and .cfg files
 SET SESSION debug_dbug="+d,fsp_flags_is_valid_failure";
 ALTER TABLE t1 IMPORT TABLESPACE;
-ERROR HY000: Internal error: Cannot reset LSNs in table `test`.`t1` : Data structure corruption
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
 SET SESSION debug_dbug=@saved_debug_dbug;
 DROP TABLE t1;
 unlink: t1.ibd
diff --git a/mysql-test/suite/innodb_zip/t/wl5522_debug_zip.test b/mysql-test/suite/innodb_zip/t/wl5522_debug_zip.test
index 282469f0a0c..a7916477e0a 100644
--- a/mysql-test/suite/innodb_zip/t/wl5522_debug_zip.test
+++ b/mysql-test/suite/innodb_zip/t/wl5522_debug_zip.test
@@ -611,7 +611,7 @@ EOF
 
 SET SESSION debug_dbug="+d,fsp_flags_is_valid_failure";
 
---error ER_INTERNAL_ERROR
+--error ER_NOT_KEYFILE
 ALTER TABLE t1 IMPORT TABLESPACE;
 
 SET SESSION debug_dbug=@saved_debug_dbug;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 405c6284df1..2dc01d973fe 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -13163,24 +13163,10 @@ ha_innobase::discard_or_import_tablespace(
 				    err, m_prebuilt->table->flags, NULL));
 	}
 
-	/* Evict and reload the table definition in order to invoke
-	btr_cur_instant_init(). */
-	table_id_t id = m_prebuilt->table->id;
-	ut_ad(id);
-	mutex_enter(&dict_sys.mutex);
-	dict_table_close(m_prebuilt->table, TRUE, FALSE);
-	dict_sys.remove(m_prebuilt->table);
-	m_prebuilt->table = dict_table_open_on_id(id, TRUE,
-						  DICT_TABLE_OP_NORMAL);
-	mutex_exit(&dict_sys.mutex);
-	if (!m_prebuilt->table) {
-		err = DB_TABLE_NOT_FOUND;
-	} else {
-		if (const Field* ai = table->found_next_number_field) {
-			initialize_auto_increment(m_prebuilt->table, ai);
-		}
-		dict_stats_init(m_prebuilt->table);
+	if (const Field* ai = table->found_next_number_field) {
+		initialize_auto_increment(m_prebuilt->table, ai);
 	}
+	dict_stats_init(m_prebuilt->table);
 
 	if (dict_stats_is_persistent_enabled(m_prebuilt->table)) {
 		dberr_t		ret;
diff --git a/storage/innobase/include/row0row.h b/storage/innobase/include/row0row.h
index b4dab3c2f1b..d813737d626 100644
--- a/storage/innobase/include/row0row.h
+++ b/storage/innobase/include/row0row.h
@@ -30,6 +30,7 @@ Created 4/20/1996 Heikki Tuuri
 #include "que0types.h"
 #include "ibuf0ibuf.h"
 #include "trx0types.h"
+#include "trx0sys.h"
 #include "mtr0mtr.h"
 #include "rem0types.h"
 #include "row0types.h"
@@ -427,6 +428,32 @@ row_mtr_start(mtr_t* mtr, dict_index_t* index, bool pessimistic)
 	log_free_check();
 }
 
+/*********************************************************************//**
+Reads the trx id field from a clustered index record.
+@return value of the field */
+static inline
+trx_id_t
+row_get_rec_trx_id(
+/*===============*/
+	const rec_t*		rec,	/*!< in: record */
+	const dict_index_t*	index,	/*!< in: clustered index */
+	const rec_offs*		offsets)/*!< in: rec_get_offsets(rec, index) */
+{
+	ulint	offset;
+
+	ut_ad(dict_index_is_clust(index));
+	ut_ad(rec_offs_validate(rec, index, offsets));
+
+	offset = index->trx_id_offset;
+
+	if (!offset) {
+		offset = row_get_trx_id_offset(index, offsets);
+	}
+
+	return(trx_read_trx_id(rec + offset));
+}
+
+
 #include "row0row.ic"
 
 #endif
diff --git a/storage/innobase/include/row0row.ic b/storage/innobase/include/row0row.ic
index e89adb581f4..68ae055fad0 100644
--- a/storage/innobase/include/row0row.ic
+++ b/storage/innobase/include/row0row.ic
@@ -52,31 +52,6 @@ row_get_trx_id_offset(
 }
 
 /*********************************************************************//**
-Reads the trx id field from a clustered index record.
-@return value of the field */
-UNIV_INLINE
-trx_id_t
-row_get_rec_trx_id(
-/*===============*/
-	const rec_t*		rec,	/*!< in: record */
-	const dict_index_t*	index,	/*!< in: clustered index */
-	const rec_offs*		offsets)/*!< in: rec_get_offsets(rec, index) */
-{
-	ulint	offset;
-
-	ut_ad(dict_index_is_clust(index));
-	ut_ad(rec_offs_validate(rec, index, offsets));
-
-	offset = index->trx_id_offset;
-
-	if (!offset) {
-		offset = row_get_trx_id_offset(index, offsets);
-	}
-
-	return(trx_read_trx_id(rec + offset));
-}
-
-/*********************************************************************//**
 Reads the roll pointer field from a clustered index record.
 @return value of the field */
 UNIV_INLINE
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 7fe5cc1802c..a431bcac32f 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -40,6 +40,7 @@ Created 2012-02-08 by Sunny Bains.
 #include "row0quiesce.h"
 #include "fil0pagecompress.h"
 #include "trx0undo.h"
+#include "row0row.h"
 #ifdef HAVE_LZO
 #include "lzo/lzo1x.h"
 #endif
@@ -47,6 +48,8 @@ Created 2012-02-08 by Sunny Bains.
 #include "snappy-c.h"
 #endif
 
+#include "scope.h"
+
 #include <vector>
 
 #ifdef HAVE_MY_AES_H
@@ -190,6 +193,9 @@ struct row_import {
 	dberr_t match_schema(
 		THD*			thd) UNIV_NOTHROW;
 
+	dberr_t match_flags(THD *thd) const ;
+
+
 	dict_table_t*	m_table;		/*!< Table instance */
 
 	ulint		m_version;		/*!< Version of config file */
@@ -605,8 +611,6 @@ AbstractCallback::init(
 	if (!fil_space_t::is_valid_flags(m_space_flags, true)) {
 		ulint cflags = fsp_flags_convert_from_101(m_space_flags);
 		if (cflags == ULINT_UNDEFINED) {
-			ib::error() << "Invalid FSP_SPACE_FLAGS="
-				<< ib::hex(m_space_flags);
 			return(DB_CORRUPTION);
 		}
 		m_space_flags = cflags;
@@ -1123,7 +1127,8 @@ row_import::match_index_columns(
 
 	for (ulint i = 0; i < index->n_fields; ++i, ++field, ++cfg_field) {
 
-		if (strcmp(field->name(), cfg_field->name()) != 0) {
+		if (field->name() && cfg_field->name()
+		     && strcmp(field->name(), cfg_field->name()) != 0) {
 			ib_errf(thd, IB_LOG_LEVEL_ERROR,
 				ER_TABLE_SCHEMA_MISMATCH,
 				"Index field name %s doesn't match"
@@ -1292,6 +1297,66 @@ row_import::match_table_columns(
 	return(err);
 }
 
+dberr_t row_import::match_flags(THD *thd) const
+{
+  ulint mismatch= (m_table->flags ^ m_flags) & ~DICT_TF_MASK_DATA_DIR;
+  if (!mismatch)
+    return DB_SUCCESS;
+
+  const char *msg;
+  if (mismatch & DICT_TF_MASK_ZIP_SSIZE)
+  {
+    if ((m_table->flags & DICT_TF_MASK_ZIP_SSIZE) &&
+        (m_flags & DICT_TF_MASK_ZIP_SSIZE))
+    {
+      switch (m_flags & DICT_TF_MASK_ZIP_SSIZE)
+      {
+      case 0U << DICT_TF_POS_ZIP_SSIZE:
+        goto uncompressed;
+      case 1U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=1";
+        break;
+      case 2U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=2";
+        break;
+      case 3U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=4";
+        break;
+      case 4U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=8";
+        break;
+      case 5U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=16";
+        break;
+      default:
+        msg= "strange KEY_BLOCK_SIZE";
+      }
+    }
+    else if (m_flags & DICT_TF_MASK_ZIP_SSIZE)
+    {
+      msg= "ROW_FORMAT=COMPRESSED";
+    }
+    else
+    {
+      goto uncompressed;
+    }
+  }
+  else
+  {
+  uncompressed:
+    msg= (m_flags & DICT_TF_MASK_ATOMIC_BLOBS) ? "ROW_FORMAT=DYNAMIC"
+         : (m_flags & DICT_TF_MASK_COMPACT)    ? "ROW_FORMAT=COMPACT"
+                                               : "ROW_FORMAT=REDUNDANT";
+  }
+
+  ib_errf(thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
+          "Table flags don't match, server table has 0x%x and the meta-data "
+          "file has 0x%zx; .cfg file uses %s",
+          m_table->flags, m_flags, msg);
+
+  return DB_ERROR;
+}
+
 /** Check if the table (and index) schema that was read from the .cfg file
 matches the in memory table definition.
 @param thd MySQL session variable
@@ -1302,60 +1367,7 @@ row_import::match_schema(
 {
 	/* Do some simple checks. */
 
-	if (ulint mismatch = (m_table->flags ^ m_flags)
-	    & ~DICT_TF_MASK_DATA_DIR) {
-		const char* msg;
-		if (mismatch & DICT_TF_MASK_ZIP_SSIZE) {
-			if ((m_table->flags & DICT_TF_MASK_ZIP_SSIZE)
-			    && (m_flags & DICT_TF_MASK_ZIP_SSIZE)) {
-				switch (m_flags & DICT_TF_MASK_ZIP_SSIZE) {
-				case 0U << DICT_TF_POS_ZIP_SSIZE:
-					goto uncompressed;
-				case 1U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=1";
-					break;
-				case 2U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=2";
-					break;
-				case 3U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=4";
-					break;
-				case 4U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=8";
-					break;
-				case 5U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=16";
-					break;
-				default:
-					msg = "strange KEY_BLOCK_SIZE";
-				}
-			} else if (m_flags & DICT_TF_MASK_ZIP_SSIZE) {
-				msg = "ROW_FORMAT=COMPRESSED";
-			} else {
-				goto uncompressed;
-			}
-		} else {
-uncompressed:
-			msg = (m_flags & DICT_TF_MASK_ATOMIC_BLOBS)
-				? "ROW_FORMAT=DYNAMIC"
-				: (m_flags & DICT_TF_MASK_COMPACT)
-				? "ROW_FORMAT=COMPACT"
-				: "ROW_FORMAT=REDUNDANT";
-		}
-
-		ib_errf(thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
-			"Table flags don't match, server table has 0x%x"
-			" and the meta-data file has 0x" ULINTPFx ";"
-			" .cfg file uses %s",
-			m_table->flags, m_flags, msg);
-
-		return(DB_ERROR);
-	} else if (m_table->n_cols != m_n_cols) {
+	if (m_table->n_cols != m_n_cols) {
 		ib_errf(thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
 			"Number of columns don't match, table has %u "
 			"columns but the tablespace meta-data file has "
@@ -1977,42 +1989,7 @@ PageConverter::update_index_page(
 	}
 
 	if (m_index->m_srv_index->is_clust()) {
-		if (page_id.page_no() == m_index->m_srv_index->page) {
-			dict_index_t* index = const_cast<dict_index_t*>(
-				m_index->m_srv_index);
-			/* Preserve the PAGE_ROOT_AUTO_INC. */
-			if (index->table->supports_instant()) {
-				if (btr_cur_instant_root_init(index, page)) {
-					return(DB_CORRUPTION);
-				}
-
-				if (index->n_core_fields > index->n_fields) {
-					/* Some columns have been dropped.
-					Refuse to IMPORT TABLESPACE for now.
-
-					NOTE: This is not an accurate check.
-					Columns could have been both
-					added and dropped instantly.
-					For an accurate check, we must read
-					the metadata BLOB page pointed to
-					by the leftmost leaf page.
-
-					But we would have to read
-					those pages in a special way,
-					bypassing the buffer pool! */
-					return DB_UNSUPPORTED;
-				}
-
-				/* Provisionally set all instantly
-				added columns to be DEFAULT NULL. */
-				for (unsigned i = index->n_core_fields;
-				     i < index->n_fields; i++) {
-					dict_col_t* col = index->fields[i].col;
-					col->def_val.len = UNIV_SQL_NULL;
-					col->def_val.data = NULL;
-				}
-			}
-		} else {
+		if (page_id.page_no() != m_index->m_srv_index->page) {
 			goto clear_page_max_trx_id;
 		}
 	} else if (page_is_leaf(page)) {
@@ -3120,6 +3097,419 @@ row_import_read_meta_data(
 	return(DB_ERROR);
 }
 
+#define BTR_BLOB_HDR_PART_LEN 0 /*!< BLOB part len on this page */
+#define BTR_BLOB_HDR_NEXT_PAGE_NO 4 /*!< next BLOB part page no,
+                                    FIL_NULL if none */
+#define BTR_BLOB_HDR_SIZE 8 /*!< Size of a BLOB part header, in bytes */
+
+/* decrypt and decompress page if needed */
+static dberr_t decrypt_decompress(fil_space_crypt_t *space_crypt,
+                                  size_t space_flags, span<byte> page,
+                                  size_t space_id, byte *page_compress_buf)
+{
+  auto *data= page.data();
+
+  if (space_crypt && space_crypt->should_encrypt())
+  {
+    if (!buf_page_verify_crypt_checksum(data, space_flags))
+      return DB_CORRUPTION;
+
+    dberr_t err;
+    if (!fil_space_decrypt(space_id, space_crypt, data, page.size(),
+                           space_flags, data, &err) ||
+        err != DB_SUCCESS)
+    {
+      return err;
+    }
+  }
+
+  bool page_compressed= false;
+
+  if (fil_space_t::full_crc32(space_flags) &&
+      fil_space_t::is_compressed(space_flags))
+  {
+    page_compressed= buf_page_is_compressed(data, space_flags);
+  }
+  else
+  {
+    switch (fil_page_get_type(data))
+    {
+    case FIL_PAGE_PAGE_COMPRESSED:
+    case FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED:
+      page_compressed= true;
+    }
+  }
+
+  if (page_compressed)
+  {
+    auto compress_length=
+        fil_page_decompress(page_compress_buf, data, space_flags);
+    ut_ad(compress_length != srv_page_size);
+
+    if (compress_length == 0)
+      return DB_CORRUPTION;
+  }
+
+  return DB_SUCCESS;
+}
+
+static size_t get_buf_size()
+{
+  return srv_page_size
+#ifdef HAVE_LZO
+         + LZO1X_1_15_MEM_COMPRESS
+#elif defined HAVE_SNAPPY
+         + snappy_max_compressed_length(srv_page_size)
+#endif
+      ;
+}
+
+/* find, parse instant metadata, performing variaous checks,
+and apply it to dict_table_t
+@return DB_SUCCESS or some error */
+static dberr_t handle_instant_metadata(dict_table_t *table,
+                                       const row_import &cfg)
+{
+  dict_get_and_save_data_dir_path(table, false);
+
+  char *filepath;
+  if (DICT_TF_HAS_DATA_DIR(table->flags))
+  {
+    ut_a(table->data_dir_path);
+
+    filepath=
+        fil_make_filepath(table->data_dir_path, table->name.m_name, IBD, true);
+  }
+  else
+  {
+    filepath= fil_make_filepath(nullptr, table->name.m_name, IBD, false);
+  }
+
+  if (!filepath)
+    return DB_OUT_OF_MEMORY;
+
+  SCOPE_EXIT([filepath]() { ut_free(filepath); });
+
+  bool success;
+  auto file= os_file_create_simple_no_error_handling(
+      innodb_data_file_key, filepath, OS_FILE_OPEN, OS_FILE_READ_WRITE, false,
+      &success);
+  if (!success)
+    return DB_IO_ERROR;
+
+  if (os_file_get_size(file) < srv_page_size * 4)
+    return DB_CORRUPTION;
+
+  SCOPE_EXIT([&file]() { os_file_close(file); });
+
+  std::unique_ptr<byte[], decltype(&aligned_free)> first_page(
+      static_cast<byte *>(aligned_malloc(srv_page_size, srv_page_size)),
+      &aligned_free);
+
+  if (dberr_t err= os_file_read_no_error_handling(IORequestReadPartial, file,
+                                                  first_page.get(), 0,
+                                                  srv_page_size, nullptr))
+  {
+    return err;
+  }
+
+  auto space_flags= fsp_header_get_flags(first_page.get());
+
+  if (!fil_space_t::is_valid_flags(space_flags, true))
+  {
+    auto cflags= fsp_flags_convert_from_101(space_flags);
+    if (cflags == ULINT_UNDEFINED)
+    {
+      ib::error() << "Invalid FSP_SPACE_FLAGS=" << ib::hex(space_flags);
+      return (DB_CORRUPTION);
+    }
+    space_flags= static_cast<decltype(space_flags)>(cflags);
+  }
+
+  if (!cfg.m_missing)
+  {
+    if (dberr_t err= cfg.match_flags(current_thd))
+      return err;
+  }
+
+  const auto zip_size= fil_space_t::zip_size(space_flags);
+  const auto physical_size= zip_size ? zip_size : srv_page_size;
+  ut_ad(physical_size <= UNIV_PAGE_SIZE_MAX);
+  const auto space_id= page_get_space_id(first_page.get());
+
+  auto *space_crypt= fil_space_read_crypt_data(zip_size, first_page.get());
+  SCOPE_EXIT([&space_crypt]() {
+    if (space_crypt)
+      fil_space_destroy_crypt_data(&space_crypt);
+  });
+
+  std::unique_ptr<byte[], decltype(&aligned_free)> page(
+      static_cast<byte *>(
+          aligned_malloc(UNIV_PAGE_SIZE_MAX, UNIV_PAGE_SIZE_MAX)),
+      &aligned_free);
+
+  if (dberr_t err= os_file_read_no_error_handling(
+          IORequestReadPartial, file, page.get(), 3 * physical_size,
+          physical_size, nullptr))
+  {
+    return err;
+  }
+
+  std::unique_ptr<byte[]> page_compress_buf(new byte[get_buf_size()]);
+
+  if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
+                                      {page.get(), physical_size}, space_id,
+                                      page_compress_buf.get()))
+  {
+    return err;
+  }
+
+  if (table->supports_instant())
+  {
+    dict_index_t *index= dict_table_get_first_index(table);
+
+    auto tmp1= table->space_id;
+    table->space_id= page_get_space_id(page.get());
+    SCOPE_EXIT([tmp1, table]() { table->space_id= tmp1; });
+
+    auto tmp2= index->page;
+    index->page= page_get_page_no(page.get());
+    SCOPE_EXIT([tmp2, index]() { index->page= tmp2; });
+
+    if (!page_is_comp(page.get()) != !dict_table_is_comp(table))
+    {
+      ib_errf(current_thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
+              "ROW_FORMAT mismatch");
+      return DB_CORRUPTION;
+    }
+
+    if (btr_cur_instant_root_init(index, page.get()))
+      return DB_ERROR;
+
+    ut_ad(index->n_core_null_bytes != dict_index_t::NO_CORE_NULL_BYTES);
+
+    if (fil_page_get_type(page.get()) == FIL_PAGE_INDEX)
+    {
+      ut_ad(!index->is_instant());
+      return DB_SUCCESS;
+    }
+
+    mem_heap_t *heap= NULL;
+    SCOPE_EXIT([&heap]() { mem_heap_free(heap); });
+
+    while (btr_page_get_level(page.get()) != 0)
+    {
+      const rec_t *rec= page_rec_get_next(page_get_infimum_rec(page.get()));
+
+      /* Relax the assertion in rec_init_offsets(). */
+      ut_ad(!index->in_instant_init);
+      ut_d(index->in_instant_init= true);
+      rec_offs *offsets=
+          rec_get_offsets(rec, index, nullptr, 0, ULINT_UNDEFINED, &heap);
+      ut_d(index->in_instant_init= false);
+
+      auto child_page_no= btr_node_ptr_get_child_page_no(rec, offsets);
+
+      if (dberr_t err= os_file_read_no_error_handling(
+              IORequestReadPartial, file, page.get(),
+              child_page_no * physical_size, physical_size, nullptr))
+      {
+        return err;
+      }
+
+      if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
+                                          {page.get(), physical_size},
+                                          space_id, page_compress_buf.get()))
+      {
+        return err;
+      }
+    }
+
+    const rec_t *rec= page_rec_get_next(page_get_infimum_rec(page.get()));
+    const ulint comp= dict_table_is_comp(index->table);
+    const ulint info_bits= rec_get_info_bits(rec, comp);
+
+    if (page_rec_is_supremum(rec) || !(info_bits & REC_INFO_MIN_REC_FLAG))
+    {
+      if (!index->is_instant())
+      {
+        /* The FIL_PAGE_TYPE_INSTANT and PAGE_INSTANT may be
+        assigned even if instant ADD COLUMN was not
+        committed. Changes to these page header fields are not
+        undo-logged, but changes to the hidden metadata record
+        are. If the server is killed and restarted, the page
+        header fields could remain set even though no metadata
+        record is present. */
+        return DB_SUCCESS;
+      }
+
+      ib::error() << "Table " << index->table->name
+                  << " is missing instant ALTER metadata";
+      index->table->corrupted= true;
+      return DB_CORRUPTION;
+    }
+
+    if ((info_bits & ~REC_INFO_DELETED_FLAG) != REC_INFO_MIN_REC_FLAG ||
+        (comp && rec_get_status(rec) != REC_STATUS_INSTANT))
+    {
+    incompatible:
+      ib::error() << "Table " << index->table->name
+                  << " contains unrecognizable instant ALTER metadata";
+      index->table->corrupted= true;
+      return DB_CORRUPTION;
+    }
+
+    if (info_bits & REC_INFO_DELETED_FLAG)
+    {
+      ulint trx_id_offset= index->trx_id_offset;
+      ut_ad(index->n_uniq);
+
+      if (trx_id_offset)
+      {
+      }
+      else if (index->table->not_redundant())
+      {
+
+        for (uint i= index->n_uniq; i--;)
+        {
+          trx_id_offset+= index->fields[i].fixed_len;
+        }
+      }
+      else if (rec_get_1byte_offs_flag(rec))
+      {
+        trx_id_offset= rec_1_get_field_end_info(rec, index->n_uniq - 1);
+        ut_ad(!(trx_id_offset & REC_1BYTE_SQL_NULL_MASK));
+        trx_id_offset&= ~REC_1BYTE_SQL_NULL_MASK;
+      }
+      else
+      {
+        trx_id_offset= rec_2_get_field_end_info(rec, index->n_uniq - 1);
+        ut_ad(!(trx_id_offset & REC_2BYTE_SQL_NULL_MASK));
+        trx_id_offset&= ~REC_2BYTE_SQL_NULL_MASK;
+      }
+
+      const byte *ptr=
+          rec + trx_id_offset + (DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
+
+      if (mach_read_from_4(ptr + BTR_EXTERN_LEN))
+      {
+        goto incompatible;
+      }
+
+      uint len= mach_read_from_4(ptr + BTR_EXTERN_LEN + 4);
+      if (!len || mach_read_from_4(ptr + BTR_EXTERN_OFFSET) != FIL_PAGE_DATA)
+      {
+        goto incompatible;
+      }
+
+      std::unique_ptr<byte[], decltype(&aligned_free)> second_page(
+          static_cast<byte *>(
+              aligned_malloc(physical_size, UNIV_PAGE_SIZE_MAX)),
+          &aligned_free);
+
+      if (dberr_t err= os_file_read_no_error_handling(
+              IORequestReadPartial, file, second_page.get(),
+              mach_read_from_4(ptr + BTR_EXTERN_PAGE_NO) * physical_size,
+              srv_page_size, nullptr))
+      {
+        return err;
+      }
+
+      if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
+                                          {second_page.get(), physical_size},
+                                          space_id, page_compress_buf.get()))
+      {
+        return err;
+      }
+
+      if (fil_page_get_type(second_page.get()) != FIL_PAGE_TYPE_BLOB ||
+          mach_read_from_4(
+              &second_page[FIL_PAGE_DATA + BTR_BLOB_HDR_NEXT_PAGE_NO]) !=
+              FIL_NULL ||
+          mach_read_from_4(
+              &second_page[FIL_PAGE_DATA + BTR_BLOB_HDR_PART_LEN]) != len)
+      {
+        goto incompatible;
+      }
+
+      /* The unused part of the BLOB page should be zero-filled. */
+      for (const byte *
+               b= second_page.get() + (FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE) +
+                  len,
+              *const end= second_page.get() + srv_page_size - BTR_EXTERN_LEN;
+           b < end;)
+      {
+        if (*b++)
+          goto incompatible;
+      }
+
+      if (index->table->deserialise_columns(
+              &second_page[FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE], len))
+      {
+        goto incompatible;
+      }
+    }
+
+    rec_offs *offsets= rec_get_offsets(
+        rec, index, nullptr, index->n_core_fields, ULINT_UNDEFINED, &heap);
+    if (rec_offs_any_default(offsets))
+    {
+    inconsistent:
+      goto incompatible;
+    }
+
+    /* In fact, because we only ever append fields to the metadata
+    record, it is also OK to perform READ UNCOMMITTED and
+    then ignore any extra fields, provided that
+    trx_sys.is_registered(DB_TRX_ID). */
+    if (rec_offs_n_fields(offsets) >
+            ulint(index->n_fields) + !!index->table->instant &&
+        !trx_sys.is_registered(current_trx(),
+                               row_get_rec_trx_id(rec, index, offsets)))
+    {
+      goto inconsistent;
+    }
+
+    for (unsigned i= index->n_core_fields; i < index->n_fields; i++)
+    {
+      dict_col_t *col= index->fields[i].col;
+      const unsigned o= i + !!index->table->instant;
+      ulint len;
+      const byte *data= rec_get_nth_field(rec, offsets, o, &len);
+      ut_ad(!col->is_added());
+      ut_ad(!col->def_val.data);
+      col->def_val.len= len;
+      switch (len)
+      {
+      case UNIV_SQL_NULL:
+        continue;
+      case 0:
+        col->def_val.data= field_ref_zero;
+        continue;
+      }
+      ut_ad(len != UNIV_SQL_DEFAULT);
+      if (!rec_offs_nth_extern(offsets, o))
+      {
+        col->def_val.data= mem_heap_dup(index->table->heap, data, len);
+      }
+      else if (len < BTR_EXTERN_FIELD_REF_SIZE ||
+               !memcmp(data + len - BTR_EXTERN_FIELD_REF_SIZE, field_ref_zero,
+                       BTR_EXTERN_FIELD_REF_SIZE))
+      {
+        col->def_val.len= UNIV_SQL_DEFAULT;
+        goto inconsistent;
+      }
+      else
+      {
+        col->def_val.data= btr_copy_externally_stored_field(
+            &col->def_val.len, data, srv_page_size, len, index->table->heap);
+      }
+    }
+  }
+
+  return DB_SUCCESS;
+}
+
 /**
 Read the contents of the <tablename>.cfg file.
 @return DB_SUCCESS or error code. */
@@ -3428,14 +3818,7 @@ dberr_t FetchIndexRootPages::run(const fil_iterator_t& iter,
 {
   const unsigned zip_size= fil_space_t::zip_size(m_space_flags);
   const unsigned size= zip_size ? zip_size : unsigned(srv_page_size);
-  const ulint buf_size=
-#ifdef HAVE_LZO
-    LZO1X_1_15_MEM_COMPRESS+
-#elif defined HAVE_SNAPPY
-    snappy_max_compressed_length(srv_page_size) +
-#endif
-    srv_page_size;
-  byte* page_compress_buf = static_cast<byte*>(malloc(buf_size));
+  byte* page_compress_buf= static_cast<byte*>(malloc(get_buf_size()));
   const bool full_crc32 = fil_space_t::full_crc32(m_space_flags);
   bool skip_checksum_check = false;
   ut_ad(!srv_read_only_mode);
@@ -3527,14 +3910,7 @@ static dberr_t fil_iterate(
 	const ulint		size = callback.physical_size();
 	ulint			n_bytes = iter.n_io_buffers * size;
 
-	const ulint buf_size = srv_page_size
-#ifdef HAVE_LZO
-		+ LZO1X_1_15_MEM_COMPRESS
-#elif defined HAVE_SNAPPY
-		+ snappy_max_compressed_length(srv_page_size)
-#endif
-		;
-	byte* page_compress_buf = static_cast<byte*>(malloc(buf_size));
+	byte* page_compress_buf= static_cast<byte*>(malloc(get_buf_size()));
 	ut_ad(!srv_read_only_mode);
 
 	if (!page_compress_buf) {
@@ -4094,6 +4470,11 @@ row_import_for_mysql(
 
 	if (err == DB_SUCCESS) {
 
+		if (dberr_t err = handle_instant_metadata(table, cfg)) {
+			rw_lock_s_unlock(&dict_sys.latch);
+			return row_import_error(prebuilt, trx, err);
+		}
+
 		/* We have a schema file, try and match it with our
 		data dictionary. */
 
@@ -4151,6 +4532,15 @@ row_import_for_mysql(
 
 			if (err == DB_SUCCESS) {
 				err = cfg.set_root_by_heuristic();
+
+				if (err == DB_SUCCESS) {
+					if (dberr_t err =
+					    handle_instant_metadata(table,
+								    cfg)) {
+						return row_import_error(
+							prebuilt, trx, err);
+					}
+				}
 			}
 		}
 	} else {