MDEV-27783 InnoDB: Failing assertion: table->get_ref_count() == 0 upon ALTER TABLE ... MODIFY COLUMN

- There is a race condition occurs between purge thread and DDL.
So purge thread can increment n_ref_count even after DDL does
purge_sys_t::stop_FTS().

- dict_table_open_on_id for purge thread should check
purge_sys.must_wait_FTS() before acquring the table.

- purge_sys.stop_FTS() does acquire dict_sys.latch for setting
the purge system flag and check table ref count on auxilary tables.

4 files changed, 57 insertions, 17 deletions

diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 4253326d46a..f64fd6f04c9 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -68,6 +68,7 @@ Created 1/8/1996 Heikki Tuuri
 #include "srv0mon.h"
 #include "srv0start.h"
 #include "trx0undo.h"
+#include "trx0purge.h"
 
 #include <vector>
 #include <algorithm>
@@ -819,12 +820,14 @@ template dict_table_t* dict_acquire_mdl_shared<true>
 (dict_table_t*,THD*,MDL_ticket**,dict_table_op_t);
 
 /** Look up a table by numeric identifier.
+@tparam purge_thd Whether the function is called by purge thread
 @param[in]      table_id        table identifier
 @param[in]      dict_locked     data dictionary locked
 @param[in]      table_op        operation to perform when opening
 @param[in,out]  thd             background thread, or NULL to not acquire MDL
 @param[out]     mdl             mdl ticket, or NULL
 @return table, NULL if does not exist */
