MDEV-26137 Improve import tablespace workflow.

Allow ALTER TABLE ... IMPORT TABLESPACE without creating the table
followed by discarding the tablespace.

That is, assuming we want to import table t1 to t2, instead of

CREATE TABLE t2 LIKE t1;
ALTER TABLE t2 DISCARD TABLESPACE;
FLUSH TABLES t1 FOR EXPORT;
--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
UNLOCK TABLES;
ALTER TABLE t2 IMPORT TABLESPACE;

We can simply do

FLUSH TABLES t1 FOR EXPORT;
--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
UNLOCK TABLES;
ALTER TABLE t2 IMPORT TABLESPACE;

We achieve this by creating a "stub" table in the second scenario
while opening the table, where t2 does not exist but needs to import
from t1. The "stub" table is similar to a table that is created but
then instructed to discard its tablespace.

We include tests with various row formats, encryption, with indexes
and auto-increment.

Signed-off-by: Yuchen Pei <yuchen.pei@mariadb.com>

22 files changed, 1194 insertions, 298 deletions

diff --git a/mysql-test/suite/encryption/r/innodb_import.result b/mysql-test/suite/encryption/r/innodb_import.result
index 54b95ab26d4..b1895b65ef7 100644
--- a/mysql-test/suite/encryption/r/innodb_import.result
+++ b/mysql-test/suite/encryption/r/innodb_import.result
@@ -20,3 +20,22 @@ SELECT * FROM t2;
 f1	f2
 1	InnoDB
 DROP TABLE t1, t2;
+#
+# MDEV-26137 ALTER TABLE IMPORT enhancement
+#
+# with encryption and page_compressed
+CREATE TABLE t1 (a int, b varchar(50)) ENGINE=InnoDB ENCRYPTED=YES ENCRYPTION_KEY_ID=6 PAGE_COMPRESSED=1;
+INSERT INTO t1 VALUES(42, "hello");
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL,
+  `b` varchar(50) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci `ENCRYPTED`=YES `ENCRYPTION_KEY_ID`=6 `PAGE_COMPRESSED`=1
+SELECT * FROM t2;
+a	b
+42	hello
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/encryption/t/innodb_import.test b/mysql-test/suite/encryption/t/innodb_import.test
index 2e5470c5568..91bae752530 100644
--- a/mysql-test/suite/encryption/t/innodb_import.test
+++ b/mysql-test/suite/encryption/t/innodb_import.test
@@ -21,3 +21,22 @@ ALTER TABLE t2 DROP KEY idx;
 ALTER TABLE t2 IMPORT TABLESPACE;
 SELECT * FROM t2;
 DROP TABLE t1, t2;
