3 files changed, 491 insertions, 174 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 4eeccdd..e468c80 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -1,3 +1,5 @@
+INCLUDE_DIRECTORIES(SYSTEM ${BOOST_PATCHES_DIR} ${BOOST_INCLUDE_DIR})
+
 MYSQL_ADD_PLUGIN(table_sizes src/table_sizes.cc
   MODULE_ONLY MODULE_OUTPUT_NAME "table_sizes")
 
diff --git a/src/mysql_inc.h b/src/mysql_inc.h
deleted file mode 100644
index 4912682..0000000
--- a/src/mysql_inc.h
+++ /dev/null
@@ -1,16 +0,0 @@
-#ifndef MYSQL_INCL_H
-#define MYSQL_INCL_H
-
-#ifndef HAVE_CONFIG_H
-#define HAVE_CONFIG_H
-#endif
-
-#ifndef MYSQL_DYNAMIC_PLUGIN
-#define MYSQL_DYNAMIC_PLUGIN
-#endif
-
-#define MYSQL_SERVER 1
-
-#include <mysql_version.h>
-
-#endif
diff --git a/src/table_sizes.cc b/src/table_sizes.cc
index b1b38bb..40d5090 100644
--- a/src/table_sizes.cc
+++ b/src/table_sizes.cc
@@ -12,31 +12,366 @@
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 */
-#include "mysql_inc.h"
-#include <sql_base.h>
-#include <sql_show.h>
-#include <sql_table.h>
-#include <sql_parse.h>
-#include <sql_db.h>
-#include <auth_common.h>
 
