MDEV-15855 Deadlock between purge thread and DDL statementbb-10.3-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.

- n_ref_count in dict_table_t does atomic operations for release, acquire,
get_ref_count()

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

18 files changed, 630 insertions, 100 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..282f64213cd
--- /dev/null
+++ b/mysql-test/suite/innodb/r/virtual_purge.result
@@ -0,0 +1,32 @@
+CREATE TABLE t1 (y YEAR, vy YEAR AS (y) VIRTUAL, pk INT,
+PRIMARY KEY(pk), UNIQUE(vy)) ENGINE=InnoDB;
+INSERT IGNORE INTO t1 (pk,y) VALUES (1,2022);
+CREATE TABLE t2(f1 INT NOT NULL, PRIMARY KEY(f1))ENGINE=InnoDB;
+BEGIN;
+INSERT INTO t2(f1) VALUES(1);
+connect  con1, localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+SET @saved_frequency = @@GLOBAL.innodb_purge_rseg_truncate_frequency;
+SET GLOBAL innodb_purge_rseg_truncate_frequency = 1;
+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		2 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';
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..3b7135b566e
--- /dev/null
+++ b/mysql-test/suite/innodb/t/virtual_purge.test
@@ -0,0 +1,48 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+--source include/have_metadata_lock_info.inc
+
+CREATE TABLE t1 (y YEAR, vy YEAR AS (y) VIRTUAL, pk INT,
+                 PRIMARY KEY(pk), UNIQUE(vy)) ENGINE=InnoDB;
+
+INSERT IGNORE INTO t1 (pk,y) VALUES (1,2022);
+CREATE TABLE t2(f1 INT NOT NULL, PRIMARY KEY(f1))ENGINE=InnoDB;
+BEGIN;
+INSERT INTO t2(f1) VALUES(1);
+
+connect (con1, localhost,root,,);
+# Stop the purge thread
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+connection default;
+# 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;
+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';
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index d2e4f66dd59..e886727fe46 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -4779,6 +4779,11 @@ TABLE *get_purge_table(THD *thd)
   return thd->open_tables;
 }
 
+/** Close the purge table for purge thread. */
+void close_purge_table(THD* thd)
+{
+	close_thread_tables(thd);
+}
 
 /** Find an open table in the list of prelocked tabled
 
diff --git a/storage/innobase/btr/btr0sea.cc b/storage/innobase/btr/btr0sea.cc
index 038e262cfad..c3b30e4d033 100644
--- a/storage/innobase/btr/btr0sea.cc
+++ b/storage/innobase/btr/btr0sea.cc
@@ -1307,7 +1307,7 @@ void btr_search_drop_page_hash_when_freed(const page_id_t& page_id)
 			/* In all our callers, the table handle should
 			be open, or we should be in the process of
 			dropping the table (preventing eviction). */
-			ut_ad(index->table->n_ref_count > 0
+			ut_ad(index->table->get_ref_count() > 0
 			      || mutex_own(&dict_sys->mutex));
 			btr_search_drop_page_hash_index(block);
 		}
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 9415624465f..89ec235a0a1 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;
@@ -1328,7 +1329,7 @@ static
 ibool
 dict_table_can_be_evicted(
 /*======================*/
-	const dict_table_t*	table)		/*!< in: table to test */
+	dict_table_t*	table)		/*!< in: table to test */
 {
 	ut_ad(mutex_own(&dict_sys->mutex));
 	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_X));
