Backport of:

------------------------------------------------------------
revno: 2617.68.24
committer: Dmitry Lenev <dlenev@mysql.com>
branch nick: mysql-next-bg-pre2-2
timestamp: Wed 2009-09-16 17:25:29 +0400
message:
  Pre-requisite patch for fixing bug #30977 "Concurrent statement
  using stored function and DROP FUNCTION breaks SBR".

  Added MDL_request for stored routine as member to Sroutine_hash_entry
  in order to be able perform metadata locking for stored routines in
  future (Sroutine_hash_entry is an equivalent of TABLE_LIST class for
  stored routines).
(WL#4284, follow up fixes).

sql/mdl.cc:
  Introduced version of MDL_request::init() method which initializes
  lock request using pre-built MDL key.
  MDL_key::table_name/table_name_length() getters were
  renamed to reflect the fact that MDL_key objects are
  now created not only for tables.
sql/mdl.h:
  Extended enum_mdl_namespace enum with values which correspond
  to namespaces for stored functions and triggers.
  Renamed MDL_key::table_name/table_name_length() getters
  to MDL_key::name() and name_length() correspondingly to
  reflect the fact that MDL_key objects are now created
  not only for tables.
  Added MDL_key::mdl_namespace() getter.
  Also added version of MDL_request::init() method which
  initializes lock request using pre-built MDL key.
sql/sp.cc:
  Added MDL_request for stored routine as member to Sroutine_hash_entry.
  Changed code to use MDL_key from this request as a key for LEX::sroutines
  set. Removed separate "key" member from Sroutine_hash_entry as it became
  unnecessary.
sql/sp.h:
  Added MDL_request for stored routine as member to Sroutine_hash_entry
  in order to be able perform metadata locking for stored routines in
  future (Sroutine_hash_entry is an equivalent of TABLE_LIST class for
  stored routines).
  Removed Sroutine_hash_entry::key member as now we can use MDL_key from
  this request as a key for LEX::sroutines set.
sql/sp_head.cc:
  Removed sp_name::m_sroutines_key member and set_routine_type() method.
  Since key for routine in LEX::sroutines set has no longer sp_name::m_qname
  as suffix we won't save anything by creating it at sp_name construction
  time.
  Adjusted sp_name constructor used for creating temporary objects for
  lookups in SP-cache to accept MDL_key as parameter and to avoid any
  memory allocation.
  Finally, removed sp_head::m_soutines_key member for reasons similar
  to why sp_name::m_sroutines_key was removed
sql/sp_head.h:
  Removed sp_name::m_sroutines_key member and set_routine_type() method.
  Since key for routine in LEX::sroutines set has no longer sp_name::m_qname
  as suffix we won't save anything by creating it at sp_name construction
  time.
  Adjusted sp_name constructor used for creating temporary objects for
  lookups in SP-cache to accept MDL_key as parameter and to avoid any
  memory allocation.
  Finally, removed sp_head::m_soutines_key member for reasons similar
  to why sp_name::m_sroutines_key was removed.
sql/sql_base.cc:
  Adjusted code to the fact that we now use MDL_key from
  Sroutine_hash_entry::mdl_request as a key for LEX::sroutines set.
  MDL_key::table_name/table_name_length() getters were
  renamed to reflect the fact that MDL_key objects are
  now created not only for tables.
sql/sql_trigger.cc:
  sp_add_used_routine() now takes MDL_key as parameter as now we use
  instance of this class as a key for LEX::sroutines set.

sql/mdl.cc

@@ -318,6 +318,24 @@ void MDL_request::init(enum_mdl_namespace mdl_namespace,
 }
 
 
+/**
+  Initialize a lock request using pre-built MDL_key.
+
+  @sa MDL_request::init(namespace, db, name, type).
+
+  @param key_arg       The pre-built MDL key for the request.
+  @param mdl_type_arg  The MDL lock type for the request.
+*/
+
+void MDL_request::init(const MDL_key *key_arg,
+                       enum enum_mdl_type mdl_type_arg)
+{
+  key.mdl_key_init(key_arg);
+  type= mdl_type_arg;
+  ticket= NULL;
+}
+
+
 /**
   Allocate and initialize one lock request.
 
@@ -1254,7 +1272,7 @@ void MDL_context::release_ticket(MDL_ticket *ticket)
   MDL_lock *lock= ticket->m_lock;
   DBUG_ENTER("release_ticket");
   DBUG_PRINT("enter", ("db=%s name=%s", lock->key.db_name(),
-                                        lock->key.table_name()));
+                                        lock->key.name()));
 
   safe_mutex_assert_owner(&LOCK_mdl);
 
@@ -1526,7 +1544,7 @@ MDL_ticket::set_cached_object(void *cached_object,
 {
   DBUG_ENTER("mdl_set_cached_object");
   DBUG_PRINT("enter", ("db=%s name=%s cached_object=%p",
-                        m_lock->key.db_name(), m_lock->key.table_name(),
+                        m_lock->key.db_name(), m_lock->key.name(),
                         cached_object));
   /*
     TODO: This assumption works now since we do get_cached_object()

sql/mdl.h

@@ -52,9 +52,14 @@ enum enum_mdl_state { MDL_PENDING, MDL_ACQUIRED };
  
   Different types of objects exist in different namespaces
    - MDL_TABLE is for tables and views.
-   - MDL_PROCEDURE is for stored procedures, stored functions and UDFs.
+   - MDL_FUNCTION is for stored functions.
+   - MDL_PROCEDURE is for stored procedures.
+   - MDL_TRIGGER is for triggers.
+  Note that although there isn't metadata locking on triggers,
+  it's necessary to have a separate namespace for them since
+  MDL_key is also used outside of the MDL subsystem.
 */
-enum enum_mdl_namespace { MDL_TABLE=0, MDL_PROCEDURE };
+enum enum_mdl_namespace { MDL_TABLE=0, MDL_FUNCTION, MDL_PROCEDURE, MDL_TRIGGER };
 
 /** Maximal length of key for metadata locking subsystem. */
 #define MAX_MDLKEY_LENGTH (1 + NAME_LEN + 1 + NAME_LEN + 1)
@@ -78,8 +83,11 @@ class MDL_key
   const char *db_name() const { return m_ptr + 1; }
   uint db_name_length() const { return m_db_name_length; }
 
-  const char *table_name() const { return m_ptr + m_db_name_length + 2; }
-  uint table_name_length() const { return m_length - m_db_name_length - 3; }
+  const char *name() const { return m_ptr + m_db_name_length + 2; }
+  uint name_length() const { return m_length - m_db_name_length - 3; }
+
+  enum_mdl_namespace mdl_namespace() const
+  { return (enum_mdl_namespace)(m_ptr[0]); }
 
   /**
     Construct a metadata lock key from a triplet (mdl_namespace, database and name).
@@ -179,6 +187,7 @@ class MDL_request
 public:
   void init(enum_mdl_namespace namespace_arg, const char *db_arg, const char *name_arg,
             enum_mdl_type mdl_type_arg);
+  void init(const MDL_key *key_arg, enum_mdl_type mdl_type_arg);
   /** Set type of lock request. Can be only applied to pending locks. */
   inline void set_type(enum_mdl_type type_arg)
   {

sql/sp.cc

@@ -1391,8 +1391,8 @@ extern "C" uchar* sp_sroutine_key(const uchar *ptr, size_t *plen,
                                   my_bool first)
 {
   Sroutine_hash_entry *rn= (Sroutine_hash_entry *)ptr;
-  *plen= rn->key.length;
-  return (uchar *)rn->key.str;
+  *plen= rn->mdl_request.key.length();
+  return (uchar *)rn->mdl_request.key.ptr();
 }
 
 
@@ -1430,23 +1430,19 @@ extern "C" uchar* sp_sroutine_key(const uchar *ptr, size_t *plen,
 */
 
 bool sp_add_used_routine(Query_tables_list *prelocking_ctx, Query_arena *arena,
-                         const LEX_STRING *key, TABLE_LIST *belong_to_view)
+                         const MDL_key *key, TABLE_LIST *belong_to_view)
 {
   my_hash_init_opt(&prelocking_ctx->sroutines, system_charset_info,
                    Query_tables_list::START_SROUTINES_HASH_SIZE,
                    0, 0, sp_sroutine_key, 0, 0);
 
-  if (!my_hash_search(&prelocking_ctx->sroutines, (uchar *)key->str,
-                      key->length))
+  if (!my_hash_search(&prelocking_ctx->sroutines, key->ptr(), key->length()))
   {
     Sroutine_hash_entry *rn=
-      (Sroutine_hash_entry *)arena->alloc(sizeof(Sroutine_hash_entry) +
-                                          key->length + 1);
+      (Sroutine_hash_entry *)arena->alloc(sizeof(Sroutine_hash_entry));
     if (!rn)              // OOM. Error will be reported using fatal_error().
       return FALSE;
-    rn->key.length= key->length;
-    rn->key.str= (char *)rn + sizeof(Sroutine_hash_entry);
-    memcpy(rn->key.str, key->str, key->length + 1);
+    rn->mdl_request.init(key, MDL_SHARED);
     my_hash_insert(&prelocking_ctx->sroutines, (uchar *)rn);
     prelocking_ctx->sroutines_list.link_in_list((uchar *)rn, (uchar **)&rn->next);
     rn->belong_to_view= belong_to_view;
@@ -1477,8 +1473,9 @@ bool sp_add_used_routine(Query_tables_list *prelocking_ctx, Query_arena *arena,
 void sp_add_used_routine(Query_tables_list *prelocking_ctx, Query_arena *arena,
                          sp_name *rt, char rt_type)
 {
-  rt->set_routine_type(rt_type);
-  (void)sp_add_used_routine(prelocking_ctx, arena, &rt->m_sroutines_key, 0);
+  MDL_key key((rt_type == TYPE_ENUM_FUNCTION) ? MDL_FUNCTION : MDL_PROCEDURE,
+              rt->m_db.str, rt->m_name.str);
+  (void)sp_add_used_routine(prelocking_ctx, arena, &key, 0);
   prelocking_ctx->sroutines_list_own_last= prelocking_ctx->sroutines_list.next;
   prelocking_ctx->sroutines_list_own_elements=
                     prelocking_ctx->sroutines_list.elements;
@@ -1535,7 +1532,8 @@ void sp_update_sp_used_routines(HASH *dst, HASH *src)
   for (uint i=0 ; i < src->records ; i++)
   {
     Sroutine_hash_entry *rt= (Sroutine_hash_entry *)my_hash_element(src, i);
-    if (!my_hash_search(dst, (uchar *)rt->key.str, rt->key.length))
+    if (!my_hash_search(dst, (uchar *)rt->mdl_request.key.ptr(),
+                        rt->mdl_request.key.length()))
       my_hash_insert(dst, (uchar *)rt);
   }
 }
@@ -1562,8 +1560,8 @@ sp_update_stmt_used_routines(THD *thd, Query_tables_list *prelocking_ctx,
   for (uint i=0 ; i < src->records ; i++)
   {
     Sroutine_hash_entry *rt= (Sroutine_hash_entry *)my_hash_element(src, i);
-    (void)sp_add_used_routine(prelocking_ctx, thd->stmt_arena, &rt->key,
-                              belong_to_view);
+    (void)sp_add_used_routine(prelocking_ctx, thd->stmt_arena,
+                              &rt->mdl_request.key, belong_to_view);
   }
 }
 
@@ -1587,8 +1585,8 @@ void sp_update_stmt_used_routines(THD *thd, Query_tables_list *prelocking_ctx,
 {
   for (Sroutine_hash_entry *rt= (Sroutine_hash_entry *)src->first;
        rt; rt= rt->next)
-    (void)sp_add_used_routine(prelocking_ctx, thd->stmt_arena, &rt->key,
-                              belong_to_view);
+    (void)sp_add_used_routine(prelocking_ctx, thd->stmt_arena,
+                              &rt->mdl_request.key, belong_to_view);
 }
 
 

sql/sp.h

@@ -73,9 +73,10 @@ class Sroutine_hash_entry
 {
 public:
   /**
-    Set key consisting of one-byte routine type and quoted routine name.
+    Metadata lock request for routine.
+    MDL_key in this request is also used as a key for set.
   */
-  LEX_STRING key;
+  MDL_request mdl_request;
   /**
     Next element in list linking all routines in set. See also comments
     for LEX::sroutine/sroutine_list and sp_head::m_sroutines.
@@ -96,7 +97,7 @@ class Sroutine_hash_entry
 void sp_add_used_routine(Query_tables_list *prelocking_ctx, Query_arena *arena,
                          sp_name *rt, char rt_type);
 bool sp_add_used_routine(Query_tables_list *prelocking_ctx, Query_arena *arena,
-                         const LEX_STRING *key, TABLE_LIST *belong_to_view);
+                         const MDL_key *key, TABLE_LIST *belong_to_view);
 void sp_remove_not_own_routines(Query_tables_list *prelocking_ctx);
 void sp_update_sp_used_routines(HASH *dst, HASH *src);
 void sp_update_stmt_used_routines(THD *thd, Query_tables_list *prelocking_ctx,

sql/sp_head.cc

@@ -382,31 +382,34 @@ sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
 }
 
 
-/*
- *
- *  sp_name
- *
- */
+/**
+  Create temporary sp_name object from MDL key.
 
-sp_name::sp_name(THD *thd, char *key, uint key_len)
+  @note The lifetime of this object is bound to the lifetime of the MDL_key.
+        This should be fine as sp_name objects created by this constructor
+        are mainly used for SP-cache lookups.
+
+  @param key         MDL key containing database and routine name.
+  @param qname_buff  Buffer to be used for storing quoted routine name
+                     (should be at least 2*NAME_LEN+1+1 bytes).
+*/
+
+sp_name::sp_name(const MDL_key *key, char *qname_buff)
 {
-  m_sroutines_key.str= key;
-  m_sroutines_key.length= key_len;
-  m_qname.str= ++key;
-  m_qname.length= key_len - 1;
-  if ((m_name.str= strchr(m_qname.str, '.')))
+  m_db.str= (char*)key->db_name();
+  m_db.length= key->db_name_length();
+  m_name.str= (char*)key->name();
+  m_name.length= key->name_length();
+  m_qname.str= qname_buff;
+  if (m_db.length)
   {
-    m_db.length= m_name.str - key;
-    m_db.str= strmake_root(thd->mem_root, key, m_db.length);
-    m_name.str++;
-    m_name.length= m_qname.length - m_db.length - 1;
+    strxmov(qname_buff, m_db.str, ".", m_name.str, NullS);
+    m_qname.length= m_db.length + 1 + m_name.length;
   }
   else
   {
-    m_name.str= m_qname.str;
-    m_name.length= m_qname.length;
-    m_db.str= 0;
-    m_db.length= 0;
+    strmov(qname_buff, m_name.str);
+    m_qname.length= m_name.length;
   }
   m_explicit_name= false;
 }
@@ -419,12 +422,10 @@ void
 sp_name::init_qname(THD *thd)
 {
   const uint dot= !!m_db.length;
-  /* m_sroutines format: m_type + [database + dot] + name + nul */
-  m_sroutines_key.length= 1 + m_db.length + dot + m_name.length;
-  if (!(m_sroutines_key.str= (char*) thd->alloc(m_sroutines_key.length + 1)))
+  /* m_qname format: [database + dot] + name + '\0' */
+  m_qname.length= m_db.length + dot + m_name.length;
+  if (!(m_qname.str= (char*) thd->alloc(m_qname.length + 1)))
     return;
-  m_qname.length= m_sroutines_key.length - 1;
-  m_qname.str= m_sroutines_key.str + 1;
   sprintf(m_qname.str, "%.*s%.*s%.*s",
           (int) m_db.length, (m_db.length ? m_db.str : ""),
           dot, ".",
@@ -585,9 +586,6 @@ sp_head::init(LEX *lex)
   m_defstr.str= NULL;
   m_defstr.length= 0;
 
-  m_sroutines_key.str= NULL;
-  m_sroutines_key.length= 0;
-
   m_return_field_def.charset= NULL;
 
   DBUG_VOID_RETURN;
@@ -617,14 +615,10 @@ sp_head::init_sp_name(THD *thd, sp_name *spname)
   if (spname->m_qname.length == 0)
     spname->init_qname(thd);
 
-  m_sroutines_key.length= spname->m_sroutines_key.length;
-  m_sroutines_key.str= (char*) memdup_root(thd->mem_root,
-                                           spname->m_sroutines_key.str,
-                                           spname->m_sroutines_key.length + 1);
-  m_sroutines_key.str[0]= static_cast<char>(m_type);
-
-  m_qname.length= m_sroutines_key.length - 1;
-  m_qname.str= m_sroutines_key.str + 1;
+  m_qname.length= spname->m_qname.length;
+  m_qname.str= (char*) memdup_root(thd->mem_root,
+                                   spname->m_qname.str,
+                                   spname->m_qname.length + 1);
 
   DBUG_VOID_RETURN;
 }

sql/sp_head.h

@@ -109,36 +109,21 @@ class sp_name : public Sql_alloc
   LEX_STRING m_db;
   LEX_STRING m_name;
   LEX_STRING m_qname;
-  /**
-    Key representing routine in the set of stored routines used by statement.
-    Consists of 1-byte routine type and m_qname (which usually refences to
-    same buffer). Note that one must complete initialization of the key by
-    calling set_routine_type().
-  */
-  LEX_STRING m_sroutines_key;
   bool       m_explicit_name;                   /**< Prepend the db name? */
 
   sp_name(LEX_STRING db, LEX_STRING name, bool use_explicit_name)
     : m_db(db), m_name(name), m_explicit_name(use_explicit_name)
   {
-    m_qname.str= m_sroutines_key.str= 0;
-    m_qname.length= m_sroutines_key.length= 0;
+    m_qname.str= 0;
+    m_qname.length= 0;
   }
 
-  /**
-    Creates temporary sp_name object from key, used mainly
-    for SP-cache lookups.
-  */
-  sp_name(THD *thd, char *key, uint key_len);
+  /** Create temporary sp_name object from MDL key. */
+  sp_name(const MDL_key *key, char *qname_buff);
 
   // Init. the qualified name from the db and name.
   void init_qname(THD *thd);	// thd for memroot allocation
 
-  void set_routine_type(char type)
-  {
-    m_sroutines_key.str[0]= type;
-  }
-
   ~sp_name()
   {}
 };
@@ -181,12 +166,6 @@ class sp_head :private Query_arena
   ulong m_sql_mode;		///< For SHOW CREATE and execution
   LEX_STRING m_qname;		///< db.name
   bool m_explicit_name;         ///< Prepend the db name? */
-  /**
-    Key representing routine in the set of stored routines used by statement.
-    [routine_type]db.name
-    @sa sp_name::m_sroutines_key
-  */
-  LEX_STRING m_sroutines_key;
   LEX_STRING m_db;
   LEX_STRING m_name;
   LEX_STRING m_params;

sql/sql_base.cc

@@ -3777,17 +3777,18 @@ open_and_process_routine(THD *thd, Query_tables_list *prelocking_ctx,
                          Prelocking_strategy *prelocking_strategy,
                          bool *need_prelocking)
 {
-  int type= rt->key.str[0];
-
   DBUG_ENTER("open_and_process_routine");
 
-  switch (type)
+  switch (rt->mdl_request.key.mdl_namespace())
   {
-  case TYPE_ENUM_FUNCTION:
-  case TYPE_ENUM_PROCEDURE:
+  case MDL_FUNCTION:
+  case MDL_PROCEDURE:
     {
-      sp_name name(thd, rt->key.str, rt->key.length);
+      char qname_buff[NAME_LEN*2+1+1];
+      sp_name name(&rt->mdl_request.key, qname_buff);
       sp_head *sp;
+      int type= (rt->mdl_request.key.mdl_namespace() == MDL_FUNCTION) ?
+                TYPE_ENUM_FUNCTION : TYPE_ENUM_PROCEDURE;
 
       if (sp_cache_routine(thd, type, &name, &sp))
         DBUG_RETURN(TRUE);
@@ -3799,7 +3800,7 @@ open_and_process_routine(THD *thd, Query_tables_list *prelocking_ctx,
       }
     }
     break;
-  case TYPE_ENUM_TRIGGER:
+  case MDL_TRIGGER:
     break;
   default:
     /* Impossible type value. */
@@ -4304,7 +4305,7 @@ handle_routine(THD *thd, Query_tables_list *prelocking_ctx,
   */
 
   if (rt != (Sroutine_hash_entry*)prelocking_ctx->sroutines_list.first ||
-      rt->key.str[0] != TYPE_ENUM_PROCEDURE)
+      rt->mdl_request.key.mdl_namespace() != MDL_PROCEDURE)
   {
     *need_prelocking= TRUE;
     sp_update_stmt_used_routines(thd, prelocking_ctx, &sp->m_sroutines,
@@ -8302,7 +8303,7 @@ tdc_wait_for_old_versions(THD *thd, MDL_request_list *mdl_requests)
     while ((mdl_request= it++))
     {
       if ((share= get_cached_table_share(mdl_request->key.db_name(),
-                                         mdl_request->key.table_name())) &&
+                                         mdl_request->key.name())) &&
           share->version != refresh_version &&
           !share->used_tables.is_empty())
         break;

sql/sql_trigger.cc

@@ -2055,17 +2055,21 @@ add_tables_and_routines_for_triggers(THD *thd,
         /* We can have only one trigger per action type currently */
         sp_head *trigger= table_list->table->triggers->bodies[i][j];
 
-        if (trigger && sp_add_used_routine(prelocking_ctx, thd->stmt_arena,
-                                           &trigger->m_sroutines_key,
-                                           table_list->belong_to_view))
+        if (trigger)
         {
-          trigger->add_used_tables_to_table_list(thd,
-                                            &prelocking_ctx->query_tables_last,
-                                            table_list->belong_to_view);
-          sp_update_stmt_used_routines(thd, prelocking_ctx,
-                                       &trigger->m_sroutines,
-                                       table_list->belong_to_view);
-          trigger->propagate_attributes(prelocking_ctx);
+          MDL_key key(MDL_TRIGGER, trigger->m_db.str, trigger->m_name.str);
+
+          if (sp_add_used_routine(prelocking_ctx, thd->stmt_arena,
+                                  &key, table_list->belong_to_view))
+          {
+            trigger->add_used_tables_to_table_list(thd,
+                       &prelocking_ctx->query_tables_last,
+                       table_list->belong_to_view);
+            sp_update_stmt_used_routines(thd, prelocking_ctx,
+                                         &trigger->m_sroutines,
+                                         table_list->belong_to_view);
+            trigger->propagate_attributes(prelocking_ctx);
+          }
         }
       }
     }