+
+--echo #
+--echo # MDEV-26137 ALTER TABLE IMPORT enhancement
+--echo #
+
+--echo # with encryption and page_compressed
+CREATE TABLE t1 (a int, b varchar(50)) ENGINE=InnoDB ENCRYPTED=YES ENCRYPTION_KEY_ID=6 PAGE_COMPRESSED=1;
+INSERT INTO t1 VALUES(42, "hello");
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t1, t2;
+
+# Embedded server uses absolute path, causing result mismatch in warning messages when .cfg is not copied over. Given we have tested importing without copying cfg in other tests, we don't do it here.
diff --git a/mysql-test/suite/innodb/include/import_begin.inc b/mysql-test/suite/innodb/include/import_begin.inc
new file mode 100644
index 00000000000..6bbbe54531f
--- /dev/null
+++ b/mysql-test/suite/innodb/include/import_begin.inc
@@ -0,0 +1,40 @@
+--source include/innodb_row_format_2.inc
+--source include/innodb_checksum_algorithm_2.inc
+
+--echo #
+--echo # MDEV-26137 ALTER TABLE IMPORT enhancement
+--echo #
+
+let $MYSQLD_DATADIR = `SELECT @@datadir`;
+
+--disable_query_log
+if($MTR_COMBINATION_ROW_FORMAT_DEFAULT) {
+  let $ROW_FORMAT_OPTION = ;
+}
+if($MTR_COMBINATION_REDUNDANT) {
+  let $ROW_FORMAT_OPTION = ROW_FORMAT=REDUNDANT;
+}
+if($MTR_COMBINATION_COMPACT) {
+  let $ROW_FORMAT_OPTION = ROW_FORMAT=COMPACT;
+}
+if($MTR_COMBINATION_DYNAMIC) {
+  let $ROW_FORMAT_OPTION = ROW_FORMAT=DYNAMIC;
+}
+if($MTR_COMBINATION_COMPRESSED) {
+  let $ROW_FORMAT_OPTION = ROW_FORMAT=COMPRESSED;
+}
+if($MTR_COMBINATION_REDUNDANT_DEFAULT) {
+  let $ROW_FORMAT_OPTION = ;
+  SET GLOBAL innodb_default_row_format='redundant';
+}
+if($MTR_COMBINATION_CRC32) {
+  SET GLOBAL innodb_checksum_algorithm=crc32;
+}
+let $CREATE_OPTIONS = ENGINE=InnoDB $ROW_FORMAT_OPTION;
+let $CREATE_OPTIONS_REGEX = /ENGINE=InnoDB.*$ROW_FORMAT_OPTION/CREATE_OPTIONS/;
+--enable_query_log
+
+--replace_regex $CREATE_OPTIONS_REGEX
+eval CREATE TABLE t1 (a int) $CREATE_OPTIONS;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
diff --git a/mysql-test/suite/innodb/include/import_end.inc b/mysql-test/suite/innodb/include/import_end.inc
new file mode 100644
index 00000000000..23d54a858c1
--- /dev/null
+++ b/mysql-test/suite/innodb/include/import_end.inc
@@ -0,0 +1,17 @@
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+--replace_regex $CREATE_OPTIONS_REGEX
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t1, t2;
+
+--disable_query_log
+if($MTR_COMBINATION_CRC32) {
+  SET GLOBAL innodb_checksum_algorithm=default;
+}
+if($MTR_COMBINATION_REDUNDANT_DEFAULT) {
+  SET GLOBAL innodb_default_row_format=default;
+}
+--enable_query_log
diff --git a/mysql-test/suite/innodb/include/innodb_checksum_algorithm_2.combinations b/mysql-test/suite/innodb/include/innodb_checksum_algorithm_2.combinations
new file mode 100644
index 00000000000..f7743e1adf6
--- /dev/null
+++ b/mysql-test/suite/innodb/include/innodb_checksum_algorithm_2.combinations
@@ -0,0 +1,2 @@
+[full_crc32]
+[crc32]
diff --git a/mysql-test/suite/innodb/include/innodb_checksum_algorithm_2.inc b/mysql-test/suite/innodb/include/innodb_checksum_algorithm_2.inc
new file mode 100644
index 00000000000..c841fece702
--- /dev/null
+++ b/mysql-test/suite/innodb/include/innodb_checksum_algorithm_2.inc
@@ -0,0 +1 @@
+--source include/have_innodb.inc
diff --git a/mysql-test/suite/innodb/include/innodb_row_format_2.combinations b/mysql-test/suite/innodb/include/innodb_row_format_2.combinations
new file mode 100644
index 00000000000..ccbd5c874b1
--- /dev/null
+++ b/mysql-test/suite/innodb/include/innodb_row_format_2.combinations
@@ -0,0 +1,6 @@
+[row_format_default]
+[redundant]
+[compact]
+[dynamic]
+[compressed]
+[redundant_default]
diff --git a/mysql-test/suite/innodb/include/innodb_row_format_2.inc b/mysql-test/suite/innodb/include/innodb_row_format_2.inc
new file mode 100644
index 00000000000..2d42f74ac66
--- /dev/null
+++ b/mysql-test/suite/innodb/include/innodb_row_format_2.inc
@@ -0,0 +1,4 @@
+# The goal of including this file is to add various row_format options
+# combinations (see include/innodb_row_format_2.combinations)
+
+--source include/have_innodb.inc
diff --git a/mysql-test/suite/innodb/r/import.result b/mysql-test/suite/innodb/r/import.result
new file mode 100644
index 00000000000..ffa9acdee8c
--- /dev/null
+++ b/mysql-test/suite/innodb/r/import.result
@@ -0,0 +1,17 @@
+#
+# MDEV-26137 ALTER TABLE IMPORT enhancement
+#
+CREATE TABLE t1 (a int) CREATE_OPTIONS;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) CREATE_OPTIONS
+SELECT * FROM t2;
+a
+42
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/innodb/r/import_no_cfg.result b/mysql-test/suite/innodb/r/import_no_cfg.result
new file mode 100644
index 00000000000..09f781e1b61
--- /dev/null
+++ b/mysql-test/suite/innodb/r/import_no_cfg.result
@@ -0,0 +1,20 @@
+#
+# MDEV-26137 ALTER TABLE IMPORT enhancement
+#
+CREATE TABLE t1 (a int) CREATE_OPTIONS;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+Warnings:
+Warning	1810	IO Read error: (2, No such file or directory) Error opening './test/t2.cfg', will attempt to import without schema verification
+Warning	1810	IO Read error: (2, No such file or directory) Error opening './test/t2.cfg', will attempt to import without schema verification
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) CREATE_OPTIONS
+SELECT * FROM t2;
+a
+42
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/innodb/r/import_recovery.result b/mysql-test/suite/innodb/r/import_recovery.result
new file mode 100644
index 00000000000..961c9d1ab95
--- /dev/null
+++ b/mysql-test/suite/innodb/r/import_recovery.result
@@ -0,0 +1,92 @@
+#
+# MDEV-26137 ALTER TABLE IMPORT enhancement
+#
+call mtr.add_suppression('InnoDB: Tablespace for table `test`.`t2` is set as discarded.');
+# Recovery from crashes
+## Creation of stub succeeds; server crashes; second import attempt succeeds
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+connect  hang,localhost,root;
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+ALTER TABLE t2 IMPORT TABLESPACE;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+# restart
+disconnect hang;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SELECT * FROM t2;
+a
+42
+DROP TABLE t1, t2;
+## Creation of stub succeeds; server crashes; drop stub succeeds
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+connect  hang,localhost,root;
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+ALTER TABLE t2 IMPORT TABLESPACE;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+# restart
+disconnect hang;
+DROP TABLE t1, t2;
+# Recovery from corruption
+call mtr.add_suppression('InnoDB: ./test/t2.ibd: Page 0 at offset 0 looks corrupted.');
+call mtr.add_suppression("mariadbd.*: Index for table 't2' is corrupt; try to repair it");
+call mtr.add_suppression("InnoDB: Corrupted page \\[page id: space=.*, page number=0\\] of datafile './test/t2.ibd' could not be found in the doublewrite buffer.");
+## Recovery from corruption, with cfg
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+# corrupting the 0th page
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR HY000: Internal error: Error importing tablespace for table `test`.`t2` : Data structure corruption
+DROP TABLE t1, t2;
+## Recovery from corruption, without cfg
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+# corrupting the 0th page
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR HY000: Schema mismatch (Expected FSP_SPACE_FLAGS=0x15, .ibd file contains 0x1010115.)
+DROP TABLE t1, t2;
+#recovery from crash and corruption
+## Creation of stub succeeds; server crashes; ibd corrupted; second import attempt fails; drops table
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+connect  hang,localhost,root;
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+ALTER TABLE t2 IMPORT TABLESPACE;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+disconnect hang;
+# corrupting the 0th page
+# Restart mysqld after the crash and reconnect.
+# restart
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR HY000: Internal error: Error importing tablespace for table `test`.`t2` : Data structure corruption
+DROP TABLE t1, t2;
+## Creation of stub without .cfg succeeds; server crashes; ibd corrupted; second import attempt fails; drops table
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+connect  hang,localhost,root;
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+ALTER TABLE t2 IMPORT TABLESPACE;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+disconnect hang;
+# corrupting the 0th page
+# Restart mysqld after the crash and reconnect.
+# restart
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR HY000: Schema mismatch (Expected FSP_SPACE_FLAGS=0x15, .ibd file contains 0x1010115.)
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/innodb/r/import_run_once.result b/mysql-test/suite/innodb/r/import_run_once.result
new file mode 100644
index 00000000000..e49340b8dfc
--- /dev/null
+++ b/mysql-test/suite/innodb/r/import_run_once.result
@@ -0,0 +1,104 @@
+#
+# MDEV-26137 ALTER TABLE IMPORT enhancement
+#
+# drop t1 before importing t2
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+DROP TABLE t1;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT * FROM t2;
+a
+42
+DROP TABLE t2;
+# created t2 but did not discard tablespace
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+CREATE TABLE t2 LIKE t1;
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+DROP TABLE t1;
+call mtr.add_suppression("InnoDB: Unable to import tablespace");
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR HY000: Tablespace for table 'test/t2' exists. Please DISCARD the tablespace before IMPORT
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT * FROM t2;
+a
+DROP TABLE t2;
+# attempt to import when there's no tablespace
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR 42S02: Table 'test.t2' doesn't exist
+# with index
+CREATE TABLE t1 (a int, b varchar(50)) ENGINE=InnoDB;
+CREATE UNIQUE INDEX ai ON t1 (a);
+INSERT INTO t1 VALUES(42, "hello");
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL,
+  `b` varchar(50) DEFAULT NULL,
+  UNIQUE KEY `ai` (`a`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT * FROM t2;
+a	b
+42	hello
+SHOW INDEX FROM t1;
+Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment	Index_comment	Ignored
+t1	0	ai	1	a	A	1	NULL	NULL	YES	BTREE			NO
+SHOW INDEX FROM t2;
+Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment	Index_comment	Ignored
+t2	0	ai	1	a	A	1	NULL	NULL	YES	BTREE			NO
+DROP TABLE t1, t2;
+# with virtual column index
+CREATE TABLE t1 (a int, b int as (a * a)) ENGINE=InnoDB;
+CREATE UNIQUE INDEX ai ON t1 (b);
+INSERT INTO t1 VALUES(42, default);
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL,
+  `b` int(11) GENERATED ALWAYS AS (`a` * `a`) VIRTUAL,
+  UNIQUE KEY `ai` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT * FROM t2;
+a	b
+42	1764
+SELECT b FROM t2 USE INDEX (ai);
+b
+1764
+SHOW INDEX FROM t1;
+Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment	Index_comment	Ignored
+t1	0	ai	1	b	A	1	NULL	NULL	YES	BTREE			NO
+SHOW INDEX FROM t2;
+Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment	Index_comment	Ignored
+t2	0	ai	1	b	A	1	NULL	NULL	YES	BTREE			NO
+CHECK TABLE t2 EXTENDED;
+Table	Op	Msg_type	Msg_text
+test.t2	check	status	OK
+DROP TABLE t1, t2;
+# with auto_increment
+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;
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+CHECK TABLE t2 EXTENDED;
+Table	Op	Msg_type	Msg_text
+test.t2	check	status	OK
+DROP TABLE t2, t1;
diff --git a/mysql-test/suite/innodb/t/import.test b/mysql-test/suite/innodb/t/import.test
new file mode 100644
index 00000000000..307d2b4064f
--- /dev/null
+++ b/mysql-test/suite/innodb/t/import.test
@@ -0,0 +1,3 @@
+--source include/import_begin.inc
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--source include/import_end.inc
diff --git a/mysql-test/suite/innodb/t/import_no_cfg.test b/mysql-test/suite/innodb/t/import_no_cfg.test
new file mode 100644
index 00000000000..d22814fef5d
--- /dev/null
+++ b/mysql-test/suite/innodb/t/import_no_cfg.test
@@ -0,0 +1,4 @@
+# embedded server uses absolute path, causing result mismatch in warning messages
+--source include/not_embedded.inc
+--source include/import_begin.inc
+--source include/import_end.inc
diff --git a/mysql-test/suite/innodb/t/import_recovery.test b/mysql-test/suite/innodb/t/import_recovery.test
new file mode 100644
index 00000000000..69913658088
--- /dev/null
+++ b/mysql-test/suite/innodb/t/import_recovery.test
@@ -0,0 +1,217 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/not_embedded.inc
+
+--echo #
+--echo # MDEV-26137 ALTER TABLE IMPORT enhancement
+--echo #
+
+let $MYSQLD_DATADIR = `SELECT @@datadir`;
+
+call mtr.add_suppression('InnoDB: Tablespace for table `test`.`t2` is set as discarded.');
+
+--echo # Recovery from crashes
+--echo ## Creation of stub succeeds; server crashes; second import attempt succeeds
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+
+connect (hang,localhost,root);
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+send
+
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+let $shutdown_timeout=0;
+--source include/restart_mysqld.inc
+disconnect hang;
+
+ALTER TABLE t2 IMPORT TABLESPACE;
+SELECT * FROM t2;
+DROP TABLE t1, t2;
+
+--echo ## Creation of stub succeeds; server crashes; drop stub succeeds
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+
+connect (hang,localhost,root);
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+send
+
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+let $shutdown_timeout=0;
+--source include/restart_mysqld.inc
+disconnect hang;
+
+DROP TABLE t1, t2;
+
+--echo # Recovery from corruption
+
+let MYSQLD_DATADIR = `SELECT @@datadir`;
+let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
+
+call mtr.add_suppression('InnoDB: ./test/t2.ibd: Page 0 at offset 0 looks corrupted.');
+call mtr.add_suppression("mariadbd.*: Index for table 't2' is corrupt; try to repair it");
+# In Windows etc.
+call mtr.add_suppression("InnoDB: Corrupted page \\[page id: space=.*, page number=0\\] of datafile './test/t2.ibd' could not be found in the doublewrite buffer.");
+
+--echo ## Recovery from corruption, with cfg
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+
+--echo # corrupting the 0th page
+perl;
+my $ps = $ENV{INNODB_PAGE_SIZE};
+
+my $file = "$ENV{MYSQLD_DATADIR}/test/t2.ibd";
+open(FILE, "+<$file") || die "Unable to open $file";
+binmode FILE;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+die "Unable to read $file" unless sysread(FILE, $page, $ps) == $ps;
+# Replace all 0 bits with 1 bits.
+$page =~ s/\x0/\x1/g;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+syswrite(FILE, $page, $ps)==$ps || die "Unable to write $file\n";
+close FILE or die "close";
+EOF
+
+--error ER_INTERNAL_ERROR
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t1, t2;
+
+
+--echo ## Recovery from corruption, without cfg
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+
+--echo # corrupting the 0th page
+perl;
+my $ps = $ENV{INNODB_PAGE_SIZE};
+
+my $file = "$ENV{MYSQLD_DATADIR}/test/t2.ibd";
+open(FILE, "+<$file") || die "Unable to open $file";
+binmode FILE;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+die "Unable to read $file" unless sysread(FILE, $page, $ps) == $ps;
+# Replace all 0 bits with 1 bits.
+$page =~ s/\x0/\x1/g;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+syswrite(FILE, $page, $ps)==$ps || die "Unable to write $file\n";
+close FILE or die "close";
+EOF
+
+--error ER_TABLE_SCHEMA_MISMATCH
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t1, t2;
+
+--echo #recovery from crash and corruption
+--echo ## Creation of stub succeeds; server crashes; ibd corrupted; second import attempt fails; drops table
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+
+connect (hang,localhost,root);
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+send
+
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+let $shutdown_timeout=0;
+--source include/shutdown_mysqld.inc
+disconnect hang;
+
+--echo # corrupting the 0th page
+perl;
+my $ps = $ENV{INNODB_PAGE_SIZE};
+
+my $file = "$ENV{MYSQLD_DATADIR}/test/t2.ibd";
+open(FILE, "+<$file") || die "Unable to open $file";
+binmode FILE;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+die "Unable to read $file" unless sysread(FILE, $page, $ps) == $ps;
+# Replace all 0 bits with 1 bits.
+$page =~ s/\x0/\x1/g;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+syswrite(FILE, $page, $ps)==$ps || die "Unable to write $file\n";
+close FILE or die "close";
+EOF
+
+--echo # Restart mysqld after the crash and reconnect.
+--source include/start_mysqld.inc
+
+--error ER_INTERNAL_ERROR
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t1, t2;
+
+
+--echo ## Creation of stub without .cfg succeeds; server crashes; ibd corrupted; second import attempt fails; drops table
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+
+connect (hang,localhost,root);
+SET DEBUG_SYNC='ib_after_create_stub_for_import SIGNAL hung WAIT_FOR ever';
+send
+
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR hung';
+let $shutdown_timeout=0;
+--source include/shutdown_mysqld.inc
+disconnect hang;
+
+--echo # corrupting the 0th page
+perl;
+my $ps = $ENV{INNODB_PAGE_SIZE};
+
+my $file = "$ENV{MYSQLD_DATADIR}/test/t2.ibd";
+open(FILE, "+<$file") || die "Unable to open $file";
+binmode FILE;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+die "Unable to read $file" unless sysread(FILE, $page, $ps) == $ps;
+# Replace all 0 bits with 1 bits.
+$page =~ s/\x0/\x1/g;
+sysseek(FILE, 0, 0) || die "Unable to seek $file\n";
+syswrite(FILE, $page, $ps)==$ps || die "Unable to write $file\n";
+close FILE or die "close";
+EOF
+
+--echo # Restart mysqld after the crash and reconnect.
+--source include/start_mysqld.inc
+
+--error ER_TABLE_SCHEMA_MISMATCH
+ALTER TABLE t2 IMPORT TABLESPACE;
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/innodb/t/import_run_once.test b/mysql-test/suite/innodb/t/import_run_once.test
new file mode 100644
index 00000000000..916851982ec
--- /dev/null
+++ b/mysql-test/suite/innodb/t/import_run_once.test
@@ -0,0 +1,90 @@
+--source include/have_innodb.inc
+--source include/have_sequence.inc
+let $MYSQLD_DATADIR = `SELECT @@datadir`;
+
+--echo #
+--echo # MDEV-26137 ALTER TABLE IMPORT enhancement
+--echo #
+
+--echo # drop t1 before importing t2
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+DROP TABLE t1;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t2;
+
+--echo # created t2 but did not discard tablespace
+CREATE TABLE t1 (a int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(42);
+CREATE TABLE t2 LIKE t1;
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--error 1
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--error 1
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+DROP TABLE t1;
+call mtr.add_suppression("InnoDB: Unable to import tablespace");
+--error ER_TABLESPACE_EXISTS
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t2;
+
+--echo # attempt to import when there's no tablespace
+--error ER_NO_SUCH_TABLE
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+--echo # with index
+CREATE TABLE t1 (a int, b varchar(50)) ENGINE=InnoDB;
+CREATE UNIQUE INDEX ai ON t1 (a);
+INSERT INTO t1 VALUES(42, "hello");
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+SHOW INDEX FROM t1;
+SHOW INDEX FROM t2;
+DROP TABLE t1, t2;
+
+--echo # with virtual column index
+CREATE TABLE t1 (a int, b int as (a * a)) ENGINE=InnoDB;
+CREATE UNIQUE INDEX ai ON t1 (b);
+INSERT INTO t1 VALUES(42, default);
+FLUSH TABLES t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+SELECT b FROM t2 USE INDEX (ai);
+SHOW INDEX FROM t1;
+SHOW INDEX FROM t2;
+CHECK TABLE t2 EXTENDED;
+DROP TABLE t1, t2;
+
+--echo # with auto_increment
+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;
+FLUSH TABLE t1 FOR EXPORT;
+--copy_file $MYSQLD_DATADIR/test/t1.frm $MYSQLD_DATADIR/test/t2.frm
+--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;
+CHECK TABLE t2 EXTENDED;
+DROP TABLE t2, t1;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 49655f8f0ce..b9b269933bb 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7212,6 +7212,7 @@ alter_commands:
             Lex->m_sql_cmd= new (thd->mem_root)
               Sql_cmd_discard_import_tablespace(
                 Sql_cmd_discard_import_tablespace::IMPORT_TABLESPACE);
+            Lex->create_info.add(DDL_options_st::OPT_IMPORT_TABLESPACE);
             if (unlikely(Lex->m_sql_cmd == NULL))
               MYSQL_YYABORT;
           }
diff --git a/sql/structs.h b/sql/structs.h
index 214fcb242ff..53bf84f8e59 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -537,7 +537,8 @@ public:
     OPT_OR_REPLACE_SLAVE_GENERATED= 32,// REPLACE was added on slave, it was
                                        // not in the original query on master.
     OPT_IF_EXISTS= 64,
-    OPT_CREATE_SELECT= 128             // CREATE ... SELECT
+    OPT_CREATE_SELECT= 128,             // CREATE ... SELECT
+    OPT_IMPORT_TABLESPACE= 256      // ALTER ... IMPORT TABLESPACE
   };
 
 private:
