MDEV-15855 Deadlock between purge thread and DDL statementbb-10.2-MDEV-15855

Problem:
=========
Truncate operation holds MDL LOCK on the table (t1) and tries to
acquire InnoDB dict_operation_lock. Purge holds InnoDB dict_operation_lock
and tries to acquire MDL LOCK on the table (t1) to calculate the virtual column
value. It leads to deadlock of purge and truncate table (DDL)

Solution:
=========
If purge tries to acquire MDL lock on the table then it should do the following:

i) Purge should release all innodb locks (including dict_operation_lock) before
acquiring metadata lock on the table.

ii) After acquiring metadata lock on the table, it should check whether the
table is dropped or renamed. If the table is dropped then purge should ignore
the record. If the table is renamed then it should release the old table mdl lock
and should acquire the new table mdl lock.

iii) Once purge acquires mdl lock, it should use the server table for all the
remaining virtual index for the purge record.

- Introduce new virtual column information in purge_node_t to know whether
the mdl lock acquired successfully.

16 files changed, 743 insertions, 141 deletions

diff --git a/mysql-test/suite/innodb/r/virtual_purge.result b/mysql-test/suite/innodb/r/virtual_purge.result
new file mode 100644
index 00000000000..e7af64289ea
--- /dev/null
+++ b/mysql-test/suite/innodb/r/virtual_purge.result
@@ -0,0 +1,35 @@
+SET @saved_frequency = @@GLOBAL.innodb_purge_rseg_truncate_frequency;
+SET GLOBAL innodb_purge_rseg_truncate_frequency = 1;
+InnoDB		0 transactions not purged
+CREATE TABLE t1 (y YEAR, vy YEAR AS (y) VIRTUAL, pk INT,
+PRIMARY KEY(pk), UNIQUE(vy)) ENGINE=InnoDB;
+INSERT INTO t1 (pk,y) VALUES (1,2022);
+CREATE TABLE t2(f1 INT NOT NULL, PRIMARY KEY(f1))ENGINE=InnoDB;
+set global debug_dbug = 'd,ib_purge_virtual_index_callback';
+BEGIN;
+INSERT INTO t2(f1) VALUES(1);
+connect  con1, localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+COMMIT;
+connect con2,localhost,root,,;
+REPLACE INTO t1(pk, y) SELECT pk,y FROM t1;
+SET DEBUG_SYNC='row_trunc_before_dict_lock SIGNAL commit_trx WAIT_FOR release_lock';
+TRUNCATE TABLE t1;
+connection con1;
+SET DEBUG_SYNC='now WAIT_FOR commit_trx';
+commit;
+SET DEBUG_SYNC='now SIGNAL purge_start';
+disconnect con1;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR purge_start';
+InnoDB		1 transactions not purged
+SET DEBUG_SYNC='now SIGNAL release_lock';
+connection con2;
+disconnect con2;
+connection default;
+InnoDB		0 transactions not purged
+drop table t1, t2;
+SET GLOBAL innodb_purge_rseg_truncate_frequency = @saved_frequency;
+SET DEBUG_SYNC='RESET';
+set global debug_dbug = 'RESET';
diff --git a/mysql-test/suite/innodb/t/virtual_purge.test b/mysql-test/suite/innodb/t/virtual_purge.test
new file mode 100644
index 00000000000..e3a1c105c6d
--- /dev/null
+++ b/mysql-test/suite/innodb/t/virtual_purge.test
@@ -0,0 +1,54 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+
+# Ensure that the history list length will actually be decremented by purge.
+SET @saved_frequency = @@GLOBAL.innodb_purge_rseg_truncate_frequency;
+SET GLOBAL innodb_purge_rseg_truncate_frequency = 1;
+
+--source include/wait_all_purged.inc
+
+CREATE TABLE t1 (y YEAR, vy YEAR AS (y) VIRTUAL, pk INT,
+                 PRIMARY KEY(pk), UNIQUE(vy)) ENGINE=InnoDB;
+
+INSERT INTO t1 (pk,y) VALUES (1,2022);
+CREATE TABLE t2(f1 INT NOT NULL, PRIMARY KEY(f1))ENGINE=InnoDB;
+
+set global debug_dbug = 'd,ib_purge_virtual_index_callback';
+
+BEGIN;
+INSERT INTO t2(f1) VALUES(1);
+
+connect (con1, localhost,root,,);
+# Stop the purge thread
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+connection default;
+COMMIT;
+
+connect(con2,localhost,root,,);
+REPLACE INTO t1(pk, y) SELECT pk,y FROM t1;
+SET DEBUG_SYNC='row_trunc_before_dict_lock SIGNAL commit_trx WAIT_FOR release_lock';
+--send TRUNCATE TABLE t1
+
+connection con1;
+SET DEBUG_SYNC='now WAIT_FOR commit_trx';
+commit;
+SET DEBUG_SYNC='now SIGNAL purge_start';
+disconnect con1;
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR purge_start';
+--source include/wait_all_purged.inc
+SET DEBUG_SYNC='now SIGNAL release_lock';
+
+connection con2;
+reap;
+disconnect con2;
+
+connection default;
+--source include/wait_all_purged.inc
+drop table t1, t2;
+SET GLOBAL innodb_purge_rseg_truncate_frequency = @saved_frequency;
+SET DEBUG_SYNC='RESET';
+set global debug_dbug = 'RESET';
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index a1010fb4c35..b62d0675145 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -67,6 +67,7 @@
 #include "wsrep_thd.h"
 #include "sql_connect.h"
 #include "my_atomic.h"
+#include "sql_base.h"
 
 #ifdef HAVE_SYS_SYSCALL_H
 #include <sys/syscall.h>
@@ -4411,7 +4412,6 @@ TABLE *get_purge_table(THD *thd)
   return thd->open_tables;
 }
 
-
 /** Find an open table in the list of prelocked tabled
 
   Used for foreign key actions, for example, in UPDATE t1 SET a=1;
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 0fd453a37ad..8f359e8ca84 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -36,6 +36,7 @@ Created 1/8/1996 Heikki Tuuri
 #include "fts0fts.h"
 #include "fil0fil.h"
 #include <algorithm>
+#include "sql_table.h"
 
 /** dummy index for ROW_FORMAT=REDUNDANT supremum and infimum records */
 dict_index_t*	dict_ind_redundant;
@@ -7365,3 +7366,48 @@ dict_table_extent_size(
 
 	return(pages_in_extent);
 }
