bzr commit into mysql-5.1 branch (tomas.ulin:2695)

[Tomas Ulin]

#At file:///home/tomas/mysql_src/mysql-5.1-telco-6.3/

 2695 Tomas Ulin	2008-10-02 [merge]
      merge
modified:
  sql/ha_ndbcluster.cc
  sql/ha_ndbcluster.h
  sql/ha_ndbcluster_binlog.cc
  sql/ha_ndbcluster_binlog.h
  sql/sql_db.cc
  sql/sql_parse.cc
  sql/sql_rename.cc
  sql/sql_table.cc

=== modified file 'sql/ha_ndbcluster.cc'
--- a/sql/ha_ndbcluster.cc	2008-10-02 06:43:17 +0000
+++ b/sql/ha_ndbcluster.cc	2008-10-02 08:19:14 +0000
@@ -68,6 +68,13 @@ TYPELIB ndb_distribution_typelib= { arra
 const char *opt_ndb_distribution= ndb_distribution_names[ND_KEYHASH];
 enum ndb_distribution opt_ndb_distribution_id= ND_KEYHASH;
 
+/*
+  Provided for testing purposes to be able to run full test suite
+  with --ndbcluster option without getting warnings about cluster
+  not being connected
+*/
+my_bool ndbcluster_silent= 0;
+
 // Default value for parallelism
 static const int parallelism= 0;
 
@@ -622,6 +629,7 @@ Thd_ndb::Thd_ndb()
   m_conflict_fn_usage_count= 0;
   global_schema_lock_trans= NULL;
   global_schema_lock_count= 0;
+  global_schema_lock_error= 0;
   init_alloc_root(&m_batch_mem_root, BATCH_FLUSH_SIZE/4, 0);
 }
 
@@ -629,19 +637,6 @@ Thd_ndb::~Thd_ndb()
 {
   if (ndb)
   {
-#ifndef DBUG_OFF
-    Ndb::Free_list_usage tmp;
-    tmp.m_name= 0;
-    while (ndb->get_free_list_usage(&tmp))
-    {
-      uint leaked= (uint) tmp.m_created - tmp.m_free;
-      if (leaked)
-        fprintf(stderr, "NDB: Found %u %s%s that %s not been released\n",
-                leaked, tmp.m_name,
-                (leaked == 1)?"":"'s",
-                (leaked == 1)?"has":"have");
-    }
-#endif
     delete ndb;
     ndb= NULL;
   }
@@ -5312,6 +5307,9 @@ int ha_ndbcluster::start_statement(THD *
     /* This is currently dead code in wait for implementation in NDB */
     /* lockThisTable(); */
     DBUG_PRINT("info", ("Locking the table..." ));
+    push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+                        ER_GET_ERRMSG, ER(ER_GET_ERRMSG), 0,
+                        "Table only locked locally in this mysqld", "NDB");
   }
   DBUG_RETURN(0);
 }
