MDEV-30655 IMPORT TABLESPACE fails with column count or index count mismatch

Problem:
========
Currently import operation fails with schema mismatch when
cfg file has hidden fts document id and hidden fts document index.

Fix:
====
To fix this issue, simply add the fts doc id column,
indexes in table definition and try to import the table.
In case of success:
1) update the fts document id in sys columns.
2) update the number of columns in sys tables.
3) insert the new fts index entry in sys indexes table
and sys fields.
4) Reload the table with new table definition

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

+call mtr.add_suppression("InnoDB: Added system generated FTS_DOC_ID and FTS_DOC_ID_INDEX while importing the tablespace");
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+f2 CHAR(2) not null, fulltext f_idx(f2),
+f3 INT as (f1) VIRTUAL, INDEX(f3),
+f4 INT as (f1) STORED, INDEX(f4),
+f5 INT as (f1) VIRTUAL)ENGINE=InnoDB;
+INSERT INTO t1(f1, f2) VALUES(1, "on");
+INSERT INTO t1(f1, f2) SELECT seq, "si" FROM seq_2_to_256;
+ALTER TABLE t1 ADD COLUMN f6 INT NOT NULL;
+ALTER TABLE t1 DROP COLUMN f6;
+ALTER TABLE t1 DROP INDEX f_idx;
+connect con1,localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+DELETE FROM t1 WHERE f1 > 1;
+FLUSH TABLE t1 FOR EXPORT;
+Warnings:
+Warning	1235	This version of MariaDB doesn't yet support 'FLUSH TABLES on a table that had an FTS index, created on a hidden column, the auxiliary tables haven't been dropped as yet. FTS auxiliary tables will not be flushed.'
+Warning	1235	This version of MariaDB doesn't yet support 'FLUSH TABLES on a table that had an FTS index, created on a hidden column, the auxiliary tables haven't been dropped as yet. FTS auxiliary tables will not be flushed.'
+backup: t1
+UNLOCK TABLES;
+Warnings:
+Warning	1235	This version of MariaDB doesn't yet support 'FLUSH TABLES on a table that had an FTS index, created on a hidden column, the auxiliary tables haven't been dropped as yet. FTS auxiliary tables will not be flushed.'
+DROP TABLE t1;
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+f2 CHAR(2) not null,
+f3 INT as (f1) VIRTUAL, INDEX(f3),
+f4 INT as (f1) STORED, INDEX(f4),
+f5 INT as (f1) VIRTUAL)ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+restore: t1 .ibd and .cfg files
+ALTER TABLE t1 IMPORT TABLESPACE;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `f1` int(11) NOT NULL,
+  `f2` char(2) NOT NULL,
+  `f3` int(11) GENERATED ALWAYS AS (`f1`) VIRTUAL,
+  `f4` int(11) GENERATED ALWAYS AS (`f1`) STORED,
+  `f5` int(11) GENERATED ALWAYS AS (`f1`) VIRTUAL,
+  PRIMARY KEY (`f1`),
+  KEY `f3` (`f3`),
+  KEY `f4` (`f4`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+DROP TABLE t1;

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

+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+f2 CHAR(2) NOT NULL, FULLTEXT f_idx(f2),
+f3 INT as (f1) VIRTUAL, INDEX(f3))ENGINE=InnoDB;
+INSERT INTO t1(f1, f2) VALUES(1, "on");
+ALTER TABLE t1 DROP INDEX f_idx;
+FLUSH TABLE t1 FOR EXPORT;
+Warnings:
+Warning	1235	This version of MariaDB doesn't yet support 'FLUSH TABLES on a table that had an FTS index, created on a hidden column, the auxiliary tables haven't been dropped as yet. FTS auxiliary tables will not be flushed.'
+Warning	1235	This version of MariaDB doesn't yet support 'FLUSH TABLES on a table that had an FTS index, created on a hidden column, the auxiliary tables haven't been dropped as yet. FTS auxiliary tables will not be flushed.'
+backup: t1
+UNLOCK TABLES;
+Warnings:
+Warning	1235	This version of MariaDB doesn't yet support 'FLUSH TABLES on a table that had an FTS index, created on a hidden column, the auxiliary tables haven't been dropped as yet. FTS auxiliary tables will not be flushed.'
+DROP TABLE t1;
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+f2 CHAR(2) NOT NULL,
+f3 CHAR(2) NOT NULL,
+f4 INT AS (f1) VIRTUAL, INDEX(f4))ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+restore: t1 .ibd and .cfg files
+ALTER TABLE t1 IMPORT TABLESPACE;
+ERROR HY000: Schema mismatch (Number of indexes don't match, table has 2 indexes but the tablespace meta-data file has 3 indexes)
+DROP TABLE t1;
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+f2 CHAR(2) NOT NULL,
+f3 INT as (f1) VIRTUAL, INDEX(f3))ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+restore: t1 .ibd and .cfg files
+SET DEBUG_DBUG="+d,ib_import_set_index_root_failure";
+ALTER TABLE t1 IMPORT TABLESPACE;
+ERROR HY000: Too many active concurrent transactions
+SET DEBUG_DBUG="-d,ib_import_set_index_root_failure";
+SET DEBUG_DBUG="+d,ib_import_vcol_update_fail";
+ALTER TABLE t1 IMPORT TABLESPACE;
+ERROR 23000: Can't write; duplicate key in table 't1'
+SET DEBUG_DBUG="-d,ib_import_vcol_update_fail";
+restore: t1 .ibd and .cfg files
+SET DEBUG_DBUG="+d,ib_import_fts_error";
+ALTER TABLE t1 IMPORT TABLESPACE;
+ERROR 23000: Can't write; duplicate key in table 't1'
+SET DEBUG_DBUG="-d,ib_import_fts_error";
+unlink: t1.ibd
+unlink: t1.cfg
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_SYS_COLUMNS
+WHERE table_id IN (SELECT table_id FROM information_schema.innodb_sys_tables where name="test/t1");
+NAME
+f1
+f2
+f3
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `f1` int(11) NOT NULL,
+  `f2` char(2) NOT NULL,
+  `f3` int(11) GENERATED ALWAYS AS (`f1`) VIRTUAL,
+  PRIMARY KEY (`f1`),
+  KEY `f3` (`f3`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+DROP TABLE t1;
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+FTS_DOC_ID BIGINT SIGNED NOT NULL,
+f2 CHAR(2) NOT NULL,
+FULLTEXT f_idx(f2))ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1, 1, "on");
+ALTER TABLE t1 DROP INDEX f_idx;
+FLUSH TABLE t1 FOR EXPORT;
+backup: t1
+UNLOCK TABLES;
+DROP TABLE t1;
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+f2 CHAR(2) NOT NULL)ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+restore: t1 .ibd and .cfg files
+ALTER TABLE t1 IMPORT TABLESPACE;
+ERROR HY000: Schema mismatch (Column f2 ordinal value mismatch, it's at 1 in the table and 2 in the tablespace meta-data file)
+DROP TABLE t1;

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

+--source include/have_innodb.inc
+--source include/have_sequence.inc
+# Table with virtual, fulltext, instant add, instant drop column
+# and purgeable rows
+call mtr.add_suppression("InnoDB: Added system generated FTS_DOC_ID and FTS_DOC_ID_INDEX while importing the tablespace");
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+		f2 CHAR(2) not null, fulltext f_idx(f2),
+		f3 INT as (f1) VIRTUAL, INDEX(f3),
+		f4 INT as (f1) STORED, INDEX(f4),
+		f5 INT as (f1) VIRTUAL)ENGINE=InnoDB;
+INSERT INTO t1(f1, f2) VALUES(1, "on");
+INSERT INTO t1(f1, f2) SELECT seq, "si" FROM seq_2_to_256;
+ALTER TABLE t1 ADD COLUMN f6 INT NOT NULL;
+ALTER TABLE t1 DROP COLUMN f6;
+ALTER TABLE t1 DROP INDEX f_idx;
+connect(con1,localhost,root,,);
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+connection default;
+DELETE FROM t1 WHERE f1 > 1;
+FLUSH TABLE t1 FOR EXPORT;
+let MYSQLD_DATADIR =`SELECT @@datadir`;
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_backup_tablespaces("test", "t1");
+EOF
+UNLOCK TABLES;
+DROP TABLE t1;
+
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+		f2 CHAR(2) not null,
+		f3 INT as (f1) VIRTUAL, INDEX(f3),
+		f4 INT as (f1) STORED, INDEX(f4),
+		f5 INT as (f1) VIRTUAL)ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_discard_tablespaces("test", "t1");
+ib_restore_tablespaces("test", "t1");
+EOF
+
+--disable_warnings
+ALTER TABLE t1 IMPORT TABLESPACE;
+--enable_warnings
+SHOW CREATE TABLE t1;
+DROP TABLE t1;

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

