MDEV-30182: Optimize open_tables to take O(N) time

======
This is an  adopted version of patch provided in the PR #MariaDB/2680
Original author: Sergey Vanislavskiy
Adopted by: Nikita Maliavin
======

Table opening stage was known to make two list traversals:
1. In find_fk_prelocked_table, all the query_tables list is traversed for each
foreign key of a table to open.
2. MDL_context::find_ticket traverses all mdl tickets, one ticket per table.

Both result in O(tables^2) time total.
This may dramatically increase the query latencty in the following known cases:
* updates/deletes on tables with many children
* DMLs in transactions involving many different tables

Also, it slows down the DROP DATABASE performance, with a big enough amount of
tables.

So to optimize this out the following is done:
* A hash table with all FK-prelocked tables is added to THD. A table is filled
and queried inside find_fk_prelocked_table, and cleaned up at the query end.
* The find_ticket implementation is replaced with two consecutive hash lookups.
* A hash table with all tickets for current context (Query_tables_list) is
added.
* find_fk_prelocked_table now makes a hash lookup as well. We have to calculate
a hash value for this lookup, so MDL_key is created earlier. It's then reused
if a new table list is created.

Given the requirement of no performance degradation for a small table value,
a new data structure is introduced: Open_address_hash.

Open_address_hash is a generic data structure, that is optimized for usage with
a few elements, similarly to the "short string optimization" in c++ standard
libraries: if a number of elements can fit the structure's class body inline,
then they are placed there. One bit is borrowed to indicate whether the inline
mode is used.

This also means that this optimization cannot work in 32-bit environments.

The hash table is implemented from scratch using open address hashing to reduce
memory fragmentation, reduce allocation pressure and memory footprint.

Speaking of the memory footprint, it is expected to be as follows:
* +4x pointer size for each connection (2 pointers per hash, two hashes total)
* +4x pointer size for each prepared statement, or cached/exected stored
procedure.
* If number of tables opened > 2, then +2x pointer size per table, because the
hash load factor is kept 50% at this moment
* If number of FK-prelocked tables opened > 2, then +2x pointer size per
FK-prelocked table.

include/open_address_hash.h