-#define STR_OR_NIL(S) ((S) ? (S) : "<nil>")
+#include <mysql/plugin.h>
+#include <boost/optional/optional.hpp>
+
+#include "sql/mysqld.h"
+#include "sql/sql_base.h"
+#include "sql/sql_show.h"
+#include "sql/sql_class.h"    // THD
+#include "sql/sql_table.h"    // filename_to_tablename
+#include "sql/strfunc.h"      // make_lex_string_root
+#include "sql/sql_parse.h"    // sql_command_flags
+#include "sql/sql_lex.h"      // LEX
+#include "sql/item_func.h"    // Item_Func
+#include "sql/item_cmpfunc.h" // Item_cond
+#include "sql/dd/cache/dictionary_client.h"  // dd::cache::Dictionary_client
+#include "sql/dd/dd_schema.h"     // dd::Schema_MDL_locker
+#include "sql/dd/string_type.h"   // dd::String_type
+#include "sql/dd/sdi_file.h"      // dd::sdi_file
+#include "sql/auth/auth_acls.h"   // *_ACLS
+#include "sql/auth/auth_common.h" // check_grant_db
+#include "mf_wcomp.h" // wild_compare,wild_one,wild_many
+#include "my_dir.h"   // MY_DIR
+
+// InnoDB FTS
+/** For storing table info when checking for orphaned tables. */
+struct fts_aux_table_t
+{
+  /** Parent table id */
+  uint64_t parent_id;
+
+  /** Table FT index id */
+  uint64_t index_id;
+};
+
+/** Check if a table is an FTS auxiliary table name.
+@param[out]     table   FTS table info
+@param[in]      name    Table name
+@param[in]      len     Length of table name
+@return true if the name matches an auxiliary table name pattern */
+bool fts_is_aux_table_name(fts_aux_table_t *table, const char *name,
+                           int len)
+{
+  /** FTS auxiliary table prefix that are common to all FT indexes.*/
+  const char *FTS_PREFIX = "fts_";
+  const char *FTS_PREFIX_5_7 = "FTS_";
+
+  /** FTS auxiliary table suffixes that are common to all FT indexes. */
+  const char *fts_common_tables[] = {"being_deleted", "being_deleted_cache",
+                                     "config",        "deleted",
+                                     "deleted_cache", NULL};
+  const char *fts_common_tables_5_7[] = {"BEING_DELETED", "BEING_DELETED_CACHE",
+                                         "CONFIG",        "DELETED",
+                                         "DELETED_CACHE", NULL};
+
+  /** FTS auxiliary INDEX split intervals. */
+  const char *fts_index_selector[] = {"index_1", "index_2", "index_3", "index_4",
+                                      "index_5", "index_6", NULL};
+  const char *fts_index_selector_5_7[] = {"INDEX_1", "INDEX_2", "INDEX_3", "INDEX_4",
+                                          "INDEX_5", "INDEX_6", NULL};
+
+  const char *ptr = name;
+  const char *end = name + len;
+
+  /* All auxiliary tables are prefixed with "FTS_" and the name
+  length will be at the very least greater than 20 bytes. */
+  if (ptr != NULL && len > 20 &&
+      (strncmp(ptr, FTS_PREFIX, 4) == 0 ||
+       strncmp(ptr, FTS_PREFIX_5_7, 4) == 0))
+  {
+    /* Skip the prefix. */
+    ptr += 4;
+    len -= 4;
+
+    /* Try and read the table id. */
+    if (sscanf(ptr, "%016" PRIx64, &table->parent_id) != 1)
+      return false;
+
+    /* Skip the table id. */
+    ptr = static_cast<const char *>(memchr(ptr, '_', len));
+    if (ptr == NULL)
+      return false;
+
+    /* Skip the underscore. */
+    ++ptr;
+    len = end - ptr;
+
+    /* First search the common table suffix array. */
+    for (int i = 0; fts_common_tables[i] != NULL; ++i) {
+      if (strncmp(ptr, fts_common_tables[i], len) == 0 ||
+          strncmp(ptr, fts_common_tables_5_7[i], len) == 0) {
+        return true;
+      }
+    }
+
+    /* Could be obsolete common tables. */
+    if (native_strncasecmp(ptr, "ADDED", len) == 0 ||
+        native_strncasecmp(ptr, "STOPWORDS", len) == 0) {
+      return true;
+    }
+
+    /* Try and read the index id. */
+    if (sscanf(ptr, "%016" PRIx64, &table->index_id) != 1)
+      return false;
+
+    /* Skip the index id. */
+    ptr = static_cast<const char *>(memchr(ptr, '_', len));
+
+   if (ptr == NULL)
+      return false;
+
+    /* Skip the underscore. */
+    ++ptr;
+    len = end - ptr;
+
+    /* Search the FT index specific array. */
+    for (int i = 0; fts_index_selector[i] != NULL; ++i) {
+      if (strncmp(ptr, fts_index_selector[i], len) == 0 ||
+          strncmp(ptr, fts_index_selector_5_7[i], len) == 0) {
+        return true;
+      }
+    }
+
+    /* Other FT index specific table(s). */
+    if (native_strncasecmp(ptr, "DOC_ID", len) == 0)
+      return true;
+  }
+
+  return false;
+}
+
+/*----------------------------------------------------------------------------*/
+
+/**
+  Condition pushdown used for INFORMATION_SCHEMA / SHOW queries.
+  This structure is to implement an optimization when
+  accessing data dictionary data in the INFORMATION_SCHEMA
+  or SHOW commands.
+  When the query contain a TABLE_SCHEMA or TABLE_NAME clause,
+  narrow the search for data based on the constraints given.
+*/
+struct LOOKUP_FIELD_VALUES {
+  /**
+    Value of a TABLE_SCHEMA clause.
+    Note that this value length may exceed @c NAME_LEN.
+    @sa wild_db_value
+  */
+  boost::optional<dd::String_type> db_value;
+  /**
+    Value of a TABLE_NAME clause.
+    Note that this value length may exceed @c NAME_LEN.
+    @sa wild_table_value
+  */
+  boost::optional<dd::String_type> table_value;
+  /**
+    True when @c db_value is a LIKE clause,
+    false when @c db_value is an '=' clause.
+  */
+  bool wild_db_value = {false};
+  /**
+    True when @c table_value is a LIKE clause,
+    false when @c table_value is an '=' clause.
+  */
+  bool wild_table_value = {false};
+};
+
+/**
+  @brief    Get lookup value from the part of 'WHERE' condition
+
+  @details This function gets lookup value from
+           the part of 'WHERE' condition if it's possible and
+           fill appropriate lookup_field_vals struct field
+           with this value.
 
-typedef struct st_lookup_field_values
+  @param[in]      thd                   thread handler
+  @param[in]      item_func             part of WHERE condition
+  @param[in]      table                 I_S table
+  @param[in, out] lookup_field_vals     Struct which holds lookup values
+
+  @return
+    0             success
+    1             error, there can be no matching records for the condition
+*/
+
+static bool get_lookup_value(Item_func *item_func, TABLE_LIST *table,
+    LOOKUP_FIELD_VALUES *lookup_field_vals)
+{
+  ST_SCHEMA_TABLE *schema_table = table->schema_table;
+  ST_FIELD_INFO *field_info = schema_table->fields_info;
+  const char *field_name1 =
+      schema_table->idx_field1 >= 0
+          ? field_info[schema_table->idx_field1].field_name
+          : "";
+  const char *field_name2 =
+      schema_table->idx_field2 >= 0
+          ? field_info[schema_table->idx_field2].field_name
+          : "";
+
+  if (item_func->functype() == Item_func::EQ_FUNC ||
+      item_func->functype() == Item_func::EQUAL_FUNC) {
+    int idx_field, idx_val;
+    char tmp[MAX_FIELD_WIDTH];
+    String *tmp_str, str_buff(tmp, sizeof(tmp), system_charset_info);
+    Item_field *item_field;
+    CHARSET_INFO *cs = system_charset_info;
+
+    if (item_func->arguments()[0]->type() == Item::FIELD_ITEM &&
+        item_func->arguments()[1]->const_item()) {
+      idx_field = 0;
+      idx_val = 1;
+    } else if (item_func->arguments()[1]->type() == Item::FIELD_ITEM &&
+               item_func->arguments()[0]->const_item()) {
+      idx_field = 1;
+      idx_val = 0;
+    } else
+      return 0;
+
+    item_field = (Item_field *)item_func->arguments()[idx_field];
+    if (table->table != item_field->field->table) return 0;
+    tmp_str = item_func->arguments()[idx_val]->val_str(&str_buff);
+
+    /* impossible value */
+    if (!tmp_str) return 1;
+
+    /* Lookup value is database name */
+    if (!cs->coll->strnncollsp(cs, (uchar *)field_name1, strlen(field_name1),
+                               (uchar *)item_field->field_name,
+                               strlen(item_field->field_name))) {
+      lookup_field_vals->db_value.emplace(tmp_str->ptr(), tmp_str->length());
+    }
+    /* Lookup value is table name */
+    else if (!cs->coll->strnncollsp(cs, (uchar *)field_name2,
+                                    strlen(field_name2),
+                                    (uchar *)item_field->field_name,
+                                    strlen(item_field->field_name))) {
+      lookup_field_vals->table_value.emplace(tmp_str->ptr(), tmp_str->length());
+    }
+  }
+  return 0;
+}
+
+/**
+  @brief    Calculates lookup values from 'WHERE' condition
+
+  @details This function calculates lookup value(database name, table name)
+           from 'WHERE' condition if it's possible and
+           fill lookup_field_vals struct fields with these values.
+
+  @param[in]      thd                   thread handler
+  @param[in]      cond                  WHERE condition
+  @param[in]      table                 I_S table
+  @param[in, out] lookup_field_vals     Struct which holds lookup values
+
+  @return
+    0             success
+    1             error, there can be no matching records for the condition
+*/
+
+static bool calc_lookup_values_from_cond(Item *cond,
+    TABLE_LIST *table, LOOKUP_FIELD_VALUES *lookup_field_vals)
+{
+  if (!cond) return 0;
+
+  if (cond->type() == Item::COND_ITEM) {
+    if (((Item_cond *)cond)->functype() == Item_func::COND_AND_FUNC) {
+      List_iterator<Item> li(*((Item_cond *)cond)->argument_list());
+      Item *item;
+      while ((item = li++)) {
+        if (item->type() == Item::FUNC_ITEM) {
+          if (get_lookup_value((Item_func *)item, table, lookup_field_vals))
+            return 1;
+        } else {
+          if (calc_lookup_values_from_cond(item, table, lookup_field_vals))
+            return 1;
+        }
+      }
+    }
+    return 0;
+  } else if (cond->type() == Item::FUNC_ITEM &&
+             get_lookup_value((Item_func *)cond, table, lookup_field_vals))
+    return 1;
+  return 0;
+}
+
+/**
+  @brief   Calculate lookup values(database name, table name)
+
+  @details This function calculates lookup values(database name, table name)
+           from 'WHERE' condition or wild values (for 'SHOW' commands only)
+           from LEX struct and fill lookup_field_vals struct field
+           with these values.
+
+  @param[in]      thd                   thread handler
+  @param[in]      cond                  WHERE condition
+  @param[in]      tables                I_S table
+  @param[in, out] lookup_field_values   Struct which holds lookup values
+
+  @return
+    0             success
+    1             error, there can be no matching records for the condition
+*/
+
+static bool get_lookup_field_values(THD *thd, Item *cond, TABLE_LIST *tables,
+    LOOKUP_FIELD_VALUES *lookup_field_values)
 {
-  LEX_STRING db_value, table_value;
-  bool wild_db_value, wild_table_value;
-} LOOKUP_FIELD_VALUES;
+  LEX *lex = thd->lex;
+  const char *wild = lex->wild ? lex->wild->ptr() : NullS;
+  bool rc = 0;
+
+  switch (lex->sql_command) {
+    case SQLCOM_SHOW_DATABASES:
+      if (wild) {
+        lookup_field_values->db_value = wild;
+        lookup_field_values->wild_db_value = true;
+      }
+      break;
+    case SQLCOM_SHOW_TABLES:
+    case SQLCOM_SHOW_TABLE_STATUS:
+    case SQLCOM_SHOW_TRIGGERS:
+    case SQLCOM_SHOW_EVENTS:
+      lookup_field_values->db_value = lex->select_lex->db;
+      if (wild) {
+        lookup_field_values->table_value = wild;
+        lookup_field_values->wild_table_value = true;
+      }
+      break;
+    default:
+      /*
+        The "default" is for queries over I_S.
+        All previous cases handle SHOW commands.
+      */
+      rc = calc_lookup_values_from_cond(cond, tables, lookup_field_values);
+      break;
+  }
 
