Implementation of Monty's idea: Now we can open mysql.proc table for lookup

of stored routines definitions even if we already have some tables open and
locked. To avoid deadlocks in this case we have to put certain restrictions
on locking of mysql.proc table.

This allows to use stored routines safely under LOCK TABLES without explicitly
mentioning mysql.proc in the list of locked tables. It also fixes bug #11554
"Server crashes on statement indirectly using non-cached function".


mysql-test/r/sp-error.result:
  Added test which checks that now we can read stored routines definitions
  under LOCK TABLES even if we have not locked mysql.proc explicitly. Also
  added check for restrictions which this ability puts on mysql.proc locking.
  Updated test for bug #9566 to correspond this new situation.
mysql-test/r/sp-threads.result:
  Added test for bug #11554 "Server crashes on statement indirectly using
  non-cached function".
mysql-test/t/sp-error.test:
  Added test which checks that now we can read stored routines definitions
  under LOCK TABLES even if we have not locked mysql.proc explicitly. Also
  added check for restrictions which this ability puts on mysql.proc locking.
  Updated test for bug #9566 to correspond this new situation.
mysql-test/t/sp-threads.test:
  Added test for bug #11554 "Server crashes on statement indirectly using
  non-cached function".
sql/lock.cc:
  get_lock_data():
   To be able to open and lock for reading system tables like 'mysql.proc',
   when we already have some tables opened and locked, and avoid deadlocks
   we have to disallow write-locking of these tables with any other tables.
sql/mysql_priv.h:
  open_table() has new parameter which allows to open table even if some-one
  has done a flush or holding namelock on it.
sql/share/errmsg.txt:
  Added error message saying that one cannot write-lock some of system tables
  with any other tables.
sql/sp.cc:
  open_proc_table_for_read()/close_proc_table():
    Added functions to be able open and close mysql.proc table when we already
    have some tables open and locked.
  open_proc_table_for_update():
    Added function to simplify opening of mysql.proc for updates.
  db_find_routine_aux()/db_find_routine()/db_update_routine()/...
    Moved responsibility for opening mysql.proc table from db_find_routine_aux()
    one level up, since this level knows better which type of table access for
    reading of for update it needs.
  sp_function_exists():
    Removed unused function.
sql/sp.h:
  sp_function_exists():
    Removed unused function.
sql/sql_base.cc:
  open_table():
    Added new parameter which allows to open table even if some-one has done a
    flush or holding namelock on it.
  open_unireg_entry():
    Mark 'mysql.proc' as a system table which has special restrictions on its
    locking, but thanks to them can be open and locked even if we already have
    some open and locked.
sql/sql_class.cc:
  Moved THD members holding information about open and locked tables to separate
  Open_tables_state class to be able to save/restore this state easier.
  Added THD::push_open_tables_state()/pop_open_tables_state() methods for
  saving/restoring this state.
sql/sql_class.h:
  Moved THD members holding information about open and locked tables to separate
  Open_tables_state class to be able to save/restore this state easier.
  Added THD::push_open_tables_state()/pop_open_tables_state() methods for
  saving/restoring this state.
sql/sql_lex.cc:
  Removed LEX::proc_table member which was not really used.
sql/sql_lex.h:
  Removed LEX::proc_table member which was not really used.
sql/sql_table.cc:
  open_table() has new parameter which allows to open table even if some-one
  has done a flush or holding namelock on it.
sql/table.h:
  Added TABLE_SHARE::system_table indicating that this table is system table
  like 'mysql.proc' and we want to be able to open and read-lock it even when
  we already have some tables open and locked (and because of this we have
  to put some restrictions on write locking it).

16 files changed, 456 insertions, 199 deletions

diff --git a/mysql-test/r/sp-error.result b/mysql-test/r/sp-error.result
index 0d624b30d9e..742e79df9df 100644
--- a/mysql-test/r/sp-error.result
+++ b/mysql-test/r/sp-error.result
@@ -286,6 +286,19 @@ ERROR 42000: OUT or INOUT argument 2 for routine test.p is not a variable
 call p(42, @tmp_y, 43)|
 ERROR 42000: OUT or INOUT argument 3 for routine test.p is not a variable
 drop procedure p|
+create procedure p() begin end|
+lock table t1 read|
+call p()|
+unlock tables|
+drop procedure p|
+lock tables t1 read, mysql.proc write|
+ERROR HY000: You can't combine write-locking of system 'mysql.proc' table with other tables
+lock tables mysql.proc write, mysql.user write|
+ERROR HY000: You can't combine write-locking of system 'mysql.proc' table with other tables
+lock tables t1 read, mysql.proc read|
+unlock tables|
+lock tables mysql.proc write|
+unlock tables|
 create procedure bug1965()
 begin
 declare c cursor for select val from t1 order by valname;
@@ -477,7 +490,7 @@ begin
 select * from t1;
 end|
 lock table t1 read|
-call bug9566()|
+alter procedure bug9566 comment 'Some comment'|
 ERROR HY000: Table 'proc' was not locked with LOCK TABLES
 unlock tables|
 drop procedure bug9566|
diff --git a/mysql-test/r/sp-threads.result b/mysql-test/r/sp-threads.result
index a9d50e6e697..e6b8128c336 100644
--- a/mysql-test/r/sp-threads.result
+++ b/mysql-test/r/sp-threads.result
@@ -55,3 +55,12 @@ call bug11158();
 unlock tables;
 drop procedure bug11158;
 drop table t1, t2;
+drop function if exists bug11554;
+drop view if exists v1;
+create table t1 (i int);
+create function bug11554 () returns int return 1;
+create view v1 as select bug11554() as f;
+insert into t1 (select f from v1);
+drop function bug11554;
+drop table t1;
+drop view v1;
diff --git a/mysql-test/t/sp-error.test b/mysql-test/t/sp-error.test
index a61c082fb30..62890757045 100644
--- a/mysql-test/t/sp-error.test
+++ b/mysql-test/t/sp-error.test
@@ -387,6 +387,29 @@ drop procedure p|
 
 
 #
+# Let us test that we can access mysql.proc table for routines
+# definitions lookup without locking it explicitly.
+#
+create procedure p() begin end|
+lock table t1 read|
+# This should succeed
+call p()|
+unlock tables|
+drop procedure p|
+# Let us check restrictions which this ability puts on mysql.proc locking.
+--error ER_WRONG_LOCK_OF_SYSTEM_TABLE
+lock tables t1 read, mysql.proc write|
+--error ER_WRONG_LOCK_OF_SYSTEM_TABLE
+lock tables mysql.proc write, mysql.user write|
+# Locking for read should be OK
+lock tables t1 read, mysql.proc read|
+unlock tables|
+# You also should be able lock only mysql.proc for write
+lock tables mysql.proc write|
+unlock tables|
+
+
+#
 # BUG#1965
 #
 create procedure bug1965()
@@ -676,9 +699,7 @@ create procedure bug6600()
 # BUG#9566: explicit LOCK TABLE and store procedures result in illegal state
 #
 # We should not think that mysql.proc table does not exist if we are unable
-# to open it under LOCK TABLE or in prelocked mode. Probably this test
-# should be removed when Monty will allow access to mysql.proc without
-# locking it.
+# to open it under LOCK TABLE or in prelocked mode.
 #
 --disable_warnings
 drop procedure if exists bug9566|
@@ -688,9 +709,11 @@ begin
   select * from t1;
 end|
 lock table t1 read|
-# This should fail because we forgot to lock mysql.proc table explicitly
+# This should fail since we forgot to lock mysql.proc for writing
+# explicitly, and we can't open mysql.proc for _writing_ if there
+# are locked tables.
 --error 1100
-call bug9566()|
+alter procedure bug9566 comment 'Some comment'|
 unlock tables|
 # This should succeed
 drop procedure bug9566|
diff --git a/mysql-test/t/sp-threads.test b/mysql-test/t/sp-threads.test
index 8fec5d14bc1..4c192f3e96f 100644
--- a/mysql-test/t/sp-threads.test
+++ b/mysql-test/t/sp-threads.test
@@ -112,6 +112,25 @@ drop procedure bug11158;
 drop table t1, t2;
 
 #
+# BUG#11554: Server crashes on statement indirectly using non-cached function
+#
+--disable_warnings
+drop function if exists bug11554;
+drop view if exists v1;
+--enable_warnings
+create table t1 (i int);
+create function bug11554 () returns int return 1;
+create view v1 as select bug11554() as f;
+connection con2root;
+# This should not crash server
+insert into t1 (select f from v1);
+# Clean-up
+connection con1root;
+drop function bug11554;
+drop table t1;
+drop view v1;
+
+#
 # BUG#NNNN: New bug synopsis
 #
 #--disable_warnings
diff --git a/sql/lock.cc b/sql/lock.cc
index e0a75a42661..7f3fe5ac5da 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -425,6 +425,19 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
       tables+=table_ptr[i]->file->lock_count();
       lock_count++;
     }
+    /*
+      To be able to open and lock for reading system tables like 'mysql.proc',
+      when we already have some tables opened and locked, and avoid deadlocks
+      we have to disallow write-locking of these tables with any other tables.
+    */
+    if (table_ptr[i]->s->system_table &&
+        table_ptr[i]->reginfo.lock_type >= TL_WRITE_ALLOW_WRITE &&
+        count != 1)
+    {
+      my_error(ER_WRONG_LOCK_OF_SYSTEM_TABLE, MYF(0), table_ptr[i]->s->db,
+               table_ptr[i]->s->table_name);
+      return 0;
+    }
   }
 
   if (!(sql_lock= (MYSQL_LOCK*)
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 9a3684c3865..b1007ea30b2 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -735,7 +735,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok);
 bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create);
 TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type update);
 TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT* mem,
-		  bool *refresh);
+		  bool *refresh, uint flags);
 TABLE *reopen_name_locked_table(THD* thd, TABLE_LIST* table);
 TABLE *find_locked_table(THD *thd, const char *db,const char *table_name);
 bool reopen_table(TABLE *table,bool locked);
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index bdcd88a7205..cb5db72e6e7 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5362,3 +5362,5 @@ ER_NO_DEFAULT_FOR_VIEW_FIELD
         eng "Field of view '%-.64s.%-.64s' underlying table doesn't have a default value"
 ER_SP_NO_RECURSION
         eng "Recursive stored routines are not allowed."
+ER_WRONG_LOCK_OF_SYSTEM_TABLE
+        eng "You can't combine write-locking of system '%-.64s.%-.64s' table with other tables"
diff --git a/sql/sp.cc b/sql/sp.cc
index 456248db66b..0270ec86ca6 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -61,57 +61,152 @@ bool mysql_proc_table_exists= 1;
 /* Tells what SP_DEFAULT_ACCESS should be mapped to */
 #define SP_DEFAULT_ACCESS_MAPPING SP_CONTAINS_SQL
 
-/* *opened=true means we opened ourselves */
-static int
-db_find_routine_aux(THD *thd, int type, sp_name *name,
-		    enum thr_lock_type ltype, TABLE **tablep, bool *opened)
+
+/*
+  Close mysql.proc, opened with open_proc_table_for_read().
+
+  SYNOPSIS
+    close_proc_table()
+      thd  Thread context
+*/
+
+static void close_proc_table(THD *thd)
 {
-  TABLE *table;
-  byte key[NAME_LEN*2+4+1];	// db, name, optional key length type
-  DBUG_ENTER("db_find_routine_aux");
-  DBUG_PRINT("enter", ("type: %d name: %*s",
-		       type, name->m_name.length, name->m_name.str));
+  close_thread_tables(thd);
+  thd->pop_open_tables_state();
+}
 
-  *opened= FALSE;
-  *tablep= 0;
+
+/*
+  Open the mysql.proc table for read.
+
+  SYNOPSIS
+    open_proc_table_for_read()
+      thd  Thread context
+
+  NOTES
+    Thanks to restrictions which we put on opening and locking of
+    this table for writing, we can open and lock it for reading
+    even when we already have some other tables open and locked.
+    One must call close_proc_table() to close table opened with
+    this call.
+
+  RETURN
+    0	Error
+    #	Pointer to TABLE object of mysql.proc
+*/
+
+static TABLE *open_proc_table_for_read(THD *thd)
+{
+  TABLE_LIST tables;
+  TABLE *table;
+  bool old_open_tables= thd->open_tables != 0;
+  bool refresh;
+  DBUG_ENTER("open_proc_table");
 
   /*
     Speed up things if mysql.proc doesn't exists. mysql_proc_table_exists
     is set when we create or read stored procedure or on flush privileges.
   */
-  if (!mysql_proc_table_exists && ltype == TL_READ)
-    DBUG_RETURN(SP_OPEN_TABLE_FAILED);
+  if (!mysql_proc_table_exists)
+    DBUG_RETURN(0);
 
-  if (thd->lex->proc_table)
-    table= thd->lex->proc_table->table;
-  else
+  if (thd->push_open_tables_state())
+    DBUG_RETURN(0);
+
+  bzero((char*) &tables, sizeof(tables));
+  tables.db= (char*) "mysql";
+  tables.table_name= tables.alias= (char*)"proc";
+  if (!(table= open_table(thd, &tables, thd->mem_root, &refresh,
+                          MYSQL_LOCK_IGNORE_FLUSH)))
   {
-    for (table= thd->open_tables ; table ; table= table->next)
-      if (strcmp(table->s->db, "mysql") == 0 &&
-          strcmp(table->s->table_name, "proc") == 0)
-        break;
+    thd->pop_open_tables_state();
+    mysql_proc_table_exists= 0;
+    DBUG_RETURN(0);
   }
-  if (!table)
-  {
-    TABLE_LIST tables;
 
-    memset(&tables, 0, sizeof(tables));
-    tables.db= (char*)"mysql";
-    tables.table_name= tables.alias= (char*)"proc";
-    if (! (table= open_ltable(thd, &tables, ltype)))
-    {
-      /*
-        Under explicit LOCK TABLES or in prelocked mode we should not
-        say that mysql.proc table does not exist if we are unable to
-        open it since this condition may be transient.
-      */
-      if (!(thd->locked_tables || thd->prelocked_mode))
-        mysql_proc_table_exists= 0;
-      DBUG_RETURN(SP_OPEN_TABLE_FAILED);
-    }
-    *opened= TRUE;
+  DBUG_ASSERT(table->s->system_table);
+
+  table->reginfo.lock_type= TL_READ;
+  /*
+    If we have other tables opened, we have to ensure we are not blocked
+    by a flush tables or global read lock, as this could lead to a deadlock
+  */
+  if (!(thd->lock= mysql_lock_tables(thd, &table, 1,
+                                     old_open_tables ?
+                                     (MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK |
+                                      MYSQL_LOCK_IGNORE_FLUSH) : 0)))
+  {
+    close_proc_table(thd);
+    DBUG_RETURN(0);
   }
-  mysql_proc_table_exists= 1;
+  DBUG_RETURN(table);
+}
+
+
+/*
+  Open the mysql.proc table for update.
+
+  SYNOPSIS
+    open_proc_table_for_update()
+      thd  Thread context
+
+  NOTES
+    Table opened with this call should closed using close_thread_tables().
+
+  RETURN
+    0	Error
+    #	Pointer to TABLE object of mysql.proc
+*/
+
+static TABLE *open_proc_table_for_update(THD *thd)
+{
+  TABLE_LIST tables;
+  TABLE *table;
+  DBUG_ENTER("open_proc_table");
+
+  bzero((char*) &tables, sizeof(tables));
+  tables.db= (char*) "mysql";
+  tables.table_name= tables.alias= (char*)"proc";
+  tables.lock_type= TL_WRITE;
+
+  table= open_ltable(thd, &tables, TL_WRITE);
+
+  /*
+    Under explicit LOCK TABLES or in prelocked mode we should not
+    say that mysql.proc table does not exist if we are unable to
+    open and lock it for writing since this condition may be
+    transient.
+  */
+  if (!(thd->locked_tables || thd->prelocked_mode) || table)
+    mysql_proc_table_exists= test(table);
+
+  DBUG_RETURN(table);
+}
+
+
+/*
+  Find row in open mysql.proc table representing stored routine.
+
+  SYNOPSIS
+    db_find_routine_aux()
+      thd    Thread context
+      type   Type of routine to find (function or procedure)
+      name   Name of routine
+      table  TABLE object for open mysql.proc table.
+
+  RETURN VALUE
+    SP_OK           - Routine found
+    SP_KEY_NOT_FOUND- No routine with given name
+*/
+
+static int
+db_find_routine_aux(THD *thd, int type, sp_name *name, TABLE *table)
+{
+  byte key[NAME_LEN*2+4+1];	// db, name, optional key length type
+  DBUG_ENTER("db_find_routine_aux");
+  DBUG_PRINT("enter", ("type: %d name: %*s",
+		       type, name->m_name.length, name->m_name.str));
 
   /*
     Create key to find row. We have to use field->store() to be able to
@@ -121,9 +216,7 @@ db_find_routine_aux(THD *thd, int type, sp_name *name,
     same fields.
   */
   if (name->m_name.length > table->field[1]->field_length)
-  {
     DBUG_RETURN(SP_KEY_NOT_FOUND);
-  }
   table->field[0]->store(name->m_db.str, name->m_db.length, &my_charset_bin);
   table->field[1]->store(name->m_name.str, name->m_name.length,
                          &my_charset_bin);
@@ -134,15 +227,33 @@ db_find_routine_aux(THD *thd, int type, sp_name *name,
   if (table->file->index_read_idx(table->record[0], 0,
 				  key, table->key_info->key_length,
 				  HA_READ_KEY_EXACT))
-  {
     DBUG_RETURN(SP_KEY_NOT_FOUND);
-  }
-  *tablep= table;
 
   DBUG_RETURN(SP_OK);
 }
 
 
+/*
+  Find routine definition in mysql.proc table and create corresponding
+  sp_head object for it.
+
+  SYNOPSIS
+    db_find_routine()
+      thd   Thread context
+      type  Type of routine (TYPE_ENUM_PROCEDURE/...)
+      name  Name of routine
+      sphp  Out parameter in which pointer to created sp_head
+            object is returned (0 in case of error).
+
+  NOTE
+    This function may damage current LEX during execution, so it is good
+    idea to create temporary LEX and make it active before calling it.
+
+  RETURN VALUE
+    0     - Success
+    non-0 - Error (may be one of special codes like SP_KEY_NOT_FOUND)
+*/
+
 static int
 db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
 {
@@ -150,7 +261,6 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
   TABLE *table;
   const char *params, *returns, *body;
   int ret;
-  bool opened;
   const char *definer;
   longlong created;
   longlong modified;
@@ -164,8 +274,11 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
   DBUG_PRINT("enter", ("type: %d name: %*s",
 		       type, name->m_name.length, name->m_name.str));
 
-  ret= db_find_routine_aux(thd, type, name, TL_READ, &table, &opened);
-  if (ret != SP_OK)
+  *sphp= 0;                                     // In case of errors
+  if (!(table= open_proc_table_for_read(thd)))
+    DBUG_RETURN(SP_OPEN_TABLE_FAILED);
+
+  if ((ret= db_find_routine_aux(thd, type, name, table)) != SP_OK)
     goto done;
 
   if (table->s->fields != MYSQL_PROC_FIELD_COUNT)
@@ -257,11 +370,8 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
   chistics.comment.str= ptr;
   chistics.comment.length= length;
 
-  if (opened)
-  {
-    opened= FALSE;
-    close_thread_tables(thd, 0, 1);
-  }
+  close_proc_table(thd);
+  table= 0;
 
   {
     String defstr;
@@ -337,9 +447,8 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
   }
 
  done:
-
-  if (opened)
-    close_thread_tables(thd);
+  if (table)
+    close_proc_table(thd);
   DBUG_RETURN(ret);
 }
 
@@ -362,7 +471,6 @@ db_create_routine(THD *thd, int type, sp_head *sp)
 {
   int ret;
   TABLE *table;
-  TABLE_LIST tables;
   char definer[HOSTNAME_LENGTH+USERNAME_LENGTH+2];
   char olddb[128];
   bool dbchanged;
@@ -377,11 +485,7 @@ db_create_routine(THD *thd, int type, sp_head *sp)
     goto done;
   }
 
-  memset(&tables, 0, sizeof(tables));
-  tables.db= (char*)"mysql";
-  tables.table_name= tables.alias= (char*)"proc";
-
-  if (! (table= open_ltable(thd, &tables, TL_WRITE)))
+  if (!(table= open_proc_table_for_update(thd)))
     ret= SP_OPEN_TABLE_FAILED;
   else
   {
@@ -491,20 +595,18 @@ db_drop_routine(THD *thd, int type, sp_name *name)
 {
   TABLE *table;
   int ret;
-  bool opened;
   DBUG_ENTER("db_drop_routine");
   DBUG_PRINT("enter", ("type: %d name: %*s",
 		       type, name->m_name.length, name->m_name.str));
 
-  ret= db_find_routine_aux(thd, type, name, TL_WRITE, &table, &opened);
-  if (ret == SP_OK)
+  if (!(table= open_proc_table_for_update(thd)))
+    DBUG_RETURN(SP_OPEN_TABLE_FAILED);
+  if ((ret= db_find_routine_aux(thd, type, name, table)) == SP_OK)
   {
     if (table->file->delete_row(table->record[0]))
       ret= SP_DELETE_ROW_FAILED;
   }
-
-  if (opened)
-    close_thread_tables(thd);
+  close_thread_tables(thd);
   DBUG_RETURN(ret);
 }
 
@@ -519,8 +621,9 @@ db_update_routine(THD *thd, int type, sp_name *name, st_sp_chistics *chistics)
   DBUG_PRINT("enter", ("type: %d name: %*s",
 		       type, name->m_name.length, name->m_name.str));
 
-  ret= db_find_routine_aux(thd, type, name, TL_WRITE, &table, &opened);
-  if (ret == SP_OK)
+  if (!(table= open_proc_table_for_update(thd)))
+    DBUG_RETURN(SP_OPEN_TABLE_FAILED);
+  if ((ret= db_find_routine_aux(thd, type, name, table)) == SP_OK)
   {
     store_record(table,record[1]);
     table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
@@ -538,8 +641,7 @@ db_update_routine(THD *thd, int type, sp_name *name, st_sp_chistics *chistics)
     if ((table->file->update_row(table->record[1],table->record[0])))
       ret= SP_WRITE_ROW_FAILED;
   }
-  if (opened)
-    close_thread_tables(thd);
+  close_thread_tables(thd);
   DBUG_RETURN(ret);
 }
 
@@ -741,20 +843,9 @@ sp_drop_db_routines(THD *thd, char *db)
   memset(key+keylen, (int)' ', 64-keylen); // Pad with space
   keylen= sizeof(key);
 
-  for (table= thd->open_tables ; table ; table= table->next)
-    if (strcmp(table->s->db, "mysql") == 0 &&
-	strcmp(table->s->table_name, "proc") == 0)
-      break;
-  if (! table)
-  {
-    TABLE_LIST tables;
-
-    memset(&tables, 0, sizeof(tables));
-    tables.db= (char*)"mysql";
-    tables.table_name= tables.alias= (char*)"proc";
-    if (! (table= open_ltable(thd, &tables, TL_WRITE)))
-      DBUG_RETURN(SP_OPEN_TABLE_FAILED);
-  }
+  ret= SP_OPEN_TABLE_FAILED;
+  if (!(table= open_proc_table_for_update(thd)))
+    goto err;
 
   ret= SP_OK;
   table->file->ha_index_init(0);
@@ -764,7 +855,8 @@ sp_drop_db_routines(THD *thd, char *db)
     int nxtres;
     bool deleted= FALSE;
 
-    do {
+    do
+    {
       if (! table->file->delete_row(table->record[0]))
 	deleted= TRUE;		/* We deleted something */
       else
@@ -784,6 +876,7 @@ sp_drop_db_routines(THD *thd, char *db)
 
   close_thread_tables(thd);
 
+err:
   DBUG_RETURN(ret);
 }
 
@@ -977,9 +1070,7 @@ sp_find_function(THD *thd, sp_name *name, bool cache_only)
   if (!(sp= sp_cache_lookup(&thd->sp_func_cache, name)) &&
       !cache_only)
   {
-    if (db_find_routine(thd, TYPE_ENUM_FUNCTION, name, &sp) != SP_OK)
-      sp= NULL;
-    else
+    if (db_find_routine(thd, TYPE_ENUM_FUNCTION, name, &sp) == SP_OK)
       sp_cache_insert(&thd->sp_func_cache, sp);
   }
   DBUG_RETURN(sp);
@@ -1057,26 +1148,6 @@ sp_show_status_function(THD *thd, const char *wild)
 }
 
 
-bool
-sp_function_exists(THD *thd, sp_name *name)
-{
-  TABLE *table;
-  bool ret= FALSE;
-  bool opened= FALSE;
-  DBUG_ENTER("sp_function_exists");
-
-  if (sp_cache_lookup(&thd->sp_func_cache, name) ||
-      db_find_routine_aux(thd, TYPE_ENUM_FUNCTION,
-			  name, TL_READ,
-			  &table, &opened) == SP_OK)
-    ret= TRUE;
-  if (opened)
-    close_thread_tables(thd, 0, 1);
-  thd->clear_error();
-  DBUG_RETURN(ret);
-}
-
-
 byte *
 sp_lex_sp_key(const byte *ptr, uint *plen, my_bool first)
 {
@@ -1185,7 +1256,6 @@ sp_cache_routines(THD *thd, LEX *lex)
 
           thd->lex= newlex;
           /* Pass hint pointer to mysql.proc table */
-          newlex->proc_table= oldlex->proc_table;
           newlex->current_select= NULL;
           name.m_name.str= strchr(name.m_qname.str, '.');
           name.m_db.length= name.m_name.str - name.m_qname.str;
diff --git a/sql/sp.h b/sql/sp.h
index 16d79026132..499d2f3d38f 100644
--- a/sql/sp.h
+++ b/sql/sp.h
@@ -74,9 +74,6 @@ sp_show_create_function(THD *thd, sp_name *name);
 int
 sp_show_status_function(THD *thd, const char *wild);
 
-bool
-sp_function_exists(THD *thd, sp_name *name);
-
 
 /*
  *  For precaching of functions and procedures
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 383adcadc6a..f19252fbbdc 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -542,10 +542,9 @@ void close_thread_tables(THD *thd, bool lock_in_use, bool skip_derived,
 
 bool close_thread_table(THD *thd, TABLE **table_ptr)
 {
-  DBUG_ENTER("close_thread_table");
-
   bool found_old_table= 0;
   TABLE *table= *table_ptr;
+  DBUG_ENTER("close_thread_table");
   DBUG_ASSERT(table->key_read == 0);
   DBUG_ASSERT(table->file->inited == handler::NONE);
 
@@ -972,18 +971,34 @@ TABLE *reopen_name_locked_table(THD* thd, TABLE_LIST* table_list)
 }
 
 
-/******************************************************************************
-** open a table
-** Uses a cache of open tables to find a table not in use.
-** If refresh is a NULL pointer, then the is no version number checking and
-** the table is not put in the thread-open-list
-** If the return value is NULL and refresh is set then one must close
-** all tables and retry the open
-******************************************************************************/
+/*
+  Open a table.
+
+  SYNOPSIS
+    open_table()
+      thd         Thread context
+      table_list  Open first table in list
+      refresh     Pointer to memory that will be set to 1 if
+                  we need to close all tables and reopen them
+                  If this is a NULL pointer, then the is no version
+                  number checking and the table is not put in the
+                  thread-open-list
+      flags       Bitmap of flags to modify how open works:
+                    MYSQL_LOCK_IGNORE_FLUSH - Open table even if someone
+                    has done a flush or namelock on it.
+
+  IMPLEMENTATION
+    Uses a cache of open tables to find a table not in use.
+
+  RETURN
+    NULL  Open failed.  If refresh is set then one should close
+          all other tables and retry the open
+    #     Success. Pointer to TABLE object for open table.
+*/
 
 
 TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
-		  bool *refresh)
+		  bool *refresh, uint flags)
 {
   reg1	TABLE *table;
   char	key[MAX_DBKEY_LENGTH];
@@ -1096,9 +1111,16 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   {
     if (table->s->version != refresh_version)
     {
+      if (flags & MYSQL_LOCK_IGNORE_FLUSH)
+      {
+        /* Force close at once after usage */
+        thd->version= table->s->version;
+        continue;
+      }
+
       /*
-      ** There is a refresh in progress for this table
-      ** Wait until the table is freed or the thread is killed.
+        There is a refresh in progress for this table
+        Wait until the table is freed or the thread is killed.
       */
       close_old_data_files(thd,thd->open_tables,0,0);
       if (table->in_use != thd)
@@ -1681,6 +1703,15 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
   if (error == 5)
     DBUG_RETURN(0);	// we have just opened VIEW
 
+  /*
+    We can't mark all tables in 'mysql' database as system since we don't
+    allow to lock such tables for writing with any other tables (even with
+    other system tables) and some privilege tables need this.
+  */
+  if (!my_strcasecmp(system_charset_info, db, "mysql") &&
+      !my_strcasecmp(system_charset_info, name, "proc"))
+    entry->s->system_table= 1;
+
   if (Table_triggers_list::check_n_load(thd, db, name, entry))
     goto err;
 
@@ -1835,7 +1866,7 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter)
     }
     (*counter)++;
     if (!tables->table &&
-	!(tables->table= open_table(thd, tables, &new_frm_mem, &refresh)))
+	!(tables->table= open_table(thd, tables, &new_frm_mem, &refresh, 0)))
     {
       free_root(&new_frm_mem, MYF(MY_KEEP_PREALLOC));
       if (tables->view)
@@ -2010,7 +2041,7 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type)
   thd->current_tablenr= 0;
   /* open_ltable can be used only for BASIC TABLEs */
   table_list->required_type= FRMTYPE_TABLE;
-  while (!(table= open_table(thd, table_list, thd->mem_root, &refresh)) &&
+  while (!(table= open_table(thd, table_list, thd->mem_root, &refresh, 0)) &&
          refresh)
     ;
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index bf6f41d00ed..d0ac1a16f6b 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -156,6 +156,14 @@ bool foreign_key_prefix(Key *a, Key *b)
 /****************************************************************************
 ** Thread specific functions
 ****************************************************************************/
+
+Open_tables_state::Open_tables_state()
+  :version(refresh_version)
+{
+  reset_open_tables_state();
+}
+
+
 /*
   Pass nominal parameters to Statement constructor only to ensure that
   the destructor works OK in case of error. The main_mem_root will be
@@ -164,6 +172,7 @@ bool foreign_key_prefix(Key *a, Key *b)
 
 THD::THD()
   :Statement(CONVENTIONAL_EXECUTION, 0, ALLOC_ROOT_MIN_BLOCK_SIZE, 0),
+   Open_tables_state(),
    user_time(0), global_read_lock(0), is_fatal_error(0),
    rand_used(0), time_zone_used(0),
    last_insert_id_used(0), insert_id_used(0), clear_next_insert_id(0),
@@ -181,10 +190,8 @@ THD::THD()
   db_length= col_access=0;
   query_error= tmp_table_used= 0;
   next_insert_id=last_insert_id=0;
-  open_tables= temporary_tables= handler_tables= derived_tables= 0;
   hash_clear(&handler_tables_hash);
   tmp_table=0;
-  lock=locked_tables=0;
   used_tables=0;
   cuted_fields= sent_row_count= 0L;
   limit_found_rows= 0;
@@ -230,7 +237,6 @@ THD::THD()
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   db_access=NO_ACCESS;
 #endif
-  version=refresh_version;			// For boot
   *scramble= '\0';
 
   init();
@@ -259,7 +265,6 @@ THD::THD()
   tablespace_op=FALSE;
   ulong tmp=sql_rnd_with_mutex();
   randominit(&rand, tmp + (ulong) &rand, tmp + (ulong) ::query_id);
-  prelocked_mode= NON_PRELOCKED;
 }
 
 
@@ -1775,3 +1780,40 @@ void THD::set_status_var_init()
 {
   bzero((char*) &status_var, sizeof(status_var));
 }
+
+/****************************************************************************
+  Handling of open and locked tables states.
+
+  This is used when we want to open/lock (and then close) some tables when
+  we already have a set of tables open and locked. We use these methods for
+  access to mysql.proc table to find definitions of stored routines.
+****************************************************************************/
+
+bool THD::push_open_tables_state()
+{
+  Open_tables_state *state;
+  DBUG_ENTER("push_open_table_state");
+  /* Currently we only push things one level */
+  DBUG_ASSERT(open_state_list.elements == 0);
+
+  if (!(state= (Open_tables_state*) alloc(sizeof(*state))))
+    DBUG_RETURN(1);                             // Fatal error is set
+  /* Store state for currently open tables */
+  state->set_open_tables_state(this);
+  if (open_state_list.push_back(state, mem_root))
+    DBUG_RETURN(1);                             // Fatal error is set
+  reset_open_tables_state();
+  DBUG_RETURN(0);
+}
+
+void THD::pop_open_tables_state()
+{
+  Open_tables_state *state;
+  DBUG_ENTER("pop_open_table_state");
+  /* Currently we only push things one level */
+  DBUG_ASSERT(open_state_list.elements == 1);
+
+  state= open_state_list.pop();
+  set_open_tables_state(state);
+  DBUG_VOID_RETURN;
+}
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 5642fa0d6af..625b9c27b44 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -933,12 +933,93 @@ enum prelocked_mode_type {NON_PRELOCKED= 0, PRELOCKED= 1,
 
 
 /*
+  Class that holds information about tables which were open and locked
+  by the thread. It is also used to save/restore this information in
+  push_open_tables_state()/pop_open_tables_state().
+*/
+
+class Open_tables_state
+{
+public:
+  /*
+    open_tables - list of regular tables in use by this thread
+    temporary_tables - list of temp tables in use by this thread
+    handler_tables - list of tables that were opened with HANDLER OPEN
+     and are still in use by this thread
+  */
+  TABLE *open_tables, *temporary_tables, *handler_tables, *derived_tables;
+  /*
+    During a MySQL session, one can lock tables in two modes: automatic
+    or manual. In automatic mode all necessary tables are locked just before
+    statement execution, and all acquired locks are stored in 'lock'
+    member. Unlocking takes place automatically as well, when the
+    statement ends.
+    Manual mode comes into play when a user issues a 'LOCK TABLES'
+    statement. In this mode the user can only use the locked tables.
+    Trying to use any other tables will give an error. The locked tables are
+    stored in 'locked_tables' member.  Manual locking is described in
+    the 'LOCK_TABLES' chapter of the MySQL manual.
+    See also lock_tables() for details.
+  */
+  MYSQL_LOCK *lock;
+  /*
+    Tables that were locked with explicit or implicit LOCK TABLES.
+    (Implicit LOCK TABLES happens when we are prelocking tables for
+     execution of statement which uses stored routines. See description
+     THD::prelocked_mode for more info.)
+  */
+  MYSQL_LOCK *locked_tables;
+  /*
+    prelocked_mode_type enum and prelocked_mode member are used for
+    indicating whenever "prelocked mode" is on, and what type of
+    "prelocked mode" is it.
+
+    Prelocked mode is used for execution of queries which explicitly
+    or implicitly (via views or triggers) use functions, thus may need
+    some additional tables (mentioned in query table list) for their
+    execution.
+
+    First open_tables() call for such query will analyse all functions
+    used by it and add all additional tables to table its list. It will
+    also mark this query as requiring prelocking. After that lock_tables()
+    will issue implicit LOCK TABLES for the whole table list and change
+    thd::prelocked_mode to non-0. All queries called in functions invoked
+    by the main query will use prelocked tables. Non-0 prelocked_mode
+    will also surpress mentioned analysys in those queries thus saving
+    cycles. Prelocked mode will be turned off once close_thread_tables()
+    for the main query will be called.
+
+    Note: Since not all "tables" present in table list are really locked
+    thd::prelocked_mode does not imply thd::locked_tables.
+  */
+  prelocked_mode_type prelocked_mode;
+  ulong	version;
+  uint current_tablenr;
+
+  Open_tables_state();
+
+  void set_open_tables_state(Open_tables_state *state)
+  {
+    *this= *state;
+  }
+
+  void reset_open_tables_state()
+  {
+    open_tables= temporary_tables= handler_tables= derived_tables= 0;
+    lock= locked_tables= 0;
+    prelocked_mode= NON_PRELOCKED;
+  }
+};
+
+
+/*
   For each client connection we create a separate thread with THD serving as
   a thread/connection descriptor
 */
 
 class THD :public ilink,
-           public Statement
+           public Statement,
+           public Open_tables_state
 {
 public:
 #ifdef EMBEDDED_LIBRARY
@@ -1006,34 +1087,6 @@ public:
   ulong master_access;			/* Global privileges from mysql.user */
   ulong db_access;			/* Privileges for current db */
 
-  /*
-    open_tables - list of regular tables in use by this thread
-    temporary_tables - list of temp tables in use by this thread
-    handler_tables - list of tables that were opened with HANDLER OPEN
-     and are still in use by this thread
-  */
-  TABLE   *open_tables,*temporary_tables, *handler_tables, *derived_tables;
-  /*
-    During a MySQL session, one can lock tables in two modes: automatic
-    or manual. In automatic mode all necessary tables are locked just before
-    statement execution, and all acquired locks are stored in 'lock'
-    member. Unlocking takes place automatically as well, when the
-    statement ends.
-    Manual mode comes into play when a user issues a 'LOCK TABLES'
-    statement. In this mode the user can only use the locked tables.
-    Trying to use any other tables will give an error. The locked tables are
-    stored in 'locked_tables' member.  Manual locking is described in
-    the 'LOCK_TABLES' chapter of the MySQL manual.
-    See also lock_tables() for details.
-  */
-  MYSQL_LOCK	*lock;				/* Current locks */
-  /*
-    Tables that were locked with explicit or implicit LOCK TABLES.
-    (Implicit LOCK TABLES happens when we are prelocking tables for
-     execution of statement which uses stored routines. See description
-     THD::prelocked_mode for more info.)
-  */
-  MYSQL_LOCK	*locked_tables;
   HASH		handler_tables_hash;
   /*
     One thread can hold up to one named user-level lock. This variable
@@ -1150,6 +1203,7 @@ public:
   List	     <MYSQL_ERROR> warn_list;
   uint	     warn_count[(uint) MYSQL_ERROR::WARN_LEVEL_END];
   uint	     total_warn_count;
+  List	     <Open_tables_state> open_state_list;
   /*
     Id of current query. Statement can be reused to execute several queries
     query_id is global in context of the whole MySQL server.
@@ -1159,7 +1213,7 @@ public:
     update auto-updatable fields (like auto_increment and timestamp).
   */
   query_id_t query_id, warn_id;
-  ulong	     version, options, thread_id, col_access;
+  ulong	     options, thread_id, col_access;
 
   /* Statement id is thread-wide. This counter is used to generate ids */
   ulong      statement_id_counter;
@@ -1167,7 +1221,7 @@ public:
   ulong      row_count;  // Row counter, mainly for errors and warnings
   long	     dbug_thread_id;
   pthread_t  real_id;
-  uint	     current_tablenr,tmp_table,global_read_lock;
+  uint	     tmp_table, global_read_lock;
   uint	     server_status,open_options,system_thread;
   uint32     db_length;
   uint       select_number;             //number of select (used for EXPLAIN)
@@ -1218,31 +1272,6 @@ public:
     long    long_value;
   } sys_var_tmp;
 
-  /*
-    prelocked_mode_type enum and prelocked_mode member are used for
-    indicating whenever "prelocked mode" is on, and what type of
-    "prelocked mode" is it.
-
-    Prelocked mode is used for execution of queries which explicitly
-    or implicitly (via views or triggers) use functions, thus may need
-    some additional tables (mentioned in query table list) for their
-    execution.
-
-    First open_tables() call for such query will analyse all functions
-    used by it and add all additional tables to table its list. It will
-    also mark this query as requiring prelocking. After that lock_tables()
-    will issue implicit LOCK TABLES for the whole table list and change
-    thd::prelocked_mode to non-0. All queries called in functions invoked
-    by the main query will use prelocked tables. Non-0 prelocked_mode
-    will also surpress mentioned analysys in those queries thus saving
-    cycles. Prelocked mode will be turned off once close_thread_tables()
-    for the main query will be called.
-
-    Note: Since not all "tables" present in table list are really locked
-    thd::relocked_mode does not imply thd::locked_tables.
-  */
-  prelocked_mode_type prelocked_mode;
-
   THD();
   ~THD();
 
@@ -1428,8 +1457,11 @@ public:
              (variables.sql_mode & MODE_STRICT_ALL_TABLES)));
   }
   void set_status_var_init();
+  bool push_open_tables_state();
+  void pop_open_tables_state();
 };
 
+
 #define tmp_disable_binlog(A)       \
   {ulong tmp_disable_binlog__save_options= (A)->options; \
   (A)->options&= ~OPTION_BIN_LOG
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 20f5092c5ce..785da5e6e64 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -145,7 +145,7 @@ void lex_start(THD *thd, uchar *buf,uint length)
   lex->found_semicolon= 0;
   lex->safe_to_cache_query= 1;
   lex->time_zone_tables_used= 0;
-  lex->leaf_tables_insert= lex->proc_table= lex->query_tables= 0;
+  lex->leaf_tables_insert= lex->query_tables= 0;
   lex->query_tables_last= &lex->query_tables;
   lex->variables_used= 0;
   lex->select_lex.parent_lex= lex;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 15306ab5b34..f7e9c29828b 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -719,7 +719,6 @@ typedef struct st_lex
     function)
   */
   TABLE_LIST **query_tables_last;
-  TABLE_LIST *proc_table; /* refer to mysql.proc if it was opened by VIEW */
   /* store original leaf_tables for INSERT SELECT and PS/SP */
   TABLE_LIST *leaf_tables_insert;
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 5a93176b922..9e6d12eba13 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1760,7 +1760,7 @@ TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
                             create_info, *extra_fields, *keys, 0,
                             select_field_count))
     {
-      if (!(table= open_table(thd, create_table, thd->mem_root, (bool*) 0)))
+      if (!(table= open_table(thd, create_table, thd->mem_root, (bool*)0, 0)))
         quick_rm_table(create_info->db_type, create_table->db,
                        table_case_name(create_info, create_table->table_name));
     }
@@ -3571,7 +3571,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       bzero((void*) &tbl, sizeof(tbl));
       tbl.db= new_db;
       tbl.table_name= tbl.alias= tmp_name;
-      new_table= open_table(thd, &tbl, thd->mem_root, 0);
+      new_table= open_table(thd, &tbl, thd->mem_root, 0, 0);
     }
     else
     {
diff --git a/sql/table.h b/sql/table.h
index d0c998c4c10..e5653a1f213 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -163,6 +163,13 @@ typedef struct st_table_share
   my_bool crashed;
   my_bool is_view;
   my_bool name_lock, replace_with_name_lock;
+  /*
+    TRUE if this is a system table like 'mysql.proc', which we want to be
+    able to open and lock even when we already have some tables open and
+    locked. To avoid deadlocks we have to put certain restrictions on
+    locking of this table for writing. FALSE - otherwise.
+  */
+  my_bool system_table;
 } TABLE_SHARE;