@@ -7257,3 +7258,45 @@ 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[in,out]  mysql_db_name   database name buffer
+@param[in,out]  mysql_tbl_name  table name buffer
+@return true if the table name is parsed properly. */
+bool dict_parse_tbl_name(
+	const char*	tbl_name,
+	char		(&mysql_db_name)[NAME_LEN + 1],
+	char		(&mysql_tbl_name)[NAME_LEN + 1])
+{
+	ulint 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(db_len > 0);
+	ut_ad(db_len <= MAX_DATABASE_NAME_LEN);
+
+	memcpy(db_buf, tbl_name, db_len);
+	db_buf[db_len] = 0;
+
+	size_t tbl_len = strlen(tbl_name) - db_len - 1;
+
+	memcpy(tbl_buf, tbl_name + db_len + 1, tbl_len);
+	tbl_buf[tbl_len] = 0;
+
+	filename_to_tablename(db_buf, mysql_db_name,
+			      MAX_DATABASE_NAME_LEN + 1, true);
+
+	if (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 66a28ea56b7..86a2d53e68a 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -129,7 +129,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_purge_table(THD *thd);
 #ifdef MYSQL_DYNAMIC_PLUGIN
 #define tc_size  400
 #define tdc_size 400
@@ -2967,7 +2967,7 @@ ha_innobase::update_thd(
 		   m_user_thd, thd));
 
 	/* The table should have been opened in ha_innobase::open(). */
-	ut_ad(m_prebuilt->table->n_ref_count > 0);
+	ut_ad(m_prebuilt->table->get_ref_count() > 0);
 
 	trx_t*	trx = check_trx_exists(thd);
 
@@ -13835,7 +13835,7 @@ ha_innobase::info_low(
 	m_prebuilt->trx->op_info = "returning various info to MariaDB";
 
 	ib_table = m_prebuilt->table;
-	ut_ad(ib_table->n_ref_count > 0);
+	ut_ad(ib_table->get_ref_count() > 0);
 
 	if (flag & HA_STATUS_TIME) {
 		if (is_analyze || innobase_stats_on_metadata) {
@@ -20536,6 +20536,92 @@ innobase_index_cond(
 	return handler_index_cond_check(file);
 }
 
+/** Acquire mdl lock for the mysql table for the given innodb table.
+@param[in]	thd	mysql thread handle
+@param[in]	table	InnoDB table
+@return TABLE if successful or NULL */
+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];
+	bool	mdl_acquire = false;
+	bool	unaccessible = false;
+
+	if (!dict_parse_tbl_name(table->name.m_name, db_buf, tbl_buf)) {
+		ut_ad(0);
+		return NULL;
+	}
+
+retry_mdl:
+
+	if (!unaccessible
+	    && (!table->is_readable() || table->corrupted)) {
+
+		if (mdl_acquire) {
+			close_purge_table(thd);
+		}
+
+		unaccessible = true;
+	}
+
+	table->release();
+
+	if (unaccessible) {
+		return NULL;
+	}
+
+	mariadb_table = open_purge_table(thd, db_buf, strlen(db_buf),
+					 tbl_buf, strlen(tbl_buf));
+
+	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. */
+		if (mdl_acquire) {
+			close_purge_table(thd);
+		}
+
+		return NULL;
+	}
+
+	if (!fil_table_accessible(table)) {
+		if (mdl_acquire) {
+			close_purge_table(thd);
+		}
+
+		unaccessible = true;
+		goto retry_mdl;
+	}
+
+	dict_parse_tbl_name(table->name.m_name, db_buf1, tbl_buf1);
+
+	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_purge_table(thd);
+	}
+
+	strcpy(tbl_buf, tbl_buf1);
+	strcpy(db_buf, db_buf1);
+	goto retry_mdl;
+}
 
 /** Find or open a mysql table for the virtual column template
 @param[in]	thd	mysql thread handle
@@ -20544,16 +20630,25 @@ innobase_index_cond(
 static TABLE *
 innobase_find_mysql_table_for_vc(
 /*=============================*/