-bool get_lookup_field_values(THD *thd, Item *cond, TABLE_LIST *table,
-    LOOKUP_FIELD_VALUES *lookup_field_vals);
+  if (lower_case_table_names && !rc) {
+    /*
+      We can safely do in-place upgrades here since all of the above cases
+      are allocating a new memory buffer for these strings.
+    */
+    if (lookup_field_values->db_value && !lookup_field_values->db_value->empty())
+      casedn(system_charset_info, *lookup_field_values->db_value);
+    if (lookup_field_values->table_value && !lookup_field_values->table_value->empty())
+      casedn(system_charset_info, *lookup_field_values->table_value);
+  }
+
+  return rc;
+}
 
 /*----------------------------------------------------------------------------*/
 
+static bool is_special_db(const dd::String_type &db_name)
+{
+  /* I_S does not exist and we hide P_S */
+  return is_infoschema_db(db_name.c_str(), db_name.size())
+    || is_perfschema_db(db_name.c_str(), db_name.size());
+}
+
 static int
-make_db_list(THD *thd, List<LEX_STRING> *db_names,
-    LOOKUP_FIELD_VALUES *lookup_field_vals,
-    MEM_ROOT *tmp_mem_root)
+make_db_list(THD *thd, std::vector<dd::String_type> *db_names,
+    LOOKUP_FIELD_VALUES *lookup_field_vals)
 {
   /*
     If we have db lookup value we just add it to list and
@@ -44,26 +379,32 @@ make_db_list(THD *thd, List<LEX_STRING> *db_names,
     We don't do this for database names longer than the maximum
     path length.
   */
-  if (lookup_field_vals->db_value.str
-      && lookup_field_vals->db_value.length <= NAME_LEN)
+  if (lookup_field_vals->db_value
+      && lookup_field_vals->db_value->length() <= NAME_LEN)
   {
-    if (db_names->push_back(&lookup_field_vals->db_value))
-      return 1;
+    const dd::String_type &db_name = lookup_field_vals->db_value.get();
+    if (!is_special_db(db_name))
+      db_names->push_back(db_name);
     return 0;
   }
 
-  return (find_files(thd, db_names, NullS,
-        mysql_data_home, NullS, TRUE, tmp_mem_root) != FIND_FILES_OK);
+  dd::cache::Dictionary_client::Auto_releaser releaser(thd->dd_client());
+  std::vector<const dd::Schema *> schemas;
+  if (thd->dd_client()->fetch_global_components(&schemas))
+    return 1;
+
+  for (const dd::Schema *schema_obj : schemas)
+  {
+    const dd::String_type &db_name = schema_obj->name();
+    if (is_special_db(db_name))
+      continue;
+    db_names->push_back(db_name);
+  }
+  return 0;
 }
 
 /*----------------------------------------------------------------------------*/
 
