MDEV-29181 Potential corruption on FK update on a table with vcol index

vc_templ->mysql_table concept is completely broken.
This table pointer persists with a global access and can be overwritten by
arbitrary thread, though mysql_table is thread-local data. Like it wasn't
enough, it also could become invalid after eviction from tc cache.

This new solution simply does the following:
* Sets up a referenced table list in TABLE instance (sql_base.cc)
* Iterates through it along with dict_table_t::referenced_set
  (row_upd_check_references_constraints)
* Passes corresponding prebuilt through a call chain to
  row_ins_foreign_check_on_constraint
* Sets up newly created upd_node_t::prebuilt field and uses it accordingly
  is cascade update.

mysql-test/suite/vcol/r/innodb_virtual_fk.result

@@ -10,3 +10,21 @@ select * from t2;
 id
 drop table t2;
 drop table t1;
+#
+# MDEV-29181 Potential corruption on FK update on a table with vcol index
+#
+connect  con2, localhost, root,,;
+connection default;
+create table t1 (a int primary key);
+insert into t1(a) select * from seq_1_to_2048;
+create table t2 (a int,  b int as (a), key(b), foreign key (a) references t1 (a)
+on update cascade);
+insert into t2(a) select * from seq_1_to_2048;
+update t1 set a = a - 100000 where a < 1024;
+connection con2;
+set lock_wait_timeout =1;
+update t1 set a= a + 1000000 where a > 1024;
+connection default;
+drop table t2, t1;
+set debug_sync= reset;
+disconnect con2;

mysql-test/suite/vcol/t/innodb_virtual_fk.test

@@ -1,4 +1,6 @@
 source include/have_innodb.inc;
+source include/have_sequence.inc;
+
 set default_storage_engine=innodb;
 
 #
@@ -14,3 +16,28 @@ select * from t1;
 select * from t2;
 drop table t2;
 drop table t1;
+
+--echo #
+--echo # MDEV-29181 Potential corruption on FK update on a table with vcol index
+--echo #
+
+--connect (con2, localhost, root,,)
+--connection default
+create table t1 (a int primary key);
+insert into t1(a) select * from seq_1_to_2048;
+create table t2 (a int,  b int as (a), key(b), foreign key (a) references t1 (a)
+  on update cascade);
+insert into t2(a) select * from seq_1_to_2048;
+
+send update t1 set a = a - 100000 where a < 1024;
+
+--connection con2
+set lock_wait_timeout =1;
+update t1 set a= a + 1000000 where a > 1024;
+
+--connection default
+--reap
+
+# Cleanup
+drop table t2, t1;
+--disconnect con2

sql/sql_base.cc

