MDEV-27017 Assertion failure 'table->get_ref_count() == 0' on DDL that involves FULLTEXT INDEX

 purge_sys.stop_FTS() does not wait for purge operation
on FTS tables to finish. InnoDB DDL does purge_sys.stop_FTS()
and lock all fts tables. It eventually fails due to
n_ref_count value.

fts_stop_purge(): Stops the purge thread to process new FTS tables,
check n_ref_count of all fts index auxiliary, common tables.
This should make sure that consecutive fts_lock_tables()
is always successful.

storage/innobase/fts/fts0fts.cc

 /*****************************************************************************
 
 Copyright (c) 2011, 2021, Oracle and/or its affiliates.
-Copyright (c) 2016, 2021, MariaDB Corporation.
+Copyright (c) 2016, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -23,9 +23,7 @@ Full Text Search interface
 ***********************************************************************/
 
 #include "trx0roll.h"
-#ifdef UNIV_DEBUG
-# include "trx0purge.h"
-#endif
+#include "trx0purge.h"
 #include "row0mysql.h"
 #include "row0upd.h"
 #include "dict0types.h"
@@ -1565,6 +1563,60 @@ dberr_t fts_lock_common_tables(trx_t *trx, const dict_table_t &table)
   return DB_SUCCESS;
 }
 
+/** This function make sure that table doesn't
+have any other reference count.
+@param	table_name	table name */
+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);
+  if (!table)
+    return;
+
+  while (table->get_ref_count() > 1)
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+  table->release();
+}
+
+/** Stop the purge thread and check n_ref_count of all auxiliary
+and common table associated with the fts table.
+@param	table	parent FTS table */
+void purge_sys_t::stop_FTS(const dict_table_t &table)
+{
+  purge_sys.stop_FTS();
+  fts_table_t fts_table;
+  char table_name[MAX_FULL_NAME_LEN];
+
+  FTS_INIT_FTS_TABLE(&fts_table, nullptr, FTS_COMMON_TABLE, (&table));
+
+  for (const char **suffix= fts_common_tables; *suffix; suffix++)
+  {
+    fts_table.suffix= *suffix;
+    fts_get_table_name(&fts_table, table_name, false);
+    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)
+  {
+    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, false);
+      fts_table_no_ref_count(table_name);
+    }
+  }
+}
+
 /** Lock the internal FTS_ tables for table, before fts_drop_tables().
 @param trx    transaction
 @param table  table containing FULLTEXT INDEX

storage/innobase/handler/ha_innodb.cc

@@ -4,7 +4,7 @@ Copyright (c) 2000, 2020, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2008, 2009 Google Inc.
 Copyright (c) 2009, Percona Inc.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2013, 2021, MariaDB Corporation.
+Copyright (c) 2013, 2022, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -13457,7 +13457,7 @@ int ha_innobase::delete_table(const char *name)
   if (fts)
   {
     fts_optimize_remove_table(table);
-    purge_sys.stop_FTS();
+    purge_sys.stop_FTS(*table);
     err= fts_lock_tables(trx, *table);
   }
 
@@ -13835,7 +13835,7 @@ int ha_innobase::truncate()
 
 	if (fts) {
 		fts_optimize_remove_table(ib_table);
-		purge_sys.stop_FTS();
+		purge_sys.stop_FTS(*ib_table);
 		error = fts_lock_tables(trx, *ib_table);
 	}
 

storage/innobase/handler/handler0alter.cc

 /*****************************************************************************
 
 Copyright (c) 2005, 2019, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2021, MariaDB Corporation.
+Copyright (c) 2013, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -6271,7 +6271,7 @@ prepare_inplace_alter_table_dict(
 	}
 
 	if (fts_exist) {
-		purge_sys.stop_FTS();
+		purge_sys.stop_FTS(*ctx->new_table);
 		if (error == DB_SUCCESS) {
 			error = fts_lock_tables(ctx->trx, *ctx->new_table);
 		}
@@ -8718,7 +8718,7 @@ inline bool rollback_inplace_alter_table(Alter_inplace_info *ha_alter_info,
     if (fts_exist)
     {
       fts_optimize_remove_table(ctx->new_table);
-      purge_sys.stop_FTS();
+      purge_sys.stop_FTS(*ctx->new_table);
     }
     if (ctx->need_rebuild())
     {
@@ -10876,14 +10876,14 @@ ha_innobase::commit_inplace_alter_table(
 		}
 	}
 
-	if (fts_exist) {
-		purge_sys.stop_FTS();
-	}
-
 	for (inplace_alter_handler_ctx** pctx = ctx_array; *pctx; pctx++) {
 		auto ctx = static_cast<ha_innobase_inplace_ctx*>(*pctx);
 		dberr_t error = DB_SUCCESS;
 
+		if (fts_exist) {
+			purge_sys.stop_FTS(*ctx->old_table);
+		}
+
 		if (new_clustered && ctx->old_table->fts) {
 			ut_ad(!ctx->old_table->fts->add_wq);
 			fts_optimize_remove_table(ctx->old_table);

storage/innobase/include/trx0purge.h

 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -283,6 +283,11 @@ class purge_sys_t
     trx_sys.clone_oldest_view(&view);
     latch.wr_unlock();
   }
+
+  /** Stop the purge thread and check n_ref_count of all auxiliary
+  and common table associated with the fts table.
+  @param table	parent FTS table */
+  void stop_FTS(const dict_table_t &table);
 };
 
 /** The global data structure coordinating a purge */

storage/innobase/row/row0mysql.cc

 /*****************************************************************************
 
 Copyright (c) 2000, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2021, MariaDB Corporation.
+Copyright (c) 2015, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -2484,7 +2484,7 @@ dberr_t row_discard_tablespace_for_mysql(dict_table_t *table, trx_t *trx)
   if (fts_exist)
   {
     fts_optimize_remove_table(table);
-    purge_sys.stop_FTS();
+    purge_sys.stop_FTS(*table);
     err= fts_lock_tables(trx, *table);
     if (err != DB_SUCCESS)
     {