-struct TABLE_DATA
-{
-  LEX_STRING name;
-  longlong size;
-};
-
 /*
   SYNOPSIS
     thd                 thread handler
@@ -77,30 +418,17 @@ struct TABLE_DATA
     FIND_FILES_OOM      out of memory error
     FIND_FILES_DIR      no such directory, or directory can't be read
 */
+typedef std::map<dd::String_type, size_t> Table_sizes_map;
 static find_files_result
-find_tables(THD *thd, List<TABLE_DATA> *tables, const char *db,
-    const char *path, const char *wild, MEM_ROOT *tmp_mem_root)
+find_tables(THD *thd, Table_sizes_map &tables, const char *db,
+    const char *path, boost::optional<dd::String_type> &wild,
+    MEM_ROOT *tmp_mem_root)
 {
   MY_DIR *dirp;
-  MEM_ROOT **root_ptr = NULL, *old_root = NULL;
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
+  MEM_ROOT **save_mem_root_ptr = THR_MALLOC;
   uint col_access = thd->col_access;
-#endif
-  size_t wild_length = 0;
-  TABLE_LIST table_list;
-  TABLE_DATA *table_data = NULL;
   DBUG_ENTER("find_files");
 
-  if (wild)
-  {
-    if (!wild[0])
-      wild = 0;
-    else
-      wild_length = strlen(wild);
-  }
-
-  memset(&table_list, 0, sizeof(table_list));
-
   if (!(dirp = my_dir(path, MYF(MY_WANT_STAT))))
   {
     if (my_errno() == ENOENT)
@@ -115,40 +443,55 @@ find_tables(THD *thd, List<TABLE_DATA> *tables, const char *db,
   }
 
   if (tmp_mem_root)
-  {
-    root_ptr = my_thread_get_THR_MALLOC();
-    old_root = *root_ptr;
-    *root_ptr = tmp_mem_root;
-  }
+    THR_MALLOC = &tmp_mem_root;
 
+  char uname[NAME_LEN + 1]; /* Unencoded name */
+  dd::String_type fts_schema_name, fts_table_name; /* FTS lookup */
+  std::map<dd::Object_id, dd::String_type> fts_cache;
+  tables.clear();
   for (uint i = 0; i < dirp->number_off_files; i++)
   {
-    char uname[NAME_LEN + 1]; /* Unencoded name */
-    FILEINFO *file;
-    size_t file_name_len;
-    char *ext;
 
-    file = dirp->dir_entry + i;
-    /* skip '.', '..', db.opt and temp files. */
+    FILEINFO *file = dirp->dir_entry + i;
+    /* skip '.', '..' and temp files. */
     if ((file->name[0] == '.' &&
         (!file->name[1] || (file->name[1] == '.' && !file->name[2])))
-        || !my_strcasecmp(files_charset_info, file->name, MY_DB_OPT_FILE)
         || is_prefix(file->name, tmp_file_prefix))
       continue;
 
+    char *ext;
     if ((ext = strrchr(file->name, '.')))
+    {
+      // sdi files are small and their name is truncated
+      // lookup by id is possible but not worth the effort
+      if (!strcmp(ext, dd::sdi_file::EXT.c_str()))
+        continue;
       *ext = 0;
-    file_name_len = filename_to_tablename(file->name, uname, sizeof(uname));
+      ++ext;
+    }
 
-    if (table_data != NULL)
+    size_t file_name_len;
+    fts_aux_table_t fts_table;
+    if (ext && !strcmp(ext, "ibd")
+        && fts_is_aux_table_name(&fts_table, file->name, strlen(file->name)))
     {
-      if (!my_strcasecmp(files_charset_info, uname, table_data->name.str))
-      {
-        table_data->size += file->mystat->st_size;
+      auto entry = fts_cache.find(fts_table.parent_id);
+      if (entry != fts_cache.end())
+        fts_table_name = entry->second;
+      else if (thd->dd_client()->get_table_name_by_se_private_id("InnoDB",
+            fts_table.parent_id, &fts_schema_name, &fts_table_name))
         continue;
-      }
-      else
-        table_data = NULL;
+
+      file_name_len = (my_stpnmov(uname, fts_table_name.c_str(), sizeof(uname)) - uname);
+    }
+    else
+      file_name_len = filename_to_tablename(file->name, uname, sizeof(uname));
+
+    auto table_data = tables.find(uname);
+    if (table_data != tables.end())
+    {
+      table_data->second += file->mystat->st_size;
+      continue;
     }
 
     if (wild)
@@ -157,81 +500,68 @@ find_tables(THD *thd, List<TABLE_DATA> *tables, const char *db,
       {
         if (my_wildcmp(files_charset_info,
               uname, uname + file_name_len,
-              wild, wild + wild_length,
+              wild->c_str(), wild->c_str() + wild->length(),
               wild_prefix, wild_one, wild_many))
           continue;
       }
-      else if (wild_compare(uname, wild, 0))
+      else if (wild_compare(uname, file_name_len, wild->c_str(), wild->length(), 0))
         continue;
     }
 
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
     /* Don't show tables where we don't have any privileges */
     if (db && !(col_access & TABLE_ACLS))
     {
+      TABLE_LIST table_list;
       table_list.db = db;
       table_list.db_length = strlen(db);
       table_list.table_name = uname;
       table_list.table_name_length = file_name_len;
       table_list.grant.privilege = col_access;
-      if (check_grant(thd, TABLE_ACLS, &table_list, TRUE, 1, TRUE))
+      if (check_grant(thd, TABLE_ACLS, &table_list, true, 1, true))
         continue;
     }
-#endif
-
-    table_data = new TABLE_DATA();
-    table_data->size = file->mystat->st_size;
-    if (!(tmp_mem_root
-          ? make_lex_string_root(tmp_mem_root, &table_data->name, uname,
-                file_name_len, FALSE)
-          : thd->make_lex_string(&table_data->name, uname,
-                file_name_len, FALSE))
-        || tables->push_back(table_data))
-    {
-      my_dirend(dirp);
-      DBUG_RETURN(FIND_FILES_OOM);
-    }
+
+    tables.emplace(uname, file->mystat->st_size);
   }
 
-  DBUG_PRINT("info", ("found: %d files", tables->elements));
+  DBUG_PRINT("info", ("found: %ld files", tables.size()));
   my_dirend(dirp);
 
-  //(void)ha_find_files(thd, db, path, wild, dir, files);
-
   if (tmp_mem_root)
-    *root_ptr = old_root;
+    THR_MALLOC = save_mem_root_ptr;
 
   DBUG_RETURN(FIND_FILES_OK);
 }
 
 static int
-make_table_list(THD *thd, List<TABLE_DATA> *table_names, LEX *lex,
-    LOOKUP_FIELD_VALUES *lookup_field_vals, LEX_STRING *db_name,
+make_table_list(THD *thd, Table_sizes_map &table_names, LEX *lex,
+    LOOKUP_FIELD_VALUES *lookup_field_vals, const dd::String_type &db_name,
     MEM_ROOT *tmp_mem_root)
 {
   char path[FN_REFLEN + 1];
-  build_table_filename(path, sizeof(path) - 1, db_name->str, "", "", 0);
+  build_table_filename(path, sizeof(path) - 1, db_name.c_str(), "", "", 0);
 
-  find_files_result res = find_tables(thd, table_names, db_name->str, path,
-      lookup_field_vals->table_value.str, tmp_mem_root);
-  if (res != FIND_FILES_OK)
+  const dd::Schema *sch_obj = NULL;
+  if (thd->dd_client()->acquire(db_name.c_str(), &sch_obj))
+    return 1;
+  if (!sch_obj)
   {
     /*
-      Downgrade errors about problems with database directory to
-      warnings if this is not a 'SHOW' command. Another thread
-      may have dropped database, and we may still have a name
-      for that directory.
+      Report missing database only if it is a 'SHOW' command.
+      Another thread may have dropped the database after we
+      got its name from the DD.
     */
-    if (res == FIND_FILES_DIR)
+    if (sql_command_flags[lex->sql_command] & CF_STATUS_COMMAND)
     {
-      if (sql_command_flags[lex->sql_command] & CF_STATUS_COMMAND)
-        return 1;
-      thd->clear_error();
-      return 2;
+      my_error(ER_BAD_DB_ERROR, MYF(0), db_name.c_str());
+      return 1;
     }
-    return 1;
+    return 2;
   }
-  return 0;
+
+  find_files_result res = find_tables(thd, table_names, db_name.c_str(), path,
+      lookup_field_vals->table_value, tmp_mem_root);
+  return (res == FIND_FILES_OK) ? 0 : 2;
 }
 
 /*----------------------------------------------------------------------------*/
@@ -254,11 +584,8 @@ static int fill_table(THD *thd, TABLE_LIST *tables, Item *cond)
   LEX *lex = thd->lex;
   TABLE *table = tables->table;
   LOOKUP_FIELD_VALUES lookup_field_vals;
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
   Security_context *sctx = thd->security_context();
-#endif
-  List<LEX_STRING> db_names;
-  List_iterator_fast<LEX_STRING> it(db_names);
+  std::vector<dd::String_type> db_names;
   int error = 1;
 
   DBUG_ENTER("fill_table");
@@ -273,9 +600,10 @@ static int fill_table(THD *thd, TABLE_LIST *tables, Item *cond)
     goto err;
   }
 
+#define STR_OR_NIL(S) ((S) ? (S->c_str()) : "<nil>")
   DBUG_PRINT("INDEX VALUES", ("db_name='%s', table_name='%s'",
-        STR_OR_NIL(lookup_field_vals.db_value.str),
-        STR_OR_NIL(lookup_field_vals.table_value.str)));
+        STR_OR_NIL(lookup_field_vals.db_value),
+        STR_OR_NIL(lookup_field_vals.table_value)));
 
   if (!lookup_field_vals.wild_db_value && !lookup_field_vals.wild_table_value)
   {
@@ -283,74 +611,76 @@ static int fill_table(THD *thd, TABLE_LIST *tables, Item *cond)
       if lookup value is empty string then
       it's impossible table name or db name
     */
-    if ((lookup_field_vals.db_value.str
-          && !lookup_field_vals.db_value.str[0])
-        || (lookup_field_vals.table_value.str
-          && !lookup_field_vals.table_value.str[0]))
+    if ((lookup_field_vals.db_value
+          && lookup_field_vals.db_value->empty())
+        || (lookup_field_vals.table_value
+          && lookup_field_vals.table_value->empty()))
     {
-      error= 0;
+      error = 0;
       goto err;
     }
   }
 