+--source include/have_innodb.inc
+--source include/have_debug.inc
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+		f2 CHAR(2) NOT NULL, FULLTEXT f_idx(f2),
+		f3 INT as (f1) VIRTUAL, INDEX(f3))ENGINE=InnoDB;
+INSERT INTO t1(f1, f2) VALUES(1, "on");
+ALTER TABLE t1 DROP INDEX f_idx;
+FLUSH TABLE t1 FOR EXPORT;
+let MYSQLD_DATADIR =`SELECT @@datadir`;
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_backup_tablespaces("test", "t1");
+EOF
+UNLOCK TABLES;
+DROP TABLE t1;
+
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+		f2 CHAR(2) NOT NULL,
+		f3 CHAR(2) NOT NULL,
+		f4 INT AS (f1) VIRTUAL, INDEX(f4))ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_discard_tablespaces("test", "t1");
+ib_restore_tablespaces("test", "t1");
+EOF
+--error ER_TABLE_SCHEMA_MISMATCH
+ALTER TABLE t1 IMPORT TABLESPACE;
+DROP TABLE t1;
+
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+		f2 CHAR(2) NOT NULL,
+		f3 INT as (f1) VIRTUAL, INDEX(f3))ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_discard_tablespaces("test", "t1");
+ib_restore_tablespaces("test", "t1");
+EOF
+
+--disable_warnings
+SET DEBUG_DBUG="+d,ib_import_set_index_root_failure";
+--error ER_TOO_MANY_CONCURRENT_TRXS
+ALTER TABLE t1 IMPORT TABLESPACE;
+SET DEBUG_DBUG="-d,ib_import_set_index_root_failure";
+
+SET DEBUG_DBUG="+d,ib_import_vcol_update_fail";
+--error ER_DUP_KEY
+ALTER TABLE t1 IMPORT TABLESPACE;
+SET DEBUG_DBUG="-d,ib_import_vcol_update_fail";
+
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_discard_tablespaces("test", "t1");
+ib_restore_tablespaces("test", "t1");
+EOF
+
+SET DEBUG_DBUG="+d,ib_import_fts_error";
+--error ER_DUP_KEY
+ALTER TABLE t1 IMPORT TABLESPACE;
+SET DEBUG_DBUG="-d,ib_import_fts_error";
+--enable_warnings
+
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_unlink_tablespace("test", "t1");
+EOF
+
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_SYS_COLUMNS
+WHERE table_id IN (SELECT table_id FROM information_schema.innodb_sys_tables where name="test/t1");
+
+SHOW CREATE TABLE t1;
+DROP TABLE t1;
+
+# Manually add the FTS_DOC_ID Column with mismatched data type
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+		FTS_DOC_ID BIGINT SIGNED NOT NULL,
+		f2 CHAR(2) NOT NULL,
+		FULLTEXT f_idx(f2))ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1, 1, "on");
+ALTER TABLE t1 DROP INDEX f_idx;
+FLUSH TABLE t1 FOR EXPORT;
+let MYSQLD_DATADIR =`SELECT @@datadir`;
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_backup_tablespaces("test", "t1");
+EOF
+UNLOCK TABLES;
+DROP TABLE t1;
+
+CREATE TABLE t1(f1 INT NOT NULL PRIMARY KEY,
+		f2 CHAR(2) NOT NULL)ENGINE=InnoDB;
+ALTER TABLE t1 DISCARD TABLESPACE;
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/innodb-util.pl";
+ib_discard_tablespaces("test", "t1");
+ib_restore_tablespaces("test", "t1");
+EOF
+--error ER_TABLE_SCHEMA_MISMATCH
+ALTER TABLE t1 IMPORT TABLESPACE;
+DROP TABLE t1;