+#pragma once
+
+#include <cstdint>
+#include <cstring>
+#include "my_dbug.h"
+#include "m_string.h"
+#include "my_global.h"
+#include "m_ctype.h"
+
+namespace traits
+{
+template<typename Key>
+struct Open_address_hash_key_trait;
+
+template<typename Value>
+struct Open_address_hash_value_trait;
+}
+
+template <typename Key, typename Value,
+          typename Key_trait=traits::Open_address_hash_key_trait<Key>,
+          typename Value_trait=traits::Open_address_hash_value_trait<Value> >
+class Open_address_hash
+{
+  static const Key *get_key(const Value &elem)
+  { return Key_trait::get_key(elem); }
+  static bool is_empty(const Value &el) { return Value_trait::is_empty(el); }
+  static bool is_equal(const Value &lhs, const Value &rhs)
+  { return Value_trait::is_equal(lhs, rhs); }
+  static constexpr Value EMPTY= Value_trait::EMPTY;
+public:
+  using Hash_value_type= typename Key_trait::Hash_value_type;
+
+  Open_address_hash()
+  {
+    first.set_mark(true);
+    first.set_ptr(EMPTY);
+    second= EMPTY;
+  }
+
+  ~Open_address_hash()
+  {
+    if (!first.mark())
+    {
+      DBUG_ASSERT(hash_array);
+      free(hash_array);
+    }
+  }
+
+private:
+  size_t _buffer_size() const
+  {
+    return (size_t)1 << capacity_power;
+  }
+  Hash_value_type to_index(const Hash_value_type &hash_value) const
+  {
+    return hash_value & (((Hash_value_type)1 << capacity_power) - 1);
+  }
+
+  Hash_value_type hash_from_value(const Value &value) const
+  {
+    return Key_trait::get_hash_value(get_key(value));
+  }
+
+  bool insert_into_bucket(const Value &value)
+  {
+    auto hash_val= to_index(hash_from_value(value));
+
+    while (!is_empty(hash_array[hash_val]))
+    {
+      if (is_equal(hash_array[hash_val], value))
+        return false;
+      hash_val= to_index(hash_val + 1);
+    }
+
+    hash_array[hash_val]= value;
+    return true;
+  };
+
+  uint rehash_subsequence(uint i)
+  {
+    for (uint j= to_index(i + 1); !is_empty(hash_array[j]); j= to_index(j + 1))
+    {
+      auto temp_el= hash_array[j];
+      if (to_index(hash_from_value(temp_el)) == j)
+        continue;
+      hash_array[j]= EMPTY;
+      insert_into_bucket(temp_el);
+    }
+
+    return i;
+  }
+
+  bool erase_from_bucket(const Value &value)
+  {
+    for (auto key= to_index(Key_trait::get_hash_value(get_key(value)));
+         !is_empty(hash_array[key]); key= to_index(key + 1))
+    {
+      if (is_equal(hash_array[key], value))
+      {
+        hash_array[key]= EMPTY;
+        rehash_subsequence(key);
+        return true;
+      }
+    }
+
+    return false;
+  }
+
+  bool grow(const uint new_capacity_power)
+  {
+    DBUG_ASSERT(new_capacity_power > capacity_power);
+    size_t past_capacity= _buffer_size();
+    size_t capacity= (size_t)1 << new_capacity_power;
+    capacity_power= new_capacity_power;
+    hash_array= (Value *) realloc(hash_array, capacity * sizeof(Value));
+    if (!hash_array)
+      return false;
+    bzero(hash_array + past_capacity,
+          (capacity - past_capacity) * sizeof(Value*));
+
+    for (size_t i= 0; i < capacity; i++)
+    {
+      if (hash_array[i] && i != to_index(hash_from_value(hash_array[i])))
+      {
+        auto temp_el= hash_array[i];
+        hash_array[i]= EMPTY;
+        insert_into_bucket(temp_el);
+      }
+    }
+    return true;
+  }
+
+  void shrink(const uint new_capacity_power)
+  {
+    DBUG_ASSERT(new_capacity_power < capacity_power);
+    size_t past_capacity= _buffer_size();
+    size_t capacity= (size_t)1 << new_capacity_power;
+    capacity_power= new_capacity_power;
+
+    for (size_t i= capacity; i < past_capacity; i++)
+    {
+      if (hash_array[i])
+      {
+        auto temp_el= hash_array[i];
+        insert_into_bucket(temp_el);
+      }
+    }
+
+    hash_array= (Value *) realloc(hash_array, capacity * sizeof(Value));
+  }
+
+
+  bool init_hash_array()
+  {
+    Value _first= first.ptr();
+    Value _second= second;
+
+    capacity_power= CAPACITY_POWER_INITIAL;
+    hash_array= (Value*)calloc(_buffer_size(), sizeof (Value*));
+    _size= 0;
+
+    if (!insert_into_bucket(_first))
+      return false;
+    _size++;
+    if (!insert_into_bucket(_second))
+      return false;
+    _size++;
+
+    return true;
+  }
+
+public:
+  Value find(const Value &elem) const
+  {
+    return find(*Key_trait::get_key(elem),
+                [&elem](const Value &rhs) { return is_equal(rhs, elem); });
+  }
+
+  template <typename Func>
+  Value find(const Key &key, const Func &elem_suits) const
+  {
+    if (first.mark())
+    {
+      if (first.ptr() && elem_suits(first.ptr()))
+        return first.ptr();
+      if (!is_empty(second) && elem_suits(second))
+        return second;
+
+      return EMPTY;
+    }
+
+    for (auto idx= to_index(Key_trait::get_hash_value(&key));
+         !is_empty(hash_array[idx]); idx= to_index(idx + 1))
+    {
+      if (elem_suits(hash_array[idx]))
+        return hash_array[idx];
+    }
+
+    return EMPTY;
+  };
+
+  bool erase(const Value &value)
+  {
+    if (first.mark())
+    {
+      if (!is_empty(first.ptr()) && is_equal(first.ptr(), value))
+      {
+        first.set_ptr(EMPTY);
+        return true;
+      }
+      else if (second && is_equal(second, value))
+      {
+        second= EMPTY;
+        return true;
+      }
+      return false;
+    }
+
+    const size_t capacity= _buffer_size();
+    if (unlikely(capacity > 7 && (_size - 1) * LOW_LOAD_FACTOR < capacity))
+      shrink(capacity_power - 1);
+
+    if (!erase_from_bucket(value))
+      return false;
+    _size--;
+    return true;
+  }
+
+  bool insert(const Value &value)
+  {
+    if (first.mark())
+    {
+      if (is_empty(first.ptr()))
+      {
+        if (is_equal(second, value))
+          return false;
+        first.set_ptr(value);
+        return true;
+      }
+      else if (is_empty(second))
+      {
+        if (is_equal(first.ptr(), value))
+          return false;
+        second= value;
+        return true;
+      }
+      else
+      {
+        first.set_mark(false);
+        if (!init_hash_array())
+          return false;
+      }
+    }
+
+    if (unlikely(_size == TABLE_SIZE_MAX))
+      return false;
+
+    bool res= true;
+    const size_t capacity= _buffer_size();
+    if (unlikely(((ulonglong)_size + 1) * MAX_LOAD_FACTOR > capacity))
+      res= grow(capacity_power + 1);
+
+    res= res && insert_into_bucket(value);
+    if (res)
+      _size++;
+    return res;
+  };
+
+  bool clear()
+  {
+    if (first.mark())
+    {
+      first.set_ptr(EMPTY);
+      second= EMPTY;
+      return true;
+    }
+    if (!hash_array)
+      return false;
+
+    free(hash_array);
+    capacity_power= CAPACITY_POWER_INITIAL;
+
+    first.set_mark(true);
+    first.set_ptr(EMPTY);
+    second= EMPTY;
+
+    return true;
+  }
+
+  size_t size() const
+  { 
+    if (first.mark())
+    {
+      size_t ret_size= 0;
+      if (!is_empty(first.ptr()))
+        ret_size++;
+      if (!is_empty(second))
+        ret_size++;
+      return ret_size;
+    }
+    else
+    {
+      return _size; 
+    }
+  }
+  size_t buffer_size() const { return first.mark() ? 0 : _buffer_size(); }
+
+  Open_address_hash &operator=(const Open_address_hash&)
+  {
+    // Do nothing. Copy operator is called by set_query_tables_list used only for backup.
+    return *this;
+  }
+private:
+  static constexpr uint CAPACITY_POWER_INITIAL= 3;
+  static constexpr ulong MAX_LOAD_FACTOR= 2;
+  static constexpr ulong LOW_LOAD_FACTOR= 10;
+  static constexpr size_t SIZE_BITS= SIZEOF_VOIDP >= 8 ? 58 : 32;
+  static constexpr size_t TABLE_SIZE_MAX= 1LL << SIZE_BITS;
+
+  class markable_reference
+  {
+  public:
+    static constexpr uint MARK_SHIFT = 63;
+    static constexpr uintptr_t MARK_MASK = (uintptr_t)1 << MARK_SHIFT;
+
+    void set_ptr(Value ptr)
+    {
+      p = reinterpret_cast<uintptr_t>(ptr) | (p & MARK_MASK);
+    }
+
+    Value ptr() const { return reinterpret_cast<Value>(p & ~MARK_MASK); }
+
+    void set_mark(bool mark)
+    {
+      p = (p & ~MARK_MASK) | (static_cast<uintptr_t>(mark) << MARK_SHIFT);
+    }
+
+    bool mark() const
+    {
+#if SIZEOF_VOIDP >= 8
+      return p & MARK_MASK;
+#else
+      return false; // 32-bit support: inlining is always disabled.
+#endif
+    }
+
+  private:
+    uintptr_t p;
+  };
+
+  union
+  {
+    struct
+    {
+      markable_reference first;
+      Value second;
+    };
+    struct
+    {
+      Value *hash_array;
+      uint capacity_power: 6;
+      size_t _size: SIZE_BITS;
+    };
+  };
+};
+
+namespace traits
+{
+
+template<typename Key>
+struct Open_address_hash_key_trait
+{
+  public:
+  using Hash_value_type= ulong;
+
+  static Hash_value_type get_hash_value(const Key *key)
+  {
+    ulong nr1= 1, nr2= 4;
+    my_ci_hash_sort(&my_charset_bin, (uchar*) key, sizeof (Key), &nr1, &nr2);
+    return (Hash_value_type) nr1;
+  }
+  /**
+   Function returning key based on value, needed to be able to rehash the table
+   on expansion. Value should be able to return Key from itself.
+   The provided instantiation implements "set", i.e. Key matches Value
+  */
+  static Key *get_key(Key *value) { return value; }
+};
+
+template<typename Value>
+struct Open_address_hash_value_trait
+{
+  static_assert(sizeof (Value) <= sizeof (uintptr_t),
+                "The plain-type Value can only be specified for elements of a bucket size. "
+                "You may have wanted to specify Value=Your_value_type*.");
+  static bool is_equal(const Value &lhs, const Value &rhs)
+  { return lhs == rhs; }
+  /** The following two methods have to be specialized for non-scalar Value */
+  static bool is_empty(const Value el)
+  { return el == 0; }
+};
+
+template<typename T>
+struct Open_address_hash_value_trait<T*>
+{
+  static bool is_equal(const T *lhs, const T *rhs)
+  { return lhs == rhs; }
+  static bool is_empty(const T* el) { return el == nullptr; }
+  static constexpr T *EMPTY= NULL;
+};
+
+}