-	THD*		thd,
-	dict_table_t*	table)
+	THD*			thd,
+	dict_table_t*		table,
+	purge_vcol_info_t*	purge_vcol=NULL)
 {
 	TABLE *mysql_table;
 	bool	bg_thread = THDVAR(thd, background_thread);
 
 	if (bg_thread) {
-		if ((mysql_table = get_purge_table(thd))) {
-			return mysql_table;
-		}
+
+		ut_ad(purge_vcol != NULL);
+
+		rw_lock_s_unlock(dict_operation_lock);
+
+		purge_vcol->use_vcol = true;
+		purge_vcol->mariadb_table = innobase_acquire_mdl_lock(
+						thd, table);
+
+		return purge_vcol->mariadb_table;
+
 	} else {
 		if (table->vc_templ->mysql_table_query_id == thd_get_query_id(thd)) {
 			return table->vc_templ->mysql_table;
@@ -20713,6 +20808,8 @@ innobase_get_field_from_update_vector(
 				or NULL.
 @param[in]	parent_update	update vector for the parent row
 @param[in]	foreign		foreign key information
+@param[in,out]	purge_vcol	virtual column information used by
+				purge thread
 @return the field filled with computed value, or NULL if just want
 to store the value in passed in "my_rec" */
 dfield_t*
@@ -20727,7 +20824,8 @@ innobase_get_computed_value(
 	TABLE*			mysql_table,
 	const dict_table_t*	old_table,
 	upd_t*			parent_update,
-	dict_foreign_t*		foreign)
+	dict_foreign_t*		foreign,
+	purge_vcol_info_t*	purge_vcol)
 {
 	byte		rec_buf2[REC_VERSION_56_MAX_INDEX_COL_LEN];
 	byte*		mysql_rec;
@@ -20760,8 +20858,17 @@ innobase_get_computed_value(
 		buf = rec_buf2;
 	}
 
+	if (purge_vcol != NULL && purge_vcol->use_vcol) {
+		mysql_table = purge_vcol->mariadb_table;
+	}
+
 	if (!mysql_table) {
-		mysql_table = innobase_find_mysql_table_for_vc(thd, index->table);
+		mysql_table = innobase_find_mysql_table_for_vc(thd, index->table,
+							       purge_vcol);
+	}
+
+	if (mysql_table == NULL && purge_vcol->use_vcol == true) {
+		return NULL;
 	}
 
 	ut_ad(mysql_table);
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index 66b64f96128..903a6afe51c 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -47,6 +47,7 @@ Created 1/8/1996 Heikki Tuuri
 #include <deque>
 #include "fsp0fsp.h"
 #include "dict0pagecompress.h"
+#include "mdl.h"
 
 extern bool innodb_table_stats_not_found;
 extern bool innodb_index_stats_not_found;
@@ -1930,6 +1931,16 @@ 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[in,out]	mysql_db_name	database name buffer
+@param[in,out]	mysql_tbl_name	table name buffer
+@return true if the table name is parsed properly. */
+bool dict_parse_tbl_name(
+	const char*	tbl_name,
+	char		(&mysql_db_name)[NAME_LEN + 1],
+	char		(&mysql_tbl_name)[NAME_LEN + 1]);
+
 #include "dict0dict.ic"
 
 #endif
diff --git a/storage/innobase/include/dict0dict.ic b/storage/innobase/include/dict0dict.ic
index db1be6513b1..29c807508c0 100644
--- a/storage/innobase/include/dict0dict.ic
+++ b/storage/innobase/include/dict0dict.ic
@@ -1405,10 +1405,9 @@ dict_table_is_file_per_table(
 @return current value of n_ref_count */
 inline
 ulint
-dict_table_t::get_ref_count() const
+dict_table_t::get_ref_count()
 {
-	ut_ad(mutex_own(&dict_sys->mutex));
-	return(n_ref_count);
+	return my_atomic_load32_explicit(&n_ref_count, MY_MEMORY_ORDER_RELAXED);
 }
 
 /** Acquire the table handle. */
@@ -1417,7 +1416,7 @@ void
 dict_table_t::acquire()
 {
 	ut_ad(mutex_own(&dict_sys->mutex));
-	++n_ref_count;
+	my_atomic_add32_explicit(&n_ref_count, 1, MY_MEMORY_ORDER_RELAXED);
 }
 
 /** Release the table handle.
@@ -1426,9 +1425,10 @@ inline
 bool
 dict_table_t::release()
 {
-	ut_ad(mutex_own(&dict_sys->mutex));
-	ut_ad(n_ref_count > 0);
-	return !--n_ref_count;
+	int32 n = my_atomic_add32_explicit(
+			&n_ref_count, -1, MY_MEMORY_ORDER_RELAXED);
+	ut_ad(n > 0);
+	return n == 1;
 }
 
 /** Encode the number of columns and number of virtual columns in a
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 154a503f1b3..e3e6a31a7c0 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -1481,7 +1481,7 @@ struct dict_table_t {
 
 	/** Get reference count.
 	@return current value of n_ref_count */
-	inline ulint get_ref_count() const;
+	inline ulint get_ref_count();
 
 	/** Acquire the table handle. */
 	inline void acquire();
@@ -1928,13 +1928,11 @@ struct dict_table_t {
 	It is protected by lock_sys.mutex. */
 	ulint					n_rec_locks;
 
-#ifndef DBUG_ASSERT_EXISTS
 private:
-#endif
 	/** Count of how many handles are opened to this table. Dropping of the
 	table is NOT allowed until this count gets to zero. MySQL does NOT
 	itself check the number of open handles at DROP. */
-	ulint					n_ref_count;
+	int32					n_ref_count;
 
 public:
 	/** List of locks on the table. Protected by lock_sys.mutex. */
diff --git a/storage/innobase/include/row0mysql.h b/storage/innobase/include/row0mysql.h
index 7c0b5d3ece9..b93dab3ab11 100644
--- a/storage/innobase/include/row0mysql.h
+++ b/storage/innobase/include/row0mysql.h
@@ -879,6 +879,38 @@ struct SysIndexCallback {
 	virtual void operator()(mtr_t* mtr, btr_pcur_t* pcur) throw() = 0;
 };
 
+/** Purge virtual column node information. */
+struct purge_vcol_info_t {
+
+	/** Used for virtual column computation. */
+	bool	use_vcol;
+	/** mariadb table opened for virtual column computation. */
+	TABLE*	mariadb_table;
+
+	/** Initialize the virtual column information. */
+	void initialize() {
+		use_vcol = false;
+		mariadb_table = NULL;
+	}
+
+	bool is_table_exists() {
+		return mariadb_table != NULL;
+	}
+
+	bool uses_vcol_info() {
+		return use_vcol == true;
+	}
+
+	/** Validate the virtual column information. */
+	bool validate() {
+		if (!use_vcol) {
+			return true;
+		}
+
+		return mariadb_table != NULL;
+	}
+};
+
 /** Get the computed value by supplying the base column values.
 @param[in,out]	row		the data row
 @param[in]	col		virtual column
@@ -892,6 +924,8 @@ struct SysIndexCallback {
 				or NULL.
 @param[in]	parent_update	update vector for the parent row
 @param[in]	foreign		foreign key information
+@param[in,out]	vcol_info	virtual column information used by
+				purge thread
 @return the field filled with computed value */
 dfield_t*
 innobase_get_computed_value(
@@ -905,7 +939,8 @@ innobase_get_computed_value(
 	TABLE*			mysql_table,
 	const dict_table_t*	old_table,
 	upd_t*			parent_update,
-	dict_foreign_t*		foreign);
+	dict_foreign_t*		foreign,
+	purge_vcol_info_t*	vcol_info=NULL);
 
 /** Get the computed value by supplying the base column values.
 @param[in,out]	table	the table whose virtual column template to be built */
diff --git a/storage/innobase/include/row0purge.h b/storage/innobase/include/row0purge.h
index e08ef8ab9f9..1d71bd0ed05 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,15 @@ struct purge_node_t{
 	ibool		done;	/* Debug flag */
 	trx_id_t	trx_id;	/*!< trx id for this purging record */
 
+	/** persistent cursor used in searching the virtual secondary
+	index record. */
+	btr_pcur_t	sec_pcur;
+	/** Heap to store virtual column information. */
+	mem_heap_t*	vcol_heap;
+
+	/** Virtual column information about opening of mariadb table. */
+	purge_vcol_info_t*	vcol_info;
+
 #ifdef UNIV_DEBUG
 	/***********************************************************//**
 	Validate the persisent cursor. The purge node has two references
@@ -127,6 +145,15 @@ 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_exist();
+
+	/** Validate the virtual column information for the purge node.
+	@return true if mariadb table opened successfully. */
+	bool	validate_vcol_info();
 };
 
 #endif
diff --git a/storage/innobase/include/row0vers.h b/storage/innobase/include/row0vers.h
index 749f42cbcf3..993932ea5a8 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 "row0mysql.h"
 
 // Forward declaration
 class ReadView;
@@ -54,27 +55,34 @@ row_vers_impl_x_locked(
 	dict_index_t*	index,
 	const ulint*	offsets);
 
-/*****************************************************************//**
-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.
+id >= purge view, and the secondary index entry and ientry are identified in
+the alphabetical ordering; 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]	vcol_info	virtual column information for purge thread.
 @return TRUE if earlier version should have */
 ibool
 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 */
+	ibool			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/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index b5494ef7313..5511940c466 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -1433,7 +1433,7 @@ lock_rec_create_low(
 	lock_rec_bitmap_reset(lock);
 	lock_rec_set_nth_bit(lock, heap_no);
 	index->table->n_rec_locks++;
-	ut_ad(index->table->n_ref_count > 0 || !index->table->can_be_evicted);
+	ut_ad(index->table->get_ref_count() > 0 || !index->table->can_be_evicted);
 
 #ifdef WITH_WSREP
 	if (c_lock && wsrep_on_trx(trx)
@@ -3531,7 +3531,7 @@ lock_table_create(
 
 	lock->un_member.tab_lock.table = table;
 
-	ut_ad(table->n_ref_count > 0 || !table->can_be_evicted);
+	ut_ad(table->get_ref_count() > 0 || !table->can_be_evicted);
 
 	UT_LIST_ADD_LAST(trx->lock.trx_locks, lock);
 
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 97b79b705b2..58d6032fe7a 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -3849,7 +3849,7 @@ row_ins_step(
 		/* No-rollback tables should only be written to by a
 		single thread at a time, but there can be multiple
 		concurrent readers. We must hold an open table handle. */
-		DBUG_ASSERT(node->table->n_ref_count > 0);
+		DBUG_ASSERT(node->table->get_ref_count() > 0);
 		DBUG_ASSERT(node->ins_type == INS_DIRECT);
 		/* No-rollback tables can consist only of a single index. */
 		DBUG_ASSERT(UT_LIST_GET_LEN(node->entry_list) == 1);
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index af8cd682288..e70da7e5501 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -78,10 +78,31 @@ 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]	node	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_zalloc(node->vcol_heap, sizeof(purge_vcol_info_t*)));
+
+	node->vcol_info->mariadb_table = NULL;
+	node->vcol_info->use_vcol = false;
+}
+
 /***********************************************************//**
 Repositions the pcur in the purge node on the clustered index record,
 if found. If the record is not found, close pcur.
@@ -136,7 +157,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->uses_vcol_info());
 
 	index = dict_table_get_first_index(node->table);
 
@@ -230,6 +252,71 @@ row_purge_remove_clust_if_poss(
 	return(false);
 }
 
+/** 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)
+{
+	if (!dict_index_has_virtual(index) || sec_mtr == NULL) {
+		return;
+	}
+
+	row_purge_vcol_create(node);
+
+	row_purge_reposition_pcur(BTR_SEARCH_LEAF, node, sec_mtr);
+
+        if (!node->found_clust) {
+		return;
+	}
+
+	btr_pcur_store_position(sec_pcur, sec_mtr);
+
+	node->sec_pcur = *sec_pcur;
+
+	node->pcur.latch_mode = BTR_NO_LATCHES;
+
+	btr_pcur_commit_specify_mtr(sec_pcur, sec_mtr);
+
+	node->pcur.pos_state = BTR_PCUR_WAS_POSITIONED;
+}
+
+/** 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	purge check happening for tree mode.
+@return restored cursor or NULL in case of failure. */
+static bool row_purge_restore_vsec_cur(
+	purge_node_t*	node,
+	dict_index_t*	index,
+	mtr_t*		sec_mtr,
+	bool		is_tree)
+{
+	if (!dict_index_has_virtual(index) || sec_mtr == NULL) {
+		return false;
+	}
+
+	sec_mtr->start();
+
+	index->set_modified(*sec_mtr);
+
+	if (btr_pcur_restore_position(is_tree ? BTR_MODIFY_TREE: BTR_PURGE_LEAF,
+				      &node->sec_pcur, sec_mtr)) {
+		return true;
+	}
+
+	btr_pcur_close(&node->sec_pcur);
+	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
@@ -244,25 +331,54 @@ 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 */
+	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;
+	bool	can_delete, use_vcol_first = false;
 	mtr_t	mtr;
 
 	ut_ad(!dict_index_is_clust(index));
+
+	row_purge_store_vsec_cur(node, index, sec_pcur, sec_mtr);
+
+retry_purge_sec:
+
+	use_vcol_first = node->is_vcol_info_exist()
+			 && !node->vcol_info->uses_vcol_info();
+
 	mtr_start(&mtr);
 
 	can_delete = !row_purge_reposition_pcur(BTR_SEARCH_LEAF, node, &mtr)
 		|| !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 (use_vcol_first && node->vcol_info->uses_vcol_info()) {
+
+		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,10 @@ row_purge_poss_sec(
 		mtr_commit(&mtr);
 	}
 
+	if (row_purge_restore_vsec_cur(node, index, sec_mtr, is_tree)) {
+		sec_pcur = &node->sec_pcur;
+	}
+
 	return(can_delete);
 }
 
@@ -354,7 +474,7 @@ row_purge_remove_sec_if_poss_tree(
 	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),
@@ -385,6 +505,10 @@ row_purge_remove_sec_if_poss_tree(
 		}
 	}
 
+	if (node->is_vcol_info_exist() && !node->validate_vcol_info()) {
+		return false;
+	}
+
 func_exit:
 	btr_pcur_close(&pcur);
 func_exit_no_pcur:
@@ -474,7 +598,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 +664,10 @@ row_purge_remove_sec_if_poss_leaf(
 				success = false;
 			}
 		}
+
+		if (node->is_vcol_info_exist() && !node->validate_vcol_info()) {
+			return false;
+		}
 		/* (The index entry is still needed,
 		or the deletion succeeded) */
 		/* fall through */
@@ -585,7 +713,12 @@ row_purge_remove_sec_if_poss(
 
 		return;
 	}
+
 retry:
+	if (node->is_vcol_info_exist() && !node->validate_vcol_info()) {
+		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 +785,13 @@ 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_info_exist()
+			    && !node->validate_vcol_info()) {
+				mem_heap_free(heap);
+				return false;
+			}
+
 			mem_heap_empty(heap);
 		}
 
@@ -671,7 +811,8 @@ static
 void
 row_purge_reset_trx_id(purge_node_t* node, mtr_t* mtr)
 {
-	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->uses_vcol_info());
 	/* Reset DB_TRX_ID, DB_ROLL_PTR for old records. */
 	mtr->start();
 
@@ -746,7 +887,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->uses_vcol_info());
 	ut_ad(!node->table->skip_alter_undo);
 
 	if (node->rec_type == TRX_UNDO_UPD_DEL_REC
@@ -1107,10 +1249,16 @@ 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_exist()
+			    || !node->vcol_info->uses_vcol_info()) {
+
+				rw_lock_s_unlock(dict_operation_lock);
+			}
 
 			if (purged
-			    || srv_shutdown_state != SRV_SHUTDOWN_NONE) {
+			    || srv_shutdown_state != SRV_SHUTDOWN_NONE
+			    || (node->is_vcol_info_exist()
+				&& !node->validate_vcol_info())) {
 				return;
 			}
 
@@ -1142,8 +1290,14 @@ row_purge_end(
 
 	node->done = TRUE;
 
+	node->vcol_info = NULL;
+
 	ut_a(thr->run_node != NULL);
 
+	if (node->vcol_heap != NULL) {
+		mem_heap_empty(node->vcol_heap);
+	}
+
 	mem_heap_empty(node->heap);
 }
 
@@ -1189,6 +1343,7 @@ row_purge_step(
 			row_purge_end(thr);
 		} else {
 			thr->run_node = node;
+			node->vcol_info = NULL;
 		}
 	} else {
 		row_purge_end(thr);
@@ -1248,3 +1403,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_exist()
+{
+	return vcol_info != NULL;
+}
+
+/** Validate the virtual column information stored in purge thread.
+@return true if purge node opens the mariadb table successfully. */
+bool purge_node_t::validate_vcol_info()
+{
+	return vcol_info->validate();
+}
diff --git a/storage/innobase/row/row0trunc.cc b/storage/innobase/row/row0trunc.cc
index 4e7b7bd9b73..9ac15e44bfe 100644
--- a/storage/innobase/row/row0trunc.cc
+++ b/storage/innobase/row/row0trunc.cc
@@ -1757,6 +1757,7 @@ row_truncate_table_for_mysql(
 		trx_set_dict_operation(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/row0upd.cc b/storage/innobase/row/row0upd.cc
index bca7464bc66..ed2b72587b7 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -2115,10 +2115,11 @@ row_upd_eval_new_vals(
 static
 void
 row_upd_store_v_row(
-	upd_node_t*	node,
-	const upd_t*	update,
-	THD*		thd,
-	TABLE*		mysql_table)
+	upd_node_t*		node,
+	const upd_t*		update,
+	THD*			thd,
+	TABLE*			mysql_table,
+	purge_vcol_info_t*	vcol_info=NULL)
 {
 	mem_heap_t*	heap = NULL;
 	dict_index_t*	index = dict_table_get_first_index(node->table);
@@ -2175,7 +2176,7 @@ row_upd_store_v_row(
 						node->row, col, index,
 						&heap, node->heap, NULL,
 						thd, mysql_table, NULL,
-						NULL, NULL);
+						NULL, NULL, vcol_info);
 				}
 			}
 		}
diff --git a/storage/innobase/row/row0vers.cc b/storage/innobase/row/row0vers.cc
index d585ef3a9d3..042c722e491 100644
--- a/storage/innobase/row/row0vers.cc
+++ b/storage/innobase/row/row0vers.cc
@@ -428,10 +428,11 @@ row_vers_impl_x_locked(
 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;
 	for (ulint i = 0; i < dict_index_get_n_fields(index); i++) {
@@ -447,7 +448,7 @@ row_vers_build_clust_v_col(
 			innobase_get_computed_value(
 				row, col, clust_index, &local_heap,
 				heap, NULL, current_thd, NULL, NULL,
-				NULL, NULL);
+				NULL, NULL, vcol_info);
 		}
 	}
 
@@ -759,16 +760,17 @@ func_exit:
 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,
-	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,
+	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;
 
@@ -787,9 +789,21 @@ row_vers_build_cur_vrow(
 		dtuple_t* row = row_build(ROW_COPY_POINTERS, clust_index,
 					  rec, *clust_offsets,
 					  NULL, NULL, NULL, NULL, heap);
+		bool	use_vcol_first = false;
+
+		if (vcol_info == NULL && !vcol_info->use_vcol) {
+			use_vcol_first = true;
+			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->validate() || use_vcol_first)) {
+			return NULL;
+		}
+
 		cur_vrow = dtuple_copy(row, v_heap);
 		dtuple_dup_v_fld(cur_vrow, v_heap);
 	} else {
@@ -804,27 +818,35 @@ 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.
+@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]	vcol_info	virtual column information for purge
+				thread.
 @return TRUE if earlier version should have */
 ibool
 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 */
+	ibool			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;
@@ -838,6 +860,7 @@ row_vers_old_has_index_entry(
 	const dtuple_t*	vrow = NULL;
 	mem_heap_t*	v_heap = NULL;
 	const dtuple_t*	cur_vrow = NULL;
+	bool		use_vcol_first = false;
 
 	ut_ad(mtr_memo_contains_page_flagged(mtr, rec, MTR_MEMO_PAGE_X_FIX
 					     | MTR_MEMO_PAGE_S_FIX));
@@ -892,8 +915,20 @@ 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->uses_vcol_info()) {
+					use_vcol_first = true;
+					mtr_commit(mtr);
+				}
+
 				row_vers_build_clust_v_col(
-					row, clust_index, index, heap);
+					row, clust_index, index, heap, vcol_info);
+
+				if (use_vcol_first) {
+					ut_ad(vcol_info->uses_vcol_info());
+					return false;
+				}
 
 				entry = row_build_index_entry(
 					row, ext, index, heap);
@@ -966,9 +1001,18 @@ 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 */
+
+		if (vcol_info != NULL && !vcol_info->uses_vcol_info()) {
+			use_vcol_first = true;
+		}
+
 		cur_vrow = row_vers_build_cur_vrow(
 			also_curr, rec, clust_index, &clust_offsets,
-			index, roll_ptr, trx_id, heap, v_heap, mtr);
+			index, roll_ptr, trx_id, heap, v_heap, mtr, vcol_info);
+
+		if (use_vcol_first && !vcol_info->validate()) {
+			return false;
+		}
 	}
 
 	version = rec;