storage/innobase/row/row0import.cc

@@ -53,7 +53,7 @@ Created 2012-02-08 by Sunny Bains.
 #include "ha_innodb.h"
 
 #include "scope.h"
-
+#include "dict0crea.h"
 #include <vector>
 
 #ifdef HAVE_MY_AES_H
@@ -198,6 +198,60 @@ struct row_import {
 
 	dberr_t match_flags(THD *thd) const ;
 
+	ulint find_fts_idx_offset() const
+	{
+	  for (ulint i= 0; i < m_n_indexes; i++)
+	  {
+            const char* index_name=
+              reinterpret_cast<const char*>(m_indexes[i].m_name);
+	    if (!strcmp(index_name, FTS_DOC_ID_INDEX_NAME))
+              return i;
+	  }
+	  return ULINT_UNDEFINED;
+	}
+
+        const row_index_t *find_index_by_name(const char *name) const
+	{
+	  for (ulint i= 0; i < m_n_indexes; i++)
+	  {
+            const char* index_name=
+              reinterpret_cast<const char*>(m_indexes[i].m_name);
+	    if (!strcmp(index_name, name))
+              return &m_indexes[i];
+	  }
+	  return nullptr;
+	}
+
+	/** @return whether cfg file has FTS_DOC_ID
+	& FTS_DOC_ID_INDEX*/
+	bool has_hidden_fts() const
+	{
+          if (m_missing) return false;
+          ulint col_offset= find_col(FTS_DOC_ID_COL_NAME);
+	  if (col_offset == ULINT_UNDEFINED) return false;
+
+          const dict_col_t *col= &m_cols[col_offset];
+	  if (col->mtype != DATA_INT
+              || (col->prtype & ~(DATA_NOT_NULL
+		                  | DATA_UNSIGNED | DATA_BINARY_TYPE
+				  | DATA_FTS_DOC_ID))
+	      || col->len != sizeof(doc_id_t))
+            return false;
+
+	  return find_index_by_name(FTS_DOC_ID_INDEX_NAME) != nullptr;
+	}
+
+        /** Need to check whether the table need to add system
+        generated fts column and system generated fts document index
+        @param table table to be imported
+        @return whether the table has to add system generated
+        fts column and fts index */
+        bool need_hidden_fts(dict_table_t *table) const
+        {
+          return has_hidden_fts() && !table->fts_doc_id_index &&
+		 m_n_cols == static_cast<ulint>(table->n_cols + 1) &&
+                 m_n_indexes == UT_LIST_GET_LEN(table->indexes) + 1;
+        }
 
 	dict_table_t*	m_table;		/*!< Table instance */
 
@@ -1092,7 +1146,6 @@ row_import::find_col(
 			return(i);
 		}
 	}