mysql-test/mariadb-test-run.pl

@@ -3111,7 +3111,7 @@ sub mysql_install_db {
   mtr_add_arg($args, "--core-file");
   mtr_add_arg($args, "--console");
   mtr_add_arg($args, "--character-set-server=latin1");
-  mtr_add_arg($args, "--disable-performance-schema");
+  mtr_add_arg($args, "--loose-disable-performance-schema");
 
   if ( $opt_debug )
   {

sql/mdl.cc

@@ -1056,16 +1056,12 @@ void MDL_request::init_by_key_with_source(const MDL_key *key_arg,
 */
 
 MDL_ticket *MDL_ticket::create(MDL_context *ctx_arg, enum_mdl_type type_arg
-#ifndef DBUG_OFF
                                , enum_mdl_duration duration_arg
-#endif
                                )
 {
   return new (std::nothrow)
              MDL_ticket(ctx_arg, type_arg
-#ifndef DBUG_OFF
                         , duration_arg
-#endif
                         );
 }
 
@@ -1944,8 +1940,9 @@ const LEX_STRING *MDL_ticket::get_type_name(enum_mdl_type type) const
 /**
   Check whether the context already holds a compatible lock ticket
   on an object.
-  Start searching from list of locks for the same duration as lock
-  being requested. If not look at lists for other durations.
+  For the MDL_EXPLICIT duration, look for the tickets with any duration.
+  For other durations, first try to find the ticket with duration other than
+  MDL_EXPLICIT. If not found, check MDL_EXPLICIT duration as well.
 
   @param mdl_request  Lock request object for lock to be acquired
   @param[out] result_duration  Duration of lock which was found.
@@ -1960,30 +1957,31 @@ MDL_ticket *
 MDL_context::find_ticket(MDL_request *mdl_request,
                          enum_mdl_duration *result_duration)
 {
-  MDL_ticket *ticket;
-  int i;
+  const auto &ticket_identical=
+      [&mdl_request](const MDL_ticket *t) {
+        return mdl_request->key.is_equal(t->get_key()) &&
+               t->has_stronger_or_equal_type(mdl_request->type) &&
+               t->m_duration == mdl_request->duration;
+      };
 
-  for (i= 0; i < MDL_DURATION_END; i++)
+  MDL_ticket *found_ticket= ticket_hash.find(&mdl_request->key,
+                                             ticket_identical);
+  if (!found_ticket)
   {
-    enum_mdl_duration duration= (enum_mdl_duration)((mdl_request->duration+i) %
-                                                    MDL_DURATION_END);
-    Ticket_iterator it(m_tickets[duration]);
+    const auto &ticket_still_good=
+        [&mdl_request](const MDL_ticket *t) {
+          return mdl_request->key.is_equal(t->get_key()) &&
+                 t->has_stronger_or_equal_type(mdl_request->type);
+        };
 
-    while ((ticket= it++))
-    {
-      if (mdl_request->key.is_equal(&ticket->m_lock->key) &&
-          ticket->has_stronger_or_equal_type(mdl_request->type))
-      {
-        DBUG_PRINT("info", ("Adding mdl lock %s to %s",
-                            get_mdl_lock_name(mdl_request->key.mdl_namespace(),
-                                              mdl_request->type)->str,
-                            ticket->get_type_name()->str));
-        *result_duration= duration;
-        return ticket;
-      }
-    }
+    found_ticket= ticket_hash.find(&mdl_request->key, ticket_still_good);
   }
-  return NULL;
+
+  if (unlikely(!found_ticket))
+    return NULL;
+
+  *result_duration= found_ticket->m_duration;
+  return found_ticket;
 }
 
 
@@ -2107,11 +2105,8 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
   if (fix_pins())
     return TRUE;
 
-  if (!(ticket= MDL_ticket::create(this, mdl_request->type
-#ifndef DBUG_OFF
-                                   , mdl_request->duration
-#endif
-                                   )))
+  if (!(ticket= MDL_ticket::create(this, mdl_request->type,
+                                   mdl_request->duration)))
     return TRUE;
 
   /* The below call implicitly locks MDL_lock::m_rwlock on success. */
@@ -2138,6 +2133,13 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
 
     mysql_prlock_unlock(&lock->m_rwlock);
 
+    bool success= ticket_hash.insert(ticket);
+    if (!success)
+    {
+      my_error(ER_OUTOFMEMORY, MYF(0));
+      return TRUE;
+    }
+
     m_tickets[mdl_request->duration].push_front(ticket);
 
     mdl_request->ticket= ticket;
@@ -2185,11 +2187,8 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
     we effectively downgrade the cloned lock to the level of
     the request.
   */
-  if (!(ticket= MDL_ticket::create(this, mdl_request->type
-#ifndef DBUG_OFF
-                                   , mdl_request->duration
-#endif
-                                   )))
+  if (!(ticket= MDL_ticket::create(this, mdl_request->type,
+                                   mdl_request->duration)))
     return TRUE;
 
   DBUG_ASSERT(ticket->m_psi == NULL);
@@ -2205,6 +2204,11 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
   DBUG_ASSERT(mdl_request->ticket->has_stronger_or_equal_type(ticket->m_type));
 
   ticket->m_lock= mdl_request->ticket->m_lock;
+  if (!ticket_hash.insert(ticket))
+  {
+    delete ticket;
+    return TRUE;
+  }
   mdl_request->ticket= ticket;
 
   mysql_prlock_wrlock(&ticket->m_lock->m_rwlock);
@@ -2474,6 +2478,8 @@ MDL_context::acquire_lock(MDL_request *mdl_request, double lock_wait_timeout)
   DBUG_ASSERT(wait_status == MDL_wait::GRANTED);
 
   m_tickets[mdl_request->duration].push_front(ticket);
+  bool success = ticket_hash.insert(ticket);
+  DBUG_ASSERT(success);
 
   mdl_request->ticket= ticket;
 
@@ -2636,6 +2642,10 @@ MDL_context::upgrade_shared_lock(MDL_ticket *mdl_ticket,
 
   if (is_new_ticket)
   {
+    bool success= ticket_hash.erase(mdl_xlock_request.ticket);
+    DBUG_ASSERT(success);
+    DBUG_ASSERT(ticket_hash.find(mdl_xlock_request.ticket) == NULL);
+
     m_tickets[MDL_TRANSACTION].remove(mdl_xlock_request.ticket);
     MDL_ticket::destroy(mdl_xlock_request.ticket);
   }
@@ -2903,6 +2913,10 @@ void MDL_context::release_lock(enum_mdl_duration duration, MDL_ticket *ticket)
   DBUG_ASSERT(this == ticket->get_ctx());
   DBUG_PRINT("mdl", ("Released: %s", dbug_print_mdl(ticket)));
 
+  bool success= ticket_hash.erase(ticket);
+  DBUG_ASSERT(success);
+  DBUG_ASSERT(ticket_hash.find(ticket) == NULL);
+
   lock->remove_ticket(m_pins, &MDL_lock::m_granted, ticket);
 
   m_tickets[duration].remove(ticket);
@@ -3202,9 +3216,7 @@ void MDL_context::set_lock_duration(MDL_ticket *mdl_ticket,
 
   m_tickets[MDL_TRANSACTION].remove(mdl_ticket);
   m_tickets[duration].push_front(mdl_ticket);
-#ifndef DBUG_OFF
   mdl_ticket->m_duration= duration;
-#endif
 }
 
 
@@ -3239,12 +3251,10 @@ void MDL_context::set_explicit_duration_for_all_locks()
     }
   }
 
-#ifndef DBUG_OFF
   Ticket_iterator exp_it(m_tickets[MDL_EXPLICIT]);
 
   while ((ticket= exp_it++))
     ticket->m_duration= MDL_EXPLICIT;
-#endif
 }
 
 
@@ -3277,12 +3287,10 @@ void MDL_context::set_transaction_duration_for_all_locks()
     m_tickets[MDL_TRANSACTION].push_front(ticket);
   }
 