@@ -566,6 +567,7 @@ public:
   bool like() const { return m_options & OPT_LIKE; }
   bool if_exists() const { return m_options & OPT_IF_EXISTS; }
   bool is_create_select() const { return m_options & OPT_CREATE_SELECT; }
+  bool import_tablespace() const { return m_options & OPT_IMPORT_TABLESPACE; }
 
   void add(const DDL_options_st::Options other)
   {
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 219d7399005..48b2c668372 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -292,6 +292,27 @@ get_row_format(
 	}
 }
 
+/** 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;
+  /* Impossible. */
+  default:
+    return ROW_TYPE_DEFAULT;
+  }
+}
+
 static ulong	innodb_default_row_format = DEFAULT_ROW_FORMAT_DYNAMIC;
 
 /** Possible values for system variable "innodb_stats_method". The values
@@ -1877,6 +1898,61 @@ static int innobase_wsrep_get_checkpoint(handlerton* hton, XID* xid);
 #define normalize_table_name(a,b) \
 	normalize_table_name_c_low(a,b,IF_WIN(true,false))
 
+
+/** Prepare the create info to create a new stub table for import.
+@param[in]	thd		Connection
+@param[in]	name		Table name, format: "db/table_name".
+@param[in,out]	create_info	The create info for creating a stub.
+@return	0 if success else error number. */
+static 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(fil_path_to_mysql_datadir, name,
+                             IO_BUFFER_SIZE(srv_page_size),
+                             fetchIndexRootPages)
+      != 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) */
+  enum rec_format_enum rec_format_from_cfg;
+  if (get_row_type_from_cfg(fil_path_to_mysql_datadir, name, thd,
+                            rec_format_from_cfg)
+      == DB_SUCCESS)
+  {
+    /* if ibd reports not_used but cfg reports compact or dynamic, go
+    with cfg. */
+    if (create_info.row_type != from_rec_format(rec_format_from_cfg) &&
+        !((rec_format_from_cfg == REC_FORMAT_COMPACT ||
+           rec_format_from_cfg == REC_FORMAT_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= from_rec_format(rec_format_from_cfg);
+  } else if (create_info.row_type == ROW_TYPE_NOT_USED)
+    create_info.row_type = ROW_TYPE_DYNAMIC;
+  DBUG_RETURN(0);
+}
+
 ulonglong ha_innobase::table_version() const
 {
   /* This is either "garbage" or something that was assigned
@@ -5847,8 +5923,22 @@ ha_innobase::open(const char* name, int, uint)
 
 	DEBUG_SYNC(thd, "ib_open_after_dict_open");
 
-	if (NULL == ib_table) {
+	/* If the table does not exist and we are trying to import, create a
+	"stub" table similar to the effects of CREATE TABLE followed by ALTER
+	TABLE ... DISCARD TABLESPACE. */
+	if (!ib_table && thd_ddl_options(thd)->import_tablespace())
+        {
+                HA_CREATE_INFO create_info;
+                if (int err= prepare_create_stub_for_import(thd, norm_name,
+                                                            create_info))
+                        DBUG_RETURN(err);
+                create(norm_name, table, &create_info, true, nullptr);
+                DEBUG_SYNC(thd, "ib_after_create_stub_for_import");
+                ib_table = open_dict_table(name, norm_name, is_part,
+                                           DICT_ERR_IGNORE_FK_NOKEY);
+        }
 
+        if (NULL == ib_table) {
 		if (is_part) {
 			sql_print_error("Failed to open table %s.\n",
 					norm_name);
@@ -10571,6 +10661,11 @@ create_table_info_t::create_table_def()
 				      ? doc_id_col : n_cols - num_v;
 	}
 
+        /* Assume the tablespace is not available until we are able to
+        import it.*/
+        if (thd_ddl_options(m_thd)->import_tablespace())
+                table->file_unreadable = true;
+
 	if (DICT_TF_HAS_DATA_DIR(m_flags)) {
 		ut_a(strlen(m_remote_path));
 
@@ -11584,6 +11679,11 @@ index_bad:
 		}
 	}
 
+        /* If we are trying to import a tablespace, mark tablespace as
+        discarded. */
+        if (thd_ddl_options(m_thd)->import_tablespace())
+                m_flags2 |= DICT_TF2_DISCARDED;
+
 	row_type = m_create_info->row_type;
 
 	if (zip_ssize && zip_allowed) {
@@ -13134,8 +13234,12 @@ ha_innobase::create(const char *name, TABLE *form, HA_CREATE_INFO *create_info,
     row_mysql_lock_data_dictionary(trx);
   }
 
+  const bool importing= thd_ddl_options(ha_thd())->import_tablespace();
   if (!error)
-    error= info.create_table(own_trx);
+    /* We can't possibly have foreign key information when creating a
+    stub table for importing .frm / .cfg / .ibd because it is not
+    stored in any of these files. */
+    error= info.create_table(own_trx && !importing);
 
   if (own_trx || (info.flags2() & DICT_TF2_TEMPORARY))
   {
@@ -13157,7 +13261,11 @@ ha_innobase::create(const char *name, TABLE *form, HA_CREATE_INFO *create_info,
 
       if (!error)
       {
-        dict_stats_update(info.table(), DICT_STATS_EMPTY_TABLE);
+        /* Skip stats update when creating a stub table for importing,
+        as it is not needed and would report error due to the table
+        not being readable yet. */
+        if (!importing)
+          dict_stats_update(info.table(), DICT_STATS_EMPTY_TABLE);
         if (!info.table()->is_temporary())
           log_write_up_to(trx->commit_lsn, true);
         info.table()->release();
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index c1927bec916..82a9596032a 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -2097,8 +2097,9 @@ public:
 	process of altering partitions */
 	unsigned                                skip_alter_undo:1;
 
-	/*!< whether this is in a single-table tablespace and the .ibd
-	file is missing or page decryption failed and page is corrupted */
+	/** whether this is in a single-table tablespace and the .ibd file
+	is believed to be missing or page decryption failed and page is
+	corrupted */
 	unsigned				file_unreadable:1;
 
 	/** TRUE if the table object has been added to the dictionary cache. */
diff --git a/storage/innobase/include/row0import.h b/storage/innobase/include/row0import.h
index fd2651da39e..7019ab2a900 100644
--- a/storage/innobase/include/row0import.h
+++ b/storage/innobase/include/row0import.h
@@ -28,11 +28,296 @@ Created 2012-02-08 by Sunny Bains
 #define row0import_h
 
 #include "dict0types.h"
+#include "ut0new.h"
+#include "os0file.h"
+#include "buf0buf.h"
+#include "trx0trx.h"
+#include "page0page.h"
 
 // Forward declarations
 struct trx_t;
 struct dict_table_t;
 struct row_prebuilt_t;
+struct fil_iterator_t;
+struct row_import;
+
+/** The size of the buffer to use for IO.
+@param n physical page size
+@return number of pages */
+#define IO_BUFFER_SIZE(n)	((1024 * 1024) / (n))
+
+/** Functor that is called for each physical page that is read from the
+tablespace file.  */
+class AbstractCallback
+{
+public:
+  /** Constructor
+  @param trx covering transaction */
+  AbstractCallback(trx_t* trx, uint32_t space_id)
+    :
+    m_zip_size(0),
+    m_trx(trx),
+    m_space(space_id),
+    m_xdes(),
+    m_xdes_page_no(UINT32_MAX),
+    m_space_flags(UINT32_MAX) UNIV_NOTHROW { }
+
+  /** Free any extent descriptor instance */
+  virtual ~AbstractCallback()
+  {
+    UT_DELETE_ARRAY(m_xdes);
+  }
+
+  /** Determine the page size to use for traversing the tablespace
+  @param file_size size of the tablespace file in bytes
+  @param block contents of the first page in the tablespace file.
+  @retval DB_SUCCESS or error code. */
+  virtual dberr_t init(
+    os_offset_t     file_size,
+    const buf_block_t*  block) UNIV_NOTHROW;
+
+  /** @return true if compressed table. */
+  bool is_compressed_table() const UNIV_NOTHROW
+  {
+    return get_zip_size();
+  }
+
+  /** @return the tablespace flags */
+  uint32_t get_space_flags() const { return m_space_flags; }
+
+  /**
+  Set the name of the physical file and the file handle that is used
+  to open it for the file that is being iterated over.
+  @param filename the physical name of the tablespace file
+  @param file OS file handle */
+  void set_file(const char* filename, pfs_os_file_t file) UNIV_NOTHROW
+  {
+    m_file = file;
+    m_filepath = filename;
+  }
+
+  ulint get_zip_size() const { return m_zip_size; }
+  ulint physical_size() const
+  {
+    return m_zip_size ? m_zip_size : srv_page_size;
+  }
+
+  const char* filename() const { return m_filepath; }
+
+  /**
+  Called for every page in the tablespace. If the page was not
+  updated then its state must be set to BUF_PAGE_NOT_USED. For
+  compressed tables the page descriptor memory will be at offset:
+    block->page.frame + srv_page_size;
+  @param block block read from file, note it is not from the buffer pool
+  @retval DB_SUCCESS or error code. */
+  virtual dberr_t operator()(buf_block_t* block) UNIV_NOTHROW = 0;
+
+  /** @return the tablespace identifier */
+  uint32_t get_space_id() const { return m_space; }
+
+  bool is_interrupted() const { return trx_is_interrupted(m_trx); }
+
+  /**
+  Get the data page depending on the table type, compressed or not.
+  @param block - block read from disk
+  @retval the buffer frame */
+  static byte* get_frame(const buf_block_t* block)
+  {
+    return block->page.zip.data
+      ? block->page.zip.data : block->page.frame;
+  }
+
+  /** Invoke the functionality for the callback */
+  virtual dberr_t run(const fil_iterator_t& iter,
+          buf_block_t* block) UNIV_NOTHROW = 0;
+
+protected:
+  /** Get the physical offset of the extent descriptor within the page.
+  @param page_no page number of the extent descriptor
+  @param page contents of the page containing the extent descriptor.
+  @return the start of the xdes array in a page */
+  const xdes_t* xdes(
+    ulint     page_no,
+    const page_t*   page) const UNIV_NOTHROW
+  {
+    ulint   offset;
+
+    offset = xdes_calc_descriptor_index(get_zip_size(), page_no);
+
+    return(page + XDES_ARR_OFFSET + XDES_SIZE * offset);
+  }
+
+  /** Set the current page directory (xdes). If the extent descriptor is
+  marked as free then free the current extent descriptor and set it to
+  0. This implies that all pages that are covered by this extent
+  descriptor are also freed.
+
+  @param page_no offset of page within the file
+  @param page page contents
+  @return DB_SUCCESS or error code. */
+  dberr_t   set_current_xdes(
+    uint32_t  page_no,
+    const page_t*   page) UNIV_NOTHROW
+  {
+    m_xdes_page_no = page_no;
+
+    UT_DELETE_ARRAY(m_xdes);
+    m_xdes = NULL;
+
+    if (mach_read_from_4(XDES_ARR_OFFSET + XDES_STATE + page)
+        != XDES_FREE) {
+      const ulint physical_size = m_zip_size
+        ? m_zip_size : srv_page_size;
+
+      m_xdes = UT_NEW_ARRAY_NOKEY(xdes_t, physical_size);
+
+      /* Trigger OOM */
+      DBUG_EXECUTE_IF(
+        "ib_import_OOM_13",
+        UT_DELETE_ARRAY(m_xdes);
+        m_xdes = NULL;
+      );
+
+      if (m_xdes == NULL) {
+        return(DB_OUT_OF_MEMORY);
+      }
+
+      memcpy(m_xdes, page, physical_size);
+    }
+
+    return(DB_SUCCESS);
+  }
+
+  /** Check if the page is marked as free in the extent descriptor.
+  @param page_no page number to check in the extent descriptor.
+  @return true if the page is marked as free */
+  bool is_free(uint32_t page_no) const UNIV_NOTHROW
+  {
+    ut_a(xdes_calc_descriptor_page(get_zip_size(), page_no)
+         == m_xdes_page_no);
+
+    if (m_xdes != 0) {
+      const xdes_t*   xdesc = xdes(page_no, m_xdes);
+      ulint     pos = page_no % FSP_EXTENT_SIZE;
+
+      return xdes_is_free(xdesc, pos);
+    }
+
+    /* If the current xdes was free, the page must be free. */
+    return(true);
+  }
+
+protected:
+  /** The ROW_FORMAT=COMPRESSED page size, or 0. */
+  ulint       m_zip_size;
+
+  /** File handle to the tablespace */
+  pfs_os_file_t     m_file;
+
+  /** Physical file path. */
+  const char*     m_filepath;
+
+  /** Covering transaction. */
+  trx_t*      m_trx;
+
+  /** Space id of the file being iterated over. */
+  uint32_t    m_space;
+
+  /** Current extent descriptor page */
+  xdes_t*       m_xdes;
+
+  /** Physical page offset in the file of the extent descriptor */
+  uint32_t    m_xdes_page_no;
+
+  /** Flags value read from the header page */
+  uint32_t    m_space_flags;
+};
+
+/**
+Try and determine the index root pages by checking if the next/prev
+pointers are both FIL_NULL. We need to ensure that skip deleted pages. */
+struct FetchIndexRootPages : public AbstractCallback {
+
+  /** Index information gathered from the .ibd file. */
+  struct Index {
+
+    Index(index_id_t id, uint32_t page_no)
+      :
+      m_id(id),
+      m_page_no(page_no) { }
+
+    index_id_t  m_id;     /*!< Index id */
+    uint32_t  m_page_no;  /*!< Root page number */
+  };
+
+  /** Constructor
+  @param trx covering (user) transaction
+  @param table table definition in server .*/
+  FetchIndexRootPages(const dict_table_t* table, trx_t* trx)
+    :
+    AbstractCallback(trx, UINT32_MAX),
+    m_table(table), m_index(0, 0) UNIV_NOTHROW { }
+
+  FetchIndexRootPages()
+    :
+    AbstractCallback(nullptr, UINT32_MAX),
+    m_table(nullptr), m_index(0, 0) UNIV_NOTHROW { }
+
+  /** Destructor */
+  ~FetchIndexRootPages() UNIV_NOTHROW override = default;
+
+  /** Fetch the clustered index root page in the tablespace
+  @param iter   Tablespace iterator
+  @param block  Block to use for IO
+  @retval DB_SUCCESS or error code */
+  dberr_t run(const fil_iterator_t& iter,
+        buf_block_t* block) UNIV_NOTHROW override;
+
+  /** 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(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)
+    {
+      if (FSP_FLAGS_GET_ZIP_SSIZE(m_space_flags))
+        return ROW_TYPE_COMPRESSED;
+      else
+        return ROW_TYPE_DYNAMIC;
+    }
+    return ROW_TYPE_COMPACT;
+  }
+
+  /** 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. 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;
+};
 
 /*****************************************************************//**
 Imports a tablespace. The space id in the .ibd file must match the space id
@@ -64,4 +349,29 @@ dberr_t
 row_import_update_index_root(trx_t* trx, dict_table_t* table, bool reset)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
+/********************************************************************//**
+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 */
+dberr_t
+fil_tablespace_iterate(
+/*===================*/
+  const char* dir_path,
+  const char*   name,
+  ulint       n_io_buffers,
+  AbstractCallback&   callback);
+
+/**
+Read the row type from a .cfg file.
+@param[in]  dir_path  Path to the data directory containing the .cfg file
+@param[in]  name      Name of the table
+@param[in]  thd       Session
+@param[out] result    The row format read from the .cfg file
+@return DB_SUCCESS or error code. */
+dberr_t get_row_type_from_cfg(const char* dir_path, const char* name, THD* thd,
+                              rec_format_enum& result);
+
 #endif /* row0import_h */
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 5d7ea475d43..817c14e0087 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -55,11 +55,6 @@ Created 2012-02-08 by Sunny Bains.
 
 using st_::span;
 
-/** The size of the buffer to use for IO.
-@param n physical page size
-@return number of pages */
-#define IO_BUFFER_SIZE(n)	((1024 * 1024) / (n))
-
 /** For gathering stats on records during phase I */
 struct row_stats_t {
 	ulint		m_n_deleted;		/*!< Number of deleted records
@@ -396,194 +391,6 @@ private:
 	ulint			m_n_rows;	/*!< Records in index */
 };
 
-/** Functor that is called for each physical page that is read from the
-tablespace file.  */
-class AbstractCallback
-{
-public:
-	/** Constructor
-	@param trx covering transaction */
-	AbstractCallback(trx_t* trx, uint32_t space_id)
-		:
-		m_zip_size(0),
-		m_trx(trx),
-		m_space(space_id),
-		m_xdes(),
-		m_xdes_page_no(UINT32_MAX),
-		m_space_flags(UINT32_MAX) UNIV_NOTHROW { }
-
-	/** Free any extent descriptor instance */
-	virtual ~AbstractCallback()
-	{
-		UT_DELETE_ARRAY(m_xdes);
-	}
-
-	/** Determine the page size to use for traversing the tablespace
-	@param file_size size of the tablespace file in bytes
-	@param block contents of the first page in the tablespace file.
-	@retval DB_SUCCESS or error code. */
-	virtual dberr_t init(
-		os_offset_t		file_size,
-		const buf_block_t*	block) UNIV_NOTHROW;
-
-	/** @return true if compressed table. */
-	bool is_compressed_table() const UNIV_NOTHROW
-	{
-		return get_zip_size();
-	}
-
-	/** @return the tablespace flags */
-	uint32_t get_space_flags() const { return m_space_flags; }
-
-	/**
-	Set the name of the physical file and the file handle that is used
-	to open it for the file that is being iterated over.
-	@param filename the physical name of the tablespace file
-	@param file OS file handle */
-	void set_file(const char* filename, pfs_os_file_t file) UNIV_NOTHROW
-	{
-		m_file = file;
-		m_filepath = filename;
-	}
-
-	ulint get_zip_size() const { return m_zip_size; }
-	ulint physical_size() const
-	{
-		return m_zip_size ? m_zip_size : srv_page_size;
-	}
-
-	const char* filename() const { return m_filepath; }
-
-	/**
-	Called for every page in the tablespace. If the page was not
-	updated then its state must be set to BUF_PAGE_NOT_USED. For
-	compressed tables the page descriptor memory will be at offset:
-		block->page.frame + srv_page_size;
-	@param block block read from file, note it is not from the buffer pool
-	@retval DB_SUCCESS or error code. */
-	virtual dberr_t operator()(buf_block_t* block) UNIV_NOTHROW = 0;
-
-	/** @return the tablespace identifier */
-	uint32_t get_space_id() const { return m_space; }
-
-	bool is_interrupted() const { return trx_is_interrupted(m_trx); }
-
-	/**
-	Get the data page depending on the table type, compressed or not.
-	@param block - block read from disk
-	@retval the buffer frame */
-	static byte* get_frame(const buf_block_t* block)
-	{
-		return block->page.zip.data
-			? block->page.zip.data : block->page.frame;
-	}
-
-	/** Invoke the functionality for the callback */
-	virtual dberr_t run(const fil_iterator_t& iter,
-			    buf_block_t* block) UNIV_NOTHROW = 0;
-
-protected:
-	/** Get the physical offset of the extent descriptor within the page.
-	@param page_no page number of the extent descriptor
-	@param page contents of the page containing the extent descriptor.
-	@return the start of the xdes array in a page */
-	const xdes_t* xdes(
-		ulint		page_no,
-		const page_t*	page) const UNIV_NOTHROW
-	{
-		ulint	offset;
-
-		offset = xdes_calc_descriptor_index(get_zip_size(), page_no);
-
-		return(page + XDES_ARR_OFFSET + XDES_SIZE * offset);
-	}
-
-	/** Set the current page directory (xdes). If the extent descriptor is
-	marked as free then free the current extent descriptor and set it to
-	0. This implies that all pages that are covered by this extent
-	descriptor are also freed.
-
-	@param page_no offset of page within the file
-	@param page page contents
-	@return DB_SUCCESS or error code. */
-	dberr_t	set_current_xdes(
-		uint32_t	page_no,
-		const page_t*	page) UNIV_NOTHROW
-	{
-		m_xdes_page_no = page_no;
-
-		UT_DELETE_ARRAY(m_xdes);
-		m_xdes = NULL;
-
-		if (mach_read_from_4(XDES_ARR_OFFSET + XDES_STATE + page)
-		    != XDES_FREE) {
-			const ulint physical_size = m_zip_size
-				? m_zip_size : srv_page_size;
-
-			m_xdes = UT_NEW_ARRAY_NOKEY(xdes_t, physical_size);
-
-			/* Trigger OOM */
-			DBUG_EXECUTE_IF(
-				"ib_import_OOM_13",
-				UT_DELETE_ARRAY(m_xdes);
-				m_xdes = NULL;
-			);
-
-			if (m_xdes == NULL) {
-				return(DB_OUT_OF_MEMORY);
-			}
-
-			memcpy(m_xdes, page, physical_size);
-		}
-
-		return(DB_SUCCESS);
-	}
-
-	/** Check if the page is marked as free in the extent descriptor.
-	@param page_no page number to check in the extent descriptor.
-	@return true if the page is marked as free */
-	bool is_free(uint32_t page_no) const UNIV_NOTHROW
-	{
-		ut_a(xdes_calc_descriptor_page(get_zip_size(), page_no)
-		     == m_xdes_page_no);
-
-		if (m_xdes != 0) {
-			const xdes_t*	xdesc = xdes(page_no, m_xdes);
-			ulint		pos = page_no % FSP_EXTENT_SIZE;
-
-			return xdes_is_free(xdesc, pos);
-		}
-
-		/* If the current xdes was free, the page must be free. */
-		return(true);
-	}
-
-protected:
-	/** The ROW_FORMAT=COMPRESSED page size, or 0. */
-	ulint			m_zip_size;
-
-	/** File handle to the tablespace */
-	pfs_os_file_t		m_file;
-
-	/** Physical file path. */
-	const char*		m_filepath;
-
-	/** Covering transaction. */
-	trx_t*			m_trx;
-
-	/** Space id of the file being iterated over. */
-	uint32_t		m_space;
-
-	/** Current extent descriptor page */
-	xdes_t*			m_xdes;
-
-	/** Physical page offset in the file of the extent descriptor */
-	uint32_t		m_xdes_page_no;
-
-	/** Flags value read from the header page */
-	uint32_t		m_space_flags;
-};
-
 ATTRIBUTE_COLD static dberr_t invalid_space_flags(uint32_t flags)
 {
   if (fsp_flags_is_incompatible_mysql(flags))
@@ -665,65 +472,19 @@ static dberr_t fil_iterate(
 	buf_block_t*		block,
 	AbstractCallback&	callback);
 
-/**
-Try and determine the index root pages by checking if the next/prev
-pointers are both FIL_NULL. We need to ensure that skip deleted pages. */
-struct FetchIndexRootPages : public AbstractCallback {
-
-	/** Index information gathered from the .ibd file. */
-	struct Index {
-
-		Index(index_id_t id, uint32_t page_no)
-			:
-			m_id(id),
-			m_page_no(page_no) { }
-
-		index_id_t	m_id;		/*!< Index id */
-		uint32_t	m_page_no;	/*!< Root page number */
-	};
-
-	/** Constructor
-	@param trx covering (user) transaction
-	@param table table definition in server .*/
-	FetchIndexRootPages(const dict_table_t* table, trx_t* trx)
-		:
-		AbstractCallback(trx, UINT32_MAX),
-		m_table(table), m_index(0, 0) UNIV_NOTHROW { }
-
-	/** Destructor */
-	~FetchIndexRootPages() UNIV_NOTHROW override = default;
-
-	/** Fetch the clustered index root page in the tablespace
-	@param iter	Tablespace iterator
-	@param block	Block to use for IO
-	@retval DB_SUCCESS or error code */
-	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.
-	@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;
-
-	/** 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. */
-	const dict_table_t*	m_table;
-
-	/** Index information */
-	Index			m_index;
-};
-
-/** Called for each block as it is read from the file. Check index pages to
-determine the exact row format. We can't get that from the tablespace
-header flags alone.
+/** Check the fsp flags from the table and header file match. Also
+check against row format mismatch between the table and the index root
+page, as we can't get that from the tablespace header flags alone.
 
 @param block block to convert, it is not from the buffer pool.
 @retval DB_SUCCESS or error code. */
 dberr_t FetchIndexRootPages::operator()(buf_block_t* block) UNIV_NOTHROW
 {
+        /* m_table is non-null iff we are trying to create a (stub)
+        table, and there's no table to compare the fsp flags and row
+        format against. */
+        ut_ad(m_table);
+
 	if (is_interrupted()) return DB_INTERRUPTED;
 
 	const page_t*	page = get_frame(block);
@@ -3338,6 +3099,43 @@ 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       Session
+@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)
+{
+  dberr_t err;
+
+  FILE* file = fopen(filename, "rb");
+
+  if (file == NULL) {
+    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);
+
+    cfg.m_missing= true;
+
+    err= DB_FAIL;
+  } else
+    {
+      cfg.m_missing= false;
+
+      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))
@@ -3348,38 +3146,37 @@ 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;
-
-		err = DB_FAIL;
-	} else {
-
-		cfg.m_missing = false;
-
-		err = row_import_read_meta_data(file, thd, cfg);
-		fclose(file);
-	}
+        return row_import_read_cfg_internal(name, thd, cfg);
+}
 
-	return(err);
+/**
+Read the row type from a .cfg file.
+@param[in]  dir_path  Path to the data directory containing the .cfg file
+@param[in]  name      Name of the table
+@param[in]  thd       Session
+@param[out] result    The row format read from the .cfg file
+@return DB_SUCCESS or error code. */
+dberr_t get_row_type_from_cfg(const char* dir_path, const char* name,
+                              THD* thd, rec_format_enum& result)
+{
+  char* filename = fil_make_filepath(dir_path,
+                                     table_name_t(const_cast<char*>(name)),
+                                     CFG, dir_path != nullptr);
+  if (!filename)
+    return DB_OUT_OF_MEMORY;
+  row_import cfg;
+  dberr_t err = row_import_read_cfg_internal(filename, thd, cfg);
+  // ASAN
+  ut_free(filename);
+  if (err == DB_SUCCESS)
+    result = dict_tf_get_rec_format(cfg.m_flags);
+  return err;
 }
 
 /** Update the root page numbers and tablespace ID of a table.
@@ -3719,7 +3516,13 @@ page_corrupted:
            && buf_page_is_corrupted(false, readptr, m_space_flags))
     goto page_corrupted;
 
-  err= this->operator()(block);
+  /* m_table_name is non-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;
@@ -4042,16 +3845,17 @@ func_exit:
 }
 
 /********************************************************************//**
-Iterate over all the pages in the tablespace.
-@param table - the table definiton in the server
+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 updates
+@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,
+        const char*             dir_path,
+	const char*	        name,
 	ulint			n_io_buffers,
 	AbstractCallback&	callback)
 {
@@ -4065,18 +3869,8 @@ 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 {
@@ -4089,8 +3883,7 @@ 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 "
+			ib::error() << "could not open the tablespace file "
 				    << filepath;
 			ut_free(filepath);
 			return DB_TABLESPACE_NOT_FOUND;
@@ -4203,6 +3996,32 @@ 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(data_dir_path, table->name.m_name,
+                                      n_io_buffers, callback);
+}
+
 /*****************************************************************//**
 Imports a tablespace. The space id in the .ibd file must match the space id
 of the table in the data dictionary.