@@ -6302,8 +6300,8 @@ int ha_ndbcluster::create(const char *na
 
   if (!((thd_ndb->options & TNO_NO_LOCK_SCHEMA_OP) ||
         ndbcluster_has_global_schema_lock(thd_ndb)))
-    ndbcluster_no_global_schema_lock_abort
-      (thd, "ha_ndbcluster::create");
+    DBUG_RETURN(ndbcluster_no_global_schema_lock_abort
+                (thd, "ha_ndbcluster::create"));
   /*
     Don't allow table creation unless
     schema distribution table is setup
@@ -6954,11 +6952,9 @@ int ha_ndbcluster::rename_table(const ch
   if (check_ndb_connection(thd))
     DBUG_RETURN(my_errno= HA_ERR_NO_CONNECTION);
 
-#ifdef HAVE_NDB_BINLOG
   if (!ndbcluster_has_global_schema_lock(get_thd_ndb(thd)))
-    ndbcluster_no_global_schema_lock_abort
-      (thd, "ha_ndbcluster::rename_table");
-#endif
+    DBUG_RETURN(ndbcluster_no_global_schema_lock_abort
+                (thd, "ha_ndbcluster::rename_table"));
 
   Ndb *ndb= get_ndb(thd);
   ndb->setDatabaseName(old_dbname);
@@ -7341,7 +7337,6 @@ int ha_ndbcluster::delete_table(const ch
   if (!ndb_schema_share)
   {
     DBUG_PRINT("info", ("Schema distribution table not setup"));
-    DBUG_ASSERT(ndb_schema_share);
     error= HA_ERR_NO_CONNECTION;
     goto err;
   }
@@ -7354,11 +7349,9 @@ int ha_ndbcluster::delete_table(const ch
 
   ndb= get_ndb(thd);
 
-#ifdef HAVE_NDB_BINLOG
   if (!ndbcluster_has_global_schema_lock(get_thd_ndb(thd)))
-    ndbcluster_no_global_schema_lock_abort
-      (thd, "ha_ndbcluster::delete_table");
-#endif
+    DBUG_RETURN(ndbcluster_no_global_schema_lock_abort
+                (thd, "ha_ndbcluster::delete_table"));
 
   /*
     Drop table in ndb.
@@ -8120,7 +8113,10 @@ err:
                              share->key, share->use_count));
     free_share(&share);
   }
-  if (ndb_error.code)
+  /*
+    ndbcluster_silent - avoid "cluster disconnected error"
+  */
+  if (ndb_error.code && (!ndbcluster_silent || ndb_error.code != 4009))
   {
     ERR_RETURN(ndb_error);
   }
@@ -8144,7 +8140,15 @@ int ndbcluster_table_exists_in_engine(ha
   NDBDICT* dict= ndb->getDictionary();
   NdbDictionary::Dictionary::List list;
   if (dict->listObjects(list, NdbDictionary::Object::UserTable) != 0)
+  {
+    /*
+      ndbcluster_silent
+      - avoid "cluster failure" warning if cluster is not connected
+    */
+    if (ndbcluster_silent && dict->getNdbError().code == 4009)
+      DBUG_RETURN(HA_ERR_NO_SUCH_TABLE);
     ERR_RETURN(dict->getNdbError());
+  }
   for (uint i= 0 ; i < list.count ; i++)
   {
     NdbDictionary::Dictionary::List::Element& elmt= list.elements[i];
@@ -8244,7 +8248,6 @@ static void ndbcluster_drop_database(han
   if (!ndb_schema_share)
   {
     DBUG_PRINT("info", ("Schema distribution table not setup"));
-    DBUG_ASSERT(ndb_schema_share);
     DBUG_VOID_RETURN;
   }
   ndbcluster_drop_database_impl(thd, path);
@@ -8421,11 +8424,9 @@ int ndbcluster_find_files(handlerton *ht
   if (dir)
     DBUG_RETURN(0); // Discover of databases not yet supported
 
-#ifdef HAVE_NDB_BINLOG
   Ndbcluster_global_schema_lock_guard ndbcluster_global_schema_lock_guard(thd);
   if (ndbcluster_global_schema_lock_guard.lock())
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
-#endif
 
   // List tables in NDB
   NDBDICT *dict= ndb->getDictionary();
@@ -8813,22 +8814,6 @@ static int ndbcluster_end(handlerton *ht
   }
   hash_free(&ndbcluster_open_tables);
 
-  if (g_ndb)
-  {
-#ifndef DBUG_OFF
-    Ndb::Free_list_usage tmp;
-    tmp.m_name= 0;
-    while (g_ndb->get_free_list_usage(&tmp))
-    {
-      uint leaked= (uint) tmp.m_created - tmp.m_free;
-      if (leaked)
-        fprintf(stderr, "NDB: Found %u %s%s that %s not been released\n",
-                leaked, tmp.m_name,
-                (leaked == 1)?"":"'s",
-                (leaked == 1)?"has":"have");
-    }
-#endif
-  }
   ndbcluster_disconnect();
 
   // cleanup ndb interface
@@ -11732,6 +11717,10 @@ int ha_ndbcluster::alter_table_phase1(TH
   adding=  adding | HA_ADD_INDEX | HA_ADD_UNIQUE_INDEX;
   dropping= dropping | HA_DROP_INDEX | HA_DROP_UNIQUE_INDEX;
 
+  if (!ndbcluster_has_global_schema_lock(get_thd_ndb(thd)))
+    DBUG_RETURN(ndbcluster_no_global_schema_lock_abort
+                (thd, "ha_ndbcluster::alter_table_phase1"));
+
   if (!(alter_data= new NDB_ALTER_DATA(dict, m_table)))
     DBUG_RETURN(HA_ERR_OUT_OF_MEM);
   old_tab= alter_data->old_table;
@@ -11922,6 +11911,10 @@ int ha_ndbcluster::alter_table_phase2(TH
   DBUG_ENTER("alter_table_phase2");
   dropping= dropping  | HA_DROP_INDEX | HA_DROP_UNIQUE_INDEX;
 
+  if (!ndbcluster_has_global_schema_lock(get_thd_ndb(thd)))
+    DBUG_RETURN(ndbcluster_no_global_schema_lock_abort
+                (thd, "ha_ndbcluster::alter_table_phase2"));
+
   if ((*alter_flags & dropping).is_set())
   {
     /* Tell the handler to finally drop the indexes. */
@@ -11952,6 +11945,10 @@ int ha_ndbcluster::alter_table_phase2(TH
 int ha_ndbcluster::alter_table_phase3(THD *thd, TABLE *table)
 {
   DBUG_ENTER("alter_table_phase3");
+
+  if (!ndbcluster_has_global_schema_lock(get_thd_ndb(thd)))
+    DBUG_RETURN(ndbcluster_no_global_schema_lock_abort
+                (thd, "ha_ndbcluster::alter_table_phase3"));
 
   const char *db= table->s->db.str;
   const char *name= table->s->table_name.str;

=== modified file 'sql/ha_ndbcluster.h'
--- a/sql/ha_ndbcluster.h	2008-09-30 10:42:32 +0000
+++ b/sql/ha_ndbcluster.h	2008-10-02 08:19:14 +0000
@@ -330,6 +330,7 @@ class Thd_ndb 
 
   NdbTransaction *global_schema_lock_trans;
   uint global_schema_lock_count;
+  uint global_schema_lock_error;
 };
 
 int ndbcluster_commit(handlerton *hton, THD *thd, bool all);

=== modified file 'sql/ha_ndbcluster_binlog.cc'
--- a/sql/ha_ndbcluster_binlog.cc	2008-09-30 10:42:32 +0000
+++ b/sql/ha_ndbcluster_binlog.cc	2008-10-02 08:19:14 +0000
@@ -38,6 +38,8 @@ extern my_bool opt_ndb_log_orig;
 extern my_bool opt_ndb_log_update_as_write;
 extern my_bool opt_ndb_log_updated_only;
 
+extern my_bool ndbcluster_silent;
+
 /*
   defines for cluster replication table names
 */
@@ -473,8 +475,14 @@ static void ndbcluster_binlog_wait(THD *
   if (ndb_binlog_running)
   {
     DBUG_ENTER("ndbcluster_binlog_wait");
-    const char *save_info= thd ? thd->proc_info : 0;
     ulonglong wait_epoch= ndb_get_latest_trans_gci();
+    /*
+      cluster not connected or no transactions done
+      so nothing to wait for
+    */
+    if (!wait_epoch)
+      DBUG_VOID_RETURN;
+    const char *save_info= thd ? thd->proc_info : 0;
     int count= 30;
     if (thd)
       thd->proc_info= "Waiting for ndbcluster binlog update to "
@@ -697,13 +705,21 @@ static void ndbcluster_reset_slave(THD *
   char *end= strmov(buf, "DELETE FROM " NDB_REP_DB "." NDB_APPLY_TABLE);
   run_query(thd, buf, end, NULL, TRUE);
   if (thd->main_da.is_error() &&
-      thd->main_da.sql_errno() == ER_NO_SUCH_TABLE)
+      ((thd->main_da.sql_errno() == ER_NO_SUCH_TABLE) ||
+       (thd->main_da.sql_errno() == ER_OPEN_AS_READONLY &&
ndbcluster_silent)))
   {
     /*
       If table does not exist ignore the error as it
       is a consistant behavior
     */
     thd->main_da.reset_diagnostics_area();
+    /*
+      ndbcluster_silent
+      - avoid "no table mysql.ndb_apply_status" warning - ER_NO_SUCH_TABLE
+      - avoid "mysql.ndb_apply_status read only" warning - ER_OPEN_AS_READONLY
+    */
+    if (ndbcluster_silent)
+      mysql_reset_errors(thd, 1);
   }
 
   DBUG_VOID_RETURN;
@@ -737,8 +753,11 @@ int ndbcluster_has_global_schema_lock(Th
   return 0;
 }
 
-void ndbcluster_no_global_schema_lock_abort(THD *thd, const char *msg)
+int ndbcluster_no_global_schema_lock_abort(THD *thd, const char *msg)
 {
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  if (thd_ndb && thd_ndb->global_schema_lock_error != 0)
+    return HA_ERR_NO_CONNECTION;
   sql_print_error("NDB: programming error, no lock taken while running "
                   "query %s. Message: %s", thd->query, msg);
   abort();
@@ -752,9 +771,11 @@ static int ndbcluster_global_schema_lock
   Ndb *ndb= check_ndb_in_thd(thd);
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
   NdbError ndb_error;
+  thd_ndb->global_schema_lock_error= 0;
   if (thd_ndb->options & TNO_NO_LOCK_SCHEMA_OP)
     return 0;
   DBUG_ENTER("ndbcluster_global_schema_lock");
+  DBUG_PRINT("enter", ("query: %s", thd->query));
   if (thd_ndb->global_schema_lock_trans)
   {
     thd_ndb->global_schema_lock_trans->refresh();
@@ -773,6 +794,11 @@ static int ndbcluster_global_schema_lock
     DBUG_RETURN(0);
   }
 
+  /*
+    ndbcluster_silent - avoid "cluster disconnected error"
+  */
+  if (ndbcluster_silent)
+    report_cluster_disconnected= 0;
   if (ndb_error.code != 4009 || report_cluster_disconnected)
   {
     sql_print_warning("NDB: Could not acquire global schema lock (%d)%s",
@@ -780,14 +806,17 @@ static int ndbcluster_global_schema_lock
     push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
                         ER_GET_ERRMSG, ER(ER_GET_ERRMSG),
                         ndb_error.code, ndb_error.message,
-                        "ndb. Could not acquire global schema lock");
+                        "NDB. Could not acquire global schema lock");
   }
+  thd_ndb->global_schema_lock_error= ndb_error.code;
   DBUG_RETURN(-1);
 }
 static int ndbcluster_global_schema_unlock(THD *thd)
 {
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
   DBUG_ASSERT(thd_ndb != 0);
+  if (thd_ndb)
+    thd_ndb->global_schema_lock_error= 0;
   if (thd_ndb == 0 || (thd_ndb->options & TNO_NO_LOCK_SCHEMA_OP))
     return 0;
   Ndb *ndb= thd_ndb->ndb;
@@ -868,6 +897,12 @@ int Ndbcluster_global_schema_lock_guard:
   m_lock= !ndbcluster_global_schema_lock(m_thd, 0);
   return !m_lock;
 }
+void Ndbcluster_global_schema_lock_guard::unlock()
+{
+  if (m_lock)
+    ndbcluster_global_schema_unlock(m_thd);
+  m_lock= 0;
+}
 
 void ndbcluster_binlog_init_handlerton()
 {
@@ -1111,8 +1146,8 @@ static int ndbcluster_find_all_databases
   DBUG_ENTER("ndbcluster_find_all_databases");
   /* Ensure that we have the right lock */
   if (!ndbcluster_has_global_schema_lock(thd_ndb))
-    ndbcluster_no_global_schema_lock_abort
-      (thd, "ndbcluster_find_all_databases");
+    DBUG_RETURN(ndbcluster_no_global_schema_lock_abort
+                (thd, "ndbcluster_find_all_databases"));
   ndb->setDatabaseName(NDB_REP_DB);
   thd_ndb_options.set(TNO_NO_LOG_SCHEMA_OP);
   thd_ndb_options.set(TNO_NO_LOCK_SCHEMA_OP);

=== modified file 'sql/ha_ndbcluster_binlog.h'
--- a/sql/ha_ndbcluster_binlog.h	2008-09-30 10:42:32 +0000
+++ b/sql/ha_ndbcluster_binlog.h	2008-10-02 08:19:14 +0000
@@ -288,7 +288,7 @@ inline void do_retry_sleep(unsigned mill
 }
 
 int ndbcluster_has_global_schema_lock(Thd_ndb *thd_ndb);
-void ndbcluster_no_global_schema_lock_abort(THD *thd, const char *msg);
+int ndbcluster_no_global_schema_lock_abort(THD *thd, const char *msg);
 
 class Ndbcluster_global_schema_lock_guard
 {
@@ -296,6 +296,7 @@ public:
   Ndbcluster_global_schema_lock_guard(THD *thd);
   ~Ndbcluster_global_schema_lock_guard();
   int lock();
+  void unlock();
 private:
   THD *m_thd;
   int m_lock;

=== modified file 'sql/sql_db.cc'
--- a/sql/sql_db.cc	2008-09-30 09:14:44 +0000
+++ b/sql/sql_db.cc	2008-10-02 06:39:01 +0000
@@ -616,6 +616,7 @@ int mysql_create_db(THD *thd, char *db, 
   MY_STAT stat_info;
   uint create_options= create_info ? create_info->options : 0;
   uint path_len;
+  Ha_global_schema_lock_guard global_schema_lock_guard(thd);
   DBUG_ENTER("mysql_create_db");
 
   /* do not create 'information_schema' db */
@@ -625,8 +626,6 @@ int mysql_create_db(THD *thd, char *db, 
     DBUG_RETURN(-1);
   }
 
-  ha_global_schema_lock(thd);
-
   /*
     Do not create database if another thread is holding read lock.
     Wait for global read lock before acquiring LOCK_mysql_create_db.
@@ -645,6 +644,8 @@ int mysql_create_db(THD *thd, char *db, 
     goto exit2;
   }
 
+  global_schema_lock_guard.lock();
+
   VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
 
   /* Check directory */
@@ -758,7 +759,6 @@ exit:
   VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
   start_waiting_global_read_lock(thd);
 exit2:
-  ha_global_schema_unlock(thd);
   DBUG_RETURN(error);
 }
 
@@ -770,10 +770,9 @@ bool mysql_alter_db(THD *thd, const char
   char path[FN_REFLEN+16];
   long result=1;
   int error= 0;
+  Ha_global_schema_lock_guard global_schema_lock_guard(thd);
   DBUG_ENTER("mysql_alter_db");
 
-  ha_global_schema_lock(thd);
-
   /*
     Do not alter database if another thread is holding read lock.
     Wait for global read lock before acquiring LOCK_mysql_create_db.
@@ -789,6 +788,8 @@ bool mysql_alter_db(THD *thd, const char
   if ((error=wait_if_global_read_lock(thd,0,1)))
     goto exit2;
 
+  global_schema_lock_guard.lock();
+
   VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
 
   /* 
@@ -837,7 +838,6 @@ exit:
   VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
   start_waiting_global_read_lock(thd);
 exit2:
-  ha_global_schema_unlock(thd);
   DBUG_RETURN(error);
 }
 
@@ -867,10 +867,9 @@ bool mysql_rm_db(THD *thd,char *db,bool 
   MY_DIR *dirp;
   uint length;
   TABLE_LIST* dropped_tables= 0;
+  Ha_global_schema_lock_guard global_schema_lock_guard(thd);
   DBUG_ENTER("mysql_rm_db");
 
-  ha_global_schema_lock(thd);
-
   /*
     Do not drop database if another thread is holding read lock.
     Wait for global read lock before acquiring LOCK_mysql_create_db.
@@ -889,6 +888,8 @@ bool mysql_rm_db(THD *thd,char *db,bool 
     goto exit2;
   }
 
+  global_schema_lock_guard.lock();
+
   VOID(pthread_mutex_lock(&LOCK_mysql_create_db));
 
   length= build_table_filename(path, sizeof(path), db, "", "", 0);
@@ -1036,7 +1037,6 @@ exit:
   VOID(pthread_mutex_unlock(&LOCK_mysql_create_db));
   start_waiting_global_read_lock(thd);
 exit2:
-  ha_global_schema_unlock(thd);
   DBUG_RETURN(error);
 }
 

=== modified file 'sql/sql_parse.cc'
--- a/sql/sql_parse.cc	2008-09-30 09:14:44 +0000
+++ b/sql/sql_parse.cc	2008-10-02 06:39:01 +0000
@@ -2419,7 +2419,6 @@ mysql_execute_command(THD *thd)
       copy.
     */
     Alter_info alter_info(lex->alter_info, thd->mem_root);
-    Ha_global_schema_lock_guard global_schema_lock_guard(thd);
 
     if (thd->is_fatal_error)
     {
@@ -2455,8 +2454,6 @@ mysql_execute_command(THD *thd)
       create_info.default_table_charset= create_info.table_charset;
       create_info.table_charset= 0;
     }
-    if (!(create_info.options & HA_LEX_CREATE_TMP_TABLE))
-      global_schema_lock_guard.lock();
     /*
       The create-select command will open and read-lock the select table
       and then create, open and write-lock the new table. If a global
@@ -2490,6 +2487,7 @@ mysql_execute_command(THD *thd)
     if (select_lex->item_list.elements)		// With select
     {
       select_result *result;
+      Ha_global_schema_lock_guard global_schema_lock_guard(thd);
 
       select_lex->options|= SELECT_NO_UNLOCK;
       unit->set_limit(select_lex);
@@ -2511,6 +2509,8 @@ mysql_execute_command(THD *thd)
       {
         lex->link_first_table_back(create_table, link_to_local);
         create_table->create= TRUE;
+        if (!thd->locked_tables)
+          global_schema_lock_guard.lock();
       }
 
       if (!(res= open_and_lock_tables(thd, lex->query_tables)))

=== modified file 'sql/sql_rename.cc'
--- a/sql/sql_rename.cc	2008-09-30 09:14:44 +0000
+++ b/sql/sql_rename.cc	2008-10-02 06:39:01 +0000
@@ -54,8 +54,6 @@ bool mysql_rename_tables(THD *thd, TABLE
 
   mysql_ha_rm_tables(thd, table_list, FALSE);
 
-  global_schema_lock_guard.lock();
-
   if (wait_if_global_read_lock(thd,0,1))
     DBUG_RETURN(1);
 
@@ -135,6 +133,13 @@ bool mysql_rename_tables(THD *thd, TABLE
       DBUG_RETURN(1);
     }
   }
+
+  /*
+    Rename table may clash with parallell or existing ndb table
+    thus a global shema lock is needed to ensure global serialization.
+    Moreover we do not know here what table type the tables have.
+  */
+  global_schema_lock_guard.lock();
 
   pthread_mutex_lock(&LOCK_open);
   if (lock_table_names_exclusively(thd, table_list))

=== modified file 'sql/sql_table.cc'
--- a/sql/sql_table.cc	2008-09-30 09:14:44 +0000
+++ b/sql/sql_table.cc	2008-10-02 06:39:01 +0000
@@ -1474,14 +1474,12 @@ bool mysql_rm_table(THD *thd,TABLE_LIST 
                     my_bool drop_temporary)
 {
   bool error= FALSE, need_start_waiters= FALSE;
-  Ha_global_schema_lock_guard global_schema_lock_guard(thd);
   DBUG_ENTER("mysql_rm_table");
 
   /* mark for close and remove all cached entries */
 
   if (!drop_temporary)
   {
-    global_schema_lock_guard.lock();
     if ((error= wait_if_global_read_lock(thd, 0, 1)))
     {
       my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), tables->table_name);
@@ -1565,6 +1563,15 @@ int mysql_rm_table_part2(THD *thd, TABLE
 
   mysql_ha_rm_tables(thd, tables, FALSE);
 
+  Ha_global_schema_lock_guard global_schema_lock_guard(thd);
+  /*
+    Here we have no way of knowing if an ndb table is one of
+    the ones that should be dropped, so we take the global
+    schema lock to be safe.
+  */
+  if (!drop_temporary)
+    global_schema_lock_guard.lock();
+
   pthread_mutex_lock(&LOCK_open);
 
   /*
@@ -3610,8 +3617,12 @@ bool mysql_create_table(THD *thd, const 
 {
   TABLE *name_lock= 0;
   bool result;
+  Ha_global_schema_lock_guard global_schema_lock_guard(thd);
   DBUG_ENTER("mysql_create_table");
 
+  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+    global_schema_lock_guard.lock();
+
   /* Wait for any database locks */
   pthread_mutex_lock(&LOCK_lock_db);
   while (!thd->killed &&
@@ -4833,9 +4844,13 @@ bool mysql_create_like_table(THD* thd, T
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   char tmp_path[FN_REFLEN];
 #endif
+  Ha_global_schema_lock_guard global_schema_lock_guard(thd);
   DBUG_ENTER("mysql_create_like_table");
 
 
+  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+    global_schema_lock_guard.lock();
+
   /*
     By opening source table we guarantee that it exists and no concurrent
     DDL operation will mess with it. Later we also take an exclusive
@@ -6571,6 +6586,11 @@ bool mysql_alter_table(THD *thd,char *ne
 
     if (wait_if_global_read_lock(thd,0,1))
       DBUG_RETURN(TRUE);
+
+    Ha_global_schema_lock_guard global_schema_lock_guard(thd);
+    if (table_type == DB_TYPE_NDBCLUSTER)
+      global_schema_lock_guard.lock();
+
     VOID(pthread_mutex_lock(&LOCK_open));
     if (lock_table_names(thd, table_list))
     {
@@ -6598,7 +6618,19 @@ view_err:
   }
 
   Ha_global_schema_lock_guard global_schema_lock_guard(thd);
-  global_schema_lock_guard.lock();
+  if (table_type == DB_TYPE_NDBCLUSTER)
+  {
+    /*
+      To avoid deadlock in this situation
+    */
+    if (thd->locked_tables)
+    {
+      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
+                 ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
+      DBUG_RETURN(TRUE);
+    }
+    global_schema_lock_guard.lock();
+  }
   if (!(table= open_n_lock_single_table(thd, table_list, TL_WRITE_ALLOW_READ)))
     DBUG_RETURN(TRUE);
   table->use_all_columns();