-#ifndef DBUG_OFF
   Ticket_iterator trans_it(m_tickets[MDL_TRANSACTION]);
 
   while ((ticket= trans_it++))
     ticket->m_duration= MDL_TRANSACTION;
-#endif
 }
 
 

sql/mdl.h

@@ -23,6 +23,7 @@
 #include <mysql_com.h>
 #include <lf.h>
 #include "lex_ident.h"
+#include "open_address_hash.h"
 
 class THD;
 
@@ -738,15 +739,10 @@ class MDL_ticket : public MDL_wait_for_subgraph, public ilist_node<>
 private:
   friend class MDL_context;
 
-  MDL_ticket(MDL_context *ctx_arg, enum_mdl_type type_arg
-#ifndef DBUG_OFF
-             , enum_mdl_duration duration_arg
-#endif
-            )
+  MDL_ticket(MDL_context *ctx_arg, enum_mdl_type type_arg,
+             enum_mdl_duration duration_arg)
    : m_type(type_arg),
-#ifndef DBUG_OFF
      m_duration(duration_arg),
-#endif
      m_ctx(ctx_arg),
      m_lock(NULL),
      m_psi(NULL)
@@ -757,22 +753,17 @@ class MDL_ticket : public MDL_wait_for_subgraph, public ilist_node<>
     DBUG_ASSERT(m_psi == NULL);
   }
 