-
 	return(ULINT_UNDEFINED);
 }
 
@@ -2115,14 +2168,30 @@ dberr_t PageConverter::operator()(buf_block_t* block) UNIV_NOTHROW
 	return DB_SUCCESS;
 }
 
-/*****************************************************************//**
-Clean up after import tablespace. */
-static	MY_ATTRIBUTE((nonnull, warn_unused_result))
+static void reload_fts_table(row_prebuilt_t *prebuilt,
+                             dict_table_t* table)
+{
+  ut_ad(prebuilt->table != table);
+  /* Reload the table in case of hidden fts column */
+  const table_id_t id= prebuilt->table->id;
+  prebuilt->table->release();
+  dict_sys.remove(prebuilt->table);
+  prebuilt->table=
+    dict_table_open_on_id(id, true, DICT_TABLE_OP_NORMAL);
+  prebuilt->table->space= table->space;
+}
+
+/** Clean up after import tablespace.
+@param  prebuilt  prebuilt from handler
+@param  err       error code
+@param  fts_table constructed table which has system generated
+                  fulltext document id
+@return error code or DB_SUCCESS */
+static
 dberr_t
-row_import_cleanup(
-/*===============*/
-	row_prebuilt_t*	prebuilt,	/*!< in/out: prebuilt from handler */
-	dberr_t		err)		/*!< in: error code */
+row_import_cleanup(row_prebuilt_t* prebuilt,
+                   dberr_t         err,
+                   dict_table_t*   fts_table = nullptr)
 {
 	if (err != DB_SUCCESS) {
 		dict_table_t* table = prebuilt->table;
@@ -2142,11 +2211,44 @@ row_import_cleanup(
 		     index = UT_LIST_GET_NEXT(indexes, index)) {
 			index->page = FIL_NULL;
 		}
-	}
 
+		prebuilt->trx->rollback();
+	}
+	else {
 		DBUG_EXECUTE_IF("ib_import_before_commit_crash", DBUG_SUICIDE(););
-
 		prebuilt->trx->commit();
+	}
+
+	if (fts_table && fts_table != prebuilt->table) {
+
+		if (err == DB_SUCCESS) {
+			reload_fts_table(prebuilt, fts_table);
+			ib::warn() << "Added system generated FTS_DOC_ID "
+				   "and FTS_DOC_ID_INDEX while importing "
+				   "the tablespace " << prebuilt->table->name;
+		} else if (fts_table->space) {
+			fil_close_tablespace(fts_table->space_id);
+			fts_table->space = NULL;
+		}
+
+		if (!prebuilt->trx->dict_operation_lock_mode) {
+			dict_sys.lock(SRW_LOCK_CALL);
+		}
+
+		dict_index_t* index = UT_LIST_GET_FIRST(
+					fts_table->indexes);
+		while (index) {
+			dict_index_t* next_index =
+				UT_LIST_GET_NEXT(indexes, index);
+			dict_index_remove_from_cache(fts_table, index);
+			index = next_index;
+		}
+		dict_mem_table_free(fts_table);
+
+		if (!prebuilt->trx->dict_operation_lock_mode) {
+			dict_sys.unlock();
+		}
+	}
 
 	if (prebuilt->trx->dict_operation_lock_mode) {
 		row_mysql_unlock_data_dictionary(prebuilt->trx);
@@ -2159,14 +2261,17 @@ row_import_cleanup(
 	return(err);
 }
 
-/*****************************************************************//**
-Report error during tablespace import. */
-static	MY_ATTRIBUTE((nonnull, warn_unused_result))
+/** Report error during tablespace import.
+@param  prebuilt  prebuilt from the handler
+@param  err       error code
+@param  fts_table table definition containing hidden FTS_DOC_ID column
+@return error code or DB_SUCCESS */
+static
 dberr_t
 row_import_error(
-/*=============*/
-	row_prebuilt_t*	prebuilt,	/*!< in/out: prebuilt from handler */
-	dberr_t		err)		/*!< in: error code */
+	row_prebuilt_t*	prebuilt,
+	dberr_t		err,
+	dict_table_t*	fts_table=nullptr)
 {
 	if (!trx_is_interrupted(prebuilt->trx)) {
 		char	table_name[MAX_FULL_NAME_LEN + 1];
@@ -2181,7 +2286,7 @@ row_import_error(
 			table_name, (ulong) err, ut_strerr(err));
 	}
 
-	return row_import_cleanup(prebuilt, err);
+	return row_import_cleanup(prebuilt, err, fts_table);
 }
 
 /*****************************************************************//**
@@ -3075,6 +3180,138 @@ static size_t get_buf_size()
       ;
 }
 
+/** Add fts index to the table
+@param table fts index to be added on the table */
+static void add_fts_index(dict_table_t *table)
+{
+  dict_index_t *fts_index= dict_mem_index_create(
+    table, FTS_DOC_ID_INDEX_NAME, DICT_UNIQUE, 2);
+  fts_index->page= FIL_NULL;
+  fts_index->cached= 1;
+  fts_index->n_uniq= 1;
+  /* Add fields for FTS_DOC_ID_INDEX */
+  dict_index_add_col(
+    fts_index, table,
+    &table->cols[table->n_cols - (DATA_N_SYS_COLS + 1)], 0);
+  dict_index_t *clust_index= UT_LIST_GET_FIRST(table->indexes);
+  for (ulint i= 0; i < clust_index->n_uniq; i++)
+    dict_index_add_col(fts_index, table, clust_index->fields[i].col,
+                       clust_index->fields[i].prefix_len);
+#ifdef BTR_CUR_HASH_ADAPT
+  fts_index->search_info= btr_search_info_create(fts_index->heap);
+  fts_index->search_info->ref_count= 0;
+#endif /* BTR_CUR_HASH_ADAPT */
+  UT_LIST_ADD_LAST(fts_index->table->indexes, fts_index);
+}
+
+/** Append the hidden fts column and fts doc index to the
+existing table
+@param  table  table to be imported
+@param  thd    thread
+@param  cfg    metadata required by import
+@return table which has fts doc id and fts doc id index */
+static dict_table_t *build_fts_hidden_table(
+  dict_table_t *table, const row_import &cfg)
+{
+  dict_table_t *new_table= dict_table_t::create(
+    {table->name.m_name, strlen(table->name.m_name)},
+    table->space, table->n_t_cols - (DATA_N_SYS_COLS - 1),
+    table->n_v_cols, table->flags,
+    table->flags2);
+
+  new_table->id= table->id;
+  new_table->space_id= table->space_id;
+  const char* col_name= &table->col_names[0];
+  /* Copy columns from old table to new fts table */
+  for (ulint new_i= 0;
+       new_i < ulint(new_table->n_cols - (DATA_N_SYS_COLS + 1));
+       new_i++)
+  {
+    dict_mem_table_add_col(new_table, new_table->heap, col_name,
+			   table->cols[new_i].mtype,
+			   table->cols[new_i].prtype,
+			   table->cols[new_i].len);
+    col_name+= strlen(col_name) + 1;
+  }
+
+  unsigned fts_col_ind= unsigned(table->n_cols - DATA_N_SYS_COLS);
+  fts_add_doc_id_column(new_table, new_table->heap);
+  new_table->cols[fts_col_ind].ind=
+    fts_col_ind & dict_index_t::MAX_N_FIELDS;
+  new_table->cols[fts_col_ind].ord_part= 1;
+  dict_table_add_system_columns(new_table, new_table->heap);
+
+  col_name= &table->v_col_names[0];
+  for (ulint new_i= 0; new_i < new_table->n_v_cols; new_i++)
+  {
+    dict_col_t old_vcol= table->v_cols[new_i].m_col;
+    dict_mem_table_add_v_col(new_table, new_table->heap, col_name,
+                             old_vcol.mtype, old_vcol.prtype,
+                             old_vcol.len, old_vcol.ind + 1,
+                             table->v_cols[new_i].num_base);
+    for (ulint i= 0; i < table->v_cols[new_i].num_base; i++)
+    {
+      dict_col_t *base_col= dict_table_get_nth_col(
+        new_table, table->v_cols[new_i].base_col[i]->ind);
+      new_table->v_cols[new_i].base_col[i]= base_col;
+    }
+    col_name+= strlen(col_name) + 1;
+  }
+
+  bool is_clustered= true;
+  /* Copy indexes from old table to new table */
+  for (dict_index_t *old_index= UT_LIST_GET_FIRST(table->indexes);
+       old_index; is_clustered= false)
+  {
+    dict_index_t *new_index= dict_mem_index_create(
+      new_table, old_index->name, old_index->type,
+      old_index->n_fields + is_clustered);
+
+    new_index->id= old_index->id;
+    new_index->n_uniq= old_index->n_uniq;
+    new_index->type= old_index->type;
+    new_index->cached= 1;
+    new_index->n_user_defined_cols= old_index->n_user_defined_cols;
+    new_index->n_core_null_bytes= old_index->n_core_null_bytes;
+    /* Copy all fields from old index to new index */
+    for (ulint i= 0; i < old_index->n_fields; i++)
+    {
+      dict_field_t *field= dict_index_get_nth_field(old_index, i);
+      dict_col_t *col= field->col;
+      if (col->is_virtual())
+      {
+        dict_v_col_t *v_col= reinterpret_cast<dict_v_col_t*>(col);
+        col= &new_table->v_cols[v_col->v_pos].m_col;
+      }
+      else
+      {
+        unsigned ind= field->col->ind;
+        if (ind >= fts_col_ind) ind++;
+        col= &new_table->cols[ind];
+      }
+      dict_index_add_col(new_index, new_table, col,
+                         field->prefix_len);
+      if (i < old_index->n_uniq) col->ord_part= 1;
+    }
+
+    if (is_clustered)
+    {
+      /* Add fts doc id in clustered index */
+      dict_index_add_col(
+        new_index, new_table, &table->cols[fts_col_ind], 0);
+      new_index->fields[old_index->n_fields].fixed_len= sizeof(doc_id_t);
+    }
+
+    new_index->search_info= old_index->search_info;
+    UT_LIST_ADD_LAST(new_index->table->indexes, new_index);
+    old_index= UT_LIST_GET_NEXT(indexes, old_index);
+    if (UT_LIST_GET_LEN(new_table->indexes)
+        == cfg.find_fts_idx_offset())
+      add_fts_index(new_table);
+  }
+  return new_table;
+}
+
 /* find, parse instant metadata, performing variaous checks,
 and apply it to dict_table_t
 @return DB_SUCCESS or some error */