@@ -1786,6 +1786,13 @@ bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx)
     {
       table= best_table;
       table->query_id= thd->query_id;
+
+      if (thd->locked_tables_mode >= LTM_PRELOCKED
+          && table->file->referenced_by_foreign_key())
+      {
+        table_list->fk_ref_list= table->pos_in_table_list->fk_ref_list;
+      }
+
       table->init(thd, table_list);
       DBUG_PRINT("info",("Using locked table"));
 #ifdef WITH_PARTITION_STORAGE_ENGINE
@@ -4541,18 +4548,18 @@ handle_routine(THD *thd, Query_tables_list *prelocking_ctx,
   @note this can be changed to use a hash, instead of scanning the linked
   list, if the performance of this function will ever become an issue
 */
-bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_CSTRING *db,
-                                LEX_CSTRING *table, thr_lock_type lock_type)
+TABLE_LIST *find_fk_prelocked_table(TABLE_LIST *tl, LEX_CSTRING *db,
+                                    LEX_CSTRING *table, thr_lock_type lock_type)
 {
   for (; tl; tl= tl->next_global )
   {
     if (tl->lock_type >= lock_type &&
         tl->prelocking_placeholder == TABLE_LIST::PRELOCK_FK &&
         strcmp(tl->db.str, db->str) == 0 &&
         strcmp(tl->table_name.str, table->str) == 0)
-      return true;
+      return tl;
   }
-  return false;
+  return NULL;
 }
 
 
@@ -4634,15 +4641,14 @@ prepare_fk_prelocking_list(THD *thd, Query_tables_list *prelocking_ctx,
                            uint8 op)
 {
   DBUG_ENTER("prepare_fk_prelocking_list");
-  List <FOREIGN_KEY_INFO> fk_list;
-  List_iterator<FOREIGN_KEY_INFO> fk_list_it(fk_list);
   FOREIGN_KEY_INFO *fk;
   Query_arena *arena, backup;
   TABLE *table= table_list->table;
 
   arena= thd->activate_stmt_arena_if_needed(&backup);
 
-  table->file->get_parent_foreign_key_list(thd, &fk_list);
+  table_list->fk_ref_list= new(thd->mem_root) List<FOREIGN_KEY_INFO>();
+  table->file->get_parent_foreign_key_list(thd, table_list->fk_ref_list);
   if (unlikely(thd->is_error()))
   {
     if (arena)
@@ -4652,6 +4658,7 @@ prepare_fk_prelocking_list(THD *thd, Query_tables_list *prelocking_ctx,
 
   *need_prelocking= TRUE;
 
+  List_iterator<FOREIGN_KEY_INFO> fk_list_it(*table_list->fk_ref_list);
   while ((fk= fk_list_it++))
   {
     // FK_OPTION_RESTRICT and FK_OPTION_NO_ACTION only need read access
@@ -4663,19 +4670,22 @@ prepare_fk_prelocking_list(THD *thd, Query_tables_list *prelocking_ctx,
     else
       lock_type= TL_READ;
 
-    if (table_already_fk_prelocked(prelocking_ctx->query_tables,
-          fk->foreign_db, fk->foreign_table,
-          lock_type))
-      continue;
+    TABLE_LIST *tl= find_fk_prelocked_table(prelocking_ctx->query_tables,
+                                            fk->foreign_db, fk->foreign_table,
+                                            lock_type);
+    if (tl == NULL)
+    {
+      tl= (TABLE_LIST *) thd->alloc(sizeof(TABLE_LIST));
+      tl->init_one_table_for_prelocking(fk->foreign_db,
+          fk->foreign_table,
+          NULL, lock_type,
+          TABLE_LIST::PRELOCK_FK,
+          table_list->belong_to_view, op,
+          &prelocking_ctx->query_tables_last,
+          table_list->for_insert_data);
+    }
 
-    TABLE_LIST *tl= (TABLE_LIST *) thd->alloc(sizeof(TABLE_LIST));
-    tl->init_one_table_for_prelocking(fk->foreign_db,
-        fk->foreign_table,
-        NULL, lock_type,
-        TABLE_LIST::PRELOCK_FK,
-        table_list->belong_to_view, op,
-        &prelocking_ctx->query_tables_last,
-        table_list->for_insert_data);
+    fk->table_list= tl;
   }
   if (arena)
     thd->restore_active_arena(arena, &backup);

sql/sql_class.cc

@@ -4950,6 +4950,7 @@ TABLE *get_purge_table(THD *thd)
   return thd->open_tables;
 }
 
+#ifndef DBUG_OFF
 /** Find an open table in the list of prelocked tabled
 
   Used for foreign key actions, for example, in UPDATE t1 SET a=1;
@@ -4970,6 +4971,7 @@ TABLE *find_fk_open_table(THD *thd, const char *db, size_t db_len,
   }
   return NULL;
 }
+#endif
 
 /* the following three functions are used in background purge threads */
 

sql/table.h

@@ -1836,6 +1836,7 @@ typedef struct st_foreign_key_info
   LEX_CSTRING *referenced_key_name;
   List<LEX_CSTRING> foreign_fields;
   List<LEX_CSTRING> referenced_fields;
+  TABLE_LIST *table_list;
 } FOREIGN_KEY_INFO;
 
 LEX_CSTRING *fk_option_name(enum_fk_option opt);
@@ -2634,6 +2635,8 @@ struct TABLE_LIST
   List<String> *partition_names;
 #endif /* WITH_PARTITION_STORAGE_ENGINE */
 
+  List<FOREIGN_KEY_INFO> *fk_ref_list;
+
   void calc_md5(char *buffer);
   int view_check_option(THD *thd, bool ignore_failure);
   bool create_field_translation(THD *thd);

storage/innobase/handler/ha_innodb.cc

@@ -124,8 +124,6 @@ extern "C" void thd_mark_transaction_to_rollback(MYSQL_THD thd, bool all);
 unsigned long long thd_get_query_id(const MYSQL_THD thd);
 void thd_clear_error(MYSQL_THD thd);
 
-TABLE *find_fk_open_table(THD *thd, const char *db, size_t db_len,
-			  const char *table, size_t table_len);
 MYSQL_THD create_background_thd();
 void destroy_background_thd(MYSQL_THD thd);
 void reset_thd(MYSQL_THD thd);
@@ -20304,27 +20302,22 @@ ha_innobase::multi_range_read_explain_info(
 	return m_ds_mrr.dsmrr_explain_info(mrr_mode, str, size);
 }
 
-/** Find or open a table handle for the virtual column template
+/** Find or open a table handle for the virtual column template for purge thread
 @param[in]	thd	thread handle
 @param[in,out]	table	InnoDB table whose virtual column template
 			is to be updated
 @return table handle
 @retval NULL if the table is dropped, unaccessible or corrupted
 for purge thread */
-static TABLE* innodb_find_table_for_vc(THD* thd, dict_table_t* table)
+TABLE* innodb_find_table_for_vc(THD* thd, dict_table_t* table)
 {
 	TABLE *mysql_table;
 	const bool  bg_thread = THDVAR(thd, background_thread);
 
-	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)) {
-			return table->vc_templ->mysql_table;
-		}
+	ut_ad(bg_thread);
+
+	if (bg_thread && (mysql_table = get_purge_table(thd))) {
+		return mysql_table;
 	}
 
 	char	db_buf[NAME_LEN + 1];
@@ -20335,16 +20328,8 @@ static TABLE* innodb_find_table_for_vc(THD* thd, dict_table_t* table)
 		return NULL;
 	}
 
-	if (bg_thread) {
-		return open_purge_table(thd, db_buf, db_buf_len,
-					tbl_buf, tbl_buf_len);
-	}
-
-	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);
-	return mysql_table;
+	return open_purge_table(thd, db_buf, db_buf_len,
+				tbl_buf, tbl_buf_len);
 }
 
 /** Get the computed value by supplying the base column values.
@@ -20428,22 +20413,14 @@ innobase_rename_vc_templ(
 */
 
 bool innobase_allocate_row_for_vcol(THD *thd, dict_index_t *index,
-                                    mem_heap_t **heap, TABLE **table,
+                                    mem_heap_t **heap, TABLE *maria_table,
                                     VCOL_STORAGE *storage)
 {
-  TABLE *maria_table;
   String *blob_value_storage;
-  if (!*table)
-    *table = innodb_find_table_for_vc(thd, index->table);
-
-  /* For purge thread, there is a possiblity that table could have
-     dropped, corrupted or unaccessible. */
-  if (!*table)
-    return false;
-  maria_table = *table;
   if (!*heap && !(*heap = mem_heap_create(srv_page_size)))
     return false;
 
+  ut_ad(maria_table);
   uchar *record = static_cast<byte *>(mem_heap_alloc(*heap,
                                                     maria_table->s->reclength));
 

storage/innobase/handler/ha_innodb.h

@@ -442,6 +442,14 @@ class ha_innobase final : public handler
 			  const KEY_PART_INFO& old_part,
 			  const KEY_PART_INFO& new_part) const override;
 