+
+/** Parse the table file name into table name and database name.
+@param[in]	tbl_name	InnoDB table name
+@param[out]	mysql_db_name	database name buffer
+@param[out]	mysql_tbl_name	table name buffer
+@param[out]	dbnamelen	database name length
+@param[out]	tblnamelen	table name length
+@return true if the table name is parsed properly. */
+bool dict_parse_tbl_name(
+	const char*	tbl_name,
+	char		*mysql_db_name,
+	char		*mysql_tbl_name,
+	uint		&mysql_db_len,
+	uint		&mysql_tbl_len)
+{
+	mysql_db_len = dict_get_db_name_len(tbl_name);
+	char db_buf[MAX_DATABASE_NAME_LEN  + 1];
+	char tbl_buf[MAX_TABLE_NAME_LEN + 1];
+
+	ut_ad(mysql_db_len > 0);
+	ut_ad(mysql_db_len <= MAX_DATABASE_NAME_LEN);
+
+	memcpy(db_buf, tbl_name, mysql_db_len);
+	db_buf[mysql_db_len] = 0;
+
+	mysql_tbl_len = strlen(tbl_name) - mysql_db_len - 1;
+	memcpy(tbl_buf, tbl_name + mysql_db_len + 1, mysql_tbl_len);
+	tbl_buf[mysql_tbl_len] = 0;
+
+	filename_to_tablename(db_buf, mysql_db_name,
+			      MAX_DATABASE_NAME_LEN + 1, true);
+
+	if (mysql_tbl_len > TEMP_FILE_PREFIX_LENGTH
+	    && !strncmp(tbl_buf, TEMP_FILE_PREFIX, TEMP_FILE_PREFIX_LENGTH)) {
+		return false;
+	}
+
+	if (char *is_part = strchr(tbl_buf, '#')) {
+		*is_part = '\0';
+	}
+
+	filename_to_tablename(tbl_buf, mysql_tbl_name,
+			      MAX_TABLE_NAME_LEN + 1, true);
+	return true;
+}
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 7cba4ef5794..ef37b20c7d3 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -130,6 +130,7 @@ void reset_thd(MYSQL_THD thd);
 TABLE *open_purge_table(THD *thd, const char *db, size_t dblen,
 			const char *tb, size_t tblen);
 TABLE *get_purge_table(THD *thd);
+void close_thread_tables(THD* thd);
 
 /** Check if user has used xtradb extended system variable that
 is not currently supported by innodb or marked as deprecated. */
@@ -21603,63 +21604,126 @@ innobase_index_cond(
 	return handler_index_cond_check(file);
 }
 