-  /* NOTE: doesn't change the output of EXPLAIN - maybe sometimes later */
-  if (lookup_field_vals.db_value.length
+  if (lookup_field_vals.db_value && !lookup_field_vals.db_value->empty()
       && !lookup_field_vals.wild_db_value)
-    tables->has_db_lookup_value = TRUE;
-  if (lookup_field_vals.table_value.length
+    tables->has_db_lookup_value = true;
+  if (lookup_field_vals.table_value && !lookup_field_vals.table_value->empty()
       && !lookup_field_vals.wild_table_value)
-    tables->has_table_lookup_value = TRUE;
+    tables->has_table_lookup_value = true;
 
-  if (lex->describe)
+  if (lex->is_explain())
   {
     /* EXPLAIN SELECT */
     error = 0;
     goto err;
   }
 
-  if (make_db_list(thd, &db_names, &lookup_field_vals, &tmp_mem_root))
+  if (make_db_list(thd, &db_names, &lookup_field_vals))
     goto err;
 
-  LEX_STRING *db_name;
-  it.rewind(); /* To get access to new elements in basis list */
-  while ((db_name = it++))
+  for (auto &db_name : db_names)
   {
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-    if ((check_access(thd, SELECT_ACL, db_name->str, &thd->col_access, NULL, 0, 1)
-          || (!thd->col_access && check_grant_db(thd, db_name->str)))
-        && !sctx->check_access(DB_ACLS | SHOW_DB_ACL, true)
-        && !acl_get(sctx->host().str, sctx->ip().str,
-          sctx->priv_user().str, db_name->str, 0))
-      continue;
-#endif
-
-    List<TABLE_DATA> tables;
-    int res = make_table_list(thd, &tables, lex, &lookup_field_vals,
-        db_name, &tmp_mem_root);
-    if (res)
-    {
-      tables.delete_elements();
-      if (res == 2) /* Not fatal error, continue */
-        continue;
-      goto err;
+    bool have_db_privileges = false;
+    if (sctx->get_active_roles()->size() > 0) {
+      LEX_CSTRING const_db_name = { db_name.c_str(), db_name.size() };
+      have_db_privileges = sctx->db_acl(const_db_name) > 0 ? true : false;
     }
-
-    TABLE_DATA *table_data;
-    List_iterator_fast<TABLE_DATA> it_tables(tables);
-    while ((table_data = it_tables++))
+    if (!(check_access(thd, SELECT_ACL, db_name.c_str(), &thd->col_access, NULL, 0, 1)
+          || (!thd->col_access && check_grant_db(thd, db_name.c_str())))
+        || sctx->check_access(DB_ACLS | SHOW_DB_ACL, true)
+        || have_db_privileges
+        || acl_get(thd, sctx->host().str, sctx->ip().str,
+          sctx->priv_user().str, db_name.c_str(), 0))
     {
-      table->field[0]->store(db_name->str, db_name->length,
-          system_charset_info);
-      table->field[1]->store(table_data->name.str, table_data->name.length,
-          system_charset_info);
-      table->field[2]->store(table_data->size, TRUE);
-      if (schema_table_store_record(thd, table))
+      // We must make sure the schema is released and unlocked in the right
+      // order. Fail if we are unable to get a meta data lock on the schema
+      // name.
+      dd::Schema_MDL_locker mdl_handler(thd);
+      if (mdl_handler.ensure_locked(db_name.c_str()))
+        goto err;
+
+      dd::cache::Dictionary_client::Auto_releaser releaser(thd->dd_client());
+      Table_sizes_map tables;
+      int res = make_table_list(thd, tables, lex, &lookup_field_vals,
+          db_name, &tmp_mem_root);
+      if (res)
       {
-        tables.delete_elements();
+        if (res == 2) /* Not fatal error, continue */
+          continue;
         goto err;
       }
+
+      for (auto &table_data : tables)
+      {
+        table->field[0]->store(db_name.c_str(), db_name.size(),
+            system_charset_info);
+        table->field[1]->store(table_data.first.c_str(), table_data.first.size(),
+            system_charset_info);
+        table->field[2]->store(table_data.second, true);
+        if (schema_table_store_record(thd, table))
+          goto err;
+      }
     }
-    tables.delete_elements();
   }
 
   error = 0;
@@ -384,7 +714,8 @@ mysql_declare_plugin(table_sizes)
   PLUGIN_LICENSE_GPL,                    /* license type */
   init,                                  /* init function */
   NULL,
-  0x0507,                                /* version = 5.7 */
+  NULL,
+  0x0800,                                /* version = 8.0 */
   NULL,                                  /* no status variables */
   NULL,                                  /* no system variables */
   NULL,                                  /* no reserved information */