-  static MDL_ticket *create(MDL_context *ctx_arg, enum_mdl_type type_arg
-#ifndef DBUG_OFF
-                            , enum_mdl_duration duration_arg
-#endif
-                            );
+  static MDL_ticket *create(MDL_context *ctx_arg, enum_mdl_type type_arg,
+                            enum_mdl_duration duration_arg);
   static void destroy(MDL_ticket *ticket);
 private:
   /** Type of metadata lock. Externally accessible. */
   enum enum_mdl_type m_type;
-#ifndef DBUG_OFF
   /**
     Duration of lock represented by this ticket.
     Context private. Debug-only.
   */
   enum_mdl_duration m_duration;
-#endif
   /**
     Context of the owner of the metadata lock ticket. Externally accessible.
   */
@@ -863,6 +854,24 @@ typedef I_P_List<MDL_request, I_P_List_adapter<MDL_request,
                  I_P_List_counter>
         MDL_request_list;
 
+
+template <typename T>
+struct MDL_key_trait
+{
+  using Hash_value_type= decltype(MDL_key().tc_hash_value());
+
+  static MDL_key *get_key(T *t) { return t->get_key(); }
+  static my_hash_value_type get_hash_value(const MDL_key *key)
+  {
+    return key->tc_hash_value();
+  }
+};
+namespace traits
+{
+template<>
+struct Open_address_hash_key_trait<MDL_key>: public MDL_key_trait<MDL_ticket>{};
+};
+
 /**
   Context of the owner of metadata locks. I.e. each server
   connection has such a context.
@@ -1059,12 +1068,18 @@ class MDL_context
 private:
   MDL_ticket *find_ticket(MDL_request *mdl_req,
                           enum_mdl_duration *duration);
+
   void release_locks_stored_before(enum_mdl_duration duration, MDL_ticket *sentinel);
   void release_lock(enum_mdl_duration duration, MDL_ticket *ticket);
   bool try_acquire_lock_impl(MDL_request *mdl_request,
                              MDL_ticket **out_ticket);
   bool fix_pins();
 
+  /**
+    Ticket hash. Stores only locked tickets.
+  */
+  Open_address_hash<MDL_key, MDL_ticket*> ticket_hash;
+
 public:
   THD *get_thd() const { return m_owner->get_thd(); }
   bool has_explicit_locks();