@@ -4279,6 +4516,76 @@ static void row_import_autoinc(dict_table_t *table, row_prebuilt_t *prebuilt,
   }
 }
 
+/** Update the virtual column position in SYS_COLUMNS and SYS_VIRTUAL
+@param  table_id   table identifier
+@param  new_pos    position value
+@param  trx        transaction
+@return DB_SUCCESS or error code */
+dberr_t update_vcol_pos(ulint table_id, ulint new_pos, trx_t *trx)
+{
+  pars_info_t *info= pars_info_create();
+  pars_info_add_ull_literal(info, "id", table_id);
+  pars_info_add_int4_literal(info, "old_pos", new_pos - 1);
+  DBUG_EXECUTE_IF("ib_import_vcol_update_fail",
+                  return DB_DUPLICATE_KEY;);
+  return que_eval_sql(info,
+                      "PROCEDURE UPDATE_VCOL () IS\n"
+                      "BEGIN\n"
+		      "UPDATE SYS_COLUMNS SET POS = POS + 1 "
+		      "WHERE TABLE_ID= :id AND POS = :old_pos;\n"
+		      "UPDATE SYS_VIRTUAL SET POS = POS + 1 "
+		      "WHERE TABLE_ID= :id AND POS = :old_pos;\n"
+		      "END\n;", trx);
+}
+
+/**
+1) Update the position of the columns and
+2) Insert the hidden fts doc id in the sys columns table
+3) Insert the hidden fts doc id in the sys indexes and
+sys_fields table
+@param  table   table to be imported
+@param  fts_pos position of fts doc id column
+@param  trx     transaction
+@return DB_SUCCESS or error code */
+static
+dberr_t innodb_insert_hidden_fts_col(dict_table_t* table,
+                                     ulint  fts_pos,
+                                     trx_t* trx)
+{
+  dict_index_t* fts_idx=
+    dict_table_get_index_on_name(table, FTS_DOC_ID_INDEX_NAME);
+  if (!fts_idx) return DB_ERROR;
+  for (ulint new_i= 0; new_i < table->n_v_cols; new_i++)
+  {
+    ulint pos= dict_create_v_col_pos(
+                 table->v_cols[new_i].v_pos,
+		 table->v_cols[new_i].m_col.ind);
+    if (dberr_t err= update_vcol_pos(table->id, pos, trx))
+      return err;
+  }
+  pars_info_t *info= pars_info_create();
+  pars_info_add_ull_literal(info, "id", table->id);
+  dict_hdr_get_new_id(NULL, &fts_idx->id, NULL);
+  pars_info_add_ull_literal(info, "idx_id", fts_idx->id);
+  pars_info_add_int4_literal(info, "pos", fts_pos);
+  pars_info_add_int4_literal(info, "space", fts_idx->table->space_id);
+  pars_info_add_int4_literal(info, "page_no", fts_idx->page);
+
+  return que_eval_sql(info,
+                      "PROCEDURE ADD_FTS_COL () IS\n"
+		      "BEGIN\n"
+		      "INSERT INTO SYS_COLUMNS VALUES"
+                      "(:id,:pos,'FTS_DOC_ID',6, 1795, 8, 0);\n"
+                      "UPDATE SYS_TABLES SET N_COLS = N_COLS + 1"
+		      " WHERE ID = :id;\n"
+		      "INSERT INTO SYS_INDEXES VALUES"
+		      "(:id, :idx_id, 'FTS_DOC_ID_INDEX', 1,"
+		      " 2, :space, :page_no, 50);\n"
+                      "INSERT INTO SYS_FIELDS VALUES"
+		      "(:idx_id, 1, 'FTS_DOC_ID');\n"
+		      "END;\n", trx);
+}
+
 /*****************************************************************//**
 Imports a tablespace. The space id in the .ibd file must match the space id
 of the table in the data dictionary.
@@ -4304,6 +4611,7 @@ row_import_for_mysql(
 	ut_ad(trx);
 	ut_ad(trx->state == TRX_STATE_ACTIVE);
 	ut_ad(!table->is_readable());
+	ut_ad(prebuilt->table == table);
 
 	ibuf_delete_for_discarded_space(table->space_id);
 
@@ -4349,7 +4657,6 @@ row_import_for_mysql(
 
 	row_import	cfg;
 	THD* thd = trx->mysql_thd;
-
 	err = row_import_read_cfg(table, thd, cfg);
 
 	/* Check if the table column definitions match the contents
@@ -4357,8 +4664,16 @@ row_import_for_mysql(
 
 	if (err == DB_SUCCESS) {
 
-		if (dberr_t err = handle_instant_metadata(table, cfg)) {
-			return row_import_error(prebuilt, err);
+		if (cfg.need_hidden_fts(table)) {
+			cfg.m_table = table = build_fts_hidden_table(
+						table, cfg);
+		}
+
+		err = handle_instant_metadata(table, cfg);
+		if (err != DB_SUCCESS) {
+import_error:
+			return row_import_error(
+					prebuilt, err, table);
 		}
 
 		/* We have a schema file, try and match it with our
@@ -4394,7 +4709,7 @@ row_import_for_mysql(
 				"table %s when .cfg file is missing.",
 				table->name.m_name);
 			err = DB_ERROR;
-			return row_import_error(prebuilt, err);
+			goto import_error;
 		}
 
 		FetchIndexRootPages	fetchIndexRootPages(table, trx);
@@ -4423,7 +4738,7 @@ row_import_for_mysql(
 	}
 
 	if (err != DB_SUCCESS) {
-		return row_import_error(prebuilt, err);
+		goto import_error;
 	}
 
 	trx->op_info = "importing tablespace";
@@ -4459,7 +4774,7 @@ row_import_for_mysql(
 				table_name, ut_strerr(err));
 		}
 
-		return row_import_cleanup(prebuilt, err);
+		goto import_error;
 	}
 
 	/* If the table is stored in a remote tablespace, we need to
@@ -4524,7 +4839,8 @@ row_import_for_mysql(
 	dict_index_t*	index = dict_table_get_first_index(table);
 
 	if (!dict_index_is_clust(index)) {
-		return row_import_error(prebuilt, DB_CORRUPTION);
+		err = DB_CORRUPTION;
+		goto import_error;
 	}
 
 	/* Update the Btree segment headers for index node and
@@ -4536,7 +4852,7 @@ row_import_for_mysql(
 			err = DB_CORRUPTION;);
 
 	if (err != DB_SUCCESS) {
-		return row_import_error(prebuilt, err);
+		goto import_error;
 	} else if (cfg.requires_purge(index->name)) {
 
 		/* Purge any delete-marked records that couldn't be
@@ -4555,7 +4871,7 @@ row_import_for_mysql(
 	DBUG_EXECUTE_IF("ib_import_cluster_failure", err = DB_CORRUPTION;);
 
 	if (err != DB_SUCCESS) {
-		return row_import_error(prebuilt, err);
+		goto import_error;
 	}
 
 	/* For secondary indexes, purge any records that couldn't be purged
@@ -4568,7 +4884,7 @@ row_import_for_mysql(
 			err = DB_CORRUPTION;);
 
 	if (err != DB_SUCCESS) {
-		return row_import_error(prebuilt, err);
+		goto import_error;
 	}
 
 	/* Ensure that the next available DB_ROW_ID is not smaller than
@@ -4583,9 +4899,9 @@ row_import_for_mysql(
 	/* Ensure that all pages dirtied during the IMPORT make it to disk.
 	The only dirty pages generated should be from the pessimistic purge
 	of delete marked records that couldn't be purged in Phase I. */