+template <bool purge_thd>
 dict_table_t*
 dict_table_open_on_id(table_id_t table_id, bool dict_locked,
                       dict_table_op_t table_op, THD *thd,
@@ -837,6 +840,12 @@ dict_table_open_on_id(table_id_t table_id, bool dict_locked,
 
   if (table)
   {
+    if (purge_thd && purge_sys.must_wait_FTS())
+    {
+      table= nullptr;
+      goto func_exit;
+    }
+
     table->acquire();
     if (thd && !dict_locked)
       table= dict_acquire_mdl_shared<false>(table, thd, mdl, table_op);
@@ -853,7 +862,14 @@ dict_table_open_on_id(table_id_t table_id, bool dict_locked,
                                  ? DICT_ERR_IGNORE_RECOVER_LOCK
                                  : DICT_ERR_IGNORE_FK_NOKEY);
     if (table)
+    {
+      if (purge_thd && purge_sys.must_wait_FTS())
+      {
+        dict_sys.unlock();
+        return nullptr;
+      }
       table->acquire();
+    }
     if (!dict_locked)
     {
       dict_sys.unlock();
@@ -867,12 +883,22 @@ dict_table_open_on_id(table_id_t table_id, bool dict_locked,
     }
   }
 
+func_exit:
   if (!dict_locked)
     dict_sys.unfreeze();
 
   return table;
 }
 
+template dict_table_t* dict_table_open_on_id<false>
+(table_id_t table_id, bool dict_locked,
+ dict_table_op_t table_op, THD *thd,
+ MDL_ticket **mdl);
+template dict_table_t* dict_table_open_on_id<true>
+(table_id_t table_id, bool dict_locked,
+ dict_table_op_t table_op, THD *thd,
+ MDL_ticket **mdl);
+
 /********************************************************************//**
 Looks for column n position in the clustered index.
 @return position in internal representation of the clustered index */
diff --git a/storage/innobase/fts/fts0fts.cc b/storage/innobase/fts/fts0fts.cc
index 964a216cd39..c3d076b81d6 100644
--- a/storage/innobase/fts/fts0fts.cc
+++ b/storage/innobase/fts/fts0fts.cc
@@ -1555,12 +1555,16 @@ have any other reference count.
 static void fts_table_no_ref_count(const char *table_name)
 {
   dict_table_t *table= dict_table_open_on_name(
-    table_name, false, DICT_ERR_IGNORE_TABLESPACE);
+    table_name, true, DICT_ERR_IGNORE_TABLESPACE);
   if (!table)
     return;
 
   while (table->get_ref_count() > 1)
+  {
+    dict_sys.unlock();
     std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    dict_sys.lock(SRW_LOCK_CALL);
+  }
 
   table->release();
 }
@@ -1572,8 +1576,10 @@ and common table associated with the fts table.
 			already stopped*/
 void purge_sys_t::stop_FTS(const dict_table_t &table, bool already_stopped)
 {
+  dict_sys.lock(SRW_LOCK_CALL);
   if (!already_stopped)
     purge_sys.stop_FTS();
+
   fts_table_t fts_table;
   char table_name[MAX_FULL_NAME_LEN];
 
@@ -1582,28 +1588,31 @@ void purge_sys_t::stop_FTS(const dict_table_t &table, bool already_stopped)
   for (const char **suffix= fts_common_tables; *suffix; suffix++)
   {
     fts_table.suffix= *suffix;
-    fts_get_table_name(&fts_table, table_name, false);
+    fts_get_table_name(&fts_table, table_name, true);
     fts_table_no_ref_count(table_name);
   }
 
-  if (!table.fts)
-    return;
-  auto indexes= table.fts->indexes;
-  if (!indexes)
-    return;
-  for (ulint i= 0;i < ib_vector_size(indexes); ++i)
+  if (table.fts)
   {
-    const dict_index_t *index= static_cast<const dict_index_t*>(
-      ib_vector_getp(indexes, i));
-    FTS_INIT_INDEX_TABLE(&fts_table, nullptr, FTS_INDEX_TABLE, index);
-    for (const fts_index_selector_t *s= fts_index_selector;
-         s->suffix; s++)
+    if (auto indexes= table.fts->indexes)
     {
-      fts_table.suffix= s->suffix;
-      fts_get_table_name(&fts_table, table_name, false);
-      fts_table_no_ref_count(table_name);
+      for (ulint i= 0;i < ib_vector_size(indexes); ++i)
+      {
+        const dict_index_t *index= static_cast<const dict_index_t*>(
+          ib_vector_getp(indexes, i));
+        FTS_INIT_INDEX_TABLE(&fts_table, nullptr, FTS_INDEX_TABLE, index);
+        for (const fts_index_selector_t *s= fts_index_selector;
+             s->suffix; s++)
+        {
+          fts_table.suffix= s->suffix;
+          fts_get_table_name(&fts_table, table_name, true);
+          fts_table_no_ref_count(table_name);
+        }
+      }
     }
   }
+
+  dict_sys.unlock();
 }
 
 /** Lock the internal FTS_ tables for table, before fts_drop_tables().
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index 07acd0ecb74..a02f4761964 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -146,6 +146,7 @@ dict_acquire_mdl_shared(dict_table_t *table,
 @param[in,out]  thd             background thread, or NULL to not acquire MDL
 @param[out]     mdl             mdl ticket, or NULL
 @return table, NULL if does not exist */
+template<bool purge_thd= false>
 dict_table_t*
 dict_table_open_on_id(table_id_t table_id, bool dict_locked,
                       dict_table_op_t table_op, THD *thd= nullptr,
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 75d497b2cf4..e6267a2023a 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -1027,10 +1027,14 @@ row_purge_parse_undo_rec(
 try_again:
 	purge_sys.check_stop_FTS();
 
-	node->table = dict_table_open_on_id(
+	node->table = dict_table_open_on_id<true>(
 		table_id, false, DICT_TABLE_OP_NORMAL, node->purge_thd,
 		&node->mdl_ticket);
 
+	if (!node->table && purge_sys.must_wait_FTS()) {
+		goto try_again;
+	}
+
 	if (!node->table) {
 		/* The table has been dropped: no need to do purge and
 		release mdl happened as a part of open process itself */