BUG#27254: Single User Mode.Mysql hangs if it tries delete a ndb table

sql/ha_ndbcluster.cc

@@ -4742,6 +4742,7 @@ int ha_ndbcluster::create(const char *name,
   bool create_from_engine= (create_info->table_options & HA_OPTION_CREATE_FROM_ENGINE);
   bool is_truncate= (thd->lex->sql_command == SQLCOM_TRUNCATE);
   char tablespace[FN_LEN];
+  NdbDictionary::Table::SingleUserMode single_user_mode= NdbDictionary::Table::SingleUserModeLocked;
 
   DBUG_ENTER("ha_ndbcluster::create");
   DBUG_PRINT("enter", ("name: %s", name));
@@ -4793,18 +4794,22 @@ int ha_ndbcluster::create(const char *name,
     schema distribution table is setup
     ( unless it is a creation of the schema dist table itself )
   */
-  if (!ndb_schema_share &&
-      !(strcmp(m_dbname, NDB_REP_DB) == 0 &&
+  if (!ndb_schema_share)
+  {
+    if (!(strcmp(m_dbname, NDB_REP_DB) == 0 &&
           strcmp(m_tabname, NDB_SCHEMA_TABLE) == 0))
     {
       DBUG_PRINT("info", ("Schema distribution table not setup"));
       DBUG_RETURN(HA_ERR_NO_CONNECTION);
     }
+    single_user_mode = NdbDictionary::Table::SingleUserModeReadWrite;
+  }
 #endif /* HAVE_NDB_BINLOG */
 
   DBUG_PRINT("table", ("name: %s", m_tabname));  
   tab.setName(m_tabname);
   tab.setLogging(!(create_info->options & HA_LEX_CREATE_TMP_TABLE));    
+  tab.setSingleUserMode(single_user_mode);
 
   // Save frm data for this table
   if (readfrm(name, &data, &length))
@@ -5569,6 +5574,7 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb,
     {
       ndb_table_id= h->m_table->getObjectId();
       ndb_table_version= h->m_table->getObjectVersion();
+      DBUG_PRINT("info", ("success 1"));
     }
     else
     {
@@ -5582,6 +5588,7 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb,
           break;
       }
       res= ndb_to_mysql_error(&dict->getNdbError());
+      DBUG_PRINT("info", ("error(1) %u", res));
     }
     h->release_metadata(thd, ndb);
   }
@@ -5598,6 +5605,8 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb,
         {
           ndb_table_id= ndbtab_g.get_table()->getObjectId();
           ndb_table_version= ndbtab_g.get_table()->getObjectVersion();
+          DBUG_PRINT("info", ("success 2"));
+          break;
         }
         else
         {
@@ -5617,8 +5626,8 @@ ha_ndbcluster::delete_table(ha_ndbcluster *h, Ndb *ndb,
           }
         }
       }
-      else
       res= ndb_to_mysql_error(&dict->getNdbError());
+      DBUG_PRINT("info", ("error(2) %u", res));
       break;
     }
   }

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -51,7 +51,7 @@
 
 #define ERR_RETURN(a,b) \
 {\
-   DBUG_PRINT("exit", ("error %d", (a).code));\
+   DBUG_PRINT("exit", ("error %d  return %d", (a).code, b));\
    DBUG_RETURN(b);\
 }