sql/sp_head.cc

@@ -3678,6 +3678,7 @@ sp_head::add_used_tables_to_table_list(THD *thd,
                                            belong_to_view,
                                            stab->trg_event_map,
                                            query_tables_last_ptr,
+                                           NULL,
                                            stab->for_insert_data);
       tab_buff+= ALIGN_SIZE(sizeof(TABLE_LIST));
       result= TRUE;

sql/sql_base.cc

@@ -4931,23 +4931,17 @@ bool DML_prelocking_strategy::handle_routine(THD *thd,
   return FALSE;
 }
 
-
-/*
-  @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)
-{
-  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 false;
+TABLE_LIST *find_fk_prelocked_table(const Query_tables_list *prelocking_ctx,
+                                    const MDL_key &key,
+                                    thr_lock_type lock_type)
+{
+  return prelocking_ctx->fk_table_hash.find(key,
+          [&key, lock_type](const TABLE_LIST *tl) {
+                  return tl->lock_type >= lock_type
+                         && tl->prelocking_placeholder == TABLE_LIST::PRELOCK_FK
+                         && strcmp(tl->table_name.str, key.name()) == 0
+                         && strcmp(tl->db.str, key.db_name()) == 0;
+  });
 }
 
 
@@ -4999,6 +4993,7 @@ add_internal_tables(THD *thd, Query_tables_list *prelocking_ctx,
                                       TABLE_LIST::PRELOCK_NONE,
                                       0, 0,
                                       &prelocking_ctx->query_tables_last,
+                                      &tables->mdl_request.key,
                                       tables->for_insert_data);
     /*
       Store link to the new table_list that will be used by open so that
@@ -5058,19 +5053,27 @@ 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;
+    MDL_key key(MDL_key::TABLE, fk->foreign_db->str, fk->foreign_table->str);
+
+    TABLE_LIST *tl= find_fk_prelocked_table(prelocking_ctx, key, 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, &key,
+          table_list->for_insert_data);
+      bool success= prelocking_ctx->fk_table_hash.insert(tl);
+      if (!success)
+      {
+        my_error(ER_OUTOFMEMORY, MYF(0));
+        if (arena)
+          thd->restore_active_arena(arena, &backup);
+        DBUG_RETURN(TRUE);
+      }
+    }
 
-    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);
   }
   if (arena)
     thd->restore_active_arena(arena, &backup);

sql/sql_class.h

@@ -54,6 +54,7 @@
 #include "xa.h"
 #include "ddl_log.h"                            /* DDL_LOG_STATE */
 #include "ha_handler_stats.h"                    // ha_handler_stats */
+#include "open_address_hash.h"
 
 extern "C"
 void set_thd_stage_info(void *thd,

sql/sql_lex.cc

@@ -4002,6 +4002,8 @@ void Query_tables_list::reset_query_tables_list(bool init)
   sroutines_list_own_elements= 0;
   binlog_stmt_flags= 0;
   stmt_accessed_table_flag= 0;
+
+  fk_table_hash.clear();
 }
 
 

sql/sql_lex.h

@@ -1767,6 +1767,13 @@ class Query_tables_list
     0 - indicates that this query does not need prelocking.
   */
   TABLE_LIST **query_tables_own_last;
+
+  /**
+    Prelocked tables hash. Stores only tables with FK_PRELOCK tag to speed up
+    table access during prelocking. Deinitializes in reset_query_tables_list.
+  */
+  Open_address_hash<MDL_key, TABLE_LIST*,
+                    MDL_key_trait<TABLE_LIST> > fk_table_hash;
   /*
     Set of stored routines called by statement.
     (Note that we use lazy-initialization for this hash).

sql/table.h

@@ -2310,19 +2310,11 @@ struct TABLE_LIST
     open_and_lock_tables
   */
   inline void reset() { bzero((void*)this, sizeof(*this)); }