+/** Acquire mdl lock for the mysql table for the given innodb table.
+@param[in,out]	thd	mysql thread handle
+@param[in,out]	table	InnoDB table
+@return MariaDB table handle
+@return NULL if the table is dropped or unaccessible or corrupted. */
+static TABLE* innobase_acquire_mdl_lock(
+	THD*		thd,
+	dict_table_t*	table)
+{
+	table_id_t	table_id = table->id;
+	ulint		db_len = dict_get_db_name_len(table->name.m_name);
+
+	ut_ad(db_len > 0);
+
+	TABLE*	mariadb_table;
+	char	db_buf[NAME_LEN + 1], db_buf1[NAME_LEN + 1];
+	char	tbl_buf[NAME_LEN + 1], tbl_buf1[NAME_LEN + 1];
+	uint	db_buf_len, db_buf1_len;
+	uint	tbl_buf_len, tbl_buf1_len;
+	bool	mdl_acquire = false;
+
+	if (!dict_parse_tbl_name(table->name.m_name, db_buf, tbl_buf,
+				 db_buf_len, tbl_buf_len)) {
+		ut_ad(0);
+		return NULL;
+	}
+
+retry_mdl:
+	const bool	unaccessible = !table->is_readable() || table->corrupted;
+	table->release();
+
+	if (unaccessible) {
+fail:
+		if (mdl_acquire) {
+			close_thread_tables(thd);
+		}
+
+		return NULL;
+	}
+
+	mariadb_table = open_purge_table(thd, db_buf, db_buf_len,
+					 tbl_buf, tbl_buf_len);
+
+	if (mariadb_table != NULL) {
+		mdl_acquire = true;
+	}
+
+	table = dict_table_open_on_id(table_id, false, DICT_TABLE_OP_NORMAL);
+
+	if (table == NULL) {
+		/* Table is dropped. */
+		goto fail;
+	}
+
+	if (!fil_table_accessible(table)) {
+		table->release();
+		goto fail;
+	}
+
+	dict_parse_tbl_name(table->name.m_name, db_buf1, tbl_buf1,
+			    db_buf1_len, tbl_buf1_len);
+
+	if (mdl_acquire && strcmp(db_buf, db_buf1) == 0
+	    && strcmp(tbl_buf, tbl_buf1) == 0) {
+		return mariadb_table;
+	}
+
+	/* Table is renamed. So release mdl lock for old name and try
+	to acquire the mdl for new table name. */
+	if (mdl_acquire) {
+		close_thread_tables(thd);
+	}
+
+	strcpy(tbl_buf, tbl_buf1);
+	strcpy(db_buf, db_buf1);
+	tbl_buf_len = tbl_buf1_len;
+	db_buf_len = db_buf1_len;
+	goto retry_mdl;
+}
 
 /** Find or open a mysql table for the virtual column template
 @param[in]	thd	mysql thread handle
-@param[in,out]	table	InnoDB table whose virtual column template is to be updated
-@return TABLE if successful or NULL */
+@param[in,out]	table	InnoDB table whose virtual column template
+			is to be updated
+@return TABLE if successful fetched
+@return NULL if the table is dropped, unaccessible or corrupted
+for purge thread */
 static TABLE *
 innobase_find_mysql_table_for_vc(
-/*=============================*/
 	THD*		thd,
 	dict_table_t*	table)
 {
-	TABLE *mysql_table;
-	bool	bg_thread = THDVAR(thd, background_thread);
+	bool	purge_thread = THDVAR(thd, background_thread);
+
+	if (purge_thread) {
+		/** Purge thread acquires dict_operation_lock while
+		processing undo log record. Release the dict_operation_lock
+		before acquiring MDL lock on the table. */
+		rw_lock_s_unlock(dict_operation_lock);
+
+		return innobase_acquire_mdl_lock(thd, table);
 
-	if (bg_thread) {
-		if ((mysql_table = get_purge_table(thd))) {
-			return mysql_table;
-		}
 	} else {
-		if (table->vc_templ->mysql_table_query_id == thd_get_query_id(thd)) {
+		if (table->vc_templ->mysql_table_query_id
+		    == thd_get_query_id(thd)) {
 			return table->vc_templ->mysql_table;
 		}
 	}
 
-	char	dbname[MAX_DATABASE_NAME_LEN + 1];
-	char	tbname[MAX_TABLE_NAME_LEN + 1];
-	char*	name = table->name.m_name;
-	uint	dbnamelen = (uint) dict_get_db_name_len(name);
-	uint	tbnamelen = (uint) strlen(name) - dbnamelen - 1;
-	char	t_dbname[MAX_DATABASE_NAME_LEN + 1];
-	char	t_tbname[MAX_TABLE_NAME_LEN + 1];
-
-	strncpy(dbname, name, dbnamelen);
-	dbname[dbnamelen] = 0;
-	strncpy(tbname, name + dbnamelen + 1, tbnamelen);
-	tbname[tbnamelen] =0;
-
-	/* For partition table, remove the partition name and use the
-	"main" table name to build the template */
-	 char*	is_part = is_partition(tbname);
-
-	if (is_part != NULL) {
-		*is_part = '\0';
-	}
+	char	db_buf[NAME_LEN + 1];
+	char	tbl_buf[NAME_LEN + 1];
+	uint	db_buf_len, tbl_buf_len;
 
-	dbnamelen = filename_to_tablename(dbname, t_dbname,
-					  MAX_DATABASE_NAME_LEN + 1);
-	tbnamelen = filename_to_tablename(tbname, t_tbname,
-					  MAX_TABLE_NAME_LEN + 1);
+	TABLE*	mysql_table;
 
-	if (bg_thread) {
-		return open_purge_table(thd, t_dbname, dbnamelen,
-					t_tbname, tbnamelen);
-	}
+	dict_parse_tbl_name(table->name.m_name, db_buf, tbl_buf,
+			    db_buf_len, tbl_buf_len);
 
-	mysql_table = find_fk_open_table(thd, t_dbname, dbnamelen,
-					t_tbname, tbnamelen);
+	mysql_table = find_fk_open_table(thd, db_buf, db_buf_len,
+					 tbl_buf, tbl_buf_len);
 
 	table->vc_templ->mysql_table = mysql_table;
 	table->vc_templ->mysql_table_query_id = thd_get_query_id(thd);
@@ -21780,7 +21844,8 @@ innobase_get_field_from_update_vector(
 @param[out]	storage         Internal storage for blobs etc
 
 @return         FALSE ok
-@return         TRUE malloc failure
+@return         TRUE malloc failure or failed to open the maria table
+		for purge thread.
 */
 
 bool innobase_allocate_row_for_vcol(
@@ -21795,6 +21860,12 @@ bool innobase_allocate_row_for_vcol(
 	String *blob_value_storage;
 	if (!*table)
 		*table= innobase_find_mysql_table_for_vc(thd, index->table);
+
+	/* For purge thread, there is a possiblity that table could have
+	dropped, corrupted or unaccessible. In that case, maria_table
+	can be NULL */
+	if (!*table)
+		return true;
 	maria_table= *table;
 	if (!*heap && !(*heap= mem_heap_create(srv_page_size)))
 	{
diff --git a/storage/innobase/include/btr0btr.h b/storage/innobase/include/btr0btr.h
index 1d7710a1496..2fccdfc431c 100644
--- a/storage/innobase/include/btr0btr.h
+++ b/storage/innobase/include/btr0btr.h
@@ -115,7 +115,15 @@ enum btr_latch_mode {
 	/** Attempt to purge a secondary index record
 	while holding the dict_index_t::lock S-latch. */
 	BTR_PURGE_LEAF_ALREADY_S_LATCHED = BTR_PURGE_LEAF
-	| BTR_ALREADY_S_LATCHED
+	| BTR_ALREADY_S_LATCHED,
+
+	/** In the case of BTR_MODIFY_TREE, the caller specifies
+	the intention to delete record only. It is used to optimize
+	block->lock range.*/
+	BTR_LATCH_FOR_DELETE = 65536,
+
+	/** Attempt to purge a secondary index record in the tree. */
+	BTR_PURGE_TREE = BTR_MODIFY_TREE | BTR_LATCH_FOR_DELETE
 };
 
 /** This flag ORed to btr_latch_mode says that we do the search in query
@@ -131,10 +139,6 @@ the insert buffer to speed up inserts */
 to insert record only. It is used to optimize block->lock range.*/
 #define BTR_LATCH_FOR_INSERT	32768U
 
-/** In the case of BTR_MODIFY_TREE, the caller specifies the intention
-to delete record only. It is used to optimize block->lock range.*/
-#define BTR_LATCH_FOR_DELETE	65536U
-
 /** This flag is for undo insert of rtree. For rtree, we need this flag
 to find proper rec to undo insert.*/
 #define BTR_RTREE_UNDO_INS	131072U
diff --git a/storage/innobase/include/btr0pcur.h b/storage/innobase/include/btr0pcur.h
index fab934ca0ee..f9f132b1733 100644
--- a/storage/innobase/include/btr0pcur.h
+++ b/storage/innobase/include/btr0pcur.h
@@ -296,6 +296,21 @@ btr_pcur_commit_specify_mtr(
 /*========================*/
 	btr_pcur_t*	pcur,	/*!< in: persistent cursor */
 	mtr_t*		mtr);	/*!< in: mtr to commit */
+
+/** Commits the mtr and sets the clustered index pcur and secondary index
+pcur latch mode to BTR_NO_LATCHES, that is, the cursor becomes detached.
+Function btr_pcur_store_position should be used for both cursor before
+calling this, if restoration of cursor is wanted later.
+@param[in]	pcur		persistent cursor
+@param[in]	sec_pcur	secondary index persistent cursor
+@param[in]	mtr		mtr to commit */
+UNIV_INLINE
+void
+btr_pcurs_commit_specify_mtr(
+	btr_pcur_t*	pcur,
+	btr_pcur_t*	sec_pcur,
+	mtr_t*		mtr);
+
 /*********************************************************//**
 Moves the persistent cursor to the next record in the tree. If no records are
 left, the cursor stays 'after last in tree'.
diff --git a/storage/innobase/include/btr0pcur.ic b/storage/innobase/include/btr0pcur.ic
index 4490942a2bb..b2a85def63d 100644
--- a/storage/innobase/include/btr0pcur.ic
+++ b/storage/innobase/include/btr0pcur.ic
@@ -389,6 +389,32 @@ btr_pcur_commit_specify_mtr(
 	pcur->pos_state = BTR_PCUR_WAS_POSITIONED;
 }
 
+/** Commits the mtr and sets the clustered index pcur and secondary index
+pcur latch mode to BTR_NO_LATCHES, that is, the cursor becomes detached.
+Function btr_pcur_store_position should be used for both cursor before
+calling this, if restoration of cursor is wanted later.
+@param[in]	pcur		persistent cursor
+@param[in]	sec_pcur	secondary index persistent cursor
+@param[in]	mtr		mtr to commit */
+UNIV_INLINE
+void
+btr_pcurs_commit_specify_mtr(
+	btr_pcur_t*	pcur,
+	btr_pcur_t*	sec_pcur,
+	mtr_t*		mtr)
+{
+	ut_ad(pcur->pos_state == BTR_PCUR_IS_POSITIONED);
+	ut_ad(sec_pcur->pos_state == BTR_PCUR_IS_POSITIONED);
+
+	pcur->latch_mode = BTR_NO_LATCHES;
+	sec_pcur->latch_mode = BTR_NO_LATCHES;
+
+	mtr_commit(mtr);
+
+	pcur->pos_state = BTR_PCUR_WAS_POSITIONED;
+	sec_pcur->pos_state = BTR_PCUR_WAS_POSITIONED;
+}
+
 /**************************************************************//**
 Sets the old_rec_buf field to NULL. */
 UNIV_INLINE
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index 9822b57d190..3fae2e1d602 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -2022,6 +2022,20 @@ bool
 dict_table_have_virtual_index(
 	dict_table_t*	table);
 
+/** Parse the table file name into table name and database name.
+@param[in]	tbl_name	InnoDB table name
+@param[out]	mysql_db_name	database name buffer
+@param[out]	mysql_tbl_name	table name buffer
+@param[out]	mysql_db_len	database name length
+@param[out]	mysql_tbl_len	table name length
+@return true if the table name is parsed properly. */
+bool dict_parse_tbl_name(
+	const char*	tbl_name,
+	char*		mysql_db_name,
+	char*		mysql_tbl_name,
+	uint		&mysql_db_len,
+	uint		&mysql_tbl_len);
+
 #include "dict0dict.ic"
 
 #endif
diff --git a/storage/innobase/include/row0purge.h b/storage/innobase/include/row0purge.h
index 3cf4a7b982a..7c63317d53c 100644
--- a/storage/innobase/include/row0purge.h
+++ b/storage/innobase/include/row0purge.h
@@ -36,6 +36,7 @@ Created 3/14/1997 Heikki Tuuri
 #include "que0types.h"
 #include "row0types.h"
 #include "ut0vec.h"
+#include "row0mysql.h"
 
 /** Create a purge node to a query graph.
 @param[in]	parent	parent node, i.e., a thr node
@@ -47,8 +48,7 @@ row_purge_node_create(
 	mem_heap_t*	heap)
 	MY_ATTRIBUTE((warn_unused_result));
 
-/***********************************************************//**
-Determines if it is possible to remove a secondary index entry.
+/** Determines if it is possible to remove a secondary index entry.
 Removal is possible if the secondary index entry does not refer to any
 not delete marked version of a clustered index record where DB_TRX_ID
 is newer than the purge view.
@@ -61,14 +61,23 @@ inserts a record that the secondary index entry would refer to.
 However, in that case, the user transaction would also re-insert the
 secondary index entry after purge has removed it and released the leaf
 page latch.
+@param[in,out]	node		row purge node
+@param[in]	index		secondary index
+@param[in]	entry		secondary index entry
+@param[in,out]	sec_pcur	secondary index cursor
+@param[in,out]	sec_mtr		mini-transaction which holds
+				secondary index entry
+@param[in]	is_tree		purge checking for tree mode
 @return true if the secondary index record can be purged */
 bool
 row_purge_poss_sec(
-/*===============*/
-	purge_node_t*	node,	/*!< in/out: row purge node */
-	dict_index_t*	index,	/*!< in: secondary index */
-	const dtuple_t*	entry)	/*!< in: secondary index entry */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
+	purge_node_t*	node,
+	dict_index_t*	index,
+	const dtuple_t*	entry,
+	btr_pcur_t*	sec_pcur=NULL,
+	mtr_t*		sec_mtr=NULL,
+	bool		is_tree=false);
+
 /***************************************************************
 Does the purge operation for a single undo log record. This is a high-level
 function used in an SQL execution graph.
@@ -117,6 +126,14 @@ struct purge_node_t{
 	ibool		done;	/* Debug flag */
 	trx_id_t	trx_id;	/*!< trx id for this purging record */
 
+	/** Heap to store virtual column information. It allocates heap when
+	it encounters first virtual index while processing. */
+	mem_heap_t*	vcol_heap;
+
+	/** Virtual column information about opening of MariaDB table.
+	It resets after processing each undo log record. */
+	purge_vcol_info_t*	vcol_info;
+
 #ifdef UNIV_DEBUG
 	/***********************************************************//**
 	Validate the persisent cursor. The purge node has two references
@@ -127,6 +144,16 @@ struct purge_node_t{
 	the ref member.*/
 	bool	validate_pcur();
 #endif
+
+	/** Check whether the virtual column information exists for
+	the purge node.
+	@return true if virtual column exists */
+	bool	is_vcol_info_exists() const;
+
+	/** Whether purge failed to open the maria table for virtual column
+	computation.
+	@return true if the table failed to open. */
+	bool	is_vcol_op_fail() const;
 };
 
 #endif
diff --git a/storage/innobase/include/row0types.h b/storage/innobase/include/row0types.h
index 52c89cb01fa..d6ed5b4e5f1 100644
--- a/storage/innobase/include/row0types.h
+++ b/storage/innobase/include/row0types.h
@@ -52,4 +52,69 @@ struct row_log_t;
 /* MySQL data types */
 struct TABLE;
 
+/** Purge virtual column node information. */
+struct purge_vcol_info_t {
+
+	/** Used for virtual column computation */
+	bool	use_vcol;
+
+	/** True if it is used for the first time. */
+        bool	use_first;
+
+	/** MariaDB table opened for virtual column computation. */
+	TABLE*	mariadb_table;
+
+	/** constructor to initialize the virtual column info. */
+	purge_vcol_info_t() {
+		use_first = use_vcol = false;
+		mariadb_table = NULL;
+	}
+
+	/** Check whether mariadb table exist
+	@return true if table exist. */
+	bool is_table_exists() const
+	{
+		return mariadb_table != NULL;
+	}
+
+	/** Check whether virtual column information is used
+	@return true if virtual column computation happened. */
+	bool is_vcol_uses() const
+	{
+		return use_vcol;
+	}
+
+	/** Validate the virtual column information.
+	@return true if the mariadb table opened successfully
+	or doesn't try to calculate virtual column. */
+	bool validate() const
+	{
+		if (!use_vcol) {
+			return true;
+		}
+
+		return mariadb_table != NULL;
+	}
+
+	/** Set the variable use_vcol and use_first as true if
+	it is used for the first time. */
+	void set_vcol_use()
+	{
+		if (use_first) {
+			use_first = false;
+			ut_ad(use_vcol);
+			return;
+		}
+
+		use_first = use_vcol = true;
+	}
+
+	/** Check whether it fetches mariadb table for the first time.
+	@return true if first time tries to open mariadb table. */
+	bool is_first_fetch() const
+	{
+		return use_first;
+	}
+};
+
 #endif
diff --git a/storage/innobase/include/row0vers.h b/storage/innobase/include/row0vers.h
index 576b53358f8..23c2e8546bc 100644
--- a/storage/innobase/include/row0vers.h
+++ b/storage/innobase/include/row0vers.h
@@ -35,6 +35,7 @@ Created 2/6/1997 Heikki Tuuri
 #include "rem0types.h"
 #include "mtr0mtr.h"
 #include "dict0mem.h"
+#include "row0types.h"
 
 // Forward declaration
 class ReadView;
@@ -68,27 +69,34 @@ row_vers_must_preserve_del_marked(
 	const table_name_t&	name,
 	mtr_t*			mtr);
 
-/*****************************************************************//**
-Finds out if a version of the record, where the version >= the current
+/** Finds out if a version of the record, where the version >= the current
 purge view, should have ientry as its secondary index entry. We check
 if there is any not delete marked version of the record where the trx
 id >= purge view, and the secondary index entry == ientry; exactly in
 this case we return TRUE.
+@param[in]	also_curr	TRUE if also rec is included in the versions
+				to search; otherwise only versions prior
+				to it are searched
+@param[in]	rec		record in the clustered index; the caller
+				must have a latch on the page
+@param[in]	mtr		mtr holding the latch on rec; it will
+				also hold the latch on purge_view
+@param[in]	index		secondary index
+@param[in]	ientry		secondary index entry
+@param[in]	roll_ptr	roll_ptr for the purge record
+@param[in]	trx_id		transaction ID on the purging record
+@param[in,out]	vcol_info	virtual column information for purge thread.
 @return TRUE if earlier version should have */
-ibool
+bool
 row_vers_old_has_index_entry(
-/*=========================*/
-	ibool		also_curr,/*!< in: TRUE if also rec is included in the
-				versions to search; otherwise only versions
-				prior to it are searched */
-	const rec_t*	rec,	/*!< in: record in the clustered index; the
-				caller must have a latch on the page */
-	mtr_t*		mtr,	/*!< in: mtr holding the latch on rec; it will
-				also hold the latch on purge_view */
-	dict_index_t*	index,	/*!< in: the secondary index */
-	const dtuple_t*	ientry,	/*!< in: the secondary index entry */
-	roll_ptr_t	roll_ptr,/*!< in: roll_ptr for the purge record */
-	trx_id_t	trx_id);/*!< in: transaction ID on the purging record */
+	bool			also_curr,
+	const rec_t*		rec,
+	mtr_t*			mtr,
+	dict_index_t*		index,
+	const dtuple_t*		ientry,
+	roll_ptr_t		roll_ptr,
+	trx_id_t		trx_id,
+	purge_vcol_info_t*	vcol_info=NULL);
 
 /*****************************************************************//**
 Constructs the version of a clustered index record which a consistent
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 448e2984b41..1fee1469ae8 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -78,10 +78,33 @@ row_purge_node_create(
 	node->common.parent = parent;
 	node->done = TRUE;
 	node->heap = mem_heap_create(256);
+	node->vcol_heap = NULL;
+	node->vcol_info = NULL;
 
 	return(node);
 }
 
+/** Create virtual column information node for the purge thread.
+@param[in,out]	node	creates virtual column information for
+			the purge node. */
+static void row_purge_vcol_create(purge_node_t* node)
+{
+	if (node->vcol_info != NULL) {
+		return;
+	}
+
+	if (node->vcol_heap == NULL) {
+		node->vcol_heap = mem_heap_create(128);
+	}
+
+	node->vcol_info = static_cast<purge_vcol_info_t*>(
+		mem_heap_alloc(node->vcol_heap, sizeof(purge_vcol_info_t*)));
+
+	node->vcol_info->use_vcol = false;
+	node->vcol_info->use_first = false;
+	node->vcol_info->mariadb_table = NULL;
+}
+
 /***********************************************************//**
 Repositions the pcur in the purge node on the clustered index record,
 if found. If the record is not found, close pcur.
@@ -136,7 +159,8 @@ row_purge_remove_clust_if_poss_low(
 	ulint			offsets_[REC_OFFS_NORMAL_SIZE];
 	rec_offs_init(offsets_);
 
-	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_S));
+	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_S)
+	      || node->vcol_info->is_vcol_uses());
 
 	index = dict_table_get_first_index(node->table);
 
@@ -230,8 +254,59 @@ row_purge_remove_clust_if_poss(
 	return(false);
 }
 
-/***********************************************************//**
-Determines if it is possible to remove a secondary index entry.
+/** Tries to store secondary index cursor before openin mysql table for
+virtual index condition computation.
+@param[in,out]	node		row purge node
+@param[in]	index		secondary index
+@param[in,out]	sec_pcur	secondary index cursor
+@param[in,out]	sec_mtr		mini-transaction which holds
+				secondary index entry */
+static void row_purge_store_vsec_cur(
+        purge_node_t*   node,
+        dict_index_t*   index,
+        btr_pcur_t*     sec_pcur,
+        mtr_t*          sec_mtr)
+{
+	row_purge_reposition_pcur(BTR_SEARCH_LEAF, node, sec_mtr);
+
+	if (!node->found_clust) {
+		return;
+	}
+
+	row_purge_vcol_create(node);
+
+	btr_pcur_store_position(sec_pcur, sec_mtr);
+
+	btr_pcurs_commit_specify_mtr(&node->pcur, sec_pcur, sec_mtr);
+}
+
+/** Tries to restore secondary index cursor after opening the mysql table
+@param[in,out]	node	row purge node
+@param[in]	index	secondary index
+@param[in,out]	sec_mtr	mini-transaction which holds secondary index entry
+@param[in]	is_tree	true=pessimistic purge,
+			false=optimistic (leaf-page only)
+@return false in case of restore failure. */
+static bool row_purge_restore_vsec_cur(
+	purge_node_t*	node,
+	dict_index_t*	index,
+	btr_pcur_t*	sec_pcur,
+	mtr_t*		sec_mtr,
+	bool		is_tree)
+{
+	mtr_start(sec_mtr);
+	sec_mtr->set_named_space(index->space);
+
+	if (btr_pcur_restore_position(
+		is_tree ? BTR_PURGE_TREE : BTR_PURGE_LEAF,
+		sec_pcur, sec_mtr)) {
+		return true;
+	}
+
+	return false;
+}
+
+/** Determines if it is possible to remove a secondary index entry.
 Removal is possible if the secondary index entry does not refer to any
 not delete marked version of a clustered index record where DB_TRX_ID
 is newer than the purge view.
@@ -244,25 +319,66 @@ inserts a record that the secondary index entry would refer to.
 However, in that case, the user transaction would also re-insert the
 secondary index entry after purge has removed it and released the leaf
 page latch.
+@param[in,out]	node		row purge node
+@param[in]	index		secondary index
+@param[in]	entry		secondary index entry
+@param[in,out]	sec_pcur	secondary index cursor or NULL
+				if it is called for purge buffering
+				operation.
+@param[in,out]	sec_mtr		mini-transaction which holds
+				secondary index entry or NULL if it is
+				called for purge buffering operation.
+@param[in]	is_tree		true=pessimistic purge,
+				false=optimistic (leaf-page only)
 @return true if the secondary index record can be purged */
 bool
 row_purge_poss_sec(
-/*===============*/
-	purge_node_t*	node,	/*!< in/out: row purge node */
-	dict_index_t*	index,	/*!< in: secondary index */
-	const dtuple_t*	entry)	/*!< in: secondary index entry */
+        purge_node_t*   node,
+	dict_index_t*	index,
+	const dtuple_t*	entry,
+	btr_pcur_t*	sec_pcur,
+	mtr_t*		sec_mtr,
+	bool		is_tree)
 {
 	bool	can_delete;
 	mtr_t	mtr;
+	bool	store_cur = false;
 
 	ut_ad(!dict_index_is_clust(index));
+
+	if (sec_mtr != NULL && dict_index_has_virtual(index)
+	    && !node->is_vcol_info_exists()) {
+		row_purge_store_vsec_cur(node, index, sec_pcur, sec_mtr);
+		store_cur = true;
+	}
+
+	/** Clustered index record not found. So it shouldn't be
+	purged. */
+	if (store_cur && !node->is_vcol_info_exists()) {
+		ut_ad(node->found_clust == 0);
+		return false;
+	}
+
+retry_purge_sec:
+
 	mtr_start(&mtr);
 
 	can_delete = !row_purge_reposition_pcur(BTR_SEARCH_LEAF, node, &mtr)
-		|| !row_vers_old_has_index_entry(TRUE,
+		|| !row_vers_old_has_index_entry(true,
 						 btr_pcur_get_rec(&node->pcur),
 						 &mtr, index, entry,
-						 node->roll_ptr, node->trx_id);
+						 node->roll_ptr, node->trx_id,
+						 node->vcol_info);
+
+	if (node->vcol_info != NULL && node->vcol_info->is_first_fetch()) {
+		if (node->vcol_info->is_table_exists()) {
+			goto retry_purge_sec;
+		}
+
+		node->table = NULL;
+		sec_pcur = NULL;
+		return false;
+	}
 
 	/* Persistent cursor is closed if reposition fails. */
 	if (node->found_clust) {
@@ -271,6 +387,14 @@ row_purge_poss_sec(
 		mtr_commit(&mtr);
 	}
 
+	if (sec_mtr != NULL
+	    && dict_index_has_virtual(index)
+	    && store_cur
+	    && !row_purge_restore_vsec_cur(
+			node, index, sec_pcur, sec_mtr, is_tree)) {
+		return false;
+	}
+
 	return(can_delete);
 }
 
@@ -287,7 +411,6 @@ row_purge_remove_sec_if_poss_tree(
 	const dtuple_t*	entry)	/*!< in: index entry */
 {
 	btr_pcur_t		pcur;
-	btr_cur_t*		btr_cur;
 	ibool			success	= TRUE;
 	dberr_t			err;
 	mtr_t			mtr;
@@ -348,16 +471,16 @@ row_purge_remove_sec_if_poss_tree(
 		ut_error;
 	}
 
-	btr_cur = btr_pcur_get_btr_cur(&pcur);
-
 	/* We should remove the index record if no later version of the row,
 	which cannot be purged yet, requires its existence. If some requires,
 	we should do nothing. */
 
-	if (row_purge_poss_sec(node, index, entry)) {
+	if (row_purge_poss_sec(node, index, entry, &pcur, &mtr, true)) {
+
 		/* Remove the index record, which should have been
 		marked for deletion. */
-		if (!rec_get_deleted_flag(btr_cur_get_rec(btr_cur),
+		if (!rec_get_deleted_flag(btr_cur_get_rec(
+						btr_pcur_get_btr_cur(&pcur)),
 					  dict_table_is_comp(index->table))) {
 			ib::error()
 				<< "tried to purge non-delete-marked record"
@@ -365,15 +488,18 @@ row_purge_remove_sec_if_poss_tree(
 				<< " of table " << index->table->name
 				<< ": tuple: " << *entry
 				<< ", record: " << rec_index_print(
-					btr_cur_get_rec(btr_cur), index);
+					btr_cur_get_rec(
+						btr_pcur_get_btr_cur(&pcur)),
+					index);
 
 			ut_ad(0);
 
 			goto func_exit;
 		}
 
-		btr_cur_pessimistic_delete(&err, FALSE, btr_cur, 0,
-					   false, &mtr);
+		btr_cur_pessimistic_delete(&err, FALSE,
+					   btr_pcur_get_btr_cur(&pcur),
+					   0, false, &mtr);
 		switch (UNIV_EXPECT(err, DB_SUCCESS)) {
 		case DB_SUCCESS:
 			break;
@@ -385,6 +511,10 @@ row_purge_remove_sec_if_poss_tree(
 		}
 	}
 
+	if (node->is_vcol_op_fail()) {
+		return false;
+	}
+
 func_exit:
 	btr_pcur_close(&pcur);
 func_exit_no_pcur:
@@ -445,8 +575,10 @@ row_purge_remove_sec_if_poss_leaf(
 		index->is_committed(). */
 		ut_ad(!dict_index_is_online_ddl(index));
 
-		/* Change buffering is disabled for spatial index. */
-		mode = dict_index_is_spatial(index)
+		/* Change buffering is disabled for spatial index and
+		virtual index. */
+		mode = (dict_index_is_spatial(index)
+			|| dict_index_has_virtual(index))
 			? BTR_MODIFY_LEAF
 			: BTR_PURGE_LEAF;
 	}
@@ -474,7 +606,7 @@ row_purge_remove_sec_if_poss_leaf(
 	case ROW_FOUND:
 		/* Before attempting to purge a record, check
 		if it is safe to do so. */
-		if (row_purge_poss_sec(node, index, entry)) {
+		if (row_purge_poss_sec(node, index, entry, &pcur, &mtr, false)) {
 			btr_cur_t* btr_cur = btr_pcur_get_btr_cur(&pcur);
 
 			/* Only delete-marked records should be purged. */
@@ -540,6 +672,11 @@ row_purge_remove_sec_if_poss_leaf(
 				success = false;
 			}
 		}
+
+		if (node->is_vcol_op_fail()) {
+			return false;
+		}
+
 		/* (The index entry is still needed,
 		or the deletion succeeded) */
 		/* fall through */
@@ -586,6 +723,10 @@ row_purge_remove_sec_if_poss(
 		return;
 	}
 retry:
+	if (node->is_vcol_op_fail()) {
+		return;
+	}
+
 	success = row_purge_remove_sec_if_poss_tree(node, index, entry);
 	/* The delete operation may fail if we have little
 	file space left: TODO: easiest to crash the database
@@ -652,6 +793,12 @@ row_purge_del_mark(
 				node->row, NULL, node->index,
 				heap, ROW_BUILD_FOR_PURGE);
 			row_purge_remove_sec_if_poss(node, node->index, entry);
+
+			if (node->is_vcol_op_fail()) {
+				mem_heap_free(heap);
+				return false;
+			}
+
 			mem_heap_empty(heap);
 		}
 
@@ -678,7 +825,8 @@ row_purge_upd_exist_or_extern_func(
 {
 	mem_heap_t*	heap;
 
-	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_S));
+	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_S)
+	      || node->vcol_info->is_vcol_uses());
 	ut_ad(!node->table->skip_alter_undo);
 
 	if (node->rec_type == TRX_UNDO_UPD_DEL_REC
@@ -1016,10 +1164,14 @@ row_purge(
 			bool purged = row_purge_record(
 				node, undo_rec, thr, updated_extern);
 
-			rw_lock_s_unlock(dict_operation_lock);
+			if (!node->is_vcol_info_exists()
+			    || !node->vcol_info->is_vcol_uses()) {
+				rw_lock_s_unlock(dict_operation_lock);
+			}
 
 			if (purged
-			    || srv_shutdown_state != SRV_SHUTDOWN_NONE) {
+			    || srv_shutdown_state != SRV_SHUTDOWN_NONE
+			    || node->is_vcol_op_fail()) {
 				return;
 			}
 
@@ -1051,8 +1203,15 @@ row_purge_end(
 
 	node->done = TRUE;
 
+	node->vcol_info = NULL;
+
 	ut_a(thr->run_node != NULL);
 
+	if (node->vcol_heap != NULL) {
+		mem_heap_free(node->vcol_heap);
+		node->vcol_heap = NULL;
+	}
+
 	mem_heap_empty(node->heap);
 }
 
@@ -1098,6 +1257,12 @@ row_purge_step(
 			row_purge_end(thr);
 		} else {
 			thr->run_node = node;
+
+			if (node->vcol_heap != NULL) {
+				mem_heap_empty(node->vcol_heap);
+			}
+
+			node->vcol_info = NULL;
 		}
 	} else {
 		row_purge_end(thr);
@@ -1157,3 +1322,18 @@ purge_node_t::validate_pcur()
 	return(true);
 }
 #endif /* UNIV_DEBUG */
+
+/** Check whether the virtual column information exist for the purge
+node.
+@return true if virtual column exist. */
+bool purge_node_t::is_vcol_info_exists() const
+{
+	return vcol_info != NULL;
+}
+
+/** Whether purge failed to open the maria table for virtual column computation.
+@return true if the table failed to open. */
+bool purge_node_t::is_vcol_op_fail() const
+{
+	return is_vcol_info_exists() && !vcol_info->validate();
+}
diff --git a/storage/innobase/row/row0trunc.cc b/storage/innobase/row/row0trunc.cc
index 94be5152596..f6be8bdfbc5 100644
--- a/storage/innobase/row/row0trunc.cc
+++ b/storage/innobase/row/row0trunc.cc
@@ -1810,6 +1810,8 @@ row_truncate_table_for_mysql(
 		trx_start_for_ddl(trx, TRX_DICT_OP_TABLE);
 	}
 
+	DEBUG_SYNC_C("row_trunc_before_dict_lock");
+
 	/* Step-3: Validate ownership of needed locks (Exclusive lock).
 	Ownership will also ensure there is no active SQL queries, INSERT,
 	SELECT, .....*/
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 6fc72f806e0..5f28f096286 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -508,7 +508,7 @@ row_undo_mod_del_mark_or_remove_sec_low(
 	any no-redo rollback segment undo logs. */
 	if (dict_table_is_temporary(node->table)
 	    || row_vers_old_has_index_entry(
-			FALSE, btr_pcur_get_rec(&(node->pcur)),
+			false, btr_pcur_get_rec(&(node->pcur)),
 			&mtr_vers, index, entry, 0, 0)) {
 		err = btr_cur_del_mark_set_sec_rec(BTR_NO_LOCKING_FLAG,
 						   btr_cur, TRUE, thr, &mtr);
diff --git a/storage/innobase/row/row0vers.cc b/storage/innobase/row/row0vers.cc
index aebcf70715f..0942d863fbb 100644
--- a/storage/innobase/row/row0vers.cc
+++ b/storage/innobase/row/row0vers.cc
@@ -436,14 +436,16 @@ row_vers_must_preserve_del_marked(
 @param[in,out]	row		the cluster index row in dtuple form
 @param[in]	clust_index	clustered index
 @param[in]	index		the secondary index
-@param[in]	heap		heap used to build virtual dtuple */
+@param[in]	heap		heap used to build virtual dtuple
+@param[in,out]	vcol_info	virtual column information. */
 static
 void
 row_vers_build_clust_v_col(
-	dtuple_t*	row,
-	dict_index_t*	clust_index,
-	dict_index_t*	index,
-	mem_heap_t*	heap)
+	dtuple_t*		row,
+	dict_index_t*		clust_index,
+	dict_index_t*		index,
+	mem_heap_t*		heap,
+	purge_vcol_info_t*	vcol_info)
 {
 	mem_heap_t*	local_heap = NULL;
 	VCOL_STORAGE	*vcol_storage= NULL;
@@ -453,12 +455,28 @@ row_vers_build_clust_v_col(
 
 	ut_ad(dict_index_has_virtual(index));
 
+	if (vcol_info != NULL) {
+		vcol_info->set_vcol_use();
+		maria_table = vcol_info->mariadb_table;
+	}
+
 	innobase_allocate_row_for_vcol(thd, index,
 				       &local_heap,
 				       &maria_table,
 				       &record,
 				       &vcol_storage);
 
+	if (vcol_info != NULL && vcol_info->mariadb_table == NULL) {
+
+		if (maria_table == NULL) {
+			return;
+		}
+
+		vcol_info->mariadb_table = maria_table;
+		ut_ad(vcol_info->is_first_fetch());
+		return;
+	}
+
 	for (ulint i = 0; i < dict_index_get_n_fields(index); i++) {
 		const dict_field_t* ind_field = dict_index_get_nth_field(
 				index, i);
@@ -497,16 +515,16 @@ row_vers_build_clust_v_col(
 static
 void
 row_vers_build_cur_vrow_low(
-	bool		in_purge,
-	const rec_t*	rec,
-	dict_index_t*	clust_index,
-	ulint*		clust_offsets,
-	dict_index_t*	index,
-	roll_ptr_t	roll_ptr,
-	trx_id_t	trx_id,
-	mem_heap_t*	v_heap,
-	const dtuple_t**vrow,
-	mtr_t*		mtr)
+	bool			in_purge,
+	const rec_t*		rec,
+	dict_index_t*		clust_index,
+	ulint*			clust_offsets,
+	dict_index_t*		index,
+	roll_ptr_t		roll_ptr,
+	trx_id_t		trx_id,
+	mem_heap_t*		v_heap,
+	const dtuple_t**	vrow,
+	mtr_t*			mtr)
 {
 	const rec_t*	version;
 	rec_t*		prev_version;
@@ -784,21 +802,23 @@ func_exit:
 @param[in,out]	heap		heap memory
 @param[in,out]	v_heap		heap memory to keep virtual colum dtuple
 @param[in]	mtr		mtr holding the latch on rec
+@param[in,out]	vcol_info	virtual column information for purge thread
 @return dtuple contains virtual column data */
 static
 const dtuple_t*
 row_vers_build_cur_vrow(
-	bool		in_purge,
-	const rec_t*	rec,
-	dict_index_t*	clust_index,
-	ulint**		clust_offsets,
-	dict_index_t*	index,
-	const dtuple_t*	ientry,
-	roll_ptr_t	roll_ptr,
-	trx_id_t	trx_id,
-	mem_heap_t*	heap,
-	mem_heap_t*	v_heap,
-	mtr_t*		mtr)
+	bool			in_purge,
+	const rec_t*		rec,
+	dict_index_t*		clust_index,
+	ulint**			clust_offsets,
+	dict_index_t*		index,
+	const dtuple_t*		ientry,
+	roll_ptr_t		roll_ptr,
+	trx_id_t		trx_id,
+	mem_heap_t*		heap,
+	mem_heap_t*		v_heap,
+	mtr_t*			mtr,
+	purge_vcol_info_t*	vcol_info)
 {
 	const dtuple_t*	cur_vrow = NULL;
 
@@ -818,8 +838,17 @@ row_vers_build_cur_vrow(
 					  rec, *clust_offsets,
 					  NULL, NULL, NULL, NULL, heap);
 
+		if (vcol_info != NULL && !vcol_info->is_vcol_uses()) {
+			mtr_commit(mtr);
+		}
+
 		row_vers_build_clust_v_col(
-			row, clust_index, index, heap);
+			row, clust_index, index, heap, vcol_info);
+
+		if (vcol_info != NULL && vcol_info->is_first_fetch()) {
+			return NULL;
+		}
+
 		cur_vrow = dtuple_copy(row, v_heap);
 		dtuple_dup_v_fld(cur_vrow, v_heap);
 	} else {
@@ -834,27 +863,34 @@ row_vers_build_cur_vrow(
 	return(cur_vrow);
 }
 
-/*****************************************************************//**
-Finds out if a version of the record, where the version >= the current
+/** Finds out if a version of the record, where the version >= the current
 purge view, should have ientry as its secondary index entry. We check
 if there is any not delete marked version of the record where the trx
-id >= purge view, and the secondary index entry and ientry are identified in
-the alphabetical ordering; exactly in this case we return TRUE.
+id >= purge view, and the secondary index entry == ientry; exactly in
+this case we return TRUE.
+@param[in]	also_curr	TRUE if also rec is included in the versions
+				to search; otherwise only versions prior
+				to it are searched
+@param[in]	rec		record in the clustered index; the caller
+				must have a latch on the page
+@param[in]	mtr		mtr holding the latch on rec; it will
+				also hold the latch on purge_view
+@param[in]	index		secondary index
+@param[in]	ientry		secondary index entry
+@param[in]	roll_ptr	roll_ptr for the purge record
+@param[in]	trx_id		transaction ID on the purging record
+@param[in,out]	vcol_info	virtual column information for purge thread.
 @return TRUE if earlier version should have */
-ibool
+bool
 row_vers_old_has_index_entry(
-/*=========================*/
-	ibool		also_curr,/*!< in: TRUE if also rec is included in the
-				versions to search; otherwise only versions
-				prior to it are searched */
-	const rec_t*	rec,	/*!< in: record in the clustered index; the
-				caller must have a latch on the page */
-	mtr_t*		mtr,	/*!< in: mtr holding the latch on rec; it will
-				also hold the latch on purge_view */
-	dict_index_t*	index,	/*!< in: the secondary index */
-	const dtuple_t*	ientry,	/*!< in: the secondary index entry */
-	roll_ptr_t	roll_ptr,/*!< in: roll_ptr for the purge record */
-	trx_id_t	trx_id)	/*!< in: transaction ID on the purging record */
+	bool			also_curr,
+	const rec_t*		rec,
+	mtr_t*			mtr,
+	dict_index_t*		index,
+	const dtuple_t*		ientry,
+	roll_ptr_t		roll_ptr,
+	trx_id_t		trx_id,
+	purge_vcol_info_t*	vcol_info)
 {
 	const rec_t*	version;
 	rec_t*		prev_version;
@@ -922,8 +958,21 @@ row_vers_old_has_index_entry(
 			columns need to be computed */
 			if (trx_undo_roll_ptr_is_insert(t_roll_ptr)
 			    || dbug_v_purge) {
+
+				if (vcol_info != NULL
+				    && !vcol_info->is_vcol_uses()) {
+					mtr_commit(mtr);
+				}
+
 				row_vers_build_clust_v_col(
-					row, clust_index, index, heap);
+					row, clust_index, index, heap,
+					vcol_info);
+
+				if (vcol_info != NULL
+				    && vcol_info->is_first_fetch()) {
+					ut_ad(vcol_info->is_vcol_uses());
+					return false;
+				}
 
 				entry = row_build_index_entry(
 					row, ext, index, heap);
@@ -988,7 +1037,7 @@ safe_to_purge:
 				if (v_heap) {
 					mem_heap_free(v_heap);
 				}
-				return(TRUE);
+				return true;
 			}
 		}
 	} else if (dict_index_has_virtual(index)) {
@@ -996,9 +1045,15 @@ safe_to_purge:
 		deleted, but the previous version of it might not. We will
 		need to get the virtual column data from undo record
 		associated with current cluster index */
+
 		cur_vrow = row_vers_build_cur_vrow(
 			also_curr, rec, clust_index, &clust_offsets,
-			index, ientry, roll_ptr, trx_id, heap, v_heap, mtr);
+			index, ientry, roll_ptr, trx_id, heap, v_heap, mtr,
+			vcol_info);
+
+		if (vcol_info != NULL && vcol_info->is_first_fetch()) {
+			return false;
+		}
 	}
 
 	version = rec;
@@ -1024,7 +1079,7 @@ safe_to_purge:
 				mem_heap_free(v_heap);
 			}
 
-			return(FALSE);
+			return false;
 		}
 
 		clust_offsets = rec_get_offsets(prev_version, clust_index,