+	/** Builds a 'template' to the prebuilt struct.
+
+	The template is used in fast retrieval of just those column
+	values MariaDB needs in its processing.
+	@param whole_row true if access is needed to a whole row,
+	false if accessing individual fields is enough */
+	void build_template(bool whole_row);
+
 protected:
 	dberr_t innobase_get_autoinc(ulonglong* value);
 	dberr_t innobase_lock_autoinc();
@@ -470,13 +478,6 @@ class ha_innobase final : public handler
 		const uchar* record0,
 		const uchar* record1);
 #endif
-	/** Builds a 'template' to the prebuilt struct.
-
-	The template is used in fast retrieval of just those column
-	values MySQL needs in its processing.
-	@param whole_row true if access is needed to a whole row,
-	false if accessing individual fields is enough */
-	void build_template(bool whole_row);
 
 	int info_low(uint, bool);
 
@@ -512,6 +513,8 @@ class ha_innobase final : public handler
 
         /** If mysql has locked with external_lock() */
         bool                    m_mysql_has_locked;
+public:
+	row_prebuilt_t *get_prebuilt() const { return m_prebuilt; }
 };
 
 
@@ -971,3 +974,5 @@ which is in the prepared state
 
 @return 0 or error number */
 int innobase_rollback_by_xid(handlerton* hton, XID* xid);