-  inline void init_one_table(const LEX_CSTRING *db_arg,
-                             const LEX_CSTRING *table_name_arg,
-                             const LEX_CSTRING *alias_arg,
-                             enum thr_lock_type lock_type_arg)
+  inline void init_one_table_no_request(const LEX_CSTRING *db_arg,
+                                        const LEX_CSTRING *table_name_arg,
+                                        const LEX_CSTRING *alias_arg,
+                                        enum thr_lock_type lock_type_arg)
   {
-    enum enum_mdl_type mdl_type;
-    if (lock_type_arg >= TL_FIRST_WRITE)
-      mdl_type= MDL_SHARED_WRITE;
-    else if (lock_type_arg == TL_READ_NO_INSERT)
-      mdl_type= MDL_SHARED_NO_WRITE;
-    else
-      mdl_type= MDL_SHARED_READ;
-
     reset();
     DBUG_ASSERT(!db_arg->str || strlen(db_arg->str) == db_arg->length);
     DBUG_ASSERT(!table_name_arg->str || strlen(table_name_arg->str) == table_name_arg->length);
@@ -2332,8 +2324,37 @@ struct TABLE_LIST
     alias= (alias_arg ? *alias_arg : *table_name_arg);
     lock_type= lock_type_arg;
     updating= lock_type >= TL_FIRST_WRITE;
+  }
+
+  static enum_mdl_type get_mdl_type(enum thr_lock_type lock_type)
+  {
+    if (lock_type >= TL_FIRST_WRITE)
+      return MDL_SHARED_WRITE;
+    if (lock_type == TL_READ_NO_INSERT)
+      return MDL_SHARED_NO_WRITE;
+    return MDL_SHARED_READ;
+  }
+
+  inline void init_one_table(const LEX_CSTRING *db_arg,
+                             const LEX_CSTRING *table_name_arg,
+                             const LEX_CSTRING *alias_arg,
+                             enum thr_lock_type lock_type_arg)
+  {
+    init_one_table_no_request(db_arg, table_name_arg, alias_arg, lock_type_arg);
+
     MDL_REQUEST_INIT(&mdl_request, MDL_key::TABLE, db.str, table_name.str,
-                     mdl_type, MDL_TRANSACTION);
+                     get_mdl_type(lock_type_arg), MDL_TRANSACTION);
+  }
+  inline void init_one_table_by_key(const LEX_CSTRING *db_arg,
+                                    const LEX_CSTRING *table_name_arg,
+                                    const LEX_CSTRING *alias_arg,
+                                    enum thr_lock_type lock_type_arg,
+                                    const MDL_key *mdl_key)
+  {
+    init_one_table_no_request(db_arg, table_name_arg, alias_arg, lock_type_arg);
+
+    MDL_REQUEST_INIT_BY_KEY(&mdl_request, mdl_key, get_mdl_type(lock_type_arg),
+                            MDL_TRANSACTION);
   }
 
   TABLE_LIST(const LEX_CSTRING *db_arg,
@@ -2360,10 +2381,14 @@ struct TABLE_LIST
                                             TABLE_LIST *belong_to_view_arg,
                                             uint8 trg_event_map_arg,
                                             TABLE_LIST ***last_ptr,
+                                            MDL_key *mdl_key,
                                             my_bool insert_data)
-
   {
-    init_one_table(db_arg, table_name_arg, alias_arg, lock_type_arg);
+    if (mdl_key == NULL)
+      init_one_table(db_arg, table_name_arg, alias_arg, lock_type_arg);
+    else
+      init_one_table_by_key(db_arg, table_name_arg, alias_arg, lock_type_arg,
+                            mdl_key);
     cacheable_table= 1;
     prelocking_placeholder= prelocking_type;
     open_type= (prelocking_type == PRELOCK_ROUTINE ?
@@ -2495,6 +2520,11 @@ struct TABLE_LIST
       table->map= table_map(1) << new_tablenr;
     }
   }