-	while (buf_flush_list_space(prebuilt->table->space));
+	while (buf_flush_list_space(table->space));
 
-	for (ulint count = 0; prebuilt->table->space->referenced(); count++) {
+	for (ulint count = 0; table->space->referenced(); count++) {
 		/* Issue a warning every 10.24 seconds, starting after
 		2.56 seconds */
 		if ((count & 511) == 128) {
@@ -4596,24 +4912,40 @@ row_import_for_mysql(
 	}
 
 	ib::info() << "Phase IV - Flush complete";
-	prebuilt->table->space->set_imported();
+	/* Set tablespace purpose as FIL_TYPE_TABLESPACE,
+	so that rollback can go ahead smoothly */
+	table->space->set_imported();
 
+	err = lock_sys_tables(trx);
+	if (err != DB_SUCCESS) {
+		goto import_error;
+	}
 	/* The dictionary latches will be released in in row_import_cleanup()
 	after the transaction commit, for both success and error. */
 
 	row_mysql_lock_data_dictionary(trx);
 
+	if (prebuilt->table != table) {
+		/* Add fts_doc_id and fts_doc_idx in data dictionary */
+		err = innodb_insert_hidden_fts_col(
+			table, cfg.find_col(FTS_DOC_ID_COL_NAME), trx);
+                DBUG_EXECUTE_IF("ib_import_fts_error",
+				err= DB_DUPLICATE_KEY;);
+		if (err != DB_SUCCESS) {
+			goto import_error;
+		}
+	}
 	/* Update the root pages of the table's indexes. */
 	err = row_import_update_index_root(trx, table, false);
 
 	if (err != DB_SUCCESS) {
-		return row_import_error(prebuilt, err);
+		goto import_error;
 	}
 
 	err = row_import_update_discarded_flag(trx, table->id, false);
 
 	if (err != DB_SUCCESS) {
-		return row_import_error(prebuilt, err);
+		goto import_error;
 	}
 
 	table->file_unreadable = false;
@@ -4623,5 +4955,5 @@ row_import_for_mysql(
 	Otherwise, read the PAGE_ROOT_AUTO_INC and set it to table autoinc. */
 	row_import_autoinc(table, prebuilt, autoinc);
 
-	return row_import_cleanup(prebuilt, err);
+	return row_import_cleanup(prebuilt, err, table);
 }