Backport of revno: 2617.69.40

A pre-requisite patch for Bug#30977 "Concurrent statement using 
stored function and DROP FUNCTION breaks SBR".

This patch changes the MDL API by introducing a namespace for
lock keys: MDL_TABLE for tables and views and MDL_PROCEDURE
for stored procedures and functions. The latter is needed for
the fix for Bug#30977.

sql/lock.cc

@@ -957,7 +957,7 @@ bool lock_table_names(THD *thd, TABLE_LIST *table_list)
 
   for (lock_table= table_list; lock_table; lock_table= lock_table->next_local)
   {
-    lock_table->mdl_request.init(0, lock_table->db, lock_table->table_name,
+    lock_table->mdl_request.init(MDL_TABLE, lock_table->db, lock_table->table_name,
                                  MDL_EXCLUSIVE);
     mdl_requests.push_front(&lock_table->mdl_request);
   }

sql/mdl.cc

@@ -301,22 +301,18 @@ void MDL_context::merge(MDL_context *src)
 
   The MDL subsystem does not own or manage memory of lock requests.
 
-  @param  type       Id of type of object to be locked
-  @param  db         Name of database to which the object belongs
-  @param  name       Name of of the object
-  @param  mdl_type   The MDL lock type for the request.
-
-  Suggested lock types: TABLE - 0 PROCEDURE - 1 FUNCTION - 2
-  Note that tables and views must have the same lock type, since
-  they share the same name space in the SQL standard.
+  @param  mdl_namespace  Id of namespace of object to be locked
+  @param  db             Name of database to which the object belongs
+  @param  name           Name of of the object
+  @param  mdl_type       The MDL lock type for the request.
 */
 
-void MDL_request::init(unsigned char type_arg,
+void MDL_request::init(enum_mdl_namespace mdl_namespace,
                        const char *db_arg,
                        const char *name_arg,
                        enum enum_mdl_type mdl_type_arg)
 {
-  key.mdl_key_init(type_arg, db_arg, name_arg);
+  key.mdl_key_init(mdl_namespace, db_arg, name_arg);
   type= mdl_type_arg;
   ticket= NULL;
 }
@@ -329,10 +325,10 @@ void MDL_request::init(unsigned char type_arg,
   on a memory root. Necessary to lock ad-hoc tables, e.g.
   mysql.* tables of grant and data dictionary subsystems.
 
-  @param  type       Id of type of object to be locked
-  @param  db         Name of database to which object belongs
-  @param  name       Name of of object
-  @param  root       MEM_ROOT on which object should be allocated
+  @param  mdl_namespace  Id of namespace of object to be locked
+  @param  db             Name of database to which object belongs
+  @param  name           Name of of object
+  @param  root           MEM_ROOT on which object should be allocated
 
   @note The allocated lock request will have MDL_SHARED type.
 
@@ -341,7 +337,7 @@ void MDL_request::init(unsigned char type_arg,
 */
 
 MDL_request *
-MDL_request::create(unsigned char type, const char *db,
+MDL_request::create(enum_mdl_namespace mdl_namespace, const char *db,
                     const char *name, enum_mdl_type mdl_type,
                     MEM_ROOT *root)
 {
@@ -350,7 +346,7 @@ MDL_request::create(unsigned char type, const char *db,
   if (!(mdl_request= (MDL_request*) alloc_root(root, sizeof(MDL_request))))
     return NULL;
 
-  mdl_request->init(type, db, name, mdl_type);
+  mdl_request->init(mdl_namespace, db, name, mdl_type);
 
   return mdl_request;
 }
@@ -1418,20 +1414,20 @@ void MDL_context::release_global_shared_lock()
   Auxiliary function which allows to check if we have exclusive lock
   on the object.
 
-  @param type    Id of object type
-  @param db      Name of the database
-  @param name    Name of the object
+  @param mdl_namespace Id of object namespace
+  @param db            Name of the database
+  @param name          Name of the object
 
   @return TRUE if current context contains exclusive lock for the object,
           FALSE otherwise.
 */
 
 bool
-MDL_context::is_exclusive_lock_owner(unsigned char type,
+MDL_context::is_exclusive_lock_owner(enum_mdl_namespace mdl_namespace,
                                      const char *db, const char *name)
 {
   MDL_request mdl_request;
-  mdl_request.init(type, db, name, MDL_EXCLUSIVE);
+  mdl_request.init(mdl_namespace, db, name, MDL_EXCLUSIVE);
   MDL_ticket *ticket= find_ticket(&mdl_request);
 
   DBUG_ASSERT(ticket == NULL || ticket->m_state == MDL_ACQUIRED);
@@ -1444,19 +1440,19 @@ MDL_context::is_exclusive_lock_owner(unsigned char type,
   Auxiliary function which allows to check if we have some kind of lock on
   a object.
 
-  @param type    Id of object type
-  @param db      Name of the database
-  @param name    Name of the object
+  @param mdl_namespace Id of object namespace
+  @param db            Name of the database
+  @param name          Name of the object
 
   @return TRUE if current context contains satisfied lock for the object,
           FALSE otherwise.
 */
 
 bool
-MDL_context::is_lock_owner(unsigned char type,
+MDL_context::is_lock_owner(enum_mdl_namespace mdl_namespace,
                            const char *db, const char *name)
 {
-  MDL_key key(type, db, name);
+  MDL_key key(mdl_namespace, db, name);
   MDL_ticket *ticket;
   MDL_context::Ticket_iterator it(m_tickets);
 

sql/mdl.h

@@ -47,6 +47,14 @@ enum enum_mdl_type {MDL_SHARED=0, MDL_SHARED_HIGH_PRIO,
 
 enum enum_mdl_state { MDL_PENDING, MDL_ACQUIRED };
 
+/**
+  Object namespaces
+ 
+  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.
+*/
+enum enum_mdl_namespace { MDL_TABLE=0, MDL_PROCEDURE };
 
 /** Maximal length of key for metadata locking subsystem. */
 #define MAX_MDLKEY_LENGTH (1 + NAME_LEN + 1 + NAME_LEN + 1)
@@ -74,18 +82,18 @@ class MDL_key
   uint table_name_length() const { return m_length - m_db_name_length - 3; }
 
   /**
-    Construct a metadata lock key from a triplet (type, database and name).
+    Construct a metadata lock key from a triplet (mdl_namespace, database and name).
 
-    @remark The key for a table is <0 (=table)>+<database name>+<table name>
+    @remark The key for a table is <mdl_namespace>+<database name>+<table name>
 
-    @param  type   Id of type of object to be locked
-    @param  db     Name of database to which the object belongs
-    @param  name   Name of of the object
-    @param  key    Where to store the the MDL key.
+    @param  mdl_namespace Id of namespace of object to be locked
+    @param  db            Name of database to which the object belongs
+    @param  name          Name of of the object
+    @param  key           Where to store the the MDL key.
   */
-  void mdl_key_init(char type, const char *db, const char *name)
+  void mdl_key_init(enum_mdl_namespace mdl_namespace, const char *db, const char *name)
   {
-    m_ptr[0]= type;
+    m_ptr[0]= (char) mdl_namespace;
     m_db_name_length= (uint) (strmov(m_ptr + 1, db) - m_ptr - 1);
     m_length= (uint) (strmov(m_ptr + m_db_name_length + 2, name) - m_ptr + 1);
   }
@@ -104,11 +112,12 @@ class MDL_key
   {
     mdl_key_init(rhs);
   }
-  MDL_key(char type_arg, const char *db_arg, const char *name_arg)
+  MDL_key(enum_mdl_namespace namespace_arg, const char *db_arg, const char *name_arg)
   {
-    mdl_key_init(type_arg, db_arg, name_arg);
+    mdl_key_init(namespace_arg, db_arg, name_arg);
   }
   MDL_key() {} /* To use when part of MDL_request. */
+
 private:
   char m_ptr[MAX_MDLKEY_LENGTH];
   uint m_length;
@@ -168,7 +177,7 @@ class MDL_request
   MDL_key key;
 
 public:
-  void init(unsigned char type_arg, const char *db_arg, const char *name_arg,
+  void init(enum_mdl_namespace namespace_arg, const char *db_arg, const char *name_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)
@@ -178,7 +187,7 @@ class MDL_request
   }
   bool is_shared() const { return type < MDL_EXCLUSIVE; }
 
-  static MDL_request *create(unsigned char type, const char *db,
+  static MDL_request *create(enum_mdl_namespace mdl_namespace, const char *db,
                              const char *name, enum_mdl_type mdl_type,
                              MEM_ROOT *root);
 
@@ -318,10 +327,10 @@ class MDL_context
   void release_lock(MDL_ticket *ticket);
   void release_global_shared_lock();
 
-  bool is_exclusive_lock_owner(unsigned char type,
+  bool is_exclusive_lock_owner(enum_mdl_namespace mdl_namespace,
                                const char *db,
                                const char *name);
-  bool is_lock_owner(unsigned char type, const char *db, const char *name);
+  bool is_lock_owner(enum_mdl_namespace mdl_namespace, const char *db, const char *name);
 
   inline bool has_locks() const
   {

sql/sp_head.cc

@@ -3981,7 +3981,7 @@ sp_head::add_used_tables_to_table_list(THD *thd,
       table->prelocking_placeholder= 1;
       table->belong_to_view= belong_to_view;
       table->trg_event_map= stab->trg_event_map;
-      table->mdl_request.init(0, table->db, table->table_name, MDL_SHARED);
+      table->mdl_request.init(MDL_TABLE, table->db, table->table_name, MDL_SHARED);
 
       /* Everyting else should be zeroed */
 
@@ -4023,7 +4023,7 @@ sp_add_to_query_tables(THD *thd, LEX *lex,
   table->lock_type= locktype;
   table->select_lex= lex->current_select;
   table->cacheable_table= 1;
-  table->mdl_request.init(0, table->db, table->table_name, MDL_SHARED);
+  table->mdl_request.init(MDL_TABLE, table->db, table->table_name, MDL_SHARED);
 
   lex->add_to_query_tables(table);
   return table;

sql/sql_base.cc

@@ -474,7 +474,7 @@ TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
     To be able perform any operation on table we should own
     some kind of metadata lock on it.
   */
-  DBUG_ASSERT(thd->mdl_context.is_lock_owner(0, table_list->db,
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_TABLE, table_list->db,
                                              table_list->table_name));
 
   /* Read table definition from cache */
@@ -2541,7 +2541,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
       TABLES breaks metadata locking protocol (potentially can lead
       to deadlocks) it should be disallowed.
     */
-    if (thd->mdl_context.is_lock_owner(0, table_list->db,
+    if (thd->mdl_context.is_lock_owner(MDL_TABLE, table_list->db,
                                        table_list->table_name))
     {
       char path[FN_REFLEN + 1];
@@ -8158,7 +8158,7 @@ void tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
   safe_mutex_assert_owner(&LOCK_open);
 
   DBUG_ASSERT(remove_type == TDC_RT_REMOVE_UNUSED ||
-              thd->mdl_context.is_exclusive_lock_owner(0, db, table_name));
+              thd->mdl_context.is_exclusive_lock_owner(MDL_TABLE, db, table_name));
 
   key_length=(uint) (strmov(strmov(key,db)+1,table_name)-key)+1;
 

sql/sql_delete.cc

@@ -1177,7 +1177,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
         thd->lex->alter_info.flags & ALTER_ADMIN_PARTITION)
       goto trunc_by_del;
 
-    mdl_request.init(0, table_list->db, table_list->table_name, MDL_EXCLUSIVE);
+    mdl_request.init(MDL_TABLE, table_list->db, table_list->table_name, MDL_EXCLUSIVE);
     if (thd->mdl_context.acquire_exclusive_lock(&mdl_request))
       DBUG_RETURN(TRUE);
 

sql/sql_handler.cc

@@ -264,7 +264,7 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
     memcpy(hash_tables->db, tables->db, dblen);
     memcpy(hash_tables->table_name, tables->table_name, namelen);
     memcpy(hash_tables->alias, tables->alias, aliaslen);
-    hash_tables->mdl_request.init(0, db, name, MDL_SHARED);
+    hash_tables->mdl_request.init(MDL_TABLE, db, name, MDL_SHARED);
 
     /* add to hash */
     if (my_hash_insert(&thd->handler_tables_hash, (uchar*) hash_tables))

sql/sql_parse.cc

@@ -6024,7 +6024,7 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
   ptr->next_name_resolution_table= NULL;
   /* Link table in global list (all used tables) */
   lex->add_to_query_tables(ptr);
-  ptr->mdl_request.init(0, ptr->db, ptr->table_name, MDL_SHARED);
+  ptr->mdl_request.init(MDL_TABLE, ptr->db, ptr->table_name, MDL_SHARED);
   DBUG_RETURN(ptr);
 }
 

sql/sql_show.cc

@@ -3068,7 +3068,7 @@ static bool
 acquire_high_prio_shared_mdl_lock(THD *thd, TABLE_LIST *table)
 {
   bool error;
-  table->mdl_request.init(0, table->db, table->table_name,
+  table->mdl_request.init(MDL_TABLE, table->db, table->table_name,
                           MDL_SHARED_HIGH_PRIO);
   while (!(error=
            thd->mdl_context.try_acquire_shared_lock(&table->mdl_request)) &&

sql/sql_table.cc

@@ -4140,7 +4140,7 @@ bool mysql_create_table(THD *thd, const char *db, const char *table_name,
 
   if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
   {
-    target_mdl_request.init(0, db, table_name, MDL_EXCLUSIVE);
+    target_mdl_request.init(MDL_TABLE, db, table_name, MDL_EXCLUSIVE);
     if (thd->mdl_context.try_acquire_exclusive_lock(&target_mdl_request))
     {
       result= TRUE;
@@ -4361,7 +4361,7 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
     uint key_length;
 
     key_length= create_table_def_key(thd, key, table_list, 0);
-    table_list->mdl_request.init(0, table_list->db, table_list->table_name,
+    table_list->mdl_request.init(MDL_TABLE, table_list->db, table_list->table_name,
                                  MDL_EXCLUSIVE);
     if (thd->mdl_context.acquire_exclusive_lock(&table_list->mdl_request))
       DBUG_RETURN(0);
@@ -5271,7 +5271,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   }
   else
   {
-    table->mdl_request.init(0, db, table_name, MDL_EXCLUSIVE);
+    table->mdl_request.init(MDL_TABLE, db, table_name, MDL_EXCLUSIVE);
     if (thd->mdl_context.try_acquire_exclusive_lock(&table->mdl_request))
       DBUG_RETURN(TRUE);
 
@@ -6637,7 +6637,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       }
       else
       {
-        target_mdl_request.init(0, new_db, new_name, MDL_EXCLUSIVE);
+        target_mdl_request.init(MDL_TABLE, new_db, new_name, MDL_EXCLUSIVE);
         if (thd->mdl_context.try_acquire_exclusive_lock(&target_mdl_request))
           DBUG_RETURN(TRUE);
         if (target_mdl_request.ticket == NULL)

sql/sql_trigger.cc

@@ -1883,7 +1883,7 @@ bool Table_triggers_list::change_table_name(THD *thd, const char *db,
     In the future, only an exclusive metadata lock will be enough.
   */
 #ifndef DBUG_OFF
-  if (thd->mdl_context.is_exclusive_lock_owner(0, db, old_table))
+  if (thd->mdl_context.is_exclusive_lock_owner(MDL_TABLE, db, old_table))
     safe_mutex_assert_owner(&LOCK_open);
 #endif
 

sql/table.cc

@@ -4822,7 +4822,7 @@ size_t max_row_length(TABLE *table, const uchar *data)
 void init_mdl_requests(TABLE_LIST *table_list)
 {
   for ( ; table_list ; table_list= table_list->next_global)
-    table_list->mdl_request.init(0, table_list->db, table_list->table_name,
+    table_list->mdl_request.init(MDL_TABLE, table_list->db, table_list->table_name,
                                  MDL_SHARED);
 }
 

sql/table.h

@@ -1125,7 +1125,7 @@ struct TABLE_LIST
     table_name_length= table_name_length_arg;
     alias= (char*) alias_arg;
     lock_type= lock_type_arg;
-    mdl_request.init(0, db, table_name, MDL_SHARED);
+    mdl_request.init(MDL_TABLE, db, table_name, MDL_SHARED);
   }
 
   /*