+
+TABLE* innodb_find_table_for_vc(THD* thd, dict_table_t* table);

storage/innobase/include/row0ins.h

@@ -51,8 +51,11 @@ row_ins_check_foreign_constraint(
 	dict_table_t*	table,	/*!< in: if check_ref is TRUE, then the foreign
 				table, else the referenced table */
 	dtuple_t*	entry,	/*!< in: index entry for index */
+	row_prebuilt_t* prebuilt,
+				/*!< in: referenced prebuilt for this table.
+				NULL if check_ref is TRUE */
 	que_thr_t*	thr)	/*!< in: query thread */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
+	MY_ATTRIBUTE((nonnull(2,3,4,6), warn_unused_result));
 
 /*********************************************************************//**
 Sets a new row to insert for an INS_DIRECT node. This function is only used

storage/innobase/include/row0mysql.h

@@ -297,7 +297,9 @@ upd_node_t*
 row_create_update_node_for_mysql(
 /*=============================*/
 	dict_table_t*	table,	/*!< in: table to update */
-	mem_heap_t*	heap);	/*!< in: mem heap from which allocated */
+	mem_heap_t*	heap,	/*!< in: mem heap from which allocated */
+	const row_prebuilt_t* prebuilt);
+				/*!< in: prebult for this table  */
 
 /**********************************************************************//**
 Does a cascaded delete or set null in a foreign key operation.
@@ -853,7 +855,7 @@ struct VCOL_STORAGE
 @param[in]	thd		MariaDB THD
 @param[in]	index		Index in use
 @param[out]	heap		Heap that holds temporary row
-@param[in,out]	mysql_table	MariaDB table
+@param[in,out]	maria_table	MariaDB table
 @param[out]	rec		Pointer to allocated MariaDB record
 @param[out]	storage		Internal storage for blobs etc
 
@@ -865,7 +867,7 @@ bool innobase_allocate_row_for_vcol(
 				    THD *	  thd,
 				    dict_index_t* index,
 				    mem_heap_t**  heap,
-				    TABLE**	  table,
+				    TABLE*	  maria_table,
 				    VCOL_STORAGE* storage);
 
 /** Free memory allocated by innobase_allocate_row_for_vcol() */
@@ -879,7 +881,7 @@ class ib_vcol_row
 
   ib_vcol_row(mem_heap_t *heap) : heap(heap) {}
 
-  byte *record(THD *thd, dict_index_t *index, TABLE **table)
+  byte *record(THD *thd, dict_index_t *index, TABLE *table)
   {
     if (!storage.innobase_record)
     {

storage/innobase/include/row0upd.h

@@ -504,6 +504,12 @@ struct upd_node_t{
 	sym_node_t*	table_sym;/* table node in symbol table */
 	que_node_t*	col_assign_list;
 				/* column assignment list */
+	const row_prebuilt_t* prebuilt;
+			/* prebuilt used to access referenced prebuilts.
+			The field is mainly set up in
+			row_create_update_node_for_mysql and is then used in
+			row_upd_check_references_constraints to access
+			referenced sql-side table list. */
 	ulint		magic_n;
 
 private: