diff --git a/mysql-test/r/mdl_sync.result b/mysql-test/r/mdl_sync.result
index e5447c32b7d..ec02f29b008 100644
--- a/mysql-test/r/mdl_sync.result
+++ b/mysql-test/r/mdl_sync.result
@@ -254,3 +254,38 @@ commit;
 # Switching to connection 'default'.
 # Clean-up
 drop table t1;
+#
+# Bug#48210 FLUSH TABLES WITH READ LOCK deadlocks 
+#           against concurrent CREATE PROCEDURE
+#
+# Test 1: CREATE PROCEDURE
+# Connection 1
+# Start CREATE PROCEDURE and open mysql.proc
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL table_opened WAIT_FOR grlwait';
+CREATE PROCEDURE p1() SELECT 1;
+# Connection 2
+SET DEBUG_SYNC= 'now WAIT_FOR table_opened';
+# Check that FLUSH must wait to get the GRL
+# and let CREATE PROCEDURE continue
+SET DEBUG_SYNC= 'wait_lock_global_read_lock SIGNAL grlwait';
+FLUSH TABLES WITH READ LOCK;
+# Connection 1
+# Connection 2
+UNLOCK TABLES;
+# Connection 1
+SET DEBUG_SYNC= 'RESET';
+# Test 2: DROP PROCEDURE
+# Start DROP PROCEDURE and open tables
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL table_opened WAIT_FOR grlwait';
+DROP PROCEDURE p1;
+# Connection 2
+SET DEBUG_SYNC= 'now WAIT_FOR table_opened';
+# Check that FLUSH must wait to get the GRL
+# and let DROP PROCEDURE continue
+SET DEBUG_SYNC= 'wait_lock_global_read_lock SIGNAL grlwait';
+FLUSH TABLES WITH READ LOCK;
+# Connection 1
+# Connection 2
+UNLOCK TABLES;
+# Connection 1
+SET DEBUG_SYNC= 'RESET';
diff --git a/mysql-test/r/partition.result b/mysql-test/r/partition.result
index 543a70f9a2a..e9bbc011f7b 100644
--- a/mysql-test/r/partition.result
+++ b/mysql-test/r/partition.result
@@ -65,6 +65,15 @@ show indexes from t1;
 Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment
 t1	1	a	1	a	A	1	NULL	NULL	YES	BTREE	
 drop table t1;
+create table t1 (a int)
+partition by hash (a);
+create index i on t1 (a);
+insert into t1 values (1);
+insert into t1 select * from t1;
+create index i on t1 (a);
+ERROR 42000: Duplicate key name 'i'
+create index i2 on t1 (a);
+drop table t1;
 CREATE TABLE t1 (a INT, FOREIGN KEY (a) REFERENCES t0 (a))
 ENGINE=MyISAM
 PARTITION BY HASH (a);
diff --git a/mysql-test/r/partition_sync.result b/mysql-test/r/partition_sync.result
index 41ca19426fe..d48362cb57c 100644
--- a/mysql-test/r/partition_sync.result
+++ b/mysql-test/r/partition_sync.result
@@ -1,2 +1,57 @@
-# Disabled until Bug#46654 False deadlock on concurrent DML/DDL 
-# with partitions, inconsistent behavior is backported
+#
+# Bug #43867 ALTER TABLE on a partitioned table 
+#            causes unnecessary deadlocks
+#
+CREATE TABLE t1 (a int) PARTITION BY RANGE (a)
+(PARTITION p0 VALUES LESS THAN (1),
+PARTITION p1 VALUES LESS THAN (2));
+INSERT INTO t1 VALUES (0),(1);
+# Connection 2
+BEGIN;
+SELECT * FROM t1;
+a
+0
+1
+# Connection 1
+ALTER TABLE t1 DROP PARTITION p3;
+ERROR HY000: Error in list of partitions to DROP
+# Connection 2
+# This failed with deadlock and should not do so.
+SELECT * FROM t1;
+a
+0
+1
+# Connection 1
+DROP TABLE t1;
+#
+# Bug #46654 False deadlock on concurrent DML/DDL 
+#            with partitions, inconsistent behavior
+#
+DROP TABLE IF EXISTS tbl_with_partitions;
+CREATE TABLE tbl_with_partitions ( i INT ) 
+PARTITION BY HASH(i);
+INSERT INTO tbl_with_partitions VALUES (1);
+# Connection 3
+LOCK TABLE tbl_with_partitions READ;
+# Connection 1
+# Access table with disabled autocommit
+SET AUTOCOMMIT = 0;
+SELECT * FROM tbl_with_partitions;
+i
+1
+# Connection 2
+# Alter table, abort after prepare
+set session debug="+d,abort_copy_table";
+ALTER TABLE tbl_with_partitions ADD COLUMN f INT;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+# Connection 1
+# Try accessing the table after Alter aborted.
+# This used to give ER_LOCK_DEADLOCK.
+SELECT * FROM tbl_with_partitions;
+i
+1
+# Connection 3
+UNLOCK TABLES;
+# Connection 1
+# Cleanup
+DROP TABLE tbl_with_partitions;
diff --git a/mysql-test/r/sp-error.result b/mysql-test/r/sp-error.result
index 00ae7dd4eca..034923bbd4f 100644
--- a/mysql-test/r/sp-error.result
+++ b/mysql-test/r/sp-error.result
@@ -1687,6 +1687,17 @@ NULL
 SELECT non_existent (a) FROM t1 WHERE b = 999999;
 ERROR 42000: FUNCTION test.non_existent does not exist
 DROP TABLE t1;
+CREATE TABLE t1 ( f2 INTEGER, f3 INTEGER );
+INSERT INTO t1 VALUES  ( 1, 1 );
+CREATE FUNCTION func_1 () RETURNS INTEGER
+BEGIN
+INSERT INTO t1 SELECT * FROM t1 ;
+RETURN 1 ;
+END|
+INSERT INTO t1 SELECT * FROM (SELECT 2 AS f1, 2 AS f2) AS A WHERE func_1() = 5;
+ERROR HY000: Can't update table 't1' in stored function/trigger because it is already used by statement which invoked this stored function/trigger.
+DROP FUNCTION func_1;
+DROP TABLE t1;
 #
 # Bug #47788: Crash in TABLE_LIST::hide_view_error on UPDATE + VIEW + 
 #   SP + MERGE + ALTER
diff --git a/mysql-test/t/mdl_sync.test b/mysql-test/t/mdl_sync.test
index fd66f6d539d..e3aceaa05fa 100644
--- a/mysql-test/t/mdl_sync.test
+++ b/mysql-test/t/mdl_sync.test
@@ -475,6 +475,73 @@ disconnect con46673;
 drop table t1;
 
 
+--echo #
+--echo # Bug#48210 FLUSH TABLES WITH READ LOCK deadlocks 
+--echo #           against concurrent CREATE PROCEDURE
+--echo #
+
+connect (con2, localhost, root);
+
+--echo # Test 1: CREATE PROCEDURE
+
+--echo # Connection 1
+connection default;
+--echo # Start CREATE PROCEDURE and open mysql.proc
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL table_opened WAIT_FOR grlwait';
+--send CREATE PROCEDURE p1() SELECT 1
+
+--echo # Connection 2
+connection con2;
+SET DEBUG_SYNC= 'now WAIT_FOR table_opened';
+--echo # Check that FLUSH must wait to get the GRL
+--echo # and let CREATE PROCEDURE continue
+SET DEBUG_SYNC= 'wait_lock_global_read_lock SIGNAL grlwait';
+--send FLUSH TABLES WITH READ LOCK
+
+--echo # Connection 1
+connection default;
+--reap
+
+--echo # Connection 2
+connection con2;
+--reap
+UNLOCK TABLES;
+
+--echo # Connection 1
+connection default;
+SET DEBUG_SYNC= 'RESET';
+
+--echo # Test 2: DROP PROCEDURE
+
+connection default;
+--echo # Start DROP PROCEDURE and open tables
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL table_opened WAIT_FOR grlwait';
+--send DROP PROCEDURE p1
+
+--echo # Connection 2
+connection con2;
+SET DEBUG_SYNC= 'now WAIT_FOR table_opened';
+--echo # Check that FLUSH must wait to get the GRL
+--echo # and let DROP PROCEDURE continue
+SET DEBUG_SYNC= 'wait_lock_global_read_lock SIGNAL grlwait';
+--send FLUSH TABLES WITH READ LOCK
+
+--echo # Connection 1
+connection default;
+--reap
+
+--echo # Connection 2
+connection con2;
+--reap
+UNLOCK TABLES;
+
+--echo # Connection 1
+connection default;
+SET DEBUG_SYNC= 'RESET';
+
+disconnect con2;
+
+
 # Check that all connections opened by test cases in this file are really
 # gone so execution of other tests won't be affected by their presence.
 --source include/wait_until_count_sessions.inc
diff --git a/mysql-test/t/partition.test b/mysql-test/t/partition.test
index 8ab91f23522..01e885a527c 100644
--- a/mysql-test/t/partition.test
+++ b/mysql-test/t/partition.test
@@ -74,6 +74,19 @@ analyze table t1;
 show indexes from t1;
 drop table t1;
 
+#
+# Bug#40181: hang if create index
+#
+create table t1 (a int)
+partition by hash (a);
+create index i on t1 (a);
+insert into t1 values (1);
+insert into t1 select * from t1;
+--error ER_DUP_KEYNAME
+create index i on t1 (a);
+create index i2 on t1 (a);
+drop table t1;
+
 #
 # Bug#36001: Partitions: spelling and using some error messages
 #
diff --git a/mysql-test/t/partition_sync.test b/mysql-test/t/partition_sync.test
index 5d2b25e87f3..85eb33ebb6b 100644
--- a/mysql-test/t/partition_sync.test
+++ b/mysql-test/t/partition_sync.test
@@ -1,42 +1,91 @@
 --source include/have_partition.inc
+--source include/have_debug.inc
 # Save the initial number of concurrent sessions.
 --source include/count_sessions.inc
 
---echo # Disabled until Bug#46654 False deadlock on concurrent DML/DDL 
---echo # with partitions, inconsistent behavior is backported
+--echo #
+--echo # Bug #43867 ALTER TABLE on a partitioned table 
+--echo #            causes unnecessary deadlocks
+--echo #
 
-#--echo #
-#--echo # Bug #43867 ALTER TABLE on a partitioned table 
-#--echo #            causes unnecessary deadlocks
-#--echo #
-#
-#CREATE TABLE t1 (a int) PARTITION BY RANGE (a)
-#(PARTITION p0 VALUES LESS THAN (1),
-# PARTITION p1 VALUES LESS THAN (2));
-#
-#INSERT INTO t1 VALUES (0),(1);
-#
-#connect(con1,localhost,root);
-#
-#--echo # Connection 2
-#connection con1;
-#BEGIN;
-#SELECT * FROM t1;
-#
-#--echo # Connection 1
-#connection default;
-#--error ER_DROP_PARTITION_NON_EXISTENT
-#ALTER TABLE t1 DROP PARTITION p3;
-#
-#--echo # Connection 2
-#connection con1;
-#--echo # This failed with deadlock and should not do so.
-#SELECT * FROM t1;
-#
-#--echo # Connection 1
-#connection default;
-#disconnect con1;
-#DROP TABLE t1;
+CREATE TABLE t1 (a int) PARTITION BY RANGE (a)
+(PARTITION p0 VALUES LESS THAN (1),
+ PARTITION p1 VALUES LESS THAN (2));
+
+INSERT INTO t1 VALUES (0),(1);
+
+connect(con1,localhost,root);
+
+--echo # Connection 2
+connection con1;
+BEGIN;
+SELECT * FROM t1;
+
+--echo # Connection 1
+connection default;
+--error ER_DROP_PARTITION_NON_EXISTENT
+ALTER TABLE t1 DROP PARTITION p3;
+
+--echo # Connection 2
+connection con1;
+--echo # This failed with deadlock and should not do so.
+SELECT * FROM t1;
+
+--echo # Connection 1
+connection default;
+disconnect con1;
+DROP TABLE t1;
+
+
+--echo #
+--echo # Bug #46654 False deadlock on concurrent DML/DDL 
+--echo #            with partitions, inconsistent behavior
+--echo #
+
+--disable_warnings
+DROP TABLE IF EXISTS tbl_with_partitions;
+--enable_warnings
+
+CREATE TABLE tbl_with_partitions ( i INT ) 
+	PARTITION BY HASH(i);
+INSERT INTO tbl_with_partitions VALUES (1);
+
+connect(con2,localhost,root);
+connect(con3,localhost,root);
+
+--echo # Connection 3
+connection con3;
+LOCK TABLE tbl_with_partitions READ;
+
+--echo # Connection 1
+--echo # Access table with disabled autocommit
+connection default;
+SET AUTOCOMMIT = 0;
+SELECT * FROM tbl_with_partitions;
+
+--echo # Connection 2
+--echo # Alter table, abort after prepare
+connection con2;
+set session debug="+d,abort_copy_table";
+--error ER_LOCK_WAIT_TIMEOUT
+ALTER TABLE tbl_with_partitions ADD COLUMN f INT;
+
+--echo # Connection 1
+--echo # Try accessing the table after Alter aborted.
+--echo # This used to give ER_LOCK_DEADLOCK.
+connection default;
+SELECT * FROM tbl_with_partitions;
+
+--echo # Connection 3
+connection con3;
+UNLOCK TABLES;
+
+--echo # Connection 1
+--echo # Cleanup
+connection default;
+disconnect con2;
+disconnect con3;
+DROP TABLE tbl_with_partitions;
 
 
 # Check that all connections opened by test cases in this file are really
diff --git a/mysql-test/t/sp-error.test b/mysql-test/t/sp-error.test
index e33adf56284..4df1118cd56 100644
--- a/mysql-test/t/sp-error.test
+++ b/mysql-test/t/sp-error.test
@@ -2490,6 +2490,35 @@ SELECT AVG (a) FROM t1 WHERE b = 999999;
 SELECT non_existent (a) FROM t1 WHERE b = 999999;
 DROP TABLE t1;
 
+
+#
+# Bug #46374 crash, INSERT INTO t1 uses function, function modifies t1
+#
+CREATE TABLE t1 ( f2 INTEGER, f3 INTEGER );
+INSERT INTO t1 VALUES  ( 1, 1 );
+
+delimiter |;
+
+CREATE FUNCTION func_1 () RETURNS INTEGER
+BEGIN
+  INSERT INTO t1 SELECT * FROM t1 ;
+  RETURN 1 ;
+END|
+
+delimiter ;|
+
+# The bug caused the following INSERT statement to trigger
+# an assertion.   Error 1442 is the correct response
+#
+--error 1442
+INSERT INTO t1 SELECT * FROM (SELECT 2 AS f1, 2 AS f2) AS A WHERE func_1() = 5;
+
+# Cleanup
+DROP FUNCTION func_1;
+DROP TABLE t1;
+
+
+
 --echo #
 --echo # Bug #47788: Crash in TABLE_LIST::hide_view_error on UPDATE + VIEW + 
 --echo #   SP + MERGE + ALTER
@@ -2513,3 +2542,4 @@ DROP VIEW v1;
 DROP TABLE t1;
 
 --echo End of 5.1 tests
+
diff --git a/sql/lock.cc b/sql/lock.cc
index bd91494f174..d414d7d6ae2 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -74,6 +74,7 @@
 */
 
 #include "mysql_priv.h"
+#include "debug_sync.h"
 #include <hash.h>
 #include <assert.h>
 
@@ -1125,9 +1126,33 @@ bool lock_global_read_lock(THD *thd)
   if (!thd->global_read_lock)
   {
     const char *old_message;
+    const char *new_message= "Waiting to get readlock";
     (void) pthread_mutex_lock(&LOCK_global_read_lock);
+
+#if defined(ENABLED_DEBUG_SYNC)
+    /*
+      The below sync point fires if we have to wait for
+      protect_against_global_read_lock.
+
+      WARNING: Beware to use WAIT_FOR with this sync point. We hold
+      LOCK_global_read_lock here.
+
+      Call the sync point before calling enter_cond() as it does use
+      enter_cond() and exit_cond() itself if a WAIT_FOR action is
+      executed in spite of the above warning.
+
+      Pre-set proc_info so that it is available immediately after the
+      sync point sends a SIGNAL. This makes tests more reliable.
+    */
+    if (protect_against_global_read_lock)
+    {
+      thd_proc_info(thd, new_message);
+      DEBUG_SYNC(thd, "wait_lock_global_read_lock");
+    }
+#endif /* defined(ENABLED_DEBUG_SYNC) */
+
     old_message=thd->enter_cond(&COND_global_read_lock, &LOCK_global_read_lock,
-                                "Waiting to get readlock");
+                                new_message);
     DBUG_PRINT("info",
 	       ("waiting_for: %d  protect_against: %d",
 		waiting_for_read_lock, protect_against_global_read_lock));
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index fad58900002..9f88393706a 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1534,8 +1534,8 @@ bool close_thread_table(THD *thd, TABLE **table_ptr)
   *table_ptr=table->next;
 
   table->mdl_ticket= NULL;
-  if (table->needs_reopen() ||
-      thd->version != refresh_version || !table->db_stat ||
+  if (table->s->needs_reopen() ||
+      thd->version != refresh_version || table->needs_reopen() ||
       table_def_shutdown_in_progress)
   {
     free_cache_entry(table);
@@ -2844,7 +2844,8 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     table exists now we should downgrade our exclusive metadata
     lock on this table to shared metadata lock.
   */
-  if (table_list->lock_strategy == TABLE_LIST::EXCLUSIVE_DOWNGRADABLE_MDL)
+  if (table_list->lock_strategy == TABLE_LIST::EXCLUSIVE_DOWNGRADABLE_MDL &&
+      !(flags & MYSQL_OPEN_HAS_MDL_LOCK))
     mdl_ticket->downgrade_exclusive_lock();
 
   table->mdl_ticket= mdl_ticket;
@@ -8190,13 +8191,13 @@ bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use)
        thd_table= thd_table->next)
   {
     /*
-      Check for TABLE::db_stat is needed since in some places we call
+      Check for TABLE::needs_reopen() is needed since in some places we call
       handler::close() for table instance (and set TABLE::db_stat to 0)
       and do not remove such instances from the THD::open_tables
       for some time, during which other thread can see those instances
       (e.g. see partitioning code).
     */
-    if (thd_table->db_stat)
+    if (!thd_table->needs_reopen())
       signalled|= mysql_lock_abort_for_thread(thd, thd_table);
   }
   pthread_mutex_unlock(&LOCK_open);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 889d7c5472b..4bfd3e66438 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3341,6 +3341,16 @@ public:
 */
 #define CF_DIAGNOSTIC_STMT        (1U << 8)
 
+/**
+  SQL statements that must be protected against impending global read lock
+  to prevent deadlock. This deadlock could otherwise happen if the statement
+  starts waiting for the GRL to go away inside mysql_lock_tables while at the
+  same time having "old" opened tables. The thread holding the GRL can be
+  waiting for these "old" opened tables to be closed, causing a deadlock
+  (FLUSH TABLES WITH READ LOCK).
+ */
+#define CF_PROTECT_AGAINST_GRL  (1U << 10)
+
 /* Bits in server_command_flags */
 
 /**
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 1966c7d8b39..94f5e84fb10 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -820,7 +820,7 @@ void mysql_ha_flush(THD *thd)
     if (hash_tables->table &&
         (hash_tables->table->mdl_ticket &&
          hash_tables->table->mdl_ticket->has_pending_conflicting_lock() ||
-         hash_tables->table->needs_reopen()))
+         hash_tables->table->s->needs_reopen()))
       mysql_ha_close_table(thd, hash_tables);
   }
 
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 8fd704c4f71..9a351085b3a 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2657,7 +2657,7 @@ bool Delayed_insert::handle_inserts(void)
 
   thd_proc_info(&thd, "insert");
   max_rows= delayed_insert_limit;
-  if (thd.killed || table->needs_reopen())
+  if (thd.killed || table->s->needs_reopen())
   {
     thd.killed= THD::KILL_CONNECTION;
     max_rows= ULONG_MAX;                     // Do as much as possible
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index e462556c133..289b6ce0da8 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -182,14 +182,15 @@ void init_update_queries(void)
   memset(sql_command_flags, 0, sizeof(sql_command_flags));
 
   sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_AUTO_COMMIT_TRANS;
+                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
-                                            CF_AUTO_COMMIT_TRANS;
+                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
                                             CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
+                                            CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_RENAME_TABLE]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
@@ -207,17 +208,17 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_DROP_TRIGGER]=   CF_AUTO_COMMIT_TRANS;
 
   sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
-                                            CF_REEXECUTION_FRAGILE;
+                                            CF_REEXECUTION_FRAGILE | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
-                                            CF_REEXECUTION_FRAGILE;
+                                            CF_REEXECUTION_FRAGILE | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
-                                            CF_REEXECUTION_FRAGILE;
+                                            CF_REEXECUTION_FRAGILE | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
-                                            CF_REEXECUTION_FRAGILE;
+                                            CF_REEXECUTION_FRAGILE | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
-                                            CF_REEXECUTION_FRAGILE;
+                                            CF_REEXECUTION_FRAGILE | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
-                                            CF_REEXECUTION_FRAGILE;
+                                            CF_REEXECUTION_FRAGILE | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
                                             CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
@@ -276,20 +277,21 @@ void init_update_queries(void)
                                                 CF_REEXECUTION_FRAGILE);
 
 
-  sql_command_flags[SQLCOM_CREATE_USER]=       CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_RENAME_USER]=       CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_DROP_USER]=         CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_CREATE_USER]=       CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_RENAME_USER]=       CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_DROP_USER]=         CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_GRANT]=             CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_REVOKE]=            CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_REVOKE_ALL]=        CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_ALTER_DB]=          CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_CREATE_FUNCTION]=   CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_DROP_FUNCTION]=     CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_DROP_FUNCTION]=     CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_OPTIMIZE]=          CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_CREATE_PROCEDURE]=  CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_CREATE_SPFUNCTION]= CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_DROP_PROCEDURE]=    CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_ALTER_PROCEDURE]=   CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_ALTER_FUNCTION]=    CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_CREATE_PROCEDURE]=  CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_CREATE_SPFUNCTION]= CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_DROP_PROCEDURE]=    CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_ALTER_PROCEDURE]=   CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_ALTER_FUNCTION]=    CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_INSTALL_PLUGIN]=    CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_UNINSTALL_PLUGIN]=  CF_CHANGES_DATA;
 
@@ -1969,6 +1971,17 @@ mysql_execute_command(THD *thd)
       thd->mdl_context.release_all_locks();
   }
 
+  /*
+    Check if this command needs protection against the global read lock
+    to avoid deadlock. See CF_PROTECT_AGAINST_GRL.
+    start_waiting_global_read_lock() is called at the end of
+    mysql_execute_command().
+  */
+  if (((sql_command_flags[lex->sql_command] & CF_PROTECT_AGAINST_GRL) != 0) &&
+      !thd->locked_tables_mode)
+    if (wait_if_global_read_lock(thd, FALSE, TRUE))
+      goto error;
+
   switch (lex->sql_command) {
 
   case SQLCOM_SHOW_EVENTS:
@@ -2309,12 +2322,9 @@ case SQLCOM_PREPARE:
       start_waiting_global_read_lock(). We protect the normal CREATE
       TABLE in the same way. That way we avoid that a new table is
       created during a global read lock.
+      Protection against grl is covered by the CF_PROTECT_AGAINST_GRL flag.
     */
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-    {
-      res= 1;
-      goto end_with_restore_list;
-    }
+
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     {
       partition_info *part_info= thd->lex->part_info;
@@ -2617,12 +2627,6 @@ end_with_restore_list:
                             "INDEX DIRECTORY");
       create_info.data_file_name= create_info.index_file_name= NULL;
 
-      if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-      {
-        res= 1;
-        break;
-      }
-
       thd->enable_slow_log= opt_log_slow_admin_statements;
       res= mysql_alter_table(thd, select_lex->db, lex->name.str,
                              &create_info,
@@ -2852,8 +2856,6 @@ end_with_restore_list:
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
     if (update_precheck(thd, all_tables))
       break;
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-      goto error;
     DBUG_ASSERT(select_lex->offset_limit == 0);
     unit->set_limit(select_lex);
     MYSQL_UPDATE_START(thd->query());
@@ -2884,15 +2886,6 @@ end_with_restore_list:
     else
       res= 0;
 
-    /*
-      Protection might have already been risen if its a fall through
-      from the SQLCOM_UPDATE case above.
-    */
-    if (!thd->locked_tables_mode &&
-        lex->sql_command == SQLCOM_UPDATE_MULTI &&
-        wait_if_global_read_lock(thd, 0, 1))
-      goto error;
-
     res= mysql_multi_update_prepare(thd);
 
 #ifdef HAVE_REPLICATION
@@ -2992,11 +2985,6 @@ end_with_restore_list:
     if ((res= insert_precheck(thd, all_tables)))
       break;
 
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-    {
-      res= 1;
-      break;
-    }
     MYSQL_INSERT_START(thd->query());
     res= mysql_insert(thd, all_tables, lex->field_list, lex->many_values,
 		      lex->update_list, lex->value_list,
@@ -3031,11 +3019,6 @@ end_with_restore_list:
 
     unit->set_limit(select_lex);
 
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-    {
-      res= 1;
-      break;
-    }
     if (!(res= open_and_lock_tables(thd, all_tables)))
     {
       MYSQL_INSERT_SELECT_START(thd->query());
@@ -3113,11 +3096,6 @@ end_with_restore_list:
     DBUG_ASSERT(select_lex->offset_limit == 0);
     unit->set_limit(select_lex);
 
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-    {
-      res= 1;
-      break;
-    }
     MYSQL_DELETE_START(thd->query());
     res = mysql_delete(thd, all_tables, select_lex->where,
                        &select_lex->order_list,
@@ -3133,12 +3111,6 @@ end_with_restore_list:
       (TABLE_LIST *)thd->lex->auxiliary_table_list.first;
     multi_delete *del_result;
 
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-    {
-      res= 1;
-      break;
-    }
-
     if ((res= multi_delete_precheck(thd, all_tables)))
       break;
 
@@ -3277,9 +3249,6 @@ end_with_restore_list:
     if (check_one_table_access(thd, privilege, all_tables))
       goto error;
 
-    if (!thd->locked_tables_mode && wait_if_global_read_lock(thd, 0, 1))
-      goto error;
-
     res= mysql_load(thd, lex->exchange, first_table, lex->field_list,
                     lex->update_list, lex->value_list, lex->duplicates,
                     lex->ignore, (bool) lex->local_file);
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 718471cc1b6..bcf81a49f56 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -4183,13 +4183,13 @@ bool mysql_unpack_partition(THD *thd,
       We need to free any memory objects allocated on item_free_list
       by the parser since we are keeping the old info from the first
       parser call in CREATE TABLE.
-      We'll ensure that this object isn't put into table cache also
-      just to ensure we don't get into strange situations with the
-      item objects.
+
+      This table object can not be used any more. However, since
+      this is CREATE TABLE, we know that it will be destroyed by the
+      caller, and rely on that.
     */
     thd->free_items();
     part_info= thd->work_part_info;
-    table->s->version= 0UL;
     *work_part_info_used= true;
   }
   table->part_info= part_info;
@@ -4482,12 +4482,11 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
 
   /*
     We are going to manipulate the partition info on the table object
-    so we need to ensure that the data structure of the table object
-    is freed by setting version to 0. table->s->version= 0 forces a
-    flush of the table object in close_thread_tables().
+    so we need to ensure that the table instance is removed from the
+    table cache.
   */
   if (table->part_info)
-    table->s->version= 0L;
+    table->m_needs_reopen= TRUE;
 
   thd->work_part_info= thd->lex->part_info;
   if (thd->work_part_info &&
@@ -6242,7 +6241,9 @@ static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
         alter_partition_lock_handling() and the table is closed
         by close_thread_tables() instead.
       */
-      table->s->version= 0;
+      tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED,
+                       table->s->db.str,
+                       table->s->table_name.str);
     }
   }
   pthread_mutex_unlock(&LOCK_open);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 5320da45322..2195719a3e9 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -11096,6 +11096,13 @@ do_select(JOIN *join,List<Item> *fields,TABLE *table,Procedure *procedure)
                                  fields);
       rc= join->result->send_data(*columns_list);
     }
+    /*
+      An error can happen when evaluating the conds 
+      (the join condition and piece of where clause 
+      relevant to this join table).
+    */
+    if (join->thd->is_error())
+      error= NESTED_LOOP_ERROR;
   }
   else
   {
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index e9474d9add6..556c0e65eee 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -7049,6 +7049,10 @@ view_err:
     new_table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
     new_table->next_number_field=new_table->found_next_number_field;
     thd_proc_info(thd, "copy to tmp table");
+    DBUG_EXECUTE_IF("abort_copy_table", {
+        my_error(ER_LOCK_WAIT_TIMEOUT, MYF(0));
+        goto err_new_table_cleanup;
+      });
     error= copy_data_between_tables(table, new_table,
                                     alter_info->create_list, ignore,
                                     order_num, order, &copied, &deleted,
diff --git a/sql/table.h b/sql/table.h
index b44c2e9f3be..4b85d8a64f4 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -294,6 +294,8 @@ TABLE_CATEGORY get_table_category(const LEX_STRING *db,
 
 struct TABLE_share;
 
+extern ulong refresh_version;
+
 /*
   This structure is shared between different table objects. There is one
   instance of table share per one table in the database.
@@ -503,6 +505,14 @@ struct TABLE_SHARE
     return table_map_id;
   }
 
+
+  /*
+    Must all TABLEs be reopened?
+  */
+  inline bool needs_reopen()
+  {
+    return version != refresh_version;
+  }
   /**
     Convert unrelated members of TABLE_SHARE to one enum
     representing its type.
@@ -605,8 +615,6 @@ struct TABLE_SHARE
 };
 
 
-extern ulong refresh_version;
-
 /* Information for one open table */
 enum index_hint_type
 {
@@ -804,6 +812,7 @@ public:
   my_bool insert_or_update;             /* Can be used by the handler */
   my_bool alias_name_used;		/* true if table_name is alias */
   my_bool get_fields_in_item_tree;      /* Signal to fix_field */
+  my_bool m_needs_reopen;
 
   REGINFO reginfo;			/* field connections */
   MEM_ROOT mem_root;
@@ -853,7 +862,7 @@ public:
     Is this instance of the table should be reopen?
   */
   inline bool needs_reopen()
-  { return s->version != refresh_version; }
+  { return !db_stat || m_needs_reopen; }
 };