+
+  MDL_key *get_key() 
+  { 
+    return &mdl_request.key;
+  }
   /*
     Reference from aux_tables to local list entry of main select of
     multi-delete statement:

unittest/sql/CMakeLists.txt

@@ -34,3 +34,7 @@ MY_ADD_TEST(mf_iocache)
 ADD_EXECUTABLE(my_json_writer-t my_json_writer-t.cc dummy_builtins.cc)
 TARGET_LINK_LIBRARIES(my_json_writer-t sql mytap)
 MY_ADD_TEST(my_json_writer)
+
+ADD_EXECUTABLE(open_address_hash-t open_address_hash-t.cc)
+TARGET_LINK_LIBRARIES(open_address_hash-t mytap dbug)
+MY_ADD_TEST(open_address_hash)

unittest/sql/open_address_hash-t.cc

+#include <my_global.h>
+#include <tap.h>
+
+#include "open_address_hash.h"
+
+struct identity_key_trait
+{
+  using Hash_value_type= uint32;
+  using Key_type= Hash_value_type;
+
+  static Key_type *get_key(Key_type *elem) { return elem; }
+  static Hash_value_type get_hash_value(const Key_type* elem) { return *elem; }
+};
+
+uint32 data[4][16]= {
+    {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
+    {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
+    {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
+    {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
+};
+
+static void test_pointer_hash_table_with_pointer_equality()
+{
+  Open_address_hash<uint32, uint32*, identity_key_trait> hashie;
+
+  auto *found= hashie.find(data[0]);
+  ok(found == nullptr, "something found in a empty hash!");
+
+  // Insert/delete into
+  ok(!hashie.erase(data[0]), "deletion unexpectedly worked out!");
+  ok(hashie.insert(data[0] + 1), "insertion into empty table failed");
+  ok(!hashie.erase(data[0]), "deletion unexpectedly worked out!");
+  ok(hashie.erase(data[0] + 1), "deletion failed");
+  ok(!hashie.erase(data[0] + 1), "deletion unexpectedly worked out!");
+  ok(hashie.insert(data[0] + 1), "insertion into empty table failed");
+  ok(hashie.insert(data[0] + 2), "insertion failed");
+  ok(hashie.find(data[0] + 1) == data[0] + 1, "find failed");
+  ok(hashie.erase(data[0] + 1), "deletion failed");
+  ok(hashie.find(data[0] + 1) == nullptr, "find after delete succeeded");
+  ok(hashie.find(data[0] + 2) == data[0] + 2, "find failed");
+
+  ok(hashie.insert(data[0] + 1), "insertion failed");
+  ok(hashie.size() == 2, "wrong size");
+  ok(hashie.erase(data[0] + 1), "deletion failed");
+  ok(hashie.find(data[0] + 2) == data[0] + 2,
+                 "find find of second element after delete of first failed");
+
+  ok(hashie.insert(data[0] + 1), "insertion into empty table failed");
+  found= hashie.find(data[1] + 1);
+  ok(found == nullptr, "wrong val with key=1 is found");
+  ok(hashie.erase(data[0] + 2), "deletion failed");
+
+  found= hashie.find(data[0] + 1);
+  ok(found && *found == 1, "1 is not found");
+
+
+  // Expand
+  hashie.insert(data[0]+4);
+  ok(hashie.size() == 2, "wrong size");
+  ok(hashie.buffer_size() == 0, "two elements, why buffer?");
+  hashie.insert(data[0]+5);
+  ok(hashie.size() == 3, "wrong size, %lu", hashie.size());
+
+  // Collision
+  hashie.insert(data[1] + 1); // 1
+  ok(!hashie.insert(data[1] + 1), "collision is not detected.");
+  auto found2= hashie.find(data[1] + 1);
+  ok(found2 != found && *found == *found2, "collision misbehavior");
+
+
+  // Expand on special occasion (offset elements to the beginning)
+  hashie.clear();
+  hashie.insert(data[0] + 14);
+  hashie.insert(data[0] + 15);
+  hashie.insert(data[1] + 15);
+  hashie.insert(data[1] + 14);
+  hashie.insert(data[2] + 15);
+  hashie.insert(data[2] + 14);
+  hashie.insert(data[0] + 1);
+  hashie.insert(data[3] + 14);
+  hashie.insert(data[0] + 2);
+  hashie.insert(data[0] + 3);
+  ok(hashie.find(data[0] + 14) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[0] + 15) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[1] + 15) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[1] + 14) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[2] + 15) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[2] + 14) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[3] + 14) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[0] + 1) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[0] + 2) != nullptr, "expand misbehavior");
+  ok(hashie.find(data[0] + 3) != nullptr, "expand misbehavior");
+}
+
+struct pointer_value_equality_trait:
+        public traits::Open_address_hash_value_trait<uint32*>
+{
+  static bool is_equal(const uint32 *lhs, const uint32 *rhs)
+  {
+    return lhs == rhs || (lhs != nullptr && rhs != nullptr && *lhs == *rhs);
+  }
+};
+
+static void test_hash_table_with_value_equality()
+{
+  Open_address_hash<uint32, uint32*,
+                    identity_key_trait,
+                    pointer_value_equality_trait> hashie;
+  ok(hashie.size() == 0, "hashie is not empty!");
+  ok(hashie.insert(data[0]), "insert to empty hash failed");
+  ok(!hashie.insert(data[0]), "collision insert succeeded");
+  ok(!hashie.insert(data[1]), "insert of the same value succeeded");
+  ok(hashie.find(data[0]) != nullptr, "item not found");
+  ok(hashie.insert(data[0] + 2), "insert to hash failed");
+  ok(hashie.insert(data[0] + 3), "insert to hash failed");
+  ok(hashie.insert(data[0] + 4), "insert to hash failed");
+  ok(hashie.insert(data[0] + 5), "insert to hash failed");
+  ok(hashie.insert(data[0] + 6), "insert to hash failed");
+  ok(hashie.insert(data[0] + 7), "insert to hash failed");
+  ok(hashie.find(data[0] + 2) != nullptr, "item not found");
+  ok(hashie.find(data[0] + 3) != nullptr, "item not found");
+  ok(hashie.find(data[0] + 4) != nullptr, "item not found");
+  ok(hashie.find(data[0] + 8) == nullptr, "item unexpectedly found");
+}
+
+
+int main(int argc __attribute__((unused)),char *argv[])
+{
+  plan(50);
+
+  test_pointer_hash_table_with_pointer_equality();
+  test_hash_table_with_value_equality();
+
+  return 0;
+}